Merge pull request 'Work items in the state store, with a transcript bound to each (M1.1)' (#87) from work-items into main
CI / checks (push) Successful in 1m15s

Reviewed-on: dongar/nisse#87
Reviewed-by: Don Garrett <dgarrett@acm.org>
This commit was merged in pull request #87.
This commit is contained in:
dongar committed 2026-10-01 23:36:36 +00:00
commit 51e144e26b
48 files changed
+1073 -431

No files matched your search

+2 -2
View File
@@ -99,7 +99,7 @@ test Matrix room, a budget-capped LiteLLM key; configuration in `~/nisse-e2e/`.
section table this plan carried for the purpose.
- Acceptance: `grep` finds no reference to the old spec or its section numbers.
**M0.2 A bring-up script.** *Pre-approved.*
**M0.2 A bring-up script.** *Pre-approved.* **Done.**
- `scripts/bringup.sh`: set aside the previous run's state, workspace and log;
export the ref under test; start the agent as a systemd unit, so it outlives the
shell that started it; collect the log and a readable transcript
@@ -115,7 +115,7 @@ These fix problems found in the live runs that **survive the loop change.** Issu
that only exist because of `control.py`'s state machine (#61, #79, #81) are folded
into Phase 2's design instead of being fixed twice.
**M1.1 Work items in the state store.** *Needs discussion* (schema).
**M1.1 Work items in the state store.** *Needs discussion* (schema). **Done.**
- A work-items table with an active pointer (L1); every canonical event carries its
work item and its actor (L7). A migration for existing stores.
- Acceptance: two work items back to back keep separate state and events; a
+2 -2
View File
@@ -25,8 +25,8 @@ llm: # L9: LiteLLM is the only LLM destination
roles: # L9: role -> LiteLLM model alias
executor: qwen3.8-27b # the local tier: best executor in the benchmark (PR #55)
planner: claude-opus-5-5 # high-end judgement: plans a weaker executor follows
delegate: claude-opus-5-5 # takes over what the executor got stuck on
consultant: claude-haiku-4-5 # one question at a time, no tools: cheap is enough
helper: claude-opus-5-5 # takes over what the executor got stuck on
expert: claude-haiku-4-5 # one question at a time, no tools: cheap is enough
matrix: # L5
homeserver: https://matrix.example
+4 -7
View File
@@ -37,7 +37,7 @@ from nisse_dev.worker import SerialWorker
# manager allows for a stop, and a hard exit mid-work is already safe.
SHUTDOWN_WAIT_S = 10.0
TOOL_DRIVING_ROLES = ("planner", "executor", "delegate")
TOOL_DRIVING_ROLES = ("planner", "executor", "helper")
def discover_models(config: AgentConfig, oplog: OpLog) -> ModelCatalog:
@@ -217,10 +217,8 @@ def build_agent(
transport = MatrixTransport(config.matrix)
announce = notify or transport.send_message
expert = ExpertDesk(
normalized, store, config.roles, budget=tuning.expert.consult_budget
)
delegate_model = config.roles.resolve("delegate")
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),
store,
@@ -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,
+13 -6
View File
@@ -168,16 +168,23 @@ 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)
model = facts.alias
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
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(
+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)
)
+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:
+5 -5
View File
@@ -36,8 +36,8 @@ REQUIRED_KEYS = (
"llm.token",
"roles.executor",
"roles.planner",
"roles.delegate",
"roles.consultant",
"roles.helper",
"roles.expert",
"matrix.homeserver",
"matrix.user_id",
"matrix.token",
@@ -312,8 +312,8 @@ class AgentConfig:
roles=RoleMap(
executor=str(values["roles.executor"]),
planner=str(values["roles.planner"]),
delegate=str(values["roles.delegate"]),
consultant=str(values["roles.consultant"]),
helper=str(values["roles.helper"]),
expert=str(values["roles.expert"]),
),
matrix=MatrixConfig(
homeserver=str(values["matrix.homeserver"]).rstrip("/"),
@@ -371,5 +371,5 @@ class AgentConfig:
"repo": self.gitea_repo,
"planner": self.roles.planner,
"executor": self.roles.executor,
"delegate": self.roles.delegate,
"helper": self.roles.helper,
}
+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
+29 -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"
@@ -185,13 +186,15 @@ class Delegation:
self._cancel = cancel or Cancel()
def run(
self, *, brief: str, branch: str, role: str = "delegate"
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,
@@ -212,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,
@@ -221,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,
@@ -258,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)
@@ -267,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,
@@ -312,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}"
@@ -342,7 +350,7 @@ class Delegation:
)
try:
produced = self._client.complete(
self._store.read_events(),
transcript.read(),
tools=DELEGATE_TOOLS,
model=self._model,
system=system,
@@ -382,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)
@@ -402,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}",
@@ -417,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=(
@@ -431,7 +439,7 @@ class Delegation:
)
continue
if handback is None:
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result=(
@@ -442,7 +450,7 @@ class Delegation:
)
)
continue
self._store.append_event(
transcript.append(
ToolResult(
call_id=call.call_id,
result="handback accepted; delegation closed",
@@ -465,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,
@@ -554,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
@@ -571,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):
+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)
+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:
+25 -19
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,21 +153,24 @@ 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.
return ToolOutcome(str(exc), success=False)
except LLMError as exc:
return ToolOutcome(f"the consultant did not answer: {exc}", success=False)
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,
+16 -15
View File
@@ -17,7 +17,7 @@ memory: it survives a restart, and it is observable from outside the model,
which is the same reasoning L9 applies to escalation triggers.
Consults are recorded as ``delegation_started`` / ``delegation_completed`` with
role ``consultant``. That is deliberate — L9 wants a "consult/delegation eval
role ``expert``. That is deliberate — L9 wants a "consult/delegation eval
set", and reusing the composite events gets consult evaluation for free rather
than inventing a parallel record.
"""
@@ -33,9 +33,9 @@ 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
CONSULTANT_ROLE = "consultant"
EXPERT_ROLE = "expert"
DEFAULT_BUDGET = 3
SYSTEM_PROMPT = """You are a consultant answering one self-contained question.
@@ -97,42 +97,43 @@ 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)
if isinstance(e.event, DelegationStarted)
and e.event.role == CONSULTANT_ROLE
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(CONSULTANT_ROLE)
self._store.append_event(
model = self._roles.resolve(EXPERT_ROLE)
transcript.append(
DelegationStarted(
brief=question,
role=CONSULTANT_ROLE,
role=EXPERT_ROLE,
model=model,
caps={"consults": self._budget, "used": used},
)
)
answer, note = self._answer(question, model)
self._store.append_event(
transcript.append(
DelegationCompleted(
handback=answer.render() if answer else note,
artifacts=[],
+3 -3
View File
@@ -109,10 +109,10 @@ class LLMClient(Protocol):
class RoleMap:
executor: str
planner: str
delegate: str
consultant: str
helper: str
expert: str
ROLES: ClassVar[tuple[str, ...]] = ("executor", "planner", "delegate", "consultant")
ROLES: ClassVar[tuple[str, ...]] = ("executor", "planner", "helper", "expert")
def resolve(self, role: str) -> str:
if role not in self.ROLES:
+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=(
+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
+372 -84
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()),
)
"""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": previous.state if previous else None,
"to": state,
"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
)
+2 -1
View File
@@ -30,6 +30,7 @@ class Planner:
def plan(
self,
transcript: object,
gate: str,
project: str,
feedback: str | None = None,
@@ -64,7 +65,7 @@ class Board:
class Executor:
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
return ExecutorOutcome("completed", "done")
+1 -1
View File
@@ -263,7 +263,7 @@ def test_a_tool_less_model_warns_but_still_boots(tmp_path: Path, monkeypatch) ->
logged = stream.getvalue()
assert "model_tools_unsupported" in logged
assert '"role": "executor"' in logged
assert '"role": "consultant"' not in logged # ask_expert drives no tools
assert '"role": "expert"' not in logged # ask_expert drives no tools
def test_the_resolved_model_is_pinned_at_task_start(
+1 -1
View File
@@ -109,7 +109,7 @@ def test_stale_feedback_is_recorded_but_rejects_nothing(tmp_path: Path) -> None:
assert gate(store)["rev"] == 2
assert "not taken as feedback" in transport.outbound[-1].text
said = [
e.event.text for e in store.read_events() if isinstance(e.event, UserMessage)
e.event.text for e in store.all_events() if isinstance(e.event, UserMessage)
]
assert "also, keep it in one file" in said # the planner still hears it
+2 -2
View File
@@ -20,7 +20,7 @@ from nisse_dev.planner import LLMPlanningAgent
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport
ROLES = RoleMap(executor="x", planner="y", delegate="y", consultant="y")
ROLES = RoleMap(executor="x", planner="y", helper="y", expert="y")
class FakeGitea:
@@ -73,7 +73,7 @@ def build(
orch = Orchestrator(
store,
transport,
LLMPlanningAgent(ScriptedClient(), store, ROLES),
LLMPlanningAgent(ScriptedClient(), ROLES),
operators={"don"},
roles=ROLES,
gitea_operators={"dongar"},
+16 -23
View File
@@ -16,7 +16,7 @@ from nisse_dev.canonical import (
from_payload,
mint_call_id,
)
from nisse_dev.state import CURRENT_SCHEMA_VERSION, StateStore
from nisse_dev.state import StateSchemaError, StateStore
ALL_EVENTS = [
UserMessage(text="use demo", source="matrix:$abc"),
@@ -27,7 +27,7 @@ ALL_EVENTS = [
GateOpened(gate="arch", reason="plan submitted (rev-1)"),
DelegationStarted(
brief="fix the flaky test",
role="delegate",
role="helper",
model="claude-opus-5",
caps={"turns": 20, "tokens": 100000},
),
@@ -45,18 +45,20 @@ ALL_EVENTS = [
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)
store.transcript("t", "executor").append(event)
with StateStore(tmp_path / "state.db") as store:
stored = store.read_events()
stored = store.transcript("t", "executor").read()
assert [s.event for s in stored] == ALL_EVENTS
assert [s.id for s in stored] == sorted(s.id for s in stored)
assert {(s.work_item, s.actor) for s in stored} == {("t", "executor")}
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)
log = store.transcript("t", "operator")
first = log.append(UserMessage(text="one", source="m:1"))
log.append(UserMessage(text="two", source="m:2"))
tail = log.read(after=first)
assert [s.event.text for s in tail] == ["two"]
@@ -102,26 +104,17 @@ V1_SCHEMA = [
]
def test_default_migration_upgrades_a_real_v1_store(tmp_path: Path) -> None:
def test_a_store_from_before_work_items_is_refused_clearly(tmp_path: Path) -> None:
# No production data predates work items, so there is no migration from
# these versions (Development Plan M1.1): the store must say so rather
# than open half-understood.
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.execute("INSERT INTO meta (key, value) VALUES ('schema_version', '3')")
conn.commit()
conn.close()
with StateStore(db) as store:
assert store.schema_version == CURRENT_SCHEMA_VERSION
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
# Two migrations were traversed, so both later tables must exist.
store.mark_outcome(1, "good", "arrived via v1 -> v3")
assert store.outcomes()[1] == ("good", "arrived via v1 -> v3")
with pytest.raises(StateSchemaError, match="predates work items"):
StateStore(db)
+3 -3
View File
@@ -130,13 +130,13 @@ def test_validate_names_every_bad_role_at_once() -> None:
{
"executor": "local-fast",
"planner": "clause-opus-5",
"delegate": "remote-architecture",
"consultant": "local-fast",
"helper": "remote-architecture",
"expert": "local-fast",
}
)
message = str(caught.value)
assert "planner='clause-opus-5'" in message
assert "delegate='remote-architecture'" in message
assert "helper='remote-architecture'" in message
assert "executor" not in message # the good ones are not noise
+4 -4
View File
@@ -16,8 +16,8 @@ COMPLETE: dict[str, Any] = {
"roles": {
"executor": "claude-haiku-4-5",
"planner": "claude-opus-5",
"delegate": "claude-opus-5",
"consultant": "claude-haiku-4-5",
"helper": "claude-opus-5",
"expert": "claude-haiku-4-5",
},
"matrix": {
"homeserver": "https://matrix.test/",
@@ -210,8 +210,8 @@ llm: {base_url: "https://litellm.test", token: "${TEST_LLM_TOKEN}"}
roles:
executor: local-fast
planner: claude-opus-5
delegate: claude-opus-5
consultant: claude-haiku-4-5
helper: claude-opus-5
expert: claude-haiku-4-5
matrix:
homeserver: https://matrix.test
user_id: "@nisse-dev-testing:test"
+10 -6
View File
@@ -248,7 +248,7 @@ def test_the_executor_sees_only_its_milestone(tmp_path: Path) -> None:
GateOpened(gate="milestone-1", reason="plan submitted"),
UserMessage(text="start milestone 1", source="t"),
):
store.append_event(event)
store.transcript("t", "executor").append(event)
class Workspace:
def describe(self) -> str:
@@ -262,8 +262,8 @@ def test_the_executor_sees_only_its_milestone(tmp_path: Path) -> None:
seen["system"] = system
return [AssistantText(text="done", model=model)]
roles = RoleMap(executor="m", planner="m", delegate="m", consultant="m")
LLMExecutor(Client(), store, roles, Workspace()).run(1)
roles = RoleMap(executor="m", planner="m", helper="m", expert="m")
LLMExecutor(Client(), roles, Workspace()).run(store.transcript("t", "executor"), 1)
texts = [getattr(e, "text", None) for e in seen["events"]]
assert "start milestone 1" in texts
@@ -278,7 +278,9 @@ def test_a_request_that_cannot_fit_stops_the_milestone(tmp_path: Path) -> None:
# Never sent: the executor stops and asks the operator, like any failed
# call, rather than handing a server a prompt it would silently cut.
store = StateStore(tmp_path / "state.db", oplog=OpLog("t", io.StringIO()))
store.append_event(UserMessage(text="start milestone 1", source="t"))
store.transcript("t", "operator").append(
UserMessage(text="start milestone 1", source="t")
)
class Workspace:
def describe(self) -> str:
@@ -286,8 +288,10 @@ def test_a_request_that_cannot_fit_stops_the_milestone(tmp_path: Path) -> None:
inner = RecordingClient()
client = BudgetedClient(inner, lambda m: (4096, 1024))
roles = RoleMap(executor="m", planner="m", delegate="m", consultant="m")
outcome = LLMExecutor(client, store, roles, Workspace()).run(1)
roles = RoleMap(executor="m", planner="m", helper="m", expert="m")
outcome = LLMExecutor(client, roles, Workspace()).run(
store.transcript("t", "executor"), 1
)
assert outcome.status == "escalate"
assert "fit" in outcome.note
+3 -1
View File
@@ -142,7 +142,7 @@ def test_a_message_is_matched_as_stripped_but_recorded_as_sent(
h.transport.receive("don", "approve", original="Nisse Dev: approve")
assert h.state == AWAIT_MILESTONE_SIGNOFF
recorded = [
e.event.text for e in h.store.read_events() if isinstance(e.event, UserMessage)
e.event.text for e in h.store.all_events() if isinstance(e.event, UserMessage)
]
assert "Nisse Dev: approve" in recorded
assert "approve" not in recorded
@@ -160,6 +160,7 @@ def test_status_reports_state_and_gate(h: Harness) -> None:
class QuestionThenPlan:
def plan(
self,
transcript: object,
gate: str,
project: str,
feedback: str | None = None,
@@ -192,6 +193,7 @@ class FailsOnce:
def plan(
self,
transcript: object,
gate: str,
project: str,
feedback: str | None = None,
+53 -41
View File
@@ -19,12 +19,13 @@ from nisse_dev.delegation import (
DELEGATION_INTENT,
Caps,
Delegation,
DelegationResult,
delegation_reconciler,
)
from nisse_dev.git import GitRepo
from nisse_dev.llm import BudgetExhausted, LLMError, UpstreamUnavailable
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.state import StateStore, Transcript
from nisse_dev.tools import Workspace
@@ -86,8 +87,34 @@ def ws(repo: GitRepo) -> Workspace:
return Workspace(repo.root, timeout_s=10)
ITEM = "t" # the work item these delegations run in
def log(store: StateStore) -> Transcript:
return store.transcript(ITEM, "helper")
class Bound:
"""A delegation run in the test's work item, as the executor runs it."""
def __init__(self, delegation: Delegation, store: StateStore) -> None:
self.delegation = delegation
self.store = store
def run(self, **kw: Any) -> DelegationResult:
return self.delegation.run(log(self.store), **kw)
def resume(self, intent: Any) -> DelegationResult:
return self.delegation.resume(intent)
def recover(self, notify: Any) -> DelegationResult | None:
return self.delegation.recover(notify)
def build(client: Any, store: StateStore, ws: Workspace, repo: GitRepo, **kw: Any):
return Delegation(client, store, ws, repo, model="claude-haiku-4-5", **kw)
return Bound(
Delegation(client, store, ws, repo, model="claude-haiku-4-5", **kw), store
)
def test_successful_delegation_returns_a_structured_handback(
@@ -169,7 +196,7 @@ def test_stop_discards_the_branch_and_settles_the_intent(
# The unrun call still has a result, so the transcript renders.
skipped = [
e.event
for e in store.read_events()
for e in log(store).read()
if isinstance(e.event, ToolResult) and e.event.result == STOPPED_RESULT
]
assert len(skipped) == 1
@@ -225,7 +252,7 @@ def test_events_are_checkpointed_as_they_arrive(
)
build(client, store, ws, repo).run(brief="add a", branch="d")
kinds = [type(e.event).__name__ for e in store.read_events()]
kinds = [type(e.event).__name__ for e in log(store).read()]
assert kinds == [
"DelegationStarted",
"ToolCall",
@@ -245,7 +272,7 @@ def test_composite_events_record_the_brief_caps_and_outcome(
caps = Caps(turns=7, tokens=1234, wall_clock_s=42.0)
build(client, store, ws, repo, caps=caps).run(brief="the brief", branch="d")
events = [e.event for e in store.read_events()]
events = [e.event for e in log(store).read()]
started = next(e for e in events if isinstance(e, DelegationStarted))
completed = next(e for e in events if isinstance(e, DelegationCompleted))
assert started.brief == "the brief"
@@ -356,6 +383,7 @@ def pending_delegation_intent(store: StateStore, branch: str = "d") -> Any:
DELEGATION_INTENT,
{"branch": branch, "base": "main", "brief": "finish it"},
cause="test",
work_item=ITEM, # owned by the work item it ran in
)
return store.pending_intents()[-1]
@@ -385,11 +413,9 @@ def test_clean_cut_after_a_tool_result_is_resumable(
repo.checkout_new_branch("d")
(repo.root / "partial.py").write_text("x")
repo.commit_all("partial work")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(ToolCall(call_id="c", tool="write_file", arguments={}))
store.append_event(ToolResult(call_id="c", result="wrote", success=True))
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(ToolCall(call_id="c", tool="write_file", arguments={}))
log(store).append(ToolResult(call_id="c", result="wrote", success=True))
intent = pending_delegation_intent(store)
assert delegation_reconciler(store, repo)(intent) is True
@@ -400,10 +426,8 @@ def test_dangling_tool_call_gates(
) -> None:
# The effect of the call is unknown and not verifiable by inspection.
repo.checkout_new_branch("d")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(ToolCall(call_id="c", tool="run_command", arguments={}))
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(ToolCall(call_id="c", tool="run_command", arguments={}))
intent = pending_delegation_intent(store)
assert delegation_reconciler(store, repo)(intent) is None
@@ -412,10 +436,8 @@ def test_dangling_tool_call_gates(
def test_missing_branch_means_nothing_started(
store: StateStore, ws: Workspace, repo: GitRepo
) -> None:
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(ToolResult(call_id="c", result="ok", success=True))
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(ToolResult(call_id="c", result="ok", success=True))
intent = pending_delegation_intent(store, branch="never-created")
assert delegation_reconciler(store, repo)(intent) is False
@@ -425,10 +447,8 @@ def test_a_finished_delegation_settles_rather_than_resumes(
store: StateStore, ws: Workspace, repo: GitRepo
) -> None:
repo.checkout_new_branch("d")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(
DelegationCompleted(
handback="done",
artifacts=[],
@@ -448,10 +468,8 @@ def test_resume_continues_the_same_delegation(
# Interrupt after one turn, then resume: no second delegation_started,
# because L7 defines a session as the delegation, not the process.
repo.checkout_new_branch("d")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(ToolResult(call_id="c", result="wrote a.py", success=True))
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(ToolResult(call_id="c", result="wrote a.py", success=True))
intent = pending_delegation_intent(store)
client = ScriptedClient(
@@ -460,7 +478,7 @@ def test_resume_continues_the_same_delegation(
result = build(client, store, ws, repo).resume(intent)
assert result.usable
starts = [e for e in store.read_events() if isinstance(e.event, DelegationStarted)]
starts = [e for e in log(store).read() if isinstance(e.event, DelegationStarted)]
assert len(starts) == 1
assert store.pending_intents() == []
assert (repo.root / "b.py").exists()
@@ -484,10 +502,8 @@ def test_recover_resumes_a_clean_cut_with_a_notice(
store: StateStore, ws: Workspace, repo: GitRepo
) -> None:
repo.checkout_new_branch("d")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(ToolResult(call_id="c", result="wrote a.py", success=True))
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(ToolResult(call_id="c", result="wrote a.py", success=True))
pending_delegation_intent(store)
notices: list[str] = []
@@ -505,10 +521,8 @@ def test_recover_settles_a_finished_delegation_without_resuming(
store: StateStore, ws: Workspace, repo: GitRepo
) -> None:
repo.checkout_new_branch("d")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(
DelegationCompleted(
handback="done",
artifacts=[],
@@ -528,7 +542,7 @@ def test_recover_settles_a_finished_delegation_without_resuming(
def test_recover_discards_an_intent_whose_branch_never_existed(
store: StateStore, ws: Workspace, repo: GitRepo
) -> None:
store.append_event(ToolResult(call_id="c", result="ok", success=True))
log(store).append(ToolResult(call_id="c", result="ok", success=True))
pending_delegation_intent(store, branch="never-made")
assert build(ScriptedClient([]), store, ws, repo).recover(lambda _m: None) is None
@@ -539,10 +553,8 @@ def test_recover_leaves_an_unclean_cut_for_the_boot_gate(
store: StateStore, ws: Workspace, repo: GitRepo
) -> None:
repo.checkout_new_branch("d")
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
)
store.append_event(ToolCall(call_id="c", tool="run_command", arguments={}))
log(store).append(DelegationStarted(brief="b", role="helper", model="m", caps={}))
log(store).append(ToolCall(call_id="c", tool="run_command", arguments={}))
pending_delegation_intent(store)
assert build(ScriptedClient([]), store, ws, repo).recover(lambda _m: None) is None
@@ -597,7 +609,7 @@ def test_tokens_spent_are_reported_not_left_at_zero(
assert result.tokens == 2000 # two calls, 1000 each
completed = next(
e.event for e in store.read_events() if isinstance(e.event, DelegationCompleted)
e.event for e in log(store).read() if isinstance(e.event, DelegationCompleted)
)
# L9 evaluates cost against outcome, which needs a real number.
assert completed.tokens == 2000
+21 -12
View File
@@ -9,34 +9,43 @@ from nisse_dev.state import StateStore
def store_with_a_run(path: Path) -> None:
store = StateStore(path, oplog=OpLog("nisse-dev", io.StringIO()))
store.transcript(None, "operator").append(
UserMessage(text="hello before use", source="matrix")
)
store.transition(
"AWAIT_ARCH_SIGNOFF",
cause="test",
project="demo",
data={"task_id": "demo-1", "gate": {"name": "arch", "rev": 1}},
)
store.append_event(UserMessage(text="Add a truncate helper", source="matrix"))
store.append_event(PlanSubmitted(plan="x" * 500, model="opus"))
store.append_event(
log = store.transcript("demo-1", "operator")
log.append(UserMessage(text="Add a truncate helper", source="matrix"))
log.as_actor("planner").append(PlanSubmitted(plan="x" * 500, model="opus"))
executor = log.as_actor("executor")
executor.append(
ToolCall(call_id="c1", tool="run_command", arguments={"command": "ls"})
)
store.append_event(ToolResult(call_id="c1", result="README.md", success=False))
store.begin_intent("open_pr", {"branch": "b"}, cause="test")
executor.append(ToolResult(call_id="c1", result="README.md", success=False))
log.begin_intent("open_pr", {"branch": "b"}, cause="test")
store.close()
def test_the_dump_shows_state_intents_and_each_event(tmp_path: Path) -> None:
def test_the_dump_shows_items_intents_and_each_event_with_its_actor(
tmp_path: Path,
) -> None:
db = tmp_path / "state.db"
store_with_a_run(db)
text = render(db)
assert "active work item: demo-1" in text
assert (
"state: AWAIT_ARCH_SIGNOFF project: demo task: demo-1 gate: arch rev-1"
in text
"demo-1 open position: AWAIT_ARCH_SIGNOFF scope: demo gate: arch rev-1"
in (text)
)
assert "pending intent #1 open_pr" in text
assert "USER [matrix]: Add a truncate helper" in text
assert "PLAN (opus, 500 chars): " in text and "[340 more]" in text
assert 'CALL run_command {"command": "ls"}' in text
assert "pending intent #1 open_pr [demo-1]" in text
assert "[agent/operator] USER [matrix]: hello before use" in text
assert "[demo-1/operator] USER [matrix]: Add a truncate helper" in text
assert "[demo-1/planner] PLAN (opus, 500 chars): " in text and "[340 more]" in text
assert '[demo-1/executor] CALL run_command {"command": "ls"}' in text
assert "result (FAILED): README.md" in text
+14 -14
View File
@@ -24,13 +24,13 @@ def record_pair(
*,
brief: str = "add greet",
model: str = "claude-haiku-4-5",
role: str = "delegate",
role: str = "helper",
status: str = "handback",
) -> int:
store.append_event(
store.transcript("t", "helper").append(
DelegationStarted(brief=brief, role=role, model=model, caps={"turns": 5})
)
store.append_event(
store.transcript("t", "helper").append(
DelegationCompleted(
handback="what changed: greet.py",
artifacts=["greet.py"],
@@ -39,7 +39,7 @@ def record_pair(
exit_status=status,
)
)
return store.read_events()[-1].id
return store.all_events()[-1].id
def test_pairs_a_start_with_its_completion(store: StateStore) -> None:
@@ -55,18 +55,18 @@ def test_pairs_a_start_with_its_completion(store: StateStore) -> None:
def test_unpaired_start_is_skipped(store: StateStore) -> None:
# A crash mid-delegation leaves a start with no completion; a half-formed
# record would pollute the set L9 exists to make trustworthy.
store.append_event(
DelegationStarted(brief="died", role="delegate", model="m", caps={})
store.transcript("t", "helper").append(
DelegationStarted(brief="died", role="helper", model="m", caps={})
)
assert collect(store) == []
def test_unrelated_events_between_the_pair_are_ignored(store: StateStore) -> None:
store.append_event(
DelegationStarted(brief="b", role="delegate", model="m", caps={})
store.transcript("t", "helper").append(
DelegationStarted(brief="b", role="helper", model="m", caps={})
)
store.append_event(AssistantText(text="working", model="m"))
store.append_event(
store.transcript("t", "helper").append(AssistantText(text="working", model="m"))
store.transcript("t", "helper").append(
DelegationCompleted(
handback="done",
artifacts=[],
@@ -97,15 +97,15 @@ def test_a_verdict_can_be_revised(store: StateStore) -> None:
def test_consults_and_delegations_share_the_set(store: StateStore) -> None:
record_pair(store, role="delegate")
record_pair(store, role="consultant", brief="which approach?")
record_pair(store, role="helper")
record_pair(store, role="expert", brief="which approach?")
roles = [r.role for r in collect(store)]
assert roles == ["delegate", "consultant"]
assert roles == ["helper", "expert"]
def test_collect_can_start_after_a_marker(store: StateStore) -> None:
record_pair(store, brief="old")
marker = store.read_events()[-1].id
marker = store.all_events()[-1].id
record_pair(store, brief="new")
assert [r.brief for r in collect(store, after=marker)] == ["new"]
+31 -13
View File
@@ -14,14 +14,14 @@ from nisse_dev.canonical import (
from nisse_dev.executor import LLMExecutor
from nisse_dev.llm import LLMError, RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.state import StateStore, Transcript
from nisse_dev.tools import Workspace
ROLES = RoleMap(
executor="llama3.2",
planner="claude-haiku-4-5",
delegate="claude-haiku-4-5",
consultant="claude-haiku-4-5",
helper="claude-haiku-4-5",
expert="claude-haiku-4-5",
)
@@ -66,8 +66,26 @@ def ws(tmp_path: Path) -> Workspace:
return Workspace(tmp_path / "work", timeout_s=10)
def build(client: Any, store: StateStore, ws: Workspace, **kw: Any) -> LLMExecutor:
return LLMExecutor(client, store, ROLES, ws, **kw)
ITEM = "t" # the work item these tests run in
def log(store: StateStore) -> Transcript:
return store.transcript(ITEM, "executor")
class Bound:
"""An executor run in the test's work item, as the orchestrator runs it."""
def __init__(self, executor: LLMExecutor, store: StateStore) -> None:
self.executor = executor
self.store = store
def run(self, milestone: int) -> Any:
return self.executor.run(log(self.store), milestone)
def build(client: Any, store: StateStore, ws: Workspace, **kw: Any) -> Bound:
return Bound(LLMExecutor(client, ROLES, ws, **kw), store)
def test_tool_calls_are_executed_and_results_logged(
@@ -84,7 +102,7 @@ def test_tool_calls_are_executed_and_results_logged(
assert outcome.status == "completed"
assert outcome.note == "wrote the file"
assert (ws.root / "app.py").read_text() == "print(1)\n"
kinds = [type(e.event).__name__ for e in store.read_events()]
kinds = [type(e.event).__name__ for e in log(store).read()]
assert kinds == ["ToolCall", "ToolResult", "ToolCall", "ToolResult"]
@@ -138,7 +156,7 @@ def test_finishing_without_any_work_is_refused_once(
assert (ws.root / "a.py").exists() # the refusal produced actual work
[refusal] = [
e.event
for e in store.read_events()
for e in log(store).read()
if isinstance(e.event, ToolResult) and e.event.call_id == "f1"
]
assert not refusal.success
@@ -165,7 +183,7 @@ def test_calls_after_finish_are_not_run(store: StateStore, ws: Workspace) -> Non
assert not (ws.root / "b.py").exists()
[late] = [
e.event
for e in store.read_events()
for e in log(store).read()
if isinstance(e.event, ToolResult) and e.event.call_id == "late"
]
assert "finish had already ended the milestone" in late.result
@@ -183,7 +201,7 @@ def test_an_invented_tool_counts_toward_escalation(
outcome = build(client, store, ws, failure_limit=2).run(milestone=1)
assert outcome.status == "escalate"
assert len(client.calls) == 2
results = [e.event for e in store.read_events() if isinstance(e.event, ToolResult)]
results = [e.event for e in log(store).read() if isinstance(e.event, ToolResult)]
assert "no such tool: chat" in results[0].result
assert "finish" in results[0].result # it is told what does exist
@@ -280,7 +298,7 @@ def test_multiple_calls_in_one_turn_all_run(store: StateStore, ws: Workspace) ->
build(client, store, ws).run(milestone=1)
assert (ws.root / "one.txt").exists() and (ws.root / "two.txt").exists()
results = [
e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
e.event for e in log(store).read() if type(e.event).__name__ == "ToolResult"
]
assert {r.call_id for r in results} == {"a", "b", "fin"}
@@ -293,7 +311,7 @@ class StubDesk:
self._answer = answer
self._error = error
def ask(self, question: str, **kw: Any) -> Any:
def ask(self, transcript: Any, question: str, **kw: Any) -> Any:
self.asked.append(question)
if self._error is not None:
raise self._error
@@ -336,7 +354,7 @@ def test_ask_expert_answer_comes_back_as_a_tool_result(
assert desk.asked == ["sqlite or postgres?"]
results = [
e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
e.event for e in log(store).read() if type(e.event).__name__ == "ToolResult"
]
assert "prefer sqlite here" in results[0].result
@@ -372,7 +390,7 @@ def test_exhausted_budget_is_reported_to_the_model_not_raised(
assert outcome.status == "completed" # it got on with the work
results = [
e.event for e in store.read_events() if type(e.event).__name__ == "ToolResult"
e.event for e in log(store).read() if type(e.event).__name__ == "ToolResult"
]
assert not results[0].success
assert "budget" in results[0].result
+35 -10
View File
@@ -20,13 +20,13 @@ from nisse_dev.expert import (
)
from nisse_dev.llm import LLMError, RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.state import StateStore
from nisse_dev.state import StateStore, Transcript
ROLES = RoleMap(
executor="llama3.2",
planner="haiku",
delegate="haiku",
consultant="claude-haiku-4-5",
helper="haiku",
expert="claude-haiku-4-5",
)
GOOD = '{"answer": "Use SQLite; it is enough here.", "confidence": "high", "caveats": "Assumes one writer."}'
@@ -57,8 +57,25 @@ def store(tmp_path: Path) -> StateStore:
return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
def desk(client: Any, store: StateStore, **kw: Any) -> ExpertDesk:
return ExpertDesk(client, store, ROLES, **kw)
ITEM = "t" # the work item the consults happen in
def log(store: StateStore, item: str = ITEM) -> Transcript:
return store.transcript(item, "expert")
class Bound:
"""The desk as the executor uses it: within one work item."""
def __init__(self, desk: ExpertDesk, store: StateStore, item: str) -> None:
self.desk, self.store, self.item = desk, store, item
def ask(self, question: str, **kw: Any) -> Any:
return self.desk.ask(log(self.store, self.item), question, **kw)
def desk(client: Any, store: StateStore, item: str = ITEM, **kw: Any) -> Bound:
return Bound(ExpertDesk(client, ROLES, **kw), store, item)
# --- parsing (the contract is prompt-level, so parsing is the guard) ------
@@ -104,7 +121,7 @@ def test_ask_returns_a_validated_answer(store: StateStore) -> None:
def test_consult_is_stateless_and_uses_the_consultant_role(
store: StateStore,
) -> None:
store.append_event(AssistantText(text="secret repo context", model="m"))
log(store).append(AssistantText(text="secret repo context", model="m"))
client = ScriptedClient([GOOD])
desk(client, store).ask("Is this API deprecated?")
@@ -134,10 +151,10 @@ def test_consults_are_recorded_as_composite_events(store: StateStore) -> None:
# L9 wants a consult/delegation eval set; reusing the pair gets consult
# evaluation for free.
desk(ScriptedClient([GOOD]), store).ask("q")
events = [e.event for e in store.read_events()]
events = [e.event for e in log(store).read()]
started = next(e for e in events if isinstance(e, DelegationStarted))
completed = next(e for e in events if isinstance(e, DelegationCompleted))
assert started.role == "consultant"
assert started.role == "expert"
assert started.brief == "q"
assert completed.exit_status == "answered"
@@ -146,7 +163,7 @@ def test_a_failed_consult_is_recorded_too(store: StateStore) -> None:
with pytest.raises(LLMError):
desk(ScriptedClient(["waffle", "waffle"]), store).ask("q")
completed = [
e.event for e in store.read_events() if isinstance(e.event, DelegationCompleted)
e.event for e in log(store).read() if isinstance(e.event, DelegationCompleted)
]
assert completed[0].exit_status == "failed"
@@ -177,10 +194,18 @@ def test_budget_survives_a_restart(tmp_path: Path) -> None:
def test_budget_counts_only_after_the_task_marker(store: StateStore) -> None:
d = desk(ScriptedClient([GOOD] * 5), store, budget=1)
d.ask("earlier task")
marker = store.read_events()[-1].id
marker = log(store).read()[-1].id
# A new task starts its allowance afresh from its own marker.
assert d.ask("new task", since_event_id=marker).confidence == "high"
def test_the_budget_is_per_work_item(store: StateStore) -> None:
# Consults in one task never use up another's allowance (L9).
desk(ScriptedClient([GOOD]), store, item="a", budget=1).ask("one")
assert desk(ScriptedClient([GOOD]), store, item="b", budget=1).ask("two")
with pytest.raises(BudgetExhausted):
desk(ScriptedClient([GOOD]), store, item="a", budget=1).ask("three")
def test_default_budget_is_small() -> None:
assert DEFAULT_BUDGET <= 5 # L9: do not let this become a second subsystem
+4 -4
View File
@@ -184,8 +184,8 @@ def test_malformed_response_raises_llm_error() -> None:
ROLES = RoleMap(
executor="local/qwen",
planner="anthropic/claude-opus-5",
delegate="anthropic/claude-opus-5",
consultant="anthropic/claude-opus-5",
helper="anthropic/claude-opus-5",
expert="anthropic/claude-opus-5",
)
@@ -195,8 +195,8 @@ def test_role_map_resolves_and_pins() -> None:
assert roles.pinned() == {
"executor": "local/qwen",
"planner": "anthropic/claude-opus-5",
"delegate": "anthropic/claude-opus-5",
"consultant": "anthropic/claude-opus-5",
"helper": "anthropic/claude-opus-5",
"expert": "anthropic/claude-opus-5",
}
+3 -2
View File
@@ -62,6 +62,7 @@ class Planner:
def plan(
self,
transcript: object,
gate: str,
project: str,
feedback: str | None = None,
@@ -74,7 +75,7 @@ class Executor:
def __init__(self, faults: Faults) -> None:
self._faults = faults
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
return self._faults.check("executor") or ExecutorOutcome("completed", "done")
@@ -189,7 +190,7 @@ def test_a_signoff_state_without_its_gate_offers_a_way_out(tmp_path: Path) -> No
class RaisingReviewExecutor:
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
raise RuntimeError("bug while addressing review")
+22 -7
View File
@@ -45,9 +45,9 @@ def test_transition_logs_from_to_and_cause(
) -> None:
store.transition("PLANNING_ARCH", cause="operator command: use demo", project="d")
store.transition("AWAIT_ARCH_SIGNOFF", cause="plan submitted", project="d")
first, second = events(stream)
assert first["event"] == "state.transition"
assert (first["from"], first["to"]) == (None, "PLANNING_ARCH")
# Opening the work item logs its own events first (work_item.open, .switch).
first, second = [e for e in events(stream) if e["event"] == "state.transition"]
assert (first["from"], first["to"]) == ("IDLE", "PLANNING_ARCH")
assert (second["from"], second["to"]) == ("PLANNING_ARCH", "AWAIT_ARCH_SIGNOFF")
assert second["cause"] == "plan submitted"
@@ -66,9 +66,9 @@ def test_task_id_threads_through_events(store: StateStore, stream: io.StringIO)
store.transition("IDLE", cause="boot")
store.oplog.task_id = "task-42"
store.transition("PLANNING_ARCH", cause="operator command")
before, during = events(stream)
before, *during = events(stream)
assert before["task_id"] is None
assert during["task_id"] == "task-42"
assert {e["task_id"] for e in during} == {"task-42"}
def test_migration_emits_event(tmp_path: Path, stream: io.StringIO) -> None:
@@ -120,7 +120,10 @@ def test_literal_log_lines_for_human_readability(
monkeypatch.setattr("nisse_dev.oplog.datetime", _FrozenDatetime)
store.transition(
"PLANNING_ARCH", cause="operator command: use demo", project="demo"
"PLANNING_ARCH",
cause="operator command: use demo",
project="demo",
data={"task_id": "demo-m2"},
)
store.oplog.task_id = "demo-m2"
intent_id = store.begin_intent(
@@ -129,11 +132,23 @@ def test_literal_log_lines_for_human_readability(
store.commit_intent(intent_id, cause="push verified by Gitea query")
assert stream.getvalue().splitlines() == [
(
'{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",'
' "event": "work_item.open", "task_id": null,'
' "cause": "operator command: use demo",'
' "work_item": "demo-m2", "scope": "demo"}'
),
(
'{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",'
' "event": "work_item.switch", "task_id": null,'
' "cause": "operator command: use demo",'
' "from": null, "to": "demo-m2"}'
),
(
'{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",'
' "event": "state.transition", "task_id": null,'
' "cause": "operator command: use demo",'
' "from": null, "to": "PLANNING_ARCH", "project": "demo"}'
' "from": "IDLE", "to": "PLANNING_ARCH", "project": "demo"}'
),
(
'{"ts": "2026-09-05T12:00:00+00:00", "agent": "nisse-dev",'
+33 -21
View File
@@ -23,14 +23,14 @@ from nisse_dev.control import (
from nisse_dev.llm import BudgetExhausted, RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.planner import PLANNER_TOOLS, LLMPlanningAgent
from nisse_dev.state import StateStore
from nisse_dev.state import StateStore, Transcript
from nisse_dev.transport import InMemoryTransport
ROLES = RoleMap(
executor="local/qwen",
planner="anthropic/claude-opus-5",
delegate="anthropic/claude-opus-5",
consultant="anthropic/claude-opus-5",
helper="anthropic/claude-opus-5",
expert="anthropic/claude-opus-5",
)
@@ -64,6 +64,13 @@ def plan_call(content: str, gate: str = "arch") -> ToolCall:
)
ITEM = "t" # the work item the unit tests plan in
def log(store: StateStore) -> Transcript:
return store.transcript(ITEM, "planner")
@pytest.fixture
def store(tmp_path: Path) -> StateStore:
return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
@@ -71,8 +78,8 @@ def store(tmp_path: Path) -> StateStore:
def test_submit_plan_tool_call_becomes_submit_plan(store: StateStore) -> None:
client = ScriptedClient([[plan_call("step 1: create src/app.py")]])
agent = LLMPlanningAgent(client, store, ROLES)
action = agent.plan("arch", "demo")
agent = LLMPlanningAgent(client, ROLES)
action = agent.plan(log(store), "arch", "demo")
assert action == SubmitPlan("step 1: create src/app.py")
@@ -88,13 +95,13 @@ def test_ask_human_tool_call_becomes_ask_human(store: StateStore) -> None:
]
]
)
action = LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
action = LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
assert action == AskHuman("postgres or sqlite?")
def test_planner_role_and_tools_are_used(store: StateStore) -> None:
client = ScriptedClient([[plan_call("plan")]])
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
call = client.calls[0]
assert call["model"] == "anthropic/claude-opus-5" # never the executor
assert [t["function"]["name"] for t in call["tools"]] == [
@@ -113,8 +120,8 @@ def test_model_output_is_appended_to_the_log(store: StateStore) -> None:
client = ScriptedClient(
[[AssistantText(text="thinking out loud", model="m"), plan_call("the plan")]]
)
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
kinds = [type(e.event).__name__ for e in store.read_events()]
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
kinds = [type(e.event).__name__ for e in log(store).read()]
assert kinds == ["AssistantText", "ToolCall", "ToolResult"]
@@ -122,8 +129,8 @@ def test_gate_tool_call_is_answered_in_the_log(store: StateStore) -> None:
# submit_plan ends the turn and has no natural result, but a tool call
# left dangling makes the next render invalid for the provider.
client = ScriptedClient([[plan_call("the plan")]])
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
events = [e.event for e in store.read_events()]
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
events = [e.event for e in log(store).read()]
assert [type(e).__name__ for e in events] == ["ToolCall", "ToolResult"]
assert events[1].call_id == events[0].call_id
assert "posted for approval" in events[1].result
@@ -137,7 +144,7 @@ def test_prose_only_turn_retries_once_then_raises(store: StateStore) -> None:
]
)
with pytest.raises(PlanningError, match="no submit_plan or ask_human"):
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
assert len(client.calls) == 2
assert "did not call a tool" in client.calls[1]["system"]
@@ -146,7 +153,7 @@ def test_retry_succeeds_after_prose(store: StateStore) -> None:
client = ScriptedClient(
[[AssistantText(text="prose", model="m")], [plan_call("the real plan")]]
)
action = LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
action = LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
assert action == SubmitPlan("the real plan")
@@ -154,13 +161,15 @@ def test_malformed_tool_arguments_are_rejected(store: StateStore) -> None:
blank = ToolCall(call_id="c", tool="submit_plan", arguments={"content": " "})
client = ScriptedClient([[blank], [blank]])
with pytest.raises(PlanningError):
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
def test_planner_sees_prior_transcript(store: StateStore) -> None:
store.append_event(UserMessage(text="use demo", source="transport:@don"))
store.transcript(ITEM, "operator").append(
UserMessage(text="use demo", source="transport:@don")
)
client = ScriptedClient([[plan_call("plan")]])
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
seen = [e.event for e in client.calls[0]["events"]]
assert UserMessage(text="use demo", source="transport:@don") in seen
@@ -171,7 +180,7 @@ def test_planner_sees_prior_transcript(store: StateStore) -> None:
def build(tmp_path: Path, client: ScriptedClient) -> tuple[Orchestrator, StateStore]:
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
transport = InMemoryTransport()
agent = LLMPlanningAgent(client, store, ROLES)
agent = LLMPlanningAgent(client, ROLES)
orch = Orchestrator(store, transport, agent, operators={"don"}, roles=ROLES)
orch.start()
return orch, store, transport
@@ -186,7 +195,10 @@ def test_use_project_produces_llm_plan_and_transcript(tmp_path: Path) -> None:
assert request.kind == "approval_request"
assert "create app.py" in request.text
events = [e.event for e in store.read_events()]
# The work item's own events: `use demo` was said while idle, and is
# carried into the item it opened (D1).
item = store.active_item()
events = [e.event for e in store.transcript(item.id, "operator").read()]
kinds = [type(e).__name__ for e in events]
assert kinds == [
"UserMessage",
@@ -237,7 +249,7 @@ def test_operator_can_recover_from_a_planning_failure(tmp_path: Path) -> None:
def test_operator_can_recover_after_a_stopped_milestone(tmp_path: Path) -> None:
class StoppedExecutor:
def run(self, milestone: int) -> Any:
def run(self, transcript: object, milestone: int) -> Any:
from nisse_dev.executor import ExecutorOutcome
return ExecutorOutcome("escalate", "tests failed twice")
@@ -248,7 +260,7 @@ def test_operator_can_recover_after_a_stopped_milestone(tmp_path: Path) -> None:
orch = Orchestrator(
store,
transport,
LLMPlanningAgent(client, store, ROLES),
LLMPlanningAgent(client, ROLES),
operators={"don"},
roles=ROLES,
executor=StoppedExecutor(),
@@ -295,5 +307,5 @@ def test_budget_exhaustion_is_not_retried(store: StateStore) -> None:
client = BrokeClient()
with pytest.raises(BudgetExhausted):
LLMPlanningAgent(client, store, ROLES).plan("arch", "demo")
LLMPlanningAgent(client, ROLES).plan(log(store), "arch", "demo")
assert client.calls == 1 # not two
+5 -5
View File
@@ -27,8 +27,8 @@ from nisse_dev.transport import InMemoryTransport
ROLES = RoleMap(
executor="llama3.2",
planner="haiku",
delegate="haiku",
consultant="haiku",
helper="haiku",
expert="haiku",
)
@@ -57,7 +57,7 @@ class ScriptedClient:
class DoneExecutor:
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
return ExecutorOutcome("completed", f"milestone {milestone} done")
@@ -91,7 +91,7 @@ def build(
orch = Orchestrator(
store,
transport,
LLMPlanningAgent(client, store, ROLES),
LLMPlanningAgent(client, ROLES),
operators={"don"},
roles=ROLES,
executor=DoneExecutor(),
@@ -183,7 +183,7 @@ def test_publisher_reconciler_is_registered_at_boot(tmp_path: Path) -> None:
orch = Orchestrator(
store,
transport,
LLMPlanningAgent(ScriptedClient([]), store, ROLES),
LLMPlanningAgent(ScriptedClient([]), ROLES),
operators={"don"},
roles=ROLES,
publisher=FakePublisher(),
+1 -1
View File
@@ -73,7 +73,7 @@ class Executor:
self.runs = 0
self.outcome = ExecutorOutcome("completed", "done")
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
self.runs += 1
return self.outcome
@@ -104,7 +104,7 @@ class Executor:
def __init__(self) -> None:
self.outcome = ExecutorOutcome("completed", "done")
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
return self.outcome
@@ -283,7 +283,7 @@ def test_crash_during_a_review_loses_the_comment(tmp_path: Path) -> None:
class Crash(BaseException):
pass
def dies(milestone: int) -> ExecutorOutcome:
def dies(transcript: object, milestone: int) -> ExecutorOutcome:
raise Crash() # the process dies mid-review
agent.executor.run = dies # type: ignore[method-assign]
@@ -348,7 +348,7 @@ class Crash(BaseException):
def crash_mid_review(agent: Agent, gitea: FakeGitea) -> None:
def dies(milestone: int) -> ExecutorOutcome:
def dies(transcript: object, milestone: int) -> ExecutorOutcome:
raise Crash()
agent.executor.run = dies # type: ignore[method-assign]
@@ -377,7 +377,7 @@ def test_an_executor_error_answers_the_comment_so_it_is_not_redelivered(
agent.say("use demo", "approve", "approve", "start milestone 1")
gitea.add(20, "don", "please add a test")
def fails(milestone: int) -> ExecutorOutcome:
def fails(transcript: object, milestone: int) -> ExecutorOutcome:
raise RuntimeError("bug in the executor")
agent.executor.run = fails # type: ignore[method-assign]
+5 -5
View File
@@ -15,7 +15,7 @@ from nisse_dev.publish import PublishResult
from nisse_dev.state import StateStore
from nisse_dev.transport import InMemoryTransport
ROLES = RoleMap(executor="x", planner="y", delegate="y", consultant="y")
ROLES = RoleMap(executor="x", planner="y", helper="y", expert="y")
class ScriptedClient:
@@ -42,10 +42,10 @@ class RecordingExecutor:
self._outcomes = outcomes or []
self.store: StateStore | None = None
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: Any, milestone: int) -> ExecutorOutcome:
self.runs.append(milestone)
if self.store is not None:
self.seen_events.append(self.store.read_events())
# What it is given: the work item's transcript, nothing wider.
self.seen_events.append(transcript.read())
status = self._outcomes.pop(0) if self._outcomes else "completed"
return ExecutorOutcome(status, f"executor said {status}")
@@ -88,7 +88,7 @@ def build(
orch = Orchestrator(
store,
transport,
LLMPlanningAgent(ScriptedClient(), store, ROLES),
LLMPlanningAgent(ScriptedClient(), ROLES),
operators={"don"},
roles=ROLES,
executor=executor,
+20 -18
View File
@@ -23,7 +23,7 @@ from nisse_dev.executor import ExecutorOutcome, LLMExecutor
from nisse_dev.llm import RoleMap
from nisse_dev.oplog import OpLog
from nisse_dev.planner import LLMPlanningAgent
from nisse_dev.state import StateStore
from nisse_dev.state import StateStore, Transcript
from nisse_dev.tools import ToolOutcome
from nisse_dev.transport import InMemoryTransport
from nisse_dev.worker import SerialWorker
@@ -31,18 +31,22 @@ from nisse_dev.worker import SerialWorker
ROLES = RoleMap(
executor="llama3.2",
planner="claude-haiku-4-5",
delegate="claude-haiku-4-5",
consultant="claude-haiku-4-5",
helper="claude-haiku-4-5",
expert="claude-haiku-4-5",
)
def log(store: StateStore, actor: str = "executor") -> Transcript:
return store.transcript("t", actor)
def new_store(path: Path) -> StateStore:
return StateStore(path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
def user_messages(store: StateStore) -> list[str]:
return [
e.event.text for e in store.read_events() if isinstance(e.event, UserMessage)
e.event.text for e in store.all_events() if isinstance(e.event, UserMessage)
]
@@ -58,7 +62,7 @@ class OperatorDuringWork:
self.says = says
self.transport: InMemoryTransport | None = None
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
assert self.transport is not None
self.transport.receive("don", self.says)
if self.cancel.requested:
@@ -158,6 +162,7 @@ class StopsBeforePlanning:
def plan(
self,
transcript: object,
gate: str,
project: str,
feedback: str | None = None,
@@ -200,7 +205,7 @@ class BlocksUntilStopped:
self.cancel = cancel
self.running = threading.Event()
def run(self, milestone: int) -> ExecutorOutcome:
def run(self, transcript: object, milestone: int) -> ExecutorOutcome:
self.running.set()
for _ in range(500):
if self.cancel.requested:
@@ -287,10 +292,8 @@ def test_executor_stops_before_a_turn_without_calling_the_model(
cancel = Cancel()
cancel.set()
client = ScriptedClient([])
executor = LLMExecutor(
client, new_store(tmp_path), ROLES, StopsAfterFirstCall(cancel), cancel=cancel
)
outcome = executor.run(1)
executor = LLMExecutor(client, ROLES, StopsAfterFirstCall(cancel), cancel=cancel)
outcome = executor.run(log(new_store(tmp_path)), 1)
assert outcome == ExecutorOutcome("stopped", "before turn 1")
assert client.calls == 0
@@ -305,14 +308,14 @@ def test_executor_stops_between_tool_calls_and_records_the_skipped_ones(
ToolCall(call_id="a", tool="run_command", arguments={"command": "make"}),
ToolCall(call_id="b", tool="write_file", arguments={"path": "x"}),
]
executor = LLMExecutor(ScriptedClient([calls]), store, ROLES, ws, cancel=cancel)
outcome = executor.run(1)
executor = LLMExecutor(ScriptedClient([calls]), ROLES, ws, cancel=cancel)
outcome = executor.run(log(store), 1)
assert outcome == ExecutorOutcome("stopped", "during turn 1")
assert ws.ran == ["run_command"]
results = {
e.event.call_id: e.event.result
for e in store.read_events()
for e in log(store).read()
if isinstance(e.event, ToolResult)
}
# Every call has a result, so the transcript still renders.
@@ -320,7 +323,7 @@ def test_executor_stops_between_tool_calls_and_records_the_skipped_ones(
class StoppedDelegate:
def run(self, *, brief: str, branch: str) -> DelegationResult:
def run(self, transcript: object, *, brief: str, branch: str) -> DelegationResult:
return DelegationResult("stopped", "stopped by the operator during turn 2")
@@ -339,12 +342,11 @@ def test_a_stopped_delegation_stops_the_milestone(tmp_path: Path) -> None:
executor = LLMExecutor(
ScriptedClient(failing),
new_store(tmp_path),
ROLES,
Failing(),
delegate=StoppedDelegate(),
)
outcome = executor.run(1)
outcome = executor.run(log(new_store(tmp_path)), 1)
assert outcome.status == "stopped"
assert "during a delegation" in outcome.note
@@ -353,9 +355,9 @@ def test_planner_does_not_start_a_turn_once_stopped(tmp_path: Path) -> None:
cancel = Cancel()
cancel.set()
client = ScriptedClient([])
planner = LLMPlanningAgent(client, new_store(tmp_path), ROLES, cancel=cancel)
planner = LLMPlanningAgent(client, ROLES, cancel=cancel)
try:
planner.plan("arch", "demo")
planner.plan(log(new_store(tmp_path), "planner"), "arch", "demo")
except PlanningStopped:
pass
else:
+177
View File
@@ -0,0 +1,177 @@
"""Work items in the state store (architecture L1, L7; Development Plan M1.1)."""
import io
from pathlib import Path
from typing import Any
import pytest
from nisse_dev.canonical import AssistantText, UserMessage
from nisse_dev.control import (
AWAIT_ARCH_SIGNOFF,
PR_OPEN,
AskHuman,
Orchestrator,
SubmitPlan,
)
from nisse_dev.oplog import OpLog
from nisse_dev.state import IDLE, StateStore
from nisse_dev.transport import InMemoryTransport
def new_store(path: Path) -> StateStore:
return StateStore(path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
# --- the store ---------------------------------------------------------------
def test_a_transcript_reads_only_its_own_work_item(tmp_path: Path) -> None:
store = new_store(tmp_path)
store.transcript("a", "operator").append(UserMessage(text="for a", source="m"))
store.transcript("b", "operator").append(UserMessage(text="for b", source="m"))
store.transcript(None, "operator").append(UserMessage(text="idle", source="m"))
assert [e.event.text for e in store.transcript("a", "planner").read()] == ["for a"]
assert [e.event.text for e in store.transcript(None, "planner").read()] == ["idle"]
def test_every_event_carries_its_work_item_and_actor(tmp_path: Path) -> None:
store = new_store(tmp_path)
log = store.transcript("a", "executor")
log.append(AssistantText(text="working", model="m"))
log.as_actor("helper").append(AssistantText(text="helping", model="f"))
assert [(e.work_item, e.actor) for e in store.all_events()] == [
("a", "executor"),
("a", "helper"),
]
def test_an_unknown_actor_is_refused(tmp_path: Path) -> None:
with pytest.raises(ValueError, match="unknown actor"):
new_store(tmp_path).transcript("a", "someone")
def test_intents_are_owned_by_their_work_item(tmp_path: Path) -> None:
store = new_store(tmp_path)
store.transcript("a", "helper").begin_intent("delegation", {}, cause="t")
[intent] = store.pending_intents()
assert intent.work_item == "a"
def test_work_items_and_the_active_one_survive_a_restart(tmp_path: Path) -> None:
store = new_store(tmp_path)
store.open_work_item("a", title="demo", scope="demo", position=IDLE, cause="t")
store.set_active("a", cause="t")
store.update_work_item("a", position="PLANNING_ARCH", data={"x": 1}, cause="t")
store.close()
store = new_store(tmp_path)
item = store.active_item()
assert item is not None
assert (item.id, item.position, item.data, item.status) == (
"a",
"PLANNING_ARCH",
{"x": 1},
"open",
)
def test_closing_records_the_outcome_and_going_idle_clears_active(
tmp_path: Path,
) -> None:
store = new_store(tmp_path)
store.open_work_item("a", title="demo", position=IDLE, cause="t")
store.set_active("a", cause="t")
store.close_work_item("a", outcome="completed", cause="t")
store.set_active(None, cause="t")
assert store.active_item() is None
item = store.work_item("a")
assert item is not None and (item.status, item.outcome) == ("closed", "completed")
with pytest.raises(KeyError):
store.close_work_item("a", outcome="abandoned", cause="t") # already closed
def test_messages_sent_while_idle_are_carried_with_their_origin(
tmp_path: Path,
) -> None:
store = new_store(tmp_path)
store.set_active(None, cause="boot")
idle = store.transcript(None, "operator")
first = idle.append(UserMessage(text="add a truncate helper", source="m:1"))
store.open_work_item("a", title="demo", position=IDLE, cause="t")
assert store.carry_idle_messages("a") == 1
[carried] = store.transcript("a", "planner").read()
assert carried.event.text == "add a truncate helper"
assert f"#{first}" in carried.event.source # where it came from
assert len(store.all_events()) == 2 # a copy; the original is untouched
# --- the orchestrator ----------------------------------------------------------
class RecordingPlanner:
"""Records what each planning turn could read."""
def __init__(self) -> None:
self.seen: list[list[str]] = []
def plan(
self,
transcript: Any,
gate: str,
project: str,
feedback: str | None = None,
answer: str | None = None,
) -> SubmitPlan | AskHuman:
self.seen.append(
[
e.event.text
for e in transcript.read()
if isinstance(e.event, UserMessage)
]
)
return SubmitPlan(f"{gate} plan for {project}")
def build(tmp_path: Path) -> tuple[StateStore, InMemoryTransport, RecordingPlanner]:
store = new_store(tmp_path)
transport = InMemoryTransport()
planner = RecordingPlanner()
Orchestrator(store, transport, planner, operators={"don"}).start()
return store, transport, planner
def test_a_second_task_never_sees_the_first(tmp_path: Path) -> None:
# The second live run's planner read every earlier task's events (#78).
store, transport, planner = build(tmp_path)
for text in ("build the first thing", "use first", "approve", "approve"):
transport.receive("don", text)
transport.receive("don", "start milestone 1")
assert store.get_state().state == PR_OPEN
transport.receive("don", "project complete")
transport.receive("don", "build the second thing")
transport.receive("don", "use second")
assert store.get_state().state == AWAIT_ARCH_SIGNOFF
second_task_view = planner.seen[-1]
assert "build the second thing" in second_task_view
assert not any("first" in text for text in second_task_view)
def test_each_task_is_its_own_work_item_with_its_outcome(tmp_path: Path) -> None:
store, transport, _ = build(tmp_path)
for text in ("use first", "approve", "approve", "start milestone 1"):
transport.receive("don", text)
transport.receive("don", "project complete")
transport.receive("don", "use second")
transport.receive("don", "abandon")
outcomes = [(item.title, item.outcome) for item in store.work_items()]
assert outcomes == [("first", "completed"), ("second", "abandoned")]
assert store.active_item() is None
def test_the_task_message_before_use_reaches_the_planner(tmp_path: Path) -> None:
# Run 2's flow: the task, then `use`. It must still be planned from.
_, transport, planner = build(tmp_path)
transport.receive("don", "add a truncate helper")
transport.receive("don", "use demo")
assert "add a truncate helper" in planner.seen[0]
+2 -2
View File
@@ -23,8 +23,8 @@ llm:
roles:
executor: claude-haiku-4-5
planner: claude-haiku-4-5
delegate: claude-haiku-4-5
consultant: claude-haiku-4-5
helper: claude-haiku-4-5
expert: claude-haiku-4-5
matrix:
homeserver: https://matrix.dongar.net
+2 -2
View File
@@ -21,7 +21,7 @@ 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"))
store.transcript("t", "operator").append(UserMessage(text="start milestone 2", source="m:1"))
os._exit(1) # die without closing anything, as a power cut would
"""
@@ -48,5 +48,5 @@ def test_state_survives_process_death(tmp_path: Path) -> None:
[pending] = store.pending_intents()
assert pending.payload == {"branch": "milestone-2"}
[stored] = store.read_events()
[stored] = store.all_events()
assert stored.event.text == "start milestone 2"
+8 -4
View File
@@ -45,11 +45,13 @@ def build(
def test_delegate_completes_a_scoped_subtask(
tmp_path: Path, passthrough: AnthropicPassthroughClient, live_model: str
) -> None:
delegation, repo, _store = build(
delegation, repo, store = build(
tmp_path, Caps(turns=12, wall_clock_s=300.0), passthrough, live_model
)
result = delegation.run(brief=BRIEF, branch="live/delegation-1")
result = delegation.run(
store.transcript("t", "helper"), brief=BRIEF, branch="live/delegation-1"
)
assert result.usable, f"delegation did not hand back: {result.status} {result.note}"
assert (repo.root / "slugify.py").exists()
@@ -75,11 +77,13 @@ def test_a_cap_breach_discards_the_work(
) -> None:
# One turn cannot finish the brief, so this exercises the abort path
# against a real model rather than a scripted one.
delegation, repo, _ = build(
delegation, repo, store = build(
tmp_path, Caps(turns=1, wall_clock_s=120.0), passthrough, live_model
)
result = delegation.run(brief=BRIEF, branch="live/delegation-2")
result = delegation.run(
store.transcript("t", "helper"), brief=BRIEF, branch="live/delegation-2"
)
assert not result.usable
assert result.status == "capped"
+11 -7
View File
@@ -22,11 +22,11 @@ pytestmark = pytest.mark.integration
def desk(tmp_path: Path, litellm: LiteLLMClient, live_model: str):
def make(budget: int = 2) -> tuple[ExpertDesk, StateStore]:
model = live_model
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
store = StateStore(
tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO())
)
return ExpertDesk(litellm, store, roles, budget=budget), store
return ExpertDesk(litellm, roles, budget=budget), store
return make
@@ -35,20 +35,24 @@ def test_live_consult_parses_against_the_local_schema(desk) -> None:
expert, store = desk()
answer = expert.ask(
store.transcript("t", "expert"),
"For a single-process agent storing a few thousand rows of task state, "
"is SQLite sufficient, or is PostgreSQL warranted? Answer briefly."
"is SQLite sufficient, or is PostgreSQL warranted? Answer briefly.",
)
assert answer.confidence in {"high", "medium", "low"}
assert len(answer.answer) > 20
# The consult is captured for L9 without any extra bookkeeping.
[record] = collect(store)
assert record.role == "consultant"
assert record.role == "expert"
assert record.exit_status == "answered"
def test_live_budget_stops_a_circling_caller(desk) -> None:
expert, _ = desk(budget=1)
expert.ask("Name one advantage of WAL mode in SQLite. Be brief.")
expert, store = desk(budget=1)
expert.ask(
store.transcript("t", "expert"),
"Name one advantage of WAL mode in SQLite. Be brief.",
)
with pytest.raises(BudgetExhausted):
expert.ask("And another?")
expert.ask(store.transcript("t", "expert"), "And another?")
+3 -3
View File
@@ -64,7 +64,7 @@ def test_plan_approve_execute_produces_a_real_file(
# Both roles use the capable model here: the point is the whole gated
# loop over a real proxy, not local-model competence (measured separately).
model = live_model
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
client = litellm
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
workspace = Workspace(tmp_path / "work", timeout_s=60)
@@ -72,10 +72,10 @@ def test_plan_approve_execute_produces_a_real_file(
orchestrator = Orchestrator(
store,
transport,
LLMPlanningAgent(client, store, roles),
LLMPlanningAgent(client, roles),
operators={"don"},
roles=roles,
executor=LLMExecutor(client, store, roles, workspace, max_turns=10),
executor=LLMExecutor(client, roles, workspace, max_turns=10),
)
orchestrator.start()
+8 -6
View File
@@ -24,7 +24,7 @@ pytestmark = pytest.mark.integration
def store_with(tmp_path: Path, *events) -> StateStore:
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
for event in events:
store.append_event(event)
store.transcript("t", "helper").append(event)
return store
@@ -35,7 +35,7 @@ def test_passthrough_completes_a_plain_turn(
tmp_path, UserMessage(text="Reply with exactly: pong", source="live")
)
result = passthrough.complete(
store.read_events(), None, live_model, system="Be terse."
store.transcript("t", "helper").read(), None, live_model, system="Be terse."
)
texts = [e.text for e in result if isinstance(e, AssistantText)]
assert texts and "pong" in texts[0].lower()
@@ -53,7 +53,7 @@ def test_passthrough_round_trips_a_tool_call(
)
client = passthrough
produced = client.complete(
store.read_events(),
store.transcript("t", "helper").read(),
EXECUTOR_TOOLS,
live_model,
system="Use a tool. Do not answer in prose.",
@@ -64,11 +64,13 @@ def test_passthrough_round_trips_a_tool_call(
# Feed the result back: the follow-up render must be accepted, which is
# what proves tool_result placement is right for this vendor.
for event in produced:
store.append_event(event)
store.append_event(
store.transcript("t", "helper").append(event)
store.transcript("t", "helper").append(
ToolResult(call_id=calls[0].call_id, result="app.py\ntests/", success=True)
)
follow_up = client.complete(store.read_events(), EXECUTOR_TOOLS, live_model)
follow_up = client.complete(
store.transcript("t", "helper").read(), EXECUTOR_TOOLS, live_model
)
assert follow_up, "the vendor rejected or ignored the tool_result turn"
+5 -5
View File
@@ -25,9 +25,9 @@ def test_live_planning_turn_returns_a_tool_action(
tmp_path: Path, litellm: LiteLLMClient, live_model: str
) -> None:
model = live_model
roles = RoleMap(executor=model, planner=model, delegate=model, consultant=model)
roles = RoleMap(executor=model, planner=model, helper=model, expert=model)
store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
store.append_event(
store.transcript("t", "operator").append(
UserMessage(
text=(
"Project 'hello-cli': a Python script that prints a greeting. "
@@ -36,12 +36,12 @@ def test_live_planning_turn_returns_a_tool_action(
source="live-test",
)
)
agent = LLMPlanningAgent(litellm, store, roles)
agent = LLMPlanningAgent(litellm, roles)
action = agent.plan("arch", "hello-cli")
action = agent.plan(store.transcript("t", "planner"), "arch", "hello-cli")
assert isinstance(action, SubmitPlan | AskHuman)
text = action.content if isinstance(action, SubmitPlan) else action.question
assert text.strip(), "tool call carried empty content"
# The turn must be durable regardless of what the model said (L8/L7).
assert len(store.read_events()) > 1
assert len(store.transcript("t", "planner").read()) > 1