Implementation Plan: T5-P4-A1-WP1 BackendCmdBuilderBase ABC and FlagVocabulary

src/autoskillit/execution/__init__.py

@@ -37,10 +37,7 @@
     sync_hooks_to_codex_config,
 )
 from autoskillit.execution.ci import DefaultCIWatcher
-from autoskillit.execution.commands import (
-    _MAX_MCP_OUTPUT_TOKENS_VALUE,  # noqa: F401
-    ClaudeHeadlessCmd,
-)
+from autoskillit.execution.commands import ClaudeHeadlessCmd
 from autoskillit.execution.db import (
     DefaultDatabaseReader,
 )

src/autoskillit/execution/backends/AGENTS.md

@@ -7,6 +7,7 @@ IL-1 backend abstraction layer — concrete `CodingAgentBackend` implementations
 | File | Purpose |
 |------|---------|
 | `__init__.py` | `BACKEND_REGISTRY`, `get_backend()` factory, re-exports |
+| `_backend_cmd_builder_base.py` | `BackendCmdBuilderBase` ABC (frozen dataclass), `FlagVocabulary` NamedTuple, `SHARED_BASELINE_ENV` — canonical location for shared env-assembly keys |
 | `claude.py` | `ClaudeCodeBackend` (incl. `validate_session_layout`), `ClaudeEnvPolicy`, `ClaudeSessionLocator`, `ClaudeStreamParser`, `ClaudeResultParser` (prompt utilities moved to `_claude_prompt.py`) |
 | `_claude_prompt.py` | Prompt injection utilities, session constants (`_ensure_skill_prefix`, `_inject_completion_directive`, `_compose_resume_prompt`, etc.), shared by claude + codex + commands |
 | `codex.py` | `CodexFlags`, `CodexBackend` (incl. `validate_session_layout`, `setup_session_dir` agent TOML generation via `_generate_agent_tomls`), `CodexEnvPolicy`, `CodexSessionLocator` (parse/config moved to `_codex_parse.py` / `_codex_config.py`) |

src/autoskillit/execution/backends/_backend_cmd_builder_base.py

@@ -0,0 +1,159 @@
+"""Canonical location for the BackendCmdBuilderBase ABC and shared env-assembly keys.
+
+Both ``ClaudeCodeBackend`` (in ``claude.py``) and ``CodexBackend`` (in ``codex.py``)
+inherit from :class:`BackendCmdBuilderBase`. The base class owns the shared
+``_assemble_shared_env_extras`` static helper and four abstract extension points
+(``_binary``, ``_sandbox_default``, ``_env_policy``, ``_flag_vocabulary``) that
+each concrete backend implements.
+
+This module is stdlib-only plus IL-0 core imports. It is IL-1 compliant — no
+imports from ``claude.py`` or ``codex.py`` (the two concrete backends).
+"""
+
+from __future__ import annotations
+
+import os
+from abc import ABC, abstractmethod
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any, NamedTuple
+
+from autoskillit.core import (
+    CAMPAIGN_ID_ENV_VAR,
+    KITCHEN_SESSION_ID_ENV_VAR,
+    SkillSessionConfig,
+)
+
+if TYPE_CHECKING:
+    from autoskillit.core import EnvPolicy
+
+
+# Injected into every AutoSkillit-launched headless and cook session.
+# Raises the Claude Code client-side MCP tool result size gate from the
+# default 25,000 tokens to 50,000, preventing open_kitchen() responses
+# from being persisted to a file instead of returned inline.
+_MAX_MCP_OUTPUT_TOKENS_VALUE: str = "50000"
+
+
+# Baseline env vars injected into EVERY AutoSkillit-launched session. Callers
+# can override via env_extras. The shim ``_codex_exec_extras`` (in
+# ``codex.py``) imports this directly for the ``include_session_baseline=True``
+# path so resume sessions do NOT inadvertently pick up ambient campaign/kitchen
+# IDs from ``os.environ`` (which ``_assemble_shared_env_extras`` would).
+SHARED_BASELINE_ENV: Mapping[str, str] = MappingProxyType(
+    {
+        "MAX_MCP_OUTPUT_TOKENS": _MAX_MCP_OUTPUT_TOKENS_VALUE,
+        "MCP_CONNECTION_NONBLOCKING": "0",
+    }
+)
+
+
+class FlagVocabulary(NamedTuple):
+    """Per-backend CLI flag metadata used by the shared ``CmdBuilder``."""
+
+    variadic_flags: frozenset[str]
+    non_variadic_flags: frozenset[str]
+    model_flag: str
+    add_dir_flag: str
+    resume_flag: str
+    config_override_flag: str
+
+
+@dataclass(frozen=True, slots=True)
+class BackendCmdBuilderBase(ABC):
+    """Base class for backend command builders.
+
+    Owns the shared env-assembly logic that both ``ClaudeCodeBackend`` and
+    ``CodexBackend`` previously duplicated. Subclasses must implement four
+    abstract extension points that capture backend-specific behavior.
+    """
+
+    @abstractmethod
+    def _binary(self) -> str:
+        """Return the CLI binary name for this backend."""
+
+    @abstractmethod
+    def _sandbox_default(self) -> str:
+        """Return the default sandbox mode string for this backend."""
+
+    @abstractmethod
+    def _env_policy(self) -> EnvPolicy:
+        """Return the backend's :class:`EnvPolicy` instance."""
+
+    @abstractmethod
+    def _flag_vocabulary(self) -> FlagVocabulary:
+        """Return the backend's :class:`FlagVocabulary`."""
+
+    @staticmethod
+    def _assemble_shared_env_extras(
+        *,
+        write_prefix: str = "",
+        write_prefixes: tuple[str, ...] = (),
+        cwd: str = "",
+        scenario_step_name: str = "",
+    ) -> dict[str, str]:
+        """Assemble the eight shared env keys consumed by both backends.
+
+        Always-on keys (two): ``MAX_MCP_OUTPUT_TOKENS``, ``MCP_CONNECTION_NONBLOCKING``.
+
+        Conditional keys (six): ``SCENARIO_STEP_NAME``,
+        ``CAMPAIGN_ID_ENV_VAR``, ``KITCHEN_SESSION_ID_ENV_VAR``,
+        ``AUTOSKILLIT_ALLOWED_WRITE_PREFIX``, ``AUTOSKILLIT_ALLOWED_WRITE_PREFIXES``,
+        ``AUTOSKILLIT_CWD``. Each is included only when its input is non-empty
+        (campaign/kitchen IDs are also read from the ambient ``os.environ``).
+        """
+        extras: dict[str, str] = dict(SHARED_BASELINE_ENV)
+        if scenario_step_name:
+            extras["SCENARIO_STEP_NAME"] = scenario_step_name
+        campaign_id = os.environ.get(CAMPAIGN_ID_ENV_VAR)
+        if campaign_id:
+            extras[CAMPAIGN_ID_ENV_VAR] = campaign_id
+        kitchen_session_id = os.environ.get(KITCHEN_SESSION_ID_ENV_VAR)
+        if kitchen_session_id:
+            extras[KITCHEN_SESSION_ID_ENV_VAR] = kitchen_session_id
+        if write_prefix:
+            extras["AUTOSKILLIT_ALLOWED_WRITE_PREFIX"] = write_prefix
+        if write_prefixes:
+            extras["AUTOSKILLIT_ALLOWED_WRITE_PREFIXES"] = ":".join(write_prefixes)
+        if cwd:
+            extras["AUTOSKILLIT_CWD"] = cwd
+        return extras
+
+    def _apply_config(self, config: SkillSessionConfig) -> dict[str, Any]:
+        """Unpack :class:`SkillSessionConfig` command-building fields into a plain dict.
+
+        Separates shared fields (consumed by ``_assemble_shared_env_extras``)
+        from backend-specific fields. The returned dict is the single source
+        of truth that backend-specific builders can read from instead of
+        accepting each field as a separate parameter.
+
+        ``backend_override`` is intentionally excluded — it is consumed upstream
+        at the headless layer before command builders are invoked.
+        """
+        return {
+            "completion_marker": config.completion_marker,
+            "model": config.model,
+            "plugin_source": config.plugin_source,
+            "output_format": config.output_format,
+            "add_dirs": config.add_dirs,
+            "exit_after_stop_delay_ms": config.exit_after_stop_delay_ms,
+            "stream_idle_timeout_ms": config.stream_idle_timeout_ms,
+            "scenario_step_name": config.scenario_step_name,
+            "temp_dir_relpath": config.temp_dir_relpath,
+            "allowed_write_prefix": config.allowed_write_prefix,
+            "allowed_write_prefixes": config.allowed_write_prefixes,
+            "provider_extras": config.provider_extras,
+            "profile_name": config.profile_name,
+            "resume_session_id": config.resume_session_id,
+            "resume_checkpoint": config.resume_checkpoint,
+            "resume_message": config.resume_message,
+            "sandbox_mode": config.sandbox_mode,
+        }
+
+
+__all__ = [
+    "BackendCmdBuilderBase",
+    "FlagVocabulary",
+    "SHARED_BASELINE_ENV",
+]

src/autoskillit/execution/backends/_claude_prompt.py

@@ -3,9 +3,7 @@
 from __future__ import annotations
 
 import os
-from collections.abc import Mapping
 from dataclasses import dataclass
-from types import MappingProxyType
 from typing import Any, NamedTuple
 
 from autoskillit.core import (
@@ -25,23 +23,6 @@ def _extract_write_artifacts(tool_uses: list[dict[str, Any]]) -> list[str]:
     ]
 
 
-# Injected into every AutoSkillit-launched headless and cook session.
-# Raises the Claude Code client-side MCP tool result size gate from the
-# default 25,000 tokens to 50,000, preventing open_kitchen() responses
-# from being persisted to a file instead of returned inline.
-_MAX_MCP_OUTPUT_TOKENS_VALUE: str = "50000"
-
-# Baseline env vars injected into EVERY AutoSkillit-launched Claude session
-# (both interactive and headless). Callers can override via env_extras.
-# Analogous to IDE_ENV_ALWAYS_EXTRAS in _claude_env.py but scoped to
-# session-level concerns rather than IDE scrubbing.
-_SESSION_BASELINE_ENV: Mapping[str, str] = MappingProxyType(
-    {
-        "MAX_MCP_OUTPUT_TOKENS": _MAX_MCP_OUTPUT_TOKENS_VALUE,
-        "MCP_CONNECTION_NONBLOCKING": "0",
-    }
-)
-
 # Non-negotiable env overrides applied to every headless subprocess launch.
 # Injected after build_agent_env() so callers cannot clobber these values via
 # extras. Scoped to headless/resume commands only — never applied to interactive

src/autoskillit/execution/backends/claude.py

@@ -18,7 +18,6 @@
     CAMPAIGN_ID_ENV_VAR,
     CLAUDE_CODE_CAPABILITIES,
     CONTEXT_EXHAUSTION_MARKER,
-    KITCHEN_SESSION_ID_ENV_VAR,
     ORCHESTRATOR_SESSION_REQUIRED_ENV,
     SESSION_TYPE_ORCHESTRATOR,
     SESSION_TYPE_SKILL,
@@ -53,13 +52,16 @@
     load_yaml,
     pkg_root,
 )
+from autoskillit.execution.backends._backend_cmd_builder_base import (
+    SHARED_BASELINE_ENV,
+    BackendCmdBuilderBase,
+    FlagVocabulary,
+)
 from autoskillit.execution.backends._claude_prompt import (
     _HEADLESS_ENV_HARDENING,
     _HEADLESS_EXCLUSIVE_VARS,
     _INTERACTIVE_ENV_EXCLUSIONS,
-    _MAX_MCP_OUTPUT_TOKENS_VALUE,
     _PROVIDER_EXTRAS_BASE_DENYLIST,
-    _SESSION_BASELINE_ENV,
     _SKILL_SESSION_EXTRAS_DENYLIST,
     PromptBuildContext,
     _apply_output_format,
@@ -270,7 +272,35 @@ def parse_stdout(self, stdout: str, *, exit_code: int = 0) -> AgentSessionResult
 
 
 @dataclass(frozen=True, slots=True)
-class ClaudeCodeBackend:
+class ClaudeCodeBackend(BackendCmdBuilderBase):
+    def _binary(self) -> str:
+        return "claude"
+
+    def _sandbox_default(self) -> str:
+        return "workspace-write"
+
+    def _env_policy(self) -> ClaudeEnvPolicy:
+        return ClaudeEnvPolicy()
+
+    def _flag_vocabulary(self) -> FlagVocabulary:
+        return FlagVocabulary(
+            variadic_flags=frozenset(
+                {ClaudeFlags.ADD_DIR, ClaudeFlags.PLUGIN_DIR, ClaudeFlags.TOOLS}
+            ),
+            non_variadic_flags=frozenset(
+                {
+                    ClaudeFlags.PRINT,
+                    ClaudeFlags.MODEL,
+                    ClaudeFlags.RESUME,
+                    ClaudeFlags.DANGEROUSLY_SKIP_PERMISSIONS,
+                }
+            ),
+            model_flag=ClaudeFlags.MODEL,
+            add_dir_flag=ClaudeFlags.ADD_DIR,
+            resume_flag=ClaudeFlags.RESUME,
+            config_override_flag="",
+        )
+
     @property
     def name(self) -> str:
         return AGENT_BACKEND_CLAUDE_CODE
@@ -429,7 +459,7 @@ def build_interactive_cmd(
             builder.variadic_pair(ClaudeFlags.ADD_DIR, str(d))
         for t in tools:
             builder.variadic_pair(ClaudeFlags.TOOLS, t)
-        merged: dict[str, str] = dict(_SESSION_BASELINE_ENV)
+        merged: dict[str, str] = dict(SHARED_BASELINE_ENV)
         if env_extras:
             merged.update(env_extras)
         interactive_base = {
@@ -468,7 +498,7 @@ def build_resume_cmd(
                 pass
             case None:
                 pass
-        merged: dict[str, str] = dict(_SESSION_BASELINE_ENV)
+        merged: dict[str, str] = dict(SHARED_BASELINE_ENV)
         if env_extras:
             merged.update(env_extras)
         env = dict(build_agent_env(base={}, extras=merged))
@@ -600,8 +630,6 @@ def _build_skill_session_cmd_impl(
         extras: dict[str, str] = {
             "AUTOSKILLIT_HEADLESS": "1",
             "AUTOSKILLIT_SESSION_TYPE": SESSION_TYPE_SKILL,
-            "MAX_MCP_OUTPUT_TOKENS": _MAX_MCP_OUTPUT_TOKENS_VALUE,
-            "MCP_CONNECTION_NONBLOCKING": "0",
             AGENT_BACKEND_ENV_VAR: AGENT_BACKEND_CLAUDE_CODE,
             AGENT_BACKEND_DYNACONF_ENV_VAR: AGENT_BACKEND_CLAUDE_CODE,
             AUTOSKILLIT_APPLICABLE_GUARDS: ",".join(sorted(self.capabilities.applicable_guards)),
@@ -613,20 +641,14 @@ def _build_skill_session_cmd_impl(
             extras["CLAUDE_CODE_EXIT_AFTER_STOP_DELAY"] = str(exit_after_stop_delay_ms)
         if stream_idle_timeout_ms > 0:
             extras["CLAUDE_STREAM_IDLE_TIMEOUT_MS"] = str(stream_idle_timeout_ms)
-        if scenario_step_name:
-            extras["SCENARIO_STEP_NAME"] = scenario_step_name
-        campaign_id = os.environ.get(CAMPAIGN_ID_ENV_VAR)
-        if campaign_id:
-            extras[CAMPAIGN_ID_ENV_VAR] = campaign_id
-        kitchen_session_id = os.environ.get(KITCHEN_SESSION_ID_ENV_VAR)
-        if kitchen_session_id:
-            extras[KITCHEN_SESSION_ID_ENV_VAR] = kitchen_session_id
-        if allowed_write_prefix:
-            extras["AUTOSKILLIT_ALLOWED_WRITE_PREFIX"] = allowed_write_prefix
-        if allowed_write_prefixes:
-            extras["AUTOSKILLIT_ALLOWED_WRITE_PREFIXES"] = ":".join(allowed_write_prefixes)
-        if cwd:
-            extras["AUTOSKILLIT_CWD"] = cwd
+        extras.update(
+            self._assemble_shared_env_extras(
+                write_prefix=allowed_write_prefix,
+                write_prefixes=allowed_write_prefixes,
+                cwd=cwd,
+                scenario_step_name=scenario_step_name,
+            )
+        )
         extras["AUTOSKILLIT_SKILL_NAME"] = extract_skill_name(skill_command) or ""
         if provider_extras:
             for k, v in provider_extras.items():
@@ -705,8 +727,6 @@ def build_food_truck_cmd(
         extras: dict[str, str] = {
             "AUTOSKILLIT_HEADLESS": "1",
             "AUTOSKILLIT_SESSION_TYPE": SESSION_TYPE_ORCHESTRATOR,
-            "MAX_MCP_OUTPUT_TOKENS": _MAX_MCP_OUTPUT_TOKENS_VALUE,
-            "MCP_CONNECTION_NONBLOCKING": "0",
             AGENT_BACKEND_ENV_VAR: AGENT_BACKEND_CLAUDE_CODE,
             AGENT_BACKEND_DYNACONF_ENV_VAR: AGENT_BACKEND_CLAUDE_CODE,
             AUTOSKILLIT_APPLICABLE_GUARDS: ",".join(sorted(self.capabilities.applicable_guards)),
@@ -718,17 +738,15 @@ def build_food_truck_cmd(
             extras["CLAUDE_CODE_EXIT_AFTER_STOP_DELAY"] = str(exit_after_stop_delay_ms)
         if stream_idle_timeout_ms > 0:
             extras["CLAUDE_STREAM_IDLE_TIMEOUT_MS"] = str(stream_idle_timeout_ms)
-        if scenario_step_name:
-            extras["SCENARIO_STEP_NAME"] = scenario_step_name
-        kitchen_session_id = os.environ.get(KITCHEN_SESSION_ID_ENV_VAR)
-        if kitchen_session_id:
-            extras[KITCHEN_SESSION_ID_ENV_VAR] = kitchen_session_id
-        if allowed_write_prefix:
-            extras["AUTOSKILLIT_ALLOWED_WRITE_PREFIX"] = allowed_write_prefix
-        if allowed_write_prefixes:
-            extras["AUTOSKILLIT_ALLOWED_WRITE_PREFIXES"] = ":".join(allowed_write_prefixes)
-        if cwd:
-            extras["AUTOSKILLIT_CWD"] = cwd
+        extras.update(
+            self._assemble_shared_env_extras(
+                write_prefix=allowed_write_prefix,
+                write_prefixes=allowed_write_prefixes,
+                cwd=cwd,
+                scenario_step_name=scenario_step_name,
+            )
+        )
+        extras.pop(CAMPAIGN_ID_ENV_VAR, None)  # food truck does not propagate campaign ID
         if env_extras:
             for k, v in env_extras.items():
                 if k not in _PROVIDER_EXTRAS_BASE_DENYLIST: