Implementation Plan: T5-P3-A5-WP1 B6 Cross-Layer Predicate Spanning Test (#4084)

## Summary

Create `tests/execution/backends/test_failure_predicate_spanning.py`
with three test classes that validate the FAILURE PREDICATES blocks
across both prompt builders (`_build_orchestrator_prompt` and
`_build_food_truck_prompt`) against the server tool TypedDicts in
`server/tools/_types.py`. No production source files are modified. Test
infrastructure registries (`.autoskillit/test-source-map.json` and
`tests/_test_filter.py` cascade entries) are updated to route source
changes to the new test.

## Implementation Plan

Plan file:
`/home/talon/projects/autoskillit-runs/impl-20260611-205810-907005/.autoskillit/temp/make-plan/t5_p3_a5_wp1_failure_predicate_spanning_plan_2026-06-11_210500.md`

Closes #4013

🤖 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 | 1.7k | 15.0k | 2.0M | 94.9k | 63 | 73.7k | 12m
10s |
| verify* | sonnet | 1 | 68 | 17.2k | 357.0k | 64.7k | 21 | 43.8k | 9m
57s |
| implement* | MiniMax-M3 | 1 | 1.1M | 17.7k | 0 | 0 | 52 | 0 | 6m 30s |
| fix* | sonnet | 1 | 174 | 6.3k | 987.8k | 56.7k | 53 | 35.9k | 4m 1s |
| audit_impl* | sonnet | 1 | 36 | 1.8k | 131.4k | 43.8k | 11 | 24.9k |
8m 4s |
| prepare_pr* | MiniMax-M3 | 1 | 198.1k | 1.7k | 0 | 0 | 12 | 0 | 52s |
| compose_pr* | MiniMax-M3 | 1 | 178.6k | 2.1k | 0 | 0 | 13 | 0 | 53s |
| **Total** | | | 1.5M | 61.8k | 3.5M | 94.9k | | 178.3k | 42m 30s |

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

## Token Efficiency

| Step | LoC Changed | cache_read/LoC | cache_write/LoC | output/LoC |
|------|-------------|----------------|-----------------|------------|
| plan | 0 | — | — | — |
| verify | 0 | — | — | — |
| implement | 240 | 0.0 | 0.0 | 73.7 |
| fix | 2 | 493875.0 | 17949.0 | 3162.5 |
| audit_impl | 0 | — | — | — |
| prepare_pr | 0 | — | — | — |
| compose_pr | 0 | — | — | — |
| **Total** | **242** | 14439.5 | 736.7 | 255.2 |

## Model Usage Breakdown

| Model | steps | uncached | output | cache_read | cache_write | time |
|-------|-------|----------|--------|------------|-------------|------|
| opus[1m] | 1 | 1.7k | 15.0k | 2.0M | 73.7k | 12m 10s |
| sonnet | 3 | 278 | 25.3k | 1.5M | 104.6k | 22m 4s |
| MiniMax-M3 | 3 | 1.5M | 21.5k | 0 | 0 | 8m 15s |

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: Trecek

.autoskillit/test-source-map.json

@@ -142,7 +142,8 @@
     "tests/contracts/test_exogenous_string_coupling.py",
     "tests/contracts/test_instruction_surface.py",
     "tests/contracts/test_sous_chef_routing.py",
-    "tests/contracts/test_sous_chef_scheduling.py"
+    "tests/contracts/test_sous_chef_scheduling.py",
+    "tests/execution/backends/test_failure_predicate_spanning.py"
   ],
   "src/autoskillit/cli/_restart.py": [
     "tests/cli/test_restart.py"
@@ -4573,6 +4574,7 @@
     "tests/cli/test_orchestrator_prompt_contract.py",
     "tests/contracts/test_exogenous_string_coupling.py",
     "tests/contracts/test_instruction_surface.py",
+    "tests/execution/backends/test_failure_predicate_spanning.py",
     "tests/fleet/test_campaign_capture.py",
     "tests/fleet/test_capture_roundtrip.py",
     "tests/fleet/test_food_truck_prompt.py",
@@ -14571,6 +14573,11 @@
     "tests/server/test_tools_run_python.py",
     "tests/server/test_tools_run_python_cwd.py"
   ],
+  "src/autoskillit/server/tools/_types.py": [
+    "tests/execution/backends/test_failure_predicate_spanning.py",
+    "tests/infra/test_pretty_output_hook_infra.py",
+    "tests/server/test_tools_types_module.py"
+  ],
   "src/autoskillit/server/tools/tools_agents.py": [
     "tests/server/test_tools_agents.py"
   ],

tests/_test_filter.py

@@ -808,6 +808,10 @@ class ImportContext(enum.StrEnum):
             "fleet",
             "server",
             "cli",
+            # file-level: cross-layer test that imports fleet prompt builder
+            "execution/backends/test_failure_predicate_spanning.py",
+            # file-level: execution test that imports fleet.build_protected_campaign_ids
+            "execution/test_session_log_retention.py",
         }
     ),
     # L3
@@ -823,12 +827,16 @@ class ImportContext(enum.StrEnum):
             "fleet/test_gate_state_persistence.py",
             "fleet/test_dispatch_identity_continuity.py",
             "fleet/test_dispatch_envelope_fields.py",
+            # file-level: cross-layer test that imports server TypedDict schemas
+            "execution/backends/test_failure_predicate_spanning.py",
         }
     ),
     "cli": frozenset(
         {
             "cli",
             "__main__",
+            # file-level: cross-layer test that imports cli prompt builder
+            "execution/backends/test_failure_predicate_spanning.py",
         }
     ),
     # Infra (non-layered)

tests/execution/backends/test_failure_predicate_spanning.py

@@ -0,0 +1,231 @@
+from __future__ import annotations
+
+import inspect
+import re
+from typing import get_type_hints
+
+import pytest
+
+from autoskillit.cli._prompts_orchestrator import _build_orchestrator_prompt
+from autoskillit.fleet._prompts import _build_food_truck_prompt
+from autoskillit.server.tools._types import (
+    MergeWorktreeResult,
+    RunCmdResult,
+    RunSkillResult,
+    TestCheckResult,
+)
+
+pytestmark = [pytest.mark.layer("execution"), pytest.mark.small]
+
+# ── Shared constants ──────────────────────────────────────────────
+
+_TOOL_TYPEDDICT_MAP: dict[str, type] = {
+    "test_check": TestCheckResult,
+    "merge_worktree": MergeWorktreeResult,
+    "run_cmd": RunCmdResult,
+    "run_skill": RunSkillResult,
+}
+
+_EXPECTED_FIELD_MAP: dict[str, str] = {
+    "test_check": "passed",
+    "merge_worktree": "error",
+    "run_cmd": "success",
+    "run_skill": "success",
+}
+
+_ALL_TOOL_NAMES = ["test_check", "merge_worktree", "run_cmd", "run_skill", "classify_fix"]
+
+
+# ── Shared helper ─────────────────────────────────────────────────
+
+
+def _extract_predicate_block(source_text: str) -> str:
+    """Locate the 'FAILURE PREDICATES' section dynamically and extract it.
+
+    Searches for the header line containing 'FAILURE PREDICATES — when to follow'
+    and extracts all subsequent '- tool_name:' lines until a blank line or
+    next section boundary.
+    """
+    match = re.search(
+        r"(FAILURE PREDICATES — when to follow on_failure:.*?)(?=\n\n|\n[A-Z])",
+        source_text,
+        re.DOTALL,
+    )
+    assert match is not None, (
+        "Could not find 'FAILURE PREDICATES — when to follow on_failure:' section"
+    )
+    return match.group(1)
+
+
+def _extract_open_kitchen_predicate_block(rendered_text: str) -> str:
+    """Extract the 'FAILURE PREDICATE — open_kitchen:' block from rendered prompt."""
+    match = re.search(
+        r"(FAILURE PREDICATE — open_kitchen:.*?)(?=\n\n|\nFAILURE PREDICATE —[^o]|\Z)",
+        rendered_text,
+        re.DOTALL,
+    )
+    assert match is not None, "Could not find 'FAILURE PREDICATE — open_kitchen:' section"
+    return match.group(1)
+
+
+def _parse_predicate_lines(block: str) -> dict[str, str]:
+    """Parse '- tool_name: ...' lines from a predicate block into a dict."""
+    lines: dict[str, str] = {}
+    for m in re.finditer(r"^- (\w+): (.+)$", block, re.MULTILINE):
+        lines[m.group(1)] = m.group(0)
+    return lines
+
+
+# ── Stub helpers ──────────────────────────────────────────────────
+
+
+def _render_orchestrator() -> str:
+    return _build_orchestrator_prompt(
+        recipe_name="stub-recipe",
+        mcp_prefix="mcp__stub",
+    )
+
+
+def _render_food_truck() -> str:
+    return _build_food_truck_prompt(
+        recipe="stub-recipe",
+        task="stub task",
+        ingredients={},
+        mcp_prefix="mcp__stub",
+        dispatch_id="d-00000000",
+        campaign_id="c-00000000",
+        l3_timeout_sec=300,
+    )
+
+
+_RENDER_HELPER_MAP = {
+    _build_orchestrator_prompt: _render_orchestrator,
+    _build_food_truck_prompt: _render_food_truck,
+}
+
+
+# ── Test classes ──────────────────────────────────────────────────
+
+
+class TestToolPredicateFieldsMatchSchema:
+    """Validate each FAILURE PREDICATE field name against its TypedDict schema."""
+
+    @pytest.mark.parametrize(
+        "builder_fn",
+        [_build_orchestrator_prompt, _build_food_truck_prompt],
+        ids=["orchestrator", "food_truck"],
+    )
+    @pytest.mark.parametrize("tool_name", list(_TOOL_TYPEDDICT_MAP.keys()))
+    def test_predicate_field_in_typeddict(self, tool_name: str, builder_fn: object) -> None:
+        rendered = _RENDER_HELPER_MAP[builder_fn]()  # type: ignore[index]
+        block = _extract_predicate_block(rendered)
+        parsed = _parse_predicate_lines(block)
+
+        assert tool_name in parsed, f"{tool_name} not found in predicate block"
+
+        expected_field = _EXPECTED_FIELD_MAP[tool_name]
+        td_cls = _TOOL_TYPEDDICT_MAP[tool_name]
+        valid_keys = set(get_type_hints(td_cls))
+
+        assert expected_field in valid_keys, (
+            f"Predicate field '{expected_field}' for {tool_name} "
+            f"is not a key in {td_cls.__name__}. Valid keys: {sorted(valid_keys)}"
+        )
+
+        line_text = parsed[tool_name]
+        assert f'"{expected_field}:' in line_text or f'"{expected_field}"' in line_text, (
+            f"Expected field '{expected_field}' not found in predicate line: {line_text}"
+        )
+
+    @pytest.mark.parametrize(
+        "builder_fn",
+        [_build_orchestrator_prompt, _build_food_truck_prompt],
+        ids=["orchestrator", "food_truck"],
+    )
+    def test_classify_fix_contains_error_colon(self, builder_fn: object) -> None:
+        rendered = _RENDER_HELPER_MAP[builder_fn]()  # type: ignore[index]
+        block = _extract_predicate_block(rendered)
+        parsed = _parse_predicate_lines(block)
+
+        assert "classify_fix" in parsed, "classify_fix not found in predicate block"
+        assert "error:" in parsed["classify_fix"], (
+            f"classify_fix predicate does not contain 'error:': {parsed['classify_fix']}"
+        )
+
+    @pytest.mark.parametrize(
+        "builder_fn",
+        [_build_orchestrator_prompt, _build_food_truck_prompt],
+        ids=["orchestrator", "food_truck"],
+    )
+    def test_predicate_block_contains_all_tools(self, builder_fn: object) -> None:
+        rendered = _RENDER_HELPER_MAP[builder_fn]()  # type: ignore[index]
+        block = _extract_predicate_block(rendered)
+        for tool_name in _ALL_TOOL_NAMES:
+            assert tool_name in block, f"Tool '{tool_name}' missing from FAILURE PREDICATES block"
+
+
+class TestOpenKitchenIngredientTableMarker:
+    """POST-C1 regression guard: open_kitchen predicate must reference JSON fields,
+    not the hook-injected display marker."""
+
+    def test_no_ingredients_table_marker(self) -> None:
+        rendered = _render_food_truck()
+        block = _extract_open_kitchen_predicate_block(rendered)
+        assert "--- INGREDIENTS TABLE ---" not in block, (
+            "open_kitchen FAILURE PREDICATE must not reference the hook-injected "
+            "'--- INGREDIENTS TABLE ---' marker"
+        )
+
+    def test_references_success_and_ingredients_table(self) -> None:
+        rendered = _render_food_truck()
+        block = _extract_open_kitchen_predicate_block(rendered)
+        assert "success" in block, (
+            "open_kitchen predicate block must reference 'success' as a JSON field"
+        )
+        assert "ingredients_table" in block, (
+            "open_kitchen predicate block must reference 'ingredients_table' as a JSON field"
+        )
+
+
+class TestPredicateParity:
+    """Assert the five shared tool predicate lines are character-identical
+    between orchestrator and food truck builders."""
+
+    def test_shared_predicate_lines_character_identical(self) -> None:
+        orch_rendered = _render_orchestrator()
+        ft_rendered = _render_food_truck()
+
+        orch_block = _extract_predicate_block(orch_rendered)
+        ft_block = _extract_predicate_block(ft_rendered)
+
+        orch_lines = _parse_predicate_lines(orch_block)
+        ft_lines = _parse_predicate_lines(ft_block)
+
+        for tool_name in _ALL_TOOL_NAMES:
+            assert tool_name in orch_lines, (
+                f"{tool_name} missing from orchestrator predicate block"
+            )
+            assert tool_name in ft_lines, f"{tool_name} missing from food truck predicate block"
+            assert orch_lines[tool_name] == ft_lines[tool_name], (
+                f"Predicate parity violation for {tool_name}:\n"
+                f"  orchestrator: {orch_lines[tool_name]!r}\n"
+                f"  food_truck:   {ft_lines[tool_name]!r}"
+            )
+
+    def test_predicate_extraction_is_dynamic(self) -> None:
+        """Verify extraction works on both source text and rendered output,
+        proving it searches by header not by hardcoded offsets."""
+        orch_source = inspect.getsource(_build_orchestrator_prompt)
+        ft_source = inspect.getsource(_build_food_truck_prompt)
+
+        for source in [orch_source, ft_source]:
+            block = _extract_predicate_block(source)
+            assert "FAILURE PREDICATES" in block
+            for tool_name in _ALL_TOOL_NAMES:
+                assert tool_name in block
+
+        for rendered in [_render_orchestrator(), _render_food_truck()]:
+            block = _extract_predicate_block(rendered)
+            assert "FAILURE PREDICATES" in block
+            for tool_name in _ALL_TOOL_NAMES:
+                assert tool_name in block