Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions docs/guides/concurrent-execution.md
Original file line number Diff line number Diff line change
Expand Up @@ -721,13 +721,13 @@ If any agent is in the `OBJECTING` readiness state (separate from BRC phase), th

If consensus is not reached within `consensus_timeout_minutes`, 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: <phase> [high]`) so the SDLC skill surfaces it as a non-blocking notification — see [issue #2264](https://github.com/jwbron/egg/issues/2264).
- **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: <phase> [medium]`) 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 the post-timeout budget — the alert is informational, not a gate; the operator can intervene with `cancel_task` / `restart_phase` / `provide_input` 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.

Expand Down
8 changes: 6 additions & 2 deletions docs/guides/sdlc-pipeline.md
Original file line number Diff line number Diff line change
Expand Up @@ -1443,8 +1443,12 @@ 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: <phase> [<priority>]`)
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` / `restart_phase` /
`provide_input` 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.
Expand Down
204 changes: 184 additions & 20 deletions orchestrator/routes/pipelines.py
Original file line number Diff line number Diff line change
Expand Up @@ -9679,20 +9679,173 @@ 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),
)
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 = f"consensus-timeout: {phase_value} [{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. If you want to "
"intervene, use `cancel_task` to stop the pipeline, `restart_phase` "
"to retry, or `provide_input` against any open agent decision."
)
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
Expand All @@ -9713,26 +9866,37 @@ 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),
)
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,
_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,
blocking_agents,
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:
Expand All @@ -9744,16 +9908,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,
)


Expand Down Expand Up @@ -11356,6 +11519,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.
Expand Down
43 changes: 31 additions & 12 deletions orchestrator/tests/test_consensus_polling.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 polling budget (#1921) also
Expand Down Expand Up @@ -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,
Expand All @@ -341,16 +355,21 @@ 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 3600s budget
# elapses, which is the realistic outcome.
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"

@patch("routes.pipelines.time.sleep")
@patch("routes.pipelines.time.monotonic")
Expand Down
Loading
Loading