dongarandClaude Opus 5.5
a921b73d34
M4.3's block, with new triggers and a way out. A VM out of reach at the check before a run, in a run (which ends there: the driver, a toolchain step and a requested step), or while preparing the workspace blocks every project on it; the item stays ready, waiting on no one. On the intake round it is tried again every tuning.build_vm.retry_s (300), and once it answers it must still answer settle_s (180) later; then the work a lost run left there is parked (pushed), the VM unblocked, and that said once. A changed host key, at the check or on a retry, blocks the VM too, until the operator's `vm ready`, which trusts the new key: `trust new host key` is retired, and the message says it's a key problem that may be an impersonation. Questions are answered while a VM is blocked, without it. Starting the agent with a VM down no longer fails. The walk breaks the VM's SSH, now or partway through a run, and checks that no model works on a VM that is down; loop.qnt models the VM going and coming back (unblocked_only_when_up). Together they found: a lost run's work set aside by the next item there; a question left behind a block; a lost VM swallowed while the toolchain step checked or recorded it; and a remark waiting to be answered that outlived its PR's merge. All fixed. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk
639 lines
22 KiB
Python
639 lines
22 KiB
Python
"""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
|
|
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 = 5
|
|
|
|
# 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]
|
|
|
|
_VM_BLOCKS = """
|
|
CREATE TABLE vm_blocks (
|
|
host TEXT PRIMARY KEY,
|
|
cause TEXT NOT NULL,
|
|
reason TEXT NOT NULL,
|
|
work_item TEXT,
|
|
since TEXT NOT NULL
|
|
)
|
|
"""
|
|
|
|
|
|
def _add_vm_blocks(conn: sqlite3.Connection) -> None:
|
|
conn.execute(_VM_BLOCKS)
|
|
|
|
|
|
# Maps version N to the migration that takes the schema from N to N+1.
|
|
MIGRATIONS: Mapping[int, Migration] = {4: _add_vm_blocks}
|
|
|
|
# Who produced an event (L7): the operator, the orchestrator's own code, or a
|
|
# model role (L9). planner and executor are the old loop's (until M2.8), kept
|
|
# so its records still read.
|
|
ACTORS = frozenset(
|
|
{
|
|
"operator",
|
|
"orchestrator",
|
|
"planner",
|
|
"executor",
|
|
"driver",
|
|
"helper",
|
|
"expert",
|
|
"classifier",
|
|
"toolchain", # the toolchain step (D3): the driver's model, apart
|
|
}
|
|
)
|
|
|
|
|
|
class StateSchemaError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@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 VMBlock:
|
|
"""A build VM nothing may run on until it is recovered (D3): every
|
|
project on it waits."""
|
|
|
|
host: str
|
|
cause: str # broken (the driver said so), unreachable, host key (#137)
|
|
reason: str
|
|
work_item: str | None # the item that found it
|
|
since: str
|
|
|
|
|
|
@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
|
|
)
|
|
""",
|
|
_VM_BLOCKS,
|
|
]
|
|
|
|
|
|
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
|
|
)
|
|
|
|
# --- build VMs (D3) ----------------------------------------------------
|
|
|
|
def block_vm(
|
|
self, host: str, *, cause: str, reason: str, work_item: str | None
|
|
) -> None:
|
|
"""Block a build VM; a second block on it keeps the first."""
|
|
row = self._conn.execute(
|
|
"INSERT INTO vm_blocks (host, cause, reason, work_item, since) "
|
|
"VALUES (?, ?, ?, ?, ?) ON CONFLICT(host) DO NOTHING RETURNING host",
|
|
(host, cause, reason, work_item, _now()),
|
|
).fetchone()
|
|
if row is not None:
|
|
self.oplog.emit("vm.block", cause=cause, host=host, work_item=work_item)
|
|
|
|
def reblock_vm(self, host: str, *, cause: str, reason: str) -> None:
|
|
"""A blocked VM's block, for another cause (an unreachable VM that
|
|
answers with a new host key)."""
|
|
self._conn.execute(
|
|
"UPDATE vm_blocks SET cause = ?, reason = ? WHERE host = ?",
|
|
(cause, reason, host),
|
|
)
|
|
self.oplog.emit("vm.block", cause=cause, host=host)
|
|
|
|
def unblock_vm(self, host: str, *, cause: str) -> VMBlock | None:
|
|
"""Lift a build VM's block; returns it, or None if there was none."""
|
|
row = self._conn.execute(
|
|
"DELETE FROM vm_blocks WHERE host = ? RETURNING *", (host,)
|
|
).fetchone()
|
|
if row is None:
|
|
return None
|
|
self.oplog.emit("vm.unblock", cause=cause, host=host)
|
|
return VMBlock(**dict(row))
|
|
|
|
def vm_blocks(self) -> dict[str, VMBlock]:
|
|
rows = self._conn.execute("SELECT * FROM vm_blocks").fetchall()
|
|
return {row["host"]: VMBlock(**dict(row)) for row in rows}
|
|
|
|
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)
|
|
|
|
# --- 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
|
|
)
|