Implementation Plan: Separate reset_dispatch State Reset from Artifact Destruction

src/autoskillit/fleet/state_recovery.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import time
 from pathlib import Path
 from typing import Any
 
@@ -58,6 +59,8 @@
     }
 )
 
+_PENDING_QUIET_PERIOD_SECONDS: float = 60.0
+
 
 def _count_consecutive_resumable_timeouts(history: list[dict[str, Any]]) -> int:
     """Count consecutive RESUMABLE entries with retriable reasons from the tail."""
@@ -332,6 +335,7 @@ def find_dispatch_for_issue(
 
     Pass 1: RUNNING dispatches (live session may own the label).
     Pass 2: terminal dispatches (FAILURE, INTERRUPTED) with labels_cleaned=False.
+    Pass 3: PENDING dispatches with a stale attempt_history entry and no active session.
 
     RUNNING takes priority — a live session should not be preempted by an old dead dispatch.
     Returns the first matching DispatchRecord, else None. Reads are filesystem-only.
@@ -375,7 +379,34 @@ def find_dispatch_for_issue(
                 entries = read_sidecar_from_path(Path(d.sidecar_path)).entries
                 if any(e.issue_url == issue_url for e in entries):
                     terminal_match = d
-    return terminal_match
+
+    if terminal_match is not None:
+        return terminal_match
+
+    for state_path in campaign_state_paths:
+        try:
+            state = read_state(state_path)
+        except Exception:
+            logger.warning("Failed to read campaign state from %s", state_path, exc_info=True)
+            continue
+        if state is None:
+            continue
+        for d in state.dispatches:
+            if d.status != DispatchStatus.PENDING:
+                continue
+            if not d.issue_url or d.issue_url != issue_url:
+                continue
+            if d.labels_cleaned:
+                continue
+            if d.dispatched_session_id:
+                continue
+            if not d.attempt_history:
+                continue
+            last_attempt = d.attempt_history[-1]
+            ended_at = last_attempt.get("ended_at", 0.0)
+            if not ended_at or (time.time() - ended_at) > _PENDING_QUIET_PERIOD_SECONDS:
+                return d
+    return None
 
 
 def derive_orchestrator_resume_spec(state: CampaignState) -> NamedResume | NoResume:

src/autoskillit/fleet/state_types.py

@@ -25,6 +25,7 @@
         "attempt_history",
         "session_chain",
         "resume_count",
+        "issue_url",
     }
 )
 

src/autoskillit/server/tools/_claim_helpers.py

@@ -10,6 +10,7 @@
 from autoskillit.fleet import (
     TERMINAL_UNCLEANED_STATUSES,
     CampaignStateMutator,
+    DispatchStatus,
     cleanup_orphaned_labels,
     discover_campaign_state_files,
     find_dispatch_for_issue,
@@ -95,6 +96,14 @@ async def _try_claim_with_liveness(
         if cleaned:
             _mark_dispatch_labels_cleaned(dispatch.name, campaign_state_paths)
         return ClaimDecision(claimed=True, stale_label_cleaned=cleaned)
+    if dispatch.status == DispatchStatus.PENDING:
+        return ClaimDecision(
+            claimed=False,
+            reason=(
+                f"Issue #{issue_number} already has '{effective_label}' label"
+                " — a reset dispatch is pending retry"
+            ),
+        )
     if is_dispatch_session_alive(dispatch):
         return ClaimDecision(
             claimed=False,

src/autoskillit/server/tools/tools_fleet_reset.py

@@ -14,6 +14,9 @@
 from autoskillit.fleet import (
     _RESETTABLE_STATUSES,
     DispatchStatus,
+    ResetReport,
+    cleanup_orphaned_labels,
+    compute_reset_labels,
     discover_campaign_state_files,
     find_dispatch_in_campaigns,
     format_resettable_statuses,
@@ -74,17 +77,20 @@ async def reset_dispatch(
     dispatch_id: str,
     reset_to: str = "queued",
     force: bool = False,
+    destroy_artifacts: bool = False,
     ctx: Context = CurrentContext(),
 ) -> str:
-    """Reset a failed L2 dispatch, cleaning up all git/PR artifacts.
+    """Reset a failed L2 dispatch for retry or abandonment.
 
-    Removes the local worktree, local and remote branches, closes any open PRs,
-    and resets issue labels. Use after a resume attempt fails or is declined.
+    By default, resets internal state only (status → PENDING) while preserving
+    git artifacts (worktree, branches, PRs). Issue labels are always swapped.
 
     Args:
         dispatch_id: The dispatch ID (UUID) or dispatch name to reset.
         reset_to: Target state — "queued" (fresh retry) or "fail" (abandon).
         force: Bypass the resume-attempt gate when resume is known to be impossible.
+        destroy_artifacts: When True, also remove the git worktree, delete local/remote
+            branches, and close open PRs. Default False for safe error recovery.
 
     Never raises.
     """
@@ -134,21 +140,36 @@ async def reset_dispatch(
         worktrees_dir = resolve_worktrees_dir(project_dir, cfg.workspace.worktree_root)
         target_state = _VALID_RESET_TARGETS[reset_to]
 
-        if tool_ctx.runner is None:
-            return fleet_error(
-                FleetErrorCode.FLEET_L3_STARTUP_OR_CRASH,
-                "No subprocess runner available in this session.",
+        if destroy_artifacts:
+            if tool_ctx.runner is None:
+                return fleet_error(
+                    FleetErrorCode.FLEET_L3_STARTUP_OR_CRASH,
+                    "No subprocess runner available in this session.",
+                )
+
+            report = await reset_dispatch_artifacts(
+                dispatch,
+                project_dir=project_dir,
+                worktrees_dir=worktrees_dir,
+                runner=tool_ctx.runner,
+                github_client=tool_ctx.github_client,
+                target_state=target_state,
+                force=force,
             )
-
-        report = await reset_dispatch_artifacts(
-            dispatch,
-            project_dir=project_dir,
-            worktrees_dir=worktrees_dir,
-            runner=tool_ctx.runner,
-            github_client=tool_ctx.github_client,
-            target_state=target_state,
-            force=force,
-        )
+        else:
+            report = ResetReport(
+                dispatch_name=dispatch.name,
+                branch_name=dispatch.branch_name or dispatch.name,
+            )
+            remove_labels, add_labels = compute_reset_labels(target_state)
+            labels_ok = await cleanup_orphaned_labels(
+                dispatch.sidecar_path,
+                tool_ctx.github_client,
+                issue_url=dispatch.issue_url,
+                remove_labels=remove_labels,
+                add_labels=add_labels,
+            )
+            report.labels_reset = labels_ok
 
         state_updated = await update_campaign_state(
             dispatch.name,
@@ -178,6 +199,7 @@ async def reset_dispatch(
                 "has_protected_artifacts": report.has_protected_artifacts,
                 "protected_prs": report.protected_prs,
                 "reset_to": reset_to,
+                "destroy_artifacts": destroy_artifacts,
             }
         )
     except Exception:

tests/fleet/test_find_dispatch_for_issue.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import time
 from dataclasses import asdict
 from pathlib import Path
 
@@ -239,3 +240,151 @@ def test_running_dispatch_takes_priority_over_failure(tmp_path):
 
     assert result is not None
     assert result.name == "task-run"
+
+
+# --- PENDING dispatch search (Pass 3) ---
+
+
+def test_pending_dispatch_stale_match(tmp_path):
+    """PENDING dispatch with stale attempt_history is returned."""
+    sp = tmp_path / "campaign.json"
+    d = DispatchRecord(
+        name="d1",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="",
+        attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
+    )
+    _write_state(sp, [d])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is not None
+    assert result.name == "d1"
+
+
+def test_pending_dispatch_too_fresh_skipped(tmp_path):
+    """PENDING dispatch within quiet period is skipped."""
+    sp = tmp_path / "campaign.json"
+    d = DispatchRecord(
+        name="d1",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="",
+        attempt_history=[{"ended_at": time.time() - 5, "status": "failure"}],
+    )
+    _write_state(sp, [d])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is None
+
+
+def test_pending_dispatch_no_history_skipped(tmp_path):
+    """PENDING dispatch with empty attempt_history is skipped (never dispatched)."""
+    sp = tmp_path / "campaign.json"
+    d = DispatchRecord(
+        name="d1",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="",
+        attempt_history=[],
+    )
+    _write_state(sp, [d])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is None
+
+
+def test_pending_dispatch_with_session_id_skipped(tmp_path):
+    """PENDING dispatch with active dispatched_session_id is skipped."""
+    sp = tmp_path / "campaign.json"
+    d = DispatchRecord(
+        name="d1",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="abc",
+        attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
+    )
+    _write_state(sp, [d])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is None
+
+
+def test_running_beats_pending(tmp_path):
+    """RUNNING dispatch wins over stale PENDING dispatch."""
+    sidecar = tmp_path / "sidecar.jsonl"
+    _write_sidecar(sidecar, [IssueSidecarEntry(issue_url=_ISSUE_URL, status="completed", ts=_TS)])
+    running = DispatchRecord(
+        name="task-run",
+        status=DispatchStatus.RUNNING,
+        sidecar_path=str(sidecar),
+    )
+    pending = DispatchRecord(
+        name="task-pending",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="",
+        attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
+    )
+    sp = tmp_path / "campaign.json"
+    _write_state(sp, [pending, running])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is not None
+    assert result.name == "task-run"
+
+
+def test_terminal_beats_pending(tmp_path):
+    """Terminal FAILURE wins over stale PENDING dispatch."""
+    sidecar = tmp_path / "sidecar.jsonl"
+    _write_sidecar(sidecar, [IssueSidecarEntry(issue_url=_ISSUE_URL, status="failed", ts=_TS)])
+    failure = DispatchRecord(
+        name="task-fail",
+        status=DispatchStatus.FAILURE,
+        sidecar_path=str(sidecar),
+        labels_cleaned=False,
+    )
+    pending = DispatchRecord(
+        name="task-pending",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="",
+        attempt_history=[{"ended_at": time.time() - 120, "status": "failure"}],
+    )
+    sp = tmp_path / "campaign.json"
+    _write_state(sp, [pending, failure])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is not None
+    assert result.name == "task-fail"
+
+
+def test_pending_dispatch_ended_at_zero_treated_as_stale(tmp_path):
+    """PENDING dispatch with ended_at=0.0 (process crashed unrecorded) is treated as stale."""
+    sp = tmp_path / "campaign.json"
+    d = DispatchRecord(
+        name="d1",
+        status=DispatchStatus.PENDING,
+        issue_url=_ISSUE_URL,
+        labels_cleaned=False,
+        dispatched_session_id="",
+        attempt_history=[{"ended_at": 0.0, "status": "interrupted"}],
+    )
+    _write_state(sp, [d])
+
+    result = find_dispatch_for_issue(_ISSUE_URL, [sp])
+
+    assert result is not None
+    assert result.name == "d1"

tests/fleet/test_retry_failed_dispatch.py

@@ -476,3 +476,23 @@ def test_snapshot_values_match_pre_clear_state(self):
                 assert snapshot[key] == expected_val, (
                     f"Snapshot[{key!r}] = {snapshot[key]!r}, expected {expected_val!r}"
                 )
+
+
+# --- identity-field coverage: issue_url ---
+
+
+def test_issue_url_in_retry_identity_fields():
+    """issue_url is an identity field — it survives retry resets."""
+    assert "issue_url" in _RETRY_IDENTITY_FIELDS
+
+
+def test_issue_url_preserved_on_clear_dispatch_for_retry():
+    """issue_url retains its value after _clear_dispatch_for_retry."""
+    d = DispatchRecord(
+        name="d1",
+        status=DispatchStatus.FAILURE,
+        issue_url="https://github.com/org/repo/issues/1",
+    )
+    _clear_dispatch_for_retry(d)
+    assert d.status == DispatchStatus.PENDING
+    assert d.issue_url == "https://github.com/org/repo/issues/1"