diff --git a/docs/guides/concurrent-execution.md b/docs/guides/concurrent-execution.md index 6fcef21f1a..948a1c3df2 100644 --- a/docs/guides/concurrent-execution.md +++ b/docs/guides/concurrent-execution.md @@ -727,13 +727,13 @@ If consensus is not reached within `consensus_timeout_minutes`, the orchestrator Once the bus and containers have been quiet for the full gate window, the BRC tracker (`PeerConsensusTracker.handle_timeout()`) evaluates blocking agents by role criticality: -- **Critical blockers** (required reviewers still unconfirmed): emits `CONSENSUS_FAILURE` and creates a HITL decision asking how to proceed. -- **Advisory-only blockers** (non-critical roles unconfirmed): emits `CONSENSUS_TIMEOUT` and proceeds automatically — no HITL created. -- **No blockers**: proceeds immediately with no HITL. +- **Critical blockers** (required reviewers still unconfirmed): emits `CONSENSUS_FAILURE` and publishes an `OVERSEER_ALERT` (subject `consensus-timeout: [high]`, where `` is the first critical-blocker role; the phase is preserved in `metadata.phase`) so the SDLC skill surfaces it as a non-blocking notification — see [issue #2264](https://github.com/jwbron/egg/issues/2264). The alert's `metadata.blocking_agents` is narrowed to the critical-blocker roles only (advisory roles, if any, are excluded so the high-priority signal isn't diluted). +- **Advisory-only blockers** (non-critical roles unconfirmed): emits `CONSENSUS_TIMEOUT` and proceeds automatically — no alert created. +- **No blockers**: proceeds immediately with no alert. After the timeout check, if the approval matrix still has unresolved NACKs (producers that exited without addressing reviewer feedback), the phase returns failure regardless of which agents are confirmed. -If the BRC tracker is unavailable, the orchestrator falls back to the old behavior and creates a generic HITL decision for any timeout. +If the BRC tracker is unavailable, the orchestrator publishes a fallback `OVERSEER_ALERT` (subject `consensus-timeout: [medium]`, falling back to `` in the role slot when no blocking agents are reported) so the timeout remains visible to the operator. The alert metadata carries `anomaly_type=consensus-timeout`, `phase`, `blocking_agents`, `latest_proposal_at`, `latest_heartbeat_at`, and `consensus_timeout_minutes`. The polling loop continues with a bounded post-timeout budget (~60 minutes before still-running containers are force-killed); the alert is informational, not a gate, and the operator can intervene with `cancel_task` or `restart_phase` if desired. Timeout handling is idempotent — if the timeout fires multiple times (e.g., due to a race with the overseer), only the first invocation takes effect. diff --git a/docs/guides/sdlc-pipeline.md b/docs/guides/sdlc-pipeline.md index 0bf0c7ff79..904670c1e8 100644 --- a/docs/guides/sdlc-pipeline.md +++ b/docs/guides/sdlc-pipeline.md @@ -1444,8 +1444,13 @@ role. Messages are filtered by `to_role` — only targeted messages and broadcas (`to_role: "all"`) are returned. **Consensus timeout**: If agents don't reach consensus within `consensus_timeout_minutes`, -a HITL decision is created. Check agent states via `egg-orch pipeline status` to -identify blocked or stuck agents. +the orchestrator publishes an `OVERSEER_ALERT` (subject `consensus-timeout: []`, +matching the SDLC skill's `: []` convention so "Check agent +logs" can extract the role) rather than gating the pipeline on a `choice` decision +(see [issue #2264](https://github.com/jwbron/egg/issues/2264)). The SDLC skill surfaces the alert via +its existing notification flow (Check agent logs / Acknowledge / Cancel pipeline). Check agent +states via `egg-orch pipeline status` to identify blocked or stuck agents; intervene with +`cancel_task` or `restart_phase` if you want to act. **Message bus empty**: Verify the pipeline has `concurrent_execution: true` in its config. The message bus is only active for concurrent pipelines. diff --git a/orchestrator/models.py b/orchestrator/models.py index 2c421322ba..4c66411d57 100644 --- a/orchestrator/models.py +++ b/orchestrator/models.py @@ -375,7 +375,9 @@ class PipelineConfig(BaseModel): default=30, ge=1, description="Suggested message polling interval for agents" ) consensus_timeout_minutes: int = Field( - default=30, ge=1, description="Timeout for consensus before HITL escalation" + default=30, + ge=1, + description="Timeout for consensus before publishing an OVERSEER_ALERT (#2264)", ) post_consensus_iteration_budget_seconds: int = Field( default=3600, @@ -405,9 +407,9 @@ class PipelineConfig(BaseModel): default=300, ge=0, description=( - "Defer the consensus-timeout HITL decision while any BRC progress signal " + "Defer the consensus-timeout OVERSEER_ALERT while any BRC progress signal " "(CONSENSUS_PROPOSE/ACK/NACK or container heartbeat) has fired within this " - "many seconds. 0 disables the gate. (#2243)" + "many seconds. 0 disables the gate. (#2243, #2264)" ), ) agent_idle_timeout_minutes: int = Field( diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 008455006f..2b058cc7df 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -9568,8 +9568,9 @@ def _check_brc_progress_gate( ) -> tuple[bool, str | None]: """Return (defer, reason) for the BRC consensus-timeout progress gate (#2243). - Defers the auto-HITL consensus-failure decision when *any* of the - following has fired within ``gate_seconds``: + Defers the consensus-timeout ``OVERSEER_ALERT`` (#2264; previously + an auto-``choice`` HITL decision) when *any* of the following has + fired within ``gate_seconds``: * The BRC tracker's most recent ``CONSENSUS_PROPOSE`` (producer proposal) timestamp. @@ -9582,18 +9583,19 @@ def _check_brc_progress_gate( :data:`consensus_timeout_minutes` we previously opened a `choice` decision unconditionally, even when producers were minutes from their first commit. With the gate, the polling loop keeps polling - while signals are alive; the decision only opens once the bus and - containers have both gone quiet for ``gate_seconds``. + while signals are alive; the alert is only published once the bus + and containers have both gone quiet for ``gate_seconds``. ``gate_seconds <= 0`` disables the gate (returns ``(False, None)``). Failures in any signal source are logged at WARNING and treated as "no signal from that source" — never as a gate defer, since a - crashed signal collector must not silently keep us off the HITL + crashed signal collector must not silently keep us off the alert surface. - Heartbeat-cadence contract: the decision-17 path (coder mid-merge- - conflict before any ``CONSENSUS_PROPOSE``) relies on container - heartbeats firing at least every ``gate_seconds``. Sandbox + Heartbeat-cadence contract: the coder-mid-merge-conflict path + (no ``CONSENSUS_PROPOSE`` yet, only container heartbeats — the + original incident's ``decision-17`` flavour, pre-#2264) relies on + container heartbeats firing at least every ``gate_seconds``. Sandbox heartbeats (see ``shared/egg_agent`` heartbeat scheduler and ``orchestrator/health_monitor.py``) cadence today is well under 300s, but a long uninterruptible subprocess (e.g. ``git rebase`` @@ -9679,20 +9681,182 @@ def _check_brc_progress_gate( return False, None +def _latest_active_role_heartbeat(active_role_names: list[str]) -> datetime | None: + """Return the most recent heartbeat timestamp across ``active_role_names``. + + Mirrors the heartbeat half of :func:`_check_brc_progress_gate` so the + consensus-timeout ``OVERSEER_ALERT`` carries a meaningful + ``latest_heartbeat_at`` value. Filters by active role to avoid + pollution from stale entries in the singleton ``HealthMonitor``. + + Returns ``None`` when no live heartbeat is available (no roles + given, no health monitor, or any failure in the lookup — failures + are logged at WARNING and treated as "no signal", consistent with + the gate). + """ + if not active_role_names: + return None + try: + from health_monitor import get_health_monitor + + hm = get_health_monitor() + if hm is None: + return None + active_set = set(active_role_names) + latest_hb: float | None = None + with hm._lock: # noqa: SLF001 — read-only snapshot + hb_snapshot = dict(hm._last_heartbeat) # noqa: SLF001 + for agent_id, hb_time in hb_snapshot.items(): + if agent_id not in active_set: + continue + if latest_hb is None or hb_time > latest_hb: + latest_hb = hb_time + if latest_hb is None: + return None + return datetime.fromtimestamp(latest_hb, tz=UTC) + except Exception as e: + logger.warning( + "Consensus-timeout alert heartbeat lookup failed", + error=str(e), + exc_info=True, + ) + return None + + +def _publish_consensus_timeout_alert( + pipeline: Pipeline, + pipeline_id: str, + consensus_timeout: float, + blocking_agents: list[str], + *, + priority: str, + latest_proposal_at: datetime | None, + latest_heartbeat_at: datetime | None, + slice_id: str | None, +) -> None: + """Publish a consensus-timeout ``OVERSEER_ALERT`` (#2264). + + Replaces the old auto-``choice`` HITL decision the orchestrator + used to open at ``consensus_timeout_minutes``. The SDLC skill's + existing ``OVERSEER_ALERT`` flow surfaces this as a non-blocking + notification (Check agent logs / Acknowledge / Cancel pipeline) + rather than gating the pipeline on a binary choice. + + Best-effort: if the message store import or write fails, log at + WARNING and return — the orchestrator log is the always-on + fallback (mirrors the slice-cascade alert path). + """ + timeout_minutes = int(consensus_timeout / 60) + phase_value = ( + pipeline.current_phase.value + if hasattr(pipeline.current_phase, "value") + else str(pipeline.current_phase) + ) + # Subject role slot follows the SDLC skill convention + # ``: []`` (skills/sdlc/SKILL.md + # §"Overseer Alert Detection") so "Check agent logs" extracts a role + # the host can pass to ``get_container_logs``. Fall back to the phase + # only when no blocking role is reported — the phase still appears in + # ``metadata.phase`` regardless. + subject_role = blocking_agents[0] if blocking_agents else phase_value + subject = f"consensus-timeout: {subject_role} [{priority}]" + blockers_render = ", ".join(blocking_agents) if blocking_agents else "(none reported)" + proposal_render = ( + latest_proposal_at.isoformat() if latest_proposal_at is not None else "no proposals seen" + ) + heartbeat_render = ( + latest_heartbeat_at.isoformat() + if latest_heartbeat_at is not None + else "no recent heartbeat" + ) + body = ( + f"BRC consensus has not converged after {timeout_minutes} minutes " + f"in phase '{phase_value}'.\n" + f"Blocking agents: {blockers_render}\n" + f"Latest proposal: {proposal_render}\n" + f"Latest heartbeat (active roles): {heartbeat_render}\n\n" + "The pipeline continues to poll for convergence (up to ~60 min " + "before still-running containers are force-killed). If you want " + "to intervene, use `cancel_task` to stop the pipeline or " + "`restart_phase` to retry." + ) + metadata: dict[str, Any] = { + "anomaly_type": "consensus-timeout", + "phase": phase_value, + "blocking_agents": list(blocking_agents), + "latest_proposal_at": ( + latest_proposal_at.isoformat() if latest_proposal_at is not None else None + ), + "latest_heartbeat_at": ( + latest_heartbeat_at.isoformat() if latest_heartbeat_at is not None else None + ), + "consensus_timeout_minutes": timeout_minutes, + "priority": priority, + } + if slice_id is not None: + metadata["slice_id"] = slice_id + + try: + try: + from message_store import Message, MessageType + except ImportError: + from ..message_store import ( # type: ignore[no-redef] + Message, + MessageType, + ) + store_fn = _get_message_store() + if store_fn is None: + logger.warning( + "Consensus-timeout alert: message store unavailable", + pipeline_id=pipeline_id, + ) + return + msg_store = store_fn() + msg_store.add_message( + Message( + pipeline_id=pipeline_id, + from_role="orchestrator", + to_role="all", + message_type=MessageType.OVERSEER_ALERT, + subject=subject, + body=body, + metadata=metadata, + phase=phase_value, + ) + ) + except Exception as e: + logger.warning( + "Failed to publish consensus-timeout OVERSEER_ALERT", + pipeline_id=pipeline_id, + error=str(e), + exc_info=True, + ) + + def _handle_brc_consensus_timeout( pipeline: Pipeline, pipeline_id: str, consensus_timeout: float, blocking_agents: list[str], - store: StateStore, + store: StateStore, # noqa: ARG001 — kept for call-site compatibility (#2264) slice_id: str | None = None, + active_role_names: list[str] | None = None, ) -> None: # Extracted from _run_concurrent_phase so k3s-style top-level-module # layouts (and tests) can exercise this path in isolation — issue #1783. # ``slice_id`` is propagated so per-slice trackers (#2137) are looked # up under the nested ``{pipeline_id}/{slice_id}`` key. + # + # Issue #2264: the auto-``choice`` HITL decision this used to open + # was the wrong protocol shape — the platform should not gate the + # pipeline on a binary choice when the operator already has the + # levers (`cancel_task`, `restart_phase`, `provide_input`). The + # two former decision paths now publish ``OVERSEER_ALERT`` messages + # so the SDLC skill's existing alert flow surfaces them as + # notifications rather than a blocking decision. _brc_handled = False _brc_timeout_result: dict | None = None + _brc_tracker = None try: try: from peer_consensus import get_peer_consensus_tracker @@ -9713,26 +9877,50 @@ def _handle_brc_consensus_timeout( ) except Exception as e: logger.warning( - "BRC timeout check failed, falling back to HITL", + "BRC timeout check failed, falling back to OVERSEER_ALERT", pipeline_id=pipeline_id, error=str(e), ) + latest_proposal_at: datetime | None = None + if _brc_tracker is not None: + try: + latest_proposal_at = _brc_tracker.get_latest_proposal_timestamp() + except Exception as e: + logger.warning( + "Consensus-timeout alert proposal lookup failed", + pipeline_id=pipeline_id, + error=str(e), + exc_info=True, + ) + latest_heartbeat_at = _latest_active_role_heartbeat(active_role_names or []) + if ( _brc_handled and _brc_timeout_result is not None and _brc_timeout_result.get("action") == "escalate" ): - _persist_hitl_decision( - pipeline_id, + # Narrow the alert's blocking_agents to the *critical* blockers + # the tracker just escalated on. The caller-supplied + # ``blocking_agents`` is the full unconfirmed-roles set + # (advisory + critical) from ``evaluate()`` — surfacing + # advisory roles on a high-priority alert dilutes the signal. + critical_entries = _brc_timeout_result.get("critical_blockers") or [] + critical_role_names: list[str] = [] + for entry in critical_entries: + for role in (entry.get("reviewer_role"), entry.get("producer_role")): + if role and role not in critical_role_names: + critical_role_names.append(role) + escalate_blocking = critical_role_names or blocking_agents + _publish_consensus_timeout_alert( pipeline, - store, - question=( - f"BRC consensus failure: critical reviewers unconfirmed after " - f"{int(consensus_timeout / 60)} minutes. How to proceed?" - ), - options=["Continue waiting", "Accept current state", "Abort phase"], - phase=pipeline.current_phase, + pipeline_id, + consensus_timeout, + escalate_blocking, + priority="high", + latest_proposal_at=latest_proposal_at, + latest_heartbeat_at=latest_heartbeat_at, + slice_id=slice_id, ) elif not _brc_handled: if _emit_event is not None: @@ -9744,16 +9932,15 @@ def _handle_brc_consensus_timeout( "blocking_agents": blocking_agents, }, ) - _persist_hitl_decision( - pipeline_id, + _publish_consensus_timeout_alert( pipeline, - store, - question=( - f"Consensus not reached after {int(consensus_timeout / 60)} minutes. " - f"How to proceed?" - ), - options=["Continue waiting", "Accept current state", "Abort phase"], - phase=pipeline.current_phase, + pipeline_id, + consensus_timeout, + blocking_agents, + priority="medium", + latest_proposal_at=latest_proposal_at, + latest_heartbeat_at=latest_heartbeat_at, + slice_id=slice_id, ) @@ -11329,12 +11516,13 @@ def _update_agents_complete() -> None: # 6. Consensus timeout if elapsed >= consensus_timeout: - # #2243 progress gate: keep polling instead of opening the - # auto-HITL decision while producer/reviewer activity is - # still live on the BRC bus or in container heartbeats. - # Without this gate, decision-15 / decision-17 on - # ``issue-1557-v2`` fired minutes before the next commit - # landed; "Continue waiting" was the only correct answer. + # #2243 progress gate: keep polling instead of publishing + # the consensus-timeout alert while producer/reviewer + # activity is still live on the BRC bus or in container + # heartbeats. Without this gate, the historical decision-15 + # / decision-17 misfires on ``issue-1557-v2`` (now + # ``OVERSEER_ALERT`` post-#2264) fired minutes before the + # next commit landed. _gate_seconds = max( 0, int(getattr(pipeline.config, "brc_consensus_progress_gate_seconds", 300)), @@ -11378,6 +11566,7 @@ def _update_agents_complete() -> None: consensus.get("blocking_agents", []), store, slice_id=slice_id, + active_role_names=[e.role.value for e in active_executions], ) # Fall back: event-driven wait for remaining containers. diff --git a/orchestrator/tests/test_consensus_polling.py b/orchestrator/tests/test_consensus_polling.py index 68306244a6..0ed1af1ca6 100644 --- a/orchestrator/tests/test_consensus_polling.py +++ b/orchestrator/tests/test_consensus_polling.py @@ -293,10 +293,16 @@ class TestConsensusTimeout: @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) - def test_timeout_creates_hitl_decision( + def test_timeout_publishes_overseer_alert( self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep ): - """When consensus times out, a HITL decision is created.""" + """When consensus times out, an OVERSEER_ALERT is published (#2264). + + Replaces the pre-#2264 HITL ``choice`` decision: the platform + no longer gates the pipeline on a binary prompt at timeout — + it surfaces a non-blocking notification via the SDLC skill's + existing alert flow. + """ # Use a callable side_effect that (a) starts before the timeout, # (b) jumps past the 30-min consensus timeout, and (c) keeps # advancing so the post-timeout per-iteration budget (#1921 / @@ -329,8 +335,16 @@ def _monotonic(): mock_lock.return_value.__enter__ = MagicMock(return_value=None) mock_lock.return_value.__exit__ = MagicMock(return_value=False) + captured_alerts: list = [] + fake_msg_store = MagicMock() + fake_msg_store.add_message.side_effect = lambda msg: captured_alerts.append(msg) or msg + msg_store_factory = MagicMock(return_value=fake_msg_store) + mock_add_decision = MagicMock(return_value=MagicMock(id="dec-1")) - with patch.object(type(pipeline), "add_decision", mock_add_decision): + with ( + patch.object(type(pipeline), "add_decision", mock_add_decision), + patch("routes.pipelines._get_message_store", return_value=msg_store_factory), + ): exit_code, logs = _run_concurrent_phase( pipeline_id="issue-999", pipeline=pipeline, @@ -341,18 +355,24 @@ def _monotonic(): ) # Timeout with no convergence → force-kill path → exit 1 (#1921). - # Prior to #1921 this returned 0 because wait_for_container was - # mocked to return a clean exit; post-#1921 the polling loop - # force-kills still-running containers when the per-iteration - # post-timeout budget elapses (#2245: budget rebaselines on - # producer progress; with no tracker proposals here the - # iteration clock never resets). assert exit_code == 1 - # HITL decision should have been created on the pipeline - mock_add_decision.assert_called_once() - call_args = mock_add_decision.call_args - question = call_args[1].get("question", call_args[0][0] if call_args[0] else "") - assert "30 minutes" in question + # No HITL decision is opened on consensus timeout post-#2264. + mock_add_decision.assert_not_called() + # Exactly one consensus-timeout OVERSEER_ALERT is published. + consensus_alerts = [ + m + for m in captured_alerts + if m.message_type == "OVERSEER_ALERT" + and m.metadata.get("anomaly_type") == "consensus-timeout" + ] + assert len(consensus_alerts) == 1 + alert = consensus_alerts[0] + assert alert.metadata["consensus_timeout_minutes"] == 30 + assert alert.metadata["blocking_agents"] == ["coder"] + assert alert.metadata["priority"] == "medium" + # Subject role slot follows the SDLC-skill convention so the + # host's "Check agent logs" can extract a real role. + assert alert.subject == "consensus-timeout: coder [medium]" @patch("routes.pipelines.time.sleep") @patch("routes.pipelines.time.monotonic") diff --git a/orchestrator/tests/test_pipelines_routes.py b/orchestrator/tests/test_pipelines_routes.py index a651977438..a48a7dde27 100644 --- a/orchestrator/tests/test_pipelines_routes.py +++ b/orchestrator/tests/test_pipelines_routes.py @@ -1,12 +1,16 @@ """Regression tests for routes/pipelines.py helpers. Covers issue #1783: the BRC/consensus timeout path used bare relative -imports that crashed under k3s's top-level-module layout, and silently -swallowed add_decision failures so the stall had no visible HITL decision. +imports that crashed under k3s's top-level-module layout. -Also covers issue #2243: the BRC progress gate must defer the auto -consensus-failure HITL decision while the bus or container heartbeats -have fired within the gate window. +Covers issue #2243: the BRC progress gate must defer the auto +consensus-failure decision while the bus or container heartbeats have +fired within the gate window. + +Covers issue #2264: the consensus-timeout fallback used to open a +``choice``-typed HITL decision; it now publishes an ``OVERSEER_ALERT`` +so the SDLC skill surfaces it as a non-blocking notification rather +than a binary gate. """ import time @@ -16,7 +20,11 @@ import pytest from events import EventType from models import Pipeline, PipelinePhase -from routes.pipelines import _check_brc_progress_gate, _handle_brc_consensus_timeout +from routes.pipelines import ( + _check_brc_progress_gate, + _handle_brc_consensus_timeout, + _latest_active_role_heartbeat, +) @pytest.fixture @@ -26,32 +34,49 @@ def pipeline(): return p -def _make_store(pipeline): - """MagicMock store whose load/save round-trip the in-memory pipeline. +def _make_store(): + """MagicMock state store kept for call-site compatibility. + + Issue #2264 removed the ``_persist_hitl_decision`` call from the + timeout handler, so the store argument is unused in the new alert + flow. The fixture is retained so the test signatures don't drift + from the production call site (which still passes the state store). + """ + return MagicMock() + + +def _capture_alerts(): + """Return ``(patch_context, alerts_list)`` for capturing OVERSEER_ALERT writes. - `_persist_hitl_decision` does load → mutate → save. Returning the - same pipeline from `load_pipeline` keeps the test assertions on - `pipeline.decisions` working without inventing a separate disk-side - Pipeline (issue #2208 review asked HITL writes go through the store). + Patches ``routes.pipelines._get_message_store`` to return a factory + that returns a fake store whose ``add_message`` appends each message + to ``alerts_list``. Tests assert against the captured list. """ - store = MagicMock() - store.load_pipeline.side_effect = lambda _pid: pipeline - store.save_pipeline.side_effect = lambda _p: None - return store + alerts: list = [] + fake_store = MagicMock() + fake_store.add_message.side_effect = lambda msg: alerts.append(msg) or msg + factory = MagicMock(return_value=fake_store) + return patch("routes.pipelines._get_message_store", return_value=factory), alerts class TestHandleBrcConsensusTimeout: - """The timeout handler must reach add_decision under k3s's module layout. + """The timeout handler publishes an OVERSEER_ALERT (issue #2264). - The test runner imports routes.pipelines via absolute imports (the same - layout k3s uses), so a regression that re-introduces a bare relative - import would surface here as an ImportError before add_decision runs. + Migration from the old auto-``choice`` HITL decision to a + non-blocking ``OVERSEER_ALERT`` notification. The SDLC skill's + existing alert flow (Check agent logs / Acknowledge / Cancel + pipeline) handles operator interaction; the platform no longer + gates the pipeline on a binary choice. """ @patch("routes.pipelines._emit_event") - def test_no_brc_tracker_queues_hitl_decision(self, mock_emit, pipeline): - store = _make_store(pipeline) - with patch("peer_consensus.get_peer_consensus_tracker", return_value=None): + def test_no_brc_tracker_publishes_overseer_alert(self, mock_emit, pipeline): + store = _make_store() + capture, alerts = _capture_alerts() + with ( + patch("peer_consensus.get_peer_consensus_tracker", return_value=None), + capture, + ): _handle_brc_consensus_timeout( pipeline, pipeline.id, @@ -60,8 +85,9 @@ def test_no_brc_tracker_queues_hitl_decision(self, mock_emit, pipeline): store=store, ) - assert len(pipeline.decisions) == 1 - assert "Consensus not reached after 30 minutes" in pipeline.decisions[0].question + # No HITL decision opened — the protocol shape is now a notification. + assert len(pipeline.decisions) == 0 + # CONSENSUS_TIMEOUT audit event still fires on the no-tracker path. mock_emit.assert_called_once_with( EventType.CONSENSUS_TIMEOUT, pipeline.id, @@ -70,35 +96,128 @@ def test_no_brc_tracker_queues_hitl_decision(self, mock_emit, pipeline): "blocking_agents": ["reviewer_code"], }, ) + # OVERSEER_ALERT published with the issue #2264 metadata schema. + assert len(alerts) == 1 + alert = alerts[0] + assert alert.message_type == "OVERSEER_ALERT" + assert alert.from_role == "orchestrator" + assert alert.to_role == "all" + # Subject role slot follows the SDLC-skill convention + # ``: []`` so "Check + # agent logs" extracts a role the host can pass to + # ``get_container_logs``. With one blocking agent reported, + # that role lands in the subject. + assert alert.subject == "consensus-timeout: reviewer_code [medium]" + assert alert.metadata["anomaly_type"] == "consensus-timeout" + assert alert.metadata["consensus_timeout_minutes"] == 30 + assert alert.metadata["blocking_agents"] == ["reviewer_code"] + assert alert.metadata["priority"] == "medium" + assert alert.metadata["phase"] == PipelinePhase.REFINE.value + # No tracker / heartbeat info available in this scenario. + assert alert.metadata["latest_proposal_at"] is None + assert alert.metadata["latest_heartbeat_at"] is None + # slice_id absent from metadata when call-site passes None. + assert "slice_id" not in alert.metadata + + @patch("routes.pipelines._emit_event") + def test_brc_escalate_publishes_high_priority_alert(self, mock_emit, pipeline): + store = _make_store() + tracker = MagicMock() + # The tracker's escalate result carries ``critical_blockers`` + # — the alert must narrow to those rather than including + # advisory roles from the caller-supplied ``blocking_agents``. + tracker.handle_timeout.return_value = { + "action": "escalate", + "critical_blockers": [ + { + "reviewer_role": "reviewer_code", + "producer_role": "coder", + "state": "pending", + } + ], + } + tracker.is_timeout_handled.return_value = True + proposal_ts = datetime(2026, 4, 29, 12, 0, 0, tzinfo=UTC) + tracker.get_latest_proposal_timestamp.return_value = proposal_ts + + capture, alerts = _capture_alerts() + with ( + patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker), + capture, + ): + _handle_brc_consensus_timeout( + pipeline, + pipeline.id, + consensus_timeout=1800.0, + # Caller-supplied list includes an advisory role; the + # alert metadata must drop it on the escalate path. + blocking_agents=["reviewer_code", "tester_advisory"], + store=store, + slice_id="slice-1", + ) + + assert len(pipeline.decisions) == 0 + # No CONSENSUS_TIMEOUT event on the escalate path — the + # tracker already emitted CONSENSUS_FAILURE internally. + mock_emit.assert_not_called() + assert len(alerts) == 1 + alert = alerts[0] + # First critical-blocker role lands in the subject's role slot. + assert alert.subject == "consensus-timeout: reviewer_code [high]" + assert alert.metadata["priority"] == "high" + assert alert.metadata["latest_proposal_at"] == proposal_ts.isoformat() + assert alert.metadata["slice_id"] == "slice-1" + # Critical blockers only — advisory ``tester_advisory`` excluded. + assert alert.metadata["blocking_agents"] == ["reviewer_code", "coder"] @patch("routes.pipelines._emit_event") - def test_brc_escalate_queues_hitl_decision(self, mock_emit, pipeline): - store = _make_store(pipeline) + def test_brc_escalate_falls_back_to_caller_blocking_agents(self, mock_emit, pipeline): + # Defensive: if the tracker's escalate result omits + # ``critical_blockers`` (older return shape, or the matrix + # cleared between handle_timeout calls), fall back to the + # caller-supplied list rather than emitting an alert with no + # roles in the subject. + store = _make_store() tracker = MagicMock() tracker.handle_timeout.return_value = {"action": "escalate"} tracker.is_timeout_handled.return_value = True + tracker.get_latest_proposal_timestamp.return_value = None - with patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker): + capture, alerts = _capture_alerts() + with ( + patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker), + capture, + ): _handle_brc_consensus_timeout( pipeline, pipeline.id, consensus_timeout=1800.0, - blocking_agents=[], + blocking_agents=["reviewer_code"], store=store, ) - assert len(pipeline.decisions) == 1 - assert "BRC consensus failure" in pipeline.decisions[0].question mock_emit.assert_not_called() + assert len(alerts) == 1 + alert = alerts[0] + assert alert.subject == "consensus-timeout: reviewer_code [high]" + assert alert.metadata["blocking_agents"] == ["reviewer_code"] @patch("routes.pipelines._emit_event") - def test_brc_handled_without_escalate_no_decision(self, mock_emit, pipeline): - store = _make_store(pipeline) + def test_brc_handled_without_escalate_no_alert(self, mock_emit, pipeline): + # The advisory-only path (proceed_with_notification) was + # silent before #2264 and stays silent: the tracker has + # already emitted CONSENSUS_TIMEOUT internally and the handler + # does not re-notify the operator. + store = _make_store() tracker = MagicMock() tracker.handle_timeout.return_value = {"action": "proceed"} tracker.is_timeout_handled.return_value = True + capture, alerts = _capture_alerts() - with patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker): + with ( + patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker), + capture, + ): _handle_brc_consensus_timeout( pipeline, pipeline.id, @@ -108,18 +227,23 @@ def test_brc_handled_without_escalate_no_decision(self, mock_emit, pipeline): ) assert len(pipeline.decisions) == 0 + assert alerts == [] mock_emit.assert_not_called() @patch("routes.pipelines.logger") @patch("routes.pipelines._emit_event") - def test_tracker_import_error_falls_back_to_hitl(self, mock_emit, mock_logger, pipeline): + def test_tracker_import_error_falls_back_to_alert(self, mock_emit, mock_logger, pipeline): # Regression test for issue #1783: bare relative import raised - # ImportError under k3s, and the outer except caught it but the - # HITL decision must still be created via the fallback path. - store = _make_store(pipeline) - with patch( - "peer_consensus.get_peer_consensus_tracker", - side_effect=ImportError("simulated k3s import failure"), + # ImportError under k3s. After #2264 the fallback is an + # OVERSEER_ALERT rather than a HITL decision. + store = _make_store() + capture, alerts = _capture_alerts() + with ( + patch( + "peer_consensus.get_peer_consensus_tracker", + side_effect=ImportError("simulated k3s import failure"), + ), + capture, ): _handle_brc_consensus_timeout( pipeline, @@ -129,14 +253,15 @@ def test_tracker_import_error_falls_back_to_hitl(self, mock_emit, mock_logger, p store=store, ) - # The warning log records the import failure - mock_logger.warning.assert_called_once() - _, warn_kwargs = mock_logger.warning.call_args - assert "simulated k3s import failure" in warn_kwargs.get("error", "") - - # Fallback HITL decision is still queued - assert len(pipeline.decisions) == 1 - assert "Consensus not reached after 30 minutes" in pipeline.decisions[0].question + # The warning log records the import failure. + assert any( + "BRC timeout check failed" in (call.args[0] if call.args else "") + for call in mock_logger.warning.call_args_list + ) + # Fallback OVERSEER_ALERT is still published. + assert len(pipeline.decisions) == 0 + assert len(alerts) == 1 + assert "[medium]" in alerts[0].subject mock_emit.assert_called_once_with( EventType.CONSENSUS_TIMEOUT, pipeline.id, @@ -148,16 +273,19 @@ def test_tracker_import_error_falls_back_to_hitl(self, mock_emit, mock_logger, p @patch("routes.pipelines.logger") @patch("routes.pipelines._emit_event") - def test_add_decision_failure_is_logged_not_swallowed(self, mock_emit, mock_logger, pipeline): - # Covers the issue #1783 second-order bug: the except Exception: pass - # at the old decision-queue call sites hid stall-causing failures. - # Post-#2208 review: persistence routes through `_persist_hitl_decision`, - # which catches and logs at WARNING with exc_info so the original - # traceback survives — still not swallowed. - store = _make_store(pipeline) + def test_message_store_failure_is_logged_not_swallowed(self, mock_emit, mock_logger, pipeline): + # Issue #2264: the publish helper must log message-store add + # failures with ``exc_info`` rather than silently dropping + # them, so a wedged broker doesn't render the consensus + # timeout invisible to operators (mirrors the #1783 contract + # for the old _persist_hitl_decision path). + store = _make_store() + bad_store = MagicMock() + bad_store.add_message.side_effect = RuntimeError("broker boom") + factory = MagicMock(return_value=bad_store) with ( patch("peer_consensus.get_peer_consensus_tracker", return_value=None), - patch.object(Pipeline, "add_decision", side_effect=RuntimeError("boom")), + patch("routes.pipelines._get_message_store", return_value=factory), ): _handle_brc_consensus_timeout( pipeline, @@ -167,17 +295,18 @@ def test_add_decision_failure_is_logged_not_swallowed(self, mock_emit, mock_logg store=store, ) - # `_persist_hitl_decision` logs the failure with exc_info so the - # traceback is preserved rather than swallowed. warning_calls = [ call for call in mock_logger.warning.call_args_list - if call.args and "Failed to persist HITL decision" in call.args[0] + if call.args and "Failed to publish consensus-timeout OVERSEER_ALERT" in call.args[0] ] - assert warning_calls, "add_decision failure must be logged with exc_info" + assert warning_calls, "message store failure must be logged with exc_info" _, kwargs = warning_calls[0] assert kwargs.get("pipeline_id") == pipeline.id assert kwargs.get("exc_info") is True + # No alert was successfully captured but the audit event + # still fired on the no-tracker path. + mock_emit.assert_called_once() class TestBrcProgressGate: @@ -352,3 +481,90 @@ def test_heartbeat_failure_logged_and_not_treated_as_defer(self): "heartbeat check failed" in (call.args[0] if call.args else "") for call in mock_logger.warning.call_args_list ) + + +class TestLatestActiveRoleHeartbeat: + """Issue #2264 — direct coverage for the alert-side heartbeat helper. + + ``_latest_active_role_heartbeat`` mirrors the heartbeat half of + ``_check_brc_progress_gate`` but writes ``latest_heartbeat_at`` + into the consensus-timeout ``OVERSEER_ALERT`` metadata. The + BRC-progress-gate tests above only exercise the gate's inline + code; this class covers the helper's branches directly. + """ + + def _patch_health_monitor(self, last_heartbeats: dict[str, float] | None): + if last_heartbeats is None: + return patch("health_monitor.get_health_monitor", return_value=None) + hm = MagicMock() + hm._lock = MagicMock() + hm._lock.__enter__ = MagicMock(return_value=hm._lock) + hm._lock.__exit__ = MagicMock(return_value=False) + hm._last_heartbeat = dict(last_heartbeats) + return patch("health_monitor.get_health_monitor", return_value=hm) + + def test_empty_active_roles_returns_none(self): + # Short-circuit before any health-monitor lookup. The alert + # then carries ``latest_heartbeat_at: null``. + assert _latest_active_role_heartbeat([]) is None + + def test_recent_heartbeat_returns_datetime(self): + recent_hb = time.time() - 30 + with self._patch_health_monitor({"coder": recent_hb}): + result = _latest_active_role_heartbeat(["coder", "tester"]) + assert isinstance(result, datetime) + assert result.tzinfo is UTC + # Within 1s of the input wall-clock value. + assert abs(result.timestamp() - recent_hb) < 1 + + def test_picks_most_recent_across_active_roles(self): + older = time.time() - 120 + newer = time.time() - 10 + with self._patch_health_monitor({"coder": older, "tester": newer}): + result = _latest_active_role_heartbeat(["coder", "tester"]) + assert result is not None + assert abs(result.timestamp() - newer) < 1 + + def test_inactive_role_heartbeats_are_ignored(self): + # Cross-phase pollution: the singleton ``HealthMonitor`` may + # still carry a ``refiner`` heartbeat during a coder phase. + # The helper must filter to the active set so the alert + # doesn't surface a ghost timestamp from a finished phase. + ghost_hb = time.time() - 30 + with self._patch_health_monitor({"refiner": ghost_hb}): + result = _latest_active_role_heartbeat(["coder", "tester"]) + assert result is None + + def test_monitor_unavailable_returns_none(self): + with self._patch_health_monitor(None): + result = _latest_active_role_heartbeat(["coder"]) + assert result is None + + def test_no_heartbeats_returns_none(self): + with self._patch_health_monitor({}): + result = _latest_active_role_heartbeat(["coder"]) + assert result is None + + def test_lookup_failure_logs_with_exc_info_and_returns_none(self): + # A crashed signal collector must NOT break the alert + # publish — the helper logs at WARNING with ``exc_info=True`` + # and returns ``None`` so the alert proceeds with a null + # ``latest_heartbeat_at``. + bad_hm = MagicMock() + bad_hm._lock = MagicMock() + bad_hm._lock.__enter__ = MagicMock(side_effect=RuntimeError("hm boom")) + bad_hm._lock.__exit__ = MagicMock(return_value=False) + with ( + patch("health_monitor.get_health_monitor", return_value=bad_hm), + patch("routes.pipelines.logger") as mock_logger, + ): + result = _latest_active_role_heartbeat(["coder"]) + assert result is None + warning_calls = [ + call + for call in mock_logger.warning.call_args_list + if call.args and "heartbeat lookup failed" in call.args[0] + ] + assert warning_calls, "heartbeat-lookup failure must be logged" + _, kwargs = warning_calls[0] + assert kwargs.get("exc_info") is True