From a7b3289bf7aa9110a82e4cebd1070303a312aaae Mon Sep 17 00:00:00 2001 From: Sollan Systems Date: Wed, 15 Jul 2026 08:20:29 -0400 Subject: [PATCH] feat(cli): read-only `loop status` and `loop replay` runtime verbs over the EventStore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New loop/runtime.py: status_report projects the event log once via the deterministic reducer and reconciles it with .loop/state.json (typed divergence findings); replay_report double-folds for determinism, reports illegal event sequences as typed findings (never a traceback), and checks the desynced-terminal window bidirectionally (file-without-event, event-without-file, value mismatch — the R-001 class). The store opens strictly read-only (file:...?mode=ro URI); loop/events.py and loop/reducer.py are byte-unchanged. --mode joins the existing _extract_mode_flag path; reducer domain checks run unconditionally. Exit codes mirror doctor: 0 healthy, 1 ran-and-found-a-problem, 2 cannot-attempt-a-report (missing/empty/corrupt store, ambiguous run_id via new RuntimeStoreError). This slice decides the previously-open §16 questions: store location .loop/events.db, one run per store (multi-run deferred). 27 new unconditional tests in scripts/test_loop_cli_status_replay.py. Refs #55 (phase 1 partial — status/replay; the run verb remains). --- loop/__main__.py | 23 ++- loop/runtime.py | 194 ++++++++++++++++++++ reference/repo-os-contract.md | 7 +- scripts/test_loop_cli_status_replay.py | 239 +++++++++++++++++++++++++ 4 files changed, 454 insertions(+), 9 deletions(-) create mode 100644 loop/runtime.py create mode 100644 scripts/test_loop_cli_status_replay.py diff --git a/loop/__main__.py b/loop/__main__.py index a8f7fad..f1856b1 100644 --- a/loop/__main__.py +++ b/loop/__main__.py @@ -7,22 +7,25 @@ from .contract import VALIDATION_MODES, ValidationModeError, doctor_report from .plan import validate_plan +from .runtime import RuntimeStoreError, replay_report, status_report _PROG = "python3 -m loop" -_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint") +_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay") # 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") +_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay") -_USAGE = f"usage: {_PROG} " +_USAGE = f"usage: {_PROG} " _HELP = f"""{_PROG} — validate, inspect, and measure a portable repo-OS loop contract. {_USAGE} {_PROG} metrics [--baseline] {_PROG} doctor|validate|verify [--mode basic|strict|release] + {_PROG} status [--mode basic|strict|release] + {_PROG} replay [--mode basic|strict|release] {_PROG} plan-lint [--mode basic|strict|release] commands: @@ -39,6 +42,8 @@ plan-lint Validate a loop-engineer/plan@1 Loop Plan IR document: task-kind fields, dependency-graph acyclicity, and the terminal-state 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. arguments: A workspace root or its .loop/ directory (all commands except plan-lint). @@ -46,7 +51,7 @@ options: --mode {{basic,strict,release}} - (doctor/validate/verify/plan-lint only) basic forces structural + (doctor/validate/verify/plan-lint/status/replay) 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. @@ -161,7 +166,7 @@ def main(argv: list[str] | None = None) -> int: return 2 mode = None - if command in {"doctor", "validate", "verify", "plan-lint"}: + if command in {"doctor", "validate", "verify", "plan-lint", "status", "replay"}: try: mode, argv = _extract_mode_flag(argv) except ValueError as exc: @@ -216,6 +221,14 @@ def main(argv: list[str] | None = None) -> int: print(f"{command}: {exc}", file=sys.stderr) return 2 + if command in {"status", "replay"}: + try: + report = status_report(target, mode=mode) if command == "status" else replay_report(target, mode=mode) + return _print_json(report) + except (ValidationModeError, RuntimeStoreError) as exc: + print(f"{command}: {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/runtime.py b/loop/runtime.py new file mode 100644 index 0000000..ba7674d --- /dev/null +++ b/loop/runtime.py @@ -0,0 +1,194 @@ +"""Read-only runtime reports over the append-only event store.""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any + +from .completion import CompletionPolicyError, criteria_satisfy_completion +from .contract import ContractIssue +from .events import EVENT_SCHEMA_ID, EVENT_TYPES, validate_event +from .paths import resolve_loop_paths +from .reducer import EventReplayError, reduce_events + + +class RuntimeStoreError(RuntimeError): + """The runtime store cannot be read well enough to construct a report.""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(f"{code}: {message}") + self.code = code + + +def _store_path(target: str | Path) -> Path: + return resolve_loop_paths(target).loop_dir / "events.db" + + +def _read_events_readonly(path: Path, run_id: str) -> list[dict[str, Any]]: + """Read the EventStore row shape without invoking its write-capable connector.""" + try: + conn = sqlite3.connect(f"{path.absolute().as_uri()}?mode=ro", uri=True) + try: + 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: + return [ + { + "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 + + +def _discover_run_id(path: Path) -> str: + if not path.exists(): + raise RuntimeStoreError("missing_store", f"event store does not exist: {path}") + try: + conn = sqlite3.connect(f"{path.absolute().as_uri()}?mode=ro", uri=True) + try: + rows = conn.execute("SELECT DISTINCT run_id FROM events ORDER BY run_id ASC").fetchall() + finally: + conn.close() + except sqlite3.DatabaseError as exc: + raise RuntimeStoreError("corrupt_store", f"cannot read event store: {exc}") from exc + if not rows: + raise RuntimeStoreError("empty_store", f"event store is empty: {path}") + if len(rows) != 1: + raise RuntimeStoreError("ambiguous_run_id", f"event store has ambiguous run_id values: {path}") + run_id = rows[0][0] + if not isinstance(run_id, str): + raise RuntimeStoreError("corrupt_store", f"event store has invalid run_id: {path}") + return run_id + + +def _events(target: str | Path, mode: str | None) -> tuple[Path, str, list[dict[str, Any]], dict[str, Any]]: + path = _store_path(target) + run_id = _discover_run_id(path) + events = _read_events_readonly(path, run_id) + validation: dict[str, Any] | None = None + for event in events: + validation = validate_event(event, mode=mode) + assert validation is not None + return path, run_id, events, validation + + +def _state_divergence(paths: Any, projection: dict[str, Any]) -> list[dict[str, Any]]: + try: + state = json.loads(paths.state.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + state = None + if not isinstance(state, dict): + return [ContractIssue("state_field_mismatch", "state.json is missing or is not an object")] + expected = { + "state": projection["state"], + "iteration_id": projection["iteration_id"], + "active_task": projection["active_task"], + "terminal_state": projection["terminal"].get("state") if projection["terminal"] else None, + } + issues: list[dict[str, Any]] = [] + for field, value in expected.items(): + if state.get(field) != value: + issues.append(ContractIssue("state_field_mismatch", f"state.json {field!r} differs from event projection")) + return issues + + +def _completion_satisfied(terminal: dict[str, Any] | None) -> bool | None: + if terminal is None: + return None + if terminal.get("state") != "Succeeded": + return False + try: + return criteria_satisfy_completion(terminal.get("criteria_met", {}), terminal.get("completion_policy")) + except CompletionPolicyError: + return False + + +def status_report(target: str | Path, *, mode: str | None = None) -> dict[str, Any]: + """Project a single event stream and reconcile it with live state.json.""" + _, run_id, events, validation = _events(target, mode) + paths = resolve_loop_paths(target) + try: + projection = reduce_events(events) + divergence = _state_divergence(paths, projection) + except EventReplayError as exc: + projection = {"state": None, "iteration_id": None, "active_task": None, "terminal": None} + divergence = [ContractIssue("illegal_event_sequence", str(exc))] + return { + "ok": not divergence, + "validation_mode": validation["validation_mode"], "requested_mode": validation["requested_mode"], + "schemas_checked": [EVENT_SCHEMA_ID], "run_id": run_id, "event_count": len(events), + "state": projection["state"], "iteration_id": projection["iteration_id"], + "active_task": projection["active_task"], "terminal": projection["terminal"], + "completion_satisfied": _completion_satisfied(projection["terminal"]), + "state_json_agrees": not divergence, "divergence": divergence, + } + + +def _terminal_desync(paths: Any, projection: dict[str, Any]) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: + event_terminal = projection["terminal"] + try: + disk_terminal = json.loads(paths.terminal.read_text(encoding="utf-8")) if paths.terminal.exists() else None + except (OSError, json.JSONDecodeError): + disk_terminal = None + if event_terminal is None and disk_terminal is None: + return None, [] + if event_terminal is None or disk_terminal is None: + return {"event": event_terminal, "file": disk_terminal}, [ + ContractIssue("desynced_terminal_window", "terminal event and terminal_state.json disagree on presence") + ] + if not isinstance(disk_terminal, dict) or disk_terminal.get("state") != event_terminal.get("state"): + return {"event": event_terminal, "file": disk_terminal}, [ + ContractIssue("desynced_terminal_window", "terminal_state.json differs from event projection") + ] + for field in ("criteria_met", "evidence", "false_completion", "completion_policy"): + if field in disk_terminal and disk_terminal.get(field) != event_terminal.get(field): + return {"event": event_terminal, "file": disk_terminal}, [ + ContractIssue("terminal_state_mismatch", f"terminal_state.json {field!r} differs from event projection") + ] + return None, [] + + +def replay_report(target: str | Path, *, mode: str | None = None) -> dict[str, Any]: + """Double-fold an event stream and check terminal-window synchronization.""" + _, run_id, events, validation = _events(target, mode) + paths = resolve_loop_paths(target) + findings: list[dict[str, Any]] = [] + deterministic = True + legal_sequence = True + projection: dict[str, Any] | None = None + try: + first = reduce_events(events) + second = reduce_events(events) + deterministic = first == second + projection = first + if not deterministic: + findings.append(ContractIssue("nondeterministic_replay", "two event folds produced different projections")) + except EventReplayError as exc: + legal_sequence = False + findings.append(ContractIssue("illegal_event_sequence", str(exc))) + terminal_desync = None + if projection is not None: + terminal_desync, terminal_findings = _terminal_desync(paths, projection) + findings.extend(terminal_findings) + return { + "ok": not findings, + "validation_mode": validation["validation_mode"], "requested_mode": validation["requested_mode"], + "schemas_checked": [EVENT_SCHEMA_ID], "run_id": run_id, "event_count": len(events), + "deterministic": deterministic, "legal_sequence": legal_sequence, + "terminal_desync": terminal_desync, "findings": findings, + } diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index 7eebd2c..1c3763e 100644 --- a/reference/repo-os-contract.md +++ b/reference/repo-os-contract.md @@ -592,10 +592,9 @@ ordered event stream into a deterministic state/runlog/receipts view — the same input sequence always produces a byte-identical result. **Scope boundary:** unlike manifest/state/tasks/terminal (§11), `event@1` is -**not yet** an artifact `loop doctor` reads from a scaffolded workspace, and -no workspace-relative on-disk location (e.g. `.loop/events.db`) has been -decided. The execution-runtime milestone that wires a live event log into -`scaffold`/`emit`/`doctor` will make that call. +**not yet** an artifact `loop doctor` reads from a scaffolded workspace; its +on-disk location is `.loop/events.db`, with one run discovered per store by +the runtime readers (multi-run support remains deferred). **Event types:** `contract_opened | iteration_appended | receipt_appended | terminal_written` — one-to-one with `loop.emit`'s four writer operations diff --git a/scripts/test_loop_cli_status_replay.py b/scripts/test_loop_cli_status_replay.py new file mode 100644 index 0000000..8e011aa --- /dev/null +++ b/scripts/test_loop_cli_status_replay.py @@ -0,0 +1,239 @@ +"""Runtime CLI contract tests for read-only event-log status and replay.""" + +import hashlib +import importlib.util +import json +import os +import sqlite3 +import subprocess +import sys +from pathlib import Path + +import pytest + +from loop.events import SQLiteEventStore +from loop.runtime import RuntimeStoreError, replay_report, status_report + + +ROOT = Path(__file__).resolve().parent.parent + + +def _run(*args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run([sys.executable, "-m", "loop", *args], cwd=ROOT, text=True, capture_output=True) + + +def _workspace(tmp_path: Path, *, state: bool = True) -> tuple[Path, SQLiteEventStore]: + workspace = tmp_path / "workspace" + loop = workspace / ".loop" + loop.mkdir(parents=True) + if state: + (loop / "state.json").write_text(json.dumps({"state": "intake", "iteration_id": 0, "active_task": None, "terminal_state": None}), encoding="utf-8") + return workspace, SQLiteEventStore(loop / "events.db") + + +def _open(store: SQLiteEventStore, run_id: str = "run-1") -> dict: + return store.append(run_id, "contract_opened", {"workspace": "workspace"}, actor="test") + + +def _terminal(store: SQLiteEventStore, run_id: str = "run-1") -> dict: + opened = _open(store, run_id) + return store.append(run_id, "terminal_written", {"state": "Succeeded", "criteria_met": {"gate": True}, "evidence": ["proof"], "false_completion": False}, actor="test", causation_id=opened["event_id"]) + + +def _write_state(workspace: Path, **values: object) -> None: + (workspace / ".loop" / "state.json").write_text(json.dumps(values), encoding="utf-8") + + +def test_status_report_missing_store_is_distinct_from_empty_store(tmp_path): + workspace, _ = _workspace(tmp_path) + with pytest.raises(RuntimeStoreError) as missing: + status_report(workspace) + SQLiteEventStore(workspace / ".loop" / "events.db")._connect().close() + with pytest.raises(RuntimeStoreError) as empty: + status_report(workspace) + assert (missing.value.code, empty.value.code) == ("missing_store", "empty_store") + + +def test_status_report_empty_store_reports_typed_finding(tmp_path): + workspace, store = _workspace(tmp_path) + store._connect().close() + with pytest.raises(RuntimeStoreError) as exc: + status_report(workspace) + assert exc.value.code == "empty_store" + + +def test_replay_report_missing_store_is_distinct_from_empty_store(tmp_path): + workspace, _ = _workspace(tmp_path) + with pytest.raises(RuntimeStoreError) as missing: + replay_report(workspace) + SQLiteEventStore(workspace / ".loop" / "events.db")._connect().close() + with pytest.raises(RuntimeStoreError) as empty: + replay_report(workspace) + assert (missing.value.code, empty.value.code) == ("missing_store", "empty_store") + + +def test_status_report_healthy_single_event_run_agrees_with_state_json(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + report = status_report(workspace) + assert report["ok"] and report["state_json_agrees"] and report["event_count"] == 1 + + +def test_status_report_detects_state_json_divergence(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + _write_state(workspace, state="verify", iteration_id=0, active_task=None, terminal_state=None) + assert status_report(workspace)["divergence"][0]["code"] == "state_field_mismatch" + + +def test_replay_report_healthy_log_is_consistent_on_double_fold(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + report = replay_report(workspace) + assert report["ok"] and report["deterministic"] and report["legal_sequence"] + + +def test_replay_report_reports_illegal_sequence_as_typed_finding_not_traceback(tmp_path): + workspace, store = _workspace(tmp_path) + store.append("run-1", "iteration_appended", {"iteration_id": 0, "outcome": "task_passed"}, actor="test") + report = replay_report(workspace) + assert not report["legal_sequence"] and report["findings"][0]["code"] == "illegal_event_sequence" + + +def test_replay_report_detects_desynced_terminal_window_file_without_event(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + (workspace / ".loop" / "terminal_state.json").write_text('{"state":"Succeeded"}', encoding="utf-8") + assert replay_report(workspace)["findings"][0]["code"] == "desynced_terminal_window" + + +def test_replay_report_detects_desynced_terminal_window_event_without_file(tmp_path): + workspace, store = _workspace(tmp_path) + _terminal(store) + assert replay_report(workspace)["findings"][0]["code"] == "desynced_terminal_window" + + +def test_replay_report_detects_terminal_state_mismatch_between_file_and_projection(tmp_path): + workspace, store = _workspace(tmp_path) + _terminal(store) + (workspace / ".loop" / "terminal_state.json").write_text('{"state":"FailedBlocked"}', encoding="utf-8") + assert replay_report(workspace)["findings"][0]["code"] == "desynced_terminal_window" + + +def test_ambiguous_run_id_is_a_typed_usage_error(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store, "run-a") + _open(store, "run-b") + with pytest.raises(RuntimeStoreError, match="ambiguous") as exc: + status_report(workspace) + assert exc.value.code == "ambiguous_run_id" + + +def test_events_db_is_opened_strictly_read_only_no_write_side_effects(tmp_path): + workspace, store = _workspace(tmp_path) + _terminal(store) + (workspace / ".loop" / "terminal_state.json").write_text('{"state":"Succeeded"}', encoding="utf-8") + files = sorted((workspace / ".loop").iterdir()) + before = {p.name: (p.stat().st_mtime_ns, hashlib.sha256(p.read_bytes()).hexdigest()) for p in files if p.is_file()} + status_report(workspace) + replay_report(workspace) + after = {p.name: (p.stat().st_mtime_ns, hashlib.sha256(p.read_bytes()).hexdigest()) for p in files if p.is_file()} + assert after == before + + +def test_status_report_reports_completion_policy_satisfaction_for_succeeded_terminal(tmp_path): + workspace, store = _workspace(tmp_path) + _terminal(store) + _write_state(workspace, state="terminal", iteration_id=0, active_task=None, terminal_state="Succeeded") + assert status_report(workspace)["completion_satisfied"] is True + + +def test_corrupt_store_file_is_a_typed_usage_error_not_a_traceback(tmp_path): + workspace, _ = _workspace(tmp_path) + (workspace / ".loop" / "events.db").write_text("not sqlite", encoding="utf-8") + with pytest.raises(RuntimeStoreError) as exc: + replay_report(workspace) + assert exc.value.code == "corrupt_store" + + +def test_help_lists_status_and_replay_commands(): + result = _run("--help") + assert result.returncode == 0 and "status" in result.stdout and "replay" in result.stdout + + +def test_status_missing_target_argument_prints_usage_and_exits_nonzero(): + result = _run("status") + assert result.returncode != 0 and "usage" in result.stderr.lower() + + +def test_replay_missing_target_argument_prints_usage_and_exits_nonzero(): + result = _run("replay") + assert result.returncode != 0 and "usage" in result.stderr.lower() + + +def test_status_nonexistent_target_gives_distinct_actionable_error(tmp_path): + result = _run("status", str(tmp_path / "missing")) + assert result.returncode == 2 and result.stdout == "" and "does not exist" in result.stderr + + +def test_replay_nonexistent_target_gives_distinct_actionable_error(tmp_path): + result = _run("replay", str(tmp_path / "missing")) + assert result.returncode == 2 and result.stdout == "" and "does not exist" in result.stderr + + +def test_status_accepts_mode_flag_like_doctor(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + result = _run("status", "--mode", "basic", str(workspace)) + assert result.returncode == 0 and json.loads(result.stdout)["requested_mode"] == "basic" + + +def test_replay_accepts_mode_flag_like_doctor(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + result = _run("replay", "--mode=basic", str(workspace)) + assert result.returncode == 0 and json.loads(result.stdout)["requested_mode"] == "basic" + + +def test_status_missing_event_store_exits_2_with_stderr_message_and_empty_stdout(tmp_path): + workspace, _ = _workspace(tmp_path) + result = _run("status", str(workspace)) + assert result.returncode == 2 and result.stdout == "" and "event store" in result.stderr + + +def test_replay_missing_event_store_exits_2_with_stderr_message_and_empty_stdout(tmp_path): + workspace, _ = _workspace(tmp_path) + result = _run("replay", str(workspace)) + assert result.returncode == 2 and result.stdout == "" and "event store" in result.stderr + + +def test_status_healthy_run_exits_0_with_json_report(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + result = _run("status", str(workspace)) + assert result.returncode == 0 and json.loads(result.stdout)["ok"] is True + + +def test_replay_healthy_log_exits_0_with_json_report(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + result = _run("replay", str(workspace)) + assert result.returncode == 0 and json.loads(result.stdout)["ok"] is True + + +def test_replay_illegal_sequence_exits_1_with_json_finding_no_traceback(tmp_path): + workspace, store = _workspace(tmp_path) + store.append("run-1", "iteration_appended", {"iteration_id": 0, "outcome": "task_passed"}, actor="test") + result = _run("replay", str(workspace)) + assert result.returncode == 1 and json.loads(result.stdout)["findings"][0]["code"] == "illegal_event_sequence" and "Traceback" not in result.stderr + + +def test_status_and_replay_mode_release_without_jsonschema_is_a_usage_error(tmp_path): + workspace, store = _workspace(tmp_path) + _open(store) + for command in ("status", "replay"): + result = _run(command, "--mode", "release", str(workspace)) + if importlib.util.find_spec("jsonschema") is None: + assert result.returncode == 2 and result.stdout == "" and "jsonschema" in result.stderr + else: + assert result.returncode == 0 and json.loads(result.stdout)["requested_mode"] == "release"