"""Local executor loop (architecture L9 escalation triggers, L9 accommodations). Runs the approved plan with the executor role — the cheap local model doing mechanical work. Every tool call and result is checkpointed to the canonical log as it happens (L9), so a crash mid-milestone leaves a resumable record. Escalation is decided from outside the model (L9): consecutive command failures and hard caps on turns, never the model's own claim that it is stuck or finished-enough. Until delegation exists (step 7), an escalation gates to the human with the failures attached rather than engaging a frontier model. """ from collections.abc import Callable from dataclasses import dataclass from typing import Any, Literal from nisse_dev.cancel import Cancel, skip_calls 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 Transcript from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, ToolOutcome, Workspace EXECUTOR_SYSTEM = """You are nisse-dev's executor, doing mechanical work. Follow the approved plan below exactly. It was written to be executed verbatim; do not redesign it, expand its scope, or skip steps. Every reply must be a tool call. Verify your work by running the project's tests. When every step of the plan is done and the tests pass, call finish with a short summary — that, and nothing else, ends the milestone.""" DEFAULT_MAX_TURNS = 24 DEFAULT_FAILURE_LIMIT = 2 FINISH_TOOL: dict[str, Any] = { "type": "function", "function": { "name": "finish", "description": ( "End the milestone: every step of the plan is done and the tests pass." ), "parameters": { "type": "object", "properties": { "summary": {"type": "string", "description": "What was done."} }, "required": ["summary"], }, }, } # Finishing before any real work has happened would record an empty # milestone as a success, so the claim is refused once first. _NOT_YET = ( "You have not made any change yet, so the plan cannot be complete. Carry " "out the next step of the plan, or run the tests to show the work is " "already done, before calling finish." ) # A reply in prose breaks the tool contract. A server that enforces # tool_choice never allows one; on one that doesn't, the model is told once. _PROSE_NUDGE = ( "\n\nYour last reply was prose. Every reply must be a tool call: carry " "out the next step, or call finish if the plan is complete." ) @dataclass(frozen=True) class ExecutorOutcome: status: Literal["completed", "capped", "escalate", "stopped"] note: str class LLMExecutor: def __init__( self, client: LLMClient, roles: RoleMap, workspace: Workspace, *, max_turns: int = DEFAULT_MAX_TURNS, failure_limit: int = DEFAULT_FAILURE_LIMIT, expert: ExpertDesk | None = None, delegate: Delegation | None = None, notify: Callable[[str], None] | None = None, cancel: Cancel | None = None, ) -> None: self._client = client self._roles = roles self._workspace = workspace self._max_turns = max_turns self._failure_limit = failure_limit self._expert = expert self._delegate = delegate # The loop runs synchronously inside the transport callback, so # without this the agent is silent for the whole milestone and a # working run looks exactly like a hung one. self._notify = notify self._cancel = cancel or Cancel() def _report(self, turn: int, calls: list[ToolCall]) -> None: if self._notify is None: return summary = ", ".join( f"{c.tool}({c.arguments.get('path') or c.arguments.get('command', '')[:40]})" if c.arguments else c.tool for c in calls ) self._notify(f"turn {turn + 1}: {summary}") def _escalate( self, transcript: Transcript, milestone: int, failures: int, last_failure: str ) -> ExecutorOutcome | None: """Hand the sub-task to a stronger model, per L9. Returns None when the delegation handed back usable work and the local loop should carry on; an ExecutorOutcome when the caller must stop. """ note = f"{failures} consecutive command failures; last: {last_failure[:500]}" if self._delegate is None: return ExecutorOutcome("escalate", note) # L9 says the failures go in the brief: the delegate must not have to # rediscover what the local model already learned the hard way. brief = ( f"Milestone {milestone} is stuck. The local executor tried and its " f"commands failed {failures} times in a row.\n\n" f"Most recent failure:\n{last_failure[:2000]}\n\n" "The approved plan and the milestone so far are in the conversation " "above. Finish this sub-task, staying inside the approved plan." ) result = self._delegate.run( transcript.as_actor("helper"), brief=brief, branch=f"delegation/milestone-{milestone}-{failures}", ) if result.usable: return None if result.status == "stopped": return ExecutorOutcome("stopped", f"during a delegation: {result.note}") # L9: a cap breach or failure leaves a half-edited tree, so the local # loop must not resume on top of it. return ExecutorOutcome( "escalate", f"{note}; delegation {result.status}: {result.note}" ) def _tools(self) -> list[dict[str, Any]]: # L9: ask_expert exists in addition to the observable escalation # triggers, never as a replacement for them. if self._expert is None: return [*EXECUTOR_TOOLS, FINISH_TOOL] return [*EXECUTOR_TOOLS, ASK_EXPERT_TOOL, FINISH_TOOL] 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: 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 expert did not answer: {exc}", success=False) 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 plan is submitted between approval and execution, or during review responses. The executor sees that plan and what came after it, and not the planning conversation or earlier milestones, which a small context window cannot afford and the plan already distils. """ latest: StoredEvent | None = None for stored in transcript.read(): if isinstance(stored.event, PlanSubmitted): latest = stored if latest is None: return None, 0 assert isinstance(latest.event, PlanSubmitted) return latest.event.plan, latest.id 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(transcript) system = ( f"{EXECUTOR_SYSTEM}\n\n{self._workspace.describe()}" f"\n\nYou are executing milestone {milestone}." ) if plan is not None: # In the system prompt, so the context budget never drops it. system += f"\n\nThe approved plan:\n{plan}" tools = self._tools() offered = {tool["function"]["name"] for tool in tools} consecutive_failures = 0 did_work = False finish_refused = False prose_nudged = False for turn in range(self._max_turns): # L2: `stop` takes effect at the next tool boundary. The start of # a turn is one, and so is the gap between two tool calls below. if self._cancel.requested: return ExecutorOutcome("stopped", f"before turn {turn + 1}") try: produced = self._client.complete( transcript.view(after=after), tools=tools, model=model, system=system, ) except LLMError as exc: return ExecutorOutcome("escalate", f"executor call failed: {exc}") calls: list[ToolCall] = [] for event in produced: transcript.append(event) if isinstance(event, ToolCall): calls.append(event) if not calls: if not prose_nudged: prose_nudged = True system += _PROSE_NUDGE continue return ExecutorOutcome( "escalate", "executor kept replying in prose instead of tools" ) self._report(turn, calls) for index, call in enumerate(calls): if self._cancel.requested: 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: finish_refused = True outcome = ToolOutcome(_NOT_YET, success=False) elif not did_work: return ExecutorOutcome( "escalate", "executor claimed the milestone was done without " "making any change", ) else: transcript.append( ToolResult( call_id=call.call_id, result="milestone finished", success=True, ) ) skip_calls( transcript, calls[index + 1 :], "not run: finish had already ended the milestone", ) summary = str(call.arguments.get("summary", "")).strip() return ExecutorOutcome( "completed", summary or "executor reported the plan done" ) elif call.tool not in offered: # A model inventing tools is not doing the work, and with # no server enforcement nothing else would stop it, so it # counts toward escalation like a failed command (L9). outcome = ToolOutcome( f"no such tool: {call.tool}; the tools are " + ", ".join(sorted(offered)), success=False, ) elif call.tool == "ask_expert": outcome = self._consult(transcript, call.arguments) else: if call.tool in MUTATING_TOOLS: did_work = True outcome = self._workspace.dispatch(call.tool, call.arguments) transcript.append( ToolResult( call_id=call.call_id, result=outcome.result, success=outcome.success, ) ) # Commands and invented tools count toward escalation: a # failed read is the model looking around, a failed command # is work not working. if call.tool == "run_command" or call.tool not in offered: consecutive_failures = ( 0 if outcome.success else consecutive_failures + 1 ) if consecutive_failures >= self._failure_limit: # L9: this is the trigger delegation exists for. # 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, ) if escalation is not None: return escalation consecutive_failures = 0 did_work = True return ExecutorOutcome( "capped", f"hit the {self._max_turns}-turn cap for milestone {milestone}" )