.gitea/workflows/ci.yml

@@ -61,8 +61,10 @@ jobs:
     # the RTX 4090 busy for ~25 minutes. On PRs only, as its own job so the
     # checks above report first; a newer push to the same PR cancels the
     # run it supersedes. Locally, run only what you're writing (-k) or
-    # diagnosing (--lf).
+    # diagnosing (--lf). Only after the checks pass: a PR that's red already
+    # doesn't spend the GPU.
     if: github.event_name != 'push'
+    needs: checks
     runs-on: ubuntu-latest
     timeout-minutes: 90
     concurrency:

docs/Development-Plan.md

@@ -31,7 +31,7 @@ Only `nisse-dev` has real code; `nisse-core` and `nisse-ops` are placeholders.
 | **L6** Context | partial | Context fitting, per-role views and orchestrator facts (M1.2); no scope or agent memory |
 | **L7** Canonical log | partial | Work item and actor on every event, `fact` (M1.1, M1.2); no `classifier_verdict`, `knowledge_retrieved` |
 | **L8** Durability | mostly done | WAL store, intents, boot reconciliation, PR adoption, comment redelivery; to move onto the new loop |
-| **L9** Models and roles | partial | LiteLLM, delegation, `ask_expert`, eval capture; roles are still planner/executor, no capability profiles |
+| **L9** Models and roles | mostly done | LiteLLM, eval capture; on the new loop: `driver`/`helper`/`expert`/`classifier` pinned per item, capability profiles in briefings, delegation and `ask_expert` with triggers (M2.6) |
 | **L10** Knowledge vault | none | |
 | **L11** Injection | partial | Operator allowlists on every channel; no classifier |
 | **L12** Secrets, logging | partial | Structured state-change log; secrets in `.env`; no per-operation handout or redaction |
@@ -222,9 +222,23 @@ deletes afterwards.
     `ARCHITECTURE.md` and the build plan. A conversation waiting on provisioning
     doesn't close as idle.
 
-**M2.6 Roles and capability profiles.** *Pre-approved.* `driver`, `helper`,
+**M2.6 Roles and capability profiles.** *Pre-approved.* **Done** (this PR).
-`expert`, `classifier` (L9); escalation triggers; delegation and `ask_expert` on the
+`driver`, `helper`, `expert`, `classifier` (L9); escalation triggers; delegation and
-new loop.
+`ask_expert` on the new loop.
+- `roles.driver` (unset: the planner's model) and `roles.classifier` (unset: the
+  expert's), pinned per item; the driver and the go-ahead reader run on the item's.
+- `profiles:` in configuration (tools, strengths, weaknesses), used in briefings:
+  the driver is told what its helper and expert are good at, the helper who will
+  read its handback; never a model's name.
+- `delegate(task)` and `ask_expert(question)` for the driver. A helper works on a
+  branch off the milestone's; its work comes back as uncommitted changes on a
+  handback and is discarded otherwise, so harness branches never hold commits only
+  the clone has (the walk found a fix stranded that way when a PR merged mid-help).
+- Triggers: `failure_limit` failed commands in a row, or the turn cap, mid-
+  milestone, call a helper in once per milestone; if it fails too, the item waits
+  on the operator as before. A delegation cut off by a crash is discarded at boot.
+- Evaluation capture: delegations, `ask_expert` and the reader's verdicts.
+- Deferred: the context-won't-fit trigger (L6); the action classifier (M3.2).
 
 **M2.7 Conversation.** *Pre-approved.* Explanations, progress in prose, asking
 mid-work (L5, #42, #77).

nisse-dev.example.yaml

@@ -45,6 +45,27 @@ roles:                                # L9: role -> LiteLLM model alias
   planner: claude-opus-5-5      # high-end judgement: plans a weaker executor follows
   helper: claude-opus-5-5     # takes over what the executor got stuck on
   expert: claude-haiku-4-5  # one question at a time, no tools: cheap is enough
+  # The new loop's roles (loop: new): unset, driver is the planner's model and
+  # classifier the expert's. Each work item pins the mapping when it opens.
+  # driver: claude-opus-5-5     # drives the loop: talks, restates, works
+  # classifier: claude-haiku-4-5  # reads plain-language go-aheads (later: actions, L4)
+
+# Capability profiles (L9), from benchmarks: what one role is told about
+# another (never a model's name). Optional; a model without one is briefed as
+# "no profile recorded".
+profiles:
+  qwen3.8-27b:
+    tools: reliable (37/40 on the executor benchmark, PR #55)
+    strengths: careful multi-step edits that follow a clear brief
+    weaknesses: speed (minutes per hard task) and open-ended design
+  claude-haiku-4-5:
+    tools: reliable
+    strengths: quick, well-scoped changes and short answers
+    weaknesses: long multi-file work (15/20 on the harder benchmark tasks)
+  claude-opus-5-5:
+    tools: reliable
+    strengths: design, debugging, and work other models got stuck on
+    weaknesses: cost; use it for what needs it
 
 matrix:                               # L5
   homeserver: https://matrix.example

nisse-dev/src/nisse_dev/config.py

@@ -46,6 +46,8 @@ REQUIRED_KEYS = (
     "gitea.operators",
 )
 OPTIONAL_KEYS = (
+    "roles.driver",
+    "roles.classifier",
     "llm.max_output_tokens",
     "llm.request_timeout_s",
     "loop",
@@ -379,6 +381,48 @@ class BuildHost:
     port: int = 22
 
 
+@dataclass(frozen=True)
+class Profile:
+    """A model's capability profile (L9), from benchmarks: what one role is
+    told about another, never the model's name."""
+
+    tools: str = ""  # how reliably it keeps the tool-call contract
+    strengths: str = ""
+    weaknesses: str = ""
+
+    def describe(self) -> str:
+        parts = [
+            f"tool calls: {self.tools}" if self.tools else "",
+            f"strong at {self.strengths}" if self.strengths else "",
+            f"weak at {self.weaknesses}" if self.weaknesses else "",
+        ]
+        return "; ".join(p for p in parts if p) or "no profile recorded"
+
+
+def _parse_profiles(raw: Any, errors: list[str]) -> dict[str, "Profile"]:
+    """`profiles:`, model alias -> {tools, strengths, weaknesses}."""
+    if raw is None:
+        return {}
+    if not isinstance(raw, dict):
+        errors.append("profiles: must be a mapping of model alias to profile")
+        return {}
+    profiles: dict[str, Profile] = {}
+    for model, spec in raw.items():
+        where = f"profiles.{model}"
+        if not isinstance(spec, dict):
+            errors.append(f"{where}: must be a mapping")
+            continue
+        unknown = set(spec) - {"tools", "strengths", "weaknesses"}
+        if unknown:
+            errors.append(f"{where}.{min(unknown)}: unknown setting")
+            continue
+        if not all(isinstance(v, str) for v in spec.values()):
+            errors.append(f"{where}: values are text")
+            continue
+        profiles[str(model)] = Profile(**{k: v.strip() for k, v in spec.items()})
+    return profiles
+
+
 @dataclass(frozen=True)
 class Secret:
     value: str = field(repr=False)
@@ -426,6 +470,7 @@ class AgentConfig:
     gitea_token: str = field(repr=False)
     gitea_operators: frozenset[str]
     projects: tuple[Project, ...]
+    profiles: Mapping[str, Profile] = field(default_factory=dict)
     max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS
     request_timeout_s: float = DEFAULT_REQUEST_TIMEOUT_S
     tuning: Tuning = field(default_factory=Tuning)
@@ -520,7 +565,10 @@ class AgentConfig:
     ) -> "AgentConfig":
         tuning = _parse_tuning(tree.get("tuning"), errors)
         projects = _parse_projects(tree.get("projects"), env, errors)
-        flat = _flatten({k: v for k, v in tree.items() if k != "projects"})
+        profiles = _parse_profiles(tree.get("profiles"), errors)
+        flat = _flatten(
+            {k: v for k, v in tree.items() if k not in ("projects", "profiles")}
+        )
         flat.pop("tuning", None)
 
         known = set(REQUIRED_KEYS) | set(OPTIONAL_KEYS)
@@ -574,6 +622,8 @@ class AgentConfig:
                 planner=str(values["roles.planner"]),
                 helper=str(values["roles.helper"]),
                 expert=str(values["roles.expert"]),
+                driver=str(values.get("roles.driver") or ""),
+                classifier=str(values.get("roles.classifier") or ""),
             ),
             matrix=MatrixConfig(
                 homeserver=str(values["matrix.homeserver"]).rstrip("/"),
@@ -586,6 +636,7 @@ class AgentConfig:
             gitea_token=str(values["gitea.token"]),
             gitea_operators=_people(values["gitea.operators"]),
             projects=projects,
+            profiles=profiles,
             max_output_tokens=max_output,
             request_timeout_s=request_timeout,
             tuning=tuning,

nisse-dev/src/nisse_dev/delegation.py

@@ -54,8 +54,8 @@ 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
+When you are done, call hand_back exactly once. Write it for {reader},
-weaker model that cannot see your reasoning and will act on your note alone:
+which cannot see your reasoning and will act on your note alone:
 
 - what changed
 - what is now true that was not before
@@ -172,10 +172,14 @@ class Delegation:
         caps: Caps | None = None,
         now: Callable[[], float] = time.monotonic,
         cancel: Cancel | None = None,
+        reader: str = "",
     ) -> None:
         self._client = client
         self._store = store
         self._workspace = workspace
+        # Who acts on the handback, from its capability profile (L9): never
+        # a model's name.
+        self._reader = reader or "a materially weaker model"
         self._git = git_repo
         self._model = model
         self._backing_model = backing_model
@@ -323,7 +327,8 @@ class Delegation:
         self, transcript: Transcript, brief: str, branch: str, started_at: float
     ) -> DelegationResult:
         system = (
-            f"{SYSTEM_PROMPT}\n\n{self._workspace.describe()}"
+            f"{SYSTEM_PROMPT.format(reader=self._reader)}\n\n"
+            f"{self._workspace.describe()}"
             f"\n\nYour sub-task:\n{brief}"
         )
         touched: set[str] = set()

nisse-dev/src/nisse_dev/llm.py

@@ -12,7 +12,7 @@ config change cannot split a task across two planners.
 
 import json
 import time
-from dataclasses import asdict, dataclass
+from dataclasses import dataclass
 from typing import Any, ClassVar, Protocol
 
 import httpx
@@ -107,21 +107,40 @@ class LLMClient(Protocol):
 
 @dataclass(frozen=True)
 class RoleMap:
-    executor: str
+    executor: str  # the old loop's (control.py), until M2.8
-    planner: str
+    planner: str  # the old loop's, until M2.8
-    helper: str
+    helper: str  # delegations (L9)
-    expert: str
+    expert: str  # ask_expert (L9)
+    driver: str = ""  # drives the new loop; unset, the planner's model
+    classifier: str = ""  # reads go-aheads now, actions later (L4); unset, expert's
 
-    ROLES: ClassVar[tuple[str, ...]] = ("executor", "planner", "helper", "expert")
+    ROLES: ClassVar[tuple[str, ...]] = (
+        "executor",
+        "planner",
+        "helper",
+        "expert",
+        "driver",
+        "classifier",
+    )
+    _FALLBACK: ClassVar[dict[str, str]] = {"driver": "planner", "classifier": "expert"}
 
     def resolve(self, role: str) -> str:
         if role not in self.ROLES:
             raise LLMError(f"unknown role: {role!r}")
-        return str(getattr(self, role))
+        model = str(getattr(self, role))
+        if not model and role in self._FALLBACK:
+            return self.resolve(self._FALLBACK[role])
+        return model
 
     def pinned(self) -> dict[str, str]:
         """The full resolved mapping, to record at task start (L9)."""
-        return asdict(self)
+        return {role: self.resolve(role) for role in self.ROLES}
+
+    @classmethod
+    def from_pinned(cls, pinned: dict[str, str], default: "RoleMap") -> "RoleMap":
+        """A work item's pinned mapping; a role it predates comes from
+        ``default``."""
+        return cls(**{r: pinned.get(r) or default.resolve(r) for r in cls.ROLES})
 
 
 class LiteLLMClient:

nisse-dev/src/nisse_dev/loop/agent.py

@@ -15,6 +15,7 @@ a message is recorded and changes its item's state, but never interrupts a
 run (L1).
 """
 
+import contextlib
 import re
 import threading
 import time
@@ -23,12 +24,22 @@ from collections.abc import Callable, Iterable, Sequence
 from dataclasses import dataclass, replace
 
 from nisse_dev.cancel import Cancel
-from nisse_dev.canonical import Fact, PlanSubmitted, UserMessage
+from nisse_dev.canonical import (
+    DelegationCompleted,
+    DelegationStarted,
+    Fact,
+    PlanSubmitted,
+    UserMessage,
+)
 from nisse_dev.config import GENERAL
+from nisse_dev.delegation import DELEGATION_INTENT, Delegation
+from nisse_dev.expert import BudgetExhausted, ExpertDesk
 from nisse_dev.gates import command_form
 from nisse_dev.git import GitError
 from nisse_dev.gitea import GiteaClient, GiteaError, Issue
+from nisse_dev.llm import LLMError, RoleMap
 from nisse_dev.loop.driver import Driver, RunOutcome
+from nisse_dev.loop.help import ASK_EXPERT, DELEGATE, Helpers
 from nisse_dev.loop.items import (
     ANSWER,
     BLOCKED,
@@ -118,7 +129,7 @@ edge cases, and the tests that are its acceptance criteria. Commit it with the
 first milestone, and mark each milestone done in it in that milestone's PR.
 Keep ARCHITECTURE.md current: a design change updates it in the same PR.
 
-{labels}Never label the issue yourself; you may suggest a label to the operator.
+{labels}{helpers}Never label the issue yourself; you may suggest a label to the operator.
 
 {workspace}"""
 
@@ -212,6 +223,7 @@ class LoopAgent:
         # A new project's entry, once the operator has added it (D1, D8):
         # the projects configured now that this agent doesn't know yet.
         reload: Callable[[set[str]], Sequence[Project]] | None = None,
+        helpers: Helpers | None = None,
         clock: Callable[[], float] = time.time,
         intake_interval_s: float = INTAKE_INTERVAL_S,
     ) -> None:
@@ -233,6 +245,7 @@ class LoopAgent:
         self._grace = bug_grace_s
         self._idle = general_idle_s
         self._reload = reload
+        self._helpers = helpers
         self._clock = clock
         self._interval = intake_interval_s
         # thread root -> revision on screen there, read on the transport's
@@ -318,6 +331,7 @@ class LoopAgent:
         self._store.set_active(None, cause="boot")
         self._reconcile_publishing()
         self._reconcile_filing()
+        self._reconcile_delegations()
         for item in self._store.work_items("open"):
             loop = loop_of(item)
             self._threads[(item.scope or "", loop.issue)] = loop.thread
@@ -338,6 +352,30 @@ class LoopAgent:
                         thread=loop.thread,
                     )
 
+    def _reconcile_delegations(self) -> None:
+        """L8, L9: a delegation cut off by a crash is discarded, its branch
+        deleted and the milestone's left as it was. The item, cut off mid-
+        run, then waits on the operator like any other."""
+        for intent in self._store.pending_intents():
+            if intent.kind != DELEGATION_INTENT or not intent.work_item:
+                continue
+            item = self._store.work_item(intent.work_item)
+            branch = str(intent.payload.get("branch", ""))
+            base = str(intent.payload.get("base", ""))
+            project = self._projects.get(item.scope or "") if item else None
+            milestones = project.milestones if project else None
+            if milestones is not None and branch and base:
+                with contextlib.suppress(GitError):
+                    milestones.git.reset_to(base, base)
+                with contextlib.suppress(GitError):
+                    milestones.git._git("branch", "-q", "-D", branch)
+            self._store.abort_intent(intent.id, cause="cut off by a restart")
+            if item is not None:
+                self._fact(
+                    item,
+                    f"a helper's work on {branch} was cut off by a restart and discarded",
+                )
+
     def _reconcile_filing(self) -> None:
         """L8: a record issue cut off by a crash. An issue carrying the new
         item's ID was opened: its item is started now. Otherwise nothing was
@@ -1096,7 +1134,30 @@ class LoopAgent:
             for e in self._store.transcript(item.id, "driver").read()
             if isinstance(e.event, PlanSubmitted)
         ]
-        return bool(proposals) and self._reader.agrees(proposals[-1], text)
+        if not proposals:
+            return False
+        classifier = self._roles_of(item).resolve("classifier")
+        agreed = self._reader.agrees(proposals[-1], text, model=classifier)
+        # L9 evaluation capture: the reading, as an input/output pair.
+        log = self._store.transcript(item.id, "classifier")
+        log.append(
+            DelegationStarted(
+                brief=f"proposal:\n{proposals[-1]}\n\nreply:\n{text}",
+                role="classifier",
+                model=classifier,
+                caps={},
+            )
+        )
+        log.append(
+            DelegationCompleted(
+                handback="a go-ahead" if agreed else "not a go-ahead",
+                artifacts=[],
+                tokens=0,
+                duration_s=0.0,
+                exit_status="answered",
+            )
+        )
+        return agreed
 
     def _which(self, command: str) -> str:
         """L5: an unnamed command at top level is ambiguous: ask."""
@@ -1217,6 +1278,160 @@ class LoopAgent:
                 )
         self._last_project = here
 
+    # -- help (L9) -------------------------------------------------------------
+
+    def _roles_of(self, item: WorkItem) -> RoleMap:
+        """The roles pinned when the item opened, so a configuration change
+        never splits one item across models (L9)."""
+        default = (
+            self._helpers.roles
+            if self._helpers
+            else RoleMap(
+                executor="",
+                planner=self._roles.get("planner", ""),
+                helper="",
+                expert="",
+            )
+        )
+        pinned = item.data.get("roles") or self._roles
+        return RoleMap.from_pinned(dict(pinned), default)
+
+    def _delegate(
+        self, item_id: str, task: str, *, trigger: str = ""
+    ) -> tuple[str, bool]:
+        """A helper takes a sub-task on a branch off the milestone's: merged
+        back on a handback, discarded otherwise."""
+        item = self._store.work_item(item_id)
+        assert item is not None
+        loop = loop_of(item)
+        milestones = self._at(item).milestones
+        workspace = self._at(item).workspace
+        if self._helpers is None or milestones is None or workspace is None:
+            return "there's no helper here", False
+        if not loop.branch:
+            return "start a milestone first: a helper works on its branch", False
+        if not task:
+            return "say what the sub-task is", False
+        roles = self._roles_of(item)
+        number = loop.helps + 1
+        branch = f"{loop.branch}-help-{number}"
+        save(
+            self._store,
+            item.id,
+            item.position,
+            replace(loop, helps=number),
+            cause=f"helper {number} called in",
+        )
+        helper = roles.resolve("helper")
+        delegation = Delegation(
+            self._helpers.delegate_client(helper),
+            self._store,
+            workspace,
+            milestones.git,
+            model=helper,
+            caps=self._helpers.caps,
+            cancel=self._cancel,
+            reader=self._helpers.reader(roles),
+        )
+        git = milestones.git
+        before = git.head_sha()  # where the milestone is, uncommitted work aside
+        brief = (
+            f'{task}\n\nThe work item: issue #{loop.issue}, "{item.title}", '
+            f"milestone {loop.milestone}." + (f" {trigger}" if trigger else "")
+        )
+        result = delegation.run(
+            self._store.transcript(item.id, "helper"), brief=brief, branch=branch
+        )
+        try:
+            # The milestone takes the helper's work (or, on failure, only
+            # its own back) as uncommitted changes: harness branches never
+            # hold commits only this clone has (publishing makes them), so
+            # parking, a later milestone and a follow-up PR all carry it.
+            git.reset_to(loop.branch, branch if result.usable else loop.branch)
+            git._git("reset", "-q", "--soft", before)
+            if branch in git.local_branches():
+                git._git("branch", "-q", "-D", branch)
+        except GitError as exc:
+            self._fact(item, f"after helper {number}: {exc}")
+        for intent in self._store.pending_intents():
+            # Cut off rather than finished: discarded all the same, never
+            # resumed blindly (L8, L9).
+            if (
+                intent.kind == DELEGATION_INTENT
+                and intent.payload.get("branch") == branch
+            ):
+                self._store.abort_intent(intent.id, cause=f"delegation {result.status}")
+        if result.usable and result.handback is not None:
+            self._fact(item, f"helper {number} finished: {result.handback.render()}")
+            return f"the helper finished:\n{result.handback.render()}", True
+        self._fact(
+            item, f"helper {number} didn't finish ({result.status}): {result.note}"
+        )
+        said = (
+            f"the helper didn't finish ({result.status}: {result.note}); its work "
+            "was discarded and the milestone is as it was"
+        )
+        return said, False
+
+    def _ask_expert(self, item_id: str, question: str) -> tuple[str, bool]:
+        item = self._store.work_item(item_id)
+        assert item is not None
+        if self._helpers is None:
+            return "there's no expert here", False
+        if not question:
+            return "ask a question", False
+        desk = ExpertDesk(
+            self._helpers.client,
+            self._roles_of(item),
+            budget=self._helpers.expert_budget,
+        )
+        try:
+            answer = desk.ask(self._store.transcript(item.id, "expert"), question)
+        except BudgetExhausted as exc:
+            return str(exc), False
+        except LLMError as exc:
+            return f"the expert couldn't answer: {exc}", False
+        return answer.render(), True
+
+    def _help_when_stuck(self, item: WorkItem, loop: Loop, outcome: RunOutcome) -> bool:
+        """L9's observable triggers: the driver stuck (commands failing in a
+        row) or out of turns mid-milestone. A helper is called in, once per
+        milestone; True when it finished and the driver goes on."""
+        if (
+            self._helpers is None
+            or outcome.kind not in ("stuck", "capped")
+            or not loop.branch
+            or loop.helped_at == loop.milestone
+        ):
+            return False
+        save(
+            self._store,
+            item.id,
+            RUNNING,
+            replace(loop, helped_at=loop.milestone),
+            cause="helper called in",
+        )
+        self._transport.post(
+            f"I got stuck ({outcome.note[:200]}), so a helper is taking over the "
+            f"rest of milestone {loop.milestone}.",
+            thread=loop.thread,
+        )
+        self._running = True
+        try:
+            _, ok = self._delegate(
+                item.id,
+                f"Finish milestone {loop.milestone}, so its tests pass.",
+                trigger=f"The driver got stuck: {outcome.note}",
+            )
+        finally:
+            self._running = False
+            self._cancel.clear()
+        if not ok:
+            return False
+        item = self._store.work_item(item.id) or item
+        save(self._store, item.id, READY, loop_of(item), cause="a helper finished")
+        return True
+
     def _design_problem(self, item_id: str) -> str | None:
         """D1, D2: a `design` item agrees its design before any code. Until
         a PR of it has merged, a milestone holds documents only."""
@@ -1340,16 +1555,23 @@ class LoopAgent:
                     FINISH_MILESTONE,
                     CLOSE_ITEM,
                     *([WAIT_FOR_REVIEW] if loop.pr else []),
+                    *([DELEGATE, ASK_EXPERT] if self._helpers else []),
                     *EXECUTOR_TOOLS,
                 ]
             ),
         ]
+        roles = self._roles_of(item)
         workspace = project.workspace
         system = SYSTEM.format(
             repo=project.repo,
             issue=loop.issue,
             title=item.title,
             labels="".join(f"{LABELS[x]}\n\n" for x in loop.labels if x in LABELS),
+            helpers=(
+                f"{self._helpers.briefing(roles)}\n\n"
+                if self._helpers and loop.started
+                else ""
+            ),
             workspace=workspace.describe() if workspace else "",
         )
         self._active_thread = loop.thread
@@ -1360,11 +1582,20 @@ class LoopAgent:
                 system=system,
                 tools=tools,
                 workspace=workspace,
-                handlers={"start_milestone": lambda args: self._start(item.id, args)},
+                handlers={
+                    "start_milestone": lambda args: self._start(item.id, args),
+                    "delegate": lambda args: self._delegate(
+                        item.id, str(args.get("task", "")).strip()
+                    ),
+                    "ask_expert": lambda args: self._ask_expert(
+                        item.id, str(args.get("question", "")).strip()
+                    ),
+                },
                 checks={
                     "restate": _restatement_problem,
                     "finish_milestone": lambda args: self._design_problem(item.id),
                 },
+                model=roles.resolve("driver"),
             )
         finally:
             self._running = False
@@ -1505,6 +1736,8 @@ class LoopAgent:
         elif outcome.kind == "stopped":
             note = f"stopped ({outcome.note}). Why did you stop it? Reply to continue."
             self._wait(item, replace(loop, wait=ANSWER, note=note), note)
+        elif self._help_when_stuck(item, loop, outcome):
+            pass  # ready again: the driver goes on from the helper's handback
         else:
             note = f"I stopped short: {outcome.note}. Reply to continue."
             self._wait(item, replace(loop, wait=ANSWER, note=note), note)

nisse-dev/src/nisse_dev/loop/build.py

@@ -15,11 +15,14 @@ from nisse_dev.agent import (
     remote_url,
 )
 from nisse_dev.cancel import Cancel
+from nisse_dev.catalog import select_client
 from nisse_dev.config import AgentConfig, ConfigError
 from nisse_dev.config import Project as ProjectEntry
+from nisse_dev.delegation import Caps
 from nisse_dev.gitea import GiteaClient
 from nisse_dev.loop.agent import LoopAgent, Project
 from nisse_dev.loop.driver import Driver
+from nisse_dev.loop.help import Helpers
 from nisse_dev.loop.milestones import Milestones
 from nisse_dev.loop.reader import GoAheadReader
 from nisse_dev.matrix import MatrixTransport
@@ -124,14 +127,28 @@ def build_loop_agent(
         project_runtime(config, entry, store, gitea, account)
         for entry in config.projects
     ]
-    normalized, _passthrough = make_llm_clients(config, catalog)
+    normalized, passthrough = make_llm_clients(config, catalog)
     cancel = Cancel()
-    # The driver is the planner's model for now (M2.6 brings the driver role).
+    # The configured driver; each work item runs on the one it pinned (L9).
     driver = Driver(
         normalized,
-        config.roles.resolve("planner"),
+        config.roles.resolve("driver"),
         max_turns=config.tuning.executor.max_turns,
         cancel=cancel,
+        failure_limit=config.tuning.executor.failure_limit,
+    )
+    tuning = config.tuning
+    helpers = Helpers(
+        client=normalized,
+        roles=config.roles,
+        client_for=lambda model: select_client(model, normalized, passthrough, catalog),
+        caps=Caps(
+            turns=tuning.delegation.max_turns,
+            tokens=tuning.delegation.max_tokens,
+            wall_clock_s=tuning.delegation.wall_clock_s,
+        ),
+        expert_budget=tuning.expert.consult_budget,
+        profiles=config.profiles,
     )
     transport = MatrixTransport(config.matrix)
     loop = LoopAgent(
@@ -147,9 +164,9 @@ def build_loop_agent(
         cancel=cancel,
         oplog=oplog,
         roles=config.roles.pinned(),
-        # The driver's model reads plain-language go-aheads until the
+        # The classifier role reads plain-language go-aheads (L5, L9).
-        # classifier role arrives (M2.6).
+        reader=GoAheadReader(normalized, config.roles.resolve("classifier")),
-        reader=GoAheadReader(normalized, config.roles.resolve("planner")),
+        helpers=helpers,
         reload=lambda known: [
             project_runtime(config, entry, store, gitea, account)
             for entry in (load or _from_file(config))().projects

nisse-dev/src/nisse_dev/loop/driver.py

@@ -29,7 +29,8 @@ _PROSE_NUDGE = (
 
 @dataclass(frozen=True)
 class RunOutcome:
-    kind: Literal["wait", "capped", "stopped", "failed"]
+    # stuck: failed commands in a row hit the failure limit (L9's trigger)
+    kind: Literal["wait", "capped", "stuck", "stopped", "failed"]
     tool: str = ""  # the waiting tool called
     arguments: dict[str, Any] | None = None
     note: str = ""
@@ -43,10 +44,12 @@ class Driver:
         *,
         max_turns: int = DEFAULT_MAX_TURNS,
         cancel: Cancel | None = None,
+        failure_limit: int = 0,  # 0: no limit
     ) -> None:
         self._client = client
         self._model = model
         self._max_turns = max_turns
+        self._failure_limit = failure_limit
         self._cancel = cancel or Cancel()
 
     def run(
@@ -59,6 +62,7 @@ class Driver:
         handlers: Mapping[str, Callable[[dict[str, Any]], tuple[str, bool]]]
         | None = None,
         checks: Mapping[str, Callable[[dict[str, Any]], str | None]] | None = None,
+        model: str | None = None,
     ) -> RunOutcome:
         """``handlers`` answer control tools that don't end the run (e.g.
         start_milestone): each returns a result and whether it succeeded.
@@ -66,6 +70,8 @@ class Driver:
         failed result and the run goes on, so the model puts it right."""
         handlers = handlers or {}
         checks = checks or {}
+        model = model or self._model  # the work item's pinned driver (L9)
+        failures: list[str] = []  # the failed commands in a row, latest last
         offered = names(tools)
         nudged = False
         for turn in range(self._max_turns):
@@ -74,7 +80,7 @@ class Driver:
                 return RunOutcome("stopped", note=f"before turn {turn + 1}")
             try:
                 produced = self._client.complete(
-                    transcript.view(), tools=tools, model=self._model, system=system
+                    transcript.view(), tools=tools, model=model, system=system
                 )
             except LLMError as exc:
                 return RunOutcome("failed", note=f"the driver's call failed: {exc}")
@@ -132,4 +138,21 @@ class Driver:
                 transcript.append(
                     ToolResult(call_id=call.call_id, result=result, success=ok)
                 )
+                if call.tool == "run_command":
+                    failures = [] if ok else [*failures, _brief(call, result)]
+                    if self._failure_limit and len(failures) >= self._failure_limit:
+                        skip_calls(
+                            transcript, calls[index + 1 :], "not run: you're stuck"
+                        )
+                        return RunOutcome(
+                            "stuck",
+                            note=f"{len(failures)} commands failed in a row: "
+                            + "; ".join(failures),
+                        )
         return RunOutcome("capped", note=f"hit the {self._max_turns}-turn cap")
+
+
+def _brief(call: ToolCall, result: str) -> str:
+    """A failed command and the end of what it said, for a helper's brief."""
+    command = str(call.arguments.get("command", ""))[:120]
+    return f"`{command}`: {result.strip()[-300:]}"

nisse-dev/src/nisse_dev/loop/help.py

@@ -0,0 +1,73 @@
+"""Help for the driver (L9): a helper for a sub-task (delegation), an expert
+for one question, both by role, briefed from capability profiles.
+
+The driver asks when it chooses to (its instructions treat asking as
+normal), and the harness asks for it on observable triggers: commands failing
+in a row, or the turn cap. Delegation runs on a branch off the milestone's,
+merged back on a handback and discarded otherwise (L9: never build on a
+half-edited tree).
+"""
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+
+from nisse_dev.config import Profile
+from nisse_dev.delegation import Caps
+from nisse_dev.expert import DEFAULT_BUDGET
+from nisse_dev.llm import LLMClient, RoleMap
+from nisse_dev.loop.tools import _tool
+
+DELEGATE = _tool(
+    "delegate",
+    "Hand a sub-task of the current milestone to a helper, who works on the "
+    "milestone's branch with the same tools and reports back what changed, "
+    "what is now true and what next. Use it when the sub-task is beyond what "
+    "you can do reliably, or you are going round in circles. Needs a "
+    "milestone started.",
+    task="The sub-task, self-contained: what to do, where, and how to know it's done.",
+)
+ASK_EXPERT = _tool(
+    "ask_expert",
+    "Ask an expert one self-contained question: a design trade-off, how a "
+    "library behaves. It can't see the repository. The budget per item is "
+    "small.",
+    question="The question, with whatever context it needs.",
+)
+
+
+@dataclass(frozen=True)
+class Helpers:
+    client: LLMClient  # for the expert
+    roles: RoleMap  # the configured mapping; each item pins its own (L9)
+    # The helper's client by model (vendor passthrough where it fits, L9);
+    # unset, ``client``.
+    client_for: Callable[[str], LLMClient] | None = None
+    caps: Caps = field(default_factory=Caps)
+    expert_budget: int = DEFAULT_BUDGET
+    profiles: Mapping[str, Profile] = field(default_factory=dict)
+
+    def delegate_client(self, model: str) -> LLMClient:
+        return self.client_for(model) if self.client_for else self.client
+
+    def profile(self, model: str) -> str:
+        """What one role is told about another: its capability profile,
+        never its name (L9)."""
+        found = self.profiles.get(model)
+        return found.describe() if found else "no profile recorded"
+
+    def briefing(self, roles: RoleMap) -> str:
+        """The driver's instructions on asking for help."""
+        return (
+            "Asking for help is normal, not a failure. delegate hands a "
+            "sub-task to a helper ("
+            + self.profile(roles.resolve("helper"))
+            + "); ask_expert asks one question of an expert ("
+            + self.profile(roles.resolve("expert"))
+            + "). If commands keep failing, a helper is called in for you."
+        )
+
+    def reader(self, roles: RoleMap) -> str:
+        """Who acts on a helper's handback: the driver, by its profile."""
+        return "the driver, a model with this profile: " + self.profile(
+            roles.resolve("driver")
+        )

nisse-dev/src/nisse_dev/loop/items.py

@@ -56,6 +56,8 @@ class Loop:
     proposal_kind: str = ""  # "work" on a project, or a new "project"
     new_project: str = ""  # a new project being provisioned (D1)
     design_merged: bool = False  # a `design` item's design PR has merged (D1, D2)
+    helps: int = 0  # delegations so far, for their branch names (L9)
+    helped_at: int = 0  # the milestone a helper was last called in for, unasked
     # Where the operator last spoke (D5: answer where asked): "" for the
     # Matrix thread, or the PR it came from, with the inline location and a
     # little of what they said, for quoting.

nisse-dev/src/nisse_dev/loop/milestones.py

@@ -161,6 +161,11 @@ class Milestones:
         if git.remote_sha(self._remote, parked, credential=self._credential):
             git.delete_remote_branch(self._remote, parked, credential=self._credential)
 
+    @property
+    def git(self) -> GitRepo:
+        """The clone, for a delegation's branch (L9)."""
+        return self._git
+
     def changed(self) -> list[str]:
         """What the current milestone changes, from the default branch."""
         return self._git.changed_paths(f"origin/{self._base}")

nisse-dev/src/nisse_dev/loop/reader.py

@@ -50,7 +50,8 @@ class GoAheadReader:
         self._client = client
         self._model = model
 
-    def agrees(self, proposal: str, reply: str) -> bool:
+    def agrees(self, proposal: str, reply: str, model: str = "") -> bool:
+        """``model``: the work item's pinned classifier (L9)."""
         if is_question(reply):
             return False
         prompt = f"The proposal:\n\n{proposal}\n\nThe operator's reply:\n\n{reply}"
@@ -65,7 +66,7 @@ class GoAheadReader:
                     )
                 ],
                 tools=[VERDICT],
-                model=self._model,
+                model=model or self._model,
                 system=SYSTEM,
             )
         except LLMError:

nisse-dev/tests/test_config.py

@@ -478,3 +478,29 @@ def test_a_workspace_of_any_project_inside_is_refused(tmp_path: Path) -> None:
             state_db=str(work / "state.db"),
             projects__other={**SECOND, "workspace": str(work)},
         )
+
+
+# --- roles and profiles (L9) -------------------------------------------------
+
+
+def test_the_new_loops_roles_fall_back() -> None:
+    config = build()
+    assert config.roles.resolve("driver") == "claude-opus-5"  # the planner's
+    assert config.roles.resolve("classifier") == "claude-haiku-4-5"  # the expert's
+    named = build(roles__driver="qwen3.8-27b", roles__classifier="gemma-4-12b")
+    assert named.roles.resolve("driver") == "qwen3.8-27b"
+    assert named.roles.pinned()["classifier"] == "gemma-4-12b"
+
+
+def test_capability_profiles_are_read() -> None:
+    config = build(
+        profiles={"qwen3.8-27b": {"tools": "reliable", "weaknesses": "speed"}}
+    )
+    profile = config.profiles["qwen3.8-27b"]
+    assert profile.describe() == "tool calls: reliable; weak at speed"
+    assert build().profiles == {}
+
+
+def test_a_bad_profile_is_reported() -> None:
+    with pytest.raises(ConfigError, match="profiles.m.colour: unknown setting"):
+        build(profiles={"m": {"colour": "red"}})

nisse-dev/tests/test_llm.py

@@ -197,6 +197,10 @@ def test_role_map_resolves_and_pins() -> None:
         "planner": "anthropic/claude-opus-5",
         "helper": "anthropic/claude-opus-5",
         "expert": "anthropic/claude-opus-5",
+        # Unset, the new loop's roles fall back: driver to the planner's
+        # model, classifier to the expert's (L9).
+        "driver": "anthropic/claude-opus-5",
+        "classifier": "anthropic/claude-opus-5",
     }
 
 

nisse-dev/tests/test_loop_conversations.py

@@ -170,7 +170,7 @@ def test_a_plain_yes_files_the_work(world: dict[str, Any]) -> None:
     quiet(world)
 
     class Reader:
-        def agrees(self, proposal: str, reply: str) -> bool:
+        def agrees(self, proposal: str, reply: str, model: str = "") -> bool:
             return reply == "sounds right"
 
     model = ScriptedModel([PROPOSAL], [call("ask_human", question="anything else?")])

nisse-dev/tests/test_loop_help.py

@@ -0,0 +1,303 @@
+"""Help for the driver on the new loop (M2.6, L9): roles pinned per item,
+delegation on a branch off the milestone's (merged back on a handback,
+discarded otherwise), the stuck trigger, ask_expert and its budget, and
+briefings from capability profiles. Real git on the throwaway build host."""
+
+import json
+from typing import Any
+
+from test_loop_milestones import (
+    RESTATE,
+    Forge,
+    ScriptedModel,
+    call,
+    git,
+    item,
+    loop_agent,
+    say,
+)
+from test_loop_milestones import world as world  # noqa: PLC0414 -- the fixture
+
+from nisse_dev.canonical import AssistantText, Fact, ToolCall, ToolResult
+from nisse_dev.config import Profile
+from nisse_dev.delegation import DELEGATION_INTENT, Caps
+from nisse_dev.llm import RoleMap
+from nisse_dev.loop.help import Helpers
+from nisse_dev.loop.items import ANSWER, MERGE, loop_of
+
+ROLES = RoleMap(executor="", planner="drv", helper="hlp", expert="exp", driver="drv")
+
+
+class Spy:
+    """A scripted model that records what each call was told."""
+
+    def __init__(self, *turns: list[Any]) -> None:
+        self.turns = list(turns)
+        self.calls: list[tuple[str, str]] = []  # (model, system)
+
+    def complete(self, events, tools, model, system=None) -> list[Any]:
+        self.calls.append((model, system or ""))
+        assert self.turns, f"{model} was called more often than scripted"
+        return self.turns.pop(0)
+
+
+def hand_back(succeeded: bool = True) -> ToolCall:
+    return ToolCall(
+        call_id=f"hb-{succeeded}",
+        tool="hand_back",
+        arguments={
+            "changed": "wrote helped.py",
+            "now_true": "the helper's part is done",
+            "next": "finish the milestone",
+            "succeeded": succeeded,
+        },
+    )
+
+
+def helpers(helper: Spy, expert: Spy | None = None, **kw: Any) -> Helpers:
+    return Helpers(
+        client=expert or Spy(),  # type: ignore[arg-type]
+        roles=ROLES,
+        client_for=lambda model: helper,  # type: ignore[arg-type,return-value]
+        caps=Caps(turns=5),
+        profiles={
+            "hlp": Profile(tools="reliable", strengths="tricky tests"),
+            "drv": Profile(tools="fair", weaknesses="long refactors"),
+        },
+        **kw,
+    )
+
+
+def driver_results(world: dict[str, Any], tool: str) -> list[str]:
+    events = world["store"].transcript(item(world).id, "driver").read()
+    ids = {
+        e.event.call_id
+        for e in events
+        if isinstance(e.event, ToolCall) and e.event.tool == tool
+    }
+    return [
+        e.event.result
+        for e in events
+        if isinstance(e.event, ToolResult) and e.event.call_id in ids
+    ]
+
+
+def local_branches(world: dict[str, Any]) -> list[str]:
+    return world["repo"].local_branches()
+
+
+def test_a_helpers_work_lands_on_the_milestone(world: dict[str, Any]) -> None:
+    forge: Forge = world["forge"]
+    helper = Spy(
+        [call("write_file", path="helped.py", content="H = 1\n")],
+        [hand_back()],
+    )
+    model = ScriptedModel(
+        RESTATE,
+        [
+            call("start_milestone", name="words"),
+            call("delegate", task="write helped.py with H = 1"),
+            call("finish_milestone", summary="with help", final="yes"),
+        ],
+    )
+    agent, transport = loop_agent(world, model, helpers=helpers(helper))
+    agent.intake()
+    say(transport, world, "go")
+
+    [result] = driver_results(world, "delegate")
+    assert result.startswith("the helper finished:") and "helped.py" in result
+    assert helper.calls[0][0] == "hlp"  # the helper role's model
+    # Briefed about the driver by its profile, never by name.
+    assert "tool calls: fair; weak at long refactors" in helper.calls[0][1]
+    [pr] = forge.prs
+    assert forge.files(pr) == ["helped.py"]
+    assert not [b for b in local_branches(world) if "-help-" in b]
+    assert loop_of(item(world)).wait == MERGE
+
+
+def test_a_failed_helpers_work_is_discarded(world: dict[str, Any]) -> None:
+    helper = Spy(
+        [call("write_file", path="half.py", content="half\n")],
+        [hand_back(succeeded=False)],
+    )
+    model = ScriptedModel(
+        RESTATE,
+        [
+            call("start_milestone", name="words"),
+            call("write_file", path="mine.py", content="M = 1\n"),
+            call("delegate", task="write half.py"),
+            call("ask_human", question="the helper couldn't; what now?"),
+        ],
+    )
+    agent, transport = loop_agent(world, model, helpers=helpers(helper))
+    agent.intake()
+    say(transport, world, "go")
+
+    [result] = driver_results(world, "delegate")
+    assert "didn't finish" in result and "discarded" in result
+    root = world["repo"].root
+    assert not (root / "half.py").exists()
+    assert (root / "mine.py").exists()  # the driver's own work stays
+    assert world["repo"].current_branch() == loop_of(item(world)).branch
+    assert not [b for b in local_branches(world) if "-help-" in b]
+    assert not [
+        i for i in world["store"].pending_intents() if i.kind == DELEGATION_INTENT
+    ]
+
+
+def test_delegating_needs_a_milestone(world: dict[str, Any]) -> None:
+    model = ScriptedModel(
+        RESTATE,
+        [
+            call("delegate", task="anything"),
+            call("ask_human", question="ok?"),
+        ],
+    )
+    agent, transport = loop_agent(world, model, helpers=helpers(Spy()))
+    agent.intake()
+    say(transport, world, "go")
+    assert driver_results(world, "delegate") == [
+        "start a milestone first: a helper works on its branch"
+    ]
+
+
+def test_a_stuck_driver_gets_a_helper_once_per_milestone(
+    world: dict[str, Any],
+) -> None:
+    forge: Forge = world["forge"]
+    helper = Spy(
+        [call("write_file", path="fixed.py", content="F = 1\n")],
+        [hand_back()],
+    )
+    failing = [
+        call("run_command", command="false"),
+        call("run_command", command="false"),
+    ]
+    model = ScriptedModel(
+        RESTATE,
+        [call("start_milestone", name="words"), *failing],
+        [call("finish_milestone", summary="fixed with help", final="yes")],
+    )
+    agent, transport = loop_agent(world, model, helpers=helpers(helper))
+    agent._driver._failure_limit = 2
+    agent.intake()
+    say(transport, world, "go")
+
+    assert any("so a helper is taking over" in m.text for m in transport.outbound)
+    assert "The driver got stuck" in str(helper.calls)  # in the brief
+    [pr] = forge.prs
+    assert forge.files(pr) == ["fixed.py"]
+    assert loop_of(item(world)).helped_at == 1
+
+
+def test_stuck_again_in_the_same_milestone_asks_the_operator(
+    world: dict[str, Any],
+) -> None:
+    helper = Spy(
+        [call("write_file", path="tried.py", content="T = 1\n")],
+        [hand_back()],
+    )
+    failing = [
+        call("run_command", command="false"),
+        call("run_command", command="false"),
+    ]
+    model = ScriptedModel(
+        RESTATE,
+        [call("start_milestone", name="words"), *failing],
+        failing,  # after the helper: stuck again
+    )
+    agent, transport = loop_agent(world, model, helpers=helpers(helper))
+    agent._driver._failure_limit = 2
+    agent.intake()
+    say(transport, world, "go")
+    loop = loop_of(item(world))
+    assert loop.wait == ANSWER and "commands failed in a row" in loop.note
+    assert len(helper.calls) == 2  # one delegation: its two turns
+
+
+def test_ask_expert_answers_within_its_budget(world: dict[str, Any]) -> None:
+    answer = {"answer": "use a dict", "confidence": "high", "caveats": ""}
+    expert = Spy(
+        [AssistantText(text=json.dumps(answer), model="exp")],
+        [AssistantText(text=json.dumps(answer), model="exp")],
+    )
+    model = ScriptedModel(
+        RESTATE,
+        [
+            call("start_milestone", name="words"),
+            call("ask_expert", question="dict or list?"),
+            call("ask_expert", question="and again?"),
+            call("ask_expert", question="once more?"),
+            call("ask_human", question="ok?"),
+        ],
+    )
+    agent, transport = loop_agent(
+        world, model, helpers=helpers(Spy(), expert, expert_budget=2)
+    )
+    agent.intake()
+    say(transport, world, "go")
+    first, second, third = driver_results(world, "ask_expert")
+    assert first.startswith("use a dict") and second.startswith("use a dict")
+    assert "budget" in third
+    assert expert.calls[0][0] == "exp"
+
+
+def test_roles_are_pinned_when_the_item_opens(world: dict[str, Any]) -> None:
+    spy = Spy(RESTATE)
+    agent, _ = loop_agent(world, spy, helpers=helpers(Spy()))  # type: ignore[arg-type]
+    agent._roles = {**ROLES.pinned(), "driver": "pinned-driver"}
+    agent.intake()
+    assert spy.calls[0][0] == "pinned-driver"
+    pinned = item(world).data["roles"]
+    assert pinned["driver"] == "pinned-driver" and pinned["classifier"] == "exp"
+
+
+def test_the_driver_is_told_about_its_helpers_by_profile(
+    world: dict[str, Any],
+) -> None:
+    spy = Spy(RESTATE, [call("ask_human", question="ok?")])
+    agent, transport = loop_agent(world, spy, helpers=helpers(Spy()))  # type: ignore[arg-type]
+    agent.intake()
+    say(transport, world, "go")
+    after_go = spy.calls[1][1]
+    assert "Asking for help is normal" in after_go
+    assert "tool calls: reliable; strong at tricky tests" in after_go
+    assert "hlp" not in after_go  # never a model's name
+
+
+def test_a_delegation_cut_off_by_a_crash_is_discarded_at_boot(
+    world: dict[str, Any],
+) -> None:
+    model = ScriptedModel(
+        RESTATE,
+        [call("start_milestone", name="words"), call("ask_human", question="?")],
+    )
+    agent, transport = loop_agent(world, model, helpers=helpers(Spy()))
+    agent.intake()
+    say(transport, world, "go")
+    milestone = loop_of(item(world)).branch
+    repo = world["repo"]
+    repo.checkout_new_branch(f"{milestone}-help-1")
+    (repo.root / "partial.py").write_text("p\n")
+    repo.commit_all("partial")
+    world["store"].begin_intent(
+        DELEGATION_INTENT,
+        {"branch": f"{milestone}-help-1", "base": milestone},
+        cause="delegating",
+        work_item=item(world).id,
+    )
+    agent2, _ = loop_agent(world, ScriptedModel(), helpers=helpers(Spy()))
+    agent2.boot()
+    assert repo.current_branch() == milestone
+    assert not (repo.root / "partial.py").exists()
+    assert f"{milestone}-help-1" not in repo.local_branches()
+    assert not [
+        i for i in world["store"].pending_intents() if i.kind == DELEGATION_INTENT
+    ]
+    facts = [
+        e.event.text
+        for e in world["store"].transcript(item(world).id, "orchestrator").read()
+        if isinstance(e.event, Fact)
+    ]
+    assert any("cut off by a restart and discarded" in f for f in facts)
+    assert git

nisse-dev/tests/test_loop_labels.py

@@ -32,7 +32,7 @@ class Reader:
         self.yes = set(yes)
         self.asked: list[tuple[str, str]] = []
 
-    def agrees(self, proposal: str, reply: str) -> bool:
+    def agrees(self, proposal: str, reply: str, model: str = "") -> bool:
         self.asked.append((proposal, reply))
         return reply in self.yes
 

nisse-dev/tests/test_loop_milestones.py

@@ -252,11 +252,11 @@ def make_world(
 
 
 def loop_agent(
-    world: dict[str, Any], model: ScriptedModel
+    world: dict[str, Any], model: ScriptedModel, **agent: Any
 ) -> tuple[LoopAgent, InMemoryTransport]:
     transport = InMemoryTransport()
     cancel = Cancel()
-    agent = LoopAgent(
+    loop = LoopAgent(
         world["store"],
         transport,
         world["forge"],
@@ -271,9 +271,10 @@ def loop_agent(
         cancel=cancel,
         oplog=OpLog("nisse-dev", io.StringIO()),
         milestones=world["milestones"],
+        **agent,
     )
-    transport.on_message(agent.on_message)
+    transport.on_message(loop.on_message)
-    return agent, transport
+    return loop, transport
 
 
 def item(world: dict[str, Any]) -> Any:

nisse-dev/tests/test_loop_properties.py

@@ -23,10 +23,13 @@ from test_loop_projects import Router
 
 from nisse_dev.cancel import Cancel
 from nisse_dev.canonical import Fact, PlanSubmitted, ToolCall, UserMessage
+from nisse_dev.delegation import DELEGATION_INTENT, Caps
 from nisse_dev.execution import SSHHost
 from nisse_dev.gitea import Comment, Issue
+from nisse_dev.llm import RoleMap
 from nisse_dev.loop.agent import LoopAgent, Project
 from nisse_dev.loop.driver import Driver
+from nisse_dev.loop.help import Helpers
 from nisse_dev.loop.items import (
     ANSWER,
     BLOCKED,
@@ -97,6 +100,8 @@ class PolicyModel:
             return [call("ask_human", question=f"talk first {self.n}?")]  # discuss
         if "finish_milestone" not in offered:
             return [call("restate", text=f"restatement {self.n}")]
+        if "delegate" in offered and self.rng.random() < 0.1:
+            return [call("delegate", task=f"help {self.n}")]
         if "wait_for_review" in offered and self.rng.random() < 0.2:
             return [call("wait_for_review", note=f"waiting {self.n}")]
         if self.rng.random() < 0.15:
@@ -120,6 +125,32 @@ class PolicyModel:
         ]
 
 
+class HelperModel:
+    """A scripted helper: does some work, then hands back, succeeding or
+    not as the seed says."""
+
+    def __init__(self, rng: random.Random) -> None:
+        self.rng = rng
+        self.n = 0
+
+    def complete(self, events, tools, model, system=None) -> list[Any]:
+        assert_renderable(events)
+        self.n += 1
+        return [
+            call("write_file", path=f"h{self.n}.py", content=f"h = {self.n}\n"),
+            ToolCall(
+                call_id=f"hb{self.n}",
+                tool="hand_back",
+                arguments={
+                    "changed": f"h{self.n}.py",
+                    "now_true": "helped",
+                    "next": "go on",
+                    "succeeded": self.rng.random() < 0.6,
+                },
+            ),
+        ]
+
+
 class Walk:
     """``worlds``: project name -> its world (forge, bare repo, clone,
     milestones), all over one store."""
@@ -174,13 +205,18 @@ class Walk:
             cancel=cancel,
             oplog=OpLog("nisse-dev", _Null()),
             reader=self,  # type: ignore[arg-type]
+            helpers=Helpers(
+                client=HelperModel(self.rng),  # type: ignore[arg-type]
+                roles=RoleMap(executor="", planner="m", helper="h", expert="x"),
+                caps=Caps(turns=4),
+            ),
             bug_grace_s=600,
             clock=lambda: self.now,
         )
         transport.on_message(agent.on_message)
         return agent, transport
 
-    def agrees(self, proposal: str, reply: str) -> bool:
+    def agrees(self, proposal: str, reply: str, model: str = "") -> bool:
         return self.reads_yes(proposal, reply)
 
     def items(self) -> list[Any]:
@@ -404,7 +440,22 @@ class Walk:
                 assert not (loop.prepared or loop.milestone or loop.pr), (
                     f"conversations_do_no_work: {item.id}: {where}"
                 )
+        # L9: a delegation never outlives its step: merged back or discarded.
+        assert not [
+            i for i in store.pending_intents() if i.kind == DELEGATION_INTENT
+        ], f"delegations_settle (intent): {where}"
         for name, w in self.worlds.items():
+            helping = [
+                b
+                for b in [
+                    *w["repo"].local_branches(),
+                    *_git(
+                        w["bare"], "for-each-ref", "--format=%(refname:short)"
+                    ).split(),
+                ]
+                if "-help-" in b
+            ]
+            assert not helping, f"delegations_settle (branch {helping}): {where}"
             self._records_have_one_item(name, w, where)
             self._own_repository_only(name, w, where)
             self._parked_work_safe(w, name, where)
@@ -519,7 +570,10 @@ class Walk:
             written = {
                 str(e.event.arguments.get("path"))
                 for e in self.world["store"].transcript(item.id, "driver").read()
-                if isinstance(e.event, ToolCall) and e.event.tool == "write_file"
+                # The driver's own: a failed helper's are discarded on purpose.
+                if isinstance(e.event, ToolCall)
+                and e.event.tool == "write_file"
+                and e.actor == "driver"
             }
             here = current.startswith(f"{item.id}-")
             for path in written: