feat: add durable loop harness primitive

course/LOOP-HARNESS-MVI.md

@@ -0,0 +1,26 @@
+# Loop Harness MVI
+
+Loop engineering is useful only after it becomes a harness:
+
+1. **One bounded cycle**: claim one task, run one worker, stop.
+2. **Durable state**: task status lives in JSON/DB, not agent context.
+3. **Event log**: every cycle appends JSONL evidence.
+4. **Budget gate**: token estimate blocks before work starts.
+5. **Verifier gate**: tests/review decide whether the next cycle may run.
+6. **Human gate**: pause on deploy, unknown failures, or budget spikes.
+
+This repo's smallest working primitive is `pipeline/harness_loop.py`.
+It intentionally does not run agents itself; callers plug in Claude Code, Pi,
+Codex, OpenClaw, or local scripts after a task is claimed.
+
+```bash
+python -m pipeline.harness_loop \
+  --tasks .runs/demo/tasks.json \
+  --events .runs/demo/events.jsonl \
+  --run-id demo \
+  --max-tokens 20000 \
+  --estimate-tokens 3000
+```
+
+Skipped for now: dashboards, Postgres, queues, schedulers, and multi-provider
+routing. Add those only after the JSONL spine proves the loop is worth running.

pipeline/harness_loop.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Tiny durable loop harness primitives.
+
+Loop engineering is just: pick next task, run one bounded step, write a receipt.
+No dashboard required until this boring JSONL spine is proven useful.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from typing import Any, Iterable, Optional
+
+
+@dataclass
+class LoopTask:
+    id: str
+    title: str
+    status: str = "todo"
+
+
+@dataclass
+class LoopEvent:
+    run_id: str
+    type: str
+    data: dict[str, Any]
+    timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
+
+
+@dataclass
+class CostBudget:
+    max_tokens: int
+    used_tokens: int = 0
+
+    def allow(self, estimate_tokens: int) -> bool:
+        return self.used_tokens + estimate_tokens <= self.max_tokens
+
+    def record(self, tokens: int) -> None:
+        self.used_tokens += tokens
+
+
+def load_tasks(path: str) -> list[LoopTask]:
+    with open(path, "r", encoding="utf-8") as f:
+        raw = json.load(f)
+    return [LoopTask(**item) for item in raw]
+
+
+def save_tasks(path: str, tasks: Iterable[LoopTask]) -> None:
+    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump([asdict(task) for task in tasks], f, indent=2)
+        f.write("\n")
+
+
+def next_task(tasks: Iterable[LoopTask]) -> Optional[LoopTask]:
+    return next((task for task in tasks if task.status == "todo"), None)
+
+
+def append_event(path: str, event: LoopEvent) -> None:
+    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
+    with open(path, "a", encoding="utf-8") as f:
+        f.write(json.dumps(asdict(event), ensure_ascii=False) + "\n")
+
+
+def read_events(path: str) -> list[LoopEvent]:
+    if not os.path.exists(path):
+        return []
+    with open(path, "r", encoding="utf-8") as f:
+        return [LoopEvent(**json.loads(line)) for line in f if line.strip()]
+
+
+def run_one_cycle(tasks_path: str, events_path: str, run_id: str, budget: CostBudget, estimate_tokens: int = 0) -> dict[str, Any]:
+    """Claim one todo task if budget allows; callers do the actual agent work."""
+    if not budget.allow(estimate_tokens):
+        event = LoopEvent(run_id, "blocked", {"reason": "token_budget", "estimate_tokens": estimate_tokens, "budget": asdict(budget)})
+        append_event(events_path, event)
+        return {"status": "blocked", "reason": "token_budget"}
+
+    tasks = load_tasks(tasks_path)
+    task = next_task(tasks)
+    if task is None:
+        event = LoopEvent(run_id, "done", {"reason": "no_todo_tasks"})
+        append_event(events_path, event)
+        return {"status": "done"}
+
+    task.status = "claimed"
+    budget.record(estimate_tokens)
+    save_tasks(tasks_path, tasks)
+    event = LoopEvent(run_id, "claimed", {"task": asdict(task), "estimate_tokens": estimate_tokens, "budget": asdict(budget)})
+    append_event(events_path, event)
+    return {"status": "claimed", "task": asdict(task), "budget": asdict(budget)}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Run one bounded loop-harness cycle")
+    parser.add_argument("--tasks", required=True, help="JSON task list")
+    parser.add_argument("--events", required=True, help="JSONL event log")
+    parser.add_argument("--run-id", required=True)
+    parser.add_argument("--max-tokens", type=int, default=20_000)
+    parser.add_argument("--used-tokens", type=int, default=0)
+    parser.add_argument("--estimate-tokens", type=int, default=0)
+    args = parser.parse_args()
+
+    receipt = run_one_cycle(
+        args.tasks,
+        args.events,
+        args.run_id,
+        CostBudget(args.max_tokens, args.used_tokens),
+        args.estimate_tokens,
+    )
+    print(json.dumps(receipt, indent=2))
+    return 1 if receipt.get("status") == "blocked" else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pipeline/test_all.py

@@ -3,6 +3,9 @@ import json
 import sys
 import os
 
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
 # Add pipeline dir to path
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
@@ -113,4 +116,6 @@ assert any("backend_port" in e[0] for e in result.type_errors), "Should report t
 print("✓ StateGate type validation")
 
 print(f"\n✅ All {8 - errors} tests passed!" if errors == 0 else f"\n❌ {errors} test(s) failed")
+
+if __name__ == "__main__":
     sys.exit(0 if errors == 0 else 1)

pipeline/test_harness_loop.py

@@ -0,0 +1,50 @@
+"""Smoke checks for durable loop-harness primitives."""
+
+import json
+import os
+import sys
+import tempfile
+
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from pipeline.harness_loop import CostBudget, append_event, read_events, run_one_cycle
+
+
+def main() -> int:
+    with tempfile.TemporaryDirectory() as tmp:
+        tasks = os.path.join(tmp, "tasks.json")
+        events = os.path.join(tmp, "events.jsonl")
+        with open(tasks, "w", encoding="utf-8") as f:
+            json.dump([{"id": "1", "title": "first"}, {"id": "2", "title": "second"}], f)
+
+        receipt = run_one_cycle(tasks, events, "run-1", CostBudget(max_tokens=100), estimate_tokens=25)
+        assert receipt["status"] == "claimed"
+        assert receipt["task"]["id"] == "1"
+        assert receipt["budget"]["used_tokens"] == 25
+        assert read_events(events)[0].type == "claimed"
+
+        with open(tasks, "r", encoding="utf-8") as f:
+            saved = json.load(f)
+        assert saved[0]["status"] == "claimed"
+        assert saved[1]["status"] == "todo"
+
+        blocked = run_one_cycle(tasks, events, "run-1", CostBudget(max_tokens=10), estimate_tokens=25)
+        assert blocked == {"status": "blocked", "reason": "token_budget"}
+        assert read_events(events)[-1].type == "blocked"
+
+        saved[1]["status"] = "claimed"
+        with open(tasks, "w", encoding="utf-8") as f:
+            json.dump(saved, f)
+        done = run_one_cycle(tasks, events, "run-1", CostBudget(max_tokens=100), estimate_tokens=0)
+        assert done == {"status": "done"}
+        assert read_events(events)[-1].type == "done"
+
+    print("✓ durable loop harness")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

pipeline/test_integration.py

@@ -12,6 +12,9 @@ import json
 import sys
 import os
 
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8")
+
 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 # Also add the parent of pipeline to ensure direct imports work
 _parent = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -131,4 +134,6 @@ purge_domain("integration-test", "pipeline")
 purge_domain("integration-test", "state-gate")
 
 print(f"\n✅ Integration test passed ({10 - errors}/10)")
+
+if __name__ == "__main__":
     sys.exit(0 if errors == 0 else 1)