fix(orchestrator): unwrap choice envelope in hard-reset recovery HITL dispatch (#2978)

orchestrator/README.md

@@ -88,7 +88,7 @@ Before each pipeline phase starts, the orchestrator syncs the agent worktree wit
 
 - **Prior phase succeeded + local ahead of remote:** Local commits are pushed to remote before resetting, preserving completed work.
 - **Prior phase failed + local ahead of remote:** Local commits are discarded and the worktree is reset to remote, removing incomplete work.
-- **Local and remote diverged:** A fast-forward merge is attempted. If the merge fails, the orchestrator logs an error and leaves the worktree unchanged (may require manual intervention).
+- **Local and remote diverged:** Local commits are rebased onto `origin/<branch>`, auto-resolving conflicts confined to `.egg-state/agent-outputs/` in favor of the remote. If the rebase cannot reconcile (a conflict outside that path), the orchestrator pins the local-only commits under a backup ref (`refs/egg-backup/sync-recovery/<pipeline_id>/<ts>`), hard-resets the worktree to `origin/<branch>`, marks the pipeline FAILED, and surfaces a hard-reset recovery HITL ack (#2792/#2797). _Note: this destructive recovery is being redesigned to reconcile non-destructively — see #2979._
 - **Local behind or in-sync with remote:** Standard reset to remote tip.
 
 ### HITL Decisions

orchestrator/routes/decisions.py

@@ -150,6 +150,40 @@ def _handle_restart_agent(pipeline_id: str, question: str) -> None:
         )
 
 
+def _normalize_choice_resolution(resolution: str) -> str:
+    """Unwrap a structured ``choice`` envelope to its bare option label (#2978).
+
+    The local SDLC HITL CLI resolves a ``choice`` decision by sending
+    ``{"action": "select", "selected": "<option>"}`` (see
+    ``sandbox/egg_lib/sdlc_hitl.py``); :func:`resolve_decision`
+    JSON-serializes that dict into ``decision.resolution``.  Dispatch
+    hooks that compare the resolution against bare option labels must
+    unwrap the envelope first — otherwise every structured selection
+    reads as an unrecognized option.  Bare-string resolutions (legacy /
+    direct-API callers) and any other shape pass through unchanged so
+    the caller's existing matching still runs.
+
+    Audit-trail note: this is a dispatch-side unwrap only.  The
+    persisted ``decision.resolution`` (and the ``DECISION_RESOLVED``
+    event / API response payload) still carries the raw envelope JSON
+    as the operator sent it.  Only the in-process value routed to the
+    Restart-agent / Continue-without / conditional-ACK / hard-reset
+    dispatch helpers — and any subsequent log line that echoes it — is
+    the normalized form.
+    """
+    if not resolution:
+        return resolution
+    try:
+        payload = json.loads(resolution)
+    except json.JSONDecodeError:
+        return resolution
+    if isinstance(payload, dict) and payload.get("action") == "select":
+        selected = payload.get("selected")
+        if isinstance(selected, str):
+            return selected
+    return resolution
+
+
 def _handle_hard_reset_recovery_resolution(
     pipeline_id: str,
     context: str,
@@ -191,10 +225,23 @@ def _handle_hard_reset_recovery_resolution(
     """
     phase_value = context.removeprefix("hard_reset_recovery:")
 
+    # #2978: the SDLC HITL CLI resolves a ``choice`` decision by sending a
+    # ``{"action": "select", "selected": "<option>"}`` envelope, which
+    # ``resolve_decision`` serializes into ``decision.resolution``.  Unwrap
+    # it to the bare option label BEFORE the ``valid_options`` cross-check
+    # and the Continue/Abort compares below — otherwise the JSON string
+    # matches neither, every structured selection routes to the
+    # unrecognized-option path, and the pipeline stays wedged in
+    # ``failed_pending_hitl`` (the phase-gate path already unwraps this
+    # envelope in ``routes.pipelines``).
+    resolution = _normalize_choice_resolution(resolution)
+
     # 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 (
+        _HARD_RESET_RECOVERY_ABORT,
+        _HARD_RESET_RECOVERY_CONTINUE,
         abort_pipeline_after_hard_reset_ack,
         resume_pipeline_after_hard_reset_ack,
     )
@@ -214,7 +261,7 @@ def _handle_hard_reset_recovery_resolution(
             f"(received: {resolution[:80]!r}; available: {valid_options!r})"
         )
 
-    if option_mismatch_reason is None and resolution == "Continue with post-reset state":
+    if option_mismatch_reason is None and resolution == _HARD_RESET_RECOVERY_CONTINUE:
         ok = resume_pipeline_after_hard_reset_ack(
             pipeline_id,
             phase_value=phase_value,
@@ -225,7 +272,7 @@ def _handle_hard_reset_recovery_resolution(
                 pipeline_id=pipeline_id,
                 phase=phase_value,
             )
-    elif option_mismatch_reason is None and resolution == "Abort pipeline":
+    elif option_mismatch_reason is None and resolution == _HARD_RESET_RECOVERY_ABORT:
         ok = abort_pipeline_after_hard_reset_ack(pipeline_id)
         if not ok:
             logger.warning(
@@ -260,7 +307,8 @@ def _handle_hard_reset_recovery_resolution(
                 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 "
+                    f"{_HARD_RESET_RECOVERY_CONTINUE!r} or "
+                    f"{_HARD_RESET_RECOVERY_ABORT!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."
@@ -334,6 +382,13 @@ def _handle_conditional_ack_gate(
         CONDITIONAL_ACK_REJECT,
     )
 
+    # #2978: defense-in-depth — unwrap the ``choice`` envelope so a future
+    # direct caller bypassing ``resolve_decision``'s dispatch-boundary
+    # normalization still sees the bare option label below.  Idempotent on
+    # already-unwrapped strings; mirrors the same call in
+    # ``_handle_hard_reset_recovery_resolution``.
+    resolution = _normalize_choice_resolution(resolution)
+
     if not context.startswith(CONDITIONAL_ACK_GATE_MARKER):
         return
     payload_str = context[len(CONDITIONAL_ACK_GATE_MARKER) :]
@@ -893,6 +948,16 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
             source=getattr(request, "egg_source", "unknown"),
         )
 
+        # #2978: normalize the choice envelope once at the dispatch
+        # boundary so every dispatch hook below sees the bare option
+        # label instead of the ``{"action": "select", "selected": ...}``
+        # envelope the SDLC HITL CLI sends.  ``decision.resolution``
+        # (persisted on disk, emitted on ``DECISION_RESOLVED``, and
+        # returned in the API response) is intentionally unchanged —
+        # the audit trail keeps the raw envelope while dispatch routes
+        # on the unwrapped label.
+        dispatch_resolution = _normalize_choice_resolution(decision.resolution or "")
+
         try:
             emit_event(
                 EventType.DECISION_RESOLVED,
@@ -915,25 +980,25 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
         #   "Agent <role> issue: <message>"
         # When the human resolves with "Restart agent", stop the old
         # container and respawn a replacement.
-        if decision.resolution == "Restart agent":
+        if dispatch_resolution == "Restart agent":
             _handle_restart_agent(pipeline_id, decision.question)
 
         # Handle the conditional-ACK 3-way HITL gate (#2004). Context
         # prefix is the discriminator — the question text is arbitrary
         # prose and mustn't be relied on for dispatch.
-        if decision.context and decision.resolution:
+        if decision.context and dispatch_resolution:
             _handle_conditional_ack_gate(
                 pipeline_id,
                 decision.context,
-                decision.resolution,
+                dispatch_resolution,
                 store.repo_path,
             )
 
         # Handle "Continue without" resolution for failed reviewer decisions.
         # The concurrent executor stores "failed_role:<role>" in the decision
         # context when a reviewer crashes. Excuse the reviewer so consensus
         # can proceed without their ACK.
-        if decision.resolution == "Continue without" and decision.context.startswith(
+        if dispatch_resolution == "Continue without" and decision.context.startswith(
             "failed_role:"
         ):
             failed_role = decision.context.removeprefix("failed_role:")
@@ -969,7 +1034,7 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
             _handle_hard_reset_recovery_resolution(
                 pipeline_id,
                 decision.context,
-                decision.resolution or "",
+                dispatch_resolution,
                 valid_options=list(decision.options or []),
             )
 

orchestrator/tests/test_conditional_ack_hitl_gate.py

@@ -361,6 +361,115 @@ def test_address_in_pipeline_invalidates_acks(
         assert resp.status_code == 200
         mock_invalidate.assert_called_once_with("pipeline-x", conditions)
 
+    @patch("routes.decisions._persist_deferred_actions")
+    @patch("routes.decisions.get_state_store_for_pipeline")
+    @patch("routes.decisions.get_decision_queue")
+    def test_approve_envelope_persists_deferred_actions(
+        self,
+        mock_get_queue,
+        mock_get_store_for_pipeline,
+        mock_persist,
+        client,
+        tmp_path,
+    ):
+        """#2978: the SDLC HITL CLI wraps the choice as
+        ``{"action": "select", "selected": "<option>"}``.  The dispatch
+        must unwrap that envelope before comparing against
+        ``CONDITIONAL_ACK_APPROVE`` — otherwise the operator's selection
+        falls into the unrecognized-option branch and the obligations
+        are never written to ``contract.pr.deferred_actions``.
+        """
+        conditions = [
+            {"reviewer": "reviewer_code", "producer": "coder", "condition": "git mv X Y"},
+        ]
+        envelope = json.dumps({"action": "select", "selected": CONDITIONAL_ACK_APPROVE})
+        resolved = HITLDecision(
+            id="decision-1",
+            question="conditional ACK",
+            status=DecisionStatus.RESOLVED,
+            resolution=envelope,
+            context=self._gate_context(conditions),
+        )
+        mock_store = MagicMock(repo_path=tmp_path)
+        mock_get_store_for_pipeline.return_value = (mock_store, MagicMock())
+        mock_get_queue.return_value.resolve_decision.return_value = resolved
+
+        resp = client.post(
+            "/api/v1/pipelines/pipeline-x/decisions/decision-1/resolve",
+            json={"resolution": {"action": "select", "selected": CONDITIONAL_ACK_APPROVE}},
+        )
+        assert resp.status_code == 200
+        mock_persist.assert_called_once_with("pipeline-x", conditions, tmp_path)
+
+    @patch("routes.decisions._force_nack_conditional_edges")
+    @patch("routes.decisions.get_state_store_for_pipeline")
+    @patch("routes.decisions.get_decision_queue")
+    def test_reject_envelope_force_nacks_edges(
+        self,
+        mock_get_queue,
+        mock_get_store_for_pipeline,
+        mock_force_nack,
+        client,
+        tmp_path,
+    ):
+        """Reject through the envelope must still drive force-NACK."""
+        conditions = [
+            {"reviewer": "reviewer_code", "producer": "coder", "condition": "git mv X Y"},
+        ]
+        envelope = json.dumps({"action": "select", "selected": CONDITIONAL_ACK_REJECT})
+        resolved = HITLDecision(
+            id="decision-1",
+            question="conditional ACK",
+            status=DecisionStatus.RESOLVED,
+            resolution=envelope,
+            context=self._gate_context(conditions),
+        )
+        mock_store = MagicMock(repo_path=tmp_path)
+        mock_get_store_for_pipeline.return_value = (mock_store, MagicMock())
+        mock_get_queue.return_value.resolve_decision.return_value = resolved
+
+        resp = client.post(
+            "/api/v1/pipelines/pipeline-x/decisions/decision-1/resolve",
+            json={"resolution": {"action": "select", "selected": CONDITIONAL_ACK_REJECT}},
+        )
+        assert resp.status_code == 200
+        mock_force_nack.assert_called_once_with("pipeline-x", conditions)
+
+    @patch("routes.decisions._invalidate_conditional_acks")
+    @patch("routes.decisions.get_state_store_for_pipeline")
+    @patch("routes.decisions.get_decision_queue")
+    def test_address_envelope_invalidates_acks(
+        self,
+        mock_get_queue,
+        mock_get_store_for_pipeline,
+        mock_invalidate,
+        client,
+        tmp_path,
+    ):
+        """Address-in-pipeline through the envelope must still
+        invalidate the conditioning ACK so the producer re-proposes."""
+        conditions = [
+            {"reviewer": "reviewer_code", "producer": "coder", "condition": "git mv X Y"},
+        ]
+        envelope = json.dumps({"action": "select", "selected": CONDITIONAL_ACK_ADDRESS})
+        resolved = HITLDecision(
+            id="decision-1",
+            question="conditional ACK",
+            status=DecisionStatus.RESOLVED,
+            resolution=envelope,
+            context=self._gate_context(conditions),
+        )
+        mock_store = MagicMock(repo_path=tmp_path)
+        mock_get_store_for_pipeline.return_value = (mock_store, MagicMock())
+        mock_get_queue.return_value.resolve_decision.return_value = resolved
+
+        resp = client.post(
+            "/api/v1/pipelines/pipeline-x/decisions/decision-1/resolve",
+            json={"resolution": {"action": "select", "selected": CONDITIONAL_ACK_ADDRESS}},
+        )
+        assert resp.status_code == 200
+        mock_invalidate.assert_called_once_with("pipeline-x", conditions)
+
     @patch("routes.decisions._persist_deferred_actions")
     @patch("routes.decisions._force_nack_conditional_edges")
     @patch("routes.decisions._invalidate_conditional_acks")