diff --git a/README.md b/README.md index a08b424..e58dc4b 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,11 @@ ships, on disk and runnable today: - an **event-sourced runtime** — `run`, `status`, `replay`, `simulate`, and approve/pause/resume/cancel over an append-only SQLite event log (`.loop/events.db`), folded by a deterministic reducer that enforces the same - completion gate as the writers, with crash-safe single-step resume. + completion gate as the writers, with crash-safe single-step resume. Events are + hash-chained; `loop doctor --expect-chain-head` verifies the log against an + externally anchored head. The chain is tamper-evident **relative to an + anchor** — an adversary with workspace write access can rewrite an unanchored + log. ![The inspector scores a self-asserted DIY loop 0/weak, then the gate-backed example 90/strong — both runs live](docs/demo.gif) diff --git a/action.yml b/action.yml index 238be05..ce8bca5 100644 --- a/action.yml +++ b/action.yml @@ -25,6 +25,18 @@ inputs: description: "Token for the optional PR scorecard comment. Empty skips the comment." required: false default: "" + expect-chain-head: + description: >- + Fail the gate unless the store's chain head equals this 64-hex value (an + externally remembered anchor). Empty performs NO cross-run tamper + detection — the gate then only records the head for a later comparison. + required: false + default: "" + +outputs: + chain-head: + description: "Chain head event_hash observed by this gate run ('' when the store has no chained events)." + value: ${{ steps.chain-head.outputs.chain-head }} runs: using: "composite" @@ -50,12 +62,17 @@ runs: shell: bash env: LOOP_PATH: "${{ inputs.path }}" + LOOP_EXPECT_HEAD: "${{ inputs.expect-chain-head }}" run: | # -eo pipefail (GitHub's bash default) makes a doctor failure fail the step # despite the tee. Then assert the strict validation path actually ran, so a # future packaging regression that drops the extras fails loudly here. The # check reads one fixed field from doctor's own JSON — no fragile parsing. - loop doctor "$LOOP_PATH" | tee "${RUNNER_TEMP}/doctor.json" + if [ -n "$LOOP_EXPECT_HEAD" ]; then + loop doctor --expect-chain-head "$LOOP_EXPECT_HEAD" "$LOOP_PATH" | tee "${RUNNER_TEMP}/doctor.json" + else + loop doctor "$LOOP_PATH" | tee "${RUNNER_TEMP}/doctor.json" + fi python - "${RUNNER_TEMP}/doctor.json" <<'PY' import json, sys mode = json.load(open(sys.argv[1])).get("validation_mode") @@ -65,6 +82,30 @@ runs: raise SystemExit(1) PY + - name: chain head (anchor surface) + id: chain-head + if: always() + shell: bash + run: | + # Runs even when doctor failed: an anchor MISMATCH is exactly the run whose + # observed head an operator needs recorded. Absent/empty doctor.json (doctor + # never got to write one) is a silent no-op, not an error. + [ -s "${RUNNER_TEMP}/doctor.json" ] || exit 0 + python - "${RUNNER_TEMP}/doctor.json" "$GITHUB_STEP_SUMMARY" "$GITHUB_OUTPUT" <<'PY' + import json, sys + try: + doctor = json.load(open(sys.argv[1])) + except (OSError, json.JSONDecodeError): + doctor = {} + chain = (doctor.get("event_store") or {}).get("chain") or {} + head = chain.get("head") or {} + value = head.get("event_hash") or "" + line = (f"**loop-engineer chain head:** `{value}` (sequence {head.get('sequence')})" + if value else "**loop-engineer chain head:** none (no chained events)") + open(sys.argv[2], "a").write(line + "\n") + open(sys.argv[3], "a").write(f"chain-head={value}\n") + PY + - name: loop inspect (scorecard) shell: bash env: diff --git a/loop/__main__.py b/loop/__main__.py index 182bdd0..4308d98 100644 --- a/loop/__main__.py +++ b/loop/__main__.py @@ -12,19 +12,20 @@ _PROG = "python3 -m loop" -_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "architect") +_COMMANDS = ("scaffold", "doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "migrate", "architect") # 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", "simulate", "run", "approve", "pause", "resume", "cancel") +_READ_COMMANDS = ("doctor", "validate", "verify", "inspect", "metrics", "plan-lint", "status", "replay", "simulate", "run", "approve", "pause", "resume", "cancel", "migrate") -_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} doctor|validate|verify [--mode basic|strict|release] + [--expect-chain-head SHA256] {_PROG} status [--mode basic|strict|release] {_PROG} replay [--mode basic|strict|release] {_PROG} simulate [--mode basic|strict|release] @@ -33,6 +34,7 @@ {_PROG} pause --reason REASON [--mode basic|strict|release] {_PROG} resume [--note NOTE] [--mode basic|strict|release] {_PROG} cancel [--reason REASON] [--mode basic|strict|release] + {_PROG} migrate {_PROG} plan-lint [--mode basic|strict|release] commands: @@ -57,6 +59,7 @@ pause Pause a non-terminal run. resume Resume a paused run. cancel Terminate a non-terminal run as AbortedByHuman. + migrate Add hash-chain columns to a legacy events.db (explicit, idempotent; the only store-upgrade path). architect Not implemented by this CLI: architecture classification and ADR authorship require agentic judgment, not deterministic code. See the loop-architect skill. @@ -69,6 +72,10 @@ --mode {{basic,strict,release}} (doctor/validate/verify/plan-lint/status/replay/simulate/run) basic forces structural checks; strict/release require jsonschema. Default: auto-detect. + --expect-chain-head SHA256 + (doctor/validate/verify) fail unless the event store's chain head + is exactly this 64-character lowercase hex hash. A missing, + unreadable, unchained, or diverged store fails the gate. --baseline (metrics only) write docs/metrics-baseline.json over a gate-backed run; exits non-zero and writes nothing otherwise. -h, --help Show this help and exit. @@ -238,6 +245,26 @@ def main(argv: list[str] | None = None) -> int: print(_USAGE, file=sys.stderr) return 2 + expect_chain_head = None + if command in {"doctor", "validate", "verify"}: + try: + expect_chain_head, argv = _extract_value_flag(argv, "--expect-chain-head") + except ValueError as exc: + print(f"{command}: {exc}", file=sys.stderr) + print(_USAGE, file=sys.stderr) + return 2 + if expect_chain_head is not None and re.fullmatch(r"[0-9a-f]{64}", expect_chain_head) is None: + print(f"{command}: --expect-chain-head must be a 64-character lowercase hex sha256", + file=sys.stderr) + return 2 + elif any(a == "--expect-chain-head" or a.startswith("--expect-chain-head=") for a in argv): + # No generic unknown-flag guard exists for the other commands, and scaffold + # would otherwise CREATE a directory named after the flag. + print(f"{command}: --expect-chain-head is only valid for doctor/validate/verify", + file=sys.stderr) + print(_USAGE, file=sys.stderr) + return 2 + decision = resume_target = reason = note = None if command in {"approve", "pause", "resume", "cancel"}: try: @@ -308,7 +335,7 @@ def main(argv: list[str] | None = None) -> int: if command in {"doctor", "validate", "verify"}: try: - return _print_json(doctor_report(target, mode=mode)) + return _print_json(doctor_report(target, mode=mode, expect_chain_head=expect_chain_head)) except ValidationModeError as exc: print(f"{command}: {exc}", file=sys.stderr) return 2 @@ -320,6 +347,14 @@ def main(argv: list[str] | None = None) -> int: print(f"{command}: {exc}", file=sys.stderr) return 2 + if command == "migrate": + from .migrate import migrate_store + try: + return _print_json(migrate_store(target)) + except RuntimeStoreError as exc: + print(f"migrate: {exc}", file=sys.stderr) + return 2 + if command in {"status", "replay", "simulate"}: from .runner import RunnerError try: diff --git a/loop/chain.py b/loop/chain.py new file mode 100644 index 0000000..dd66475 --- /dev/null +++ b/loop/chain.py @@ -0,0 +1,88 @@ +"""Pure hash-chain canonicalization and verification for event@1 records. + +Stdlib-only and import-free of other loop modules: verify_chain() must work over +any ordered event list (a SQLite read or a JSONL export) so third parties can +re-verify a chain without this package's store code. Canonical form is +json.dumps(sort_keys, separators=(",",":"), ensure_ascii=False, allow_nan=False) +encoded UTF-8 — pinned normatively in reference/repo-os-contract.md #16. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any, Iterable, Mapping + +_PREIMAGE_FIELDS = ( + "schema", "run_id", "sequence", "event_id", "type", "actor", "ts", + "causation_id", "correlation_id", "payload", "artifact_hashes", + "prev_event_hash", +) + + +class ChainHashError(ValueError): + """A value cannot be canonically hashed (non-JSON type, non-finite float, lone surrogate).""" + + +def canonical_json(value: Any) -> str: + try: + text = json.dumps(value, sort_keys=True, separators=(",", ":"), + ensure_ascii=False, allow_nan=False) + except (TypeError, ValueError) as exc: + raise ChainHashError(f"value is not canonically serializable: {exc}") from exc + try: + text.encode("utf-8") + except UnicodeEncodeError as exc: + raise ChainHashError(f"value contains a lone surrogate: {exc}") from exc + return text + + +def compute_event_hash(record: Mapping[str, Any]) -> str: + preimage = {field: record.get(field) for field in _PREIMAGE_FIELDS} + return hashlib.sha256(canonical_json(preimage).encode("utf-8")).hexdigest() + + +def link_issue(record: Mapping[str, Any], prev_head: Mapping[str, Any] | None) -> str | None: + """One incremental chain check; None means record legally extends prev_head.""" + sequence = record.get("sequence") + stored = record.get("event_hash") + if stored is None: + if prev_head is None: + return None + return (f"unchained event after chained prefix at sequence {sequence!r} " + "(a pre-0.10.0 writer appended to a chained store, or the row was tampered)") + expected_prev = prev_head["event_hash"] if prev_head is not None else None + if record.get("prev_event_hash") != expected_prev: + return f"prev_event_hash mismatch at sequence {sequence!r}" + try: + recomputed = compute_event_hash(record) + except ChainHashError as exc: + return f"unhashable record at sequence {sequence!r}: {exc}" + if recomputed != stored: + return f"event_hash mismatch at sequence {sequence!r}" + return None + + +def verify_chain(events: Iterable[Mapping[str, Any]], *, expected_head: str | None = None) -> dict[str, Any]: + """Verify a COMPLETE run stream's hash chain (sequence 0 onward); pure, I/O-free.""" + issues: list[str] = [] + unchained_prefix = 0 + chained_events = 0 + head: dict[str, Any] | None = None + for record in events: + issue = link_issue(record, head) + if issue is not None: + issues.append(issue) + break + if record.get("event_hash") is None: + unchained_prefix += 1 + continue + chained_events += 1 + head = {"sequence": record.get("sequence"), "event_hash": record["event_hash"]} + if expected_head is not None and not issues: + if head is None: + issues.append("expected chain head, but the stream has no chained events") + elif head["event_hash"] != expected_head: + issues.append(f"chain head {head['event_hash']} does not match expected {expected_head}") + return {"ok": not issues, "issues": issues, "chained_events": chained_events, + "unchained_prefix": unchained_prefix, "head": head} diff --git a/loop/contract.py b/loop/contract.py index 4b7e4e6..b523400 100644 --- a/loop/contract.py +++ b/loop/contract.py @@ -749,11 +749,13 @@ 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]: +def doctor_report(target: str | Path, *, mode: str | None = None, + expect_chain_head: str | None = None) -> dict[str, Any]: report = validate_contract(target, mode=mode) from .runtime import event_consistency_issues - event_store, event_issues = event_consistency_issues(target, mode=mode) + event_store, event_issues = event_consistency_issues( + target, mode=mode, expect_chain_head=expect_chain_head) 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} diff --git a/loop/events.py b/loop/events.py index 85b1e72..b504beb 100644 --- a/loop/events.py +++ b/loop/events.py @@ -10,6 +10,7 @@ from pathlib import Path from typing import Any, Mapping, Protocol, Sequence, runtime_checkable +from . import chain from .contract import ContractIssue, _resolve_requested_mode, _schemas_dir from .emit import _ITERATION_OUTCOMES, _RECEIPT_OUTCOMES, _RECEIPT_ROLES @@ -41,6 +42,8 @@ ts TEXT NOT NULL, payload TEXT NOT NULL, artifact_hashes TEXT NOT NULL, + prev_event_hash TEXT, + event_hash TEXT NOT NULL, PRIMARY KEY (run_id, sequence) ) """ @@ -66,6 +69,14 @@ class SequenceConflictError(ValueError): """expected_sequence differed from the atomically assigned next sequence.""" +class EventRowDecodeError(ValueError): + """A stored payload/artifact_hashes column is not valid JSON.""" + + +class EventStoreOperationalError(RuntimeError): + """The events table exists but cannot service this operation (schema drift, lock).""" + + @runtime_checkable class EventStore(Protocol): def append(self, run_id: str, event_type: str, payload: Mapping[str, Any], *, actor: str, @@ -191,6 +202,11 @@ def _structural_validate_event(data: dict[str, Any]) -> list[str]: for field in ("causation_id", "correlation_id"): if field in data and data[field] is not None and not isinstance(data[field], str): issues.append(f"{field} must be a string or null") + for field in ("prev_event_hash", "event_hash"): + if field in data and data[field] is not None and ( + not isinstance(data[field], str) + or re.search(r"^[0-9a-f]{64}$", data[field]) is None): + issues.append(f"{field} must be null or a 64-character lowercase hex sha256") hashes = data.get("artifact_hashes", []) if not isinstance(hashes, list): issues.append("artifact_hashes must be an array") @@ -234,6 +250,45 @@ def validate_event(data: dict[str, Any], *, mode: str | None = None) -> dict[str return _validate_event_dict(data, mode=mode) +def has_chain_columns(conn: sqlite3.Connection) -> bool: + return any(row[1] == "event_hash" for row in conn.execute("PRAGMA table_info(events)")) + + +def store_user_version(conn: sqlite3.Connection) -> int: + return int(conn.execute("PRAGMA user_version").fetchone()[0]) + + +_BASE_COLUMNS = ("run_id", "sequence", "event_id", "type", "actor", "causation_id", + "correlation_id", "ts", "payload", "artifact_hashes") + + +def read_event_rows(conn: sqlite3.Connection, run_id: str, *, + since_sequence: int | None = None) -> list[dict[str, Any]]: + """The single event-row projection shared by store, runtime, and runner reads. + + Records always carry prev_event_hash/event_hash keys; legacy stores project + None. Owns the JSON-decode translation every call site used to repeat. + """ + chained = has_chain_columns(conn) + columns = _BASE_COLUMNS + (("prev_event_hash", "event_hash") if chained else ()) + operator, cursor = (">=", 0) if since_sequence is None else (">", since_sequence) + rows = conn.execute( + f"SELECT {', '.join(columns)} FROM events WHERE run_id = ? AND sequence {operator} ? " + "ORDER BY sequence ASC", (run_id, cursor)).fetchall() + records: list[dict[str, Any]] = [] + try: + for row in rows: + records.append({ + "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]), + "prev_event_hash": row[10] if chained else None, + "event_hash": row[11] if chained else None}) + except (TypeError, json.JSONDecodeError) as exc: + raise EventRowDecodeError(f"event row is not decodable: {exc}") from exc + return records + + class SQLiteEventStore: """A transactional SQLite/WAL event store with DB-enforced append-only rows.""" @@ -245,9 +300,13 @@ def _connect(self) -> sqlite3.Connection: conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=FULL") conn.execute("PRAGMA busy_timeout=5000") + fresh = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='events'").fetchone() is None conn.execute(_CREATE_EVENTS_TABLE) conn.execute(_CREATE_NO_UPDATE_TRIGGER) conn.execute(_CREATE_NO_DELETE_TRIGGER) + if fresh: + conn.execute("PRAGMA user_version = 2") return conn def append(self, run_id: str, event_type: str, payload: Mapping[str, Any], *, actor: str, @@ -275,15 +334,39 @@ def append(self, run_id: str, event_type: str, payload: Mapping[str, Any], *, ac conn.execute("ROLLBACK") raise SequenceConflictError(f"expected next sequence {next_sequence} for run_id {run_id!r}, caller supplied {expected_sequence}") record["sequence"] = next_sequence + chained = has_chain_columns(conn) + if chained: + prev_row = conn.execute( + "SELECT event_hash FROM events WHERE run_id = ? AND sequence = ?", + (run_id, next_sequence - 1)).fetchone() if next_sequence else None + record["prev_event_hash"] = prev_row[0] if prev_row else None + try: + record["event_hash"] = chain.compute_event_hash(record) + except chain.ChainHashError as exc: + conn.execute("ROLLBACK") + raise EventValidationError(str(exc)) from exc + else: + record["prev_event_hash"] = None + record["event_hash"] = None payload_json = json.dumps(record["payload"], sort_keys=True) hashes_json = json.dumps([dict(item) for item in record["artifact_hashes"]], sort_keys=True) try: - conn.execute("INSERT INTO events (run_id, sequence, event_id, type, actor, causation_id, correlation_id, ts, payload, artifact_hashes) VALUES (?,?,?,?,?,?,?,?,?,?)", - (record["run_id"], record["sequence"], record["event_id"], record["type"], record["actor"], record["causation_id"], record["correlation_id"], record["ts"], payload_json, hashes_json)) + if chained: + conn.execute("INSERT INTO events (run_id, sequence, event_id, type, actor, causation_id, correlation_id, ts, payload, artifact_hashes, prev_event_hash, event_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", + (record["run_id"], record["sequence"], record["event_id"], record["type"], record["actor"], record["causation_id"], record["correlation_id"], record["ts"], payload_json, hashes_json, record["prev_event_hash"], record["event_hash"])) + else: + conn.execute("INSERT INTO events (run_id, sequence, event_id, type, actor, causation_id, correlation_id, ts, payload, artifact_hashes) VALUES (?,?,?,?,?,?,?,?,?,?)", + (record["run_id"], record["sequence"], record["event_id"], record["type"], record["actor"], record["causation_id"], record["correlation_id"], record["ts"], payload_json, hashes_json)) except sqlite3.IntegrityError as exc: conn.execute("ROLLBACK") raise DuplicateEventError(record["event_id"]) from exc conn.execute("COMMIT") + except sqlite3.OperationalError as exc: + try: + conn.execute("ROLLBACK") + except sqlite3.Error: + pass + raise EventStoreOperationalError(f"event store cannot accept appends: {exc}") from exc finally: conn.close() return record @@ -292,11 +375,9 @@ def read(self, run_id: str, *, since_sequence: int | None = None) -> list[dict[s """Read the full stream for None, or events strictly after an integer cursor.""" conn = self._connect() try: - operator, cursor = (">=", 0) if since_sequence is None else (">", since_sequence) - rows = conn.execute(f"SELECT run_id, sequence, event_id, type, actor, causation_id, correlation_id, ts, payload, artifact_hashes FROM events WHERE run_id = ? AND sequence {operator} ? ORDER BY sequence ASC", (run_id, cursor)).fetchall() + return read_event_rows(conn, run_id, since_sequence=since_sequence) finally: conn.close() - 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] def latest_sequence(self, run_id: str) -> int | None: conn = self._connect() diff --git a/loop/migrate.py b/loop/migrate.py new file mode 100644 index 0000000..6c2ebdc --- /dev/null +++ b/loop/migrate.py @@ -0,0 +1,45 @@ +"""Explicit legacy-store migration: add chain columns; never rewrites rows. + +Backfilling hashes onto existing rows is deliberately impossible — the +append-only triggers forbid UPDATE — so migration only widens the table and +stamps user_version=2. Pre-migration rows stay an *unchained prefix* that +doctor reports explicitly; the first post-migration append is a chain genesis. +Migrated columns stay nullable (legacy rows are NULL), so unlike a fresh store +a migrated store cannot refuse a pre-0.10.0 writer at the DB layer — see the +compatibility rule in reference/repo-os-contract.md #16. +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path +from typing import Any + +from .events import has_chain_columns +from .paths import resolve_loop_paths +from .runtime import RuntimeStoreError + + +def migrate_store(target: str | Path) -> dict[str, Any]: + path = resolve_loop_paths(target).loop_dir / "events.db" + if not path.exists(): + raise RuntimeStoreError("missing_store", f"event store does not exist: {path}") + try: + conn = sqlite3.connect(str(path), isolation_level=None, timeout=5.0) + try: + conn.execute("PRAGMA busy_timeout=5000") + already = has_chain_columns(conn) + if not already: + conn.execute("BEGIN IMMEDIATE") + conn.execute("ALTER TABLE events ADD COLUMN prev_event_hash TEXT") + conn.execute("ALTER TABLE events ADD COLUMN event_hash TEXT") + conn.execute("COMMIT") + conn.execute("PRAGMA user_version = 2") + unchained = conn.execute("SELECT COUNT(*) FROM events WHERE event_hash IS NULL").fetchone()[0] + top = conn.execute("SELECT MAX(sequence) FROM events").fetchone()[0] + finally: + conn.close() + except sqlite3.DatabaseError as exc: + raise RuntimeStoreError("corrupt_store", f"cannot migrate event store: {exc}") from exc + return {"ok": True, "migrated": not already, "store": str(path), "user_version": 2, + "unchained_rows": unchained, "chained_from_sequence": 0 if top is None else top + 1} diff --git a/loop/reducer.py b/loop/reducer.py index bfa0ec8..569044f 100644 --- a/loop/reducer.py +++ b/loop/reducer.py @@ -4,7 +4,7 @@ from typing import Any, Iterable, Mapping -from . import fsm +from . import chain, fsm from .completion import CompletionPolicyError, criteria_satisfy_completion, normalize_completion_policy from .contract import TERMINAL_STATES from .events import _structural_validate_event @@ -14,10 +14,19 @@ class EventReplayError(ValueError): """An event stream is malformed or violates a replay domain invariant.""" +class ChainBreakError(EventReplayError): + """The event stream's hash link is broken or forged. + + Not a gap detector: a mid-stream deletion surfaces as a sequence error, and a + truncated tail is caught only by an anchored head. + """ + + def _empty_projection(run_id: str | None) -> dict[str, Any]: return {"run_id": run_id, "state": None, "iteration_id": None, "active_task": None, "terminal": None, "runlog_entries": [], "receipts": [], "superseded_history": [], "event_count": 0, - "last_sequence": None, "paused": False, "pause_reason": None, "pending_approval": None} + "last_sequence": None, "paused": False, "pause_reason": None, "pending_approval": None, + "chain_head": None, "unchained_prefix": 0} def _validate_terminal_payload_semantics(payload: Mapping[str, Any]) -> None: @@ -80,6 +89,9 @@ def _reduce_one(state: dict[str, Any], event: Mapping[str, Any]) -> dict[str, An expected_sequence = 0 if state["last_sequence"] is None else state["last_sequence"] + 1 if event["sequence"] != expected_sequence: raise EventReplayError(f"non-monotonic sequence: expected {expected_sequence}, got {event['sequence']!r}") + issue = chain.link_issue(event, state["chain_head"]) + if issue is not None: + raise ChainBreakError(f"event chain broken: {issue}") event_type = event["type"] if state["terminal"] is not None: if event_type != "terminal_superseded": @@ -91,6 +103,10 @@ def _reduce_one(state: dict[str, Any], event: Mapping[str, Any]) -> dict[str, An if event_type != "contract_opened" and state["state"] is None: raise EventReplayError(f"{event_type} event before contract_opened") new_state = {**state, "run_id": run_id, "last_sequence": event["sequence"], "event_count": state["event_count"] + 1} + if event.get("event_hash") is None: + new_state["unchained_prefix"] = state["unchained_prefix"] + 1 + else: + new_state["chain_head"] = {"sequence": event["sequence"], "event_hash": event["event_hash"]} payload = event["payload"] if event_type == "contract_opened": new_state["state"] = "intake" diff --git a/loop/runcontrol.py b/loop/runcontrol.py index cad43e4..2c6aff8 100644 --- a/loop/runcontrol.py +++ b/loop/runcontrol.py @@ -5,8 +5,14 @@ from typing import Any from . import emit, fsm, runner -from .events import EventValidationError, SequenceConflictError, SQLiteEventStore +from .events import ( + EventStoreOperationalError, + EventValidationError, + SequenceConflictError, + SQLiteEventStore, +) from .paths import resolve_loop_paths +from .runtime import RuntimeStoreError class RunControlError(RuntimeError): @@ -36,6 +42,8 @@ def _append_event(target: str | Path, run_id: str, projection: dict[str, Any], e raise RunControlConflictError("retry: another writer advanced the run") from exc except EventValidationError as exc: raise RunControlUsageError(str(exc)) from exc + except EventStoreOperationalError as exc: + raise RuntimeStoreError("event_store_unusable", str(exc)) from exc def approve_run(target: str | Path, *, decision: str, resume_target: str | None = None, diff --git a/loop/runner.py b/loop/runner.py index a7ff02c..fd89b1f 100644 --- a/loop/runner.py +++ b/loop/runner.py @@ -11,10 +11,11 @@ from typing import Any, Callable from . import emit -from .events import EVENT_SCHEMA_ID, SQLiteEventStore, validate_event +from .events import (EventRowDecodeError, EventStoreOperationalError, SQLiteEventStore, + read_event_rows, validate_event) from .paths import resolve_loop_paths -from .reducer import reduce_events -from .runtime import RuntimeStoreError +from .reducer import ChainBreakError, reduce_events +from .runtime import RuntimeStoreError, _read_store class RunnerError(RuntimeError): @@ -98,6 +99,14 @@ def _subprocess_verifier(task: dict[str, Any], workspace: Path) -> VerifyOutcome return VerifyOutcome(proc.returncode == 0, summary=(proc.stdout + proc.stderr)[-2000:]) +def _store_append(store: SQLiteEventStore, *args: Any, **kwargs: Any) -> dict[str, Any]: + """Keep an unusable store (schema drift, lock) inside the typed runtime family.""" + try: + return store.append(*args, **kwargs) + except EventStoreOperationalError as exc: + raise RuntimeStoreError("event_store_unusable", str(exc)) from exc + + def _load_tasks(paths: Any) -> list[dict]: try: raw = json.loads(paths.tasks.read_text(encoding="utf-8")) @@ -110,31 +119,24 @@ def _load_tasks(paths: Any) -> list[dict]: 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.""" + """Read through the shared read path so a dispatch attempt writes nothing and never + mistakes a lost race with a live appender for a corrupt store (design change D4).""" 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 + + def read_stream(conn: sqlite3.Connection) -> tuple[str, list[dict[str, Any]]]: + 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] + return run_id, read_event_rows(conn, run_id) + 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: + run_id, events = _read_store(path, read_stream) + except EventRowDecodeError as exc: raise RuntimeStoreError("corrupt_store", f"cannot read event store: {exc}") from exc for event in events: report = validate_event(event, mode=mode) @@ -142,6 +144,8 @@ def _projection(target: str | Path, mode: str | None) -> tuple[str, dict[str, An raise RuntimeStoreError("invalid_event", f"event store contains invalid event: {report['issues']}") try: return run_id, reduce_events(events) + except ChainBreakError as exc: + raise RuntimeStoreError("event_chain_broken", str(exc)) from exc except ValueError as exc: raise RuntimeStoreError("invalid_event_stream", str(exc)) from exc @@ -220,8 +224,8 @@ def dispatch_once( "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) + _store_append(store, 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} @@ -237,8 +241,8 @@ def dispatch_once( "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) + _store_append(store, 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"], diff --git a/loop/runtime.py b/loop/runtime.py index 37ca33f..05e4613 100644 --- a/loop/runtime.py +++ b/loop/runtime.py @@ -4,14 +4,25 @@ import json import sqlite3 +from contextlib import closing from pathlib import Path -from typing import Any +from typing import Any, Callable, TypeVar from .completion import CompletionPolicyError, criteria_satisfy_completion from .contract import ContractIssue -from .events import EVENT_SCHEMA_ID, EVENT_TYPES, validate_event +from .events import ( + EVENT_SCHEMA_ID, + EVENT_TYPES, + EventRowDecodeError, + has_chain_columns, + read_event_rows, + store_user_version, + validate_event, +) from .paths import resolve_loop_paths -from .reducer import EventReplayError, reduce_events +from .reducer import ChainBreakError, EventReplayError, reduce_events + +_T = TypeVar("_T") class RuntimeStoreError(RuntimeError): @@ -26,51 +37,49 @@ def _store_path(target: str | Path) -> Path: return resolve_loop_paths(target).loop_dir / "events.db" -def _readonly_query(path: Path) -> str: - """Avoid creating sidecars on clean stores while preserving crash-left WAL reads.""" - return "mode=ro" if (path.parent / (path.name + "-wal")).exists() else "mode=ro&immutable=1" +def _read_only_connect(path: Path) -> sqlite3.Connection: + """Read-only connection; immutable when no WAL sidecar exists so reads leave no files. + immutable=1 assumes no concurrent writer. A live append can surface as + SQLITE_CORRUPT, so a failed immutable open is retried once as plain mode=ro + before the caller may conclude corruption: real corruption fails both. + """ + uri = path.absolute().as_uri() + if not path.with_name(path.name + "-wal").exists(): + try: + return sqlite3.connect(f"{uri}?mode=ro&immutable=1", uri=True) + except sqlite3.DatabaseError: + pass + return sqlite3.connect(f"{uri}?mode=ro", uri=True) -def _read_events_readonly(path: Path, run_id: str) -> list[dict[str, Any]]: - """Read the EventStore row shape without invoking its write-capable connector.""" + +def _read_store(path: Path, read: Callable[[sqlite3.Connection], _T]) -> _T: + """Run one read; a lost immutable=1 race retries plainly before counting as corruption.""" try: - conn = sqlite3.connect(f"{path.absolute().as_uri()}?{_readonly_query(path)}", 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() + with closing(_read_only_connect(path)) as conn: + return read(conn) + except sqlite3.DatabaseError: + pass + try: + with closing(sqlite3.connect(f"{path.absolute().as_uri()}?mode=ro", uri=True)) as conn: + return read(conn) except sqlite3.DatabaseError as exc: raise RuntimeStoreError("corrupt_store", f"cannot read event store: {exc}") from exc + + +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: - 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: + return _read_store(path, lambda conn: read_event_rows(conn, run_id)) + except EventRowDecodeError 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()}?{_readonly_query(path)}", 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 + rows = _read_store( + path, lambda conn: conn.execute("SELECT DISTINCT run_id FROM events ORDER BY run_id ASC").fetchall()) if not rows: raise RuntimeStoreError("empty_store", f"event store is empty: {path}") if len(rows) != 1: @@ -88,7 +97,11 @@ def _events(target: str | Path, mode: str | None) -> tuple[Path, str, list[dict[ validation: dict[str, Any] | None = None for event in events: validation = validate_event(event, mode=mode) - assert validation is not None + if not validation["ok"]: + raise RuntimeStoreError("invalid_event", + f"event store contains invalid event: {validation['issues']}") + if validation is None: + raise RuntimeStoreError("empty_store", f"event store is empty: {path}") return path, run_id, events, validation @@ -130,11 +143,16 @@ def status_report(target: str | Path, *, mode: str | None = None) -> dict[str, A """Project a single event stream and reconcile it with live state.json.""" _, run_id, events, validation = _events(target, mode) paths = resolve_loop_paths(target) + degraded = {"state": None, "iteration_id": None, "active_task": None, "terminal": None, + "chain_head": None, "unchained_prefix": 0} try: projection = reduce_events(events) divergence = _state_divergence(paths, projection) + except ChainBreakError as exc: + projection = degraded + divergence = [ContractIssue("event_chain_broken", str(exc))] except EventReplayError as exc: - projection = {"state": None, "iteration_id": None, "active_task": None, "terminal": None} + projection = degraded divergence = [ContractIssue("illegal_event_sequence", str(exc))] return { "ok": not divergence, @@ -143,6 +161,8 @@ def status_report(target: str | Path, *, mode: str | None = None) -> dict[str, A "state": projection["state"], "iteration_id": projection["iteration_id"], "active_task": projection["active_task"], "terminal": projection["terminal"], "completion_satisfied": _completion_satisfied(projection["terminal"]), + "chain_head": projection.get("chain_head"), + "unchained_prefix": projection.get("unchained_prefix", 0), "state_json_agrees": not divergence, "divergence": divergence, } @@ -186,6 +206,9 @@ def replay_report(target: str | Path, *, mode: str | None = None) -> dict[str, A projection = first if not deterministic: findings.append(ContractIssue("nondeterministic_replay", "two event folds produced different projections")) + except ChainBreakError as exc: + legal_sequence = False + findings.append(ContractIssue("event_chain_broken", str(exc))) except EventReplayError as exc: legal_sequence = False findings.append(ContractIssue("illegal_event_sequence", str(exc))) @@ -198,25 +221,66 @@ def replay_report(target: str | Path, *, mode: str | None = None) -> dict[str, A "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, + "chain_head": (projection or {}).get("chain_head"), + "unchained_prefix": (projection or {}).get("unchained_prefix", 0), "terminal_desync": terminal_desync, "findings": findings, } +def _store_declares_chain_without_columns(path: Path) -> bool: + """True when the store still declares generation 2 but its chain columns are gone. + + PRAGMA-only probe, routed through _read_store so a lost immutable=1 race + retries plainly instead of being mistaken for corruption (the D4 hole). + """ + return _read_store(path, lambda conn: store_user_version(conn) >= 2 and not has_chain_columns(conn)) + + +def _anchor_mismatch(message: str) -> dict[str, Any]: + return ContractIssue("chain_anchor_mismatch", message) + + def event_consistency_issues( - target: str | Path, *, mode: str | None = None + target: str | Path, *, mode: str | None = None, expect_chain_head: 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(): + absent_issues: list[dict[str, Any]] = [] + residue = [suffix for suffix in ("-wal", "-shm") + if path.with_name(path.name + suffix).exists()] + if residue: + absent_issues.append(ContractIssue( + "missing_event_store", + "events.db is absent but SQLite sidecar files remain — the store was deleted")) + if expect_chain_head is not None: + absent_issues.append(_anchor_mismatch( + "an anchored chain head was supplied but no event store is present")) + if absent_issues: + return {"present": False, "sidecar_residue": bool(residue)}, absent_issues return {"present": False}, [] try: status = status_report(target, mode=mode) replay = replay_report(target, mode=mode) + declares_chain_without_columns = _store_declares_chain_without_columns(path) except RuntimeStoreError as exc: - return {"present": True, "readable": False, "error_code": exc.code}, [ - ContractIssue(exc.code, str(exc)) - ] + unreadable_issues = [ContractIssue(exc.code, str(exc))] + if expect_chain_head is not None: + unreadable_issues.append(_anchor_mismatch( + "an anchored chain head was supplied but the event store cannot be read")) + return {"present": True, "readable": False, "error_code": exc.code}, unreadable_issues issues = list(status["divergence"]) + list(replay["findings"]) + if declares_chain_without_columns: + issues.append(ContractIssue( + "chain_columns_missing", + "store declares user_version >= 2 but the chain columns are absent — " + "the chain was dropped (this check only catches the lazy downgrade; " + "an anchored head is the real control)")) + if expect_chain_head is not None: + actual = (status["chain_head"] or {}).get("event_hash") + if actual != expect_chain_head: + issues.append(_anchor_mismatch( + f"chain head {actual!r} does not match expected {expect_chain_head!r}")) return { "present": True, "readable": True, @@ -225,4 +289,5 @@ def event_consistency_issues( "state_json_agrees": status["state_json_agrees"], "deterministic": replay["deterministic"], "legal_sequence": replay["legal_sequence"], + "chain": {"head": status["chain_head"], "unchained_prefix": status["unchained_prefix"]}, }, issues diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index f614df7..ba3e186 100644 --- a/reference/repo-os-contract.md +++ b/reference/repo-os-contract.md @@ -586,7 +586,10 @@ the negative tests). event log (ADR 0001). `loop.events.SQLiteEventStore` persists events in a SQLite database in WAL mode with `synchronous=FULL` (every committed `append()` survives a crash) and DB-level `BEFORE UPDATE`/`BEFORE DELETE` triggers that -refuse mutation or removal of a committed row, regardless of caller. +refuse mutation or removal of a committed row **through the store API**, +regardless of caller; a process with direct write access to the database file +can `DROP TRIGGER` — the triggers are an anti-footgun, not a security control +(see Integrity boundary). `loop.reducer.reduce_events()` is a pure, resumable left-fold that projects an ordered event stream into a deterministic state/runlog/receipts view — the same input sequence always produces a byte-identical result. @@ -630,7 +633,180 @@ enforces *domain* semantics at replay time — FSM transition legality `loop.contract`/`loop.emit` already enforce at file-write time, never re-implemented. A store back-end therefore never needs domain awareness to be conformant; the reducer is a second, independent enforcement point that a -tampered or foreign-sourced event stream still cannot talk past. +tampered or foreign-sourced event stream still cannot talk past **without +constructing a stream that is itself FSM-legal, G1-satisfying and +hash-chain-consistent; a determined in-workspace rewriter can construct one — +see Integrity boundary.** + +### Hash chain (v0.10.0+) + +`event@1` carries two **additive, optional** fields: `prev_event_hash` — the +immediately preceding event's `event_hash` within the same run, `null` at +genesis — and `event_hash`, this event's own digest. Both are +`["string", "null"]` constrained to `^[0-9a-f]{64}$`. A fresh v0.10.0 store +declares `PRAGMA user_version = 2`, chains every append, and holds +`event_hash NOT NULL` at the database layer; a pre-existing unchained store is +widened by the explicit `loop migrate` verb (§22), which never rewrites rows. + +**Canonical form (normative).** `event_hash` is the lowercase-hex SHA-256 of +the UTF-8 encoding of + +```python +json.dumps(preimage, sort_keys=True, separators=(",", ":"), + ensure_ascii=False, allow_nan=False) +``` + +where `preimage` is exactly these **twelve** fields: + +`schema`, `run_id`, `sequence`, `event_id`, `type`, `actor`, `ts`, +`causation_id`, `correlation_id`, `payload`, `artifact_hashes`, +`prev_event_hash`. + +Insertion order is irrelevant (`sort_keys=True` fixes the serialized order). +`event_hash` is never part of its own preimage. An **absent optional field is +hashed as `null`, never omitted** — the preimage object always carries all +twelve keys. A genesis event (the run's `sequence` 0) hashes +`prev_event_hash: null`. + +Two caveats for non-Python re-implementations. (1) Floats serialize through +Python's shortest-round-trip `repr`, which other languages' default float +formatting does not necessarily reproduce; keep payload numbers integral or +string-encoded if you need cross-language digests. `allow_nan=False` means +`NaN`/`Infinity` are a hard error, not a serialized token. (2) +`ensure_ascii=False` emits non-ASCII characters literally in UTF-8 and +`sort_keys` orders keys by code point, so **ASCII-only object keys are +recommended** for interop. + +**Conformance vectors.** Three records and their digests, generated by +`loop/chain.py` and pinned against it by +`scripts/test_event_chain.py::test_documented_conformance_vectors` — docs and +code cannot drift. Each `Preimage` line is the exact canonical string that is +UTF-8 encoded and SHA-256'd. + +*Vector 1 — genesis (`prev_event_hash: null`):* + +```json +{"schema":"loop-engineer/event@1","run_id":"run-1","sequence":0,"event_id":"e0","type":"contract_opened","actor":"operator","ts":"2026-07-24T00:00:00+00:00","causation_id":null,"correlation_id":null,"payload":{"workspace":"ws"},"artifact_hashes":[],"prev_event_hash":null} +``` + +Preimage: `{"actor":"operator","artifact_hashes":[],"causation_id":null,"correlation_id":null,"event_id":"e0","payload":{"workspace":"ws"},"prev_event_hash":null,"run_id":"run-1","schema":"loop-engineer/event@1","sequence":0,"ts":"2026-07-24T00:00:00+00:00","type":"contract_opened"}` + +`event_hash` = `3ca65d4da7d87a98616441a86c6866ff39b5513ccd156d8526abfd6df7ec88a7` + +*Vector 2 — second event, linked to vector 1:* + +```json +{"schema":"loop-engineer/event@1","run_id":"run-1","sequence":1,"event_id":"e1","type":"iteration_appended","actor":"operator","ts":"2026-07-24T00:00:01+00:00","causation_id":null,"correlation_id":null,"payload":{"iteration_id":1,"outcome":"task_passed","state":"execute-task"},"artifact_hashes":[],"prev_event_hash":"3ca65d4da7d87a98616441a86c6866ff39b5513ccd156d8526abfd6df7ec88a7"} +``` + +Preimage: `{"actor":"operator","artifact_hashes":[],"causation_id":null,"correlation_id":null,"event_id":"e1","payload":{"iteration_id":1,"outcome":"task_passed","state":"execute-task"},"prev_event_hash":"3ca65d4da7d87a98616441a86c6866ff39b5513ccd156d8526abfd6df7ec88a7","run_id":"run-1","schema":"loop-engineer/event@1","sequence":1,"ts":"2026-07-24T00:00:01+00:00","type":"iteration_appended"}` + +`event_hash` = `bb40984d1b98bda565d93dd90a39ea212be999078a66cf013f37cbed650c155d` + +*Vector 3 — non-ASCII payload (pins `ensure_ascii=False`):* + +```json +{"schema":"loop-engineer/event@1","run_id":"run-1","sequence":2,"event_id":"e2","type":"receipt_appended","actor":"operator","ts":"2026-07-24T00:00:02+00:00","causation_id":null,"correlation_id":null,"payload":{"iteration_id":1,"note":"café — naïve ✅","summary":"日本語"},"artifact_hashes":[],"prev_event_hash":"bb40984d1b98bda565d93dd90a39ea212be999078a66cf013f37cbed650c155d"} +``` + +Preimage: `{"actor":"operator","artifact_hashes":[],"causation_id":null,"correlation_id":null,"event_id":"e2","payload":{"iteration_id":1,"note":"café — naïve ✅","summary":"日本語"},"prev_event_hash":"bb40984d1b98bda565d93dd90a39ea212be999078a66cf013f37cbed650c155d","run_id":"run-1","schema":"loop-engineer/event@1","sequence":2,"ts":"2026-07-24T00:00:02+00:00","type":"receipt_appended"}` + +`event_hash` = `0d0413aa0a1903a46a802f98f0a28abafd10ca09d5e312622f729482cfc40a19` + +**Third-party re-verification.** +`loop.chain.verify_chain(events, expected_head=...)` is the normative entry +point for re-verifying a chain outside this package's store code: it is pure, +I/O-free, imports no other `loop` module, and accepts any ordered sequence of +event mappings (a SQLite read, a JSONL export, a JSON API response). It +returns `{ok, issues, chained_events, unchained_prefix, head}`, and with +`expected_head` set it additionally fails when the stream's final chained head +is absent or differs. **Scope:** `verify_chain` verifies a *complete run +stream beginning at sequence 0*. It cannot validate a suffix or a slice — a +window that starts mid-run has no genesis to anchor `prev_event_hash: null` +against, and its first record's link is unverifiable by construction. + +**Interop rule (normative).** Populating the chain fields is optional per run +but all-or-nothing after the first chained event: once an event carries +`event_hash`, every later event in that run must too and must match the +canonical preimage exactly, or the reference implementation hard-fails the +store. + +**Compatibility rule.** A pre-0.10.0 writer must not append to a chained +store. A fresh v0.10.0 store refuses such an append at the database +(`event_hash NOT NULL`); a migrated store cannot, and an unchained row +appended after a chained prefix is reported as `event_chain_broken` and is +unrepairable, because UPDATE is trigger-blocked. Pin your loop-engineer (and +action) version per store. + +### Integrity boundary + +The chain is **tamper-evident relative to an anchored head**. That is a +detection property, not a prevention one, and it is scoped to the anchor: +nothing here stops a writer from changing the log. Stating the boundary in +both directions is part of the contract — a reader must know exactly which +claims a clean chain supports. + +**It detects:** splicing an event into the middle of a log; reordering events; +editing a committed row without recomputing every downstream digest; byte +corruption of any hashed field; and — given an externally remembered anchor — +*any* divergence of the log from the head that anchor names, including +truncation of the tail and wholesale replacement of the history. Note the +asymmetry: truncation is detected **only** with an anchor, because deleting +trailing events leaves a shorter but internally valid chain. + +**It does not detect:** + +- **A full in-workspace recompute.** A process with write access to the + workspace can rewrite history, re-chain from genesis, and forge + `.loop/state.json` and `terminal_state.json` to agree. With no anchor + supplied, the event-store block of such a report is wholly clean — + `state_json_agrees`, `deterministic`, and `legal_sequence` all `true`, a + `FailedBlocked` run laundered into `Succeeded` — as pinned by + `scripts/test_adversarial_chain.py::test_full_rewrite_with_recompute_passes_without_anchor_pinned`. + (In the probe that produced that fixture the report was also globally `ok` + with zero issues. What the committed pin actually asserts is narrower and + event-store-scoped: that `event_chain_broken` is absent, that the three + projection-disagreement codes `state_field_mismatch`, + `desynced_terminal_window` and `terminal_state_mismatch` stay absent, and + that the three event-store cleanliness flags `state_json_agrees`, + `deterministic` and `legal_sequence` stay `true`. A future standalone + event-store cross-check — a new issue code appended directly to `issues`, as + `chain_columns_missing` is — would not move any of those, so it must be added + to this pin's assertions when it is introduced.) +- **A chain-column downgrade.** Dropping `event_hash`/`prev_event_hash`, or + rebuilding the store without them, silently downgrades a chained history to + an unchained one. An unchained or legacy doctor report is *not* proof of + provenance. The `chain_columns_missing` check catches only the lazy variant + — columns dropped while `user_version` still declares generation 2; a + downgrade that also resets `user_version` leaves nothing but the anchor. +- **Deleting the store outright**, when no SQLite sidecars remain and no + `--expect-chain-head` is supplied: a bare `loop doctor` reads that as a + valid never-ran contract (§22). +- **Well-formed lies.** Nothing in the chain judges whether a payload is + *true*. A truthfully-recorded, correctly-hashed event asserting a test + passed when it did not is chain-clean by construction; that is the job of + evidence@1, the held-out gate, and the verifier — not of the digest. +- **Anything in a never-migrated prefix.** Rows written before migration have + no hashes to break, so there is no retroactive coverage: doctor reports them + as `unchained_prefix` and never elides them. + +**The mid-run window.** An anchor certifies the log only up to the anchored +head. Everything appended after the last externally-read anchor — including a +rewrite of the suffix — is unverified until the next anchor is read and +remembered outside the workspace. The chain narrows the tampering window; it +does not close it. + +Three closing notes. The append-only `BEFORE UPDATE`/`BEFORE DELETE` triggers +are an anti-footgun, not a security control: any writer holding the database +file can `DROP TRIGGER` first. The chain is one of several cross-checks a full +rewrite must satisfy *simultaneously* — `_state_divergence` (state.json +agreement), `_terminal_desync` (terminal-file agreement), and G1 completion +all still apply, which raises the cost of a convincing forgery without +bounding it. And `scripts/test_adversarial_chain.py` pins **both** sides of +this boundary: the attacks that are caught, and four `PINNED LIMITATION` +cases that are not — the full in-workspace rewrite, tail truncation without an +anchor, tampering inside a never-migrated prefix, and the chain-column +downgrade. --- @@ -767,14 +943,101 @@ Plan-Then-Execute (2605.14290), Plan Compliance (2604.12147), and Code as Agent 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 +logic — and folds their findings into its own `issues`/`ok`. An absent store +**with no SQLite sidecar residue and no `--expect-chain-head`** is conformant: +doctor reports `"event_store": {"present": false}` and every other key is +byte-identical to a store-less report; sidecar residue +(`missing_event_store`) or a supplied anchor (`chain_anchor_mismatch`) fails +doctor. When an absent store *does* raise one of those, the block gains the +residue flag — `{"present": false, "sidecar_residue": true}` for a deleted +store whose `-wal`/`-shm` files remain, and `sidecar_residue: false` when the +only finding is the anchor. A present, readable store adds `"event_store": {"present": true, "readable": true, "run_id", "event_count", -"state_json_agrees", "deterministic", "legal_sequence"}`; any of +"state_json_agrees", "deterministic", "legal_sequence", "chain"}`; any of `state_field_mismatch`, `desynced_terminal_window`, `terminal_state_mismatch`, -or `illegal_event_sequence` fails doctor (`ok: false`) with the identical issue +`illegal_event_sequence`, `event_chain_broken`, `chain_columns_missing`, or +`chain_anchor_mismatch` 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 +all — `corrupt_store`, `empty_store`, `invalid_event`, or `ambiguous_run_id` — +also fails doctor rather than being silently skipped; `"event_store"` reports `{"present": true, "readable": false, "error_code": }` in that case. +Note the ordering consequence: event@1 validation runs before the fold, so a +tamper that *also* breaks the envelope or payload shape (a payload edit that +drops a required field, say) surfaces as `invalid_event` on an unreadable +store, never as `event_chain_broken`. + +**The `chain` block.** A present, readable store nests +`"chain": {"head": {"sequence", "event_hash"} | null, "unchained_prefix": }` +under `event_store`. `head` is `null` for a store with no chained events at +all (a legacy or fully downgraded store) **and also for a store whose chain is +broken**, because the fold stops at the break and never establishes a head. The +block alone therefore does not distinguish "never chained" from "chain broken"; +the `event_chain_broken` issue is what separates them, and a `null` head with +no such issue is the honest never-chained case. `unchained_prefix` counts the +leading events that carry no `event_hash`; it is **never elided** — a migrated +store legitimately reports a non-zero prefix, and silently hiding it would let +a legacy tail read as chained provenance. A prefix is not an issue by itself; +it is the honest statement of how much of the log the chain does not cover. + +**New issue codes.** + +| Code | Meaning | +|---|---| +| `event_chain_broken` | A link check failed: `prev_event_hash` mismatch, recomputed `event_hash` mismatch, an unhashable record, or an unchained row appended after a chained prefix. Unrepairable — UPDATE is trigger-blocked. | +| `chain_anchor_mismatch` | `--expect-chain-head` was supplied and the store's actual head is absent, unreadable, unchained, or a different digest. | +| `chain_columns_missing` | The store declares `user_version >= 2` but the chain columns are gone — the lazy downgrade. A downgrade that also resets `user_version` is invisible here; the anchor is the real control. | +| `missing_event_store` | `events.db` is absent but `-wal`/`-shm` sidecars remain — the store was deleted. Distinct from the pre-existing `missing_store`, which `status`/`replay`/`run`/`migrate` raise when a verb that *requires* a store is pointed at a workspace that has none; `missing_event_store` is a doctor finding about a store that evidently once existed. | + +**`loop migrate`.** `loop migrate ` is the only store-upgrade path: +explicit, idempotent, and non-rewriting. It widens `events` with the two +nullable chain columns and stamps `user_version = 2`; it never backfills +hashes onto existing rows, because the append-only triggers forbid UPDATE and +a backfilled digest would be an assertion about history the store cannot +support. Pre-migration rows therefore remain an unchained prefix forever, and +the first post-migration append is a chain genesis +(`chained_from_sequence` in the migrate report names it). Because migrated +columns stay nullable, a migrated store — unlike a fresh one — cannot refuse a +pre-0.10.0 writer at the database layer (see §16's compatibility rule). + +**Concurrency assumption.** The read verbs assume no concurrent writer — the +sidecar-free guarantee holds only for a store whose last writer closed +cleanly. `loop doctor`/`status`/`replay`/`simulate` open the store read-only +and immutable when no WAL sidecar exists, so a clean store is read without +leaving a byte behind; against a store with a live or crashed writer, a +`-wal`/`-shm` sidecar is the sole permitted coordination-file difference (§20) +and durable event content still never changes. + +### Anchor trust assumptions + +`--expect-chain-head` (valid on `doctor`, and identically on its `validate` +and `verify` aliases) fails hard when the store is missing, unreadable, has no +chained head, or has a head that differs from the supplied digest. It is the +only cross-run tamper control in the toolchain — and it is only as trustworthy +as the place the anchor is read from. The anchor is outside the worker's trust +domain when **all three** hold: + +1. **The gate is not installed from the worker's own checkout.** The + `loop-engineer` GitHub Action is pinned to a released tag or SHA of a repo + the worker cannot write, **and** its `version` input is non-empty — with an + empty `version` the `Install loop-engineer` step takes its fallback branch + and installs the kernel from `github.action_path`, i.e. the checkout under + test, so the code doing the verifying is code the worker may have edited. +2. **The invoking workflow is protected from worker edits** — a required + workflow, CODEOWNERS on `.github/`, or a branch ruleset. For a same-repo + pull request the workflow that runs is *the PR head's*, so an unprotected + workflow file is an unprotected anchor. +3. **The expected head is remembered outside the workspace** — a repository + variable, a protected environment, or an operator's own record. A digest + read out of the tree it is meant to certify certifies nothing. + +State it plainly: **with the default empty `expect-chain-head`, this action +performs no cross-run tamper detection; it records the head for a comparison +someone else must make.** That recording is the action's `chain-head` output +and a line in its step summary — the place an operator reads the observed head +in order to re-anchor the next run. When an anchor *is* supplied and does not +match, the mismatch itself surfaces in doctor's issue list as +`chain_anchor_mismatch`. + +Always pass an anchor in CI. A bare `loop doctor` treats a fully deleted +store — no database, no sidecars — as a valid never-ran contract, so +"delete the evidence" is a passing run without one. diff --git a/schemas/event.schema.json b/schemas/event.schema.json index 090b6f7..cd92fd8 100644 --- a/schemas/event.schema.json +++ b/schemas/event.schema.json @@ -2,7 +2,7 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "loop-engineer/event@1", "title": "Loop Engineer Event @1", - "description": "One immutable, append-only fact in a run's event log (ADR 0001). SQLiteEventStore (loop/events.py) persists these; loop.reducer projects an ordered sequence into a deterministic state/runlog/receipts view. Cross-field domain semantics (FSM legality, G1 completion, terminal immutability) are enforced by the reducer at replay time, not by the store at write time. Not yet a loop-doctor-validated workspace artifact — see reference/repo-os-contract.md #16.", + "description": "One immutable, append-only fact in a run's event log (ADR 0001). SQLiteEventStore (loop/events.py) persists these; loop.reducer projects an ordered sequence into a deterministic state/runlog/receipts view. Cross-field domain semantics (FSM legality, G1 completion, terminal immutability) are enforced by the reducer at replay time, not by the store at write time. Not yet a loop-doctor-validated workspace artifact — see reference/repo-os-contract.md #16. prev_event_hash/event_hash (additive, optional) carry the per-run hash chain; loop/chain.py pins the canonical preimage.", "type": "object", "required": ["schema", "event_id", "run_id", "sequence", "type", "actor", "ts", "payload"], "properties": { @@ -16,6 +16,8 @@ "correlation_id": { "type": ["string", "null"] }, "ts": { "type": "string", "minLength": 1 }, "payload": { "type": "object" }, + "prev_event_hash": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$" }, + "event_hash": { "type": ["string", "null"], "pattern": "^[0-9a-f]{64}$" }, "artifact_hashes": { "type": "array", "items": { diff --git a/scripts/chain_fixtures.py b/scripts/chain_fixtures.py new file mode 100644 index 0000000..baa900d --- /dev/null +++ b/scripts/chain_fixtures.py @@ -0,0 +1,66 @@ +"""Shared test fixtures for chain work: byte-faithful v0.9.0 store builders. + +Imported by test_event_chain.py, test_adversarial_chain.py, +test_doctor_eventstore.py and test_loop_simulate_zero_writes.py as +`from chain_fixtures import make_legacy_store` — pytest's prepend import mode +puts scripts/ on sys.path (there is no scripts/__init__.py). +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +LEGACY_DDL = """ +CREATE TABLE events ( + run_id TEXT NOT NULL, sequence INTEGER NOT NULL, event_id TEXT NOT NULL UNIQUE, + type TEXT NOT NULL, actor TEXT NOT NULL, causation_id TEXT, correlation_id TEXT, + ts TEXT NOT NULL, payload TEXT NOT NULL, artifact_hashes TEXT NOT NULL, + PRIMARY KEY (run_id, sequence) +)""" + +LEGACY_TRIGGERS = ( + "CREATE TRIGGER events_no_update BEFORE UPDATE ON events " + "BEGIN SELECT RAISE(ABORT, 'events table is append-only: UPDATE is forbidden'); END", + "CREATE TRIGGER events_no_delete BEFORE DELETE ON events " + "BEGIN SELECT RAISE(ABORT, 'events table is append-only: DELETE is forbidden'); END", +) + + +def make_legacy_store(path: str | Path, *, run_id: str = "r1") -> Path: + """Write a v0.9.0-shaped store holding one contract_opened event.""" + path = Path(path) + conn = sqlite3.connect(str(path)) + try: + conn.execute(LEGACY_DDL) + for trigger in LEGACY_TRIGGERS: + conn.execute(trigger) + conn.execute( + "INSERT INTO events VALUES (?, 0, 'legacy-e0', 'contract_opened', 'operator', " + "NULL, NULL, '2026-07-24T00:00:00+00:00', '{\"workspace\":\"ws\"}', '[]')", + (run_id,)) + conn.commit() + finally: + conn.close() + return path + + +def drop_triggers(path: str | Path) -> None: + """Adversary helper: remove the append-only triggers (they are DDL, not a security control).""" + conn = sqlite3.connect(str(path)) + try: + conn.execute("DROP TRIGGER IF EXISTS events_no_update") + conn.execute("DROP TRIGGER IF EXISTS events_no_delete") + conn.commit() + finally: + conn.close() + + +def restore_triggers(path: str | Path) -> None: + conn = sqlite3.connect(str(path)) + try: + for trigger in LEGACY_TRIGGERS: + conn.execute(trigger.replace("CREATE TRIGGER", "CREATE TRIGGER IF NOT EXISTS")) + conn.commit() + finally: + conn.close() diff --git a/scripts/test_adversarial_chain.py b/scripts/test_adversarial_chain.py new file mode 100644 index 0000000..232d3c4 --- /dev/null +++ b/scripts/test_adversarial_chain.py @@ -0,0 +1,345 @@ +"""Adversarial chain tests: what the chain catches, and — pinned deliberately — +what it does NOT catch without an external anchor. + +If a *_pinned test starts FAILING, the kernel gained a stronger property: update +reference/repo-os-contract.md #16 (Integrity boundary) in the same commit. +""" +import json +import sqlite3 + +import pytest + +from chain_fixtures import drop_triggers, make_legacy_store, restore_triggers +from loop.chain import compute_event_hash +from loop.contract import doctor_report +from loop.events import SQLiteEventStore +from loop.scaffold import scaffold + +_EVENT_SCHEMA_ID = "loop-engineer/event@1" + +# SQLite gained ALTER TABLE ... DROP COLUMN in 3.35; the downgrade attack cannot +# be staged below that, so those tests are skipped rather than errored. +_DROP_COLUMN = pytest.mark.skipif( + sqlite3.sqlite_version_info < (3, 35), + reason="ALTER TABLE ... DROP COLUMN requires SQLite >= 3.35", +) + +# Same shape as the live chain DDL minus the NOT NULL on event_hash. The adversary +# owns the file: a table rebuild is how a row loses its hash in practice (a +# pre-0.10.0 writer appending to a chained store leaves the same footprint). +_NULLABLE_CHAIN_DDL = """ +CREATE TABLE events ( + run_id TEXT NOT NULL, sequence INTEGER NOT NULL, event_id TEXT NOT NULL UNIQUE, + type TEXT NOT NULL, actor TEXT NOT NULL, causation_id TEXT, correlation_id TEXT, + ts TEXT NOT NULL, payload TEXT NOT NULL, artifact_hashes TEXT NOT NULL, + prev_event_hash TEXT, event_hash TEXT, + PRIMARY KEY (run_id, sequence) +)""" + + +def _codes(report): + return {issue["code"] for issue in report["issues"]} + + +def _chain_block(report): + return report["event_store"]["chain"] + + +def _store_path(target): + return target / ".loop" / "events.db" + + +def _sync_state(target, **fields): + """Write projection-agreeing values into state.json so _state_divergence stays quiet.""" + path = target / ".loop" / "state.json" + state = json.loads(path.read_text(encoding="utf-8")) + state.update(fields) + path.write_text(json.dumps(state), encoding="utf-8") + + +def _chained_workspace(tmp_path, name="workspace"): + """A synced workspace over a 4-event chained store (>= 3 events, spliceable middle). + + Deliberately self-contained rather than imported from test_doctor_eventstore: + that module imports loop.migrate/loop.runner at module level, which the + negative-control overlay (chain.py + chain_fixtures.py + this file, on main) + cannot satisfy. + """ + target = tmp_path / name + scaffold(target) + store = SQLiteEventStore(_store_path(target)) + store.append("run-1", "contract_opened", {"workspace": name}, actor="test") + store.append("run-1", "iteration_appended", {"iteration_id": 1, "outcome": "task_failed"}, actor="test") + store.append("run-1", "iteration_appended", {"iteration_id": 2, "outcome": "task_passed"}, actor="test") + store.append("run-1", "receipt_appended", + {"iteration_id": 2, "role": "write", "model": "test-model", "outcome": "ok"}, actor="test") + _sync_state(target, iteration_id=2, active_task=None) + return target + + +def _terminal_workspace(tmp_path, name="workspace"): + """A chained run that honestly ended FailedBlocked, with both projection files in sync.""" + target = tmp_path / name + scaffold(target) + store = SQLiteEventStore(_store_path(target)) + store.append("run-1", "contract_opened", {"workspace": name}, actor="test") + store.append("run-1", "iteration_appended", {"iteration_id": 1, "outcome": "task_failed"}, actor="test") + store.append("run-1", "terminal_written", + {"state": "FailedBlocked", "criteria_met": {"gate": False}, + "evidence": ["red-bundle.json"], "false_completion": False}, actor="test") + _sync_state(target, iteration_id=1, active_task=None, state="terminal", terminal_state="FailedBlocked") + _write_terminal_file(target, "FailedBlocked", {"gate": False}, ["red-bundle.json"]) + return target + + +def _write_terminal_file(target, state, criteria_met, evidence): + (target / ".loop" / "terminal_state.json").write_text(json.dumps({ + "schema": "loop-engineer/terminal@1", "state": state, "criteria_met": criteria_met, + "evidence": evidence, "false_completion": False, + }), encoding="utf-8") + + +def _record_at(conn, sequence, prev_event_hash): + """Rebuild one row into the record dict read_event_rows projects (hash preimage shape).""" + row = conn.execute( + "SELECT run_id, sequence, event_id, type, actor, causation_id, correlation_id, ts, " + "payload, artifact_hashes FROM events WHERE sequence = ?", (sequence,)).fetchone() + 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]), + "prev_event_hash": prev_event_hash} + + +def _head(target): + return _chain_block(doctor_report(target))["head"]["event_hash"] + + +def test_splice_detected(tmp_path): + ws = _chained_workspace(tmp_path) + store_path = ws / ".loop" / "events.db" + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute("UPDATE events SET payload = '{\"iteration_id\":1,\"outcome\":\"task_passed\"}' " + "WHERE sequence = 1") + # recompute ONLY the spliced row's own hash: its successor still cites the original + row = conn.execute("SELECT run_id, sequence, event_id, type, actor, causation_id, " + "correlation_id, ts, payload, artifact_hashes, prev_event_hash " + "FROM events WHERE sequence = 1").fetchone() + record = {"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]), "prev_event_hash": row[10]} + conn.execute("UPDATE events SET event_hash = ? WHERE sequence = 1", + (compute_event_hash(record),)) + conn.commit() + finally: + conn.close() + restore_triggers(store_path) + assert "event_chain_broken" in _codes(doctor_report(ws)) + + +def test_reorder_detected(tmp_path): + """Swapping two same-type rows' payloads leaves both hashes citing the wrong content.""" + ws = _chained_workspace(tmp_path) + store_path = _store_path(ws) + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + first, second = (conn.execute( + "SELECT payload FROM events WHERE sequence = ?", (sequence,)).fetchone()[0] + for sequence in (1, 2)) + conn.execute("UPDATE events SET payload = ? WHERE sequence = 1", (second,)) + conn.execute("UPDATE events SET payload = ? WHERE sequence = 2", (first,)) + conn.commit() + finally: + conn.close() + restore_triggers(store_path) + assert "event_chain_broken" in _codes(doctor_report(ws)) + + +def test_midstream_hash_strip_breaks_chain(tmp_path): + """An unchained row after a chained prefix is a break, not a legacy prefix.""" + ws = _chained_workspace(tmp_path) + store_path = _store_path(ws) + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute("ALTER TABLE events RENAME TO events_old") + conn.execute(_NULLABLE_CHAIN_DDL) + conn.execute("INSERT INTO events SELECT * FROM events_old") + conn.execute("DROP TABLE events_old") + conn.execute("UPDATE events SET event_hash = NULL WHERE sequence = 2") + conn.commit() + finally: + conn.close() + restore_triggers(store_path) + report = doctor_report(ws) + assert "event_chain_broken" in _codes(report) + assert _chain_block(report)["head"] is None + + +def test_full_rewrite_with_recompute_passes_without_anchor_pinned(tmp_path): + """The competent adversary: rewrite history, re-chain from genesis, and forge the + projection files too. The chain alone does NOT catch this — the anchor does.""" + ws = _terminal_workspace(tmp_path) + store_path = _store_path(ws) + original_head = _chain_block(doctor_report(ws))["head"]["event_hash"] + forged_terminal = {"state": "Succeeded", "criteria_met": {"gate": True}, + "evidence": ["forged-bundle.json"], "false_completion": False} + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute("UPDATE events SET payload = replace(payload, '\"task_failed\"', " + "'\"task_passed\"') WHERE type = 'iteration_appended'") + conn.execute("UPDATE events SET payload = ? WHERE type = 'terminal_written'", + (json.dumps(forged_terminal, sort_keys=True),)) + prev = None + for row in conn.execute("SELECT sequence FROM events ORDER BY sequence ASC").fetchall(): + record = _record_at(conn, row[0], prev) + digest = compute_event_hash(record) + conn.execute("UPDATE events SET prev_event_hash = ?, event_hash = ? WHERE sequence = ?", + (prev, digest, row[0])) + prev = digest + conn.commit() + finally: + conn.close() + restore_triggers(store_path) + _sync_state(ws, terminal_state="Succeeded") + _write_terminal_file(ws, "Succeeded", forged_terminal["criteria_met"], forged_terminal["evidence"]) + + unanchored = doctor_report(ws) + assert "event_chain_broken" not in _codes(unanchored) # PINNED LIMITATION + assert _chain_block(unanchored)["head"] is not None + # the forge is complete: no projection check fires either, so nothing but the anchor is + # left. These are the event-store block's own flags, so ANY new event-store-layer + # detection flips the pin — an absence-of-known-codes assertion would not. + assert not _codes(unanchored) & {"state_field_mismatch", "desynced_terminal_window", + "terminal_state_mismatch"} + store_block = unanchored["event_store"] + assert store_block["state_json_agrees"] is True + assert store_block["deterministic"] is True + assert store_block["legal_sequence"] is True + + anchored = doctor_report(ws, expect_chain_head=original_head) + assert "chain_anchor_mismatch" in _codes(anchored) # the anchor is the control + + +def test_truncation_alone_not_detected_but_anchor_catches_it(tmp_path): + """Dropping the trailing receipt leaves a shorter but internally valid chain.""" + ws = _chained_workspace(tmp_path) + store_path = _store_path(ws) + original_head = _head(ws) + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute("DELETE FROM events WHERE type = 'receipt_appended'") + conn.commit() + finally: + conn.close() + restore_triggers(store_path) + + unanchored = doctor_report(ws) + assert "event_chain_broken" not in _codes(unanchored) # PINNED LIMITATION + assert _chain_block(unanchored)["head"] is not None + assert "state_field_mismatch" not in _codes(unanchored) + + anchored = doctor_report(ws, expect_chain_head=original_head) + assert "chain_anchor_mismatch" in _codes(anchored) # the anchor is the control + + +def test_legacy_store_tamper_is_undetectable_pinned(tmp_path): + """A never-migrated store has no hashes to break: there is no retroactive coverage.""" + target = tmp_path / "workspace" + scaffold(target) + store_path = _store_path(target) + make_legacy_store(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute( + "INSERT INTO events VALUES ('r1', 1, 'legacy-e1', 'iteration_appended', 'operator', " + "NULL, NULL, '2026-07-24T00:00:01+00:00', " + "'{\"iteration_id\": 1, \"outcome\": \"task_failed\", \"summary\": \"gate red\"}', '[]')") + conn.commit() + finally: + conn.close() + _sync_state(target, iteration_id=1, active_task=None) + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute("UPDATE events SET payload = replace(payload, 'gate red', 'gate green') " + "WHERE sequence = 1") + conn.commit() + tampered = conn.execute("SELECT payload FROM events WHERE sequence = 1").fetchone()[0] + finally: + conn.close() + restore_triggers(store_path) + # staging self-guard: a non-detection claim is worthless if the tamper never landed + assert "gate green" in tampered + + report = doctor_report(target) + assert "event_chain_broken" not in _codes(report) # PINNED LIMITATION + assert _chain_block(report) == {"head": None, "unchained_prefix": 2} + + +def test_unhashable_record_breaks_chain(tmp_path): + """A row whose payload json.loads accepts but canonical_json refuses (bare NaN). + + Pins link_issue's "unhashable record" branch end-to-end: the recompute raises + before it can compare, and doctor must report a broken chain rather than + propagate a ChainHashError. The required payload fields are kept intact on + purpose — a payload that also violates event@1 is refused by validate_event + before the fold and surfaces as invalid_event instead (§22). + """ + ws = _chained_workspace(tmp_path) + store_path = _store_path(ws) + drop_triggers(store_path) + conn = sqlite3.connect(str(store_path)) + try: + conn.execute("UPDATE events SET payload = ? WHERE sequence = 1", + ('{"iteration_id": 1, "outcome": "task_failed", "x": NaN}',)) + conn.commit() + finally: + conn.close() + restore_triggers(store_path) + + report = doctor_report(ws) # must not raise ChainHashError + assert "event_chain_broken" in _codes(report) + assert any("unhashable record" in issue["message"] for issue in report["issues"] + if issue["code"] == "event_chain_broken") + assert _chain_block(report)["head"] is None + + +def test_never_chained_store_with_anchor_fails(tmp_path): + """An anchor over a store that never chained cannot match: fail hard, never skip.""" + target = tmp_path / "workspace" + scaffold(target) + make_legacy_store(_store_path(target)) + _sync_state(target, active_task=None) + report = doctor_report(target, expect_chain_head="a" * 64) + assert not report["ok"] + assert "chain_anchor_mismatch" in _codes(report) + + +@_DROP_COLUMN +def test_column_drop_downgrade_is_silent_without_anchor_pinned(tmp_path): + """Dropping the columns AND the user_version defeats the D2 cross-check; only the + anchor survives a full downgrade.""" + ws = _chained_workspace(tmp_path) + original_head = _head(ws) + conn = sqlite3.connect(str(_store_path(ws))) + try: + conn.execute("ALTER TABLE events DROP COLUMN event_hash") + conn.execute("ALTER TABLE events DROP COLUMN prev_event_hash") + conn.execute("PRAGMA user_version = 0") + conn.commit() + finally: + conn.close() + + unanchored = doctor_report(ws) + assert "event_chain_broken" not in _codes(unanchored) # PINNED LIMITATION + assert "chain_columns_missing" not in _codes(unanchored) + assert _chain_block(unanchored)["head"] is None + + anchored = doctor_report(ws, expect_chain_head=original_head) + assert "chain_anchor_mismatch" in _codes(anchored) # the anchor is the control diff --git a/scripts/test_doctor_eventstore.py b/scripts/test_doctor_eventstore.py index 4d77f65..640c844 100644 --- a/scripts/test_doctor_eventstore.py +++ b/scripts/test_doctor_eventstore.py @@ -1,11 +1,17 @@ """Doctor integration tests for the read-only EventStore consistency gate.""" import json +import sqlite3 import pytest +from chain_fixtures import drop_triggers, make_legacy_store +from loop.__main__ import main from loop.contract import doctor_report, validate_contract from loop.events import SQLiteEventStore +from loop.migrate import migrate_store +from loop.runner import dispatch_once +from loop.runtime import RuntimeStoreError, replay_report, status_report from loop.scaffold import scaffold @@ -22,6 +28,13 @@ def _sync_active_task(target): path.write_text(json.dumps(state), encoding="utf-8") +def _sync_iteration(target, iteration_id): + path = target / ".loop" / "state.json" + state = json.loads(path.read_text(encoding="utf-8")) + state["iteration_id"] = iteration_id + path.write_text(json.dumps(state), encoding="utf-8") + + def _store(target): return SQLiteEventStore(target / ".loop" / "events.db") @@ -59,6 +72,38 @@ def _terminal_file(state, *, evidence): }) +def _store_path(target): + return target / ".loop" / "events.db" + + +def _chained_workspace(tmp_path, name="workspace"): + """The synced happy-path workspace, whose store is a fresh (chained) generation.""" + target = _fresh_contract(tmp_path, name) + _sync_active_task(target) + _open(_store(target)) + return target + + +def _legacy_workspace(tmp_path, name="workspace"): + """Same contract, but its store is a byte-faithful v0.9.0 unchained file.""" + target = _fresh_contract(tmp_path, name) + _sync_active_task(target) + make_legacy_store(_store_path(target)) + return target + + +def _tamper_payload(target, payload): + """Rewrite a stored payload the way an in-workspace adversary would.""" + path = _store_path(target) + drop_triggers(path) + conn = sqlite3.connect(str(path)) + try: + conn.execute("UPDATE events SET payload = ? WHERE sequence = 0", (payload,)) + conn.commit() + finally: + conn.close() + + def test_absent_event_store_matches_pre_slice_doctor_shape(tmp_path): target = _fresh_contract(tmp_path) file_only = validate_contract(target) @@ -83,6 +128,8 @@ def test_synced_happy_path_is_doctor_clean(tmp_path, monkeypatch, mode): assert report["event_store"]["state_json_agrees"] is True assert report["event_store"]["deterministic"] is True assert report["event_store"]["legal_sequence"] is True + assert report["event_store"]["chain"]["head"]["sequence"] == 0 + assert report["event_store"]["chain"]["unchained_prefix"] == 0 @pytest.mark.parametrize("mode", ["jsonschema", "structural-fallback"]) @@ -166,6 +213,202 @@ def test_ambiguous_run_id_fails_doctor(tmp_path): assert "ambiguous_run_id" in _codes(report) +def test_status_and_replay_expose_chain_head(tmp_path): + ws = _chained_workspace(tmp_path) + report = status_report(ws) + assert report["chain_head"] is not None + assert replay_report(ws)["chain_head"] == report["chain_head"] + + +def test_doctor_nests_chain_under_event_store(tmp_path): + ws = _chained_workspace(tmp_path) + report = doctor_report(ws) + assert report["ok"] is True, report["issues"] + assert report["event_store"]["chain"]["head"]["sequence"] >= 0 + assert report["event_store"]["chain"]["unchained_prefix"] == 0 + + +def test_legacy_store_doctor_ok_and_chain_null(tmp_path): + ws = _legacy_workspace(tmp_path) + report = doctor_report(ws) + assert report["ok"], report["issues"] + assert report["event_store"]["chain"] == {"head": None, "unchained_prefix": 1} + + +def test_migrated_store_doctor_reports_unchained_prefix(tmp_path): + ws = _legacy_workspace(tmp_path) + migrate_store(ws) + report = doctor_report(ws) + assert report["ok"], report["issues"] + assert report["event_store"]["chain"]["head"] is None + assert report["event_store"]["chain"]["unchained_prefix"] == 1 + + +def test_migrated_store_after_append_reports_genesis_head(tmp_path): + ws = _legacy_workspace(tmp_path) + migrate_store(ws) + SQLiteEventStore(_store_path(ws)).append( + "r1", "iteration_appended", {"iteration_id": 1, "outcome": "task_passed"}, actor="test") + _sync_iteration(ws, 1) + chain_block = doctor_report(ws)["event_store"]["chain"] + assert chain_block["head"]["sequence"] == 1 and chain_block["unchained_prefix"] == 1 + + +def test_tampered_store_fails_doctor_status_and_replay_with_event_chain_broken(tmp_path): + ws = _chained_workspace(tmp_path) + _tamper_payload(ws, '{"workspace":"tampered"}') + for report in (doctor_report(ws), status_report(ws), replay_report(ws)): + codes = {issue["code"] for issue in + report.get("issues", report.get("divergence", []) + report.get("findings", []))} + assert "event_chain_broken" in codes + + +def test_run_on_tampered_store_reports_event_chain_broken(tmp_path): + """Design change D3: runner must not relabel it invalid_event_stream.""" + ws = _chained_workspace(tmp_path) + _tamper_payload(ws, '{"workspace":"tampered"}') + with pytest.raises(RuntimeStoreError) as excinfo: + dispatch_once(ws) + assert excinfo.value.code == "event_chain_broken" + + +def test_invalid_event_now_fails_status_instead_of_being_discarded(tmp_path): + ws = _legacy_workspace(tmp_path) + conn = sqlite3.connect(str(_store_path(ws))) + try: + conn.execute( + "INSERT INTO events VALUES ('r1', 1, 'legacy-e1', 'iteration_appended', 'operator', " + "NULL, NULL, '2026-07-24T00:00:01+00:00', '{\"iteration_id\": 1}', '[]')") + conn.commit() + finally: + conn.close() + with pytest.raises(RuntimeStoreError) as excinfo: + status_report(ws) + assert excinfo.value.code == "invalid_event" + + +def test_in_row_json_corruption_fails_doctor_without_traceback(tmp_path): + """Design change D5: read_event_rows owns the decode translation.""" + ws = _chained_workspace(tmp_path) + _tamper_payload(ws, "not json") + report = doctor_report(ws) + assert not report["ok"] and report["event_store"]["error_code"] == "corrupt_store" + + +def test_run_on_in_row_json_corruption_reports_corrupt_store(tmp_path): + """Without runner's EventRowDecodeError clause dispatch_once leaks a bare ValueError.""" + ws = _chained_workspace(tmp_path) + _tamper_payload(ws, "not json") + with pytest.raises(RuntimeStoreError) as excinfo: + dispatch_once(ws) + assert excinfo.value.code == "corrupt_store" + + +def _head_hash(target): + return doctor_report(target)["event_store"]["chain"]["head"]["event_hash"] + + +def test_expect_chain_head_matching_passes(tmp_path): + ws = _chained_workspace(tmp_path) + report = doctor_report(ws, expect_chain_head=_head_hash(ws)) + assert report["ok"], report["issues"] + + +def test_expect_chain_head_mismatch_fails_doctor(tmp_path): + ws = _chained_workspace(tmp_path) + report = doctor_report(ws, expect_chain_head="a" * 64) + assert not report["ok"] + assert "chain_anchor_mismatch" in _codes(report) + + +def test_expect_chain_head_with_missing_store_fails_doctor(tmp_path): + target = _fresh_contract(tmp_path) + report = doctor_report(target, expect_chain_head="a" * 64) + assert not report["ok"] + assert "chain_anchor_mismatch" in _codes(report) + + +def test_expect_chain_head_with_unreadable_store_fails_doctor(tmp_path): + ws = _chained_workspace(tmp_path) + _store_path(ws).write_text("not sqlite", encoding="utf-8") + report = doctor_report(ws, expect_chain_head="a" * 64) + assert not report["ok"] + assert {"corrupt_store", "chain_anchor_mismatch"} <= _codes(report) + + +def test_expect_chain_head_on_tampered_store_fails_doctor(tmp_path): + """A broken chain degrades chain_head to None, so no anchor can ever match it.""" + ws = _chained_workspace(tmp_path) + head = _head_hash(ws) + _tamper_payload(ws, '{"workspace":"tampered"}') + report = doctor_report(ws, expect_chain_head=head) + assert not report["ok"] + assert {"event_chain_broken", "chain_anchor_mismatch"} <= _codes(report) + + +def test_sidecar_residue_without_db_fails_doctor(tmp_path): + ws = _chained_workspace(tmp_path) + _store_path(ws).unlink() + (ws / ".loop" / "events.db-wal").write_bytes(b"") + report = doctor_report(ws) + assert not report["ok"] + assert "missing_event_store" in _codes(report) + assert report["event_store"] == {"present": False, "sidecar_residue": True} + + +@pytest.mark.skipif(sqlite3.sqlite_version_info < (3, 35), + reason="ALTER TABLE ... DROP COLUMN requires SQLite >= 3.35") +def test_chain_columns_dropped_but_version_2_fails_doctor(tmp_path): + """Design change D2: the lazy downgrade attack.""" + ws = _chained_workspace(tmp_path) + conn = sqlite3.connect(str(_store_path(ws))) + try: + conn.execute("ALTER TABLE events DROP COLUMN event_hash") + conn.commit() + finally: + conn.close() + report = doctor_report(ws) + assert not report["ok"] + assert "chain_columns_missing" in _codes(report) + + +def test_absent_store_without_flag_or_sidecars_stays_byte_stable(tmp_path): + target = _fresh_contract(tmp_path) + assert doctor_report(target)["event_store"] == {"present": False} + + +def test_cli_doctor_accepts_flag_before_target(tmp_path): + """The action.yml invocation shape: flag BEFORE the positional target.""" + ws = _chained_workspace(tmp_path) + head = _head_hash(ws) + assert main(["doctor", "--expect-chain-head", head, str(ws)]) == 0 + assert main(["doctor", "--expect-chain-head", "a" * 64, str(ws)]) == 1 + + +def test_cli_rejects_flag_on_other_commands_and_creates_nothing(tmp_path): + ws = _chained_workspace(tmp_path) + target = tmp_path / "fresh" + assert main(["scaffold", "--expect-chain-head", "a" * 64, str(target)]) == 2 + assert not target.exists() + assert main(["status", "--expect-chain-head", "a" * 64, str(ws)]) == 2 + + +def test_cli_rejects_malformed_anchor_value(tmp_path, capsys): + ws = _chained_workspace(tmp_path) + assert main(["doctor", "--expect-chain-head", "nothex", str(ws)]) == 2 + # the flag must be REJECTED, not swallowed as a positional target + assert "must be a 64-character lowercase hex sha256" in capsys.readouterr().err + + +def test_read_verbs_leave_no_wal_sidecars_on_clean_store(tmp_path): + ws = _chained_workspace(tmp_path) + status_report(ws) + replay_report(ws) + doctor_report(ws) + assert not (ws / ".loop" / "events.db-wal").exists() + assert not (ws / ".loop" / "events.db-shm").exists() + + def test_doctor_event_store_reads_do_not_leave_wal_or_shm_sidecars(tmp_path): target = _fresh_contract(tmp_path) _sync_active_task(target) diff --git a/scripts/test_event_chain.py b/scripts/test_event_chain.py new file mode 100644 index 0000000..1f9aa2b --- /dev/null +++ b/scripts/test_event_chain.py @@ -0,0 +1,487 @@ +"""scripts/test_event_chain.py — chain canonicalization, store chaining, migration.""" +import pytest + +from loop.chain import _PREIMAGE_FIELDS, ChainHashError, canonical_json, compute_event_hash + + +def _record(**overrides): + base = { + "schema": "loop-engineer/event@1", "event_id": "e1", "run_id": "r1", + "sequence": 0, "type": "contract_opened", "actor": "operator", + "causation_id": None, "correlation_id": None, "ts": "2026-07-24T00:00:00+00:00", + "payload": {"workspace": "ws"}, "artifact_hashes": [], "prev_event_hash": None, + } + base.update(overrides) + return base + + +def test_canonical_json_is_compact_sorted_utf8(): + assert canonical_json({"b": 1, "a": [1, "é"]}) == '{"a":[1,"é"],"b":1}' + + +def test_canonical_json_rejects_non_finite_floats(): + with pytest.raises(ChainHashError): + canonical_json({"x": float("nan")}) + + +def test_canonical_json_rejects_lone_surrogates(): + with pytest.raises(ChainHashError): + canonical_json({"x": "\ud800"}) + + +def test_canonical_json_rejects_non_json_values(): + with pytest.raises(ChainHashError): + canonical_json({"x": object()}) + + +def test_event_hash_is_stable_and_key_order_independent(): + a = _record() + b = dict(reversed(list(_record().items()))) + assert compute_event_hash(a) == compute_event_hash(b) + assert len(compute_event_hash(a)) == 64 + + +def test_event_hash_excludes_event_hash_but_includes_prev_and_ts_and_actor(): + base = _record() + with_own_hash = dict(base, event_hash="f" * 64) + assert compute_event_hash(base) == compute_event_hash(with_own_hash) + assert compute_event_hash(base) != compute_event_hash(dict(base, prev_event_hash="a" * 64)) + assert compute_event_hash(base) != compute_event_hash(dict(base, ts="2026-07-25T00:00:00+00:00")) + assert compute_event_hash(base) != compute_event_hash(dict(base, actor="worker")) + + +def test_event_hash_treats_absent_optionals_as_null(): + explicit = _record() + implicit = {k: v for k, v in _record().items() + if k not in ("causation_id", "correlation_id", "prev_event_hash")} + assert compute_event_hash(explicit) == compute_event_hash(implicit) + + +from loop.chain import link_issue, verify_chain + + +def _chained(seq, prev_hash, **overrides): + rec = _record(sequence=seq, event_id=f"e{seq}", prev_event_hash=prev_hash, + type="iteration_appended" if seq else "contract_opened", + payload={"iteration_id": seq, "outcome": "task_passed"} if seq else {"workspace": "ws"}) + rec.update(overrides) + rec["event_hash"] = compute_event_hash(rec) + return rec + + +def test_link_issue_genesis_requires_null_prev(): + assert link_issue(_chained(0, None), None) is None + assert "prev_event_hash mismatch" in link_issue(_chained(0, "a" * 64), None) + + +def test_link_issue_detects_recompute_mismatch(): + rec = _chained(0, None) + rec["payload"] = {"workspace": "tampered"} + assert "event_hash mismatch" in link_issue(rec, None) + + +def test_link_issue_unchained_after_chained_is_a_break_and_names_the_likely_cause(): + head = {"sequence": 0, "event_hash": "b" * 64} + unchained = _record(sequence=1, event_id="e1") + message = link_issue(unchained, head) + assert "unchained event after chained prefix" in message + assert "pre-0.10.0 writer" in message # self-diagnosing per design change D1 + assert link_issue(unchained, None) is None + + +def test_verify_chain_happy_path_and_head(): + e0 = _chained(0, None) + e1 = _chained(1, e0["event_hash"]) + report = verify_chain([e0, e1]) + assert report["ok"] and report["chained_events"] == 2 and report["unchained_prefix"] == 0 + assert report["head"] == {"sequence": 1, "event_hash": e1["event_hash"]} + + +def test_verify_chain_legacy_prefix_then_genesis(): + legacy = _record(sequence=0) # no event_hash key at all + e1 = _chained(1, None) # genesis after unchained prefix + report = verify_chain([legacy, e1]) + assert report["ok"] and report["unchained_prefix"] == 1 and report["chained_events"] == 1 + + +def test_verify_chain_detects_splice(): + e0 = _chained(0, None) + e1 = _chained(1, e0["event_hash"]) + forged = dict(e1, payload={"iteration_id": 1, "outcome": "task_failed"}) + forged["event_hash"] = compute_event_hash(forged) # recomputed own hash... + e2 = _chained(2, e1["event_hash"]) # ...but successor cites the original + report = verify_chain([e0, forged, e2]) + assert not report["ok"] and any("prev_event_hash mismatch" in i for i in report["issues"]) + + +def test_verify_chain_reports_first_record_failure_without_counting_it(): + bad = _chained(0, "a" * 64) # bad genesis + report = verify_chain([bad]) + assert not report["ok"] and report["chained_events"] == 0 + assert report["unchained_prefix"] == 0 and report["head"] is None + + +def test_verify_chain_truncation_needs_expected_head(): + e0 = _chained(0, None) + e1 = _chained(1, e0["event_hash"]) + assert verify_chain([e0])["ok"] # honest limit: shorter valid chain verifies + report = verify_chain([e0], expected_head=e1["event_hash"]) + assert not report["ok"] and any("chain head" in i for i in report["issues"]) + + +def test_verify_chain_reports_missing_head_when_anchor_supplied_on_unchained_stream(): + report = verify_chain([_record(sequence=0)], expected_head="a" * 64) + assert not report["ok"] and any("no chained events" in i for i in report["issues"]) + + +import sqlite3 + +from chain_fixtures import make_legacy_store +from loop.events import SQLiteEventStore, has_chain_columns, read_event_rows, store_user_version + + +def test_fresh_store_has_chain_columns_and_user_version_2(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + conn = sqlite3.connect(str(tmp_path / "events.db")) + try: + assert has_chain_columns(conn) and store_user_version(conn) == 2 + notnull = {row[1]: row[3] for row in conn.execute("PRAGMA table_info(events)")} + assert notnull["event_hash"] == 1 and notnull["prev_event_hash"] == 0 + finally: + conn.close() + + +def test_legacy_store_is_not_upgraded_by_connect(tmp_path): + path = make_legacy_store(tmp_path / "events.db") + SQLiteEventStore(path).read("r1") # any connect on a legacy store + conn = sqlite3.connect(str(path)) + try: + assert not has_chain_columns(conn) and store_user_version(conn) == 0 + finally: + conn.close() + + +def test_read_event_rows_projects_hash_keys_on_legacy_store(tmp_path): + path = make_legacy_store(tmp_path / "events.db") + conn = sqlite3.connect(str(path)) + try: + rows = read_event_rows(conn, "r1") + finally: + conn.close() + assert rows[0]["prev_event_hash"] is None and rows[0]["event_hash"] is None + + +def test_read_event_rows_raises_typed_error_on_corrupt_payload_json(tmp_path): + from loop.events import EventRowDecodeError + path = make_legacy_store(tmp_path / "events.db") + conn = sqlite3.connect(str(path)) + try: + conn.execute("DROP TRIGGER events_no_update") + conn.execute("UPDATE events SET payload = 'not json' WHERE sequence = 0") + conn.commit() + with pytest.raises(EventRowDecodeError): + read_event_rows(conn, "r1") + finally: + conn.close() + + +from loop.chain import compute_event_hash as _hash +from loop.events import EventStoreOperationalError + + +def test_read_projects_store_computed_hash_on_fresh_store(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + record = store.append("r2", "contract_opened", {"workspace": "ws"}, actor="operator") + assert store.read("r2")[0]["event_hash"] == record["event_hash"] + + +def test_append_chains_on_fresh_store(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + e0 = store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + e1 = store.append("r1", "iteration_appended", {"iteration_id": 1, "outcome": "task_passed"}, + actor="operator") + assert e0["prev_event_hash"] is None and e0["event_hash"] == _hash(e0) + assert e1["prev_event_hash"] == e0["event_hash"] and e1["event_hash"] == _hash(e1) + + +def test_append_ignores_caller_supplied_chain_fields(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + smuggled = store.append("r1", "iteration_appended", + {"iteration_id": 1, "outcome": "task_passed", "event_hash": "f" * 64}, + actor="operator") + assert smuggled["event_hash"] != "f" * 64 and smuggled["event_hash"] == _hash(smuggled) + + +def test_append_on_legacy_store_stays_unchained_and_working(tmp_path): + path = make_legacy_store(tmp_path / "events.db") + record = SQLiteEventStore(path).append( + "r1", "iteration_appended", {"iteration_id": 1, "outcome": "task_passed"}, actor="operator") + assert record["prev_event_hash"] is None and record["event_hash"] is None + assert SQLiteEventStore(path).read("r1")[1]["event_hash"] is None + + +def test_legacy_style_ten_column_insert_is_refused_by_a_fresh_store(tmp_path): + """Design change D1: a pre-0.10.0 writer cannot silently unchain a v2 store.""" + path = tmp_path / "events.db" + SQLiteEventStore(path).append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + conn = sqlite3.connect(str(path)) + try: + with pytest.raises(sqlite3.IntegrityError): + conn.execute( + "INSERT INTO events (run_id, sequence, event_id, type, actor, causation_id, " + "correlation_id, ts, payload, artifact_hashes) VALUES " + "('r1',1,'old-writer','iteration_appended','worker',NULL,NULL," + "'2026-07-24T00:00:00+00:00','{\"iteration_id\":1,\"outcome\":\"task_passed\"}','[]')") + finally: + conn.close() + + +def test_append_wraps_schema_drift_as_typed_error(tmp_path): + path = tmp_path / "events.db" + conn = sqlite3.connect(str(path)) + conn.execute("CREATE TABLE events (run_id TEXT, sequence INTEGER)") # wrong shape entirely + conn.commit(); conn.close() + with pytest.raises(EventStoreOperationalError): + SQLiteEventStore(path).append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + + +from pathlib import Path + +from loop.runtime import RuntimeStoreError + +_ROOT = Path(__file__).resolve().parent.parent + + +def _drifted_store(path): + """An events table that reads and writes nothing the kernel expects.""" + conn = sqlite3.connect(str(path)) + try: + conn.execute("CREATE TABLE events (run_id TEXT, sequence INTEGER)") + conn.execute("INSERT INTO events VALUES ('r1', 0)") + conn.commit() + finally: + conn.close() + return Path(path) + + +def test_runcontrol_append_translates_operational_error_to_typed_store_error(tmp_path): + from loop import runcontrol + + workspace = tmp_path / "workspace" + (workspace / ".loop").mkdir(parents=True) + _drifted_store(workspace / ".loop" / "events.db") + with pytest.raises(RuntimeStoreError, match="event_store_unusable"): + runcontrol._append_event(workspace, "r1", {"last_sequence": 0}, "contract_opened", + {"workspace": "ws"}) + + +def test_runner_append_translates_operational_error_to_typed_store_error(tmp_path): + from loop.runner import _store_append + + path = _drifted_store(tmp_path / "events.db") + with pytest.raises(RuntimeStoreError, match="event_store_unusable"): + _store_append(SQLiteEventStore(path), "r1", "contract_opened", {"workspace": "ws"}, + actor="loop.run") + + +def test_runner_read_path_retries_plain_mode_ro_before_declaring_corruption(tmp_path, monkeypatch): + """Design change D4 on the run/simulate surface: one lost race is not corruption.""" + from loop import runner + + workspace = tmp_path / "workspace" + (workspace / ".loop").mkdir(parents=True) + _drifted_store(workspace / ".loop" / "events.db") + seen = [] + real_connect = sqlite3.connect + + def record(*args, **kwargs): + seen.append(args[0]) + return real_connect(*args, **kwargs) + + monkeypatch.setattr(sqlite3, "connect", record) + with pytest.raises(RuntimeStoreError) as excinfo: + runner.dispatch_once(workspace) + assert excinfo.value.code == "corrupt_store" # real corruption fails BOTH attempts + assert len(seen) == 2 and all("mode=ro" in uri for uri in seen) + assert "immutable=1" in seen[0] and "immutable=1" not in seen[1] + + +from loop.migrate import migrate_store + + +def _workspace_with_legacy_store(tmp_path): + loop_dir = tmp_path / ".loop" + loop_dir.mkdir() + make_legacy_store(loop_dir / "events.db") + return tmp_path + + +def test_migrate_adds_columns_sets_version_and_reports_unchained(tmp_path): + ws = _workspace_with_legacy_store(tmp_path) + report = migrate_store(ws) + assert report["ok"] and report["migrated"] is True + assert report["user_version"] == 2 and report["unchained_rows"] == 1 + assert report["chained_from_sequence"] == 1 + conn = sqlite3.connect(str(ws / ".loop" / "events.db")) + try: + assert has_chain_columns(conn) and store_user_version(conn) == 2 + finally: + conn.close() + + +def test_migrate_is_idempotent(tmp_path): + ws = _workspace_with_legacy_store(tmp_path) + migrate_store(ws) + assert migrate_store(ws)["migrated"] is False + + +def test_migrate_missing_store_raises_typed(tmp_path): + (tmp_path / ".loop").mkdir() + with pytest.raises(RuntimeStoreError): + migrate_store(tmp_path) + + +def test_post_migration_appends_chain_with_genesis_after_legacy_prefix(tmp_path): + ws = _workspace_with_legacy_store(tmp_path) + migrate_store(ws) + record = SQLiteEventStore(ws / ".loop" / "events.db").append( + "r1", "iteration_appended", {"iteration_id": 1, "outcome": "task_passed"}, actor="operator") + assert record["prev_event_hash"] is None # genesis after unchained prefix + assert record["event_hash"] == _hash(record) + + +from loop.events import validate_event + + +@pytest.mark.parametrize("mode", ["strict", "basic"]) +def test_chain_fields_validate_in_both_modes(mode): + if mode == "strict": + pytest.importorskip("jsonschema") + good = _chained(0, None) + assert validate_event(good, mode=mode)["ok"] + report = validate_event(dict(good, event_hash="not-hex"), mode=mode) + assert not report["ok"] + assert validate_event(dict(good, prev_event_hash=17), mode=mode)["ok"] is False + + +from loop.chain import verify_chain +from loop.reducer import ChainBreakError, reduce_events + + +def test_reducer_folds_chained_stream_and_exposes_head(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + last = store.append("r1", "iteration_appended", + {"iteration_id": 1, "outcome": "task_passed"}, actor="operator") + projection = reduce_events(store.read("r1")) + assert projection["chain_head"] == {"sequence": 1, "event_hash": last["event_hash"]} + assert projection["unchained_prefix"] == 0 + + +def test_reducer_raises_chain_break_on_tampered_payload(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + events = store.read("r1") + events[0]["payload"] = {"workspace": "tampered"} + with pytest.raises(ChainBreakError): + reduce_events(events) + + +def test_reducer_accepts_legacy_unchained_stream(tmp_path): + make_legacy_store(tmp_path / "events.db") + projection = reduce_events(SQLiteEventStore(tmp_path / "events.db").read("r1")) + assert projection["chain_head"] is None and projection["unchained_prefix"] == 1 + + +def test_reducer_resume_from_initial_chain_head(tmp_path): + store = SQLiteEventStore(tmp_path / "events.db") + store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + store.append("r1", "iteration_appended", {"iteration_id": 1, "outcome": "task_passed"}, actor="operator") + events = store.read("r1") + snapshot = reduce_events(events[:1]) + resumed = reduce_events(events[1:], initial=snapshot) + assert resumed["chain_head"] == reduce_events(events)["chain_head"] + forged = dict(events[1], prev_event_hash="a" * 64) + forged["event_hash"] = compute_event_hash(forged) + with pytest.raises(ChainBreakError): + reduce_events([forged], initial=snapshot) + + +@pytest.mark.parametrize("generation", ["fresh", "legacy", "migrated"]) +def test_verify_chain_agrees_with_reducer(tmp_path, generation): + """Two verifiers, one truth — guards against lockstep drift (design decision 8).""" + path = tmp_path / "events.db" + if generation == "fresh": + store = SQLiteEventStore(path) + else: + make_legacy_store(path) + if generation == "migrated": + (tmp_path / ".loop").mkdir(exist_ok=True) + # migrate_store takes a workspace; migrate this file in place via the same DDL + conn = sqlite3.connect(str(path)) + conn.execute("ALTER TABLE events ADD COLUMN prev_event_hash TEXT") + conn.execute("ALTER TABLE events ADD COLUMN event_hash TEXT") + conn.execute("PRAGMA user_version = 2") + conn.commit(); conn.close() + store = SQLiteEventStore(path) + if generation == "fresh": + store.append("r1", "contract_opened", {"workspace": "ws"}, actor="operator") + store.append("r1", "iteration_appended", {"iteration_id": 1, "outcome": "task_passed"}, + actor="operator") + events = store.read("r1") + projection = reduce_events(events) + report = verify_chain(events) + assert report["head"] == projection["chain_head"] + assert report["unchained_prefix"] == projection["unchained_prefix"] + + +# --- documented conformance vectors (reference/repo-os-contract.md #16) ------- +# Literal records + literal digests. The digests are also asserted to appear in +# the contract document, so the spec and the implementation cannot drift apart. + +_VECTOR_GENESIS = { + "schema": "loop-engineer/event@1", "run_id": "run-1", "sequence": 0, + "event_id": "e0", "type": "contract_opened", "actor": "operator", + "ts": "2026-07-24T00:00:00+00:00", "causation_id": None, "correlation_id": None, + "payload": {"workspace": "ws"}, "artifact_hashes": [], "prev_event_hash": None, +} +_DIGEST_GENESIS = "3ca65d4da7d87a98616441a86c6866ff39b5513ccd156d8526abfd6df7ec88a7" + +_VECTOR_SECOND = { + "schema": "loop-engineer/event@1", "run_id": "run-1", "sequence": 1, + "event_id": "e1", "type": "iteration_appended", "actor": "operator", + "ts": "2026-07-24T00:00:01+00:00", "causation_id": None, "correlation_id": None, + "payload": {"iteration_id": 1, "outcome": "task_passed", "state": "execute-task"}, + "artifact_hashes": [], "prev_event_hash": _DIGEST_GENESIS, +} +_DIGEST_SECOND = "bb40984d1b98bda565d93dd90a39ea212be999078a66cf013f37cbed650c155d" + +_VECTOR_UNICODE = { + "schema": "loop-engineer/event@1", "run_id": "run-1", "sequence": 2, + "event_id": "e2", "type": "receipt_appended", "actor": "operator", + "ts": "2026-07-24T00:00:02+00:00", "causation_id": None, "correlation_id": None, + "payload": {"iteration_id": 1, "note": "café — naïve ✅", "summary": "日本語"}, + "artifact_hashes": [], "prev_event_hash": _DIGEST_SECOND, +} +_DIGEST_UNICODE = "0d0413aa0a1903a46a802f98f0a28abafd10ca09d5e312622f729482cfc40a19" + +_CONFORMANCE_VECTORS = ( + ("genesis", _VECTOR_GENESIS, _DIGEST_GENESIS), + ("second", _VECTOR_SECOND, _DIGEST_SECOND), + ("unicode-payload", _VECTOR_UNICODE, _DIGEST_UNICODE), +) + + +def test_documented_conformance_vectors(): + """The three vectors published in the contract are exactly what chain.py computes, + and the published digests AND canonical preimages are still literally in the document.""" + contract = (_ROOT / "reference" / "repo-os-contract.md").read_text(encoding="utf-8") + for name, record, digest in _CONFORMANCE_VECTORS: + assert compute_event_hash(record) == digest, f"vector {name} drifted from chain.py" + assert digest in contract, f"vector {name} digest is not documented in the contract" + preimage = canonical_json({field: record.get(field) for field in _PREIMAGE_FIELDS}) + assert preimage in contract, f"vector {name} preimage is not documented in the contract" + chained = [dict(record, event_hash=digest) for _, record, digest in _CONFORMANCE_VECTORS] + assert verify_chain(chained, expected_head=_DIGEST_UNICODE)["ok"] is True diff --git a/scripts/test_loop_cli.py b/scripts/test_loop_cli.py index ed53f3a..9ac836f 100644 --- a/scripts/test_loop_cli.py +++ b/scripts/test_loop_cli.py @@ -50,7 +50,7 @@ def test_short_help_flag_matches_long_help(): def test_help_lists_every_command_with_a_description(): out = _run("--help").stdout - for command in ("scaffold", "doctor", "validate", "verify", "inspect"): + for command in ("scaffold", "doctor", "validate", "verify", "inspect", "migrate"): assert command in out, f"help omits command {command!r}" # Per-command descriptions, not a bare command list. assert "Validate" in out or "validate the contract" in out.lower() diff --git a/scripts/test_loop_simulate_zero_writes.py b/scripts/test_loop_simulate_zero_writes.py index 2f56682..5c2bbc8 100644 --- a/scripts/test_loop_simulate_zero_writes.py +++ b/scripts/test_loop_simulate_zero_writes.py @@ -1,4 +1,4 @@ -"""Proofs that simulate observes a workspace without mutating it.""" +"""Proofs that the read verbs observe a workspace without mutating it.""" from __future__ import annotations import hashlib @@ -8,8 +8,12 @@ import sys from pathlib import Path +from chain_fixtures import make_legacy_store + from loop import emit, runner +from loop.contract import doctor_report from loop.events import SQLiteEventStore +from loop.runtime import replay_report, status_report from loop.simulate import simulate_run ROOT = Path(__file__).resolve().parent.parent @@ -20,10 +24,15 @@ def _task(status="pending"): return {"id": "T-1", "title": "T-1", "status": status, "criterion_ref": "T-1", "verify": "true", "depends_on": [], "attempts": 0, "evidence": None} -def _ws(tmp_path, task=None): +def _ws(tmp_path, task=None, *, legacy=False): w = tmp_path / "workspace"; emit.open_contract(w) (w / "TASKS.json").write_text(json.dumps({"schema": "loop-engineer/tasks@1", "tasks": [task or _task()]}), encoding="utf-8") - store = SQLiteEventStore(w / ".loop" / "events.db"); store.append(RUN_ID, "contract_opened", {"workspace": "workspace"}, actor="test") + # legacy=True seeds the v0.9.0 unchained generation; its contract_opened stands in for the chained append. + if legacy: + make_legacy_store(w / ".loop" / "events.db", run_id=RUN_ID) + store = SQLiteEventStore(w / ".loop" / "events.db") + if not legacy: + store.append(RUN_ID, "contract_opened", {"workspace": "workspace"}, actor="test") for n, state in enumerate(("plan", "critique-plan", "queue-tasks", "execute-task"), 1): store.append(RUN_ID, "iteration_appended", {"iteration_id": n, "outcome": "replanned", "state": state}, actor="test") emit.append_iteration(w, iteration_id=n, outcome="replanned", state=state) @@ -49,6 +58,22 @@ def test_simulate_on_pristine_store_creates_zero_new_files_and_full_workspace_tr assert report["would"]["action"] == "would_dispatch" and _without_shm(before) == _without_shm(after) and len(store.read(RUN_ID)) == 5 +def test_simulate_on_legacy_unchained_store_creates_zero_new_files_and_full_workspace_tree_byte_hash_unchanged(tmp_path): + w, store = _ws(tmp_path, legacy=True); (w / "subdir").mkdir(); (w / "subdir" / "sentinel.txt").write_text("same", encoding="utf-8") + before = _tree_hashes(w); report = simulate_run(w); after = _tree_hashes(w) + assert status_report(w)["chain_head"] is None + # Clean checkpointed store: assert the whole tree, so an shm-only regression cannot slip through. + assert report["would"]["action"] == "would_dispatch" and before == after and len(store.read(RUN_ID)) == 5 + + +def test_doctor_status_and_replay_on_clean_chained_store_leave_the_full_workspace_tree_byte_identical(tmp_path): + w, _ = _ws(tmp_path) + assert not (w / ".loop" / "events.db-wal").exists() and not (w / ".loop" / "events.db-shm").exists() + before = _tree_hashes(w); reports = (doctor_report(w), status_report(w), replay_report(w)); after = _tree_hashes(w) + assert all(report["ok"] for report in reports) and reports[1]["chain_head"] is not None + assert before == after + + def test_simulate_on_crash_left_wal_sidecar_leaves_events_db_and_wal_bytes_unchanged_shm_exempted(tmp_path): w, _ = _ws(tmp_path) # A child keeps its WAL frames by exiting without closing the connection. diff --git a/scripts/test_migrate_cli.py b/scripts/test_migrate_cli.py new file mode 100644 index 0000000..a459adb --- /dev/null +++ b/scripts/test_migrate_cli.py @@ -0,0 +1,128 @@ +"""CLI contract tests for `python3 -m loop migrate` — the only store-upgrade path. + +`migrate` is the one explicitly write-classed verb over an existing store: connect +never upgrades a legacy events.db, so a v0.9.0 store stays unchained until an +operator runs this command. These tests drive the real entry point as a +subprocess so the exit code and the stdout/stderr split are exercised exactly as +an operator sees them. +""" + +import json +import os +import sqlite3 +import subprocess +import sys +from pathlib import Path + +import pytest + +from chain_fixtures import make_legacy_store + +ROOT = Path(__file__).resolve().parent.parent + + +def _cli(*args: str) -> subprocess.CompletedProcess[str]: + env = dict(os.environ) + env["PYTHONPATH"] = str(ROOT) + os.pathsep + env.get("PYTHONPATH", "") + return subprocess.run([sys.executable, "-B", "-m", "loop", *args], cwd=ROOT, env=env, + text=True, capture_output=True, timeout=120) + + +def _legacy_workspace(tmp_path: Path) -> Path: + workspace = tmp_path / "workspace" + (workspace / ".loop").mkdir(parents=True) + make_legacy_store(workspace / ".loop" / "events.db") + return workspace + + +def _drifted_store(path: Path) -> Path: + """An events table that reads and writes nothing the kernel expects.""" + conn = sqlite3.connect(str(path)) + try: + conn.execute("CREATE TABLE events (run_id TEXT, sequence INTEGER)") + conn.execute("INSERT INTO events VALUES ('r1', 0)") + conn.commit() + finally: + conn.close() + return path + + +def test_migrate_on_a_legacy_workspace_exits_zero_and_reports_migrated_true(tmp_path): + result = _cli("migrate", str(_legacy_workspace(tmp_path))) + assert result.returncode == 0, result.stderr + report = json.loads(result.stdout) + assert report["ok"] is True and report["migrated"] is True + assert report["user_version"] == 2 and report["unchained_rows"] == 1 + assert report["chained_from_sequence"] == 1 + + +def test_second_migrate_exits_zero_and_reports_migrated_false(tmp_path): + workspace = _legacy_workspace(tmp_path) + assert _cli("migrate", str(workspace)).returncode == 0 + second = _cli("migrate", str(workspace)) + assert second.returncode == 0, second.stderr + assert json.loads(second.stdout)["migrated"] is False + + +def test_migrate_nonexistent_target_exits_two_with_the_exists_guard_hint(tmp_path): + missing = tmp_path / "does-not-exist" + result = _cli("migrate", str(missing)) + assert result.returncode == 2 + assert result.stdout == "" + assert "does not exist" in result.stderr and str(missing) in result.stderr + assert "scaffold" in result.stderr # the _READ_COMMANDS exists-guard hint + assert "Traceback" not in result.stderr + + +def test_migrate_missing_store_in_an_existing_workspace_is_a_typed_error(tmp_path): + workspace = tmp_path / "workspace" + (workspace / ".loop").mkdir(parents=True) + result = _cli("migrate", str(workspace)) + assert result.returncode == 2 + assert result.stdout == "" + assert result.stderr.startswith("migrate: missing_store: ") + assert "Traceback" not in result.stderr + + +def test_migrate_corrupt_store_is_a_typed_error_not_a_traceback(tmp_path): + workspace = tmp_path / "workspace" + (workspace / ".loop").mkdir(parents=True) + (workspace / ".loop" / "events.db").write_text("not sqlite", encoding="utf-8") + result = _cli("migrate", str(workspace)) + assert result.returncode == 2 + assert result.stdout == "" + assert result.stderr.startswith("migrate: corrupt_store: ") + assert "Traceback" not in result.stderr + + +def test_migrate_accepts_the_loop_dir_as_target_like_every_other_verb(tmp_path): + workspace = _legacy_workspace(tmp_path) + result = _cli("migrate", str(workspace / ".loop")) + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout)["migrated"] is True + + +def test_help_lists_migrate_and_describes_it_as_the_only_upgrade_path(): + result = _cli("--help") + assert result.returncode == 0 + assert "migrate" in result.stdout + assert "only store-upgrade path" in result.stdout + + +def test_migrate_missing_target_argument_prints_usage_and_exits_nonzero(): + result = _cli("migrate") + assert result.returncode == 2 + assert "usage" in result.stderr.lower() + assert "Traceback" not in result.stderr + + +@pytest.mark.parametrize("command,extra", [("run", []), ("pause", ["--reason", "drift probe"])]) +def test_cli_refuses_a_schema_drifted_store_without_a_traceback(tmp_path, command, extra): + """Task-4 probe: the drift path stays typed at the CLI boundary.""" + workspace = tmp_path / "workspace" + (workspace / ".loop").mkdir(parents=True) + _drifted_store(workspace / ".loop" / "events.db") + result = _cli(command, *extra, str(workspace)) + assert result.returncode == 2 + assert "Traceback" not in result.stderr + assert result.stderr.strip().startswith(f"{command}: ")