dev/nisse
2
1 Fork 0
Code Issues 23 Pull Requests 1 Actions Packages Projects Releases 1 Wiki Activity

Put work items in the state store, and bind the log to them (M1.1)
CI / checks (push) Skipped
CI / checks (pull_request) Successful in 1m14s
Details

Browse Source 
Development Plan M1.1, as discussed.

- Schema v4: a work_items table (id, scope, title, status, outcome,
  position, data) and an agent row whose active_item points at one
  (NULL: idle). Canonical events and intents carry their work item, and
  events their actor. Older stores are refused with a clear message:
  there was no production data, so no migration (the mechanism stays).
- The log is reached through a Transcript bound to one work item and
  one actor: store.transcript(item, actor) reads only that item and
  writes with both. The planner, executor, delegation and expert desk
  get one per call from the orchestrator instead of holding the store,
  so a component can neither read another task's events (the planner
  read every earlier task's, #78) nor write without attribution.
  store.all_events() is kept for readers that span items (evaluation
  capture, dumps). Intents are begun through the transcript or with
  work_item=, so each belongs to its item.
- The orchestrator opens a work item on 'use', closes it with its
  outcome (completed or abandoned), and carries what the operator said
  while idle into the new item as copies with their origin, so a task
  sent before 'use' is still planned from (D1). control.py keeps its
  state machine through get_state()/transition(), now over the active
  work item, until M2.6 retires it.
- nisse_dev.dump shows work items and each event's item and actor.

Tests: test_work_items.py (scoping, attribution, intent ownership,
restart, outcomes, carried messages, a second task never seeing the
first); the suites updated for the per-call transcript.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
This commit is contained in:
dongarandClaude Opus 5.5 committed 2026-10-01 23:20:29 +00:00
1 parent 3cf4cefacd
commit 83804fe02a
40 files changed
+1026 -363
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

nisse-dev/src/nisse_dev/agent.py
+2 -5
View File
@@ -217,9 +217,7 @@ def build_agent(
transport = MatrixTransport(config.matrix)
announce = notify or transport.send_message
expert = ExpertDesk(
normalized, store, config.roles, budget=tuning.expert.consult_budget
)
expert = ExpertDesk(normalized, config.roles, budget=tuning.expert.consult_budget)
delegate_model = config.roles.resolve("helper")
delegation = Delegation(
select_client(delegate_model, normalized, passthrough, catalog),
@@ -237,7 +235,6 @@ def build_agent(
)
executor = LLMExecutor(
normalized,
store,
config.roles,
workspace,
expert=expert,
@@ -272,7 +269,7 @@ def build_agent(
orchestrator = Orchestrator(
store,
transport,
LLMPlanningAgent(normalized, store, config.roles, cancel=cancel),
LLMPlanningAgent(normalized, config.roles, cancel=cancel),
operators=config.operators,
roles=config.roles,
executor=executor,
nisse-dev/src/nisse_dev/bench/runner.py
+12 -5
View File
@@ -168,8 +168,16 @@ def run_task(
)
task.seed_into(workspace.root)
store = StateStore(root / "state.db", oplog=OpLog("nisse-bench", io.StringIO()))
store.append_event(PlanSubmitted(plan=task.plan, model="benchmark"))
store.append_event(UserMessage(text="start milestone 1", source="benchmark"))
# One work item per run, as the agent's own tasks are (L1).
store.open_work_item(
"bench", title=task.name, position="EXECUTING", cause="benchmark run"
)
store.transcript("bench", "planner").append(
PlanSubmitted(plan=task.plan, model="benchmark")
)
store.transcript("bench", "operator").append(
UserMessage(text="start milestone 1", source="benchmark")
)
meter = MeteringClient(client, facts)
trims_before = getattr(client, "trimmed_requests", 0)
@@ -177,7 +185,6 @@ def run_task(
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
executor = LLMExecutor(
meter,
store,
roles,
workspace,
max_turns=tuning.executor.max_turns,
@@ -185,14 +192,14 @@ def run_task(
)
started = time.monotonic()
try:
outcome = executor.run(1)
outcome = executor.run(store.transcript("bench", "executor"), 1)
status, detail = outcome.status, outcome.note
except Exception as exc: # noqa: BLE001 -- recorded, the benchmark goes on
status, detail = "error", repr(exc)
wall = time.monotonic() - started
success, check_output = task.run_check(workspace.root)
failed, invented = _count(store.read_events())
failed, invented = _count(store.transcript("bench", "executor").read())
store.close()
return RunRecord(
nisse-dev/src/nisse_dev/cancel.py
+3 -3
View File
@@ -9,7 +9,7 @@ record of a stop is written by the worker when the stopped work returns.
import threading
from nisse_dev.canonical import ToolCall, ToolResult
from nisse_dev.state import StateStore
from nisse_dev.state import Transcript
STOPPED_RESULT = "not run: the operator stopped the agent"
@@ -30,11 +30,11 @@ class Cancel:
def skip_calls(
store: StateStore, calls: list[ToolCall], reason: str = STOPPED_RESULT
transcript: Transcript, calls: list[ToolCall], reason: str = STOPPED_RESULT
) -> None:
"""Record calls left unrun. Every tool call needs a result for the
transcript to render again, and the result says what happened."""
for call in calls:
store.append_event(
transcript.append(
ToolResult(call_id=call.call_id, result=reason, success=False)
)
nisse-dev/src/nisse_dev/canonical.py
+4
View File
@@ -112,6 +112,10 @@ class StoredEvent:
id: int
ts: str
event: CanonicalEvent
# The work item it belongs to (None: agent-level, L6) and the role that
# produced it (L7). Rendering uses both.
work_item: str | None = None
actor: str = ""
def mint_call_id() -> str:
nisse-dev/src/nisse_dev/control.py
+47 -11
View File
@@ -35,7 +35,7 @@ from nisse_dev.git import GitError
from nisse_dev.gitea import GiteaError
from nisse_dev.llm import RoleMap
from nisse_dev.reconcile import Reconciler, reconcile_on_boot
from nisse_dev.state import StateStore
from nisse_dev.state import StateStore, Transcript
from nisse_dev.transport import InboundMessage, Transport
from nisse_dev.worker import InlineWorker, Worker
@@ -109,7 +109,7 @@ class MilestoneExecutor(Protocol):
lives in nisse_dev.executor (imported there, not here, to keep the
dependency pointing one way)."""
def run(self, milestone: int) -> Any: ...
def run(self, transcript: Transcript, milestone: int) -> Any: ...
class MilestonePublisher(Protocol):
@@ -149,6 +149,7 @@ class PlanBoard(Protocol):
class PlanningAgent(Protocol):
def plan(
self,
transcript: Transcript,
gate: str,
project: str,
feedback: str | None = None,
@@ -161,6 +162,7 @@ class StubAgent:
def plan(
self,
transcript: Transcript,
gate: str,
project: str,
feedback: str | None = None,
@@ -449,8 +451,19 @@ class Orchestrator:
except Exception as exc: # noqa: BLE001 -- nothing left to tell
traceback.print_exception(exc, file=sys.stderr)
def _log(self, actor: str) -> Transcript:
"""The active work item's transcript as one actor writes it; the
agent-level one when no work item is active (L6, L7)."""
item = self._store.active_item()
return self._store.transcript(item.id if item else None, actor)
def _close_item(self, outcome: str, cause: str) -> None:
item = self._store.active_item()
if item is not None and item.status == "open":
self._store.close_work_item(item.id, outcome=outcome, cause=cause)
def _record(self, msg: InboundMessage) -> None:
self._store.append_event(
self._log("operator").append(
UserMessage(
text=(msg.original or msg.text).strip(),
source=f"transport:{msg.sender}",
@@ -495,7 +508,7 @@ class Orchestrator:
# Every operator message enters the canonical transcript, so planning
# turns see feedback and answers as conversation rather than as
# out-of-band parameters (L7).
self._store.append_event(
self._log("operator").append(
UserMessage(
text=(msg.original or msg.text).strip(),
source=f"transport:{msg.sender}",
@@ -571,6 +584,23 @@ class Orchestrator:
}
self._project = project
self._store.oplog.task_id = task_id
# A work item of its own (L1). One left active by an unfinished task
# (reachable only through a lost pending item) is closed first.
replaced = self._store.active_item()
if replaced is not None:
self._close_item("abandoned", f"replaced by use {project}")
self._store.set_active(None, cause=f"replaced by use {project}")
self._store.open_work_item(
task_id,
title=project,
scope=project,
position=IDLE,
data=self._data,
cause=f"operator command: use {project}",
)
self._store.set_active(task_id, cause=f"operator command: use {project}")
# What the operator said before `use` is part of this task (D1).
self._store.carry_idle_messages(task_id)
if self._board is not None:
try:
self._data["board"] = {
@@ -594,7 +624,11 @@ class Orchestrator:
try:
with self._long_work():
action = self._agent.plan(
gate_name, self._project, feedback=feedback, answer=answer
self._log("planner"),
gate_name,
self._project,
feedback=feedback,
answer=answer,
)
except PlanningStopped as exc:
# Answering resumes planning with the answer, so the reason for
@@ -633,10 +667,10 @@ class Orchestrator:
AWAIT_ARCH_SIGNOFF if gate_name == "arch" else AWAIT_MILESTONE_SIGNOFF
)
roles = self._data.get("roles") or {}
self._store.append_event(
self._log("planner").append(
PlanSubmitted(plan=action.content, model=roles.get("planner", "stub"))
)
self._store.append_event(
self._log("orchestrator").append(
GateOpened(gate=gate_name, reason=f"plan submitted (rev-{rev})")
)
self._transition(await_state, cause=f"plan submitted ({gate_name} rev-{rev})")
@@ -747,7 +781,7 @@ class Orchestrator:
return
self._transport.send_message(f"executing milestone {milestone}")
with self._long_work():
outcome = self._executor.run(milestone)
outcome = self._executor.run(self._log("executor"), milestone)
if outcome.status == "stopped":
self._halt_for_stop(
what=f"milestone {milestone}",
@@ -826,7 +860,7 @@ class Orchestrator:
):
return # not taken: a watch attached later still sees it
self._take_comment(board_state, comment)
self._store.append_event(
self._log("operator").append(
UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}")
)
self._apply_gate_reply(
@@ -881,12 +915,12 @@ class Orchestrator:
self._save(f"answering review comment {comment.id}")
# The comment enters the transcript, so the executor sees review
# feedback as conversation rather than as an out-of-band instruction.
self._store.append_event(
self._log("operator").append(
UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}")
)
try:
with self._long_work():
outcome = self._executor.run(milestone)
outcome = self._executor.run(self._log("executor"), milestone)
except Exception:
# Reported to the operator by _settled: answered, if not well.
self._done_reviewing(pr)
@@ -1002,6 +1036,7 @@ class Orchestrator:
pr = self._data.get("pr") or {}
project = self._project
self.stop_watching()
self._close_item("abandoned", f"operator: abandon ({sender})")
self._data = {**_fresh_data()}
self._project = None
self._store.oplog.task_id = None
@@ -1084,6 +1119,7 @@ class Orchestrator:
elif command == "project complete":
board_state = self._data.get("board") or {}
self.stop_watching()
self._close_item("completed", "operator: project complete")
self._data = {**_fresh_data()}
self._project = None
self._store.oplog.task_id = None
nisse-dev/src/nisse_dev/delegation.py
+31 -20
View File
@@ -33,13 +33,14 @@ from nisse_dev.canonical import (
AssistantText,
DelegationCompleted,
DelegationStarted,
StoredEvent,
ToolCall,
ToolResult,
)
from nisse_dev.git import GitError, GitRepo
from nisse_dev.llm import LLMClient, LLMError, UpstreamUnavailable
from nisse_dev.reconcile import Reconciler
from nisse_dev.state import Intent, StateStore
from nisse_dev.state import Intent, StateStore, Transcript
from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, Workspace
DELEGATION_INTENT = "delegation"
@@ -184,12 +185,16 @@ class Delegation:
# a reason, and the next step is to talk about that reason.
self._cancel = cancel or Cancel()
def run(self, *, brief: str, branch: str, role: str = "helper") -> DelegationResult:
def run(
self, transcript: Transcript, *, brief: str, branch: str, role: str = "helper"
) -> DelegationResult:
"""Run a delegation in the work item whose transcript this is,
writing as the helper (L6, L7); its intent is owned by that item."""
started_at = self._now()
base_branch = self._git.current_branch()
# Recorded before the branch exists, so a crash anywhere after this
# leaves a durable statement of what was in flight (L8).
intent_id = self._store.begin_intent(
intent_id = transcript.begin_intent(
DELEGATION_INTENT,
{
"branch": branch,
@@ -210,7 +215,7 @@ class Delegation:
self._store.abort_intent(intent_id, cause=f"could not branch: {exc}")
return DelegationResult("failed", f"could not branch for delegation: {exc}")
self._store.append_event(
transcript.append(
DelegationStarted(
brief=brief,
role=role,
@@ -219,9 +224,9 @@ class Delegation:
backing_model=self._backing_model,
)
)
result = self._loop(brief, branch, started_at)
result = self._loop(transcript, brief, branch, started_at)
result.duration_s = self._now() - started_at
self._store.append_event(
transcript.append(
DelegationCompleted(
handback=result.handback.render() if result.handback else result.note,
artifacts=result.artifacts,
@@ -256,6 +261,7 @@ class Delegation:
"""
payload = intent.payload
branch = str(payload["branch"])
transcript = self._store.transcript(intent.work_item, "helper")
started_at = self._now()
try:
self._git.checkout_existing_branch(branch)
@@ -265,9 +271,11 @@ class Delegation:
)
return DelegationResult("failed", f"could not resume on {branch}: {exc}")
result = self._loop(str(payload.get("brief", "")), branch, started_at)
result = self._loop(
transcript, str(payload.get("brief", "")), branch, started_at
)
result.duration_s = self._now() - started_at
self._store.append_event(
transcript.append(
DelegationCompleted(
handback=result.handback.render() if result.handback else result.note,
artifacts=result.artifacts,
@@ -310,7 +318,9 @@ class Delegation:
return self.resume(intent)
return None
def _loop(self, brief: str, branch: str, started_at: float) -> DelegationResult:
def _loop(
self, transcript: Transcript, brief: str, branch: str, started_at: float
) -> DelegationResult:
system = (
f"{SYSTEM_PROMPT}\n\n{self._workspace.describe()}"
f"\n\nYour sub-task:\n{brief}"
@@ -340,7 +350,7 @@ class Delegation:
)
try:
produced = self._client.complete(
self._store.read_events(),
transcript.read(),
tools=DELEGATE_TOOLS,
model=self._model,
system=system,
@@ -380,7 +390,7 @@ class Delegation:
for event in produced:
# Checkpoint as it arrives, never batched at handback (L9):
# this is what makes the delegation resumable after a crash.
self._store.append_event(event)
transcript.append(event)
if isinstance(event, ToolCall):
calls.append(event)
@@ -400,7 +410,7 @@ class Delegation:
for index, call in enumerate(calls):
if self._cancel.requested:
skip_calls(self._store, calls[index:])
skip_calls(transcript, calls[index:])
return DelegationResult(
"stopped",
f"stopped by the operator during turn {turn + 1}",
@@ -415,7 +425,7 @@ class Delegation:
# and the file state are the whole interface back, so
# an unbacked claim would hand the caller a note about
# work that does not exist.
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result=(
@@ -429,7 +439,7 @@ class Delegation:
)
continue
if handback is None:
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result=(
@@ -440,7 +450,7 @@ class Delegation:
)
)
continue
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result="handback accepted; delegation closed",
@@ -463,7 +473,7 @@ class Delegation:
path = call.arguments.get("path")
if isinstance(path, str):
touched.add(path)
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result=outcome.result,
@@ -552,11 +562,12 @@ def delegation_verdict(
branch = intent.payload.get("branch")
if not branch:
return "gate"
if _completed_after_start(store):
# The delegation's own work item: an intent is owned by the item it ran in.
events = store.transcript(intent.work_item, "helper").read()
if _completed_after_start(events):
return "settled" # it finished; the intent merely outlived its commit
if not git_repo.branch_exists(str(branch)):
return "absent" # the crash landed before any work existed
events = store.read_events()
if not events:
return "gate"
last = events[-1].event
@@ -569,9 +580,9 @@ def delegation_verdict(
return "resume"
def _completed_after_start(store: StateStore) -> bool:
def _completed_after_start(events: list[StoredEvent]) -> bool:
"""True when the last delegation event in the log is a completion."""
for stored in reversed(store.read_events()):
for stored in reversed(events):
if isinstance(stored.event, DelegationCompleted):
return True
if isinstance(stored.event, DelegationStarted):
nisse-dev/src/nisse_dev/dump.py
+21 -13
View File
@@ -3,9 +3,10 @@
python -m nisse_dev.dump STATE_DB [--full]
Opens the store read-only (SQLite `mode=ro`), so it is safe to run against a
live agent's database: it never migrates, writes or locks it. Prints the
agent's state, pending intents, and the canonical log, one event per line;
long texts are shortened unless `--full` is given.
live agent's database: it never migrates, writes or locks it. Prints the work
items and which is active, pending intents, and the canonical log, one event
per line with its work item and actor; long texts are shortened unless
`--full` is given.
"""
import argparse
@@ -55,27 +56,34 @@ def _event_line(kind: str, payload: dict[str, Any], full: bool) -> str:
def render(path: Path, *, full: bool = False) -> str:
conn = _connect(path)
lines: list[str] = []
state = conn.execute("SELECT * FROM agent_state WHERE id = 1").fetchone()
if state is None:
lines.append("state: (none)")
else:
data = json.loads(state["data"])
agent = conn.execute("SELECT * FROM agent WHERE id = 1").fetchone()
active = agent["active_item"] if agent else None
lines.append(f"active work item: {active or '(none: idle)'}")
for item in conn.execute("SELECT * FROM work_items ORDER BY opened_at"):
data = json.loads(item["data"])
gate = data.get("gate") or {}
status = item["status"] + (f" ({item['outcome']})" if item["outcome"] else "")
lines.append(
f"state: {state['state']} project: {state['project']} "
f"task: {data.get('task_id')} gate: "
f" {item['id']} {status} position: {item['position']} "
f"scope: {item['scope']} gate: "
+ (f"{gate.get('name')} rev-{gate.get('rev')}" if gate else "none")
)
pending = conn.execute(
"SELECT id, kind, payload FROM intent_log WHERE status = 'pending'"
"SELECT id, kind, payload, work_item FROM intent_log WHERE status = 'pending'"
).fetchall()
for row in pending:
lines.append(f"pending intent #{row['id']} {row['kind']}: {row['payload']}")
lines.append(
f"pending intent #{row['id']} {row['kind']} "
f"[{row['work_item'] or 'agent'}]: {row['payload']}"
)
lines.append("")
for row in conn.execute("SELECT * FROM canonical_events ORDER BY id"):
payload = json.loads(row["payload"])
time = row["ts"][11:19]
lines.append(f"#{row['id']} {time} {_event_line(row['type'], payload, full)}")
where = f"[{row['work_item'] or 'agent'}/{row['actor']}]"
lines.append(
f"#{row['id']} {time} {where} {_event_line(row['type'], payload, full)}"
)
conn.close()
return "\n".join(lines)
nisse-dev/src/nisse_dev/evals.py
+1 -1
View File
@@ -46,7 +46,7 @@ def collect(store: StateStore, after: int = 0) -> list[EvalRecord]:
verdicts = store.outcomes()
records: list[EvalRecord] = []
pending: tuple[int, DelegationStarted] | None = None
for stored in store.read_events(after=after):
for stored in store.all_events(after=after):
if isinstance(stored.event, DelegationStarted):
pending = (stored.id, stored.event)
elif isinstance(stored.event, DelegationCompleted) and pending is not None:
nisse-dev/src/nisse_dev/executor.py
+24 -18
View File
@@ -19,7 +19,7 @@ from nisse_dev.canonical import PlanSubmitted, StoredEvent, ToolCall, ToolResult
from nisse_dev.delegation import Delegation
from nisse_dev.expert import ASK_EXPERT_TOOL, BudgetExhausted, ExpertDesk
from nisse_dev.llm import LLMClient, LLMError, RoleMap
from nisse_dev.state import StateStore
from nisse_dev.state import Transcript
from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, ToolOutcome, Workspace
EXECUTOR_SYSTEM = """You are nisse-dev's executor, doing mechanical work.
@@ -76,7 +76,6 @@ class LLMExecutor:
def __init__(
self,
client: LLMClient,
store: StateStore,
roles: RoleMap,
workspace: Workspace,
*,
@@ -88,7 +87,6 @@ class LLMExecutor:
cancel: Cancel | None = None,
) -> None:
self._client = client
self._store = store
self._roles = roles
self._workspace = workspace
self._max_turns = max_turns
@@ -113,7 +111,7 @@ class LLMExecutor:
self._notify(f"turn {turn + 1}: {summary}")
def _escalate(
self, milestone: int, failures: int, last_failure: str
self, transcript: Transcript, milestone: int, failures: int, last_failure: str
) -> ExecutorOutcome | None:
"""Hand the sub-task to a stronger model, per L9.
@@ -134,7 +132,9 @@ class LLMExecutor:
"above. Finish this sub-task, staying inside the approved plan."
)
result = self._delegate.run(
brief=brief, branch=f"delegation/milestone-{milestone}-{failures}"
transcript.as_actor("helper"),
brief=brief,
branch=f"delegation/milestone-{milestone}-{failures}",
)
if result.usable:
return None
@@ -153,13 +153,16 @@ class LLMExecutor:
return [*EXECUTOR_TOOLS, FINISH_TOOL]
return [*EXECUTOR_TOOLS, ASK_EXPERT_TOOL, FINISH_TOOL]
def _consult(self, arguments: dict[str, Any]) -> ToolOutcome:
def _consult(
self, transcript: Transcript, arguments: dict[str, Any]
) -> ToolOutcome:
assert self._expert is not None
question = str(arguments.get("question", "")).strip()
if not question:
return ToolOutcome("ask_expert needs a question", success=False)
try:
return ToolOutcome(self._expert.ask(question).render(), success=True)
answer = self._expert.ask(transcript.as_actor("expert"), question)
return ToolOutcome(answer.render(), success=True)
except BudgetExhausted as exc:
# Not an error the model should route around: it is the signal
# that it is circling and should get on with the work or fail.
@@ -167,7 +170,7 @@ class LLMExecutor:
except LLMError as exc:
return ToolOutcome(f"the expert did not answer: {exc}", success=False)
def _scope(self) -> tuple[str | None, int]:
def _scope(self, transcript: Transcript) -> tuple[str | None, int]:
"""The approved plan, and where this milestone's events begin.
The most recent plan in the log is the approved milestone plan: no
@@ -177,7 +180,7 @@ class LLMExecutor:
context window cannot afford and the plan already distils.
"""
latest: StoredEvent | None = None
for stored in self._store.read_events():
for stored in transcript.read():
if isinstance(stored.event, PlanSubmitted):
latest = stored
if latest is None:
@@ -185,9 +188,11 @@ class LLMExecutor:
assert isinstance(latest.event, PlanSubmitted)
return latest.event.plan, latest.id
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: Transcript, milestone: int) -> ExecutorOutcome:
"""Run a milestone of the work item whose transcript this is, writing
as the executor (L6, L7)."""
model = self._roles.resolve("executor")
plan, after = self._scope()
plan, after = self._scope(transcript)
system = (
f"{EXECUTOR_SYSTEM}\n\n{self._workspace.describe()}"
f"\n\nYou are executing milestone {milestone}."
@@ -209,7 +214,7 @@ class LLMExecutor:
return ExecutorOutcome("stopped", f"before turn {turn + 1}")
try:
produced = self._client.complete(
self._store.read_events(after=after),
transcript.read(after=after),
tools=tools,
model=model,
system=system,
@@ -219,7 +224,7 @@ class LLMExecutor:
calls: list[ToolCall] = []
for event in produced:
self._store.append_event(event)
transcript.append(event)
if isinstance(event, ToolCall):
calls.append(event)
@@ -235,7 +240,7 @@ class LLMExecutor:
self._report(turn, calls)
for index, call in enumerate(calls):
if self._cancel.requested:
skip_calls(self._store, calls[index:])
skip_calls(transcript, calls[index:])
return ExecutorOutcome("stopped", f"during turn {turn + 1}")
if call.tool == "finish":
if not did_work and not finish_refused:
@@ -248,7 +253,7 @@ class LLMExecutor:
"making any change",
)
else:
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result="milestone finished",
@@ -256,7 +261,7 @@ class LLMExecutor:
)
)
skip_calls(
self._store,
transcript,
calls[index + 1 :],
"not run: finish had already ended the milestone",
)
@@ -274,12 +279,12 @@ class LLMExecutor:
success=False,
)
elif call.tool == "ask_expert":
outcome = self._consult(call.arguments)
outcome = self._consult(transcript, call.arguments)
else:
if call.tool in MUTATING_TOOLS:
did_work = True
outcome = self._workspace.dispatch(call.tool, call.arguments)
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result=outcome.result,
@@ -298,6 +303,7 @@ class LLMExecutor:
# Escalating to the human is the fallback when there
# is no delegate, not the intended path.
escalation = self._escalate(
transcript,
milestone,
consecutive_failures,
outcome.result,
nisse-dev/src/nisse_dev/expert.py
+11 -9
View File
@@ -33,7 +33,7 @@ from nisse_dev.canonical import (
UserMessage,
)
from nisse_dev.llm import LLMClient, LLMError, RoleMap
from nisse_dev.state import StateStore
from nisse_dev.state import Transcript
EXPERT_ROLE = "expert"
DEFAULT_BUDGET = 3
@@ -97,32 +97,34 @@ class ExpertDesk:
def __init__(
self,
client: LLMClient,
store: StateStore,
roles: RoleMap,
*,
budget: int = DEFAULT_BUDGET,
) -> None:
self._client = client
self._store = store
self._roles = roles
self._budget = budget
def consults_used(self, since_event_id: int = 0) -> int:
def consults_used(self, transcript: Transcript, since_event_id: int = 0) -> int:
return sum(
1
for e in self._store.read_events(after=since_event_id)
for e in transcript.read(after=since_event_id)
if isinstance(e.event, DelegationStarted) and e.event.role == EXPERT_ROLE
)
def ask(self, question: str, *, since_event_id: int = 0) -> ExpertAnswer:
used = self.consults_used(since_event_id)
def ask(
self, transcript: Transcript, question: str, *, since_event_id: int = 0
) -> ExpertAnswer:
"""The consult is recorded in the work item's transcript, as the
expert, and the budget counts that work item's consults (L9)."""
used = self.consults_used(transcript, since_event_id)
if used >= self._budget:
raise BudgetExhausted(
f"consult budget of {self._budget} is spent for this task "
f"({used} used); escalate or ask the operator instead"
)
model = self._roles.resolve(EXPERT_ROLE)
self._store.append_event(
transcript.append(
DelegationStarted(
brief=question,
role=EXPERT_ROLE,
@@ -131,7 +133,7 @@ class ExpertDesk:
)
)
answer, note = self._answer(question, model)
self._store.append_event(
transcript.append(
DelegationCompleted(
handback=answer.render() if answer else note,
artifacts=[],
nisse-dev/src/nisse_dev/planner.py
+16 -12
View File
@@ -15,7 +15,7 @@ from nisse_dev.cancel import Cancel
from nisse_dev.canonical import AssistantText, ToolCall, ToolResult
from nisse_dev.control import AskHuman, PlanningError, PlanningStopped, SubmitPlan
from nisse_dev.llm import BudgetExhausted, LLMClient, LLMError, RoleMap
from nisse_dev.state import StateStore
from nisse_dev.state import Transcript
# Plans may be executed by a materially weaker model, which cannot
# reconstruct intent. This predates the architecture's capability profiles
@@ -83,14 +83,12 @@ class LLMPlanningAgent:
def __init__(
self,
client: LLMClient,
store: StateStore,
roles: RoleMap,
*,
system_prompt: str = SYSTEM_PROMPT,
cancel: Cancel | None = None,
) -> None:
self._client = client
self._store = store
self._roles = roles
self._system_prompt = system_prompt
# Checked before each turn only. A turn is one model call, and a turn
@@ -99,6 +97,7 @@ class LLMPlanningAgent:
def plan(
self,
transcript: Transcript,
gate: str,
project: str,
feedback: str | None = None,
@@ -106,18 +105,21 @@ class LLMPlanningAgent:
) -> SubmitPlan | AskHuman:
"""Operator feedback and answers are already durable in the canonical
log (the orchestrator appends every inbound message), so they are not
re-injected here; the log is the single source of context."""
re-injected here; the log is the single source of context. The
transcript is the work item's, written as the planner (L6, L7)."""
model = self._roles.resolve("planner")
instruction = (
f"{self._system_prompt}\n\n"
f"Current task: draft the {gate} plan for project {project!r}."
)
self._check_stop()
action, first_error = self._try_turn(model, instruction)
action, first_error = self._try_turn(transcript, model, instruction)
if action is not None:
return action
self._check_stop()
action, second_error = self._try_turn(model, f"{instruction}\n\n{_NUDGE}")
action, second_error = self._try_turn(
transcript, model, f"{instruction}\n\n{_NUDGE}"
)
if action is None:
detail = second_error or first_error
raise PlanningError(
@@ -131,13 +133,13 @@ class LLMPlanningAgent:
raise PlanningStopped("stopped by the operator")
def _try_turn(
self, model: str, system: str
self, transcript: Transcript, model: str, system: str
) -> tuple[SubmitPlan | AskHuman | None, str | None]:
"""A malformed or failed call is a bad turn, not a crash: the caller
retries once and then gates to the human (L9 — judgment about model
output is observable, never the model's own claim)."""
try:
return self._turn(model, system), None
return self._turn(transcript, model, system), None
except BudgetExhausted:
# Retrying cannot help: the budget will not reappear within a
# turn. Surface it now so the operator is told rather than the
@@ -146,8 +148,10 @@ class LLMPlanningAgent:
except LLMError as exc:
return None, str(exc)
def _turn(self, model: str, system: str) -> SubmitPlan | AskHuman | None:
events = self._store.read_events()
def _turn(
self, transcript: Transcript, model: str, system: str
) -> SubmitPlan | AskHuman | None:
events = transcript.read()
produced = self._client.complete(
events, tools=PLANNER_TOOLS, model=model, system=system
)
@@ -155,7 +159,7 @@ class LLMPlanningAgent:
for event in produced:
# Append before interpreting: the transcript must be truthful even
# if the call below rejects the arguments.
self._store.append_event(event)
transcript.append(event)
if isinstance(event, AssistantText) or action is not None:
continue
if isinstance(event, ToolCall):
@@ -164,7 +168,7 @@ class LLMPlanningAgent:
# The gate is this call's outcome. Recording it keeps the
# transcript renderable: providers reject a tool call with
# no result, and the next planning turn replays this log.
self._store.append_event(
transcript.append(
ToolResult(
call_id=event.call_id,
result=(
nisse-dev/src/nisse_dev/publish.py
+1
View File
@@ -82,6 +82,7 @@ class MilestonePublisher:
OPEN_PR_INTENT,
{"branch": branch, "milestone": milestone, "sha": sha},
cause=f"publishing milestone {milestone}",
work_item=task_id, # owned by the work item it publishes (L1)
)
self._git.push(self._remote, branch, token=self._token)
# Idempotent: if a PR for this branch is already open (a crash after
nisse-dev/src/nisse_dev/state.py
+380 -92
View File
@@ -12,6 +12,7 @@ resolved against external truth on boot).
import json
import sqlite3
import uuid
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
@@ -21,49 +22,41 @@ 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
CURRENT_SCHEMA_VERSION = 3
CURRENT_SCHEMA_VERSION = 4
# 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]
_CANONICAL_EVENTS_DDL = """
CREATE TABLE canonical_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts TEXT NOT NULL,
type TEXT NOT NULL,
payload TEXT NOT NULL
)
"""
_EVAL_OUTCOMES_DDL = """
CREATE TABLE eval_outcomes (
event_id INTEGER PRIMARY KEY,
verdict TEXT NOT NULL,
note TEXT NOT NULL DEFAULT '',
marked_at TEXT NOT NULL
)
"""
def _migrate_v1_add_canonical_events(conn: sqlite3.Connection) -> None:
conn.execute(_CANONICAL_EVENTS_DDL)
def _migrate_v2_add_eval_outcomes(conn: sqlite3.Connection) -> None:
conn.execute(_EVAL_OUTCOMES_DDL)
# Maps version N to the migration that takes the schema from N to N+1.
MIGRATIONS: Mapping[int, Migration] = {
1: _migrate_v1_add_canonical_events,
2: _migrate_v2_add_eval_outcomes,
}
MIGRATIONS: Mapping[int, Migration] = {}
IDLE = "IDLE"
# Who produced an event (L7): the operator, the orchestrator's own code, or a
# model role (L9). planner and executor become driver in the new loop (M2.3).
ACTORS = frozenset(
{
"operator",
"orchestrator",
"planner",
"executor",
"driver",
"helper",
"expert",
"classifier",
}
)
class StateSchemaError(RuntimeError):
@@ -72,12 +65,29 @@ class StateSchemaError(RuntimeError):
@dataclass(frozen=True)
class AgentState:
"""The state machine's view of the active work item (control.py, until
Development Plan M2.6 retires it)."""
state: str
project: str | None
data: dict[str, Any]
updated_at: str
@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 Intent:
id: int
@@ -86,16 +96,30 @@ class Intent:
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 agent_state (
id INTEGER PRIMARY KEY CHECK (id = 1),
state TEXT NOT NULL,
project TEXT,
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
)
""",
@@ -106,11 +130,29 @@ _SCHEMA = [
payload TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
created_at TEXT NOT NULL,
resolved_at TEXT
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
)
""",
_CANONICAL_EVENTS_DDL,
_EVAL_OUTCOMES_DDL,
]
@@ -150,6 +192,12 @@ class StateStore:
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 "
@@ -200,17 +248,177 @@ class StateStore:
).fetchone()
return int(row["value"])
def get_state(self) -> AgentState | None:
row = self._conn.execute("SELECT * FROM agent_state WHERE id = 1").fetchone()
if row is None:
return None
return AgentState(
state=row["state"],
project=row["project"],
data=json.loads(row["data"]),
updated_at=row["updated_at"],
# --- 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
)
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)
# --- the state machine's view (control.py, until M2.6) -----------------
def get_state(self) -> AgentState | None:
"""The active work item as the state machine sees it: its position
is the state, IDLE when there is none. None only for a store the
agent has never started on."""
agent = self._conn.execute("SELECT * FROM agent WHERE id = 1").fetchone()
if agent is None:
return None
item = self.active_item()
if item is None:
return AgentState(IDLE, None, {}, agent["updated_at"])
return AgentState(item.position, item.scope, item.data, item.updated_at)
def transition(
self,
state: str,
@@ -219,36 +427,55 @@ class StateStore:
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)
VALUES (1, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
state = excluded.state,
project = excluded.project,
data = excluded.data,
updated_at = excluded.updated_at
""",
(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,
},
)
"""The state machine's transition, over work items: IDLE closes and
deactivates the active item (its outcome should already be recorded
with close_work_item); any other state updates the active item,
opening one if there is none."""
self._ensure_agent_row()
active = self.active_item()
if state == IDLE:
if active is not None:
if active.status == "open":
self.close_work_item(active.id, outcome="closed", cause=cause)
self.set_active(None, cause=cause)
self.oplog.emit(
"state.transition",
cause=cause,
**{
"from": active.position if active else None,
"to": IDLE,
"project": project,
},
)
return
if active is None:
item_id = str((data or {}).get("task_id") or f"task-{uuid.uuid4().hex[:8]}")
active = self.open_work_item(
item_id,
title=project or item_id,
scope=project,
position=IDLE,
cause=cause,
)
self.set_active(item_id, cause=cause)
self.update_work_item(active.id, position=state, data=data or {}, cause=cause)
def begin_intent(self, kind: str, payload: dict[str, Any], *, cause: str) -> int:
"""Durably record that a unit of work is about to start."""
# --- 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) VALUES (?, ?, ?)",
(kind, json.dumps(payload), _now()),
"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(
@@ -286,32 +513,51 @@ class StateStore:
status=row["status"],
created_at=row["created_at"],
resolved_at=row["resolved_at"],
work_item=row["work_item"],
)
for row in rows
]
def append_event(self, event: CanonicalEvent) -> int:
"""Append to the canonical transcript (L7); committed and fsynced
before return, like every other durable write."""
# --- 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) VALUES (?, ?, ?)",
(_now(), event_type(event), json.dumps(to_payload(event))),
"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_events(self, after: int = 0) -> list[StoredEvent]:
def _read(self, work_item: str | None, after: int) -> list[StoredEvent]:
rows = self._conn.execute(
"SELECT * FROM canonical_events WHERE id > ? ORDER BY id", (after,)
"SELECT * FROM canonical_events WHERE work_item IS ? AND id > ? "
"ORDER BY id",
(work_item, after),
).fetchall()
return [
StoredEvent(
id=row["id"],
ts=row["ts"],
event=from_payload(row["type"], json.loads(row["payload"])),
)
for row in rows
]
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).
@@ -343,3 +589,45 @@ class StateStore:
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]:
return self._store._read(self.work_item, after)
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
)