From 4319882f3560a2f765031010522e4132aa50bbde Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sat, 5 Sep 2026 18:59:15 +0000 Subject: [PATCH] Add durable state store and boot-time reconciliation (A2/A3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Build-order step 1: the critical path everything else sits on. - StateStore: SQLite in WAL mode with synchronous=FULL, so every state transition and intent-log write is committed and fsynced before returning; a power cut cannot tear state - Durable intent log: begin_intent records a unit of work before it starts; commit/abort resolve it; pending intents survive restart - schema_version stamped from the first commit, with a sequential migration hook; a newer state file than the code supports is refused rather than misread - reconcile_on_boot (A3 first cut): pending intents are checked against external truth by registered reconcilers (observable checks, never model self-assessment); verified effects are committed, verified no-ops aborted, anything unverifiable gates to AWAIT_INPUT — never blind-resume - Lives in nisse_dev per spec 0.1: the library is extracted after the dev agent works, not designed up front - Integration test kills a subprocess mid-work with os._exit and verifies state integrity and intent visibility on reopen Co-Authored-By: Claude Fable 5 --- nisse-dev/src/nisse_dev/reconcile.py | 53 +++++ nisse-dev/src/nisse_dev/state.py | 230 +++++++++++++++++++++ nisse-dev/tests/test_reconcile.py | 68 ++++++ nisse-dev/tests/test_state.py | 118 +++++++++++ tests/integration/test_crash_durability.py | 47 +++++ 5 files changed, 516 insertions(+) create mode 100644 nisse-dev/src/nisse_dev/reconcile.py create mode 100644 nisse-dev/src/nisse_dev/state.py create mode 100644 nisse-dev/tests/test_reconcile.py create mode 100644 nisse-dev/tests/test_state.py create mode 100644 tests/integration/test_crash_durability.py diff --git a/nisse-dev/src/nisse_dev/reconcile.py b/nisse-dev/src/nisse_dev/reconcile.py new file mode 100644 index 0000000..af33076 --- /dev/null +++ b/nisse-dev/src/nisse_dev/reconcile.py @@ -0,0 +1,53 @@ +"""Boot-time reconciliation (spec A3): turn an ambiguous crash into a safe, +known state. + +Every pending intent found at startup is checked against external truth by a +registered reconciler — an observable check (query Gitea, inspect the tree), +never model self-assessment. Anything unverifiable gates to AWAIT_INPUT; +the agent never blind-resumes. +""" + +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import Literal + +from nisse_dev.state import Intent, StateStore + +# Returns True if the intent's external effect verifiably happened (do not +# repeat it), False if it verifiably never started (safe to discard), or +# None if it cannot be verified. +Reconciler = Callable[[Intent], bool | None] + + +@dataclass(frozen=True) +class BootReport: + action: Literal["resume", "await_input"] + notes: list[str] + + +def reconcile_on_boot( + store: StateStore, reconcilers: Mapping[str, Reconciler] +) -> BootReport: + notes: list[str] = [] + unresolved = False + for intent in store.pending_intents(): + label = f"intent {intent.id} ({intent.kind})" + reconciler = reconcilers.get(intent.kind) + if reconciler is None: + unresolved = True + notes.append(f"{label}: no reconciler registered") + continue + verdict = reconciler(intent) + if verdict is True: + store.commit_intent(intent.id) + notes.append(f"{label}: external effect verified, marked committed") + elif verdict is False: + store.abort_intent(intent.id) + notes.append(f"{label}: never took effect, marked aborted") + else: + unresolved = True + notes.append(f"{label}: unverifiable") + return BootReport( + action="await_input" if unresolved else "resume", + notes=notes, + ) diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py new file mode 100644 index 0000000..57e5fd5 --- /dev/null +++ b/nisse-dev/src/nisse_dev/state.py @@ -0,0 +1,230 @@ +"""Durable state store (spec A2). + +SQLite in WAL mode with ``synchronous=FULL``: every commit is fsynced, so a +power cut mid-write can never produce torn state. Every state transition and +intent-log write is a single immediately-committed statement — in-memory state +that hasn't hit disk is assumed lost. + +The intent log records a unit of work *before* it starts, so a restart knows +what was in flight (see ``nisse_dev.reconcile`` for how pending intents are +resolved against external truth on boot). +""" + +import json +import sqlite3 +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Self + +CURRENT_SCHEMA_VERSION = 1 + +Migration = Callable[[sqlite3.Connection], None] + +# Maps version N to the migration that takes the schema from N to N+1. +MIGRATIONS: Mapping[int, Migration] = {} + + +class StateSchemaError(RuntimeError): + pass + + +@dataclass(frozen=True) +class AgentState: + state: str + project: str | None + data: dict[str, Any] + updated_at: str + + +@dataclass(frozen=True) +class Intent: + id: int + kind: str + payload: dict[str, Any] + status: str + created_at: str + resolved_at: str | None + + +_SCHEMA = [ + "CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)", + """ + CREATE TABLE agent_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + state TEXT NOT NULL, + project TEXT, + data TEXT NOT NULL DEFAULT '{}', + updated_at TEXT NOT NULL + ) + """, + """ + CREATE TABLE intent_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + created_at TEXT NOT NULL, + resolved_at TEXT + ) + """, +] + + +def _now() -> str: + return datetime.now(UTC).isoformat() + + +class StateStore: + def __init__( + self, + path: str | Path, + *, + schema_version: int = CURRENT_SCHEMA_VERSION, + migrations: Mapping[int, Migration] = MIGRATIONS, + ) -> None: + self._conn = sqlite3.connect(path, isolation_level=None) + self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA journal_mode=WAL") + self._conn.execute("PRAGMA synchronous=FULL") + self._conn.execute("PRAGMA busy_timeout=5000") + self._target_version = schema_version + self._init_or_migrate(migrations) + + def _init_or_migrate(self, migrations: Mapping[int, Migration]) -> None: + fresh = ( + self._conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name='meta'" + ).fetchone() + is None + ) + if fresh: + self._in_transaction(self._create_schema) + return + stored = self.schema_version + if stored > self._target_version: + raise StateSchemaError( + f"state schema version {stored} is newer than this code " + f"supports ({self._target_version}); refusing to open" + ) + for version in range(stored, self._target_version): + migrate = migrations.get(version) + if migrate is None: + raise StateSchemaError( + f"no migration registered from schema version {version}" + ) + self._in_transaction(lambda m=migrate, v=version: self._apply(m, v)) + + def _create_schema(self) -> None: + for statement in _SCHEMA: + self._conn.execute(statement) + self._conn.execute( + "INSERT INTO meta (key, value) VALUES ('schema_version', ?)", + (str(self._target_version),), + ) + + def _apply(self, migrate: Migration, from_version: int) -> None: + migrate(self._conn) + self._conn.execute( + "UPDATE meta SET value = ? WHERE key = 'schema_version'", + (str(from_version + 1),), + ) + + def _in_transaction(self, work: Callable[[], None]) -> None: + self._conn.execute("BEGIN IMMEDIATE") + try: + work() + except BaseException: + self._conn.execute("ROLLBACK") + raise + self._conn.execute("COMMIT") + + @property + def schema_version(self) -> int: + row = self._conn.execute( + "SELECT value FROM meta WHERE key = 'schema_version'" + ).fetchone() + return int(row["value"]) + + def get_state(self) -> AgentState | None: + row = self._conn.execute("SELECT * FROM agent_state WHERE id = 1").fetchone() + if row is None: + return None + return AgentState( + state=row["state"], + project=row["project"], + data=json.loads(row["data"]), + updated_at=row["updated_at"], + ) + + def transition( + self, + state: str, + *, + project: str | None = None, + data: dict[str, Any] | None = None, + ) -> None: + """Record the new state-machine position; the full position is passed + every time (no partial updates), committed and fsynced before return.""" + self._conn.execute( + """ + INSERT INTO agent_state (id, state, project, data, updated_at) + VALUES (1, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + state = excluded.state, + project = excluded.project, + data = excluded.data, + updated_at = excluded.updated_at + """, + (state, project, json.dumps(data or {}), _now()), + ) + + def begin_intent(self, kind: str, payload: dict[str, Any]) -> int: + """Durably record that a unit of work is about to start.""" + cursor = self._conn.execute( + "INSERT INTO intent_log (kind, payload, created_at) VALUES (?, ?, ?)", + (kind, json.dumps(payload), _now()), + ) + assert cursor.lastrowid is not None + return cursor.lastrowid + + def commit_intent(self, intent_id: int) -> None: + self._resolve_intent(intent_id, "committed") + + def abort_intent(self, intent_id: int) -> None: + self._resolve_intent(intent_id, "aborted") + + def _resolve_intent(self, intent_id: int, status: str) -> None: + cursor = self._conn.execute( + "UPDATE intent_log SET status = ?, resolved_at = ? " + "WHERE id = ? AND status = 'pending'", + (status, _now(), intent_id), + ) + if cursor.rowcount != 1: + raise KeyError(f"no pending intent with id {intent_id}") + + def pending_intents(self) -> list[Intent]: + rows = self._conn.execute( + "SELECT * FROM intent_log WHERE status = 'pending' ORDER BY id" + ).fetchall() + return [ + Intent( + id=row["id"], + kind=row["kind"], + payload=json.loads(row["payload"]), + status=row["status"], + created_at=row["created_at"], + resolved_at=row["resolved_at"], + ) + for row in rows + ] + + def close(self) -> None: + self._conn.close() + + def __enter__(self) -> Self: + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() diff --git a/nisse-dev/tests/test_reconcile.py b/nisse-dev/tests/test_reconcile.py new file mode 100644 index 0000000..cbd2eea --- /dev/null +++ b/nisse-dev/tests/test_reconcile.py @@ -0,0 +1,68 @@ +from pathlib import Path + +import pytest + +from nisse_dev.reconcile import reconcile_on_boot +from nisse_dev.state import Intent, StateStore + + +@pytest.fixture +def store(tmp_path: Path) -> StateStore: + return StateStore(tmp_path / "state.db") + + +def test_no_pending_intents_resumes(store: StateStore) -> None: + report = reconcile_on_boot(store, {}) + assert report.action == "resume" + assert report.notes == [] + + +def test_verified_effect_commits_intent_and_resumes(store: StateStore) -> None: + store.begin_intent("push_branch", {"branch": "m2"}) + report = reconcile_on_boot(store, {"push_branch": lambda intent: True}) + assert report.action == "resume" + assert store.pending_intents() == [] + + +def test_verified_no_effect_aborts_intent_and_resumes(store: StateStore) -> None: + store.begin_intent("push_branch", {"branch": "m2"}) + report = reconcile_on_boot(store, {"push_branch": lambda intent: False}) + assert report.action == "resume" + assert store.pending_intents() == [] + + +def test_unverifiable_intent_gates_to_await_input(store: StateStore) -> None: + store.begin_intent("push_branch", {"branch": "m2"}) + report = reconcile_on_boot(store, {"push_branch": lambda intent: None}) + assert report.action == "await_input" + assert [i.kind for i in store.pending_intents()] == ["push_branch"] + + +def test_unknown_intent_kind_gates_to_await_input(store: StateStore) -> None: + store.begin_intent("mystery", {}) + report = reconcile_on_boot(store, {}) + assert report.action == "await_input" + assert "no reconciler registered" in report.notes[0] + + +def test_reconciler_receives_the_intent(store: StateStore) -> None: + store.begin_intent("push_branch", {"branch": "m2"}) + seen: list[Intent] = [] + + def check(intent: Intent) -> bool: + seen.append(intent) + return True + + reconcile_on_boot(store, {"push_branch": check}) + assert seen[0].payload == {"branch": "m2"} + + +def test_mixed_verdicts_still_gate(store: StateStore) -> None: + store.begin_intent("push_branch", {}) + store.begin_intent("open_pr", {}) + report = reconcile_on_boot( + store, + {"push_branch": lambda intent: True, "open_pr": lambda intent: None}, + ) + assert report.action == "await_input" + assert [i.kind for i in store.pending_intents()] == ["open_pr"] diff --git a/nisse-dev/tests/test_state.py b/nisse-dev/tests/test_state.py new file mode 100644 index 0000000..4512647 --- /dev/null +++ b/nisse-dev/tests/test_state.py @@ -0,0 +1,118 @@ +import sqlite3 +from pathlib import Path + +import pytest + +from nisse_dev.state import StateSchemaError, StateStore + + +@pytest.fixture +def db(tmp_path: Path) -> Path: + return tmp_path / "state.db" + + +def test_wal_mode_and_full_sync(db: Path) -> None: + with StateStore(db) as store: + journal = store._conn.execute("PRAGMA journal_mode").fetchone()[0] + sync = store._conn.execute("PRAGMA synchronous").fetchone()[0] + assert journal == "wal" + assert sync == 2 # FULL + + +def test_fresh_store_stamps_schema_version(db: Path) -> None: + with StateStore(db, schema_version=1) as store: + assert store.schema_version == 1 + + +def test_transition_persists_across_reopen(db: Path) -> None: + with StateStore(db) as store: + store.transition("EXECUTING", project="demo", data={"milestone": 2}) + with StateStore(db) as store: + state = store.get_state() + assert state is not None + assert state.state == "EXECUTING" + assert state.project == "demo" + assert state.data == {"milestone": 2} + + +def test_get_state_on_fresh_store_is_none(db: Path) -> None: + with StateStore(db) as store: + assert store.get_state() is None + + +def test_intent_lifecycle(db: Path) -> None: + with StateStore(db) as store: + intent_id = store.begin_intent("push_branch", {"branch": "m2"}) + [pending] = store.pending_intents() + assert pending.id == intent_id + assert pending.kind == "push_branch" + assert pending.payload == {"branch": "m2"} + store.commit_intent(intent_id) + assert store.pending_intents() == [] + + +def test_pending_intent_survives_reopen(db: Path) -> None: + with StateStore(db) as store: + store.begin_intent("open_pr", {"title": "hello"}) + with StateStore(db) as store: + [pending] = store.pending_intents() + assert pending.kind == "open_pr" + + +def test_aborted_intent_is_not_pending(db: Path) -> None: + with StateStore(db) as store: + intent_id = store.begin_intent("push_branch", {}) + store.abort_intent(intent_id) + assert store.pending_intents() == [] + + +def test_resolving_nonpending_intent_raises(db: Path) -> None: + with StateStore(db) as store: + intent_id = store.begin_intent("push_branch", {}) + store.commit_intent(intent_id) + with pytest.raises(KeyError): + store.commit_intent(intent_id) + with pytest.raises(KeyError): + store.abort_intent(9999) + + +def test_migration_runs_and_bumps_version(db: Path) -> None: + with StateStore(db, schema_version=1): + pass + + def add_notes_table(conn: sqlite3.Connection) -> None: + conn.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT)") + + with StateStore(db, schema_version=2, migrations={1: add_notes_table}) as store: + assert store.schema_version == 2 + store._conn.execute("SELECT * FROM notes") + + +def test_missing_migration_raises(db: Path) -> None: + with StateStore(db, schema_version=1): + pass + with pytest.raises(StateSchemaError, match="no migration"): + StateStore(db, schema_version=2, migrations={}) + + +def test_failed_migration_leaves_version_unchanged(db: Path) -> None: + with StateStore(db, schema_version=1): + pass + + def broken(conn: sqlite3.Connection) -> None: + conn.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY)") + raise RuntimeError("boom") + + with pytest.raises(RuntimeError, match="boom"): + StateStore(db, schema_version=2, migrations={1: broken}) + with StateStore(db, schema_version=1) as store: + assert store.schema_version == 1 + with pytest.raises(sqlite3.OperationalError): + store._conn.execute("SELECT * FROM notes") + + +def test_newer_schema_is_refused(db: Path) -> None: + with StateStore(db, schema_version=3): + pass + with pytest.raises(StateSchemaError, match="newer"): + StateStore(db, schema_version=1) diff --git a/tests/integration/test_crash_durability.py b/tests/integration/test_crash_durability.py new file mode 100644 index 0000000..8972fbd --- /dev/null +++ b/tests/integration/test_crash_durability.py @@ -0,0 +1,47 @@ +"""Integration: state written before an abrupt process death must be intact +and the in-flight intent visible on the next boot (spec A2/A3).""" + +import subprocess +import sys +from pathlib import Path + +import pytest + +from nisse_dev.reconcile import reconcile_on_boot +from nisse_dev.state import StateStore + +pytestmark = pytest.mark.integration + +CRASH_SCRIPT = """ +import os +import sys +from nisse_dev.state import StateStore + +store = StateStore(sys.argv[1]) +store.transition("EXECUTING", project="demo", data={"milestone": 2}) +store.begin_intent("push_branch", {"branch": "milestone-2"}) +os._exit(1) # die without closing anything, as a power cut would +""" + + +def test_state_survives_process_death(tmp_path: Path) -> None: + db = tmp_path / "state.db" + result = subprocess.run( + [sys.executable, "-c", CRASH_SCRIPT, str(db)], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + assert result.returncode == 1, result.stderr + + with StateStore(db) as store: + state = store.get_state() + assert state is not None + assert state.state == "EXECUTING" + assert state.data == {"milestone": 2} + + report = reconcile_on_boot(store, {"push_branch": lambda intent: None}) + assert report.action == "await_input" + [pending] = store.pending_intents() + assert pending.payload == {"branch": "milestone-2"}