Implementation Plan: T5-P4-A8-WP2 Prove planner.yaml composes valid:true under codex backend

tests/_test_filter.py

@@ -769,6 +769,7 @@ class ImportContext(enum.StrEnum):
             # Execution file-level entries:
             "execution/test_headless_path_validation.py",
             "execution/test_zero_write_detection.py",
+            "execution/test_smoke_codex.py",
             # Fleet file-level entries (9 of N import autoskillit.recipe):
             "fleet/test_fleet_e2e.py",
             "fleet/test_fleet_e2e_codex.py",

tests/arch/test_layer_enforcement.py

@@ -1460,6 +1460,8 @@ def test_default_classes_only_instantiated_inside_factory_or_allowlist() -> None
     ),
     # write detection sync guard validates recipe contract patterns against test fixtures
     "tests/execution/test_zero_write_detection.py": frozenset({"autoskillit.recipe"}),
+    # smoke composition tests validate recipe validity under codex backend — needs recipe API
+    "tests/execution/test_smoke_codex.py": frozenset({"autoskillit.recipe"}),
     # quota tests cross into config to validate the contract between vocab constants
     # (execution layer) and config defaults — intentional, documented cross-ref
     "tests/execution/test_quota_binding.py": frozenset({"autoskillit.config"}),

tests/execution/test_smoke_codex.py

@@ -7,18 +7,23 @@
 
 from __future__ import annotations
 
+import json
 import os
 import subprocess
 from pathlib import Path
+from typing import NamedTuple
 
 import pytest
 
 from autoskillit.core import BackendEventKind, DirectInstall, SessionEvent
+from autoskillit.core.types import Severity
+from autoskillit.execution.backends import CompositeSessionLocator
 from autoskillit.execution.backends.codex import (
     CodexBackend,
     CodexResultParser,
     CodexStreamParser,
 )
+from autoskillit.recipe._api import load_and_validate
 
 pytestmark = [pytest.mark.layer("execution"), pytest.mark.large]
 
@@ -27,6 +32,15 @@
     " or ~/.codex/auth.json to run Codex smoke tests"
 )
 
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+
+
+class _CodexSessionData(NamedTuple):
+    result: subprocess.CompletedProcess
+    events: list[SessionEvent]
+    thread_id: str
+
+
 _skip_unless_codex_smoke = pytest.mark.skipif(
     not os.environ.get("CODEX_SMOKE_TEST")
     or (
@@ -119,3 +133,196 @@ def test_food_truck_cmd_has_required_flags(self) -> None:
         assert "--json" in cmd.cmd
         assert "--sandbox" in cmd.cmd
         assert cmd.cmd[cmd.cmd.index("--sandbox") + 1] == "read-only"
+
+
+@_skip_unless_codex_smoke
+@pytest.mark.smoke
+class TestCodexSmokeRecipeComposition:
+    """Recipe composition validity and session diagnostics under codex backend.
+
+    Tests 1-2 are pure recipe validation (no codex CLI needed, but gated
+    with the class for organizational grouping).
+    Tests 3-5 use a shared single codex exec invocation.
+    """
+
+    @pytest.fixture(scope="class")
+    def codex_session(self, tmp_path_factory):
+        """Single codex exec invocation shared across smoke tests."""
+        proc = subprocess.run(
+            [
+                "codex",
+                "exec",
+                "--json",
+                "--sandbox",
+                "workspace-write",
+                "Respond with exactly: hello",
+            ],
+            capture_output=True,
+            text=True,
+            timeout=int(os.environ.get("CODEX_SMOKE_TIMEOUT", "30")),
+        )
+        parser = CodexStreamParser()
+        events: list[SessionEvent] = []
+        thread_id = ""
+        for line in proc.stdout.splitlines():
+            evt = parser.parse_line(line)
+            if evt is not None:
+                events.append(evt)
+            if thread_id:
+                continue
+            try:
+                obj = json.loads(line)
+            except (json.JSONDecodeError, ValueError):
+                continue
+            if obj.get("type") == "thread.started":
+                thread_id = obj.get("thread_id", "")
+            elif obj.get("type") == "session_meta":
+                thread_id = obj.get("payload", {}).get("id", "")
+
+        tmp_dir = tmp_path_factory.mktemp("codex_smoke")
+        rollout = tmp_dir / "codex-sessions" / "rollout.jsonl"
+        rollout.parent.mkdir(parents=True)
+        rollout.write_text(proc.stdout)
+
+        return _CodexSessionData(result=proc, events=events, thread_id=thread_id), rollout
+
+    def test_open_kitchen_implementation_valid(self) -> None:
+        result = load_and_validate(
+            "implementation",
+            project_dir=_PROJECT_ROOT,
+            backend_name="codex",
+            ingredient_overrides={"backend_supports_git_write": "false"},
+        )
+        assert result["valid"] is True, "implementation recipe invalid on codex: " + "; ".join(
+            f"[{s.get('rule')}] {s.get('message', '')[:80]}"
+            for s in result.get("suggestions", [])
+            if s.get("severity") == Severity.ERROR
+        )
+        assert len(result.get("content", "")) > 0
+        backend_compat_errors = [
+            s
+            for s in result.get("suggestions", [])
+            if s.get("rule") == "backend-incompatible-skill"
+            and s.get("severity") == Severity.ERROR
+        ]
+        assert not backend_compat_errors
+
+    def test_open_kitchen_planner_valid(self) -> None:
+        result = load_and_validate(
+            "planner",
+            project_dir=_PROJECT_ROOT,
+            backend_name="codex",
+            ingredient_overrides={"backend_supports_git_write": "false"},
+        )
+        assert result["valid"] is True
+        assert len(result.get("content", "")) > 0
+
+    def test_reduced_codex_smoke_pipeline_no_refusals(self, codex_session) -> None:
+        session_data, _rollout = codex_session
+        assert session_data.result.returncode == 0, (
+            f"codex exec failed with rc={session_data.result.returncode}: "
+            f"{session_data.result.stderr}"
+        )
+        completion_events = [
+            e for e in session_data.events if e.kind == BackendEventKind.COMPLETION
+        ]
+        assert len(completion_events) >= 1, (
+            f"Expected at least one COMPLETION event, got kinds: "
+            f"{[e.kind for e in session_data.events]}"
+        )
+        for line in session_data.result.stdout.splitlines():
+            try:
+                obj = json.loads(line)
+            except (json.JSONDecodeError, ValueError):
+                continue
+            if obj.get("type") == "turn.failed":
+                error = obj.get("error", {})
+                pytest.fail(
+                    f"Capability-gated refusal detected: "
+                    f"{error.get('code', 'unknown')}: {error.get('message', '')}"
+                )
+
+    def test_session_diagnostics_field_completeness(self, codex_session, tmp_path) -> None:
+        session_data, rollout = codex_session
+        assert session_data.thread_id, "thread_id not resolved from codex NDJSON"
+
+        from autoskillit.core.types._type_results import ProviderOutcome
+        from autoskillit.core.types._type_results_execution import (
+            RecipeIdentity,
+            SessionTelemetry,
+        )
+        from autoskillit.execution.session_log import flush_session_log
+
+        class _FakeLocator:
+            def __init__(self, path: Path) -> None:
+                self._path = path
+
+            def locate_session(self, session_id: str) -> Path:
+                return self._path
+
+            def project_log_dir(self, cwd: str) -> Path:
+                return Path(cwd)
+
+            def session_log_path(self, cwd: str, session_id: str) -> Path:
+                return self._path
+
+        flush_session_log(
+            log_dir=str(tmp_path),
+            backend="codex",
+            channel_b_capable=False,
+            session_locator=_FakeLocator(rollout),
+            cwd="/tmp/smoke",
+            session_id=session_data.thread_id,
+            pid=os.getpid(),
+            skill_command="/autoskillit:implement",
+            success=True,
+            subtype="completed",
+            exit_code=0,
+            start_ts="2026-06-28T00:00:00",
+            proc_snapshots=None,
+            telemetry=SessionTelemetry.empty(),
+            provider_outcome=ProviderOutcome.none_used(),
+            recipe_identity=RecipeIdentity.empty(),
+        )
+        index_text = (tmp_path / "sessions.jsonl").read_text().strip()
+        entry = json.loads(index_text)
+        assert entry["session_id"] == session_data.thread_id
+        assert entry["codex_log"] is not None
+        assert entry["backend"] == "codex"
+        assert entry["success"] is True
+
+    def test_composite_locator_resolves_codex_session(self, codex_session, monkeypatch) -> None:
+        session_data, rollout = codex_session
+        assert session_data.thread_id, "thread_id not resolved"
+
+        class _StubLocator:
+            def __init__(self, path: Path) -> None:
+                self._path = path
+
+            def locate_session(self, session_id: str) -> Path:
+                return self._path
+
+            def project_log_dir(self, cwd: str) -> Path:
+                return Path("/stub")
+
+            def session_log_path(self, cwd: str, session_id: str) -> Path:
+                return self._path
+
+        def _fake_backend(locator):
+            from unittest.mock import Mock
+
+            backend = Mock()
+            backend.session_locator.return_value = locator
+            cls = Mock(return_value=backend)
+            return cls
+
+        import autoskillit.execution.backends as backends_mod
+
+        monkeypatch.setattr(
+            backends_mod,
+            "BACKEND_REGISTRY",
+            {"codex": _fake_backend(_StubLocator(rollout))},
+        )
+        result = CompositeSessionLocator().locate_session(session_data.thread_id)
+        assert result is not None
+        assert result.exists()

tests/recipe/test_planner_recipe.py

@@ -1,15 +1,20 @@
 """Structural assertions for the planner recipe."""
 
 import importlib
+from pathlib import Path
 
 import pytest
 
+from autoskillit.core.types import Severity
+from autoskillit.recipe._api import load_and_validate
 from autoskillit.recipe.io import builtin_recipes_dir, load_recipe
 from autoskillit.recipe.validator import run_semantic_rules
 from tests.recipe.conftest import assert_no_rule_errors
 
 pytestmark = [pytest.mark.layer("recipe"), pytest.mark.small]
 
+_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+
 
 @pytest.fixture(scope="module")
 def planner_recipe():
@@ -374,3 +379,28 @@ def test_refine_tier_steps_all_use_dispatch_items(planner_recipe, step_name):
         f"{step_name} must use dispatch_items for per-phase parallel dispatch. "
         "All refine_* steps at the assignment/WP tier must shard by phase."
     )
+
+
+def test_planner_recipe_valid_on_codex_backend():
+    """Planner recipe must compose valid=True under codex backend capability overrides."""
+    result = load_and_validate(
+        "planner",
+        project_dir=_PROJECT_ROOT,
+        backend_name="codex",
+        ingredient_overrides={"backend_supports_git_write": "false"},
+    )
+    assert result["valid"] is True, "planner recipe invalid on codex: " + "; ".join(
+        f"[{s.get('rule')}] {s.get('message', '')[:80]}"
+        for s in result.get("suggestions", [])
+        if s.get("severity") == Severity.ERROR
+    )
+    assert len(result.get("content", "")) > 0, "planner recipe produced empty content on codex"
+    backend_compat_errors = [
+        s
+        for s in result.get("suggestions", [])
+        if s.get("rule") == "backend-incompatible-skill" and s.get("severity") == Severity.ERROR
+    ]
+    assert not backend_compat_errors, (
+        "planner recipe has backend-incompatible-skill errors on codex: "
+        + "; ".join(s.get("message", "") for s in backend_compat_errors)
+    )