diff --git a/src/autoskillit/fleet/state_recovery.py b/src/autoskillit/fleet/state_recovery.py index 0e957a465..d255680b0 100644 --- a/src/autoskillit/fleet/state_recovery.py +++ b/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: diff --git a/src/autoskillit/fleet/state_types.py b/src/autoskillit/fleet/state_types.py index f386e66a0..fc6b136f3 100644 --- a/src/autoskillit/fleet/state_types.py +++ b/src/autoskillit/fleet/state_types.py @@ -25,6 +25,7 @@ "attempt_history", "session_chain", "resume_count", + "issue_url", } ) diff --git a/src/autoskillit/server/tools/_claim_helpers.py b/src/autoskillit/server/tools/_claim_helpers.py index a17680cd2..54ffa7190 100644 --- a/src/autoskillit/server/tools/_claim_helpers.py +++ b/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, diff --git a/src/autoskillit/server/tools/tools_fleet_reset.py b/src/autoskillit/server/tools/tools_fleet_reset.py index 8c3f72779..f8282dda8 100644 --- a/src/autoskillit/server/tools/tools_fleet_reset.py +++ b/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: diff --git a/tests/fleet/test_find_dispatch_for_issue.py b/tests/fleet/test_find_dispatch_for_issue.py index 9bb4c5420..aaf66bc47 100644 --- a/tests/fleet/test_find_dispatch_for_issue.py +++ b/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" diff --git a/tests/fleet/test_retry_failed_dispatch.py b/tests/fleet/test_retry_failed_dispatch.py index f27f66dae..5d48fe1b2 100644 --- a/tests/fleet/test_retry_failed_dispatch.py +++ b/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" diff --git a/tests/server/test_claim_liveness.py b/tests/server/test_claim_liveness.py index 75d26bd31..f24911275 100644 --- a/tests/server/test_claim_liveness.py +++ b/tests/server/test_claim_liveness.py @@ -342,3 +342,36 @@ async def test_claim_helper_failure_dispatch_recovers_via_issue_url(self, tmp_pa assert decision.stale_label_cleaned is True cleanup_mock.assert_called_once() assert cleanup_mock.call_args.kwargs["issue_url"] == _ISSUE_URL + + +class TestClaimHelperPendingDispatch: + @pytest.mark.anyio + async def test_pending_dispatch_blocks_claim(self): + """PENDING dispatch (from a retry reset) blocks claim to prevent double-dispatch.""" + from autoskillit.server.tools._claim_helpers import _try_claim_with_liveness + + pending = DispatchRecord( + name="d-pending", + status=DispatchStatus.PENDING, + issue_url=_ISSUE_URL, + attempt_history=[{"ended_at": 0.0, "status": "failure"}], + ) + cleanup_mock = AsyncMock() + + with ( + patch(f"{_CLAIM_MODULE}.find_dispatch_for_issue", return_value=pending), + patch(f"{_CLAIM_MODULE}.cleanup_orphaned_labels", cleanup_mock), + ): + decision = await _try_claim_with_liveness( + issue_url=_ISSUE_URL, + issue_number=42, + effective_label="in-progress", + current_labels=["in-progress"], + allow_reentry=False, + github_client=AsyncMock(), + campaign_state_paths=[], + ) + + assert decision.claimed is False + assert "pending retry" in decision.reason + cleanup_mock.assert_not_called() diff --git a/tests/server/test_tools_fleet_reset.py b/tests/server/test_tools_fleet_reset.py index 7be62c7ac..1fc241630 100644 --- a/tests/server/test_tools_fleet_reset.py +++ b/tests/server/test_tools_fleet_reset.py @@ -4,7 +4,7 @@ import json from pathlib import Path -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, patch import pytest @@ -103,7 +103,9 @@ async def test_reset_dispatch_happy_path_queued( from autoskillit.server.tools.tools_fleet_reset import reset_dispatch - raw = await reset_dispatch(dispatch_id="d-abc123", reset_to="queued") + raw = await reset_dispatch( + dispatch_id="d-abc123", reset_to="queued", destroy_artifacts=True + ) result = json.loads(raw) assert result["success"] is True @@ -366,7 +368,7 @@ async def _runner(cmd, **_kwargs): from autoskillit.server.tools.tools_fleet_reset import reset_dispatch - raw = await reset_dispatch(dispatch_id="d-abc123") + raw = await reset_dispatch(dispatch_id="d-abc123", destroy_artifacts=True) result = json.loads(raw) assert result["success"] is True assert "https://github.com/owner/repo/pull/99" in result["prs_closed"] @@ -443,9 +445,144 @@ async def _runner(cmd, **_kwargs): from autoskillit.server.tools.tools_fleet_reset import reset_dispatch - raw = await reset_dispatch(dispatch_id="d-abc123", reset_to="queued") + raw = await reset_dispatch( + dispatch_id="d-abc123", reset_to="queued", destroy_artifacts=True + ) result = json.loads(raw) assert result["success"] is True assert result["has_protected_artifacts"] is True assert "https://github.com/owner/repo/pull/1" in result["protected_prs"] + + +class TestResetDispatchStateOnly: + @pytest.mark.anyio + async def test_state_only_default(self, build_ctx_open, tmp_path, monkeypatch) -> None: + """Default reset_dispatch swaps labels but preserves worktree/branch/PRs.""" + sidecar = tmp_path / "sidecar.jsonl" + _write_sidecar(sidecar, pr_url="https://github.com/owner/repo/pull/1") + state_path = _setup_state(tmp_path, sidecar_path=str(sidecar)) + tool_ctx = build_ctx_open() + _setup_tool(tool_ctx, monkeypatch, state_path) + + from autoskillit.server.tools.tools_fleet_reset import reset_dispatch + + with ( + patch( + "autoskillit.server.tools.tools_fleet_reset.reset_dispatch_artifacts" + ) as mock_rda, + patch( + "autoskillit.server.tools.tools_fleet_reset.cleanup_orphaned_labels", + new_callable=AsyncMock, + return_value=True, + ) as mock_col, + ): + raw = await reset_dispatch(dispatch_id="d-abc123", reset_to="queued") + result = json.loads(raw) + + assert result["success"] is True + assert result["destroy_artifacts"] is False + assert result["worktree_removed"] is False + assert result["local_branch_deleted"] is False + assert result["remote_branch_deleted"] is False + assert result["prs_closed"] == [] + mock_rda.assert_not_called() + mock_col.assert_called_once() + + @pytest.mark.anyio + async def test_destroy_artifacts_true(self, build_ctx_open, tmp_path, monkeypatch) -> None: + """destroy_artifacts=True preserves the existing full-destruction behavior.""" + sidecar = tmp_path / "sidecar.jsonl" + _write_sidecar(sidecar, pr_url="https://github.com/owner/repo/pull/1") + state_path = _setup_state(tmp_path, sidecar_path=str(sidecar)) + tool_ctx = build_ctx_open() + _setup_tool(tool_ctx, monkeypatch, state_path) + + from autoskillit.server.tools.tools_fleet_reset import reset_dispatch + + raw = await reset_dispatch( + dispatch_id="d-abc123", reset_to="queued", destroy_artifacts=True + ) + result = json.loads(raw) + + assert result["success"] is True + assert result["destroy_artifacts"] is True + assert result["worktree_removed"] is True + assert result["sidecar_removed"] is True + assert result["local_branch_deleted"] is True + assert result["remote_branch_deleted"] is True + assert "https://github.com/owner/repo/pull/1" in result["prs_closed"] + + @pytest.mark.anyio + async def test_state_only_still_cleans_labels( + self, build_ctx_open, tmp_path, monkeypatch + ) -> None: + """State-only reset swaps labels using dispatch.issue_url.""" + sidecar = tmp_path / "sidecar.jsonl" + _write_sidecar(sidecar, pr_url=None) + state_path = _setup_state(tmp_path, sidecar_path=str(sidecar)) + # Inject issue_url onto the dispatch + from autoskillit.fleet.state import read_state + + state = read_state(state_path) + assert state is not None + state.dispatches[0].issue_url = "https://github.com/owner/repo/issues/5" + from autoskillit.fleet.state import CampaignStateMutator + + with CampaignStateMutator(state_path) as m: + assert m.state is not None + m.state.dispatches[0].issue_url = "https://github.com/owner/repo/issues/5" + m.mark_dirty() + + tool_ctx = build_ctx_open() + _setup_tool(tool_ctx, monkeypatch, state_path) + + from autoskillit.server.tools.tools_fleet_reset import reset_dispatch + + with ( + patch("autoskillit.server.tools.tools_fleet_reset.reset_dispatch_artifacts"), + patch( + "autoskillit.server.tools.tools_fleet_reset.cleanup_orphaned_labels", + new_callable=AsyncMock, + return_value=True, + ) as mock_col, + ): + raw = await reset_dispatch(dispatch_id="d-abc123", reset_to="queued") + result = json.loads(raw) + + assert result["success"] is True + assert result["labels_reset"] is True + mock_col.assert_called_once() + call_kwargs = mock_col.call_args.kwargs + assert call_kwargs["issue_url"] == "https://github.com/owner/repo/issues/5" + assert "queued" in call_kwargs["add_labels"] + + @pytest.mark.anyio + async def test_force_without_destroy(self, build_ctx_open, tmp_path, monkeypatch) -> None: + """force=True bypasses resume gate without destroying artifacts.""" + sidecar = tmp_path / "sidecar.jsonl" + _write_sidecar(sidecar, pr_url="https://github.com/owner/repo/pull/1") + state_path = _setup_state(tmp_path, sidecar_path=str(sidecar)) + tool_ctx = build_ctx_open() + _setup_tool(tool_ctx, monkeypatch, state_path) + + from autoskillit.server.tools.tools_fleet_reset import reset_dispatch + + with ( + patch( + "autoskillit.server.tools.tools_fleet_reset.reset_dispatch_artifacts" + ) as mock_rda, + patch( + "autoskillit.server.tools.tools_fleet_reset.cleanup_orphaned_labels", + new_callable=AsyncMock, + return_value=True, + ), + ): + raw = await reset_dispatch(dispatch_id="d-abc123", force=True) + result = json.loads(raw) + + assert result["success"] is True + assert result["destroy_artifacts"] is False + assert result["worktree_removed"] is False + assert result["prs_closed"] == [] + mock_rda.assert_not_called()