From e4b9ad26eec8574e918ca001906a6d14de915634 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Tue, 7 Jul 2026 16:04:44 -0700 Subject: [PATCH 1/4] Downgrade noop-park alert when the parked role waits on a live producer The agent-invocation-noop-streak alert (#3425) fired at priority=high for a dependent role no-oping while its upstream producer worked toward its first proposal, which is normal BRC choreography in every phase with a consumer role. Routine false positives at [high] undermine the alert feed's trustworthiness (#3364). At the park transition, when no gating contract decision (cq-N) is visible, the supervisor now consults a new waiting_probe: the parked role's latest current-phase HEARTBEAT on the bus. If it self-reports WAITING_ON_ROLE and every waited-on role shows bus activity within SUPERVISION_WAITING_ROLE_LIVE_SECONDS, the alert is downgraded to a low-priority agent-parked-waiting-on-role notice. If the waited-on role is NOT live the alert stays high and names the stalled producer (the parked role's own escalation threshold can no longer fire once its pod stops spawning). No self-report, an unknown probe result, or a visible gating cq-N keep the existing high-priority alert. Parking behavior itself is unchanged. Closes #3520 --- orchestrator/concurrent_executor.py | 70 ++++++++- orchestrator/event_loop/__init__.py | 13 ++ orchestrator/event_loop/_supervisor.py | 89 ++++++++++- orchestrator/supervision_policy.py | 11 ++ .../tests/test_concurrent_executor.py | 145 ++++++++++++++++++ orchestrator/tests/test_event_loop.py | 124 +++++++++++++++ 6 files changed, 450 insertions(+), 2 deletions(-) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index b1ae243dc7..d59a586c00 100644 --- a/orchestrator/concurrent_executor.py +++ b/orchestrator/concurrent_executor.py @@ -11,7 +11,7 @@ import threading from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING, Any @@ -540,6 +540,11 @@ def _agent_free(*, action: str, role: str, payload: Any = None) -> None: # parked while racing its upstream producer). ``getattr`` keeps # test doubles without the method on the heartbeat-only path. brc_probe=getattr(tracker, "consensus_state_fingerprint", None), + # #3520: at the park transition the parked role's latest + # WAITING_ON_ROLE heartbeat decides the alert's severity — + # waiting on a live upstream producer is choreography, not a + # wedge, so it must not fire at [high]. + waiting_probe=self._role_waiting_status, ) # #3064 slice-5: convergence-stall notifier re-uses the same @@ -1073,6 +1078,69 @@ def _unresolved_contract_decision_ids(self) -> frozenset[str] | None: ) return None + def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: + """Return ``role``'s latest WAITING_ON_ROLE self-report status (#3520). + + Wired as the :class:`JobSupervisor`'s ``waiting_probe``, consulted + once at the no-op park transition to pick the park alert's severity. + Reads the message bus for ``role``'s most recent HEARTBEAT in the + current phase; when that heartbeat self-reports + ``state=WAITING_ON_ROLE`` this returns ``(waiting_on, + waited_on_live)``, where ``waited_on_live`` is True iff every role + named in ``metadata.waiting_on`` (comma-tolerant) emitted any bus + message within ``SUPERVISION_WAITING_ROLE_LIVE_SECONDS``. + + Latest-heartbeat semantics are sound despite server-side dedup + (``routes/messages.py``): only *consecutive identical* states are + deduped, so the newest HEARTBEAT on the bus is always the role's + current self-reported state. The phase filter keeps a previous + phase's WAITING_ON_ROLE report (same pipeline stream) from being + mistaken for current evidence. + + Returns ``None`` (unknown / no self-report) when the latest + heartbeat is any other state, the role has no heartbeat this phase, + or the read fails — the supervisor then falls back to the + wedge-shaped high-priority alert, so a probe failure can only make + the alert MORE alarming, never quieter. + """ + try: + from supervision_policy import SUPERVISION_WAITING_ROLE_LIVE_SECONDS + + messages = get_message_store().get_messages( + self.pipeline.id, limit=10000, slice_id=self._slice_id + ) + phase = self.pipeline.current_phase.value + latest = None + for msg in messages: + if ( + msg.message_type == MessageType.HEARTBEAT + and msg.from_role == role + and msg.phase == phase + ): + latest = msg + if latest is None or latest.metadata.get("state") != "WAITING_ON_ROLE": + return None + waiting_on = str(latest.metadata.get("waiting_on") or "") + waited_roles = [r.strip() for r in waiting_on.split(",") if r.strip()] + if not waited_roles: + return None + cutoff = datetime.now(UTC) - timedelta(seconds=SUPERVISION_WAITING_ROLE_LIVE_SECONDS) + recent_senders = { + msg.from_role + for msg in messages + if (msg.timestamp if msg.timestamp.tzinfo else msg.timestamp.replace(tzinfo=UTC)) + >= cutoff + } + return (waiting_on, all(r in recent_senders for r in waited_roles)) + except Exception: # noqa: BLE001 — probing is best-effort + logger.warning( + "Failed to probe WAITING_ON_ROLE heartbeat for no-op park alert", + pipeline_id=self.pipeline.id, + role=role, + exc_info=True, + ) + return None + def _handle_arms_exhausted( self, *, report: list[dict[str, Any]], blocked_arms: list[tuple[str, str]] ) -> None: diff --git a/orchestrator/event_loop/__init__.py b/orchestrator/event_loop/__init__.py index f6bec5d31a..e3f4477f01 100644 --- a/orchestrator/event_loop/__init__.py +++ b/orchestrator/event_loop/__init__.py @@ -390,6 +390,7 @@ def __init__( on_exhausted: Callable[..., Any] | None = None, hitl_probe: Callable[[], Iterable[str] | None] | None = None, brc_probe: Callable[[], str | None] | None = None, + waiting_probe: Callable[[str], tuple[str, bool] | None] | None = None, ) -> None: self.clock = clock self._overseer_alert = overseer_alert @@ -410,6 +411,17 @@ def __init__( # dedupe key, so without this probe the only wake path is the retry # heartbeat. ``None`` means "unknown", same semantics as ``hitl_probe``. self._brc_probe = brc_probe + # #3520: best-effort probe consulted at the no-op park transition to + # pick the alert's severity. Called with the parking role's name; when + # that role's latest HEARTBEAT self-reports ``WAITING_ON_ROLE`` it + # returns ``(waiting_on, waited_on_live)`` — the self-reported + # waited-on role(s) and whether they show recent bus activity. + # Waiting on a LIVE upstream producer's first proposal is normal BRC + # choreography in every phase with a consumer role, so that shape + # emits a low-priority notice instead of the high-priority wedge + # alert. ``None`` means "no such self-report or unknown" — the alert + # then keeps its wedge-shaped high priority. + self._waiting_probe = waiting_probe # #3064 slice-4: fired once when a dedupe key crosses into the # exhausted set (the ``_exhausted`` transition). The orchestrator # wires this to tear down the role's reused gateway session — an @@ -505,6 +517,7 @@ def backoff_cap(self) -> int: noop_parked = _supervisor.noop_parked _probe_hitl_fingerprint = _supervisor._probe_hitl_fingerprint _probe_brc_fingerprint = _supervisor._probe_brc_fingerprint + _probe_waiting_on = _supervisor._probe_waiting_on reconcile = _supervisor.reconcile # ------------------------------------------------------------------ diff --git a/orchestrator/event_loop/_supervisor.py b/orchestrator/event_loop/_supervisor.py index 5ebc1a79fa..d7fd697708 100644 --- a/orchestrator/event_loop/_supervisor.py +++ b/orchestrator/event_loop/_supervisor.py @@ -70,7 +70,13 @@ def record_success(self, dedupe_key: str, *, action: str = "", role: str = "") - action, role, ) - self._emit_noop_alert(dedupe_key, streak, action, role, fingerprint) + # #3520: with no gating contract decision visible, the role's own + # WAITING_ON_ROLE self-report decides the alert's severity — a + # dependent role no-oping while its live upstream producer works + # toward its first proposal is choreography, not a wedge. A visible + # cq-N keeps the original wedge hypothesis, so the probe is skipped. + waiting = None if fingerprint else self._probe_waiting_on(role) + self._emit_noop_alert(dedupe_key, streak, action, role, fingerprint, waiting) def retire(self, dedupe_key: str) -> None: @@ -499,6 +505,31 @@ def _probe_hitl_fingerprint(self) -> frozenset[str] | None: return frozenset(result) +def _probe_waiting_on(self, role: str) -> tuple[str, bool] | None: + """Snapshot ``role``'s latest WAITING_ON_ROLE self-report (best-effort, #3520). + + Returns ``(waiting_on, waited_on_live)`` when the role's most recent + HEARTBEAT in the current phase self-reports ``WAITING_ON_ROLE``: + ``waiting_on`` names the waited-on role(s) as self-reported and + ``waited_on_live`` is True iff every waited-on role shows recent bus + activity. ``None`` means "no such self-report or unknown" (no probe + wired, probe failed, or the latest heartbeat is some other state) — + the caller then keeps the wedge-shaped high-priority alert. + """ + if self._waiting_probe is None: + return None + try: + return self._waiting_probe(role) + except Exception as exc: # noqa: BLE001 — probing is best-effort + logger.warning( + "JobSupervisor: waiting-on probe failed for role=%s — " + "treating self-report as unknown: %s", + role, + exc, + ) + return None + + def _probe_brc_fingerprint(self) -> str | None: """Snapshot the consensus-state fingerprint (best-effort, #3465). @@ -582,6 +613,7 @@ def _emit_noop_alert( action: str, role: str, fingerprint: frozenset[str] | None, + waiting: tuple[str, bool] | None = None, ) -> None: """Emit a named, once-per-key alert for a successful-no-op park (#3425). @@ -590,9 +622,64 @@ def _emit_noop_alert( respawn cannot resolve (typically an unresolved operator HITL ``cq-N``), so the message points the operator at the pending decision rather than at agent health. + + #3520 severity split: ``waiting`` is the parked role's latest + WAITING_ON_ROLE self-report, probed by the caller only when no gating + contract decision was visible. A role waiting on a LIVE upstream + producer's first proposal is normal BRC choreography (the park still + saves the pod spawns; the arm un-parks on BRC movement), so that shape + emits a low-priority ``agent-parked-waiting-on-role`` notice — a + routine [high] would train operators to skim past the alert feed + (#3364). High priority is kept for the genuine wedges: a visible + gating ``cq-N``, a WAITING_ON_ROLE report whose waited-on role shows + no recent bus activity (a real stall — the parked role's own + escalation threshold can no longer fire once its pod stops spawning), + or a streak with no self-report at all (silent wedge). """ if self._overseer_alert is None: return + if not fingerprint and waiting is not None: + waited_on, waited_on_live = waiting + if waited_on_live: + self._overseer_alert( + anomaly="agent-parked-waiting-on-role", + priority="low", + summary=(f"agent parked waiting on {waited_on} (action={action}, streak={streak})"), + detail=( + f"Event-pump for role={role} has had {streak} consecutive " + f"one-shot invocations on action={action} with zero BRC " + f"progress and is parked (dedupe key {dedupe_key}). Its " + f"latest HEARTBEAT self-reports WAITING_ON_ROLE on " + f"{waited_on}, which is live on the bus — this is normal " + f"dependency choreography, not a wedge. The arm un-parks " + f"as soon as the BRC state moves (e.g. {waited_on} " + f"proposes); a probe spawn is retried every " + f"{SUPERVISION_NOOP_PARK_RETRY_SECONDS}s as a backstop. " + f"No operator action needed." + ), + ) + return + self._overseer_alert( + anomaly="agent-invocation-noop-streak", + priority="high", + summary=( + f"agent parked waiting on {waited_on}, which shows no recent " + f"bus activity (action={action}, streak={streak})" + ), + detail=( + f"Event-pump for role={role} has had {streak} consecutive " + f"one-shot invocations on action={action} with zero BRC " + f"progress and is parked (dedupe key {dedupe_key}). Its " + f"latest HEARTBEAT self-reports WAITING_ON_ROLE on " + f"{waited_on}, but {waited_on} has emitted nothing on the " + f"bus recently — the waited-on producer looks stalled, and " + f"the parked role's own escalation threshold cannot fire " + f"while its pod no longer spawns. Check the {waited_on} " + f"arm's health (pod status, failure streaks) and the " + f"slice's BRC transcript." + ), + ) + return if fingerprint: gating = ( f" Unresolved contract HITL decision(s) likely gating it: " diff --git a/orchestrator/supervision_policy.py b/orchestrator/supervision_policy.py index 7046ee8977..47951be475 100644 --- a/orchestrator/supervision_policy.py +++ b/orchestrator/supervision_policy.py @@ -43,3 +43,14 @@ # through the contract-decision fingerprint; it bounds the burn to ~2 pods/h # instead of deadlocking the arm. SUPERVISION_NOOP_PARK_RETRY_SECONDS = 1800 + +# A parked role whose latest HEARTBEAT self-reports ``WAITING_ON_ROLE`` is +# normal BRC choreography (a consumer waiting for its upstream producer's +# first proposal), PROVIDED the waited-on role is live — so the park alert is +# downgraded to low priority in that shape (#3520). "Live" means the waited-on +# role emitted any bus message within this window: under the orchestrator-owned +# event loop a working producer's pod emits WORKING heartbeats (dedup-exempt, +# see routes/messages.py) plus progress/consensus traffic, so a healthy +# producer is visible well inside it. Matches the health monitor's default +# non-implement heartbeat timeout (600s). +SUPERVISION_WAITING_ROLE_LIVE_SECONDS = 600 diff --git a/orchestrator/tests/test_concurrent_executor.py b/orchestrator/tests/test_concurrent_executor.py index 90a712eef4..9d96d16847 100644 --- a/orchestrator/tests/test_concurrent_executor.py +++ b/orchestrator/tests/test_concurrent_executor.py @@ -1960,3 +1960,148 @@ def _capturing_supervisor(**kwargs): executor._start_event_loop([AgentRole.CODER], tracker=MagicMock()) assert captured.get("hitl_probe") == executor._unresolved_contract_decision_ids + + +class TestRoleWaitingStatusProbe: + """#3520: the JobSupervisor ``waiting_probe`` the executor wires in. + + Decides the no-op park alert's severity: a parked role whose latest + HEARTBEAT self-reports WAITING_ON_ROLE on a live upstream role is normal + BRC choreography and must not fire at [high]. ``None`` (no self-report / + unknown) keeps the wedge-shaped high-priority alert, so failure semantics + can only make the alert more alarming, never quieter. + """ + + def _executor(self): + from concurrent_executor import ConcurrentPhaseExecutor + + return ConcurrentPhaseExecutor(_make_pipeline(), spawn_fn=MagicMock()) + + def _probe_with(self, executor, messages): + mock_store = MagicMock() + mock_store.get_messages.return_value = messages + with patch("concurrent_executor.get_message_store", return_value=mock_store): + return executor._role_waiting_status("simplifier") + + @staticmethod + def _heartbeat(role, state, phase, *, waiting_on=None, age_seconds=0): + from datetime import UTC, datetime, timedelta + + from message_store import Message, MessageType + + metadata = {"state": state} + if waiting_on: + metadata["waiting_on"] = waiting_on + return Message( + pipeline_id="issue-999", + from_role=role, + to_role="all", + message_type=MessageType.HEARTBEAT, + subject=f"heartbeat: {state}", + metadata=metadata, + timestamp=datetime.now(UTC) - timedelta(seconds=age_seconds), + phase=phase, + ) + + def test_waiting_on_live_role(self): + """WAITING_ON_ROLE + recent bus traffic from the waited-on role. + + The self-report itself is OLD (server-side dedup suppresses repeats + of an unchanged state, so the newest WAITING_ON_ROLE entry can long + predate the park) — only the waited-on role's liveness is + recency-gated. + """ + executor = self._executor() + phase = executor.pipeline.current_phase.value + messages = [ + self._heartbeat( + "simplifier", "WAITING_ON_ROLE", phase, waiting_on="refiner", age_seconds=3600 + ), + self._heartbeat("refiner", "WORKING", phase, age_seconds=30), + ] + assert self._probe_with(executor, messages) == ("refiner", True) + + def test_waited_on_role_not_live(self): + """No recent bus activity from the waited-on role → live=False.""" + executor = self._executor() + phase = executor.pipeline.current_phase.value + messages = [ + self._heartbeat("refiner", "WORKING", phase, age_seconds=7200), + self._heartbeat( + "simplifier", "WAITING_ON_ROLE", phase, waiting_on="refiner", age_seconds=3600 + ), + ] + assert self._probe_with(executor, messages) == ("refiner", False) + + def test_comma_separated_waiting_on_requires_all_live(self): + executor = self._executor() + phase = executor.pipeline.current_phase.value + messages = [ + self._heartbeat("simplifier", "WAITING_ON_ROLE", phase, waiting_on="refiner, planner"), + self._heartbeat("refiner", "WORKING", phase, age_seconds=30), + self._heartbeat("planner", "WORKING", phase, age_seconds=7200), # dead + ] + assert self._probe_with(executor, messages) == ("refiner, planner", False) + + def test_latest_heartbeat_wins_over_older_waiting_report(self): + """A role that moved on from WAITING_ON_ROLE (state changes always + land on the bus — only consecutive identical states dedup) reports + its CURRENT state, so an older waiting entry must not match.""" + executor = self._executor() + phase = executor.pipeline.current_phase.value + messages = [ + self._heartbeat( + "simplifier", "WAITING_ON_ROLE", phase, waiting_on="refiner", age_seconds=600 + ), + self._heartbeat("simplifier", "WORKING", phase, age_seconds=60), + ] + assert self._probe_with(executor, messages) is None + + def test_previous_phase_heartbeat_is_ignored(self): + """The pipeline stream persists across phases; a refine-phase + WAITING_ON_ROLE report is not evidence about the current phase.""" + executor = self._executor() + messages = [ + self._heartbeat( + "simplifier", "WAITING_ON_ROLE", "refine", waiting_on="refiner", age_seconds=60 + ), + ] + assert self._probe_with(executor, messages) is None + + def test_no_heartbeats_is_none(self): + executor = self._executor() + assert self._probe_with(executor, []) is None + + def test_missing_waiting_on_metadata_is_none(self): + executor = self._executor() + phase = executor.pipeline.current_phase.value + messages = [self._heartbeat("simplifier", "WAITING_ON_ROLE", phase)] + assert self._probe_with(executor, messages) is None + + def test_read_failure_is_none(self): + executor = self._executor() + mock_store = MagicMock() + mock_store.get_messages.side_effect = OSError("bus down") + with patch("concurrent_executor.get_message_store", return_value=mock_store): + assert executor._role_waiting_status("simplifier") is None + + def test_wired_as_the_supervisor_waiting_probe(self): + """_start_event_loop hands the probe to the JobSupervisor (#3520).""" + import event_loop as event_loop_mod + + executor = self._executor() + captured: dict = {} + real_supervisor = event_loop_mod.JobSupervisor + + def _capturing_supervisor(**kwargs): + captured.update(kwargs) + return real_supervisor(**kwargs) + + with ( + patch.object(event_loop_mod, "JobSupervisor", _capturing_supervisor), + patch.object(event_loop_mod.OrchestratorEventLoop, "start", lambda self: None), + ): + from egg_orchestrator.types import AgentRole + + executor._start_event_loop([AgentRole.CODER], tracker=MagicMock()) + assert captured.get("waiting_probe") == executor._role_waiting_status diff --git a/orchestrator/tests/test_event_loop.py b/orchestrator/tests/test_event_loop.py index ed81b229bb..c414090877 100644 --- a/orchestrator/tests/test_event_loop.py +++ b/orchestrator/tests/test_event_loop.py @@ -1973,6 +1973,130 @@ def test_alert_fires_once_with_named_anomaly(self): supervisor.record_success("key-n", action="propose", role="coder") assert len(alerts) == 1 + def test_waiting_on_live_role_downgrades_to_low_priority(self): + """#3520: a consumer parked while its live upstream producer works + toward its first proposal is normal BRC choreography in every phase + with a dependent role — firing the routine shape at [high] trains + operators to skim past the alert feed (#3364).""" + import event_loop + + alerts: list[dict] = [] + supervisor = event_loop.JobSupervisor( + clock=_ManualClock(), + overseer_alert=lambda **kw: alerts.append(kw), + hitl_probe=lambda: set(), + waiting_probe=lambda role: ("refiner", True), + ) + self._park(supervisor, "key-n", role="simplifier") + assert len(alerts) == 1 + assert alerts[0]["anomaly"] == "agent-parked-waiting-on-role" + assert alerts[0]["priority"] == "low" + assert "refiner" in alerts[0]["detail"] + assert "No operator action needed" in alerts[0]["detail"] + # Only the alert changes — the park itself (pod-spawn saving) stays. + assert supervisor.noop_parked("key-n") + + def test_waiting_on_non_live_role_stays_high_priority(self): + """#3520: WAITING_ON_ROLE on a role with no recent bus activity is a + genuine stall — the parked role's own escalation threshold can never + fire once its pod stops spawning, so the alert must stay [high].""" + import event_loop + + alerts: list[dict] = [] + supervisor = event_loop.JobSupervisor( + clock=_ManualClock(), + overseer_alert=lambda **kw: alerts.append(kw), + hitl_probe=lambda: set(), + waiting_probe=lambda role: ("refiner", False), + ) + self._park(supervisor, "key-n", role="simplifier") + assert len(alerts) == 1 + assert alerts[0]["anomaly"] == "agent-invocation-noop-streak" + assert alerts[0]["priority"] == "high" + assert "refiner" in alerts[0]["detail"] + + def test_gating_decision_wins_over_waiting_probe(self): + """A visible unresolved cq-N keeps the original wedge alert verbatim; + the waiting probe is not consulted at all (#3520).""" + import event_loop + + alerts: list[dict] = [] + probed: list[str] = [] + + def _waiting_probe(role: str): + probed.append(role) + return ("refiner", True) + + supervisor = event_loop.JobSupervisor( + clock=_ManualClock(), + overseer_alert=lambda **kw: alerts.append(kw), + hitl_probe=lambda: {"cq-3"}, + waiting_probe=_waiting_probe, + ) + self._park(supervisor, "key-n", role="simplifier") + assert len(alerts) == 1 + assert alerts[0]["anomaly"] == "agent-invocation-noop-streak" + assert alerts[0]["priority"] == "high" + assert "cq-3" in alerts[0]["detail"] + assert probed == [] + + def test_no_waiting_self_report_keeps_high_priority(self): + """No WAITING_ON_ROLE self-report (probe returns None) is the silent + wedge — the empty-fingerprint high alert is unchanged (#3520).""" + import event_loop + + alerts: list[dict] = [] + supervisor = event_loop.JobSupervisor( + clock=_ManualClock(), + overseer_alert=lambda **kw: alerts.append(kw), + hitl_probe=lambda: set(), + waiting_probe=lambda role: None, + ) + self._park(supervisor, "key-n") + assert len(alerts) == 1 + assert alerts[0]["anomaly"] == "agent-invocation-noop-streak" + assert alerts[0]["priority"] == "high" + + def test_waiting_probe_failure_keeps_high_priority(self): + """A probe crash maps to unknown → the alert can only get MORE + alarming on failure, never quieter (#3520).""" + import event_loop + + alerts: list[dict] = [] + + def _boom(role: str): + raise RuntimeError("bus unreachable") + + supervisor = event_loop.JobSupervisor( + clock=_ManualClock(), + overseer_alert=lambda **kw: alerts.append(kw), + hitl_probe=lambda: set(), + waiting_probe=_boom, + ) + self._park(supervisor, "key-n") + assert len(alerts) == 1 + assert alerts[0]["anomaly"] == "agent-invocation-noop-streak" + assert alerts[0]["priority"] == "high" + + def test_waiting_probe_receives_parking_role(self): + """The probe is called with the role whose arm parked (#3520).""" + import event_loop + + probed: list[str] = [] + + def _waiting_probe(role: str): + probed.append(role) + return ("refiner", True) + + supervisor = event_loop.JobSupervisor( + clock=_ManualClock(), + overseer_alert=lambda **kw: None, + hitl_probe=lambda: set(), + waiting_probe=_waiting_probe, + ) + self._park(supervisor, "key-n", role="simplifier") + assert probed == ["simplifier"] + def test_park_never_engages_agent_failed(self): """The wedge is operator-bound and already alerted; creating another HITL decision via AGENT_FAILED would be noise.""" From be3e32ddd7c013b179d562394497abfc798b4dde Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Tue, 7 Jul 2026 23:19:37 +0000 Subject: [PATCH 2/4] Make waited-on liveness window phase-aware (mirror health monitor) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on #3555: the flat SUPERVISION_WAITING_ROLE_LIVE_SECONDS=600 was mis-sized for the refine/plan phases the alert most affects, and its comment wrongly claimed 600s matched the non-implement heartbeat timeout (that timeout is 120s; 600s is the *implement* timeout). Read the liveness window per-probe from the pipeline's own heartbeat-timeout config, mirroring health_monitor._get_heartbeat_threshold: 120s in refine/plan/pr, 600s in implement. This keeps the park path in lockstep with the monitor (including operator overrides), so a producer the probe calls 'live' is exactly one the monitor would not yet flag stale — the low-priority park notice can no longer contradict a fresh heartbeat_timeout alert about the same stalled producer. Removed the now-unused flat constant and its incorrect comment; added a phase-aware test asserting a 300s-old producer reads live in implement (600s window) but not live in refine (120s window). --- orchestrator/concurrent_executor.py | 24 ++++++++-- orchestrator/supervision_policy.py | 21 +++++---- .../tests/test_concurrent_executor.py | 46 +++++++++++++++++++ 3 files changed, 77 insertions(+), 14 deletions(-) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index d59a586c00..4b62703428 100644 --- a/orchestrator/concurrent_executor.py +++ b/orchestrator/concurrent_executor.py @@ -1088,7 +1088,8 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: ``state=WAITING_ON_ROLE`` this returns ``(waiting_on, waited_on_live)``, where ``waited_on_live`` is True iff every role named in ``metadata.waiting_on`` (comma-tolerant) emitted any bus - message within ``SUPERVISION_WAITING_ROLE_LIVE_SECONDS``. + message within the health monitor's phase-aware staleness window + (120s default in refine/plan/pr, 600s in implement — see below). Latest-heartbeat semantics are sound despite server-side dedup (``routes/messages.py``): only *consecutive identical* states are @@ -1104,8 +1105,6 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: the alert MORE alarming, never quieter. """ try: - from supervision_policy import SUPERVISION_WAITING_ROLE_LIVE_SECONDS - messages = get_message_store().get_messages( self.pipeline.id, limit=10000, slice_id=self._slice_id ) @@ -1124,7 +1123,24 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: waited_roles = [r.strip() for r in waiting_on.split(",") if r.strip()] if not waited_roles: return None - cutoff = datetime.now(UTC) - timedelta(seconds=SUPERVISION_WAITING_ROLE_LIVE_SECONDS) + # #3520: mirror the health monitor's phase-aware staleness + # threshold (``health_monitor._get_heartbeat_threshold``): the + # implement phase tolerates the longer implement heartbeat timeout + # (default 600s), every other phase the shorter default (120s). + # Reading the SAME ``PipelineConfig`` fields the monitor reads + # keeps the two in lockstep under operator overrides — so a + # producer this probe calls "live" is exactly one the monitor + # would not yet have flagged stale, and the low-priority park + # notice can never contradict a fresh ``heartbeat_timeout`` alert + # about the same producer. A flat 600s would have called a + # producer "live" for 5x the monitor's 120s window in refine/plan + # — the very phases this PR targets. + live_window_seconds = ( + self.pipeline.config.orchestrator_implement_heartbeat_timeout_seconds + if phase == "implement" + else self.pipeline.config.orchestrator_heartbeat_timeout_seconds + ) + cutoff = datetime.now(UTC) - timedelta(seconds=live_window_seconds) recent_senders = { msg.from_role for msg in messages diff --git a/orchestrator/supervision_policy.py b/orchestrator/supervision_policy.py index 47951be475..23224ed785 100644 --- a/orchestrator/supervision_policy.py +++ b/orchestrator/supervision_policy.py @@ -44,13 +44,14 @@ # instead of deadlocking the arm. SUPERVISION_NOOP_PARK_RETRY_SECONDS = 1800 -# A parked role whose latest HEARTBEAT self-reports ``WAITING_ON_ROLE`` is -# normal BRC choreography (a consumer waiting for its upstream producer's -# first proposal), PROVIDED the waited-on role is live — so the park alert is -# downgraded to low priority in that shape (#3520). "Live" means the waited-on -# role emitted any bus message within this window: under the orchestrator-owned -# event loop a working producer's pod emits WORKING heartbeats (dedup-exempt, -# see routes/messages.py) plus progress/consensus traffic, so a healthy -# producer is visible well inside it. Matches the health monitor's default -# non-implement heartbeat timeout (600s). -SUPERVISION_WAITING_ROLE_LIVE_SECONDS = 600 +# NOTE (#3520): the "waited-on role is live" window that downgrades the no-op +# park alert to low priority is NOT a constant here. It is read per-probe from +# the pipeline's own heartbeat-timeout config so it stays in lockstep with the +# health monitor's phase-aware staleness threshold — 120s in refine/plan/pr, +# 600s in implement (``orchestrator_heartbeat_timeout_seconds`` / +# ``orchestrator_implement_heartbeat_timeout_seconds``, mirrored by +# ``health_monitor._get_heartbeat_threshold`` and +# ``ConcurrentPhaseExecutor._role_waiting_status``). A flat constant here +# previously hard-coded 600s for every phase, which called a producer "live" +# for 5x the monitor's 120s window in the refine/plan phases this alert most +# affects; keeping the value config-derived avoids that contradiction. diff --git a/orchestrator/tests/test_concurrent_executor.py b/orchestrator/tests/test_concurrent_executor.py index 9d96d16847..a0767ddea6 100644 --- a/orchestrator/tests/test_concurrent_executor.py +++ b/orchestrator/tests/test_concurrent_executor.py @@ -2043,6 +2043,52 @@ def test_comma_separated_waiting_on_requires_all_live(self): ] assert self._probe_with(executor, messages) == ("refiner, planner", False) + def _executor_in_phase(self, phase): + from concurrent_executor import ConcurrentPhaseExecutor + + config = PipelineConfig() + config.__dict__["concurrent_execution"] = True + pipeline = Pipeline( + id="issue-999", + repo="test/repo", + issue_number=999, + status=PipelineStatus.RUNNING, + current_phase=phase, + config=config, + ) + return ConcurrentPhaseExecutor(pipeline, spawn_fn=MagicMock()) + + def test_live_window_is_phase_aware(self): + """#3520: the liveness window mirrors the health monitor's phase-aware + heartbeat timeout — 600s in implement, 120s in refine/plan — rather + than a flat 600s, so the low-priority park notice never calls a + producer "live" that the monitor would already have flagged stale. + + A waited-on producer whose newest bus message is 300s old sits INSIDE + the 600s implement window (→ live, low-priority notice) but OUTSIDE the + 120s refine window (→ not live, high-priority wedge). A flat 600s + constant would wrongly report ``live`` in refine — the very phase this + alert most affects. + """ + + def _messages(phase_value): + return [ + self._heartbeat( + "simplifier", + "WAITING_ON_ROLE", + phase_value, + waiting_on="refiner", + age_seconds=3600, + ), + self._heartbeat("refiner", "WORKING", phase_value, age_seconds=300), + ] + + implement = self._executor_in_phase(PipelinePhase.IMPLEMENT) + assert self._probe_with(implement, _messages("implement")) == ("refiner", True) + + refine = self._executor_in_phase(PipelinePhase.REFINE) + assert self._probe_with(refine, _messages("refine")) == ("refiner", False) + def test_latest_heartbeat_wins_over_older_waiting_report(self): """A role that moved on from WAITING_ON_ROLE (state changes always land on the bus — only consecutive identical states dedup) reports From 1c32380421ec05a75bc7c2178a1322596b371002 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Tue, 7 Jul 2026 23:23:55 +0000 Subject: [PATCH 3/4] Anchor waited-on liveness read to stream tip and phase-filter it Address review notes #2 and #4 on #3555. #2: the latest-heartbeat fetch starts at the stream head (XRANGE from 0-0), so on a >30k-entry stream it returns the oldest window and can miss the tip. Compute liveness from a dedicated since=cutoff read that anchors near the tip, so the waited-on role's liveness is correct and cheap regardless of stream length. The broad head-anchored read stays only for finding the role's latest heartbeat, which server-side dedup can make arbitrarily old (so it must be found outside any liveness window). #4: phase-filter the recent-senders set to match the phase filter already applied to the latest heartbeat, so a producer that emitted only in the prior phase within the window is not miscounted as live right after a phase boundary. Both trims only shrink the live set, so they stay fail-safe (toward the high-priority alert). Added a phase-filtered-liveness test. --- orchestrator/concurrent_executor.py | 32 ++++++++++++++++--- .../tests/test_concurrent_executor.py | 18 +++++++++++ 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index 4b62703428..eb1fbed55f 100644 --- a/orchestrator/concurrent_executor.py +++ b/orchestrator/concurrent_executor.py @@ -1087,9 +1087,10 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: current phase; when that heartbeat self-reports ``state=WAITING_ON_ROLE`` this returns ``(waiting_on, waited_on_live)``, where ``waited_on_live`` is True iff every role - named in ``metadata.waiting_on`` (comma-tolerant) emitted any bus - message within the health monitor's phase-aware staleness window - (120s default in refine/plan/pr, 600s in implement — see below). + named in ``metadata.waiting_on`` (comma-tolerant) emitted a bus + message IN THE CURRENT PHASE within the health monitor's phase-aware + staleness window (120s default in refine/plan/pr, 600s in implement — + see below). Latest-heartbeat semantics are sound despite server-side dedup (``routes/messages.py``): only *consecutive identical* states are @@ -1141,10 +1142,31 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: else self.pipeline.config.orchestrator_heartbeat_timeout_seconds ) cutoff = datetime.now(UTC) - timedelta(seconds=live_window_seconds) + # #3520 (review notes #2/#4): compute liveness from a dedicated + # window-bounded, phase-filtered read rather than re-scanning the + # broad latest-heartbeat fetch. ``since=cutoff`` anchors the read + # near the stream tip, so the waited-on role's liveness stays + # correct and cheap regardless of stream length — the unbounded + # ``limit`` fetch above starts at the stream HEAD and can miss the + # tip on a >30k-entry stream (note #2). The explicit ``>= cutoff`` + # filter is kept for precise sub-millisecond bounding on top of the + # ``since`` stream-ID resolution. Phase-filtering (note #4) mirrors + # the latest-heartbeat filter so a producer that emitted only in + # the PRIOR phase, still inside the window, is not miscounted as + # live right after a phase boundary. Both trims only ever SHRINK + # the live set → fail-safe (toward the high-priority alert), never + # a false low-priority notice. The latest-heartbeat read above + # stays broad on purpose: server-side dedup can make the role's + # current WAITING_ON_ROLE self-report arbitrarily old, so it must + # be found outside any liveness window. + recent = get_message_store().get_messages( + self.pipeline.id, since=cutoff, limit=10000, slice_id=self._slice_id + ) recent_senders = { msg.from_role - for msg in messages - if (msg.timestamp if msg.timestamp.tzinfo else msg.timestamp.replace(tzinfo=UTC)) + for msg in recent + if msg.phase == phase + and (msg.timestamp if msg.timestamp.tzinfo else msg.timestamp.replace(tzinfo=UTC)) >= cutoff } return (waiting_on, all(r in recent_senders for r in waited_roles)) diff --git a/orchestrator/tests/test_concurrent_executor.py b/orchestrator/tests/test_concurrent_executor.py index a0767ddea6..d80d382da5 100644 --- a/orchestrator/tests/test_concurrent_executor.py +++ b/orchestrator/tests/test_concurrent_executor.py @@ -2089,6 +2089,24 @@ def _messages(phase_value): refine = self._executor_in_phase(PipelinePhase.REFINE) assert self._probe_with(refine, _messages("refine")) == ("refiner", False) + def test_waited_on_liveness_is_phase_filtered(self): + """#3520 (review note #4): liveness counts only current-phase traffic. + + A waited-on producer whose sole in-window message landed in the PRIOR + phase must not read as "live" — otherwise, right after a phase + boundary, a role that went quiet in the new phase would be mistaken + for a working upstream and downgrade the alert. + """ + executor = self._executor() # implement phase + messages = [ + self._heartbeat( + "simplifier", "WAITING_ON_ROLE", "implement", waiting_on="refiner", age_seconds=60 + ), + # refiner's only recent message is from the prior (plan) phase. + self._heartbeat("refiner", "WORKING", "plan", age_seconds=30), + ] + assert self._probe_with(executor, messages) == ("refiner", False) + def test_latest_heartbeat_wins_over_older_waiting_report(self): """A role that moved on from WAITING_ON_ROLE (state changes always land on the bus — only consecutive identical states dedup) reports From ac28194c98c9876a3353a5f4ea06747425c321cb Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Tue, 7 Jul 2026 23:37:58 +0000 Subject: [PATCH 4/4] Guard latest-heartbeat tip-miss so the noop-park probe degrades to high MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The broad latest-heartbeat read is HEAD-anchored, so on a >30k-entry single-phase stream it can stop before the tip and resolve to a stale WAITING_ON_ROLE self-report the role has since moved off of — yielding a false low-priority 'healthy wait' notice for a genuine wedge. Reuse the tip-anchored liveness read to detect supersession: if the parked role's own latest in-window heartbeat is not WAITING_ON_ROLE, treat the self-report as stale and return None (-> high alert). The healthy-wait case (a dedup-aged self-report older than the window) is preserved because the role then has no in-window heartbeat and the guard is inert. --- orchestrator/concurrent_executor.py | 65 ++++++++++++++++--- .../tests/test_concurrent_executor.py | 34 ++++++++++ 2 files changed, 91 insertions(+), 8 deletions(-) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index eb1fbed55f..e64a0586a0 100644 --- a/orchestrator/concurrent_executor.py +++ b/orchestrator/concurrent_executor.py @@ -1097,13 +1097,20 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: deduped, so the newest HEARTBEAT on the bus is always the role's current self-reported state. The phase filter keeps a previous phase's WAITING_ON_ROLE report (same pipeline stream) from being - mistaken for current evidence. + mistaken for current evidence. The broad latest-heartbeat read is + HEAD-anchored so a dedup-aged (arbitrarily old but still current) + WAITING_ON_ROLE self-report is always found; its residual tip-miss on + a >30k-entry single-phase stream is closed by a supersession guard — + if the tip-anchored liveness read shows the parked role's own latest + in-window heartbeat is NOT WAITING_ON_ROLE, the self-report is treated + as stale and this returns ``None`` (→ high alert). Returns ``None`` (unknown / no self-report) when the latest heartbeat is any other state, the role has no heartbeat this phase, + the self-report has been superseded by a newer in-window heartbeat, or the read fails — the supervisor then falls back to the - wedge-shaped high-priority alert, so a probe failure can only make - the alert MORE alarming, never quieter. + wedge-shaped high-priority alert, so a probe failure (or a stale + self-report) can only make the alert MORE alarming, never quieter. """ try: messages = get_message_store().get_messages( @@ -1162,12 +1169,54 @@ def _role_waiting_status(self, role: str) -> tuple[str, bool] | None: recent = get_message_store().get_messages( self.pipeline.id, since=cutoff, limit=10000, slice_id=self._slice_id ) + + def _at_or_after_cutoff(msg: Message) -> bool: + ts = msg.timestamp if msg.timestamp.tzinfo else msg.timestamp.replace(tzinfo=UTC) + return ts >= cutoff + + # #3520 (re-review note): guard the broad latest-heartbeat read's + # residual tip-miss. That read is HEAD-anchored (``limit`` from + # ``0-0``), so on a >30k-entry SINGLE-phase stream it can stop + # before the tip and resolve ``latest`` to a stale WAITING_ON_ROLE + # the role has since moved off of (e.g. dedup-exempt WORKING beats + # at the tip while its arm is genuinely wedged). Left unguarded that + # yields a low-priority "healthy wait" notice for a real wedge — the + # one direction this probe must never take. The tip-anchored + # ``recent`` read (``since=cutoff``) DOES see the tip, so if the + # PARKED role's own latest in-window, current-phase heartbeat is not + # WAITING_ON_ROLE, the self-report has been superseded → return + # None → the high-priority wedge alert. When the role emitted + # nothing in the window (its WAITING_ON_ROLE is deduped-old but + # still current) the guard is inert and the downgrade still fires, + # so the healthy-wait case (a self-report arbitrarily older than the + # window) is preserved. This makes the "always degrades to high" + # characterization hold in both directions. + role_tip_state: str | None = None + for msg in recent: + if ( + msg.message_type == MessageType.HEARTBEAT + and msg.from_role == role + and msg.phase == phase + and _at_or_after_cutoff(msg) + ): + role_tip_state = msg.metadata.get("state") + if role_tip_state is not None and role_tip_state != "WAITING_ON_ROLE": + return None + + # #3520 (review notes #2/#4): compute liveness from the same + # dedicated window-bounded, phase-filtered read rather than + # re-scanning the broad latest-heartbeat fetch. ``since=cutoff`` + # anchors the read near the stream tip, so the waited-on role's + # liveness stays correct and cheap regardless of stream length. The + # explicit ``>= cutoff`` filter is kept for precise sub-millisecond + # bounding on top of the ``since`` stream-ID resolution. + # Phase-filtering (note #4) mirrors the latest-heartbeat filter so a + # producer that emitted only in the PRIOR phase, still inside the + # window, is not miscounted as live right after a phase boundary. + # Both trims only ever SHRINK the live set → fail-safe (toward the + # high-priority alert), never a false low-priority notice. recent_senders = { - msg.from_role - for msg in recent - if msg.phase == phase - and (msg.timestamp if msg.timestamp.tzinfo else msg.timestamp.replace(tzinfo=UTC)) - >= cutoff + msg.from_role for msg in recent if msg.phase == phase and _at_or_after_cutoff(msg) } return (waiting_on, all(r in recent_senders for r in waited_roles)) except Exception: # noqa: BLE001 — probing is best-effort diff --git a/orchestrator/tests/test_concurrent_executor.py b/orchestrator/tests/test_concurrent_executor.py index d80d382da5..bc6474ad1c 100644 --- a/orchestrator/tests/test_concurrent_executor.py +++ b/orchestrator/tests/test_concurrent_executor.py @@ -2121,6 +2121,40 @@ def test_latest_heartbeat_wins_over_older_waiting_report(self): ] assert self._probe_with(executor, messages) is None + def test_stale_waiting_report_superseded_by_tip_heartbeat(self): + """#3520 (re-review note): the broad latest-heartbeat read is + HEAD-anchored, so on a >30k-entry single-phase stream it can stop + before the tip and resolve ``latest`` to a WAITING_ON_ROLE the role + has since moved off of. The tip-anchored liveness read (``since``) + does see the tip; when the parked role's own latest in-window + heartbeat is not WAITING_ON_ROLE the self-report is stale, so the + probe returns ``None`` (→ high alert), never a false low-priority + "healthy wait" notice for a genuine wedge. + + The two reads are mocked separately: the broad read returns only the + aged WAITING_ON_ROLE (tip missed), the ``since`` read returns the + role's superseding WORKING beat at the tip. + """ + executor = self._executor() + phase = executor.pipeline.current_phase.value + broad = [ + self._heartbeat( + "simplifier", "WAITING_ON_ROLE", phase, waiting_on="refiner", age_seconds=3600 + ), + ] + tip = [ + # simplifier's genuine current state at the tip — its arm is + # wedged, but it is no longer self-reporting WAITING_ON_ROLE. + self._heartbeat("simplifier", "WORKING", phase, age_seconds=10), + self._heartbeat("refiner", "WORKING", phase, age_seconds=10), + ] + mock_store = MagicMock() + # First call: broad HEAD-anchored latest-heartbeat read (no ``since``). + # Second call: tip-anchored liveness read (``since=cutoff``). + mock_store.get_messages.side_effect = [broad, tip] + with patch("concurrent_executor.get_message_store", return_value=mock_store): + assert executor._role_waiting_status("simplifier") is None + def test_previous_phase_heartbeat_is_ignored(self): """The pipeline stream persists across phases; a refine-phase WAITING_ON_ROLE report is not evidence about the current phase."""