[FIX] Dispatch-Feasibility Preflight — Claim-Before-Validate Immunity

src/autoskillit/hooks/formatters/_fmt_recipe.py

@@ -43,6 +43,7 @@
         "recipe_version",  # internal identity metadata
         "stop_step_semantics",  # delivered via open_kitchen response Channel B; not redisplayed
         "deferred_guards",  # internal deferral metadata; not displayed to agent
+        "post_prune_step_names",  # internal preflight field; not displayed to agent
     }
 )
 
@@ -183,6 +184,7 @@ def _fmt_load_recipe(data: LoadRecipeResult, pipeline: bool) -> str:
         "stop_step_semantics",  # delivered via open_kitchen Channel B
         "hook_warning",  # edge-case diagnostic; not rendered in standard path
         "deferred_guards",  # internal deferral metadata; not displayed to agent
+        "post_prune_step_names",  # internal preflight field; not displayed to agent
     }
 )
 

src/autoskillit/recipe/_api.py

@@ -580,6 +580,8 @@ def load_and_validate(
                 )
     if _deferred_guard_list:
         result["deferred_guards"] = _deferred_guard_list
+    if active_recipe is not None:
+        result["post_prune_step_names"] = list(active_recipe.steps.keys())
 
     # Write to cache (only when recipe was found and fully processed)
     if match is not None:

src/autoskillit/recipe/_recipe_ingredients.py

@@ -126,6 +126,7 @@ class LoadRecipeResult(TypedDict, total=False):
     composite_hash: str
     recipe_version: str | None
     deferred_guards: list[DeferredGuard]
+    post_prune_step_names: list[str]
 
 
 class OpenKitchenResult(TypedDict, total=False):
@@ -134,7 +135,7 @@ class OpenKitchenResult(TypedDict, total=False):
     Extends LoadRecipeResult with three post-return keys injected by the handler.
     """
 
-    # Inherited from LoadRecipeResult (15 keys)
+    # Inherited from LoadRecipeResult (17 keys)
     content: str
     errors: list[str]
     diagram: str | None
@@ -151,6 +152,7 @@ class OpenKitchenResult(TypedDict, total=False):
     composite_hash: str
     recipe_version: str | None
     deferred_guards: list[DeferredGuard]
+    post_prune_step_names: list[str]
     # Post-return keys injected by open_kitchen handler (4 keys)
     success: bool
     kitchen: str

src/autoskillit/server/tools/AGENTS.md

@@ -19,6 +19,7 @@ MCP `@mcp.tool()` handlers registered on import (20 tool modules).
 | `tools_clone.py` | `clone_repo`, `remove_clone`, `push_to_remote`, `register_clone_status`, `batch_cleanup_clones`, `bootstrap_clone` |
 | `_claim_helpers.py` | `ClaimDecision`, `_try_claim_with_liveness`, `_get_campaign_state_paths` — shared claiming logic for `claim_issue` and `claim_and_resolve_issue` |
 | `_execution_helpers.py` | `_import_and_call`, `_coerce_scalar`, `maybe_promote_work_dir`, `validate_path_arg_anchoring`, `resolve_relative_path_args` — execution helpers for `run_python` (no MCP tools) |
+| `_preflight.py` | `_check_dispatch_feasibility`, `_get_fix_required_hook_matchers`, `filter_steps_by_post_prune` — shared preflight for `open_kitchen` and `dispatch_food_truck` |
 | `tools_execution.py` | `run_cmd`, `run_python`, `run_skill` |
 | `tools_fleet_dispatch.py` | `dispatch_food_truck`, `record_gate_dispatch` |
 | `tools_fleet_reset.py` | `reset_dispatch` (full dispatch artifact cleanup) |

src/autoskillit/server/tools/_preflight.py

@@ -0,0 +1,114 @@
+"""Dispatch-feasibility preflight — shared by open_kitchen and dispatch_food_truck."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import Any
+
+from autoskillit.hook_registry import HOOK_REGISTRY
+
+
+def _get_fix_required_hook_matchers(applicable_guards: frozenset[str]) -> list[str]:
+    """Return matchers of fix-required hooks not enforced by the given guard set."""
+    return [
+        h.matcher
+        for h in HOOK_REGISTRY
+        if h.codex_status == "fix-required"
+        and (
+            not h.scripts
+            or not frozenset(Path(s).stem for s in h.scripts).issubset(applicable_guards)
+        )
+    ]
+
+
+def filter_steps_by_post_prune(
+    raw_steps: dict[str, Any],
+    post_prune_step_names: list[str],
+) -> dict[str, Any]:
+    """Return only the steps whose names survived skip_when_false pruning."""
+    keep = set(post_prune_step_names)
+    return {k: v for k, v in raw_steps.items() if k in keep}
+
+
+def _check_dispatch_feasibility(
+    post_prune_step_names: list[str],
+    active_recipe_steps: dict[str, Any],
+    backend: Any | None,
+    config_providers: Any,
+    recipe_name: str = "",
+) -> str | None:
+    """Fail-closed dispatch-feasibility preflight.
+
+    Evaluated at open_kitchen time (and dispatch_food_truck time) to detect
+    recipe/backend combinations where HOOK_REGISTRY has fix-required entries
+    that the current backend cannot enforce. Returns a JSON error envelope
+    string on failure, None on pass.
+    """
+    if backend is None:
+        return None
+    if not post_prune_step_names:
+        return None
+
+    run_skill_step_names: list[str] = []
+    for step_name in post_prune_step_names:
+        step = active_recipe_steps.get(step_name)
+        if step is not None and getattr(step, "tool", None) == "run_skill":
+            run_skill_step_names.append(step_name)
+
+    if not run_skill_step_names:
+        return None
+
+    from autoskillit.server._guards import _resolve_provider_profile  # circular-break
+
+    feasible_step_names: list[str] = []
+    for step_name in run_skill_step_names:
+        step = active_recipe_steps.get(step_name)
+        step_provider = getattr(step, "provider", "") or ""
+        _profile_name, provider_extras = _resolve_provider_profile(
+            step_name,
+            recipe_name,
+            config_providers,
+            step_provider=step_provider,
+        )
+        if (
+            provider_extras
+            and "ANTHROPIC_BASE_URL" in provider_extras
+            and not backend.capabilities.anthropic_provider_capable
+        ):
+            continue
+        feasible_step_names.append(step_name)
+
+    if not feasible_step_names:
+        return None
+
+    fix_required_matchers = _get_fix_required_hook_matchers(
+        backend.capabilities.applicable_guards,
+    )
+    if not fix_required_matchers:
+        return None
+
+    return json.dumps(
+        {
+            "success": False,
+            "kitchen": "preflight_failed",
+            "user_visible_message": (
+                f"Cannot dispatch recipe: backend {backend.name!r} cannot enforce "
+                f"HOOK_REGISTRY fix-required entries "
+                f"[{', '.join(fix_required_matchers)}]. "
+                f"Add a per-step provider override that sets ANTHROPIC_BASE_URL "
+                f"to reroute dispatch to claude-code, which has full hook enforcement."
+            ),
+            "error": (
+                f"Dispatch infeasible on backend {backend.name!r}: "
+                f"fix-required hooks {fix_required_matchers} are not enforceable."
+            ),
+            "stage": "dispatch_feasibility_preflight",
+            "unfixable_matchers": fix_required_matchers,
+            "backend": backend.name,
+            "escape_hatch": (
+                "Per-step provider: override with ANTHROPIC_BASE_URL set to "
+                "reroute dispatch to claude-code (which enforces all fix-required hooks)."
+            ),
+        }
+    )

src/autoskillit/server/tools/tools_execution.py

@@ -37,7 +37,6 @@
 from autoskillit.core import current_order_id as _current_order_id
 from autoskillit.core import current_step_name as _current_step_name
 from autoskillit.core import resolve_skill_temp_dir as _resolve_skill_temp_dir
-from autoskillit.hook_registry import HOOK_REGISTRY
 from autoskillit.pipeline import canonical_step_name as _canonical_step_name
 from autoskillit.server import mcp
 from autoskillit.server._guards import (
@@ -65,6 +64,7 @@
     resolve_relative_path_args,
     validate_path_arg_anchoring,
 )
+from autoskillit.server.tools._preflight import _get_fix_required_hook_matchers
 
 logger = get_logger(__name__)
 
@@ -87,19 +87,6 @@ def _is_backend_incompatible(skill_info: object, effective_backend: str) -> bool
     return bool(reqs and effective_backend not in reqs)
 
 
-def _get_fix_required_hook_matchers(applicable_guards: frozenset[str]) -> list[str]:
-    """Return matchers of fix-required hooks not enforced by the given guard set."""
-    return [
-        h.matcher
-        for h in HOOK_REGISTRY
-        if h.codex_status == "fix-required"
-        and (
-            not h.scripts
-            or not frozenset(Path(s).stem for s in h.scripts).issubset(applicable_guards)
-        )
-    ]
-
-
 def _check_backend_compat(
     skill_command: str,
     resolved_command: str,

src/autoskillit/server/tools/tools_fleet_dispatch.py

@@ -48,6 +48,10 @@
 from autoskillit.server._misc import resolve_log_dir
 from autoskillit.server._notify import track_response_size
 from autoskillit.server.tools._cancellation_shield import _cancellation_shield
+from autoskillit.server.tools._preflight import (
+    _check_dispatch_feasibility,
+    filter_steps_by_post_prune,
+)
 
 logger = get_logger(__name__)
 
@@ -312,6 +316,47 @@ async def dispatch_food_truck(
                     "Dispatch skipped: skip_when condition evaluated to true",
                 )
 
+        # Dispatch-feasibility preflight: verify the backend can enforce
+        # all fix-required hooks for the recipe's run_skill steps before
+        # spawning a subprocess.
+        _fleet_load_result: dict[str, Any] = {}
+        if tool_ctx.recipes is not None:
+            try:
+                _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,
+                    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)
+
+        _active_recipe_steps: dict[str, Any] | None = None
+        if _fleet_load_result and tool_ctx.recipes is not None:
+            try:
+                _recipe_info = tool_ctx.recipes.find(recipe, tool_ctx.project_dir)
+                if _recipe_info is not None:
+                    _recipe_obj = tool_ctx.recipes.load(_recipe_info.path)
+                    _active_recipe_steps = filter_steps_by_post_prune(
+                        _recipe_obj.steps,
+                        _fleet_load_result.get("post_prune_step_names", []),
+                    )
+            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:
+            _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,
+                config_providers=tool_ctx.config.providers,
+                recipe_name=recipe,
+            )
+            if _preflight_err is not None:
+                return _preflight_err
+
         result = await execute_dispatch(
             tool_ctx=tool_ctx,
             recipe=recipe,

src/autoskillit/server/tools/tools_kitchen.py

@@ -61,6 +61,10 @@
     _promote_capability_keys,
 )
 from autoskillit.server.tools._cancellation_shield import _cancellation_shield
+from autoskillit.server.tools._preflight import (
+    _check_dispatch_feasibility,
+    filter_steps_by_post_prune,
+)
 
 logger = get_logger(__name__)
 
@@ -625,7 +629,9 @@ async def open_kitchen(
                         try:
                             recipe_obj = tool_ctx.recipes.load(recipe_info.path)
                             _deferred_recipe_obj = recipe_obj
-                            tool_ctx.active_recipe_steps = dict(recipe_obj.steps)
+                            tool_ctx.active_recipe_steps = filter_steps_by_post_prune(
+                                recipe_obj.steps, result.get("post_prune_step_names", [])
+                            )
                             tool_ctx.active_recipe_ingredients = frozenset(
                                 recipe_obj.ingredients.keys()
                             )
@@ -638,7 +644,22 @@ async def open_kitchen(
                         tool_ctx.active_recipe_ingredients = None
                 # Default to False for missing 'valid' so a absent key is treated as invalid
                 if not result.get("valid", False) or not result.get("content", ""):
+                    tool_ctx.gate.disable()
                     return _recipe_validation_error_response(name, result)
+                # Dispatch-feasibility preflight: verify the backend can enforce
+                # all fix-required hooks for the recipe's run_skill steps.
+                if tool_ctx.active_recipe_steps is not None:
+                    _preflight_err = _check_dispatch_feasibility(
+                        post_prune_step_names=result.get("post_prune_step_names", []),
+                        active_recipe_steps=tool_ctx.active_recipe_steps,
+                        backend=tool_ctx.backend,
+                        config_providers=tool_ctx.config.providers,
+                        recipe_name=name,
+                    )
+                    if _preflight_err is not None:
+                        tool_ctx.gate.disable()
+                        await ctx.disable_components(tags={"kitchen"})
+                        return _preflight_err
                 result["success"] = True
                 result["kitchen"] = "open"
                 result["version"] = __version__
@@ -708,7 +729,9 @@ async def open_kitchen(
                 try:
                     recipe_obj = tool_ctx.recipes.load(recipe_info.path)
                     _normal_recipe_obj = recipe_obj
-                    tool_ctx.active_recipe_steps = dict(recipe_obj.steps)
+                    tool_ctx.active_recipe_steps = filter_steps_by_post_prune(
+                        recipe_obj.steps, result.get("post_prune_step_names", [])
+                    )
                     tool_ctx.active_recipe_ingredients = frozenset(recipe_obj.ingredients.keys())
                 except Exception:
                     logger.warning("open_kitchen_recipe_steps_cache_failed", exc_info=True)
@@ -725,8 +748,24 @@ async def open_kitchen(
                 return _kitchen_failure_envelope(exc, stage="apply_triage_gate")
 
             if not result.get("valid", False) or not result.get("content", ""):
+                tool_ctx.gate.disable()
                 return _recipe_validation_error_response(name, result)
 
+            # Dispatch-feasibility preflight: verify the backend can enforce
+            # all fix-required hooks for the recipe's run_skill steps.
+            if tool_ctx.active_recipe_steps is not None:
+                _preflight_err = _check_dispatch_feasibility(
+                    post_prune_step_names=result.get("post_prune_step_names", []),
+                    active_recipe_steps=tool_ctx.active_recipe_steps,
+                    backend=tool_ctx.backend,
+                    config_providers=tool_ctx.config.providers,
+                    recipe_name=name,
+                )
+                if _preflight_err is not None:
+                    tool_ctx.gate.disable()
+                    await ctx.disable_components(tags={"kitchen"})
+                    return _preflight_err
+
             result["success"] = True
             result["kitchen"] = "open"
             result["version"] = __version__

tests/_test_filter.py

@@ -858,8 +858,10 @@ class ImportContext(enum.StrEnum):
             "hooks",
             "cli",
             "execution",
-            # server/ narrowed to 3 files
+            # server/ narrowed to 4 files
             "server/test_tools_kitchen_envelope.py",
+            "server/test_tools_kitchen_preflight.py",
+            "server/test_tools_fleet_dispatch_preflight.py",
             "server/test_lifespan.py",
             "server/test_run_skill_backend_compat.py",
             # infra/ narrowed to 7 files

tests/arch/test_subpackage_isolation.py

@@ -876,7 +876,7 @@ def test_no_subpackage_exceeds_10_files() -> None:
         "pipeline": 12,
         "fleet": 22,  # REQ-CNST-003-E9: _dispatch_reaper.py; +_sidecar_synthesis.py; +_reset.py
         "recipe/rules": 51,  # +commit_guard_regression_route +rules_model +rules_gitignored_deliverable  # noqa: E501
-        "server/tools": 26,  # _auto_overrides.py + _cancellation_shield.py + tools_fleet_reset.py
+        "server/tools": 27,  # +_preflight.py (dispatch-feasibility preflight)
         "hooks/guards": 32,  # +fleet_claim_guard, +reset_resume_gate, +recipe_read_guard
     }
     violations: list[str] = []
@@ -962,13 +962,14 @@ def test_data_directories_are_not_python_packages() -> None:
         "co-located with the execution engine that calls them",
     ),
     "tools_kitchen.py": (
-        1160,
+        1200,
         "REQ-CNST-010-E7: kitchen tool handlers — open_kitchen and lock_ingredients require "
         "inline validation helpers (_check_override_keys, _build_ingredient_key_suggestions) "
         "for ingredient key validation; splitting would cross import-layer boundaries; "
         "backend capability promotion delegated to _promote_capability_keys in _auto_overrides; "
         "fail-closed validity gate on both deferred-recall and normal paths adds structural "
-        "error propagation from LoadRecipeResult; get_recipe validity guard adds 9 lines",
+        "error propagation from LoadRecipeResult; get_recipe validity guard adds 9 lines; "
+        "dispatch-feasibility preflight wiring on both paths with gate-close on failure",
     ),
     "tools_execution.py": (
         1130,