fix: supervise headless sessions with operation leases and retain Codex errors

docs/audit/surface-freeze-checklist.md

@@ -174,7 +174,6 @@ autoskillit.core
 | 15 | `tests/execution/test_process_env_boundary.py` | `ClaudeHeadlessCmd` | Deferred (101) |
 | 16 | `tests/execution/backends/test_claude_code_backend.py` | `build_headless_cmd` | Deferred (47) |
 | 17 | `tests/execution/test_process_session_log_monitor_stale_suppression.py` | `_session_log_monitor`, `ChannelBStatus` | Module-level |
-| 18 | `tests/execution/test_process_session_log_monitor_dispatch_marker.py` | `_session_log_monitor`, `ChannelBStatus` | Module-level |
 
 **Additional indirect consumer:**
 - `tests/cli/test_cook_env_scrub.py` — imports `_MAX_MCP_OUTPUT_TOKENS_VALUE` from `autoskillit.execution` (package re-export path)

docs/design/forward-obligations/inspector-contract-impact.md

@@ -32,7 +32,7 @@ Already defined in `core/types/execution/_type_inspector.py`:
 
 - **`InspectorEvidence`** (lines 15–24, frozen dataclass): `idle_seconds`, `stdout_path`,
   `jsonl_lines`, `cpu_trend`, `rss_trend`, `connection_summary`,
-  `execution_marker_present`, `dispatch_context`
+  `operation_lease_active`, `dispatch_context`
 - **`InspectorVerdict`** (lines 27–32, frozen dataclass): `action`, `reasoning`,
   `confidence`, `elapsed_seconds`
 - **`InspectorCallback`** (line 35): `Callable[[InspectorEvidence], Awaitable[InspectorVerdict]]`

docs/developer/diagnostics.md

@@ -65,7 +65,7 @@ Logs are stored in a **global** directory (not per-project), so they persist acr
 
 ### Session Summary Fields
 
-`summary.json` contains: `session_id`, `dir_name`, `pid`, `cwd`, `evidence_worktree_path`, `evidence_worktree_source`, `evidence_worktree_detail`, `skill_command`, `success`, `subtype`, `exit_code`, `start_ts`, `snapshot_count`, `anomaly_count`, `peak_rss_kb`, `peak_oom_score`, `peak_fd_ratio`, `session_type`, `child_outcomes` (see [Child Terminal Reasons](#child-terminal-reasons)).
+`summary.json` contains: `session_id`, `dir_name`, `pid`, `cwd`, `evidence_worktree_path`, `evidence_worktree_source`, `evidence_worktree_detail`, `skill_command`, `success`, `subtype`, `exit_code`, `start_ts`, `snapshot_count`, `anomaly_count`, `peak_rss_kb`, `peak_oom_score`, `peak_fd_ratio`, `session_type`, `session_errors`, `child_outcomes` (see [Child Terminal Reasons](#child-terminal-reasons)).
 
 ### Execution-candidate manifests
 
@@ -178,7 +178,7 @@ Completed sessions with per-request evidence publish `turn_usage.jsonl` before
 version-4 token descriptor carries `turn_usage_file`, `turn_usage_count`, and
 `turn_usage_schema_version`. The file reference is `null` with count zero when
 no series was observed. The JSONL row schema is version 2; the retained
-`sessions.jsonl` index uses schema version 15.
+`sessions.jsonl` index uses schema version 16.
 
 Each token-bearing row carries non-empty `backend` and `provider_used`, plus
 nullable source `message_id`, `request_id`, `timestamp`, and observed `model`.

docs/execution/orchestration.md

@@ -28,6 +28,20 @@ The boundary is enforced three ways: FastMCP visibility, the
 `tools_execution/`. `run_skill` uses `_require_orchestrator_exact()`; `run_cmd` and
 `run_python` use `_require_orchestrator_or_higher()`. All three layers must agree.
 
+## Food-truck idle timeouts
+
+Food-truck supervision and the child tools use separate idle policies. The
+per-dispatch `idle_output_timeout` controls the outer supervisor. Its effective
+value is at least `run_skill.timeout`, so a zero idle setting still leaves that
+supervisor backstop enabled. Active operation leases refresh the supervisor's
+liveness clock until their deadlines expire.
+
+The child's `AUTOSKILLIT_IDLE_OUTPUT_TIMEOUT` comes from `fleet.idle_output_timeout`,
+falling back to `run_skill.idle_output_timeout` when the Fleet value is zero.
+An explicit value in `env_extras` takes precedence. A per-dispatch idle override
+does not change this child policy. Setting both configured idle values to zero
+omits the child idle override while retaining the outer supervisor floor.
+
 ## Recipe as a program
 
 A recipe (`recipe/schema.py:Recipe`) is a sequenced list of `RecipeStep`

scripts/generate_core_stub.py

@@ -54,6 +54,7 @@
     ".io",
     ".logging",
     ".pipeline._execution_marker",
+    ".plugins._operation_lease",
     ".pipeline._step_context",
     ".pipeline.pipeline_tracker",
     ".planset",

src/autoskillit/cli/app.py

@@ -31,22 +31,20 @@
 from autoskillit.cli.session._session_order import _recipes_dir_for
 from autoskillit.core import (
     AUDIT_ADMISSION_AUTHORITY_PATH_ENV_VAR,
+    OPERATION_LEASE_DIR_ENV_VAR,
     AuditAdmissionStoreAuthority,
     RecipeSource,
     atomic_write,
     get_logger,
 )
-from autoskillit.execution import _has_active_execution_marker
 
 if TYPE_CHECKING:
-    from autoskillit.core import ManagedWorkerCapacity, SkillResult
+    from autoskillit.core import InFlightOperationsProtocol, ManagedWorkerCapacity, SkillResult
     from autoskillit.migration import MigrationEngine, MigrationFile
     from autoskillit.workspace import SkillInfo
 
 logger = get_logger(__name__)
 
-logger = get_logger(__name__)
-
 app = App(
     name="autoskillit",
     help="MCP server for executing recipes with Claude Code.",
@@ -69,20 +67,32 @@
 
 
 def is_server_active(
-    marker_dir: Path | None,
     worker_capacity: ManagedWorkerCapacity | None,
+    in_flight_operations: InFlightOperationsProtocol,
 ) -> bool:
-    if worker_capacity is not None and worker_capacity.active_count > 0:
-        return True
-    if marker_dir is not None and _has_active_execution_marker(marker_dir):
-        return True
-    return False
+    return in_flight_operations.active_count > 0 or (
+        worker_capacity is not None and worker_capacity.active_count > 0
+    )
 
 
 class CliError(Exception):
     """Raised by CLI helpers to signal a user-facing error that should abort the command."""
 
 
+def _operation_lease_channel_from_env() -> Path | None:
+    value = os.environ.get(OPERATION_LEASE_DIR_ENV_VAR)
+    if value is None:
+        return None
+    try:
+        candidate = Path(value)
+        if candidate.is_absolute() and candidate.is_dir():
+            return candidate
+    except (OSError, ValueError):
+        pass
+    logger.warning("operation_lease_channel_rejected", value=value)
+    return None
+
+
 @app.default
 def serve(*, verbose: Annotated[bool, Parameter(name=["--verbose", "-v"])] = False):
     """Start the MCP server (default command)."""
@@ -165,23 +175,21 @@ def serve(*, verbose: Annotated[bool, Parameter(name=["--verbose", "-v"])] = Fal
         cfg,
         project_dir=_explicit_project_dir,
         audit_admission_store_authority=audit_admission_store_authority,
+        operation_lease_channel=_operation_lease_channel_from_env(),
         plugin_retirement_coordinator=default_plugin_retirement_coordinator(),
     )
     _initialize(ctx)
 
     try:
         backend_options = {"use_uvloop": sys.platform != "win32"}
 
-        _marker_dir: Path | None = (
-            ctx.backend.session_locator().project_log_dir(str(ctx.project_dir))
-            if ctx.backend is not None
-            else None
-        )
-
         async def _guarded_serve() -> None:
             await serve_with_signal_guard(
                 mcp,
-                activity_check=lambda: is_server_active(_marker_dir, ctx.worker_capacity),
+                activity_check=lambda: is_server_active(
+                    ctx.worker_capacity,
+                    ctx.in_flight_operations,
+                ),
             )
 
         anyio.run(

src/autoskillit/cli/session/_session_process.py

@@ -10,6 +10,7 @@
 import time
 from collections.abc import Callable, Iterator, Sequence
 from dataclasses import dataclass
+from functools import partial
 from pathlib import Path
 from typing import Final, TextIO
 from uuid import uuid4
@@ -20,6 +21,7 @@
 from autoskillit.cli.ui._terminal import terminal_guard, terminal_last_activity
 from autoskillit.config import ProcessTetherConfig
 from autoskillit.core import (
+    OPERATION_LEASE_DIR_ENV_VAR,
     SESSION_LIFETIME_NOTICE_ENV_VAR,
     CmdSpec,
     ProcessCleanupResult,
@@ -55,9 +57,11 @@
 )
 
 
-def _default_activity(pid: int, fd: int | None) -> frozenset[str]:
+def _default_activity(
+    pid: int, fd: int | None, *, operation_lease_dir: Path | None = None
+) -> frozenset[str]:
     """Read process liveness and recent kernel TTY activity."""
-    signals = set(_active_liveness_signals(pid, None, None))
+    signals = set(_active_liveness_signals(pid, operation_lease_dir=operation_lease_dir))
     if fd is not None:
         last_activity = terminal_last_activity(fd)
         if last_activity is not None and time.time() - last_activity <= _IDLE_WINDOW_SECONDS:
@@ -75,6 +79,7 @@ def __init__(
         clock: Callable[[], float] = time.monotonic,
         wall: Callable[[], float] = time.time,
         activity_probe: Callable[[int, int | None], frozenset[str]] | None = None,
+        operation_lease_dir: str | None = None,
     ) -> None:
         policy.validate()
         soft_seconds = policy.cook_ceiling_seconds
@@ -92,7 +97,14 @@ def __init__(
         self._hard_cap_seconds = hard_cap_seconds
         self._clock = clock
         self._wall = wall
-        self._activity_probe = _default_activity if activity_probe is None else activity_probe
+        channel = Path(operation_lease_dir) if operation_lease_dir else None
+        if channel is not None and not channel.is_absolute():
+            channel = None
+        self._activity_probe = (
+            partial(_default_activity, operation_lease_dir=channel)
+            if activity_probe is None
+            else activity_probe
+        )
         self._started_at: float | None = None
         self._soft_deadline: float | None = None
         self._hard_deadline: float | None = None
@@ -342,7 +354,9 @@ def run_cook_attempt(
     _require_posix_process_ownership()
     cwd = _canonical_cwd(spec.cwd)
     inherited_fds = _normalize_pass_fds(pass_fds)
-    life = InteractiveLifetime(lifetime)
+    life = InteractiveLifetime(
+        lifetime, operation_lease_dir=spec.env.get(OPERATION_LEASE_DIR_ENV_VAR)
+    )
     terminal_fd = _interactive_terminal_fd()
     notice_path = ensure_project_temp(Path(cwd)) / "session_lifetime" / f"{uuid4().hex}.json"
 

src/autoskillit/config/_automation_config.py

@@ -31,7 +31,6 @@
     _claude_mcp_timeout_coherence_gate,
     _codex_mcp_timeout_coherence_gate,
     _process_tether_coherence_gate,
-    _timeout_coherence_gate,
 )
 from autoskillit.config._dataclasses_diagnostics import (
     DiagnosticsConfig,
@@ -307,7 +306,6 @@ def sec(name: str) -> dict[str, Any]:
             raise ValueError(
                 "providers.execution_candidates requires the providers feature to be enabled"
             )
-        _timeout_coherence_gate(result.run_skill)
         _codex_mcp_timeout_coherence_gate(
             result.run_skill, result.fleet, tool_timeout=result.run_skill.mcp_tool_timeout_sec
         )

src/autoskillit/config/_coherence.py

@@ -1,11 +1,7 @@
 """Timeout-coherence gates for cross-cutting runtime configuration.
 
-Owns: ``_timeout_coherence_gate`` (idle_output_timeout vs known long-poll tools),
-``_codex_mcp_timeout_coherence_gate`` and ``_claude_mcp_timeout_coherence_gate``
-(MCP tool-timeout vs session-max), ``_process_tether_coherence_gate`` (orphan
-ceiling vs session-max), the helper ``compute_codex_mcp_tool_timeout``, and the
-shared time-limit constants ``_MERGE_QUEUE_DEFAULT``, ``_MERGE_QUEUE_RECIPE_MAX``,
-``_CI_WATCH_DEFAULT``.
+Owns the MCP tool-timeout and process-tether coherence gates plus the helper
+``compute_codex_mcp_tool_timeout``.
 
 The gates are warning-only: existing configs continue to work even when the gate
 fires. The implementation simply emits a structlog ``logger.warning`` so on-call
@@ -22,43 +18,6 @@
 
 logger = get_logger(__name__)
 
-# Known tool timeouts for coherence validation.
-# These are the maximum observed blocking durations for tools that may produce
-# zero stdout during execution — used to validate idle_output_timeout coherence.
-_MERGE_QUEUE_DEFAULT = 600
-_MERGE_QUEUE_RECIPE_MAX = 900
-_CI_WATCH_DEFAULT = 300
-
-
-def _timeout_coherence_gate(run_skill: RunSkillConfig) -> None:
-    """Warn when idle_output_timeout is too low relative to known long-polling tool durations.
-
-    The idle stall watchdog monitors raw stdout byte growth with no awareness of MCP tool
-    execution state. When idle_output_timeout <= a known tool's max duration, the watchdog
-    can fire and kill legitimate sessions that are simply waiting on a long poll.
-
-    This is a WARNING-only gate — existing configs continue working.
-    """
-    idle = run_skill.idle_output_timeout
-    if idle == 0:
-        return
-    if idle <= _MERGE_QUEUE_RECIPE_MAX:
-        logger.warning(
-            "idle_output_timeout_coherence",
-            idle_output_timeout=idle,
-            merge_queue_recipe_max=_MERGE_QUEUE_RECIPE_MAX,
-            merge_queue_default=_MERGE_QUEUE_DEFAULT,
-            ci_watch_default=_CI_WATCH_DEFAULT,
-            message=(
-                f"idle_output_timeout={idle}s is at or below the maximum known blocking tool "
-                f"duration ({_MERGE_QUEUE_RECIPE_MAX}s for wait_for_merge_queue recipe override). "
-                f"This creates a race condition where the idle stall watchdog fires before the "
-                f"long-polling tool returns. Consider raising idle_output_timeout to at least "
-                f"{_MERGE_QUEUE_RECIPE_MAX + 100}s, or set it to 0 to disable the watchdog "
-                f"for L2 food truck sessions."
-            ),
-        )
-
 
 def compute_codex_mcp_tool_timeout(
     run_skill: RunSkillConfig | None = None,

src/autoskillit/config/settings.py

@@ -41,15 +41,6 @@
 from autoskillit.config._coercion import (
     _preprocess_agent_backend as _preprocess_agent_backend,
 )
-from autoskillit.config._coherence import (
-    _CI_WATCH_DEFAULT as _CI_WATCH_DEFAULT,
-)
-from autoskillit.config._coherence import (
-    _MERGE_QUEUE_DEFAULT as _MERGE_QUEUE_DEFAULT,
-)
-from autoskillit.config._coherence import (
-    _MERGE_QUEUE_RECIPE_MAX as _MERGE_QUEUE_RECIPE_MAX,
-)
 from autoskillit.config._coherence import (
     _claude_mcp_timeout_coherence_gate as _claude_mcp_timeout_coherence_gate,
 )
@@ -59,9 +50,6 @@
 from autoskillit.config._coherence import (
     _process_tether_coherence_gate as _process_tether_coherence_gate,
 )
-from autoskillit.config._coherence import (
-    _timeout_coherence_gate as _timeout_coherence_gate,
-)
 from autoskillit.config._coherence import (
     compute_codex_mcp_tool_timeout as compute_codex_mcp_tool_timeout,
 )

src/autoskillit/core/__init__.pyi

@@ -226,6 +226,18 @@ from .plugins._active_kitchens import (
     sample_kitchen_process_identity as sample_kitchen_process_identity,
 )
 from .plugins._active_kitchens import unregister_active_kitchen as unregister_active_kitchen
+from .plugins._operation_lease import (
+    OPERATION_LEASE_FRESHNESS_SECONDS as OPERATION_LEASE_FRESHNESS_SECONDS,
+)
+from .plugins._operation_lease import (
+    OPERATION_LEASE_HEARTBEAT_SECONDS as OPERATION_LEASE_HEARTBEAT_SECONDS,
+)
+from .plugins._operation_lease import InFlightOperations as InFlightOperations
+from .plugins._operation_lease import OperationLeaseHandle as OperationLeaseHandle
+from .plugins._operation_lease import OperationLeaseRecord as OperationLeaseRecord
+from .plugins._operation_lease import current_operation_lease as current_operation_lease
+from .plugins._operation_lease import operation_lease as operation_lease
+from .plugins._operation_lease import read_active_operation_leases as read_active_operation_leases
 from .plugins._plugin_artifact_identity import (
     INSTALLED_PLUGIN_ARTIFACT_MANIFEST_FIELDS as INSTALLED_PLUGIN_ARTIFACT_MANIFEST_FIELDS,
 )
@@ -613,6 +625,7 @@ from .types import MAXIMUM_FIELDS as MAXIMUM_FIELDS
 from .types import MCP_CLIENT_BACKEND_ENV_VAR as MCP_CLIENT_BACKEND_ENV_VAR
 from .types import NATIVE_SHELL_CAPTURE_MODE_ENV_VAR as NATIVE_SHELL_CAPTURE_MODE_ENV_VAR
 from .types import NON_VARIADIC_CLAUDE_FLAGS as NON_VARIADIC_CLAUDE_FLAGS
+from .types import OPERATION_LEASE_DIR_ENV_VAR as OPERATION_LEASE_DIR_ENV_VAR
 from .types import (
     ORCHESTRATOR_FACING_INSTRUCTION_SURFACES as ORCHESTRATOR_FACING_INSTRUCTION_SURFACES,
 )
@@ -1018,6 +1031,7 @@ from .types import HostClientAttestation as HostClientAttestation
 from .types import IdempotencyExpiredEffect as IdempotencyExpiredEffect
 from .types import IdempotencyNamespace as IdempotencyNamespace
 from .types import IdempotencyRecord as IdempotencyRecord
+from .types import InFlightOperationsProtocol as InFlightOperationsProtocol
 from .types import InfraExitCategory as InfraExitCategory
 from .types import InfraOutcome as InfraOutcome
 from .types import InfrastructureFaultError as InfrastructureFaultError
@@ -1263,6 +1277,7 @@ from .types import SerializedTokenMeasure as SerializedTokenMeasure
 from .types import ServeOverridesSnapshot as ServeOverridesSnapshot
 from .types import SessionAttemptHandle as SessionAttemptHandle
 from .types import SessionCheckpoint as SessionCheckpoint
+from .types import SessionErrorOutcome as SessionErrorOutcome
 from .types import SessionEvent as SessionEvent
 from .types import SessionHookRoot as SessionHookRoot
 from .types import SessionInvariantAdaptationRefusal as SessionInvariantAdaptationRefusal