[FIX] Rectify: Fleet Dispatch Resume — Centralized Precondition Chokepoint (#4214)

## Summary

The new headless `fleet run --resume-session-id` CLI path (PR #4191)
inherits the dispatch state machine but bypasses the campaign-level
auto-reset precondition gate. When the prior dispatch is in `FAILURE`,
`INTERRUPTED`, or `REFUSED`, the headless path reads
`prior_session_chain` from the stale record, spawns a child subprocess,
then `mark_dispatch_running` raises `ValueError` on the illegal
`FAILURE→RUNNING` transition. The exception is swallowed in `_on_spawn`
(`fleet/_api.py:153`), the child runs for ~44 minutes on top of a stale
`FAILURE` record, and the eventual envelope reports `dispatch_status:
"failure"` while real work was completed.

Part A introduces a **single precondition chokepoint**
(`prepare_resume`) that every resume entry point must call before
spawning, **a fail-closed spawn path** that kills the child on
transition failure, and **`MAX_CONSECUTIVE_RESUME_ATTEMPTS` enforcement
on the headless path** so the existing campaign-level cap is not
silently bypassed. Together these make the immediate bug class
architecturally impossible rather than papering over the immediate
crash.

## Implementation Plan

Plan file:
`/home/talon/projects/generic_automation_mcp/.autoskillit/temp/rectify/rectify_fleet-resume-precondition-chokepoint_2026-07-08_143000.md`

Closes #4199

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

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: Trecek

pyproject.toml

@@ -365,6 +365,11 @@ ignore_imports = [
     # _dispatch_reaper calls kill_process_tree for orphan reaping; psutil is non-stdlib
     # so the reaper cannot live in core/. This mirrors the workspace path-utility exception.
     "autoskillit.fleet._dispatch_reaper -> autoskillit.execution",
+    # _write_pid (the _on_spawn callback) calls kill_process_tree to fail-closed-spawn
+    # when mark_dispatch_running raises an illegal transition. This is the IL-2 fleet-side
+    # counterpart to the reaper's kill primitive; the import is deferred (function-body)
+    # and deliberate. Do not expand _api's execution access beyond this single kill call.
+    "autoskillit.fleet._api -> autoskillit.execution",
     # _run_dispatch enforces config-authoritative ingredient injection at dispatch time.
     # apply_config_authoritative_overrides lives in config/ to share resolve_ingredient_defaults
     # without duplication. The import is deferred (function-body) and deliberate — it is the

src/autoskillit/cli/fleet/_fleet_run.py

@@ -19,8 +19,26 @@
 logger = get_logger(__name__)
 
 
-def _fleet_run_error(error: str, message: str, exit_code: int = 1) -> NoReturn:
-    envelope = {"success": False, "error": error, "user_visible_message": message}
+def _fleet_run_error(
+    error: str,
+    message: str,
+    exit_code: int = 1,
+    *,
+    dispatch_status: str | None = None,
+) -> NoReturn:
+    """Emit a CLI error envelope and exit.
+
+    Every envelope carries a ``dispatch_status`` field so downstream consumers
+    can distinguish crash outcomes from real dispatch outcomes without parsing
+    the message string. When ``dispatch_status`` is omitted, the default
+    ``"rejected"`` is used (no subprocess was launched).
+    """
+    envelope: dict[str, object] = {
+        "success": False,
+        "error": error,
+        "user_visible_message": message,
+        "dispatch_status": "rejected" if dispatch_status is None else dispatch_status,
+    }
     print(json.dumps(envelope))
     raise SystemExit(exit_code)
 

src/autoskillit/fleet/__init__.py

@@ -4,6 +4,7 @@
 ``autoskillit.fleet``, not from sub-modules.
 """
 
+from ._api import DispatchSpawnFailed as DispatchSpawnFailed
 from ._api import _build_capability_overrides as _build_capability_overrides
 from ._api import _write_pid as _write_pid
 from ._api import execute_dispatch
@@ -56,6 +57,7 @@
     DispatchStateHandle,
     DispatchStatus,
     GateRecordResult,
+    ResumeCountExceeded,
     ResumeDecision,
     append_dispatch_record,
     build_protected_campaign_ids,
@@ -75,13 +77,20 @@
     write_captured_values,
     write_initial_state,
 )
+from .state_recovery import (
+    MAX_CONSECUTIVE_RESUME_ATTEMPTS as MAX_CONSECUTIVE_RESUME_ATTEMPTS,
+)
+from .state_recovery import (
+    ResumePreflight as ResumePreflight,
+)
 from .state_recovery import (
     classify_stale_dispatch,
     derive_orchestrator_resume_spec,
     find_completed_dispatch,
     find_dispatch_for_issue,
     has_blocking_dispatch,
     has_completed_dispatch,
+    prepare_resume,
     resolve_stale_running,
 )
 from .state_types import (
@@ -138,9 +147,11 @@
     "DispatchRecord",
     "DispatchRejected",
     "DispatchResult",
+    "DispatchSpawnFailed",
     "DispatchStateHandle",
     "DispatchStatus",
     "DispatchOutcome",
+    "ResumeCountExceeded",
     "ResumeDecision",
     "GateRecordResult",
     "append_dispatch_record",
@@ -165,6 +176,9 @@
     "normalize_dispatch_token_usage",
     "classify_stale_dispatch",
     "FLEET_STATE_SCHEMA_VERSION",
+    "MAX_CONSECUTIVE_RESUME_ATTEMPTS",
+    "prepare_resume",
+    "ResumePreflight",
     "derive_orchestrator_resume_spec",
     "find_dispatch_for_issue",
     "checkpoint_from_sidecar",

src/autoskillit/fleet/_api.py

@@ -28,6 +28,7 @@
 from autoskillit.fleet._outcome import _checkpoint_to_dict, classify_dispatch_outcome
 from autoskillit.fleet.result_parser import parse_l3_result_block
 from autoskillit.fleet.state import DispatchStatus
+from autoskillit.fleet.state_recovery import ResumePreflight, prepare_resume
 from autoskillit.fleet.state_types import (
     DispatchCompleted,
     DispatchRejected,
@@ -45,6 +46,20 @@
 ENVELOPE_STDERR_MAX = 2000
 
 
+class DispatchSpawnFailed(RuntimeError):
+    """Spawn-time failure signal for the ``execute_dispatch`` callback path.
+
+    Attributes:
+        error_code: FleetErrorCode identifying the failure category.
+        message: Human-readable description of the spawn failure.
+    """
+
+    def __init__(self, error_code: FleetErrorCode, message: str) -> None:
+        super().__init__(message)
+        self.error_code = error_code
+        self.message = message
+
+
 @asynccontextmanager
 async def _dispatch_heartbeat(
     dispatches_dir: Path,
@@ -132,9 +147,28 @@ def _write_pid(
     dispatched_create_time: float = 0.0,
     identity_degraded: bool = False,
     issue_url: str = "",
-) -> None:
-    """on_spawn callback: atomically mark dispatch as running with dispatched_pid."""
+    *,
+    enforce_max_resume_attempts: bool = False,
+) -> str | None:
+    """on_spawn callback: atomically mark dispatch as running with dispatched_pid.
+
+    L2 — Fail-closed: if ``mark_dispatch_running`` raises (e.g. illegal state
+    transition), the spawned child is killed via ``kill_process_tree`` (the
+    canonical sync kill primitive used by ``_dispatch_reaper``) and the
+    exception's message string is returned to the caller via closure-scoped
+    state. Raising the exception from ``_on_spawn`` is NOT safe because
+    ``_execute_claude_headless`` catches runner exceptions and returns
+    ``SkillResult.crashed`` — the propagated exception would never reach the
+    outer ``execute_dispatch`` wrapper. The caller therefore inspects the
+    returned error string (or the closure-scoped ``_spawn_error`` list) and
+    translates it into a ``FLEET_L3_STARTUP_OR_CRASH`` envelope.
+
+    Returns:
+        None on success; the formatted error message string on failure (also
+        recorded via the side-effect of having killed the child).
+    """
     from autoskillit.core import read_boot_id
+    from autoskillit.execution import kill_process_tree
     from autoskillit.fleet import mark_dispatch_running
 
     try:
@@ -149,9 +183,23 @@ def _write_pid(
             sidecar_path=sidecar_path,
             identity_degraded=identity_degraded,
             issue_url=issue_url,
+            enforce_max_resume_attempts=enforce_max_resume_attempts,
         )
-    except Exception:
-        logger.warning("_write_pid: failed to mark dispatch running", exc_info=True)
+        return None
+    except Exception as exc:
+        # Fail-closed: kill the child before the state record can diverge.
+        if pid:
+            try:
+                kill_process_tree(pid, timeout=2.0)
+            except Exception:
+                logger.warning(
+                    "_write_pid: kill_process_tree failed for pid=%d",
+                    pid,
+                    exc_info=True,
+                )
+        cause = getattr(exc, "__cause__", None) or getattr(exc, "__context__", None)
+        cause_str = f" caused by {type(cause).__name__}: {cause}" if cause is not None else ""
+        return f"_on_spawn transition failed: {type(exc).__name__}: {exc}{cause_str}"
 
 
 def _post_dispatch_cleanup(
@@ -479,24 +527,60 @@ async def _run_dispatch(
     }
     prior_session_chain: list[str] = []
     prior_dispatched_session_id = ""
+    # Hoisted: closure-captured by _on_spawn to derive is_resume_branch for
+    # the enforce_max_resume_attempts kwarg on _write_pid. A non-None value
+    # means the resume branch executed prepare_resume (vs. fresh dispatch path).
+    preflight: ResumePreflight | None = None
     if resume_session_id and prior_dispatch_id:
         try:
             handle = DispatchStateHandle.open_continued(dispatches_dir, prior_dispatch_id)
             prior_state = read_state(handle.state_path)
-            if prior_state:
-                for d in prior_state.dispatches:
-                    if d.name == effective_name:
-                        if d.status == DispatchStatus.SUCCESS:
-                            logger.info(
-                                "resume_skipped_prior_success",
-                                dispatch_name=effective_name,
-                                prior_dispatch_id=prior_dispatch_id,
-                            )
-                            return _build_success_short_circuit(d, handle)
-                        prior_session_chain = list(d.session_chain)
-                        prior_dispatched_session_id = d.dispatched_session_id
-                        break
-        except (OSError, ValueError, KeyError, TypeError):
+            if prior_state is None:
+                # Corrupt or missing prior state — degrade to fresh.
+                handle = DispatchStateHandle.create_fresh(
+                    dispatches_dir,
+                    campaign_id,
+                    effective_name,
+                    "",
+                    [DispatchRecord(name=effective_name, caller_session_id=caller_session_id)],
+                    recipe_snapshot,
+                )
+            else:
+                # Single precondition chokepoint. Funnels every resume entry
+                # point through one function so the bug class becomes
+                # structurally impossible.
+                preflight = prepare_resume(
+                    handle.state_path,
+                    effective_name,
+                    continue_on_failure=True,
+                )
+                if preflight is not None:
+                    if preflight.short_circuit is not None:
+                        logger.info(
+                            "resume_skipped_prior_success",
+                            dispatch_name=effective_name,
+                            prior_dispatch_id=prior_dispatch_id,
+                        )
+                        return _build_success_short_circuit(preflight.short_circuit, handle)
+                    if preflight.halt:
+                        return DispatchResult(
+                            outcome=DispatchRejected(
+                                error_code=FleetErrorCode.FLEET_CAMPAIGN_HALTED,
+                                message=preflight.halted_reason
+                                or "Resume refused by precondition chokepoint",
+                                dispatch_id=handle.identity.dispatch_id,
+                            ),
+                            per_dispatch_state_path=None,
+                        )
+                    prior_session_chain = preflight.prior_session_chain
+                    prior_dispatched_session_id = preflight.prior_dispatched_session_id
+        except (OSError, KeyError, TypeError):
+            # OSError covers FileNotFoundError from DispatchStateHandle.open_continued
+            # when the prior state file is missing (fail-open to fresh dispatch) and
+            # write failures from reset_blocking_dispatch. ValueError is intentionally
+            # NOT caught so ResumeCountExceeded (a ValueError subclass raised by the
+            # cap check in mark_dispatch_running) propagates instead of being silently
+            # degraded to a create_fresh path.
             logger.warning("failed to read prior session chain from state", exc_info=True)
             handle = DispatchStateHandle.create_fresh(
                 dispatches_dir,
@@ -668,6 +752,9 @@ def _reject_with_state(error_code: FleetErrorCode, message: str) -> DispatchResu
     _dispatched_create_time: list[float] = []
     _dispatched_boot_id: list[str] = []
     _dispatched_session_id: list[str] = []
+    # Closure-scoped spawn error state (layer L2). See _write_pid docstring
+    # for why raising from on_spawn would not propagate.
+    _spawn_error: list[str] = []
 
     # Collect prior dispatch_ids from attempt_history for defense-in-depth parsing
     prior_ids: list[str] = []
@@ -705,7 +792,16 @@ def _on_spawn(pid: int, ticks: int) -> None:
         _dispatched_ticks.append(ticks)
         _dispatched_create_time.append(create_time)
         _dispatched_boot_id.append(boot_id)
-        _write_pid(
+        # Resume branch iff preflight was returned by prepare_resume above.
+        # Cap enforcement (MAX_CONSECUTIVE_RESUME_ATTEMPTS) lives one layer down
+        # in mark_dispatch_running.
+        is_resume_branch = preflight is not None
+        # _write_pid returns None on success or an error string on failure.
+        # We record the error in closure-scoped _spawn_error rather than
+        # raising — raising from on_spawn would be swallowed by the executor
+        # and converted to SkillResult.crashed, never reaching the outer
+        # execute_dispatch wrapper.
+        err = _write_pid(
             state_path,
             effective_name,
             dispatch_id,
@@ -715,7 +811,10 @@ def _on_spawn(pid: int, ticks: int) -> None:
             create_time,
             identity_degraded=(ticks == 0),
             issue_url=_issue_urls_raw,
+            enforce_max_resume_attempts=is_resume_branch,
         )
+        if err is not None:
+            _spawn_error.append(err)
 
     def _on_session_id(session_id: str) -> None:
         from autoskillit.fleet.state import mark_dispatch_session_identity
@@ -777,6 +876,16 @@ def _on_session_id(session_id: str) -> None:
                     else None,
                 )
 
+        # L2 fail-closed spawn gate: check closure-scoped error state.
+        # If _on_spawn recorded a transition failure (and killed the child
+        # via kill_process_tree), translate it to a structured envelope
+        # instead of letting the dispatch proceed on a stale record.
+        if _spawn_error:
+            return _reject_with_state(
+                FleetErrorCode.FLEET_L3_STARTUP_OR_CRASH,
+                _spawn_error[0],
+            )
+
         ended_at = max(time.time(), started_at + 1e-6)
         _dispatch_completed_normally = True
     except asyncio.CancelledError:

src/autoskillit/fleet/state.py

@@ -350,6 +350,16 @@ def _write_state(state_path: Path, state: CampaignState) -> None:
     write_versioned_json(state_path, payload, schema_version=FLEET_STATE_SCHEMA_VERSION)
 
 
+class ResumeCountExceeded(ValueError):
+    """Raised when ``mark_dispatch_running`` is invoked with ``enforce_max_resume_attempts=True``
+    and the dispatch has already accumulated ``MAX_CONSECUTIVE_RESUME_ATTEMPTS``
+    consecutive RESUMABLE entries in its ``attempt_history``.
+
+    Distinct from the ValueError raised by ``_validate_transition`` so callers
+    can differentiate state-machine violations from cap exhaustion.
+    """
+
+
 def mark_dispatch_running(
     state_path: Path,
     dispatch_name: str,
@@ -362,8 +372,22 @@ def mark_dispatch_running(
     sidecar_path: str | None = None,
     identity_degraded: bool = False,
     issue_url: str = "",
+    enforce_max_resume_attempts: bool = False,
 ) -> None:
-    """Atomically mark a dispatch as running with its dispatch_id and dispatched_pid."""
+    """Atomically mark a dispatch as running with its dispatch_id and dispatched_pid.
+
+    L3 — Resume-count cap on the headless path: when ``enforce_max_resume_attempts``
+    is True, the transition is gated by ``_count_consecutive_resumable_timeouts``
+    over the dispatch's ``attempt_history``. The cap check is computed from the
+    history (preserved across FAILURE→PENDING auto-resets by ``_RETRY_IDENTITY_FIELDS``),
+    so the existing campaign-level cap semantics are extended to the headless
+    path without skipping the cap on a reset-rewriting-the-status path.
+    """
+    from autoskillit.fleet.state_recovery import (  # noqa: PLC0415
+        MAX_CONSECUTIVE_RESUME_ATTEMPTS,
+        _count_consecutive_resumable_timeouts,
+    )
+
     with CampaignStateMutator(state_path) as m:
         if m.state is None:
             raise FileNotFoundError(f"State file not found or corrupted: {state_path}")
@@ -372,6 +396,13 @@ def mark_dispatch_running(
                 d.retry_reason = ""
                 d.infra_exit_category = ""
                 _validate_transition(d.status, DispatchStatus.RUNNING, d.name)
+                if enforce_max_resume_attempts:
+                    consecutive_timeouts = _count_consecutive_resumable_timeouts(d.attempt_history)
+                    if consecutive_timeouts >= MAX_CONSECUTIVE_RESUME_ATTEMPTS:
+                        raise ResumeCountExceeded(
+                            f"Resume cap ({MAX_CONSECUTIVE_RESUME_ATTEMPTS}) "
+                            f"exhausted for {dispatch_name!r}"
+                        )
                 was_resumable = d.status == DispatchStatus.RESUMABLE
                 d.status = DispatchStatus.RUNNING
                 if was_resumable: