Map A17 onto Outline's actual primitives, and record the MCP decision

Observed rather than assumed, after checking the running instance:
collections carry per-user permissions in three tiers and permissions
are collection-scoped, not per-document. So one collection per writing
principal, with read membership for everyone else.

- States precisely what that enforces: no other agent can write a
  given agent's world model. It does not enforce operator-versus-agent
  exclusion, since the operator holds manager on their own wiki; that
  half is procedural, via the break-glass routine that stops the
  agents first. Two different strengths, so the spec no longer implies
  one guarantee covers both
- Per-user grants, not groups: Outline groups are not federated with
  LLDAP, so using them would put membership somewhere nothing syncs.
  Three principals make the manual mapping trivial
- Outline API keys are per-user, so each agent's token must belong to
  that agent's own Outline user. A token minted from an admin account
  voids every boundary above while appearing to work — the same shape
  as the Gitea permission surprise
- Collections identified by ID from config, documents by ID recorded
  in the state store, which also makes writes idempotent across a
  retry
- The conditional-write requirement softens to a revision *read*:
  break-glass stops the agents, so compare-and-swap is unnecessary and
  only divergence detection is needed for A3

Also records Outline's MCP server as considered and deferred. A17's
enforcement depends on this library sitting between the model and the
store; MCP's purpose is direct model access, and a procedure the model
may decline to fetch does not constrain it. Still a candidate for the
discretionary search tool, with two things to verify first.

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

.gitea/workflows/ci.yml

@@ -1,7 +1,14 @@
 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:
 
 jobs:
   checks:

docs/Architecture-Build-Spec.md

@@ -175,7 +175,9 @@ Requirements:
     task. The agent has no path to approve itself.
 - **Approval sources.** A gate waits on an approval event from any registered
   **`ApprovalSource`** — the Matrix transport (A1), the Gitea comment poller (B2), or
-  the Vikunja comment stream (C4). First valid approval wins and is consumed; the
+  the Vikunja comment stream (C4). **The knowledge store (A17) is never an approval
+  source**, and never an instruction channel: a wiki page has neither the identity
+  guarantees nor the revision binding an approval requires. First valid approval wins and is consumed; the
   gate records which channel satisfied it, and the orchestrator echoes the outcome to
   the other channels so the records converge. Each source authenticates the approver
   against a per-channel operator allowlist (Matrix MXID, Gitea username, Vikunja
@@ -333,6 +335,7 @@ The durable record is a **canonical event log** in SQLite, not a provider-shaped
 | `delegation_started` | brief, target role, resolved model, caps |
 | `delegation_completed` | handback note, artifacts, tokens, duration, exit status |
 | `compaction` | summary, producing model, range of events replaced |
+| `knowledge_retrieved` | document text as read, document ID, revision, retrieved-at (A17) |
 
 Rules:
 - **Canonical tool-call IDs are minted locally.** Provider IDs are mapped at the
@@ -507,6 +510,167 @@ additional cost is marking outcomes as good or bad.
 Without this, "best model this week" is decided by benchmark chatter, and swapping the
 frontier tier becomes something to avoid because the effect is unmeasurable. With it, a
 new model is a config change and a re-run.
+### A17. Knowledge Store (Outline)
+
+**Purpose:** durable knowledge that outlives a single task — an agent-maintained
+**world model** (what infrastructure exists, how it is arranged) and, later,
+operator-authored **procedures**. Distinct from A2, which holds *task* state: A2
+is the agent's memory of what it is doing, A17 is its memory of what the world
+looks like. Neither is a fallback for the other.
+
+Part D describes the service; this defines the contract.
+
+**Layout — sharing is read access, never write access.**
+- Every document has **exactly one writer**: one agent, or the operator. Sharing
+  governs who may *read*. Concurrent writers therefore never exist, which
+  disposes of the last-write-wins objection Part D raises, without needing
+  distributed-write machinery.
+- **Agent-private** space per agent: its own world model.
+- **Shared** space: readable by the operator and designated agents. Default
+  membership is **operator + `nisse-ops`**. `nisse-dev` gets private space only —
+  B3 seals it from RFC1918 precisely so it cannot reach the NAS, Proxmox, or the
+  UDM, and a shared network-architecture document is a map of exactly those
+  things. Cross-agent read access is justified per collection, never assumed.
+- The dev agent's project conventions live in the **repo it is working on**, not
+  here: versioned with the code, reviewed in the PR, and impossible to
+  desynchronise from the project.
+
+**Agent-owned knowledge is always a cache, never authority.** An agent may rely on
+it to know *what exists and where to look*; it must confirm a specific fact
+against the live system *before acting on it*. Hard requirement for anything in
+C2's confirm-first tier; advisory for read-only and reversible actions.
+
+**Enforcement is structural, because a prompt is not enforcement.** "Treat this as
+a cache" is exactly the advisory guarantee A4 and A12 reject, and the model cannot
+be prevented from *believing* a fact it has read. The enforceable goal is
+narrower: **an unverified belief must not become an irreversible action.**
+1. **The store decides what it serves.** Facts carry provenance and an observed-at
+   timestamp; retrieval marks or withholds anything past a staleness bound for
+   high-consequence classes.
+2. **Verification as a precondition, checked against the log.** For confirm-first
+   actions, refuse unless a verifying observation exists *in this task's canonical
+   log*, recorded after the task began, naming the action's target. Inspectable
+   without asking the model anything — the same move as A12's observable triggers.
+   Approximate by nature, so build it for the confirm-first tier only.
+3. **Provenance at the human gate** — the primary reliance, because C2 already
+   stops these actions for confirmation. The prompt states evidence quality:
+   "target recalled from the world model, observed 2026-08-02, not verified this
+   session" versus "confirmed by `zfs list` two minutes ago". The operator is
+   already the authority there; this makes the risk legible rather than invisible.
+4. **Audit, not prevention, for the cheap tiers.** Log which facts were retrieved
+   and which actions followed, so a wrong fact is traceable afterwards.
+
+Residual risk, stated plainly: a model can retype a fact from context with no
+retrieval event, and no mechanism here catches that. That is why the world model
+stays out of the authority path entirely.
+
+**Never an instruction channel, and never an approval channel.**
+- **Task instructions never come from the knowledge store.** Work is driven by
+  Vikunja tickets (C4) and operator commands (B2). An agent must never read the
+  store looking for work to do.
+- **Approvals never come from the knowledge store.** The registered
+  `ApprovalSource`s are Matrix (A1), Gitea (B2), and Vikunja (C4). A wiki page is
+  not one and must not become one.
+- An agent must never write here intending to cause another agent to act. This
+  extends Part D's rejection of the store as a message bus: the erosion path is
+  one well-meaning "note for the other agent" at a time.
+
+**Corrections flow through transport, not through direct edits.** The operator
+corrects a fact by telling the agent, which is strictly better than editing the
+page by hand: a hand edit is invisible to the agent's record, whereas an
+instruction arrives on an authenticated channel from an allowlisted identity, is
+recorded in the canonical log, and can be reconciled against what the agent
+believed. Corrections are stored **verbatim with attribution**, never paraphrased —
+a model rewriting a precise fact loses precision invisibly.
+- **Break-glass:** the operator keeps write permission. The procedure reuses Part
+  F's idle flag — wait for idle, stop the agent, edit freely, restart. Single
+  writer holds trivially while the agent is down, and A2/A3 make stopping safe.
+- Because that permission exists, an agent re-reads and compares the revision
+  immediately before writing, and re-merges rather than clobbering if it moved.
+  A safety net for the rare edit, not a designed workflow.
+
+**Canonical log representation (A10).** Retrieved content is **snapshotted into the
+log** — the text, plus document id and revision for provenance — never referenced
+by handle alone: A10 forbids a provider handle being the only reference to content,
+and these documents are mutable. Snapshot once per task at first use, not per turn.
+**A3 consequence:** a task resumed across a restart carries the old snapshot, so
+boot reconciliation re-fetches snapshotted knowledge and flags divergence rather
+than silently continuing on a stale copy.
+
+**Retrieval is structural first, discretionary second** — the A12/A14 pattern.
+- Procedures are **bound to C2 action classes**: the policy taxonomy and the
+  procedure index are the same list. Binding is deterministic and auditable
+  ("which procedure governed this action?" has an exact answer), and costs nothing
+  extra because the taxonomy must exist anyway.
+- A model-invoked **search** exists *in addition* for discovery, never as a
+  replacement for the bound retrieval — a procedure the model may skip does not
+  constrain it.
+
+**Outage posture is asymmetric.** Unlike B2's poller, "never fails a task" does not
+generalise here.
+- **Write failure never blocks:** retry, or record locally and reconcile.
+- **Read failure for a procedure bound to a confirm-first action gates.**
+  Proceeding is not a delayed reaction; it is acting unconstrained on precisely the
+  operation someone wrote a procedure for.
+- Read failure for read-only or reversible actions proceeds, logging the gap.
+
+**Accepted single point of failure.** Once the agent is the sole author and Outline
+the only home, losing Outline loses the world model. No local mirror: infrastructure
+facts are re-derivable by observation, which is the same premise the cache rule
+rests on.
+
+**Deferred:** the operator-writable procedures collection. The interface is defined
+here; the content and the C2 binding list arrive with the policy taxonomy (Part H).
+
+**Concrete mapping onto Outline.** Observed in the running instance: a **collection**
+carries per-user permissions with three tiers — visible-to (read), read-write, and
+manager. Permissions are collection-scoped, not per-document.
+- Therefore **one collection per writing principal**, with read membership granted to
+  everyone else who needs it: `nisse-ops-world` (write: `nisse-ops`),
+  `nisse-dev-world` (write: `nisse-dev`), `procedures` (write: operator).
+- **What this enforces, precisely:** no *other agent* can write a given agent's world
+  model — the collision that would otherwise be silent and unattributable. It does not
+  enforce operator-versus-agent exclusion, because the operator holds manager on their
+  own wiki and can always grant themselves write. That half is procedural: the
+  break-glass routine stops the agents first, so concurrent writers do not exist while
+  editing.
+- **Assign per user, not per group.** Outline groups appear to be internal to Outline
+  rather than federated with LLDAP (A8), so using them would put membership in a second
+  place that nothing syncs. With three principals the manual mapping is trivial and
+  obvious; revisit only if the cast grows.
+- **Each agent's API token must belong to that agent's own Outline user.** Outline API
+  keys are per-user, so the token *is* the identity every permission above evaluates
+  against. A token minted from the operator's account — or an admin's — silently voids
+  the entire boundary while appearing to work.
+- **Identify collections and documents by ID, from config and state.** Collection IDs
+  come from env (`NISSE_OUTLINE_*_COLLECTION`), never discovered from "whatever the
+  token can write": a permission change must not silently redefine what an agent
+  considers its own. Document IDs are recorded in the state store (A2) when created and
+  used for updates, which also makes writing idempotent — a retry after a crash updates
+  the document instead of creating a second one.
+- **Revision reads, not conditional writes.** Because break-glass stops the agents,
+  compare-and-swap is unnecessary; what is needed is the ability to *observe* that a
+  document changed since the agent last read it, feeding the A3 divergence check.
+  Whether Outline exposes a revision identifier or only a modified timestamp is
+  unverified — either satisfies this.
+
+#### Considered and deferred: Outline's MCP server
+
+Outline can expose search, read, create and edit over MCP (remote, Streamable HTTP,
+enabled workspace-wide). **Not adopted for the retrieval path**, because A17's
+enforcement depends on this library sitting *between* the model and the store: staleness
+marking, snapshotting into the canonical log, binding procedures to action classes, and
+gating a failed confirm-first read all happen to a retrieval before the model sees it.
+MCP's purpose is the opposite — direct model access — and a procedure the model may
+decline to fetch does not constrain it (A12).
+
+It remains a reasonable candidate for the *discretionary* search tool later, and unlike
+the vendor agent SDKs A13 rejects, MCP is a neutral standard rather than lock-in: one
+client would serve many servers if others are ever added. Before adopting it, verify
+that it honours the key owner's **collection** permissions rather than mere workspace
+membership, and that the workspace-level toggle does not force broader exposure than
+intended.
+
 
 ---
 
@@ -550,6 +714,25 @@ PR_OPEN
   - *Outbound:* `git` + Gitea API to push branches and open PRs, via a dedicated
     Gitea **service account** with a scoped token. Record any runnable test service's
     URL/port in the PR description.
+  - **The project repo's default branch must be protected, and this is what makes
+    the PR gate real.** The service account needs write access to push branches —
+    and in Gitea, *write access includes merging pull requests*. Repository
+    ownership does not help: an account that merely collaborates can still merge,
+    and one that owns the repo can merge and delete it. So without branch
+    protection the agent can merge its own PR, and the review gate holds only
+    because the agent's code does not call the merge endpoint. That is precisely
+    the advisory guarantee A4 exists to rule out.
+    - Required on the default branch: **block direct pushes** by the agent
+      account, and **restrict merging** to the operator (a merge whitelist, or
+      required approvals that the agent's own account cannot satisfy).
+    - Feature-branch pushes are unaffected, so the agent's normal workflow is
+      unchanged.
+    - **Verify it rather than assume it.** The property is observable from
+      outside: the agent's own token must be refused when it attempts to merge
+      its own PR and when it pushes to the default branch. Assert both, so that
+      relaxing the protection later fails a test instead of silently removing
+      the gate. (Discovered the hard way — before protection was configured, the
+      agent's token merged its own PR successfully.)
   - *Inbound:* react to PR comments, commits, pushes. **Poll** the Gitea API
     (15–30s) rather than take webhooks, to keep the VM outbound-only. Intake is
     structured JSON — never via a chat bridge.
@@ -670,6 +853,10 @@ Requirements:
   back to Outline: searchable, off-box, and an extra externalized record that
   survives volume loss.
 - Access via Outline **REST API**, dedicated service account, token from Bitwarden.
+- **The library contract is A17**, which settles what this section left implicit:
+  per-agent private space plus a shared space, one writer per document, agent-owned
+  knowledge treated as a cache rather than authority, and the structural means of
+  keeping it that way.
 
 **Deliberately rejected:** using a shared notes store (Outline *or* an Obsidian vault)
 as an **agent-to-agent message bus**. It's unstructured, race-prone (last-write-wins,
@@ -748,7 +935,8 @@ each agent is built — not standing up the service itself.
    here). Crash-atomic writes and reconcile-on-boot must be in place first.
 10. **Extract `nisse-core`**; build the **Sysadmin Agent** (`nisse-ops`): policy
     engine (C2), SSO + scoped browser (A8/A6), secrets (A7), Vikunja control (C4),
-    Outline runbooks (D), scheduled checks, alert intake.
+    knowledge store (A17/D) — world model first, operator procedures when the
+    C2 taxonomy they bind to exists — scheduled checks, alert intake.
     - With the monorepo layout (Part H, decided), extraction is a **package split**
       within one tree: move code behind a package boundary, no release machinery
       required.
@@ -816,6 +1004,7 @@ nisse-core/              # shared library (extracted after nisse-dev works)
   secrets/               #   Bitwarden machine-account client         (A7)
   sso/                   #   LLDAP identity + auth path               (A8)
   logging/               #   structured operational logs              (A9)
+  knowledge/             #   Outline client, world model, procedures  (A17)
 
 nisse-dev/               # app: sealed VM                        (Part B)
   workflow (arch/milestone gating), Gitea PR layer, allow-all sandbox

nisse-dev/src/nisse_dev/approvals.py

@@ -0,0 +1,80 @@
+"""Plan externalisation and Gitea-backed approvals (spec A4, Part H).
+
+Plans are posted as comments on a per-project Gitea issue. The approval
+comment — its author, timestamp, and revision tag — is the durable off-box
+record, authenticated by Gitea's own auth. The state store keeps only a
+pointer to it, so boot reconciliation can re-derive gate state from Gitea
+even after the state volume is lost: external truth wins.
+
+Approvals arriving here bind to a revision exactly as they do over transport
+(A4). The first valid approval on any channel wins; the orchestrator echoes
+the outcome to the other channel so the records converge.
+"""
+
+from dataclasses import dataclass
+
+from nisse_dev.gitea import Comment, GiteaClient, GiteaError
+
+APPROVAL_HINT = (
+    "Reply `approve {tag}` to approve this revision, or comment with feedback "
+    "to reject it."
+)
+
+
+@dataclass(frozen=True)
+class PlanPost:
+    issue: int
+    comment_id: int
+
+
+class PlanBoard:
+    """The Gitea side of plan gating: one issue per project, one comment per
+    plan revision."""
+
+    def __init__(self, client: GiteaClient, repo: str) -> None:
+        self._client = client
+        self._repo = repo
+
+    @property
+    def repo(self) -> str:
+        return self._repo
+
+    def open_project_issue(self, project: str, task_id: str) -> int:
+        return self._client.create_issue(
+            self._repo,
+            title=f"nisse-dev: {project}",
+            body=(
+                f"Plan gate for project `{project}` (task `{task_id}`).\n\n"
+                "Each plan revision is posted here as a comment. Approving a "
+                "revision by comment is equivalent to approving it in Matrix; "
+                "the first valid approval on either channel wins."
+            ),
+        )
+
+    def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> PlanPost:
+        tag = f"rev-{rev}"
+        body = "\n".join(
+            [
+                f"## Plan for `{gate}` ({tag})",
+                "",
+                plan,
+                "",
+                "---",
+                APPROVAL_HINT.format(tag=tag),
+            ]
+        )
+        comment = self._client.post_comment(self._repo, issue, body)
+        return PlanPost(issue=issue, comment_id=comment.id)
+
+    def echo(self, issue: int, text: str) -> None:
+        """Mirror an outcome decided on another channel, so the Gitea record
+        does not silently disagree with Matrix."""
+        try:
+            self._client.post_comment(self._repo, issue, text)
+        except GiteaError:
+            pass
+
+    def comments_since(self, issue: int, after_id: int) -> list[Comment]:
+        return [
+            c for c in self._client.list_comments(self._repo, issue) if c.id > after_id
+        ]

nisse-dev/src/nisse_dev/control.py

@@ -94,6 +94,17 @@ class MilestonePublisher(Protocol):
     def comment(self, index: int, body: str) -> None: ...
 
 
+class PlanBoard(Protocol):
+    """The Gitea side of plan gating; the concrete one is in
+    nisse_dev.approvals."""
+
+    def open_project_issue(self, project: str, task_id: str) -> int: ...
+
+    def post_plan(self, issue: int, gate: str, rev: int, plan: str) -> Any: ...
+
+    def echo(self, issue: int, text: str) -> None: ...
+
+
 class PlanningAgent(Protocol):
     def plan(
         self,
@@ -133,6 +144,10 @@ def _fresh_data() -> dict[str, Any]:
         "task_id": None,
         "roles": None,  # A11 mapping, pinned at task start
         "pr": None,  # the open PR for the current milestone, once published
+        # Pointer to the Gitea plan record (Part H): the state store holds
+        # only this, so gate state can be re-derived from Gitea after a
+        # state-volume loss.
+        "board": None,
     }
 
 
@@ -147,6 +162,7 @@ class Orchestrator:
         executor: "MilestoneExecutor | None" = None,
         publisher: "MilestonePublisher | None" = None,
         gitea_operators: Iterable[str] = (),
+        board: "PlanBoard | None" = None,
     ) -> None:
         self._store = store
         self._transport = transport
@@ -159,6 +175,7 @@ class Orchestrator:
         # are different names for the same person, and neither implies
         # the other.
         self._gitea_operators = set(gitea_operators)
+        self._board = board
         # Transport and the comment poller deliver on separate threads;
         # state transitions must not interleave.
         self._lock = threading.RLock()
@@ -260,6 +277,19 @@ class Orchestrator:
         self._data = {**_fresh_data(), "task_id": task_id, "roles": pinned}
         self._project = project
         self._store.oplog.task_id = task_id
+        if self._board is not None:
+            try:
+                self._data["board"] = {
+                    "issue": self._board.open_project_issue(project, task_id),
+                    "last_comment": 0,
+                }
+            except GiteaError as exc:
+                # Gitea being down must not block planning; Matrix remains a
+                # complete approval channel on its own.
+                self._transport.send_message(
+                    f"note: could not open the Gitea plan issue ({exc}); "
+                    "approvals will be Matrix-only for this task"
+                )
         self._transition(PLANNING_ARCH, cause=f"operator command: use {project}")
         self._run_planning_turn("arch")
 
@@ -308,6 +338,7 @@ class Orchestrator:
             f"{action.content}\n\nreply 'approve' to approve rev-{rev}; "
             "any other reply is rejection feedback"
         )
+        self._post_plan_to_board(gate_name, rev, action.content)
         self._store.oplog.emit(
             "gate.open",
             cause=f"plan submitted (rev-{rev})",
@@ -317,6 +348,12 @@ class Orchestrator:
         )
 
     def _handle_gate_reply(self, msg: InboundMessage, text: str) -> None:
+        self._apply_gate_reply(sender=msg.sender, text=text, channel="transport")
+
+    def _apply_gate_reply(self, *, sender: str, text: str, channel: str) -> None:
+        """Shared by every approval source (A4). The allowlist is per channel:
+        a Matrix MXID and a Gitea login are different names, and holding one
+        says nothing about holding the other."""
         gate_data = self._data["gate"]
         name, rev = gate_data["name"], gate_data["rev"]
         milestone = self._data["milestone"]
@@ -332,9 +369,8 @@ class Orchestrator:
                 )
             return
         revision = PlanRevision(gate=name, rev=rev, content=gate_data["plan"])
-        result = Gate(revision, self._operators).evaluate(
-            msg.sender, text, channel="transport"
-        )
+        allowlist = self._operators if channel == "transport" else self._gitea_operators
+        result = Gate(revision, allowlist).evaluate(sender, text, channel=channel)
         if isinstance(result, Ignored):
             if result.notice:
                 self._transport.send_message(result.notice)
@@ -347,12 +383,19 @@ class Orchestrator:
                 rev=f"rev-{rev}",
                 channel=result.channel,
             )
+            self._echo_outcome(
+                channel,
+                f"{name} rev-{rev} rejected by {result.approver}: {result.feedback}",
+            )
             self._data["gate"] = None
             back = PLANNING_ARCH if name == "arch" else MILESTONE_PLANNING
             self._transition(back, cause=f"plan rejected ({name} rev-{rev})")
             self._run_planning_turn(name, feedback=result.feedback)
             return
         assert isinstance(result, Approved)
+        self._echo_outcome(
+            channel, f"{name} rev-{rev} approved by {result.approver} via {channel}"
+        )
         self._store.oplog.emit(
             "gate.approve",
             cause=f"approved by {result.approver}",
@@ -395,6 +438,52 @@ class Orchestrator:
             pending={"kind": "milestone_stopped", "milestone": milestone},
         )
 
+    def _echo_outcome(self, deciding_channel: str, text: str) -> None:
+        """A gate satisfied on one channel is echoed to the other, so neither
+        record silently disagrees about what was approved (A4)."""
+        board_state = self._data.get("board")
+        if deciding_channel != "gitea" and self._board is not None and board_state:
+            self._board.echo(board_state["issue"], text)
+        elif deciding_channel == "gitea":
+            self._transport.send_message(text)
+
+    def _post_plan_to_board(self, gate_name: str, rev: int, plan: str) -> None:
+        board_state = self._data.get("board")
+        if self._board is None or not board_state:
+            return
+        try:
+            post = self._board.post_plan(board_state["issue"], gate_name, rev, plan)
+        except GiteaError:
+            # Matrix already carries the plan; a Gitea outage degrades the
+            # durability of the record, not the ability to approve.
+            return
+        board_state["comment_id"] = post.comment_id
+
+    def handle_plan_comment(self, comment: Any) -> None:
+        """A comment on the project issue may satisfy an open plan gate (A4).
+
+        Approvals bind to a revision here exactly as over transport, and the
+        first valid approval on either channel wins.
+        """
+        with self._lock:
+            self._handle_plan_comment_locked(comment)
+
+    def _handle_plan_comment_locked(self, comment: Any) -> None:
+        if comment.author not in self._gitea_operators:
+            return
+        state = self._store.get_state()
+        if state is None or state.state not in (
+            AWAIT_ARCH_SIGNOFF,
+            AWAIT_MILESTONE_SIGNOFF,
+        ):
+            return
+        self._store.append_event(
+            UserMessage(text=comment.body, source=f"gitea:comment:{comment.id}")
+        )
+        self._apply_gate_reply(
+            sender=comment.author, text=comment.body.strip(), channel="gitea"
+        )
+
     def handle_review_comment(self, comment: Any) -> None:
         """React to a PR review comment (B2). Called from the poller thread.
 

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,6 +30,15 @@ plain-text summary and no tool call — that ends the milestone."""
 DEFAULT_MAX_TURNS = 24
 DEFAULT_FAILURE_LIMIT = 2
 
+# 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.
+_NUDGE = (
+    "\n\nYou have not made any change yet, so the plan cannot be complete. "
+    "Carry out the next step of the plan with a tool call, or run the tests "
+    "to show the work is already done."
+)
+
 
 @dataclass(frozen=True)
 class ExecutorOutcome:
@@ -57,8 +66,13 @@ 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
 
         for _turn in range(self._max_turns):
             try:
@@ -78,9 +92,24 @@ class LLMExecutor:
                     calls.append(event)
 
             if not calls:
-                return ExecutorOutcome("completed", "executor reported the plan done")
+                if did_work:
+                    return ExecutorOutcome(
+                        "completed", "executor reported the plan done"
+                    )
+                if not challenged:
+                    # Claiming completion without having touched anything is
+                    # not evidence of completion; ask once before believing it.
+                    challenged = True
+                    system += _NUDGE
+                    continue
+                return ExecutorOutcome(
+                    "escalate",
+                    "executor claimed the milestone was done without making any change",
+                )
 
             for call in calls:
+                if call.tool in MUTATING_TOOLS:
+                    did_work = True
                 outcome = self._workspace.dispatch(call.tool, call.arguments)
                 self._store.append_event(
                     ToolResult(

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

@@ -151,6 +151,12 @@ class GiteaClient:
             created_at=data.get("created_at", ""),
         )
 
+    def create_issue(self, repo: str, title: str, body: str) -> int:
+        data = self._request(
+            "POST", f"/repos/{repo}/issues", json={"title": title, "body": body}
+        )
+        return int(data["number"])
+
     def close_pull_request(self, repo: str, index: int) -> None:
         self._request("PATCH", f"/repos/{repo}/pulls/{index}", json={"state": "closed"})
 

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
@@ -26,12 +27,21 @@ from nisse_dev.canonical import (
     mint_call_id,
 )
 from nisse_dev.render import OpenAIChatRenderer
+from nisse_dev.render_anthropic import AnthropicMessagesRenderer
 
 
 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,
@@ -81,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("/"),
@@ -89,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":
@@ -106,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]}"
@@ -157,3 +168,111 @@ class LiteLLMClient:
 
     def close(self) -> None:
         self._http.close()
+
+
+class AnthropicPassthroughClient:
+    """LiteLLM's Anthropic passthrough (A5). Same interface as LiteLLMClient,
+    so a caller chooses a route rather than a different kind of client.
+
+    Delegations take this path because fidelity matters there (A13);
+    ask_expert and fallbacks use the normalized route, where it does not.
+    """
+
+    def __init__(
+        self,
+        base_url: str,
+        token: str,
+        *,
+        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("/"),
+            headers={
+                "Authorization": f"Bearer {token}",
+                "x-api-key": token,
+                "anthropic-version": "2023-06-01",
+            },
+            timeout=timeout_s,
+            transport=transport,
+        )
+        self._renderer = AnthropicMessagesRenderer()
+        self._max_tokens = max_tokens
+        self._sleep = sleep
+
+    @classmethod
+    def from_env(
+        cls, env: Mapping[str, str] = os.environ
+    ) -> "AnthropicPassthroughClient":
+        base_url = env.get("NISSE_LLM_BASE_URL")
+        token = env.get("NISSE_LLM_TOKEN")
+        if not base_url or not token:
+            raise LLMError("set NISSE_LLM_BASE_URL and NISSE_LLM_TOKEN")
+        return cls(base_url, token)
+
+    def complete(
+        self,
+        events: list[StoredEvent],
+        tools: list[dict[str, Any]] | None,
+        model: str,
+        system: str | None = None,
+    ) -> list[CanonicalEvent]:
+        body = self._renderer.render(
+            events, tools, model, system=system, max_tokens=self._max_tokens
+        )
+        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]}"
+            )
+        return self._parse(response.json(), requested_model=model)
+
+    def _parse(
+        self, data: dict[str, Any], requested_model: str
+    ) -> list[CanonicalEvent]:
+        content = data.get("content")
+        if not isinstance(content, list):
+            raise LLMError(f"malformed passthrough response: {str(data)[:300]}")
+        produced_by = data.get("model") or requested_model
+        out: list[CanonicalEvent] = []
+        for block in content:
+            kind = block.get("type")
+            if kind == "text" and block.get("text"):
+                out.append(AssistantText(text=block["text"], model=produced_by))
+            elif kind == "tool_use":
+                # Provider ids are mapped at the boundary and never used as a
+                # join key (A10); the canonical log owns identity.
+                out.append(
+                    ToolCall(
+                        call_id=mint_call_id(),
+                        tool=block["name"],
+                        arguments=block.get("input") or {},
+                    )
+                )
+            # Vendor-opaque reasoning blocks (thinking, redacted_thinking) are
+            # deliberately dropped for now: A10 requires storing them tagged
+            # with the producing model and never replaying them outside the
+            # session that made them, and the canonical schema has no event
+            # for that yet. Nothing requests extended thinking, so none arrive.
+        return out
+
+    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/render_anthropic.py

@@ -0,0 +1,184 @@
+"""Anthropic Messages renderer (spec A10, A13).
+
+The second adapter, and the reason A13 puts normalisation in the renderer
+rather than on the wire: the two vendors disagree *structurally*, not
+cosmetically.
+
+| | OpenAI-compatible | Anthropic Messages |
+|---|---|---|
+| system prompt | a message with role `system` | a top-level `system` field |
+| tool call | `tool_calls` array on the message | a `tool_use` content block |
+| tool result | a message with role `tool` | a `tool_result` block in a **user** message |
+| tool schema | `function.parameters` | `input_schema` |
+
+A middlebox that normalised between these would be lossy exactly where a
+long-running delegation lives, which is why delegations take the passthrough
+route and this renderer builds the vendor-native body.
+"""
+
+from typing import Any
+
+from nisse_dev.canonical import (
+    AssistantText,
+    Compaction,
+    DelegationCompleted,
+    DelegationStarted,
+    GateOpened,
+    PlanSubmitted,
+    StoredEvent,
+    ToolCall,
+    ToolResult,
+    UserMessage,
+)
+from nisse_dev.render import effective_events
+
+DEFAULT_MAX_TOKENS = 4096
+NO_RESULT_PLACEHOLDER = "[no result recorded]"
+
+
+def to_anthropic_tools(tools: list[dict[str, Any]] | None) -> list[dict[str, Any]]:
+    """Convert the OpenAI-shaped manifests the tool modules already define,
+    so a tool is declared once and rendered per vendor."""
+    converted: list[dict[str, Any]] = []
+    for tool in tools or []:
+        fn = tool.get("function", tool)
+        converted.append(
+            {
+                "name": fn["name"],
+                "description": fn.get("description", ""),
+                "input_schema": fn.get(
+                    "parameters", {"type": "object", "properties": {}}
+                ),
+            }
+        )
+    return converted
+
+
+class AnthropicMessagesRenderer:
+    def render(
+        self,
+        events: list[StoredEvent],
+        tools: list[dict[str, Any]] | None,
+        model: str,
+        system: str | None = None,
+        *,
+        max_tokens: int = DEFAULT_MAX_TOKENS,
+    ) -> dict[str, Any]:
+        kept = effective_events(events)
+        answered = {e.event.call_id for e in kept if isinstance(e.event, ToolResult)}
+        blocks: list[tuple[str, dict[str, Any]]] = []
+        for stored in kept:
+            blocks.extend(self._blocks_for(stored.event))
+            # A dangling tool_use is rejected outright by this API — more
+            # strictly than the OpenAI shape — so a crash between a call and
+            # its result must not poison the transcript (A3).
+            if isinstance(stored.event, ToolCall) and (
+                stored.event.call_id not in answered
+            ):
+                blocks.append(
+                    (
+                        "user",
+                        {
+                            "type": "tool_result",
+                            "tool_use_id": stored.event.call_id,
+                            "content": NO_RESULT_PLACEHOLDER,
+                        },
+                    )
+                )
+
+        body: dict[str, Any] = {
+            "model": model,
+            "max_tokens": max_tokens,
+            "messages": _coalesce(blocks),
+        }
+        if system is not None:
+            body["system"] = system
+        if tools:
+            body["tools"] = to_anthropic_tools(tools)
+        return body
+
+    def _blocks_for(self, event: object) -> list[tuple[str, dict[str, Any]]]:
+        match event:
+            case UserMessage(text=text):
+                return [("user", {"type": "text", "text": text})]
+            case AssistantText(text=text):
+                return [("assistant", {"type": "text", "text": text})]
+            case ToolCall(call_id=call_id, tool=tool, arguments=arguments):
+                return [
+                    (
+                        "assistant",
+                        {
+                            "type": "tool_use",
+                            "id": call_id,
+                            "name": tool,
+                            "input": arguments,
+                        },
+                    )
+                ]
+            case ToolResult(call_id=call_id, result=result):
+                return [
+                    (
+                        "user",
+                        {
+                            "type": "tool_result",
+                            "tool_use_id": call_id,
+                            "content": result,
+                        },
+                    )
+                ]
+            case PlanSubmitted(plan=plan):
+                return [("assistant", {"type": "text", "text": plan})]
+            case GateOpened(gate=gate, reason=reason):
+                return [
+                    (
+                        "user",
+                        {"type": "text", "text": f"[gate {gate} opened: {reason}]"},
+                    )
+                ]
+            case DelegationStarted(brief=brief, model=model):
+                return [
+                    (
+                        "assistant",
+                        {
+                            "type": "text",
+                            "text": f"[delegation started ({model}): {brief}]",
+                        },
+                    )
+                ]
+            case DelegationCompleted(handback=handback):
+                return [
+                    (
+                        "assistant",
+                        {"type": "text", "text": f"[delegation handback]\n{handback}"},
+                    )
+                ]
+            case Compaction(summary=summary):
+                return [
+                    (
+                        "user",
+                        {
+                            "type": "text",
+                            "text": f"[summary of earlier conversation]\n{summary}",
+                        },
+                    )
+                ]
+            case _:
+                raise ValueError(f"unrenderable event: {event!r}")
+
+
+def _coalesce(
+    blocks: list[tuple[str, dict[str, Any]]],
+) -> list[dict[str, Any]]:
+    """Merge adjacent blocks of the same role into one message.
+
+    The Messages API expects alternating roles, and a tool_result must sit in
+    the user message directly after the assistant's tool_use — which happens
+    naturally once same-role runs are merged.
+    """
+    messages: list[dict[str, Any]] = []
+    for role, block in blocks:
+        if messages and messages[-1]["role"] == role:
+            messages[-1]["content"].append(block)
+        else:
+            messages.append({"role": role, "content": [block]})
+    return messages

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_approvals.py

@@ -0,0 +1,210 @@
+"""Plan externalisation to Gitea and approvals arriving on either channel
+(spec A4, Part H)."""
+
+import io
+from pathlib import Path
+from typing import Any
+
+from nisse_dev.approvals import PlanBoard
+from nisse_dev.canonical import CanonicalEvent, ToolCall
+from nisse_dev.control import (
+    AWAIT_ARCH_SIGNOFF,
+    AWAIT_MILESTONE_SIGNOFF,
+    MILESTONE_PLANNING,
+    Orchestrator,
+)
+from nisse_dev.gitea import Comment, GiteaError
+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.transport import InMemoryTransport
+
+ROLES = RoleMap(executor="x", planner="y", delegate="y", consultant="y")
+
+
+class FakeGitea:
+    def __init__(self, *, fail: bool = False) -> None:
+        self.issues: list[dict[str, str]] = []
+        self.comments: list[tuple[int, str]] = []
+        self._fail = fail
+        self._next_id = 100
+
+    def create_issue(self, repo: str, title: str, body: str) -> int:
+        if self._fail:
+            raise GiteaError("gitea is down")
+        self.issues.append({"title": title, "body": body})
+        return len(self.issues)
+
+    def post_comment(self, repo: str, index: int, body: str) -> Comment:
+        if self._fail:
+            raise GiteaError("gitea is down")
+        self.comments.append((index, body))
+        self._next_id += 1
+        return Comment(
+            id=self._next_id, author="nisse-dev-testing", body=body, created_at="t"
+        )
+
+    def list_comments(self, repo: str, index: int, since: str | None = None):
+        return []
+
+
+class ScriptedClient:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    def complete(self, events, tools, model, system=None) -> list[CanonicalEvent]:
+        self.calls += 1
+        return [
+            ToolCall(
+                call_id=f"c{self.calls}",
+                tool="submit_plan",
+                arguments={"type": "x", "content": f"plan {self.calls}"},
+            )
+        ]
+
+
+def build(
+    tmp_path: Path, gitea: Any
+) -> tuple[Orchestrator, StateStore, InMemoryTransport, PlanBoard]:
+    store = StateStore(tmp_path / "state.db", oplog=OpLog("nisse-dev", io.StringIO()))
+    transport = InMemoryTransport()
+    board = PlanBoard(gitea, "o/sandbox")
+    orch = Orchestrator(
+        store,
+        transport,
+        LLMPlanningAgent(ScriptedClient(), store, ROLES),
+        operators={"don"},
+        roles=ROLES,
+        gitea_operators={"dongar"},
+        board=board,
+    )
+    orch.start()
+    return orch, store, transport, board
+
+
+def comment(body: str, author: str = "dongar", cid: int = 900) -> Comment:
+    return Comment(id=cid, author=author, body=body, created_at="t")
+
+
+def test_project_start_opens_an_issue_and_posts_the_plan(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    _, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+
+    assert gitea.issues[0]["title"] == "nisse-dev: demo"
+    issue, body = gitea.comments[0]
+    assert issue == 1
+    assert "plan 1" in body
+    assert "approve rev-1" in body  # the tag is printed for copy-paste (A4)
+    # The store keeps a pointer, not the plan record itself (Part H).
+    assert store.get_state().data["board"]["issue"] == 1
+
+
+def test_approval_by_gitea_comment_satisfies_the_gate(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    orch, store, _, _ = build(tmp_path, gitea)
+    orch._transport.receive("don", "use demo")
+    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
+
+    orch.handle_plan_comment(comment("approve rev-1"))
+
+    assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF
+
+
+def test_stale_revision_from_gitea_is_refused(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    orch, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+    transport.receive("don", "too vague")  # now at rev-2
+
+    orch.handle_plan_comment(comment("approve rev-1"))
+    assert store.get_state().state == AWAIT_ARCH_SIGNOFF  # not advanced
+
+    orch.handle_plan_comment(comment("approve rev-2", cid=901))
+    assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF
+
+
+def test_non_operator_gitea_comment_cannot_approve(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    orch, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+
+    orch.handle_plan_comment(comment("approve rev-1", author="drive-by"))
+    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
+
+
+def test_agents_own_comment_cannot_approve_its_plan(tmp_path: Path) -> None:
+    # The agent posts plans as nisse-dev-testing; that login is not on the
+    # operator allowlist, so its own comments can never satisfy its gate (A4).
+    gitea = FakeGitea()
+    orch, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+
+    orch.handle_plan_comment(comment("approve rev-1", author="nisse-dev-testing"))
+    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
+
+
+def test_rejection_by_gitea_comment_replans(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    orch, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+
+    orch.handle_plan_comment(comment("needs more detail on the schema"))
+
+    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
+    assert "plan 2" in transport.outbound[-1].text  # replanned
+
+
+def test_matrix_approval_is_echoed_to_gitea(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    _, _, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+    transport.receive("don", "approve")
+
+    echoes = [body for _, body in gitea.comments if "approved by don" in body]
+    assert echoes, "the Gitea record must not silently disagree with Matrix"
+
+
+def test_gitea_approval_is_echoed_to_matrix(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    orch, _, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+    orch.handle_plan_comment(comment("approve rev-1"))
+
+    assert any("approved by dongar via gitea" in m.text for m in transport.outbound)
+
+
+def test_gitea_outage_leaves_matrix_a_complete_channel(tmp_path: Path) -> None:
+    gitea = FakeGitea(fail=True)
+    _, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+
+    # Planning proceeded and the gate is usable despite Gitea being down.
+    assert store.get_state().state == AWAIT_ARCH_SIGNOFF
+    assert any("Matrix-only" in m.text for m in transport.outbound)
+    transport.receive("don", "approve")
+    assert store.get_state().state == AWAIT_MILESTONE_SIGNOFF
+
+
+def test_comments_are_ignored_outside_a_gate(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    orch, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+    transport.receive("don", "approve")  # arch approved; now at milestone gate
+    transport.receive("don", "approve")  # milestone approved, awaiting start
+    before = store.get_state().state
+
+    orch.handle_plan_comment(comment("approve rev-1", cid=902))
+    assert store.get_state().state == before
+
+
+def test_board_records_the_milestone_plan_too(tmp_path: Path) -> None:
+    gitea = FakeGitea()
+    _, store, transport, _ = build(tmp_path, gitea)
+    transport.receive("don", "use demo")
+    transport.receive("don", "approve")
+
+    assert store.get_state().state in (MILESTONE_PLANNING, AWAIT_MILESTONE_SIGNOFF)
+    posted = [body for _, body in gitea.comments if "milestone-1" in body]
+    assert posted, "the milestone plan must be externalised as well"

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_executor.py

@@ -88,11 +88,46 @@ def test_uses_executor_role_and_reduced_manifest(
     assert "milestone 2" in client.calls[0]["system"]
 
 
-def test_no_tool_call_ends_the_milestone(store: StateStore, ws: Workspace) -> None:
-    client = ScriptedClient([[AssistantText(text="all steps done", model="m")]])
+def test_no_tool_call_after_work_ends_the_milestone(
+    store: StateStore, ws: Workspace
+) -> None:
+    client = ScriptedClient(
+        [
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="all steps done", model="m")],
+        ]
+    )
     assert build(client, store, ws).run(milestone=1).status == "completed"
 
 
+def test_completion_claimed_without_any_work_is_challenged(
+    store: StateStore, ws: Workspace
+) -> None:
+    # A model that answers with prose on turn one has done nothing; believing
+    # it would record an empty milestone as a success.
+    client = ScriptedClient(
+        [
+            [AssistantText(text="looks done to me", model="m")],
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="now it is done", model="m")],
+        ]
+    )
+    outcome = build(client, store, ws).run(milestone=1)
+    assert outcome.status == "completed"
+    assert (ws.root / "a.py").exists()  # the nudge produced actual work
+    assert "not made any change yet" in client.calls[1]["system"]
+
+
+def test_persisting_in_an_empty_claim_escalates(
+    store: StateStore, ws: Workspace
+) -> None:
+    prose = [AssistantText(text="nothing to do", model="m")]
+    client = ScriptedClient([prose, prose])
+    outcome = build(client, store, ws).run(milestone=1)
+    assert outcome.status == "escalate"
+    assert "without making any change" in outcome.note
+
+
 def test_two_consecutive_command_failures_escalate(
     store: StateStore, ws: Workspace
 ) -> None:
@@ -126,11 +161,14 @@ def test_a_success_resets_the_failure_count(store: StateStore, ws: Workspace) ->
 def test_failed_reads_do_not_count_toward_escalation(
     store: StateStore, ws: Workspace
 ) -> None:
+    # Two failed reads in a row must not trip the failure limit: looking for
+    # a file that is not there is the model orienting itself, not failing.
     client = ScriptedClient(
         [
             [call("read_file", path="missing-a.txt")],
             [call("read_file", path="missing-b.txt")],
-            [AssistantText(text="looked around, done", model="m")],
+            [call("write_file", path="a.py", content="x")],
+            [AssistantText(text="done", model="m")],
         ]
     )
     assert build(client, store, ws, failure_limit=2).run(milestone=1).status == (
@@ -138,6 +176,20 @@ def test_failed_reads_do_not_count_toward_escalation(
     )
 
 
+def test_reading_alone_is_not_doing_the_work(store: StateStore, ws: Workspace) -> None:
+    # Exploration is not evidence a milestone was carried out, so a
+    # completion claim backed only by reads is still challenged.
+    client = ScriptedClient(
+        [
+            [call("list_dir", path=".")],
+            [AssistantText(text="looks done to me", model="m")],
+            [AssistantText(text="really, done", model="m")],
+        ]
+    )
+    outcome = build(client, store, ws).run(milestone=1)
+    assert outcome.status == "escalate"
+
+
 def test_turn_cap_is_enforced(store: StateStore, ws: Workspace) -> None:
     forever = [[call("list_dir", path=".")] for _ in range(50)]
     client = ScriptedClient(forever)

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_render_anthropic.py

@@ -0,0 +1,196 @@
+"""The Anthropic Messages shape differs structurally from the OpenAI one;
+these tests pin the differences that a normalising middlebox would blur."""
+
+import httpx
+import pytest
+
+from nisse_dev.canonical import (
+    AssistantText,
+    Compaction,
+    StoredEvent,
+    ToolCall,
+    ToolResult,
+    UserMessage,
+)
+from nisse_dev.llm import AnthropicPassthroughClient, LLMError
+from nisse_dev.render_anthropic import (
+    AnthropicMessagesRenderer,
+    to_anthropic_tools,
+)
+
+TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "run_command",
+            "description": "Run a shell command.",
+            "parameters": {
+                "type": "object",
+                "properties": {"command": {"type": "string"}},
+                "required": ["command"],
+            },
+        },
+    }
+]
+
+
+def stored(events: list) -> list[StoredEvent]:
+    return [StoredEvent(id=i + 1, ts="t", event=e) for i, e in enumerate(events)]
+
+
+def render(events: list, **kw) -> dict:
+    return AnthropicMessagesRenderer().render(
+        stored(events), kw.pop("tools", None), "claude-haiku-4-5", **kw
+    )
+
+
+def test_system_is_a_top_level_field_not_a_message() -> None:
+    body = render([UserMessage(text="hi", source="m")], system="you are nisse")
+    assert body["system"] == "you are nisse"
+    assert all(m["role"] != "system" for m in body["messages"])
+
+
+def test_max_tokens_is_always_present() -> None:
+    # The Messages API rejects a request without it.
+    assert render([UserMessage(text="hi", source="m")])["max_tokens"] > 0
+
+
+def test_tool_call_is_a_content_block_not_an_array() -> None:
+    body = render(
+        [ToolCall(call_id="call-1", tool="run_command", arguments={"command": "ls"})]
+    )
+    block = body["messages"][0]["content"][0]
+    assert block["type"] == "tool_use"
+    assert block["id"] == "call-1"
+    assert block["input"] == {"command": "ls"}
+    assert "tool_calls" not in body["messages"][0]
+
+
+def test_tool_result_is_a_user_block_directly_after_the_call() -> None:
+    body = render(
+        [
+            ToolCall(call_id="call-1", tool="run_command", arguments={}),
+            ToolResult(call_id="call-1", result="exit code: 0", success=True),
+        ]
+    )
+    assistant, user = body["messages"]
+    assert assistant["role"] == "assistant"
+    assert assistant["content"][0]["type"] == "tool_use"
+    assert user["role"] == "user"  # not a "tool" role, as OpenAI would have
+    assert user["content"][0] == {
+        "type": "tool_result",
+        "tool_use_id": "call-1",
+        "content": "exit code: 0",
+    }
+
+
+def test_adjacent_same_role_events_are_merged() -> None:
+    body = render(
+        [
+            UserMessage(text="one", source="m"),
+            UserMessage(text="two", source="m"),
+            AssistantText(text="reply", model="m"),
+        ]
+    )
+    assert [m["role"] for m in body["messages"]] == ["user", "assistant"]
+    assert [b["text"] for b in body["messages"][0]["content"]] == ["one", "two"]
+
+
+def test_dangling_tool_call_gets_a_placeholder_result() -> None:
+    body = render([ToolCall(call_id="call-x", tool="submit_plan", arguments={})])
+    assert [m["role"] for m in body["messages"]] == ["assistant", "user"]
+    assert body["messages"][1]["content"][0]["tool_use_id"] == "call-x"
+
+
+def test_tools_are_converted_to_input_schema() -> None:
+    [tool] = to_anthropic_tools(TOOLS)
+    assert tool["name"] == "run_command"
+    assert tool["input_schema"]["properties"] == {"command": {"type": "string"}}
+    assert "parameters" not in tool
+    assert "function" not in tool
+
+
+def test_compaction_still_replaces_its_range() -> None:
+    events = stored(
+        [
+            UserMessage(text="old", source="m"),
+            UserMessage(text="kept", source="m"),
+        ]
+    )
+    events.append(
+        StoredEvent(
+            id=3,
+            ts="t",
+            event=Compaction(summary="summary", model="m", replaces=[1, 1]),
+        )
+    )
+    body = AnthropicMessagesRenderer().render(events, None, "claude-haiku-4-5")
+    texts = [b["text"] for m in body["messages"] for b in m["content"]]
+    assert "old" not in texts
+    assert any("summary" in t for t in texts)
+
+
+# --- passthrough client --------------------------------------------------
+
+
+def client(handler) -> AnthropicPassthroughClient:
+    return AnthropicPassthroughClient(
+        "https://litellm.test", "tok", transport=httpx.MockTransport(handler)
+    )
+
+
+def test_client_posts_to_the_messages_route_with_version_header() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        seen["version"] = request.headers.get("anthropic-version")
+        return httpx.Response(
+            200,
+            json={
+                "model": "claude-haiku-4-5",
+                "content": [{"type": "text", "text": "hi"}],
+            },
+        )
+
+    events = stored([UserMessage(text="hello", source="m")])
+    [event] = client(handler).complete(events, None, "claude-haiku-4-5")
+    assert seen["url"] == "https://litellm.test/v1/messages"
+    assert seen["version"] == "2023-06-01"
+    assert event == AssistantText(text="hi", model="claude-haiku-4-5")
+
+
+def test_client_maps_tool_use_to_local_call_ids() -> None:
+    response = {
+        "model": "claude-haiku-4-5",
+        "content": [
+            {"type": "text", "text": "running it"},
+            {
+                "type": "tool_use",
+                "id": "toolu_vendor_xyz",
+                "name": "run_command",
+                "input": {"command": "pytest"},
+            },
+        ],
+    }
+    events = stored([UserMessage(text="go", source="m")])
+    text, call = client(lambda r: httpx.Response(200, json=response)).complete(
+        events, TOOLS, "claude-haiku-4-5"
+    )
+    assert isinstance(text, AssistantText)
+    assert isinstance(call, ToolCall)
+    assert call.arguments == {"command": "pytest"}
+    assert call.call_id != "toolu_vendor_xyz"  # provider id never a join key
+    assert call.call_id.startswith("call-")
+
+
+def test_client_errors_are_llm_errors() -> None:
+    events = stored([UserMessage(text="go", source="m")])
+    with pytest.raises(LLMError, match="400"):
+        client(lambda r: httpx.Response(400, text="bad")).complete(
+            events, None, "claude-haiku-4-5"
+        )
+    with pytest.raises(LLMError, match="malformed"):
+        client(lambda r: httpx.Response(200, json={"oops": 1})).complete(
+            events, None, "claude-haiku-4-5"
+        )

nisse-dev/tests/test_tools.py

@@ -1,4 +1,5 @@
 import os
+import time
 from pathlib import Path
 
 import pytest
@@ -97,6 +98,12 @@ def test_timeout_kills_the_whole_process_group(tmp_path: Path) -> None:
     assert pidfile.exists()
     pid = int(pidfile.read_text().strip())
     assert _is_running(os.getpid()), "liveness check is broken, not the kill"
+    # SIGKILL to a process group is asynchronous, and the grandchild is an
+    # orphan by now, so its exit is not synchronised with our return. Wait
+    # for death rather than assuming it is instantaneous.
+    deadline = time.monotonic() + 5
+    while _is_running(pid) and time.monotonic() < deadline:
+        time.sleep(0.05)
     assert not _is_running(pid)
 
 
@@ -127,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()

tests/integration/test_gitea_live.py

@@ -12,7 +12,7 @@ from pathlib import Path
 import pytest
 
 from nisse_dev.envfile import load_env_file
-from nisse_dev.git import GitRepo, authenticated_url
+from nisse_dev.git import GitError, GitRepo, authenticated_url
 from nisse_dev.gitea import GiteaClient, pull_request_reconciler
 from nisse_dev.state import Intent
 
@@ -88,6 +88,19 @@ def test_push_open_pr_and_reconcile(tmp_path: Path) -> None:
 
         # Now the same reconciler must see the effect and refuse to repeat it.
         assert reconcile(intent_for(branch)) is True
+
+        # A4 is only real if the server enforces it. The agent's own token
+        # must not be able to merge the PR it just opened, and must not be
+        # able to push to the protected default branch at all.
+        merged = client._http.post(
+            f"/repos/{repo_path}/pulls/{pr.number}/merge", json={"Do": "merge"}
+        )
+        assert merged.status_code >= 400, (
+            "the agent merged its own PR: branch protection is not enforcing "
+            "the human gate"
+        )
+        with pytest.raises(GitError):
+            repo.push(url, "main", token=token)
     finally:
         if pr is not None:
             client.close_pull_request(repo_path, pr.number)

tests/integration/test_passthrough_live.py

@@ -0,0 +1,108 @@
+"""Live Anthropic passthrough: the vendor-native route delegations will use.
+
+Proves the renderer's body is accepted as-is, that a tool call round-trips
+through the canonical log, and — per A5 — that prompt caching survives the
+passthrough rather than being flattened by the proxy.
+"""
+
+import io
+import os
+from pathlib import Path
+
+import httpx
+import pytest
+
+from nisse_dev.canonical import AssistantText, ToolCall, ToolResult, UserMessage
+from nisse_dev.envfile import load_env_file
+from nisse_dev.llm import AnthropicPassthroughClient
+from nisse_dev.oplog import OpLog
+from nisse_dev.state import StateStore
+from nisse_dev.tools import EXECUTOR_TOOLS
+
+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)",
+    ),
+]
+
+
+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)
+    return store
+
+
+def test_passthrough_completes_a_plain_turn(tmp_path: Path) -> None:
+    store = store_with(
+        tmp_path, UserMessage(text="Reply with exactly: pong", source="live")
+    )
+    result = AnthropicPassthroughClient.from_env().complete(
+        store.read_events(), None, os.environ[SENTINEL], system="Be terse."
+    )
+    texts = [e.text for e in result if isinstance(e, AssistantText)]
+    assert texts and "pong" in texts[0].lower()
+
+
+def test_passthrough_round_trips_a_tool_call(tmp_path: Path) -> None:
+    store = store_with(
+        tmp_path,
+        UserMessage(
+            text="List the files in the current directory using your tools.",
+            source="live",
+        ),
+    )
+    client = AnthropicPassthroughClient.from_env()
+    produced = client.complete(
+        store.read_events(),
+        EXECUTOR_TOOLS,
+        os.environ[SENTINEL],
+        system="Use a tool. Do not answer in prose.",
+    )
+    calls = [e for e in produced if isinstance(e, ToolCall)]
+    assert calls, f"expected a tool call, got {produced!r}"
+
+    # 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(
+        ToolResult(call_id=calls[0].call_id, result="app.py\ntests/", success=True)
+    )
+    follow_up = client.complete(
+        store.read_events(), EXECUTOR_TOOLS, os.environ[SENTINEL]
+    )
+    assert follow_up, "the vendor rejected or ignored the tool_result turn"
+
+
+def test_prompt_caching_survives_the_passthrough() -> None:
+    # A5 asks for this to be verified rather than assumed: a proxy that
+    # flattened the request would silently drop cache control and turn a
+    # cost optimisation into a per-call expense.
+    base = os.environ["NISSE_LLM_BASE_URL"].rstrip("/")
+    token = os.environ["NISSE_LLM_TOKEN"]
+    response = httpx.post(
+        f"{base}/v1/messages",
+        headers={
+            "Authorization": f"Bearer {token}",
+            "x-api-key": token,
+            "anthropic-version": "2023-06-01",
+        },
+        json={
+            "model": os.environ[SENTINEL],
+            "max_tokens": 16,
+            "messages": [{"role": "user", "content": "Say pong."}],
+        },
+        timeout=120,
+    )
+    response.raise_for_status()
+    usage = response.json().get("usage", {})
+    assert "cache_read_input_tokens" in usage, (
+        f"cache accounting absent from passthrough usage: {usage}"
+    )