Add canonical event log and renderer skeleton (A10) #7

Merged
dongar merged 1 commits from canonical-log into main 2026-09-06 20:46:47 +00:00
6 changed files with 501 additions and 2 deletions
Showing only changes of commit ac81dd6234 - Show all commits

No files matched your search

+133
View File
@@ -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)
+114
View File
@@ -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}")
+47 -2
View File
@@ -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()
+124
View File
@@ -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
+78
View File
@@ -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"]
@@ -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"