fix(run): enforce read-only seat capability (#383)

Author: solomonneas

docs/seat-catalog.md

@@ -45,6 +45,29 @@ role = "Open-weight implementation worker on the flat sub; overflow relief for t
 
 Known gap: Composer models return empty text in read-only plan mode (issue #206). Pin a non-composer model for read-only seats.
 
+Mark direct Cursor Composer and Grok seats as incapable so the chef cannot route
+`--read-only` work to them:
+
+```toml
+[agents.composer]
+cli = "cursor"
+model = "composer-2.5"
+read_only_capable = false
+role = "Fast implementation worker for writable runs."
+
+[agents.grok]
+cli = "cursor"
+model = "grok-4.5-xhigh"
+read_only_capable = false
+role = "Alternate implementation worker for writable runs."
+```
+
+Keep those direct seats set to `false`. To use either model for read-only findings,
+create a separate ACP seat and set `transport = "acpx"`,
+`transport_version = "0.12.0"`, and `read_only_capable = true` on that seat. ACP
+model IDs differ from direct Cursor aliases, so copy the reviewed model ID from
+the [technical guide](technical-guide.md#run-a-brigade).
+
 ### Claude subscription
 
 Two seats, two costs. The heavier model reviews. A lighter sibling takes routine passes at a fraction of the quota burn:
@@ -111,10 +134,14 @@ Several open-weight providers expose Anthropic-compatible endpoints, which means
 [agents.k3]
 cli = "claude"
 model = "kimi-k3"
+read_only_capable = false
 role = "Open-weight worker on a coding-plan quota; relief for orchestrator-tier work."
 env = { ANTHROPIC_BASE_URL = "https://api.moonshot.ai/anthropic", ANTHROPIC_AUTH_TOKEN_REF = "KIMI_API_KEY", CLAUDE_CONFIG_DIR = "/home/operator/.claude-lanes" }
 ```
 
+This proxy example opts out of read-only dispatch. Set the value to `true` only after
+the validation smoke below returns a usable worker result in read-only mode.
+
 Overrides apply to the spawned CLI process only, `run.json` records the override names and endpoint host (never values), and a missing referenced variable fails the worker before dispatch. If the CLI echoes any resolved override value, Brigade replaces the exact value with its target name in brackets before worker text, detail, stdout, or stderr can be stored. Direct CLI seats only: acpx and codex-cloud seats manage their own environment.
 
 The isolated `CLAUDE_CONFIG_DIR` is load-bearing: with the default config directory, the `claude` CLI prefers its subscription OAuth over env auth, the upstream returns 401, and the CLI retries silently, which presents as an indefinite hang.

docs/technical-guide.md

@@ -279,6 +279,11 @@ a large one). Pull the model yourself (`ollama pull llama3.2:3b`) or name one
 `ollama list` already shows; `brigade roster doctor` warns about missing ones.
 `limits.timeout_seconds` is the default per-agent timeout.
 `agents.<name>.timeout_seconds` overrides it for one agent.
+`agents.<name>.read_only_capable` is an optional boolean that defaults to `true`.
+Set it to `false` when a seat cannot return usable output under `--read-only`.
+The chef sees the value in its planning prompt, plan validation rejects an incapable
+assignment during read-only runs, and `--worker` rejects the seat before run artifacts
+are created. The field does not restrict writable runs.
 `limits.sandbox` is optional. When set to `read-only`, `workspace-write`, or
 `danger-full-access`, `brigade run` uses it as the native Codex sandbox mode
 unless the run also passes `--sandbox`.

src/brigade/aboyeur.py

@@ -37,7 +37,7 @@
     write_worker_logs as _write_worker_logs,
 )
 from .run_transport import Assignment, WorkerResult
-from .roster import Agent, Roster, is_cli_allowed, timeout_for, workers
+from .roster import Agent, Roster, is_cli_allowed, read_only_capability_error, timeout_for, workers
 from .route_catalog import RouteBrief, route_brief, uncovered_stages, unknown_covers
 
 CODE_GRAPH_HEADING = "## Code graph context (GraphTrail, read-only)"
@@ -368,12 +368,18 @@ def build_plan_prompt(
     evidence: EvidenceBrief | None = None,
     route: RouteBrief | None = None,
 ) -> str:
-    worker_lines = "\n".join(f"- {agent.name}: cli={agent.cli}; role={agent.role}" for agent in workers(roster))
+    worker_lines = "\n".join(
+        f"- {agent.name}: cli={agent.cli}; "
+        + (f"read_only_capable={str(agent.read_only_capable).lower()}; " if read_only else "")
+        + f"role={agent.role}"
+        for agent in workers(roster)
+    )
     if not worker_lines:
         worker_lines = "- no workers configured"
 
     note = f"\nCorrection needed: {corrective_note}\n" if corrective_note else ""
     policy = f"\n\n{_read_only_rules()}\n" if read_only else ""
+    capability_rule = "- Assign only workers with read_only_capable=true.\n" if read_only else ""
     route_section = ""
     route_rule = ""
     if route is not None and route.attached and route.text:
@@ -395,6 +401,7 @@ def build_plan_prompt(
         "- Assignments in the same stage run in parallel; later stages receive earlier-stage worker results.\n"
         "- Omit stage only for backwards-compatible stage 1 assignments.\n"
         "- Assign only listed workers.\n"
+        f"{capability_rule}"
         "- Use zero assignments only if no worker is useful."
         f"{route_rule}"
         f"{policy}"
@@ -571,7 +578,7 @@ def _read_only_rules() -> str:
     )
 
 
-def parse_plan(text: str, roster: Roster) -> list[Assignment]:
+def parse_plan(text: str, roster: Roster, *, read_only: bool = False) -> list[Assignment]:
     try:
         payload = _extract_json(text)
     except json.JSONDecodeError as exc:
@@ -601,6 +608,10 @@ def parse_plan(text: str, roster: Roster) -> list[Assignment]:
             raise ValueError(f"assignment references unknown worker: {worker!r}")
         if worker == roster.orchestrator:
             raise ValueError("assignment cannot target the orchestrator")
+        if read_only:
+            capability_error = read_only_capability_error(roster.agents[worker])
+            if capability_error is not None:
+                raise ValueError(capability_error)
         if not isinstance(subtask, str) or not subtask.strip():
             raise ValueError("assignment.task must be a non-empty string")
         raw_covers = item.get("covers", [])
@@ -824,7 +835,7 @@ def plan(
         _record_plan_attempt(attempts, stage="initial", result=first)
         raise RuntimeError(f"orchestrator failed during plan: {first.detail}")
     try:
-        assignments = parse_plan(first.text, roster)
+        assignments = parse_plan(first.text, roster, read_only=read_only)
         _record_plan_attempt(
             attempts,
             stage="initial",
@@ -859,7 +870,7 @@ def plan(
             _record_plan_attempt(attempts, stage="correction", result=second)
             raise RuntimeError(f"orchestrator failed during plan correction: {second.detail}") from exc
         try:
-            assignments = parse_plan(second.text, roster)
+            assignments = parse_plan(second.text, roster, read_only=read_only)
             _record_plan_attempt(
                 attempts,
                 stage="correction",
@@ -911,7 +922,7 @@ def plan(
         _record_plan_attempt(attempts, stage="coverage-correction", result=revised_result)
         return assignments
     try:
-        revised = parse_plan(revised_result.text, roster)
+        revised = parse_plan(revised_result.text, roster, read_only=read_only)
     except ValueError as exc:
         _record_plan_attempt(attempts, stage="coverage-correction", result=revised_result, parse_error=str(exc))
         return assignments
@@ -1838,6 +1849,7 @@ def _roster_payload(roster: Roster) -> dict[str, object]:
                 "role": agent.role,
                 "timeout_seconds": agent.timeout_seconds,
                 "invalid_final_fallback": agent.invalid_final_fallback,
+                "read_only_capable": agent.read_only_capable,
                 # env tables hold names and references only, never secret
                 # values (enforced at roster load), so persisting them for
                 # resume is safe.
@@ -1960,12 +1972,16 @@ def _run_payload(
     return payload
 
 
-def _direct_worker_error(worker: str, roster: Roster) -> str | None:
+def _direct_worker_error(worker: str, roster: Roster, *, read_only: bool = False) -> str | None:
     agent = roster.agents.get(worker)
     if agent is None:
         return f"unknown worker: {worker}"
     if worker == roster.orchestrator:
         return f"--worker cannot target orchestrator seat: {worker}"
+    if read_only:
+        capability_error = read_only_capability_error(agent)
+        if capability_error is not None:
+            return capability_error
     if agent.cli is None:
         return f"worker has no CLI adapter: {worker}"
     if not is_cli_allowed(agent.cli, roster):
@@ -2181,7 +2197,7 @@ def _payload(**kwargs: Any) -> dict[str, object]:
         return _run_payload(lock_workspace=lock_workspace, **kwargs)
 
     if worker is not None:
-        worker_error = _direct_worker_error(worker, roster)
+        worker_error = _direct_worker_error(worker, roster, read_only=read_only)
         if worker_error is not None:
             print(f"error: {worker_error}", file=sys.stderr)
             return 2

src/brigade/cli/run.py

@@ -306,7 +306,12 @@ def dispatch(args) -> int:
             file=sys.stderr,
         )
     if args.worker is not None:
-        worker_error = _direct_worker_error(args.worker, loaded_roster, roster_mod)
+        worker_error = _direct_worker_error(
+            args.worker,
+            loaded_roster,
+            roster_mod,
+            read_only=args.read_only,
+        )
         if worker_error is not None:
             print(f"error: {worker_error}", file=sys.stderr)
             return 2
@@ -766,12 +771,16 @@ def _detached_child_argv(args, *, run_cwd: Path, roster_resolution, output_dir:
     return argv
 
 
-def _direct_worker_error(worker: str, loaded_roster, roster_mod) -> str | None:
+def _direct_worker_error(worker: str, loaded_roster, roster_mod, *, read_only: bool = False) -> str | None:
     agent = loaded_roster.agents.get(worker)
     if agent is None:
         return f"unknown worker: {worker}"
     if worker == loaded_roster.orchestrator:
         return f"--worker cannot target orchestrator seat: {worker}"
+    if read_only:
+        capability_error = roster_mod.read_only_capability_error(agent)
+        if capability_error is not None:
+            return capability_error
     if agent.cli is None:
         return f"worker has no CLI adapter: {worker}"
     if not roster_mod.is_cli_allowed(agent.cli, loaded_roster):

src/brigade/roster.py

@@ -39,6 +39,7 @@ class Agent:
     transport_version: str | None = None
     env: dict[str, str] | None = None
     invalid_final_fallback: str | None = None
+    read_only_capable: bool = True
 
 
 @dataclass(frozen=True)
@@ -68,6 +69,12 @@ def _as_positive_number(value: object, field: str) -> float:
     return float(value)
 
 
+def _as_bool(value: object, field: str) -> bool:
+    if not isinstance(value, bool):
+        raise ValueError(f"{field} must be a boolean")
+    return value
+
+
 def _as_sandbox(value: object) -> str | None:
     if value is None:
         return None
@@ -246,6 +253,10 @@ def load_roster(path: Path, *, resolution: RosterResolution | None = None) -> Ro
         invalid_final_fallback = (
             _as_str(fallback_raw, f"agents.{agent_name}.invalid_final_fallback") if fallback_raw is not None else None
         )
+        read_only_capable = _as_bool(
+            raw_agent.get("read_only_capable", True),
+            f"agents.{agent_name}.read_only_capable",
+        )
 
         cli_raw = raw_agent.get("cli")
         has_endpoint = endpoint is not None and model is not None
@@ -301,6 +312,7 @@ def load_roster(path: Path, *, resolution: RosterResolution | None = None) -> Ro
             transport_version=transport_version,
             env=env,
             invalid_final_fallback=invalid_final_fallback,
+            read_only_capable=read_only_capable,
         )
 
     if orchestrator not in parsed_agents:
@@ -341,3 +353,9 @@ def load_roster(path: Path, *, resolution: RosterResolution | None = None) -> Ro
 
 def workers(roster: Roster) -> list[Agent]:
     return [agent for name, agent in roster.agents.items() if name != roster.orchestrator]
+
+
+def read_only_capability_error(agent: Agent) -> str | None:
+    if agent.read_only_capable:
+        return None
+    return f"worker {agent.name!r} cannot run in read-only mode: agents.{agent.name}.read_only_capable is false"

src/brigade/run_resume.py

@@ -13,7 +13,7 @@
 from pathlib import Path
 
 from . import aboyeur, agents, codex_appserver, runguard
-from .roster import Agent, Roster, _as_env
+from .roster import Agent, Roster, _as_bool, _as_env
 
 _RESUMABLE_STATUSES = ("interrupted", "failed")
 _NONTERMINAL_RUN_STATUSES = frozenset(
@@ -55,6 +55,10 @@ def _roster_from_snapshot(snapshot: dict) -> Roster:
             transport_version=raw.get("transport_version"),
             env=_as_env(raw.get("env"), name),
             invalid_final_fallback=raw.get("invalid_final_fallback"),
+            read_only_capable=_as_bool(
+                raw.get("read_only_capable", True),
+                f"agents.{name}.read_only_capable",
+            ),
         )
     return Roster(
         orchestrator=snapshot["orchestrator"],

tests/test_aboyeur.py

@@ -31,6 +31,23 @@ def _roster():
     )
 
 
+def _roster_with_incapable_worker():
+    return Roster(
+        orchestrator="chef",
+        agents={
+            "chef": Agent("chef", "codex", "plan and synthesize"),
+            "coder": Agent(
+                "coder",
+                "cursor",
+                "write code",
+                model="composer-2.5",
+                read_only_capable=False,
+            ),
+        },
+        max_workers=1,
+    )
+
+
 def _timeout_roster():
     return Roster(
         orchestrator="chef",
@@ -151,6 +168,20 @@ def test_parse_plan_accepts_plain_json():
     assert plan == [aboyeur.Assignment(worker="coder", task="implement it")]
 
 
+def test_parse_plan_rejects_incapable_worker_only_in_read_only_mode():
+    text = '{"assignments":[{"worker":"coder","task":"inspect it"}]}'
+
+    with pytest.raises(
+        ValueError,
+        match=r"coder.*read-only mode.*agents\.coder\.read_only_capable is false",
+    ):
+        aboyeur.parse_plan(text, _roster_with_incapable_worker(), read_only=True)
+
+    assert aboyeur.parse_plan(text, _roster_with_incapable_worker()) == [
+        aboyeur.Assignment(worker="coder", task="inspect it")
+    ]
+
+
 def test_parse_plan_accepts_staged_json_and_defaults_missing_stage():
     plan = aboyeur.parse_plan(
         json.dumps(
@@ -255,6 +286,19 @@ def test_build_plan_prompt_describes_stage_contract():
     assert "later stages receive earlier-stage worker results" in prompt
 
 
+def test_build_plan_prompt_exposes_read_only_capability():
+    prompt = aboyeur.build_plan_prompt("inspect feature", _roster_with_incapable_worker(), read_only=True)
+
+    assert "coder: cli=cursor; read_only_capable=false; role=write code" in prompt
+    assert "Assign only workers with read_only_capable=true" in prompt
+
+
+def test_build_plan_prompt_hides_read_only_capability_for_writable_runs():
+    prompt = aboyeur.build_plan_prompt("build feature", _roster_with_incapable_worker())
+
+    assert "read_only_capable" not in prompt
+
+
 def test_worker_prompt_without_prior_context_keeps_original_contract():
     assignment = aboyeur.Assignment(worker="coder", task="implement it")
     prompt = aboyeur._worker_prompt(_roster().agents["coder"], assignment)
@@ -2290,6 +2334,37 @@ def test_roster_payload_includes_invalid_final_fallback():
     assert payload["agents"]["grok_cli"]["invalid_final_fallback"] == "cursor_grok"
 
 
+def test_roster_payload_includes_read_only_capability():
+    payload = aboyeur._roster_payload(_roster_with_incapable_worker())
+
+    assert payload["agents"]["chef"]["read_only_capable"] is True
+    assert payload["agents"]["coder"]["read_only_capable"] is False
+
+
+def test_direct_worker_rejects_incapable_read_only_seat_before_artifacts(monkeypatch, tmp_path, capsys):
+    output_dir = tmp_path / "run"
+    monkeypatch.setattr(
+        aboyeur,
+        "code_graph_brief",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("validation happened too late")),
+    )
+
+    rc = aboyeur.run(
+        "inspect feature",
+        _roster_with_incapable_worker(),
+        worker="coder",
+        read_only=True,
+        cwd=tmp_path,
+        output_dir=output_dir,
+    )
+
+    assert rc == 2
+    assert not output_dir.exists()
+    err = capsys.readouterr().err
+    assert "coder" in err
+    assert "agents.coder.read_only_capable is false" in err
+
+
 def test_roster_payload_includes_sandbox():
     roster = Roster(
         orchestrator="chef",