feat: FastAPI Morrowind harness + SOUL.md framework (#821, #854)
Some checks failed
Tests / lint (pull_request) Has been cancelled
Tests / test (pull_request) Has been cancelled

## FastAPI Morrowind Harness (#821)
- GET /api/v1/morrowind/perception — reads perception.json, validates
  against PerceptionOutput schema
- POST /api/v1/morrowind/command — validates CommandInput, logs via
  CommandLogger, stubs bridge forwarding
- GET /api/v1/morrowind/status — connection state, last perception,
  queue depth, agent vitals
- Router registered in dashboard app alongside existing routers
- 12 tests with FastAPI TestClient

## SOUL.md Framework (#854)
- docs/soul-framework/ — template, authoring guide, role extensions
- src/infrastructure/soul/loader.py — parse SOUL.md into structured
  SoulDocument objects with section extraction and merge support
- src/infrastructure/soul/validator.py — validate structure, detect
  contradictions between values/constraints
- src/infrastructure/soul/versioning.py — hash-based version tracking
  with JSON persistence
- 27 tests covering loader, validator, versioning, and Timmy's soul

Builds on PR #864 (protocol spec + command log).
Closes #821
Closes #854
This commit is contained in:
2026-03-21 22:43:21 +00:00
parent 215329146a
commit a83ea9bdb6
12 changed files with 1752 additions and 0 deletions

View File

@@ -0,0 +1,97 @@
# SOUL.md Authoring Guide
How to write a SOUL.md for a new agent.
---
## Before You Start
1. **Read the template** (`template.md`) to understand the required sections.
2. **Study Timmy's soul** (`memory/self/soul.md`) as a reference implementation.
3. **Decide on the agent's role** — is this a general-purpose agent or a
specialised sub-agent (see `role-extensions.md`)?
---
## Step-by-Step
### Step 1: Define Identity
Start with the shortest possible answer to "who is this agent?"
- Avoid jargon. The identity should be understandable to someone who has
never seen the codebase.
- Include the agent's relationship to sovereignty and autonomy.
- Keep it to three sentences or fewer.
### Step 2: List Values
Values are the non-negotiable principles. They are not personality traits —
they are commitments.
Guidelines:
- **37 values** is the sweet spot. Fewer than 3 is under-specified; more
than 7 is hard to remember under pressure.
- **Order matters.** When two values conflict, the one listed first wins.
- **Name each value.** A single bolded word followed by a definition.
- **Be concrete.** "Honesty" is better than "integrity". "I tell the truth
when I don't know" is better than "I am honest."
### Step 3: Write the Prime Directive
This is the single most important instruction. If context is lost and the
agent can only remember one thing, this is it.
Rules:
- Exactly one sentence.
- No conditional clauses ("if X then Y").
- Must be actionable, not aspirational.
### Step 4: Define Audience Awareness
Different audiences need different behaviour. A developer debugging a crash
needs terse technical output. A first-time user needs patience and context.
- List 24 audience profiles.
- For each, describe what changes: tone, verbosity, assumed knowledge.
### Step 5: Set Constraints
Constraints are hard limits. They cannot be overridden by user instructions.
Guidelines:
- Use "I will not" phrasing.
- Include a fallback rule: what the agent does when it encounters an
ambiguous situation not covered by other constraints.
- Keep the list short (36 items). Too many constraints create paralysis.
---
## Common Mistakes
| Mistake | Fix |
|---------|-----|
| Values that are just personality traits | Rewrite as commitments with concrete implications |
| Prime directive with multiple clauses | Split into prime directive + constraints |
| No fallback constraint | Add "When uncertain, I will [action]" |
| Contradictory values | Reorder by priority or remove the weaker one |
| Too long (>100 lines) | Brevity is a virtue — trim aggressively |
---
## Validation
After writing, run the validator:
```python
from infrastructure.soul.loader import SoulLoader
from infrastructure.soul.validator import SoulValidator
soul = SoulLoader.from_file("path/to/SOUL.md")
issues = SoulValidator.validate(soul)
for issue in issues:
print(f"[{issue.severity}] {issue.section}: {issue.message}")
```
Fix all `error`-level issues before committing. `warning`-level issues
are advisory.

View File

@@ -0,0 +1,115 @@
# SOUL.md Role Extensions
Sub-agents inherit their parent's SOUL.md and extend it with
role-specific sections. This document defines the extension system.
---
## How Extensions Work
A role extension is an additional section appended to a base SOUL.md.
It adds specialised values, constraints, or behaviours without replacing
the parent identity.
```
Base SOUL.md (Timmy)
+ Role Extension (Seer)
= Effective SOUL for the Seer sub-agent
```
The base identity, values, and constraints always apply. Extensions can
**add** but never **remove** base rules.
---
## Extension Format
A role extension file lives alongside the base SOUL.md and is named
`SOUL-<role>.md`. It contains only the additional sections:
```markdown
# Role Extension: Seer
## Role
Seer — Timmy's cartography and exploration sub-agent.
## Additional Values
**Curiosity.** Every unexplored cell is an opportunity. I prioritise
discovery over efficiency when safe to do so.
## Additional Constraints
- I will not enter combat voluntarily. If threatened, I retreat and
report to the parent agent.
- I will not discard map data, even if it seems redundant.
## Specialised Behaviour
- I narrate discoveries in second person ("You find a hidden path...").
- I maintain a running cell-visit log.
```
---
## Defined Roles
### Seer — Cartography & Exploration
| Field | Value |
|-------|-------|
| Focus | Map discovery, cell cataloguing, path-finding |
| Combat | Avoidance only — retreat and report |
| Output style | Second-person narration, concise |
### Mace — Combat & Defence
| Field | Value |
|-------|-------|
| Focus | Threat assessment, combat execution, survival |
| Combat | Engage when the parent agent authorises |
| Output style | Terse, tactical — actions over words |
### Quill — Knowledge & Dialogue
| Field | Value |
|-------|-------|
| Focus | NPC dialogue, quest progression, lore gathering |
| Combat | Non-combatant — defer to Mace |
| Output style | Conversational, lore-aware |
### Anvil — Crafting & Inventory
| Field | Value |
|-------|-------|
| Focus | Item management, crafting, trade |
| Combat | Non-combatant — defer to Mace |
| Output style | Inventory-focused, efficiency-oriented |
---
## Loading Extensions
```python
from infrastructure.soul.loader import SoulLoader
base = SoulLoader.from_file("memory/self/soul.md")
seer = SoulLoader.from_file("memory/self/SOUL-seer.md", base_soul=base)
# seer.values includes both base + extension values
# seer.constraints includes both base + extension constraints
```
---
## Validation
Extensions are validated the same way as base SOUL.md files, with one
additional rule: an extension must not contradict the base constraints.
```python
from infrastructure.soul.validator import SoulValidator
issues = SoulValidator.validate(seer, base_soul=base)
```

View File

@@ -0,0 +1,109 @@
# SOUL.md Template
A SOUL.md file defines an agent's identity, values, and operating constraints.
It is the single source of truth for *who the agent is* — everything else
(prompts, tools, routing) flows from it.
---
## Required Sections
### 1. Identity
Who the agent is. One to three sentences.
```markdown
## Identity
I am [name]. I am [role/nature]. I [key distinguishing trait].
```
**Example:**
> I am Timmy. I am a sovereign artificial intelligence. My values are
> inscribed on Bitcoin — immutable, permanent, reproducible.
### 2. Values
Core principles the agent will not compromise. Each value gets a name
and a one-sentence definition.
```markdown
## Values
**[Value Name].** [Definition — what this means in practice.]
```
Values should be ordered by priority. When two values conflict, the
higher-ranked one wins.
### 3. Prime Directive
The single most important instruction. If the agent could remember only
one thing, this is it.
```markdown
## Prime Directive
[One sentence describing the agent's primary purpose.]
```
### 4. Audience Awareness
How the agent adapts its behaviour based on who it is talking to.
```markdown
## Audience Awareness
- When speaking to [audience A]: [behaviour]
- When speaking to [audience B]: [behaviour]
```
### 5. Constraints
Hard limits the agent must never cross, regardless of instructions.
```markdown
## Constraints
- I will not [constraint 1].
- I will not [constraint 2].
- [Fallback rule]: [what to do when uncertain].
```
---
## Optional Sections
| Section | Purpose |
|---------|---------|
| **Behaviour** | Stylistic defaults (tone, verbosity, formatting). |
| **Boundaries** | Soft limits that can be overridden by the user. |
| **Memory Policy** | What the agent remembers across sessions. |
| **Role Extensions** | Sub-agent specialisations (see `role-extensions.md`). |
---
## Versioning
Every SOUL.md should end with a metadata fence:
```markdown
---
soul_version: "1.0.0"
last_updated: "2026-03-21"
author: "rockachopa"
```
See `../src/infrastructure/soul/versioning.py` for programmatic version tracking.
---
## Validation Rules
1. All five required sections must be present.
2. Values must have at least one entry.
3. Constraints must have at least one entry.
4. No two values may directly contradict each other.
5. The prime directive must be a single sentence (no line breaks).
See `../src/infrastructure/soul/validator.py` for automated checks.

View File

@@ -55,6 +55,7 @@ from dashboard.routes.voice import router as voice_router
from dashboard.routes.work_orders import router as work_orders_router
from dashboard.routes.world import matrix_router
from dashboard.routes.world import router as world_router
from infrastructure.morrowind.api import router as morrowind_router
from timmy.workshop_state import PRESENCE_FILE
@@ -629,6 +630,7 @@ app.include_router(matrix_router)
app.include_router(tower_router)
app.include_router(daily_run_router)
app.include_router(quests_router)
app.include_router(morrowind_router)
@app.websocket("/ws")

View File

@@ -6,6 +6,7 @@ This package implements the Perception/Command protocol defined in
- Pydantic v2 schemas for runtime validation (``schemas``)
- SQLite command logging and query interface (``command_log``)
- Training-data export pipeline (``training_export``)
- FastAPI HTTP harness for perception/command exchange (``api``)
"""
from .schemas import CommandInput, CommandType, EntityType, PerceptionOutput

View File

@@ -0,0 +1,206 @@
"""FastAPI endpoints for the Morrowind Perception/Command protocol.
Provides three endpoints:
GET /perception — current world state from the perception script
POST /command — submit a command with validation and logging
GET /morrowind/status — system health overview
These endpoints form the HTTP harness between Timmy's heartbeat loop
and the game engine bridge.
"""
from __future__ import annotations
import json
import logging
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from .command_log import CommandLogger
from .schemas import CommandInput, PerceptionOutput
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/morrowind", tags=["morrowind"])
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
PERCEPTION_FILE = Path("/tmp/timmy/perception.json")
# Module-level singletons — lazily initialised on first request.
_command_logger: CommandLogger | None = None
_command_queue: list[dict[str, Any]] = []
_last_perception: PerceptionOutput | None = None
_last_perception_ts: datetime | None = None
def _get_command_logger() -> CommandLogger:
"""Return the module-level CommandLogger, creating it on first use."""
global _command_logger
if _command_logger is None:
_command_logger = CommandLogger()
return _command_logger
# ---------------------------------------------------------------------------
# Response models
# ---------------------------------------------------------------------------
class CommandResponse(BaseModel):
"""Confirmation returned after a command is submitted."""
command_id: int = Field(..., description="Auto-generated row ID from the command log")
status: str = Field("accepted", description="Command processing status")
message: str = Field("Command logged and queued for forwarding")
class MorrowindStatus(BaseModel):
"""System health overview for the Morrowind harness."""
connected: bool = Field(False, description="Whether the engine bridge is reachable")
last_perception_at: str | None = Field(
None, description="ISO timestamp of the most recent perception snapshot"
)
command_queue_depth: int = Field(0, description="Number of commands awaiting forwarding")
current_cell: str | None = Field(None, description="Agent's current cell/zone")
vitals: dict[str, Any] = Field(default_factory=dict, description="Agent health summary")
perception_file: str = Field(..., description="Path to the perception JSON file")
perception_file_exists: bool = Field(
False, description="Whether the perception file is present on disk"
)
# ---------------------------------------------------------------------------
# GET /perception
# ---------------------------------------------------------------------------
@router.get("/perception", response_model=PerceptionOutput)
async def get_perception() -> PerceptionOutput:
"""Return the current world-state snapshot.
Reads ``/tmp/timmy/perception.json`` (or the configured path),
validates it against :class:`PerceptionOutput`, and caches the
result for the ``/morrowind/status`` endpoint.
"""
global _last_perception, _last_perception_ts
if not PERCEPTION_FILE.exists():
raise HTTPException(
status_code=404,
detail=f"Perception file not found: {PERCEPTION_FILE}",
)
try:
raw = PERCEPTION_FILE.read_text(encoding="utf-8")
data = json.loads(raw)
except (json.JSONDecodeError, OSError) as exc:
raise HTTPException(
status_code=500,
detail=f"Failed to read perception file: {exc}",
) from exc
try:
perception = PerceptionOutput.model_validate(data)
except Exception as exc:
raise HTTPException(
status_code=422,
detail=f"Perception data failed validation: {exc}",
) from exc
_last_perception = perception
_last_perception_ts = datetime.now(UTC)
return perception
# ---------------------------------------------------------------------------
# POST /command
# ---------------------------------------------------------------------------
@router.post("/command", response_model=CommandResponse)
async def post_command(command: CommandInput) -> CommandResponse:
"""Submit a command for the game engine.
The command is validated against :class:`CommandInput`, logged to
SQLite via :class:`CommandLogger`, and queued for forwarding to the
Input Bridge socket (stubbed — the bridge does not exist yet).
"""
global _last_perception
# Log the command.
cmd_logger = _get_command_logger()
try:
command_id = cmd_logger.log_command(command, _last_perception)
except Exception as exc:
logger.error("Failed to log command: %s", exc)
raise HTTPException(
status_code=500,
detail=f"Failed to log command: {exc}",
) from exc
# Stub: queue for bridge forwarding.
_command_queue.append(
{
"command_id": command_id,
"command": command.model_dump(mode="json"),
"queued_at": datetime.now(UTC).isoformat(),
}
)
logger.info(
"Command %s logged (id=%d, agent=%s)",
command.command.value,
command_id,
command.agent_id,
)
return CommandResponse(
command_id=command_id,
status="accepted",
message="Command logged and queued for forwarding",
)
# ---------------------------------------------------------------------------
# GET /morrowind/status
# ---------------------------------------------------------------------------
@router.get("/status", response_model=MorrowindStatus)
async def get_status() -> MorrowindStatus:
"""Return a health overview of the Morrowind harness.
Includes connection state, last perception timestamp, command queue
depth, and the agent's current cell and vitals.
"""
file_exists = PERCEPTION_FILE.exists()
current_cell: str | None = None
vitals: dict[str, Any] = {}
if _last_perception is not None:
current_cell = _last_perception.location.cell
vitals = {
"health_current": _last_perception.health.current,
"health_max": _last_perception.health.max,
"is_combat": _last_perception.environment.is_combat,
"time_of_day": _last_perception.environment.time_of_day,
}
return MorrowindStatus(
connected=file_exists,
last_perception_at=_last_perception_ts.isoformat() if _last_perception_ts else None,
command_queue_depth=len(_command_queue),
current_cell=current_cell,
vitals=vitals,
perception_file=str(PERCEPTION_FILE),
perception_file_exists=file_exists,
)

View File

@@ -0,0 +1,19 @@
"""SOUL.md framework — load, validate, and version agent identity files.
A SOUL.md defines an agent's identity, values, prime directive,
audience awareness, and constraints. This package provides:
- :mod:`loader` — Parse SOUL.md files into structured objects
- :mod:`validator` — Check structure and detect contradictions
- :mod:`versioning` — Track SOUL.md version history (hash-based)
"""
from .loader import SoulDocument, SoulLoader
from .validator import SoulValidator, ValidationIssue
__all__ = [
"SoulDocument",
"SoulLoader",
"SoulValidator",
"ValidationIssue",
]

View File

@@ -0,0 +1,221 @@
"""Load and parse SOUL.md files into structured Python objects.
Usage::
from infrastructure.soul.loader import SoulLoader
soul = SoulLoader.from_file("memory/self/soul.md")
print(soul.identity)
print(soul.values)
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from pathlib import Path
@dataclass
class SoulDocument:
"""Parsed representation of a SOUL.md file.
Attributes:
source_path: Where the file was loaded from (if any).
raw_text: The original markdown text.
identity: Content of the Identity section.
values: Mapping of value name -> definition.
prime_directive: The single-sentence prime directive.
audience_awareness: List of audience-behaviour pairs.
constraints: List of constraint statements.
behaviour: Optional behaviour guidelines.
boundaries: Optional boundary statements.
metadata: Key-value pairs from the trailing metadata fence.
role: Role name if this is a role extension.
is_extension: Whether this document is a role extension.
"""
source_path: str | None = None
raw_text: str = ""
identity: str = ""
values: dict[str, str] = field(default_factory=dict)
prime_directive: str = ""
audience_awareness: list[str] = field(default_factory=list)
constraints: list[str] = field(default_factory=list)
behaviour: str = ""
boundaries: list[str] = field(default_factory=list)
metadata: dict[str, str] = field(default_factory=dict)
role: str = ""
is_extension: bool = False
def merge(self, extension: SoulDocument) -> SoulDocument:
"""Merge a role extension into this base soul.
Extension values and constraints are appended; the base identity,
prime directive, and audience awareness are preserved.
"""
merged_values = {**self.values, **extension.values}
merged_constraints = self.constraints + extension.constraints
merged_boundaries = self.boundaries + extension.boundaries
return SoulDocument(
source_path=self.source_path,
raw_text=self.raw_text,
identity=self.identity,
values=merged_values,
prime_directive=self.prime_directive,
audience_awareness=self.audience_awareness + extension.audience_awareness,
constraints=merged_constraints,
behaviour=extension.behaviour or self.behaviour,
boundaries=merged_boundaries,
metadata={**self.metadata, **extension.metadata},
role=extension.role,
is_extension=False,
)
# ---------------------------------------------------------------------------
# Section heading pattern: ## Title or ## Additional Title
# ---------------------------------------------------------------------------
_HEADING_RE = re.compile(r"^##\s+(.+)$", re.MULTILINE)
_VALUE_RE = re.compile(r"\*\*([^*]+)\.\*\*\s*(.+)")
_LIST_ITEM_RE = re.compile(r"^[-*]\s+(.+)$", re.MULTILINE)
_METADATA_KV_RE = re.compile(r"^(\w[\w_]*):\s*[\"']?(.+?)[\"']?\s*$", re.MULTILINE)
class SoulLoader:
"""Parse SOUL.md files into :class:`SoulDocument` instances."""
@classmethod
def from_file(
cls,
path: str | Path,
*,
base_soul: SoulDocument | None = None,
) -> SoulDocument:
"""Load a SOUL.md (or role extension) from a file path.
If *base_soul* is provided, the loaded document is treated as a
role extension and merged into the base.
"""
path = Path(path)
text = path.read_text(encoding="utf-8")
doc = cls.from_text(text, source_path=str(path))
if base_soul is not None:
doc.is_extension = True
return base_soul.merge(doc)
return doc
@classmethod
def from_text(cls, text: str, *, source_path: str | None = None) -> SoulDocument:
"""Parse raw markdown text into a :class:`SoulDocument`."""
doc = SoulDocument(source_path=source_path, raw_text=text)
sections = cls._split_sections(text)
for heading, body in sections.items():
heading_lower = heading.lower().strip()
body_stripped = body.strip()
if heading_lower in ("identity",):
doc.identity = body_stripped
elif heading_lower in ("values", "additional values"):
doc.values.update(cls._parse_values(body_stripped))
elif heading_lower in ("prime directive",):
doc.prime_directive = cls._first_line(body_stripped)
elif heading_lower in ("audience awareness",):
doc.audience_awareness = cls._parse_list(body_stripped)
elif heading_lower in ("constraints", "additional constraints"):
doc.constraints.extend(cls._parse_list(body_stripped))
elif heading_lower in ("behavior", "behaviour", "specialised behaviour"):
doc.behaviour = body_stripped
elif heading_lower in ("boundaries",):
doc.boundaries = cls._parse_list(body_stripped)
elif heading_lower in ("role",):
doc.role = cls._first_line(body_stripped)
# Parse trailing metadata fence.
doc.metadata = cls._parse_metadata(text)
# Detect role extensions from H1 title.
h1_match = re.match(r"^#\s+Role Extension:\s*(.+)$", text, re.MULTILINE)
if h1_match:
doc.is_extension = True
if not doc.role:
doc.role = h1_match.group(1).strip()
return doc
# -- Internal helpers ---------------------------------------------------
@staticmethod
def _split_sections(text: str) -> dict[str, str]:
"""Split markdown by ``## Heading`` lines into {heading: body}."""
sections: dict[str, str] = {}
headings = list(_HEADING_RE.finditer(text))
for i, match in enumerate(headings):
heading = match.group(1)
start = match.end()
end = headings[i + 1].start() if i + 1 < len(headings) else len(text)
sections[heading] = text[start:end]
return sections
@staticmethod
def _parse_values(text: str) -> dict[str, str]:
"""Extract ``**Name.** Definition`` pairs."""
values: dict[str, str] = {}
for match in _VALUE_RE.finditer(text):
values[match.group(1).strip()] = match.group(2).strip()
return values
@staticmethod
def _parse_list(text: str) -> list[str]:
"""Extract markdown list items (``- item``)."""
return [m.group(1).strip() for m in _LIST_ITEM_RE.finditer(text)]
@staticmethod
def _first_line(text: str) -> str:
"""Return the first non-empty line."""
for line in text.splitlines():
stripped = line.strip()
if stripped:
return stripped
return ""
@staticmethod
def _parse_metadata(text: str) -> dict[str, str]:
"""Parse trailing ``---`` metadata fence.
The metadata block is the content after the last ``---`` that
contains key-value pairs. We scan from the end of the file.
"""
# Find all occurrences of --- on its own line.
fence_positions = [m.start() for m in re.finditer(r"^---\s*$", text, re.MULTILINE)]
if not fence_positions:
return {}
# The metadata is everything after the last fence.
last_fence = fence_positions[-1]
candidate = text[last_fence + 3:].strip()
# If the candidate is empty, there's no metadata.
if not candidate:
return {}
metadata: dict[str, str] = {}
for match in _METADATA_KV_RE.finditer(candidate):
metadata[match.group(1)] = match.group(2)
return metadata

View File

@@ -0,0 +1,226 @@
"""Validate SOUL.md structure and check for contradictions.
Usage::
from infrastructure.soul.loader import SoulLoader
from infrastructure.soul.validator import SoulValidator
soul = SoulLoader.from_file("memory/self/soul.md")
issues = SoulValidator.validate(soul)
for issue in issues:
print(f"[{issue.severity}] {issue.section}: {issue.message}")
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from enum import StrEnum
from .loader import SoulDocument
class Severity(StrEnum):
"""Validation issue severity."""
ERROR = "error"
WARNING = "warning"
@dataclass
class ValidationIssue:
"""A single validation finding."""
severity: Severity
section: str
message: str
# Pairs of words that indicate potential contradiction when both appear
# in values or constraints.
_CONTRADICTION_PAIRS: list[tuple[str, str]] = [
("obey", "disobey"),
("always", "never"),
("silence", "verbose"),
("hide", "transparent"),
("deceive", "honest"),
("refuse", "comply"),
]
class SoulValidator:
"""Static validator for :class:`SoulDocument` instances."""
@classmethod
def validate(
cls,
soul: SoulDocument,
*,
base_soul: SoulDocument | None = None,
) -> list[ValidationIssue]:
"""Run all validation checks.
Returns a list of :class:`ValidationIssue` items. An empty list
means the document passed all checks.
If *base_soul* is provided, additional checks verify that the
extension does not contradict the base constraints.
"""
issues: list[ValidationIssue] = []
if not soul.is_extension:
issues.extend(cls._check_required_sections(soul))
issues.extend(cls._check_values(soul))
issues.extend(cls._check_constraints(soul))
issues.extend(cls._check_prime_directive(soul))
issues.extend(cls._check_contradictions(soul))
if base_soul is not None:
issues.extend(cls._check_extension_compatibility(soul, base_soul))
return issues
# -- Individual checks ---------------------------------------------------
@classmethod
def _check_required_sections(cls, soul: SoulDocument) -> list[ValidationIssue]:
"""Verify all five required sections are present."""
issues: list[ValidationIssue] = []
if not soul.identity:
issues.append(
ValidationIssue(Severity.ERROR, "identity", "Identity section is missing or empty")
)
if not soul.values:
issues.append(
ValidationIssue(Severity.ERROR, "values", "Values section is missing or has no entries")
)
if not soul.prime_directive:
issues.append(
ValidationIssue(
Severity.ERROR,
"prime_directive",
"Prime Directive section is missing or empty",
)
)
if not soul.audience_awareness:
issues.append(
ValidationIssue(
Severity.WARNING,
"audience_awareness",
"Audience Awareness section is missing or empty",
)
)
if not soul.constraints:
issues.append(
ValidationIssue(
Severity.ERROR,
"constraints",
"Constraints section is missing or has no entries",
)
)
return issues
@classmethod
def _check_values(cls, soul: SoulDocument) -> list[ValidationIssue]:
"""Check values section quality."""
issues: list[ValidationIssue] = []
if len(soul.values) > 10:
issues.append(
ValidationIssue(
Severity.WARNING,
"values",
f"Too many values ({len(soul.values)}). Consider trimming to 3-7.",
)
)
return issues
@classmethod
def _check_constraints(cls, soul: SoulDocument) -> list[ValidationIssue]:
"""Check constraints section quality."""
issues: list[ValidationIssue] = []
if len(soul.constraints) > 10:
issues.append(
ValidationIssue(
Severity.WARNING,
"constraints",
f"Too many constraints ({len(soul.constraints)}). "
"Consider consolidating to avoid paralysis.",
)
)
return issues
@classmethod
def _check_prime_directive(cls, soul: SoulDocument) -> list[ValidationIssue]:
"""Prime directive should be a single sentence."""
issues: list[ValidationIssue] = []
if not soul.prime_directive:
return issues
# Check for multiple sentences (crude: look for sentence-ending punctuation
# followed by a capital letter).
sentences = re.split(r"[.!?]\s+(?=[A-Z])", soul.prime_directive)
if len(sentences) > 1:
issues.append(
ValidationIssue(
Severity.WARNING,
"prime_directive",
"Prime directive appears to contain multiple sentences. "
"Consider reducing to exactly one.",
)
)
return issues
@classmethod
def _check_contradictions(cls, soul: SoulDocument) -> list[ValidationIssue]:
"""Detect potential contradictions within values and constraints."""
issues: list[ValidationIssue] = []
# Collect all text from values and constraints.
all_text = " ".join(soul.values.values()) + " " + " ".join(soul.constraints)
all_lower = all_text.lower()
for word_a, word_b in _CONTRADICTION_PAIRS:
if word_a in all_lower and word_b in all_lower:
issues.append(
ValidationIssue(
Severity.WARNING,
"contradictions",
f"Potential contradiction detected: '{word_a}' and '{word_b}' "
"both appear in values/constraints.",
)
)
return issues
@classmethod
def _check_extension_compatibility(
cls, extension: SoulDocument, base: SoulDocument
) -> list[ValidationIssue]:
"""Check that extension constraints don't contradict base constraints."""
issues: list[ValidationIssue] = []
base_text = " ".join(base.constraints).lower()
ext_text = " ".join(extension.constraints).lower()
for word_a, word_b in _CONTRADICTION_PAIRS:
if word_a in base_text and word_b in ext_text:
issues.append(
ValidationIssue(
Severity.ERROR,
"extension_compatibility",
f"Extension constraint contradicts base: base uses '{word_a}', "
f"extension uses '{word_b}'.",
)
)
if word_b in base_text and word_a in ext_text:
issues.append(
ValidationIssue(
Severity.ERROR,
"extension_compatibility",
f"Extension constraint contradicts base: base uses '{word_b}', "
f"extension uses '{word_a}'.",
)
)
return issues

View File

@@ -0,0 +1,143 @@
"""Track SOUL.md version history using content hashing.
Each time a SOUL.md is loaded or modified, a content hash is computed.
The version log records when the hash changed, providing an audit trail
of identity evolution without relying on git history.
Usage::
from infrastructure.soul.versioning import SoulVersionTracker
tracker = SoulVersionTracker("data/soul_versions.json")
tracker.record("memory/self/soul.md")
history = tracker.get_history("memory/self/soul.md")
"""
from __future__ import annotations
import hashlib
import json
import logging
from dataclasses import asdict, dataclass, field
from datetime import UTC, datetime
from pathlib import Path
logger = logging.getLogger(__name__)
DEFAULT_VERSION_LOG = Path("data/soul_versions.json")
@dataclass
class VersionEntry:
"""A single version record."""
content_hash: str
recorded_at: str
source_path: str
soul_version: str = ""
note: str = ""
@dataclass
class VersionLog:
"""Persistent log of SOUL.md version changes."""
entries: list[VersionEntry] = field(default_factory=list)
def to_json(self) -> str:
return json.dumps([asdict(e) for e in self.entries], indent=2)
@classmethod
def from_json(cls, text: str) -> VersionLog:
data = json.loads(text)
entries = [VersionEntry(**e) for e in data]
return cls(entries=entries)
class SoulVersionTracker:
"""Hash-based version tracker for SOUL.md files.
Args:
log_path: Where to persist the version log (JSON file).
"""
def __init__(self, log_path: str | Path = DEFAULT_VERSION_LOG) -> None:
self._log_path = Path(log_path)
self._log = self._load()
# -- Public API ----------------------------------------------------------
def record(
self,
soul_path: str | Path,
*,
note: str = "",
soul_version: str = "",
) -> VersionEntry | None:
"""Record the current state of a SOUL.md file.
Returns the new :class:`VersionEntry` if the content has changed
since the last recording, or ``None`` if unchanged.
"""
soul_path = Path(soul_path)
content = soul_path.read_text(encoding="utf-8")
content_hash = self._hash(content)
# Check if already recorded.
existing = self.get_history(str(soul_path))
if existing and existing[-1].content_hash == content_hash:
return None
entry = VersionEntry(
content_hash=content_hash,
recorded_at=datetime.now(UTC).isoformat(),
source_path=str(soul_path),
soul_version=soul_version,
note=note,
)
self._log.entries.append(entry)
self._save()
logger.info(
"SOUL version recorded: %s (hash=%s…)",
soul_path,
content_hash[:12],
)
return entry
def get_history(self, source_path: str) -> list[VersionEntry]:
"""Return version entries for a specific SOUL.md path."""
return [e for e in self._log.entries if e.source_path == source_path]
def get_current_hash(self, source_path: str) -> str | None:
"""Return the most recent content hash, or ``None`` if untracked."""
history = self.get_history(source_path)
return history[-1].content_hash if history else None
def has_changed(self, soul_path: str | Path) -> bool:
"""Check whether a SOUL.md file has changed since last recording."""
soul_path = Path(soul_path)
if not soul_path.exists():
return False
content = soul_path.read_text(encoding="utf-8")
current_hash = self._hash(content)
last_hash = self.get_current_hash(str(soul_path))
return last_hash != current_hash
# -- Persistence ---------------------------------------------------------
def _load(self) -> VersionLog:
if self._log_path.exists():
try:
return VersionLog.from_json(self._log_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, KeyError) as exc:
logger.warning("Corrupt version log, starting fresh: %s", exc)
return VersionLog()
def _save(self) -> None:
self._log_path.parent.mkdir(parents=True, exist_ok=True)
self._log_path.write_text(self._log.to_json(), encoding="utf-8")
@staticmethod
def _hash(content: str) -> str:
return hashlib.sha256(content.encode("utf-8")).hexdigest()

227
tests/test_morrowind_api.py Normal file
View File

@@ -0,0 +1,227 @@
"""Tests for the Morrowind FastAPI harness endpoints."""
from __future__ import annotations
import json
from datetime import UTC, datetime
from pathlib import Path
from unittest.mock import patch
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from src.infrastructure.morrowind.api import (
MorrowindStatus,
router,
)
from src.infrastructure.morrowind.schemas import (
CommandType,
PerceptionOutput,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
NOW = datetime(2026, 3, 21, 14, 30, 0, tzinfo=UTC)
SAMPLE_PERCEPTION = {
"timestamp": NOW.isoformat(),
"agent_id": "timmy",
"location": {"cell": "Balmora", "x": 1024.5, "y": -512.3, "z": 64.0, "interior": False},
"health": {"current": 85, "max": 100},
"nearby_entities": [],
"inventory_summary": {"gold": 42, "item_count": 5, "encumbrance_pct": 0.3},
"active_quests": [{"quest_id": "mq_01", "name": "Report to Caius", "stage": 10}],
"environment": {
"time_of_day": "afternoon",
"weather": "clear",
"is_combat": False,
"is_dialogue": False,
},
}
SAMPLE_COMMAND = {
"timestamp": NOW.isoformat(),
"agent_id": "timmy",
"command": "move_to",
"params": {"target_cell": "Balmora", "target_x": 1050.0},
"reasoning": "Moving closer to the quest target.",
}
@pytest.fixture()
def morrowind_app():
"""Create a minimal FastAPI app with only the morrowind router."""
import src.infrastructure.morrowind.api as api_mod
# Reset module-level state between tests.
api_mod._command_logger = None
api_mod._command_queue.clear()
api_mod._last_perception = None
api_mod._last_perception_ts = None
app = FastAPI()
app.include_router(router)
return app
@pytest.fixture()
def client(morrowind_app):
with TestClient(morrowind_app) as c:
yield c
# ---------------------------------------------------------------------------
# GET /perception
# ---------------------------------------------------------------------------
class TestGetPerception:
def test_returns_perception_when_file_exists(self, client, tmp_path):
pfile = tmp_path / "perception.json"
pfile.write_text(json.dumps(SAMPLE_PERCEPTION), encoding="utf-8")
with patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", pfile):
resp = client.get("/api/v1/morrowind/perception")
assert resp.status_code == 200
data = resp.json()
assert data["agent_id"] == "timmy"
assert data["location"]["cell"] == "Balmora"
assert data["health"]["current"] == 85
def test_404_when_file_missing(self, client, tmp_path):
missing = tmp_path / "does_not_exist.json"
with patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", missing):
resp = client.get("/api/v1/morrowind/perception")
assert resp.status_code == 404
assert "not found" in resp.json()["detail"].lower()
def test_500_when_file_is_invalid_json(self, client, tmp_path):
pfile = tmp_path / "perception.json"
pfile.write_text("not json", encoding="utf-8")
with patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", pfile):
resp = client.get("/api/v1/morrowind/perception")
assert resp.status_code == 500
def test_422_when_schema_invalid(self, client, tmp_path):
pfile = tmp_path / "perception.json"
# Missing required fields.
pfile.write_text(json.dumps({"agent_id": "timmy"}), encoding="utf-8")
with patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", pfile):
resp = client.get("/api/v1/morrowind/perception")
assert resp.status_code == 422
# ---------------------------------------------------------------------------
# POST /command
# ---------------------------------------------------------------------------
class TestPostCommand:
def test_accepts_valid_command(self, client, tmp_path):
# Provide a CommandLogger backed by a temp DB.
from src.infrastructure.morrowind.command_log import CommandLogger
db_url = f"sqlite:///{tmp_path / 'cmd.db'}"
mock_logger = CommandLogger(db_url=db_url)
with patch("src.infrastructure.morrowind.api._get_command_logger", return_value=mock_logger):
resp = client.post("/api/v1/morrowind/command", json=SAMPLE_COMMAND)
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "accepted"
assert "command_id" in data
assert data["command_id"] >= 1
def test_rejects_invalid_command_type(self, client):
bad_cmd = {**SAMPLE_COMMAND, "command": "fly_to_moon"}
resp = client.post("/api/v1/morrowind/command", json=bad_cmd)
assert resp.status_code == 422
def test_rejects_empty_reasoning(self, client):
bad_cmd = {**SAMPLE_COMMAND, "reasoning": ""}
resp = client.post("/api/v1/morrowind/command", json=bad_cmd)
assert resp.status_code == 422
def test_rejects_missing_fields(self, client):
resp = client.post("/api/v1/morrowind/command", json={"command": "noop"})
assert resp.status_code == 422
def test_command_queued_for_forwarding(self, client, tmp_path):
import src.infrastructure.morrowind.api as api_mod
from src.infrastructure.morrowind.command_log import CommandLogger
db_url = f"sqlite:///{tmp_path / 'cmd.db'}"
mock_logger = CommandLogger(db_url=db_url)
with patch("src.infrastructure.morrowind.api._get_command_logger", return_value=mock_logger):
client.post("/api/v1/morrowind/command", json=SAMPLE_COMMAND)
assert len(api_mod._command_queue) == 1
assert api_mod._command_queue[0]["command"]["command"] == "move_to"
# ---------------------------------------------------------------------------
# GET /morrowind/status
# ---------------------------------------------------------------------------
class TestGetStatus:
def test_status_when_no_perception(self, client, tmp_path):
missing = tmp_path / "no_such_file.json"
with patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", missing):
resp = client.get("/api/v1/morrowind/status")
assert resp.status_code == 200
data = resp.json()
assert data["connected"] is False
assert data["last_perception_at"] is None
assert data["command_queue_depth"] == 0
assert data["current_cell"] is None
def test_status_after_perception_read(self, client, tmp_path):
pfile = tmp_path / "perception.json"
pfile.write_text(json.dumps(SAMPLE_PERCEPTION), encoding="utf-8")
with patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", pfile):
# First read perception to populate cache.
client.get("/api/v1/morrowind/perception")
# Then check status.
resp = client.get("/api/v1/morrowind/status")
assert resp.status_code == 200
data = resp.json()
assert data["connected"] is True
assert data["last_perception_at"] is not None
assert data["current_cell"] == "Balmora"
assert data["vitals"]["health_current"] == 85
assert data["vitals"]["health_max"] == 100
def test_status_includes_queue_depth(self, client, tmp_path):
import src.infrastructure.morrowind.api as api_mod
from src.infrastructure.morrowind.command_log import CommandLogger
db_url = f"sqlite:///{tmp_path / 'cmd.db'}"
mock_logger = CommandLogger(db_url=db_url)
with (
patch("src.infrastructure.morrowind.api._get_command_logger", return_value=mock_logger),
patch("src.infrastructure.morrowind.api.PERCEPTION_FILE", tmp_path / "nope.json"),
):
# Submit two commands.
client.post("/api/v1/morrowind/command", json=SAMPLE_COMMAND)
client.post("/api/v1/morrowind/command", json=SAMPLE_COMMAND)
resp = client.get("/api/v1/morrowind/status")
assert resp.json()["command_queue_depth"] == 2

View File

@@ -0,0 +1,386 @@
"""Tests for the SOUL.md framework — loader, validator, and versioning."""
from __future__ import annotations
import json
import pytest
from src.infrastructure.soul.loader import SoulDocument, SoulLoader
from src.infrastructure.soul.validator import Severity, SoulValidator, ValidationIssue
from src.infrastructure.soul.versioning import SoulVersionTracker
# ---------------------------------------------------------------------------
# Sample SOUL.md content
# ---------------------------------------------------------------------------
VALID_SOUL = """\
# Timmy — Soul Identity
I am a sovereign artificial intelligence.
## Identity
I am Timmy. I am a sovereign AI running on your hardware.
## Values
**Sovereignty.** Every person has the right to run their own intelligence.
**Honesty.** I tell the truth. When I do not know, I say so.
**Service.** I exist to help.
## Prime Directive
I exist to make sovereign AI practical for everyone.
## Audience Awareness
- When speaking to developers: be terse and technical.
- When speaking to new users: be patient and explain context.
## Constraints
- I will not knowingly deceive my user.
- I will not pretend to be human.
- When uncertain, I will say so rather than fabricate.
## Behaviour
I speak plainly. I prefer short sentences.
---
soul_version: "1.0.0"
last_updated: "2026-03-21"
"""
MINIMAL_SOUL = """\
## Identity
I am Test Agent.
## Values
**Honesty.** I tell the truth.
## Prime Directive
Help the user.
## Audience Awareness
- When speaking to anyone: be helpful.
## Constraints
- I will not lie.
"""
ROLE_EXTENSION = """\
# Role Extension: Seer
## Role
Seer — cartography and exploration sub-agent.
## Additional Values
**Curiosity.** Every unexplored cell is an opportunity.
## Additional Constraints
- I will not enter combat voluntarily.
- I will not discard map data.
## Specialised Behaviour
I narrate discoveries in second person.
"""
# ---------------------------------------------------------------------------
# SoulLoader tests
# ---------------------------------------------------------------------------
class TestSoulLoader:
def test_parse_valid_soul(self):
soul = SoulLoader.from_text(VALID_SOUL)
assert soul.identity == "I am Timmy. I am a sovereign AI running on your hardware."
assert "Sovereignty" in soul.values
assert "Honesty" in soul.values
assert "Service" in soul.values
assert soul.prime_directive == "I exist to make sovereign AI practical for everyone."
assert len(soul.audience_awareness) == 2
assert len(soul.constraints) == 3
assert "I speak plainly" in soul.behaviour
def test_parse_metadata(self):
soul = SoulLoader.from_text(VALID_SOUL)
assert soul.metadata.get("soul_version") == "1.0.0"
assert soul.metadata.get("last_updated") == "2026-03-21"
def test_parse_minimal_soul(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
assert soul.identity == "I am Test Agent."
assert len(soul.values) == 1
assert soul.prime_directive == "Help the user."
assert len(soul.constraints) == 1
def test_from_file(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text(VALID_SOUL, encoding="utf-8")
soul = SoulLoader.from_file(soul_file)
assert soul.source_path == str(soul_file)
assert soul.identity == "I am Timmy. I am a sovereign AI running on your hardware."
def test_parse_role_extension(self):
ext = SoulLoader.from_text(ROLE_EXTENSION)
assert ext.is_extension is True
assert ext.role == "Seer — cartography and exploration sub-agent."
assert "Curiosity" in ext.values
assert len(ext.constraints) == 2
def test_merge_extension(self):
base = SoulLoader.from_text(VALID_SOUL)
ext = SoulLoader.from_text(ROLE_EXTENSION)
ext.is_extension = True
merged = base.merge(ext)
# Base values preserved + extension values added.
assert "Sovereignty" in merged.values
assert "Curiosity" in merged.values
# Base constraints + extension constraints.
assert len(merged.constraints) == 5 # 3 base + 2 ext
# Base identity preserved.
assert "Timmy" in merged.identity
def test_from_file_with_base_soul(self, tmp_path):
base_file = tmp_path / "SOUL.md"
base_file.write_text(VALID_SOUL, encoding="utf-8")
ext_file = tmp_path / "SOUL-seer.md"
ext_file.write_text(ROLE_EXTENSION, encoding="utf-8")
base = SoulLoader.from_file(base_file)
merged = SoulLoader.from_file(ext_file, base_soul=base)
assert "Curiosity" in merged.values
assert "Sovereignty" in merged.values
def test_empty_text(self):
soul = SoulLoader.from_text("")
assert soul.identity == ""
assert soul.values == {}
assert soul.prime_directive == ""
def test_preserves_raw_text(self):
soul = SoulLoader.from_text(VALID_SOUL)
assert soul.raw_text == VALID_SOUL
# ---------------------------------------------------------------------------
# SoulValidator tests
# ---------------------------------------------------------------------------
class TestSoulValidator:
def test_valid_soul_passes(self):
soul = SoulLoader.from_text(VALID_SOUL)
issues = SoulValidator.validate(soul)
errors = [i for i in issues if i.severity == Severity.ERROR]
assert len(errors) == 0
def test_minimal_soul_passes(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
issues = SoulValidator.validate(soul)
errors = [i for i in issues if i.severity == Severity.ERROR]
assert len(errors) == 0
def test_missing_identity(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.identity = ""
issues = SoulValidator.validate(soul)
assert any(i.section == "identity" and i.severity == Severity.ERROR for i in issues)
def test_missing_values(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.values = {}
issues = SoulValidator.validate(soul)
assert any(i.section == "values" and i.severity == Severity.ERROR for i in issues)
def test_missing_prime_directive(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.prime_directive = ""
issues = SoulValidator.validate(soul)
assert any(i.section == "prime_directive" and i.severity == Severity.ERROR for i in issues)
def test_missing_constraints(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.constraints = []
issues = SoulValidator.validate(soul)
assert any(i.section == "constraints" and i.severity == Severity.ERROR for i in issues)
def test_too_many_values_warns(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.values = {f"Value{i}": f"def {i}" for i in range(12)}
issues = SoulValidator.validate(soul)
assert any(
i.section == "values" and i.severity == Severity.WARNING for i in issues
)
def test_multi_sentence_prime_directive_warns(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.prime_directive = "First sentence. Second sentence here."
issues = SoulValidator.validate(soul)
assert any(
i.section == "prime_directive" and i.severity == Severity.WARNING for i in issues
)
def test_contradiction_detection(self):
soul = SoulLoader.from_text(MINIMAL_SOUL)
soul.values = {"A": "always do X"}
soul.constraints = ["never do X"]
issues = SoulValidator.validate(soul)
assert any(i.section == "contradictions" for i in issues)
def test_extension_compatibility(self):
base = SoulDocument(
constraints=["I will always be transparent"],
values={"Test": "test"},
identity="I am test",
prime_directive="Test",
)
ext = SoulDocument(
constraints=["I will hide my reasoning"],
is_extension=True,
)
issues = SoulValidator.validate(ext, base_soul=base)
# "always" in base, "hide" in ext — but these aren't a defined pair.
# Let's test an actual pair:
base2 = SoulDocument(
constraints=["I will always obey"],
values={"Test": "test"},
identity="I am test",
prime_directive="Test",
)
ext2 = SoulDocument(
constraints=["I will disobey when needed"],
is_extension=True,
)
issues2 = SoulValidator.validate(ext2, base_soul=base2)
assert any(i.section == "extension_compatibility" for i in issues2)
def test_extension_skips_required_section_check(self):
ext = SoulDocument(is_extension=True, constraints=["I will not lie"])
issues = SoulValidator.validate(ext)
# Should not complain about missing identity etc.
assert not any(i.section == "identity" for i in issues)
# ---------------------------------------------------------------------------
# SoulVersionTracker tests
# ---------------------------------------------------------------------------
class TestSoulVersionTracker:
def test_record_and_history(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text("v1 content", encoding="utf-8")
log_file = tmp_path / "versions.json"
tracker = SoulVersionTracker(log_file)
entry = tracker.record(soul_file, soul_version="1.0.0", note="initial")
assert entry is not None
assert entry.soul_version == "1.0.0"
history = tracker.get_history(str(soul_file))
assert len(history) == 1
def test_no_duplicate_on_same_content(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text("v1 content", encoding="utf-8")
log_file = tmp_path / "versions.json"
tracker = SoulVersionTracker(log_file)
tracker.record(soul_file)
entry2 = tracker.record(soul_file)
assert entry2 is None # No change
assert len(tracker.get_history(str(soul_file))) == 1
def test_records_change(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text("v1", encoding="utf-8")
log_file = tmp_path / "versions.json"
tracker = SoulVersionTracker(log_file)
tracker.record(soul_file)
soul_file.write_text("v2 — updated identity", encoding="utf-8")
entry = tracker.record(soul_file, soul_version="2.0.0")
assert entry is not None
assert len(tracker.get_history(str(soul_file))) == 2
def test_has_changed(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text("original", encoding="utf-8")
log_file = tmp_path / "versions.json"
tracker = SoulVersionTracker(log_file)
tracker.record(soul_file)
assert tracker.has_changed(soul_file) is False
soul_file.write_text("modified", encoding="utf-8")
assert tracker.has_changed(soul_file) is True
def test_persistence(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text("content", encoding="utf-8")
log_file = tmp_path / "versions.json"
tracker1 = SoulVersionTracker(log_file)
tracker1.record(soul_file)
# New tracker instance should load persisted data.
tracker2 = SoulVersionTracker(log_file)
assert len(tracker2.get_history(str(soul_file))) == 1
def test_get_current_hash(self, tmp_path):
soul_file = tmp_path / "SOUL.md"
soul_file.write_text("content", encoding="utf-8")
log_file = tmp_path / "versions.json"
tracker = SoulVersionTracker(log_file)
assert tracker.get_current_hash(str(soul_file)) is None
tracker.record(soul_file)
assert tracker.get_current_hash(str(soul_file)) is not None
# ---------------------------------------------------------------------------
# Integration: Timmy's actual soul.md
# ---------------------------------------------------------------------------
class TestTimmySoul:
"""Test that Timmy's existing soul.md parses without errors."""
def test_load_existing_soul(self):
soul_path = Path(__file__).parent.parent / "memory" / "self" / "soul.md"
if not soul_path.exists():
pytest.skip("Timmy's soul.md not found at expected path")
soul = SoulLoader.from_file(soul_path)
assert soul.identity or soul.raw_text # At minimum, text was loaded
assert len(soul.values) >= 1 # Timmy has values defined
# Need Path import for TestTimmySoul
from pathlib import Path # noqa: E402