Fix #2792: auto-recover from sync divergence with HITL ack

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:<phase>``; ``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:<phase>``; 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={

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,