diff --git a/docs/Development-Plan.md b/docs/Development-Plan.md index 7b0592a..43fca1a 100644 --- a/docs/Development-Plan.md +++ b/docs/Development-Plan.md @@ -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 diff --git a/nisse-dev.example.yaml b/nisse-dev.example.yaml index 6f78d01..2fbd6bf 100644 --- a/nisse-dev.example.yaml +++ b/nisse-dev.example.yaml @@ -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 diff --git a/nisse-dev/src/nisse_dev/agent.py b/nisse-dev/src/nisse_dev/agent.py index cb9693d..a711a09 100644 --- a/nisse-dev/src/nisse_dev/agent.py +++ b/nisse-dev/src/nisse_dev/agent.py @@ -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, diff --git a/nisse-dev/src/nisse_dev/bench/runner.py b/nisse-dev/src/nisse_dev/bench/runner.py index 0ae3c13..26ad4ed 100644 --- a/nisse-dev/src/nisse_dev/bench/runner.py +++ b/nisse-dev/src/nisse_dev/bench/runner.py @@ -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( diff --git a/nisse-dev/src/nisse_dev/cancel.py b/nisse-dev/src/nisse_dev/cancel.py index 9e459a3..7d9ee5e 100644 --- a/nisse-dev/src/nisse_dev/cancel.py +++ b/nisse-dev/src/nisse_dev/cancel.py @@ -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) ) diff --git a/nisse-dev/src/nisse_dev/canonical.py b/nisse-dev/src/nisse_dev/canonical.py index d3cfe29..0be2098 100644 --- a/nisse-dev/src/nisse_dev/canonical.py +++ b/nisse-dev/src/nisse_dev/canonical.py @@ -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: diff --git a/nisse-dev/src/nisse_dev/config.py b/nisse-dev/src/nisse_dev/config.py index bb733d2..ac208b6 100644 --- a/nisse-dev/src/nisse_dev/config.py +++ b/nisse-dev/src/nisse_dev/config.py @@ -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, } diff --git a/nisse-dev/src/nisse_dev/control.py b/nisse-dev/src/nisse_dev/control.py index 7b8e322..bc67379 100644 --- a/nisse-dev/src/nisse_dev/control.py +++ b/nisse-dev/src/nisse_dev/control.py @@ -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 diff --git a/nisse-dev/src/nisse_dev/delegation.py b/nisse-dev/src/nisse_dev/delegation.py index 5535e7d..e36c403 100644 --- a/nisse-dev/src/nisse_dev/delegation.py +++ b/nisse-dev/src/nisse_dev/delegation.py @@ -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): diff --git a/nisse-dev/src/nisse_dev/dump.py b/nisse-dev/src/nisse_dev/dump.py index 301f895..731fbcd 100644 --- a/nisse-dev/src/nisse_dev/dump.py +++ b/nisse-dev/src/nisse_dev/dump.py @@ -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) diff --git a/nisse-dev/src/nisse_dev/evals.py b/nisse-dev/src/nisse_dev/evals.py index 56e6a8e..285b5e3 100644 --- a/nisse-dev/src/nisse_dev/evals.py +++ b/nisse-dev/src/nisse_dev/evals.py @@ -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: diff --git a/nisse-dev/src/nisse_dev/executor.py b/nisse-dev/src/nisse_dev/executor.py index 3c6d8c7..de23dc5 100644 --- a/nisse-dev/src/nisse_dev/executor.py +++ b/nisse-dev/src/nisse_dev/executor.py @@ -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, diff --git a/nisse-dev/src/nisse_dev/expert.py b/nisse-dev/src/nisse_dev/expert.py index 68d6039..0735f4c 100644 --- a/nisse-dev/src/nisse_dev/expert.py +++ b/nisse-dev/src/nisse_dev/expert.py @@ -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=[], diff --git a/nisse-dev/src/nisse_dev/llm.py b/nisse-dev/src/nisse_dev/llm.py index 249ff28..e5b5bd6 100644 --- a/nisse-dev/src/nisse_dev/llm.py +++ b/nisse-dev/src/nisse_dev/llm.py @@ -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: diff --git a/nisse-dev/src/nisse_dev/planner.py b/nisse-dev/src/nisse_dev/planner.py index 2de895c..1739f57 100644 --- a/nisse-dev/src/nisse_dev/planner.py +++ b/nisse-dev/src/nisse_dev/planner.py @@ -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=( diff --git a/nisse-dev/src/nisse_dev/publish.py b/nisse-dev/src/nisse_dev/publish.py index 44763e4..189dd21 100644 --- a/nisse-dev/src/nisse_dev/publish.py +++ b/nisse-dev/src/nisse_dev/publish.py @@ -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 diff --git a/nisse-dev/src/nisse_dev/state.py b/nisse-dev/src/nisse_dev/state.py index 3187e96..eaf4f51 100644 --- a/nisse-dev/src/nisse_dev/state.py +++ b/nisse-dev/src/nisse_dev/state.py @@ -12,6 +12,7 @@ resolved against external truth on boot). import json import sqlite3 +import uuid from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import UTC, datetime @@ -21,49 +22,41 @@ from typing import Any, Self from nisse_dev.canonical import ( CanonicalEvent, StoredEvent, + UserMessage, event_type, from_payload, to_payload, ) from nisse_dev.oplog import OpLog -CURRENT_SCHEMA_VERSION = 3 +CURRENT_SCHEMA_VERSION = 4 + +# Stores older than this predate work items (Development Plan M1.1). There was +# no production data to carry over, so there is no migration from them: start +# a fresh store (scripts/bringup.sh start sets the old one aside). +OLDEST_SUPPORTED_VERSION = 4 Migration = Callable[[sqlite3.Connection], None] -_CANONICAL_EVENTS_DDL = """ -CREATE TABLE canonical_events ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - ts TEXT NOT NULL, - type TEXT NOT NULL, - payload TEXT NOT NULL -) -""" - - -_EVAL_OUTCOMES_DDL = """ -CREATE TABLE eval_outcomes ( - event_id INTEGER PRIMARY KEY, - verdict TEXT NOT NULL, - note TEXT NOT NULL DEFAULT '', - marked_at TEXT NOT NULL -) -""" - - -def _migrate_v1_add_canonical_events(conn: sqlite3.Connection) -> None: - conn.execute(_CANONICAL_EVENTS_DDL) - - -def _migrate_v2_add_eval_outcomes(conn: sqlite3.Connection) -> None: - conn.execute(_EVAL_OUTCOMES_DDL) - - # Maps version N to the migration that takes the schema from N to N+1. -MIGRATIONS: Mapping[int, Migration] = { - 1: _migrate_v1_add_canonical_events, - 2: _migrate_v2_add_eval_outcomes, -} +MIGRATIONS: Mapping[int, Migration] = {} + +IDLE = "IDLE" + +# Who produced an event (L7): the operator, the orchestrator's own code, or a +# model role (L9). planner and executor become driver in the new loop (M2.3). +ACTORS = frozenset( + { + "operator", + "orchestrator", + "planner", + "executor", + "driver", + "helper", + "expert", + "classifier", + } +) class StateSchemaError(RuntimeError): @@ -72,12 +65,29 @@ class StateSchemaError(RuntimeError): @dataclass(frozen=True) class AgentState: + """The state machine's view of the active work item (control.py, until + Development Plan M2.6 retires it).""" + state: str project: str | None data: dict[str, Any] updated_at: str +@dataclass(frozen=True) +class WorkItem: + id: str + scope: str | None + title: str + status: str # open | closed + outcome: str | None # completed | abandoned, once closed + position: str # where the loop is for this item + data: dict[str, Any] + opened_at: str + updated_at: str + closed_at: str | None + + @dataclass(frozen=True) class Intent: id: int @@ -86,16 +96,30 @@ class Intent: status: str created_at: str resolved_at: str | None + work_item: str | None = None _SCHEMA = [ "CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)", """ - CREATE TABLE agent_state ( - id INTEGER PRIMARY KEY CHECK (id = 1), - state TEXT NOT NULL, - project TEXT, + CREATE TABLE work_items ( + id TEXT PRIMARY KEY, + scope TEXT, + title TEXT NOT NULL, + status TEXT NOT NULL CHECK (status IN ('open', 'closed')), + outcome TEXT, + position TEXT NOT NULL, data TEXT NOT NULL DEFAULT '{}', + opened_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + closed_at TEXT + ) + """, + """ + CREATE TABLE agent ( + id INTEGER PRIMARY KEY CHECK (id = 1), + active_item TEXT REFERENCES work_items(id), + idle_from_event INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL ) """, @@ -106,11 +130,29 @@ _SCHEMA = [ payload TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', created_at TEXT NOT NULL, - resolved_at TEXT + resolved_at TEXT, + work_item TEXT + ) + """, + """ + CREATE TABLE canonical_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + type TEXT NOT NULL, + payload TEXT NOT NULL, + work_item TEXT, + actor TEXT NOT NULL + ) + """, + "CREATE INDEX canonical_events_by_item ON canonical_events (work_item, id)", + """ + CREATE TABLE eval_outcomes ( + event_id INTEGER PRIMARY KEY, + verdict TEXT NOT NULL, + note TEXT NOT NULL DEFAULT '', + marked_at TEXT NOT NULL ) """, - _CANONICAL_EVENTS_DDL, - _EVAL_OUTCOMES_DDL, ] @@ -150,6 +192,12 @@ class StateStore: self._in_transaction(self._create_schema) return stored = self.schema_version + if stored < OLDEST_SUPPORTED_VERSION <= self._target_version: + raise StateSchemaError( + f"state store schema version {stored} predates work items " + f"(version {OLDEST_SUPPORTED_VERSION}); start a fresh store " + "(scripts/bringup.sh start sets the old one aside)" + ) if stored > self._target_version: raise StateSchemaError( f"state schema version {stored} is newer than this code " @@ -200,17 +248,177 @@ class StateStore: ).fetchone() return int(row["value"]) - def get_state(self) -> AgentState | None: - row = self._conn.execute("SELECT * FROM agent_state WHERE id = 1").fetchone() - if row is None: - return None - return AgentState( - state=row["state"], - project=row["project"], - data=json.loads(row["data"]), - updated_at=row["updated_at"], + # --- work items (L1) --------------------------------------------------- + + def _ensure_agent_row(self) -> None: + self._conn.execute( + "INSERT INTO agent (id, active_item, updated_at) VALUES (1, NULL, ?) " + "ON CONFLICT(id) DO NOTHING", + (_now(),), ) + @staticmethod + def _item(row: sqlite3.Row) -> WorkItem: + return WorkItem( + id=row["id"], + scope=row["scope"], + title=row["title"], + status=row["status"], + outcome=row["outcome"], + position=row["position"], + data=json.loads(row["data"]), + opened_at=row["opened_at"], + updated_at=row["updated_at"], + closed_at=row["closed_at"], + ) + + def work_item(self, item_id: str) -> WorkItem | None: + row = self._conn.execute( + "SELECT * FROM work_items WHERE id = ?", (item_id,) + ).fetchone() + return None if row is None else self._item(row) + + def work_items(self, status: str | None = None) -> list[WorkItem]: + query = "SELECT * FROM work_items" + args: tuple[Any, ...] = () + if status is not None: + query += " WHERE status = ?" + args = (status,) + rows = self._conn.execute(query + " ORDER BY opened_at", args).fetchall() + return [self._item(row) for row in rows] + + def active_item(self) -> WorkItem | None: + row = self._conn.execute( + "SELECT w.* FROM agent a JOIN work_items w ON w.id = a.active_item " + "WHERE a.id = 1" + ).fetchone() + return None if row is None else self._item(row) + + def open_work_item( + self, + item_id: str, + *, + title: str, + scope: str | None = None, + position: str, + data: dict[str, Any] | None = None, + cause: str, + ) -> WorkItem: + now = _now() + self._conn.execute( + "INSERT INTO work_items (id, scope, title, status, position, data, " + "opened_at, updated_at) VALUES (?, ?, ?, 'open', ?, ?, ?, ?)", + (item_id, scope, title, position, json.dumps(data or {}), now, now), + ) + self.oplog.emit("work_item.open", cause=cause, work_item=item_id, scope=scope) + item = self.work_item(item_id) + assert item is not None + return item + + def set_active(self, item_id: str | None, *, cause: str) -> None: + """Point the agent at a work item, or at none (idle). Going idle + records where in the log idleness began, so messages sent while idle + can be carried into the next work item (carry_idle_messages).""" + self._ensure_agent_row() + previous = self._conn.execute( + "SELECT active_item FROM agent WHERE id = 1" + ).fetchone()["active_item"] + if item_id is None: + last = self._conn.execute( + "SELECT COALESCE(MAX(id), 0) AS last FROM canonical_events" + ).fetchone()["last"] + self._conn.execute( + "UPDATE agent SET active_item = NULL, idle_from_event = ?, " + "updated_at = ? WHERE id = 1", + (last, _now()), + ) + else: + self._conn.execute( + "UPDATE agent SET active_item = ?, updated_at = ? WHERE id = 1", + (item_id, _now()), + ) + if previous != item_id: + self.oplog.emit( + "work_item.switch", + cause=cause, + **{"from": previous, "to": item_id}, + ) + + def update_work_item( + self, + item_id: str, + *, + position: str, + data: dict[str, Any], + cause: str, + ) -> None: + """Record a work item's new position and data; the full data is + passed every time (no partial updates).""" + previous = self.work_item(item_id) + if previous is None: + raise KeyError(f"no work item {item_id!r}") + self._conn.execute( + "UPDATE work_items SET position = ?, data = ?, updated_at = ? WHERE id = ?", + (position, json.dumps(data), _now(), item_id), + ) + self.oplog.emit( + "state.transition", + cause=cause, + **{"from": previous.position, "to": position, "project": previous.scope}, + ) + + def close_work_item(self, item_id: str, *, outcome: str, cause: str) -> None: + now = _now() + row = self._conn.execute( + "UPDATE work_items SET status = 'closed', outcome = ?, closed_at = ?, " + "updated_at = ? WHERE id = ? AND status = 'open' RETURNING id", + (outcome, now, now, item_id), + ).fetchone() + if row is None: + raise KeyError(f"no open work item {item_id!r}") + self.oplog.emit( + "work_item.close", cause=cause, work_item=item_id, outcome=outcome + ) + + def carry_idle_messages(self, item_id: str) -> int: + """Copy what the operator said while the agent was idle into a work + item that has just opened (D1: messages sent before `use` are kept + for it). Copied as new events, with the original's ID in the source, + never by reassigning the originals: the log is never edited (L7).""" + self._ensure_agent_row() + since = self._conn.execute( + "SELECT idle_from_event FROM agent WHERE id = 1" + ).fetchone()["idle_from_event"] + rows = self._conn.execute( + "SELECT * FROM canonical_events WHERE work_item IS NULL " + "AND actor = 'operator' AND type = 'user_message' AND id > ? ORDER BY id", + (since,), + ).fetchall() + log = self.transcript(item_id, "operator") + for row in rows: + payload = json.loads(row["payload"]) + log.append( + UserMessage( + text=payload["text"], + source=f"{payload.get('source', '')} (sent while idle, #{row['id']})", + ) + ) + return len(rows) + + # --- the state machine's view (control.py, until M2.6) ----------------- + + def get_state(self) -> AgentState | None: + """The active work item as the state machine sees it: its position + is the state, IDLE when there is none. None only for a store the + agent has never started on.""" + agent = self._conn.execute("SELECT * FROM agent WHERE id = 1").fetchone() + if agent is None: + return None + item = self.active_item() + if item is None: + return AgentState(IDLE, None, {}, agent["updated_at"]) + return AgentState(item.position, item.scope, item.data, item.updated_at) + def transition( self, state: str, @@ -219,36 +427,55 @@ class StateStore: project: str | None = None, data: dict[str, Any] | None = None, ) -> None: - """Record the new state-machine position; the full position is passed - every time (no partial updates), committed and fsynced before return.""" - previous = self.get_state() - self._conn.execute( - """ - INSERT INTO agent_state (id, state, project, data, updated_at) - VALUES (1, ?, ?, ?, ?) - ON CONFLICT(id) DO UPDATE SET - state = excluded.state, - project = excluded.project, - data = excluded.data, - updated_at = excluded.updated_at - """, - (state, project, json.dumps(data or {}), _now()), - ) - self.oplog.emit( - "state.transition", - cause=cause, - **{ - "from": previous.state if previous else None, - "to": state, - "project": project, - }, - ) + """The state machine's transition, over work items: IDLE closes and + deactivates the active item (its outcome should already be recorded + with close_work_item); any other state updates the active item, + opening one if there is none.""" + self._ensure_agent_row() + active = self.active_item() + if state == IDLE: + if active is not None: + if active.status == "open": + self.close_work_item(active.id, outcome="closed", cause=cause) + self.set_active(None, cause=cause) + self.oplog.emit( + "state.transition", + cause=cause, + **{ + "from": active.position if active else None, + "to": IDLE, + "project": project, + }, + ) + return + if active is None: + item_id = str((data or {}).get("task_id") or f"task-{uuid.uuid4().hex[:8]}") + active = self.open_work_item( + item_id, + title=project or item_id, + scope=project, + position=IDLE, + cause=cause, + ) + self.set_active(item_id, cause=cause) + self.update_work_item(active.id, position=state, data=data or {}, cause=cause) - def begin_intent(self, kind: str, payload: dict[str, Any], *, cause: str) -> int: - """Durably record that a unit of work is about to start.""" + # --- intents (L8) ------------------------------------------------------ + + def begin_intent( + self, + kind: str, + payload: dict[str, Any], + *, + cause: str, + work_item: str | None = None, + ) -> int: + """Durably record that a unit of work is about to start, owned by + the work item it belongs to (L1).""" cursor = self._conn.execute( - "INSERT INTO intent_log (kind, payload, created_at) VALUES (?, ?, ?)", - (kind, json.dumps(payload), _now()), + "INSERT INTO intent_log (kind, payload, created_at, work_item) " + "VALUES (?, ?, ?, ?)", + (kind, json.dumps(payload), _now(), work_item), ) assert cursor.lastrowid is not None self.oplog.emit( @@ -286,32 +513,51 @@ class StateStore: status=row["status"], created_at=row["created_at"], resolved_at=row["resolved_at"], + work_item=row["work_item"], ) for row in rows ] - def append_event(self, event: CanonicalEvent) -> int: - """Append to the canonical transcript (L7); committed and fsynced - before return, like every other durable write.""" + # --- the canonical log (L7) ------------------------------------------- + + def transcript(self, work_item: str | None, actor: str) -> "Transcript": + """The log as one role sees it for one work item: what it reads is + that item's events only, and what it writes carries the item and the + actor. None is the agent-level scope (L6).""" + if actor not in ACTORS: + raise ValueError(f"unknown actor {actor!r}") + return Transcript(self, work_item, actor) + + def all_events(self, after: int = 0) -> list[StoredEvent]: + """Every event across work items, for the few readers whose job + spans them (evaluation capture, dumps). Model calls never use this.""" + rows = self._conn.execute( + "SELECT * FROM canonical_events WHERE id > ? ORDER BY id", (after,) + ).fetchall() + return [_stored(row) for row in rows] + + def _append(self, event: CanonicalEvent, work_item: str | None, actor: str) -> int: cursor = self._conn.execute( - "INSERT INTO canonical_events (ts, type, payload) VALUES (?, ?, ?)", - (_now(), event_type(event), json.dumps(to_payload(event))), + "INSERT INTO canonical_events (ts, type, payload, work_item, actor) " + "VALUES (?, ?, ?, ?, ?)", + ( + _now(), + event_type(event), + json.dumps(to_payload(event)), + work_item, + actor, + ), ) assert cursor.lastrowid is not None return cursor.lastrowid - def read_events(self, after: int = 0) -> list[StoredEvent]: + def _read(self, work_item: str | None, after: int) -> list[StoredEvent]: rows = self._conn.execute( - "SELECT * FROM canonical_events WHERE id > ? ORDER BY id", (after,) + "SELECT * FROM canonical_events WHERE work_item IS ? AND id > ? " + "ORDER BY id", + (work_item, after), ).fetchall() - return [ - StoredEvent( - id=row["id"], - ts=row["ts"], - event=from_payload(row["type"], json.loads(row["payload"])), - ) - for row in rows - ] + return [_stored(row) for row in rows] def mark_outcome(self, event_id: int, verdict: str, note: str = "") -> None: """Record whether a delegation or consult was good (L9). @@ -343,3 +589,45 @@ class StateStore: def __exit__(self, *exc_info: object) -> None: self.close() + + +def _stored(row: sqlite3.Row) -> StoredEvent: + return StoredEvent( + id=row["id"], + ts=row["ts"], + event=from_payload(row["type"], json.loads(row["payload"])), + work_item=row["work_item"], + actor=row["actor"], + ) + + +class Transcript: + """One work item's log, as one actor reads and writes it (L6, L7). + + Handed to each component per call by the orchestrator, which knows the + active work item, so a component can neither read another item's events + nor write without attribution. + """ + + def __init__(self, store: StateStore, work_item: str | None, actor: str) -> None: + self._store = store + self.work_item = work_item + self.actor = actor + + def append(self, event: CanonicalEvent) -> int: + """Committed and fsynced before return, like every durable write.""" + return self._store._append(event, self.work_item, self.actor) + + def read(self, after: int = 0) -> list[StoredEvent]: + return self._store._read(self.work_item, after) + + def as_actor(self, actor: str) -> "Transcript": + """The same work item, written by another role: e.g. the executor + handing a sub-task to the helper.""" + return self._store.transcript(self.work_item, actor) + + def begin_intent(self, kind: str, payload: dict[str, Any], *, cause: str) -> int: + """An intent owned by this transcript's work item.""" + return self._store.begin_intent( + kind, payload, cause=cause, work_item=self.work_item + ) diff --git a/nisse-dev/tests/test_abandon.py b/nisse-dev/tests/test_abandon.py index f4543a7..c0dc7aa 100644 --- a/nisse-dev/tests/test_abandon.py +++ b/nisse-dev/tests/test_abandon.py @@ -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") diff --git a/nisse-dev/tests/test_agent_build.py b/nisse-dev/tests/test_agent_build.py index 3fefb2c..c43ae32 100644 --- a/nisse-dev/tests/test_agent_build.py +++ b/nisse-dev/tests/test_agent_build.py @@ -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( diff --git a/nisse-dev/tests/test_approval_binding.py b/nisse-dev/tests/test_approval_binding.py index 4124a93..2605c85 100644 --- a/nisse-dev/tests/test_approval_binding.py +++ b/nisse-dev/tests/test_approval_binding.py @@ -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 diff --git a/nisse-dev/tests/test_approvals.py b/nisse-dev/tests/test_approvals.py index 58c6d9b..6734999 100644 --- a/nisse-dev/tests/test_approvals.py +++ b/nisse-dev/tests/test_approvals.py @@ -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"}, diff --git a/nisse-dev/tests/test_canonical.py b/nisse-dev/tests/test_canonical.py index 930982b..f740770 100644 --- a/nisse-dev/tests/test_canonical.py +++ b/nisse-dev/tests/test_canonical.py @@ -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) diff --git a/nisse-dev/tests/test_catalog.py b/nisse-dev/tests/test_catalog.py index e502d4f..1b8913a 100644 --- a/nisse-dev/tests/test_catalog.py +++ b/nisse-dev/tests/test_catalog.py @@ -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 diff --git a/nisse-dev/tests/test_config.py b/nisse-dev/tests/test_config.py index ca0a1b1..f17198e 100644 --- a/nisse-dev/tests/test_config.py +++ b/nisse-dev/tests/test_config.py @@ -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" diff --git a/nisse-dev/tests/test_context.py b/nisse-dev/tests/test_context.py index 463afa7..7fb6437 100644 --- a/nisse-dev/tests/test_context.py +++ b/nisse-dev/tests/test_context.py @@ -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 diff --git a/nisse-dev/tests/test_control.py b/nisse-dev/tests/test_control.py index 02cd139..9867401 100644 --- a/nisse-dev/tests/test_control.py +++ b/nisse-dev/tests/test_control.py @@ -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, diff --git a/nisse-dev/tests/test_delegation.py b/nisse-dev/tests/test_delegation.py index ed2f49d..084aa52 100644 --- a/nisse-dev/tests/test_delegation.py +++ b/nisse-dev/tests/test_delegation.py @@ -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 diff --git a/nisse-dev/tests/test_dump.py b/nisse-dev/tests/test_dump.py index 7e6d99d..1688e47 100644 --- a/nisse-dev/tests/test_dump.py +++ b/nisse-dev/tests/test_dump.py @@ -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 diff --git a/nisse-dev/tests/test_evals.py b/nisse-dev/tests/test_evals.py index 6f09c56..282b42a 100644 --- a/nisse-dev/tests/test_evals.py +++ b/nisse-dev/tests/test_evals.py @@ -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"] diff --git a/nisse-dev/tests/test_executor.py b/nisse-dev/tests/test_executor.py index df024c1..72c22aa 100644 --- a/nisse-dev/tests/test_executor.py +++ b/nisse-dev/tests/test_executor.py @@ -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 diff --git a/nisse-dev/tests/test_expert.py b/nisse-dev/tests/test_expert.py index 5434556..5097c61 100644 --- a/nisse-dev/tests/test_expert.py +++ b/nisse-dev/tests/test_expert.py @@ -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 diff --git a/nisse-dev/tests/test_llm.py b/nisse-dev/tests/test_llm.py index 970d226..c49d2a5 100644 --- a/nisse-dev/tests/test_llm.py +++ b/nisse-dev/tests/test_llm.py @@ -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", } diff --git a/nisse-dev/tests/test_no_stranding.py b/nisse-dev/tests/test_no_stranding.py index 58a097b..cc1761e 100644 --- a/nisse-dev/tests/test_no_stranding.py +++ b/nisse-dev/tests/test_no_stranding.py @@ -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") diff --git a/nisse-dev/tests/test_oplog.py b/nisse-dev/tests/test_oplog.py index 4e1404b..f533fa7 100644 --- a/nisse-dev/tests/test_oplog.py +++ b/nisse-dev/tests/test_oplog.py @@ -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",' diff --git a/nisse-dev/tests/test_planner.py b/nisse-dev/tests/test_planner.py index 0334242..8d60742 100644 --- a/nisse-dev/tests/test_planner.py +++ b/nisse-dev/tests/test_planner.py @@ -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 diff --git a/nisse-dev/tests/test_pr_open.py b/nisse-dev/tests/test_pr_open.py index c29aaab..4384c91 100644 --- a/nisse-dev/tests/test_pr_open.py +++ b/nisse-dev/tests/test_pr_open.py @@ -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(), diff --git a/nisse-dev/tests/test_pr_open_handling.py b/nisse-dev/tests/test_pr_open_handling.py index 0ecc91d..213af6b 100644 --- a/nisse-dev/tests/test_pr_open_handling.py +++ b/nisse-dev/tests/test_pr_open_handling.py @@ -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 diff --git a/nisse-dev/tests/test_quint_counterexamples.py b/nisse-dev/tests/test_quint_counterexamples.py index bdca1c7..6ddc20a 100644 --- a/nisse-dev/tests/test_quint_counterexamples.py +++ b/nisse-dev/tests/test_quint_counterexamples.py @@ -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] diff --git a/nisse-dev/tests/test_review_loop.py b/nisse-dev/tests/test_review_loop.py index 9a3f5c3..5bedce9 100644 --- a/nisse-dev/tests/test_review_loop.py +++ b/nisse-dev/tests/test_review_loop.py @@ -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, diff --git a/nisse-dev/tests/test_stop.py b/nisse-dev/tests/test_stop.py index 4137600..97a25e4 100644 --- a/nisse-dev/tests/test_stop.py +++ b/nisse-dev/tests/test_stop.py @@ -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: diff --git a/nisse-dev/tests/test_work_items.py b/nisse-dev/tests/test_work_items.py new file mode 100644 index 0000000..be03177 --- /dev/null +++ b/nisse-dev/tests/test_work_items.py @@ -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] diff --git a/tests/integration/nisse-test.yaml b/tests/integration/nisse-test.yaml index 2d4451f..5d36210 100644 --- a/tests/integration/nisse-test.yaml +++ b/tests/integration/nisse-test.yaml @@ -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 diff --git a/tests/integration/test_crash_durability.py b/tests/integration/test_crash_durability.py index a0aaeda..dad16db 100644 --- a/tests/integration/test_crash_durability.py +++ b/tests/integration/test_crash_durability.py @@ -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" diff --git a/tests/integration/test_delegation_live.py b/tests/integration/test_delegation_live.py index 4c73f6e..4cceb50 100644 --- a/tests/integration/test_delegation_live.py +++ b/tests/integration/test_delegation_live.py @@ -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" diff --git a/tests/integration/test_expert_live.py b/tests/integration/test_expert_live.py index 39856b3..a61fb6b 100644 --- a/tests/integration/test_expert_live.py +++ b/tests/integration/test_expert_live.py @@ -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?") diff --git a/tests/integration/test_milestone_live.py b/tests/integration/test_milestone_live.py index 18337b1..b196fc3 100644 --- a/tests/integration/test_milestone_live.py +++ b/tests/integration/test_milestone_live.py @@ -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() diff --git a/tests/integration/test_passthrough_live.py b/tests/integration/test_passthrough_live.py index 2e7c949..8ec328a 100644 --- a/tests/integration/test_passthrough_live.py +++ b/tests/integration/test_passthrough_live.py @@ -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" diff --git a/tests/integration/test_planning_live.py b/tests/integration/test_planning_live.py index 65d8c30..0ee41ae 100644 --- a/tests/integration/test_planning_live.py +++ b/tests/integration/test_planning_live.py @@ -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