Skip to content

Commit 80f6551

Browse files
Trecekclaude
andcommitted
fix: stale RUNNING dispatch recovery for interactive sessions (#4133)
Extract resolve_stale_running helper to centralize the RUNNING→check liveness pattern. reset_dispatch now verifies process liveness before blocking RUNNING dispatches, open_kitchen runs reap_stale_dispatches_async for interactive sessions, and structural guards prevent future regressions. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 3d7a765 commit 80f6551

11 files changed

Lines changed: 344 additions & 26 deletions

src/autoskillit/fleet/__init__.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@
8181
find_dispatch_for_issue,
8282
has_blocking_dispatch,
8383
has_completed_dispatch,
84+
resolve_stale_running,
8485
)
8586
from .state_types import (
8687
_INFRASTRUCTURE_FAILURE_REASONS, # noqa: F401
@@ -126,6 +127,7 @@
126127
"parse_campaign_summary",
127128
"serialize_campaign_summary",
128129
"validate_campaign_summary",
130+
"resolve_stale_running",
129131
"TERMINAL_DISPATCH_STATUSES",
130132
"TERMINAL_UNCLEANED_STATUSES",
131133
"FLEET_HALTED_SENTINEL",

src/autoskillit/fleet/_dispatch_reaper.py

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -44,13 +44,9 @@ def _apply_stale_dispatch(
4444
m: CampaignStateMutator,
4545
reaper_dispatch_id: str = "",
4646
) -> None:
47-
from autoskillit.fleet import classify_stale_dispatch # noqa: PLC0415
47+
from autoskillit.fleet import resolve_stale_running # noqa: PLC0415
4848

49-
new_status, sidecar = classify_stale_dispatch(dispatch)
50-
dispatch.status = new_status
51-
dispatch.reason = reason
52-
if sidecar:
53-
dispatch.sidecar_path = sidecar
49+
resolve_stale_running(dispatch, m, reason=reason)
5450
dispatch.ended_at = time.time()
5551

5652
dispatch.reaper_reason = reason
@@ -74,7 +70,6 @@ def _apply_stale_dispatch(
7470
logger.warning("reaper: failed to write tombstone", exc_info=True)
7571

7672
_append_reaper_event(dispatch, reason, reaper_dispatch_id)
77-
m.mark_dirty()
7873

7974

8075
def _mark_dead_pid(

src/autoskillit/fleet/state_recovery.py

Lines changed: 41 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
import time
66
from pathlib import Path
7-
from typing import Any
7+
from typing import TYPE_CHECKING, Any
88

99
from autoskillit.core import (
1010
FleetErrorCode,
@@ -28,6 +28,9 @@
2828

2929
MAX_CONSECUTIVE_RESUME_ATTEMPTS = 3
3030

31+
if TYPE_CHECKING:
32+
from autoskillit.fleet.state import CampaignStateMutator
33+
3134
__all__ = [
3235
"classify_stale_dispatch",
3336
"derive_orchestrator_resume_spec",
@@ -36,6 +39,7 @@
3639
"has_blocking_dispatch",
3740
"has_completed_dispatch",
3841
"has_failed_dispatch",
42+
"resolve_stale_running",
3943
"resume_campaign_from_state",
4044
]
4145

@@ -209,6 +213,38 @@ def classify_stale_dispatch(
209213
return (DispatchStatus.INTERRUPTED, "")
210214

211215

216+
def resolve_stale_running(
217+
dispatch: DispatchRecord,
218+
mutator: CampaignStateMutator,
219+
*,
220+
reason: str = "stale_running_resolved",
221+
) -> bool:
222+
"""Check whether a RUNNING dispatch is actually alive.
223+
224+
Returns True if the dispatch is confirmed alive (caller should block).
225+
Returns False if the dispatch is dead — reclassifies it in-place within
226+
the mutator (to INTERRUPTED or RESUMABLE via classify_stale_dispatch)
227+
and marks the mutator dirty. Caller can proceed with the demoted status.
228+
229+
Precondition: dispatch.status == DispatchStatus.RUNNING.
230+
Precondition: dispatch is an element of mutator.state.dispatches
231+
(not a copy from an unlocked read).
232+
Precondition: called within a CampaignStateMutator context.
233+
"""
234+
from autoskillit.fleet import is_dispatch_session_alive
235+
236+
if is_dispatch_session_alive(dispatch):
237+
return True
238+
239+
new_status, sidecar_path = classify_stale_dispatch(dispatch)
240+
dispatch.status = new_status
241+
dispatch.reason = reason
242+
if sidecar_path:
243+
dispatch.sidecar_path = sidecar_path
244+
mutator.mark_dirty()
245+
return False
246+
247+
212248
def resume_campaign_from_state(
213249
state_path: Path,
214250
continue_on_failure: bool,
@@ -230,10 +266,9 @@ def resume_campaign_from_state(
230266
ResumeDecision with next_dispatch_name="" if all dispatches are
231267
complete or the campaign is halted.
232268
"""
233-
from autoskillit.fleet import is_dispatch_session_alive # noqa: PLC0415
234-
from autoskillit.fleet.state import (
235-
CampaignStateMutator, # noqa: PLC0415
236-
_clear_dispatch_for_retry, # noqa: PLC0415
269+
from autoskillit.fleet.state import ( # noqa: PLC0415
270+
CampaignStateMutator,
271+
_clear_dispatch_for_retry,
237272
)
238273

239274
with CampaignStateMutator(state_path) as m:
@@ -242,14 +277,7 @@ def resume_campaign_from_state(
242277

243278
for d in m.state.dispatches:
244279
if d.status == DispatchStatus.RUNNING:
245-
if is_dispatch_session_alive(d):
246-
continue
247-
new_status, sidecar_path = classify_stale_dispatch(d)
248-
d.status = new_status
249-
d.reason = "stale_running_on_resume"
250-
if sidecar_path:
251-
d.sidecar_path = sidecar_path
252-
m.mark_dirty()
280+
resolve_stale_running(d, m, reason="stale_running_on_resume")
253281

254282
if continue_on_failure and reset_on_retry:
255283
for d in m.state.dispatches:

src/autoskillit/server/tools/tools_fleet_reset.py

Lines changed: 34 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
find_dispatch_in_campaigns,
2222
format_resettable_statuses,
2323
reset_dispatch_artifacts,
24+
resolve_stale_running,
2425
resolve_worktrees_dir,
2526
update_campaign_state,
2627
)
@@ -123,11 +124,39 @@ async def reset_dispatch(
123124
dispatch, state_path = result
124125

125126
if dispatch.status == DispatchStatus.RUNNING:
126-
return fleet_error(
127-
FleetErrorCode.FLEET_RESET_STILL_RUNNING,
128-
f"Dispatch {dispatch_id!r} is still RUNNING. "
129-
"Wait for it to finish or reap it first.",
130-
)
127+
from autoskillit.fleet.state import CampaignStateMutator # circular-break
128+
129+
with CampaignStateMutator(state_path) as m:
130+
if m.state is None:
131+
return fleet_error(
132+
FleetErrorCode.FLEET_RESET_NOT_FOUND,
133+
f"No dispatch found matching {dispatch_id!r} in any campaign state file.",
134+
)
135+
locked_dispatch = next(
136+
(
137+
d
138+
for d in m.state.dispatches
139+
if (d.dispatch_id and d.dispatch_id == dispatch_id)
140+
or d.name == dispatch_id
141+
),
142+
None,
143+
)
144+
if locked_dispatch is None:
145+
return fleet_error(
146+
FleetErrorCode.FLEET_RESET_NOT_FOUND,
147+
f"No dispatch found matching {dispatch_id!r} in any campaign state file.",
148+
)
149+
if locked_dispatch.status != DispatchStatus.RUNNING:
150+
dispatch = locked_dispatch
151+
elif resolve_stale_running(locked_dispatch, m):
152+
return fleet_error(
153+
FleetErrorCode.FLEET_RESET_STILL_RUNNING,
154+
f"Dispatch {dispatch_id!r} is still RUNNING "
155+
f"(process {locked_dispatch.dispatched_pid} is alive). "
156+
"Wait for it to finish.",
157+
)
158+
else:
159+
dispatch = locked_dispatch
131160

132161
if dispatch.status not in _RESETTABLE_STATUSES:
133162
return fleet_error(

src/autoskillit/server/tools/tools_kitchen.py

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,11 @@
3939
resolve_kitchen_id,
4040
unregister_active_kitchen,
4141
)
42-
from autoskillit.fleet import FleetSemaphore
42+
from autoskillit.fleet import (
43+
FleetSemaphore,
44+
discover_campaign_state_files,
45+
reap_stale_dispatches_async,
46+
)
4347
from autoskillit.pipeline import create_background_task
4448
from autoskillit.server import mcp
4549
from autoskillit.server._guards import _backend_supports_quota, _require_orchestrator_exact
@@ -306,6 +310,17 @@ async def _open_kitchen_handler() -> str | None:
306310
except Exception:
307311
logger.warning("open_kitchen_registry_failed", exc_info=True)
308312

313+
try:
314+
_campaign_state_paths = discover_campaign_state_files(ctx.project_dir)
315+
if _campaign_state_paths:
316+
await reap_stale_dispatches_async(
317+
_campaign_state_paths,
318+
min_reap_age_seconds=60.0,
319+
heartbeat_grace_seconds=90.0,
320+
)
321+
except Exception:
322+
logger.warning("open_kitchen_reap_failed", exc_info=True)
323+
309324
ctx.gate_infrastructure_ready = True
310325
return None
311326

tests/arch/AGENTS.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ AST enforcement, sub-package layer contracts, and architectural invariant tests.
3636
| `test_judge_eval_skill.py` | Structural integrity tests for the judge-eval skill |
3737
| `test_agent_prompt_structure.py` | Structural validation: agent definition MUST-gate fallback, quote-verification gates, inconclusive output format, and empty-array documentation |
3838
| `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 |
39+
| `test_running_status_liveness_guard.py` | AST guard: DispatchStatus.RUNNING checks in server/tools/ must be guarded by liveness verification |
3940
| `test_bundled_recipes_split.py` | Enforcement: test_bundled_recipes.py split structure guard |
4041
| `test_cascade_map_guard.py` | REQ-GUARD-001..003, 005: CI guard validating cascade maps against AST-derived reverse import graph |
4142
| `test_channel_b_timeout_guard.py` | AST guard: Channel B tests must use timeout >= TimeoutTier.CHANNEL_B |

tests/arch/test_boot_step_symmetry.py

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@
1010
pytestmark = [pytest.mark.layer("arch"), pytest.mark.small]
1111

1212
LIFESPAN_PATH = Path(__file__).parents[2] / "src" / "autoskillit" / "server" / "_lifespan.py"
13+
KITCHEN_PATH = (
14+
Path(__file__).parents[2] / "src" / "autoskillit" / "server" / "tools" / "tools_kitchen.py"
15+
)
1316

1417
_REQUIRED_BOOT_STEPS: list[tuple[str, tuple[str, ...]]] = [
1518
("sweep_stale_dispatch_labels", ("_fleet_auto_gate_boot", "_food_truck_auto_gate_boot")),
@@ -99,3 +102,18 @@ def test_boot_step_ordering(
99102
f"{node.name}: {before_symbol} (line {before_lineno}) must appear "
100103
f"before {after_symbol} (line {after_lineno})"
101104
)
105+
106+
def test_open_kitchen_handler_calls_reaper(self) -> None:
107+
"""AST guard: _open_kitchen_handler must call reap_stale_dispatches_async.
108+
109+
Interactive sessions need the reaper at kitchen-open time since they
110+
never go through _fleet_auto_gate_boot or _food_truck_auto_gate_boot.
111+
"""
112+
assert KITCHEN_PATH.exists(), f"Production file not found: {KITCHEN_PATH}"
113+
tree = ast.parse(KITCHEN_PATH.read_text())
114+
assert _function_body_contains_symbol(
115+
tree, "_open_kitchen_handler", "reap_stale_dispatches_async"
116+
), (
117+
"_open_kitchen_handler must call reap_stale_dispatches_async "
118+
"to provide dispatch recovery for interactive sessions"
119+
)
Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
"""Arch guard: any function in server/tools/ that checks DispatchStatus.RUNNING
2+
must also verify liveness via is_dispatch_session_alive or resolve_stale_running.
3+
4+
Without this guard, new MCP tools can silently encode the bug pattern that
5+
caused issue #4133 — treating a status claim as a fact without verifying
6+
the owning process is alive.
7+
"""
8+
9+
from __future__ import annotations
10+
11+
import ast
12+
13+
import pytest
14+
15+
pytestmark = [pytest.mark.layer("arch"), pytest.mark.small]
16+
17+
_LIVENESS_HELPERS: frozenset[str] = frozenset(
18+
{"is_dispatch_session_alive", "resolve_stale_running"}
19+
)
20+
21+
22+
class _RunningStatusChecker(ast.NodeVisitor):
23+
"""Find DispatchStatus.RUNNING comparisons and verify liveness helper usage."""
24+
25+
def __init__(self, fn_node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
26+
self.fn_node = fn_node
27+
self.running_check_linenos: list[int] = []
28+
self.has_liveness_helper = False
29+
30+
def visit_Compare(self, node: ast.Compare) -> None:
31+
if self._is_running_status_check(node):
32+
self.running_check_linenos.append(node.lineno)
33+
self.generic_visit(node)
34+
35+
def visit_Call(self, node: ast.Call) -> None:
36+
if self._is_liveness_helper_call(node):
37+
self.has_liveness_helper = True
38+
self.generic_visit(node)
39+
40+
def _is_running_status_check(self, node: ast.Compare) -> bool:
41+
if not isinstance(node.left, ast.Attribute):
42+
return False
43+
if node.left.attr != "status":
44+
return False
45+
for comparator in node.comparators:
46+
if not isinstance(comparator, ast.Attribute):
47+
continue
48+
if comparator.attr != "RUNNING":
49+
continue
50+
if not isinstance(node.ops[0], (ast.Eq, ast.Is)):
51+
continue
52+
return True
53+
return False
54+
55+
def _is_liveness_helper_call(self, node: ast.Call) -> bool:
56+
if isinstance(node.func, ast.Name) and node.func.id in _LIVENESS_HELPERS:
57+
return True
58+
if isinstance(node.func, ast.Attribute) and node.func.attr in _LIVENESS_HELPERS:
59+
return True
60+
return False
61+
62+
63+
def _get_function_defs(tree: ast.Module) -> list[ast.FunctionDef | ast.AsyncFunctionDef]:
64+
fns: list[ast.FunctionDef | ast.AsyncFunctionDef] = []
65+
for node in ast.walk(tree):
66+
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
67+
fns.append(node)
68+
return fns
69+
70+
71+
def test_running_status_checks_require_liveness():
72+
"""Every DispatchStatus.RUNNING check in server/tools/ must be guarded
73+
by a liveness verification (is_dispatch_session_alive or resolve_stale_running).
74+
"""
75+
from autoskillit.core import paths
76+
77+
tools_dir = paths.pkg_root() / "server" / "tools"
78+
assert tools_dir.is_dir(), f"server/tools/ directory not found at {tools_dir}"
79+
80+
violations: list[str] = []
81+
for py_file in sorted(tools_dir.glob("*.py")):
82+
try:
83+
tree = ast.parse(py_file.read_text())
84+
except SyntaxError:
85+
continue
86+
87+
for fn in _get_function_defs(tree):
88+
checker = _RunningStatusChecker(fn)
89+
checker.visit(fn)
90+
91+
if not checker.running_check_linenos:
92+
continue
93+
94+
if checker.has_liveness_helper:
95+
continue
96+
97+
relpath = str(py_file.relative_to(paths.pkg_root()))
98+
for lineno in checker.running_check_linenos:
99+
violations.append(
100+
f"{relpath}:{lineno}{fn.name}() checks "
101+
"DispatchStatus.RUNNING without verifying liveness. "
102+
f"Use {' or '.join(sorted(_LIVENESS_HELPERS))} "
103+
"before treating RUNNING as a blocking fact."
104+
)
105+
106+
assert not violations, "RUNNING status checks without liveness verification:\n" + "\n".join(
107+
f" {v}" for v in violations
108+
)

tests/server/conftest.py

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -152,6 +152,28 @@ async def _noop(*_args, **_kwargs):
152152
)
153153

154154

155+
@pytest.fixture(autouse=True)
156+
def _patch_kitchen_reaper(monkeypatch):
157+
"""Neutralize reaper calls inside _open_kitchen_handler for unit tests.
158+
159+
_open_kitchen_handler now calls discover_campaign_state_files and
160+
reap_stale_dispatches_async. Existing kitchen handler tests rely on
161+
filesystem absence to make these no-ops; this fixture makes that
162+
guarantee explicit and stable regardless of host filesystem state.
163+
"""
164+
monkeypatch.setattr(
165+
"autoskillit.server.tools.tools_kitchen.discover_campaign_state_files",
166+
lambda _project_dir: [],
167+
)
168+
169+
async def _noop_reaper(*_args, **_kwargs):
170+
return None
171+
172+
monkeypatch.setattr(
173+
"autoskillit.server.tools.tools_kitchen.reap_stale_dispatches_async", _noop_reaper
174+
)
175+
176+
155177
@pytest.fixture
156178
def build_ctx(tmp_path):
157179
"""Factory: build_ctx(**overrides) → minimal ToolContext with overrides applied."""

0 commit comments

Comments
 (0)