Implementation Plan: T5-P5-A4-WP1 — Result Validation Helper with Fail-Closed Envelope (#4144)

## Summary

Add a `_validate_result` function to
`src/autoskillit/server/tools/_types.py` that enforces output shape
invariants on MCP tool result dicts before serialization. The function
checks required-key presence, non-None values, `success==True` when
present, and non-empty `content` when present. On violation it returns a
`json.dumps`-serialized `ToolFailureEnvelope` (fail-closed); on pass it
returns `None`. Wire this validator into `open_kitchen` (normal
named-recipe branch only — deferred-recall branch exempted per
acceptance criteria) and `load_recipe` (unified exit, gated on result
validity) with structured logging at each fail-closed site.

Closes #4036

## Implementation Plan

Plan file:
`/home/talon/projects/autoskillit-runs/impl-20260628-093915-207021/.autoskillit/temp/make-plan/t5-p5-a4-wp1-validate-result-helper_plan_2026-06-28_094800.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 | 843 | 26.4k | 985.6k | 97.6k | 44 | 86.5k | 14m
10s |
| verify* | sonnet | 1 | 60 | 11.9k | 304.3k | 62.3k | 22 | 43.8k | 6m
34s |
| implement* | MiniMax-M3 | 1 | 76.2k | 10.9k | 2.4M | 0 | 93 | 0 | 5m
31s |
| fix* | sonnet | 1 | 438 | 51.6k | 5.1M | 147.2k | 140 | 129.2k | 18m
44s |
| audit_impl* | sonnet | 1 | 1.3k | 15.2k | 223.8k | 57.0k | 16 | 42.3k
| 8m 35s |
| prepare_pr* | MiniMax-M3 | 1 | 44.1k | 2.2k | 232.2k | 0 | 16 | 0 | 1m
40s |
| compose_pr* | MiniMax-M3 | 1 | 37.2k | 1.8k | 177.0k | 0 | 13 | 0 |
31s |
| **Total** | | | 160.2k | 119.9k | 9.4M | 147.2k | | 301.7k | 55m 48s |

\* *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 | 193 | 12177.9 | 0.0 | 56.4 |
| fix | 309 | 16587.2 | 418.0 | 166.9 |
| audit_impl | 0 | — | — | — |
| prepare_pr | 0 | — | — | — |
| compose_pr | 0 | — | — | — |
| **Total** | **502** | 18722.5 | 601.0 | 238.9 |

## Model Usage Breakdown

| Model | steps | uncached | output | cache_read | cache_write | time |
|-------|-------|----------|--------|------------|-------------|------|
| opus[1m] | 1 | 843 | 26.4k | 985.6k | 86.5k | 14m 10s |
| sonnet | 3 | 1.8k | 78.7k | 5.7M | 215.2k | 33m 54s |
| MiniMax-M3 | 3 | 157.5k | 14.9k | 2.8M | 0 | 7m 43s |

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Trecek

src/autoskillit/server/tools/_types.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import Any, Literal, TypedDict
 
 from autoskillit.core import ModelTotalEntry, RetryReason
@@ -23,6 +24,7 @@
     "ToolFailureEnvelope",
     "server_failure_envelope",
     "input_failure_envelope",
+    "_validate_result",
 ]
 
 
@@ -244,3 +246,56 @@ def input_failure_envelope(
         stage=stage,
         retriable=False,
     )
+
+
+def _validate_result(
+    result: dict[str, Any],
+    *,
+    required_keys: frozenset[str],
+    tool_name: str,
+    retriable: bool = False,
+) -> str | None:
+    """Validate a tool result dict and return a fail-closed envelope on violation.
+
+    Returns ``None`` when all invariants hold, or a ``json.dumps``-serialized
+    ``ToolFailureEnvelope`` on the first violation detected.
+    """
+    _stage = f"validate_result:{tool_name}"
+    for key in sorted(required_keys):
+        if key not in result:
+            return json.dumps(
+                ToolFailureEnvelope(
+                    success=False,
+                    error=f"Missing required key: {key}",
+                    stage=_stage,
+                    retriable=retriable,
+                )
+            )
+        if result[key] is None:
+            return json.dumps(
+                ToolFailureEnvelope(
+                    success=False,
+                    error=f"Required key is None: {key}",
+                    stage=_stage,
+                    retriable=retriable,
+                )
+            )
+    if "success" in result and result["success"] is not True:
+        return json.dumps(
+            ToolFailureEnvelope(
+                success=False,
+                error=f"Result reports failure: success={result['success']!r}",
+                stage=_stage,
+                retriable=retriable,
+            )
+        )
+    if "content" in result and not result["content"]:
+        return json.dumps(
+            ToolFailureEnvelope(
+                success=False,
+                error="Result content is empty",
+                stage=_stage,
+                retriable=retriable,
+            )
+        )
+    return None

src/autoskillit/server/tools/tools_kitchen.py

@@ -68,6 +68,7 @@
     _check_dispatch_feasibility,
     filter_steps_by_post_prune,
 )
+from autoskillit.server.tools._types import _validate_result
 
 logger = get_logger(__name__)
 
@@ -888,6 +889,20 @@ async def open_kitchen(
             if warning:
                 result["hook_warning"] = warning.strip()
 
+            _required_keys = frozenset({"success", "content", "valid"})
+            if ingredients_only:
+                _required_keys = _required_keys - {"content"}
+            _validation_err = _validate_result(
+                result, required_keys=_required_keys, tool_name="open_kitchen"
+            )
+            if _validation_err is not None:
+                logger.warning(
+                    "open_kitchen_fail_closed",
+                    tool="open_kitchen",
+                    stage="validate_result",
+                )
+                return _validation_err
+
             return json.dumps(result)
 
         text = (

src/autoskillit/server/tools/tools_recipe.py

@@ -29,6 +29,7 @@
     _promote_capability_keys,
 )
 from autoskillit.server.tools._cancellation_shield import _cancellation_shield
+from autoskillit.server.tools._types import _validate_result
 
 logger = get_logger(__name__)
 
@@ -244,6 +245,19 @@ async def load_recipe(
                 )
             if ingredients_only:
                 result = strip_ingredients_only_keys(result)
+            _required_keys: frozenset[str] = frozenset()
+            if not ingredients_only and result.get("valid", False):
+                _required_keys = frozenset({"content"})
+            _validation_err = _validate_result(
+                result, required_keys=_required_keys, tool_name="load_recipe"
+            )
+            if _validation_err is not None:
+                logger.warning(
+                    "load_recipe_fail_closed",
+                    tool="load_recipe",
+                    stage="validate_result",
+                )
+                return _validation_err
             return json.dumps(result)
     except Exception as exc:
         logger.error("load_recipe unhandled exception", exc_info=True)

tests/arch/test_subpackage_isolation.py

@@ -967,7 +967,7 @@ def test_data_directories_are_not_python_packages() -> None:
         "co-located with the execution engine that calls them",
     ),
     "tools_kitchen.py": (
-        1290,
+        1310,
         "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; "

tests/infra/test_schema_version_convention.py

@@ -119,11 +119,11 @@ def _is_yaml_dump(node: ast.expr) -> bool:
     # _lifespan.py — hooks.json self-heal on startup drift (co-owned with Claude plugin system)
     ("src/autoskillit/server/_lifespan.py", 90),
     # tools_kitchen.py — hook config, quota guard, git_ops_policy, ingredient locks overlay
-    ("src/autoskillit/server/tools/tools_kitchen.py", 204),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 223),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 257),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 1013),
-    ("src/autoskillit/server/tools/tools_kitchen.py", 1073),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 205),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 224),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 258),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 1028),
+    ("src/autoskillit/server/tools/tools_kitchen.py", 1088),
     # tools_pipeline_tracker.py — tracker_data dict
     ("src/autoskillit/server/tools/tools_pipeline_tracker.py", 166),
     # tools_status.py — mcp_data dict

tests/server/AGENTS.md

@@ -133,6 +133,7 @@ Server tool handler unit tests — kitchen, execution, CI, clone, workspace tool
 | `test_tools_status_quota_and_db.py` | Tests for server status tools: quota events, telemetry writing, and DB access |
 | `test_tools_status_summaries.py` | Tests for server status tools: token and timing summaries |
 | `test_tools_types_module.py` | Tests for server/tools/_types.py TypedDict imports and failure envelope factory functions |
+| `test_tools_types_validate.py` | Tests for _validate_result helper in server/tools/_types.py — shape invariant enforcement and fail-closed envelope validation |
 | `test_tools_workspace.py` | Tests for autoskillit server workspace tools |
 | `test_tools_agents.py` | Tests for agent pack registry, MCP resources, and `unlock_agent_pack` |
 | `test_track_response_size.py` | Tests for the track_response_size decorator in autoskillit.server._notify |

tests/server/test_tools_kitchen_gate.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import json
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -341,3 +341,77 @@ async def test_close_kitchen_drains_orphaned_github_api_entries(tmp_path, monkey
     data = json.loads(orphan_path.read_text())
     assert data["total_requests"] == 1
     assert not log._entries
+
+
+@pytest.mark.anyio
+async def test_open_kitchen_fail_closed_empty_content(monkeypatch, tmp_path):
+    """open_kitchen returns fail-closed when content is empty (caught at validity gate)."""
+    monkeypatch.chdir(tmp_path)
+    ctx = _make_mock_ctx()
+    ctx.gate.enabled = True
+    ctx.gate_infrastructure_ready = True
+    _load_result = {"valid": True, "content": "", "dispatch_feasible": True}
+    ctx.recipes.load_and_validate.return_value = _load_result
+    ctx.recipes.find.return_value = MagicMock(path=tmp_path / "r.yaml")
+    ctx.config.providers = MagicMock()
+    ctx.backend = MagicMock()
+    ctx.backend.name = "test"
+    with (
+        patch("autoskillit.server._get_ctx", return_value=ctx),
+        patch("autoskillit.server.logger"),
+        patch(
+            "autoskillit.server.tools.tools_kitchen._apply_triage_gate",
+            new=AsyncMock(return_value=_load_result),
+        ),
+        patch(
+            "autoskillit.server.tools.tools_kitchen._prime_quota_cache",
+            new_callable=AsyncMock,
+        ),
+        patch("autoskillit.server.tools.tools_kitchen._write_hook_config"),
+    ):
+        from autoskillit.server.tools.tools_kitchen import open_kitchen
+
+        raw = await open_kitchen(name="test-recipe")
+
+    parsed = json.loads(raw)
+    assert parsed["success"] is False
+    assert parsed.get("kitchen") == "failed"
+    assert "error" in parsed
+    assert "failed validation" in parsed["error"]
+
+
+@pytest.mark.anyio
+async def test_open_kitchen_fail_closed_missing_content(monkeypatch, tmp_path):
+    """open_kitchen returns fail-closed when content key is missing (caught at validity gate)."""
+    monkeypatch.chdir(tmp_path)
+    ctx = _make_mock_ctx()
+    ctx.gate.enabled = True
+    ctx.gate_infrastructure_ready = True
+    _load_result = {"valid": True, "dispatch_feasible": True}
+    ctx.recipes.load_and_validate.return_value = _load_result
+    ctx.recipes.find.return_value = MagicMock(path=tmp_path / "r.yaml")
+    ctx.config.providers = MagicMock()
+    ctx.backend = MagicMock()
+    ctx.backend.name = "test"
+    with (
+        patch("autoskillit.server._get_ctx", return_value=ctx),
+        patch("autoskillit.server.logger"),
+        patch(
+            "autoskillit.server.tools.tools_kitchen._apply_triage_gate",
+            new=AsyncMock(return_value=_load_result),
+        ),
+        patch(
+            "autoskillit.server.tools.tools_kitchen._prime_quota_cache",
+            new_callable=AsyncMock,
+        ),
+        patch("autoskillit.server.tools.tools_kitchen._write_hook_config"),
+    ):
+        from autoskillit.server.tools.tools_kitchen import open_kitchen
+
+        raw = await open_kitchen(name="test-recipe")
+
+    parsed = json.loads(raw)
+    assert parsed["success"] is False
+    assert parsed.get("kitchen") == "failed"
+    assert "error" in parsed
+    assert "failed validation" in parsed["error"]

tests/server/test_tools_load_recipe.py

@@ -580,3 +580,47 @@ async def test_load_recipe_surfaces_validation_failure(
         )
         assert "errors" in result
         assert len(result["errors"]) > 0
+
+
+class TestLoadRecipeFailClosed:
+    """Fail-closed validation for empty and missing content."""
+
+    @pytest.fixture(autouse=True)
+    def _ensure_ctx(self, tool_ctx_kitchen_open):
+        self.ctx = tool_ctx_kitchen_open
+
+    @pytest.mark.anyio
+    async def test_load_recipe_fail_closed_empty_content(self, monkeypatch, tmp_path):
+        monkeypatch.chdir(tmp_path)
+        from autoskillit.recipe._api_cache import _LOAD_CACHE
+
+        _LOAD_CACHE.clear()
+        test_result = {"valid": True, "content": "", "dispatch_feasible": True}
+        monkeypatch.setattr(self.ctx.recipes, "load_and_validate", lambda *a, **kw: test_result)
+        monkeypatch.setattr(self.ctx.recipes, "find", lambda *a, **kw: None)
+        with patch(
+            "autoskillit.server.tools.tools_recipe._apply_triage_gate",
+            new=AsyncMock(return_value=test_result),
+        ):
+            raw = await load_recipe(name="test-recipe")
+        parsed = json.loads(raw)
+        assert parsed["success"] is False
+        assert "content" in parsed["error"].lower()
+
+    @pytest.mark.anyio
+    async def test_load_recipe_fail_closed_missing_content(self, monkeypatch, tmp_path):
+        monkeypatch.chdir(tmp_path)
+        from autoskillit.recipe._api_cache import _LOAD_CACHE
+
+        _LOAD_CACHE.clear()
+        test_result = {"valid": True, "dispatch_feasible": True}
+        monkeypatch.setattr(self.ctx.recipes, "load_and_validate", lambda *a, **kw: test_result)
+        monkeypatch.setattr(self.ctx.recipes, "find", lambda *a, **kw: None)
+        with patch(
+            "autoskillit.server.tools.tools_recipe._apply_triage_gate",
+            new=AsyncMock(return_value=test_result),
+        ):
+            raw = await load_recipe(name="test-recipe")
+        parsed = json.loads(raw)
+        assert parsed["success"] is False
+        assert "content" in parsed["error"]

tests/server/test_tools_types_module.py

@@ -126,6 +126,7 @@ def test_tool_failure_envelope_in_all(self):
         assert "ToolFailureEnvelope" in types_all
         assert "server_failure_envelope" in types_all
         assert "input_failure_envelope" in types_all
+        assert "_validate_result" in types_all
 
     def test_tool_failure_envelope_distinct_from_kitchen_envelope(self):
         from autoskillit.server.tools._types import ToolFailureEnvelope