"""Durable state store (architecture L8). 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 import uuid from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any, Self from nisse_dev.canonical import ( CanonicalEvent, StoredEvent, UserMessage, event_type, from_payload, to_payload, ) from nisse_dev.oplog import OpLog from nisse_dev.views import project CURRENT_SCHEMA_VERSION = 4 # Stores older than this predate work items (Development Plan M1.1). There was # no production data to carry over, so there is no migration from them: start # a fresh store (scripts/bringup.sh start sets the old one aside). OLDEST_SUPPORTED_VERSION = 4 Migration = Callable[[sqlite3.Connection], None] # Maps version N to the migration that takes the schema from N to N+1. MIGRATIONS: Mapping[int, Migration] = {} IDLE = "IDLE" # Who produced an event (L7): the operator, the orchestrator's own code, or a # model role (L9). planner and executor become driver in the new loop (M2.3). ACTORS = frozenset( { "operator", "orchestrator", "planner", "executor", "driver", "helper", "expert", "classifier", } ) class StateSchemaError(RuntimeError): pass @dataclass(frozen=True) class AgentState: """The state machine's view of the active work item (control.py, until Development Plan M2.6 retires it).""" state: str project: str | None data: dict[str, Any] updated_at: str @dataclass(frozen=True) class WorkItem: id: str scope: str | None title: str status: str # open | closed outcome: str | None # completed | abandoned, once closed position: str # where the loop is for this item data: dict[str, Any] opened_at: str updated_at: str closed_at: str | None @dataclass(frozen=True) class Intent: id: int kind: str payload: dict[str, Any] status: str created_at: str resolved_at: str | None work_item: str | None = None _SCHEMA = [ "CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)", """ CREATE TABLE work_items ( id TEXT PRIMARY KEY, scope TEXT, title TEXT NOT NULL, status TEXT NOT NULL CHECK (status IN ('open', 'closed')), outcome TEXT, position TEXT NOT NULL, data TEXT NOT NULL DEFAULT '{}', opened_at TEXT NOT NULL, updated_at TEXT NOT NULL, closed_at TEXT ) """, """ CREATE TABLE agent ( id INTEGER PRIMARY KEY CHECK (id = 1), active_item TEXT REFERENCES work_items(id), idle_from_event INTEGER NOT NULL DEFAULT 0, 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, work_item TEXT ) """, """ CREATE TABLE canonical_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, type TEXT NOT NULL, payload TEXT NOT NULL, work_item TEXT, actor TEXT NOT NULL ) """, "CREATE INDEX canonical_events_by_item ON canonical_events (work_item, id)", """ CREATE TABLE eval_outcomes ( event_id INTEGER PRIMARY KEY, verdict TEXT NOT NULL, note TEXT NOT NULL DEFAULT '', marked_at TEXT NOT NULL ) """, ] 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, oplog: OpLog | None = None, ) -> None: # Log emission is owned by the store (L12): the same code path that # performs a durable write emits its log line, right after commit. self.oplog = oplog if oplog is not None else OpLog("nisse-dev") 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.closed = False 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 < OLDEST_SUPPORTED_VERSION <= self._target_version: raise StateSchemaError( f"state store schema version {stored} predates work items " f"(version {OLDEST_SUPPORTED_VERSION}); start a fresh store " "(scripts/bringup.sh start sets the old one aside)" ) 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)) self.oplog.emit( "schema.migrate", cause="startup", from_version=version, to_version=version + 1, ) 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"]) # --- work items (L1) --------------------------------------------------- def _ensure_agent_row(self) -> None: self._conn.execute( "INSERT INTO agent (id, active_item, updated_at) VALUES (1, NULL, ?) " "ON CONFLICT(id) DO NOTHING", (_now(),), ) @staticmethod def _item(row: sqlite3.Row) -> WorkItem: return WorkItem( id=row["id"], scope=row["scope"], title=row["title"], status=row["status"], outcome=row["outcome"], position=row["position"], data=json.loads(row["data"]), opened_at=row["opened_at"], updated_at=row["updated_at"], closed_at=row["closed_at"], ) def work_item(self, item_id: str) -> WorkItem | None: row = self._conn.execute( "SELECT * FROM work_items WHERE id = ?", (item_id,) ).fetchone() return None if row is None else self._item(row) def work_items(self, status: str | None = None) -> list[WorkItem]: query = "SELECT * FROM work_items" args: tuple[Any, ...] = () if status is not None: query += " WHERE status = ?" args = (status,) rows = self._conn.execute(query + " ORDER BY opened_at", args).fetchall() return [self._item(row) for row in rows] def active_item(self) -> WorkItem | None: row = self._conn.execute( "SELECT w.* FROM agent a JOIN work_items w ON w.id = a.active_item " "WHERE a.id = 1" ).fetchone() return None if row is None else self._item(row) def open_work_item( self, item_id: str, *, title: str, scope: str | None = None, position: str, data: dict[str, Any] | None = None, cause: str, ) -> WorkItem: now = _now() self._conn.execute( "INSERT INTO work_items (id, scope, title, status, position, data, " "opened_at, updated_at) VALUES (?, ?, ?, 'open', ?, ?, ?, ?)", (item_id, scope, title, position, json.dumps(data or {}), now, now), ) self.oplog.emit("work_item.open", cause=cause, work_item=item_id, scope=scope) item = self.work_item(item_id) assert item is not None return item def set_active(self, item_id: str | None, *, cause: str) -> None: """Point the agent at a work item, or at none (idle). Going idle records where in the log idleness began, so messages sent while idle can be carried into the next work item (carry_idle_messages).""" self._ensure_agent_row() previous = self._conn.execute( "SELECT active_item FROM agent WHERE id = 1" ).fetchone()["active_item"] if item_id is None: last = self._conn.execute( "SELECT COALESCE(MAX(id), 0) AS last FROM canonical_events" ).fetchone()["last"] self._conn.execute( "UPDATE agent SET active_item = NULL, idle_from_event = ?, " "updated_at = ? WHERE id = 1", (last, _now()), ) else: self._conn.execute( "UPDATE agent SET active_item = ?, updated_at = ? WHERE id = 1", (item_id, _now()), ) if previous != item_id: self.oplog.emit( "work_item.switch", cause=cause, **{"from": previous, "to": item_id}, ) def update_work_item( self, item_id: str, *, position: str, data: dict[str, Any], cause: str, ) -> None: """Record a work item's new position and data; the full data is passed every time (no partial updates).""" previous = self.work_item(item_id) if previous is None: raise KeyError(f"no work item {item_id!r}") self._conn.execute( "UPDATE work_items SET position = ?, data = ?, updated_at = ? WHERE id = ?", (position, json.dumps(data), _now(), item_id), ) self.oplog.emit( "state.transition", cause=cause, **{"from": previous.position, "to": position, "project": previous.scope}, ) def close_work_item(self, item_id: str, *, outcome: str, cause: str) -> None: now = _now() row = self._conn.execute( "UPDATE work_items SET status = 'closed', outcome = ?, closed_at = ?, " "updated_at = ? WHERE id = ? AND status = 'open' RETURNING id", (outcome, now, now, item_id), ).fetchone() if row is None: raise KeyError(f"no open work item {item_id!r}") self.oplog.emit( "work_item.close", cause=cause, work_item=item_id, outcome=outcome ) def carry_idle_messages(self, item_id: str) -> int: """Copy what the operator said while the agent was idle into a work item that has just opened (D1: messages sent before `use` are kept for it). Copied as new events, with the original's ID in the source, never by reassigning the originals: the log is never edited (L7).""" self._ensure_agent_row() since = self._conn.execute( "SELECT idle_from_event FROM agent WHERE id = 1" ).fetchone()["idle_from_event"] rows = self._conn.execute( "SELECT * FROM canonical_events WHERE work_item IS NULL " "AND actor = 'operator' AND type = 'user_message' AND id > ? ORDER BY id", (since,), ).fetchall() log = self.transcript(item_id, "operator") for row in rows: payload = json.loads(row["payload"]) log.append( UserMessage( text=payload["text"], source=f"{payload.get('source', '')} (sent while idle, #{row['id']})", ) ) return len(rows) # --- the state machine's view (control.py, until M2.6) ----------------- def get_state(self) -> AgentState | None: """The active work item as the state machine sees it: its position is the state, IDLE when there is none. None only for a store the agent has never started on.""" agent = self._conn.execute("SELECT * FROM agent WHERE id = 1").fetchone() if agent is None: return None item = self.active_item() if item is None: return AgentState(IDLE, None, {}, agent["updated_at"]) return AgentState(item.position, item.scope, item.data, item.updated_at) def transition( self, state: str, *, cause: str, project: str | None = None, data: dict[str, Any] | None = None, ) -> None: """The state machine's transition, over work items: IDLE closes and deactivates the active item (its outcome should already be recorded with close_work_item); any other state updates the active item, opening one if there is none.""" self._ensure_agent_row() active = self.active_item() if state == IDLE: if active is not None: if active.status == "open": self.close_work_item(active.id, outcome="closed", cause=cause) self.set_active(None, cause=cause) self.oplog.emit( "state.transition", cause=cause, **{ "from": active.position if active else None, "to": IDLE, "project": project, }, ) return if active is None: item_id = str((data or {}).get("task_id") or f"task-{uuid.uuid4().hex[:8]}") active = self.open_work_item( item_id, title=project or item_id, scope=project, position=IDLE, cause=cause, ) self.set_active(item_id, cause=cause) self.update_work_item(active.id, position=state, data=data or {}, cause=cause) # --- intents (L8) ------------------------------------------------------ def begin_intent( self, kind: str, payload: dict[str, Any], *, cause: str, work_item: str | None = None, ) -> int: """Durably record that a unit of work is about to start, owned by the work item it belongs to (L1).""" cursor = self._conn.execute( "INSERT INTO intent_log (kind, payload, created_at, work_item) " "VALUES (?, ?, ?, ?)", (kind, json.dumps(payload), _now(), work_item), ) assert cursor.lastrowid is not None self.oplog.emit( "intent.begin", cause=cause, intent_id=cursor.lastrowid, kind=kind ) return cursor.lastrowid def commit_intent(self, intent_id: int, *, cause: str) -> None: self._resolve_intent(intent_id, "committed", "intent.commit", cause) def abort_intent(self, intent_id: int, *, cause: str) -> None: self._resolve_intent(intent_id, "aborted", "intent.abort", cause) def _resolve_intent( self, intent_id: int, status: str, event: str, cause: str ) -> None: row = self._conn.execute( "UPDATE intent_log SET status = ?, resolved_at = ? " "WHERE id = ? AND status = 'pending' RETURNING kind", (status, _now(), intent_id), ).fetchone() if row is None: raise KeyError(f"no pending intent with id {intent_id}") self.oplog.emit(event, cause=cause, intent_id=intent_id, kind=row["kind"]) 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"], work_item=row["work_item"], ) for row in rows ] # --- the canonical log (L7) ------------------------------------------- def transcript(self, work_item: str | None, actor: str) -> "Transcript": """The log as one role sees it for one work item: what it reads is that item's events only, and what it writes carries the item and the actor. None is the agent-level scope (L6).""" if actor not in ACTORS: raise ValueError(f"unknown actor {actor!r}") return Transcript(self, work_item, actor) def all_events(self, after: int = 0) -> list[StoredEvent]: """Every event across work items, for the few readers whose job spans them (evaluation capture, dumps). Model calls never use this.""" rows = self._conn.execute( "SELECT * FROM canonical_events WHERE id > ? ORDER BY id", (after,) ).fetchall() return [_stored(row) for row in rows] def _append(self, event: CanonicalEvent, work_item: str | None, actor: str) -> int: cursor = self._conn.execute( "INSERT INTO canonical_events (ts, type, payload, work_item, actor) " "VALUES (?, ?, ?, ?, ?)", ( _now(), event_type(event), json.dumps(to_payload(event)), work_item, actor, ), ) assert cursor.lastrowid is not None return cursor.lastrowid def _read(self, work_item: str | None, after: int) -> list[StoredEvent]: rows = self._conn.execute( "SELECT * FROM canonical_events WHERE work_item IS ? AND id > ? " "ORDER BY id", (work_item, after), ).fetchall() return [_stored(row) for row in rows] def mark_outcome(self, event_id: int, verdict: str, note: str = "") -> None: """Record whether a delegation or consult was good (L9). Kept out of the canonical log on purpose: the log is an append-only record of what happened, and a verdict is a later judgement about it that may be revised. """ self._conn.execute( "INSERT INTO eval_outcomes (event_id, verdict, note, marked_at) " "VALUES (?, ?, ?, ?) ON CONFLICT(event_id) DO UPDATE SET " "verdict = excluded.verdict, note = excluded.note, " "marked_at = excluded.marked_at", (event_id, verdict, note, _now()), ) def outcomes(self) -> dict[int, tuple[str, str]]: rows = self._conn.execute( "SELECT event_id, verdict, note FROM eval_outcomes" ).fetchall() return {r["event_id"]: (r["verdict"], r["note"]) for r in rows} def close(self) -> None: self._conn.close() self.closed = True def __enter__(self) -> Self: return self def __exit__(self, *exc_info: object) -> None: self.close() def _stored(row: sqlite3.Row) -> StoredEvent: return StoredEvent( id=row["id"], ts=row["ts"], event=from_payload(row["type"], json.loads(row["payload"])), work_item=row["work_item"], actor=row["actor"], ) class Transcript: """One work item's log, as one actor reads and writes it (L6, L7). Handed to each component per call by the orchestrator, which knows the active work item, so a component can neither read another item's events nor write without attribution. """ def __init__(self, store: StateStore, work_item: str | None, actor: str) -> None: self._store = store self.work_item = work_item self.actor = actor def append(self, event: CanonicalEvent) -> int: """Committed and fsynced before return, like every durable write.""" return self._store._append(event, self.work_item, self.actor) def read(self, after: int = 0) -> list[StoredEvent]: """The log as recorded: every role's events, for the record.""" return self._store._read(self.work_item, after) def view(self, after: int = 0) -> list[StoredEvent]: """The log as this actor is shown it: what a model's context is built from (L6). See ``views``.""" return project(self.read(after), self.actor) def as_actor(self, actor: str) -> "Transcript": """The same work item, written by another role: e.g. the executor handing a sub-task to the helper.""" return self._store.transcript(self.work_item, actor) def begin_intent(self, kind: str, payload: dict[str, Any], *, cause: str) -> int: """An intent owned by this transcript's work item.""" return self._store.begin_intent( kind, payload, cause=cause, work_item=self.work_item )