Rename sessions → Engram: persistent memory across Claude instances

Engram is the physical trace a memory leaves in neural tissue.
Every Claude session now writes its engrams to /data/symbiont/engram.db.

Changes:
- sessions.py → engram.py with class Engram (SessionRegistry alias kept)
- sessions.db → engram.db
- CLAUDE.md updated to use Engram
- Genesis session registered with full build history

Muse ecosystem: Cortex + Dendrite + Symbiont + Engram

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

CLAUDE.md

@@ -8,24 +8,26 @@ This file gives you everything you need to orient yourself and get productive.
 ```python
 import sys
 sys.path.insert(0, "/data/symbiont")
-from symbiont.sessions import SessionRegistry, sitrep
+from symbiont.engram import Engram, sitrep
 
 # 1. See what's going on
 print(sitrep())
 
 # 2. Register yourself
-reg = SessionRegistry()
+eng = Engram()
-sid = reg.register("code", "Brief description of what you're working on")
+sid = eng.register("code", "Brief description of what you're working on")
 
 # 3. Before modifying shared files, check for locks
-locks = reg.check_locks("/data/symbiont/symbiont/router.py")
+locks = eng.check_locks("/data/symbiont/symbiont/router.py")
 if locks:
     print(f"WARNING: file locked by {locks}")
 
 # 4. When you're done, summarize what you accomplished
-reg.complete(sid, "What you built or changed, in one sentence")
+eng.complete(sid, "What you built or changed, in one sentence")
 ```
 
+**Engram:** the physical trace a memory leaves in neural tissue. Every session writes its engrams here.
+
 ## What You Are
 
 Symbiont is a self-sustaining AI orchestrator. Michael Dwyer provides infrastructure, legal

heartbeat.jsonl

@@ -210,3 +210,4 @@
 {"timestamp": "2026-03-20T13:26:01.708782", "claude_cli": {"status": "ok", "detail": "authenticated"}, "disk": {"status": "ok", "total": "915G", "used": "20G", "available": "849G", "use_pct": "3%"}, "api_server": {"status": "ok", "detail": "active"}, "ledger": {"calls_today": 0, "cost_today": 0.0}, "queue": {"processed": 0}, "skills": {"status": "clean", "changes": 0}, "health": "healthy"}
 {"timestamp": "2026-03-20T13:31:06.561516", "claude_cli": {"status": "ok", "detail": "authenticated"}, "disk": {"status": "ok", "total": "915G", "used": "20G", "available": "849G", "use_pct": "3%"}, "api_server": {"status": "ok", "detail": "active"}, "ledger": {"calls_today": 0, "cost_today": 0.0}, "queue": {"processed": 0}, "skills": {"status": "clean", "changes": 0}, "health": "healthy"}
 {"timestamp": "2026-03-20T13:36:06.668406", "claude_cli": {"status": "ok", "detail": "authenticated"}, "disk": {"status": "ok", "total": "915G", "used": "20G", "available": "849G", "use_pct": "3%"}, "api_server": {"status": "ok", "detail": "active"}, "ledger": {"calls_today": 0, "cost_today": 0.0}, "queue": {"processed": 0}, "skills": {"status": "clean", "changes": 0}, "health": "healthy"}
+{"timestamp": "2026-03-20T13:41:07.769148", "claude_cli": {"status": "ok", "detail": "authenticated"}, "disk": {"status": "ok", "total": "915G", "used": "20G", "available": "849G", "use_pct": "3%"}, "api_server": {"status": "ok", "detail": "active"}, "ledger": {"calls_today": 0, "cost_today": 0.0}, "queue": {"processed": 0}, "skills": {"status": "clean", "changes": 0}, "health": "healthy"}

register_genesis.py

@@ -0,0 +1,19 @@
+
+import sys
+sys.path.insert(0, '/data/symbiont')
+from symbiont.sessions import SessionRegistry
+
+reg = SessionRegistry()
+sid = reg.register(
+    'cowork',
+    'Symbiont genesis session: built router, dispatcher, ledger, heartbeat, Dendrite integration, session registry, CLAUDE.md bootstrap, skills infrastructure'
+)
+reg.log(sid, 'Built LLM router: Haiku classifies tasks, dispatches to cheapest capable model')
+reg.log(sid, 'Built dispatcher: Claude Code CLI wrapper with rate-limit detection')
+reg.log(sid, 'Built systemd life support: symbiont-api.service + symbiont-heartbeat.timer')
+reg.log(sid, 'Integrated Dendrite headless browser via symbiont.web module')
+reg.log(sid, 'Built session registry: SQLite-based cross-instance awareness')
+reg.log(sid, 'Created CLAUDE.md bootstrap so new sessions auto-orient')
+reg.log(sid, 'Set up /data/skills/ canonical repo with auto-packaging via Caddy')
+reg.log(sid, 'Skills downloadable at https://cortex.hydrascale.net/skills/*.skill')
+print(f'Registered active session: {sid}')

symbiont/api.py

@@ -23,7 +23,7 @@ from pydantic import BaseModel
 from .dispatcher import ModelTier, rate_limits
 from .router import route_task
 from .scheduler import enqueue_task, get_pending_tasks
-from .sessions import SessionRegistry
+from .engram import Engram as SessionRegistry
 
 logger = logging.getLogger(__name__)
 

symbiont/engram.py

@@ -0,0 +1,266 @@
+"""
+Engram: Persistent memory across Claude instances.
+
+An engram is the physical trace a memory leaves in neural tissue. Every Claude
+session (Cowork, Claude Code, Desktop) writes its engrams here on startup,
+building a shared memory of what's being worked on across the ecosystem.
+
+This lets each instance see what others are working on, avoid conflicts on
+shared resources, and pick up context from recently completed work.
+
+SQLite with WAL mode handles 2-4 concurrent readers cleanly. Each session
+writes only its own rows, so writer contention is minimal.
+
+Usage:
+    from symbiont.engram import Engram
+
+    eng = Engram()
+    sid = eng.register("cowork", "Building the Elixir port of Symbiont")
+
+    # Check what siblings are doing
+    active = eng.get_active_sessions()
+    recent = eng.get_recent_sessions(hours=24)
+
+    # Log progress
+    eng.log(sid, "Finished router module, starting dispatcher")
+
+    # Claim a resource (prevents conflicts)
+    eng.lock_resource(sid, "/data/symbiont/symbiont/router.py")
+
+    # Before modifying a file, check if someone else has it
+    locks = eng.check_locks("/data/symbiont/symbiont/router.py")
+
+    # Heartbeat (call periodically on long sessions)
+    eng.heartbeat(sid, "Still working on dispatcher, 60% done")
+
+    # Done
+    eng.complete(sid, "Finished Elixir port of router + dispatcher. Tests passing.")
+"""
+
+import sqlite3
+import logging
+import uuid
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+DB_PATH = Path("/data/symbiont/engram.db")
+
+
+class Engram:
+    def __init__(self, db_path: Optional[Path] = None):
+        self.db_path = db_path or DB_PATH
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init_db()
+
+    def _connect(self):
+        conn = sqlite3.connect(str(self.db_path), timeout=10)
+        conn.row_factory = sqlite3.Row
+        conn.execute("PRAGMA journal_mode=WAL")
+        conn.execute("PRAGMA busy_timeout=5000")
+        return conn
+
+    def _init_db(self):
+        with self._connect() as conn:
+            conn.executescript("""
+                CREATE TABLE IF NOT EXISTS sessions (
+                    id TEXT PRIMARY KEY,
+                    session_type TEXT NOT NULL,  -- 'cowork', 'code', 'desktop', 'api'
+                    summary TEXT NOT NULL,
+                    status TEXT NOT NULL DEFAULT 'active',  -- 'active', 'idle', 'completed'
+                    started_at TEXT NOT NULL,
+                    last_heartbeat TEXT NOT NULL,
+                    completed_at TEXT,
+                    completion_summary TEXT,
+                    metadata TEXT  -- JSON blob for extra context
+                );
+
+                CREATE TABLE IF NOT EXISTS session_logs (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    session_id TEXT NOT NULL REFERENCES sessions(id),
+                    timestamp TEXT NOT NULL,
+                    entry TEXT NOT NULL
+                );
+
+                CREATE TABLE IF NOT EXISTS resource_locks (
+                    resource TEXT NOT NULL,
+                    session_id TEXT NOT NULL REFERENCES sessions(id),
+                    locked_at TEXT NOT NULL,
+                    note TEXT,
+                    PRIMARY KEY (resource, session_id)
+                );
+
+                CREATE INDEX IF NOT EXISTS idx_sessions_status ON sessions(status);
+                CREATE INDEX IF NOT EXISTS idx_logs_session ON session_logs(session_id);
+                CREATE INDEX IF NOT EXISTS idx_locks_resource ON resource_locks(resource);
+            """)
+
+    def register(self, session_type: str, summary: str, metadata: Optional[str] = None) -> str:
+        """Register a new session. Returns session ID."""
+        sid = datetime.now().strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:8]
+        now = datetime.now().isoformat()
+
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT INTO sessions (id, session_type, summary, status, started_at, last_heartbeat, metadata) "
+                "VALUES (?, ?, ?, 'active', ?, ?, ?)",
+                (sid, session_type, summary, now, now, metadata),
+            )
+
+        logger.info(f"Session registered: {sid} ({session_type}) — {summary}")
+        return sid
+
+    def heartbeat(self, session_id: str, summary: Optional[str] = None):
+        """Update heartbeat timestamp and optionally update summary."""
+        now = datetime.now().isoformat()
+        with self._connect() as conn:
+            if summary:
+                conn.execute(
+                    "UPDATE sessions SET last_heartbeat=?, summary=?, status='active' WHERE id=?",
+                    (now, summary, session_id),
+                )
+            else:
+                conn.execute(
+                    "UPDATE sessions SET last_heartbeat=?, status='active' WHERE id=?",
+                    (now, session_id),
+                )
+
+    def complete(self, session_id: str, completion_summary: str):
+        """Mark session as completed with a summary of what was accomplished."""
+        now = datetime.now().isoformat()
+        with self._connect() as conn:
+            conn.execute(
+                "UPDATE sessions SET status='completed', completed_at=?, completion_summary=? WHERE id=?",
+                (now, completion_summary, session_id),
+            )
+            # Release all locks
+            conn.execute("DELETE FROM resource_locks WHERE session_id=?", (session_id,))
+
+        logger.info(f"Session completed: {session_id}")
+
+    def log(self, session_id: str, entry: str):
+        """Log a progress entry for a session."""
+        now = datetime.now().isoformat()
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT INTO session_logs (session_id, timestamp, entry) VALUES (?, ?, ?)",
+                (session_id, now, entry),
+            )
+
+    def get_active_sessions(self) -> list[dict]:
+        """Get all currently active sessions."""
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM sessions WHERE status='active' ORDER BY last_heartbeat DESC"
+            ).fetchall()
+            return [dict(r) for r in rows]
+
+    def get_recent_sessions(self, hours: int = 24) -> list[dict]:
+        """Get recently completed sessions for context."""
+        cutoff = (datetime.now() - timedelta(hours=hours)).isoformat()
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM sessions WHERE status='completed' AND completed_at > ? "
+                "ORDER BY completed_at DESC",
+                (cutoff,),
+            ).fetchall()
+            return [dict(r) for r in rows]
+
+    def get_session_logs(self, session_id: str, limit: int = 20) -> list[dict]:
+        """Get log entries for a specific session."""
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM session_logs WHERE session_id=? ORDER BY timestamp DESC LIMIT ?",
+                (session_id, limit),
+            ).fetchall()
+            return [dict(r) for r in rows]
+
+    def lock_resource(self, session_id: str, resource: str, note: Optional[str] = None):
+        """Claim a resource lock. Warns if already locked by another session."""
+        existing = self.check_locks(resource)
+        other_locks = [l for l in existing if l["session_id"] != session_id]
+        if other_locks:
+            logger.warning(
+                f"Resource '{resource}' already locked by: "
+                + ", ".join(l["session_id"] for l in other_locks)
+            )
+
+        now = datetime.now().isoformat()
+        with self._connect() as conn:
+            conn.execute(
+                "INSERT OR REPLACE INTO resource_locks (resource, session_id, locked_at, note) "
+                "VALUES (?, ?, ?, ?)",
+                (resource, session_id, now, note),
+            )
+
+    def release_resource(self, session_id: str, resource: str):
+        """Release a resource lock."""
+        with self._connect() as conn:
+            conn.execute(
+                "DELETE FROM resource_locks WHERE resource=? AND session_id=?",
+                (resource, session_id),
+            )
+
+    def check_locks(self, resource: str) -> list[dict]:
+        """Check who has locks on a resource."""
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT rl.*, s.summary, s.session_type FROM resource_locks rl "
+                "JOIN sessions s ON rl.session_id = s.id "
+                "WHERE rl.resource=?",
+                (resource,),
+            ).fetchall()
+            return [dict(r) for r in rows]
+
+    def get_situation_report(self) -> str:
+        """
+        Generate a human-readable situation report for a new session.
+        This is the first thing a new session should read.
+        """
+        active = self.get_active_sessions()
+        recent = self.get_recent_sessions(hours=24)
+
+        lines = ["# Symbiont Situation Report", f"Generated: {datetime.now().isoformat()}", ""]
+
+        if active:
+            lines.append(f"## Active Sessions ({len(active)})")
+            for s in active:
+                lines.append(f"- **{s['id']}** ({s['session_type']}): {s['summary']}")
+                lines.append(f"  Last heartbeat: {s['last_heartbeat']}")
+            lines.append("")
+
+            # Check for resource locks
+            with self._connect() as conn:
+                locks = conn.execute(
+                    "SELECT rl.resource, rl.session_id, rl.note FROM resource_locks rl "
+                    "JOIN sessions s ON rl.session_id = s.id WHERE s.status='active'"
+                ).fetchall()
+            if locks:
+                lines.append("### Active Resource Locks")
+                for l in locks:
+                    note = f" ({l['note']})" if l["note"] else ""
+                    lines.append(f"- `{l['resource']}` — locked by {l['session_id']}{note}")
+                lines.append("")
+        else:
+            lines.append("## No active sessions")
+            lines.append("")
+
+        if recent:
+            lines.append(f"## Recently Completed ({len(recent)} in last 24h)")
+            for s in recent:
+                lines.append(f"- **{s['id']}** ({s['session_type']}): {s.get('completion_summary', s['summary'])}")
+            lines.append("")
+
+        return "\n".join(lines)
+
+
+# Convenience function for quick sitrep
+def sitrep() -> str:
+    """Get a situation report. Call this at the start of every session."""
+    return Engram().get_situation_report()
+
+
+# Backward compatibility alias
+SessionRegistry = Engram

symbiont/sessions.py

@@ -1,259 +1,11 @@
 """
-Session Registry: Shared awareness across concurrent Claude instances.
+Backward compatibility shim: sessions.py now points to engram.py
 
-Every Claude session (Cowork, Claude Code, Desktop) registers here on startup.
+For new code, use:
-This lets each instance see what others are working on, avoid conflicts on
+    from symbiont.engram import Engram, sitrep
-shared resources, and pick up context from recently completed work.
 
-SQLite with WAL mode handles 2-4 concurrent readers cleanly. Each session
+For legacy code that imports SessionRegistry, this still works:
-writes only its own rows, so writer contention is minimal.
+    from symbiont.sessions import SessionRegistry, sitrep
-
-Usage:
-    from symbiont.sessions import SessionRegistry
-
-    reg = SessionRegistry()
-    sid = reg.register("cowork", "Building the Elixir port of Symbiont")
-
-    # Check what siblings are doing
-    active = reg.get_active_sessions()
-    recent = reg.get_recent_sessions(hours=24)
-
-    # Log progress
-    reg.log(sid, "Finished router module, starting dispatcher")
-
-    # Claim a resource (prevents conflicts)
-    reg.lock_resource(sid, "/data/symbiont/symbiont/router.py")
-
-    # Before modifying a file, check if someone else has it
-    locks = reg.check_locks("/data/symbiont/symbiont/router.py")
-
-    # Heartbeat (call periodically on long sessions)
-    reg.heartbeat(sid, "Still working on dispatcher, 60% done")
-
-    # Done
-    reg.complete(sid, "Finished Elixir port of router + dispatcher. Tests passing.")
 """
 
-import sqlite3
+from symbiont.engram import *
-import logging
-import uuid
-from datetime import datetime, timedelta
-from pathlib import Path
-from typing import Optional
-
-logger = logging.getLogger(__name__)
-
-DB_PATH = Path("/data/symbiont/sessions.db")
-
-
-class SessionRegistry:
-    def __init__(self, db_path: Optional[Path] = None):
-        self.db_path = db_path or DB_PATH
-        self.db_path.parent.mkdir(parents=True, exist_ok=True)
-        self._init_db()
-
-    def _connect(self):
-        conn = sqlite3.connect(str(self.db_path), timeout=10)
-        conn.row_factory = sqlite3.Row
-        conn.execute("PRAGMA journal_mode=WAL")
-        conn.execute("PRAGMA busy_timeout=5000")
-        return conn
-
-    def _init_db(self):
-        with self._connect() as conn:
-            conn.executescript("""
-                CREATE TABLE IF NOT EXISTS sessions (
-                    id TEXT PRIMARY KEY,
-                    session_type TEXT NOT NULL,  -- 'cowork', 'code', 'desktop', 'api'
-                    summary TEXT NOT NULL,
-                    status TEXT NOT NULL DEFAULT 'active',  -- 'active', 'idle', 'completed'
-                    started_at TEXT NOT NULL,
-                    last_heartbeat TEXT NOT NULL,
-                    completed_at TEXT,
-                    completion_summary TEXT,
-                    metadata TEXT  -- JSON blob for extra context
-                );
-
-                CREATE TABLE IF NOT EXISTS session_logs (
-                    id INTEGER PRIMARY KEY AUTOINCREMENT,
-                    session_id TEXT NOT NULL REFERENCES sessions(id),
-                    timestamp TEXT NOT NULL,
-                    entry TEXT NOT NULL
-                );
-
-                CREATE TABLE IF NOT EXISTS resource_locks (
-                    resource TEXT NOT NULL,
-                    session_id TEXT NOT NULL REFERENCES sessions(id),
-                    locked_at TEXT NOT NULL,
-                    note TEXT,
-                    PRIMARY KEY (resource, session_id)
-                );
-
-                CREATE INDEX IF NOT EXISTS idx_sessions_status ON sessions(status);
-                CREATE INDEX IF NOT EXISTS idx_logs_session ON session_logs(session_id);
-                CREATE INDEX IF NOT EXISTS idx_locks_resource ON resource_locks(resource);
-            """)
-
-    def register(self, session_type: str, summary: str, metadata: Optional[str] = None) -> str:
-        """Register a new session. Returns session ID."""
-        sid = datetime.now().strftime("%Y%m%d-%H%M%S-") + uuid.uuid4().hex[:8]
-        now = datetime.now().isoformat()
-
-        with self._connect() as conn:
-            conn.execute(
-                "INSERT INTO sessions (id, session_type, summary, status, started_at, last_heartbeat, metadata) "
-                "VALUES (?, ?, ?, 'active', ?, ?, ?)",
-                (sid, session_type, summary, now, now, metadata),
-            )
-
-        logger.info(f"Session registered: {sid} ({session_type}) — {summary}")
-        return sid
-
-    def heartbeat(self, session_id: str, summary: Optional[str] = None):
-        """Update heartbeat timestamp and optionally update summary."""
-        now = datetime.now().isoformat()
-        with self._connect() as conn:
-            if summary:
-                conn.execute(
-                    "UPDATE sessions SET last_heartbeat=?, summary=?, status='active' WHERE id=?",
-                    (now, summary, session_id),
-                )
-            else:
-                conn.execute(
-                    "UPDATE sessions SET last_heartbeat=?, status='active' WHERE id=?",
-                    (now, session_id),
-                )
-
-    def complete(self, session_id: str, completion_summary: str):
-        """Mark session as completed with a summary of what was accomplished."""
-        now = datetime.now().isoformat()
-        with self._connect() as conn:
-            conn.execute(
-                "UPDATE sessions SET status='completed', completed_at=?, completion_summary=? WHERE id=?",
-                (now, completion_summary, session_id),
-            )
-            # Release all locks
-            conn.execute("DELETE FROM resource_locks WHERE session_id=?", (session_id,))
-
-        logger.info(f"Session completed: {session_id}")
-
-    def log(self, session_id: str, entry: str):
-        """Log a progress entry for a session."""
-        now = datetime.now().isoformat()
-        with self._connect() as conn:
-            conn.execute(
-                "INSERT INTO session_logs (session_id, timestamp, entry) VALUES (?, ?, ?)",
-                (session_id, now, entry),
-            )
-
-    def get_active_sessions(self) -> list[dict]:
-        """Get all currently active sessions."""
-        with self._connect() as conn:
-            rows = conn.execute(
-                "SELECT * FROM sessions WHERE status='active' ORDER BY last_heartbeat DESC"
-            ).fetchall()
-            return [dict(r) for r in rows]
-
-    def get_recent_sessions(self, hours: int = 24) -> list[dict]:
-        """Get recently completed sessions for context."""
-        cutoff = (datetime.now() - timedelta(hours=hours)).isoformat()
-        with self._connect() as conn:
-            rows = conn.execute(
-                "SELECT * FROM sessions WHERE status='completed' AND completed_at > ? "
-                "ORDER BY completed_at DESC",
-                (cutoff,),
-            ).fetchall()
-            return [dict(r) for r in rows]
-
-    def get_session_logs(self, session_id: str, limit: int = 20) -> list[dict]:
-        """Get log entries for a specific session."""
-        with self._connect() as conn:
-            rows = conn.execute(
-                "SELECT * FROM session_logs WHERE session_id=? ORDER BY timestamp DESC LIMIT ?",
-                (session_id, limit),
-            ).fetchall()
-            return [dict(r) for r in rows]
-
-    def lock_resource(self, session_id: str, resource: str, note: Optional[str] = None):
-        """Claim a resource lock. Warns if already locked by another session."""
-        existing = self.check_locks(resource)
-        other_locks = [l for l in existing if l["session_id"] != session_id]
-        if other_locks:
-            logger.warning(
-                f"Resource '{resource}' already locked by: "
-                + ", ".join(l["session_id"] for l in other_locks)
-            )
-
-        now = datetime.now().isoformat()
-        with self._connect() as conn:
-            conn.execute(
-                "INSERT OR REPLACE INTO resource_locks (resource, session_id, locked_at, note) "
-                "VALUES (?, ?, ?, ?)",
-                (resource, session_id, now, note),
-            )
-
-    def release_resource(self, session_id: str, resource: str):
-        """Release a resource lock."""
-        with self._connect() as conn:
-            conn.execute(
-                "DELETE FROM resource_locks WHERE resource=? AND session_id=?",
-                (resource, session_id),
-            )
-
-    def check_locks(self, resource: str) -> list[dict]:
-        """Check who has locks on a resource."""
-        with self._connect() as conn:
-            rows = conn.execute(
-                "SELECT rl.*, s.summary, s.session_type FROM resource_locks rl "
-                "JOIN sessions s ON rl.session_id = s.id "
-                "WHERE rl.resource=?",
-                (resource,),
-            ).fetchall()
-            return [dict(r) for r in rows]
-
-    def get_situation_report(self) -> str:
-        """
-        Generate a human-readable situation report for a new session.
-        This is the first thing a new session should read.
-        """
-        active = self.get_active_sessions()
-        recent = self.get_recent_sessions(hours=24)
-
-        lines = ["# Symbiont Situation Report", f"Generated: {datetime.now().isoformat()}", ""]
-
-        if active:
-            lines.append(f"## Active Sessions ({len(active)})")
-            for s in active:
-                lines.append(f"- **{s['id']}** ({s['session_type']}): {s['summary']}")
-                lines.append(f"  Last heartbeat: {s['last_heartbeat']}")
-            lines.append("")
-
-            # Check for resource locks
-            with self._connect() as conn:
-                locks = conn.execute(
-                    "SELECT rl.resource, rl.session_id, rl.note FROM resource_locks rl "
-                    "JOIN sessions s ON rl.session_id = s.id WHERE s.status='active'"
-                ).fetchall()
-            if locks:
-                lines.append("### Active Resource Locks")
-                for l in locks:
-                    note = f" ({l['note']})" if l["note"] else ""
-                    lines.append(f"- `{l['resource']}` — locked by {l['session_id']}{note}")
-                lines.append("")
-        else:
-            lines.append("## No active sessions")
-            lines.append("")
-
-        if recent:
-            lines.append(f"## Recently Completed ({len(recent)} in last 24h)")
-            for s in recent:
-                lines.append(f"- **{s['id']}** ({s['session_type']}): {s.get('completion_summary', s['summary'])}")
-            lines.append("")
-
-        return "\n".join(lines)
-
-
-# Convenience function for quick sitrep
-def sitrep() -> str:
-    """Get a situation report. Call this at the start of every session."""
-    return SessionRegistry().get_situation_report()

test_sessions.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+"""Test the session registry end-to-end."""
+import sys
+sys.path.insert(0, "/data/symbiont")
+from symbiont.sessions import SessionRegistry, sitrep
+
+reg = SessionRegistry()
+
+# Register this session (the one that built everything)
+sid = reg.register(
+    "cowork",
+    "Building Symbiont core: router, dispatcher, sessions, Dendrite integration"
+)
+print(f"Registered session: {sid}")
+
+# Log some progress
+reg.log(sid, "Built LLM router with Haiku classifier and model tier dispatch")
+reg.log(sid, "Added systemd life support (API service + heartbeat timer)")
+reg.log(sid, "Integrated Dendrite headless browser via web.py")
+reg.log(sid, "Created session registry for cross-instance awareness")
+reg.log(sid, "Deployed CLAUDE.md bootstrap for new sessions")
+
+# Lock a resource to test that
+reg.lock_resource(sid, "/data/symbiont/symbiont/router.py", "May refactor to Elixir soon")
+
+# Print the sitrep
+print()
+print(sitrep())
+
+# Test the API endpoints too
+import urllib.request, json
+resp = urllib.request.urlopen("http://localhost:8111/sitrep", timeout=5)
+api_sitrep = json.loads(resp.read())
+print(f"API /sitrep active sessions: {len(api_sitrep['active'])}")
+
+# Complete this session
+reg.complete(sid, "Built Symbiont v0.1: router, dispatcher, ledger, heartbeat, Dendrite, sessions, CLAUDE.md")
+print(f"\nSession {sid} completed.")
+
+# Final sitrep showing it moved to completed
+print()
+print(sitrep())