.gitea/workflows/ci.yml

@@ -1,6 +1,11 @@
 name: CI
 on:
+  # Only main on push. A PR branch is already covered by the pull_request
+  # event, and running both fired two identical suites concurrently — which
+  # doubled the LLM tokens the live tests spend and tripped the proxy's
+  # per-minute limit.
   push:
+    branches: [main]
   pull_request:
   # Manual re-run, e.g. after rotating a credential or raising a budget cap.
   workflow_dispatch:

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/llm.py

@@ -12,6 +12,7 @@ config change cannot split a task across two planners.
 
 import json
 import os
+import time
 from collections.abc import Mapping
 from dataclasses import asdict, dataclass
 from typing import Any, ClassVar, Protocol
@@ -33,6 +34,14 @@ class LLMError(RuntimeError):
     pass
 
 
+# A rate limit is a transient condition, not a bad request. Losing a
+# ten-turn delegation to one wastes far more tokens than waiting out the
+# window, so retry once before surfacing it (A13's caps still bound the
+# total).
+RATE_LIMIT_STATUS = 429
+RATE_LIMIT_RETRY_S = 20.0
+
+
 class LLMClient(Protocol):
     def complete(
         self,
@@ -82,6 +91,7 @@ class LiteLLMClient:
         *,
         timeout_s: float = 120.0,
         transport: httpx.BaseTransport | None = None,
+        sleep: Any = time.sleep,
     ) -> None:
         self._http = httpx.Client(
             base_url=base_url.rstrip("/"),
@@ -90,6 +100,7 @@ class LiteLLMClient:
             transport=transport,
         )
         self._renderer = OpenAIChatRenderer()
+        self._sleep = sleep
 
     @classmethod
     def from_env(cls, env: Mapping[str, str] = os.environ) -> "LiteLLMClient":
@@ -107,10 +118,9 @@ class LiteLLMClient:
         system: str | None = None,
     ) -> list[CanonicalEvent]:
         body = self._renderer.render(events, tools, model, system=system)
-        try:
-            response = self._http.post("/chat/completions", json=body)
-        except httpx.HTTPError as exc:
-            raise LLMError(f"LLM request failed: {exc}") from exc
+        response = _post_with_rate_limit_retry(
+            self._http, "/chat/completions", body, self._sleep
+        )
         if response.status_code != 200:
             raise LLMError(
                 f"LLM request returned {response.status_code}: {response.text[:500]}"
@@ -176,6 +186,7 @@ class AnthropicPassthroughClient:
         timeout_s: float = 300.0,
         max_tokens: int = 4096,
         transport: httpx.BaseTransport | None = None,
+        sleep: Any = time.sleep,
     ) -> None:
         self._http = httpx.Client(
             base_url=base_url.rstrip("/"),
@@ -189,6 +200,7 @@ class AnthropicPassthroughClient:
         )
         self._renderer = AnthropicMessagesRenderer()
         self._max_tokens = max_tokens
+        self._sleep = sleep
 
     @classmethod
     def from_env(
@@ -210,10 +222,9 @@ class AnthropicPassthroughClient:
         body = self._renderer.render(
             events, tools, model, system=system, max_tokens=self._max_tokens
         )
-        try:
-            response = self._http.post("/v1/messages", json=body)
-        except httpx.HTTPError as exc:
-            raise LLMError(f"passthrough request failed: {exc}") from exc
+        response = _post_with_rate_limit_retry(
+            self._http, "/v1/messages", body, self._sleep
+        )
         if response.status_code != 200:
             raise LLMError(
                 f"passthrough returned {response.status_code}: {response.text[:500]}"
@@ -251,3 +262,17 @@ class AnthropicPassthroughClient:
 
     def close(self) -> None:
         self._http.close()
+
+
+def _post_with_rate_limit_retry(
+    http: httpx.Client, path: str, body: dict[str, Any], sleep: Any
+) -> httpx.Response:
+    for attempt in (1, 2):
+        try:
+            response = http.post(path, json=body)
+        except httpx.HTTPError as exc:
+            raise LLMError(f"LLM request failed: {exc}") from exc
+        if response.status_code != RATE_LIMIT_STATUS or attempt == 2:
+            return response
+        sleep(RATE_LIMIT_RETRY_S)
+    raise AssertionError("unreachable")

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

nisse-dev/tests/test_delegation.py

@@ -0,0 +1,304 @@
+import io
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    CanonicalEvent,
+    DelegationCompleted,
+    DelegationStarted,
+    StoredEvent,
+    ToolCall,
+)
+from nisse_dev.delegation import (
+    DELEGATE_TOOLS,
+    Caps,
+    Delegation,
+)
+from nisse_dev.git import GitRepo
+from nisse_dev.llm import LLMError
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+from nisse_dev.tools import Workspace
+
+
+class ScriptedClient:
+    def __init__(self, turns: list[list[CanonicalEvent]]) -> None:
+        self._turns = turns
+        self.calls: list[dict[str, Any]] = []
+
+    def complete(
+        self,
+        events: list[StoredEvent],
+        tools: list[dict[str, Any]] | None,
+        model: str,
+        system: str | None = None,
+    ) -> list[CanonicalEvent]:
+        self.calls.append({"tools": tools, "model": model, "system": system})
+        if not self._turns:
+            return [AssistantText(text="still thinking", model=model)]
+        return self._turns.pop(0)
+
+
+class FailingClient:
+    def complete(self, *args: Any, **kwargs: Any) -> list[CanonicalEvent]:
+        raise LLMError("frontier unavailable")
+
+
+def call(tool: str, cid: str = "c1", **arguments: Any) -> ToolCall:
+    return ToolCall(call_id=cid, tool=tool, arguments=arguments)
+
+
+def hand_back(succeeded: bool = True, cid: str = "hb", **overrides: Any) -> ToolCall:
+    args = {
+        "changed": "added greet.py",
+        "now_true": "greet() returns a greeting and its test passes",
+        "next": "open the PR",
+        "succeeded": succeeded,
+    }
+    args.update(overrides)
+    return ToolCall(call_id=cid, tool="hand_back", arguments=args)
+
+
+@pytest.fixture
+def store(tmp_path: Path) -> StateStore:
+    return StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+
+
+@pytest.fixture
+def repo(tmp_path: Path) -> GitRepo:
+    r = GitRepo(tmp_path / "work")
+    r.init()
+    r.configure_identity("bot", "bot@example.test")
+    (r.root / "seed.txt").write_text("seed")
+    r.commit_all("seed")
+    return r
+
+
+@pytest.fixture
+def ws(repo: GitRepo) -> Workspace:
+    return Workspace(repo.root, timeout_s=10)
+
+
+def build(client: Any, store: StateStore, ws: Workspace, repo: GitRepo, **kw: Any):
+    return Delegation(client, store, ws, repo, model="claude-haiku-4-5", **kw)
+
+
+def test_successful_delegation_returns_a_structured_handback(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [
+            [call("write_file", path="greet.py", content="def greet(): ...")],
+            [hand_back()],
+        ]
+    )
+    result = build(client, store, ws, repo).run(brief="add greet", branch="d1")
+
+    assert result.usable
+    assert result.handback is not None
+    assert "greet.py" in result.handback.changed
+    assert "what to do next" in result.handback.render().lower()
+    assert result.artifacts == ["greet.py"]
+
+
+def test_delegation_works_on_its_own_branch(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [[call("write_file", path="a.py", content="x")], [hand_back()]]
+    )
+    build(client, store, ws, repo).run(brief="b", branch="deleg-1")
+    assert repo.current_branch() == "deleg-1"
+
+
+def test_abort_discards_the_branch_and_its_work(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # A13: abort must leave nothing to resume on top of.
+    client = ScriptedClient(
+        [
+            [call("write_file", path="half-done.py", content="oops")],
+            [hand_back(succeeded=False, changed="could not finish")],
+        ]
+    )
+    result = build(client, store, ws, repo).run(brief="b", branch="deleg-2")
+
+    assert not result.usable
+    assert repo.current_branch() == "main"
+    assert not (repo.root / "half-done.py").exists()
+
+
+def test_turn_cap_is_hard(store: StateStore, ws: Workspace, repo: GitRepo) -> None:
+    forever = [[call("list_dir", path=".")] for _ in range(20)]
+    client = ScriptedClient(forever)
+    result = build(client, store, ws, repo, caps=Caps(turns=3)).run(
+        brief="b", branch="d"
+    )
+
+    assert result.status == "capped"
+    assert "turn cap of 3" in result.note
+    assert len(client.calls) == 3
+
+
+def test_wall_clock_cap_is_hard(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    ticks = iter([0.0, 1.0, 99.0, 99.0, 99.0])
+    client = ScriptedClient([[call("list_dir", path=".")] for _ in range(10)])
+    result = build(
+        client,
+        store,
+        ws,
+        repo,
+        caps=Caps(turns=10, wall_clock_s=60.0),
+        now=lambda: next(ticks),
+    ).run(brief="b", branch="d")
+
+    assert result.status == "capped"
+    assert "wall-clock cap" in result.note
+
+
+def test_cap_breach_is_not_usable_and_discards_work(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient([[call("write_file", path="partial.py", content="x")]] * 5)
+    result = build(client, store, ws, repo, caps=Caps(turns=2)).run(
+        brief="b", branch="d"
+    )
+
+    assert not result.usable  # the caller must gate, not resume (A13)
+    assert not (repo.root / "partial.py").exists()
+
+
+def test_events_are_checkpointed_as_they_arrive(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [[call("write_file", path="a.py", content="x")], [hand_back()]]
+    )
+    build(client, store, ws, repo).run(brief="add a", branch="d")
+
+    kinds = [type(e.event).__name__ for e in store.read_events()]
+    assert kinds == [
+        "DelegationStarted",
+        "ToolCall",
+        "ToolResult",
+        "ToolCall",  # hand_back
+        "ToolResult",
+        "DelegationCompleted",
+    ]
+
+
+def test_composite_events_record_the_brief_caps_and_outcome(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [[call("write_file", path="a.py", content="x")], [hand_back()]]
+    )
+    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()]
+    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"
+    assert started.model == "claude-haiku-4-5"
+    assert started.caps == {"turns": 7, "tokens": 1234, "wall_clock_s": 42.0}
+    # A16: this pair is the eval record, so it must carry the outcome.
+    assert completed.exit_status == "handback"
+    assert "greet()" in completed.handback
+
+
+def test_hand_back_tool_is_offered_alongside_the_executor_tools() -> None:
+    names = [t["function"]["name"] for t in DELEGATE_TOOLS]
+    assert names[-1] == "hand_back"
+    assert "run_command" in names  # full tool access, not a consult (A13)
+
+
+def test_malformed_handback_is_rejected_and_the_delegation_continues(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [
+            [hand_back(cid="bad", changed="   ")],  # blank field
+            [call("write_file", path="a.py", content="x")],
+            [hand_back(cid="good")],
+        ]
+    )
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.usable  # recovered rather than accepting a useless note
+    assert len(client.calls) == 3
+
+
+def test_non_boolean_success_flag_is_rejected(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    client = ScriptedClient(
+        [
+            [call("write_file", path="a.py", content="x")],
+            [hand_back(succeeded="yes")],  # not a bool: unparseable
+            [hand_back(cid="ok")],
+        ]
+    )
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+    assert result.usable
+    assert len(client.calls) == 3
+
+
+def test_client_failure_is_a_defined_return_not_a_crash(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    result = build(FailingClient(), store, ws, repo).run(brief="b", branch="d")
+    assert result.status == "failed"
+    assert "frontier unavailable" in result.note
+    assert not result.usable
+
+
+def test_a_delegate_that_never_hands_back_is_capped_not_believed(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # Prose forever: the delegate never reports, so its tree state is unknown.
+    client = ScriptedClient([])
+    result = build(client, store, ws, repo, caps=Caps(turns=3)).run(
+        brief="b", branch="d"
+    )
+    assert result.status == "capped"
+    assert not result.usable
+
+
+def test_success_claimed_without_any_change_is_refused(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # The handback plus file state is the whole interface back, so a note
+    # describing work that was never done is worse than no note.
+    client = ScriptedClient(
+        [
+            [hand_back(cid="early")],
+            [call("write_file", path="real.py", content="x")],
+            [hand_back(cid="earned")],
+        ]
+    )
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.usable
+    assert len(client.calls) == 3  # the first claim was sent back
+    assert (repo.root / "real.py").exists()
+
+
+def test_failure_may_be_reported_without_doing_work(
+    store: StateStore, ws: Workspace, repo: GitRepo
+) -> None:
+    # "This sub-task is impossible" is a legitimate finding that requires no
+    # changes; only a claim of success needs evidence.
+    client = ScriptedClient(
+        [[hand_back(succeeded=False, changed="nothing: the API does not exist")]]
+    )
+    result = build(client, store, ws, repo).run(brief="b", branch="d")
+
+    assert result.status == "failed"
+    assert not result.usable
+    assert len(client.calls) == 1  # believed immediately, not challenged

nisse-dev/tests/test_llm.py

@@ -11,16 +11,17 @@ from nisse_dev.canonical import (
     UserMessage,
 )
 from nisse_dev.envfile import load_env_file
-from nisse_dev.llm import LiteLLMClient, LLMError, RoleMap
+from nisse_dev.llm import RATE_LIMIT_RETRY_S, LiteLLMClient, LLMError, RoleMap
 
 EVENTS = [StoredEvent(id=1, ts="t", event=UserMessage(text="hi", source="m:1"))]
 
 
-def make_client(handler) -> LiteLLMClient:
+def make_client(handler, sleep=lambda _s: None) -> LiteLLMClient:
     return LiteLLMClient(
         "http://litellm.test/v1",
         "sk-test",
         transport=httpx.MockTransport(handler),
+        sleep=sleep,  # never wait in unit tests
     )
 
 
@@ -109,9 +110,34 @@ def test_non_object_tool_arguments_raise_llm_error() -> None:
 
 
 def test_http_error_status_raises_llm_error() -> None:
-    client = make_client(lambda request: httpx.Response(429, text="rate limited"))
+    client = make_client(lambda request: httpx.Response(500, text="boom"))
+    with pytest.raises(LLMError, match="500"):
+        client.complete(EVENTS, tools=None, model="local/qwen")
+
+
+def test_rate_limit_is_retried_once_then_surfaced() -> None:
+    # A 429 is transient. Losing a long delegation to one costs far more
+    # than waiting out the window, but retrying forever would hide a real
+    # ceiling, so it is retried exactly once.
+    attempts: list[int] = []
+    waited: list[float] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        attempts.append(1)
+        return httpx.Response(429, text="rate limited")
+
+    client = make_client(handler, sleep=waited.append)
     with pytest.raises(LLMError, match="429"):
         client.complete(EVENTS, tools=None, model="local/qwen")
+    assert len(attempts) == 2
+    assert waited == [RATE_LIMIT_RETRY_S]
+
+
+def test_rate_limit_that_clears_succeeds() -> None:
+    responses = [httpx.Response(429, text="slow down"), completion({"content": "hi"})]
+    client = make_client(lambda request: responses.pop(0), sleep=lambda _s: None)
+    [event] = client.complete(EVENTS, tools=None, model="local/qwen")
+    assert isinstance(event, AssistantText)
 
 
 def test_connection_failure_raises_llm_error() -> None:

nisse-dev/tests/test_tools.py

@@ -134,3 +134,25 @@ def test_dispatch_unknown_tool_and_missing_argument(ws: Workspace) -> None:
     missing = ws.dispatch("read_file", {})
     assert not missing.success
     assert "missing argument" in missing.result
+
+
+def test_write_outside_the_workspace_is_refused(ws: Workspace, tmp_path: Path) -> None:
+    # Not sandboxing: a file written outside the repo is invisible to git, so
+    # the work would vanish while the model truthfully reported success.
+    # Models really do invent absolute paths like /workspace.
+    stray = tmp_path / "elsewhere" / "slugify.py"
+    outcome = ws.write_file(str(stray), "x")
+    assert not outcome.success
+    assert "outside the workspace" in outcome.result
+    assert str(ws.root) in outcome.result  # tells it where to write instead
+    assert not stray.exists()
+
+
+def test_escaping_via_parent_traversal_is_refused(ws: Workspace) -> None:
+    outcome = ws.write_file("../escaped.txt", "x")
+    assert not outcome.success
+    assert not (ws.root.parent / "escaped.txt").exists()
+
+
+def test_describe_names_the_root(ws: Workspace) -> None:
+    assert str(ws.root) in ws.describe()

tests/integration/test_delegation_live.py

@@ -0,0 +1,89 @@
+"""A real delegation over the Anthropic passthrough: scoped sub-task in,
+working code and a structured handback out."""
+
+import io
+import os
+from pathlib import Path
+
+import pytest
+
+from nisse_dev.delegation import Caps, Delegation
+from nisse_dev.envfile import load_env_file
+from nisse_dev.git import GitRepo
+from nisse_dev.llm import AnthropicPassthroughClient
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+from nisse_dev.tools import Workspace
+
+load_env_file(Path(__file__).resolve().parents[2] / ".env")
+
+SENTINEL = "NISSE_LIVE_PLANNER_MODEL"
+
+pytestmark = [
+    pytest.mark.integration,
+    pytest.mark.skipif(
+        not os.environ.get(SENTINEL),
+        reason=f"no live frontier route configured ({SENTINEL} unset)",
+    ),
+]
+
+BRIEF = """Create `slugify.py` in the workspace root containing a function
+`slugify(text)` that lowercases the text and replaces every run of
+non-alphanumeric characters with a single hyphen, with no leading or trailing
+hyphen. Then create `test_slugify.py` with at least two assertions covering
+it, and run `python -m pytest test_slugify.py -q` until it passes."""
+
+
+def build(tmp_path: Path, caps: Caps) -> tuple[Delegation, GitRepo, StateStore]:
+    repo = GitRepo(tmp_path / "work")
+    repo.init()
+    repo.configure_identity("nisse-dev-testing", "n@t.test")
+    (repo.root / "README.md").write_text("sandbox\n")
+    repo.commit_all("seed")
+    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+    delegation = Delegation(
+        AnthropicPassthroughClient.from_env(),
+        store,
+        Workspace(repo.root, timeout_s=60),
+        repo,
+        model=os.environ[SENTINEL],
+        caps=caps,
+    )
+    return delegation, repo, store
+
+
+def test_delegate_completes_a_scoped_subtask(tmp_path: Path) -> None:
+    delegation, repo, _store = build(tmp_path, Caps(turns=12, wall_clock_s=300.0))
+
+    result = delegation.run(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()
+
+    # The handback is the entire interface back, so it must actually say
+    # something in each part rather than be structurally present but empty.
+    handback = result.handback
+    assert handback is not None
+    assert len(handback.changed) > 10
+    assert len(handback.now_true) > 10
+    assert len(handback.next) > 5
+
+    # The delegate's own work is on its branch, and it really runs.
+    assert repo.current_branch() == "live/delegation-1"
+    check = Workspace(repo.root, timeout_s=60).run_command(
+        "python -m pytest test_slugify.py -q"
+    )
+    assert check.success, f"delegate's own tests do not pass: {check.result[:400]}"
+
+
+def test_a_cap_breach_discards_the_work(tmp_path: Path) -> 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(tmp_path, Caps(turns=1, wall_clock_s=120.0))
+
+    result = delegation.run(brief=BRIEF, branch="live/delegation-2")
+
+    assert not result.usable
+    assert result.status == "capped"
+    assert repo.current_branch() == "main"
+    assert not (repo.root / "slugify.py").exists()