From a83ea9bdb6af5519849fdeaf96cff1e911a81e41 Mon Sep 17 00:00:00 2001 From: Perplexity Computer Date: Sat, 21 Mar 2026 22:43:21 +0000 Subject: [PATCH] feat: FastAPI Morrowind harness + SOUL.md framework (#821, #854) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- docs/soul-framework/authoring-guide.md | 97 ++++++ docs/soul-framework/role-extensions.md | 115 +++++++ docs/soul-framework/template.md | 109 +++++++ src/dashboard/app.py | 2 + src/infrastructure/morrowind/__init__.py | 1 + src/infrastructure/morrowind/api.py | 206 ++++++++++++ src/infrastructure/soul/__init__.py | 19 ++ src/infrastructure/soul/loader.py | 221 +++++++++++++ src/infrastructure/soul/validator.py | 226 +++++++++++++ src/infrastructure/soul/versioning.py | 143 +++++++++ tests/test_morrowind_api.py | 227 +++++++++++++ tests/test_soul_framework.py | 386 +++++++++++++++++++++++ 12 files changed, 1752 insertions(+) create mode 100644 docs/soul-framework/authoring-guide.md create mode 100644 docs/soul-framework/role-extensions.md create mode 100644 docs/soul-framework/template.md create mode 100644 src/infrastructure/morrowind/api.py create mode 100644 src/infrastructure/soul/__init__.py create mode 100644 src/infrastructure/soul/loader.py create mode 100644 src/infrastructure/soul/validator.py create mode 100644 src/infrastructure/soul/versioning.py create mode 100644 tests/test_morrowind_api.py create mode 100644 tests/test_soul_framework.py diff --git a/docs/soul-framework/authoring-guide.md b/docs/soul-framework/authoring-guide.md new file mode 100644 index 00000000..750c68b8 --- /dev/null +++ b/docs/soul-framework/authoring-guide.md @@ -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: +- **3–7 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 2–4 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 (3–6 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. diff --git a/docs/soul-framework/role-extensions.md b/docs/soul-framework/role-extensions.md new file mode 100644 index 00000000..e565cc5a --- /dev/null +++ b/docs/soul-framework/role-extensions.md @@ -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-.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) +``` diff --git a/docs/soul-framework/template.md b/docs/soul-framework/template.md new file mode 100644 index 00000000..ae5d2cbd --- /dev/null +++ b/docs/soul-framework/template.md @@ -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. diff --git a/src/dashboard/app.py b/src/dashboard/app.py index 43c980fa..a5c145c8 100644 --- a/src/dashboard/app.py +++ b/src/dashboard/app.py @@ -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") diff --git a/src/infrastructure/morrowind/__init__.py b/src/infrastructure/morrowind/__init__.py index 80a3a5b4..d0a98ca4 100644 --- a/src/infrastructure/morrowind/__init__.py +++ b/src/infrastructure/morrowind/__init__.py @@ -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 diff --git a/src/infrastructure/morrowind/api.py b/src/infrastructure/morrowind/api.py new file mode 100644 index 00000000..60eff604 --- /dev/null +++ b/src/infrastructure/morrowind/api.py @@ -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, + ) diff --git a/src/infrastructure/soul/__init__.py b/src/infrastructure/soul/__init__.py new file mode 100644 index 00000000..767e2c91 --- /dev/null +++ b/src/infrastructure/soul/__init__.py @@ -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", +] diff --git a/src/infrastructure/soul/loader.py b/src/infrastructure/soul/loader.py new file mode 100644 index 00000000..7fd9545e --- /dev/null +++ b/src/infrastructure/soul/loader.py @@ -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 diff --git a/src/infrastructure/soul/validator.py b/src/infrastructure/soul/validator.py new file mode 100644 index 00000000..134e51b3 --- /dev/null +++ b/src/infrastructure/soul/validator.py @@ -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 diff --git a/src/infrastructure/soul/versioning.py b/src/infrastructure/soul/versioning.py new file mode 100644 index 00000000..a843ef2f --- /dev/null +++ b/src/infrastructure/soul/versioning.py @@ -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() diff --git a/tests/test_morrowind_api.py b/tests/test_morrowind_api.py new file mode 100644 index 00000000..52c51f33 --- /dev/null +++ b/tests/test_morrowind_api.py @@ -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 diff --git a/tests/test_soul_framework.py b/tests/test_soul_framework.py new file mode 100644 index 00000000..ea6c07c9 --- /dev/null +++ b/tests/test_soul_framework.py @@ -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