Fix #2079: wake stuck producers via brc_confirmation_timeout escalation

orchestrator/health_monitor.py

@@ -160,6 +160,11 @@ def set_current_phase(self, phase: str) -> None:
         with self._lock:
             self._current_phase = phase
 
+    def get_current_phase(self) -> str | None:
+        """Return the phase last set via :meth:`set_current_phase`."""
+        with self._lock:
+            return self._current_phase
+
     def _get_heartbeat_threshold(self) -> int:
         """Return the heartbeat timeout threshold for the current phase.
 
@@ -768,6 +773,19 @@ def check_brc_progress(self) -> list[dict[str, Any]]:
                 if agent_state:
                     agent_state.brc_progress_escalated = False
 
+            # Per-iteration breadcrumb so post-mortems can verify the check
+            # ran and what it observed (#2079).
+            if fully_acked:
+                logger.info(
+                    "BRC progress check observed fully-acked producers",
+                    pipeline_id=self._pipeline_id,
+                    producers={
+                        p: round(now - self._fully_acked_first_seen.get(p, now), 1)
+                        for p in fully_acked
+                    },
+                    timeout_seconds=timeout,
+                )
+
             # Timeout is measured from when the monitor first observed the
             # fully-acked state (_fully_acked_first_seen), not from the
             # original proposal timestamp, because the monitor may start
@@ -783,9 +801,20 @@ def check_brc_progress(self) -> list[dict[str, Any]]:
                 if elapsed <= timeout:
                     continue
 
-                # Skip producers with no agent state (never emitted heartbeat/progress)
+                # Skip producers with no agent state (never emitted
+                # heartbeat/progress).  This branch is unexpected in
+                # practice — every producer in the fully-acked set has
+                # at minimum proposed, which routes through MESSAGE_SENT
+                # and registers agent_state.  Log loudly so future
+                # post-mortems can audit (#2079).
                 agent_state = self._agents.get(producer)
                 if not agent_state:
+                    logger.warning(
+                        "BRC progress timeout but producer has no agent_state",
+                        pipeline_id=self._pipeline_id,
+                        producer=producer,
+                        elapsed_seconds=int(elapsed),
+                    )
                     continue
                 if agent_state.brc_progress_escalated:
                     continue
@@ -802,11 +831,21 @@ def check_brc_progress(self) -> list[dict[str, Any]]:
                 }
                 actions.append(action)
 
+                logger.warning(
+                    "BRC progress timeout — escalating",
+                    pipeline_id=self._pipeline_id,
+                    producer=producer,
+                    elapsed_seconds=int(elapsed),
+                    timeout_seconds=timeout,
+                )
+
                 escalation_type = "overseer" if self._config.overseer_enabled else "hitl"
                 escalation = {
                     "type": escalation_type,
                     "agent_id": producer,
                     "reason": action["reason"],
+                    "alert_type": "brc_confirmation_timeout",
+                    "elapsed_seconds": int(elapsed),
                     "timestamp": datetime.now(UTC).isoformat(),
                 }
                 escalations.append(escalation)

orchestrator/routes/pipelines.py

@@ -507,6 +507,92 @@ def _check_and_respawn_overseer(
     return overseer_container_id, overseer_respawn_count
 
 
+def _send_brc_confirmation_nudge(
+    escalation: dict[str, Any],
+    pipeline_id: str,
+    phase: str | None,
+) -> bool:
+    """Wake a producer stuck post-ACK with a directed OVERSEER_ALERT (#2079).
+
+    Wired as an escalation callback for HealthMonitor's
+    ``brc_confirmation_timeout`` alert.  The deterministic detector in
+    ``check_brc_progress`` knows the exact remediation, so we deliver
+    it directly to the stuck producer rather than relying on the
+    overseer agent's discretion.
+
+    Uses ``OVERSEER_ALERT`` (not ``STATUS`` or ``NUDGE``) because the
+    producer's post-confirm wait_loop filter is
+    ``CONSENSUS_CONFIRMED,CONSENSUS_RE_REVIEW,OVERSEER_ALERT`` — only
+    those types wake the wait.  The subject calls out that the alert
+    originated from the orchestrator's deterministic detector rather
+    than the overseer agent.
+
+    Returns True when a message was posted, False otherwise (wrong
+    alert type, missing fields, message store unavailable, send error).
+    """
+    if escalation.get("alert_type") != "brc_confirmation_timeout":
+        return False
+
+    producer = escalation.get("agent_id")
+    if not producer:
+        return False
+
+    elapsed = escalation.get("elapsed_seconds", 0)
+
+    store_fn = _get_message_store()
+    if store_fn is None:
+        return False
+
+    try:
+        from message_store import Message, MessageType
+    except ImportError:
+        return False
+
+    body = (
+        f"You are PROPOSED and fully ACKed but have not confirmed in "
+        f"{elapsed}s. Call `mcp__brc__confirm` now. If it returns "
+        "status='pending_acks' (e.g., another producer hasn't proposed "
+        "yet, or the global zero-proposal guard is blocking you), wait "
+        "on the prerequisite events instead — CONSENSUS_PROPOSE from "
+        "missing producers, CONSENSUS_ACK from your reviewers, or "
+        "CONSENSUS_RE_REVIEW — then retry confirm."
+    )
+
+    try:
+        msg_store = store_fn()
+        msg_store.add_message(
+            Message(
+                pipeline_id=pipeline_id,
+                from_role="orchestrator",
+                to_role=producer,
+                message_type=MessageType.OVERSEER_ALERT,
+                subject="BRC confirmation timeout — call mcp__brc__confirm",
+                body=body,
+                phase=phase,
+                metadata={
+                    "alert_type": "brc_confirmation_timeout",
+                    "elapsed_seconds": elapsed,
+                    "source": "health_monitor",
+                },
+            )
+        )
+        logger.info(
+            "Sent BRC confirmation-timeout nudge",
+            pipeline_id=pipeline_id,
+            producer=producer,
+            elapsed_seconds=elapsed,
+        )
+        return True
+    except Exception as send_err:
+        logger.warning(
+            "Failed to send BRC confirmation-timeout nudge (non-fatal)",
+            pipeline_id=pipeline_id,
+            producer=producer,
+            error=str(send_err),
+        )
+        return False
+
+
 def _teardown_phase_overseer(
     spawner: "ContainerSpawner",
     container_id: str,
@@ -11454,6 +11540,17 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None:
             # Sync the phase-aware threshold with the current pipeline phase
             health_monitor_instance.set_current_phase(pipeline.current_phase.value)
 
+            # Wake stuck producers directly when check_brc_progress fires
+            # so the deterministic detector actually drives remediation
+            # instead of relying on the overseer agent's discretion (#2079).
+            # The closure reads the monitor's current phase at fire time so
+            # the message records the phase the producer is actually in.
+            def _on_health_escalation(escalation: dict[str, Any]) -> None:
+                phase = health_monitor_instance.get_current_phase()
+                _send_brc_confirmation_nudge(escalation, pipeline_id, phase)
+
+            health_monitor_instance.on_escalation(_on_health_escalation)
+
             # Start a background polling thread for time-based tripwires
             health_monitor_timer = threading.Event()
 

orchestrator/tests/test_brc_confirmation_nudge.py

@@ -0,0 +1,208 @@
+"""
+Tests for the BRC confirmation-timeout nudge wired into HealthMonitor escalations.
+
+Covers ``_send_brc_confirmation_nudge`` in ``orchestrator/routes/pipelines.py``,
+which wakes producers stuck post-ACK by posting a directed OVERSEER_ALERT
+when ``check_brc_progress`` fires the deterministic timeout (#2079).
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# ---------------------------------------------------------------------------
+# Path setup
+# ---------------------------------------------------------------------------
+_orchestrator_path = Path(__file__).parent.parent
+if str(_orchestrator_path) not in sys.path:
+    sys.path.insert(0, str(_orchestrator_path))
+
+_shared_path = Path(__file__).parent.parent.parent / "shared"
+if _shared_path.exists() and str(_shared_path) not in sys.path:
+    sys.path.insert(0, str(_shared_path))
+
+# Mock docker before importing route modules that depend on it.
+sys.modules.setdefault("docker", MagicMock())
+sys.modules.setdefault("docker.errors", MagicMock())
+sys.modules.setdefault("docker.types", MagicMock())
+
+try:
+    from message_store import Message, MessageType
+    from routes.pipelines import _send_brc_confirmation_nudge
+except ImportError as exc:
+    pytest.skip(
+        f"Required orchestrator modules not available: {exc}",
+        allow_module_level=True,
+    )
+
+
+PIPELINE_ID = "issue-2079"
+
+
+def _make_escalation(
+    *,
+    alert_type: str = "brc_confirmation_timeout",
+    agent_id: str = "documenter",
+    elapsed_seconds: int = 240,
+) -> dict:
+    """Build a HealthMonitor escalation dict matching check_brc_progress."""
+    return {
+        "type": "overseer",
+        "agent_id": agent_id,
+        "alert_type": alert_type,
+        "elapsed_seconds": elapsed_seconds,
+        "reason": (
+            f"Producer {agent_id} fully ACKed but not confirmed "
+            f"for {elapsed_seconds}s (timeout: 180s)"
+        ),
+        "timestamp": "2026-04-25T18:07:22+00:00",
+    }
+
+
+class TestSendBRCConfirmationNudge:
+    """The nudge posts a directed OVERSEER_ALERT to the stuck producer."""
+
+    def test_posts_directed_overseer_alert(self):
+        """A valid escalation results in an OVERSEER_ALERT addressed to the producer."""
+        msg_store = MagicMock()
+        store_factory = MagicMock(return_value=msg_store)
+
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=store_factory,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                _make_escalation(agent_id="documenter", elapsed_seconds=240),
+                PIPELINE_ID,
+                phase="implement",
+            )
+
+        assert posted is True
+        msg_store.add_message.assert_called_once()
+        message = msg_store.add_message.call_args[0][0]
+        assert isinstance(message, Message)
+        assert message.pipeline_id == PIPELINE_ID
+        assert message.from_role == "orchestrator"
+        assert message.to_role == "documenter"
+        # Must be OVERSEER_ALERT to wake the producer's post-confirm wait_loop,
+        # whose filter is CONSENSUS_CONFIRMED,CONSENSUS_RE_REVIEW,OVERSEER_ALERT.
+        assert message.message_type == MessageType.OVERSEER_ALERT
+        assert message.phase == "implement"
+        assert "mcp__brc__confirm" in message.body
+        assert "240s" in message.body
+        assert message.metadata["alert_type"] == "brc_confirmation_timeout"
+        assert message.metadata["elapsed_seconds"] == 240
+        assert message.metadata["source"] == "health_monitor"
+
+    def test_ignores_non_brc_escalations(self):
+        """Escalations from other tripwires are ignored — no message posted."""
+        msg_store = MagicMock()
+        store_factory = MagicMock(return_value=msg_store)
+
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=store_factory,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                _make_escalation(alert_type="heartbeat_timeout"),
+                PIPELINE_ID,
+                phase="implement",
+            )
+
+        assert posted is False
+        msg_store.add_message.assert_not_called()
+
+    def test_ignores_escalation_with_no_alert_type(self):
+        """Escalations missing alert_type (legacy tripwires) are ignored."""
+        msg_store = MagicMock()
+        store_factory = MagicMock(return_value=msg_store)
+
+        escalation = _make_escalation()
+        escalation.pop("alert_type")
+
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=store_factory,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                escalation,
+                PIPELINE_ID,
+                phase="implement",
+            )
+
+        assert posted is False
+        msg_store.add_message.assert_not_called()
+
+    def test_ignores_escalation_missing_agent_id(self):
+        """Defensive: escalation without agent_id is skipped, not crashed."""
+        msg_store = MagicMock()
+        store_factory = MagicMock(return_value=msg_store)
+
+        escalation = _make_escalation()
+        escalation.pop("agent_id")
+
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=store_factory,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                escalation,
+                PIPELINE_ID,
+                phase="implement",
+            )
+
+        assert posted is False
+        msg_store.add_message.assert_not_called()
+
+    def test_returns_false_when_message_store_unavailable(self):
+        """Returns False (does not raise) when the message store factory is None."""
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=None,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                _make_escalation(),
+                PIPELINE_ID,
+                phase="implement",
+            )
+
+        assert posted is False
+
+    def test_returns_false_when_send_raises(self):
+        """A send error is logged + swallowed; nudge call itself does not raise."""
+        msg_store = MagicMock()
+        msg_store.add_message.side_effect = RuntimeError("boom")
+        store_factory = MagicMock(return_value=msg_store)
+
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=store_factory,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                _make_escalation(),
+                PIPELINE_ID,
+                phase="implement",
+            )
+
+        assert posted is False
+
+    def test_phase_none_passes_through(self):
+        """phase=None is permitted (Message.phase is Optional)."""
+        msg_store = MagicMock()
+        store_factory = MagicMock(return_value=msg_store)
+
+        with patch(
+            "routes.pipelines._get_message_store",
+            return_value=store_factory,
+        ):
+            posted = _send_brc_confirmation_nudge(
+                _make_escalation(),
+                PIPELINE_ID,
+                phase=None,
+            )
+
+        assert posted is True
+        message = msg_store.add_message.call_args[0][0]
+        assert message.phase is None