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,
+    )

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()

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"]

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)

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"}