diff --git a/loop/__main__.py b/loop/__main__.py index f1856b1..c7b29c6 100644 --- a/loop/__main__.py +++ b/loop/__main__.py @@ -11,13 +11,13 @@ _PROG = "python3 -m loop" -_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay") +_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "run") # Read commands operate on an EXISTING contract dir; scaffold CREATES one, so it # is exempt from the "target must exist" guard. -_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay") +_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "run") -_USAGE = f"usage: {_PROG} " +_USAGE = f"usage: {_PROG} " _HELP = f"""{_PROG} — validate, inspect, and measure a portable repo-OS loop contract. @@ -26,6 +26,7 @@ {_PROG} doctor|validate|verify [--mode basic|strict|release] {_PROG} status [--mode basic|strict|release] {_PROG} replay [--mode basic|strict|release] + {_PROG} run [--mode basic|strict|release] {_PROG} plan-lint [--mode basic|strict|release] commands: @@ -44,6 +45,7 @@ mapping. --mode selects validation strength, same as doctor. status Project the read-only event log and reconcile it with state.json. replay Double-fold the read-only event log and check terminal synchronization. + run Perform one event-sourced execute-task dispatch step. arguments: A workspace root or its .loop/ directory (all commands except plan-lint). @@ -51,7 +53,7 @@ options: --mode {{basic,strict,release}} - (doctor/validate/verify/plan-lint/status/replay) basic forces structural + (doctor/validate/verify/plan-lint/status/replay/run) basic forces structural checks; strict/release require jsonschema. Default: auto-detect. --baseline (metrics only) write docs/metrics-baseline.json over a gate-backed run; exits non-zero and writes nothing otherwise. @@ -166,7 +168,7 @@ def main(argv: list[str] | None = None) -> int: return 2 mode = None - if command in {"doctor", "validate", "verify", "plan-lint", "status", "replay"}: + if command in {"doctor", "validate", "verify", "plan-lint", "status", "replay", "run"}: try: mode, argv = _extract_mode_flag(argv) except ValueError as exc: @@ -229,6 +231,15 @@ def main(argv: list[str] | None = None) -> int: print(f"{command}: {exc}", file=sys.stderr) return 2 + if command == "run": + from .runner import RunnerError, dispatch_once + + try: + return _print_json(dispatch_once(target, mode=mode)) + except (RunnerError, RuntimeStoreError, ValidationModeError) as exc: + print(f"run: {exc}", file=sys.stderr) + return 2 + # command == "inspect": keep the historical inspector script as the scoring # UI over the same contract artifacts; import lazily to avoid making # scripts/ a package. diff --git a/loop/emit.py b/loop/emit.py index 99d8a41..65dfb57 100644 --- a/loop/emit.py +++ b/loop/emit.py @@ -183,8 +183,10 @@ def append_iteration( if not runlog.exists(): runlog = paths.workspace / "RUNLOG.md" runlog.write_text(f"# RUNLOG.md — {paths.workspace.name}\n", encoding="utf-8") - with runlog.open("a", encoding="utf-8") as fh: - fh.write("\n".join(lines)) + header = f"## Iteration {iteration_id} —" + if header not in runlog.read_text(encoding="utf-8"): + with runlog.open("a", encoding="utf-8") as fh: + fh.write("\n".join(lines)) current["iteration_id"] = iteration_id if task_id: diff --git a/loop/runner.py b/loop/runner.py new file mode 100644 index 0000000..78a8ea3 --- /dev/null +++ b/loop/runner.py @@ -0,0 +1,210 @@ +"""One event-sourced, crash-resumable execute-task dispatch step.""" + +from __future__ import annotations + +import json +import sqlite3 +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable + +from . import emit +from .events import EVENT_SCHEMA_ID, SQLiteEventStore, validate_event +from .paths import resolve_loop_paths +from .reducer import reduce_events +from .runtime import RuntimeStoreError + + +class RunnerError(RuntimeError): + """A dispatch request could not be attempted.""" + + +class NotReadyError(RunnerError): + """The event projection has not reached execute-task.""" + + +class VerifierNotImplementedError(RunnerError): + """S3a deliberately has no process-isolated verifier yet.""" + + +@dataclass(frozen=True) +class VerifyOutcome: + passed: bool + summary: str = "" + + +Verifier = Callable[[dict[str, Any], Path], VerifyOutcome] + + +def done_task_ids(tasks: list[dict], projection: dict) -> set[str]: + """Return declaratively done tasks plus durable successful dispatches.""" + done = {task["id"] for task in tasks if task.get("status") == "done"} + done.update( + entry["task_id"] + for entry in projection.get("runlog_entries", []) + if entry.get("outcome") == "task_passed" and isinstance(entry.get("task_id"), str) + ) + return done + + +def select_next_task(tasks: list[dict], projection: dict) -> dict | None: + """Select the first pending task whose declared dependencies are done.""" + done = done_task_ids(tasks, projection) + for task in tasks: + if task.get("status") != "pending" or task.get("id") in done: + continue + if all(dependency in done for dependency in task.get("depends_on", [])): + return task + return None + + +def _default_verifier(task: dict[str, Any], workspace: Path) -> VerifyOutcome: + raise VerifierNotImplementedError( + "verification is not implemented yet; supply a verifier through dispatch_once()" + ) + + +def _load_tasks(paths: Any) -> list[dict]: + try: + raw = json.loads(paths.tasks.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise RunnerError(f"cannot read TASKS.json: {exc}") from exc + tasks = raw.get("tasks") if isinstance(raw, dict) else None + if not isinstance(tasks, list) or not all(isinstance(task, dict) for task in tasks): + raise RunnerError("TASKS.json must contain a tasks array of objects") + return tasks + + +def _projection(target: str | Path, mode: str | None) -> tuple[str, dict[str, Any]]: + """Read with SQLite's immutable URI so failed attempts create no WAL files.""" + path = resolve_loop_paths(target).loop_dir / "events.db" + if not path.exists(): + raise RuntimeStoreError("missing_store", f"event store does not exist: {path}") + # A clean, closed WAL store can be read immutable without creating SQLite + # sidecars. A post-COMMIT crash deliberately leaves a WAL sidecar, which + # must be read in ordinary read-only mode so its durable frames are replayed. + query = "mode=ro" if path.with_name(path.name + "-wal").exists() else "mode=ro&immutable=1" + try: + conn = sqlite3.connect(f"{path.absolute().as_uri()}?{query}", uri=True) + try: + run_ids = conn.execute("SELECT DISTINCT run_id FROM events ORDER BY run_id ASC").fetchall() + if not run_ids: + raise RuntimeStoreError("empty_store", f"event store is empty: {path}") + if len(run_ids) != 1: + raise RuntimeStoreError("ambiguous_run_id", f"event store has ambiguous run_id values: {path}") + run_id = run_ids[0][0] + rows = conn.execute("SELECT run_id, sequence, event_id, type, actor, causation_id, correlation_id, ts, payload, artifact_hashes FROM events WHERE run_id = ? ORDER BY sequence ASC", (run_id,)).fetchall() + finally: + conn.close() + except sqlite3.DatabaseError as exc: + raise RuntimeStoreError("corrupt_store", f"cannot read event store: {exc}") from exc + try: + events = [{"schema": EVENT_SCHEMA_ID, "run_id": row[0], "sequence": row[1], "event_id": row[2], "type": row[3], "actor": row[4], "causation_id": row[5], "correlation_id": row[6], "ts": row[7], "payload": json.loads(row[8]), "artifact_hashes": json.loads(row[9])} for row in rows] + except (TypeError, json.JSONDecodeError) as exc: + raise RuntimeStoreError("corrupt_store", f"cannot read event store: {exc}") from exc + for event in events: + report = validate_event(event, mode=mode) + if not report["ok"]: + raise RuntimeStoreError("invalid_event", f"event store contains invalid event: {report['issues']}") + try: + return run_id, reduce_events(events) + except ValueError as exc: + raise RuntimeStoreError("invalid_event_stream", str(exc)) from exc + + +def _reconcile_legacy_iteration(target: str | Path, projection: dict[str, Any]) -> None: + """Materialize every event-log iteration not yet reflected in state.json.""" + paths = resolve_loop_paths(target) + try: + state = json.loads(paths.state.read_text(encoding="utf-8")) + current_id = state.get("iteration_id") if isinstance(state, dict) else None + except (OSError, json.JSONDecodeError) as exc: + raise RunnerError(f"cannot read state.json: {exc}") from exc + if not isinstance(current_id, int): + raise RunnerError("state.json iteration_id must be an integer") + for entry in projection["runlog_entries"]: + iteration_id = entry.get("iteration_id") + if isinstance(iteration_id, int) and iteration_id > current_id: + emit.append_iteration( + target, + iteration_id=iteration_id, + outcome=entry["outcome"], + state=entry.get("state"), + task_id=entry.get("task_id", ""), + notes=entry.get("summary", ""), + ) + current_id = iteration_id + + +def _reconcile_legacy_terminal(target: str | Path, projection: dict[str, Any]) -> None: + """Replay the terminal's already-recorded payload into existing emit APIs.""" + terminal = projection.get("terminal") + if terminal is None: + return + paths = resolve_loop_paths(target) + if not paths.terminal.exists(): + emit.terminate( + target, + state=terminal["state"], + criteria_met=terminal["criteria_met"], + evidence=terminal["evidence"], + false_completion=terminal["false_completion"], + iteration_id=terminal.get("iteration_id"), + completion_policy=terminal.get("completion_policy"), + ) + try: + state = json.loads(paths.state.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + raise RunnerError(f"cannot read state.json: {exc}") from exc + if not isinstance(state, dict) or state.get("state") != "terminal" or state.get("terminal_state") != terminal["state"]: + emit.sync_state_to_terminal(target) + + +def dispatch_once( + target: str | Path, *, verifier: Verifier | None = None, mode: str | None = None, +) -> dict[str, Any]: + """Run at most one durable selection/verification/recording dispatch.""" + run_id, projection = _projection(target, mode) + if projection.get("terminal") is not None: + _reconcile_legacy_terminal(target, projection) + return {"ok": True, "action": "noop_terminal", "run_id": run_id} + if projection.get("state") != "execute-task": + raise NotReadyError(f"dispatch requires state 'execute-task', got {projection.get('state')!r}") + + _reconcile_legacy_iteration(target, projection) + paths = resolve_loop_paths(target) + tasks = _load_tasks(paths) + task = select_next_task(tasks, projection) + if task is None: + done = done_task_ids(tasks, projection) + if all(task.get("id") in done for task in tasks): + iteration_id = projection["iteration_id"] + payload = { + "state": "Succeeded", "criteria_met": {task["id"]: True for task in tasks}, + "evidence": ["RUNLOG.md"], "false_completion": False, + "completion_policy": {"mode": "all_required"}, "iteration_id": iteration_id, + } + store = SQLiteEventStore(paths.loop_dir / "events.db") + store.append(run_id, "terminal_written", payload, actor="loop.run", + expected_sequence=projection["last_sequence"] + 1) + _reconcile_legacy_terminal(target, {**projection, "terminal": payload}) + return {"ok": True, "action": "terminal_written", "iteration_id": iteration_id, "run_id": run_id} + return {"ok": False, "action": "blocked", "run_id": run_id} + + outcome = (verifier or _default_verifier)(task, paths.workspace) + if not isinstance(outcome, VerifyOutcome): + raise RunnerError("verifier must return VerifyOutcome") + iteration_id = projection["iteration_id"] + 1 + payload = { + "iteration_id": iteration_id, + "outcome": "task_passed" if outcome.passed else "task_failed", + "task_id": task["id"], + "summary": outcome.summary, + } + store = SQLiteEventStore(paths.loop_dir / "events.db") + store.append(run_id, "iteration_appended", payload, actor="loop.run", + expected_sequence=projection["last_sequence"] + 1) + emit.append_iteration(target, iteration_id=iteration_id, outcome=payload["outcome"], + task_id=payload["task_id"], notes=payload["summary"]) + return {"ok": True, "action": "dispatched", "task_id": task["id"], + "outcome": payload["outcome"], "iteration_id": iteration_id, "run_id": run_id} diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index 1c3763e..a1e14f2 100644 --- a/reference/repo-os-contract.md +++ b/reference/repo-os-contract.md @@ -596,6 +596,16 @@ same input sequence always produces a byte-identical result. on-disk location is `.loop/events.db`, with one run discovered per store by the runtime readers (multi-run support remains deferred). +**Dispatch crash boundary:** `loop run` verifies first, then commits its +`iteration_appended` (or `terminal_written`) event with a compare-and-swap +sequence. That committed event is the source of truth; only afterwards are +`RUNLOG.md` and `.loop/state.json` materialized from the exact recorded +payload. A later `loop run` replays missing legacy materialization before +selecting work, so a crash after the event commit never duplicates a dispatch. +TASKS.json is read-only declarative input for dispatch: event-log +`task_passed` facts supply dynamic completion and dispatch does not rewrite +task status or evidence. + **Event types:** `contract_opened | iteration_appended | receipt_appended | terminal_written` — one-to-one with `loop.emit`'s four writer operations (`open_contract`/`append_iteration`/`append_receipt`/`terminate`), so a diff --git a/scripts/test_runner_dispatch.py b/scripts/test_runner_dispatch.py new file mode 100644 index 0000000..2413470 --- /dev/null +++ b/scripts/test_runner_dispatch.py @@ -0,0 +1,129 @@ +"""Regression coverage for the bounded event-sourced runner.""" +from __future__ import annotations + +import hashlib +import json +import signal +import subprocess +import sys +from pathlib import Path + +import pytest + +from loop import emit +from loop.events import SQLiteEventStore, SequenceConflictError +from loop.runner import NotReadyError, VerifyOutcome, VerifierNotImplementedError, dispatch_once, select_next_task +from loop.runtime import replay_report + +ROOT = Path(__file__).resolve().parent.parent + + +def _task(i, deps=(), status="pending"): + return {"id": i, "title": i, "status": status, "criterion_ref": i, "verify": "true", "depends_on": list(deps), "attempts": 0, "evidence": None} + + +def _ws(tmp_path, tasks=None, ready=True): + w = tmp_path / "workspace"; emit.open_contract(w) + (w / "TASKS.json").write_text(json.dumps({"schema": "loop-engineer/tasks@1", "tasks": tasks or [_task("T-1")]}), encoding="utf-8") + s = SQLiteEventStore(w / ".loop" / "events.db") + s.append("run-1", "contract_opened", {"workspace": "workspace"}, actor="test") + if ready: + for n, state in enumerate(("plan", "critique-plan", "queue-tasks", "execute-task"), 1): + s.append("run-1", "iteration_appended", {"iteration_id": n, "outcome": "replanned", "state": state}, actor="test") + emit.append_iteration(w, iteration_id=n, outcome="replanned", state=state) + return w, s + + +def _pass(task, workspace): return VerifyOutcome(True, "verified") +def _hashes(w): return {str(p.relative_to(w)): hashlib.sha256(p.read_bytes()).hexdigest() for p in w.rglob("*") if p.is_file() and not p.name.endswith((".db-wal", ".db-shm"))} +def _cli(*args): return subprocess.run([sys.executable, "-m", "loop", *args], cwd=ROOT, text=True, capture_output=True, timeout=15) + + +def test_select_next_task_respects_depends_on_and_declaration_order(): + assert select_next_task([_task("T-2", ("T-1",)), _task("T-1")], {"runlog_entries": []})["id"] == "T-1" + +def test_select_next_task_returns_none_when_all_tasks_done(): + assert select_next_task([_task("T-1", status="done")], {"runlog_entries": []}) is None + +def test_select_next_task_skips_task_already_recorded_task_passed_in_event_log_even_if_tasks_json_says_pending(): + assert select_next_task([_task("T-1"), _task("T-2")], {"runlog_entries": [{"task_id": "T-1", "outcome": "task_passed"}]})["id"] == "T-2" + +def test_select_next_task_reports_blocked_when_a_pending_task_has_a_never_satisfiable_dependency(): + assert select_next_task([_task("T-1", ("missing",))], {"runlog_entries": []}) is None + +def test_dispatch_once_raises_not_ready_when_projection_state_is_not_execute_task(tmp_path): + w, _ = _ws(tmp_path, ready=False); before = _hashes(w) + with pytest.raises(NotReadyError): dispatch_once(w, verifier=_pass) + assert _hashes(w) == before + +def test_dispatch_once_default_verifier_raises_and_persists_no_event_or_legacy_write(tmp_path): + w, _ = _ws(tmp_path); before = _hashes(w) + with pytest.raises(VerifierNotImplementedError): dispatch_once(w) + assert _hashes(w) == before + +def test_dispatch_once_appends_exactly_one_iteration_event_and_materializes_runlog_and_state(tmp_path): + w, s = _ws(tmp_path); assert dispatch_once(w, verifier=_pass)["outcome"] == "task_passed" + assert len(s.read("run-1")) == 6 and "## Iteration 5 —" in (w / "RUNLOG.md").read_text() and json.loads((w / ".loop" / "state.json").read_text())["iteration_id"] == 5 + +def test_dispatch_once_writes_terminal_and_syncs_legacy_artifacts_when_all_tasks_done(tmp_path): + w, _ = _ws(tmp_path, [_task("T-1", status="done")]); assert dispatch_once(w, verifier=_pass)["action"] == "terminal_written" + assert json.loads((w / ".loop" / "state.json").read_text())["terminal_state"] == "Succeeded" + +def test_dispatch_once_second_call_after_terminal_is_a_clean_noop(tmp_path): + w, _ = _ws(tmp_path, [_task("T-1", status="done")]); dispatch_once(w, verifier=_pass) + assert dispatch_once(w, verifier=_pass)["action"] == "noop_terminal" + +def test_append_iteration_retry_with_same_iteration_id_does_not_duplicate_runlog_block(tmp_path): + w, _ = _ws(tmp_path); emit.append_iteration(w, iteration_id=5, outcome="task_passed", task_id="T-1"); emit.append_iteration(w, iteration_id=5, outcome="task_passed", task_id="T-1") + assert (w / "RUNLOG.md").read_text().count("## Iteration 5 —") == 1 + +def test_dispatch_once_sequential_calls_advance_through_all_tasks_without_repeats(tmp_path): + w, _ = _ws(tmp_path, [_task("T-1"), _task("T-2", ("T-1",))]); a = dispatch_once(w, verifier=_pass); b = dispatch_once(w, verifier=_pass); c = dispatch_once(w, verifier=_pass) + assert (a["task_id"], b["task_id"], c["action"]) == ("T-1", "T-2", "terminal_written") + +def test_dispatch_once_surfaces_sequence_conflict_from_a_concurrent_external_append_as_typed_error(tmp_path): + w, s = _ws(tmp_path) + def conflict(task, root): + s.append("run-1", "receipt_appended", {"iteration_id": 4, "role": "write", "model": "test", "outcome": "ok"}, actor="other") + return VerifyOutcome(True) + with pytest.raises(SequenceConflictError): dispatch_once(w, verifier=conflict) + +def test_replay_report_after_normal_multi_task_run_is_ok_and_deterministic(tmp_path): + w, _ = _ws(tmp_path, [_task("T-1"), _task("T-2", ("T-1",))]); dispatch_once(w, verifier=_pass); dispatch_once(w, verifier=_pass); dispatch_once(w, verifier=_pass) + r = replay_report(w); assert r["ok"] and r["deterministic"] and r["terminal_desync"] is None + + +def _crash(tmp_path, workspace, after): + p = tmp_path / ("after.py" if after else "before.py") + body = "result = super().execute(sql, *args, **kwargs)\n if isinstance(sql, str) and sql.strip().upper() == 'COMMIT': os.kill(os.getpid(), signal.SIGKILL)\n return result" if after else "if isinstance(sql, str) and sql.strip().upper() == 'COMMIT': os.kill(os.getpid(), signal.SIGKILL)\n return super().execute(sql, *args, **kwargs)" + p.write_text("import os,sys,signal,sqlite3\nsys.path.insert(0, os.getcwd())\nreal=sqlite3.connect\nclass Barrier(sqlite3.Connection):\n def execute(self,sql,*args,**kwargs):\n " + body + "\nsqlite3.connect=lambda *a,**kw: real(*a,factory=Barrier,**kw)\nfrom loop.runner import dispatch_once,VerifyOutcome\ndispatch_once(sys.argv[1],verifier=lambda t,w: VerifyOutcome(True))\n", encoding="utf-8") + return subprocess.run([sys.executable, "-B", str(p), str(workspace)], cwd=ROOT, timeout=15) + +def test_crash_injection_before_iteration_event_commit_leaves_no_partial_dispatch(tmp_path): + w, s = _ws(tmp_path); before = _hashes(w); p = _crash(tmp_path, w, False) + assert p.returncode == -signal.SIGKILL and _hashes(w) == before and len(s.read("run-1")) == 5 + assert dispatch_once(w, verifier=_pass)["action"] == "dispatched" + +def test_crash_injection_after_iteration_event_commit_before_legacy_sync_resumes_without_duplicate_event(tmp_path): + w, s = _ws(tmp_path); p = _crash(tmp_path, w, True) + assert p.returncode == -signal.SIGKILL and len(s.read("run-1")) == 6 and json.loads((w / ".loop" / "state.json").read_text())["iteration_id"] == 4 + assert dispatch_once(w, verifier=_pass)["action"] == "terminal_written" and len(s.read("run-1")) == 7 and (w / "RUNLOG.md").read_text().count("## Iteration 5 —") == 1 + +def test_crash_injection_after_terminal_event_commit_before_legacy_write_resumes_via_sync(tmp_path): + w, s = _ws(tmp_path, [_task("T-1", status="done")]); p = _crash(tmp_path, w, True); terminal = w / ".loop" / "terminal_state.json" + assert p.returncode == -signal.SIGKILL and s.read("run-1")[-1]["type"] == "terminal_written" and not terminal.exists() + assert dispatch_once(w, verifier=_pass)["action"] == "noop_terminal"; stamp = terminal.stat().st_mtime_ns + assert dispatch_once(w, verifier=_pass)["action"] == "noop_terminal" and terminal.stat().st_mtime_ns == stamp and replay_report(w)["terminal_desync"] is None + +def test_run_command_listed_in_help_and_usage(): + r = _cli("--help"); assert r.returncode == 0 and "run" in r.stdout and "python3 -m loop run [--mode basic|strict|release] " in r.stdout + +def test_run_missing_target_argument_prints_usage_and_exits_nonzero(): + r = _cli("run"); assert r.returncode != 0 and "usage:" in r.stderr + +def test_run_nonexistent_target_gives_actionable_error_exit_2(tmp_path): + r = _cli("run", str(tmp_path / "missing")); assert r.returncode == 2 and "does not exist" in r.stderr + +def test_run_cli_default_verifier_not_implemented_exits_2_no_traceback(tmp_path): + w, _ = _ws(tmp_path); r = _cli("run", str(w)) + assert r.returncode == 2 and "verification is not implemented" in r.stderr and r.stdout == "" and "Traceback" not in r.stderr