From 18d4fcd701664deb16a3300162ad6e1bf6af4b59 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sat, 5 Sep 2026 21:13:36 +0000 Subject: [PATCH 1/2] Implement A9 state-change logging in the state store Retrofit for the conventions merged in #4 (issue #3): - OpLog: JSON-lines emitter with the fixed envelope (ts, agent, event, task_id, cause); task_id set by the orchestrator at unit-of-work start - StateStore owns emission: transition, intent begin/commit/abort, and schema migrations each emit their event right after the durable write commits; callers pass the cause - reconcile_on_boot emits boot.reconcile with action and verdicts, and resolves intents with reconciliation-verdict causes - Failed resolutions emit nothing (a line exists iff the write committed) Co-Authored-By: Claude Fable 5 --- nisse-dev/src/nisse_dev/oplog.py | 33 ++++++ nisse-dev/src/nisse_dev/reconcile.py | 14 ++- nisse-dev/src/nisse_dev/state.py | 49 +++++++-- nisse-dev/tests/test_oplog.py | 112 +++++++++++++++++++++ nisse-dev/tests/test_reconcile.py | 14 +-- nisse-dev/tests/test_state.py | 22 ++-- tests/integration/test_crash_durability.py | 4 +- 7 files changed, 216 insertions(+), 32 deletions(-) create mode 100644 nisse-dev/src/nisse_dev/oplog.py create mode 100644 nisse-dev/tests/test_oplog.py diff --git a/nisse-dev/src/nisse_dev/oplog.py b/nisse-dev/src/nisse_dev/oplog.py new file mode 100644 index 0000000..6af9090 --- /dev/null +++ b/nisse-dev/src/nisse_dev/oplog.py @@ -0,0 +1,33 @@ +"""Structured operational logging (spec A9). + +One JSON object per line to stderr (journald captures it when running as a +service). Every event carries the fixed envelope: ts, agent, event, task_id, +cause. Logs are diagnostic, never load-bearing — nothing may read them back +to derive state. +""" + +import json +import sys +from datetime import UTC, datetime +from typing import Any, TextIO + + +class OpLog: + def __init__(self, agent: str, stream: TextIO | None = None) -> None: + self.agent = agent + # Correlation ID for the enclosing unit of work; the orchestrator sets + # it when a unit starts and clears it when the unit ends (A9). + self.task_id: str | None = None + self._stream = stream if stream is not None else sys.stderr + + def emit(self, event: str, *, cause: str, **fields: Any) -> None: + record: dict[str, Any] = { + "ts": datetime.now(UTC).isoformat(), + "agent": self.agent, + "event": event, + "task_id": self.task_id, + "cause": cause, + **fields, + } + self._stream.write(json.dumps(record) + "\n") + self._stream.flush() diff --git a/nisse-dev/src/nisse_dev/reconcile.py b/nisse-dev/src/nisse_dev/reconcile.py index af33076..15b34eb 100644 --- a/nisse-dev/src/nisse_dev/reconcile.py +++ b/nisse-dev/src/nisse_dev/reconcile.py @@ -39,15 +39,23 @@ def reconcile_on_boot( continue verdict = reconciler(intent) if verdict is True: - store.commit_intent(intent.id) + store.commit_intent( + intent.id, cause="reconciliation verdict: effect verified" + ) notes.append(f"{label}: external effect verified, marked committed") elif verdict is False: - store.abort_intent(intent.id) + store.abort_intent( + intent.id, cause="reconciliation verdict: never took effect" + ) notes.append(f"{label}: never took effect, marked aborted") else: unresolved = True notes.append(f"{label}: unverifiable") - return BootReport( + report = BootReport( action="await_input" if unresolved else "resume", notes=notes, ) + store.oplog.emit( + "boot.reconcile", cause="startup", action=report.action, verdicts=notes + ) + return report diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py index 57e5fd5..d112cec 100644 --- a/nisse-dev/src/nisse_dev/state.py +++ b/nisse-dev/src/nisse_dev/state.py @@ -18,6 +18,8 @@ from datetime import UTC, datetime from pathlib import Path from typing import Any, Self +from nisse_dev.oplog import OpLog + CURRENT_SCHEMA_VERSION = 1 Migration = Callable[[sqlite3.Connection], None] @@ -83,7 +85,11 @@ class StateStore: *, schema_version: int = CURRENT_SCHEMA_VERSION, migrations: Mapping[int, Migration] = MIGRATIONS, + oplog: OpLog | None = None, ) -> None: + # Log emission is owned by the store (A9): 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") @@ -115,6 +121,12 @@ class StateStore: 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: @@ -162,11 +174,13 @@ class StateStore: self, state: str, *, + cause: 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.""" + previous = self.get_state() self._conn.execute( """ INSERT INTO agent_state (id, state, project, data, updated_at) @@ -179,30 +193,45 @@ class StateStore: """, (state, project, json.dumps(data or {}), _now()), ) + self.oplog.emit( + "state.transition", + cause=cause, + **{ + "from": previous.state if previous else None, + "to": state, + "project": project, + }, + ) - def begin_intent(self, kind: str, payload: dict[str, Any]) -> int: + def begin_intent(self, kind: str, payload: dict[str, Any], *, cause: str) -> 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 + self.oplog.emit( + "intent.begin", cause=cause, intent_id=cursor.lastrowid, kind=kind + ) return cursor.lastrowid - def commit_intent(self, intent_id: int) -> None: - self._resolve_intent(intent_id, "committed") + 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) -> None: - self._resolve_intent(intent_id, "aborted") + 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) -> None: - cursor = self._conn.execute( + 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'", + "WHERE id = ? AND status = 'pending' RETURNING kind", (status, _now(), intent_id), - ) - if cursor.rowcount != 1: + ).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( diff --git a/nisse-dev/tests/test_oplog.py b/nisse-dev/tests/test_oplog.py new file mode 100644 index 0000000..5b57480 --- /dev/null +++ b/nisse-dev/tests/test_oplog.py @@ -0,0 +1,112 @@ +"""State-change log conventions (spec A9): every durable mutation emits one +JSON-lines event with the fixed envelope, from the store's own write path.""" + +import io +import json +import sqlite3 +from pathlib import Path + +import pytest + +from nisse_dev.oplog import OpLog +from nisse_dev.reconcile import reconcile_on_boot +from nisse_dev.state import StateStore + +ENVELOPE_FIELDS = {"ts", "agent", "event", "task_id", "cause"} + + +@pytest.fixture +def stream() -> io.StringIO: + return io.StringIO() + + +@pytest.fixture +def store(tmp_path: Path, stream: io.StringIO) -> StateStore: + return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", stream)) + + +def events(stream: io.StringIO) -> list[dict]: + return [json.loads(line) for line in stream.getvalue().splitlines()] + + +def test_every_event_carries_the_envelope( + store: StateStore, stream: io.StringIO +) -> None: + store.transition("PLANNING_ARCH", cause="operator command: use demo") + intent_id = store.begin_intent("push_branch", {}, cause="milestone execution") + store.commit_intent(intent_id, cause="push verified") + for event in events(stream): + assert ENVELOPE_FIELDS <= event.keys() + assert event["agent"] == "nisse-dev" + + +def test_transition_logs_from_to_and_cause( + store: StateStore, stream: io.StringIO +) -> None: + store.transition("PLANNING_ARCH", cause="operator command: use demo", project="d") + store.transition("AWAIT_ARCH_SIGNOFF", cause="plan submitted", project="d") + first, second = events(stream) + assert first["event"] == "state.transition" + assert (first["from"], first["to"]) == (None, "PLANNING_ARCH") + assert (second["from"], second["to"]) == ("PLANNING_ARCH", "AWAIT_ARCH_SIGNOFF") + assert second["cause"] == "plan submitted" + + +def test_intent_lifecycle_events(store: StateStore, stream: io.StringIO) -> None: + intent_id = store.begin_intent("push_branch", {"b": "m2"}, cause="milestone") + store.abort_intent(intent_id, cause="discarded") + begin, abort = events(stream) + assert begin["event"] == "intent.begin" + assert abort["event"] == "intent.abort" + assert begin["intent_id"] == abort["intent_id"] == intent_id + assert begin["kind"] == abort["kind"] == "push_branch" + + +def test_task_id_threads_through_events(store: StateStore, stream: io.StringIO) -> None: + store.transition("IDLE", cause="boot") + store.oplog.task_id = "task-42" + store.transition("PLANNING_ARCH", cause="operator command") + before, during = events(stream) + assert before["task_id"] is None + assert during["task_id"] == "task-42" + + +def test_migration_emits_event(tmp_path: Path, stream: io.StringIO) -> None: + db = tmp_path / "state.db" + with StateStore(db, schema_version=1): + pass + + def add_notes(conn: sqlite3.Connection) -> None: + conn.execute("CREATE TABLE notes (id INTEGER PRIMARY KEY)") + + with StateStore( + db, + schema_version=2, + migrations={1: add_notes}, + oplog=OpLog("nisse-dev", stream), + ): + pass + [event] = events(stream) + assert event["event"] == "schema.migrate" + assert (event["from_version"], event["to_version"]) == (1, 2) + + +def test_boot_reconcile_emits_action_and_verdicts( + store: StateStore, stream: io.StringIO +) -> None: + store.begin_intent("push_branch", {}, cause="milestone") + reconcile_on_boot(store, {"push_branch": lambda intent: True}) + boot = [e for e in events(stream) if e["event"] == "boot.reconcile"] + assert len(boot) == 1 + assert boot[0]["action"] == "resume" + assert "effect verified" in boot[0]["verdicts"][0] + commit = [e for e in events(stream) if e["event"] == "intent.commit"] + assert commit[0]["cause"] == "reconciliation verdict: effect verified" + + +def test_failed_resolution_emits_nothing( + store: StateStore, stream: io.StringIO +) -> None: + with pytest.raises(KeyError): + store.commit_intent(9999, cause="test") + assert events(stream) == [] diff --git a/nisse-dev/tests/test_reconcile.py b/nisse-dev/tests/test_reconcile.py index cbd2eea..a99b447 100644 --- a/nisse-dev/tests/test_reconcile.py +++ b/nisse-dev/tests/test_reconcile.py @@ -18,35 +18,35 @@ def test_no_pending_intents_resumes(store: StateStore) -> None: def test_verified_effect_commits_intent_and_resumes(store: StateStore) -> None: - store.begin_intent("push_branch", {"branch": "m2"}) + store.begin_intent("push_branch", {"branch": "m2"}, cause="test") 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"}) + store.begin_intent("push_branch", {"branch": "m2"}, cause="test") 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"}) + store.begin_intent("push_branch", {"branch": "m2"}, cause="test") 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", {}) + store.begin_intent("mystery", {}, cause="test") 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"}) + store.begin_intent("push_branch", {"branch": "m2"}, cause="test") seen: list[Intent] = [] def check(intent: Intent) -> bool: @@ -58,8 +58,8 @@ def test_reconciler_receives_the_intent(store: StateStore) -> None: def test_mixed_verdicts_still_gate(store: StateStore) -> None: - store.begin_intent("push_branch", {}) - store.begin_intent("open_pr", {}) + store.begin_intent("push_branch", {}, cause="test") + store.begin_intent("open_pr", {}, cause="test") report = reconcile_on_boot( store, {"push_branch": lambda intent: True, "open_pr": lambda intent: None}, diff --git a/nisse-dev/tests/test_state.py b/nisse-dev/tests/test_state.py index 4512647..29a61c6 100644 --- a/nisse-dev/tests/test_state.py +++ b/nisse-dev/tests/test_state.py @@ -26,7 +26,9 @@ def test_fresh_store_stamps_schema_version(db: Path) -> None: def test_transition_persists_across_reopen(db: Path) -> None: with StateStore(db) as store: - store.transition("EXECUTING", project="demo", data={"milestone": 2}) + store.transition( + "EXECUTING", cause="test", project="demo", data={"milestone": 2} + ) with StateStore(db) as store: state = store.get_state() assert state is not None @@ -42,18 +44,18 @@ def test_get_state_on_fresh_store_is_none(db: Path) -> None: def test_intent_lifecycle(db: Path) -> None: with StateStore(db) as store: - intent_id = store.begin_intent("push_branch", {"branch": "m2"}) + intent_id = store.begin_intent("push_branch", {"branch": "m2"}, cause="test") [pending] = store.pending_intents() assert pending.id == intent_id assert pending.kind == "push_branch" assert pending.payload == {"branch": "m2"} - store.commit_intent(intent_id) + store.commit_intent(intent_id, cause="test") 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"}) + store.begin_intent("open_pr", {"title": "hello"}, cause="test") with StateStore(db) as store: [pending] = store.pending_intents() assert pending.kind == "open_pr" @@ -61,19 +63,19 @@ def test_pending_intent_survives_reopen(db: Path) -> None: 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) + intent_id = store.begin_intent("push_branch", {}, cause="test") + store.abort_intent(intent_id, cause="test") 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) + intent_id = store.begin_intent("push_branch", {}, cause="test") + store.commit_intent(intent_id, cause="test") with pytest.raises(KeyError): - store.commit_intent(intent_id) + store.commit_intent(intent_id, cause="test") with pytest.raises(KeyError): - store.abort_intent(9999) + store.abort_intent(9999, cause="test") def test_migration_runs_and_bumps_version(db: Path) -> None: diff --git a/tests/integration/test_crash_durability.py b/tests/integration/test_crash_durability.py index 8972fbd..bcde3cf 100644 --- a/tests/integration/test_crash_durability.py +++ b/tests/integration/test_crash_durability.py @@ -18,8 +18,8 @@ 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"}) +store.transition("EXECUTING", cause="test", project="demo", data={"milestone": 2}) +store.begin_intent("push_branch", {"branch": "milestone-2"}, cause="test") os._exit(1) # die without closing anything, as a power cut would """ -- 2.54.0 From 7e89a791d2ff0f45ee46f51119d4dd5766c92c33 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sat, 5 Sep 2026 21:20:41 +0000 Subject: [PATCH 2/2] Add literal-format test showing exact log lines Freezes the clock and asserts the exact bytes emitted per event, so the on-disk log format is reviewable in the test file itself. Co-Authored-By: Claude Fable 5 --- nisse-dev/tests/test_oplog.py | 46 +++++++++++++++++++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/nisse-dev/tests/test_oplog.py b/nisse-dev/tests/test_oplog.py index 5b57480..5759ff3 100644 --- a/nisse-dev/tests/test_oplog.py +++ b/nisse-dev/tests/test_oplog.py @@ -104,6 +104,52 @@ def test_boot_reconcile_emits_action_and_verdicts( assert commit[0]["cause"] == "reconciliation verdict: effect verified" +def test_literal_log_lines_for_human_readability( + store: StateStore, stream: io.StringIO, monkeypatch: pytest.MonkeyPatch +) -> None: + """Asserts the exact bytes written per line, so the on-disk format is + visible here rather than only in a running system.""" + + class _FrozenDatetime: + @staticmethod + def now(tz: object) -> object: + from datetime import UTC, datetime + + return datetime(2026, 9, 5, 12, 0, 0, tzinfo=UTC) + + monkeypatch.setattr("nisse_dev.oplog.datetime", _FrozenDatetime) + + store.transition( + "PLANNING_ARCH", cause="operator command: use demo", project="demo" + ) + store.oplog.task_id = "demo-m2" + intent_id = store.begin_intent( + "push_branch", {"branch": "milestone-2"}, cause="milestone 2 execution" + ) + store.commit_intent(intent_id, cause="push verified by Gitea query") + + assert stream.getvalue().splitlines() == [ + ( + '{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",' + ' "event": "state.transition", "task_id": null,' + ' "cause": "operator command: use demo",' + ' "from": null, "to": "PLANNING_ARCH", "project": "demo"}' + ), + ( + '{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",' + ' "event": "intent.begin", "task_id": "demo-m2",' + ' "cause": "milestone 2 execution", "intent_id": 1,' + ' "kind": "push_branch"}' + ), + ( + '{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",' + ' "event": "intent.commit", "task_id": "demo-m2",' + ' "cause": "push verified by Gitea query",' + ' "intent_id": 1, "kind": "push_branch"}' + ), + ] + + def test_failed_resolution_emits_nothing( store: StateStore, stream: io.StringIO ) -> None: -- 2.54.0