Implementation Plan: T5-P4-A1-WP1 Create BackendCmdBuilderBase with Shared Env Assembly

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, `FlagVocabulary` NamedTuple — shared env assembly base class for all backends |
 | `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,145 @@
+"""Shared base class and types for backend command builder implementations.
+
+Owns the eight env keys that every backend's skill-session and food-truck
+cmd builders must inject (campaign id, kitchen session id, scenario step name,
+allowed write prefixes, cwd, MCP token ceiling, MCP connection nonblocking).
+Concrete backends (``ClaudeCodeBackend``, ``CodexBackend``) inherit from
+:class:`BackendCmdBuilderBase` and contribute backend-specific extension
+points (binary, sandbox default, env policy, flag vocabulary).
+
+This module is IL-1 — it imports from ``autoskillit.core`` and
+same-package modules only. It must NOT import from ``claude.py`` or
+``codex.py`` to avoid a cyclic import.
+"""
+
+from __future__ import annotations
+
+import abc
+import os
+from dataclasses import asdict, dataclass
+from typing import Any, NamedTuple
+
+from autoskillit.core import (
+    CAMPAIGN_ID_ENV_VAR,
+    KITCHEN_SESSION_ID_ENV_VAR,
+    EnvPolicy,
+    SkillSessionConfig,
+)
+from autoskillit.execution.backends._claude_prompt import _MAX_MCP_OUTPUT_TOKENS_VALUE
+
+
+class FlagVocabulary(NamedTuple):
+    """Per-backend CLI flag structure.
+
+    Captures which flags a backend treats as variadic (may repeat with
+    distinct values, e.g. ``--add-dir``) vs non-variadic, plus the canonical
+    flag spellings for the four flags every backend shares (model, add-dir,
+    resume, config override).
+    """
+
+    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.ABC):
+    """Abstract base for backend cmd builders.
+
+    Concrete subclasses (e.g. ``ClaudeCodeBackend``) supply the four
+    extension points below and inherit the two concrete helpers. The base
+    is intentionally minimal — it owns ONLY the eight shared env keys and
+    the SkillSessionConfig unpacking helper. Backend-specific keys
+    (``AUTOSKILLIT_HEADLESS``, ``AUTOSKILLIT_SESSION_TYPE``, Claude's
+    ``CLAUDE_CODE_EXIT_AFTER_STOP_DELAY``, Codex's
+    ``AUTOSKILLIT_COMPLETION_MARKER``, etc.) remain in each caller's local
+    dict.
+    """
+
+    @property
+    @abc.abstractmethod
+    def _binary(self) -> str:
+        """Return the CLI binary name (e.g. ``"claude"`` or ``"codex"``)."""
+
+    @property
+    @abc.abstractmethod
+    def _sandbox_default(self) -> str:
+        """Return the backend's default sandbox mode string.
+
+        ``""`` for backends without a sandbox concept; ``"workspace-write"``
+        for backends that default to workspace-write sandbox.
+        """
+
+    @property
+    @abc.abstractmethod
+    def _env_policy(self) -> EnvPolicy:
+        """Return the backend's env policy instance."""
+
+    @property
+    @abc.abstractmethod
+    def _flag_vocabulary(self) -> FlagVocabulary:
+        """Return the backend's flag vocabulary describing its CLI surface."""
+
+    def _assemble_shared_env_extras(
+        self,
+        *,
+        scenario_step_name: str = "",
+        allowed_write_prefix: str = "",
+        allowed_write_prefixes: tuple[str, ...] = (),
+        cwd: str = "",
+    ) -> dict[str, str]:
+        """Build the dict of env keys every backend injects.
+
+        Always-set keys:
+          - ``MAX_MCP_OUTPUT_TOKENS`` -> ``_MAX_MCP_OUTPUT_TOKENS_VALUE``
+          - ``MCP_CONNECTION_NONBLOCKING`` -> ``"0"``
+
+        Conditional keys (present only when their input is truthy or
+        present in ``os.environ``):
+          - ``CAMPAIGN_ID_ENV_VAR`` -> ``os.environ[...]``
+          - ``KITCHEN_SESSION_ID_ENV_VAR`` -> ``os.environ[...]``
+          - ``SCENARIO_STEP_NAME`` -> ``scenario_step_name``
+          - ``AUTOSKILLIT_ALLOWED_WRITE_PREFIX`` -> ``allowed_write_prefix``
+          - ``AUTOSKILLIT_ALLOWED_WRITE_PREFIXES`` -> colon-joined tuple
+          - ``AUTOSKILLIT_CWD`` -> ``cwd``
+
+        ``AUTOSKILLIT_SESSION_TYPE`` is intentionally NOT assembled here.
+        It is a backend-specific concern: claude.py writes it from a local
+        constant (``SESSION_TYPE_SKILL`` / ``SESSION_TYPE_ORCHESTRATOR``)
+        and codex.py writes it via ``_codex_exec_extras``. Including it as
+        a parameter here would be a structural trap for future callers.
+        """
+        extras: dict[str, str] = {
+            "MAX_MCP_OUTPUT_TOKENS": _MAX_MCP_OUTPUT_TOKENS_VALUE,
+            "MCP_CONNECTION_NONBLOCKING": "0",
+        }
+        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 scenario_step_name:
+            extras["SCENARIO_STEP_NAME"] = scenario_step_name
+        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
+        return extras
+
+    def _apply_config(self, config: SkillSessionConfig) -> dict[str, Any]:
+        """Unpack a :class:`SkillSessionConfig` into a flat ``dict[str, Any]``.
+
+        Round-trips all 18 fields of ``SkillSessionConfig``. Backends
+        consume this dict when forwarding fields into per-backend env vars
+        or CLI flags.
+        """
+        return asdict(config)
+
+
+__all__ = ["BackendCmdBuilderBase", "FlagVocabulary"]

src/autoskillit/execution/backends/claude.py

@@ -15,14 +15,14 @@
     AGENT_BACKEND_ENV_VAR,
     AUTOSKILLIT_APPLICABLE_GUARDS,
     AUTOSKILLIT_WRITE_GUARD_TOOL_NAMES,
-    CAMPAIGN_ID_ENV_VAR,
     CLAUDE_CODE_CAPABILITIES,
     CONTEXT_EXHAUSTION_MARKER,
-    KITCHEN_SESSION_ID_ENV_VAR,
+    NON_VARIADIC_CLAUDE_FLAGS,
     ORCHESTRATOR_SESSION_REQUIRED_ENV,
     SESSION_TYPE_ORCHESTRATOR,
     SESSION_TYPE_SKILL,
     SKILL_SESSION_REQUIRED_ENV,
+    VARIADIC_CLAUDE_FLAGS,
     AgentSessionResult,
     BackendCapabilities,
     BackendConventions,
@@ -33,6 +33,7 @@
     ClaudeFlags,
     CmdSpec,
     DirectInstall,
+    EnvPolicy,
     MarketplaceInstall,
     NamedResume,
     NoResume,
@@ -53,11 +54,14 @@
     load_yaml,
     pkg_root,
 )
+from autoskillit.execution.backends._backend_cmd_builder_base import (
+    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,
@@ -83,6 +87,16 @@
 ]
 
 
+CLAUDE_FLAG_VOCABULARY = FlagVocabulary(
+    variadic_flags=VARIADIC_CLAUDE_FLAGS,
+    non_variadic_flags=NON_VARIADIC_CLAUDE_FLAGS,
+    model_flag=ClaudeFlags.MODEL,
+    add_dir_flag=ClaudeFlags.ADD_DIR,
+    resume_flag=ClaudeFlags.RESUME,
+    config_override_flag="",
+)
+
+
 @dataclass(frozen=True, slots=True)
 class ClaudeEnvPolicy:
     def build_env(
@@ -270,7 +284,23 @@ def parse_stdout(self, stdout: str, *, exit_code: int = 0) -> AgentSessionResult
 
 
 @dataclass(frozen=True, slots=True)
-class ClaudeCodeBackend:
+class ClaudeCodeBackend(BackendCmdBuilderBase):
+    @property
+    def _binary(self) -> str:
+        return "claude"
+
+    @property
+    def _sandbox_default(self) -> str:
+        return ""
+
+    @property
+    def _env_policy(self) -> EnvPolicy:
+        return ClaudeEnvPolicy()
+
+    @property
+    def _flag_vocabulary(self) -> FlagVocabulary:
+        return CLAUDE_FLAG_VOCABULARY
+
     @property
     def name(self) -> str:
         return AGENT_BACKEND_CLAUDE_CODE
@@ -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(
+                scenario_step_name=scenario_step_name,
+                allowed_write_prefix=allowed_write_prefix,
+                allowed_write_prefixes=allowed_write_prefixes,
+                cwd=cwd,
+            )
+        )
         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,14 @@ 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(
+                scenario_step_name=scenario_step_name,
+                allowed_write_prefix=allowed_write_prefix,
+                allowed_write_prefixes=allowed_write_prefixes,
+                cwd=cwd,
+            )
+        )
         if env_extras:
             for k, v in env_extras.items():
                 if k not in _PROVIDER_EXTRAS_BASE_DENYLIST: