Add durable state store and boot-time reconciliation (A2/A3)
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 <noreply@anthropic.com>
This commit is contained in:
1 parent
5962aaee57
commit
4319882f35
5 files changed
+516
No files matched your search
@@ -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,
|
||||
)
|
||||
@@ -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()
|
||||
Reference in new issue
Block a user