fix(hitl): dedupe contract cq-N decisions + surface them in get_status

docs/hitl-decisions.md

@@ -183,12 +183,41 @@ router. Like `feedback-N`, those decisions are bridged into the orchestrator
 queue only *after* the phase gate is approved. An agent blocked on such a
 question before proposing never reaches the gate, so:
 
-- the decision does not appear in `get_status(...).pending_decisions`;
+- the decision does not appear in `get_status(...).pending_decisions` (which
+  lists only orchestrator-queue decisions);
 - calling `provide_input(decision_id="cq-N", ...)` previously returned
   **HTTP 404** (not in the queue), leaving the operator with no resolution
   channel and the pipeline deadlocking (#3071, observed on
   pipeline-c2faf164).
 
+**Surfacing (#3374).** Unresolved `cq-N` HITL decisions that are not yet
+bridged into the queue are surfaced in their own `get_status(...)` field,
+`pending_contract_decisions` — each entry carries `id`, `question`, `phase`,
+`options`, and `scope: "contract"`, plus `type: "hitl"` (mirroring the
+queue-decision shape) and a `note` string pointing the operator at the
+`provide_input` resolution flow. It is kept distinct from `pending_decisions`
+so the two-wave resolve flow is unaffected; an operator driving via
+`get_status` no longer has to call `get_contract` out of band to discover them.
+Already-bridged questions are filtered out of this field so a mirrored `cq-N`
+is not listed twice.
+
+**Deduplication (#3374).** A re-registration of a question already open and
+unresolved *under the same phase* adopts the existing `cq-N` rather than
+minting a duplicate: `register_open_question` and the impasse-escalation
+router both dedupe on the normalized question text **keyed by phase** via
+`egg_contracts.decisions.find_duplicate_open_question`. This covers the
+observed repro — a re-run agent, or a re-escalated impasse, re-asking within a
+single phase — and any re-ask explicitly tagged for that same phase. A genuine
+*cross-phase* re-ask (a different phase tag) is treated as a distinct question
+and mints a fresh `cq-N` by design, because its decision context differs: in
+the agent path the phase defaults to the contract's `current_phase`, so a
+question first posed in `refine` and re-posed in `plan` does **not** dedupe.
+The dedup key is (question, phase) only — the option set is not part of it, so
+a re-registration carrying a *different* option set adopts the stored options
+and logs a warning (the new options are not merged). The registration is
+idempotent — the response carries `deduped: true` and no second contract
+write occurs.
+
 `provide_input` now falls back to the contract when the id is not found in
 the queue. It writes the resolution fields straight onto the contract
 (`resolved=True`, `resolved_by="human"`, stripped resolution string), so the
@@ -321,6 +350,8 @@ Two complementary bridges ensure contract-scoped decisions created by agents via
 
 **Client-side bridge (prompt-driven mode only):** In prompt-driven mode, the phase gate menu displays a `[q] Answer open questions` option when unanswered decisions exist in the contract JSON, letting humans respond from the terminal before approving. Approving a phase gate with unanswered questions triggers a warning prompt.
 
+**Gate-approval guard (#3374):** The bridge only promotes the *current* phase's `cq-N`; questions tagged for a later phase remain unanswered and otherwise invisible until that phase's gate. To avoid narrating an approval as "nothing else pending" while such questions sit outstanding, the `provide_input` response for a resolved `phase_gate` includes an `outstanding_contract_decisions` list of those later-phase unresolved `cq-N` (`id` / `question` / `phase`). They also remain visible in every `get_status` snapshot under `pending_contract_decisions`.
+
 ### Draft Document Display
 
 When multiple HITL decisions are pending (e.g., agent-created choice/feedback questions plus the phase gate approval), the CLI presents them in FIFO order. To ensure humans have context when answering agent questions before seeing the phase gate:

orchestrator/impasse_routing.py

@@ -37,7 +37,7 @@
     sys.path.insert(0, str(_shared_path))
 
 from egg_contracts.agent_roles import AgentRole as ContractAgentRole
-from egg_contracts.decisions import next_cq_id
+from egg_contracts.decisions import find_duplicate_open_question, next_cq_id
 from egg_contracts.impasse import Impasse, ImpasseCategory
 from egg_contracts.loader import load_contract, save_contract
 from egg_contracts.models import (
@@ -527,6 +527,37 @@ def _record_escalate(
     field_path, decision = _build_hitl_decision(
         contract, slice_obj, task, impasse, impassed_role, reason
     )
+
+    # Dedupe: a re-escalation of an impasse whose HITL question is already
+    # open and unanswered should adopt the existing ``cq-N`` rather than
+    # append an identical one (parity with ``register_open_question``).
+    # ``_build_hitl_decision`` did not mutate the contract, so scan the
+    # current decisions for an equivalent open question.
+    duplicate = find_duplicate_open_question(contract.decisions, decision.question, decision.phase)
+    if duplicate is not None:
+        # The dedup key is (normalized question, phase); the option set is
+        # not. If the re-escalation produced a different option set, the
+        # operator only ever sees the stored options — log it so the loss
+        # is not invisible (#3374 review).
+        new_labels = [getattr(o, "label", None) for o in (decision.options or [])]
+        existing_labels = [getattr(o, "label", None) for o in (duplicate.options or [])]
+        logger.info(
+            "Impasse escalation deduped onto existing HITL decision",
+            slice_id=slice_obj.id if slice_obj else None,
+            task_id=task.id if task else impasse.task_id,
+            decision_id=duplicate.id,
+            options_differ=new_labels != existing_labels,
+        )
+        return RoutingDecision(
+            action=ImpasseAction.ESCALATE,
+            impasse=impasse,
+            role=impassed_role,
+            task_id=task.id if task else impasse.task_id,
+            new_role=None,
+            reason=reason,
+            hitl_decision_id=duplicate.id,
+        )
+
     # decisions.* is owned by IMPLEMENTER per FIELD_OWNERSHIP — mirror
     # the existing ``register_open_question`` MCP tool's role choice.
     # The audit log records the orchestrator-side actor so it stays

orchestrator/mcp_tools/__init__.py

@@ -149,6 +149,7 @@ def __init__(
     _handle_get_status = _status._handle_get_status
     _live_running_agents_fallback = _status._live_running_agents_fallback
     _build_status_snapshot = _status._build_status_snapshot
+    _pending_contract_decisions = _status._pending_contract_decisions
     _enrich_pending_decisions = _status._enrich_pending_decisions
     _read_reviewer_feedback = _status._read_reviewer_feedback
     _handle_get_phase = _status._handle_get_phase

orchestrator/mcp_tools/_status.py

@@ -17,6 +17,7 @@
 from typing import Any
 from urllib.parse import quote
 
+from egg_contracts.decisions import truncate_question
 from mcp_tools import logger
 
 
@@ -216,9 +217,102 @@ def _build_status_snapshot(self, raw_task_id: str) -> dict[str, Any]:
     # Enrichment: attach draft content to pending decisions (optional)
     self._enrich_pending_decisions(status, raw_task_id, pipeline_data)
 
+    # Surface contract-resident HITL decisions (``cq-N``) that the
+    # orchestrator queue does not yet know about. Agents register these
+    # via ``register_open_question``; they live on the SDLC contract and
+    # only reach ``pending_decisions`` (above) once the post-gate bridge
+    # promotes them. Until then they gate upcoming work but are invisible
+    # to an operator driving via ``get_status`` / ``wait-status`` — they
+    # had to call ``get_contract`` out of band to find them. Expose them
+    # as a sibling field (kept distinct from the queue so the documented
+    # two-wave resolve flow is unaffected). Issued last and best-effort:
+    # any failure leaves the field absent rather than breaking the
+    # snapshot. ``task_id`` is already URL-quoted above.
+    contract_pending = self._pending_contract_decisions(task_id, status["pending_decisions"])
+    if contract_pending:
+        status["pending_contract_decisions"] = contract_pending
+
     return status
 
 
+def _pending_contract_decisions(
+    self,
+    task_id: str,
+    queue_pending: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Return unresolved contract-resident HITL (``cq-N``) decisions.
+
+    ``task_id`` must already be URL-quoted (the caller quotes it once).
+    Reads the SDLC contract via the orchestrator's ``/api/v1/contracts``
+    endpoint and returns the unresolved ``type == "hitl"`` decisions that
+    are *not already mirrored* into the orchestrator queue — the post-gate
+    bridge mirrors a ``cq-N`` into a fresh ``decision-M`` whose context is
+    ``"Open contract question {cq-id}, ..."`` and the contract entry stays
+    unresolved until that queue decision is answered and written back, so
+    without this filter a bridged question would appear twice.
+
+    Best-effort: returns ``[]`` on any failure (missing contract, request
+    error) so a status snapshot is never broken by this enrichment.
+    """
+    try:
+        result = self._make_request(f"/api/v1/contracts/{task_id}?pipeline_id={task_id}")
+    except Exception as e:
+        logger.debug(
+            "Contract fetch for pending-decision surfacing failed",
+            task_id=task_id,
+            error=str(e),
+        )
+        return []
+
+    contract = result.get("data") if isinstance(result.get("data"), dict) else result
+    if not isinstance(contract, dict):
+        return []
+    decisions = contract.get("decisions") or []
+
+    # Ids already bridged into the queue (so we don't double-list them).
+    bridged_ids: set[str] = set()
+    for q in queue_pending:
+        context = q.get("context") or ""
+        match = re.match(r"Open contract question (cq-\d+),", context)
+        if match:
+            bridged_ids.add(match.group(1))
+
+    surfaced: list[dict[str, Any]] = []
+    for d in decisions:
+        if not isinstance(d, dict):
+            continue
+        if d.get("type") != "hitl" or d.get("resolved"):
+            continue
+        decision_id = d.get("id", "")
+        if not isinstance(decision_id, str) or not decision_id.startswith("cq-"):
+            continue
+        if decision_id in bridged_ids:
+            continue
+        question = d.get("question") or ""
+        if isinstance(question, str):
+            question = truncate_question(question)
+        surfaced.append(
+            {
+                "id": decision_id,
+                "question": question,
+                "phase": d.get("phase"),
+                "type": "hitl",
+                "options": [
+                    opt.get("label", "")
+                    for opt in (d.get("options") or [])
+                    if isinstance(opt, dict)
+                ],
+                "scope": "contract",
+                "note": (
+                    "Agent-registered contract question, not yet bridged into the "
+                    "decision queue. Resolve with the same decision-resolve flow "
+                    "(provide_input) using this id."
+                ),
+            }
+        )
+    return surfaced
+
+
 def _enrich_pending_decisions(
     self,
     status: dict[str, Any],

orchestrator/mcp_tools/_tool_defs.py

@@ -90,7 +90,11 @@
             "Get the current status of a pipeline task. "
             "Returns pipeline state (current_phase, status, agents, decisions), "
             "pipeline details (id, repo, issue_number, created_at, mode), "
-            "and recent_messages (from_role, type, subject, timestamp)."
+            "and recent_messages (from_role, type, subject, timestamp). "
+            "Also surfaces pending_contract_decisions: agent-registered "
+            "contract-resident HITL questions (`cq-N`) that are unresolved and "
+            "not yet bridged into the decision queue — these gate upcoming "
+            "work and are resolved via the same provide_input flow."
         ),
         "inputSchema": {
             "type": "object",

orchestrator/routes/decisions/__init__.py

@@ -90,7 +90,10 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
     _maybe_complete_task_from_resolution,
     _normalize_choice_resolution,
 )
-from ._resolve import _resolve_contract_decision  # noqa: E402,F401
+from ._resolve import (  # noqa: E402,F401
+    _outstanding_contract_hitl,
+    _resolve_contract_decision,
+)
 from ._responses import (  # noqa: E402,F401
     make_error_response,
     make_success_response,

orchestrator/routes/decisions/_resolve.py

@@ -6,6 +6,7 @@
 
 import routes.decisions as _pkg
 from decision_queue import DecisionAlreadyResolvedError, DecisionNotFoundError
+from egg_contracts.decisions import truncate_question
 from events import EventType
 from flask import Response, request
 from state_store import InvalidPipelineIdError, PipelineNotFoundError
@@ -182,6 +183,7 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
         # treats the absent resolution as approval and marks the current
         # phase complete. Only a phase_gate resolution is meant to drive the
         # phase forward, so only it triggers the revival.
+        outstanding_contract_decisions: list[dict[str, Any]] = []
         if decision.decision_type == "phase_gate":
             try:
                 from routes.pipelines import maybe_revive_orphaned_awaiting_human_driver
@@ -196,6 +198,14 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
                     exc_info=True,
                 )
 
+            # Gate-approval guard: report contract-resident HITL questions
+            # (``cq-N``) that this gate resolution leaves outstanding, so the
+            # operator is not told "proceeding" while agent-registered HITL
+            # questions sit unanswered. The post-gate bridge promotes the
+            # current phase's ``cq-N`` into the queue; questions tagged for a
+            # later phase remain outstanding and are the ones worth flagging.
+            outstanding_contract_decisions = _outstanding_contract_hitl(pipeline_id, _pipeline)
+
         return make_success_response(
             "Decision resolved",
             data={
@@ -209,6 +219,11 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
                     "scope": "queue",
                 },
                 **({"executed_action": executed_action} if executed_action else {}),
+                **(
+                    {"outstanding_contract_decisions": outstanding_contract_decisions}
+                    if outstanding_contract_decisions
+                    else {}
+                ),
             },
         )
 
@@ -235,6 +250,52 @@ def resolve_decision(pipeline_id: str, decision_id: str) -> tuple[Response, int]
         return make_error_response(str(e), status_code=409)
 
 
+def _outstanding_contract_hitl(pipeline_id: str, pipeline: Any) -> list[dict[str, Any]]:
+    """Return unresolved contract HITL (``cq-N``) decisions a gate leaves open.
+
+    Best-effort summary attached to a phase_gate resolve response (#3374):
+    contract-resident HITL questions that are *not* the current phase (so
+    the post-gate bridge will not promote them into the queue) and remain
+    unanswered. Reports ``id`` / ``question`` (truncated) / ``phase`` so the
+    operator gets explicit signal instead of a silent "proceeding". Any
+    failure yields ``[]`` — this guard must never block a resolution.
+    """
+    try:
+        import contract_store
+        from egg_contracts import load_contract
+        from routes.pipelines import _pipeline_identifier
+
+        worktree = contract_store.resolve_pipeline_worktree(pipeline_id)
+        if worktree is None:
+            return []
+        identifier = _pipeline_identifier(getattr(pipeline, "issue_number", None), pipeline_id)
+        contract = load_contract(identifier, worktree)
+    except Exception:
+        logger.debug(
+            "Outstanding-contract-decision scan failed after gate resolve",
+            pipeline_id=pipeline_id,
+            exc_info=True,
+        )
+        return []
+
+    current_phase = getattr(contract, "current_phase", None)
+    current_phase_val = getattr(current_phase, "value", current_phase)
+
+    outstanding: list[dict[str, Any]] = []
+    for d in contract.decisions or []:
+        if d.resolved or getattr(d.type, "value", d.type) != "hitl":
+            continue
+        d_phase = getattr(d.phase, "value", d.phase)
+        # The bridge promotes the current phase (and phase-less) questions
+        # into the queue on approval; only later-phase questions are left
+        # genuinely outstanding and invisible.
+        if d_phase is None or d_phase == current_phase_val:
+            continue
+        question = truncate_question(d.question or "")
+        outstanding.append({"id": d.id, "question": question, "phase": d_phase})
+    return outstanding
+
+
 def _resolve_contract_decision(
     pipeline_id: str,
     decision_id: str,

orchestrator/tests/test_impasse_routing.py

@@ -230,6 +230,41 @@ def test_external_blocker_escalates_with_hitl_options(self, tmp_path):
         assert "external_blocker" in decision.question
         assert "upstream PR not merged" in decision.question
 
+    def test_repeated_escalation_dedupes_onto_existing_decision(self, tmp_path):
+        """Re-escalating the same impasse adopts the open ``cq-N`` rather
+        than appending an identical duplicate (#3374)."""
+        pid = _seed_contract(tmp_path)
+        contract = load_contract(pid, tmp_path)
+        contract.slices[0].tasks[0].delegation_attempts = DELEGATION_LIMIT
+        save_contract(contract, tmp_path)
+
+        imp = Impasse(
+            category=ImpasseCategory.PLAN_BUG,
+            reason="acceptance criteria contradict each other",
+            task_id="task-1-1",
+        )
+
+        def _escalate_once():
+            return route_impasses(
+                repo_path=tmp_path,
+                pipeline_id=pid,
+                contract_identifier=pid,
+                impasses=[(ContractAgentRole.CODER, imp)],
+                slice_id="slice-1",
+            )
+
+        first = _escalate_once()
+        second = _escalate_once()
+
+        assert first[0].action == ImpasseAction.ESCALATE
+        assert second[0].action == ImpasseAction.ESCALATE
+        # Both routing decisions point at the same HITL id...
+        assert first[0].hitl_decision_id == second[0].hitl_decision_id
+        # ...and only one decision was ever written to the contract.
+        contract = load_contract(pid, tmp_path)
+        hitl_ids = [d.id for d in contract.decisions if d.id.startswith("cq-")]
+        assert hitl_ids == [first[0].hitl_decision_id]
+
     def test_self_delegation_escalates(self, tmp_path):
         # Even though report_impasse rejects suggested_role==role at the
         # handler boundary, defense-in-depth: if a malformed payload