From 2675594422706f533742301a3e5d0f551e82fd87 Mon Sep 17 00:00:00 2001 From: artale Date: Thu, 18 Jun 2026 03:30:27 +0200 Subject: [PATCH] feat: add durable loop harness primitive --- course/LOOP-HARNESS-MVI.md | 26 ++++++++ pipeline/harness_loop.py | 119 ++++++++++++++++++++++++++++++++++ pipeline/test_all.py | 7 +- pipeline/test_harness_loop.py | 50 ++++++++++++++ pipeline/test_integration.py | 7 +- 5 files changed, 207 insertions(+), 2 deletions(-) create mode 100644 course/LOOP-HARNESS-MVI.md create mode 100644 pipeline/harness_loop.py create mode 100644 pipeline/test_harness_loop.py diff --git a/course/LOOP-HARNESS-MVI.md b/course/LOOP-HARNESS-MVI.md new file mode 100644 index 0000000..34d17d4 --- /dev/null +++ b/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. diff --git a/pipeline/harness_loop.py b/pipeline/harness_loop.py new file mode 100644 index 0000000..8119d75 --- /dev/null +++ b/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()) diff --git a/pipeline/test_all.py b/pipeline/test_all.py index cc580b6..59982a9 100644 --- a/pipeline/test_all.py +++ b/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") -sys.exit(0 if errors == 0 else 1) + +if __name__ == "__main__": + sys.exit(0 if errors == 0 else 1) diff --git a/pipeline/test_harness_loop.py b/pipeline/test_harness_loop.py new file mode 100644 index 0000000..8f9440d --- /dev/null +++ b/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()) diff --git a/pipeline/test_integration.py b/pipeline/test_integration.py index a50e818..e242381 100644 --- a/pipeline/test_integration.py +++ b/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)") -sys.exit(0 if errors == 0 else 1) + +if __name__ == "__main__": + sys.exit(0 if errors == 0 else 1)