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 <noreply@anthropic.com>
This commit is contained in:
1 parent
ba848281ec
commit
18d4fcd701
7 files changed
+216
-32
No files matched your search
@@ -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()
|
||||
@@ -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
|
||||
@@ -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(
|
||||
|
||||
@@ -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) == []
|
||||
@@ -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},
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
"""
|
||||
|
||||
|
||||
Reference in new issue
Block a user