fix: stale RUNNING dispatch recovery with session-type-agnostic liveness

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"
+        )