diff --git a/orchestrator/routes/decisions.py b/orchestrator/routes/decisions.py index 7848283137..2f1508384d 100644 --- a/orchestrator/routes/decisions.py +++ b/orchestrator/routes/decisions.py @@ -150,6 +150,148 @@ def _handle_restart_agent(pipeline_id: str, question: str) -> None: ) +def _handle_hard_reset_recovery_resolution( + pipeline_id: str, + context: str, + resolution: str, + valid_options: list[str] | None = None, +) -> None: + """Dispatch the hard-reset recovery HITL ack (#2792). + + ``context`` is ``hard_reset_recovery:``; ``resolution`` is + one of the two options the HITL exposed (``"Continue with post-reset + state"`` or ``"Abort pipeline"``). + + Continue → :func:`routes.pipelines.resume_pipeline_after_hard_reset_ack` + resets the failed phase exec state and spawns a fresh + ``_run_pipeline`` thread so the populator (and downstream phase + work) re-runs against the reconciled worktree. + + Abort → :func:`routes.pipelines.abort_pipeline_after_hard_reset_ack` + transitions the pipeline to CANCELLED. Cleanup of containers / + worktrees / other pending decisions still happens via the existing + PATCH ``update_pipeline`` flow the operator drives next. + + ``valid_options`` is the decision's actual options list (defense + in depth, #2797 follow-up): the doubly-failed branch emits a HITL + whose options list is ``["Abort pipeline"]`` only. An operator + hitting the resolution API directly with ``"Continue with + post-reset state"`` would otherwise route to the resume helper + and re-spawn a ``_run_pipeline`` thread that loops back into the + same divergence. When the resolution is not in ``valid_options``, + routing is suppressed and the unknown-resolution path runs + instead. ``valid_options=None`` keeps backward-compatible + behavior (skip the check) for legacy callers. + + Unknown resolutions are logged at WARN and broadcast as an + ``OVERSEER_ALERT`` so the operator notices the dispatch was skipped + — the decision is already marked RESOLVED at this point, so the + human's intent is preserved in state regardless, but the pipeline + will stay stuck in ``failed_pending_hitl`` until someone intervenes. + """ + phase_value = context.removeprefix("hard_reset_recovery:") + + # Local import to avoid circular import at module load + # (routes.pipelines imports routes.decisions via the blueprint + # registration path in some test setups). + from routes.pipelines import ( + abort_pipeline_after_hard_reset_ack, + resume_pipeline_after_hard_reset_ack, + ) + + # #2797 follow-up: refuse to route a resolution that wasn't in the + # decision's options list. Catches a direct-API operator picking + # "Continue" on a doubly-failed HITL whose options collapsed to + # ["Abort pipeline"] only — without this gate, the dispatch would + # happily restart the phase and loop straight back into the same + # divergence at the next sync. ``valid_options=None`` keeps the + # legacy behavior (skip the cross-check) for callers that don't + # have the decision in hand. + option_mismatch_reason: str | None = None + if valid_options is not None and resolution not in valid_options: + option_mismatch_reason = ( + f"resolution not in the decision's options list " + f"(received: {resolution[:80]!r}; available: {valid_options!r})" + ) + + if option_mismatch_reason is None and resolution == "Continue with post-reset state": + ok = resume_pipeline_after_hard_reset_ack( + pipeline_id, + phase_value=phase_value, + ) + if not ok: + logger.warning( + "hard_reset_recovery 'Continue' dispatch returned False", + pipeline_id=pipeline_id, + phase=phase_value, + ) + elif option_mismatch_reason is None and resolution == "Abort pipeline": + ok = abort_pipeline_after_hard_reset_ack(pipeline_id) + if not ok: + logger.warning( + "hard_reset_recovery 'Abort' dispatch returned False", + pipeline_id=pipeline_id, + ) + else: + logger.warning( + "hard_reset_recovery resolved with unrecognized option", + pipeline_id=pipeline_id, + resolution=resolution[:80], + mismatch=option_mismatch_reason, + ) + try: + try: + from message_store import Message, get_message_store + except ImportError: + from orchestrator.message_store import ( # type: ignore[no-redef] + Message, + get_message_store, + ) + if option_mismatch_reason is not None: + alert_body = ( + "Hard-reset recovery HITL resolved with an option not in " + f"the decision's options list (received: {resolution[:80]!r}; " + f"available: {valid_options!r}). The decision is marked " + "RESOLVED but no dispatch ran; the pipeline will stay in " + "failed_pending_hitl until the operator re-resolves with a " + "valid option." + ) + else: + alert_body = ( + "Hard-reset recovery HITL resolved with an unrecognized " + f"option (received: {resolution[:80]!r}). Expected one of " + "'Continue with post-reset state' or 'Abort pipeline'. The " + "decision is marked RESOLVED but no dispatch ran; the " + "pipeline will stay in failed_pending_hitl until the " + "operator re-resolves with a valid option." + ) + msg = Message( + pipeline_id=pipeline_id, + from_role="orchestrator", + to_role="all", + message_type="OVERSEER_ALERT", + subject="hard-reset-recovery-unknown-resolution", + body=alert_body, + metadata={ + "anomaly": "hard_reset_recovery_unknown_resolution", + "priority": "high", + "context": context, + "resolution": resolution[:80], + }, + phase=phase_value or None, + ) + get_message_store().add_message(msg) + except Exception: # noqa: BLE001 + # N5 follow-up: a broadcast failure here means the operator + # never sees the alert. Log at WARN so the bus-down path + # leaves a trace alongside the unknown-resolution warning. + logger.warning( + "hard_reset_recovery OVERSEER_ALERT broadcast failed", + pipeline_id=pipeline_id, + exc_info=True, + ) + + def _handle_conditional_ack_gate( pipeline_id: str, context: str, @@ -813,6 +955,24 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int] exc_info=True, ) + # #2792: hard-reset recovery ack. ``context`` is + # ``hard_reset_recovery:``; dispatch on the prefix so the + # branch is independent of the prose-y question text. + # ``decision.options`` is passed so the dispatch handler can + # cross-check the resolution against the decision's actual + # options list (defense-in-depth, #2797 follow-up): the + # doubly-failed HITL collapses options to ``["Abort pipeline"]`` + # only, and a direct-API "Continue" on that decision must not + # silently restart a phase that would loop straight back into + # the same divergence. + if decision.context and decision.context.startswith("hard_reset_recovery:"): + _handle_hard_reset_recovery_resolution( + pipeline_id, + decision.context, + decision.resolution or "", + valid_options=list(decision.options or []), + ) + return make_success_response( "Decision resolved", data={ diff --git a/orchestrator/routes/phases.py b/orchestrator/routes/phases.py index d73c439887..020e03cced 100644 --- a/orchestrator/routes/phases.py +++ b/orchestrator/routes/phases.py @@ -1059,11 +1059,25 @@ def populate_contract(pipeline_id: str) -> tuple[Response, int]: ``store.repo_path``) and the operator must commit and push themselves before respawning. + Pre-populate sync (#2792): the route runs + :func:`_sync_worktree_with_remote` before reading the draft. If + that helper falls through to the destructive hard-reset recovery + (rebase autoresolve failed), the route emits the hard-reset + recovery HITL itself (via + :func:`_fail_pipeline_and_emit_hard_reset_recovery`) and returns + HTTP 409 with ``reason="hard_reset_recovery_unacked"`` and + ``backup_ref`` / ``discarded_commit_shas`` in ``details``. The + operator must ack the hard-reset recovery HITL (visible in + ``/sdlc``) before re-running this endpoint — a 2xx with the + destructive recovery flag hidden in the body would let automation + silently miss the discard. + Error responses include a machine-readable ``reason`` code (#1939, #2627): - 400 ``invalid_pipeline_id`` - 404 ``pipeline_not_found`` / ``draft_missing`` / ``no_draft_path`` + - 409 ``hard_reset_recovery_unacked`` (#2792) - 422 ``parse_failed`` / ``empty_result`` / forest violations (structured body) - 500 ``contract_load_failed`` / ``egg_contracts_unavailable`` / @@ -1080,13 +1094,131 @@ def populate_contract(pipeline_id: str) -> tuple[Response, int]: # Import and call the populate function from routes.pipelines import ( PopulateOutcome, + SyncRebaseAndResetFailedError, _commit_statefiles_to_worktree, _compute_gateway_mode, + _fail_pipeline_and_emit_hard_reset_recovery, _get_spawner, _pipeline_identifier, _populate_contract_from_plan, + _sync_worktree_with_remote, ) + # #2792: reconcile the worktree before reading the draft so an + # auto-recovery here can rescue a divergent worktree the same + # way the phase-boundary sync does. When the rebase fails and + # the helper falls through to its hard-reset path, emit the + # same hard-reset recovery HITL the phase-boundary sites do, + # pin the pipeline+phase to FAILED, and refuse to populate. + # All three triggers of the destructive recovery (phase-start, + # post-phase, populate_contract) now expose the operator the + # same surface — see #2797 review B4. A 200 with a deep-buried + # ``hard_reset_performed`` flag would let automation silently + # miss the discard. + populate_sync_outcome = None + if pipeline.branch and worktree_path != store.repo_path: + gateway_mode_for_sync, _ = _compute_gateway_mode(pipeline) + try: + populate_sync_outcome = _sync_worktree_with_remote( + _get_spawner(), + pipeline_id, + worktree_path, + gateway_mode=gateway_mode_for_sync, + base_branch=pipeline.base_branch, + pipeline_branch=pipeline.branch, + ) + except SyncRebaseAndResetFailedError as sync_terminal_err: + # #2792 review B5: rebase AND hard-reset both failed — + # the worktree is still divergent. Pin the pipeline to + # FAILED, emit the hard-reset recovery HITL, and surface + # the terminal failure with a distinct 409 reason code + # so callers can tell it apart from the + # successful-recovery-but-unacked case. + _doubly_failed_msg = ( + f"Sync helper could not reconcile {pipeline.branch} during " + f"populate_contract pre-sync: {sync_terminal_err}" + ) + logger.error( + "populate_contract: pre-populate sync doubly failed", + pipeline_id=pipeline_id, + backup_ref=sync_terminal_err.backup_ref, + discarded_commit_count=len(sync_terminal_err.discarded_commit_shas), + ) + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + store, + phase=pipeline.current_phase, + error_message=_doubly_failed_msg, + backup_ref=sync_terminal_err.backup_ref, + discarded_commit_shas=sync_terminal_err.discarded_commit_shas, + reset_succeeded=False, + ) + return make_error_response( + f"Worktree sync helper exhausted recovery options: {sync_terminal_err}", + status_code=409, + reason="sync_rebase_and_reset_failed", + details={ + "hard_reset_performed": False, + "backup_ref": sync_terminal_err.backup_ref, + "discarded_commit_shas": list(sync_terminal_err.discarded_commit_shas), + }, + ) + except Exception as sync_err: # noqa: BLE001 + logger.warning( + "populate_contract: pre-populate sync raised (continuing)", + pipeline_id=pipeline_id, + error=str(sync_err), + ) + + hard_reset_performed = bool( + populate_sync_outcome and populate_sync_outcome.hard_reset_performed + ) + hard_reset_backup_ref = populate_sync_outcome.backup_ref if populate_sync_outcome else None + hard_reset_discarded = ( + list(populate_sync_outcome.discarded_commit_shas) if populate_sync_outcome else [] + ) + + if hard_reset_performed: + # Do NOT run the populator on a worktree that was just + # hard-reset. Pin the pipeline+phase to FAILED and emit + # the same hard-reset recovery HITL the phase-boundary + # sites do so the operator's ack surface is uniform across + # all three triggers of the destructive recovery, then + # return 409 (#2797 review B4). + _hard_reset_msg = ( + f"Sync helper hard-reset {pipeline.branch} to origin during " + f"populate_contract pre-sync (rebase autoresolve could not " + f"reconcile divergence); " + f"{len(hard_reset_discarded)} local-only commit(s) preserved " + f"under {hard_reset_backup_ref or '(backup ref write failed)'}" + ) + logger.error( + "populate_contract: refusing to populate after hard-reset recovery", + pipeline_id=pipeline_id, + backup_ref=hard_reset_backup_ref, + discarded_commit_count=len(hard_reset_discarded), + ) + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + store, + phase=pipeline.current_phase, + error_message=_hard_reset_msg, + backup_ref=hard_reset_backup_ref, + discarded_commit_shas=hard_reset_discarded, + ) + return make_error_response( + "Worktree was hard-reset during pre-populate sync; " + "operator must ack the hard-reset recovery HITL " + "(visible in /sdlc) before re-running populate_contract.", + status_code=409, + reason="hard_reset_recovery_unacked", + details={ + "hard_reset_performed": True, + "backup_ref": hard_reset_backup_ref, + "discarded_commit_shas": hard_reset_discarded, + }, + ) + _populate_endpoint_result = _populate_contract_from_plan( repo_path=worktree_path, pipeline_id=pipeline_id, diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index a6f3fa0be5..5d20a109b3 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -3338,6 +3338,274 @@ def restart_phase(pipeline_id: str, phase: str) -> tuple[Response, int]: ) +def resume_pipeline_after_hard_reset_ack( + pipeline_id: str, + *, + phase_value: str, + reason: str = "operator ack'd sync-recovery hard reset (#2792)", +) -> bool: + """Programmatic phase-restart for the hard-reset-recovery HITL (#2792). + + Mirrors the in-lock state reset that :func:`restart_phase` performs + (phase exec status → PENDING, pipeline status → RUNNING, bump + ``run_epoch``) and spawns a fresh ``_run_pipeline`` driver thread. + Also mirrors :func:`restart_phase`'s consensus / restart-count / + health-monitor cleanup so a re-spawn after a post-phase hard reset + does not short-circuit against the prior round's CONFIRMED tracker + state or fire stale-elapsed Tier-1 health alerts (the #2084 fix + class that the original slim implementation skipped). + + Slimmer than the HTTP route on purpose: + + * No container teardown — the hard-reset path is reached after BRC + consensus already completed (or never spawned containers for this + run), so there are no live containers to stop. + * No per-agent worktree deletion — the pipeline worktree was + reconciled in-place by the sync helper; per-agent worktrees are + managed by the next phase spawn. + * No salvage step — the discarded local commits were already + pinned under the ``refs/egg-backup/sync-recovery/...`` ref by + :func:`_sync_worktree_with_remote`. + + Returns True if the resume kicked off, False on any precondition + failure (logged). The caller (decision-resolution dispatch) treats + False as best-effort — the decision itself is already marked + RESOLVED, so a transient resume failure leaves the operator with a + FAILED pipeline they can manually ``restart_phase`` against. + """ + repo_path = get_repo_path() + try: + store, _ = _resolve_pipeline(pipeline_id, repo_path) + except (InvalidPipelineIdError, PipelineNotFoundError) as exc: + logger.warning( + "resume_pipeline_after_hard_reset_ack: pipeline lookup failed", + pipeline_id=pipeline_id, + error=str(exc), + ) + return False + + try: + PipelinePhase(phase_value) + except ValueError: + logger.warning( + "resume_pipeline_after_hard_reset_ack: invalid phase", + pipeline_id=pipeline_id, + phase=phase_value, + ) + return False + + agent_role_values: list[str] = [] + with get_pipeline_state_lock(pipeline_id): + pipeline = store.load_pipeline(pipeline_id) + if phase_value != pipeline.current_phase.value: + logger.warning( + "resume_pipeline_after_hard_reset_ack: phase mismatch", + pipeline_id=pipeline_id, + requested_phase=phase_value, + current_phase=pipeline.current_phase.value, + ) + return False + phase_exec = pipeline.phases.get(phase_value) + if phase_exec is None: + logger.warning( + "resume_pipeline_after_hard_reset_ack: phase exec missing", + pipeline_id=pipeline_id, + phase=phase_value, + ) + return False + + # Collect roster of agent roles for health-monitor reset (#2084). + # The cache on phase_exec.agents may be the most recent spawn's + # roster (post-phase emission site) or stale-from-prior-run + # (phase-start site). Fall back to the deterministic per-phase + # roster source so the reset covers both cases. + for agent in phase_exec.agents: + try: + role = ( + agent.role + if isinstance(getattr(agent, "role", None), AgentRole) + else AgentRole(agent.role) + ) + agent_role_values.append(role.value) + except ValueError, AttributeError: + continue + if not agent_role_values: + try: + from egg_contracts.agent_roles import ( + get_roles_for_phase as _get_roles_for_phase, + ) + + for r in _get_roles_for_phase( + phase_value, + include_reviewers=True, + repo=pipeline.repo, + has_contract=getattr(pipeline, "has_contract", True), + ): + agent_role_values.append(r.value) + except Exception as exc: # noqa: BLE001 + logger.warning( + "resume_pipeline_after_hard_reset_ack: roster fallback failed", + pipeline_id=pipeline_id, + phase=phase_value, + error=str(exc), + ) + + # Mirror the state reset in restart_phase (lines 3140-3155) so + # the new _run_pipeline thread treats this as a fresh phase. + phase_exec.containers = [] + phase_exec.agents = [] + phase_exec.review_cycles = 0 + phase_exec.hitl_review_cycles = 0 + phase_exec.status = PipelineStatus.PENDING + phase_exec.started_at = None + phase_exec.work_started_at = None + phase_exec.completed_at = None + phase_exec.error = None + phase_exec.cycle_timings = [] + pipeline.status = PipelineStatus.RUNNING + pipeline.error = None + pipeline.run_epoch = datetime.now(UTC) + store.update_pipeline(pipeline_id, pipeline.model_dump(mode="json")) + + # Outside the lock: reset BRC tracker, legacy evaluator, restart + # counts, and health-monitor anchors so a re-spawn does NOT + # short-circuit against the prior round's CONFIRMED tracker state + # or fire stale-elapsed health alerts. This mirrors restart_phase + # lines 3250-3312 — the bug class is #2084. (#2792 review B1.) + try: + try: + from peer_consensus import get_peer_consensus_tracker + except ImportError: + from ..peer_consensus import ( # type: ignore[import-not-found] + get_peer_consensus_tracker, + ) + + tracker = get_peer_consensus_tracker(pipeline_id) + if tracker: + tracker.clear() + logger.info( + "Cleared peer consensus tracker after hard-reset ack", + pipeline_id=pipeline_id, + ) + except ImportError: + pass + except Exception as e: # noqa: BLE001 + logger.warning( + "Failed to clear peer consensus after hard-reset ack", + pipeline_id=pipeline_id, + error=str(e), + ) + + try: + try: + from consensus import get_consensus_evaluator + except ImportError: + from ..consensus import ( # type: ignore[import-not-found] + get_consensus_evaluator, + ) + + evaluator = get_consensus_evaluator() + evaluator.clear(pipeline_id) + except ImportError: + pass + except Exception as e: # noqa: BLE001 + logger.warning( + "Failed to clear legacy consensus after hard-reset ack", + pipeline_id=pipeline_id, + error=str(e), + ) + + try: + _get_spawner().reset_restart_counts(pipeline_id) + except Exception as e: # noqa: BLE001 + logger.warning( + "Failed to reset restart counts after hard-reset ack", + pipeline_id=pipeline_id, + error=str(e), + ) + + try: + try: + from health_monitor import get_health_monitor + except ImportError: + from ..health_monitor import ( # type: ignore[import-not-found] + get_health_monitor, + ) + _hm = get_health_monitor() + if _hm is not None: + for _role_value in agent_role_values: + _hm.reset_agent(_role_value) + except Exception as e: # noqa: BLE001 + logger.warning( + "Failed to reset health-monitor state after hard-reset ack", + pipeline_id=pipeline_id, + phase=phase_value, + error=str(e), + ) + + _spawn_pipeline_run_thread(pipeline_id, store.repo_path, pipeline.run_epoch) + logger.info( + "Resumed pipeline after hard-reset recovery ack", + pipeline_id=pipeline_id, + phase=phase_value, + reason=reason, + agent_roles_reset=agent_role_values, + ) + return True + + +def abort_pipeline_after_hard_reset_ack( + pipeline_id: str, + *, + reason: str = "operator aborted after sync-recovery hard reset (#2792)", +) -> bool: + """Programmatic abort for the hard-reset-recovery HITL (#2792). + + Sets ``pipeline.status = CANCELLED`` and records ``pipeline.error`` + so observers see a structured terminal state instead of FAILED- + with-pending-decision drifting forever. Cancellation of other + pending decisions and container/worktree cleanup are intentionally + left to the existing :func:`update_pipeline` PATCH path that the + operator (or downstream automation) drives next — replicating that + full cleanup pipeline here would duplicate ~80 lines of logic that + is already battle-tested. The backup ref preserves the discarded + commits for offline inspection independently of cleanup timing. + + Returns True on a successful state write, False on precondition + failure. + """ + repo_path = get_repo_path() + try: + store, _ = _resolve_pipeline(pipeline_id, repo_path) + except (InvalidPipelineIdError, PipelineNotFoundError) as exc: + logger.warning( + "abort_pipeline_after_hard_reset_ack: pipeline lookup failed", + pipeline_id=pipeline_id, + error=str(exc), + ) + return False + + with get_pipeline_state_lock(pipeline_id): + pipeline = store.load_pipeline(pipeline_id) + pipeline.status = PipelineStatus.CANCELLED + pipeline.error = reason + store.save_pipeline(pipeline) + try: + _emit_pipeline_event(pipeline, "pipeline.cancelled") + except Exception: # noqa: BLE001 + logger.debug( + "Failed to emit pipeline.cancelled after hard-reset abort", + pipeline_id=pipeline_id, + exc_info=True, + ) + logger.info( + "Aborted pipeline after hard-reset recovery ack", + pipeline_id=pipeline_id, + reason=reason, + ) + return True + + def _filter_salvage_worktrees( worktrees: list[Any], *, @@ -6439,6 +6707,142 @@ def _aggregate_review_verdicts( ) +class WorktreeSyncOutcome(NamedTuple): + """Structured outcome from :func:`_sync_worktree_with_remote` (#2792). + + Phase-boundary callers inspect ``hard_reset_performed`` to decide + whether to emit the destructive-recovery HITL ack. Best-effort + callers can ignore the return value entirely — every field has a + safe default and the sync still does the same in-band work whether + or not the outcome is consumed. + + ``case`` is the same discriminator the function emits to its + ``worktree_sync_outcome`` log line, so the field can be cross- + referenced against operator-grep patterns. + + ``backup_ref`` is the full ref name (``refs/egg-backup/sync-recovery/ + /``) when the hard-reset fallback fires and the + backup write succeeded. ``None`` means the reset still happened but + the backup write failed — the discarded SHAs go into the WARN log + inline so they're at least in the audit trail (see the helper body). + + ``discarded_commit_shas`` is the list of local-only short SHAs (with + summaries) that were on HEAD before the hard reset and are now + reachable only via ``backup_ref``. Empty when the local-only-commit + rev-list itself failed; the hard reset still runs, but the operator + can't be told exactly what was discarded. + """ + + case: str + hard_reset_performed: bool = False + backup_ref: str | None = None + discarded_commit_shas: tuple[str, ...] = () + + +def _build_sync_recovery_backup_ref(pipeline_id: str, unix_ts: int) -> str: + """Return the canonical ``refs/egg-backup/sync-recovery//`` name (#2792). + + Pulled out so the test, the writer, and any future opportunistic + pruner share a single ref-name convention. The slash-segment + layout lets ``git for-each-ref refs/egg-backup/sync-recovery/`` + enumerate just this pipeline's backups. + """ + return f"refs/egg-backup/sync-recovery/{pipeline_id}/{unix_ts}" + + +def _collect_local_only_commits( + git_base: list[str], + *, + pipeline_id: str, + branch: str, + remote_branch: str, +) -> tuple[str, ...]: + """Enumerate local-only commits between HEAD and ``origin/``. + + Returns a tuple of ``" "`` strings, oldest first. + A failure (subprocess error, nonzero rc, parse error) returns an + empty tuple and emits a WARN — the hard-reset fallback proceeds + with an unknown discard list rather than blocking on best-effort + forensic enumeration (#2792 section 5). + """ + try: + result = subprocess.run( + [ + *git_base, + "rev-list", + "--reverse", + "--pretty=format:%h %s", + "--no-commit-header", + f"origin/{remote_branch}..HEAD", + ], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + if result.returncode != 0: + logger.warning( + "Failed to enumerate local-only commits before hard reset", + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + rc=result.returncode, + stderr=result.stderr.strip()[:200], + ) + return () + lines = [ln.strip() for ln in result.stdout.splitlines() if ln.strip()] + return tuple(lines) + except Exception as exc: + logger.warning( + "Local-only commit enumeration raised before hard reset", + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + error=str(exc), + ) + return () + + +def _create_sync_recovery_backup_ref( + git_base: list[str], + *, + pipeline_id: str, + ref_name: str, +) -> bool: + """Pin current HEAD under ``ref_name`` via ``git update-ref`` (#2792). + + Returns True on success. On failure logs WARN and returns False; + the caller proceeds with the destructive reset regardless — the + backup is best-effort, the reset is the reconcile primitive. + """ + try: + result = subprocess.run( + [*git_base, "update-ref", ref_name, "HEAD"], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + if result.returncode != 0: + logger.warning( + "Failed to create sync-recovery backup ref", + pipeline_id=pipeline_id, + ref_name=ref_name, + rc=result.returncode, + stderr=result.stderr.strip()[:200], + ) + return False + return True + except Exception as exc: + logger.warning( + "Sync-recovery backup-ref write raised", + pipeline_id=pipeline_id, + ref_name=ref_name, + error=str(exc), + ) + return False + + def _sync_worktree_with_remote( spawner: "ContainerSpawner", # noqa: UP037 pipeline_id: str, @@ -6448,7 +6852,7 @@ def _sync_worktree_with_remote( base_branch: str | None = None, *, pipeline_branch: str | None = None, -) -> None: +) -> WorktreeSyncOutcome: """Sync a worktree with its remote branch (best-effort). After an orchestrator restart or a phase boundary, the local worktree @@ -6486,6 +6890,17 @@ def _sync_worktree_with_remote( implementation silently left the worktree stale and downstream populator/decision-sync paths consumed the stale state. + When the rebase itself fails (#2792), fall through to a destructive + hard-reset recovery so the worktree is reconciled *before* the + populator and other downstream consumers read it. Local-only + commits are pinned to ``refs/egg-backup/sync-recovery/ + /`` before the reset so they remain reachable + for forensic inspection (``git log ``); the reset + discards them from HEAD. Callers at phase boundaries inspect + ``hard_reset_performed`` on the returned :class:`WorktreeSyncOutcome` + and surface an HITL ack — recovery is automatic, *acknowledgement* + is the human gate. + Every return path emits at least one ``worktree_sync_outcome`` log line with a ``case`` discriminator so production logs name which path fired. Paths that fall through to the step-4 reset @@ -6496,6 +6911,11 @@ def _sync_worktree_with_remote( Safe to call on every pipeline start because it is idempotent when the local branch is already up to date. + + Returns a :class:`WorktreeSyncOutcome` describing what the helper + did. Most callers can ignore the return value; phase-boundary + callers inspect ``hard_reset_performed`` to decide whether to emit + the destructive-recovery HITL ack (#2792). """ base_branch_for_reconcile = base_branch git_base = [ @@ -6520,7 +6940,7 @@ def _sync_worktree_with_remote( pipeline_id=pipeline_id, case="fetch_failed", ) - return + return WorktreeSyncOutcome(case="fetch_failed") # Step 2: Determine current branch try: @@ -6538,7 +6958,7 @@ def _sync_worktree_with_remote( pipeline_id=pipeline_id, case="detached_head", ) - return + return WorktreeSyncOutcome(case="detached_head") except Exception as branch_err: logger.info( "worktree_sync_outcome", @@ -6546,7 +6966,7 @@ def _sync_worktree_with_remote( case="branch_detect_failed", error=str(branch_err), ) - return + return WorktreeSyncOutcome(case="branch_detect_failed") # ``branch`` is the **local** branch name (e.g. ``egg//work`` on # orchestrator worktrees). ``remote_branch`` is the remote-side name @@ -6573,7 +6993,7 @@ def _sync_worktree_with_remote( remote_branch=remote_branch, case="no_remote_tracking", ) - return + return WorktreeSyncOutcome(case="no_remote_tracking") except Exception as rev_parse_err: logger.info( "worktree_sync_outcome", @@ -6583,7 +7003,7 @@ def _sync_worktree_with_remote( case="rev_parse_failed", error=str(rev_parse_err), ) - return + return WorktreeSyncOutcome(case="rev_parse_failed") # Step 3b: Check divergence between local and remote. local_ahead = 0 @@ -6643,7 +7063,7 @@ def _sync_worktree_with_remote( local_ahead=0, remote_ahead=0, ) - return + return WorktreeSyncOutcome(case="already_in_sync") if local_ahead > 0 and remote_ahead == 0: # Local is strictly ahead of remote (no divergence). @@ -6678,7 +7098,7 @@ def _sync_worktree_with_remote( local_ahead=local_ahead, remote_ahead=remote_ahead, ) - return + return WorktreeSyncOutcome(case="local_ahead_pushed") else: logger.warning( "worktree_sync_outcome", @@ -6760,7 +7180,7 @@ def _sync_worktree_with_remote( local_ahead=local_ahead, remote_ahead=remote_ahead, ) - return + return WorktreeSyncOutcome(case="divergence_rebased") logger.error( "worktree_sync_outcome", pipeline_id=pipeline_id, @@ -6772,7 +7192,114 @@ def _sync_worktree_with_remote( category=rebase_outcome.category, detail=rebase_outcome.detail, ) - return + + # #2792: hard-reset auto-recovery. The rebase failed to + # reconcile divergence (typically because the agent-output + # autoresolve only handles ``.egg-state/agent-outputs/`` and + # contracts/brc-history conflicts fall outside that allowlist). + # Without recovery, downstream callers — populator, + # decision-sync, plan-complete — would consume the stale + # worktree state, exactly the silent-failure path #2337 + # raised an explicit error for and #2792 is closing the + # recurrence loop on. + # + # Step A: enumerate the local-only commits we're about to + # discard so the operator-facing HITL can list them. + discarded = _collect_local_only_commits( + git_base, + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + ) + # Step B: pin them under a backup ref so they remain reachable + # for forensic inspection after the reset. Best-effort; a + # backup-write failure inlines the SHA list into the WARN log + # so they at least land in the audit trail (#2792 section 5). + # Use nanosecond precision so two recoveries within the same + # second on the same pipeline (orchestrator restart loop, HITL + # ack racing a phase-start emission) cannot collide on the + # ref name and silently overwrite the first backup (#2797 + # review N1). + unix_ts = time.time_ns() + backup_ref = _build_sync_recovery_backup_ref(pipeline_id, unix_ts) + backup_ok = _create_sync_recovery_backup_ref( + git_base, + pipeline_id=pipeline_id, + ref_name=backup_ref, + ) + if not backup_ok and discarded: + logger.warning( + "Sync-recovery hard reset proceeding without backup ref; " + "discarded SHAs inlined for audit", + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + discarded_commit_shas=list(discarded), + ) + # Step C: destructive reset to reconcile. + try: + reset_result = subprocess.run( + [*git_base, "reset", "--hard", f"origin/{remote_branch}"], + capture_output=True, + text=True, + timeout=30, + check=False, + ) + reset_rc = reset_result.returncode + reset_err = reset_result.stderr.strip() + except Exception as reset_exc: + reset_rc = -1 + reset_err = str(reset_exc) + + if reset_rc != 0: + logger.error( + "worktree_sync_outcome", + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + case="divergence_rebase_and_reset_failed", + local_ahead=local_ahead, + remote_ahead=remote_ahead, + rebase_category=rebase_outcome.category, + rebase_detail=rebase_outcome.detail, + reset_rc=reset_rc, + reset_error=reset_err[:200], + ) + # #2792 review B5: raise a typed error so callers route the + # doubly-failed path through the same FAILED-cleanup flow + # as other terminal sync failures. Returning an outcome + # with hard_reset_performed=False would be indistinguishable + # from a happy no-op at every caller — the worktree is + # still divergent, but the pipeline would continue with no + # signal, re-opening the silent-failure loop this PR + # closes. + raise SyncRebaseAndResetFailedError( + f"Worktree sync helper exhausted recovery options: " + f"rebase failed ({rebase_outcome.category}) and " + f"hard-reset to origin/{remote_branch} also failed " + f"(rc={reset_rc}, stderr={reset_err[:120]})", + backup_ref=backup_ref if backup_ok else None, + discarded_commit_shas=discarded, + ) + + logger.warning( + "worktree_sync_outcome", + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + case="divergence_recovered_via_reset", + local_ahead=local_ahead, + remote_ahead=remote_ahead, + backup_ref=backup_ref if backup_ok else None, + discarded_commit_count=len(discarded), + rebase_category=rebase_outcome.category, + ) + return WorktreeSyncOutcome( + case="divergence_recovered_via_reset", + hard_reset_performed=True, + backup_ref=backup_ref if backup_ok else None, + discarded_commit_shas=discarded, + ) # Step 4: Reset local branch to remote. # This handles: local behind remote, post-push reset, and rev-list-failed @@ -6796,16 +7323,17 @@ def _sync_worktree_with_remote( remote_ahead=remote_ahead, error=result.stderr.strip(), ) - else: - logger.info( - "worktree_sync_outcome", - pipeline_id=pipeline_id, - branch=branch, - remote_branch=remote_branch, - case="reset_succeeded", - local_ahead=local_ahead, - remote_ahead=remote_ahead, - ) + return WorktreeSyncOutcome(case="reset_failed") + logger.info( + "worktree_sync_outcome", + pipeline_id=pipeline_id, + branch=branch, + remote_branch=remote_branch, + case="reset_succeeded", + local_ahead=local_ahead, + remote_ahead=remote_ahead, + ) + return WorktreeSyncOutcome(case="reset_succeeded") except Exception as sync_err: logger.warning( "worktree_sync_outcome", @@ -6817,6 +7345,7 @@ def _sync_worktree_with_remote( remote_ahead=remote_ahead, error=str(sync_err), ) + return WorktreeSyncOutcome(case="reset_failed") class StalePipelineBranchError(RuntimeError): @@ -6830,6 +7359,29 @@ class StalePipelineBranchError(RuntimeError): """ +class SyncRebaseAndResetFailedError(RuntimeError): + """Raised when the rebase AND the hard-reset fallback both failed (#2792). + + The sync helper attempts a rebase-then-reset cascade to reconcile a + divergent worktree. When both halves fail, the worktree is still + divergent — proceeding silently would re-open the silent-failure + loop #2792 was opened to close. Callers convert this into a FAILED + pipeline + HITL ack so the operator knows the helper exhausted its + auto-recovery options without reconciling. + """ + + def __init__( + self, + message: str, + *, + backup_ref: str | None, + discarded_commit_shas: tuple[str, ...], + ) -> None: + super().__init__(message) + self.backup_ref = backup_ref + self.discarded_commit_shas = discarded_commit_shas + + def _rebase_pipeline_branch_onto_base( spawner: "ContainerSpawner", # noqa: UP037 pipeline_id: str, @@ -14130,6 +14682,7 @@ def _persist_hitl_decision( question: str, options: list[str], phase: PipelinePhase | None = None, + context: str | None = None, ): """Create and persist an HITL decision under the pipeline state lock. @@ -14146,6 +14699,11 @@ def _persist_hitl_decision( notifications are not needed for the issue-2203 path. The in-memory `pipeline` argument is also synced so callers observe consistent state. + ``context`` is set on the persisted decision before save so dispatch + handlers in :mod:`routes.decisions` can route on a stable string + discriminator rather than the prose-y ``question`` text (see the + ``failed_role:`` / ``hard_reset_recovery:`` patterns). + Returns the created decision, or None if persistence failed (logged; callers should not raise — losing an HITL decision is bad but losing the rest of the cleanup path is worse). @@ -14158,6 +14716,8 @@ def _persist_hitl_decision( options=options, phase=phase or disk_pipeline.current_phase, ) + if context is not None: + decision.context = context store.save_pipeline(disk_pipeline) # Defensive copy: avoid sharing the list reference with the # disk-loaded copy, which is local and goes out of scope. @@ -14173,6 +14733,228 @@ def _persist_hitl_decision( return None +_HARD_RESET_RECOVERY_CONTEXT_PREFIX = "hard_reset_recovery:" + +_HARD_RESET_RECOVERY_HITL_OPTIONS = [ + "Continue with post-reset state", + "Abort pipeline", +] + +_HARD_RESET_RECOVERY_CONTINUE = "Continue with post-reset state" +_HARD_RESET_RECOVERY_ABORT = "Abort pipeline" + + +def _hard_reset_recovery_hitl_question( + *, + pipeline_id: str, + phase: PipelinePhase | None, + backup_ref: str | None, + discarded_commit_shas: tuple[str, ...] | list[str], + reset_succeeded: bool = True, +) -> str: + """Build the HITL question for the destructive sync-recovery ack (#2792). + + When ``reset_succeeded`` is True the worktree has already been + reconciled — the rebase couldn't resolve divergence, so the helper + hard-reset HEAD to the remote after pinning the local-only commits + under ``backup_ref``. The operator's job here is not to recover; + it's to acknowledge that the recovery happened and either continue + from the reconciled state or abort the pipeline outright. + + When ``reset_succeeded`` is False the rebase *and* the hard-reset + fallback both failed (``SyncRebaseAndResetFailedError``). The + worktree is still divergent — "Continue" would only loop back into + the same failure on the next sync — so the wording reflects the + actual state and only the abort option is offered (the caller + suppresses ``_HARD_RESET_RECOVERY_CONTINUE`` in + :data:`_HARD_RESET_RECOVERY_HITL_OPTIONS` via + :func:`_hard_reset_recovery_hitl_options`). Operators who want to + intervene manually should abort, fix the worktree on the + orchestrator side, then resubmit the task. + """ + phase_label = phase.value if phase is not None else "current phase" + backup_line = ( + f"Backup ref: {backup_ref} (inspect with `git log {backup_ref}`)." + if backup_ref + else "Backup ref: not written — see WARN log for inlined SHAs." + ) + if discarded_commit_shas: + commits_label = ( + "Discarded commits (now reachable only via the backup ref):" + if reset_succeeded + else "Local-only commits captured under the backup ref (worktree still divergent):" + ) + discarded_block = commits_label + "\n - " + "\n - ".join(discarded_commit_shas) + else: + discarded_block = ( + "Discarded commit list could not be enumerated; check the WARN log " + "and the backup ref for the exact set." + ) + if reset_succeeded: + return ( + f"Pipeline {pipeline_id}: the worktree had diverged from origin at " + f"{phase_label}, the rebase autoresolve could not reconcile it, and " + f"the sync helper hard-reset HEAD to origin to keep downstream " + f"populator/decision-sync reads against a consistent state (#2792). " + f"{backup_line}\n{discarded_block}\n\n" + f"How to proceed?\n" + f"- '{_HARD_RESET_RECOVERY_CONTINUE}' — restart_phase {phase_label} " + f"so the populator and BRC agents re-run against the reconciled " + f"worktree.\n" + f"- '{_HARD_RESET_RECOVERY_ABORT}' — cancel_task; the backup ref " + f"preserves the discarded commits for offline inspection." + ) + return ( + f"Pipeline {pipeline_id}: the worktree had diverged from origin at " + f"{phase_label}, the rebase autoresolve could not reconcile it, and " + f"the subsequent hard-reset to origin ALSO failed — the worktree is " + f"still divergent (#2792). Continuing would only loop back into the " + f"same failure on the next sync. {backup_line}\n{discarded_block}\n\n" + f"How to proceed?\n" + f"- '{_HARD_RESET_RECOVERY_ABORT}' — cancel_task; the backup ref " + f"preserves the local-only commits for offline inspection. To recover, " + f"fix the worktree on the orchestrator side manually and resubmit the " + f"task." + ) + + +def _hard_reset_recovery_hitl_options(*, reset_succeeded: bool) -> list[str]: + """Return the HITL options list for the hard-reset recovery ack. + + The "Continue with post-reset state" option only makes sense when + the reset actually completed — when the rebase AND the reset both + failed (``SyncRebaseAndResetFailedError``) the worktree is still + divergent and restarting the phase would loop straight back into + the same failure, so the doubly-failed branch suppresses it (#2797 + follow-up). + """ + if reset_succeeded: + return list(_HARD_RESET_RECOVERY_HITL_OPTIONS) + return [_HARD_RESET_RECOVERY_ABORT] + + +def _emit_hard_reset_recovery_hitl( + pipeline_id: str, + pipeline: Pipeline, + store: StateStore, + *, + phase: PipelinePhase | None, + backup_ref: str | None, + discarded_commit_shas: tuple[str, ...] | list[str], + reset_succeeded: bool = True, +): + """Persist the dedicated hard-reset-recovery HITL (#2792). + + Mirrors :func:`_emit_empty_contract_hitl` — load/mutate/save under + the pipeline state lock, with a ``context`` discriminator so the + decisions dispatch hook in :mod:`routes.decisions` can route on a + stable string instead of the prose-y question text. + + Phase is embedded in the context (``hard_reset_recovery:``) + so the ``Continue`` resolution knows which phase to restart without + re-walking the pipeline state at dispatch time. + + When ``reset_succeeded`` is False (the doubly-failed + ``SyncRebaseAndResetFailedError`` path) the question wording and + the options list both branch to reflect that the worktree is still + divergent — see :func:`_hard_reset_recovery_hitl_question` and + :func:`_hard_reset_recovery_hitl_options`. + """ + phase_label = phase.value if phase is not None else "unknown" + context = f"{_HARD_RESET_RECOVERY_CONTEXT_PREFIX}{phase_label}" + return _persist_hitl_decision( + pipeline_id, + pipeline, + store, + question=_hard_reset_recovery_hitl_question( + pipeline_id=pipeline_id, + phase=phase, + backup_ref=backup_ref, + discarded_commit_shas=tuple(discarded_commit_shas), + reset_succeeded=reset_succeeded, + ), + options=_hard_reset_recovery_hitl_options(reset_succeeded=reset_succeeded), + phase=phase, + context=context, + ) + + +def _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id: str, + store, # noqa: ANN001 — StateStore (avoid import cycle) + *, + phase: PipelinePhase | None, + error_message: str, + backup_ref: str | None, + discarded_commit_shas: tuple[str, ...] | list[str], + reset_succeeded: bool = True, + pre_event_hook: Callable[[], None] | None = None, +) -> None: + """Pin pipeline+phase to FAILED, emit the hard-reset HITL, broadcast events. + + Shared across all four ``_run_pipeline`` emission sites (phase-start + + post-phase, success + doubly-failed) and the ``populate_contract`` + route (#2792 review B4, #2797 follow-up) so every trigger of the + destructive recovery surfaces the same operator-facing state: + ``pipeline.status=FAILED`` + ``phase_execution.status=FAILED`` + persisted under lock, the dedicated hard-reset HITL written under + the same lock so observers never see ``status=FAILED`` without the + pending decision, then a ``pipeline.failed`` event/StatusReporter + dispatch. + + ``reset_succeeded`` distinguishes the successful-recovery branch + (``hard_reset_performed=True``) from the doubly-failed branch + (``SyncRebaseAndResetFailedError``) — the HITL question wording and + the options list both branch on it so the operator isn't offered a + "Continue" option that would only loop into the same failure. + + ``pre_event_hook`` runs after the FAILED-write + HITL persist but + before the ``pipeline.failed`` broadcast. The two post-phase + ``_run_pipeline`` sites use it to tear down the per-phase overseer + container under their existing overseer lock — keeping the + teardown ordered before the public event matches the pre-helper + inline layout. + + Acquires ``get_pipeline_state_lock(pipeline_id)`` as an RLock and + holds it across both the FAILED write and the HITL persist so a + reader observing the pipeline never sees ``status=FAILED`` without + the corresponding pending decision (#2797 follow-up). Callers + must not already hold a non-reentrant lock that conflicts. + """ + with get_pipeline_state_lock(pipeline_id): + pipeline = store.load_pipeline(pipeline_id) + if phase is not None: + phase_execution = pipeline.get_phase_execution(phase) + if phase_execution is not None: + phase_execution.status = PipelineStatus.FAILED + phase_execution.error = error_message + phase_execution.completed_at = datetime.now(UTC) + pipeline.status = PipelineStatus.FAILED + pipeline.error = error_message + store.save_pipeline(pipeline) + # Persist the HITL while still holding the (reentrant) lock so + # a reader between the two writes can't observe FAILED without + # the pending decision. ``_persist_hitl_decision`` re-acquires + # the same RLock — safe under reentrance. + _emit_hard_reset_recovery_hitl( + pipeline_id, + pipeline, + store, + phase=phase, + backup_ref=backup_ref, + discarded_commit_shas=discarded_commit_shas, + reset_succeeded=reset_succeeded, + ) + if pre_event_hook is not None: + pre_event_hook() + report_pipeline_status( + pipeline, + event_type="pipeline.failed", + message=f"Pipeline failed: {error_message[:100]}", + ) + _emit_pipeline_event(pipeline, "pipeline.failed") + + def _emit_empty_contract_hitl( pipeline_id: str, pipeline: Pipeline, @@ -18246,10 +19028,12 @@ def _empty_contract_hitl_question( ) return ( f"Pipeline blocked at {gate}: {divergence_line} " - f"(reason={reason}). The populate-from-plan step silently failed " - f"earlier (#2337 / #2627), so pipeline state and the contract have " - f"diverged. Plain restart_phase implement will respawn into the " - f"same broken state. How to proceed?\n" + f"(reason={reason}). The sync helper's auto-reconcile path " + f"(#2792) tried to bring the worktree forward before the " + f"populator ran; if you're seeing this, that reconcile either " + f"didn't fire or didn't restore the draft, so pipeline state " + f"and the contract have diverged. Plain restart_phase implement " + f"will respawn into the same broken state. How to proceed?\n" f"- 'Repopulate contract from plan draft and retry' — run " f"POST /pipelines/{pipeline_id}/phase/populate-contract, then " f"restart_phase implement.\n" @@ -19662,6 +20446,44 @@ def _run_pipeline( pipeline_mode = "issue" if pipeline.issue_number is not None else "prompt" transitions = PHASE_TRANSITIONS + def _make_overseer_teardown_hook( + *, + reason: str, + container_id: str | None, + phase: PipelinePhase, + ) -> Callable[[], None]: + """Build a pre_event_hook that tears down the per-phase overseer. + + ``container_id`` and ``phase`` are snapshotted as function + parameters (frozen per-call), so the returned closure binds + the loop-iteration values that were current when the + post-phase cleanup branch fired — late binding would race a + subsequent loop iteration. ``reason`` differs between the + doubly-failed and hard-reset-recovered call sites and is + forwarded to :func:`_teardown_phase_overseer`. + + #2797 follow-up: collapses the two duplicated closure + definitions at the two post-phase hard-reset emission sites + into one shared factory. The closure remains inside + ``_run_pipeline`` because the ``phase_overseer_active`` + bool is a local nonlocal of this function. + """ + + def _hook() -> None: + nonlocal phase_overseer_active + with overseer_lock: + if container_id and phase_overseer_active: + phase_overseer_active = False + _teardown_phase_overseer( + spawner, + container_id, + pipeline_id, + phase_label=str(phase), + reason=reason, + ) + + return _hook + # Map pipeline to gateway session mode. gateway_mode, detected_visibility = _compute_gateway_mode(pipeline) if not pipeline.network_mode and pipeline.repo: @@ -19852,15 +20674,77 @@ def _run_pipeline( ): prior_phase_succeeded = False - _sync_worktree_with_remote( - spawner, - pipeline_id, - worktree_repo_path, - prior_phase_succeeded=prior_phase_succeeded, - gateway_mode=gateway_mode, - base_branch=pipeline.base_branch, - pipeline_branch=pipeline.branch, - ) + try: + phase_start_sync_outcome = _sync_worktree_with_remote( + spawner, + pipeline_id, + worktree_repo_path, + prior_phase_succeeded=prior_phase_succeeded, + gateway_mode=gateway_mode, + base_branch=pipeline.base_branch, + pipeline_branch=pipeline.branch, + ) + except SyncRebaseAndResetFailedError as sync_terminal_err: + # #2792 review B5: rebase AND hard-reset both failed. + # The worktree is still divergent; we cannot let the + # phase proceed. Route through the shared helper so + # all four FAILED-recovery sites stay in sync (#2797 + # follow-up: drift-risk elimination). + _doubly_failed_msg = ( + f"Sync helper could not reconcile {pipeline.branch} at " + f"{current_phase.value} phase start: {sync_terminal_err}" + ) + logger.error( + "OVERSEER_ALERT worktree_sync_doubly_failed", + pipeline_id=pipeline_id, + phase=current_phase.value, + backup_ref=sync_terminal_err.backup_ref, + discarded_commit_count=len(sync_terminal_err.discarded_commit_shas), + ) + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + store, + phase=current_phase, + error_message=_doubly_failed_msg, + backup_ref=sync_terminal_err.backup_ref, + discarded_commit_shas=sync_terminal_err.discarded_commit_shas, + reset_succeeded=False, + ) + return + + # #2792: phase-start sync fell through to the destructive + # hard-reset recovery. The worktree is now reconciled but + # local-only commits were discarded; pin the pipeline to + # FAILED and surface an HITL ack before any phase work + # spawns against the post-reset state. + if phase_start_sync_outcome.hard_reset_performed: + _hard_reset_msg = ( + f"Sync helper hard-reset {pipeline.branch} to origin at " + f"{current_phase.value} phase start (rebase autoresolve " + f"could not reconcile divergence); " + f"{len(phase_start_sync_outcome.discarded_commit_shas)} " + f"local-only commit(s) preserved under " + f"{phase_start_sync_outcome.backup_ref or '(backup ref write failed)'}" + ) + logger.error( + "OVERSEER_ALERT worktree_sync_hard_reset_recovery", + pipeline_id=pipeline_id, + phase=current_phase.value, + backup_ref=phase_start_sync_outcome.backup_ref, + discarded_commit_count=len(phase_start_sync_outcome.discarded_commit_shas), + ) + # Mirror the post-phase emission site (B3 + B2) via the + # shared helper: same FAILED state, same HITL, same + # pipeline.failed event order (#2797 follow-up). + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + store, + phase=current_phase, + error_message=_hard_reset_msg, + backup_ref=phase_start_sync_outcome.backup_ref, + discarded_commit_shas=phase_start_sync_outcome.discarded_commit_shas, + ) + return # When resuming a stale pipeline branch (cancelled run from # days/weeks ago), rebase origin/ onto origin/ @@ -21405,15 +22289,21 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float = # their on-disk modifications. Running the sync first also # ensures _populate_contract_from_plan can read agent-produced # draft files that only exist on the remote. + post_phase_sync_outcome: WorktreeSyncOutcome | None = None + post_phase_sync_doubly_failed: SyncRebaseAndResetFailedError | None = None if pipeline.branch and worktree_repo_path != repo_path: - # Best-effort: a sync failure must not strand the - # auto-advance. Without this guard, a gateway HTTP error - # or git subprocess failure inside the helper propagates - # to the outer Exception handler and (if marking FAILED - # also fails) leaves the pipeline wedged with phase - # COMPLETE but no successor (#2219). + # Best-effort for transient failures: a sync failure + # must not strand the auto-advance. Without this guard, + # a gateway HTTP error or git subprocess failure inside + # the helper propagates to the outer Exception handler + # and (if marking FAILED also fails) leaves the pipeline + # wedged with phase COMPLETE but no successor (#2219). + # SyncRebaseAndResetFailedError is the structured + # signal that rebase AND hard-reset both failed (#2792 + # review B5) — capture it explicitly so the same HITL + # path runs as the successful-recovery case. try: - _sync_worktree_with_remote( + post_phase_sync_outcome = _sync_worktree_with_remote( spawner, pipeline_id, worktree_repo_path, @@ -21421,6 +22311,8 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float = base_branch=pipeline.base_branch, pipeline_branch=pipeline.branch, ) + except SyncRebaseAndResetFailedError as sync_terminal_err: + post_phase_sync_doubly_failed = sync_terminal_err except Exception as sync_err: logger.warning( "Failed to sync worktree with remote after phase (continuing)", @@ -21429,6 +22321,83 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float = error=str(sync_err), ) + # #2792 review B5: rebase AND hard-reset both failed. The + # worktree is still divergent; populator and decision-sync + # consumers would read stale state. Route through the + # shared FAILED-recovery helper so all four sites stay in + # lockstep (#2797 follow-up). ``pre_event_hook`` tears + # down the per-phase overseer under its own lock between + # the HITL persist and the public ``pipeline.failed`` + # event — same ordering as the inline implementation. + if post_phase_sync_doubly_failed is not None: + _doubly_failed_msg = ( + f"Sync helper could not reconcile {pipeline.branch} after " + f"{current_phase.value} phase: {post_phase_sync_doubly_failed}" + ) + logger.error( + "OVERSEER_ALERT worktree_sync_doubly_failed", + pipeline_id=pipeline_id, + phase=current_phase.value, + backup_ref=post_phase_sync_doubly_failed.backup_ref, + discarded_commit_count=len(post_phase_sync_doubly_failed.discarded_commit_shas), + ) + + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + store, + phase=current_phase, + error_message=_doubly_failed_msg, + backup_ref=post_phase_sync_doubly_failed.backup_ref, + discarded_commit_shas=post_phase_sync_doubly_failed.discarded_commit_shas, + reset_succeeded=False, + pre_event_hook=_make_overseer_teardown_hook( + reason="sync helper rebase+reset doubly failed", + container_id=overseer_container_id, + phase=current_phase, + ), + ) + break + + # #2792: when the post-phase sync fell through to the + # destructive hard-reset recovery, the worktree is now + # reconciled but local agent commits were discarded (pinned + # under ``backup_ref``). Surface this as a dedicated HITL + # ack so the operator decides whether to continue from the + # reconciled state or abort — keeping the populator from + # silently running after a destructive reset matches the + # #2337 / #2627 fail-loud posture. + if post_phase_sync_outcome is not None and post_phase_sync_outcome.hard_reset_performed: + _hard_reset_msg = ( + f"Sync helper hard-reset {pipeline.branch} to origin " + f"after rebase autoresolve failed at {current_phase.value} " + f"phase boundary; " + f"{len(post_phase_sync_outcome.discarded_commit_shas)} " + f"local-only commit(s) preserved under " + f"{post_phase_sync_outcome.backup_ref or '(backup ref write failed)'}" + ) + logger.error( + "OVERSEER_ALERT worktree_sync_hard_reset_recovery", + pipeline_id=pipeline_id, + phase=current_phase.value, + backup_ref=post_phase_sync_outcome.backup_ref, + discarded_commit_count=len(post_phase_sync_outcome.discarded_commit_shas), + ) + + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + store, + phase=current_phase, + error_message=_hard_reset_msg, + backup_ref=post_phase_sync_outcome.backup_ref, + discarded_commit_shas=post_phase_sync_outcome.discarded_commit_shas, + pre_event_hook=_make_overseer_teardown_hook( + reason="sync helper hard-reset recovery", + container_id=overseer_container_id, + phase=current_phase, + ), + ) + break + # After plan phase: populate contract with task structure. # NOTE: worktree_repo_path is used for both draft reads and # contract load/save inside _populate_contract_from_plan. diff --git a/orchestrator/tests/test_advance_phase_thread.py b/orchestrator/tests/test_advance_phase_thread.py index ff73bf07b0..c80dcfdd3e 100644 --- a/orchestrator/tests/test_advance_phase_thread.py +++ b/orchestrator/tests/test_advance_phase_thread.py @@ -437,8 +437,11 @@ def test_sync_worktree_with_remote_is_wrapped(self): # ``try`` whose ``except`` matches ``Exception`` so any failure # mode is swallowed with a warning rather than killing the thread. # Indentation-tolerant: ``\s+`` between ``try:`` and the call. + # Allow either the original direct call or the ``outcome = ...`` + # capture introduced in #2792 (the call is the same; we only + # capture the return now so the hard-reset HITL can fire). assert re.search( - r"try:\s*\n\s*_sync_worktree_with_remote\(", + r"try:\s*\n\s*(?:\w[\w.\[\] |]*\s*=\s*)?_sync_worktree_with_remote\(", source, ), ( "_sync_worktree_with_remote(...) call after BRC return must be " diff --git a/orchestrator/tests/test_hard_reset_recovery.py b/orchestrator/tests/test_hard_reset_recovery.py new file mode 100644 index 0000000000..cca7813aef --- /dev/null +++ b/orchestrator/tests/test_hard_reset_recovery.py @@ -0,0 +1,1185 @@ +"""Tests for the #2792 sync-helper hard-reset recovery surface. + +Covers: + +* ``WorktreeSyncOutcome`` shape returned by + :func:`_sync_worktree_with_remote` on the non-recovery branches + (already-in-sync, behind-only, push-ahead). +* The dedicated HITL question text, options, context discriminator. +* The decision-resolution dispatch hook that wires + ``hard_reset_recovery:`` resolutions to + :func:`resume_pipeline_after_hard_reset_ack` / + :func:`abort_pipeline_after_hard_reset_ack`. + +The end-to-end sync helper subprocess scenarios already live in +``test_sync_worktree.py``; this file focuses on the HITL layer and the +dispatch wiring so a regression in either is caught by a unit test that +doesn't need a real git worktree. +""" + +import subprocess +import sys +import threading +from pathlib import Path +from unittest.mock import MagicMock, patch + +_orchestrator_path = Path(__file__).parent.parent +if str(_orchestrator_path) not in sys.path: + sys.path.insert(0, str(_orchestrator_path)) + +_shared_path = Path(__file__).parent.parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +sys.modules.setdefault("docker", MagicMock()) +sys.modules.setdefault("docker.errors", MagicMock()) +sys.modules.setdefault("docker.types", MagicMock()) + +from gateway_client import PushResult # noqa: E402 +from routes.pipelines import ( # noqa: E402 + WorktreeSyncOutcome, + _build_sync_recovery_backup_ref, + _emit_hard_reset_recovery_hitl, + _fail_pipeline_and_emit_hard_reset_recovery, + _hard_reset_recovery_hitl_options, + _hard_reset_recovery_hitl_question, + _sync_worktree_with_remote, +) + +_PUSH_OK = PushResult(ok=True, category="", detail="") + + +def _make_spawner(fetch_ok: bool = True, push_ok: bool = True) -> MagicMock: + spawner = MagicMock() + spawner.gateway.fetch_worktree_branch.return_value = fetch_ok + spawner.gateway.push_worktree_branch.return_value = ( + _PUSH_OK if push_ok else PushResult(ok=False, category="t", detail="d") + ) + return spawner + + +def _make_subprocess_result( + returncode: int = 0, + stdout: str = "", + stderr: str = "", +) -> subprocess.CompletedProcess: + return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr) + + +class TestBackupRefName: + """``refs/egg-backup/sync-recovery//`` is the contract.""" + + def test_ref_name_layout(self): + ref = _build_sync_recovery_backup_ref("pipeline-abc", 1717000000) + assert ref == "refs/egg-backup/sync-recovery/pipeline-abc/1717000000" + + def test_ref_name_supports_for_each_ref_filter(self): + """A ``for-each-ref refs/egg-backup/sync-recovery/`` filter + must select only this pipeline's backups (the slash-segment + layout, not a flat name with dashes).""" + ref = _build_sync_recovery_backup_ref("pipeline-xyz", 100) + assert ref.startswith("refs/egg-backup/sync-recovery/pipeline-xyz/") + + +class TestSyncOutcomeShape: + """Non-recovery branches return outcomes with hard_reset_performed=False.""" + + def test_already_in_sync_returns_outcome(self): + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = [ + _make_subprocess_result(stdout="egg/issue-42\n"), + _make_subprocess_result(returncode=0), + _make_subprocess_result(stdout="0\t0\n"), + ] + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert isinstance(outcome, WorktreeSyncOutcome) + assert outcome.case == "already_in_sync" + assert outcome.hard_reset_performed is False + assert outcome.backup_ref is None + assert outcome.discarded_commit_shas == () + + def test_behind_only_returns_reset_succeeded(self): + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = [ + _make_subprocess_result(stdout="egg/issue-42\n"), + _make_subprocess_result(returncode=0), + _make_subprocess_result(stdout="0\t3\n"), + _make_subprocess_result(returncode=0), + ] + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert outcome.case == "reset_succeeded" + assert outcome.hard_reset_performed is False + + def test_divergence_rebased_returns_outcome(self): + spawner = _make_spawner() + with ( + patch("routes.pipelines.subprocess.run") as mock_run, + patch("routes.pipelines._rebase_with_agent_output_autoresolve") as mock_rebase, + ): + mock_run.side_effect = [ + _make_subprocess_result(stdout="egg/issue-42\n"), + _make_subprocess_result(returncode=0), + _make_subprocess_result(stdout="2\t3\n"), + ] + mock_rebase.return_value = PushResult(ok=True, category="", detail="") + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert outcome.case == "divergence_rebased" + assert outcome.hard_reset_performed is False + + +class TestHardResetHitlQuestion: + """The HITL question text names the backup ref, lists discarded + SHAs, and exposes exactly two options.""" + + def test_lists_backup_ref_and_discarded_commits(self): + from routes.pipelines import PipelinePhase + + question = _hard_reset_recovery_hitl_question( + pipeline_id="pipeline-zzz", + phase=PipelinePhase.PLAN, + backup_ref="refs/egg-backup/sync-recovery/pipeline-zzz/123", + discarded_commit_shas=("abc1234 add foo", "def5678 add bar"), + ) + assert "refs/egg-backup/sync-recovery/pipeline-zzz/123" in question + assert "abc1234 add foo" in question + assert "def5678 add bar" in question + assert "plan" in question + assert "pipeline-zzz" in question + # Both option labels present in the prose so the SDLC skill + # renders them when no separate options list is shown. + assert "Continue with post-reset state" in question + assert "Abort pipeline" in question + + def test_handles_missing_backup_ref(self): + from routes.pipelines import PipelinePhase + + question = _hard_reset_recovery_hitl_question( + pipeline_id="pipeline-zzz", + phase=PipelinePhase.PLAN, + backup_ref=None, + discarded_commit_shas=(), + ) + # The "backup ref not written" branch must NOT claim a fake ref. + assert "not written" in question + # The "couldn't enumerate" branch is the fallback when rev-list + # itself failed. + assert "could not be enumerated" in question + + def test_options_are_two_distinct_strings(self): + from routes.pipelines import _HARD_RESET_RECOVERY_HITL_OPTIONS + + assert _HARD_RESET_RECOVERY_HITL_OPTIONS == [ + "Continue with post-reset state", + "Abort pipeline", + ] + assert len(_HARD_RESET_RECOVERY_HITL_OPTIONS) == 2 + + +class TestHardResetHitlEmission: + """The emission helper persists with the canonical context discriminator.""" + + def test_context_prefix_used_for_dispatch(self): + """The context must be ``hard_reset_recovery:`` so the + decisions dispatch hook routes on a stable string, not prose.""" + from routes.pipelines import PipelinePhase + + captured: dict = {} + + def fake_persist( + pipeline_id, pipeline, store, *, question, options, phase=None, context=None + ): # noqa: ANN001 + captured["context"] = context + captured["options"] = options + captured["question"] = question + return MagicMock(id="decision-9", context=context) + + with patch("routes.pipelines._persist_hitl_decision", side_effect=fake_persist): + _emit_hard_reset_recovery_hitl( + "pipeline-xyz", + MagicMock(), + MagicMock(), + phase=PipelinePhase.PLAN, + backup_ref="refs/egg-backup/sync-recovery/pipeline-xyz/100", + discarded_commit_shas=("abc1234 foo",), + ) + + assert captured["context"] == "hard_reset_recovery:plan" + assert captured["options"] == [ + "Continue with post-reset state", + "Abort pipeline", + ] + assert "abc1234 foo" in captured["question"] + + +class TestDispatchResolution: + """``_handle_hard_reset_recovery_resolution`` routes Continue/Abort.""" + + def test_continue_triggers_resume_helper(self): + from routes.decisions import _handle_hard_reset_recovery_resolution + + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_abort, + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Continue with post-reset state", + ) + + mock_resume.assert_called_once() + kwargs = mock_resume.call_args.kwargs + assert kwargs["phase_value"] == "plan" + mock_abort.assert_not_called() + + def test_abort_triggers_abort_helper(self): + from routes.decisions import _handle_hard_reset_recovery_resolution + + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_abort, + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Abort pipeline", + ) + + mock_abort.assert_called_once_with("pipeline-abc") + mock_resume.assert_not_called() + + def test_unknown_resolution_is_logged_and_skipped(self): + from routes.decisions import _handle_hard_reset_recovery_resolution + + mock_store = MagicMock() + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_abort, + patch("routes.decisions.logger") as mock_logger, + patch("message_store.get_message_store", return_value=mock_store), + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Something else entirely", + ) + + mock_resume.assert_not_called() + mock_abort.assert_not_called() + # N5: unknown resolution now logs at WARN and emits an + # OVERSEER_ALERT so the operator notices the stuck pipeline. + mock_logger.warning.assert_called() + mock_store.add_message.assert_called_once() + sent_msg = mock_store.add_message.call_args.args[0] + assert sent_msg.message_type == "OVERSEER_ALERT" + assert sent_msg.metadata.get("anomaly") == ("hard_reset_recovery_unknown_resolution") + assert sent_msg.metadata.get("priority") == "high" + + def test_continue_rejected_when_not_in_valid_options(self): + """#2797 follow-up: a "Continue with post-reset state" resolution + on a doubly-failed HITL whose options list collapsed to + ``["Abort pipeline"]`` only must not route to the resume helper + (which would loop straight back into the same divergence). + The cross-check routes the call into the unknown-resolution + path: WARN log + OVERSEER_ALERT, no dispatch. + """ + from routes.decisions import _handle_hard_reset_recovery_resolution + + mock_store = MagicMock() + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_abort, + patch("routes.decisions.logger") as mock_logger, + patch("message_store.get_message_store", return_value=mock_store), + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Continue with post-reset state", + valid_options=["Abort pipeline"], + ) + + mock_resume.assert_not_called() + mock_abort.assert_not_called() + mock_logger.warning.assert_called() + mock_store.add_message.assert_called_once() + sent_msg = mock_store.add_message.call_args.args[0] + assert sent_msg.message_type == "OVERSEER_ALERT" + assert sent_msg.metadata.get("anomaly") == "hard_reset_recovery_unknown_resolution" + assert sent_msg.metadata.get("priority") == "high" + # The alert body should call out the options-list mismatch + # so the operator sees why dispatch was suppressed. + assert "options list" in sent_msg.body + assert "Abort pipeline" in sent_msg.body + + def test_abort_accepted_when_in_valid_options(self): + """The valid-options cross-check must not block a legitimate + Abort on a doubly-failed HITL — "Abort pipeline" is the only + option offered in that branch and must still dispatch. + """ + from routes.decisions import _handle_hard_reset_recovery_resolution + + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_abort, + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Abort pipeline", + valid_options=["Abort pipeline"], + ) + + mock_abort.assert_called_once_with("pipeline-abc") + mock_resume.assert_not_called() + + def test_continue_accepted_when_in_valid_options(self): + """Successful-recovery HITLs offer both options — Continue must + still dispatch to the resume helper when it's in the list. + """ + from routes.decisions import _handle_hard_reset_recovery_resolution + + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_abort, + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Continue with post-reset state", + valid_options=["Continue with post-reset state", "Abort pipeline"], + ) + + mock_resume.assert_called_once() + mock_abort.assert_not_called() + + def test_valid_options_none_keeps_legacy_behavior(self): + """``valid_options=None`` (the default) skips the cross-check — + legacy callers that don't pass options keep dispatching on the + known whitelist alone. + """ + from routes.decisions import _handle_hard_reset_recovery_resolution + + with ( + patch( + "routes.pipelines.resume_pipeline_after_hard_reset_ack", + return_value=True, + ) as mock_resume, + patch( + "routes.pipelines.abort_pipeline_after_hard_reset_ack", + return_value=True, + ), + ): + _handle_hard_reset_recovery_resolution( + "pipeline-abc", + "hard_reset_recovery:plan", + "Continue with post-reset state", + # valid_options omitted → defaults to None + ) + + mock_resume.assert_called_once() + + +class TestEmptyContractHitlWordingNoLongerNamesPriorPopulator: + """#2792: drop the misleading 'populate-from-plan step silently + failed earlier' wording from :func:`_empty_contract_hitl_question`. + + Reason: when the hard-reset recovery fires inline at sync time, the + populator never ran in the first place — there's no 'earlier' step + that silently failed. The current wording invents a phantom + earlier failure that confuses operators investigating the HITL. + """ + + def test_question_text_does_not_blame_phantom_earlier_failure(self): + from routes.pipelines import _empty_contract_hitl_question + + question = _empty_contract_hitl_question( + pipeline_id="p-x", + reason="plan_draft_missing_on_local", + draft_slice_count=None, + gate="plan_complete", + ) + assert "populate-from-plan step silently failed earlier" not in question + # The replacement wording must still tell the operator that + # state and contract have diverged so the next action picker + # has the context it needs. + assert "diverged" in question + + +class TestResumeHelperResetsConsensusAndHealth: + """#2792 review B1: ``resume_pipeline_after_hard_reset_ack`` must + mirror ``restart_phase``'s consensus / restart-count / health-monitor + cleanup so a re-spawn after a post-phase hard reset does not + short-circuit against the prior round's CONFIRMED tracker state and + does not fire stale-elapsed Tier-1 health alerts (#2084 bug class). + """ + + def _make_pipeline_with_agents(self): + from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + PhaseExecution, + Pipeline, + PipelinePhase, + PipelineStatus, + ) + + pipeline = Pipeline( + id="issue-2792", + issue_number=2792, + repo="owner/repo", + branch="egg/issue-2792", + status=PipelineStatus.FAILED, + current_phase=PipelinePhase.IMPLEMENT, + ) + pipeline.phases = { + PipelinePhase.IMPLEMENT.value: PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.FAILED, + error="hard-reset recovery pending", + review_cycles=2, + agents=[ + AgentExecution(role=AgentRole.CODER, status=AgentExecutionStatus.RUNNING), + AgentExecution(role=AgentRole.TESTER, status=AgentExecutionStatus.RUNNING), + AgentExecution(role=AgentRole.DOCUMENTER, status=AgentExecutionStatus.RUNNING), + ], + ), + } + return pipeline + + def test_resume_clears_tracker_evaluator_restart_counts_health(self): + pipeline = self._make_pipeline_with_agents() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + + mock_spawner = MagicMock() + mock_tracker = MagicMock() + mock_evaluator = MagicMock() + mock_hm = MagicMock() + + with ( + patch("routes.pipelines.get_repo_path", return_value=Path("/repo")), + patch("routes.pipelines._resolve_pipeline", return_value=(mock_store, pipeline)), + patch("routes.pipelines.get_pipeline_state_lock"), + patch("routes.pipelines._get_spawner", return_value=mock_spawner), + patch("routes.pipelines._spawn_pipeline_run_thread") as mock_spawn_thread, + patch.dict( + "sys.modules", + { + "peer_consensus": MagicMock( + get_peer_consensus_tracker=MagicMock(return_value=mock_tracker) + ), + "consensus": MagicMock( + get_consensus_evaluator=MagicMock(return_value=mock_evaluator) + ), + "health_monitor": MagicMock(get_health_monitor=MagicMock(return_value=mock_hm)), + }, + ), + ): + from routes.pipelines import resume_pipeline_after_hard_reset_ack + + ok = resume_pipeline_after_hard_reset_ack( + "issue-2792", + phase_value="implement", + ) + + assert ok is True + mock_tracker.clear.assert_called_once() + mock_evaluator.clear.assert_called_once_with("issue-2792") + mock_spawner.reset_restart_counts.assert_called_once_with("issue-2792") + reset_calls = {call.args[0] for call in mock_hm.reset_agent.call_args_list} + assert reset_calls == {"coder", "tester", "documenter"} + mock_spawn_thread.assert_called_once() + + def test_resume_falls_back_to_role_table_when_phase_agents_empty(self): + """When ``phase_exec.agents`` is empty (phase-start hard reset + path), the resume helper must fall back to the deterministic + per-phase roster source so health-monitor cleanup still covers + the roles the next spawn will create.""" + from models import ( + AgentRole, + PhaseExecution, + Pipeline, + PipelinePhase, + PipelineStatus, + ) + + pipeline = Pipeline( + id="issue-2792b", + issue_number=2792, + repo="owner/repo", + branch="egg/issue-2792b", + status=PipelineStatus.FAILED, + current_phase=PipelinePhase.IMPLEMENT, + ) + pipeline.phases = { + PipelinePhase.IMPLEMENT.value: PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.FAILED, + error="hard-reset recovery pending", + agents=[], + ), + } + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + + mock_spawner = MagicMock() + mock_hm = MagicMock() + fake_roles_module = MagicMock() + fake_roles_module.get_roles_for_phase.return_value = [ + AgentRole.CODER, + AgentRole.TESTER, + ] + + with ( + patch("routes.pipelines.get_repo_path", return_value=Path("/repo")), + patch("routes.pipelines._resolve_pipeline", return_value=(mock_store, pipeline)), + patch("routes.pipelines.get_pipeline_state_lock"), + patch("routes.pipelines._get_spawner", return_value=mock_spawner), + patch("routes.pipelines._spawn_pipeline_run_thread"), + patch.dict( + "sys.modules", + { + "peer_consensus": MagicMock( + get_peer_consensus_tracker=MagicMock(return_value=None) + ), + "consensus": MagicMock( + get_consensus_evaluator=MagicMock(return_value=MagicMock()) + ), + "health_monitor": MagicMock(get_health_monitor=MagicMock(return_value=mock_hm)), + "egg_contracts.agent_roles": fake_roles_module, + }, + ), + ): + from routes.pipelines import resume_pipeline_after_hard_reset_ack + + ok = resume_pipeline_after_hard_reset_ack( + "issue-2792b", + phase_value="implement", + ) + + assert ok is True + reset_calls = {call.args[0] for call in mock_hm.reset_agent.call_args_list} + assert reset_calls == {"coder", "tester"} + + def test_resume_returns_false_on_phase_mismatch(self): + """Phase-mismatch (operator resolved a stale recovery decision + after the pipeline already advanced) must not clear consensus — + the active phase would lose live tracker state.""" + pipeline = self._make_pipeline_with_agents() + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + mock_store.repo_path = Path("/repo") + + mock_tracker = MagicMock() + mock_evaluator = MagicMock() + + with ( + patch("routes.pipelines.get_repo_path", return_value=Path("/repo")), + patch("routes.pipelines._resolve_pipeline", return_value=(mock_store, pipeline)), + patch("routes.pipelines.get_pipeline_state_lock"), + patch("routes.pipelines._get_spawner") as mock_get_spawner, + patch("routes.pipelines._spawn_pipeline_run_thread") as mock_spawn, + patch.dict( + "sys.modules", + { + "peer_consensus": MagicMock( + get_peer_consensus_tracker=MagicMock(return_value=mock_tracker) + ), + "consensus": MagicMock( + get_consensus_evaluator=MagicMock(return_value=mock_evaluator) + ), + }, + ), + ): + from routes.pipelines import resume_pipeline_after_hard_reset_ack + + # Pipeline is currently on IMPLEMENT, ack names PLAN. + ok = resume_pipeline_after_hard_reset_ack( + "issue-2792", + phase_value="plan", + ) + + assert ok is False + mock_tracker.clear.assert_not_called() + mock_evaluator.clear.assert_not_called() + mock_get_spawner.assert_not_called() + mock_spawn.assert_not_called() + + +class TestPopulateContractHardResetSurfacing: + """#2797 review B4 / N3: ``populate_contract`` must surface the + destructive recovery the same way the phase-boundary sites do — + pin pipeline+phase to FAILED, emit the hard-reset HITL, broadcast + ``pipeline.failed``, and return 409 — so the operator's ack surface + is uniform across all three triggers (phase-start, post-phase, + populate_contract). + """ + + def _make_app_client(self): + from flask import Flask + from routes.phases import phases_bp + + app = Flask(__name__) + app.register_blueprint(phases_bp) + app.config["TESTING"] = True + return app.test_client() + + def _make_pipeline(self): + from models import Pipeline, PipelinePhase + + pipeline = Pipeline( + id="issue-2792", + issue_number=2792, + repo="owner/repo", + branch="egg/issue-2792", + ) + pipeline.current_phase = PipelinePhase.IMPLEMENT + return pipeline + + def test_hard_reset_recovery_returns_409_and_emits_hitl(self): + """When ``_sync_worktree_with_remote`` reports + ``hard_reset_performed=True``, the route must call the shared + failure-and-HITL helper (so the operator gets the same ack + surface as the phase-boundary sites), skip the populator, and + return HTTP 409 with ``reason="hard_reset_recovery_unacked"``. + """ + from routes.pipelines import WorktreeSyncOutcome + + client = self._make_app_client() + pipeline = self._make_pipeline() + mock_store = MagicMock() + mock_store.repo_path = Path("/home/egg/repos/egg") + + outcome = WorktreeSyncOutcome( + case="divergence_recovered_via_reset", + hard_reset_performed=True, + backup_ref="refs/egg-backup/sync-recovery/issue-2792/123", + discarded_commit_shas=("abc1234 add foo",), + ) + + with ( + patch( + "routes.phases.get_state_store_for_pipeline", + return_value=(mock_store, pipeline), + ), + patch( + "routes.resolve_worktree_path", + return_value=Path("/home/egg/.egg-worktrees/issue-2792/egg"), + ), + patch("routes.pipelines._sync_worktree_with_remote", return_value=outcome), + patch("routes.pipelines._compute_gateway_mode", return_value=("public", None)), + patch("routes.pipelines._get_spawner", return_value=MagicMock()), + patch("routes.pipelines._populate_contract_from_plan") as mock_populate, + patch("routes.pipelines._fail_pipeline_and_emit_hard_reset_recovery") as mock_fail_emit, + ): + resp = client.post("/api/v1/pipelines/issue-2792/phase/populate-contract") + + assert resp.status_code == 409 + import json + + body = json.loads(resp.data) + assert body["success"] is False + assert body["reason"] == "hard_reset_recovery_unacked" + assert body["details"]["hard_reset_performed"] is True + assert body["details"]["backup_ref"] == "refs/egg-backup/sync-recovery/issue-2792/123" + assert body["details"]["discarded_commit_shas"] == ["abc1234 add foo"] + + # Populator MUST NOT run on a worktree that was just hard-reset. + mock_populate.assert_not_called() + # The shared failure-and-HITL helper MUST have been invoked so + # the operator sees the same recovery HITL as the phase-boundary + # sites — without this the 409 body would lie about a HITL that + # was never emitted (the original B4 wording-vs-reality bug). + mock_fail_emit.assert_called_once() + from models import PipelinePhase + + kwargs = mock_fail_emit.call_args.kwargs + assert kwargs["phase"] == PipelinePhase.IMPLEMENT + assert kwargs["backup_ref"] == "refs/egg-backup/sync-recovery/issue-2792/123" + # Helper accepts ``tuple`` or ``list``; the route normalises to + # list (to share the value with the JSON response body). + assert list(kwargs["discarded_commit_shas"]) == ["abc1234 add foo"] + + def test_doubly_failed_returns_409_and_emits_hitl(self): + """When ``_sync_worktree_with_remote`` raises + ``SyncRebaseAndResetFailedError`` (rebase AND hard-reset both + failed — worktree still divergent), the route must call the + shared failure-and-HITL helper, skip the populator, and return + HTTP 409 with ``reason="sync_rebase_and_reset_failed"``. + """ + from routes.pipelines import SyncRebaseAndResetFailedError + + client = self._make_app_client() + pipeline = self._make_pipeline() + mock_store = MagicMock() + mock_store.repo_path = Path("/home/egg/repos/egg") + + terminal_err = SyncRebaseAndResetFailedError( + "rebase failed (conflicts) and hard-reset failed (rc=128, stderr=…)", + backup_ref="refs/egg-backup/sync-recovery/issue-2792/456", + discarded_commit_shas=("def5678 add bar",), + ) + + with ( + patch( + "routes.phases.get_state_store_for_pipeline", + return_value=(mock_store, pipeline), + ), + patch( + "routes.resolve_worktree_path", + return_value=Path("/home/egg/.egg-worktrees/issue-2792/egg"), + ), + patch( + "routes.pipelines._sync_worktree_with_remote", + side_effect=terminal_err, + ), + patch("routes.pipelines._compute_gateway_mode", return_value=("public", None)), + patch("routes.pipelines._get_spawner", return_value=MagicMock()), + patch("routes.pipelines._populate_contract_from_plan") as mock_populate, + patch("routes.pipelines._fail_pipeline_and_emit_hard_reset_recovery") as mock_fail_emit, + ): + resp = client.post("/api/v1/pipelines/issue-2792/phase/populate-contract") + + assert resp.status_code == 409 + import json + + body = json.loads(resp.data) + assert body["success"] is False + assert body["reason"] == "sync_rebase_and_reset_failed" + # ``hard_reset_performed`` is False on this branch because the + # hard reset itself failed (it was attempted but did not + # complete) — distinct from the unacked-recovery case above. + assert body["details"]["hard_reset_performed"] is False + assert body["details"]["backup_ref"] == "refs/egg-backup/sync-recovery/issue-2792/456" + assert body["details"]["discarded_commit_shas"] == ["def5678 add bar"] + + # Populator MUST NOT run on a worktree that is still divergent. + mock_populate.assert_not_called() + # The shared failure-and-HITL helper MUST have been invoked so + # the operator gets the same recovery surface across all three + # triggers of the hard reset (#2797 B4). + mock_fail_emit.assert_called_once() + from models import PipelinePhase + + kwargs = mock_fail_emit.call_args.kwargs + assert kwargs["phase"] == PipelinePhase.IMPLEMENT + assert kwargs["backup_ref"] == "refs/egg-backup/sync-recovery/issue-2792/456" + assert kwargs["discarded_commit_shas"] == ("def5678 add bar",) + # Doubly-failed branch must pass reset_succeeded=False so the + # HITL wording reflects the still-divergent state and the + # "Continue" option is suppressed. + assert kwargs["reset_succeeded"] is False + + +class TestHardResetHitlDoublyFailedBranch: + """#2797 follow-up: when ``SyncRebaseAndResetFailedError`` fires the + HITL question must reflect the still-divergent state and the options + must suppress "Continue" — otherwise the operator is offered a + restart that would loop back into the same failure. + """ + + def test_options_drop_continue_when_reset_failed(self): + # Success branch keeps both options. + assert _hard_reset_recovery_hitl_options(reset_succeeded=True) == [ + "Continue with post-reset state", + "Abort pipeline", + ] + # Doubly-failed branch only offers the abort option. + assert _hard_reset_recovery_hitl_options(reset_succeeded=False) == ["Abort pipeline"] + + def test_question_text_branches_on_reset_succeeded(self): + from routes.pipelines import PipelinePhase + + succeeded = _hard_reset_recovery_hitl_question( + pipeline_id="pipeline-xyz", + phase=PipelinePhase.PLAN, + backup_ref="refs/egg-backup/sync-recovery/pipeline-xyz/100", + discarded_commit_shas=("abc1234 add foo",), + reset_succeeded=True, + ) + failed = _hard_reset_recovery_hitl_question( + pipeline_id="pipeline-xyz", + phase=PipelinePhase.PLAN, + backup_ref="refs/egg-backup/sync-recovery/pipeline-xyz/100", + discarded_commit_shas=("abc1234 add foo",), + reset_succeeded=False, + ) + # Success wording claims reconciliation completed; failed + # wording says the reset itself failed and the worktree is + # still divergent. + assert "hard-reset HEAD to origin to keep downstream" in succeeded + assert "still divergent" not in succeeded + assert "subsequent hard-reset to origin ALSO failed" in failed + assert "still divergent" in failed + # The Continue option label MUST NOT appear in the doubly-failed + # prose — the helper suppresses it and the question shouldn't + # advertise it either (operators copy-paste resolutions). + assert "Continue with post-reset state" not in failed + assert "Abort pipeline" in failed + # Success prose still lists both options. + assert "Continue with post-reset state" in succeeded + assert "Abort pipeline" in succeeded + + def test_emit_passes_reset_succeeded_to_options_and_question(self): + """The emission helper must thread ``reset_succeeded`` through to + both the question builder and the options list — otherwise the + operator could see doubly-failed wording with a "Continue" + option, or vice versa.""" + from routes.pipelines import PipelinePhase + + captured: dict = {} + + def fake_persist( + pipeline_id, pipeline, store, *, question, options, phase=None, context=None + ): # noqa: ANN001 + captured["options"] = options + captured["question"] = question + return MagicMock(id="decision-9", context=context) + + with patch("routes.pipelines._persist_hitl_decision", side_effect=fake_persist): + _emit_hard_reset_recovery_hitl( + "pipeline-xyz", + MagicMock(), + MagicMock(), + phase=PipelinePhase.PLAN, + backup_ref="refs/egg-backup/sync-recovery/pipeline-xyz/100", + discarded_commit_shas=("abc1234 foo",), + reset_succeeded=False, + ) + + assert captured["options"] == ["Abort pipeline"] + assert "still divergent" in captured["question"] + + +class TestFailPipelineAndEmitHelper: + """#2797 follow-up: direct unit test of + :func:`_fail_pipeline_and_emit_hard_reset_recovery` — the previous + tests mocked the helper out at the call sites, so the lock / + save / emit / hook / event sequence had no isolated coverage. + """ + + def _make_pipeline(self): + from models import ( + PhaseExecution, + Pipeline, + PipelinePhase, + PipelineStatus, + ) + + pipeline = Pipeline( + id="issue-2792", + issue_number=2792, + repo="owner/repo", + branch="egg/issue-2792", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + pipeline.phases = { + PipelinePhase.IMPLEMENT.value: PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.RUNNING, + ), + } + return pipeline + + def test_writes_failed_under_lock_then_emits_hitl_and_events(self): + """Happy path: helper acquires the pipeline state lock, writes + ``status=FAILED`` on both pipeline and phase_exec, persists the + HITL (under the same reentrant lock), then broadcasts the + ``pipeline.failed`` event via both the StatusReporter and the + pipeline event stream. + """ + from models import PipelinePhase, PipelineStatus + + pipeline = self._make_pipeline() + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + + call_order: list[str] = [] + + def fake_save(p): # noqa: ANN001 + call_order.append("save_pipeline") + assert p.status == PipelineStatus.FAILED + assert p.error == "boom" + phase_exec = p.get_phase_execution(PipelinePhase.IMPLEMENT) + assert phase_exec is not None + assert phase_exec.status == PipelineStatus.FAILED + assert phase_exec.error == "boom" + assert phase_exec.completed_at is not None + + mock_store.save_pipeline.side_effect = fake_save + + def fake_emit(*args, **kwargs): # noqa: ANN001 + call_order.append("emit_hitl") + return MagicMock() + + def fake_report(*args, **kwargs): # noqa: ANN001 + call_order.append("report_pipeline_status") + + def fake_event(*args, **kwargs): # noqa: ANN001 + call_order.append("emit_pipeline_event") + + with ( + patch( + "routes.pipelines._emit_hard_reset_recovery_hitl", side_effect=fake_emit + ) as m_emit, + patch("routes.pipelines.report_pipeline_status", side_effect=fake_report) as m_report, + patch("routes.pipelines._emit_pipeline_event", side_effect=fake_event) as m_event, + ): + _fail_pipeline_and_emit_hard_reset_recovery( + "issue-2792", + mock_store, + phase=PipelinePhase.IMPLEMENT, + error_message="boom", + backup_ref="refs/egg-backup/sync-recovery/issue-2792/42", + discarded_commit_shas=("abc1234 add foo",), + ) + + # Order matters: FAILED must be persisted before the HITL is + # written, and the HITL must be written before the public + # pipeline.failed event so subscribers reading state on the + # event see both the FAILED status and the pending decision. + assert call_order == [ + "save_pipeline", + "emit_hitl", + "report_pipeline_status", + "emit_pipeline_event", + ] + + emit_kwargs = m_emit.call_args.kwargs + assert emit_kwargs["phase"] == PipelinePhase.IMPLEMENT + assert emit_kwargs["backup_ref"] == "refs/egg-backup/sync-recovery/issue-2792/42" + assert emit_kwargs["discarded_commit_shas"] == ("abc1234 add foo",) + # Default is reset_succeeded=True. + assert emit_kwargs["reset_succeeded"] is True + + report_kwargs = m_report.call_args.kwargs + assert report_kwargs["event_type"] == "pipeline.failed" + assert "boom" in report_kwargs["message"] + + m_event.assert_called_once() + event_args = m_event.call_args.args + assert event_args[1] == "pipeline.failed" + + def test_pre_event_hook_runs_between_hitl_and_public_event(self): + """The post-phase ``_run_pipeline`` sites pass a hook that tears + down the per-phase overseer container — it must run after the + HITL is persisted (so the operator sees the decision before the + container disappears) and before the public event (so observers + don't race the teardown).""" + from models import PipelinePhase + + pipeline = self._make_pipeline() + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + + call_order: list[str] = [] + mock_store.save_pipeline.side_effect = lambda p: call_order.append("save") + hook = MagicMock(side_effect=lambda: call_order.append("hook")) + + with ( + patch( + "routes.pipelines._emit_hard_reset_recovery_hitl", + side_effect=lambda *a, **k: call_order.append("emit_hitl"), + ), + patch( + "routes.pipelines.report_pipeline_status", + side_effect=lambda *a, **k: call_order.append("report"), + ), + patch( + "routes.pipelines._emit_pipeline_event", + side_effect=lambda *a, **k: call_order.append("event"), + ), + ): + _fail_pipeline_and_emit_hard_reset_recovery( + "issue-2792", + mock_store, + phase=PipelinePhase.IMPLEMENT, + error_message="boom", + backup_ref=None, + discarded_commit_shas=(), + pre_event_hook=hook, + ) + + hook.assert_called_once() + assert call_order == ["save", "emit_hitl", "hook", "report", "event"] + + def test_reset_succeeded_false_threads_through_to_emit(self): + """The helper must forward ``reset_succeeded=False`` to the HITL + emission so the question/options reflect the doubly-failed + state — otherwise the operator could see "Continue" on a + worktree that's still divergent.""" + from models import PipelinePhase + + pipeline = self._make_pipeline() + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + + with ( + patch("routes.pipelines._emit_hard_reset_recovery_hitl") as m_emit, + patch("routes.pipelines.report_pipeline_status"), + patch("routes.pipelines._emit_pipeline_event"), + ): + _fail_pipeline_and_emit_hard_reset_recovery( + "issue-2792", + mock_store, + phase=PipelinePhase.IMPLEMENT, + error_message="rebase+reset both failed", + backup_ref="refs/egg-backup/sync-recovery/issue-2792/99", + discarded_commit_shas=("def5678 add bar",), + reset_succeeded=False, + ) + + assert m_emit.call_args.kwargs["reset_succeeded"] is False + + def test_failed_write_and_hitl_persist_held_under_same_lock(self): + """#2797 follow-up: the helper must hold + ``get_pipeline_state_lock(pipeline_id)`` across both the outer + ``store.save_pipeline`` write that pins ``status=FAILED`` *and* + the inner ``_persist_hitl_decision`` save that writes the + decision. A concurrent reader from another thread attempting + to acquire the same lock between the two writes must be + blocked until both have landed — otherwise an observer could + see ``status=FAILED`` without the pending decision (the + invariant the helper exists to enforce). + + This test exercises the real ``threading.RLock`` from + ``state_store.get_pipeline_state_lock`` and the real + ``_persist_hitl_decision`` (only ``store`` and the event + broadcasters are mocked), so a future refactor that + accidentally drops the lock-spanning behavior would be caught + here even if the mock-based call-order tests still pass. + """ + from models import PipelinePhase, PipelineStatus + from routes.pipelines import ( + _persist_hitl_decision, # noqa: F401 # ensure real symbol present + ) + from state_store import get_pipeline_state_lock + + # A unique pipeline_id keeps the per-pipeline lock isolated + # from any other test that may share the module-global lock + # registry in ``state_store``. + pipeline_id = "real-lock-test-issue-2792" + pipeline = self._make_pipeline() + pipeline.id = pipeline_id + + # Real per-pipeline RLock — same instance the helper acquires. + lock = get_pipeline_state_lock(pipeline_id) + + # At every ``save_pipeline`` call, spawn a worker thread that + # tries to acquire the lock non-blocking. RLock allows + # reentrance from the same thread but blocks others — so if + # the helper still holds the lock at that point, the worker's + # ``acquire(blocking=False)`` returns False. + acquisitions_during_save: list[bool] = [] + + def probe_other_thread() -> None: + result = {"acquired": False} + + def attempt() -> None: + if lock.acquire(blocking=False): + try: + result["acquired"] = True + finally: + lock.release() + + t = threading.Thread(target=attempt) + t.start() + t.join(timeout=2.0) + acquisitions_during_save.append(result["acquired"]) + + mock_store = MagicMock() + mock_store.load_pipeline.return_value = pipeline + + def fake_save(_p): # noqa: ANN001 + probe_other_thread() + + mock_store.save_pipeline.side_effect = fake_save + + # Patch only the event broadcasters — let ``_persist_hitl_decision`` + # run for real so its inner ``load → add_decision → save`` is + # the second save call we probe. + with ( + patch("routes.pipelines.report_pipeline_status"), + patch("routes.pipelines._emit_pipeline_event"), + ): + _fail_pipeline_and_emit_hard_reset_recovery( + pipeline_id, + mock_store, + phase=PipelinePhase.IMPLEMENT, + error_message="boom", + backup_ref="refs/egg-backup/sync-recovery/test/123", + discarded_commit_shas=("abc1234 add foo",), + ) + + # Two save_pipeline calls: outer pins FAILED, inner persists + # the HITL. The lock must have been held (un-acquirable from + # another thread) at *both* points. + assert mock_store.save_pipeline.call_count == 2, ( + f"expected 2 save_pipeline calls (outer FAILED + inner HITL); " + f"got {mock_store.save_pipeline.call_count}" + ) + assert acquisitions_during_save == [False, False], ( + f"lock was acquirable from another thread during save_pipeline; " + f"per-call results: {acquisitions_during_save} " + f"(False=held by helper, True=lock dropped between writes)" + ) + + # After the helper returns, the lock must be released so the + # next operator action (e.g. resolve_decision) can acquire it. + assert lock.acquire(blocking=False), ( + "helper did not release pipeline state lock after returning" + ) + lock.release() + + # Sanity-check the persisted state mirrors what the helper + # claimed to write: FAILED + a decision visible on the same + # pipeline object the inner save received. + assert pipeline.status == PipelineStatus.FAILED + assert len(pipeline.decisions) == 1 + assert pipeline.decisions[0].context == "hard_reset_recovery:implement" diff --git a/orchestrator/tests/test_pipeline_failure_path.py b/orchestrator/tests/test_pipeline_failure_path.py index 9cf6cbccc4..a6815a3280 100644 --- a/orchestrator/tests/test_pipeline_failure_path.py +++ b/orchestrator/tests/test_pipeline_failure_path.py @@ -1951,6 +1951,7 @@ def test_stale_branch_error_marks_pipeline_failed_with_actionable_message( from routes.pipelines import ( WORKTREE_BASE_DIR, StalePipelineBranchError, + WorktreeSyncOutcome, _run_pipeline, ) @@ -1970,6 +1971,13 @@ def test_stale_branch_error_marks_pipeline_failed_with_actionable_message( pipeline, ) + # #2792: the helper now returns a structured outcome; the default + # mock would return a MagicMock whose ``hard_reset_performed`` + # attribute is truthy and trip the new hard-reset HITL path. + # Pin a non-recovery outcome so this test stays scoped to the + # ``StalePipelineBranchError`` propagation it's asserting. + mock_sync.return_value = WorktreeSyncOutcome(case="already_in_sync") + worktree_dir = WORKTREE_BASE_DIR / "issue-42" / "repo" mock_gateway.create_worktrees.return_value = MagicMock( success=True, diff --git a/orchestrator/tests/test_sync_worktree.py b/orchestrator/tests/test_sync_worktree.py index 6b0f206656..44f0d58418 100644 --- a/orchestrator/tests/test_sync_worktree.py +++ b/orchestrator/tests/test_sync_worktree.py @@ -224,8 +224,8 @@ def test_diverged_attempts_rebase(self): # No reset was attempted (rebase succeeded → early return) assert mock_run.call_count == 3 - def test_diverged_rebase_fails_signals_error(self): - """(g) diverged + rebase fails → ERROR log with worktree_sync_outcome.""" + def test_diverged_rebase_fails_triggers_hard_reset_recovery(self): + """(g) diverged + rebase fails → backup ref + hard reset (#2792).""" spawner = _make_spawner() with ( patch("routes.pipelines.subprocess.run") as mock_run, @@ -233,22 +233,55 @@ def test_diverged_rebase_fails_signals_error(self): patch("routes.pipelines._rebase_with_agent_output_autoresolve") as mock_rebase, ): mock_run.side_effect = [ + # Step 2: branch name _make_subprocess_result(stdout="egg/issue-42\n"), + # Step 3: rev-parse succeeds _make_subprocess_result(returncode=0), + # Step 3b: local is 2 ahead, 3 behind (diverged) _make_subprocess_result(stdout="2\t3\n"), + # _collect_local_only_commits: rev-list HEAD ^origin + _make_subprocess_result(stdout="abc1234 commit A\ndef5678 commit B\n"), + # _create_sync_recovery_backup_ref: update-ref + _make_subprocess_result(returncode=0), + # Hard reset to origin + _make_subprocess_result(returncode=0), ] mock_rebase.return_value = PushResult( ok=False, category="reconcile_rebase_conflict", detail="conflicts outside agent-outputs", ) - _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) - mock_logger.error.assert_called() - err_kwargs = mock_logger.error.call_args.kwargs - assert err_kwargs.get("case") == "divergence_rebase_failed" - assert err_kwargs.get("category") == "reconcile_rebase_conflict" - # No reset was attempted (rebase failure short-circuits) - assert mock_run.call_count == 3 + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + # Pre-fallback ERROR with the original rebase-failed + # discriminator is still emitted so the audit log shape + # is preserved. + divergence_rebase_failed_err = next( + ( + c + for c in mock_logger.error.call_args_list + if c.kwargs.get("case") == "divergence_rebase_failed" + ), + None, + ) + assert divergence_rebase_failed_err is not None + assert ( + divergence_rebase_failed_err.kwargs.get("category") == "reconcile_rebase_conflict" + ) + # The hard-reset recovery path fired and reconciled. + assert outcome.case == "divergence_recovered_via_reset" + assert outcome.hard_reset_performed is True + assert outcome.backup_ref is not None + assert outcome.backup_ref.startswith("refs/egg-backup/sync-recovery/pipe-1/") + assert outcome.discarded_commit_shas == ( + "abc1234 commit A", + "def5678 commit B", + ) + # 3 (head detect / rev-parse / rev-list-counts) + 3 (rev-list local-only, + # update-ref, reset) = 6 subprocess calls. + assert mock_run.call_count == 6 + reset_call = mock_run.call_args_list[5] + assert "reset" in reset_call[0][0] + assert "--hard" in reset_call[0][0] def test_successful_reset(self): """Happy path: fetch, detect branch, verify remote, reset.""" @@ -301,8 +334,8 @@ def test_handles_subprocess_timeout(self): # Should not raise _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) - def test_diverged_rebase_returns_failure_result(self): - """Diverged + rebase helper returns failure → no reset, ERROR logged.""" + def test_diverged_rebase_failure_runs_hard_reset_even_on_timeout(self): + """Diverged + rebase timeout still triggers the hard-reset recovery (#2792).""" spawner = _make_spawner() with ( patch("routes.pipelines.subprocess.run") as mock_run, @@ -313,16 +346,23 @@ def test_diverged_rebase_returns_failure_result(self): _make_subprocess_result(stdout="egg/issue-42\n"), _make_subprocess_result(returncode=0), _make_subprocess_result(stdout="2\t3\n"), + # _collect_local_only_commits returns empty (parse failure) + _make_subprocess_result(returncode=0, stdout=""), + # backup-ref update-ref succeeds + _make_subprocess_result(returncode=0), + # Hard reset succeeds + _make_subprocess_result(returncode=0), ] mock_rebase.return_value = PushResult( ok=False, category="reconcile_rebase_timeout", detail="timed out", ) - _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) mock_logger.error.assert_called() - # No reset attempt — failed rebase short-circuits the function - assert mock_run.call_count == 3 + assert outcome.hard_reset_performed is True + assert outcome.case == "divergence_recovered_via_reset" + assert mock_run.call_count == 6 def test_rev_list_non_numeric_output_defaults_to_reset(self): """Non-numeric rev-list output falls through to reset.""" @@ -620,8 +660,13 @@ def test_case_diverged_rebased(self): _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) assert _outcome_cases(mock_logger) == ["divergence_rebased"] - def test_case_diverged_rebase_failed(self): - """Rebase failure emits divergence_rebase_failed; the function returns without falling through to reset.""" + def test_case_diverged_rebase_failed_then_recovers_via_reset(self): + """Rebase failure now triggers hard-reset recovery (#2792). + + The original ``divergence_rebase_failed`` ERROR is still emitted + so audit-log greps for that discriminator keep working, but the + terminal outcome is ``divergence_recovered_via_reset``. + """ spawner = _make_spawner() with ( patch("routes.pipelines.subprocess.run") as mock_run, @@ -632,14 +677,109 @@ def test_case_diverged_rebase_failed(self): _make_subprocess_result(stdout="egg/issue-42\n"), _make_subprocess_result(returncode=0), _make_subprocess_result(stdout="2\t3\n"), + # _collect_local_only_commits + _make_subprocess_result(stdout="abc1234 foo\n"), + # _create_sync_recovery_backup_ref + _make_subprocess_result(returncode=0), + # Hard reset + _make_subprocess_result(returncode=0), ] mock_rebase.return_value = PushResult( ok=False, category="reconcile_rebase_conflict", detail="conflict", ) - _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) - assert _outcome_cases(mock_logger) == ["divergence_rebase_failed"] + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert _outcome_cases(mock_logger) == [ + "divergence_rebase_failed", + "divergence_recovered_via_reset", + ] + assert outcome.hard_reset_performed is True + + def test_case_diverged_rebase_and_reset_failed(self): + """Rebase fails AND the hard-reset fallback also fails → raises + ``SyncRebaseAndResetFailedError`` with the backup ref and the + discarded SHA list attached (#2792 review B5).""" + import pytest + from routes.pipelines import SyncRebaseAndResetFailedError + + spawner = _make_spawner() + with ( + patch("routes.pipelines.subprocess.run") as mock_run, + patch("routes.pipelines.logger") as mock_logger, + patch("routes.pipelines._rebase_with_agent_output_autoresolve") as mock_rebase, + ): + mock_run.side_effect = [ + _make_subprocess_result(stdout="egg/issue-42\n"), + _make_subprocess_result(returncode=0), + _make_subprocess_result(stdout="2\t3\n"), + # _collect_local_only_commits + _make_subprocess_result(stdout="abc1234 foo\n"), + # _create_sync_recovery_backup_ref + _make_subprocess_result(returncode=0), + # Hard reset fails + _make_subprocess_result(returncode=1, stderr="reset blew up"), + ] + mock_rebase.return_value = PushResult( + ok=False, + category="reconcile_rebase_conflict", + detail="conflict", + ) + with pytest.raises(SyncRebaseAndResetFailedError) as exc_info: + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert _outcome_cases(mock_logger) == [ + "divergence_rebase_failed", + "divergence_rebase_and_reset_failed", + ] + # Backup ref was written successfully even though the reset + # itself failed, so it's still surfaced for forensic use. + assert exc_info.value.backup_ref is not None + assert exc_info.value.discarded_commit_shas == ("abc1234 foo",) + + def test_case_diverged_recovered_via_reset_with_backup_failure(self): + """Backup-ref write failure → reset still runs, ``backup_ref=None`` on outcome.""" + spawner = _make_spawner() + with ( + patch("routes.pipelines.subprocess.run") as mock_run, + patch("routes.pipelines.logger") as mock_logger, + patch("routes.pipelines._rebase_with_agent_output_autoresolve") as mock_rebase, + ): + mock_run.side_effect = [ + _make_subprocess_result(stdout="egg/issue-42\n"), + _make_subprocess_result(returncode=0), + _make_subprocess_result(stdout="2\t3\n"), + # _collect_local_only_commits succeeds — non-empty so the + # WARN inlines them when backup write fails. + _make_subprocess_result(stdout="abc1234 foo\ndef5678 bar\n"), + # _create_sync_recovery_backup_ref fails (e.g. disk error) + _make_subprocess_result(returncode=1, stderr="ref write failed"), + # Hard reset still runs and succeeds + _make_subprocess_result(returncode=0), + ] + mock_rebase.return_value = PushResult( + ok=False, + category="reconcile_rebase_conflict", + detail="conflict", + ) + outcome = _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + # Reset still ran; outcome flags hard_reset_performed but + # backup_ref is None and the SHA list is inlined into the WARN. + assert outcome.hard_reset_performed is True + assert outcome.backup_ref is None + assert outcome.discarded_commit_shas == ( + "abc1234 foo", + "def5678 bar", + ) + inlined = [ + c + for c in mock_logger.warning.call_args_list + if c.args and "Sync-recovery hard reset proceeding without backup ref" in c.args[0] + ] + assert inlined, "Expected the inlined-SHAs WARN when backup-ref write fails" + assert inlined[0].kwargs.get("discarded_commit_shas") == [ + "abc1234 foo", + "def5678 bar", + ] def test_divergence_with_base_branch_none_logs_contamination_warning(self): """#2337 review S7: divergence rebase with ``base_branch=None``