fix: stale RUNNING dispatch recovery with session-type-agnostic liveness (#4136)

## Summary

Interactive sessions have zero path to recover stale RUNNING dispatch
state. The bug is a convergence of two gaps: (1) `reset_dispatch` blocks
on RUNNING status without checking process liveness, and (2) neither
`open_kitchen` nor any interactive boot path runs the dispatch reaper.
The architectural weakness is that liveness verification before
status-based blocking is enforced by convention, not by structure —
three callers do it correctly (`resume_campaign_from_state`,
`_try_claim_with_liveness`, `reap_stale_dispatches`) while
`reset_dispatch` does not.

The immunity plan extracts a shared `resolve_stale_running` helper that
centralizes the "RUNNING → check liveness → classify if dead" pattern,
adds the reaper to `open_kitchen`, extends the existing AST boot-step
symmetry guard to cover `open_kitchen`, and adds integration tests that
would have caught this bug immediately.

The remediation plan replaces mocked `resolve_stale_running` in two
tests (`test_reset_dispatch_running_dead_process_succeeds` and
`test_reset_dispatch_running_alive_process_still_blocked`) with real
PID/boot-id fields on the dispatch record, exercising the full liveness
detection path: `reset_dispatch` → `find_locked_dispatch` →
`resolve_stale_running` → `is_dispatch_session_alive` →
`is_session_alive`.

Closes #4133

## Implementation Plan

Plan files:
-
`/home/talon/projects/autoskillit-runs/remediation-20260627-224639-242652/.autoskillit/temp/rectify/rectify_stale_running_dispatch_recovery_2026-06-27_224639.md`
-
`/home/talon/projects/autoskillit-runs/remediation-20260627-224639-242652/.autoskillit/temp/make-plan/remediation_stale_running_dispatch_recovery_plan_2026-06-28_001600.md`

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

## Token Usage Summary

| Step | Model | count | uncached | output | cache_read | peak_ctx |
turns | cache_write | time |

|------|-------|-------|----------|--------|------------|----------|-------|-------------|------|
| plan* | opus[1m] | 3 | 9.1k | 37.2k | 4.4M | 123.0k | 132 | 231.7k |
45m 36s |
| verify* | sonnet | 3 | 228 | 24.3k | 1.3M | 71.9k | 70 | 123.2k | 13m
37s |
| implement* | MiniMax-M3 | 3 | 185.8k | 16.1k | 2.5M | 60.5k | 140 | 0
| 7m 54s |
| retry_worktree* | sonnet | 1 | 76 | 3.6k | 314.5k | 40.7k | 24 | 22.2k
| 2m 2s |
| audit_impl* | sonnet | 3 | 756 | 16.1k | 723.1k | 42.9k | 46 | 79.7k |
10m 33s |
| dispatch:1133b9e6-5319-48e2-b878-5f30f979fbcb* | sonnet | 1 | 170 |
5.8k | 1.4M | 77.7k | 51 | 77.9k | 20m 52s |
| prepare_pr* | MiniMax-M3 | 2 | 85.9k | 4.6k | 386.8k | 0 | 29 | 0 | 1m
45s |
| compose_pr* | MiniMax-M3 | 2 | 73.1k | 4.5k | 430.1k | 38.5k | 25 | 0
| 1m 45s |
| review_pr* | sonnet | 4 | 482 | 46.6k | 2.7M | 59.5k | 155 | 161.1k |
13m 55s |
| dispatch:9f9aa203-f3ea-44fd-9d3d-8f7ce0265f6f* | sonnet | 1 | 298 |
10.5k | 2.8M | 84.3k | 77 | 59.8k | 1h 5m |
| merge_gate_fix* | sonnet | 1 | 110 | 5.5k | 527.3k | 47.6k | 35 |
29.5k | 2m 55s |
| **Total** | | | 356.0k | 174.8k | 17.4M | 123.0k | | 785.2k | 3h 6m |

\* *Step used a non-Anthropic provider; caching behavior may differ.*

## Token Efficiency

| Step | LoC Changed | cache_read/LoC | cache_write/LoC | output/LoC |
|------|-------------|----------------|-----------------|------------|
| plan | 0 | — | — | — |
| verify | 0 | — | — | — |
| implement | 16 | 156056.6 | 0.0 | 1008.7 |
| retry_worktree | 0 | — | — | — |
| audit_impl | 0 | — | — | — |
| dispatch:1133b9e6-5319-48e2-b878-5f30f979fbcb | 0 | — | — | — |
| prepare_pr | 0 | — | — | — |
| compose_pr | 0 | — | — | — |
| review_pr | 0 | — | — | — |
| dispatch:9f9aa203-f3ea-44fd-9d3d-8f7ce0265f6f | 0 | — | — | — |
| merge_gate_fix | 7 | 75335.4 | 4210.6 | 791.0 |
| **Total** | **23** | 755144.3 | 34140.5 | 7600.8 |

## Model Usage Breakdown

| Model | steps | uncached | output | cache_read | cache_write | time |
|-------|-------|----------|--------|------------|-------------|------|
| opus[1m] | 1 | 9.1k | 37.2k | 4.4M | 231.7k | 45m 36s |
| sonnet | 7 | 2.1k | 112.4k | 9.7M | 553.5k | 2h 9m |
| MiniMax-M3 | 3 | 344.7k | 25.3k | 3.3M | 0 | 11m 25s |

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Trecek

src/autoskillit/fleet/__init__.py

@@ -24,6 +24,7 @@
 from ._reset import (
     ResetReport,
     find_dispatch_in_campaigns,
+    find_locked_dispatch,
     format_resettable_statuses,
     reset_dispatch_artifacts,
     resolve_worktrees_dir,
@@ -81,6 +82,7 @@
     find_dispatch_for_issue,
     has_blocking_dispatch,
     has_completed_dispatch,
+    resolve_stale_running,
 )
 from .state_types import (
     _INFRASTRUCTURE_FAILURE_REASONS,  # noqa: F401
@@ -126,6 +128,7 @@
     "parse_campaign_summary",
     "serialize_campaign_summary",
     "validate_campaign_summary",
+    "resolve_stale_running",
     "TERMINAL_DISPATCH_STATUSES",
     "TERMINAL_UNCLEANED_STATUSES",
     "FLEET_HALTED_SENTINEL",
@@ -179,6 +182,7 @@
     "ResetReport",
     "compute_reset_labels",
     "find_dispatch_in_campaigns",
+    "find_locked_dispatch",
     "format_resettable_statuses",
     "reset_dispatch_artifacts",
     "resolve_worktrees_dir",

src/autoskillit/fleet/_dispatch_reaper.py

@@ -44,13 +44,9 @@ def _apply_stale_dispatch(
     m: CampaignStateMutator,
     reaper_dispatch_id: str = "",
 ) -> None:
-    from autoskillit.fleet import classify_stale_dispatch  # noqa: PLC0415
+    from autoskillit.fleet import resolve_stale_running  # noqa: PLC0415
 
-    new_status, sidecar = classify_stale_dispatch(dispatch)
-    dispatch.status = new_status
-    dispatch.reason = reason
-    if sidecar:
-        dispatch.sidecar_path = sidecar
+    resolve_stale_running(dispatch, m, reason=reason)
     dispatch.ended_at = time.time()
 
     dispatch.reaper_reason = reason
@@ -74,6 +70,7 @@ def _apply_stale_dispatch(
                 logger.warning("reaper: failed to write tombstone", exc_info=True)
 
     _append_reaper_event(dispatch, reason, reaper_dispatch_id)
+
     m.mark_dirty()
 
 

src/autoskillit/fleet/_reset.py

@@ -35,6 +35,7 @@
 __all__ = [
     "ResetReport",
     "find_dispatch_in_campaigns",
+    "find_locked_dispatch",
     "compute_reset_labels",
     "format_resettable_statuses",
     "reset_dispatch_artifacts",
@@ -80,6 +81,16 @@ def find_dispatch_in_campaigns(
     return None
 
 
+def find_locked_dispatch(dispatch_id: str, mutator: CampaignStateMutator) -> DispatchRecord | None:
+    """Locate a dispatch inside a locked mutator's state by dispatch_id or name."""
+    if mutator.state is None:
+        return None
+    for d in mutator.state.dispatches:
+        if (d.dispatch_id and d.dispatch_id == dispatch_id) or d.name == dispatch_id:
+            return d
+    return None
+
+
 def compute_reset_labels(target_state: IssueLabelState) -> tuple[list[str], list[str]]:
     label_def = LABEL_LIFECYCLE_REGISTRY[target_state]
     remove = sorted(s.value for s in label_def.removes_on_entry | {IssueLabelState.IN_PROGRESS})

src/autoskillit/fleet/state_recovery.py

@@ -4,7 +4,7 @@
 
 import time
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from autoskillit.core import (
     FleetErrorCode,
@@ -28,6 +28,9 @@
 
 MAX_CONSECUTIVE_RESUME_ATTEMPTS = 3
 
+if TYPE_CHECKING:
+    from autoskillit.fleet.state import CampaignStateMutator
+
 __all__ = [
     "classify_stale_dispatch",
     "derive_orchestrator_resume_spec",
@@ -36,6 +39,7 @@
     "has_blocking_dispatch",
     "has_completed_dispatch",
     "has_failed_dispatch",
+    "resolve_stale_running",
     "resume_campaign_from_state",
 ]
 
@@ -209,6 +213,38 @@ def classify_stale_dispatch(
     return (DispatchStatus.INTERRUPTED, "")
 
 
+def resolve_stale_running(
+    dispatch: DispatchRecord,
+    mutator: CampaignStateMutator,
+    *,
+    reason: str = "stale_running_resolved",
+) -> bool:
+    """Check whether a RUNNING dispatch is actually alive.
+
+    Returns True if the dispatch is confirmed alive (caller should block).
+    Returns False if the dispatch is dead — reclassifies it in-place within
+    the mutator (to INTERRUPTED or RESUMABLE via classify_stale_dispatch)
+    and marks the mutator dirty. Caller can proceed with the demoted status.
+
+    Precondition: dispatch.status == DispatchStatus.RUNNING.
+    Precondition: dispatch is an element of mutator.state.dispatches
+                  (not a copy from an unlocked read).
+    Precondition: called within a CampaignStateMutator context.
+    """
+    from autoskillit.fleet import is_dispatch_session_alive
+
+    if is_dispatch_session_alive(dispatch):
+        return True
+
+    new_status, sidecar_path = classify_stale_dispatch(dispatch)
+    dispatch.status = new_status
+    dispatch.reason = reason
+    if sidecar_path:
+        dispatch.sidecar_path = sidecar_path
+    mutator.mark_dirty()
+    return False
+
+
 def resume_campaign_from_state(
     state_path: Path,
     continue_on_failure: bool,
@@ -230,10 +266,9 @@ def resume_campaign_from_state(
     ResumeDecision with next_dispatch_name="" if all dispatches are
     complete or the campaign is halted.
     """
-    from autoskillit.fleet import is_dispatch_session_alive  # noqa: PLC0415
-    from autoskillit.fleet.state import (
-        CampaignStateMutator,  # noqa: PLC0415
-        _clear_dispatch_for_retry,  # noqa: PLC0415
+    from autoskillit.fleet.state import (  # noqa: PLC0415
+        CampaignStateMutator,
+        _clear_dispatch_for_retry,
     )
 
     with CampaignStateMutator(state_path) as m:
@@ -242,14 +277,7 @@ def resume_campaign_from_state(
 
         for d in m.state.dispatches:
             if d.status == DispatchStatus.RUNNING:
-                if is_dispatch_session_alive(d):
-                    continue
-                new_status, sidecar_path = classify_stale_dispatch(d)
-                d.status = new_status
-                d.reason = "stale_running_on_resume"
-                if sidecar_path:
-                    d.sidecar_path = sidecar_path
-                m.mark_dirty()
+                resolve_stale_running(d, m, reason="stale_running_on_resume")
 
         if continue_on_failure and reset_on_retry:
             for d in m.state.dispatches:

src/autoskillit/server/tools/tools_fleet_reset.py

@@ -13,14 +13,17 @@
 from autoskillit.core import FleetErrorCode, IssueLabelState, fleet_error, get_logger
 from autoskillit.fleet import (
     _RESETTABLE_STATUSES,
+    CampaignStateMutator,
     DispatchStatus,
     ResetReport,
     cleanup_orphaned_labels,
     compute_reset_labels,
     discover_campaign_state_files,
     find_dispatch_in_campaigns,
+    find_locked_dispatch,
     format_resettable_statuses,
     reset_dispatch_artifacts,
+    resolve_stale_running,
     resolve_worktrees_dir,
     update_campaign_state,
 )
@@ -123,11 +126,24 @@ async def reset_dispatch(
         dispatch, state_path = result
 
         if dispatch.status == DispatchStatus.RUNNING:
-            return fleet_error(
-                FleetErrorCode.FLEET_RESET_STILL_RUNNING,
-                f"Dispatch {dispatch_id!r} is still RUNNING. "
-                "Wait for it to finish or reap it first.",
-            )
+            with CampaignStateMutator(state_path) as m:
+                locked_dispatch = find_locked_dispatch(dispatch_id, m)
+                if locked_dispatch is None:
+                    return fleet_error(
+                        FleetErrorCode.FLEET_RESET_NOT_FOUND,
+                        f"No dispatch found matching {dispatch_id!r} in any campaign state file.",
+                    )
+                if locked_dispatch.status != DispatchStatus.RUNNING:
+                    dispatch = locked_dispatch
+                elif resolve_stale_running(locked_dispatch, m):
+                    return fleet_error(
+                        FleetErrorCode.FLEET_RESET_STILL_RUNNING,
+                        f"Dispatch {dispatch_id!r} is still RUNNING "
+                        f"(process {locked_dispatch.dispatched_pid} is alive). "
+                        "Wait for it to finish.",
+                    )
+                else:
+                    dispatch = locked_dispatch
 
         if dispatch.status not in _RESETTABLE_STATUSES:
             return fleet_error(

src/autoskillit/server/tools/tools_kitchen.py

@@ -39,7 +39,11 @@
     resolve_kitchen_id,
     unregister_active_kitchen,
 )
-from autoskillit.fleet import FleetSemaphore
+from autoskillit.fleet import (
+    FleetSemaphore,
+    discover_campaign_state_files,
+    reap_stale_dispatches_async,
+)
 from autoskillit.pipeline import create_background_task
 from autoskillit.server import mcp
 from autoskillit.server._guards import _backend_supports_quota, _require_orchestrator_exact
@@ -306,6 +310,17 @@ async def _open_kitchen_handler() -> str | None:
     except Exception:
         logger.warning("open_kitchen_registry_failed", exc_info=True)
 
+    try:
+        _campaign_state_paths = discover_campaign_state_files(ctx.project_dir)
+        if _campaign_state_paths:
+            await reap_stale_dispatches_async(
+                _campaign_state_paths,
+                min_reap_age_seconds=60.0,
+                heartbeat_grace_seconds=90.0,
+            )
+    except Exception:
+        logger.warning("open_kitchen_reap_failed", exc_info=True)
+
     ctx.gate_infrastructure_ready = True
     return None
 

tests/arch/AGENTS.md

@@ -36,6 +36,7 @@ AST enforcement, sub-package layer contracts, and architectural invariant tests.
 | `test_judge_eval_skill.py` | Structural integrity tests for the judge-eval skill |
 | `test_agent_prompt_structure.py` | Structural validation: agent definition MUST-gate fallback, quote-verification gates, inconclusive output format, and empty-array documentation |
 | `test_boot_step_symmetry.py` | AST guard: both boot functions (_fleet_auto_gate_boot, _food_truck_auto_gate_boot) must call sweep_stale_dispatch_labels |
+| `test_running_status_liveness_guard.py` | AST guard: DispatchStatus.RUNNING checks in server/tools/ must be guarded by liveness verification |
 | `test_bundled_recipes_split.py` | Enforcement: test_bundled_recipes.py split structure guard |
 | `test_cascade_map_guard.py` | REQ-GUARD-001..003, 005: CI guard validating cascade maps against AST-derived reverse import graph |
 | `test_channel_b_timeout_guard.py` | AST guard: Channel B tests must use timeout >= TimeoutTier.CHANNEL_B |

tests/arch/test_boot_step_symmetry.py

@@ -10,6 +10,9 @@
 pytestmark = [pytest.mark.layer("arch"), pytest.mark.small]
 
 LIFESPAN_PATH = Path(__file__).parents[2] / "src" / "autoskillit" / "server" / "_lifespan.py"
+KITCHEN_PATH = (
+    Path(__file__).parents[2] / "src" / "autoskillit" / "server" / "tools" / "tools_kitchen.py"
+)
 
 _REQUIRED_BOOT_STEPS: list[tuple[str, tuple[str, ...]]] = [
     ("sweep_stale_dispatch_labels", ("_fleet_auto_gate_boot", "_food_truck_auto_gate_boot")),
@@ -99,3 +102,18 @@ def test_boot_step_ordering(
                 f"{node.name}: {before_symbol} (line {before_lineno}) must appear "
                 f"before {after_symbol} (line {after_lineno})"
             )
+
+    def test_open_kitchen_handler_calls_reaper(self) -> None:
+        """AST guard: _open_kitchen_handler must call reap_stale_dispatches_async.
+
+        Interactive sessions need the reaper at kitchen-open time since they
+        never go through _fleet_auto_gate_boot or _food_truck_auto_gate_boot.
+        """
+        assert KITCHEN_PATH.exists(), f"Production file not found: {KITCHEN_PATH}"
+        tree = ast.parse(KITCHEN_PATH.read_text())
+        assert _function_body_contains_symbol(
+            tree, "_open_kitchen_handler", "reap_stale_dispatches_async"
+        ), (
+            "_open_kitchen_handler must call reap_stale_dispatches_async "
+            "to provide dispatch recovery for interactive sessions"
+        )