Rectify: Codex L2 Liveness Policy (#4217)

## Summary

Codex L2 food trucks can be killed by the raw stdout byte-idle watchdog
while doing legitimate long MCP work. The immediate failure is not just
a low timeout; it is an architectural mismatch between independent
timeout scalars and actual backend liveness.

Today, Codex builders copy `run_skill.stream_idle_timeout_ms` into
`CmdSpec.process_idle_timeout_ms`, and `_execute_claude_headless()`
treats that value as a lower cap on the outer AutoSkillit process idle
watchdog. For Codex, that collapses default L2 food-truck idle from
fleet's intended `1800s` to `600s`, while
`run_skill.mcp_tool_timeout_sec` and fleet session deadlines allow much
longer blocking calls. Codex `item.started` / `item.updated
status=in_progress` MCP records are ignored as liveness, so a healthy
in-flight MCP call can look identical to a hung stdout stream.

The immunity plan is to replace raw timeout-field propagation with one
resolved session liveness contract plus a process-level liveness
supervisor. The contract must be resolved once per backend and session
kind, then consumed by every watcher. Codex operation lifecycle events
must become first-class liveness inputs, bounded by MCP/session
deadlines rather than by an arbitrary marker-suppression cap.

## Implementation Plan

Plan file:
`/home/talon/projects/autoskillit-runs/remediation-20260708-205019-355728/.autoskillit/temp/rectify/rectify_codex_l2_liveness_policy_2026-07-08_211800.md`

Closes #4202

🤖 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
-->

Author: Trecek

.autoskillit/test-source-map.json

@@ -3753,6 +3753,9 @@
     "tests/server/test_tools_run_skill_retry.py",
     "tests/server/test_tools_status_kitchen.py"
   ],
+  "src/autoskillit/execution/headless/_headless_liveness.py": [
+    "tests/contracts/test_headless_liveness_contracts.py"
+  ],
   "src/autoskillit/execution/headless/_headless_path_tokens.py": [
     "tests/arch/test_pty_coherence.py",
     "tests/execution/test_api_error_signal_invariants.py",
@@ -3950,6 +3953,9 @@
     "tests/execution/test_termination_action.py",
     "tests/execution/test_termination_executor.py"
   ],
+  "src/autoskillit/execution/process/_liveness_supervisor.py": [
+    "tests/execution/process/test_liveness_supervisor.py"
+  ],
   "src/autoskillit/execution/process/_process_io.py": [
     "tests/execution/test_boundary_pty_dispatch.py",
     "tests/execution/test_linux_tracing_pty_integration.py",

src/autoskillit/core/__init__.pyi

@@ -287,6 +287,7 @@ from .types import IssueLabelState as IssueLabelState
 from .types import KillReason as KillReason
 from .types import LabelDef as LabelDef
 from .types import LensEntry as LensEntry
+from .types import LivenessSource as LivenessSource
 from .types import LoadReport as LoadReport
 from .types import LoadResult as LoadResult
 from .types import MarketplaceInstall as MarketplaceInstall
@@ -301,6 +302,8 @@ from .types import ModelTranslation as ModelTranslation
 from .types import NamedResume as NamedResume
 from .types import NdjsonDriftOutcome as NdjsonDriftOutcome
 from .types import NoResume as NoResume
+from .types import OperationLiveness as OperationLiveness
+from .types import OperationStatus as OperationStatus
 from .types import OutputFormat as OutputFormat
 from .types import OutputPatternResolver as OutputPatternResolver
 from .types import PackDef as PackDef
@@ -328,6 +331,7 @@ from .types import RetryReason as RetryReason
 from .types import ServeOverridesSnapshot as ServeOverridesSnapshot
 from .types import SessionCheckpoint as SessionCheckpoint
 from .types import SessionEvent as SessionEvent
+from .types import SessionLivenessSpec as SessionLivenessSpec
 from .types import SessionLocator as SessionLocator
 from .types import SessionOutcome as SessionOutcome
 from .types import SessionSkillManager as SessionSkillManager

src/autoskillit/core/types/AGENTS.md

@@ -27,7 +27,7 @@ Type re-export hub and all typed building blocks for the autoskillit package (IL
 | `_type_protocols_infra.py` | Protocols: `GateState`, `BackgroundSupervisor`, `FleetLock`, `QuotaRefreshTask`, `TokenFactory`, `CampaignProtector` |
 | `_type_protocols_backend.py` | Protocols: `StreamParser`, `ResultParser`, `EnvPolicy`, `SessionLocator`, `CodingAgentBackend` |
 | `_type_checkpoint.py` | `SessionCheckpoint` frozen dataclass and `compute_remaining()` helper for session resume |
-| `_type_backend.py` | `BackendCapabilities` frozen dataclass, `CLAUDE_CODE_CAPABILITIES` constant, `CmdSpec`, `SkillSessionConfig`, `ClaudeEventData`, `CodexEventData`, `SessionEvent`, `AgentSessionResult` |
+| `_type_backend.py` | `BackendCapabilities` frozen dataclass, `CLAUDE_CODE_CAPABILITIES` constant, `CmdSpec`, `SkillSessionConfig`, `OperationLiveness`, `SessionLivenessSpec`, `ClaudeEventData`, `CodexEventData`, `SessionEvent`, `AgentSessionResult` |
 | `_type_capture.py` | `CaptureEntrySpec` and `CaptureValueTypeError` for typed capture contract enforcement |
 | `_type_dispatch_identity.py` | `DispatchIdentity` frozen value object, `PromptContractError`, and `assert_prompt_sentinel` for sentinel contract enforcement |
 | `_type_helpers.py` | Text processing and skill-name extraction utilities |

src/autoskillit/core/types/_type_backend.py

@@ -9,7 +9,7 @@
 from typing import Any
 
 from ._type_checkpoint import SessionCheckpoint
-from ._type_enums import BackendEventKind, OutputFormat
+from ._type_enums import BackendEventKind, LivenessSource, OperationStatus, OutputFormat
 from ._type_plugin_source import PluginSource
 from ._type_results import ValidatedAddDir
 
@@ -25,6 +25,8 @@
     "CmdOrigin",
     "CmdSpec",
     "ModelTranslation",
+    "OperationLiveness",
+    "SessionLivenessSpec",
     "SkillSessionConfig",
     "ClaudeEventData",
     "CodexEventData",
@@ -357,6 +359,46 @@ class CodexEventData:
     command: str | None = None
 
 
+@dataclass(frozen=True, slots=True)
+class OperationLiveness:
+    """Liveness signal for a single in-flight backend operation."""
+
+    operation_id: str
+    item_type: str
+    status: OperationStatus
+    started_monotonic: float | None = None
+    updated_monotonic: float | None = None
+    raw: Mapping[str, Any] = field(default_factory=dict)
+
+
+@dataclass(frozen=True, slots=True)
+class SessionLivenessSpec:
+    """Resolved per-session liveness contract for process watchers."""
+
+    stdout_idle_timeout_sec: float | None
+    stale_threshold_sec: float
+    operation_deadline_sec: float
+    mcp_tool_timeout_sec: float
+    wall_timeout_sec: float
+    explicit_idle_disabled: bool
+    caller_session_id: str
+    authorized_sources: frozenset[LivenessSource] = field(
+        default_factory=lambda: frozenset(
+            {
+                LivenessSource.STDOUT_GROWTH,
+                LivenessSource.CHANNEL_B_GROWTH,
+                LivenessSource.EXECUTION_MARKER,
+                LivenessSource.OPERATION_IN_FLIGHT,
+            }
+        )
+    )
+
+    @property
+    def is_idle_disabled(self) -> bool:
+        """Whether the outer stdout-idle watchdog is disabled."""
+        return self.explicit_idle_disabled or self.stdout_idle_timeout_sec is None
+
+
 @dataclass(frozen=True, slots=True)
 class SessionEvent:
     """A single parsed event emitted by a running backend session."""
@@ -367,6 +409,7 @@ class SessionEvent:
     session_id: str | None = None
     exit_code: int | None = None
     backend_data: ClaudeEventData | CodexEventData | None = None
+    operation_liveness: OperationLiveness | None = None
 
 
 @dataclass(frozen=True, slots=True)

src/autoskillit/core/types/_type_enums.py

@@ -37,6 +37,8 @@
     "BackendEventKind",
     "CodexEventType",
     "CodexItemType",
+    "OperationStatus",
+    "LivenessSource",
     "SynthesisStrategy",
 ]
 
@@ -589,3 +591,34 @@ class SynthesisStrategy(StrEnum):
     ELECTRE_III = "electre_iii"
     DEX = "dex"
     CUSTOM = "custom"
+
+
+class OperationStatus(StrEnum):
+    """Lifecycle states for a backend operation (tool call, MCP call, etc.).
+
+    Emitted by backend stream parsers as first-class liveness signals. Used
+    by ``ProcessLivenessSupervisor`` to suppress byte-idle termination while
+    an in-flight operation is legitimately silent on stdout.
+    """
+
+    STARTED = "started"
+    PROGRESS = "progress"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+class LivenessSource(StrEnum):
+    """Authorized sources that count as legitimate liveness for process supervision.
+
+    Centralizes the liveness taxonomy so that no watcher can invent its own
+    predicate. The supervisor owns the decision; watchers only report their
+    raw observations and consult the supervisor.
+    """
+
+    STDOUT_GROWTH = "stdout_growth"
+    CHANNEL_B_GROWTH = "channel_b_growth"
+    EXECUTION_MARKER = "execution_marker"
+    CHILD_PROCESS_ACTIVITY = "child_process_activity"
+    API_CONNECTION = "api_connection"
+    OPERATION_IN_FLIGHT = "operation_in_flight"

src/autoskillit/execution/backends/AGENTS.md

@@ -37,3 +37,17 @@ IL-1 backend abstraction layer — concrete `CodingAgentBackend` implementations
 6. **Add a `FeatureDef`** — add an entry to `FEATURE_REGISTRY` in `core/types/_type_constants_features.py` with `default_enabled=False` and `requires_backend_alignment=True`.
 
 7. **Extend test coverage** — add tests to `tests/execution/backends/test_backend_registry.py`, `tests/contracts/test_backend_compliance.py`, and `tests/contracts/test_backend_protocol.py`.
+
+## Backend liveness contract
+
+- Backend stream parsers emit raw observations only. They must not decide whether a
+  process is healthy, stale, or killable.
+- Codex lifecycle records for long-running tool items (`item.started`, `item.updated`,
+  `item.completed`) must preserve the raw backend record and emit `OperationLiveness`
+  when `item.id` is stable and `item.type` is one of the long-running operation types.
+- `item.started` and `item.updated` remain `BackendEventKind.IGNORED`; they are liveness
+  signals, not result evidence, write evidence, or command/MCP completion evidence.
+- Parent stdout-idle behavior is owned by the resolved `SessionLivenessSpec` in
+  `execution/headless/_headless_liveness.py`. Backend builders may set child env hints,
+  but `_execute_claude_headless()` must normalize `AUTOSKILLIT_IDLE_OUTPUT_TIMEOUT`
+  from the resolved spec before invoking the runner.

src/autoskillit/execution/backends/_codex_parse.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import json
-from collections.abc import Sequence
+import time
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -16,12 +17,29 @@
     CodexEventData,
     CodexEventType,
     CodexItemType,
+    OperationLiveness,
+    OperationStatus,
     SessionEvent,
     fast_loads,
     get_logger,
 )
 from autoskillit.execution.process import _marker_is_standalone
 
+# Item-type enum members whose lifecycle records are eligible for
+# OperationLiveness emission. Membership drives liveness, not result
+# extraction. Membership must include every Codex item type that can produce
+# a legitimately silent stdout period while in-flight (long-running tool calls
+# or subagent delegations).
+_LIVENESS_ITEM_TYPES: frozenset[str] = frozenset(
+    {
+        CodexItemType.MCP_TOOL_CALL.value,
+        CodexItemType.FUNCTION_CALL.value,
+        CodexItemType.COMMAND_EXECUTION.value,
+        CodexItemType.COLLAB_TOOL_CALL.value,
+        CodexItemType.WEB_SEARCH.value,
+    }
+)
+
 logger = get_logger(__name__)
 
 
@@ -213,6 +231,80 @@ def _check_marker_text(self, text: str) -> None:
         if self.completion_marker and _marker_is_standalone(text, self.completion_marker):
             self._saw_marker = True
 
+    @staticmethod
+    def _extract_item_id(obj: Mapping[str, Any]) -> str:
+        """Return the stable ``item.id`` from a Codex NDJSON record, or ``""``.
+
+        The Codex NDJSON envelope places operation identity on ``item.id``
+        for ``item.started``/``item.updated``/``item.completed`` records.
+        Records without an id are skipped from OperationLiveness emission so
+        the supervisor never carries an invented correlation handle.
+        """
+        if not isinstance(obj, Mapping):
+            return ""
+        item = obj.get("item")
+        if isinstance(item, Mapping):
+            raw_id = item.get("id")
+            if isinstance(raw_id, str) and raw_id:
+                return raw_id
+        return ""
+
+    def _make_operation_liveness(
+        self,
+        obj: Mapping[str, Any],
+        *,
+        status: OperationStatus,
+    ) -> OperationLiveness | None:
+        """Build an ``OperationLiveness`` for long-running in-flight items.
+
+        Returns ``None`` when the record lacks a stable operation id or the
+        item type is not in ``_LIVENESS_ITEM_TYPES``. Callers MUST treat
+        ``None`` as "no liveness signal" — never invent an id to recover a
+        non-emission.
+        """
+        if not isinstance(obj, Mapping):
+            return None
+        op_id = self._extract_item_id(obj)
+        if not op_id:
+            return None
+        item = obj.get("item")
+        item_type: str = ""
+        if isinstance(item, Mapping):
+            raw_type = item.get("type")
+            if isinstance(raw_type, str):
+                item_type = raw_type
+        if item_type not in _LIVENESS_ITEM_TYPES:
+            return None
+        now = time.monotonic()
+        started = now if status == OperationStatus.STARTED else None
+        updated = now if status in (OperationStatus.PROGRESS, OperationStatus.COMPLETED) else None
+        return OperationLiveness(
+            operation_id=op_id,
+            item_type=item_type,
+            status=status,
+            started_monotonic=started,
+            updated_monotonic=updated,
+            raw=obj,
+        )
+
+    @staticmethod
+    def _make_item_event_data(
+        obj: Mapping[str, Any],
+        *,
+        record_type: str,
+    ) -> CodexEventData | None:
+        item = obj.get("item")
+        if not isinstance(item, Mapping):
+            return None
+        raw_type = item.get("type")
+        item_type = raw_type if isinstance(raw_type, str) else ""
+        return CodexEventData(
+            record_type=record_type,
+            thread_id="",
+            item_type=item_type,
+            raw=obj,
+        )
+
     def parse_line(self, line: str) -> SessionEvent | None:
         line = line.strip()
         if not line:
@@ -242,11 +334,25 @@ def parse_line(self, line: str) -> SessionEvent | None:
                 session_id=obj.get("payload", {}).get("id", "") or None,
             )
 
-        if event_type in (CodexEventType.TURN_STARTED, CodexEventType.ITEM_STARTED):
+        if event_type == CodexEventType.TURN_STARTED:
+            return SessionEvent(
+                kind=BackendEventKind.IGNORED,
+                is_terminal=False,
+                has_marker=False,
+            )
+
+        if event_type == CodexEventType.ITEM_STARTED:
+            # ``item.started`` may also emit OperationLiveness for long-running
+            # item types. We only emit when ``item["id"]`` is a non-empty
+            # stable string — inventing an id would break supervisor
+            # correlation and double-count lifecycle records.
+            op_liveness = self._make_operation_liveness(obj, status=OperationStatus.STARTED)
             return SessionEvent(
                 kind=BackendEventKind.IGNORED,
                 is_terminal=False,
                 has_marker=False,
+                backend_data=self._make_item_event_data(obj, record_type="item.started"),
+                operation_liveness=op_liveness,
             )
 
         if event_type == CodexEventType.ITEM_COMPLETED:
@@ -297,6 +403,7 @@ def parse_line(self, line: str) -> SessionEvent | None:
                 CodexItemType.COLLAB_TOOL_CALL,
                 CodexItemType.WEB_SEARCH,
             ):
+                op_liveness = self._make_operation_liveness(obj, status=OperationStatus.COMPLETED)
                 return SessionEvent(
                     kind=BackendEventKind.TOOL_OUTPUT,
                     is_terminal=False,
@@ -307,6 +414,7 @@ def parse_line(self, line: str) -> SessionEvent | None:
                         item_type=item_type.value,
                         raw=obj,
                     ),
+                    operation_liveness=op_liveness,
                 )
 
             if item_type in (CodexItemType.REASONING, CodexItemType.TODO_LIST):
@@ -365,10 +473,13 @@ def parse_line(self, line: str) -> SessionEvent | None:
             )
 
         if event_type == CodexEventType.ITEM_UPDATED:
+            op_liveness = self._make_operation_liveness(obj, status=OperationStatus.PROGRESS)
             return SessionEvent(
                 kind=BackendEventKind.IGNORED,
                 is_terminal=False,
                 has_marker=False,
+                backend_data=self._make_item_event_data(obj, record_type="item.updated"),
+                operation_liveness=op_liveness,
             )
 
         self.ndjson_unknown_event_count += 1

src/autoskillit/execution/backends/codex.py

@@ -815,7 +815,7 @@ def build_skill_session_cmd(
             env=env,
             cwd=cwd,
             is_resume=bool(resume_session_id),
-            process_idle_timeout_ms=stream_idle_timeout_ms,
+            process_idle_timeout_ms=0,
         )
 
     def build_food_truck_cmd(
@@ -920,7 +920,7 @@ def build_food_truck_cmd(
             env=env,
             cwd=cwd,
             is_resume=bool(resume_session_id),
-            process_idle_timeout_ms=stream_idle_timeout_ms,
+            process_idle_timeout_ms=0,
         )
 
     def build_interactive_cmd(

src/autoskillit/execution/headless/AGENTS.md

@@ -9,6 +9,7 @@ Headless Claude session orchestration — command prep, subprocess invocation, r
 | `__init__.py` | Facade: `run_headless_core()`, `DefaultHeadlessExecutor`, re-exports |
 | `_headless_helpers.py` | Session helpers: `_session_log_dir`, `_resolve_model`, `PostSessionMetrics` |
 | `_headless_execute.py` | Subprocess core: `_execute_claude_headless()` — shared skill/fleet path |
+| `_headless_liveness.py` | Resolves per-session liveness specs for parent process watchers |
 | `_headless_git.py` | Git LOC tracking: `_capture_git_head_sha()`, `_compute_loc_changed()` |
 | `_headless_path_tokens.py` | Path-token extraction and output-path validation from assistant messages |
 | `_headless_recovery.py` | Session recovery: `_recover_from_separate_marker`, `_synthesize_from_write_artifacts` |

src/autoskillit/execution/headless/__init__.py

@@ -510,4 +510,5 @@ async def dispatch_food_truck(
             model_identity=model_identity,
             on_session_id_resolved=on_session_id_resolved,
             step_backend=dispatch_backend,
+            _is_food_truck=True,
         )