Dedup AgentCLI adapters: hoist read_hook_payload + resolve_model to base, drop dead CONFIG_MOUNT

src/panopticon/container/cli/base.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import json
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
 from pathlib import Path
@@ -67,6 +68,10 @@ class AgentCLI(ABC):
     name: ClassVar[str]
     #: The CLI's config dir, relative to the container home (the launcher mounts it per-task).
     config_dirname: ClassVar[str]
+    #: The control plane's abstract model **tiers** mapped to this CLI's concrete model ids (ADR
+    #: 0014 §3a). Subclasses set the mapping; :meth:`resolve_model` reads it. Unknown tiers pass
+    #: through unchanged so a raw model id set directly still reaches ``--model`` verbatim.
+    MODEL_TIERS: ClassVar[Mapping[str, str]]
 
     @abstractmethod
     def render_skills(self, client: _Client, task_id: str, home: Path) -> list[Path]:
@@ -109,13 +114,30 @@ def write_credentials(self, config_dir: Path, env: Mapping[str, str]) -> Path |
         env (claude) or a credential file is already present. Never clobbers an existing one.
         """
 
-    @abstractmethod
     def resolve_model(self, tier: str) -> str:
-        """Map the control plane's abstract model **tier** to this CLI's concrete model id (§3a)."""
+        """Map the control plane's abstract model **tier** to this CLI's concrete model id (§3a).
+
+        The control plane stores a CLI-agnostic tier (e.g. ``"primary"``); adapters declare their
+        :attr:`MODEL_TIERS` mapping and this is the only place a tier becomes a provider model name,
+        keeping model vocabulary out of ``core``/``workflows``. Reserved tiers that are absent from
+        the mapping raise, so a stale image running pre-resolution code fails loud rather than leaking
+        the raw tier to ``--model`` (see :func:`resolve_tier`).
+        """
+        return resolve_tier(tier, self.MODEL_TIERS)
 
-    @abstractmethod
     def read_hook_payload(self, stdin: TextIO) -> dict[str, Any]:
-        """Tolerantly parse the turn-flip hook's stdin payload (empty/invalid → ``{}``)."""
+        """Tolerantly parse the hook's stdin JSON; empty/invalid input yields an empty payload."""
+        try:
+            raw = stdin.read()
+        except (OSError, ValueError):
+            return {}
+        if not raw or not raw.strip():
+            return {}
+        try:
+            data = json.loads(raw)
+        except json.JSONDecodeError:
+            return {}
+        return data if isinstance(data, dict) else {}
 
     @abstractmethod
     def has_live_background_task(self, payload: dict[str, Any]) -> bool:

src/panopticon/container/cli/claude.py

@@ -15,9 +15,9 @@
 import subprocess
 from collections.abc import Mapping
 from pathlib import Path
-from typing import Any, ClassVar, TextIO
+from typing import Any, ClassVar
 
-from panopticon.container.cli.base import AgentCLI, _Client, resolve_tier
+from panopticon.container.cli.base import AgentCLI, _Client
 from panopticon.container.config import update_json_config
 from panopticon.container.hooks import write_settings
 from panopticon.container.skills import write_commands, write_operation_commands
@@ -31,17 +31,13 @@
 #: Sent to claude as the first message when a container restarts mid-task on the agent's turn.
 INTERRUPT_PROMPT = "You were interrupted. Continue."
 
-#: The control plane's abstract model **tiers** mapped to claude's concrete ``--model`` ids (ADR
-#: 0014 §3a). This is the only place a provider model name appears; ``core``/``workflows`` name only
-#: the tier. Unknown values pass through unchanged (see ``resolve_model``).
-_MODEL_TIERS = {"primary": "opus"}
-
 
 class ClaudeAgentCLI(AgentCLI):
     """The `claude` adapter — every ``container/`` seam claude satisfies today, unchanged in effect."""
 
     name = "claude"
     config_dirname = ".claude"
+    MODEL_TIERS: ClassVar[Mapping[str, str]] = {"primary": "opus"}
 
     #: claude's main config file. Holds (besides per-container state) per-project trust acceptance.
     CONFIG_FILE: ClassVar[str] = ".claude.json"
@@ -130,32 +126,6 @@ def write_credentials(self, config_dir: Path, env: Mapping[str, str]) -> Path |
         """No-op: claude authenticates from the env var itself, with no on-disk credential to write."""
         return None
 
-    def resolve_model(self, tier: str) -> str:
-        """Map the control plane's abstract model tier to claude's concrete model id (ADR 0014 §3a).
-
-        The control plane stores a CLI-agnostic tier (e.g. ``"primary"``); this is the only place
-        that tier becomes a provider's model name (``"primary"`` → ``"opus"``), keeping model
-        vocabulary out of ``core``/``workflows``. Unknown values pass through unchanged so a raw
-        model id set directly (or a tier already persisted as its resolved name) still reaches
-        ``--model`` verbatim, while an unmapped **reserved** tier fails loud instead of leaking
-        through unresolved (see :func:`~panopticon.container.cli.base.resolve_tier`).
-        """
-        return resolve_tier(tier, _MODEL_TIERS)
-
-    def read_hook_payload(self, stdin: TextIO) -> dict[str, Any]:
-        """Tolerantly parse the hook's stdin JSON; empty/invalid input yields an empty payload."""
-        try:
-            raw = stdin.read()
-        except (OSError, ValueError):
-            return {}
-        if not raw or not raw.strip():
-            return {}
-        try:
-            data = json.loads(raw)
-        except json.JSONDecodeError:
-            return {}
-        return data if isinstance(data, dict) else {}
-
     def has_live_background_task(self, payload: dict[str, Any]) -> bool:
         """Whether the Stop payload reports a still-running background task.
 

src/panopticon/container/cli/codex.py

@@ -36,9 +36,9 @@
 import subprocess
 from collections.abc import Mapping
 from pathlib import Path
-from typing import Any, ClassVar, TextIO
+from typing import Any, ClassVar
 
-from panopticon.container.cli.base import AgentCLI, _Client, resolve_tier
+from panopticon.container.cli.base import AgentCLI, _Client
 from panopticon.container.config import update_toml_config
 from panopticon.container.hooks import HOOK_COMMAND
 from panopticon.container.skills import write_agent_operation_skills, write_agent_skills
@@ -93,14 +93,6 @@ def _find_resume_target(sessions_dir: Path) -> str | None:
     return best_id
 
 
-#: The control plane's abstract model **tiers** mapped to codex's concrete model ids (ADR 0014 §3a).
-#: The only place a provider model name appears; ``core``/``workflows`` name only the tier. ``primary``
-#: maps to codex's flagship (``gpt-5.6-sol``), verified against the pinned codex release (the
-#: ``CODEX_VERSION`` build arg in ``docker/Dockerfile``); unknown values pass through unchanged (see
-#: :meth:`CodexAgentCLI.resolve_model`).
-_MODEL_TIERS = {"primary": "gpt-5.6-sol"}
-
-
 def _command_hook(actor: str, event: str) -> dict[str, Any]:
     """One codex hook group: run the shared turn-flip callback with ``<actor> <event>``.
 
@@ -116,6 +108,7 @@ class CodexAgentCLI(AgentCLI):
 
     name = "codex"
     config_dirname = ".codex"
+    MODEL_TIERS: ClassVar[Mapping[str, str]] = {"primary": "gpt-5.6-sol"}
 
     #: codex's single config file, under the config dir. MCP, trust, and the unattended posture all
     #: merge into it (each adapter method touches only its own keys, via :func:`update_toml_config`).
@@ -283,31 +276,6 @@ def write_credentials(self, config_dir: Path, env: Mapping[str, str]) -> Path |
         auth.chmod(0o600)
         return auth
 
-    def resolve_model(self, tier: str) -> str:
-        """Map the control plane's abstract model tier to codex's concrete model id (ADR 0014 §3a).
-
-        The only place the tier (e.g. ``"primary"``) becomes a provider model name, keeping model
-        vocabulary out of ``core``/``workflows``. Unknown values pass through unchanged so a raw model
-        id set directly still reaches ``--model`` verbatim, while an unmapped **reserved** tier fails
-        loud instead of leaking through unresolved (see
-        :func:`~panopticon.container.cli.base.resolve_tier`).
-        """
-        return resolve_tier(tier, _MODEL_TIERS)
-
-    def read_hook_payload(self, stdin: TextIO) -> dict[str, Any]:
-        """Tolerantly parse the hook's stdin JSON; empty/invalid input yields an empty payload."""
-        try:
-            raw = stdin.read()
-        except (OSError, ValueError):
-            return {}
-        if not raw or not raw.strip():
-            return {}
-        try:
-            data = json.loads(raw)
-        except json.JSONDecodeError:
-            return {}
-        return data if isinstance(data, dict) else {}
-
     def has_live_background_task(self, payload: dict[str, Any]) -> bool:
         """Whether the Stop payload reports still-running background work (gates the turn flip).
 

src/panopticon/sessionservice/local_runner.py

@@ -77,12 +77,6 @@ def session_name(task_id: str) -> str:
 #: names it so the pane runs as that same user (ADR 0008 / the unprivileged-user work).
 CONTAINER_USER = "panopticon"
 
-#: The claude config-volume mount path — kept as a module constant for callers/tests that assume
-#: the default CLI; the spawn path derives the mount per-CLI via :func:`config_mount`. A **per-task**
-#: named volume is mounted here so the CLI's session history survives respawn/recreate (the container
-#: layer is thrown away each spawn, but the volume persists); per-task so concurrent tasks don't share.
-CONFIG_MOUNT = config_mount(DEFAULT_AGENT_CLI)
-
 
 class CommandRunner(Protocol):
     """Runs an external command and returns its stdout; ``check`` raises on non-zero exit.

tests/container/test_claude.py

@@ -4,7 +4,6 @@
 
 from __future__ import annotations
 
-import io
 import json
 from pathlib import Path
 
@@ -141,21 +140,6 @@ def test_resolve_model_maps_the_primary_tier_to_opus() -> None:
     assert ClaudeAgentCLI().resolve_model("primary") == "opus"
 
 
-def test_resolve_model_passes_unknown_values_through() -> None:
-    # A raw model id set directly (or a tier already persisted as its resolved name) reaches
-    # --model verbatim — so back-compat with a stored "opus" holds.
-    assert ClaudeAgentCLI().resolve_model("opus") == "opus"
-
-
-def test_resolve_model_rejects_an_unmapped_reserved_tier(monkeypatch: pytest.MonkeyPatch) -> None:
-    # A reserved tier the adapter doesn't map must fail loud, not leak through to --model — the
-    # backstop for the stale-image bug (a container running pre-resolution code forwarded the raw
-    # tier). We simulate it by stripping the tier map.
-    monkeypatch.setattr("panopticon.container.cli.claude._MODEL_TIERS", {})
-    with pytest.raises(ValueError, match="primary"):
-        ClaudeAgentCLI().resolve_model("primary")
-
-
 def test_built_in_workflow_tier_resolves_to_a_concrete_claude_model() -> None:
     # End to end across the two halves: the tier the control plane declares (never a model name)
     # resolves through the claude adapter to today's concrete model.
@@ -252,14 +236,6 @@ def test_write_credentials_is_a_no_op_for_claude(tmp_path: Path) -> None:
 # -- hook payload seam (background-task gating) --------------------------------------------------
 
 
-def test_read_hook_payload_tolerates_empty_and_invalid() -> None:
-    cli = ClaudeAgentCLI()
-    assert cli.read_hook_payload(io.StringIO("")) == {}
-    assert cli.read_hook_payload(io.StringIO("not json")) == {}
-    assert cli.read_hook_payload(io.StringIO("[]")) == {}  # JSON, but not an object
-    assert cli.read_hook_payload(io.StringIO('{"a": 1}')) == {"a": 1}
-
-
 @pytest.mark.parametrize(
     "payload,live",
     [

tests/container/test_cli_base.py

@@ -1,9 +1,11 @@
 """The agent-CLI adapter seam (ADR 0014): the ABC + the name-keyed registry. A CLI drops in by
-implementing :class:`AgentCLI` and registering under its name, with no launcher edit. The claude
-adapter's own behavior lives in :mod:`tests.container.test_claude`."""
+implementing :class:`AgentCLI` and registering under its name, with no launcher edit. Shared
+base-class behavior (read_hook_payload, resolve_model passthrough) lives here; per-adapter
+MODEL_TIERS mapping assertions live in their own modules."""
 
 from __future__ import annotations
 
+import io
 from pathlib import Path
 
 import pytest
@@ -38,6 +40,7 @@ def test_registering_an_adapter_makes_it_resolvable_without_a_launcher_edit() ->
     class _Fake(AgentCLI):
         name = "fake-cli"
         config_dirname = ".fake"
+        MODEL_TIERS = {}
 
         def render_skills(self, client: object, task_id: str, home: Path) -> list[Path]:
             return []
@@ -63,12 +66,6 @@ def auth_missing_detail(self, env: object, config_dir: object) -> str | None:
         def write_credentials(self, config_dir: Path, env: object) -> Path | None:
             return None
 
-        def resolve_model(self, tier: str) -> str:
-            return tier
-
-        def read_hook_payload(self, stdin: object) -> dict[str, object]:
-            return {}
-
         def has_live_background_task(self, payload: dict[str, object]) -> bool:
             return False
 
@@ -78,3 +75,28 @@ def launch(self, config_dir: Path) -> None:  # pragma: no cover - not exercised
     register_agent_cli(_Fake)
     resolved = get_agent_cli("fake-cli")
     assert isinstance(resolved, _Fake) and resolved.config_dirname == ".fake"
+
+
+# -- shared base-class behaviour ----------------------------------------------------------------
+
+
+def test_read_hook_payload_tolerates_empty_and_invalid() -> None:
+    # Shared implementation on the base — tested once; adapter tests cover only their own seams.
+    cli = ClaudeAgentCLI()
+    assert cli.read_hook_payload(io.StringIO("")) == {}
+    assert cli.read_hook_payload(io.StringIO("not json")) == {}
+    assert cli.read_hook_payload(io.StringIO("[]")) == {}  # JSON, but not an object
+    assert cli.read_hook_payload(io.StringIO('{"a": 1}')) == {"a": 1}
+
+
+def test_resolve_model_passes_unknown_tiers_through() -> None:
+    # The passthrough fallback lives on the base; adapters supply only their own mapping.
+    assert ClaudeAgentCLI().resolve_model("some-raw-model-id") == "some-raw-model-id"
+
+
+def test_resolve_model_rejects_an_unmapped_reserved_tier(monkeypatch: pytest.MonkeyPatch) -> None:
+    # A reserved tier absent from the adapter's MODEL_TIERS must fail loud — the stale-image
+    # backstop (resolve_tier raises instead of leaking the raw tier to --model).
+    monkeypatch.setattr(ClaudeAgentCLI, "MODEL_TIERS", {})
+    with pytest.raises(ValueError, match="primary"):
+        ClaudeAgentCLI().resolve_model("primary")

tests/container/test_codex.py

@@ -4,14 +4,11 @@
 
 from __future__ import annotations
 
-import io
 import json
 import time
 import tomllib
 from pathlib import Path
 
-import pytest
-
 from panopticon.container.cli.codex import CodexAgentCLI, _find_resume_target
 
 
@@ -234,18 +231,6 @@ def test_resolve_model_maps_the_primary_tier_to_a_codex_model() -> None:
     assert CodexAgentCLI().resolve_model("primary") == "gpt-5.6-sol"
 
 
-def test_resolve_model_passes_unknown_values_through() -> None:
-    assert CodexAgentCLI().resolve_model("gpt-5.6") == "gpt-5.6"
-
-
-def test_resolve_model_rejects_an_unmapped_reserved_tier(monkeypatch: pytest.MonkeyPatch) -> None:
-    # A reserved tier the adapter doesn't map fails loud rather than leaking to --model (the
-    # stale-image backstop). Simulated by stripping the tier map.
-    monkeypatch.setattr("panopticon.container.cli.codex._MODEL_TIERS", {})
-    with pytest.raises(ValueError, match="primary"):
-        CodexAgentCLI().resolve_model("primary")
-
-
 def test_built_in_workflow_tier_resolves_to_a_concrete_codex_model() -> None:
     from panopticon.workflows.github_self_reviewed import GithubSelfReviewed
 
@@ -527,14 +512,6 @@ def test_launch_argv_omits_effort_config_on_resume(tmp_path: Path) -> None:
 # -- hook seam (M3.6) ---------------------------------------------------------------------------
 
 
-def test_read_hook_payload_tolerates_empty_and_invalid() -> None:
-    cli = CodexAgentCLI()
-    assert cli.read_hook_payload(io.StringIO("")) == {}
-    assert cli.read_hook_payload(io.StringIO("not json")) == {}
-    assert cli.read_hook_payload(io.StringIO("[]")) == {}  # JSON, but not an object
-    assert cli.read_hook_payload(io.StringIO('{"a": 1}')) == {"a": 1}
-
-
 def test_has_live_background_task_always_false_for_codexs_stop_payload() -> None:
     # Codex's documented Stop payload carries no background-task array, so a real Stop flips the turn.
     cli = CodexAgentCLI()