Dashboard: pick a per-task agent CLI in the n new-task flow

The per-task `agent_cli` override has had a full data path since #384/#385 (model,
store, service, REST, client, MCP) but no way to set it interactively — the dashboard's
`n` flow went repo -> workflow -> memo and always inherited the repo default.

Insert a CLI step between workflow and memo, pre-selected to the repo's own default so
the common path is still Enter/Enter/Enter-and-type. Picking the default sends
`agent_cli=None` rather than a redundant explicit value, matching `resolve_agent_cli`'s
contract.

- `KNOWN_AGENT_CLIS` in `core/models.py`. The CLI name set is a control-plane concept
  (it drives the base-image variant + config mount), and `terminal`/`taskservice` can't
  import `container/cli` at runtime without breaking the determinism invariant, so the
  list is restated in the LLM-free core and guarded by a test.
- `registered_agent_clis()` on the adapter registry — the registry is populated lazily,
  so a drift test comparing against `_REGISTRY` directly would pass vacuously. The new
  guard also pins the two long-duplicated `DEFAULT_AGENT_CLI` constants together.
- `_OptionListModal` grows an `initial=` param; it had no way to set the starting
  highlight, which the pre-select needs. The repo default is named in the picker title
  rather than baked into an option label, since `ChoiceScreen` dismisses the option text.
- Read the repo default from the `list_repos()` payload `action_new_task` already
  fetches, not `self._repo_clis` — that cache is only filled on a refresh pass and can
  be empty or stale at this point.
- Swap four hardcoded `"claude"` literals in `dashboard.py` for `DEFAULT_AGENT_CLI`.

The fake-adapter registration test now cleans up after itself; it was leaking `fake-cli`
into the module-global registry, which would have made the drift guard order-dependent.

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

Author: claude

src/panopticon/container/cli/__init__.py

@@ -8,6 +8,13 @@
     AgentCLI,
     get_agent_cli,
     register_agent_cli,
+    registered_agent_clis,
 )
 
-__all__ = ["DEFAULT_AGENT_CLI", "AgentCLI", "get_agent_cli", "register_agent_cli"]
+__all__ = [
+    "DEFAULT_AGENT_CLI",
+    "AgentCLI",
+    "get_agent_cli",
+    "register_agent_cli",
+    "registered_agent_clis",
+]

src/panopticon/container/cli/base.py

@@ -172,6 +172,17 @@ def get_agent_cli(name: str | None = None) -> AgentCLI:
         raise KeyError(f"unknown agent CLI {key!r}; registered: {sorted(_REGISTRY)}") from None
 
 
+def registered_agent_clis() -> tuple[str, ...]:
+    """Return the registered adapter names, sorted (loading the built-ins first).
+
+    The registry is populated lazily, so callers that only want the *names* need this rather than
+    reaching into :data:`_REGISTRY` (which is empty until an adapter module is imported). Used by
+    the drift test that keeps :data:`panopticon.core.models.KNOWN_AGENT_CLIS` honest.
+    """
+    _load_builtin_adapters()
+    return tuple(sorted(_REGISTRY))
+
+
 def _load_builtin_adapters() -> None:
     """Register the built-in adapters (imported lazily so this module holds only the contract)."""
     from panopticon.container.cli.claude import ClaudeAgentCLI

src/panopticon/core/models.py

@@ -226,6 +226,15 @@ class Repo:
 #: The CLI a task falls back to when neither the task nor its repo names one (ADR 0014 §2/§3).
 DEFAULT_AGENT_CLI = "claude"
 
+#: The CLI names an operator can pick from (the dashboard's new-task picker offers these).
+#:
+#: The set of CLIs is a **control-plane** concept — it drives the base-image variant and the
+#: config-dir mount — so it lives here, in the LLM-free core, rather than being read from the
+#: adapter registry in :mod:`panopticon.container.cli` (the only LLM-bearing package; importing it
+#: from ``terminal``/``taskservice`` would break the determinism invariant). The two are kept in
+#: sync by a test — see ``tests/container/test_cli_base.py``.
+KNOWN_AGENT_CLIS: tuple[str, ...] = ("claude", "codex")
+
 
 def resolve_agent_cli(task_agent_cli: str | None, repo_agent_cli: str | None) -> str:
     """Resolve a task's effective agent CLI: its own override → the repo default → ``"claude"``.

src/panopticon/terminal/dashboard.py

@@ -105,7 +105,7 @@
 from panopticon.client import JsonObj, TaskServiceClient
 from panopticon.core.artifacts import InvalidArtifactName, validate_segment
 from panopticon.core.dirs import ARTIFACTS_DIR
-from panopticon.core.models import resolve_agent_cli
+from panopticon.core.models import DEFAULT_AGENT_CLI, KNOWN_AGENT_CLIS, resolve_agent_cli
 from panopticon.core.state import TERMINAL_LABELS
 from panopticon.sessionservice.local_runner import session_name
 from panopticon.taskservice.artifacts_fs import FilesystemArtifactStore
@@ -624,10 +624,13 @@ class _OptionListModal(ModalScreen[_ResultT | None]):
     BINDINGS = [("escape", "cancel", "Cancel")]
     BOX_ID = "list-box"
 
-    def __init__(self, title: str, options: list[str]) -> None:
+    def __init__(self, title: str, options: list[str], *, initial: str | None = None) -> None:
         super().__init__()
         self._title = title
         self._options = options
+        #: Which option starts highlighted, so Enter alone picks the expected one. Ignored when it
+        #: isn't in ``options``.
+        self._initial = initial
 
     def compose(self) -> ComposeResult:
         with Vertical(id=self.BOX_ID):
@@ -639,7 +642,10 @@ def _extra_widgets(self) -> Iterable[Widget]:
         return ()
 
     def on_mount(self) -> None:
-        self.query_one(OptionList).focus()
+        option_list = self.query_one(OptionList)
+        if self._initial is not None and self._initial in self._options:
+            option_list.highlighted = self._options.index(self._initial)
+        option_list.focus()
 
     def action_cancel(self) -> None:
         self.dismiss(None)
@@ -1407,13 +1413,13 @@ def __init__(
 
     def _initial(self, name: str) -> str:
         """A field's pre-populated value: the repo's stored value, else (create mode only)
-        ``main`` for ``default_base`` / ``claude`` for ``agent_cli``, else blank."""
+        ``main`` for ``default_base`` / the default CLI for ``agent_cli``, else blank."""
         stored = self._repo.get(name)
         if stored:
             return str(stored)
         if self._editing:
             return ""
-        return {"default_base": "main", "agent_cli": "claude"}.get(name, "")
+        return {"default_base": "main", "agent_cli": DEFAULT_AGENT_CLI}.get(name, "")
 
     def _wf_checked(self, wf: dict[str, Any]) -> bool:
         name = wf["name"]
@@ -1617,7 +1623,7 @@ def create(values: dict[str, Any]) -> str | None:
                     capabilities={"docker_in_docker": values["docker_in_docker"]},
                     enabled_workflows=values["enabled_workflows"],
                     disabled_workflows=values["disabled_workflows"],
-                    agent_cli=values["agent_cli"] or "claude",
+                    agent_cli=values["agent_cli"] or DEFAULT_AGENT_CLI,
                 )
             except httpx.HTTPStatusError as exc:
                 return f"Can't create: {_detail(exc)}"
@@ -1653,7 +1659,7 @@ def save(values: dict[str, Any]) -> str | None:
                     capabilities=capabilities,
                     enabled_workflows=values["enabled_workflows"],
                     disabled_workflows=values["disabled_workflows"],
-                    agent_cli=values["agent_cli"] or "claude",
+                    agent_cli=values["agent_cli"] or DEFAULT_AGENT_CLI,
                 )
             except httpx.HTTPStatusError as exc:
                 return f"Can't update: {_detail(exc)}"
@@ -1985,7 +1991,9 @@ def _load_repo_names(self) -> None:
         try:
             repos = self._client.list_repos()
             self._repo_names = {str(r["id"]): str(r["name"]) for r in repos}
-            self._repo_clis = {str(r["id"]): str(r.get("agent_cli") or "claude") for r in repos}
+            self._repo_clis = {
+                str(r["id"]): str(r.get("agent_cli") or DEFAULT_AGENT_CLI) for r in repos
+            }
         except Exception:
             pass
 
@@ -2242,11 +2250,20 @@ def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
         self.action_refresh()
 
     def action_new_task(self) -> None:
-        """`n`: create a task — pick a repo, a workflow, describe the work, then POST it."""
-        repos = [str(r["id"]) for r in self._client.list_repos()]
-        if not repos:
+        """`n`: create a task — pick a repo, a workflow, an agent CLI, describe the work, then POST it.
+
+        The CLI step is pre-selected to the repo's own default, so the common path stays
+        Enter/Enter/Enter-and-type; picking that default sends no override at all (ADR 0014 §3)."""
+        repo_records = self._client.list_repos()
+        if not repo_records:
             self.notify("Need at least one repo to create a task.", severity="warning")
             return
+        repos = [str(r["id"]) for r in repo_records]
+        # Read each repo's CLI from the payload we just fetched rather than `self._repo_clis`, which
+        # is only filled in on a refresh pass and so can be empty or stale here.
+        repo_clis = {
+            str(r["id"]): str(r.get("agent_cli") or DEFAULT_AGENT_CLI) for r in repo_records
+        }
 
         def pick_workflow(repo: str | None) -> None:
             if repo is None:
@@ -2256,34 +2273,57 @@ def pick_workflow(repo: str | None) -> None:
                 self.notify(f"No workflows enabled for repo {repo!r}.", severity="warning")
                 return
 
-            def describe(workflow: str | None) -> None:
+            def pick_cli(workflow: str | None) -> None:
                 if workflow is None:
                     return
+                repo_default = repo_clis.get(repo, DEFAULT_AGENT_CLI)
 
-                def create(result: tuple[str, bool, dict[str, str]] | None) -> None:
-                    if result is None:  # backed out
+                def describe(cli: str | None) -> None:
+                    if cli is None:
                         return
-                    memo_text, submit, artifacts_b64 = result
-                    stripped = memo_text.strip()
-                    if _apply_memo_filter(stripped):
-                        return
-                    if submit and stripped:
-                        self._client.create_task(
-                            repo,
-                            workflow,
-                            stripped,
-                            initial_prompt=stripped,
-                            artifacts_b64=artifacts_b64,
-                        )
-                    else:
-                        self._client.create_task(
-                            repo, workflow, stripped or None, artifacts_b64=artifacts_b64
-                        )
-                    self.action_refresh()
-
-                self.push_screen(MemoScreen(), create)
+                    # None = "no override, use the repo default" — what `resolve_agent_cli` expects.
+                    agent_cli = None if cli == repo_default else cli
+
+                    def create(result: tuple[str, bool, dict[str, str]] | None) -> None:
+                        if result is None:  # backed out
+                            return
+                        memo_text, submit, artifacts_b64 = result
+                        stripped = memo_text.strip()
+                        if _apply_memo_filter(stripped):
+                            return
+                        if submit and stripped:
+                            self._client.create_task(
+                                repo,
+                                workflow,
+                                stripped,
+                                initial_prompt=stripped,
+                                artifacts_b64=artifacts_b64,
+                                agent_cli=agent_cli,
+                            )
+                        else:
+                            self._client.create_task(
+                                repo,
+                                workflow,
+                                stripped or None,
+                                artifacts_b64=artifacts_b64,
+                                agent_cli=agent_cli,
+                            )
+                        self.action_refresh()
+
+                    self.push_screen(MemoScreen(), create)
+
+                # The default is named in the title, not baked into an option label — ChoiceScreen
+                # dismisses the option's text, so a decorated label would need stripping back off.
+                self.push_screen(
+                    ChoiceScreen(
+                        f"agent CLI (repo default: {repo_default})",
+                        list(KNOWN_AGENT_CLIS),
+                        initial=repo_default,
+                    ),
+                    describe,
+                )
 
-            self.push_screen(WorkflowScreen(workflows), describe)
+            self.push_screen(WorkflowScreen(workflows), pick_cli)
 
         self.push_screen(ChoiceScreen("repo", repos), pick_workflow)
 

tests/container/test_cli_base.py

@@ -13,11 +13,14 @@
 from panopticon.container.cli import (
     DEFAULT_AGENT_CLI,
     AgentCLI,
+    base,
     get_agent_cli,
     register_agent_cli,
+    registered_agent_clis,
 )
 from panopticon.container.cli.claude import ClaudeAgentCLI
 from panopticon.container.cli.codex import CodexAgentCLI
+from panopticon.core import models as core_models
 
 
 def test_default_resolves_to_claude() -> None:
@@ -26,6 +29,17 @@ def test_default_resolves_to_claude() -> None:
     assert isinstance(get_agent_cli("claude"), ClaudeAgentCLI)
 
 
+def test_core_agent_cli_constants_match_the_adapter_registry() -> None:
+    """The control plane can't import this package at runtime (the determinism invariant — `core`
+    stays LLM-free), so `core.models` re-states the CLI name list and the default. This is the guard
+    that keeps the two copies from drifting: add an adapter, and it fails until `core` knows about it.
+    """
+    assert list(registered_agent_clis()) == sorted(core_models.KNOWN_AGENT_CLIS)
+    assert core_models.DEFAULT_AGENT_CLI == DEFAULT_AGENT_CLI
+    # The fallback must itself be a selectable CLI.
+    assert core_models.DEFAULT_AGENT_CLI in core_models.KNOWN_AGENT_CLIS
+
+
 def test_codex_is_a_registered_built_in_adapter() -> None:
     # The second built-in CLI: registering it makes it resolvable with no launcher edit (ADR 0014 §2).
     assert isinstance(get_agent_cli("codex"), CodexAgentCLI)
@@ -73,8 +87,13 @@ def launch(self, config_dir: Path) -> None:  # pragma: no cover - not exercised
             pass
 
     register_agent_cli(_Fake)
-    resolved = get_agent_cli("fake-cli")
-    assert isinstance(resolved, _Fake) and resolved.config_dirname == ".fake"
+    try:
+        resolved = get_agent_cli("fake-cli")
+        assert isinstance(resolved, _Fake) and resolved.config_dirname == ".fake"
+    finally:
+        # The registry is module-global: leaving the fake in it would leak into any other test that
+        # inspects the registered names (e.g. the KNOWN_AGENT_CLIS drift guard above).
+        base._REGISTRY.pop("fake-cli", None)
 
 
 # -- shared base-class behaviour ----------------------------------------------------------------