[FIX] Dispatch Feasibility Immunity — Capability-Aware Admission Control

src/autoskillit/core/__init__.pyi

@@ -134,6 +134,7 @@ from .types import AUTOSKILLIT_WRITE_GUARD_TOOL_NAMES as AUTOSKILLIT_WRITE_GUARD
 from .types import BACKEND_CAPABILITY_INGREDIENTS as BACKEND_CAPABILITY_INGREDIENTS
 from .types import CAMPAIGN_ID_ENV_VAR as CAMPAIGN_ID_ENV_VAR
 from .types import CAPABILITY_GATE_CALLABLES as CAPABILITY_GATE_CALLABLES
+from .types import CAPABILITY_INGREDIENT_TO_SKIP_GUARD as CAPABILITY_INGREDIENT_TO_SKIP_GUARD
 from .types import CAPTURE_VALID_VALUE_TYPES as CAPTURE_VALID_VALUE_TYPES
 from .types import CATEGORY_TAGS as CATEGORY_TAGS
 from .types import CLAUDE_CODE_CAPABILITIES as CLAUDE_CODE_CAPABILITIES

src/autoskillit/core/types/_type_constants.py

@@ -34,6 +34,7 @@
     "SCOPE_DIRECTION_SOURCE_TYPES",
     "WORKTREE_SKILLS",
     "BACKEND_CAPABILITY_INGREDIENTS",
+    "CAPABILITY_INGREDIENT_TO_SKIP_GUARD",
     "CAPABILITY_GATE_CALLABLES",
     "SkillFamilyDef",
     "GITHUB_API_SKILL_FAMILIES",
@@ -97,6 +98,13 @@
     }
 )
 
+# Maps each BACKEND_CAPABILITY_INGREDIENTS key to its corresponding
+# skip_when_false ingredient reference string. Used by capability-feasibility
+# to detect vacuous gates — a gate whose guarded steps were all pruned.
+CAPABILITY_INGREDIENT_TO_SKIP_GUARD: dict[str, str] = {
+    "backend_supports_git_write": "inputs.backend_supports_git_write",
+}
+
 # Bare (un-dotted) callable names whose run_python steps are capability gates.
 # Each entry corresponds to a callable in smoke_utils that reads a
 # BACKEND_CAPABILITY_INGREDIENTS key and returns a verdict dict. When a

src/autoskillit/fleet/__init__.py

@@ -4,6 +4,7 @@
 ``autoskillit.fleet``, not from sub-modules.
 """
 
+from ._api import _build_capability_overrides as _build_capability_overrides
 from ._api import _write_pid as _write_pid
 from ._api import execute_dispatch
 from ._capture import CaptureCompletenessError
@@ -101,6 +102,7 @@
 )
 
 __all__ = [
+    "_build_capability_overrides",
     "_write_pid",
     "cleanup_orphaned_labels",
     "discover_campaign_state_files",

src/autoskillit/fleet/_api.py

@@ -347,10 +347,13 @@ async def _run_dispatch(
         )
 
     try:
+        _capability_overrides = _build_capability_overrides(tool_ctx.backend)
+        _merged_ingredients = {**(ingredients or {}), **_capability_overrides}
         validation_result = tool_ctx.recipes.load_and_validate(
             recipe,
             tool_ctx.project_dir,
             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,
         )
@@ -388,6 +391,19 @@ async def _run_dispatch(
             per_dispatch_state_path=None,
         )
 
+    if not validation_result.get("dispatch_feasible", True):
+        infeasible_steps = validation_result.get("infeasible_steps", [])
+        return DispatchResult(
+            DispatchRejected(
+                error_code=FleetErrorCode.FLEET_RECIPE_INVALID,
+                message=(
+                    f"Recipe '{recipe}' is dispatch-infeasible: "
+                    f"infeasible_steps={infeasible_steps}"
+                ),
+            ),
+            per_dispatch_state_path=None,
+        )
+
     try:
         full_recipe = tool_ctx.recipes.load(recipe_obj.path)
     except Exception as exc:

src/autoskillit/recipe/_api.py

@@ -407,6 +407,8 @@ def load_and_validate(
                 ingredient_overrides or {},
                 capability_ingredient_keys=BACKEND_CAPABILITY_INGREDIENTS,
                 capability_gate_callables=CAPABILITY_GATE_CALLABLES,
+                skip_resolutions=_skip_resolutions,
+                pre_prune_steps=_pre_prune_steps,
             )
             t0 = _t("capability_feasibility", t0, name)
 

src/autoskillit/recipe/_recipe_composition.py

@@ -9,7 +9,7 @@
 
 import regex as re
 
-from autoskillit.core import YAMLError, load_yaml
+from autoskillit.core import CAPABILITY_INGREDIENT_TO_SKIP_GUARD, YAMLError, load_yaml
 from autoskillit.recipe._contracts_types import INPUT_REF_RE
 from autoskillit.recipe.io import find_sub_recipe_by_name
 from autoskillit.recipe.io import load_recipe as _load_recipe_from_path
@@ -112,6 +112,8 @@ def _compute_capability_feasibility(
     *,
     capability_ingredient_keys: frozenset[str],
     capability_gate_callables: frozenset[str],
+    skip_resolutions: dict[str, bool | None] | None = None,
+    pre_prune_steps: dict[str, Any] | None = None,
 ) -> tuple[bool, list[str]]:
     """Detect dead-on-arrival pipelines caused by capability-gated run_python steps.
 
@@ -124,6 +126,10 @@ def _compute_capability_feasibility(
     3. The step's on_result / on_failure routes the falsy case to a terminal
        failure target ("escalate") rather than recovery.
 
+    A gate is vacuous (and removed from infeasible) when all steps gated by
+    the same capability ingredient were pruned by skip_when_false and no
+    surviving step references that capability.
+
     Returns (is_feasible, infeasible_step_names).
     """
     if not post_prune_recipe or not getattr(post_prune_recipe, "steps", None):
@@ -154,12 +160,59 @@ def _compute_capability_feasibility(
             continue
         on_failure = getattr(step, "on_failure", None) or "escalate"
         on_exhausted = getattr(step, "on_exhausted", None) or "escalate"
-        if on_failure == "escalate" or on_exhausted == "escalate":
-            infeasible.append(step_name)
+        if on_failure != "escalate" and on_exhausted != "escalate":
+            continue
+        if _is_vacuous_gate(
+            gate_input_keys,
+            gate_step_name=step_name,
+            skip_resolutions=skip_resolutions,
+            pre_prune_steps=pre_prune_steps,
+            post_prune_steps=steps,
+        ):
+            continue
+        infeasible.append(step_name)
 
     return (not infeasible), infeasible
 
 
+def _is_vacuous_gate(
+    gate_input_keys: set[str],
+    *,
+    gate_step_name: str,
+    skip_resolutions: dict[str, bool | None] | None,
+    pre_prune_steps: dict[str, Any] | None,
+    post_prune_steps: dict[str, Any],
+) -> bool:
+    """Return True if the gate is vacuous — its guarded operations were all pruned."""
+    if skip_resolutions is None or pre_prune_steps is None:
+        return False
+    for cap_key in gate_input_keys:
+        guard_ref = CAPABILITY_INGREDIENT_TO_SKIP_GUARD.get(cap_key)
+        if guard_ref is None:
+            return False
+        pruned_for_capability = False
+        for pre_step_name, pre_step in pre_prune_steps.items():
+            skip_guard = getattr(pre_step, "skip_when_false", None)
+            if skip_guard != guard_ref:
+                continue
+            if skip_resolutions.get(pre_step_name) is False:
+                pruned_for_capability = True
+                break
+        if not pruned_for_capability:
+            return False
+        live_uses_capability = False
+        for sn, s in post_prune_steps.items():
+            if s is None or sn == gate_step_name:
+                continue
+            with_args = getattr(s, "with_args", None) or {}
+            if cap_key in with_args:
+                live_uses_capability = True
+                break
+        if live_uses_capability:
+            return False
+    return True
+
+
 def _drop_sub_recipe_step(recipe: Any, step_name: str) -> Any:
     """Return a new Recipe with the named sub_recipe placeholder step removed."""
     new_steps = {k: v for k, v in recipe.steps.items() if k != step_name}

src/autoskillit/server/tools/tools_fleet_dispatch.py

@@ -32,6 +32,7 @@
     DispatchRejected,
     DispatchResult,
     DispatchStatus,
+    _build_capability_overrides,
     _build_food_truck_prompt,
     evaluate_skip_when,
     execute_dispatch,
@@ -321,17 +322,32 @@ async def dispatch_food_truck(
         _fleet_load_result: dict[str, Any] = {}
         if tool_ctx.recipes is not None:
             try:
+                _capability_overrides = _build_capability_overrides(tool_ctx.backend)
+                _merged_ingredients = {**(ingredients or {}), **_capability_overrides}
                 _fleet_load_result = tool_ctx.recipes.load_and_validate(
                     recipe,
                     tool_ctx.project_dir,
                     suppressed=tool_ctx.config.migration.suppressed if tool_ctx.config else None,
-                    ingredient_overrides=ingredients,
+                    ingredient_overrides=_merged_ingredients,
                     temp_dir=tool_ctx.temp_dir,
                     backend_name=tool_ctx.backend.name if tool_ctx.backend else None,
                 )
             except Exception:
                 logger.warning("dispatch_food_truck_preflight_load_failed", exc_info=True)
 
+        if not _fleet_load_result.get("dispatch_feasible", True):
+            _infeasible_steps = _fleet_load_result.get("infeasible_steps", [])
+            _infeasible_msg = (
+                f"Recipe '{recipe}' is dispatch-infeasible: capability gate(s) "
+                f"blocked at preflight. Infeasible steps: {_infeasible_steps}"
+            )
+            logger.warning(
+                "dispatch_food_truck_capability_infeasible",
+                recipe=recipe,
+                infeasible_steps=_infeasible_steps,
+            )
+            return fleet_error(FleetErrorCode.FLEET_RECIPE_INVALID, _infeasible_msg)
+
         _active_recipe_steps: dict[str, Any] | None = None
         if _fleet_load_result and tool_ctx.recipes is not None:
             try:

tests/arch/test_capability_admission_control.py

@@ -1,10 +1,12 @@
 """Architectural structural tests for capability admission control.
 
 Verifies that:
-- _compute_capability_feasibility is called inside load_and_validate
+- _compute_capability_feasibility is called inside load_and_validate with skip_resolutions
 - All four content-serving surfaces gate on dispatch_feasible
+  (open_kitchen, get_recipe, load_recipe, dispatch_food_truck)
 - CAPABILITY_GATE_CALLABLES entries have a corresponding BACKEND_CAPABILITY_INGREDIENTS input
 - Every run_python step using a CAPABILITY_GATE_CALLABLES callable reads a capability ingredient
+- dispatch_food_truck injects _build_capability_overrides for backend capability signals
 """
 
 from __future__ import annotations
@@ -104,6 +106,73 @@ def test_load_recipe_gates_on_dispatch_feasible() -> None:
     pytest.fail("load_recipe function not found in tools_recipe.py")
 
 
+def test_dispatch_food_truck_gates_on_dispatch_feasible() -> None:
+    """dispatch_food_truck must check dispatch_feasible before executing."""
+    fleet_path = SRC_ROOT / "server" / "tools" / "tools_fleet_dispatch.py"
+    src = fleet_path.read_text(encoding="utf-8")
+    tree = ast.parse(src)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.AsyncFunctionDef) and node.name == "dispatch_food_truck":
+            found = False
+            for child in ast.walk(node):
+                if isinstance(child, ast.Constant) and child.value == "dispatch_feasible":
+                    found = True
+                    break
+            assert found, (
+                "dispatch_food_truck must reference 'dispatch_feasible' to gate "
+                "capability-DOA fleet dispatches before subprocess launch."
+            )
+            return
+    pytest.fail("dispatch_food_truck function not found in tools_fleet_dispatch.py")
+
+
+def test_compute_capability_feasibility_receives_skip_resolutions() -> None:
+    """load_and_validate must pass skip_resolutions to _compute_capability_feasibility
+    so the vacuous-gate detection has access to pruning context."""
+    api_path = SRC_ROOT / "recipe" / "_api.py"
+    tree = ast.parse(api_path.read_text(encoding="utf-8"))
+    for node in ast.walk(tree):
+        if (
+            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+            and node.name == "load_and_validate"
+        ):
+            for child in ast.walk(node):
+                if (
+                    isinstance(child, ast.Call)
+                    and isinstance(child.func, ast.Name)
+                    and child.func.id == "_compute_capability_feasibility"
+                ):
+                    kw_names = {kw.arg for kw in child.keywords if kw.arg is not None}
+                    assert "skip_resolutions" in kw_names, (
+                        "load_and_validate must pass skip_resolutions kwarg "
+                        "to _compute_capability_feasibility for vacuous-gate detection."
+                    )
+                    return
+            pytest.fail("_compute_capability_feasibility call not found in load_and_validate")
+    pytest.fail("load_and_validate function not found in _api.py")
+
+
+def test_dispatch_food_truck_injects_capability_overrides() -> None:
+    """dispatch_food_truck must reference _build_capability_overrides to inject
+    backend capability signals into the load_and_validate call."""
+    fleet_path = SRC_ROOT / "server" / "tools" / "tools_fleet_dispatch.py"
+    src = fleet_path.read_text(encoding="utf-8")
+    tree = ast.parse(src)
+    for node in ast.walk(tree):
+        if isinstance(node, ast.AsyncFunctionDef) and node.name == "dispatch_food_truck":
+            found = False
+            for child in ast.walk(node):
+                if isinstance(child, ast.Name) and child.id == "_build_capability_overrides":
+                    found = True
+                    break
+            assert found, (
+                "dispatch_food_truck must reference _build_capability_overrides "
+                "to inject backend capability signals into load_and_validate."
+            )
+            return
+    pytest.fail("dispatch_food_truck function not found in tools_fleet_dispatch.py")
+
+
 def test_capability_gate_callables_have_matching_ingredient() -> None:
     """Every CAPABILITY_GATE_CALLABLES callable must map to a BACKEND_CAPABILITY_INGREDIENTS input.
 
@@ -129,3 +198,21 @@ def test_capability_gate_callables_have_matching_ingredient() -> None:
     assert len(CAPABILITY_GATE_CALLABLES) > 0, (
         "CAPABILITY_GATE_CALLABLES must declare at least one capability gate callable."
     )
+
+
+def test_capability_ingredient_to_skip_guard_keys_subset() -> None:
+    """CAPABILITY_INGREDIENT_TO_SKIP_GUARD keys must be BACKEND_CAPABILITY_INGREDIENTS subset."""
+    from autoskillit.core import (
+        BACKEND_CAPABILITY_INGREDIENTS,
+        CAPABILITY_INGREDIENT_TO_SKIP_GUARD,
+    )
+
+    orphaned = set(CAPABILITY_INGREDIENT_TO_SKIP_GUARD) - set(BACKEND_CAPABILITY_INGREDIENTS)
+    assert not orphaned, (
+        f"CAPABILITY_INGREDIENT_TO_SKIP_GUARD keys {orphaned} are not in "
+        f"BACKEND_CAPABILITY_INGREDIENTS — vacuous-gate detection will silently "
+        f"malfunction for these keys."
+    )
+    assert len(CAPABILITY_INGREDIENT_TO_SKIP_GUARD) > 0, (
+        "CAPABILITY_INGREDIENT_TO_SKIP_GUARD must declare at least one mapping."
+    )

tests/arch/test_subpackage_structure.py

@@ -60,8 +60,8 @@ def test_type_constants_split_completeness(self):
         assert len(combined) == len(remaining) + len(env) + len(features) + len(registries), (
             "Duplicate symbols across split modules"
         )
-        assert len(combined) == 93, (
-            f"Expected 93 symbols total, got {len(combined)} "
+        assert len(combined) == 94, (
+            f"Expected 94 symbols total, got {len(combined)} "
             f"(remaining={len(remaining)}, env={len(env)}, "
             f"features={len(features)}, registries={len(registries)})"
         )

tests/fleet/test_dispatch_ingredient_validation.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from unittest.mock import AsyncMock, MagicMock
+
 import pytest
 
 from tests.fleet._helpers import (
@@ -548,3 +550,46 @@ def test_capability_overrides_dict_covers_registry_keys():
     assert not missing, (
         f"Capability registry keys missing from dispatch capability_overrides: {missing}"
     )
+
+
+@pytest.mark.anyio
+async def test_run_dispatch_rejects_dispatch_infeasible(tool_ctx):
+    """When _run_dispatch's inner load_and_validate returns dispatch_feasible=False,
+    it must return DispatchRejected without proceeding to subprocess launch."""
+    from unittest.mock import patch
+
+    tool_ctx.recipes = MagicMock()
+    tool_ctx.recipes.load_and_validate.return_value = {
+        "valid": True,
+        "dispatch_feasible": False,
+        "infeasible_steps": ["gate_backend_write"],
+    }
+    tool_ctx.recipes.find.return_value = MagicMock()
+    tool_ctx.recipes.load.return_value = MagicMock()
+
+    with patch(
+        "autoskillit.fleet._api.execute_dispatch",
+        new_callable=AsyncMock,
+    ) as mock_exec:
+        from autoskillit.fleet._api import _run_dispatch
+
+        result = await _run_dispatch(
+            tool_ctx=tool_ctx,
+            recipe="test-recipe",
+            task="test task",
+            ingredients={},
+            dispatch_name=None,
+            timeout_sec=None,
+            prompt_builder=lambda **kw: "prompt",
+            quota_checker=_no_sleep_quota_checker,
+            quota_refresher=_noop_quota_refresher,
+        )
+
+    mock_exec.assert_not_called()
+    from autoskillit.fleet.state_types import DispatchRejected
+
+    assert isinstance(result.outcome, DispatchRejected)
+    assert (
+        "gate_backend_write" in result.outcome.message
+        or "dispatch" in result.outcome.message.lower()
+    )