diff --git a/loop/__init__.py b/loop/__init__.py index 26f95cf..82e5125 100644 --- a/loop/__init__.py +++ b/loop/__init__.py @@ -7,16 +7,24 @@ from .paths import LoopPaths, resolve_loop_paths from .contract import TERMINAL_STATES, VALIDATION_MODES, doctor_report, validate_contract +from .events import EVENT_SCHEMA_ID, EVENT_TYPES, EventStore, SQLiteEventStore, validate_event from .plan import PLAN_SCHEMA_ID, TASK_KINDS, validate_plan +from .reducer import reduce_events __all__ = [ + "EVENT_SCHEMA_ID", + "EVENT_TYPES", + "EventStore", "LoopPaths", "PLAN_SCHEMA_ID", + "SQLiteEventStore", "TASK_KINDS", "TERMINAL_STATES", "VALIDATION_MODES", "doctor_report", "resolve_loop_paths", + "reduce_events", "validate_contract", + "validate_event", "validate_plan", ] diff --git a/loop/events.py b/loop/events.py new file mode 100644 index 0000000..6b8b7c0 --- /dev/null +++ b/loop/events.py @@ -0,0 +1,250 @@ +"""event@1 EventStore protocol and SQLite/WAL implementation (ADR 0001).""" + +from __future__ import annotations + +import json +import re +import sqlite3 +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping, Protocol, Sequence, runtime_checkable + +from .contract import ContractIssue, _resolve_requested_mode, _schemas_dir +from .emit import _ITERATION_OUTCOMES, _RECEIPT_OUTCOMES, _RECEIPT_ROLES + +EVENT_SCHEMA_ID = "loop-engineer/event@1" +EVENT_TYPES = ("contract_opened", "iteration_appended", "receipt_appended", "terminal_written") + +_PAYLOAD_REQUIRED_FIELDS: dict[str, tuple[str, ...]] = { + "contract_opened": ("workspace",), + "iteration_appended": ("iteration_id", "outcome"), + "receipt_appended": ("iteration_id", "role", "model", "outcome"), + "terminal_written": ("state", "criteria_met", "evidence", "false_completion"), +} + +_CREATE_EVENTS_TABLE = """ +CREATE TABLE IF NOT EXISTS 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) +) +""" +_CREATE_NO_UPDATE_TRIGGER = """ +CREATE TRIGGER IF NOT EXISTS events_no_update BEFORE UPDATE ON events +BEGIN SELECT RAISE(ABORT, 'events table is append-only: UPDATE is forbidden'); END +""" +_CREATE_NO_DELETE_TRIGGER = """ +CREATE TRIGGER IF NOT EXISTS events_no_delete BEFORE DELETE ON events +BEGIN SELECT RAISE(ABORT, 'events table is append-only: DELETE is forbidden'); END +""" + + +class EventValidationError(ValueError): + """A candidate event failed event@1 validation and was refused before write.""" + + +class DuplicateEventError(ValueError): + """An event_id already exists in the store; callers may treat this as a retry.""" + + +class SequenceConflictError(ValueError): + """expected_sequence differed from the atomically assigned next sequence.""" + + +@runtime_checkable +class EventStore(Protocol): + def append(self, run_id: str, event_type: str, payload: Mapping[str, Any], *, actor: str, + event_id: str | None = None, causation_id: str | None = None, + correlation_id: str | None = None, + artifact_hashes: Sequence[Mapping[str, Any]] | None = None, + expected_sequence: int | None = None, ts: str | None = None) -> dict[str, Any]: ... + + def read(self, run_id: str, *, since_sequence: int | None = None) -> list[dict[str, Any]]: ... + + def latest_sequence(self, run_id: str) -> int | None: ... + + +def _load_event_schema() -> dict[str, Any]: + return json.loads((_schemas_dir() / "event.schema.json").read_text(encoding="utf-8")) + + +def _payload_issues(event_type: str, payload: Any) -> list[str]: + """Cross-field payload validation shared by jsonschema and structural modes.""" + if not isinstance(payload, dict): + return [f"payload must be an object for type {event_type!r}"] + issues: list[str] = [] + for field in _PAYLOAD_REQUIRED_FIELDS.get(event_type, ()): + if field not in payload: + issues.append(f"{event_type} payload missing {field!r}") + if event_type == "contract_opened": + if "workspace" in payload and (not isinstance(payload["workspace"], str) or not payload["workspace"]): + issues.append("contract_opened.workspace must be a non-empty string") + elif event_type == "iteration_appended": + value = payload.get("iteration_id") + if "iteration_id" in payload and (not isinstance(value, int) or isinstance(value, bool) or value < 0): + issues.append("iteration_appended.iteration_id must be a non-negative integer") + if payload.get("outcome") not in _ITERATION_OUTCOMES: + issues.append(f"iteration_appended.outcome must be one of {_ITERATION_OUTCOMES}") + if "state" in payload and payload["state"] is not None and not isinstance(payload["state"], str): + issues.append("iteration_appended.state must be a string or null") + elif event_type == "receipt_appended": + value = payload.get("iteration_id") + if "iteration_id" in payload and (not isinstance(value, int) or isinstance(value, bool) or value < 0): + issues.append("receipt_appended.iteration_id must be a non-negative integer") + if payload.get("role") not in _RECEIPT_ROLES: + issues.append(f"receipt_appended.role must be one of {_RECEIPT_ROLES}") + if "model" in payload and (not isinstance(payload["model"], str) or not payload["model"]): + issues.append("receipt_appended.model must be a non-empty string") + if payload.get("outcome") not in _RECEIPT_OUTCOMES: + issues.append(f"receipt_appended.outcome must be one of {_RECEIPT_OUTCOMES}") + elif event_type == "terminal_written": + if "state" in payload and not isinstance(payload["state"], str): + issues.append("terminal_written.state must be a string") + if "criteria_met" in payload and not isinstance(payload["criteria_met"], dict): + issues.append("terminal_written.criteria_met must be an object") + if "evidence" in payload and not isinstance(payload["evidence"], list): + issues.append("terminal_written.evidence must be an array") + if "false_completion" in payload and not isinstance(payload["false_completion"], bool): + issues.append("terminal_written.false_completion must be a boolean") + return issues + + +def _structural_validate_event(data: dict[str, Any]) -> list[str]: + """Stdlib fallback equivalent to the schema's required envelope surface.""" + issues: list[str] = [] + if data.get("schema") != EVENT_SCHEMA_ID: + issues.append(f"expected schema {EVENT_SCHEMA_ID!r}, got {data.get('schema')!r}") + for field in ("event_id", "run_id", "actor", "ts"): + if not isinstance(data.get(field), str) or not data[field]: + issues.append(f"{field} must be a non-empty string") + sequence = data.get("sequence") + if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 0: + issues.append("sequence must be a non-negative integer") + if not isinstance(data.get("type"), str) or data["type"] not in EVENT_TYPES: + issues.append(f"type must be one of {EVENT_TYPES}") + 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") + hashes = data.get("artifact_hashes", []) + if not isinstance(hashes, list): + issues.append("artifact_hashes must be an array") + else: + for item in hashes: + path = item.get("path") if isinstance(item, dict) else None + digest = item.get("sha256") if isinstance(item, dict) else None + if not isinstance(path, str) or not path or not isinstance(digest, str) or re.search(r"^[0-9a-f]{64}$", digest) is None: + issues.append("artifact_hashes entries need non-empty string path + 64-character lowercase hex sha256") + break + if isinstance(data.get("type"), str): + issues.extend(_payload_issues(data["type"], data.get("payload"))) + return issues + + +def _jsonschema_validate_event(data: dict[str, Any]) -> list[str]: + import jsonschema # type: ignore + + validator = jsonschema.Draft202012Validator(_load_event_schema()) + issues = [f"{'/'.join(str(p) for p in error.absolute_path) or ''}: {error.message}" for error in validator.iter_errors(data)] + if isinstance(data.get("type"), str): + issues.extend(_payload_issues(data["type"], data.get("payload"))) + return issues + + +def _validate_event_dict(data: Any, *, mode: str | None = None) -> dict[str, Any]: + requested_mode, resolved_mode = _resolve_requested_mode(mode) + if not isinstance(data, dict): + issues = ["event record must be an object"] + elif resolved_mode == "jsonschema": + issues = _jsonschema_validate_event(data) + else: + issues = _structural_validate_event(data) + return {"ok": not issues, "validation_mode": resolved_mode, "requested_mode": requested_mode, + "schemas_checked": [EVENT_SCHEMA_ID], + "issues": [ContractIssue("invalid_event", issue) for issue in issues]} + + +def validate_event(data: dict[str, Any], *, mode: str | None = None) -> dict[str, Any]: + """Validate a standalone event@1 record in the requested validation mode.""" + return _validate_event_dict(data, mode=mode) + + +class SQLiteEventStore: + """A transactional SQLite/WAL event store with DB-enforced append-only rows.""" + + def __init__(self, path: str | Path) -> None: + self._path = Path(path) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(str(self._path), isolation_level=None, timeout=5.0) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=FULL") + conn.execute("PRAGMA busy_timeout=5000") + conn.execute(_CREATE_EVENTS_TABLE) + conn.execute(_CREATE_NO_UPDATE_TRIGGER) + conn.execute(_CREATE_NO_DELETE_TRIGGER) + return conn + + def append(self, run_id: str, event_type: str, payload: Mapping[str, Any], *, actor: str, + event_id: str | None = None, causation_id: str | None = None, + correlation_id: str | None = None, + artifact_hashes: Sequence[Mapping[str, Any]] | None = None, + expected_sequence: int | None = None, ts: str | None = None) -> dict[str, Any]: + normalized_payload: Any = dict(payload) if isinstance(payload, Mapping) else payload + normalized_hashes: Any = [] if artifact_hashes is None else [ + dict(item) if isinstance(item, Mapping) else item for item in artifact_hashes + ] + record = {"schema": EVENT_SCHEMA_ID, "event_id": event_id if event_id is not None else uuid.uuid4().hex, "run_id": run_id, + "sequence": 0, "type": event_type, "actor": actor, "causation_id": causation_id, + "correlation_id": correlation_id, "ts": ts if ts is not None else datetime.now(timezone.utc).isoformat(timespec="seconds"), + "payload": normalized_payload, "artifact_hashes": normalized_hashes} + report = _validate_event_dict(record) + if not report["ok"]: + raise EventValidationError(f"event failed validation: {report['issues']}") + conn = self._connect() + try: + conn.execute("BEGIN IMMEDIATE") + row = conn.execute("SELECT MAX(sequence) FROM events WHERE run_id = ?", (run_id,)).fetchone() + next_sequence = 0 if row[0] is None else row[0] + 1 + if expected_sequence is not None and expected_sequence != next_sequence: + 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 + 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)) + except sqlite3.IntegrityError as exc: + conn.execute("ROLLBACK") + raise DuplicateEventError(record["event_id"]) from exc + conn.execute("COMMIT") + finally: + conn.close() + return record + + def read(self, run_id: str, *, since_sequence: int | None = None) -> list[dict[str, Any]]: + """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() + 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() + try: + row = conn.execute("SELECT MAX(sequence) FROM events WHERE run_id = ?", (run_id,)).fetchone() + finally: + conn.close() + return row[0] diff --git a/loop/reducer.py b/loop/reducer.py new file mode 100644 index 0000000..7d86728 --- /dev/null +++ b/loop/reducer.py @@ -0,0 +1,97 @@ +"""Pure deterministic event@1 reducer; persistence is deliberately not involved.""" + +from __future__ import annotations + +from typing import Any, Iterable, Mapping + +from . import fsm +from .completion import CompletionPolicyError, criteria_satisfy_completion, normalize_completion_policy +from .contract import TERMINAL_STATES +from .events import _structural_validate_event + + +class EventReplayError(ValueError): + """An event stream is malformed or violates a replay domain invariant.""" + + +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": [], "event_count": 0, + "last_sequence": None} + + +def _validate_terminal_payload_semantics(payload: Mapping[str, Any]) -> None: + state = payload.get("state") + if state not in TERMINAL_STATES: + raise EventReplayError(f"terminal_written payload has non-canonical state {state!r}") + if state != "Succeeded": + return + if payload.get("false_completion") is True: + raise EventReplayError("refusing Succeeded terminal_written with false_completion=True (G1)") + try: + policy = normalize_completion_policy(payload.get("completion_policy")) + except CompletionPolicyError as exc: + raise EventReplayError(f"invalid completion_policy: {exc}") from exc + criteria = payload.get("criteria_met") + if not isinstance(criteria, dict) or not criteria_satisfy_completion(criteria, policy): + raise EventReplayError("Succeeded terminal_written does not satisfy the completion policy (G1)") + evidence = payload.get("evidence") + if not isinstance(evidence, list) or not evidence: + raise EventReplayError("Succeeded terminal_written has empty evidence (G1)") + + +def _reduce_one(state: dict[str, Any], event: Mapping[str, Any]) -> dict[str, Any]: + if not isinstance(event, Mapping): + raise EventReplayError("event must be a mapping") + issues = _structural_validate_event(dict(event)) + if issues: + raise EventReplayError(f"malformed event: {issues}") + run_id = event["run_id"] + if state["run_id"] is not None and state["run_id"] != run_id: + raise EventReplayError(f"mixed run_id in one replay: {state['run_id']!r} vs {run_id!r}") + 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}") + if state["terminal"] is not None: + raise EventReplayError("event appended after terminal — terminal is immutable") + event_type = event["type"] + if event_type == "contract_opened" and state["last_sequence"] is not None: + raise EventReplayError("contract_opened must be the first event in a run") + 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} + payload = event["payload"] + if event_type == "contract_opened": + new_state["state"] = "intake" + new_state["iteration_id"] = 0 + elif event_type == "iteration_appended": + target = payload.get("state") + if target is not None: + if target not in fsm.ALL_STATES: + raise EventReplayError(f"illegal FSM transition {new_state['state']!r} -> {target!r}") + if not fsm.is_legal_transition(new_state["state"], target): + raise EventReplayError(f"illegal FSM transition {new_state['state']!r} -> {target!r}") + new_state["state"] = target + new_state["iteration_id"] = payload["iteration_id"] + if payload.get("task_id"): + new_state["active_task"] = payload["task_id"] + entry = dict(payload, event_id=event["event_id"], causation_id=event.get("causation_id"), correlation_id=event.get("correlation_id"), ts=event["ts"]) + new_state["runlog_entries"] = new_state["runlog_entries"] + [entry] + elif event_type == "receipt_appended": + entry = dict(payload, event_id=event["event_id"], causation_id=event.get("causation_id"), correlation_id=event.get("correlation_id"), ts=event["ts"]) + new_state["receipts"] = new_state["receipts"] + [entry] + elif event_type == "terminal_written": + if not fsm.is_legal_transition(new_state["state"], fsm.TERMINAL_MARKER): + raise EventReplayError(f"illegal FSM transition {new_state['state']!r} -> {fsm.TERMINAL_MARKER!r}") + _validate_terminal_payload_semantics(payload) + new_state["state"] = fsm.TERMINAL_MARKER + new_state["terminal"] = dict(payload, event_id=event["event_id"], ts=event["ts"]) + return new_state + + +def reduce_events(events: Iterable[Mapping[str, Any]], *, initial: Mapping[str, Any] | None = None) -> dict[str, Any]: + """Fold events without mutating the supplied stream or initial projection.""" + state = ({**initial, "runlog_entries": list(initial["runlog_entries"]), "receipts": list(initial["receipts"])} if initial is not None else _empty_projection(None)) + for event in events: + state = _reduce_one(state, event) + return state diff --git a/reference/repo-os-contract.md b/reference/repo-os-contract.md index 437887f..dcaaefd 100644 --- a/reference/repo-os-contract.md +++ b/reference/repo-os-contract.md @@ -580,6 +580,42 @@ the negative tests). --- +## 16. `loop-engineer/event@1` — EventStore + deterministic reducer + +`schemas/event.schema.json` defines one immutable, append-only fact in a run's +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. +`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. + +**Scope boundary:** unlike manifest/state/tasks/terminal (§11), `event@1` is +**not yet** an artifact `loop doctor` reads from a scaffolded workspace, and +no workspace-relative on-disk location (e.g. `.loop/events.db`) has been +decided. The execution-runtime milestone that wires a live event log into +`scaffold`/`emit`/`doctor` will make that call. + +**Event types:** `contract_opened | iteration_appended | receipt_appended | +terminal_written` — one-to-one with `loop.emit`'s four writer operations +(`open_contract`/`append_iteration`/`append_receipt`/`terminate`), so a +future write-through migration targets an already-matching payload shape. + +**Two-layer enforcement, deliberately split:** the store validates event@1 +envelope/payload *shape* only (`loop/events.py::validate_event`, both +validation modes, both type-checked in structural fallback); the reducer +enforces *domain* semantics at replay time — FSM transition legality +(`loop.fsm.is_legal_transition`), G1 completion +(`loop.completion.criteria_satisfy_completion`), and terminal immutability +(no event may follow a `terminal_written`) — reusing the exact functions +`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. + +--- + Sources: "Designing a Loop Engineer Skill for Frontier Agent Workflows" (2026), synthesizing Anthropic guidance on long-running agent harnesses (anthropic.com, 2025), OpenAI Agents/Codex guidance, Google Conductor, and arXiv PreFlect (2602.07187), SWE-Marathon (2606.07682), Web Agents diff --git a/schemas/event.schema.json b/schemas/event.schema.json new file mode 100644 index 0000000..04d6359 --- /dev/null +++ b/schemas/event.schema.json @@ -0,0 +1,33 @@ +{ + "$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.", + "type": "object", + "required": ["schema", "event_id", "run_id", "sequence", "type", "actor", "ts", "payload"], + "properties": { + "schema": { "const": "loop-engineer/event@1" }, + "event_id": { "type": "string", "minLength": 1 }, + "run_id": { "type": "string", "minLength": 1 }, + "sequence": { "type": "integer", "minimum": 0 }, + "type": { "enum": ["contract_opened", "iteration_appended", "receipt_appended", "terminal_written"] }, + "actor": { "type": "string", "minLength": 1 }, + "causation_id": { "type": ["string", "null"] }, + "correlation_id": { "type": ["string", "null"] }, + "ts": { "type": "string", "minLength": 1 }, + "payload": { "type": "object" }, + "artifact_hashes": { + "type": "array", + "items": { + "type": "object", + "required": ["path", "sha256"], + "properties": { + "path": { "type": "string", "minLength": 1 }, + "sha256": { "type": "string", "pattern": "^[0-9a-f]{64}$" } + }, + "additionalProperties": true + } + } + }, + "additionalProperties": true +} diff --git a/scripts/test_eventstore.py b/scripts/test_eventstore.py new file mode 100644 index 0000000..c754cac --- /dev/null +++ b/scripts/test_eventstore.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import sqlite3 +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from loop.events import ( + EVENT_SCHEMA_ID, + DuplicateEventError, + EventStore, + EventValidationError, + SQLiteEventStore, + SequenceConflictError, + validate_event, +) + + +def event(event_type: str = "contract_opened", **overrides: object) -> dict[str, object]: + payloads = { + "contract_opened": {"workspace": "demo"}, + "iteration_appended": {"iteration_id": 0, "outcome": "task_passed"}, + "receipt_appended": {"iteration_id": 0, "role": "read", "model": "test", "outcome": "ok"}, + "terminal_written": {"state": "Succeeded", "criteria_met": {"done": True}, "evidence": ["proof"], "false_completion": False}, + } + result: dict[str, object] = { + "schema": EVENT_SCHEMA_ID, "event_id": "event-1", "run_id": "run-1", "sequence": 0, + "type": event_type, "actor": "test", "ts": "2026-01-01T00:00:00+00:00", "payload": payloads[event_type], + } + result.update(overrides) + return result + + +@pytest.mark.parametrize("event_type", ["contract_opened", "iteration_appended", "receipt_appended", "terminal_written"]) +def test_basic_validation_accepts_all_event_types(event_type: str) -> None: + report = validate_event(event(event_type), mode="basic") + assert report["ok"] is True + assert report["issues"] == [] + + +@pytest.mark.parametrize("event_type", ["contract_opened", "iteration_appended", "receipt_appended", "terminal_written"]) +def test_release_validation_accepts_event_when_jsonschema_is_available(event_type: str) -> None: + pytest.importorskip("jsonschema") + assert validate_event(event(event_type), mode="release")["ok"] is True + + +def test_structural_validation_type_checks_required_surface() -> None: + report = validate_event(event(sequence="0", payload="not-an-object", artifact_hashes=[{"path": 1, "sha256": 1}]), mode="basic") + messages = " ".join(issue["message"] for issue in report["issues"]) + assert report["ok"] is False + assert "sequence" in messages and "payload" in messages and "artifact_hashes" in messages + + +@pytest.mark.parametrize("mode", ["basic", "release"]) +@pytest.mark.parametrize("iteration_id", [None, True, -1]) +def test_receipt_validation_requires_a_non_negative_integer_iteration_id(mode: str, iteration_id: int | bool | None) -> None: + if mode == "release": + pytest.importorskip("jsonschema") + payload = {"role": "read", "model": "test", "outcome": "ok"} + if iteration_id is not None: + payload["iteration_id"] = iteration_id + assert validate_event(event("receipt_appended", payload=payload), mode=mode)["ok"] is False + + +def test_store_assigns_sequences_round_trips_and_is_a_protocol(tmp_path) -> None: + store = SQLiteEventStore(tmp_path / "events.db") + assert isinstance(store, EventStore) + assert store.latest_sequence("run") is None + records = [store.append("run", "contract_opened", {"workspace": "w"}, actor="t"), + store.append("run", "iteration_appended", {"iteration_id": 0, "outcome": "task_passed"}, actor="t"), + store.append("run", "receipt_appended", {"iteration_id": 0, "role": "read", "model": "m", "outcome": "ok"}, actor="t", artifact_hashes=[{"path": "a", "sha256": "a" * 64}])] + assert [record["sequence"] for record in records] == [0, 1, 2] + assert [record["sequence"] for record in store.read("run")] == [0, 1, 2] + assert [record["sequence"] for record in store.read("run", since_sequence=0)] == [1, 2] + assert store.read("run", since_sequence=1)[0]["sequence"] == 2 + assert store.read("run")[2]["artifact_hashes"] == [{"path": "a", "sha256": "a" * 64}] + + +def test_store_rejects_duplicates_conflicts_and_malformed_events(tmp_path) -> None: + store = SQLiteEventStore(tmp_path / "events.db") + store.append("run", "contract_opened", {"workspace": "w"}, actor="t", event_id="same") + with pytest.raises(DuplicateEventError): + store.append("other", "contract_opened", {"workspace": "w"}, actor="t", event_id="same") + with pytest.raises(SequenceConflictError): + store.append("run", "iteration_appended", {"iteration_id": 0, "outcome": "task_passed"}, actor="t", expected_sequence=0) + appended = store.append("run", "iteration_appended", {"iteration_id": 0, "outcome": "task_passed"}, actor="t", expected_sequence=1) + assert appended["sequence"] == 1 + with pytest.raises(EventValidationError): + store.append("run", "bogus", {}, actor="t") + assert store.latest_sequence("run") == 1 + + +def test_store_is_wal_and_raw_sql_cannot_mutate_rows(tmp_path) -> None: + path = tmp_path / "events.db" + store = SQLiteEventStore(path) + store.append("run", "contract_opened", {"workspace": "w"}, actor="t") + raw = store._connect() + try: + assert raw.execute("PRAGMA journal_mode").fetchone()[0] == "wal" + assert raw.execute("PRAGMA synchronous").fetchone()[0] == 2 + assert raw.execute("PRAGMA busy_timeout").fetchone()[0] == 5000 + with pytest.raises((sqlite3.IntegrityError, sqlite3.OperationalError), match="append-only"): + raw.execute("UPDATE events SET type = 'x'") + with pytest.raises((sqlite3.IntegrityError, sqlite3.OperationalError), match="append-only"): + raw.execute("DELETE FROM events") + finally: + raw.close() + + +def test_concurrent_appends_are_serialized(tmp_path) -> None: + path = tmp_path / "events.db" + + def append(index: int) -> int: + return SQLiteEventStore(path).append("run", "iteration_appended", {"iteration_id": index, "outcome": "task_passed"}, actor="t")["sequence"] + + SQLiteEventStore(path).append("run", "contract_opened", {"workspace": "w"}, actor="t") + with ThreadPoolExecutor(max_workers=8) as pool: + sequences = list(pool.map(append, range(8))) + assert set(sequences) == set(range(1, 9)) diff --git a/scripts/test_reducer.py b/scripts/test_reducer.py new file mode 100644 index 0000000..fce606c --- /dev/null +++ b/scripts/test_reducer.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +import json + +import pytest + +from loop.events import EVENT_SCHEMA_ID, SQLiteEventStore +from loop.reducer import EventReplayError, reduce_events + + +def stream(run_id: str = "run") -> list[dict[str, object]]: + base = {"schema": EVENT_SCHEMA_ID, "run_id": run_id, "actor": "t", "causation_id": None, "correlation_id": None} + return [ + {**base, "event_id": "e0", "sequence": 0, "type": "contract_opened", "ts": "2026-01-01T00:00:00+00:00", "payload": {"workspace": "w"}}, + {**base, "event_id": "e1", "sequence": 1, "type": "iteration_appended", "ts": "2026-01-01T00:00:01+00:00", "payload": {"iteration_id": 1, "outcome": "task_passed", "state": "plan"}}, + {**base, "event_id": "e2", "sequence": 2, "type": "iteration_appended", "ts": "2026-01-01T00:00:02+00:00", "payload": {"iteration_id": 2, "outcome": "task_passed", "state": "critique-plan"}}, + {**base, "event_id": "e3", "sequence": 3, "type": "receipt_appended", "ts": "2026-01-01T00:00:03+00:00", "payload": {"iteration_id": 2, "role": "read", "model": "m", "outcome": "ok"}}, + {**base, "event_id": "e4", "sequence": 4, "type": "terminal_written", "ts": "2026-01-01T00:00:04+00:00", "payload": {"state": "Succeeded", "criteria_met": {"done": True}, "evidence": ["proof"], "false_completion": False}}, + ] + + +def test_reducer_is_deterministic_and_resumable() -> None: + events = stream() + whole = reduce_events(events) + assert json.dumps(whole, sort_keys=True) == json.dumps(reduce_events(events), sort_keys=True) + assert reduce_events(events[2:], initial=reduce_events(events[:2])) == whole + assert reduce_events([])["event_count"] == 0 + + +def test_reducer_is_deterministic_after_store_round_trip(tmp_path) -> None: + store = SQLiteEventStore(tmp_path / "events.db") + for item in stream(): + store.append(item["run_id"], item["type"], item["payload"], actor=item["actor"], event_id=item["event_id"], ts=item["ts"]) + assert json.dumps(reduce_events(store.read("run")), sort_keys=True) == json.dumps(reduce_events(store.read("run")), sort_keys=True) + + +def test_reducer_resumes_from_an_explicit_sequence_zero_cursor(tmp_path) -> None: + store = SQLiteEventStore(tmp_path / "events.db") + for item in stream(): + store.append(item["run_id"], item["type"], item["payload"], actor=item["actor"], event_id=item["event_id"], ts=item["ts"]) + whole = reduce_events(store.read("run")) + prior = reduce_events([store.read("run")[0]]) + resumed = reduce_events(store.read("run", since_sequence=0), initial=prior) + assert json.dumps(resumed, sort_keys=True) == json.dumps(whole, sort_keys=True) + + +@pytest.mark.parametrize("mutate, message", [ + (lambda events: events.__setitem__(1, {**events[1], "sequence": 2}), "non-monotonic"), + (lambda events: events.__setitem__(1, {**events[1], "run_id": "other"}), "mixed run_id"), + (lambda events: events.__setitem__(0, {**events[0], "type": "iteration_appended", "payload": {"iteration_id": 0, "outcome": "task_passed"}}), "before contract_opened"), + (lambda events: events.__setitem__(1, {**events[1], "type": "contract_opened", "payload": {"workspace": "w"}}), "contract_opened must be the first"), + (lambda events: events.__setitem__(1, {**events[1], "payload": {"iteration_id": 1, "outcome": "task_passed", "state": "verify"}}), "illegal FSM transition"), + (lambda events: events.__setitem__(4, {**events[4], "payload": {"state": "Succeeded", "criteria_met": {"done": True}, "evidence": [], "false_completion": False}}), "empty evidence"), + (lambda events: events.__setitem__(4, {**events[4], "payload": {"state": "Succeeded", "criteria_met": {"done": True}, "evidence": ["proof"], "false_completion": True}}), "false_completion"), + (lambda events: events.__setitem__(4, {**events[4], "payload": {"state": "Succeeded", "criteria_met": {"done": False}, "evidence": ["proof"], "false_completion": False}}), "completion policy"), + (lambda events: events.__setitem__(4, {**events[4], "payload": {"state": "succeeded", "criteria_met": {"done": True}, "evidence": ["proof"], "false_completion": False}}), "non-canonical state"), +]) +def test_reducer_rejects_tampered_streams(mutate, message: str) -> None: + events = stream() + mutate(events) + with pytest.raises(EventReplayError, match=message): + reduce_events(events) + + +def test_reducer_rejects_event_after_terminal() -> None: + events = stream() + events.append({**events[-1], "event_id": "later", "sequence": 5, "type": "receipt_appended", "payload": {"iteration_id": 2, "role": "read", "model": "m", "outcome": "ok"}}) + with pytest.raises(EventReplayError, match="immutable"): + reduce_events(events)