feat(cli): event-sourced loop run single-step dispatch with crash-safe resume (Refs #55)

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} <scaffold|doctor|validate|verify|inspect|metrics|plan-lint|status|replay> <target>"
+_USAGE = f"usage: {_PROG} <scaffold|doctor|validate|verify|inspect|metrics|plan-lint|status|replay|run> <target>"
 
 _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] <workspace-or-.loop>
        {_PROG} status [--mode basic|strict|release] <workspace>
        {_PROG} replay [--mode basic|strict|release] <workspace>
+       {_PROG} run [--mode basic|strict|release] <workspace>
        {_PROG} plan-lint [--mode basic|strict|release] <plan-file>
 
 commands:
@@ -44,14 +45,15 @@
              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:
   <target>     A workspace root or its .loop/ directory (all commands except plan-lint).
   <plan-file>  A single loop-engineer/plan@1 JSON file (plan-lint only).
 
 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.

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:

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}

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