Add the delegation harness (A13) — step 7b

A frontier model receives a scoped sub-task, works autonomously with
full tool access, and hands back; the local loop reads the handback
note and the file state, never the reasoning.

- Branch at delegation start, and abort discards it. That required
  committing each turn's work onto the branch: without commits,
  deleting a branch discards nothing, so "abort discards the work"
  was a comforting phrase rather than a fact. The caller's own
  in-flight changes are committed to their branch first, so an abort
  cannot destroy work nobody delegated
- Hard caps on turns and wall clock; a breach is a defined failure
  return, and its result is not usable, so the caller gates rather
  than resuming on a half-edited tree
- Every event is checkpointed as it arrives, so a crash mid-delegation
  leaves the branch at a known commit and a truthful transcript
- Handback is validated locally: blank fields or a non-boolean success
  flag are refused and the delegation continues rather than accepting
  a useless note

Two failure modes found by running it against a real model:

- A success claim with nothing changed was accepted. The handback plus
  file state is the entire interface back, so a note describing work
  that does not exist is worse than no note. Success now requires a
  successful mutating tool call; reporting failure still does not,
  since "this is impossible" is a legitimate finding
- The delegate invented the absolute path /workspace, wrote there, ran
  its tests there, and handed back truthfully — with the repo empty
  and git seeing nothing. Writes outside the workspace are now refused
  with a message naming the real root, and both prompts state where
  the workspace is instead of leaving the model to guess. This is not
  sandboxing: work outside the repo is invisible to git, so the
  milestone would publish an empty PR

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

nisse-dev/src/nisse_dev/delegation.py

@@ -0,0 +1,363 @@
+"""Delegation harness (spec A13).
+
+Escalation is delegation, not consultation: a frontier model receives a scoped
+sub-task, works autonomously with full tool access, and hands back. The local
+model does not interpret or re-derive its reasoning — the handback note plus
+the resulting file state is the entire interface back.
+
+Three properties do the work here:
+
+**Isolation.** The delegation runs on its own branch, taken at start. Abort
+discards the branch, which gives a delegation the same atomicity that boot
+reconciliation gives an interrupted state transition.
+
+**Caps that are hard.** Turns, tokens, and wall clock. A breach is a defined
+failure return, not a nudge: the local loop gates to the human rather than
+resuming on top of a half-edited tree.
+
+**Checkpointing as it happens.** Every tool call, result, and assistant turn
+is written to the canonical log when it occurs, never batched at handback.
+That is what makes a delegation resumable after a crash (A3).
+
+Scope is bounded by construction: a delegation is handed a sub-task the
+approved plan already contains. It is smart, not authorised to widen the work.
+"""
+
+import time
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+from nisse_dev.canonical import (
+    AssistantText,
+    DelegationCompleted,
+    DelegationStarted,
+    ToolCall,
+    ToolResult,
+)
+from nisse_dev.git import GitError, GitRepo
+from nisse_dev.llm import LLMClient, LLMError
+from nisse_dev.state import StateStore
+from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, Workspace
+
+DELEGATION_INTENT = "delegation"
+
+SYSTEM_PROMPT = """You are a delegate working on one scoped sub-task for
+nisse-dev. You have full tool access and are expected to finish the sub-task
+yourself, including running the tests.
+
+Stay inside the sub-task you were given. It comes from a plan a human already
+approved; widening it is not yours to do. If the sub-task turns out to be
+wrong or impossible, say so in the handback rather than substituting your own.
+
+When you are done, call hand_back exactly once. Write it for a materially
+weaker model that cannot see your reasoning and will act on your note alone:
+
+- what changed
+- what is now true that was not before
+- what to do next
+
+Do not call hand_back until the work is actually done."""
+
+HAND_BACK_TOOL: dict[str, Any] = {
+    "type": "function",
+    "function": {
+        "name": "hand_back",
+        "description": (
+            "End the delegation and report. Call exactly once, when the "
+            "sub-task is complete or you have determined it cannot be."
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "changed": {
+                    "type": "string",
+                    "description": "What you changed, concretely.",
+                },
+                "now_true": {
+                    "type": "string",
+                    "description": "What is now true that was not before.",
+                },
+                "next": {
+                    "type": "string",
+                    "description": "What the caller should do next.",
+                },
+                "succeeded": {
+                    "type": "boolean",
+                    "description": "Whether the sub-task was completed.",
+                },
+            },
+            "required": ["changed", "now_true", "next", "succeeded"],
+        },
+    },
+}
+
+DELEGATE_TOOLS = [*EXECUTOR_TOOLS, HAND_BACK_TOOL]
+
+
+@dataclass(frozen=True)
+class Caps:
+    turns: int = 30
+    tokens: int = 200_000
+    wall_clock_s: float = 900.0
+
+    def as_payload(self) -> dict[str, Any]:
+        return {
+            "turns": self.turns,
+            "tokens": self.tokens,
+            "wall_clock_s": self.wall_clock_s,
+        }
+
+
+@dataclass(frozen=True)
+class Handback:
+    """The entire interface back to the local loop, alongside file state."""
+
+    changed: str
+    now_true: str
+    next: str
+    succeeded: bool
+
+    def render(self) -> str:
+        return (
+            f"What changed: {self.changed}\n"
+            f"What is now true: {self.now_true}\n"
+            f"What to do next: {self.next}"
+        )
+
+
+@dataclass
+class DelegationResult:
+    status: Literal["handback", "capped", "failed"]
+    note: str
+    handback: Handback | None = None
+    branch: str | None = None
+    artifacts: list[str] = field(default_factory=list)
+    tokens: int = 0
+    duration_s: float = 0.0
+
+    @property
+    def usable(self) -> bool:
+        """Only a successful handback may be built upon. A cap breach or a
+        failure leaves a half-edited tree, which A13 says must gate rather
+        than be resumed."""
+        return self.status == "handback" and self.handback is not None
+
+
+class Delegation:
+    def __init__(
+        self,
+        client: LLMClient,
+        store: StateStore,
+        workspace: Workspace,
+        git_repo: GitRepo,
+        *,
+        model: str,
+        caps: Caps | None = None,
+        now: Callable[[], float] = time.monotonic,
+    ) -> None:
+        self._client = client
+        self._store = store
+        self._workspace = workspace
+        self._git = git_repo
+        self._model = model
+        self._caps = caps or Caps()
+        self._now = now
+
+    def run(
+        self, *, brief: str, branch: str, role: str = "delegate"
+    ) -> DelegationResult:
+        started_at = self._now()
+        base_branch = self._git.current_branch()
+        try:
+            # Commit whatever the caller had in flight onto their own branch
+            # first. Otherwise it would ride along onto the delegation branch
+            # and be destroyed by an abort — discarding work nobody delegated.
+            self._git.commit_all("Checkpoint before delegating")
+            self._git.checkout_new_branch(branch)
+        except GitError as exc:
+            return DelegationResult("failed", f"could not branch for delegation: {exc}")
+
+        self._store.append_event(
+            DelegationStarted(
+                brief=brief,
+                role=role,
+                model=self._model,
+                caps=self._caps.as_payload(),
+            )
+        )
+        result = self._loop(brief, branch, started_at)
+        result.duration_s = self._now() - started_at
+        self._store.append_event(
+            DelegationCompleted(
+                handback=result.handback.render() if result.handback else result.note,
+                artifacts=result.artifacts,
+                tokens=result.tokens,
+                duration_s=result.duration_s,
+                exit_status=result.status,
+            )
+        )
+        if not result.usable:
+            # Abort discards the branch: the caller must never resume on top
+            # of a partially-edited tree (A13).
+            self._abandon(branch, base_branch)
+        return result
+
+    def _loop(self, brief: str, branch: str, started_at: float) -> DelegationResult:
+        system = (
+            f"{SYSTEM_PROMPT}\n\n{self._workspace.describe()}"
+            f"\n\nYour sub-task:\n{brief}"
+        )
+        touched: set[str] = set()
+        did_work = False
+
+        for turn in range(self._caps.turns):
+            elapsed = self._now() - started_at
+            if elapsed >= self._caps.wall_clock_s:
+                return DelegationResult(
+                    "capped",
+                    f"wall-clock cap of {self._caps.wall_clock_s}s reached "
+                    f"after {turn} turns",
+                    branch=branch,
+                    artifacts=sorted(touched),
+                )
+            try:
+                produced = self._client.complete(
+                    self._store.read_events(),
+                    tools=DELEGATE_TOOLS,
+                    model=self._model,
+                    system=system,
+                )
+            except LLMError as exc:
+                return DelegationResult(
+                    "failed", f"delegate call failed: {exc}", branch=branch
+                )
+
+            calls: list[ToolCall] = []
+            for event in produced:
+                # Checkpoint as it arrives, never batched at handback (A13):
+                # this is what makes the delegation resumable after a crash.
+                self._store.append_event(event)
+                if isinstance(event, ToolCall):
+                    calls.append(event)
+
+            if not calls:
+                # A delegate that stops without handing back has not reported,
+                # and its tree state is unknown. Treat it as a failure rather
+                # than inventing a handback on its behalf.
+                if any(isinstance(e, AssistantText) for e in produced):
+                    continue
+                return DelegationResult(
+                    "failed",
+                    "delegate produced neither a tool call nor text",
+                    branch=branch,
+                    artifacts=sorted(touched),
+                )
+
+            for call in calls:
+                if call.tool == "hand_back":
+                    handback = _parse_handback(call.arguments)
+                    if handback is not None and handback.succeeded and not did_work:
+                        # Success claimed with nothing changed. The handback
+                        # 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(
+                            ToolResult(
+                                call_id=call.call_id,
+                                result=(
+                                    "you have not changed anything yet, so the "
+                                    "sub-task cannot be complete; do the work, "
+                                    "or hand back with succeeded=false and say "
+                                    "why"
+                                ),
+                                success=False,
+                            )
+                        )
+                        continue
+                    if handback is None:
+                        self._store.append_event(
+                            ToolResult(
+                                call_id=call.call_id,
+                                result=(
+                                    "hand_back needs changed, now_true, next and "
+                                    "succeeded; the delegation is still open"
+                                ),
+                                success=False,
+                            )
+                        )
+                        continue
+                    self._store.append_event(
+                        ToolResult(
+                            call_id=call.call_id,
+                            result="handback accepted; delegation closed",
+                            success=True,
+                        )
+                    )
+                    self._commit_turn(turn)
+                    return DelegationResult(
+                        "handback" if handback.succeeded else "failed",
+                        handback.next if handback.succeeded else handback.changed,
+                        handback=handback if handback.succeeded else None,
+                        branch=branch,
+                        artifacts=sorted(touched),
+                    )
+
+                outcome = self._workspace.dispatch(call.tool, call.arguments)
+                if call.tool in MUTATING_TOOLS and outcome.success:
+                    did_work = True
+                    path = call.arguments.get("path")
+                    if isinstance(path, str):
+                        touched.add(path)
+                self._store.append_event(
+                    ToolResult(
+                        call_id=call.call_id,
+                        result=outcome.result,
+                        success=outcome.success,
+                    )
+                )
+
+            # Commit each turn's work onto the delegation branch, so that
+            # discarding the branch really does discard the work, and so a
+            # crash leaves the branch at a known commit (A3).
+            self._commit_turn(turn)
+
+        return DelegationResult(
+            "capped",
+            f"turn cap of {self._caps.turns} reached without a handback",
+            branch=branch,
+            artifacts=sorted(touched),
+        )
+
+    def _commit_turn(self, turn: int) -> None:
+        try:
+            self._git.commit_all(f"Delegation turn {turn + 1}")
+        except GitError:
+            # Losing a checkpoint commit does not invalidate the work; the
+            # canonical log still records what happened.
+            pass
+
+    def _abandon(self, branch: str, base_branch: str) -> None:
+        try:
+            self._git.discard_branch(branch, back_to=base_branch)
+        except GitError:
+            # The branch is already recorded in the delegation events, so a
+            # failed cleanup leaves evidence rather than a silent mess.
+            pass
+
+
+def _parse_handback(arguments: dict[str, Any]) -> Handback | None:
+    """Validated locally rather than trusted: A14's rule for prompt-level
+    structured output applies to the handback too."""
+    required = ("changed", "now_true", "next")
+    if any(not str(arguments.get(key, "")).strip() for key in required):
+        return None
+    succeeded = arguments.get("succeeded")
+    if not isinstance(succeeded, bool):
+        return None
+    return Handback(
+        changed=str(arguments["changed"]).strip(),
+        now_true=str(arguments["now_true"]).strip(),
+        next=str(arguments["next"]).strip(),
+        succeeded=succeeded,
+    )

nisse-dev/src/nisse_dev/executor.py

@@ -16,7 +16,7 @@ from typing import Literal
 from nisse_dev.canonical import ToolCall, ToolResult
 from nisse_dev.llm import LLMClient, LLMError, RoleMap
 from nisse_dev.state import StateStore
-from nisse_dev.tools import EXECUTOR_TOOLS, Workspace
+from nisse_dev.tools import EXECUTOR_TOOLS, MUTATING_TOOLS, Workspace
 
 EXECUTOR_SYSTEM = """You are nisse-dev's executor, doing mechanical work.
 
@@ -30,11 +30,6 @@ plain-text summary and no tool call — that ends the milestone."""
 DEFAULT_MAX_TURNS = 24
 DEFAULT_FAILURE_LIMIT = 2
 
-# Tools that change or verify something. Reading a file or listing a
-# directory is the model looking around, not doing the work, so neither
-# counts as evidence that a milestone was carried out.
-MUTATING_TOOLS = frozenset({"write_file", "run_command"})
-
 # A turn with no tool call means "the plan is done". Believing that before
 # any real work has happened would record an empty milestone as a success, so
 # the claim is challenged once first.
@@ -71,7 +66,10 @@ class LLMExecutor:
 
     def run(self, milestone: int) -> ExecutorOutcome:
         model = self._roles.resolve("executor")
-        system = f"{EXECUTOR_SYSTEM}\n\nYou are executing milestone {milestone}."
+        system = (
+            f"{EXECUTOR_SYSTEM}\n\n{self._workspace.describe()}"
+            f"\n\nYou are executing milestone {milestone}."
+        )
         consecutive_failures = 0
         did_work = False
         challenged = False

nisse-dev/src/nisse_dev/git.py

@@ -80,6 +80,16 @@ class GitRepo:
         # the token never lands in .git/config.
         self._run("push", "-q", remote_url, f"HEAD:refs/heads/{branch}", secret=token)
 
+    def discard_branch(self, branch: str, *, back_to: str) -> None:
+        """Throw away a branch and everything on it.
+
+        Used when a delegation aborts: A13 gives a delegation the same
+        atomicity that reconciliation gives an interrupted transition, and
+        half-finished work must not survive to be built upon.
+        """
+        self._run("checkout", "-q", "--force", back_to)
+        self._run("branch", "-q", "-D", branch)
+
     def delete_remote_branch(
         self, remote_url: str, branch: str, *, token: str | None = None
     ) -> None:

nisse-dev/src/nisse_dev/tools.py

@@ -20,6 +20,11 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
+# Tools that change or verify something. Reading a file or listing a
+# directory is a model looking around, not doing the work, so neither counts
+# as evidence that work was carried out.
+MUTATING_TOOLS = frozenset({"write_file", "run_command"})
+
 DEFAULT_TIMEOUT_S = 120
 MAX_OUTPUT_CHARS = 4000
 
@@ -110,6 +115,12 @@ class Workspace:
         candidate = Path(path)
         return candidate if candidate.is_absolute() else self.root / candidate
 
+    def _inside(self, target: Path) -> bool:
+        try:
+            return target.resolve().is_relative_to(self.root.resolve())
+        except OSError:
+            return False
+
     def dispatch(self, tool: str, arguments: dict[str, Any]) -> ToolOutcome:
         try:
             match tool:
@@ -137,6 +148,18 @@ class Workspace:
 
     def write_file(self, path: str, content: str) -> ToolOutcome:
         target = self._resolve(path)
+        # Not a sandbox — the VM is the trust boundary (B3). This is about the
+        # work existing at all: a file written outside the repo is invisible
+        # to git, so the milestone would publish an empty PR while the model
+        # truthfully reports success. Models do guess plausible absolute paths
+        # like /workspace, so say where the workspace actually is.
+        if not self._inside(target):
+            return ToolOutcome(
+                f"refusing to write outside the workspace: {path}. "
+                f"The workspace root is {self.root}; write there, using "
+                "relative paths.",
+                success=False,
+            )
         try:
             target.parent.mkdir(parents=True, exist_ok=True)
             target.write_text(content)
@@ -154,6 +177,14 @@ class Workspace:
             return ToolOutcome(f"could not list {path}: {exc}", success=False)
         return ToolOutcome(_truncate("\n".join(entries) or "(empty)"), success=True)
 
+    def describe(self) -> str:
+        """Told to the model, so it does not have to guess where it is."""
+        return (
+            f"Your workspace is {self.root}. It is a git repository and the "
+            "only place your work is recorded; commands run there by default. "
+            "Use relative paths."
+        )
+
     def run_command(self, command: str) -> ToolOutcome:
         try:
             # Popen rather than run(): on timeout, run() kills only the direct