dispatch_food_truck per-dispatch backend override (heterogeneous routing R1) (#4182)

## Summary

Add an optional `backend` parameter to the `dispatch_food_truck` MCP
tool that selects which coding-agent backend executes the dispatched
child session. When omitted, behavior is byte-for-byte identical to
today (uses `ctx.backend`). When supplied, the per-dispatch backend is
used for: capability gating (`food_truck_capable`), command building
(`build_food_truck_cmd`), pre-launch validation (`ensure_pre_launch`),
recipe admission control (capability overrides,
`backend-incompatible-skill` rule, `effective_backend_map`), session-log
resolution, and record persistence. The `DispatchRecord` gains a
`backend_name` field so downstream consumers (reset, diagnostics,
session-log lookup) resolve against the correct backend. The field is
added to `_RETRY_IDENTITY_FIELDS` so retried dispatches preserve the
backend override.

The implementation follows the existing `run_headless_core` call-local
override pattern: `ctx.backend` is never mutated; a resolved
`CodingAgentBackend` instance is threaded through the dispatch path as a
function argument.

## Requirements

**REQ-HBR-001:** The `dispatch_food_truck` MCP tool
(`server/tools/tools_fleet_dispatch.py`) must accept an optional
`backend` parameter whose accepted values are the names registered in
`BACKEND_REGISTRY` (`"claude-code"`, `"codex"`).

**REQ-HBR-002:** When the `backend` parameter is omitted, dispatch
behavior must be byte-for-byte identical to the current `ctx.backend`
path — zero behavior change for existing single-backend flows.

**REQ-HBR-003:** The child session command must be built via the
per-dispatch backend's `build_food_truck_cmd`, with backend-correct env
injection (`AUTOSKILLIT_AGENT_BACKEND__BACKEND`,
`AUTOSKILLIT_AGENT_BACKEND`, `AUTOSKILLIT_MCP_CLIENT_BACKEND`) so the
child process resolves the per-dispatch backend at boot.

**REQ-HBR-004:** `ensure_pre_launch()` must be invoked on the
per-dispatch backend before spawn (e.g., Codex MCP registration in
`config.toml`).

**REQ-HBR-005:** The `food_truck_capable` capability gate must be
evaluated against the per-dispatch backend, not the kitchen backend.

**REQ-HBR-006:** Recipe admission control and capability overrides must
be computed against the per-dispatch backend:
`_provider_aware_capability_overrides`, the `backend_supports_git_write`
ingredient, `_compute_effective_backend_map`, and the
`backend-incompatible-skill` validation rule
(`server/tools/_auto_overrides.py`,
`recipe/rules/rules_backend_compat.py`).

**REQ-HBR-007:** Per-dispatch backend selection must compose with
existing provider profiles (e.g., a MiniMax implement-step provider
override via `ANTHROPIC_BASE_URL`) and with the R0 capability-driven
auto-route (#4176) inside the child session — existing provider recipe
overrides must behave identically.

**REQ-HBR-008:** The `DispatchRecord` must persist which backend the
dispatch ran on, so that `reset_dispatch`, diagnostics, and session-log
lookup (Channel B for Claude Code vs `CodexSessionLocator` for Codex)
resolve against the correct backend.

**REQ-HBR-009:** Any `BackendCapabilities` field addition required by
this work must be forward-declared with a tracking entry in
`_FORWARD_DECLARED` in `test_capability_consumption.py`
(frozen-dataclass constraint noted in the forward-obligations doc).

Closes #4179

## Implementation Plan

Plan file:
`.autoskillit/temp/make-plan/dispatch_food_truck_per_dispatch_backend_override_plan_2026-07-04_152400.md`

🤖 Generated with [Claude Code](https://claude.com/claude-code) via
AutoSkillit
<!-- autoskillit:pipeline-signature
steps=prepare_pr,run_arch_lenses,compose_pr,annotate_pr_diff,review_pr
-->

## Token Usage Summary

| Step | Model | count | uncached | output | cache_read | peak_ctx |
turns | cache_write | time |

|------|-------|-------|----------|--------|------------|----------|-------|-------------|------|
| plan* | opus[1m] | 1 | 2.2k | 44.9k | 4.1M | 168.5k | 87 | 149.6k |
22m 37s |
| review_approach* | opus | 1 | 9.2k | 5.3k | 199.4k | 49.0k | 12 |
61.8k | 6m 1s |
| **Total** | | | 11.4k | 50.2k | 4.3M | 168.5k | | 211.4k | 28m 39s |

\* *Step used a non-Anthropic provider; caching behavior may differ.*

## Model Usage Breakdown

| Model | steps | uncached | output | cache_read | cache_write | time |
|-------|-------|----------|--------|------------|-------------|------|
| opus[1m] | 1 | 2.2k | 44.9k | 4.1M | 149.6k | 22m 37s |
| opus | 1 | 9.2k | 5.3k | 199.4k | 61.8k | 6m 1s |

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Trecek

src/autoskillit/core/types/_type_enums.py

@@ -450,6 +450,7 @@ class FleetErrorCode(StrEnum):
     FLEET_RESET_NOT_FOUND = "fleet_reset_not_found"
     FLEET_RESET_INVALID_TARGET = "fleet_reset_invalid_target"
     FLEET_RESET_STILL_RUNNING = "fleet_reset_still_running"
+    FLEET_INVALID_BACKEND = "fleet_invalid_backend"
 
 
 class FeatureLifecycle(StrEnum):

src/autoskillit/execution/headless/__init__.py

@@ -382,11 +382,28 @@ async def dispatch_food_truck(
         backend_override: str | None = None,
         on_session_id_resolved: Callable[[str], None] | None = None,
     ) -> SkillResult:
-        if self._ctx.backend is not None and not self._ctx.backend.capabilities.food_truck_capable:
+        dispatch_backend: CodingAgentBackend | None
+        if backend_override is not None:
+            from autoskillit.execution.backends import get_backend
+
+            dispatch_backend = get_backend(backend_override)
+        else:
+            dispatch_backend = self._ctx.backend
+
+        if dispatch_backend is not None and not dispatch_backend.capabilities.food_truck_capable:
             raise RuntimeError(
                 f"backend does not support food truck dispatch "
-                f"(food_truck_capable=False); got {self._ctx.backend.name!r}"
+                f"(food_truck_capable=False); got {dispatch_backend.name!r}"
             )
+        if backend_override is not None:
+            assert dispatch_backend is not None
+            pre_launch_errors = dispatch_backend.ensure_pre_launch()
+            if pre_launch_errors:
+                raise RuntimeError(
+                    f"Pre-launch check failed for dispatch backend "
+                    f"{dispatch_backend.name!r}: {'; '.join(pre_launch_errors)}"
+                )
+
         cfg = self._ctx.config
         model_identity = resolve_model_identity(
             model, cfg, step_name=step_name, profile_name=profile_name
@@ -412,8 +429,11 @@ async def dispatch_food_truck(
             if idle_cfg_val > 0:
                 merged_extras.setdefault("AUTOSKILLIT_IDLE_OUTPUT_TIMEOUT", str(idle_cfg_val))
 
-        assert self._ctx.backend is not None, "ctx.backend must be set before dispatch_food_truck"
-        backend = self._ctx.backend
+        if dispatch_backend is None:
+            raise RuntimeError(
+                "dispatch_backend must be resolved before dispatch_food_truck execution"
+            )
+        backend = dispatch_backend
         cmd_spec = backend.build_food_truck_cmd(
             orchestrator_prompt=orchestrator_prompt,
             plugin_source=self._ctx.plugin_source,
@@ -451,7 +471,7 @@ async def dispatch_food_truck(
         )
 
         effective_marker_dir: Path | None = marker_dir or (
-            _resolve_session_log_dir(cwd, cast(CodingAgentBackend, self._ctx.backend))
+            _resolve_session_log_dir(cwd, cast(CodingAgentBackend, dispatch_backend))
             if cwd
             else None
         )
@@ -485,4 +505,5 @@ async def dispatch_food_truck(
             session_id=session_id,
             model_identity=model_identity,
             on_session_id_resolved=on_session_id_resolved,
+            step_backend=dispatch_backend,
         )

src/autoskillit/fleet/_api.py

@@ -35,6 +35,7 @@
 )
 
 if TYPE_CHECKING:
+    from autoskillit.core import CodingAgentBackend
     from autoskillit.fleet.sidecar import IssueSidecarEntry
     from autoskillit.fleet.state import DispatchRecord, DispatchStateHandle
     from autoskillit.pipeline.context import ToolContext
@@ -190,6 +191,7 @@ async def execute_dispatch(
     resume_message: str | None = None,
     caller_instructions: str | None = None,
     provider_capability_overrides: dict[str, str] | None = None,
+    dispatch_backend: CodingAgentBackend | None = None,
 ) -> DispatchResult:
     """Execute a single food truck dispatch.
 
@@ -257,6 +259,7 @@ def _reject(error_code: FleetErrorCode, message: str, **kwargs: Any) -> Dispatch
             resume_message=resume_message,
             caller_instructions=caller_instructions,
             provider_capability_overrides=provider_capability_overrides,
+            dispatch_backend=dispatch_backend,
         )
     except asyncio.CancelledError:
         raise
@@ -318,6 +321,7 @@ async def _run_dispatch(
     resume_message: str | None = None,
     caller_instructions: str | None = None,
     provider_capability_overrides: dict[str, str] | None = None,
+    dispatch_backend: CodingAgentBackend | None = None,
 ) -> DispatchResult:
     """Inner dispatch body — called after lock acquisition."""
     from autoskillit.fleet.state import (
@@ -349,9 +353,11 @@ async def _run_dispatch(
             per_dispatch_state_path=None,
         )
 
+    _effective_backend = dispatch_backend if dispatch_backend is not None else tool_ctx.backend
+
     try:
         _capability_overrides = provider_capability_overrides or _build_capability_overrides(
-            tool_ctx.backend
+            _effective_backend
         )
         _merged_ingredients = {**(ingredients or {}), **_capability_overrides}
         validation_result = tool_ctx.recipes.load_and_validate(
@@ -360,7 +366,7 @@ async def _run_dispatch(
             suppressed=tool_ctx.config.migration.suppressed if tool_ctx.config else None,
             ingredient_overrides=_merged_ingredients,
             temp_dir=tool_ctx.temp_dir,
-            backend_name=tool_ctx.backend.name if tool_ctx.backend else None,
+            backend_name=_effective_backend.name if _effective_backend else None,
         )
     except ProcessStaleError as exc:
         return DispatchResult(
@@ -442,7 +448,7 @@ async def _run_dispatch(
     )
 
     _capability_overrides = provider_capability_overrides or _build_capability_overrides(
-        tool_ctx.backend
+        _effective_backend
     )
     effective_ingredients = apply_config_authoritative_overrides(
         effective_ingredients,
@@ -573,7 +579,7 @@ def _reject_with_state(error_code: FleetErrorCode, message: str) -> DispatchResu
     if quota_result.get("should_sleep"):
         await asyncio.sleep(quota_result.get("sleep_seconds", 0))
 
-    _locator = tool_ctx.backend.session_locator() if tool_ctx.backend is not None else None
+    _locator = _effective_backend.session_locator() if _effective_backend is not None else None
 
     if resume_session_id:
         _primary_jsonl = (
@@ -758,6 +764,9 @@ def _on_session_id(session_id: str) -> None:
                     on_spawn=_on_spawn,
                     sentinel_contract=sentinel_contract,
                     resume_message=resume_message,
+                    backend_override=dispatch_backend.name
+                    if dispatch_backend is not None
+                    else None,
                 )
 
         ended_at = time.time()
@@ -957,6 +966,7 @@ def _on_session_id(session_id: str) -> None:
         labels_cleaned=_labels_cleaned,
         issue_url=_issue_urls_raw,
         branch_name=_branch_name,
+        backend_name=_effective_backend.name if _effective_backend else "",
         resume_checkpoint=_checkpoint_to_dict(dispatch_checkpoint),
     )
 

src/autoskillit/fleet/state_types.py

@@ -26,6 +26,7 @@
         "session_chain",
         "resume_count",
         "issue_url",
+        "backend_name",
     }
 )
 
@@ -130,6 +131,7 @@ class DispatchRecord:
     wait_seconds: float | None = None
     resets_at: str = ""
     resume_count: int = 0
+    backend_name: str = ""
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -164,6 +166,7 @@ def to_dict(self) -> dict[str, Any]:
             "wait_seconds": self.wait_seconds,
             "resets_at": self.resets_at,
             "resume_count": self.resume_count,
+            "backend_name": self.backend_name,
         }
 
     @classmethod
@@ -224,6 +227,7 @@ def from_dict(cls, d: dict[str, Any]) -> DispatchRecord:
             wait_seconds=d.get("wait_seconds"),
             resets_at=d.get("resets_at", ""),
             resume_count=d.get("resume_count", 0),
+            backend_name=d.get("backend_name", ""),
         )
 
     @classmethod

src/autoskillit/server/_misc.py

@@ -20,14 +20,15 @@
     get_logger,
 )
 from autoskillit.execution import (
-    SCENARIO_STEP_NAME_ENV as SCENARIO_STEP_NAME_ENV,
-)
-from autoskillit.execution import (
+    BACKEND_REGISTRY,
     SessionState,
     clear_session_state,
     persist_session_state,
     resolve_remote_repo,
 )
+from autoskillit.execution import (
+    SCENARIO_STEP_NAME_ENV as SCENARIO_STEP_NAME_ENV,
+)
 from autoskillit.execution import (
     _refresh_quota_cache as _refresh_quota_cache,
 )
@@ -69,10 +70,22 @@
 
 if TYPE_CHECKING:
     from autoskillit.config import QuotaGuardConfig
-    from autoskillit.core import SkillResult
+    from autoskillit.core import CodingAgentBackend, SkillResult
 
 logger = get_logger(__name__)
 
+
+def resolve_backend_override(name: str) -> CodingAgentBackend:
+    """Resolve a backend name to a CodingAgentBackend instance.
+
+    Raises ValueError if the name is not in BACKEND_REGISTRY.
+    """
+    if name not in BACKEND_REGISTRY:
+        valid = ", ".join(sorted(BACKEND_REGISTRY))
+        raise ValueError(f"Unknown backend {name!r}. Valid names: {valid}")
+    return get_backend(name)
+
+
 _HOOK_CONFIG_FILENAME: str = _HOOK_CONFIG_PATH_COMPONENTS[-1]
 _HOOK_CONFIG_OVERLAY_FILENAME: str = ".hook_config_overlay.json"
 _HOOK_DIR_COMPONENTS: tuple[str, ...] = _HOOK_CONFIG_PATH_COMPONENTS[:-1]

src/autoskillit/server/tools/tools_fleet_dispatch.py

@@ -20,6 +20,7 @@
 
 from autoskillit.core import (
     CapabilityResolutionDetail,
+    CodingAgentBackend,
     FleetErrorCode,
     SessionCheckpoint,
     detect_autoskillit_mcp_prefix,
@@ -48,7 +49,7 @@
 )
 from autoskillit.server import mcp
 from autoskillit.server._guards import _require_enabled, _require_fleet
-from autoskillit.server._misc import resolve_log_dir
+from autoskillit.server._misc import resolve_backend_override, resolve_log_dir
 from autoskillit.server._notify import track_response_size
 from autoskillit.server.tools._auto_overrides import (
     _compute_effective_backend_map,
@@ -175,6 +176,7 @@ async def dispatch_food_truck(
     skip_when: str | None = None,
     resume_message: str | None = None,
     caller_instructions: str | None = None,
+    backend: str | None = None,
     ctx: Context = CurrentContext(),
 ) -> str:
     """Dispatch a single food truck L2 session for one recipe.
@@ -226,6 +228,16 @@ async def dispatch_food_truck(
         if caller_instructions and len(caller_instructions) > _MAX_CALLER_INSTRUCTIONS_LEN:
             caller_instructions = caller_instructions[:_MAX_CALLER_INSTRUCTIONS_LEN]
 
+        dispatch_backend: CodingAgentBackend | None = None
+        if backend is not None:
+            try:
+                dispatch_backend = resolve_backend_override(backend)
+            except ValueError as exc:
+                return fleet_error(
+                    FleetErrorCode.FLEET_INVALID_BACKEND,
+                    str(exc),
+                )
+
         # Feature guard: config authority check independent of MCP visibility state.
         # Fleet sessions open the gate unconditionally at boot; this catch-all ensures
         # dispatch_food_truck never executes when features.fleet is disabled in config.
@@ -284,6 +296,7 @@ async def dispatch_food_truck(
             SessionCheckpoint.from_dict(resume_checkpoint) if resume_checkpoint else None
         )
         tool_ctx = _get_ctx()
+        _override_backend = dispatch_backend if dispatch_backend is not None else tool_ctx.backend
         caller_session_id = find_caller_session_id(project_dir=tool_ctx.project_dir)
         effective_name = dispatch_name or recipe
 
@@ -337,7 +350,7 @@ async def dispatch_food_truck(
                     else None
                 )
                 _capability_overrides, _cap_detail = _provider_aware_capability_overrides(
-                    tool_ctx.backend,
+                    _override_backend,
                     recipe,
                     tool_ctx.config.providers,
                     _preflight_raw_steps,
@@ -346,7 +359,7 @@ async def dispatch_food_truck(
                 _merged_ingredients = {**(ingredients or {}), **_capability_overrides}
                 _effective_backend_map = _compute_effective_backend_map(
                     _preflight_raw_steps,
-                    tool_ctx.backend.name if tool_ctx.backend else None,
+                    _override_backend.name if _override_backend else None,
                     tool_ctx.config.providers,
                     recipe,
                     skill_resolver=tool_ctx.skill_resolver,
@@ -357,7 +370,7 @@ async def dispatch_food_truck(
                     suppressed=tool_ctx.config.migration.suppressed if tool_ctx.config else None,
                     ingredient_overrides=_merged_ingredients,
                     temp_dir=tool_ctx.temp_dir,
-                    backend_name=tool_ctx.backend.name if tool_ctx.backend else None,
+                    backend_name=_override_backend.name if _override_backend else None,
                     effective_backend_map=_effective_backend_map,
                 )
             except Exception:
@@ -411,20 +424,20 @@ async def dispatch_food_truck(
             except Exception:
                 logger.warning("dispatch_food_truck_preflight_recipe_load_failed", exc_info=True)
 
-        if tool_ctx.backend is not None and _active_recipe_steps is not None:
+        if _override_backend is not None and _active_recipe_steps is not None:
             _preflight_err = _check_dispatch_feasibility(
                 post_prune_step_names=_fleet_load_result.get("post_prune_step_names", []),
                 active_recipe_steps=_active_recipe_steps,
-                backend=tool_ctx.backend,
+                backend=_override_backend,
                 config_providers=tool_ctx.config.providers,
                 recipe_name=recipe,
             )
             if _preflight_err is not None:
                 return _preflight_err
 
         _supports_quota = (
-            tool_ctx.backend is not None
-            and tool_ctx.backend.capabilities.anthropic_provider_capable
+            _override_backend is not None
+            and _override_backend.capabilities.anthropic_provider_capable
         )
         try:
             with anyio.fail_after(tool_ctx.config.run_skill.mcp_tool_timeout_sec):
@@ -437,8 +450,8 @@ async def dispatch_food_truck(
                     timeout_sec=timeout_sec,
                     prompt_builder=_get_food_truck_prompt_builder(
                         has_unguarded_filesystem_access=(
-                            tool_ctx.backend.capabilities.has_unguarded_filesystem_access
-                            if tool_ctx.backend
+                            _override_backend.capabilities.has_unguarded_filesystem_access
+                            if _override_backend
                             else False
                         ),
                     ),
@@ -457,6 +470,7 @@ async def dispatch_food_truck(
                     resume_message=resume_message,
                     caller_instructions=caller_instructions,
                     provider_capability_overrides=_capability_overrides,
+                    dispatch_backend=dispatch_backend,
                 )
         except TimeoutError:
             logger.error(

tests/_test_filter.py

@@ -203,7 +203,7 @@ class ImportContext(enum.StrEnum):
     "_type_helpers": frozenset({"core", "execution", "fleet", "pipeline", "recipe", "server"}),
     "_type_protocols_workspace": frozenset({"core", "pipeline", "recipe", "server", "workspace"}),
     "_type_protocols_backend": frozenset(
-        {"_llm_triage", "cli", "core", "execution", "pipeline", "server", "workspace"}
+        {"_llm_triage", "cli", "core", "execution", "fleet", "pipeline", "server", "workspace"}
     ),
     "_type_inspector": frozenset({"core", "execution"}),
     "_type_invariant_registry": frozenset({"core"}),
@@ -830,6 +830,7 @@ class ImportContext(enum.StrEnum):
             "cli",
             # file-level: fleet tests that import server tool handlers directly
             "fleet/test_api.py",
+            "fleet/test_dispatch_backend_override.py",
             "fleet/test_dispatch_crash_diagnostics.py",
             "fleet/test_fleet_e2e.py",
             "fleet/test_pack_enforcement.py",

tests/arch/test_execution_source_split.py

@@ -16,7 +16,7 @@
     "_headless_execute.py",
 ]
 HEADLESS_SIZE_BUDGETS = {
-    "headless/__init__.py": 490,
+    "headless/__init__.py": 510,
     "headless/_headless_helpers.py": 220,
     "headless/_headless_execute.py": 616,
     "headless/_headless_recovery.py": 370,

tests/execution/test_headless_dispatch.py

@@ -407,7 +407,7 @@ async def test_dispatch_food_truck_raises_for_none_backend(
         minimal_ctx.plugin_source = DirectInstall(plugin_dir=tmp_path)
         executor = DefaultHeadlessExecutor(minimal_ctx)
 
-        with pytest.raises(AssertionError):
+        with pytest.raises(RuntimeError, match="dispatch_backend must be resolved"):
             await executor.dispatch_food_truck(
                 "some prompt",
                 str(tmp_path),