From ac81dd62344e327adf0a13b6e5eb170c6ee37e17 Mon Sep 17 00:00:00 2001 From: Don Garrett Date: Sun, 6 Sep 2026 19:51:58 +0000 Subject: [PATCH] Add canonical event log and renderer skeleton (A10) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Build-order step 4, sequenced before the LLM client (step 3) because the client interface consumes the canonical log. - canonical.py: typed events for the full A10 table (user_message through compaction); canonical tool-call IDs minted locally, provider IDs never a join key; strict payload validation on load - StateStore: canonical_events table via schema v2 — the first real migration, exercised in a test against a hand-built v1 database — with append_event/read_events on the same fsync-per-commit path - render.py: request bodies rendered from the log at call time; first adapter targets the OpenAI-compatible shape of LiteLLM's normalized route; compaction events replace their id range at render time while staying in the durable log - Crash integration test now also verifies transcript appends survive process death Co-Authored-By: Claude Fable 5 --- nisse-dev/src/nisse_dev/canonical.py | 133 +++++++++++++++++++++ nisse-dev/src/nisse_dev/render.py | 114 ++++++++++++++++++ nisse-dev/src/nisse_dev/state.py | 49 +++++++- nisse-dev/tests/test_canonical.py | 124 +++++++++++++++++++ nisse-dev/tests/test_render.py | 78 ++++++++++++ tests/integration/test_crash_durability.py | 5 + 6 files changed, 501 insertions(+), 2 deletions(-) create mode 100644 nisse-dev/src/nisse_dev/canonical.py create mode 100644 nisse-dev/src/nisse_dev/render.py create mode 100644 nisse-dev/tests/test_canonical.py create mode 100644 nisse-dev/tests/test_render.py diff --git a/nisse-dev/src/nisse_dev/canonical.py b/nisse-dev/src/nisse_dev/canonical.py new file mode 100644 index 0000000..95b729a --- /dev/null +++ b/nisse-dev/src/nisse_dev/canonical.py @@ -0,0 +1,133 @@ +"""Canonical transcript schema (spec A10). + +The durable record is this event log, never a provider-shaped ``messages`` +array; provider request bodies are rendered from it at call time. Canonical +tool-call IDs are minted locally — provider IDs are mapped at the boundary and +never used as a join key. + +Transcript appends are the conversation record itself, not state-machine +mutations, so they are deliberately outside the A9 state-change taxonomy. +""" + +import uuid +from dataclasses import asdict, dataclass, fields +from typing import Any + + +@dataclass(frozen=True) +class UserMessage: + text: str + source: str # e.g. a Matrix event ID + + +@dataclass(frozen=True) +class AssistantText: + text: str + model: str + + +@dataclass(frozen=True) +class ToolCall: + call_id: str # canonical, minted locally via mint_call_id() + tool: str + arguments: dict[str, Any] + + +@dataclass(frozen=True) +class ToolResult: + call_id: str + result: str + success: bool + + +@dataclass(frozen=True) +class PlanSubmitted: + plan: str + model: str + + +@dataclass(frozen=True) +class GateOpened: + gate: str + reason: str + + +@dataclass(frozen=True) +class DelegationStarted: + brief: str + role: str + model: str + caps: dict[str, Any] # turns / tokens / wall-clock limits + + +@dataclass(frozen=True) +class DelegationCompleted: + handback: str + artifacts: list[str] + tokens: int + duration_s: float + exit_status: str + + +@dataclass(frozen=True) +class Compaction: + summary: str + model: str + replaces: list[int] # [first_id, last_id] inclusive range replaced + + +CanonicalEvent = ( + UserMessage + | AssistantText + | ToolCall + | ToolResult + | PlanSubmitted + | GateOpened + | DelegationStarted + | DelegationCompleted + | Compaction +) + +EVENT_TYPES: dict[str, type] = { + "user_message": UserMessage, + "assistant_text": AssistantText, + "tool_call": ToolCall, + "tool_result": ToolResult, + "plan_submitted": PlanSubmitted, + "gate_opened": GateOpened, + "delegation_started": DelegationStarted, + "delegation_completed": DelegationCompleted, + "compaction": Compaction, +} + +_TYPE_NAMES = {cls: name for name, cls in EVENT_TYPES.items()} + + +@dataclass(frozen=True) +class StoredEvent: + id: int + ts: str + event: CanonicalEvent + + +def mint_call_id() -> str: + return f"call-{uuid.uuid4().hex}" + + +def event_type(event: CanonicalEvent) -> str: + return _TYPE_NAMES[type(event)] + + +def to_payload(event: CanonicalEvent) -> dict[str, Any]: + return asdict(event) + + +def from_payload(type_name: str, payload: dict[str, Any]) -> CanonicalEvent: + cls = EVENT_TYPES.get(type_name) + if cls is None: + raise ValueError(f"unknown canonical event type: {type_name!r}") + known = {f.name for f in fields(cls)} + unknown = payload.keys() - known + if unknown: + raise ValueError(f"unknown fields for {type_name}: {sorted(unknown)}") + return cls(**payload) diff --git a/nisse-dev/src/nisse_dev/render.py b/nisse-dev/src/nisse_dev/render.py new file mode 100644 index 0000000..21db51b --- /dev/null +++ b/nisse-dev/src/nisse_dev/render.py @@ -0,0 +1,114 @@ +"""A10 renderer: provider request bodies are built from the canonical log at +call time, never stored provider-shaped. + +First adapter: the OpenAI-compatible chat format LiteLLM's normalized route +speaks — the executor role's path (A5). The Anthropic passthrough adapter +arrives with delegations (A13). Token budgeting and prompt-cache breakpoints +are render-time concerns and are never stored (A10); neither exists yet in +this skeleton. +""" + +import json +from typing import Any + +from nisse_dev.canonical import ( + AssistantText, + Compaction, + DelegationCompleted, + DelegationStarted, + GateOpened, + PlanSubmitted, + StoredEvent, + ToolCall, + ToolResult, + UserMessage, +) + + +def effective_events(stored: list[StoredEvent]) -> list[StoredEvent]: + """Apply compactions: events whose ids fall in a compaction's replaced + range are dropped; the compaction event itself stays as their summary.""" + replaced: list[tuple[int, int]] = [ + (e.event.replaces[0], e.event.replaces[1]) + for e in stored + if isinstance(e.event, Compaction) + ] + + def is_replaced(event_id: int) -> bool: + return any(first <= event_id <= last for first, last in replaced) + + return [ + e for e in stored if isinstance(e.event, Compaction) or not is_replaced(e.id) + ] + + +class OpenAIChatRenderer: + """Mappings for orchestrator-side events (gates, plans, delegations, + compaction summaries) are provisional; they get tuned when the executor + loop (step 5) and delegations (step 7) consume them for real.""" + + def render( + self, + events: list[StoredEvent], + tools: list[dict[str, Any]] | None, + model: str, + system: str | None = None, + ) -> dict[str, Any]: + messages: list[dict[str, Any]] = [] + if system is not None: + messages.append({"role": "system", "content": system}) + for stored in effective_events(events): + messages.append(self._message_for(stored.event)) + body: dict[str, Any] = {"model": model, "messages": messages} + if tools: + body["tools"] = tools + return body + + def _message_for(self, event: object) -> dict[str, Any]: + match event: + case UserMessage(text=text): + return {"role": "user", "content": text} + case AssistantText(text=text): + return {"role": "assistant", "content": text} + case ToolCall(call_id=call_id, tool=tool, arguments=arguments): + return { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": call_id, + "type": "function", + "function": { + "name": tool, + "arguments": json.dumps(arguments), + }, + } + ], + } + case ToolResult(call_id=call_id, result=result): + return { + "role": "tool", + "tool_call_id": call_id, + "content": result, + } + case PlanSubmitted(plan=plan): + return {"role": "assistant", "content": plan} + case GateOpened(gate=gate, reason=reason): + return {"role": "user", "content": f"[gate {gate} opened: {reason}]"} + case DelegationStarted(brief=brief, model=model): + return { + "role": "assistant", + "content": f"[delegation started ({model}): {brief}]", + } + case DelegationCompleted(handback=handback): + return { + "role": "assistant", + "content": f"[delegation handback]\n{handback}", + } + case Compaction(summary=summary): + return { + "role": "user", + "content": f"[summary of earlier conversation]\n{summary}", + } + case _: + raise ValueError(f"unrenderable event: {event!r}") diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py index d112cec..978c0d5 100644 --- a/nisse-dev/src/nisse_dev/state.py +++ b/nisse-dev/src/nisse_dev/state.py @@ -18,14 +18,35 @@ from datetime import UTC, datetime from pathlib import Path from typing import Any, Self +from nisse_dev.canonical import ( + CanonicalEvent, + StoredEvent, + event_type, + from_payload, + to_payload, +) from nisse_dev.oplog import OpLog -CURRENT_SCHEMA_VERSION = 1 +CURRENT_SCHEMA_VERSION = 2 Migration = Callable[[sqlite3.Connection], None] +_CANONICAL_EVENTS_DDL = """ +CREATE TABLE canonical_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + type TEXT NOT NULL, + payload TEXT NOT NULL +) +""" + + +def _migrate_v1_add_canonical_events(conn: sqlite3.Connection) -> None: + conn.execute(_CANONICAL_EVENTS_DDL) + + # Maps version N to the migration that takes the schema from N to N+1. -MIGRATIONS: Mapping[int, Migration] = {} +MIGRATIONS: Mapping[int, Migration] = {1: _migrate_v1_add_canonical_events} class StateSchemaError(RuntimeError): @@ -71,6 +92,7 @@ _SCHEMA = [ resolved_at TEXT ) """, + _CANONICAL_EVENTS_DDL, ] @@ -249,6 +271,29 @@ class StateStore: for row in rows ] + def append_event(self, event: CanonicalEvent) -> int: + """Append to the canonical transcript (A10); committed and fsynced + before return, like every other durable write.""" + cursor = self._conn.execute( + "INSERT INTO canonical_events (ts, type, payload) VALUES (?, ?, ?)", + (_now(), event_type(event), json.dumps(to_payload(event))), + ) + assert cursor.lastrowid is not None + return cursor.lastrowid + + def read_events(self, after: int = 0) -> list[StoredEvent]: + rows = self._conn.execute( + "SELECT * FROM canonical_events WHERE id > ? ORDER BY id", (after,) + ).fetchall() + return [ + StoredEvent( + id=row["id"], + ts=row["ts"], + event=from_payload(row["type"], json.loads(row["payload"])), + ) + for row in rows + ] + def close(self) -> None: self._conn.close() diff --git a/nisse-dev/tests/test_canonical.py b/nisse-dev/tests/test_canonical.py new file mode 100644 index 0000000..1de85e9 --- /dev/null +++ b/nisse-dev/tests/test_canonical.py @@ -0,0 +1,124 @@ +import sqlite3 +from pathlib import Path + +import pytest + +from nisse_dev.canonical import ( + AssistantText, + Compaction, + DelegationCompleted, + DelegationStarted, + GateOpened, + PlanSubmitted, + ToolCall, + ToolResult, + UserMessage, + from_payload, + mint_call_id, +) +from nisse_dev.state import StateStore + +ALL_EVENTS = [ + UserMessage(text="use demo", source="matrix:$abc"), + AssistantText(text="starting", model="local/qwen"), + ToolCall(call_id="call-1", tool="run_tests", arguments={"path": "tests/"}), + ToolResult(call_id="call-1", result="3 passed", success=True), + PlanSubmitted(plan="the plan", model="claude-opus-5"), + GateOpened(gate="arch", reason="plan submitted (rev-1)"), + DelegationStarted( + brief="fix the flaky test", + role="delegate", + model="claude-opus-5", + caps={"turns": 20, "tokens": 100000}, + ), + DelegationCompleted( + handback="test fixed; root cause was a shared tmpdir", + artifacts=["nisse-dev/tests/test_x.py"], + tokens=51234, + duration_s=61.5, + exit_status="success", + ), + Compaction(summary="earlier work summarized", model="local/qwen", replaces=[1, 4]), +] + + +def test_all_event_types_round_trip(tmp_path: Path) -> None: + with StateStore(tmp_path / "state.db") as store: + for event in ALL_EVENTS: + store.append_event(event) + with StateStore(tmp_path / "state.db") as store: + stored = store.read_events() + assert [s.event for s in stored] == ALL_EVENTS + assert [s.id for s in stored] == sorted(s.id for s in stored) + + +def test_read_events_after_id(tmp_path: Path) -> None: + with StateStore(tmp_path / "state.db") as store: + first = store.append_event(UserMessage(text="one", source="m:1")) + store.append_event(UserMessage(text="two", source="m:2")) + tail = store.read_events(after=first) + assert [s.event.text for s in tail] == ["two"] + + +def test_mint_call_id_is_unique_and_prefixed() -> None: + ids = {mint_call_id() for _ in range(100)} + assert len(ids) == 100 + assert all(i.startswith("call-") for i in ids) + + +def test_unknown_event_type_rejected() -> None: + with pytest.raises(ValueError, match="unknown canonical event type"): + from_payload("mystery", {}) + + +def test_unknown_fields_rejected() -> None: + with pytest.raises(ValueError, match="unknown fields"): + from_payload("user_message", {"text": "hi", "source": "m", "extra": 1}) + + +# The v1 schema as it shipped (PR #2), frozen here to prove a newer image +# opens and upgrades a database written by the older one (A2). +V1_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 test_default_migration_upgrades_a_real_v1_store(tmp_path: Path) -> None: + db = tmp_path / "state.db" + conn = sqlite3.connect(db) + for statement in V1_SCHEMA: + conn.execute(statement) + conn.execute("INSERT INTO meta (key, value) VALUES ('schema_version', '1')") + conn.execute( + "INSERT INTO agent_state (id, state, project, data, updated_at) " + "VALUES (1, 'IDLE', NULL, '{}', 't')" + ) + conn.commit() + conn.close() + + with StateStore(db) as store: + assert store.schema_version == 2 + state = store.get_state() + assert state is not None + assert state.state == "IDLE" # v1 data intact + store.append_event(UserMessage(text="hello", source="m:1")) + assert len(store.read_events()) == 1 diff --git a/nisse-dev/tests/test_render.py b/nisse-dev/tests/test_render.py new file mode 100644 index 0000000..9cf7654 --- /dev/null +++ b/nisse-dev/tests/test_render.py @@ -0,0 +1,78 @@ +import json + +from nisse_dev.canonical import ( + AssistantText, + Compaction, + GateOpened, + StoredEvent, + ToolCall, + ToolResult, + UserMessage, +) +from nisse_dev.render import OpenAIChatRenderer, effective_events + + +def stored(events: list) -> list[StoredEvent]: + return [StoredEvent(id=i + 1, ts="t", event=e) for i, e in enumerate(events)] + + +def test_renders_mixed_transcript_in_order() -> None: + events = stored( + [ + UserMessage(text="run the tests", source="m:1"), + ToolCall(call_id="call-9", tool="run_tests", arguments={"path": "x"}), + ToolResult(call_id="call-9", result="1 passed", success=True), + AssistantText(text="all green", model="local/qwen"), + ] + ) + body = OpenAIChatRenderer().render(events, tools=None, model="local/qwen") + assert body["model"] == "local/qwen" + roles = [m["role"] for m in body["messages"]] + assert roles == ["user", "assistant", "tool", "assistant"] + call = body["messages"][1]["tool_calls"][0] + assert call["id"] == "call-9" + assert call["function"]["name"] == "run_tests" + assert json.loads(call["function"]["arguments"]) == {"path": "x"} + assert body["messages"][2]["tool_call_id"] == "call-9" + assert "tools" not in body + + +def test_system_prompt_and_tools_pass_through() -> None: + tools = [{"type": "function", "function": {"name": "run_tests"}}] + body = OpenAIChatRenderer().render( + stored([UserMessage(text="hi", source="m:1")]), + tools=tools, + model="local/qwen", + system="you are nisse-dev", + ) + assert body["messages"][0] == {"role": "system", "content": "you are nisse-dev"} + assert body["tools"] == tools + + +def test_compaction_replaces_its_range_only() -> None: + events = stored( + [ + UserMessage(text="old-1", source="m:1"), + AssistantText(text="old-2", model="q"), + UserMessage(text="kept", source="m:3"), + ] + ) + events.append( + StoredEvent( + id=4, + ts="t", + event=Compaction(summary="the old stuff", model="q", replaces=[1, 2]), + ) + ) + remaining = effective_events(events) + assert [e.id for e in remaining] == [3, 4] + body = OpenAIChatRenderer().render(events, tools=None, model="q") + contents = [m["content"] for m in body["messages"]] + assert contents == ["kept", "[summary of earlier conversation]\nthe old stuff"] + + +def test_gate_event_renders_as_context() -> None: + events = stored([GateOpened(gate="arch", reason="plan submitted (rev-1)")]) + body = OpenAIChatRenderer().render(events, tools=None, model="q") + assert body["messages"][0]["role"] == "user" + assert "arch" in body["messages"][0]["content"] diff --git a/tests/integration/test_crash_durability.py b/tests/integration/test_crash_durability.py index bcde3cf..f80bb13 100644 --- a/tests/integration/test_crash_durability.py +++ b/tests/integration/test_crash_durability.py @@ -15,11 +15,13 @@ pytestmark = pytest.mark.integration CRASH_SCRIPT = """ import os import sys +from nisse_dev.canonical import UserMessage from nisse_dev.state import StateStore store = StateStore(sys.argv[1]) store.transition("EXECUTING", cause="test", project="demo", data={"milestone": 2}) store.begin_intent("push_branch", {"branch": "milestone-2"}, cause="test") +store.append_event(UserMessage(text="start milestone 2", source="m:1")) os._exit(1) # die without closing anything, as a power cut would """ @@ -45,3 +47,6 @@ def test_state_survives_process_death(tmp_path: Path) -> None: assert report.action == "await_input" [pending] = store.pending_intents() assert pending.payload == {"branch": "milestone-2"} + + [stored] = store.read_events() + assert stored.event.text == "start milestone 2" -- 2.54.0