Downgrade noop-park alert when the parked role waits on a live producer

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,156 @@ 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 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
+        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. 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 (or a stale
+        self-report) can only make the alert MORE alarming, never quieter.
+        """
+        try:
+            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
+            # #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)
+            # #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
+            )
+
+            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 _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
+            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:

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
 
     # ------------------------------------------------------------------

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: "

orchestrator/supervision_policy.py

@@ -43,3 +43,15 @@
 # through the contract-decision fingerprint; it bounds the burn to ~2 pods/h
 # instead of deadlocking the arm.
 SUPERVISION_NOOP_PARK_RETRY_SECONDS = 1800
+
+# 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.