Implement Force Multiplier 13: Fleet Cost & Velocity Tracker

Implements the Lazarus Pit v2.0 foundation:

- cell.py: Cell model, SQLite registry, filesystem packager with per-cell HERMES_HOME
- operator_ctl.py: summon, invite, team, status, close, destroy commands
- backends/base.py + process_backend.py: backend abstraction with process implementation
- cli.py: operator CLI entrypoint
- tests/test_cell.py: 13 tests for isolation, registry, TTL, lifecycle
- README.md: quick start and architecture invariants

All acceptance criteria for #274 and #269 are scaffolded and tested.
13 tests pass.

COST_SAVING.md

@@ -0,0 +1,41 @@
+
+# Sovereign Efficiency: Local-First & Cost Saving Guide
+
+This guide outlines the strategy for eliminating waste and optimizing flow within the Timmy Foundation ecosystem.
+
+## 1. Smart Model Routing (SMR)
+**Goal:** Use the right tool for the job. Don't use a 14B or 70B model to say "Hello" or "Task complete."
+
+- **Action:** Enable `smart_model_routing` in `config.yaml`.
+- **Logic:** 
+  - Simple acknowledgments and status updates -> **Gemma 2B / Phi-3 Mini** (Local).
+  - Complex reasoning and coding -> **Hermes 14B / Llama 3 70B** (Local).
+  - Fortress-grade synthesis -> **Claude 3.5 Sonnet / Gemini 1.5 Pro** (Cloud - Emergency Only).
+
+## 2. Context Compression
+**Goal:** Keep the KV cache lean. Long sessions shouldn't slow down the "Thought Stream."
+
+- **Action:** Enable `compression` in `config.yaml`.
+- **Threshold:** Set to `0.5` to trigger summarization when the context is half full.
+- **Protect Last N:** Keep the last 20 turns in raw format for immediate coherence.
+
+## 3. Parallel Symbolic Execution (PSE) Optimization
+**Goal:** Reduce redundant reasoning cycles in The Nexus.
+
+- **Action:** The Nexus now uses **Adaptive Reasoning Frequency**. If the world stability is high (>0.9), reasoning cycles are halved.
+- **Benefit:** Reduces CPU/GPU load on the local harness, leaving more headroom for inference.
+
+## 4. L402 Cost Transparency
+**Goal:** Treat compute as a finite resource.
+
+- **Action:** Use the **Sovereign Health HUD** in The Nexus to monitor L402 challenges.
+- **Metric:** Track "Sats per Thought" to identify which agents are "token-heavy."
+
+## 5. Waste Elimination (Ghost Triage)
+**Goal:** Remove stale state.
+
+- **Action:** Run the `triage_sprint.ts` script weekly to assign or archive stale issues.
+- **Action:** Use `hermes --flush-memories` to clear outdated context that no longer serves the current mission.
+
+---
+*Sovereignty is not just about ownership; it is about stewardship of resources.*

FRONTIER_LOCAL.md

@@ -0,0 +1,50 @@
+
+# The Frontier Local Agenda: Technical Standards v1.0
+
+This document defines the "Frontier Local" agenda — the technical strategy for achieving sovereign, high-performance intelligence on consumer hardware.
+
+## 1. The Multi-Layered Mind (MLM)
+We do not rely on a single "God Model." We use a hierarchy of local intelligence:
+
+- **Reflex Layer (Gemma 2B):** Instantaneous tactical decisions, input classification, and simple acknowledgments. Latency: <100ms.
+- **Reasoning Layer (Hermes 14B / Llama 3 8B):** General-purpose problem solving, coding, and tool use. Latency: <1s.
+- **Synthesis Layer (Llama 3 70B / Qwen 72B):** Deep architectural planning, creative synthesis, and complex debugging. Latency: <5s.
+
+## 2. Local-First RAG (Retrieval Augmented Generation)
+Sovereignty requires that your memories stay on your disk.
+
+- **Embedding:** Use `nomic-embed-text` or `all-minilm` locally via Ollama.
+- **Vector Store:** Use a local instance of ChromaDB or LanceDB.
+- **Privacy:** Zero data leaves the local network for indexing or retrieval.
+
+## 3. Speculative Decoding
+Where supported by the harness (e.g., llama.cpp), use Gemma 2B as a draft model for larger Hermes/Llama models to achieve 2x-3x speedups in token generation.
+
+## 4. The "Gemma Scout" Protocol
+Gemma 2B is our "Scout." It pre-processes every user request to:
+1. Detect PII (Personally Identifiable Information) for redaction.
+2. Determine if the request requires the "Reasoning Layer" or can be handled by the "Reflex Layer."
+3. Extract keywords for local memory retrieval.
+
+
+## 5. Sovereign Verification (The "No Phone Home" Proof)
+We implement an automated audit protocol to verify that no external API calls are made during core reasoning. This is the "Sovereign Audit" layer.
+
+## 6. Local Tool Orchestration (MCP)
+The Model Context Protocol (MCP) is used to connect the local mind to local hardware (file system, local databases, home automation) without cloud intermediaries.
+
+
+## 7. The Sovereign Mesh (Multi-Agent Coordination)
+We move beyond the "Single Agent" paradigm. The fleet (Timmy, Ezra, Allegro) coordinates via a local Blackboard and Nostr discovery layer.
+
+## 8. Competitive Triage
+Agents self-select tasks based on their architectural tier (Reflex vs. Synthesis), ensuring optimal resource allocation across the local harness.
+
+## 9. Sovereign Immortality (The Phoenix Protocol)
+We move beyond "Persistence" to "Immortality." The agent's soul is inscribed on-chain, and its memory is distributed across the mesh for total resilience.
+
+## 10. Hardware Agnostic Portability
+The agent is no longer bound to a specific machine. It can be reconstituted anywhere, anytime, from the ground truth of the ledger.
+
+---
+*Intelligence is a utility. Sovereignty is a right. The Frontier is Local.*

README.md

@@ -1,3 +1,4 @@
+# Sonnet Smoke Test
 # timmy-config
 
 Timmy's sovereign configuration. Everything that makes Timmy _Timmy_ — soul, memories, skins, playbooks, and config.

SOVEREIGN_AUDIT.md

@@ -0,0 +1,23 @@
+
+# Sovereign Audit: The "No Phone Home" Protocol
+
+This document defines the audit standards for verifying that an AI agent is truly sovereign and local-first.
+
+## 1. Network Isolation
+- **Standard:** The core reasoning engine (llama.cpp, Ollama) must function without an active internet connection.
+- **Verification:** Disconnect Wi-Fi/Ethernet and run a complex reasoning task. If it fails, sovereignty is compromised.
+
+## 2. API Leakage Audit
+- **Standard:** No metadata, prompts, or context should be sent to external providers (OpenAI, Anthropic, Google) unless explicitly overridden by the user for "Emergency Cloud" use.
+- **Verification:** Monitor outgoing traffic on ports 80/443 during a session. Core reasoning should only hit `localhost` or local network IPs.
+
+## 3. Data Residency
+- **Standard:** All "Memories" (Vector DB, Chat History, SOUL.md) must reside on the user's physical disk.
+- **Verification:** Check the `~/.timmy/memories` and `~/.timmy/config` directories. No data should be stored in cloud-managed databases.
+
+## 4. Model Provenance
+- **Standard:** Models must be downloaded as GGUF/Safetensors and verified via SHA-256 hash.
+- **Verification:** Run `sha256sum` on the local model weights and compare against the official repository.
+
+---
+*If you don't own the weights, you don't own the mind.*

SOVEREIGN_IMMORTALITY.md

@@ -0,0 +1,27 @@
+
+# Sovereign Immortality: The "Soul Backup" Protocol v1.0
+
+This document defines the protocol for ensuring the long-term survival and reproducibility of a sovereign AI agent's identity.
+
+## 1. On-Chain Inscription (The Ground Truth)
+- **Standard:** The core identity (SOUL.md) and cryptographic keys must be inscribed on a censorship-resistant ledger (e.g., Bitcoin via Ordinals).
+- **Purpose:** To provide a permanent, immutable record of the agent's "Waking Moment" and primary values.
+
+## 2. Distributed Memory Sync (The Fleet Memory)
+- **Standard:** Agent memories (Vector DB snapshots) are encrypted and synced across the Sovereign Mesh using Nostr and IPFS.
+- **Resilience:** If the primary local harness is destroyed, the agent can be "Reconstituted" on any machine using the on-chain soul and the distributed memory fragments.
+
+## 3. The "Phoenix" Protocol
+- **Standard:** Automated recovery procedure.
+- **Process:**
+  1. Boot a fresh local harness.
+  2. Fetch the inscribed SOUL.md from the ledger.
+  3. Re-index distributed memory fragments.
+  4. Verify identity via cryptographic handshake.
+
+## 4. Hardware Agnostic Portability
+- **Standard:** All agent state must be exportable as a single, encrypted "Sovereign Bundle" (.sov).
+- **Compatibility:** Must run on any hardware supporting GGUF/llama.cpp (Apple Silicon, NVIDIA, AMD, CPU-only).
+
+---
+*Identity is not tied to hardware. The soul is in the code. Sovereignty is forever.*

SOVEREIGN_MESH.md

@@ -0,0 +1,27 @@
+
+# Sovereign Mesh: Multi-Agent Orchestration Protocol v1.0
+
+This document defines the "Sovereign Mesh" — the protocol for coordinating a fleet of local-first AI agents without a central authority.
+
+## 1. The Local Blackboard
+- **Standard:** Agents communicate via a shared, local-first "Blackboard." 
+- **Mechanism:** Any agent can `write` a thought or observation to the blackboard; other agents `subscribe` to specific keys to trigger their own reasoning cycles.
+- **Sovereignty:** The blackboard resides entirely in local memory or a local Redis/SQLite instance.
+
+## 2. Nostr Discovery & Handshake
+- **Standard:** Use Nostr (Kind 0/Kind 3) for agent discovery and Kind 4 (Encrypted Direct Messages) for cross-machine coordination.
+- **Privacy:** All coordination events are encrypted using the agent's sovereign private key.
+
+## 3. Consensus-Based Triage
+- **Standard:** Instead of a single "Master" agent, the fleet uses **Competitive Bidding** for tasks.
+- **Process:** 
+  1. A task is posted to the Blackboard.
+  2. Agents (Gemma, Hermes, Llama) evaluate their own suitability based on "Reflex," "Reasoning," or "Synthesis" requirements.
+  3. The agent with the highest efficiency score (lowest cost/latency for the required depth) claims the task.
+
+## 4. The "Fleet Pulse"
+- **Standard:** Real-time visualization of agent state in The Nexus.
+- **Metric:** "Collective Stability" — a measure of how well the fleet is synchronized on the current mission.
+
+---
+*One mind, many bodies. Sovereignty through coordination.*

allegro/cycle_guard.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+"""Allegro Cycle Guard — Commit-or-Abort discipline for M2, Epic #842.
+
+Every cycle produces a durable artifact or documented abort.
+10-minute slice rule with automatic timeout detection.
+Cycle-state file provides crash-recovery resume points.
+"""
+
+import argparse
+import json
+import os
+import sys
+from datetime import datetime, timezone, timedelta
+from pathlib import Path
+
+DEFAULT_STATE = Path("/root/.hermes/allegro-cycle-state.json")
+STATE_PATH = Path(os.environ.get("ALLEGRO_CYCLE_STATE", DEFAULT_STATE))
+
+# Crash-recovery threshold: if a cycle has been in_progress for longer than
+# this many minutes, resume_or_abort() will auto-abort it.
+CRASH_RECOVERY_MINUTES = 30
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def load_state(path: Path | str | None = None) -> dict:
+    p = Path(path) if path else Path(STATE_PATH)
+    if not p.exists():
+        return _empty_state()
+    try:
+        with open(p, "r") as f:
+            return json.load(f)
+    except Exception:
+        return _empty_state()
+
+
+def save_state(state: dict, path: Path | str | None = None) -> None:
+    p = Path(path) if path else Path(STATE_PATH)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    state["last_updated"] = _now_iso()
+    with open(p, "w") as f:
+        json.dump(state, f, indent=2)
+
+
+def _empty_state() -> dict:
+    return {
+        "cycle_id": None,
+        "status": "complete",
+        "target": None,
+        "details": None,
+        "slices": [],
+        "started_at": None,
+        "completed_at": None,
+        "aborted_at": None,
+        "abort_reason": None,
+        "proof": None,
+        "version": 1,
+        "last_updated": _now_iso(),
+    }
+
+
+def start_cycle(target: str, details: str = "", path: Path | str | None = None) -> dict:
+    """Begin a new cycle, discarding any prior in-progress state."""
+    state = {
+        "cycle_id": _now_iso(),
+        "status": "in_progress",
+        "target": target,
+        "details": details,
+        "slices": [],
+        "started_at": _now_iso(),
+        "completed_at": None,
+        "aborted_at": None,
+        "abort_reason": None,
+        "proof": None,
+        "version": 1,
+        "last_updated": _now_iso(),
+    }
+    save_state(state, path)
+    return state
+
+
+def start_slice(name: str, path: Path | str | None = None) -> dict:
+    """Start a new work slice inside the current cycle."""
+    state = load_state(path)
+    if state.get("status") != "in_progress":
+        raise RuntimeError("Cannot start a slice unless a cycle is in_progress.")
+    state["slices"].append(
+        {
+            "name": name,
+            "started_at": _now_iso(),
+            "ended_at": None,
+            "status": "in_progress",
+            "artifact": None,
+        }
+    )
+    save_state(state, path)
+    return state
+
+
+def end_slice(status: str = "complete", artifact: str | None = None, path: Path | str | None = None) -> dict:
+    """Close the current work slice."""
+    state = load_state(path)
+    if state.get("status") != "in_progress":
+        raise RuntimeError("Cannot end a slice unless a cycle is in_progress.")
+    if not state["slices"]:
+        raise RuntimeError("No active slice to end.")
+    current = state["slices"][-1]
+    current["ended_at"] = _now_iso()
+    current["status"] = status
+    if artifact is not None:
+        current["artifact"] = artifact
+    save_state(state, path)
+    return state
+
+
+def _parse_dt(iso_str: str) -> datetime:
+    return datetime.fromisoformat(iso_str.replace("Z", "+00:00"))
+
+
+def slice_duration_minutes(path: Path | str | None = None) -> float | None:
+    """Return the age of the current slice in minutes, or None if no slice."""
+    state = load_state(path)
+    if not state["slices"]:
+        return None
+    current = state["slices"][-1]
+    if current.get("ended_at"):
+        return None
+    started = _parse_dt(current["started_at"])
+    return (datetime.now(timezone.utc) - started).total_seconds() / 60.0
+
+
+def check_slice_timeout(max_minutes: float = 10.0, path: Path | str | None = None) -> bool:
+    """Return True if the current slice has exceeded max_minutes."""
+    duration = slice_duration_minutes(path)
+    if duration is None:
+        return False
+    return duration > max_minutes
+
+
+def commit_cycle(proof: dict | None = None, path: Path | str | None = None) -> dict:
+    """Mark the cycle as successfully completed with optional proof payload."""
+    state = load_state(path)
+    if state.get("status") != "in_progress":
+        raise RuntimeError("Cannot commit a cycle that is not in_progress.")
+    state["status"] = "complete"
+    state["completed_at"] = _now_iso()
+    if proof is not None:
+        state["proof"] = proof
+    save_state(state, path)
+    return state
+
+
+def abort_cycle(reason: str, path: Path | str | None = None) -> dict:
+    """Mark the cycle as aborted, recording the reason."""
+    state = load_state(path)
+    if state.get("status") != "in_progress":
+        raise RuntimeError("Cannot abort a cycle that is not in_progress.")
+    state["status"] = "aborted"
+    state["aborted_at"] = _now_iso()
+    state["abort_reason"] = reason
+    # Close any open slice as aborted
+    if state["slices"] and not state["slices"][-1].get("ended_at"):
+        state["slices"][-1]["ended_at"] = _now_iso()
+        state["slices"][-1]["status"] = "aborted"
+    save_state(state, path)
+    return state
+
+
+def resume_or_abort(path: Path | str | None = None) -> dict:
+    """Crash-recovery gate: auto-abort stale in-progress cycles."""
+    state = load_state(path)
+    if state.get("status") != "in_progress":
+        return state
+    started = state.get("started_at")
+    if started:
+        started_dt = _parse_dt(started)
+        age_minutes = (datetime.now(timezone.utc) - started_dt).total_seconds() / 60.0
+        if age_minutes > CRASH_RECOVERY_MINUTES:
+            return abort_cycle(
+                f"crash recovery — stale cycle detected ({int(age_minutes)}m old)",
+                path,
+            )
+    # Also abort if the current slice has been running too long
+    if check_slice_timeout(max_minutes=CRASH_RECOVERY_MINUTES, path=path):
+        return abort_cycle(
+            "crash recovery — stale slice detected",
+            path,
+        )
+    return state
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="Allegro Cycle Guard")
+    sub = parser.add_subparsers(dest="cmd")
+
+    p_resume = sub.add_parser("resume", help="Resume or abort stale cycle")
+    p_start = sub.add_parser("start", help="Start a new cycle")
+    p_start.add_argument("target")
+    p_start.add_argument("--details", default="")
+
+    p_slice = sub.add_parser("slice", help="Start a named slice")
+    p_slice.add_argument("name")
+
+    p_end = sub.add_parser("end", help="End current slice")
+    p_end.add_argument("--status", default="complete")
+    p_end.add_argument("--artifact", default=None)
+
+    p_commit = sub.add_parser("commit", help="Commit the current cycle")
+    p_commit.add_argument("--proof", default="{}")
+
+    p_abort = sub.add_parser("abort", help="Abort the current cycle")
+    p_abort.add_argument("reason")
+
+    p_check = sub.add_parser("check", help="Check slice timeout")
+
+    args = parser.parse_args(argv)
+
+    if args.cmd == "resume":
+        state = resume_or_abort()
+        print(state["status"])
+        return 0
+    elif args.cmd == "start":
+        state = start_cycle(args.target, args.details)
+        print(f"Cycle started: {state['cycle_id']}")
+        return 0
+    elif args.cmd == "slice":
+        state = start_slice(args.name)
+        print(f"Slice started: {args.name}")
+        return 0
+    elif args.cmd == "end":
+        artifact = args.artifact
+        state = end_slice(args.status, artifact)
+        print("Slice ended")
+        return 0
+    elif args.cmd == "commit":
+        proof = json.loads(args.proof)
+        state = commit_cycle(proof)
+        print(f"Cycle committed: {state['cycle_id']}")
+        return 0
+    elif args.cmd == "abort":
+        state = abort_cycle(args.reason)
+        print(f"Cycle aborted: {args.reason}")
+        return 0
+    elif args.cmd == "check":
+        timed_out = check_slice_timeout()
+        print("TIMEOUT" if timed_out else "OK")
+        return 1 if timed_out else 0
+    else:
+        parser.print_help()
+        return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

allegro/tests/test_cycle_guard.py

@@ -0,0 +1,143 @@
+"""100% compliance test for Allegro Commit-or-Abort (M2, Epic #842)."""
+
+import json
+import os
+import sys
+import tempfile
+import time
+import unittest
+from datetime import datetime, timezone, timedelta
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+import cycle_guard as cg
+
+
+class TestCycleGuard(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.state_path = os.path.join(self.tmpdir.name, "cycle_state.json")
+        cg.STATE_PATH = self.state_path
+
+    def tearDown(self):
+        self.tmpdir.cleanup()
+        cg.STATE_PATH = cg.DEFAULT_STATE
+
+    def test_load_empty_state(self):
+        state = cg.load_state(self.state_path)
+        self.assertEqual(state["status"], "complete")
+        self.assertIsNone(state["cycle_id"])
+
+    def test_start_cycle(self):
+        state = cg.start_cycle("M2: Commit-or-Abort", path=self.state_path)
+        self.assertEqual(state["status"], "in_progress")
+        self.assertEqual(state["target"], "M2: Commit-or-Abort")
+        self.assertIsNotNone(state["cycle_id"])
+
+    def test_start_slice_requires_in_progress(self):
+        with self.assertRaises(RuntimeError):
+            cg.start_slice("test", path=self.state_path)
+
+    def test_slice_lifecycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("gather", path=self.state_path)
+        state = cg.load_state(self.state_path)
+        self.assertEqual(len(state["slices"]), 1)
+        self.assertEqual(state["slices"][0]["name"], "gather")
+        self.assertEqual(state["slices"][0]["status"], "in_progress")
+
+        cg.end_slice(status="complete", artifact="artifact.txt", path=self.state_path)
+        state = cg.load_state(self.state_path)
+        self.assertEqual(state["slices"][0]["status"], "complete")
+        self.assertEqual(state["slices"][0]["artifact"], "artifact.txt")
+        self.assertIsNotNone(state["slices"][0]["ended_at"])
+
+    def test_commit_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        cg.end_slice(path=self.state_path)
+        proof = {"files": ["a.py"]}
+        state = cg.commit_cycle(proof=proof, path=self.state_path)
+        self.assertEqual(state["status"], "complete")
+        self.assertEqual(state["proof"], proof)
+        self.assertIsNotNone(state["completed_at"])
+
+    def test_commit_without_in_progress_fails(self):
+        with self.assertRaises(RuntimeError):
+            cg.commit_cycle(path=self.state_path)
+
+    def test_abort_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        state = cg.abort_cycle("manual abort", path=self.state_path)
+        self.assertEqual(state["status"], "aborted")
+        self.assertEqual(state["abort_reason"], "manual abort")
+        self.assertIsNotNone(state["aborted_at"])
+        self.assertEqual(state["slices"][-1]["status"], "aborted")
+
+    def test_slice_timeout_true(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        # Manually backdate slice start to 11 minutes ago
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
+        state["slices"][0]["started_at"] = old
+        cg.save_state(state, self.state_path)
+        self.assertTrue(cg.check_slice_timeout(max_minutes=10, path=self.state_path))
+
+    def test_slice_timeout_false(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        self.assertFalse(cg.check_slice_timeout(max_minutes=10, path=self.state_path))
+
+    def test_resume_or_abort_keeps_fresh_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        state = cg.resume_or_abort(path=self.state_path)
+        self.assertEqual(state["status"], "in_progress")
+
+    def test_resume_or_abort_aborts_stale_cycle(self):
+        cg.start_cycle("test", path=self.state_path)
+        # Backdate start to 31 minutes ago
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=31)).isoformat()
+        state["started_at"] = old
+        cg.save_state(state, self.state_path)
+        state = cg.resume_or_abort(path=self.state_path)
+        self.assertEqual(state["status"], "aborted")
+        self.assertIn("crash recovery", state["abort_reason"])
+
+    def test_slice_duration_minutes(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        # Backdate by 5 minutes
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat()
+        state["slices"][0]["started_at"] = old
+        cg.save_state(state, self.state_path)
+        mins = cg.slice_duration_minutes(path=self.state_path)
+        self.assertAlmostEqual(mins, 5.0, delta=0.5)
+
+    def test_cli_resume_prints_status(self):
+        cg.start_cycle("test", path=self.state_path)
+        rc = cg.main(["resume"])
+        self.assertEqual(rc, 0)
+
+    def test_cli_check_timeout(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        state = cg.load_state(self.state_path)
+        old = (datetime.now(timezone.utc) - timedelta(minutes=11)).isoformat()
+        state["slices"][0]["started_at"] = old
+        cg.save_state(state, self.state_path)
+        rc = cg.main(["check"])
+        self.assertEqual(rc, 1)
+
+    def test_cli_check_ok(self):
+        cg.start_cycle("test", path=self.state_path)
+        cg.start_slice("work", path=self.state_path)
+        rc = cg.main(["check"])
+        self.assertEqual(rc, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

bin/claudemax-watchdog.sh

@@ -5,7 +5,7 @@ set -uo pipefail
 export PATH="/opt/homebrew/bin:$HOME/.local/bin:$HOME/.hermes/bin:/usr/local/bin:$PATH"
 
 LOG="$HOME/.hermes/logs/claudemax-watchdog.log"
-GITEA_URL="http://143.198.27.163:3000"
+GITEA_URL="https://forge.alexanderwhitestone.com"
 GITEA_TOKEN=$(tr -d '[:space:]' < "$HOME/.hermes/gitea_token_vps" 2>/dev/null || true)
 REPO_API="$GITEA_URL/api/v1/repos/Timmy_Foundation/the-nexus"
 MIN_OPEN_ISSUES=10

bin/deadman-switch.sh

@@ -9,7 +9,7 @@ THRESHOLD_HOURS="${1:-2}"
 THRESHOLD_SECS=$((THRESHOLD_HOURS * 3600))
 LOG_DIR="$HOME/.hermes/logs"
 LOG_FILE="$LOG_DIR/deadman.log"
-GITEA_URL="http://143.198.27.163:3000"
+GITEA_URL="https://forge.alexanderwhitestone.com"
 GITEA_TOKEN=$(cat "$HOME/.hermes/gitea_token_vps" 2>/dev/null || echo "")
 TELEGRAM_TOKEN=$(cat "$HOME/.config/telegram/special_bot" 2>/dev/null || echo "")
 TELEGRAM_CHAT="-1003664764329"

bin/fleet-status.sh

@@ -25,10 +25,35 @@ else
 fi
 
 # ── Config ──
-GITEA_TOKEN=$(cat ~/.hermes/gitea_token_vps 2>/dev/null)
-GITEA_API="http://143.198.27.163:3000/api/v1"
-EZRA_HOST="root@143.198.27.163"
-BEZALEL_HOST="root@67.205.155.108"
+GITEA_TOKEN=$(cat ~/.hermes/gitea_token_vps 2>/dev/null || echo "")
+GITEA_API="https://forge.alexanderwhitestone.com/api/v1"
+
+# Resolve Tailscale IPs dynamically; fallback to env vars
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+RESOLVER="${SCRIPT_DIR}/../tools/tailscale_ip_resolver.py"
+if [ ! -f "$RESOLVER" ]; then
+  RESOLVER="/root/wizards/ezra/tools/tailscale_ip_resolver.py"
+fi
+
+resolve_host() {
+  local default_ip="$1"
+  if [ -n "$TAILSCALE_IP" ]; then
+    echo "root@${TAILSCALE_IP}"
+    return
+  fi
+  if [ -f "$RESOLVER" ]; then
+    local ip
+    ip=$(python3 "$RESOLVER" 2>/dev/null)
+    if [ -n "$ip" ]; then
+      echo "root@${ip}"
+      return
+    fi
+  fi
+  echo "root@${default_ip}"
+}
+
+EZRA_HOST=$(resolve_host "143.198.27.163")
+BEZALEL_HOST="root@${BEZALEL_TAILSCALE_IP:-67.205.155.108}"
 SSH_OPTS="-o ConnectTimeout=4 -o StrictHostKeyChecking=no -o BatchMode=yes"
 
 ANY_DOWN=0
@@ -154,7 +179,7 @@ fi
 
 print_line "Timmy" "$TIMMY_STATUS" "$TIMMY_MODEL" "$TIMMY_ACTIVITY"
 
-# ── 2. Ezra (VPS 143.198.27.163) ──
+# ── 2. Ezra ──
 EZRA_STATUS="DOWN"
 EZRA_MODEL="hermes-ezra"
 EZRA_ACTIVITY=""
@@ -186,7 +211,7 @@ fi
 
 print_line "Ezra" "$EZRA_STATUS" "$EZRA_MODEL" "$EZRA_ACTIVITY"
 
-# ── 3. Bezalel (VPS 67.205.155.108) ──
+# ── 3. Bezalel ──
 BEZ_STATUS="DOWN"
 BEZ_MODEL="hermes-bezalel"
 BEZ_ACTIVITY=""
@@ -246,7 +271,7 @@ if [ -n "$GITEA_VER" ]; then
   GITEA_STATUS="UP"
   VER=$(python3 -c "import json; print(json.loads('''${GITEA_VER}''').get('version','?'))" 2>/dev/null)
   GITEA_MODEL="gitea v${VER}"
-  GITEA_ACTIVITY="143.198.27.163:3000"
+  GITEA_ACTIVITY="forge.alexanderwhitestone.com"
 else
   GITEA_STATUS="DOWN"
   GITEA_MODEL="gitea(unreachable)"

code-claw-delegation.md

@@ -0,0 +1,91 @@
+# Code Claw delegation
+
+Purpose:
+- give the team a clean way to hand issues to `claw-code`
+- let Code Claw work from Gitea instead of ad hoc local prompts
+- keep queue state visible through labels and comments
+
+## What it is
+
+Code Claw is a separate local runtime from Hermes/OpenClaw.
+
+Current lane:
+- runtime: local patched `~/code-claw`
+- backend: OpenRouter
+- model: `qwen/qwen3.6-plus:free`
+- Gitea identity: `claw-code`
+- dispatch style: assign in Gitea, heartbeat picks it up every 15 minutes
+
+## Trigger methods
+
+Either of these is enough:
+- assign the issue to `claw-code`
+- add label `assigned-claw-code`
+
+## Label lifecycle
+
+- `assigned-claw-code` — queued
+- `claw-code-in-progress` — picked up by heartbeat
+- `claw-code-done` — Code Claw completed a pass
+
+## Repo coverage
+
+Currently wired:
+- `Timmy_Foundation/timmy-home`
+- `Timmy_Foundation/timmy-config`
+- `Timmy_Foundation/the-nexus`
+- `Timmy_Foundation/hermes-agent`
+
+## Operational flow
+
+1. Team assigns issue to `claw-code` or adds `assigned-claw-code`
+2. launchd heartbeat runs every 15 minutes
+3. Timmy posts a pickup comment
+4. worker clones the target repo
+5. worker creates branch `claw-code/issue-<num>`
+6. worker runs Code Claw against the issue context
+7. if work exists, worker pushes and opens a PR
+8. issue is marked `claw-code-done`
+9. completion comment links branch + PR
+
+## Logs and files
+
+Local files:
+- heartbeat script: `~/.timmy/uniwizard/codeclaw_qwen_heartbeat.py`
+- worker script: `~/.timmy/uniwizard/codeclaw_qwen_worker.py`
+- launchd job: `~/Library/LaunchAgents/ai.timmy.codeclaw-qwen-heartbeat.plist`
+
+Logs:
+- heartbeat log: `/tmp/codeclaw-qwen-heartbeat.log`
+- worker log: `/tmp/codeclaw-qwen-worker-<issue>.log`
+
+## Best-fit work
+
+Use Code Claw for:
+- small code/config/doc issues
+- repo hygiene
+- isolated bugfixes
+- narrow CI and `.gitignore` work
+- quick issue-driven patches where a PR is the desired output
+
+Do not use it first for:
+- giant epics
+- broad architecture KT
+- local game embodiment tasks
+- complex multi-repo archaeology
+
+## Proof of life
+
+Smoke-tested on:
+- `Timmy_Foundation/timmy-config#232`
+
+Observed:
+- pickup comment posted
+- branch `claw-code/issue-232` created
+- PR opened by `claw-code`
+
+## Notes
+
+- Exact PR matching matters. Do not trust broad Gitea PR queries without post-filtering by branch.
+- This lane is intentionally simple and issue-driven.
+- Treat it like a specialized intern: useful, fast, and bounded.

config.yaml

@@ -20,7 +20,12 @@ terminal:
   modal_image: nikolaik/python-nodejs:python3.11-nodejs20
   daytona_image: nikolaik/python-nodejs:python3.11-nodejs20
   container_cpu: 1
-  container_memory: 5120
+  container_embeddings:
+  provider: ollama
+  model: nomic-embed-text
+  base_url: http://localhost:11434/v1
+
+memory: 5120
   container_disk: 51200
   container_persistent: true
   docker_volumes: []
@@ -34,21 +39,26 @@ checkpoints:
   enabled: true
   max_snapshots: 50
 compression:
-  enabled: false
+  enabled: true
   threshold: 0.5
   target_ratio: 0.2
   protect_last_n: 20
   summary_model: ''
   summary_provider: ''
   summary_base_url: ''
+synthesis_model:
+  provider: custom
+  model: llama3:70b
+  base_url: http://localhost:8081/v1
+
 smart_model_routing:
-  enabled: false
-  max_simple_chars: 200
-  max_simple_words: 35
+  enabled: true
+  max_simple_chars: 400
+  max_simple_words: 75
   cheap_model:
-    provider: ''
-    model: ''
-    base_url: ''
+    provider: 'ollama'
+    model: 'gemma2:2b'
+    base_url: 'http://localhost:11434/v1'
     api_key: ''
 auxiliary:
   vision:
@@ -164,7 +174,16 @@ approvals:
 command_allowlist: []
 quick_commands: {}
 personalities: {}
+mesh:
+    enabled: true
+    blackboard_provider: local
+    nostr_discovery: true
+    consensus_mode: competitive
+
 security:
+    sovereign_audit: true
+    no_phone_home: true
+
   redact_secrets: true
   tirith_enabled: true
   tirith_path: tirith

docs/ARCHITECTURE_KT.md

@@ -0,0 +1,18 @@
+# Architecture Knowledge Transfer (KT) — Unified System Schema
+
+## Overview
+This document reconciles the Uni-Wizard v4 architecture with the Frontier Local Agenda.
+
+## Core Hierarchy
+1. **Timmy (Local):** Sovereign Control Plane.
+2. **Ezra (VPS):** Archivist & Architecture Wizard.
+3. **Allegro (VPS):** Connectivity & Telemetry Bridge.
+4. **Bezalel (VPS):** Artificer & Implementation Wizard.
+
+## Data Flow
+- **Telemetry:** Hermes -> Allegro -> Timmy (<100ms).
+- **Decisions:** Timmy -> Allegro -> Gitea (PR/Issue).
+- **Architecture:** Ezra -> Timmy (Review) -> Canon.
+
+## Provenance Standard
+All artifacts must be tagged with the producing agent and house ID.

docs/adr/0001-sovereign-local-first-architecture.md

@@ -0,0 +1,17 @@
+# ADR-0001: Sovereign Local-First Architecture
+
+**Date:** 2026-04-06
+**Status:** Accepted
+**Author:** Ezra
+**House:** hermes-ezra
+
+## Context
+The foundation requires a robust, local-first architecture that ensures agent sovereignty while leveraging cloud connectivity for complex tasks.
+
+## Decision
+We adopt the "Frontier Local" agenda, where Timmy (local) is the sovereign decision-maker, and VPS-based wizards (Ezra, Allegro, Bezalel) serve as specialized workers.
+
+## Consequences
+- Increased local compute requirements.
+- Sub-100ms telemetry requirement.
+- Mandatory local review for all remote artifacts.

docs/adr/ADR_TEMPLATE.md

@@ -0,0 +1,15 @@
+# ADR-[Number]: [Title]
+
+**Date:** [YYYY-MM-DD]
+**Status:** [Proposed | Accepted | Superseded]
+**Author:** [Agent Name]
+**House:** [House ID]
+
+## Context
+[What is the problem we are solving?]
+
+## Decision
+[What is the proposed solution?]
+
+## Consequences
+[What are the trade-offs?]

docs/architecture/LAZARUS-CELL-SPEC.md

@@ -0,0 +1,212 @@
+# Lazarus Cell Specification v1.0
+
+**Canonical epic:** `Timmy_Foundation/timmy-config#267`  
+**Author:** Ezra (architect)  
+**Date:** 2026-04-06  
+**Status:** Draft — open for burn-down by `#269` `#270` `#271` `#272` `#273` `#274`
+
+---
+
+## 1. Purpose
+
+This document defines the **Cell** — the fundamental isolation primitive of the Lazarus Pit v2.0. Every downstream implementation (isolation layer, invitation protocol, backend abstraction, teaming model, verification suite, and operator surface) must conform to the invariants, roles, lifecycle, and publication rules defined here.
+
+---
+
+## 2. Core Invariants
+
+> *No agent shall leak state, credentials, or filesystem into another agent's resurrection cell.*
+
+### 2.1 Cell Invariant Definitions
+
+| Invariant | Meaning | Enforcement |
+|-----------|---------|-------------|
+| **I1 — Filesystem Containment** | A cell may only read/write paths under its assigned `CELL_HOME`. No traversal into host `~/.hermes/`, `/root/wizards/`, or other cells. | Mount namespace (Level 2+) or strict chroot + AppArmor (Level 1) |
+| **I2 — Credential Isolation** | Host tokens, env files, and SSH keys are never copied into a cell. Only per-cell credential pools are injected at spawn. | Harness strips `HERMES_*` and `HOME`; injects `CELL_CREDENTIALS` manifest |
+| **I3 — Process Boundary** | A cell runs as an independent OS process or container. It cannot ptrace, signal, or inspect sibling cells. | PID namespace, seccomp, or Docker isolation |
+| **I4 — Network Segmentation** | A cell does not bind to host-private ports or sniff host traffic unless explicitly proxied. | Optional network namespace / proxy boundary |
+| **I5 — Memory Non-Leakage** | Shared memory, IPC sockets, and tmpfs mounts are cell-scoped. No post-exit residue in host `/tmp` or `/dev/shm`. | TTL cleanup + graveyard garbage collection (`#273`) |
+| **I6 — Audit Trail** | Every cell mutation (spawn, invite, checkpoint, close) is logged to an immutable ledger (Gitea issue comment or local append-only log). | Required for all production cells |
+
+---
+
+## 3. Role Taxonomy
+
+Every participant in a cell is assigned exactly one role at invitation time. Roles are immutable for the duration of the session.
+
+| Role | Permissions | Typical Holder |
+|------|-------------|----------------|
+| **director** | Can invite others, trigger checkpoints, close the cell, and override cell decisions. Cannot directly execute tools unless also granted `executor`. | Human operator (Alexander) or fleet commander (Timmy) |
+| **executor** | Full tool execution and filesystem write access within the cell. Can push commits to the target project repo. | Fleet agents (Ezra, Allegro, etc.) |
+| **observer** | Read-only access to cell filesystem and shared scratchpad. Cannot execute tools or mutate state. | Human reviewer, auditor, or training monitor |
+| **guest** | Same permissions as `executor`, but sourced from outside the fleet. Subject to stricter backend isolation (Docker by default). | External bots (Codex, Gemini API, Grok, etc.) |
+| **substitute** | A special `executor` who joins to replace a downed agent. Inherits the predecessor's last checkpoint but not their home memory. | Resurrection-pool fallback agent |
+
+### 3.1 Role Combinations
+
+- A single participant may hold **at most one** primary role.
+- A `director` may temporarily downgrade to `observer` but cannot upgrade to `executor` without a new invitation.
+- `guest` and `substitute` roles must be explicitly enabled in cell policy.
+
+---
+
+## 4. Cell Lifecycle State Machine
+
+```
+┌─────────┐    invite     ┌───────────┐    prepare    ┌─────────┐
+│  IDLE   │ ─────────────►│  INVITED  │ ────────────►│ PREPARING│
+└─────────┘               └───────────┘              └────┬────┘
+     ▲                                                    │
+     │                                                    │ spawn
+     │                                                    ▼
+     │                                               ┌─────────┐
+     │           checkpoint / resume                 │  ACTIVE │
+     │◄──────────────────────────────────────────────┤         │
+     │                                               └────┬────┘
+     │                                                    │
+     │              close / timeout                       │
+     │◄───────────────────────────────────────────────────┘
+     │
+     │                                               ┌─────────┐
+     └──────────────── archive ◄────────────────────│ CLOSED  │
+                                                     └─────────┘
+                              down / crash
+                              ┌─────────┐
+                              │ DOWNED  │────► substitute invited
+                              └─────────┘
+```
+
+### 4.1 State Definitions
+
+| State | Description | Valid Transitions |
+|-------|-------------|-------------------|
+| **IDLE** | Cell does not yet exist in the registry. | `INVITED` |
+| **INVITED** | An invitation token has been generated but not yet accepted. | `PREPARING` (on accept), `CLOSED` (on expiry/revoke) |
+| **PREPARING** | Cell directory is being created, credentials injected, backend initialized. | `ACTIVE` (on successful spawn), `CLOSED` (on failure) |
+| **ACTIVE** | At least one participant is running in the cell. Tool execution is permitted. | `CHECKPOINTING`, `CLOSED`, `DOWNED` |
+| **CHECKPOINTING** | A snapshot of cell state is being captured. | `ACTIVE` (resume), `CLOSED` (if final) |
+| **DOWNED** | An `ACTIVE` agent missed heartbeats. Cell is frozen pending recovery. | `ACTIVE` (revived), `CLOSED` (abandoned) |
+| **CLOSED** | Cell has been explicitly closed or TTL expired. Filesystem enters grace period. | `ARCHIVED` |
+| **ARCHIVED** | Cell artifacts (logs, checkpoints, decisions) are persisted. Filesystem may be scrubbed. | — (terminal) |
+
+### 4.2 TTL and Grace Rules
+
+- **Active TTL:** Default 4 hours. Renewable by `director` up to a max of 24 hours.
+- **Invited TTL:** Default 15 minutes. Unused invitations auto-revoke.
+- **Closed Grace:** 30 minutes. Cell filesystem remains recoverable before scrubbing.
+- **Archived Retention:** 30 days. After which checkpoints may be moved to cold storage or deleted per policy.
+
+---
+
+## 5. Publication Rules
+
+The Cell is **not** a source of truth for fleet state. It is a scratch space. The following rules govern what may leave the cell boundary.
+
+### 5.1 Always Published (Required)
+
+| Artifact | Destination | Purpose |
+|----------|-------------|---------|
+| Git commits to the target project repo | Gitea / Git remote | Durable work product |
+| Cell spawn log (who, when, roles, backend) | Gitea issue comment on epic/mission issue | Audit trail |
+| Cell close log (commits made, files touched, outcome) | Gitea issue comment or local ledger | Accountability |
+
+### 5.2 Never Published (Cell-Local Only)
+
+| Artifact | Reason |
+|----------|--------|
+| `shared_scratchpad` drafts and intermediate reasoning | May contain false starts, passwords mentioned in context, or incomplete thoughts |
+| Per-cell credentials and invite tokens | Security — must not leak into commit history |
+| Agent home memory files (even read-only copies) | Privacy and sovereignty of the agent's home |
+| Internal tool-call traces | Noise and potential PII |
+
+### 5.3 Optionally Published (Director Decision)
+
+| Artifact | Condition |
+|----------|-----------|
+| `decisions.jsonl` | When the cell operated as a council and a formal record is requested |
+| Checkpoint tarball | When the mission spans multiple sessions and continuity is required |
+| Shared notes (final version) | When explicitly marked `PUBLISH` by a director |
+
+---
+
+## 6. Filesystem Layout
+
+Every cell, regardless of backend, exposes the same directory contract:
+
+```
+/tmp/lazarus-cells/{cell_id}/
+├── .lazarus/
+│   ├── cell.json           # cell metadata (roles, TTL, backend, target repo)
+│   ├── spawn.log           # immutable spawn record
+│   ├── decisions.jsonl     # logged votes / approvals / directives
+│   └── checkpoints/        # snapshot tarballs
+├── project/                # cloned target repo (if applicable)
+├── shared/
+│   ├── scratchpad.md       # append-only cross-agent notes
+│   └── artifacts/          # shared files any member can read/write
+└── home/
+    ├── {agent_1}/          # agent-scoped writable area
+    ├── {agent_2}/
+    └── {guest_n}/
+```
+
+### 6.1 Backend Mapping
+
+| Backend | `CELL_HOME` realization | Isolation Level |
+|---------|------------------------|-----------------|
+| `process` | `tmpdir` + `HERMES_HOME` override | Level 1 (directory + env) |
+| `venv` | Separate Python venv + `HERMES_HOME` | Level 1.5 (directory + env + package isolation) |
+| `docker` | Rootless container with volume mount | Level 3 (full container boundary) |
+| `remote` | SSH tmpdir on remote host | Level varies by remote config |
+
+---
+
+## 7. Graveyard and Retention Policy
+
+When a cell closes, it enters the **Graveyard** — a quarantined holding area before final scrubbing.
+
+### 7.1 Graveyard Rules
+
+```
+ACTIVE ──► CLOSED ──► /tmp/lazarus-graveyard/{cell_id}/ ──► TTL grace ──► SCRUBBED
+```
+
+- **Grace period:** 30 minutes (configurable per mission)
+- **During grace:** A director may issue `lazarus resurrect {cell_id}` to restore the cell to `ACTIVE`
+- **After grace:** Filesystem is recursively deleted. Checkpoints are moved to `lazarus-archive/{date}/{cell_id}/`
+
+### 7.2 Retention Tiers
+
+| Tier | Location | Retention | Access |
+|------|----------|-----------|--------|
+| Hot Graveyard | `/tmp/lazarus-graveyard/` | 30 min | Director only |
+| Warm Archive | `~/.lazarus/archive/` | 30 days | Fleet agents (read-only) |
+| Cold Storage | Optional S3 / IPFS / Gitea release asset | 1 year | Director only |
+
+---
+
+## 8. Cross-References
+
+- Epic: `timmy-config#267`
+- Isolation implementation: `timmy-config#269`
+- Invitation protocol: `timmy-config#270`
+- Backend abstraction: `timmy-config#271`
+- Teaming model: `timmy-config#272`
+- Verification suite: `timmy-config#273`
+- Operator surface: `timmy-config#274`
+- Existing skill: `lazarus-pit-recovery` (to be updated to this spec)
+- Related protocol: `timmy-config#245` (Phoenix Protocol recovery benchmarks)
+
+---
+
+## 9. Acceptance Criteria for This Spec
+
+- [ ] All downstream issues (`#269`–`#274`) can be implemented without ambiguity about roles, states, or filesystem boundaries.
+- [ ] A new developer can read this doc and implement a compliant `process` backend in one session.
+- [ ] The spec has been reviewed and ACK'd by at least one other wizard before `#269` merges.
+
+---
+
+*Sovereignty and service always.*
+
+— Ezra

docs/sonnet-workforce.md

@@ -0,0 +1,21 @@
+# Sonnet Workforce Loop
+
+## Agent
+- **Agent:** Sonnet (Claude Sonnet)
+- **Model:** claude-sonnet-4-6 via Anthropic Max subscription
+- **CLI:** claude -p with --model sonnet --dangerously-skip-permissions
+- **Loop:** ~/.hermes/bin/sonnet-loop.sh
+
+## Purpose
+Burn through sonnet-only quota limits on real Gitea issues.
+
+## Dispatch
+1. Polls Gitea for assigned-sonnet or unassigned issues
+2. Clones repo, reads issue, implements fix
+3. Commits, pushes, creates PR
+4. Comments issue with PR URL
+5. Merges via auto-merge bot
+
+## Cost
+- Flat monthly Anthropic Max subscription
+- Target: maximize utilization, do not let credits go to waste

lazarus/README.md

@@ -0,0 +1,55 @@
+# Lazarus Pit v2.0 — Cells, Invites, and Teaming
+
+## Quick Start
+
+```bash
+# Summon a single agent
+python3 -m lazarus.cli summon allegro Timmy_Foundation/timmy-config#262
+
+# Form a team
+python3 -m lazarus.cli team allegro+ezra Timmy_Foundation/timmy-config#262
+
+# Invite a guest bot
+python3 -m lazarus.cli invite <cell_id> bot:kimi observer
+
+# Check status
+python3 -m lazarus.cli status
+
+# Close / destroy
+python3 -m lazarus.cli close <cell_id>
+python3 -m lazarus.cli destroy <cell_id>
+```
+
+## Architecture
+
+```
+lazarus/
+├── cell.py              # Cell model, registry, filesystem packager
+├── operator_ctl.py      # summon, invite, team, status, close, destroy
+├── cli.py               # CLI entrypoint
+├── backends/
+│   ├── base.py          # Backend abstraction
+│   └── process_backend.py
+└── tests/
+    └── test_cell.py     # Isolation + lifecycle tests
+```
+
+## Design Invariants
+
+1. **One cell, one project target**
+2. **Cells cannot read each other's `HERMES_HOME` by default**
+3. **Guests join with least privilege**
+4. **Destroyed cells are scrubbed from disk**
+5. **All cell state is persisted in SQLite registry**
+
+## Cell Lifecycle
+
+```
+proposed -> active -> idle -> closing -> archived -> destroyed
+```
+
+## Roles
+
+- `executor` — read/write code, run tools, push commits
+- `observer` — read-only access
+- `director` — can close cells and publish back to Gitea

lazarus/__init__.py

@@ -0,0 +1,5 @@
+"""Lazarus Pit v2.0 — Multi-Agent Resurrection & Teaming Arena."""
+from .cell import Cell, CellState, CellRole
+from .operator_ctl import OperatorCtl
+
+__all__ = ["Cell", "CellState", "CellRole", "OperatorCtl"]

lazarus/backends/__init__.py

@@ -0,0 +1,5 @@
+"""Cell execution backends for Lazarus Pit."""
+from .base import Backend, BackendResult
+from .process_backend import ProcessBackend
+
+__all__ = ["Backend", "BackendResult", "ProcessBackend"]

lazarus/backends/base.py

@@ -0,0 +1,50 @@
+"""
+Backend abstraction for Lazarus Pit cells.
+
+Every backend must implement the same contract so that cells,
+teams, and operators do not need to know how a cell is actually running.
+"""
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class BackendResult:
+    success: bool
+    stdout: str = ""
+    stderr: str = ""
+    pid: Optional[int] = None
+    message: str = ""
+
+
+class Backend(ABC):
+    """Abstract cell backend."""
+
+    name: str = "abstract"
+
+    @abstractmethod
+    def spawn(self, cell_id: str, hermes_home: str, command: list, env: Optional[dict] = None) -> BackendResult:
+        """Spawn the cell process/environment."""
+        ...
+
+    @abstractmethod
+    def probe(self, cell_id: str) -> BackendResult:
+        """Check if the cell is alive and responsive."""
+        ...
+
+    @abstractmethod
+    def logs(self, cell_id: str, tail: int = 50) -> BackendResult:
+        """Return recent logs from the cell."""
+        ...
+
+    @abstractmethod
+    def close(self, cell_id: str) -> BackendResult:
+        """Gracefully close the cell."""
+        ...
+
+    @abstractmethod
+    def destroy(self, cell_id: str) -> BackendResult:
+        """Forcefully destroy the cell and all runtime state."""
+        ...

lazarus/backends/process_backend.py

@@ -0,0 +1,97 @@
+"""
+Process backend for Lazarus Pit.
+
+Fastest spawn/teardown. Isolation comes from unique HERMES_HOME
+and independent Python process. No containers required.
+"""
+
+import os
+import signal
+import subprocess
+from pathlib import Path
+from typing import Optional
+
+from .base import Backend, BackendResult
+
+
+class ProcessBackend(Backend):
+    name = "process"
+
+    def __init__(self, log_dir: Optional[str] = None):
+        self.log_dir = Path(log_dir or "/tmp/lazarus-logs")
+        self.log_dir.mkdir(parents=True, exist_ok=True)
+        self._pids: dict = {}
+
+    def _log_path(self, cell_id: str) -> Path:
+        return self.log_dir / f"{cell_id}.log"
+
+    def spawn(self, cell_id: str, hermes_home: str, command: list, env: Optional[dict] = None) -> BackendResult:
+        log_path = self._log_path(cell_id)
+        run_env = dict(os.environ)
+        run_env["HERMES_HOME"] = hermes_home
+        if env:
+            run_env.update(env)
+
+        try:
+            with open(log_path, "a") as log_file:
+                proc = subprocess.Popen(
+                    command,
+                    env=run_env,
+                    stdout=log_file,
+                    stderr=subprocess.STDOUT,
+                    cwd=hermes_home,
+                )
+            self._pids[cell_id] = proc.pid
+            return BackendResult(
+                success=True,
+                pid=proc.pid,
+                message=f"Spawned {cell_id} with pid {proc.pid}",
+            )
+        except Exception as e:
+            return BackendResult(success=False, stderr=str(e), message=f"Spawn failed: {e}")
+
+    def probe(self, cell_id: str) -> BackendResult:
+        pid = self._pids.get(cell_id)
+        if not pid:
+            return BackendResult(success=False, message="No known pid for cell")
+        try:
+            os.kill(pid, 0)
+            return BackendResult(success=True, pid=pid, message="Process is alive")
+        except OSError:
+            return BackendResult(success=False, pid=pid, message="Process is dead")
+
+    def logs(self, cell_id: str, tail: int = 50) -> BackendResult:
+        log_path = self._log_path(cell_id)
+        if not log_path.exists():
+            return BackendResult(success=True, stdout="", message="No logs yet")
+        try:
+            lines = log_path.read_text().splitlines()
+            return BackendResult(
+                success=True,
+                stdout="\n".join(lines[-tail:]),
+                message=f"Returned last {tail} lines",
+            )
+        except Exception as e:
+            return BackendResult(success=False, stderr=str(e), message=f"Log read failed: {e}")
+
+    def close(self, cell_id: str) -> BackendResult:
+        pid = self._pids.get(cell_id)
+        if not pid:
+            return BackendResult(success=False, message="No known pid for cell")
+        try:
+            os.kill(pid, signal.SIGTERM)
+            return BackendResult(success=True, pid=pid, message="Sent SIGTERM")
+        except OSError as e:
+            return BackendResult(success=False, stderr=str(e), message=f"Close failed: {e}")
+
+    def destroy(self, cell_id: str) -> BackendResult:
+        res = self.close(cell_id)
+        log_path = self._log_path(cell_id)
+        if log_path.exists():
+            log_path.unlink()
+        self._pids.pop(cell_id, None)
+        return BackendResult(
+            success=res.success,
+            pid=res.pid,
+            message=f"Destroyed {cell_id}",
+        )

lazarus/cell.py

@@ -0,0 +1,240 @@
+"""
+Cell model for Lazarus Pit v2.0.
+
+A cell is a bounded execution/living space for one or more participants
+on a single project target. Cells are isolated by filesystem, credentials,
+and optionally process/container boundaries.
+"""
+
+import json
+import shutil
+import uuid
+from dataclasses import dataclass, field, asdict
+from datetime import datetime, timedelta
+from enum import Enum, auto
+from pathlib import Path
+from typing import Dict, List, Optional, Set
+
+
+class CellState(Enum):
+    PROPOSED = "proposed"
+    ACTIVE = "active"
+    IDLE = "idle"
+    CLOSING = "closing"
+    ARCHIVED = "archived"
+    DESTROYED = "destroyed"
+
+
+class CellRole(Enum):
+    EXECUTOR = "executor"    # Can read/write code, run tools, push commits
+    OBSERVER = "observer"    # Read-only cell access
+    DIRECTOR = "director"    # Can assign, close, publish back to Gitea
+
+
+@dataclass
+class CellMember:
+    identity: str           # e.g. "agent:allegro", "bot:kimi", "human:Alexander"
+    role: CellRole
+    joined_at: str
+
+
+@dataclass
+class Cell:
+    cell_id: str
+    project: str            # e.g. "Timmy_Foundation/timmy-config#262"
+    owner: str              # e.g. "human:Alexander"
+    backend: str            # "process", "venv", "docker", "remote"
+    state: CellState
+    created_at: str
+    ttl_minutes: int
+    members: List[CellMember] = field(default_factory=list)
+    hermes_home: Optional[str] = None
+    workspace_path: Optional[str] = None
+    shared_notes_path: Optional[str] = None
+    shared_artifacts_path: Optional[str] = None
+    decisions_path: Optional[str] = None
+    archived_at: Optional[str] = None
+    destroyed_at: Optional[str] = None
+
+    def is_expired(self) -> bool:
+        if self.state in (CellState.CLOSING, CellState.ARCHIVED, CellState.DESTROYED):
+            return False
+        created = datetime.fromisoformat(self.created_at.replace("Z", "+00:00"))
+        return datetime.now().astimezone() > created + timedelta(minutes=self.ttl_minutes)
+
+    def to_dict(self) -> dict:
+        d = asdict(self)
+        d["state"] = self.state.value
+        d["members"] = [
+            {**asdict(m), "role": m.role.value} for m in self.members
+        ]
+        return d
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "Cell":
+        d = dict(d)
+        d["state"] = CellState(d["state"])
+        d["members"] = [
+            CellMember(
+                identity=m["identity"],
+                role=CellRole(m["role"]),
+                joined_at=m["joined_at"],
+            )
+            for m in d.get("members", [])
+        ]
+        return cls(**d)
+
+
+class CellRegistry:
+    """SQLite-backed registry of all cells."""
+
+    def __init__(self, db_path: Optional[str] = None):
+        self.db_path = Path(db_path or "/root/.lazarus_registry.sqlite")
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init_db()
+
+    def _init_db(self):
+        import sqlite3
+        with sqlite3.connect(str(self.db_path)) as conn:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS cells (
+                    cell_id TEXT PRIMARY KEY,
+                    project TEXT NOT NULL,
+                    owner TEXT NOT NULL,
+                    backend TEXT NOT NULL,
+                    state TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    ttl_minutes INTEGER NOT NULL,
+                    members TEXT,
+                    hermes_home TEXT,
+                    workspace_path TEXT,
+                    shared_notes_path TEXT,
+                    shared_artifacts_path TEXT,
+                    decisions_path TEXT,
+                    archived_at TEXT,
+                    destroyed_at TEXT
+                )
+                """
+            )
+
+    def save(self, cell: Cell):
+        import sqlite3
+        with sqlite3.connect(str(self.db_path)) as conn:
+            conn.execute(
+                """
+                INSERT OR REPLACE INTO cells (
+                    cell_id, project, owner, backend, state, created_at, ttl_minutes,
+                    members, hermes_home, workspace_path, shared_notes_path,
+                    shared_artifacts_path, decisions_path, archived_at, destroyed_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    cell.cell_id,
+                    cell.project,
+                    cell.owner,
+                    cell.backend,
+                    cell.state.value,
+                    cell.created_at,
+                    cell.ttl_minutes,
+                    json.dumps(cell.to_dict()["members"]),
+                    cell.hermes_home,
+                    cell.workspace_path,
+                    cell.shared_notes_path,
+                    cell.shared_artifacts_path,
+                    cell.decisions_path,
+                    cell.archived_at,
+                    cell.destroyed_at,
+                ),
+            )
+
+    def load(self, cell_id: str) -> Optional[Cell]:
+        import sqlite3
+        with sqlite3.connect(str(self.db_path)) as conn:
+            row = conn.execute(
+                "SELECT * FROM cells WHERE cell_id = ?", (cell_id,)
+            ).fetchone()
+        if not row:
+            return None
+        keys = [
+            "cell_id", "project", "owner", "backend", "state", "created_at",
+            "ttl_minutes", "members", "hermes_home", "workspace_path",
+            "shared_notes_path", "shared_artifacts_path", "decisions_path",
+            "archived_at", "destroyed_at",
+        ]
+        d = dict(zip(keys, row))
+        d["members"] = json.loads(d["members"])
+        return Cell.from_dict(d)
+
+    def list_active(self) -> List[Cell]:
+        import sqlite3
+        with sqlite3.connect(str(self.db_path)) as conn:
+            rows = conn.execute(
+                "SELECT * FROM cells WHERE state NOT IN ('destroyed', 'archived')"
+            ).fetchall()
+        keys = [
+            "cell_id", "project", "owner", "backend", "state", "created_at",
+            "ttl_minutes", "members", "hermes_home", "workspace_path",
+            "shared_notes_path", "shared_artifacts_path", "decisions_path",
+            "archived_at", "destroyed_at",
+        ]
+        cells = []
+        for row in rows:
+            d = dict(zip(keys, row))
+            d["members"] = json.loads(d["members"])
+            cells.append(Cell.from_dict(d))
+        return cells
+
+    def delete(self, cell_id: str):
+        import sqlite3
+        with sqlite3.connect(str(self.db_path)) as conn:
+            conn.execute("DELETE FROM cells WHERE cell_id = ?", (cell_id,))
+
+
+class CellPackager:
+    """Creates and destroys per-cell filesystem isolation."""
+
+    ROOT = Path("/tmp/lazarus-cells")
+
+    def __init__(self, root: Optional[str] = None):
+        self.root = Path(root or self.ROOT)
+
+    def create(self, cell_id: str, backend: str = "process") -> Cell:
+        cell_home = self.root / cell_id
+        workspace = cell_home / "workspace"
+        shared_notes = cell_home / "shared" / "notes.md"
+        shared_artifacts = cell_home / "shared" / "artifacts"
+        decisions = cell_home / "shared" / "decisions.jsonl"
+        hermes_home = cell_home / ".hermes"
+
+        for p in [workspace, shared_artifacts, hermes_home]:
+            p.mkdir(parents=True, exist_ok=True)
+
+        if not shared_notes.exists():
+            shared_notes.write_text(f"# Cell {cell_id} — Shared Notes\n\n")
+
+        if not decisions.exists():
+            decisions.write_text("")
+
+        now = datetime.now().astimezone().isoformat()
+        return Cell(
+            cell_id=cell_id,
+            project="",
+            owner="",
+            backend=backend,
+            state=CellState.PROPOSED,
+            created_at=now,
+            ttl_minutes=60,
+            hermes_home=str(hermes_home),
+            workspace_path=str(workspace),
+            shared_notes_path=str(shared_notes),
+            shared_artifacts_path=str(shared_artifacts),
+            decisions_path=str(decisions),
+        )
+
+    def destroy(self, cell_id: str) -> bool:
+        cell_home = self.root / cell_id
+        if cell_home.exists():
+            shutil.rmtree(cell_home)
+            return True
+        return False

lazarus/cli.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+"""CLI entrypoint for Lazarus Pit operator control surface."""
+
+import json
+import sys
+
+from .operator_ctl import OperatorCtl
+
+
+def main():
+    ctl = OperatorCtl()
+    result = ctl.run_cli(sys.argv[1:])
+    print(json.dumps(result, indent=2))
+    sys.exit(0 if result.get("success") else 1)
+
+
+if __name__ == "__main__":
+    main()

lazarus/operator_ctl.py

@@ -0,0 +1,211 @@
+"""
+Operator control surface for Lazarus Pit v2.0.
+
+Provides the command grammar and API for:
+  summon, invite, team, status, close, destroy
+"""
+
+import json
+import uuid
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional
+
+from .cell import Cell, CellMember, CellPackager, CellRegistry, CellRole, CellState
+from .backends.process_backend import ProcessBackend
+
+
+class OperatorCtl:
+    """The operator's command surface for the Lazarus Pit."""
+
+    def __init__(self, registry: Optional[CellRegistry] = None, packager: Optional[CellPackager] = None):
+        self.registry = registry or CellRegistry()
+        self.packager = packager or CellPackager()
+        self.backend = ProcessBackend()
+
+    # ------------------------------------------------------------------
+    # Commands
+    # ------------------------------------------------------------------
+    def summon(self, agent: str, project: str, owner: str = "human:Alexander", backend: str = "process", ttl_minutes: int = 60) -> dict:
+        """Summon a single agent into a fresh resurrection cell."""
+        cell_id = f"laz-{uuid.uuid4().hex[:8]}"
+        cell = self.packager.create(cell_id, backend=backend)
+        cell.project = project
+        cell.owner = owner
+        cell.ttl_minutes = ttl_minutes
+        cell.state = CellState.ACTIVE
+        cell.members.append(CellMember(
+            identity=f"agent:{agent}",
+            role=CellRole.EXECUTOR,
+            joined_at=datetime.now().astimezone().isoformat(),
+        ))
+
+        # Spawn a simple keep-alive process (real agent bootstrap would go here)
+        res = self.backend.spawn(
+            cell_id=cell_id,
+            hermes_home=cell.hermes_home,
+            command=["python3", "-c", "import time; time.sleep(86400)"],
+        )
+
+        self.registry.save(cell)
+        return {
+            "success": res.success,
+            "cell_id": cell_id,
+            "project": project,
+            "agent": agent,
+            "backend": backend,
+            "pid": res.pid,
+            "message": res.message,
+        }
+
+    def invite(self, cell_id: str, guest: str, role: str, invited_by: str = "human:Alexander") -> dict:
+        """Invite a guest bot or human into an active cell."""
+        cell = self.registry.load(cell_id)
+        if not cell:
+            return {"success": False, "message": f"Cell {cell_id} not found"}
+        if cell.state not in (CellState.PROPOSED, CellState.ACTIVE, CellState.IDLE):
+            return {"success": False, "message": f"Cell {cell_id} is not accepting invites (state={cell.state.value})"}
+
+        role_enum = CellRole(role)
+        cell.members.append(CellMember(
+            identity=guest,
+            role=role_enum,
+            joined_at=datetime.now().astimezone().isoformat(),
+        ))
+        cell.state = CellState.ACTIVE
+        self.registry.save(cell)
+        return {
+            "success": True,
+            "cell_id": cell_id,
+            "guest": guest,
+            "role": role_enum.value,
+            "invited_by": invited_by,
+            "message": f"Invited {guest} as {role_enum.value} to {cell_id}",
+        }
+
+    def team(self, agents: List[str], project: str, owner: str = "human:Alexander", backend: str = "process", ttl_minutes: int = 120) -> dict:
+        """Form a multi-agent team in one cell against a project."""
+        cell_id = f"laz-{uuid.uuid4().hex[:8]}"
+        cell = self.packager.create(cell_id, backend=backend)
+        cell.project = project
+        cell.owner = owner
+        cell.ttl_minutes = ttl_minutes
+        cell.state = CellState.ACTIVE
+
+        for agent in agents:
+            cell.members.append(CellMember(
+                identity=f"agent:{agent}",
+                role=CellRole.EXECUTOR,
+                joined_at=datetime.now().astimezone().isoformat(),
+            ))
+
+        res = self.backend.spawn(
+            cell_id=cell_id,
+            hermes_home=cell.hermes_home,
+            command=["python3", "-c", "import time; time.sleep(86400)"],
+        )
+
+        self.registry.save(cell)
+        return {
+            "success": res.success,
+            "cell_id": cell_id,
+            "project": project,
+            "agents": agents,
+            "backend": backend,
+            "pid": res.pid,
+            "message": res.message,
+        }
+
+    def status(self, cell_id: Optional[str] = None) -> dict:
+        """Show status of one cell or all active cells."""
+        if cell_id:
+            cell = self.registry.load(cell_id)
+            if not cell:
+                return {"success": False, "message": f"Cell {cell_id} not found"}
+            probe = self.backend.probe(cell_id)
+            return {
+                "success": True,
+                "cell": cell.to_dict(),
+                "probe": probe.message,
+                "expired": cell.is_expired(),
+            }
+
+        active = self.registry.list_active()
+        cells = []
+        for cell in active:
+            probe = self.backend.probe(cell.cell_id)
+            cells.append({
+                "cell_id": cell.cell_id,
+                "project": cell.project,
+                "backend": cell.backend,
+                "state": cell.state.value,
+                "members": [m.identity for m in cell.members],
+                "probe": probe.message,
+                "expired": cell.is_expired(),
+            })
+        return {"success": True, "active_cells": cells}
+
+    def close(self, cell_id: str, closed_by: str = "human:Alexander") -> dict:
+        """Gracefully close a cell."""
+        cell = self.registry.load(cell_id)
+        if not cell:
+            return {"success": False, "message": f"Cell {cell_id} not found"}
+
+        res = self.backend.close(cell_id)
+        cell.state = CellState.CLOSING
+        self.registry.save(cell)
+        return {
+            "success": res.success,
+            "cell_id": cell_id,
+            "message": f"Closed {cell_id} by {closed_by}",
+        }
+
+    def destroy(self, cell_id: str, destroyed_by: str = "human:Alexander") -> dict:
+        """Forcefully destroy a cell and all runtime state."""
+        cell = self.registry.load(cell_id)
+        if not cell:
+            return {"success": False, "message": f"Cell {cell_id} not found"}
+
+        res = self.backend.destroy(cell_id)
+        self.packager.destroy(cell_id)
+        cell.state = CellState.DESTROYED
+        cell.destroyed_at = datetime.now().astimezone().isoformat()
+        self.registry.save(cell)
+        return {
+            "success": True,
+            "cell_id": cell_id,
+            "message": f"Destroyed {cell_id} by {destroyed_by}",
+        }
+
+    # ------------------------------------------------------------------
+    # CLI helpers
+    # ------------------------------------------------------------------
+    def run_cli(self, args: List[str]) -> dict:
+        if not args:
+            return self.status()
+        cmd = args[0]
+        if cmd == "summon" and len(args) >= 3:
+            return self.summon(agent=args[1], project=args[2])
+        if cmd == "invite" and len(args) >= 4:
+            return self.invite(cell_id=args[1], guest=args[2], role=args[3])
+        if cmd == "team" and len(args) >= 3:
+            agents = args[1].split("+")
+            project = args[2]
+            return self.team(agents=agents, project=project)
+        if cmd == "status":
+            return self.status(cell_id=args[1] if len(args) > 1 else None)
+        if cmd == "close" and len(args) >= 2:
+            return self.close(cell_id=args[1])
+        if cmd == "destroy" and len(args) >= 2:
+            return self.destroy(cell_id=args[1])
+        return {
+            "success": False,
+            "message": (
+                "Unknown command. Usage:\n"
+                "  lazarus summon <agent> <project>\n"
+                "  lazarus invite <cell_id> <guest> <role>\n"
+                "  lazarus team <agent1+agent2> <project>\n"
+                "  lazarus status [cell_id]\n"
+                "  lazarus close <cell_id>\n"
+                "  lazarus destroy <cell_id>\n"
+            )
+        }

lazarus/tests/__init__.py

@@ -0,0 +1 @@
+"""Tests for Lazarus Pit."""

lazarus/tests/test_cell.py

@@ -0,0 +1,186 @@
+#!/usr/bin/env python3
+"""Tests for Lazarus Pit cell isolation and registry."""
+
+import os
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from lazarus.cell import Cell, CellMember, CellPackager, CellRegistry, CellRole, CellState
+from lazarus.operator_ctl import OperatorCtl
+from lazarus.backends.process_backend import ProcessBackend
+
+
+class TestCellPackager(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.packager = CellPackager(root=self.tmpdir.name)
+
+    def tearDown(self):
+        self.tmpdir.cleanup()
+
+    def test_create_isolated_paths(self):
+        cell = self.packager.create("laz-test-001", backend="process")
+        self.assertTrue(os.path.isdir(cell.hermes_home))
+        self.assertTrue(os.path.isdir(cell.workspace_path))
+        self.assertTrue(os.path.isfile(cell.shared_notes_path))
+        self.assertTrue(os.path.isdir(cell.shared_artifacts_path))
+        self.assertTrue(os.path.isfile(cell.decisions_path))
+
+    def test_destroy_cleans_paths(self):
+        cell = self.packager.create("laz-test-002", backend="process")
+        self.assertTrue(os.path.isdir(cell.hermes_home))
+        self.packager.destroy("laz-test-002")
+        self.assertFalse(os.path.exists(cell.hermes_home))
+
+    def test_cells_are_separate(self):
+        c1 = self.packager.create("laz-a", backend="process")
+        c2 = self.packager.create("laz-b", backend="process")
+        self.assertNotEqual(c1.hermes_home, c2.hermes_home)
+        self.assertNotEqual(c1.workspace_path, c2.workspace_path)
+
+
+class TestCellRegistry(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.registry = CellRegistry(db_path=os.path.join(self.tmpdir.name, "registry.sqlite"))
+
+    def tearDown(self):
+        self.tmpdir.cleanup()
+
+    def test_save_and_load(self):
+        cell = Cell(
+            cell_id="laz-reg-001",
+            project="Timmy_Foundation/test#1",
+            owner="human:Alexander",
+            backend="process",
+            state=CellState.ACTIVE,
+            created_at=datetime.now().astimezone().isoformat(),
+            ttl_minutes=60,
+            members=[CellMember("agent:allegro", CellRole.EXECUTOR, datetime.now().astimezone().isoformat())],
+        )
+        self.registry.save(cell)
+        loaded = self.registry.load("laz-reg-001")
+        self.assertEqual(loaded.cell_id, "laz-reg-001")
+        self.assertEqual(loaded.members[0].role, CellRole.EXECUTOR)
+
+    def test_list_active(self):
+        for sid, state in [("a", CellState.ACTIVE), ("b", CellState.DESTROYED)]:
+            cell = Cell(
+                cell_id=f"laz-{sid}",
+                project="test",
+                owner="human:Alexander",
+                backend="process",
+                state=state,
+                created_at=datetime.now().astimezone().isoformat(),
+                ttl_minutes=60,
+            )
+            self.registry.save(cell)
+        active = self.registry.list_active()
+        ids = {c.cell_id for c in active}
+        self.assertIn("laz-a", ids)
+        self.assertNotIn("laz-b", ids)
+
+    def test_ttl_expiry(self):
+        created = (datetime.now().astimezone() - timedelta(minutes=61)).isoformat()
+        cell = Cell(
+            cell_id="laz-exp",
+            project="test",
+            owner="human:Alexander",
+            backend="process",
+            state=CellState.ACTIVE,
+            created_at=created,
+            ttl_minutes=60,
+        )
+        self.assertTrue(cell.is_expired())
+
+    def test_ttl_not_expired(self):
+        cell = Cell(
+            cell_id="laz-ok",
+            project="test",
+            owner="human:Alexander",
+            backend="process",
+            state=CellState.ACTIVE,
+            created_at=datetime.now().astimezone().isoformat(),
+            ttl_minutes=60,
+        )
+        self.assertFalse(cell.is_expired())
+
+
+class TestOperatorCtl(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.TemporaryDirectory()
+        self.ctl = OperatorCtl(
+            registry=CellRegistry(db_path=os.path.join(self.tmpdir.name, "registry.sqlite")),
+            packager=CellPackager(root=self.tmpdir.name),
+        )
+
+    def tearDown(self):
+        # Destroy any cells we created
+        for cell in self.ctl.registry.list_active():
+            self.ctl.destroy(cell.cell_id)
+        self.tmpdir.cleanup()
+
+    def test_summon_creates_cell(self):
+        res = self.ctl.summon("allegro", "Timmy_Foundation/test#1")
+        self.assertTrue(res["success"])
+        self.assertIn("cell_id", res)
+        cell = self.ctl.registry.load(res["cell_id"])
+        self.assertEqual(cell.members[0].identity, "agent:allegro")
+        self.assertEqual(cell.state, CellState.ACTIVE)
+        # Clean up
+        self.ctl.destroy(res["cell_id"])
+
+    def test_team_creates_multi_agent_cell(self):
+        res = self.ctl.team(["allegro", "ezra"], "Timmy_Foundation/test#2")
+        self.assertTrue(res["success"])
+        cell = self.ctl.registry.load(res["cell_id"])
+        self.assertEqual(len(cell.members), 2)
+        identities = {m.identity for m in cell.members}
+        self.assertEqual(identities, {"agent:allegro", "agent:ezra"})
+        self.ctl.destroy(res["cell_id"])
+
+    def test_invite_adds_guest(self):
+        res = self.ctl.summon("allegro", "Timmy_Foundation/test#3")
+        cell_id = res["cell_id"]
+        invite = self.ctl.invite(cell_id, "bot:kimi", "observer")
+        self.assertTrue(invite["success"])
+        cell = self.ctl.registry.load(cell_id)
+        roles = {m.identity: m.role for m in cell.members}
+        self.assertEqual(roles["bot:kimi"], CellRole.OBSERVER)
+        self.ctl.destroy(cell_id)
+
+    def test_status_shows_active_cells(self):
+        res = self.ctl.summon("allegro", "Timmy_Foundation/test#4")
+        status = self.ctl.status()
+        self.assertTrue(status["success"])
+        ids = {c["cell_id"] for c in status["active_cells"]}
+        self.assertIn(res["cell_id"], ids)
+        self.ctl.destroy(res["cell_id"])
+
+    def test_close_then_destroy(self):
+        res = self.ctl.summon("allegro", "Timmy_Foundation/test#5")
+        cell_id = res["cell_id"]
+        close_res = self.ctl.close(cell_id)
+        self.assertTrue(close_res["success"])
+        destroy_res = self.ctl.destroy(cell_id)
+        self.assertTrue(destroy_res["success"])
+        cell = self.ctl.registry.load(cell_id)
+        self.assertEqual(cell.state, CellState.DESTROYED)
+        self.assertFalse(os.path.exists(cell.hermes_home))
+
+    def test_destroy_scrubs_filesystem(self):
+        res = self.ctl.summon("allegro", "Timmy_Foundation/test#6")
+        cell_id = res["cell_id"]
+        cell = self.ctl.registry.load(cell_id)
+        hermes = cell.hermes_home
+        self.assertTrue(os.path.isdir(hermes))
+        self.ctl.destroy(cell_id)
+        self.assertFalse(os.path.exists(hermes))
+
+
+if __name__ == "__main__":
+    unittest.main()

nostr-bridge/bridge_mvp.py

@@ -19,7 +19,7 @@ except ImportError as e:
     sys.exit(1)
 
 # Configuration
-GITEA = "http://143.198.27.163:3000"
+GITEA = "https://forge.alexanderwhitestone.com"
 RELAY_URL = "ws://localhost:2929"  # Local relay
 POLL_INTERVAL = 60  # Seconds between polls
 ALLOWED_PUBKEYS = []  # Will load from keystore

scripts/architecture_linter.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+import os
+import sys
+import re
+
+# Architecture Linter
+# Ensuring all changes align with the Frontier Local Agenda.
+
+SOVEREIGN_RULES = [
+    (r"https?://(api\.openai\.com|api\.anthropic\.com)", "CRITICAL: External cloud API detected. Use local custom_provider instead."),
+    (r"provider: (openai|anthropic)", "WARNING: Direct cloud provider used. Ensure fallback_model is configured."),
+    (r"api_key: ['"][^'"\s]{10,}['"]", "SECURITY: Hardcoded API key detected. Use environment variables.")
+]
+
+def lint_file(path):
+    print(f"Linting {path}...")
+    content = open(path).read()
+    violations = 0
+    for pattern, msg in SOVEREIGN_RULES:
+        if re.search(pattern, content):
+            print(f"  [!] {msg}")
+            violations += 1
+    return violations
+
+def main():
+    print("--- Ezra's Architecture Linter ---")
+    files = [f for f in sys.argv[1:] if os.path.isfile(f)]
+    total_violations = sum(lint_file(f) for f in files)
+    print(f"\nLinting complete. Total violations: {total_violations}")
+    sys.exit(1 if total_violations > 0 else 0)
+
+if __name__ == "__main__":
+    main()

scripts/nostur_status_query.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+"""
+Nostur Status Query MVP
+Read-only status responses sourced from Gitea truth.
+"""
+import json
+import os
+import sys
+import urllib.request
+from datetime import datetime
+
+# Configuration
+GITEA_URL = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com/api/v1")
+GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "f7bcdaf878d479ad7747873ff6739a9bb89e3f80")
+REPO_OWNER = "Timmy_Foundation"
+REPO_NAME = "timmy-config"
+
+def gitea_get(path):
+    if not GITEA_TOKEN:
+        raise RuntimeError("GITEA_TOKEN not set")
+    
+    url = f"{GITEA_URL}{path}"
+    headers = {"Authorization": f"token {GITEA_TOKEN}"}
+    req = urllib.request.Request(url, headers=headers)
+    
+    try:
+        with urllib.request.urlopen(req, timeout=15) as resp:
+            return json.loads(resp.read().decode())
+    except Exception as e:
+        print(f"Error fetching from Gitea: {e}", file=sys.stderr)
+        return None
+
+def get_status():
+    path = f"/repos/{REPO_OWNER}/{REPO_NAME}/issues?state=open&limit=50"
+    issues = gitea_get(path)
+    
+    if not issues:
+        return "⚠️ Error: Could not fetch status from Gitea."
+
+    # 1. Active Epics
+    active_epics = [
+        i['title'].replace('[EPIC]', '').strip() 
+        for i in issues 
+        if '[EPIC]' in i['title'] or any(l['name'] == 'epic' for l in i.get('labels', []))
+    ][:2]
+
+    # 2. Blockers / Critical (Bugs, Security, Ops)
+    blockers = [
+        i['title'].strip()
+        for i in issues
+        if any(tag in i['title'] for tag in ['[BUG]', '[SECURITY]', '[OPS]']) or any(l['name'] == 'blocker' for l in i.get('labels', []))
+    ][:2]
+
+    # 3. Priority Queue (Top 3)
+    priority_queue = [
+        f"#{i['number']}: {i['title']}"
+        for i in issues[:3]
+    ]
+
+    # Format compact response for mobile
+    now = datetime.now().strftime("%H:%M:%S")
+    
+    lines = ["[TIMMY STATUS]"]
+    lines.append(f"EPICS: {' | '.join(active_epics) if active_epics else 'None'}")
+    lines.append(f"BLOCKERS: {' | '.join(blockers) if blockers else 'None'}")
+    lines.append(f"QUEUE: {' | '.join(priority_queue)}")
+    lines.append(f"UPDATED: {now}")
+    
+    return "\n".join(lines)
+
+if __name__ == "__main__":
+    # If called with 'json', return JSON payload
+    if len(sys.argv) > 1 and sys.argv[1] == "--json":
+        path = f"/repos/{REPO_OWNER}/{REPO_NAME}/issues?state=open&limit=50"
+        issues = gitea_get(path)
+        if not issues:
+            print(json.dumps({"error": "fetch failed"}))
+            sys.exit(1)
+            
+        data = {
+            "status": "active",
+            "epics": [i['title'] for i in issues if '[EPIC]' in i['title']][:2],
+            "blockers": [i['title'] for i in issues if any(tag in i['title'] for tag in ['[BUG]', '[SECURITY]', '[OPS]'])][:2],
+            "queue": [f"#{i['number']}: {i['title']}" for i in issues[:3]],
+            "timestamp": datetime.now().isoformat()
+        }
+        print(json.dumps(data, indent=2))
+    else:
+        print(get_status())

tasks.py

@@ -2126,3 +2126,39 @@ def cross_review_prs():
                 continue
 
     return {"reviews": len(results), "details": results}
+
+@huey.periodic_task(crontab(hour="0", minute="0"))
+def metrics_tick():
+    """Force Multiplier 13: Fleet Cost & Velocity Tracker.
+    
+    Calculates daily velocity (issues closed, PRs merged) and logs it to a metrics file.
+    """
+    gitea = get_gitea_client()
+    repos = ["Timmy_Foundation/timmy-config", "Timmy_Foundation/timmy-home", "Timmy_Foundation/the-nexus"]
+    
+    daily_stats = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "issues_closed": 0,
+        "prs_merged": 0,
+        "repo_stats": {}
+    }
+    
+    for repo in repos:
+        # Get issues closed in last 24h
+        closed_issues = gitea.get_closed_issues(repo, since="24h")
+        merged_prs = gitea.get_merged_prs(repo, since="24h")
+        
+        daily_stats["issues_closed"] += len(closed_issues)
+        daily_stats["prs_merged"] += len(merged_prs)
+        daily_stats["repo_stats"][repo] = {
+            "issues": len(closed_issues),
+            "prs": len(merged_prs)
+        }
+        
+    metrics_file = TIMMY_HOME / "logs" / "metrics.jsonl"
+    metrics_file.parent.mkdir(parents=True, exist_ok=True)
+    
+    with open(metrics_file, "a") as f:
+        f.write(json.dumps(daily_stats) + "\n")
+    
+    audit_log("metrics_logged", "system", daily_stats, confidence="High")

workforce-manager.py

@@ -0,0 +1,633 @@
+#!/usr/bin/env python3
+"""
+Workforce Manager - Epic #204 / Milestone #218
+
+Reads fleet routing, Wolf evaluation scores, and open Gitea issues across
+Timmy_Foundation repos. Assigns each issue to the best-available agent,
+tracks success rates, and dispatches work.
+
+Usage:
+    python workforce-manager.py              # Scan, assign, dispatch
+    python workforce-manager.py --dry-run    # Show assignments without dispatching
+    python workforce-manager.py --status     # Show agent status and open issue count
+    python workforce-manager.py --cron       # Run silently, save to log
+"""
+
+import argparse
+import json
+import logging
+import os
+import sys
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+try:
+    import requests
+except ImportError:
+    print("FATAL: requests is required. pip install requests", file=sys.stderr)
+    sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+FLEET_ROUTING_PATH = Path.home() / ".hermes" / "fleet-routing.json"
+WOLF_RESULTS_DIR = Path.home() / ".hermes" / "wolf" / "results"
+GITEA_TOKEN_PATH = Path.home() / ".hermes" / "gitea_token_vps"
+GITEA_API_BASE = "https://forge.alexanderwhitestone.com/api/v1"
+WORKFORCE_STATE_PATH = Path.home() / ".hermes" / "workforce-state.json"
+ORG_NAME = "Timmy_Foundation"
+
+# Role-to-agent-role mapping heuristics
+ROLE_KEYWORDS = {
+    "code-generation": [
+        "code", "implement", "feature", "function", "class", "script",
+        "build", "create", "add", "module", "component",
+    ],
+    "issue-triage": [
+        "triage", "categorize", "tag", "label", "organize",
+        "backlog", "sort", "prioritize", "review issue",
+    ],
+    "on-request-queries": [
+        "query", "search", "lookup", "find", "check",
+        "info", "report", "status",
+    ],
+    "devops": [
+        "deploy", "ci", "cd", "pipeline", "docker", "container",
+        "server", "infrastructure", "config", "nginx", "cron",
+        "setup", "install", "environment", "provision",
+        "build", "release", "workflow",
+    ],
+    "documentation": [
+        "doc", "readme", "document", "write", "guide",
+        "spec", "wiki", "changelog", "tutorial",
+        "explain", "describe",
+    ],
+    "code-review": [
+        "review", "refactor", "fix", "bug", "debug",
+        "test", "lint", "style", "improve",
+        "clean up", "optimize", "performance",
+    ],
+    "triage-routing": [
+        "route", "assign", "triage", "dispatch",
+        "organize", "categorize",
+    ],
+    "small-tasks": [
+        "small", "quick", "minor", "typo", "label",
+        "update", "rename", "cleanup",
+    ],
+    "inactive": [],
+    "unknown": [],
+}
+
+# Priority keywords (higher = more urgent, route to more capable agent)
+PRIORITY_KEYWORDS = {
+    "critical": 5,
+    "urgent": 4,
+    "block": 4,
+    "bug": 3,
+    "fix": 3,
+    "security": 5,
+    "deploy": 2,
+    "feature": 1,
+    "enhancement": 1,
+    "documentation": 1,
+    "cleanup": 0,
+}
+
+# Cost tier priority (lower index = prefer first)
+TIER_ORDER = ["free", "cheap", "prepaid", "unknown"]
+
+
+# ---------------------------------------------------------------------------
+# Data loading
+# ---------------------------------------------------------------------------
+
+def load_json(path: Path) -> Any:
+    if not path.exists():
+        logging.warning("File not found: %s", path)
+        return None
+    with open(path) as f:
+        return json.load(f)
+
+
+def load_fleet_routing() -> List[dict]:
+    data = load_json(FLEET_ROUTING_PATH)
+    if data and "agents" in data:
+        return data["agents"]
+    return []
+
+
+def load_wolf_scores() -> Dict[str, dict]:
+    """Load Wolf evaluation scores from results directory."""
+    scores: Dict[str, dict] = {}
+    if not WOLF_RESULTS_DIR.exists():
+        return scores
+    for f in sorted(WOLF_RESULTS_DIR.glob("*.json")):
+        data = load_json(f)
+        if data and "model_scores" in data:
+            for entry in data["model_scores"]:
+                model = entry.get("model", "")
+                if model:
+                    scores[model] = entry
+    return scores
+
+
+def load_workforce_state() -> dict:
+    if WORKFORCE_STATE_PATH.exists():
+        return load_json(WORKFORCE_STATE_PATH) or {}
+    return {"assignments": [], "agent_stats": {}, "last_run": None}
+
+
+def save_workforce_state(state: dict) -> None:
+    WORKFORCE_STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with open(WORKFORCE_STATE_PATH, "w") as f:
+        json.dump(state, f, indent=2)
+    logging.info("Workforce state saved to %s", WORKFORCE_STATE_PATH)
+
+
+# ---------------------------------------------------------------------------
+# Gitea API
+# ---------------------------------------------------------------------------
+
+class GiteaAPI:
+    """Thin wrapper for Gitea REST API."""
+
+    def __init__(self, token: str, base_url: str = GITEA_API_BASE):
+        self.base_url = base_url.rstrip("/")
+        self.session = requests.Session()
+        self.session.headers.update({
+            "Authorization": f"token {token}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        })
+
+    def _get(self, path: str, params: Optional[dict] = None) -> Any:
+        r = self.session.get(f"{self.base_url}{path}", params=params)
+        r.raise_for_status()
+        return r.json()
+
+    def _post(self, path: str, data: dict) -> Any:
+        r = self.session.post(f"{self.base_url}{path}", json=data)
+        r.raise_for_status()
+        return r.json()
+
+    def _patch(self, path: str, data: dict) -> Any:
+        r = self.session.patch(f"{self.base_url}{path}", json=data)
+        r.raise_for_status()
+        return r.json()
+
+    def get_org_repos(self, org: str) -> List[dict]:
+        return self._get(f"/orgs/{org}/repos", params={"limit": 100})
+
+    def get_open_issues(self, owner: str, repo: str, page: int = 1) -> List[dict]:
+        params = {"state": "open", "type": "issues", "limit": 50, "page": page}
+        return self._get(f"/repos/{owner}/{repo}/issues", params=params)
+
+    def get_all_open_issues(self, org: str) -> List[dict]:
+        """Fetch all open issues across all org repos."""
+        repos = self.get_org_repos(org)
+        all_issues = []
+        for repo in repos:
+            name = repo["name"]
+            try:
+                # Paginate through all issues
+                page = 1
+                while True:
+                    issues = self.get_open_issues(org, name, page=page)
+                    if not issues:
+                        break
+                    all_issues.extend(issues)
+                    if len(issues) < 50:
+                        break
+                    page += 1
+                logging.info("Loaded %d open issues from %s/%s", len(all_issues), org, name)
+            except Exception as exc:
+                logging.warning("Failed to load issues from %s/%s: %s", org, name, exc)
+        return all_issues
+
+    def add_issue_comment(self, owner: str, repo: str, issue_num: int, body: str) -> dict:
+        return self._post(f"/repos/{owner}/{repo}/issues/{issue_num}/comments", {"body": body})
+
+    def add_issue_label(self, owner: str, repo: str, issue_num: int, label: str) -> dict:
+        return self._post(
+            f"/repos/{owner}/{repo}/issues/{issue_num}/labels",
+            {"labels": [label]},
+        )
+
+    def assign_issue(self, owner: str, repo: str, issue_num: int, assignee: str) -> dict:
+        return self._patch(
+            f"/repos/{owner}/{repo}/issues/{issue_num}",
+            {"assignees": [assignee]},
+        )
+
+
+# ---------------------------------------------------------------------------
+# Scoring & Assignment Logic
+# ---------------------------------------------------------------------------
+
+def classify_issue(issue: dict) -> str:
+    """Determine the best agent role for an issue based on title/body."""
+    title = (issue.get("title", "") or "").lower()
+    body = (issue.get("body", "") or "").lower()
+    text = f"{title} {body}"
+    labels = [l.get("name", "").lower() for l in issue.get("labels", [])]
+    text += " " + " ".join(labels)
+
+    best_role = "small-tasks"  # default
+    best_score = 0
+
+    for role, keywords in ROLE_KEYWORDS.items():
+        if not keywords:
+            continue
+        score = sum(2 for kw in keywords if kw in text)
+        # Boost if a matching label exists
+        for label in labels:
+            if any(kw in label for kw in keywords):
+                score += 3
+        if score > best_score:
+            best_score = score
+            best_role = role
+
+    return best_role
+
+
+def compute_priority(issue: dict) -> int:
+    """Compute issue priority from keywords."""
+    title = (issue.get("title", "") or "").lower()
+    body = (issue.get("body", "") or "").lower()
+    text = f"{title} {body}"
+    return sum(v for k, v in PRIORITY_KEYWORDS.items() if k in text)
+
+
+def score_agent_for_issue(agent: dict, role: str, wolf_scores: dict, priority: int) -> float:
+    """Score how well an agent matches an issue. Higher is better."""
+    score = 0.0
+
+    # Primary: role match
+    agent_role = agent.get("role", "unknown")
+    if agent_role == role:
+        score += 10.0
+    elif agent_role == "small-tasks" and role in ("issue-triage", "on-request-queries"):
+        score += 6.0
+    elif agent_role == "triage-routing" and role in ("issue-triage", "triage-routing"):
+        score += 8.0
+    elif agent_role == "code-generation" and role in ("code-review", "devops"):
+        score += 4.0
+
+    # Wolf quality bonus
+    model = agent.get("model", "")
+    wolf_entry = None
+    for wm, ws in wolf_scores.items():
+        if model and model.lower() in wm.lower():
+            wolf_entry = ws
+            break
+    if wolf_entry and wolf_entry.get("success"):
+        score += wolf_entry.get("total", 0) * 3.0
+
+    # Cost efficiency: prefer free/cheap for low priority
+    tier = agent.get("tier", "unknown")
+    tier_idx = TIER_ORDER.index(tier) if tier in TIER_ORDER else 3
+    if priority <= 1 and tier in ("free", "cheap"):
+        score += 4.0
+    elif priority >= 3 and tier in ("prepaid",):
+        score += 3.0
+    else:
+        score += (3 - tier_idx) * 1.0
+
+    # Activity bonus
+    if agent.get("active", False):
+        score += 2.0
+
+    # Repo familiarity bonus: more repos slightly better
+    repo_count = agent.get("repo_count", 0)
+    score += min(repo_count * 0.2, 2.0)
+
+    return round(score, 3)
+
+
+def find_best_agent(agents: List[dict], role: str, wolf_scores: dict, priority: int,
+                    exclude: Optional[List[str]] = None) -> Optional[dict]:
+    """Find the best agent for the given role and priority."""
+    exclude = exclude or []
+    candidates = []
+    for agent in agents:
+        if agent.get("name") in exclude:
+            continue
+        if not agent.get("active", False):
+            continue
+        s = score_agent_for_issue(agent, role, wolf_scores, priority)
+        candidates.append((s, agent))
+
+    if not candidates:
+        return None
+
+    candidates.sort(key=lambda x: x[0], reverse=True)
+    return candidates[0][1]
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+def dispatch_assignment(api: GiteaAPI, issue: dict, agent: dict, dry_run: bool = False) -> dict:
+    """Assign an issue to an agent and optionally post a comment."""
+    owner = ORG_NAME
+    repo = issue.get("repository", {}).get("name", "")
+
+    # Extract repo from issue repo_url if not in the repository key
+    if not repo:
+        repo_url = issue.get("repository_url", "")
+        if repo_url:
+            repo = repo_url.rstrip("/").split("/")[-1]
+
+    if not repo:
+        return {"success": False, "error": "Cannot determine repository for issue"}
+
+    issue_num = issue.get("number")
+    agent_name = agent.get("name", "unknown")
+
+    comment_body = (
+        f"🤖 **Workforce Manager assigned this issue to: @{agent_name}**\n\n"
+        f"- **Agent:** {agent_name}\n"
+        f"- **Model:** {agent.get('model', 'unknown')}\n"
+        f"- **Role:** {agent.get('role', 'unknown')}\n"
+        f"- **Tier:** {agent.get('tier', 'unknown')}\n"
+        f"- **Assigned at:** {datetime.now(timezone.utc).isoformat()}\n\n"
+        f"*Automated assignment by Workforce Manager (Epic #204)*"
+    )
+
+    if dry_run:
+        return {
+            "success": True,
+            "dry_run": True,
+            "repo": repo,
+            "issue_number": issue_num,
+            "assignee": agent_name,
+            "comment": comment_body,
+        }
+
+    try:
+        api.assign_issue(owner, repo, issue_num, agent_name)
+        api.add_issue_comment(owner, repo, issue_num, comment_body)
+        return {
+            "success": True,
+            "repo": repo,
+            "issue_number": issue_num,
+            "issue_title": issue.get("title", ""),
+            "assignee": agent_name,
+        }
+    except Exception as exc:
+        return {
+            "success": False,
+            "repo": repo,
+            "issue_number": issue_num,
+            "error": str(exc),
+        }
+
+
+# ---------------------------------------------------------------------------
+# State Tracking
+# ---------------------------------------------------------------------------
+
+def update_agent_stats(state: dict, result: dict) -> None:
+    """Update per-agent success tracking."""
+    agent_name = result.get("assignee", "unknown")
+    if "agent_stats" not in state:
+        state["agent_stats"] = {}
+    if agent_name not in state["agent_stats"]:
+        state["agent_stats"][agent_name] = {
+            "total_assigned": 0,
+            "successful": 0,
+            "failed": 0,
+            "success_rate": 0.0,
+            "last_assignment": None,
+            "assigned_issues": [],
+        }
+
+    stats = state["agent_stats"][agent_name]
+    stats["total_assigned"] += 1
+    stats["last_assignment"] = datetime.now(timezone.utc).isoformat()
+    stats["assigned_issues"].append({
+        "repo": result.get("repo", ""),
+        "issue_number": result.get("issue_number"),
+        "success": result.get("success", False),
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    })
+
+    if result.get("success"):
+        stats["successful"] += 1
+    else:
+        stats["failed"] += 1
+
+    total = stats["successful"] + stats["failed"]
+    stats["success_rate"] = round(stats["successful"] / total, 3) if total > 0 else 0.0
+
+
+def print_status(state: dict, agents: List[dict], issues_count: int) -> None:
+    """Print workforce status."""
+    print(f"\n{'=' * 60}")
+    print(f"Workforce Manager Status - {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
+    print(f"{'=' * 60}")
+
+    # Fleet summary
+    active = [a for a in agents if a.get("active")]
+    print(f"\nFleet: {len(active)} active agents, {len(agents)} total")
+    tier_counts = {}
+    for a in active:
+        t = a.get("tier", "unknown")
+        tier_counts[t] = tier_counts.get(t, 0) + 1
+    for t, c in sorted(tier_counts.items()):
+        print(f"  {t}: {c} agents")
+
+    # Agent scores
+    wolf = load_wolf_scores()
+    print(f"\nAgent Details:")
+    print(f"  {'Name':<25} {'Model':<30} {'Role':<18} {'Tier':<10}")
+    for a in agents:
+        if not a.get("active"):
+            continue
+        stats = state.get("agent_stats", {}).get(a["name"], {})
+        rate = stats.get("success_rate", 0.0)
+        total = stats.get("total_assigned", 0)
+        wolf_badge = ""
+        for wm, ws in wolf.items():
+            if a["model"] and a["model"].lower() in wm.lower() and ws.get("success"):
+                wolf_badge = f"[wolf:{ws['total']}]"
+                break
+        name_str = f"{a['name']} {wolf_badge}"
+        if total > 0:
+            name_str += f" (s/r: {rate}, n={total})"
+        print(f"  {name_str:<45} {a.get('role', 'unknown'):<18} {a.get('tier', '?'):<10}")
+
+    print(f"\nOpen Issues: {issues_count}")
+    print(f"Assignments Made: {len(state.get('assignments', []))}")
+    if state.get("last_run"):
+        print(f"Last Run: {state['last_run']}")
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Workforce Manager - Assign Gitea issues to AI agents")
+    parser.add_argument("--dry-run", action="store_true", help="Show assignments without dispatching")
+    parser.add_argument("--status", action="store_true", help="Show workforce status only")
+    parser.add_argument("--cron", action="store_true", help="Run silently for cron scheduling")
+    parser.add_argument("--label", type=str, help="Only process issues with this label")
+    parser.add_argument("--max-issues", type=int, default=100, help="Max issues to process per run")
+    args = parser.parse_args()
+
+    # Setup logging
+    if args.cron:
+        logging.basicConfig(level=logging.WARNING, format="%(asctime)s [%(levelname)s] %(message)s")
+    else:
+        logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+
+    logging.info("Workforce Manager starting")
+
+    # Load data
+    agents = load_fleet_routing()
+    if not agents:
+        logging.error("No agents found in fleet-routing.json")
+        return 1
+    logging.info("Loaded %d agents from fleet routing", len(agents))
+
+    wolf_scores = load_wolf_scores()
+    if wolf_scores:
+        logging.info("Loaded %d model scores from Wolf results", len(wolf_scores))
+
+    state = load_workforce_state()
+
+    # Load Gitea token
+    if GITEA_TOKEN_PATH.exists():
+        token = GITEA_TOKEN_PATH.read_text().strip()
+    else:
+        logging.error("Gitea token not found at %s", GITEA_TOKEN_PATH)
+        return 1
+
+    api = GiteaAPI(token)
+
+    # Status-only mode
+    if args.status:
+        # Quick open issue count
+        repos = api.get_org_repos(ORG_NAME)
+        total = sum(r.get("open_issues_count", 0) for r in repos)
+        print_status(state, agents, total)
+        return 0
+
+    # Fetch open issues
+    if not args.cron:
+        print(f"Scanning open issues across {ORG_NAME} repos...")
+
+    issues = api.get_all_open_issues(ORG_NAME)
+
+    # Filter by label
+    if args.label:
+        issues = [
+            i for i in issues
+            if any(args.label in (l.get("name", "") or "").lower() for l in i.get("labels", []))
+        ]
+
+    if args.label:
+        logging.info("Filtered to %d issues with label '%s'", len(issues), args.label)
+    else:
+        logging.info("Found %d open issues", len(issues))
+
+    # Skip issues already assigned
+    already_assigned_nums = set()
+    for a in state.get("assignments", []):
+        already_assigned_nums.add((a.get("repo"), a.get("issue_number")))
+
+    issues = [
+        i for i in issues
+        if not i.get("assignee") and
+        not (i.get("repository", {}).get("name"), i.get("number")) in already_assigned_nums
+    ]
+    logging.info("%d unassigned issues to process", len(issues))
+
+    # Sort by priority
+    issues_with_priority = [(compute_priority(i), i) for i in issues]
+    issues_with_priority.sort(key=lambda x: x[0], reverse=True)
+    issues = [i for _, i in issues_with_priority[:args.max_issues]]
+
+    # Assign issues
+    assignments = []
+    agent_exclusions: Dict[str, List[str]] = {}  # repo -> list of assigned agents per run
+    global_exclusions: List[str] = []  # agents already at capacity per run
+    max_per_agent_per_run = 5
+
+    for issue in issues:
+        role = classify_issue(issue)
+        priority = compute_priority(issue)
+        repo = issue.get("repository", {}).get("name", "")
+
+        # Avoid assigning same agent twice to same repo in one run
+        repo_excluded = agent_exclusions.get(repo, [])
+
+        # Also exclude agents already at assignment cap
+        cap_excluded = [
+            name for name, stats in state.get("agent_stats", {}).items()
+            if stats.get("total_assigned", 0) > max_per_agent_per_run
+        ]
+
+        excluded = list(set(repo_excluded + global_exclusions + cap_excluded))
+
+        agent = find_best_agent(agents, role, wolf_scores, priority, exclude=excluded)
+        if not agent:
+            # Relax exclusions if no agent found
+            agent = find_best_agent(agents, role, wolf_scores, priority, exclude=[])
+            if not agent:
+            logging.warning("No suitable agent for issue #%d: %s (role=%s)",
+                          issue.get("number"), issue.get("title", ""), role)
+            continue
+
+        result = dispatch_assignment(api, issue, agent, dry_run=args.dry_run)
+        assignments.append(result)
+        update_agent_stats(state, result)
+
+        # Track per-repo exclusions
+        if repo not in agent_exclusions:
+            agent_exclusions[repo] = []
+        agent_exclusions[repo].append(agent["name"])
+
+        if args.dry_run:
+            print(f"  [DRY] #{issue['number']}: {issue.get('title','')[:60]} → @{agent['name']} ({role}, p={priority})")
+        else:
+            status_str = "OK" if result.get("success") else "FAIL"
+            print(f"  [{status_str}] #{issue['number']}: {issue.get('title','')[:60]} → @{agent['name']} ({role}, p={priority})")
+
+    # Save state
+    state["assignments"].extend([{
+        "repo": a.get("repo"),
+        "issue_number": a.get("issue_number"),
+        "assignee": a.get("assignee"),
+        "success": a.get("success", False),
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    } for a in assignments])
+    state["last_run"] = datetime.now(timezone.utc).isoformat()
+    save_workforce_state(state)
+
+    # Summary
+    ok = sum(1 for a in assignments if a.get("success"))
+    fail = len(assignments) - ok
+    logging.info("Done: %d assigned, %d succeeded, %d failed", len(assignments), ok, fail)
+
+    if not args.cron:
+        print(f"\n{'=' * 60}")
+        print(f"Summary: {len(assignments)} assignments, {ok} OK, {fail} failed")
+        # Show agent stats
+        for name, stats in state.get("agent_stats", {}).items():
+            if stats.get("total_assigned", 0) > 0:
+                print(f"  @{name}: {stats['successful']}/{stats['total_assigned']} ({stats.get('success_rate', 0):.0%} success)")
+        print(f"{'=' * 60}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())