dongarandClaude Opus 5.5
4f152af170
CI / checks (push) Skipped
CI / checks (pull_request) Failing after 55s
Live run 3: the executor ran `git checkout -b <task id>` itself, and publishing then failed, because git can't create '<task>/milestone-1' while a branch named '<task>' exists. Milestone and delegation branches are now '<task>-milestone-N' and 'delegation-milestone-N-F'. D4's harness-owned branches (M2.1) remove the cause; this removes the collision on the current loop. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE
319 lines
13 KiB
Python
319 lines
13 KiB
Python
"""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}"
|
|
)
|