Roles, capability profiles and help on the new loop (M2.6)

- roles.driver (unset: the planner's model) and roles.classifier (unset:
  the expert's); the mapping is pinned when an item opens, and the driver
  and the go-ahead reader run on the item's own (L9).
- profiles: in configuration (tools, strengths, weaknesses), used in
  briefings: the driver's instructions say what its helper and expert are
  good at, the helper's brief who reads its handback; never a model name.
- delegate(task) and ask_expert(question) for the driver. A helper works
  on a branch off the milestone's; on a handback its work comes back as
  uncommitted changes, otherwise it is discarded, so harness branches
  never hold commits only the clone has. ask_expert keeps its budget.
- Triggers: failure_limit failed commands in a row (Driver's new "stuck"
  outcome) or the turn cap, mid-milestone, call a helper in once per
  milestone; if it fails too, the item waits on the operator. A
  delegation cut off by a crash is discarded at boot.
- Evaluation capture: delegations and ask_expert as before, and the
  reader's verdicts as classifier pairs.

Tests: help tests (merge-back, discard, needs a milestone, the stuck
trigger and its once-per-milestone limit, the expert and its budget,
pinning, briefings by profile, boot discarding a cut-off delegation),
config tests for roles and profiles, and the walk delegating with a
scripted helper that sometimes fails, with a delegations_settle property.
The walk found helper commits stranded on a merged PR's branch, fixed by
returning a helper's work uncommitted.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPQ14L3LikyefCMFBqEbKE

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)
-    return agent, transport
+    transport.on_message(loop.on_message)
+    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: