Implementation Plan: T5-P4-A6-WP1 Enforce the contract — remove _FORWARD_DECLARED exemption, add capability gate branch coverage

src/autoskillit/execution/headless/_headless_result.py

@@ -551,7 +551,7 @@ def _build_skill_result(
     success = outcome == SessionOutcome.SUCCEEDED
     needs_retry = outcome == SessionOutcome.RETRIABLE
 
-    infra_category = classify_infra_exit(session, result)
+    infra_category = classify_infra_exit(session, result, capabilities=backend.capabilities)
     api_retry = _build_api_retry_outcome(session)
 
     # API error override: when the session failed due to an API infrastructure error

src/autoskillit/execution/session/_exit_classification.py

@@ -9,7 +9,7 @@
 from autoskillit.core import CODEX_CONTEXT_EXHAUSTION_MARKER, InfraExitCategory, get_logger
 
 if TYPE_CHECKING:
-    from autoskillit.core import SubprocessResult
+    from autoskillit.core import BackendCapabilities, SubprocessResult
     from autoskillit.execution.session._session_model import ClaudeSessionResult
 
 logger = get_logger(__name__)
@@ -94,6 +94,8 @@ def is_signal_death_code(returncode: int) -> bool:
 def classify_infra_exit(
     session: ClaudeSessionResult,
     result: SubprocessResult,
+    *,
+    capabilities: BackendCapabilities | None = None,
 ) -> InfraExitCategory:
     """Classify why a headless session exited at the infrastructure level.
 
@@ -102,11 +104,19 @@ def classify_infra_exit(
     (HTTP 429) is checked before other API errors so transient rate limits are
     distinguished from structural failures and routed to on_rate_limit instead of
     on_context_limit.
+
+    When ``capabilities`` is provided, context-exhaustion detection is gated on
+    ``supports_context_exhaustion_detection``; backends that do not support
+    context-exhaustion telemetry (e.g., Claude's local session JSONL marker is
+    already authoritative) skip those checks. ``capabilities=None`` is permissive
+    and preserves the prior behavior for callers that have not yet threaded a
+    capability object through.
     """
-    if session._is_context_exhausted():
-        return InfraExitCategory.CONTEXT_EXHAUSTED
-    if _CODEX_CONTEXT_EXHAUSTION_PATTERN.search(result.stderr):
-        return InfraExitCategory.CONTEXT_EXHAUSTED
+    if capabilities is None or capabilities.supports_context_exhaustion_detection:
+        if session._is_context_exhausted():
+            return InfraExitCategory.CONTEXT_EXHAUSTED
+        if _CODEX_CONTEXT_EXHAUSTION_PATTERN.search(result.stderr):
+            return InfraExitCategory.CONTEXT_EXHAUSTED
     # Rate limit detection — must precede all API_ERROR checks (including
     # api_retry_exhausted) so 429s are classified as RATE_LIMITED even
     # when retries are exhausted.

tests/arch/test_capability_consumption.py

@@ -29,11 +29,6 @@ class ForwardDeclaredField:
         rationale="thinking-block rendering gating",
         added_date=date(2026, 5, 31),
     ),
-    "supports_context_exhaustion_detection": ForwardDeclaredField(
-        issue=3384,
-        rationale="context exhaustion and recovery paths",
-        added_date=date(2026, 5, 31),
-    ),
     "min_version": ForwardDeclaredField(
         issue=3122,
         rationale="version validation via BackendCapabilities fields",

tests/execution/backends/test_coding_agent_backend_conformance.py

@@ -33,7 +33,6 @@
         "patch_format",
         "required_session_files",
         "session_dir_symlinks",
-        "supports_context_exhaustion_detection",
         "supports_thinking_blocks",
     }
 )
@@ -352,6 +351,10 @@ def test_setup_session_dir_and_locator_round_trip(
                 f" for backend {self.backend.name!r}"
             )
 
+    def test_supports_context_exhaustion_detection_is_bool(self) -> None:
+        """BackendCapabilities.supports_context_exhaustion_detection — capability is bool-typed."""
+        assert isinstance(self.backend.capabilities.supports_context_exhaustion_detection, bool)
+
 
 def test_every_capability_field_exercised_or_not_yet_live() -> None:
     fields = {f.name for f in dataclasses.fields(BackendCapabilities)}

tests/execution/test_exit_classification.py

@@ -7,6 +7,8 @@
 import pytest
 
 from autoskillit.core.types import (
+    CODEX_CONTEXT_EXHAUSTION_MARKER,
+    BackendCapabilities,
     ChannelConfirmation,
     InfraExitCategory,
     SubprocessResult,
@@ -304,6 +306,72 @@ def test_rate_limit_exceeded_is_rate_limited_not_context_exhausted(self):
         assert classify_infra_exit(session, result) == InfraExitCategory.RATE_LIMITED
 
 
+class TestContextExhaustionCapabilityGate:
+    """Capability gate: supports_context_exhaustion_detection controls detection."""
+
+    def test_capability_false_suppresses_context_exhausted(self):
+        """capability=False + context exhaustion signals → NOT CONTEXT_EXHAUSTED."""
+        caps = BackendCapabilities(supports_context_exhaustion_detection=False)
+        session = ClaudeSessionResult(
+            subtype="success",
+            is_error=True,
+            result="prompt is too long",
+            session_id="s1",
+            jsonl_context_exhausted=True,
+        )
+        result = _sr(returncode=1)
+        assert (
+            classify_infra_exit(session, result, capabilities=caps)
+            != InfraExitCategory.CONTEXT_EXHAUSTED
+        )
+
+    def test_capability_true_produces_context_exhausted(self):
+        """capability=True + context exhaustion signals → CONTEXT_EXHAUSTED."""
+        caps = BackendCapabilities(supports_context_exhaustion_detection=True)
+        session = ClaudeSessionResult(
+            subtype="success",
+            is_error=True,
+            result="prompt is too long",
+            session_id="s1",
+            jsonl_context_exhausted=True,
+        )
+        result = _sr(returncode=1)
+        assert (
+            classify_infra_exit(session, result, capabilities=caps)
+            == InfraExitCategory.CONTEXT_EXHAUSTED
+        )
+
+    def test_capability_false_suppresses_context_exhausted_via_stderr(self):
+        """capability=False + Codex stderr pattern → NOT CONTEXT_EXHAUSTED."""
+        caps = BackendCapabilities(supports_context_exhaustion_detection=False)
+        session = ClaudeSessionResult(
+            subtype="success",
+            is_error=True,
+            result="",
+            session_id="s1",
+        )
+        result = _sr(returncode=1, stderr=CODEX_CONTEXT_EXHAUSTION_MARKER)
+        assert (
+            classify_infra_exit(session, result, capabilities=caps)
+            != InfraExitCategory.CONTEXT_EXHAUSTED
+        )
+
+    def test_capability_true_detects_context_exhausted_via_stderr(self):
+        """capability=True + Codex stderr pattern → CONTEXT_EXHAUSTED."""
+        caps = BackendCapabilities(supports_context_exhaustion_detection=True)
+        session = ClaudeSessionResult(
+            subtype="success",
+            is_error=True,
+            result="",
+            session_id="s1",
+        )
+        result = _sr(returncode=1, stderr=CODEX_CONTEXT_EXHAUSTION_MARKER)
+        assert (
+            classify_infra_exit(session, result, capabilities=caps)
+            == InfraExitCategory.CONTEXT_EXHAUSTED
+        )
+
+
 @pytest.mark.parametrize("category", list(InfraExitCategory))
 def test_all_infra_categories_handled(category: InfraExitCategory) -> None:
     """Every InfraExitCategory value has a distinct test above."""