Skip to content

Commit 0d13966

Browse files
Trecekclaude
andauthored
Implementation Plan: Separate reset_dispatch State Reset from Artifact Destruction (#4115)
## Summary Refactor `reset_dispatch` to decouple internal state reset from external artifact destruction by adding a `destroy_artifacts` parameter (default `False`). Promote `issue_url` to an identity field so it survives retry resets. Extend `find_dispatch_for_issue` to search PENDING dispatches with a staleness guard to prevent double-dispatch races. These changes make `force=true` orthogonal to artifact preservation — `force` bypasses the resume gate, `destroy_artifacts` controls whether worktrees/branches/PRs are removed. Campaign-level artifact sweep is deferred to a follow-up task. ## Implementation Plan Plan file: `/home/talon/projects/autoskillit-runs/impl-20260626-092004-754802/.autoskillit/temp/make-plan/reset_dispatch_preserve_state_plan_2026-06-26_093000.md` Closes #4114 🤖 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] | 1 | 5.6k | 20.4k | 1.2M | 110.3k | 45 | 82.6k | 22m 11s | | review_approach* | sonnet | 1 | 6.6k | 6.2k | 222.7k | 54.9k | 16 | 40.4k | 8m 55s | | verify* | opus[1m] | 1 | 66 | 15.7k | 2.2M | 100.2k | 55 | 81.5k | 10m 33s | | **Total** | | | 12.3k | 42.3k | 3.7M | 110.3k | | 204.6k | 41m 40s | \* *Step used a non-Anthropic provider; caching behavior may differ.* ## Model Usage Breakdown | Model | steps | uncached | output | cache_read | cache_write | time | |-------|-------|----------|--------|------------|-------------|------| | opus[1m] | 2 | 5.7k | 36.1k | 3.5M | 164.1k | 32m 45s | | sonnet | 1 | 6.6k | 6.2k | 222.7k | 40.4k | 8m 55s | --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent bb32cde commit 0d13966

8 files changed

Lines changed: 424 additions & 22 deletions

File tree

src/autoskillit/fleet/state_recovery.py

Lines changed: 32 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
from __future__ import annotations
44

5+
import time
56
from pathlib import Path
67
from typing import Any
78

@@ -58,6 +59,8 @@
5859
}
5960
)
6061

62+
_PENDING_QUIET_PERIOD_SECONDS: float = 60.0
63+
6164

6265
def _count_consecutive_resumable_timeouts(history: list[dict[str, Any]]) -> int:
6366
"""Count consecutive RESUMABLE entries with retriable reasons from the tail."""
@@ -332,6 +335,7 @@ def find_dispatch_for_issue(
332335
333336
Pass 1: RUNNING dispatches (live session may own the label).
334337
Pass 2: terminal dispatches (FAILURE, INTERRUPTED) with labels_cleaned=False.
338+
Pass 3: PENDING dispatches with a stale attempt_history entry and no active session.
335339
336340
RUNNING takes priority — a live session should not be preempted by an old dead dispatch.
337341
Returns the first matching DispatchRecord, else None. Reads are filesystem-only.
@@ -375,7 +379,34 @@ def find_dispatch_for_issue(
375379
entries = read_sidecar_from_path(Path(d.sidecar_path)).entries
376380
if any(e.issue_url == issue_url for e in entries):
377381
terminal_match = d
378-
return terminal_match
382+
383+
if terminal_match is not None:
384+
return terminal_match
385+
386+
for state_path in campaign_state_paths:
387+
try:
388+
state = read_state(state_path)
389+
except Exception:
390+
logger.warning("Failed to read campaign state from %s", state_path, exc_info=True)
391+
continue
392+
if state is None:
393+
continue
394+
for d in state.dispatches:
395+
if d.status != DispatchStatus.PENDING:
396+
continue
397+
if not d.issue_url or d.issue_url != issue_url:
398+
continue
399+
if d.labels_cleaned:
400+
continue
401+
if d.dispatched_session_id:
402+
continue
403+
if not d.attempt_history:
404+
continue
405+
last_attempt = d.attempt_history[-1]
406+
ended_at = last_attempt.get("ended_at", 0.0)
407+
if not ended_at or (time.time() - ended_at) > _PENDING_QUIET_PERIOD_SECONDS:
408+
return d
409+
return None
379410

380411

381412
def derive_orchestrator_resume_spec(state: CampaignState) -> NamedResume | NoResume:

src/autoskillit/fleet/state_types.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
"attempt_history",
2626
"session_chain",
2727
"resume_count",
28+
"issue_url",
2829
}
2930
)
3031

src/autoskillit/server/tools/_claim_helpers.py

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
from autoskillit.fleet import (
1111
TERMINAL_UNCLEANED_STATUSES,
1212
CampaignStateMutator,
13+
DispatchStatus,
1314
cleanup_orphaned_labels,
1415
discover_campaign_state_files,
1516
find_dispatch_for_issue,
@@ -95,6 +96,14 @@ async def _try_claim_with_liveness(
9596
if cleaned:
9697
_mark_dispatch_labels_cleaned(dispatch.name, campaign_state_paths)
9798
return ClaimDecision(claimed=True, stale_label_cleaned=cleaned)
99+
if dispatch.status == DispatchStatus.PENDING:
100+
return ClaimDecision(
101+
claimed=False,
102+
reason=(
103+
f"Issue #{issue_number} already has '{effective_label}' label"
104+
" — a reset dispatch is pending retry"
105+
),
106+
)
98107
if is_dispatch_session_alive(dispatch):
99108
return ClaimDecision(
100109
claimed=False,

src/autoskillit/server/tools/tools_fleet_reset.py

Lines changed: 39 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,9 @@
1414
from autoskillit.fleet import (
1515
_RESETTABLE_STATUSES,
1616
DispatchStatus,
17+
ResetReport,
18+
cleanup_orphaned_labels,
19+
compute_reset_labels,
1720
discover_campaign_state_files,
1821
find_dispatch_in_campaigns,
1922
format_resettable_statuses,
@@ -74,17 +77,20 @@ async def reset_dispatch(
7477
dispatch_id: str,
7578
reset_to: str = "queued",
7679
force: bool = False,
80+
destroy_artifacts: bool = False,
7781
ctx: Context = CurrentContext(),
7882
) -> str:
79-
"""Reset a failed L2 dispatch, cleaning up all git/PR artifacts.
83+
"""Reset a failed L2 dispatch for retry or abandonment.
8084
81-
Removes the local worktree, local and remote branches, closes any open PRs,
82-
and resets issue labels. Use after a resume attempt fails or is declined.
85+
By default, resets internal state only (status → PENDING) while preserving
86+
git artifacts (worktree, branches, PRs). Issue labels are always swapped.
8387
8488
Args:
8589
dispatch_id: The dispatch ID (UUID) or dispatch name to reset.
8690
reset_to: Target state — "queued" (fresh retry) or "fail" (abandon).
8791
force: Bypass the resume-attempt gate when resume is known to be impossible.
92+
destroy_artifacts: When True, also remove the git worktree, delete local/remote
93+
branches, and close open PRs. Default False for safe error recovery.
8894
8995
Never raises.
9096
"""
@@ -134,21 +140,36 @@ async def reset_dispatch(
134140
worktrees_dir = resolve_worktrees_dir(project_dir, cfg.workspace.worktree_root)
135141
target_state = _VALID_RESET_TARGETS[reset_to]
136142

137-
if tool_ctx.runner is None:
138-
return fleet_error(
139-
FleetErrorCode.FLEET_L3_STARTUP_OR_CRASH,
140-
"No subprocess runner available in this session.",
143+
if destroy_artifacts:
144+
if tool_ctx.runner is None:
145+
return fleet_error(
146+
FleetErrorCode.FLEET_L3_STARTUP_OR_CRASH,
147+
"No subprocess runner available in this session.",
148+
)
149+
150+
report = await reset_dispatch_artifacts(
151+
dispatch,
152+
project_dir=project_dir,
153+
worktrees_dir=worktrees_dir,
154+
runner=tool_ctx.runner,
155+
github_client=tool_ctx.github_client,
156+
target_state=target_state,
157+
force=force,
141158
)
142-
143-
report = await reset_dispatch_artifacts(
144-
dispatch,
145-
project_dir=project_dir,
146-
worktrees_dir=worktrees_dir,
147-
runner=tool_ctx.runner,
148-
github_client=tool_ctx.github_client,
149-
target_state=target_state,
150-
force=force,
151-
)
159+
else:
160+
report = ResetReport(
161+
dispatch_name=dispatch.name,
162+
branch_name=dispatch.branch_name or dispatch.name,
163+
)
164+
remove_labels, add_labels = compute_reset_labels(target_state)
165+
labels_ok = await cleanup_orphaned_labels(
166+
dispatch.sidecar_path,
167+
tool_ctx.github_client,
168+
issue_url=dispatch.issue_url,
169+
remove_labels=remove_labels,
170+
add_labels=add_labels,
171+
)
172+
report.labels_reset = labels_ok
152173

153174
state_updated = await update_campaign_state(
154175
dispatch.name,
@@ -178,6 +199,7 @@ async def reset_dispatch(
178199
"has_protected_artifacts": report.has_protected_artifacts,
179200
"protected_prs": report.protected_prs,
180201
"reset_to": reset_to,
202+
"destroy_artifacts": destroy_artifacts,
181203
}
182204
)
183205
except Exception:

tests/fleet/test_find_dispatch_for_issue.py

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
from __future__ import annotations
44

55
import json
6+
import time
67
from dataclasses import asdict
78
from pathlib import Path
89

@@ -239,3 +240,151 @@ def test_running_dispatch_takes_priority_over_failure(tmp_path):
239240

240241
assert result is not None
241242
assert result.name == "task-run"
243+
244+
245+
# --- PENDING dispatch search (Pass 3) ---
246+
247+
248+
def test_pending_dispatch_stale_match(tmp_path):
249+
"""PENDING dispatch with stale attempt_history is returned."""
250+
sp = tmp_path / "campaign.json"
251+
d = DispatchRecord(
252+
name="d1",
253+
status=DispatchStatus.PENDING,
254+
issue_url=_ISSUE_URL,
255+
labels_cleaned=False,
256+
dispatched_session_id="",
257+
attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
258+
)
259+
_write_state(sp, [d])
260+
261+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
262+
263+
assert result is not None
264+
assert result.name == "d1"
265+
266+
267+
def test_pending_dispatch_too_fresh_skipped(tmp_path):
268+
"""PENDING dispatch within quiet period is skipped."""
269+
sp = tmp_path / "campaign.json"
270+
d = DispatchRecord(
271+
name="d1",
272+
status=DispatchStatus.PENDING,
273+
issue_url=_ISSUE_URL,
274+
labels_cleaned=False,
275+
dispatched_session_id="",
276+
attempt_history=[{"ended_at": time.time() - 5, "status": "failure"}],
277+
)
278+
_write_state(sp, [d])
279+
280+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
281+
282+
assert result is None
283+
284+
285+
def test_pending_dispatch_no_history_skipped(tmp_path):
286+
"""PENDING dispatch with empty attempt_history is skipped (never dispatched)."""
287+
sp = tmp_path / "campaign.json"
288+
d = DispatchRecord(
289+
name="d1",
290+
status=DispatchStatus.PENDING,
291+
issue_url=_ISSUE_URL,
292+
labels_cleaned=False,
293+
dispatched_session_id="",
294+
attempt_history=[],
295+
)
296+
_write_state(sp, [d])
297+
298+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
299+
300+
assert result is None
301+
302+
303+
def test_pending_dispatch_with_session_id_skipped(tmp_path):
304+
"""PENDING dispatch with active dispatched_session_id is skipped."""
305+
sp = tmp_path / "campaign.json"
306+
d = DispatchRecord(
307+
name="d1",
308+
status=DispatchStatus.PENDING,
309+
issue_url=_ISSUE_URL,
310+
labels_cleaned=False,
311+
dispatched_session_id="abc",
312+
attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
313+
)
314+
_write_state(sp, [d])
315+
316+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
317+
318+
assert result is None
319+
320+
321+
def test_running_beats_pending(tmp_path):
322+
"""RUNNING dispatch wins over stale PENDING dispatch."""
323+
sidecar = tmp_path / "sidecar.jsonl"
324+
_write_sidecar(sidecar, [IssueSidecarEntry(issue_url=_ISSUE_URL, status="completed", ts=_TS)])
325+
running = DispatchRecord(
326+
name="task-run",
327+
status=DispatchStatus.RUNNING,
328+
sidecar_path=str(sidecar),
329+
)
330+
pending = DispatchRecord(
331+
name="task-pending",
332+
status=DispatchStatus.PENDING,
333+
issue_url=_ISSUE_URL,
334+
labels_cleaned=False,
335+
dispatched_session_id="",
336+
attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
337+
)
338+
sp = tmp_path / "campaign.json"
339+
_write_state(sp, [pending, running])
340+
341+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
342+
343+
assert result is not None
344+
assert result.name == "task-run"
345+
346+
347+
def test_terminal_beats_pending(tmp_path):
348+
"""Terminal FAILURE wins over stale PENDING dispatch."""
349+
sidecar = tmp_path / "sidecar.jsonl"
350+
_write_sidecar(sidecar, [IssueSidecarEntry(issue_url=_ISSUE_URL, status="failed", ts=_TS)])
351+
failure = DispatchRecord(
352+
name="task-fail",
353+
status=DispatchStatus.FAILURE,
354+
sidecar_path=str(sidecar),
355+
labels_cleaned=False,
356+
)
357+
pending = DispatchRecord(
358+
name="task-pending",
359+
status=DispatchStatus.PENDING,
360+
issue_url=_ISSUE_URL,
361+
labels_cleaned=False,
362+
dispatched_session_id="",
363+
attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
364+
)
365+
sp = tmp_path / "campaign.json"
366+
_write_state(sp, [pending, failure])
367+
368+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
369+
370+
assert result is not None
371+
assert result.name == "task-fail"
372+
373+
374+
def test_pending_dispatch_ended_at_zero_treated_as_stale(tmp_path):
375+
"""PENDING dispatch with ended_at=0.0 (process crashed unrecorded) is treated as stale."""
376+
sp = tmp_path / "campaign.json"
377+
d = DispatchRecord(
378+
name="d1",
379+
status=DispatchStatus.PENDING,
380+
issue_url=_ISSUE_URL,
381+
labels_cleaned=False,
382+
dispatched_session_id="",
383+
attempt_history=[{"ended_at": 0.0, "status": "interrupted"}],
384+
)
385+
_write_state(sp, [d])
386+
387+
result = find_dispatch_for_issue(_ISSUE_URL, [sp])
388+
389+
assert result is not None
390+
assert result.name == "d1"

tests/fleet/test_retry_failed_dispatch.py

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -476,3 +476,23 @@ def test_snapshot_values_match_pre_clear_state(self):
476476
assert snapshot[key] == expected_val, (
477477
f"Snapshot[{key!r}] = {snapshot[key]!r}, expected {expected_val!r}"
478478
)
479+
480+
481+
# --- identity-field coverage: issue_url ---
482+
483+
484+
def test_issue_url_in_retry_identity_fields():
485+
"""issue_url is an identity field — it survives retry resets."""
486+
assert "issue_url" in _RETRY_IDENTITY_FIELDS
487+
488+
489+
def test_issue_url_preserved_on_clear_dispatch_for_retry():
490+
"""issue_url retains its value after _clear_dispatch_for_retry."""
491+
d = DispatchRecord(
492+
name="d1",
493+
status=DispatchStatus.FAILURE,
494+
issue_url="https://github.com/org/repo/issues/1",
495+
)
496+
_clear_dispatch_for_retry(d)
497+
assert d.status == DispatchStatus.PENDING
498+
assert d.issue_url == "https://github.com/org/repo/issues/1"

0 commit comments

Comments
 (0)