feat(doctor): event-store consistency gate — compose status/replay when .loop/events.db exists

loop/contract.py

@@ -750,4 +750,10 @@ def validate_contract(target: str | Path, *, mode: str | None = None) -> dict[st
 
 
 def doctor_report(target: str | Path, *, mode: str | None = None) -> dict[str, Any]:
-    return validate_contract(target, mode=mode)
+    report = validate_contract(target, mode=mode)
+    from .runtime import event_consistency_issues
+
+    event_store, event_issues = event_consistency_issues(target, mode=mode)
+    issues = report["issues"] + list(event_issues) if event_issues else report["issues"]
+    return {**report, "event_store": event_store, "issues": issues,
+            "ok": report["ok"] and not event_issues}

loop/runtime.py

@@ -195,3 +195,29 @@ def replay_report(target: str | Path, *, mode: str | None = None) -> dict[str, A
         "deterministic": deterministic, "legal_sequence": legal_sequence,
         "terminal_desync": terminal_desync, "findings": findings,
     }
+
+
+def event_consistency_issues(
+    target: str | Path, *, mode: str | None = None
+) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    """Return event-store health and the existing status/replay findings."""
+    path = _store_path(target)
+    if not path.exists():
+        return {"present": False}, []
+    try:
+        status = status_report(target, mode=mode)
+        replay = replay_report(target, mode=mode)
+    except RuntimeStoreError as exc:
+        return {"present": True, "readable": False, "error_code": exc.code}, [
+            ContractIssue(exc.code, str(exc))
+        ]
+    issues = list(status["divergence"]) + list(replay["findings"])
+    return {
+        "present": True,
+        "readable": True,
+        "run_id": status["run_id"],
+        "event_count": status["event_count"],
+        "state_json_agrees": status["state_json_agrees"],
+        "deterministic": replay["deterministic"],
+        "legal_sequence": replay["legal_sequence"],
+    }, issues

reference/repo-os-contract.md

@@ -591,10 +591,11 @@ refuse mutation or removal of a committed row, regardless of caller.
 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; its
-on-disk location is `.loop/events.db`, with one run discovered per store by
-the runtime readers (multi-run support remains deferred).
+**Scope boundary:** `loop doctor` reads `event@1` when `.loop/events.db`
+exists (§22) by composing the exact `status`/`replay` read-only verbs, not by
+duplicating their logic; an absent store is conformant and adds no issues
+(§12.1's terminal-file-iff rule has an analogous "absent is fine" shape). One
+run is discovered per store; 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
@@ -759,3 +760,21 @@ Anthropic guidance on long-running agent harnesses (anthropic.com, 2025), OpenAI
 Conductor, and arXiv PreFlect (2602.07187), SWE-Marathon (2606.07682), Web Agents
 Plan-Then-Execute (2605.14290), Plan Compliance (2604.12147), and Code as Agent Harness
 (2605.18747).
+
+---
+
+## 22. `loop doctor` — event-store consistency gate
+
+When `.loop/events.db` exists, `loop doctor` composes the exact read-only
+`status`/`replay` verbs (§16, §20) — never duplicating their fold/divergence
+logic — and folds their findings into its own `issues`/`ok`. An absent store is
+conformant: doctor reports `"event_store": {"present": false}` and every other
+key is byte-identical to a store-less report. A present, readable store adds
+`"event_store": {"present": true, "readable": true, "run_id", "event_count",
+"state_json_agrees", "deterministic", "legal_sequence"}`; any of
+`state_field_mismatch`, `desynced_terminal_window`, `terminal_state_mismatch`,
+or `illegal_event_sequence` fails doctor (`ok: false`) with the identical issue
+code the `status`/`replay` verbs already use. A store that cannot be read at
+all — `corrupt_store`, `empty_store`, or `ambiguous_run_id` — also fails doctor
+rather than being silently skipped; `"event_store"` reports
+`{"present": true, "readable": false, "error_code": <code>}` in that case.

scripts/test_doctor_eventstore.py

@@ -0,0 +1,166 @@
+"""Doctor integration tests for the read-only EventStore consistency gate."""
+
+import json
+
+import pytest
+
+from loop.contract import doctor_report, validate_contract
+from loop.events import SQLiteEventStore
+from loop.scaffold import scaffold
+
+
+def _fresh_contract(tmp_path, name="workspace"):
+    target = tmp_path / name
+    scaffold(target)
+    return target
+
+
+def _sync_active_task(target):
+    path = target / ".loop" / "state.json"
+    state = json.loads(path.read_text(encoding="utf-8"))
+    state["active_task"] = None
+    path.write_text(json.dumps(state), encoding="utf-8")
+
+
+def _store(target):
+    return SQLiteEventStore(target / ".loop" / "events.db")
+
+
+def _open(store, run_id="run-1"):
+    return store.append(run_id, "contract_opened", {"workspace": "workspace"}, actor="test")
+
+
+def _terminal(store):
+    opened = _open(store)
+    return store.append(
+        "run-1", "terminal_written",
+        {"state": "Succeeded", "criteria_met": {"gate": True}, "evidence": ["proof"], "false_completion": False},
+        actor="test", causation_id=opened["event_id"],
+    )
+
+
+def _force_structural_mode(monkeypatch):
+    import loop.contract as contract
+
+    monkeypatch.setattr(contract, "_validation_mode", lambda: "structural-fallback")
+
+
+def _codes(report):
+    return {issue["code"] for issue in report["issues"]}
+
+
+def _terminal_file(state, *, evidence):
+    return json.dumps({
+        "schema": "loop-engineer/terminal@1",
+        "state": state,
+        "criteria_met": {"gate": True},
+        "evidence": evidence,
+        "false_completion": False,
+    })
+
+
+def test_absent_event_store_matches_pre_slice_doctor_shape(tmp_path):
+    target = _fresh_contract(tmp_path)
+    file_only = validate_contract(target)
+    report = doctor_report(target)
+    assert report["event_store"] == {"present": False}
+    assert {key: value for key, value in report.items() if key != "event_store"} == file_only
+
+
+@pytest.mark.parametrize("mode", ["jsonschema", "structural-fallback"])
+def test_synced_happy_path_is_doctor_clean(tmp_path, monkeypatch, mode):
+    if mode == "jsonschema":
+        pytest.importorskip("jsonschema")
+    else:
+        _force_structural_mode(monkeypatch)
+    target = _fresh_contract(tmp_path)
+    _sync_active_task(target)
+    _open(_store(target))
+    report = doctor_report(target)
+    assert report["validation_mode"] == mode
+    assert report["ok"] is True, report["issues"]
+    assert report["event_store"]["present"] is True
+    assert report["event_store"]["state_json_agrees"] is True
+    assert report["event_store"]["deterministic"] is True
+    assert report["event_store"]["legal_sequence"] is True
+
+
+@pytest.mark.parametrize("mode", ["jsonschema", "structural-fallback"])
+def test_state_field_mismatch_fails_doctor(tmp_path, monkeypatch, mode):
+    if mode == "jsonschema":
+        pytest.importorskip("jsonschema")
+    else:
+        _force_structural_mode(monkeypatch)
+    target = _fresh_contract(tmp_path)
+    _sync_active_task(target)
+    _open(_store(target))
+    path = target / ".loop" / "state.json"
+    state = json.loads(path.read_text(encoding="utf-8"))
+    state["state"] = "plan"
+    path.write_text(json.dumps(state), encoding="utf-8")
+    report = doctor_report(target)
+    assert report["validation_mode"] == mode
+    assert report["ok"] is False
+    assert "state_field_mismatch" in _codes(report)
+
+
+def test_desynced_terminal_window_fails_doctor(tmp_path):
+    target = _fresh_contract(tmp_path)
+    _sync_active_task(target)
+    _terminal(_store(target))
+    (target / ".loop" / "terminal_state.json").write_text(
+        _terminal_file("FailedBlocked", evidence=[]), encoding="utf-8"
+    )
+    report = doctor_report(target)
+    assert report["ok"] is False
+    assert "desynced_terminal_window" in _codes(report)
+
+
+def test_terminal_state_mismatch_fails_doctor(tmp_path):
+    target = _fresh_contract(tmp_path)
+    _sync_active_task(target)
+    _terminal(_store(target))
+    (target / ".loop" / "terminal_state.json").write_text(
+        _terminal_file("Succeeded", evidence=["different"]), encoding="utf-8"
+    )
+    report = doctor_report(target)
+    assert report["ok"] is False
+    assert "terminal_state_mismatch" in _codes(report)
+
+
+def test_illegal_event_sequence_fails_doctor(tmp_path):
+    target = _fresh_contract(tmp_path)
+    _store(target).append("run-1", "iteration_appended", {"iteration_id": 0, "outcome": "task_passed"}, actor="test")
+    report = doctor_report(target)
+    assert report["ok"] is False
+    assert "illegal_event_sequence" in _codes(report)
+
+
+def test_corrupt_store_fails_doctor_without_traceback(tmp_path):
+    target = _fresh_contract(tmp_path)
+    path = target / ".loop" / "events.db"
+    path.write_text("not sqlite", encoding="utf-8")
+    report = doctor_report(target)
+    assert report["ok"] is False
+    assert report["event_store"]["error_code"] == "corrupt_store"
+    assert "corrupt_store" in _codes(report)
+
+
+def test_empty_store_fails_doctor(tmp_path):
+    target = _fresh_contract(tmp_path)
+    _store(target)._connect().close()
+    report = doctor_report(target)
+    assert report["ok"] is False
+    assert report["event_store"]["error_code"] == "empty_store"
+    assert "empty_store" in _codes(report)
+
+
+def test_ambiguous_run_id_fails_doctor(tmp_path):
+    target = _fresh_contract(tmp_path)
+    store = _store(target)
+    _open(store, "run-a")
+    _open(store, "run-b")
+    report = doctor_report(target)
+    assert report["ok"] is False
+    assert report["event_store"]["error_code"] == "ambiguous_run_id"
+    assert "ambiguous_run_id" in _codes(report)