From dd477281f7779d52fa6682eb9833475bf4e3c106 Mon Sep 17 00:00:00 2001
From: James Wiesebron <jameswiesebron@khanacademy.org>
Date: Thu, 2 Jul 2026 21:47:39 -0700
Subject: [PATCH 1/2] Surface explicit-none decision attestations as
 confirmable HITL decisions (#3462)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

An agent could bypass the entire register -> bridge -> resolve chain by
attesting no_decisions_rationale and documenting decisions as draft
prose — the attestation rode along as a sentence in the phase_gate
question, so the operator never confirmed it as a decision.

Gate side: when a refine/plan phase reaches its gate with an
explicit-none attestation standing in for a ledger, queue a dedicated
confirmable choice decision quoting the role and rationale. Confirm
(bare keyword or full option label; anything else conservatively
rejects) proceeds to the normal gate with "Operator confirmed the
attestation" recorded on the ledger note; reject re-runs the phase with
a directive to register each decision — including believed-resolved
ones, registered with the recommended answer as the first option.
Idempotent across converge rounds (same question -> reuse prior
confirmation or pending decision). Autonomous pipelines emit a
phase.decision_ledger_explicit_none event instead of blocking,
mirroring the #3392 gate-skip posture.

Prompt side: task-named decisions are non-optional — a "surface as
HITL" directive or operator-owned decision named by the task must be
registered even when believed resolved; belief about resolution is a
recommended disposition (recommended option citing the resolving
context), not a reason to skip. The #2481 skip-already-resolved rule is
narrowed to answers from THIS pipeline's pre-refine HITL round.

Reviewer side: reviewer_refine (§7) and reviewer_plan (§14) NACK an
explicit-none ledger on a task that names decisions to surface.
---
 docs/hitl-decisions.md                        |  38 +++
 orchestrator/routes/pipelines.py              | 317 +++++++++++++++++-
 .../tests/test_decision_ledger_gate.py        |  99 +++++-
 orchestrator/tests/test_pipeline_prompts.py   |  62 ++++
 4 files changed, 493 insertions(+), 23 deletions(-)

diff --git a/docs/hitl-decisions.md b/docs/hitl-decisions.md
index cd6351f4cc..753671d654 100644
--- a/docs/hitl-decisions.md
+++ b/docs/hitl-decisions.md
@@ -435,6 +435,44 @@ normal gate. On autonomous pipelines (`hitl_gates: false`) the missing ledger
 is surfaced as a loud `phase.decision_ledger_missing` event but never blocks,
 mirroring the autonomous gate-skip posture.
 
+### Explicit-none attestations are confirmed, not trusted (#3462)
+
+An explicit-none attestation is itself a judgment call about what *is* a
+judgment call — exactly the class of decision the HITL contract assigns to
+the operator, and the one escape hatch through which an agent under
+convergence pressure can bypass the whole register → bridge → resolve chain.
+So the gate does not fold it into the `phase_gate` question as prose: when a
+refine/plan phase reaches its gate with an explicit-none attestation standing
+in for a ledger, the orchestrator first surfaces a dedicated **confirmable
+`choice` decision** quoting the role and rationale ("the &lt;role&gt; attests
+this phase deliberately raises no operator decisions — confirm?").
+
+- **Confirm** (the bare keyword or the full option label — anything else is
+  conservatively treated as a rejection, mirroring the phase_gate's
+  "bare approve advances" posture) proceeds to the normal phase gate, and the
+  gate's ledger note records "Operator confirmed the attestation".
+- **Re-run** (or any free-text reply, which rides along as an operator note)
+  kicks the phase back via the converge loop's standard re-run, with a
+  directive telling producers to register each decision — including ones they
+  believe prior context already resolves, registered with the recommended
+  answer as the first option rather than attested away.
+
+The confirmation is idempotent across converge rounds: a re-entered gate with
+the *same* attestation reuses the operator's prior confirmation (or a pending
+confirmation decision) instead of re-asking; a changed rationale is a new
+claim and is asked again. On autonomous pipelines the unconfirmed attestation
+is surfaced as a `phase.decision_ledger_explicit_none` event but never
+blocks.
+
+Prompt-side, the same issue closes the loophole at the source: decisions the
+task description names as operator-owned (or covered by any
+"surface as HITL" directive) **must be registered** even when the producer
+believes they are already resolved, non-blocking, or deferred — belief about
+resolution is a *recommended disposition* (recommended option citing the
+resolving context), never a reason to skip registration — and
+`reviewer_refine` (§7) / `reviewer_plan` (§14) NACK an explicit-none ledger
+on a task that names decisions to surface.
+
 ### Judgment: reviewer obligation against un-surfaced decisions
 
 A draft that quietly **commits** to an operator-grade choice ("we will drop
diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py
index 93fc058b16..88b1453044 100644
--- a/orchestrator/routes/pipelines.py
+++ b/orchestrator/routes/pipelines.py
@@ -5856,7 +5856,16 @@ def _get_refine_review_criteria() -> str:
         "- If the ledger is deliberately empty (the producer attested "
         "`no_decisions_rationale`), verify the rationale holds: requirements "
         "genuinely unambiguous, no assumptions made silently. NACK if you "
-        "find a hidden operator-grade choice.\n\n"
+        "find a hidden operator-grade choice.\n"
+        "- **Task-named decisions — NACK an explicit-none ledger (#3462).** "
+        "If the task description names decisions as the operator's to make "
+        "(or directs that decisions be surfaced as HITL questions), each "
+        "must have a registered `cq-N` — even when the draft argues prior "
+        'context already resolves it. "Already resolved" is a recommended '
+        "disposition to register (recommended option citing the resolving "
+        "context), not a reason to skip; a `no_decisions_rationale` "
+        "attestation on such a task is a **NACK** regardless of how "
+        "defensible the rationale reads.\n\n"
         + _human_companion_review_criteria(
             companion="`*-analysis-human.md`",
             parent="the refine analysis",
@@ -6291,7 +6300,16 @@ def _get_plan_review_criteria() -> str:
         "boundaries, external commitments, user-visible behavior).\n"
         "- A deliberately empty ledger arrives as a producer's "
         "`no_decisions_rationale` attestation — verify it holds; NACK if "
-        "the plan hides an operator-grade choice.\n\n"
+        "the plan hides an operator-grade choice.\n"
+        "- **Task-named decisions — NACK an explicit-none ledger (#3462).** "
+        "If the task description or refine analysis names decisions as the "
+        "operator's to make (or directs that decisions be surfaced as HITL "
+        "questions), each must have a registered `cq-N` — even when the "
+        'plan argues prior context already resolves it. "Already '
+        'resolved" is a recommended disposition to register (recommended '
+        "option citing the resolving context), not a reason to skip; a "
+        "`no_decisions_rationale` attestation on such a task is a **NACK** "
+        "regardless of how defensible the rationale reads.\n\n"
         + _human_companion_review_criteria(
             companion="`*-plan-human.md`",
             parent="the implementation plan",
@@ -14141,7 +14159,26 @@ def _build_phase_prompt(
                 "anything, list those items in a `### Resolved in Pre-Refine` "
                 "subsection at the top of `## Open Questions` (one bullet per resolved "
                 "item, citing the answer). Only register questions that go beyond what "
-                "`## Additional Context` covers.\n",
+                "`## Additional Context` covers. This skip rule is NARROW: it covers "
+                "only answers THIS pipeline's operator recorded in "
+                "`## Additional Context`. It never covers decisions the task "
+                "description names as operator-owned, and never answers inherited "
+                "from a prior or cancelled run's seeded context — register those "
+                "(see the next rule).\n",
+                "**Task-named decisions are non-optional (#3462).** If the task "
+                "description or contract names specific decisions as the operator's "
+                "to make — or contains any directive to surface decisions as HITL "
+                "questions — you MUST register each one via "
+                "`egg-contract add-decision`, even when you believe prior context "
+                "already resolves it, or that it is non-blocking or deferred. "
+                "Belief about resolution is a *recommended disposition*, not a "
+                "reason to skip registration: make your recommended answer the "
+                "first option (suffix its label with `(recommended)`) and cite the "
+                "resolving context in that option's description, so the operator "
+                "can confirm in one click while retaining the authority to choose "
+                "differently. Documenting a decision in draft prose is a "
+                "supplement to registration, never a substitute — unregistered "
+                "decisions never reach the operator's decision surface.\n",
                 "Surface uncertainties, ambiguities, and assumptions **that genuinely "
                 "need a human to answer**. Filter ruthlessly: a good open question is "
                 "one the operator must answer because the answer changes what we're "
@@ -14210,6 +14247,13 @@ def _build_phase_prompt(
                 "`<!-- DECISION: ... -->` instead of the contract CLI",
                 "- Skip registration because you think the questions are minor — "
                 "register every question",
+                "- Skip registration because you believe a decision is already "
+                "resolved, non-blocking, or deferred — register it with your "
+                "recommended disposition instead (#3462)",
+                "- Attest `no_decisions_rationale` when the task names decisions "
+                "to surface — the attestation is presented to the operator as its "
+                "own confirmable decision, and a rejected 'none' sends the phase "
+                "back for a re-run (#3462)",
                 "- Transcribe this `## How to Populate Open Questions` section "
                 "into your analysis document — it is meta-guidance, not template "
                 "content\n",
@@ -14226,7 +14270,13 @@ def _build_phase_prompt(
                 "request (no `cq-N` decisions), attest the rationale form and "
                 "name the feedback request in it. This is what lets the "
                 "operator trust that an empty gate means *deliberately no "
-                "decisions*, not *forgot to register*.\n",
+                "decisions*, not *forgot to register*. The explicit-none form "
+                "is not a shortcut (#3462): the orchestrator surfaces it to "
+                "the operator as its own confirmable decision before the "
+                "phase gate, and it is only valid when the phase genuinely "
+                "raises no meaningful decision — never when the task names "
+                "decisions to surface, and never as a substitute for "
+                "registering a decision you believe is already resolved.\n",
                 "",
             ]
         )
@@ -14972,7 +15022,13 @@ def _build_brc_preamble(
                 "`cq-N` (copying the `--format markdown` output into the "
                 "draft satisfies this). A decision your draft commits to "
                 "without a registered `cq-N` is a reviewer NACK — register "
-                "it or remove the unilateral commitment."
+                "it or remove the unilateral commitment. The rationale form "
+                "is not a shortcut (#3462): the operator is asked to confirm "
+                "it as its own decision before the phase gate, and a rejected "
+                "'none' re-runs the phase. If the task names decisions to "
+                "surface — or you believe a decision is already resolved by "
+                "prior context — register it with your recommended answer as "
+                "the first option instead of attesting none."
             )
         if phase == "implement":
             propose_line += (
@@ -24750,6 +24806,64 @@ def _sync_pipeline_decisions_to_contract(
 _LEDGER_BACKSTOP_RERUN_OPTION = "Re-run phase to register decisions"
 _LEDGER_BACKSTOP_PROCEED_OPTION = "Proceed without a decision ledger"
 
+# Explicit-none attestation confirmation option (#3462). Paired with
+# ``_LEDGER_BACKSTOP_RERUN_OPTION`` on the confirmation decision; only a
+# resolution that IS a confirmation (the bare keyword or the full label)
+# proceeds — any other text is treated as a re-run directive, mirroring
+# the phase_gate's "bare approve advances, notes request changes" posture.
+_LEDGER_ATTESTATION_CONFIRM_OPTION = "Confirm — no open decisions this phase"
+
+
+def _ledger_attestation_question(role: str, rationale: str, phase_value: str) -> str:
+    """Compose the explicit-none confirmation question (#3462).
+
+    A producer's claim that a phase raises no operator decisions is itself
+    a judgment call about what *is* a judgment call — exactly the class of
+    decision the HITL contract assigns to the operator. It therefore
+    surfaces as its own confirmable decision, not a sentence embedded in
+    the phase_gate question.
+    """
+    return (
+        f"The {role} attests the {phase_value} phase deliberately raises "
+        f"no operator decisions (#3462):\n\n"
+        f"> {rationale}\n\n"
+        f"Confirm to proceed to the phase gate, or choose "
+        f"“{_LEDGER_BACKSTOP_RERUN_OPTION}” to send the phase back so its "
+        f"agents register the decisions as first-class contract entries "
+        f"(cq-N). Any free-text reply is treated as a re-run directive and "
+        f"forwarded to the agents."
+    )
+
+
+def _unwrap_choice_resolution(resolution: str) -> str:
+    """Unwrap the ``{"action":"select","selected":<label>}`` envelope.
+
+    The SDLC HITL CLI resolves a ``choice`` decision with that structured
+    envelope (mirrors ``routes.decisions._normalize_choice_resolution``);
+    a bare string / non-JSON resolution passes through unchanged.
+    """
+    try:
+        payload = json.loads(resolution)
+        if isinstance(payload, dict) and payload.get("action") == "select":
+            selected = payload.get("selected")
+            if isinstance(selected, str):
+                return selected
+    except ValueError, TypeError:
+        pass
+    return resolution
+
+
+def _ledger_attestation_confirmed(resolution: str) -> bool:
+    """Return True when ``resolution`` confirms the explicit-none attestation.
+
+    Conservative on purpose (#3462): only the bare keyword ``confirm`` or
+    the full confirm-option label counts. Anything else — the re-run
+    option, or free text naming decisions the operator expected — kicks
+    the phase back, with the text riding along as the directive.
+    """
+    normalized = _unwrap_choice_resolution(resolution).strip().lower()
+    return normalized in ("confirm", _LEDGER_ATTESTATION_CONFIRM_OPTION.lower())
+
 
 def _find_explicit_none_attestation(
     pipeline_id: str,
@@ -24801,10 +24915,10 @@ def _collect_decision_ledger_status(
     pipeline_id: str,
     pipeline_identifier: int | str,
     phase: PipelinePhase,
-) -> tuple[str, bool]:
+) -> tuple[str, bool, tuple[str, str] | None]:
     """Summarize the phase's decision ledger for the gate surface (#3390).
 
-    Returns ``(note, missing)``:
+    Returns ``(note, missing, explicit_none)``:
 
     - ``note`` — an operator-visible one-liner appended to the phase_gate
       question so "N registered" vs "explicitly none" vs "MISSING" is
@@ -24815,6 +24929,11 @@ def _collect_decision_ledger_status(
       bypassed consensus (force-advance, resume) or the producer's claim
       was lost — the caller surfaces a dedicated backstop HITL rather
       than silently advancing.
+    - ``explicit_none`` — the ``(role, rationale)`` of a producer's
+      explicit-none attestation when that is what stands in for a ledger
+      (zero registered decisions), else ``None``. The caller surfaces it
+      as its own confirmable decision (#3462) rather than trusting the
+      self-attestation. Mutually exclusive with ``missing``.
     """
     phase_value = phase.value
     registered_ids: list[str] = []
@@ -24849,6 +24968,7 @@ def _collect_decision_ledger_status(
             f"Decision ledger: {len(registered_ids)} decision(s) registered this "
             f"phase ({', '.join(registered_ids)}), {resolved} resolved.",
             False,
+            None,
         )
 
     explicit_none = _find_explicit_none_attestation(pipeline_id, phase_value)
@@ -24857,6 +24977,7 @@ def _collect_decision_ledger_status(
         return (
             f"Decision ledger: explicitly none — {role} attested: {rationale}",
             False,
+            explicit_none,
         )
 
     return (
@@ -24865,6 +24986,7 @@ def _collect_decision_ledger_status(
         "“0 decisions” here cannot be distinguished from “failed "
         "to register”.",
         True,
+        None,
     )
 
 
@@ -28022,11 +28144,13 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                 # the gate-skip posture — surface a missing ledger loudly
                 # (event + warning) but never block.
                 try:
-                    _ledger_note, _ledger_missing = _collect_decision_ledger_status(
-                        worktree_repo_path,
-                        pipeline_id,
-                        _pipeline_identifier(pipeline.issue_number, pipeline_id),
-                        current_phase,
+                    _ledger_note, _ledger_missing, _ledger_explicit_none = (
+                        _collect_decision_ledger_status(
+                            worktree_repo_path,
+                            pipeline_id,
+                            _pipeline_identifier(pipeline.issue_number, pipeline_id),
+                            current_phase,
+                        )
                     )
                     if _ledger_missing:
                         logger.warning(
@@ -28042,6 +28166,19 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                                 f"with no decision ledger — {_ledger_note}"
                             ),
                         )
+                    elif _ledger_explicit_none is not None:
+                        # No human is present to confirm the attestation
+                        # (#3462) — mirror the gate-skip posture: surface
+                        # loudly, never block.
+                        report_pipeline_status(
+                            pipeline,
+                            event_type="phase.decision_ledger_explicit_none",
+                            message=(
+                                f"{current_phase.value} phase advanced autonomously "
+                                f"on an unconfirmed no-decisions attestation — "
+                                f"{_ledger_note}"
+                            ),
+                        )
                 except Exception as ledger_err:  # noqa: BLE001
                     logger.warning(
                         "Decision-ledger check raised on autonomous path (continuing)",
@@ -28061,12 +28198,15 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                 # explicit operator override to proceed.
                 _ledger_note = ""
                 _ledger_missing = False
+                _ledger_explicit_none: tuple[str, str] | None = None
                 try:
-                    _ledger_note, _ledger_missing = _collect_decision_ledger_status(
-                        worktree_repo_path,
-                        pipeline_id,
-                        _pipeline_identifier(pipeline.issue_number, pipeline_id),
-                        current_phase,
+                    _ledger_note, _ledger_missing, _ledger_explicit_none = (
+                        _collect_decision_ledger_status(
+                            worktree_repo_path,
+                            pipeline_id,
+                            _pipeline_identifier(pipeline.issue_number, pipeline_id),
+                            current_phase,
+                        )
                     )
                 except Exception as ledger_err:  # noqa: BLE001
                     # Never let a helper bug strand the pipeline — the
@@ -28179,6 +28319,149 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                         phase=current_phase.value,
                         resolution=_backstop_resolution[:200],
                     )
+                elif _ledger_explicit_none is not None:
+                    # --- Explicit-none attestation confirmation (#3462) ---
+                    # The producer's claim that this phase raises no operator
+                    # decisions bypasses the entire register → bridge →
+                    # resolve chain, and the claim is itself a judgment call
+                    # the HITL contract assigns to the operator. Surface it
+                    # as its own confirmable decision instead of a sentence
+                    # embedded in the phase_gate question: confirming records
+                    # the operator's endorsement on the ledger note;
+                    # rejecting re-runs the phase so producers register the
+                    # decisions as first-class cq-N entries.
+                    _attest_role, _attest_rationale = _ledger_explicit_none
+                    _attest_question = _ledger_attestation_question(
+                        _attest_role, _attest_rationale, current_phase.value
+                    )
+                    # A converge-loop round (or a resume) re-enters this gate
+                    # with the same attestation — do not re-ask a question
+                    # the operator already answered, and reuse a pending one
+                    # instead of queueing a duplicate (mirrors the
+                    # phase_gate's #1152 guard).
+                    _prior_confirm = next(
+                        (
+                            d
+                            for d in reversed(pipeline.decisions)
+                            if d.decision_type == "choice"
+                            and d.phase == current_phase
+                            and d.question == _attest_question
+                            and d.status == DecisionStatus.RESOLVED
+                            and _ledger_attestation_confirmed(str(d.resolution or ""))
+                        ),
+                        None,
+                    )
+                    if _prior_confirm is not None:
+                        _ledger_note += " Operator confirmed the attestation."
+                    else:
+                        dq = get_decision_queue(pipeline_id, repo_path)
+                        _pending_attest = next(
+                            (
+                                d
+                                for d in reversed(pipeline.decisions)
+                                if d.decision_type == "choice"
+                                and d.phase == current_phase
+                                and d.question == _attest_question
+                                and d.status == DecisionStatus.PENDING
+                            ),
+                            None,
+                        )
+                        if _pending_attest is not None:
+                            _attest_decision = _pending_attest
+                        else:
+                            _attest_decision = dq.queue_decision(
+                                question=_attest_question,
+                                context=_ledger_note,
+                                options=[
+                                    _LEDGER_ATTESTATION_CONFIRM_OPTION,
+                                    _LEDGER_BACKSTOP_RERUN_OPTION,
+                                ],
+                                decision_type="choice",
+                                phase=current_phase,
+                            )
+                        with get_pipeline_state_lock(pipeline_id):
+                            pipeline = store.load_pipeline(pipeline_id)
+                            pipeline.status = PipelineStatus.AWAITING_HUMAN
+                            phase_execution = pipeline.get_phase_execution(current_phase)
+                            phase_execution.status = PipelineStatus.AWAITING_HUMAN
+                            store.save_pipeline(pipeline)
+                        report_pipeline_status(
+                            pipeline,
+                            event_type="decision.created",
+                            message=(
+                                f"{current_phase.value} phase attests no operator "
+                                f"decisions — awaiting operator confirmation (#3462)"
+                            ),
+                        )
+                        _emit_pipeline_event(pipeline, "decision.created")
+
+                        _attest_resolved = dq.wait_for_decision(_attest_decision.id)
+                        _attest_resolution = _unwrap_choice_resolution(
+                            str(getattr(_attest_resolved, "resolution", None) or "")
+                        ).strip()
+                        # Fail open to the phase gate on a non-RESOLVED
+                        # terminal state (cancel): the gate itself still
+                        # blocks for approval, mirroring the missing-ledger
+                        # backstop's posture.
+                        _confirmed = (
+                            _attest_resolved.status != DecisionStatus.RESOLVED
+                            or _ledger_attestation_confirmed(_attest_resolution)
+                        )
+                        if not _confirmed:
+                            _rerun_directive = (
+                                f"The operator declined to confirm the "
+                                f"{current_phase.value} phase's no-decisions "
+                                f"attestation (#3462). The phase claimed: "
+                                f"“{_attest_rationale}”. Register each "
+                                f"operator-grade decision via `egg-contract "
+                                f"add-decision` — including decisions you "
+                                f"believe prior context already resolves: "
+                                f"register those with your recommended answer "
+                                f"as the first option and cite the resolving "
+                                f"context in its description. Belief about "
+                                f"resolution is a recommended disposition, not "
+                                f"a reason to skip registration."
+                            )
+                            if _attest_resolution.lower() != (
+                                _LEDGER_BACKSTOP_RERUN_OPTION.lower()
+                            ):
+                                _rerun_directive += f"\n\nOperator note: {_attest_resolution}"
+                            logger.info(
+                                "Explicit-none attestation rejected: re-running phase (#3462)",
+                                pipeline_id=pipeline_id,
+                                phase=current_phase.value,
+                            )
+                            with get_pipeline_state_lock(pipeline_id):
+                                pipeline = store.load_pipeline(pipeline_id)
+                                pipeline.status = PipelineStatus.RUNNING
+                                phase_execution = pipeline.get_phase_execution(current_phase)
+                                phase_execution.status = PipelineStatus.RUNNING
+                                phase_execution.completed_at = None
+                                phase_execution.hitl_review_cycles += 1
+                                _alert_threshold = pipeline.config.max_hitl_review_cycles
+                                if phase_execution.hitl_review_cycles >= _alert_threshold:
+                                    _broadcast_hitl_nonconvergence_alert(
+                                        pipeline_id,
+                                        pipeline,
+                                        current_phase,
+                                        phase_execution.hitl_review_cycles,
+                                        _alert_threshold,
+                                    )
+                                _perform_hitl_phase_rerun(
+                                    store=store,
+                                    spawner=spawner,
+                                    pipeline=pipeline,
+                                    phase_execution=phase_execution,
+                                    pipeline_id=pipeline_id,
+                                    current_phase=current_phase,
+                                    feedback_text=_rerun_directive,
+                                    event_message=(
+                                        f"Re-running {current_phase.value}: "
+                                        f"no-decisions attestation rejected (#3462)"
+                                    ),
+                                )
+                            continue  # Re-enter outer loop → re-run phase
+                        _ledger_note += " Operator confirmed the attestation."
 
                 # Check for an existing pending phase_gate decision for this
                 # phase.  A prior agent-exit event may
diff --git a/orchestrator/tests/test_decision_ledger_gate.py b/orchestrator/tests/test_decision_ledger_gate.py
index 865813a936..0bc0fe565d 100644
--- a/orchestrator/tests/test_decision_ledger_gate.py
+++ b/orchestrator/tests/test_decision_ledger_gate.py
@@ -1,10 +1,13 @@
-"""Tests for the gate-side decision-ledger backstop helpers (#3390).
+"""Tests for the gate-side decision-ledger backstop helpers (#3390, #3462).
 
 ``_collect_decision_ledger_status`` summarizes the phase's decision ledger
 for the phase_gate surface: "N registered" (from the contract), "explicitly
 none" (from a producer's ``no_decisions_rationale`` propose attestation), or
 MISSING (neither — the backstop-HITL trigger). ``_find_explicit_none_
 attestation`` is the message-store scan behind the explicit-none case.
+``_ledger_attestation_question`` / ``_ledger_attestation_confirmed`` back the
+explicit-none confirmation decision the gate surfaces to the operator
+(#3462).
 """
 
 from __future__ import annotations
@@ -98,10 +101,11 @@ def test_registered_decisions_counted(self, tmp_path: Path):
                 _decision("cq-3", phase="plan"),
             ],
         )
-        note, missing = _collect_decision_ledger_status(
+        note, missing, explicit_none = _collect_decision_ledger_status(
             tmp_path, "pipeline-id", "issue-42", PipelinePhase.REFINE
         )
         assert missing is False
+        assert explicit_none is None
         assert "2 decision(s) registered" in note
         assert "cq-1" in note and "cq-2" in note and "cq-3" not in note
         assert "1 resolved" in note
@@ -116,10 +120,11 @@ def test_zero_registered_with_explicit_none_attestation(self, tmp_path: Path):
             )
         ]
         with _patch_message_store(messages):
-            note, missing = _collect_decision_ledger_status(
+            note, missing, explicit_none = _collect_decision_ledger_status(
                 tmp_path, "pipeline-id", "issue-42", PipelinePhase.REFINE
             )
         assert missing is False
+        assert explicit_none == ("refiner", "mechanical change, no choices")
         assert "explicitly none" in note
         assert "refiner" in note
         assert "mechanical change" in note
@@ -129,10 +134,11 @@ def test_zero_registered_no_attestation_is_missing(self, tmp_path: Path):
 
         _make_contract_file(tmp_path, "issue-42", decisions=[])
         with _patch_message_store([]):
-            note, missing = _collect_decision_ledger_status(
+            note, missing, explicit_none = _collect_decision_ledger_status(
                 tmp_path, "pipeline-id", "issue-42", PipelinePhase.REFINE
             )
         assert missing is True
+        assert explicit_none is None
         assert "MISSING" in note
 
     def test_contract_unloadable_falls_back_to_attestation(self, tmp_path: Path):
@@ -141,10 +147,11 @@ def test_contract_unloadable_falls_back_to_attestation(self, tmp_path: Path):
         # No contract file written at all.
         messages = [_propose_message(attestation={"no_decisions_rationale": "none needed"})]
         with _patch_message_store(messages):
-            note, missing = _collect_decision_ledger_status(
+            note, missing, explicit_none = _collect_decision_ledger_status(
                 tmp_path, "pipeline-id", "issue-42", PipelinePhase.REFINE
             )
         assert missing is False
+        assert explicit_none == ("refiner", "none needed")
         assert "explicitly none" in note
 
     def test_message_store_outage_fails_closed(self, tmp_path: Path):
@@ -152,10 +159,11 @@ def test_message_store_outage_fails_closed(self, tmp_path: Path):
 
         _make_contract_file(tmp_path, "issue-42", decisions=[])
         with patch("message_store.get_message_store", side_effect=RuntimeError("redis down")):
-            note, missing = _collect_decision_ledger_status(
+            note, missing, explicit_none = _collect_decision_ledger_status(
                 tmp_path, "pipeline-id", "issue-42", PipelinePhase.REFINE
             )
         assert missing is True
+        assert explicit_none is None
         assert "MISSING" in note
 
 
@@ -203,3 +211,82 @@ def test_registered_form_attestation_is_not_explicit_none(self):
         messages = [_propose_message(attestation={"decisions_registered": ["cq-1"]})]
         with _patch_message_store(messages):
             assert _find_explicit_none_attestation("pipeline-id", "refine") is None
+
+
+class TestLedgerAttestationConfirmation:
+    """The explicit-none confirmation decision surface (#3462).
+
+    An explicit-none attestation is surfaced to the operator as its own
+    confirmable decision. ``_ledger_attestation_question`` composes it;
+    ``_ledger_attestation_confirmed`` decides whether a resolution
+    confirms (proceed to the gate) or rejects (re-run the phase).
+    """
+
+    def test_question_quotes_role_phase_and_rationale(self):
+        from routes.pipelines import _ledger_attestation_question
+
+        question = _ledger_attestation_question(
+            "refiner", "answers present in seeded context", "refine"
+        )
+        assert "refiner" in question
+        assert "refine phase" in question
+        assert "answers present in seeded context" in question
+        assert "#3462" in question
+
+    def test_question_is_stable_for_identical_claims(self):
+        # Converge-round dedupe keys on question equality: the same
+        # (role, rationale, phase) must compose the identical question.
+        from routes.pipelines import _ledger_attestation_question
+
+        a = _ledger_attestation_question("refiner", "no choices", "refine")
+        b = _ledger_attestation_question("refiner", "no choices", "refine")
+        assert a == b
+        assert a != _ledger_attestation_question("refiner", "other claim", "refine")
+
+    def test_bare_confirm_keyword_confirms(self):
+        from routes.pipelines import _ledger_attestation_confirmed
+
+        assert _ledger_attestation_confirmed("confirm") is True
+        assert _ledger_attestation_confirmed("  Confirm  ") is True
+
+    def test_full_option_label_confirms(self):
+        from routes.pipelines import (
+            _LEDGER_ATTESTATION_CONFIRM_OPTION,
+            _ledger_attestation_confirmed,
+        )
+
+        assert _ledger_attestation_confirmed(_LEDGER_ATTESTATION_CONFIRM_OPTION) is True
+        assert _ledger_attestation_confirmed(_LEDGER_ATTESTATION_CONFIRM_OPTION.upper()) is True
+
+    def test_choice_envelope_unwrapped(self):
+        # The SDLC HITL CLI resolves a choice as
+        # {"action": "select", "selected": "<label>"} (#2978) — the
+        # confirm match must unwrap it.
+        from routes.pipelines import (
+            _LEDGER_ATTESTATION_CONFIRM_OPTION,
+            _ledger_attestation_confirmed,
+        )
+
+        envelope = json.dumps({"action": "select", "selected": _LEDGER_ATTESTATION_CONFIRM_OPTION})
+        assert _ledger_attestation_confirmed(envelope) is True
+
+    def test_rerun_option_and_free_text_reject(self):
+        from routes.pipelines import (
+            _LEDGER_BACKSTOP_RERUN_OPTION,
+            _ledger_attestation_confirmed,
+        )
+
+        assert _ledger_attestation_confirmed(_LEDGER_BACKSTOP_RERUN_OPTION) is False
+        # Free text is a re-run directive, even when it contains the
+        # word "confirm" in a negating sense.
+        assert _ledger_attestation_confirmed("do not confirm — register cq for deploy") is False
+        assert _ledger_attestation_confirmed("") is False
+
+    def test_envelope_with_rerun_selection_rejects(self):
+        from routes.pipelines import (
+            _LEDGER_BACKSTOP_RERUN_OPTION,
+            _ledger_attestation_confirmed,
+        )
+
+        envelope = json.dumps({"action": "select", "selected": _LEDGER_BACKSTOP_RERUN_OPTION})
+        assert _ledger_attestation_confirmed(envelope) is False
diff --git a/orchestrator/tests/test_pipeline_prompts.py b/orchestrator/tests/test_pipeline_prompts.py
index 4059c9b874..4d57b0068d 100644
--- a/orchestrator/tests/test_pipeline_prompts.py
+++ b/orchestrator/tests/test_pipeline_prompts.py
@@ -4157,6 +4157,68 @@ def test_plan_reviewer_carries_unsurfaced_decision_obligation(self):
         assert "do not over-NACK" in criteria
 
 
+class TestTaskNamedDecisionRegistrationSurface:
+    """Prompt/reviewer surface of the task-named registration rule (#3462).
+
+    An agent must register task-named decisions even when it believes prior
+    context resolves them — belief about resolution is a recommended
+    disposition, not a reason to skip — and self-attesting an empty ledger
+    is confirmable by the operator, never a silent bypass.
+    """
+
+    @staticmethod
+    def _refine_prompt() -> str:
+        return _build_phase_prompt(
+            phase="refine",
+            pipeline_id="test-pipe",
+            pipeline_mode="issue",
+            prompt="Analyze this issue.",
+            issue_number=100,
+        )
+
+    def test_refine_prompt_requires_registering_task_named_decisions(self):
+        prompt = self._refine_prompt()
+        assert "Task-named decisions are non-optional" in prompt
+        assert "recommended disposition" in prompt
+        assert "(recommended)" in prompt
+        # Prose is a supplement, never a substitute.
+        assert "never a substitute" in prompt
+
+    def test_refine_prompt_narrows_the_skip_already_resolved_rule(self):
+        prompt = self._refine_prompt()
+        # The #2481 skip rule must not cover task-named decisions or
+        # answers inherited from a prior/cancelled run's seeded context —
+        # that over-read is exactly what #3462's motivating failure did.
+        assert "This skip rule is NARROW" in prompt
+        assert "prior or cancelled run" in prompt
+
+    def test_refine_prompt_do_not_list_forbids_attesting_away(self):
+        prompt = self._refine_prompt()
+        assert "Skip registration because you believe a decision is already" in prompt
+        assert "Attest `no_decisions_rationale` when the task names decisions" in prompt
+
+    def test_producer_preamble_warns_rationale_is_confirmable(self):
+        for role, phase in (
+            ("refiner", "refine"),
+            ("task_planner", "plan"),
+            ("architect", "plan"),
+            ("risk_analyst", "plan"),
+        ):
+            preamble = _build_brc_preamble(role, phase)
+            assert "not a shortcut (#3462)" in preamble, (role, phase)
+            assert "confirm" in preamble, (role, phase)
+
+    def test_refine_reviewer_nacks_explicit_none_on_task_named_decisions(self):
+        criteria = _get_refine_review_criteria()
+        assert "Task-named decisions — NACK an explicit-none ledger (#3462)" in criteria
+        assert "recommended disposition" in criteria
+
+    def test_plan_reviewer_nacks_explicit_none_on_task_named_decisions(self):
+        criteria = _get_plan_review_criteria()
+        assert "Task-named decisions — NACK an explicit-none ledger (#3462)" in criteria
+        assert "recommended disposition" in criteria
+
+
 class TestPlannerPromptTreatsRefinerSlicingAsAdvisory:
     """Planner prompts must frame any refiner-side slice sketch as advisory.
 

From 51da6e10274374a5034c248a023c8dfbdbc74d6b Mon Sep 17 00:00:00 2001
From: "egg-reviewer[bot]"
 <261018737+egg-reviewer[bot]@users.noreply.github.com>
Date: Fri, 3 Jul 2026 05:04:48 +0000
Subject: [PATCH 2/2] Address review: extract + test explicit-none attestation
 gate (#3462)

Extract the health-monitor's explicit-none confirmation orchestration into
a module-level _handle_explicit_none_attestation_gate helper so the
confirm/reject/reuse branches are unit-testable, and cover them with a new
TestHandleExplicitNoneAttestationGate class (confirm falls through, reject
re-runs with the directive, free-text rides along as an operator note,
prior confirmation reused without re-asking, pending decision reused without
re-emitting decision.created, cancel fails open).

Also fix the audit-note inaccuracy on cancel: a non-RESOLVED terminal state
now records 'confirmation was cancelled; deferring to the phase gate' rather
than falsely claiming the operator confirmed, and guard the decision.created
emit so reusing a pending decision across polls no longer re-announces it.
---
 orchestrator/routes/pipelines.py              | 345 +++++++++++-------
 .../tests/test_decision_ledger_gate.py        | 257 +++++++++++++
 2 files changed, 464 insertions(+), 138 deletions(-)

diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py
index 88b1453044..a18213e58b 100644
--- a/orchestrator/routes/pipelines.py
+++ b/orchestrator/routes/pipelines.py
@@ -24865,6 +24865,195 @@ def _ledger_attestation_confirmed(resolution: str) -> bool:
     return normalized in ("confirm", _LEDGER_ATTESTATION_CONFIRM_OPTION.lower())
 
 
+def _ledger_attestation_rerun_directive(phase_value: str, rationale: str, resolution: str) -> str:
+    """Compose the re-run directive for a rejected explicit-none attestation (#3462).
+
+    The operator declined to confirm that the phase raises no operator
+    decisions, so the phase re-runs with an instruction to register each
+    decision — including ones the producer believes prior context already
+    resolves. Any free-text resolution (i.e. not the bare re-run option) is
+    an operator note and rides along verbatim so the agents see the specific
+    concern.
+    """
+    directive = (
+        f"The operator declined to confirm the {phase_value} phase's "
+        f"no-decisions attestation (#3462). The phase claimed: "
+        f"“{rationale}”. Register each operator-grade decision via "
+        f"`egg-contract add-decision` — including decisions you believe "
+        f"prior context already resolves: register those with your "
+        f"recommended answer as the first option and cite the resolving "
+        f"context in its description. Belief about resolution is a "
+        f"recommended disposition, not a reason to skip registration."
+    )
+    if resolution.strip().lower() != _LEDGER_BACKSTOP_RERUN_OPTION.lower():
+        directive += f"\n\nOperator note: {resolution.strip()}"
+    return directive
+
+
+def _handle_explicit_none_attestation_gate(
+    *,
+    pipeline,
+    pipeline_id: str,
+    repo_path,
+    current_phase: PipelinePhase,
+    ledger_note: str,
+    explicit_none: tuple[str, str],
+    store,
+    spawner,
+):
+    """Surface an explicit-none attestation as a confirmable HITL decision (#3462).
+
+    A producer's claim that a refine/plan phase raises no operator decisions
+    bypasses the entire register → bridge → resolve chain, and the claim is
+    itself a judgment call the HITL contract assigns to the operator. Rather
+    than folding it into the phase_gate question as prose, surface it as its
+    own confirmable ``choice`` decision: confirming records the operator's
+    endorsement on the ledger note; rejecting re-runs the phase so producers
+    register the decisions as first-class ``cq-N`` entries.
+
+    Returns ``(rerun_requested, ledger_note, pipeline)``:
+
+    - ``rerun_requested`` — True when the operator rejected the attestation
+      and the phase has already been re-run here; the caller must ``continue``
+      its poll loop. False when the attestation was confirmed (or fail-open on
+      a cancelled/non-RESOLVED terminal state); the caller proceeds to the
+      phase gate.
+    - ``ledger_note`` — the note to thread into the phase_gate question,
+      annotated with the confirmation outcome.
+    - ``pipeline`` — the (possibly reloaded) pipeline the caller must rebind,
+      since queuing the confirmation decision reloads and mutates state.
+    """
+    attest_role, attest_rationale = explicit_none
+    attest_question = _ledger_attestation_question(
+        attest_role, attest_rationale, current_phase.value
+    )
+    # A converge-loop round (or a resume) re-enters this gate with the same
+    # attestation — do not re-ask a question the operator already answered,
+    # and reuse a pending one instead of queueing a duplicate (mirrors the
+    # phase_gate's #1152 guard).
+    prior_confirm = next(
+        (
+            d
+            for d in reversed(pipeline.decisions)
+            if d.decision_type == "choice"
+            and d.phase == current_phase
+            and d.question == attest_question
+            and d.status == DecisionStatus.RESOLVED
+            and _ledger_attestation_confirmed(str(d.resolution or ""))
+        ),
+        None,
+    )
+    if prior_confirm is not None:
+        return False, ledger_note + " Operator confirmed the attestation.", pipeline
+
+    dq = get_decision_queue(pipeline_id, repo_path)
+    pending_attest = next(
+        (
+            d
+            for d in reversed(pipeline.decisions)
+            if d.decision_type == "choice"
+            and d.phase == current_phase
+            and d.question == attest_question
+            and d.status == DecisionStatus.PENDING
+        ),
+        None,
+    )
+    if pending_attest is not None:
+        attest_decision = pending_attest
+        newly_created = False
+    else:
+        attest_decision = dq.queue_decision(
+            question=attest_question,
+            context=ledger_note,
+            options=[
+                _LEDGER_ATTESTATION_CONFIRM_OPTION,
+                _LEDGER_BACKSTOP_RERUN_OPTION,
+            ],
+            decision_type="choice",
+            phase=current_phase,
+        )
+        newly_created = True
+    with get_pipeline_state_lock(pipeline_id):
+        pipeline = store.load_pipeline(pipeline_id)
+        pipeline.status = PipelineStatus.AWAITING_HUMAN
+        phase_execution = pipeline.get_phase_execution(current_phase)
+        phase_execution.status = PipelineStatus.AWAITING_HUMAN
+        store.save_pipeline(pipeline)
+    # Only announce a freshly-created decision. Reusing a pending decision
+    # across polls must not re-emit ``decision.created`` — a duplicate event
+    # for a decision the operator is already looking at (#3462 review).
+    if newly_created:
+        report_pipeline_status(
+            pipeline,
+            event_type="decision.created",
+            message=(
+                f"{current_phase.value} phase attests no operator "
+                f"decisions — awaiting operator confirmation (#3462)"
+            ),
+        )
+        _emit_pipeline_event(pipeline, "decision.created")
+
+    attest_resolved = dq.wait_for_decision(attest_decision.id)
+    attest_resolution = _unwrap_choice_resolution(
+        str(getattr(attest_resolved, "resolution", None) or "")
+    ).strip()
+    resolved_ok = attest_resolved.status == DecisionStatus.RESOLVED
+    confirmed = resolved_ok and _ledger_attestation_confirmed(attest_resolution)
+
+    if resolved_ok and not confirmed:
+        # Rejected — re-run the phase so producers register the decisions as
+        # first-class cq-N entries.
+        rerun_directive = _ledger_attestation_rerun_directive(
+            current_phase.value, attest_rationale, attest_resolution
+        )
+        logger.info(
+            "Explicit-none attestation rejected: re-running phase (#3462)",
+            pipeline_id=pipeline_id,
+            phase=current_phase.value,
+        )
+        with get_pipeline_state_lock(pipeline_id):
+            pipeline = store.load_pipeline(pipeline_id)
+            pipeline.status = PipelineStatus.RUNNING
+            phase_execution = pipeline.get_phase_execution(current_phase)
+            phase_execution.status = PipelineStatus.RUNNING
+            phase_execution.completed_at = None
+            phase_execution.hitl_review_cycles += 1
+            _alert_threshold = pipeline.config.max_hitl_review_cycles
+            if phase_execution.hitl_review_cycles >= _alert_threshold:
+                _broadcast_hitl_nonconvergence_alert(
+                    pipeline_id,
+                    pipeline,
+                    current_phase,
+                    phase_execution.hitl_review_cycles,
+                    _alert_threshold,
+                )
+            _perform_hitl_phase_rerun(
+                store=store,
+                spawner=spawner,
+                pipeline=pipeline,
+                phase_execution=phase_execution,
+                pipeline_id=pipeline_id,
+                current_phase=current_phase,
+                feedback_text=rerun_directive,
+                event_message=(
+                    f"Re-running {current_phase.value}: no-decisions attestation rejected (#3462)"
+                ),
+            )
+        return True, ledger_note, pipeline
+
+    if confirmed:
+        return False, ledger_note + " Operator confirmed the attestation.", pipeline
+    # Fail open to the phase gate on a non-RESOLVED terminal state (cancel):
+    # the gate itself still blocks for approval, mirroring the missing-ledger
+    # backstop's posture. Record the outcome accurately — do not claim a
+    # confirmation the operator never gave (#3462 review).
+    return (
+        False,
+        ledger_note + " Attestation confirmation was cancelled; deferring to the phase gate.",
+        pipeline,
+    )
+
+
 def _find_explicit_none_attestation(
     pipeline_id: str,
     phase_value: str,
@@ -28323,145 +28512,25 @@ def _health_monitor_poll(monitor, stop_event: threading.Event, interval: float =
                     # --- Explicit-none attestation confirmation (#3462) ---
                     # The producer's claim that this phase raises no operator
                     # decisions bypasses the entire register → bridge →
-                    # resolve chain, and the claim is itself a judgment call
-                    # the HITL contract assigns to the operator. Surface it
-                    # as its own confirmable decision instead of a sentence
-                    # embedded in the phase_gate question: confirming records
-                    # the operator's endorsement on the ledger note;
-                    # rejecting re-runs the phase so producers register the
-                    # decisions as first-class cq-N entries.
-                    _attest_role, _attest_rationale = _ledger_explicit_none
-                    _attest_question = _ledger_attestation_question(
-                        _attest_role, _attest_rationale, current_phase.value
-                    )
-                    # A converge-loop round (or a resume) re-enters this gate
-                    # with the same attestation — do not re-ask a question
-                    # the operator already answered, and reuse a pending one
-                    # instead of queueing a duplicate (mirrors the
-                    # phase_gate's #1152 guard).
-                    _prior_confirm = next(
-                        (
-                            d
-                            for d in reversed(pipeline.decisions)
-                            if d.decision_type == "choice"
-                            and d.phase == current_phase
-                            and d.question == _attest_question
-                            and d.status == DecisionStatus.RESOLVED
-                            and _ledger_attestation_confirmed(str(d.resolution or ""))
-                        ),
-                        None,
-                    )
-                    if _prior_confirm is not None:
-                        _ledger_note += " Operator confirmed the attestation."
-                    else:
-                        dq = get_decision_queue(pipeline_id, repo_path)
-                        _pending_attest = next(
-                            (
-                                d
-                                for d in reversed(pipeline.decisions)
-                                if d.decision_type == "choice"
-                                and d.phase == current_phase
-                                and d.question == _attest_question
-                                and d.status == DecisionStatus.PENDING
-                            ),
-                            None,
-                        )
-                        if _pending_attest is not None:
-                            _attest_decision = _pending_attest
-                        else:
-                            _attest_decision = dq.queue_decision(
-                                question=_attest_question,
-                                context=_ledger_note,
-                                options=[
-                                    _LEDGER_ATTESTATION_CONFIRM_OPTION,
-                                    _LEDGER_BACKSTOP_RERUN_OPTION,
-                                ],
-                                decision_type="choice",
-                                phase=current_phase,
-                            )
-                        with get_pipeline_state_lock(pipeline_id):
-                            pipeline = store.load_pipeline(pipeline_id)
-                            pipeline.status = PipelineStatus.AWAITING_HUMAN
-                            phase_execution = pipeline.get_phase_execution(current_phase)
-                            phase_execution.status = PipelineStatus.AWAITING_HUMAN
-                            store.save_pipeline(pipeline)
-                        report_pipeline_status(
-                            pipeline,
-                            event_type="decision.created",
-                            message=(
-                                f"{current_phase.value} phase attests no operator "
-                                f"decisions — awaiting operator confirmation (#3462)"
-                            ),
-                        )
-                        _emit_pipeline_event(pipeline, "decision.created")
-
-                        _attest_resolved = dq.wait_for_decision(_attest_decision.id)
-                        _attest_resolution = _unwrap_choice_resolution(
-                            str(getattr(_attest_resolved, "resolution", None) or "")
-                        ).strip()
-                        # Fail open to the phase gate on a non-RESOLVED
-                        # terminal state (cancel): the gate itself still
-                        # blocks for approval, mirroring the missing-ledger
-                        # backstop's posture.
-                        _confirmed = (
-                            _attest_resolved.status != DecisionStatus.RESOLVED
-                            or _ledger_attestation_confirmed(_attest_resolution)
+                    # resolve chain, and is itself a judgment call the HITL
+                    # contract assigns to the operator. Surface it as its own
+                    # confirmable decision (see the helper): confirming records
+                    # the operator's endorsement on the ledger note; rejecting
+                    # re-runs the phase to register cq-N entries.
+                    _rerun_requested, _ledger_note, pipeline = (
+                        _handle_explicit_none_attestation_gate(
+                            pipeline=pipeline,
+                            pipeline_id=pipeline_id,
+                            repo_path=repo_path,
+                            current_phase=current_phase,
+                            ledger_note=_ledger_note,
+                            explicit_none=_ledger_explicit_none,
+                            store=store,
+                            spawner=spawner,
                         )
-                        if not _confirmed:
-                            _rerun_directive = (
-                                f"The operator declined to confirm the "
-                                f"{current_phase.value} phase's no-decisions "
-                                f"attestation (#3462). The phase claimed: "
-                                f"“{_attest_rationale}”. Register each "
-                                f"operator-grade decision via `egg-contract "
-                                f"add-decision` — including decisions you "
-                                f"believe prior context already resolves: "
-                                f"register those with your recommended answer "
-                                f"as the first option and cite the resolving "
-                                f"context in its description. Belief about "
-                                f"resolution is a recommended disposition, not "
-                                f"a reason to skip registration."
-                            )
-                            if _attest_resolution.lower() != (
-                                _LEDGER_BACKSTOP_RERUN_OPTION.lower()
-                            ):
-                                _rerun_directive += f"\n\nOperator note: {_attest_resolution}"
-                            logger.info(
-                                "Explicit-none attestation rejected: re-running phase (#3462)",
-                                pipeline_id=pipeline_id,
-                                phase=current_phase.value,
-                            )
-                            with get_pipeline_state_lock(pipeline_id):
-                                pipeline = store.load_pipeline(pipeline_id)
-                                pipeline.status = PipelineStatus.RUNNING
-                                phase_execution = pipeline.get_phase_execution(current_phase)
-                                phase_execution.status = PipelineStatus.RUNNING
-                                phase_execution.completed_at = None
-                                phase_execution.hitl_review_cycles += 1
-                                _alert_threshold = pipeline.config.max_hitl_review_cycles
-                                if phase_execution.hitl_review_cycles >= _alert_threshold:
-                                    _broadcast_hitl_nonconvergence_alert(
-                                        pipeline_id,
-                                        pipeline,
-                                        current_phase,
-                                        phase_execution.hitl_review_cycles,
-                                        _alert_threshold,
-                                    )
-                                _perform_hitl_phase_rerun(
-                                    store=store,
-                                    spawner=spawner,
-                                    pipeline=pipeline,
-                                    phase_execution=phase_execution,
-                                    pipeline_id=pipeline_id,
-                                    current_phase=current_phase,
-                                    feedback_text=_rerun_directive,
-                                    event_message=(
-                                        f"Re-running {current_phase.value}: "
-                                        f"no-decisions attestation rejected (#3462)"
-                                    ),
-                                )
-                            continue  # Re-enter outer loop → re-run phase
-                        _ledger_note += " Operator confirmed the attestation."
+                    )
+                    if _rerun_requested:
+                        continue  # Re-enter outer loop → re-run phase
 
                 # Check for an existing pending phase_gate decision for this
                 # phase.  A prior agent-exit event may
diff --git a/orchestrator/tests/test_decision_ledger_gate.py b/orchestrator/tests/test_decision_ledger_gate.py
index 0bc0fe565d..3be834e981 100644
--- a/orchestrator/tests/test_decision_ledger_gate.py
+++ b/orchestrator/tests/test_decision_ledger_gate.py
@@ -290,3 +290,260 @@ def test_envelope_with_rerun_selection_rejects(self):
 
         envelope = json.dumps({"action": "select", "selected": _LEDGER_BACKSTOP_RERUN_OPTION})
         assert _ledger_attestation_confirmed(envelope) is False
+
+
+class TestLedgerAttestationRerunDirective:
+    """The re-run directive built when an explicit-none attestation is
+    rejected (#3462). Free-text resolutions ride along as an operator note;
+    the bare re-run option does not.
+    """
+
+    def test_directive_names_phase_rationale_and_registration_rule(self):
+        from routes.pipelines import _ledger_attestation_rerun_directive
+
+        directive = _ledger_attestation_rerun_directive(
+            "refine", "no choices this phase", "Re-run phase to register decisions"
+        )
+        assert "refine phase" in directive
+        assert "no choices this phase" in directive
+        assert "egg-contract add-decision" in directive
+        assert "recommended disposition" in directive
+
+    def test_bare_rerun_option_adds_no_operator_note(self):
+        from routes.pipelines import (
+            _LEDGER_BACKSTOP_RERUN_OPTION,
+            _ledger_attestation_rerun_directive,
+        )
+
+        directive = _ledger_attestation_rerun_directive(
+            "plan", "nothing to decide", _LEDGER_BACKSTOP_RERUN_OPTION
+        )
+        assert "Operator note:" not in directive
+
+    def test_free_text_resolution_rides_along_as_operator_note(self):
+        from routes.pipelines import _ledger_attestation_rerun_directive
+
+        directive = _ledger_attestation_rerun_directive(
+            "plan", "nothing to decide", "register the deploy-target choice"
+        )
+        assert "Operator note: register the deploy-target choice" in directive
+
+
+class TestHandleExplicitNoneAttestationGate:
+    """The orchestration around the explicit-none confirmation decision (#3462).
+
+    ``_handle_explicit_none_attestation_gate`` queues the confirmable choice,
+    waits on it, and either falls through to the phase gate (confirm / cancel
+    fail-open) or re-runs the phase (reject). The dedup branches — a prior
+    confirmation reused without re-asking, and a pending decision reused
+    without re-emitting ``decision.created`` — are the net-new integration
+    logic this covers.
+    """
+
+    _ROLE = "refiner"
+    _RATIONALE = "requirements are unambiguous; no operator choices"
+
+    def _fake_phase_execution(self):
+        from types import SimpleNamespace
+
+        return SimpleNamespace(
+            status=None,
+            completed_at="2026-04-01T00:00:00+00:00",
+            hitl_review_cycles=0,
+        )
+
+    def _fake_pipeline(self, decisions=None):
+        from types import SimpleNamespace
+
+        phase_exec = self._fake_phase_execution()
+        return SimpleNamespace(
+            decisions=list(decisions or []),
+            status=None,
+            config=SimpleNamespace(max_hitl_review_cycles=3),
+            get_phase_execution=lambda _phase: phase_exec,
+        )
+
+    def _choice_decision(self, question, status, resolution=None):
+        from types import SimpleNamespace
+
+        from models import PipelinePhase
+
+        return SimpleNamespace(
+            decision_type="choice",
+            phase=PipelinePhase.REFINE,
+            question=question,
+            status=status,
+            resolution=resolution,
+        )
+
+    def _call(self, *, pipeline, dq, store=None, spawner=None):
+        """Invoke the helper with all module-level collaborators patched."""
+        from unittest.mock import MagicMock, patch
+
+        from models import PipelinePhase
+        from routes.pipelines import _handle_explicit_none_attestation_gate
+
+        store = store or MagicMock()
+        store.load_pipeline.return_value = pipeline
+        spawner = spawner or MagicMock()
+
+        with (
+            patch("routes.pipelines.get_decision_queue", return_value=dq) as m_get_dq,
+            patch("routes.pipelines.get_pipeline_state_lock", return_value=MagicMock()),
+            patch("routes.pipelines.report_pipeline_status") as m_report,
+            patch("routes.pipelines._emit_pipeline_event") as m_emit,
+            patch("routes.pipelines._perform_hitl_phase_rerun") as m_rerun,
+            patch("routes.pipelines._broadcast_hitl_nonconvergence_alert") as m_alert,
+        ):
+            rerun_requested, note, out_pipeline = _handle_explicit_none_attestation_gate(
+                pipeline=pipeline,
+                pipeline_id="pipeline-id",
+                repo_path=Path("/tmp/repo"),
+                current_phase=PipelinePhase.REFINE,
+                ledger_note="Decision ledger: explicitly none — refiner attested: x",
+                explicit_none=(self._ROLE, self._RATIONALE),
+                store=store,
+                spawner=spawner,
+            )
+        return {
+            "rerun_requested": rerun_requested,
+            "note": note,
+            "pipeline": out_pipeline,
+            "get_dq": m_get_dq,
+            "report": m_report,
+            "emit": m_emit,
+            "rerun": m_rerun,
+            "alert": m_alert,
+            "store": store,
+        }
+
+    def _question(self):
+        from models import PipelinePhase
+        from routes.pipelines import _ledger_attestation_question
+
+        return _ledger_attestation_question(self._ROLE, self._RATIONALE, PipelinePhase.REFINE.value)
+
+    def test_confirm_falls_through_with_note(self):
+        from unittest.mock import MagicMock
+
+        from models import DecisionStatus
+
+        dq = MagicMock()
+        dq.queue_decision.return_value = MagicMock(id="attest-1")
+        dq.wait_for_decision.return_value = MagicMock(
+            status=DecisionStatus.RESOLVED, resolution="confirm"
+        )
+        pipeline = self._fake_pipeline()
+
+        res = self._call(pipeline=pipeline, dq=dq)
+
+        assert res["rerun_requested"] is False
+        assert res["note"].endswith("Operator confirmed the attestation.")
+        dq.queue_decision.assert_called_once()
+        # A freshly-created decision is announced exactly once.
+        assert res["emit"].call_count == 1
+        res["rerun"].assert_not_called()
+
+    def test_reject_rerun_option_triggers_phase_rerun(self):
+        from unittest.mock import MagicMock
+
+        from models import DecisionStatus, PipelineStatus
+        from routes.pipelines import _LEDGER_BACKSTOP_RERUN_OPTION
+
+        dq = MagicMock()
+        dq.queue_decision.return_value = MagicMock(id="attest-1")
+        dq.wait_for_decision.return_value = MagicMock(
+            status=DecisionStatus.RESOLVED, resolution=_LEDGER_BACKSTOP_RERUN_OPTION
+        )
+        pipeline = self._fake_pipeline()
+
+        res = self._call(pipeline=pipeline, dq=dq)
+
+        assert res["rerun_requested"] is True
+        res["rerun"].assert_called_once()
+        directive = res["rerun"].call_args.kwargs["feedback_text"]
+        assert "egg-contract add-decision" in directive
+        assert "Operator note:" not in directive
+        assert pipeline.status == PipelineStatus.RUNNING
+
+    def test_reject_free_text_forwards_operator_note(self):
+        from unittest.mock import MagicMock
+
+        from models import DecisionStatus
+
+        dq = MagicMock()
+        dq.queue_decision.return_value = MagicMock(id="attest-1")
+        dq.wait_for_decision.return_value = MagicMock(
+            status=DecisionStatus.RESOLVED,
+            resolution="you skipped the deploy-target choice",
+        )
+        pipeline = self._fake_pipeline()
+
+        res = self._call(pipeline=pipeline, dq=dq)
+
+        assert res["rerun_requested"] is True
+        directive = res["rerun"].call_args.kwargs["feedback_text"]
+        assert "Operator note: you skipped the deploy-target choice" in directive
+
+    def test_prior_confirmation_is_reused_without_reasking(self):
+        from unittest.mock import MagicMock
+
+        from models import DecisionStatus
+
+        prior = self._choice_decision(
+            self._question(), DecisionStatus.RESOLVED, resolution="confirm"
+        )
+        pipeline = self._fake_pipeline(decisions=[prior])
+        dq = MagicMock()
+
+        res = self._call(pipeline=pipeline, dq=dq)
+
+        assert res["rerun_requested"] is False
+        assert res["note"].endswith("Operator confirmed the attestation.")
+        # No new decision is queued and the queue is never even fetched.
+        res["get_dq"].assert_not_called()
+        res["emit"].assert_not_called()
+        res["rerun"].assert_not_called()
+
+    def test_pending_decision_reused_without_reemitting_created(self):
+        from unittest.mock import MagicMock
+
+        from models import DecisionStatus
+
+        pending = self._choice_decision(self._question(), DecisionStatus.PENDING)
+        pending.id = "attest-pending"
+        pipeline = self._fake_pipeline(decisions=[pending])
+        dq = MagicMock()
+        dq.wait_for_decision.return_value = MagicMock(
+            status=DecisionStatus.RESOLVED, resolution="confirm"
+        )
+
+        res = self._call(pipeline=pipeline, dq=dq)
+
+        # The pending decision is reused, not re-queued...
+        dq.queue_decision.assert_not_called()
+        dq.wait_for_decision.assert_called_once_with("attest-pending")
+        # ...and reusing it must not re-announce decision.created.
+        res["emit"].assert_not_called()
+        assert res["rerun_requested"] is False
+
+    def test_cancel_fails_open_with_accurate_note(self):
+        from unittest.mock import MagicMock
+
+        from models import DecisionStatus
+
+        dq = MagicMock()
+        dq.queue_decision.return_value = MagicMock(id="attest-1")
+        dq.wait_for_decision.return_value = MagicMock(
+            status=DecisionStatus.CANCELLED, resolution=None
+        )
+        pipeline = self._fake_pipeline()
+
+        res = self._call(pipeline=pipeline, dq=dq)
+
+        # Fail open to the phase gate, but record the cancel accurately —
+        # never claim a confirmation the operator did not give (#3462 review).
+        assert res["rerun_requested"] is False
+        assert "Operator confirmed the attestation." not in res["note"]
+        assert "cancelled" in res["note"].lower()
+        res["rerun"].assert_not_called()