Fix #2079: wake stuck producers via brc_confirmation_timeout escalation

[jwbron]

Summary

The Tier-1 detector in check_brc_progress (orchestrator/health_monitor.py:731) already tracks fully-ACKed-but-not-confirmed producers and fires after 180s, but its escalation only added an entry to _active_alerts — _escalation_callbacks was never registered in production. The only consumer of _active_alerts is the overseer agent's discretionary poll of /api/v1/pipelines/<id>/health/alerts. On pipeline issue-1965, the overseer chose not to act on the alert, and documenter sat fully-ACKed-but-not-confirmed for ~37 minutes until the generic agent-heartbeat-stall alert finally broke the deadlock.

This PR turns the detector's output into a deterministic remediation:

• Wires an escalation callback in _run_pipeline after init_health_monitor that posts a directed OVERSEER_ALERT to the stuck producer. Body explains the state and the recovery path (call mcp__brc__confirm; if it returns pending_acks, wait on the prerequisite events instead of CONSENSUS_CONFIRMED).
• Uses OVERSEER_ALERT because it's the only message type in the producer's post-confirm wait_loop filter (CONSENSUS_CONFIRMED,CONSENSUS_RE_REVIEW,OVERSEER_ALERT) — a STATUS or NUDGE message would not wake it.
• Adds alert_type + elapsed_seconds to the escalation dict so callbacks can discriminate without parsing the reason string.
• Adds a per-iteration INFO breadcrumb in check_brc_progress so future post-mortems can verify the check ran and what it observed (suggested investigation Phases 1-2: Repository setup, docs, and gateway extraction (partial) #1 in the issue).
• Logs WARNING instead of silent skip when a fully-acked producer past timeout has no agent_state (suggested investigation Phase 1: Repository setup and CI infrastructure #2). The branch is unexpected — every producer has at minimum proposed, which routes through MESSAGE_SENT and registers state — but worth surfacing.
• Adds HealthMonitor.get_current_phase() so the callback records the current phase on the message without reaching into private state.

Independent of PR #2077 (server-side wait_loop guard for #2064 — a different layer of fix at the wait endpoint).

Files

• orchestrator/health_monitor.py — alert_type/elapsed_seconds in escalation dict, breadcrumb + skip logs, get_current_phase() accessor
• orchestrator/routes/pipelines.py — _send_brc_confirmation_nudge helper + escalation callback wiring in _run_pipeline
• orchestrator/tests/test_health_monitor.py — escalation dict shape, breadcrumb log, no-agent-state warning, get_current_phase round-trip
• orchestrator/tests/test_brc_confirmation_nudge.py — 7 unit tests for _send_brc_confirmation_nudge covering valid escalation, alert_type filter, missing fields, message-store unavailable, send error, phase=None

Test plan

• pytest orchestrator/tests/test_health_monitor.py orchestrator/tests/test_brc_confirmation_nudge.py — 97/97 pass
• pytest orchestrator/tests/test_pipelines_routes.py orchestrator/tests/test_pipelines_api.py orchestrator/tests/test_overseer_spawn.py — 81/81 pass (no regressions in touched code paths)
• ruff check clean
• Live BRC pipeline run that wedges a producer post-ACK to confirm the new OVERSEER_ALERT actually wakes the wait_loop end-to-end (CI / next pipeline)

🤖 Generated with Claude Code

[egg-reviewer]

Verdict: Approve — no blocking issues

I traced the end-to-end path:

• check_brc_progress already includes alert_type and elapsed_seconds in the escalation dict (verified at orchestrator/health_monitor.py:843-851), and only check_brc_progress populates alert_type, so the callback's filter (escalation.get("alert_type") != "brc_confirmation_timeout") correctly ignores escalations from _check_infra_errors, _on_container_stopped, check_heartbeats, etc., which all build escalation dicts without that key.
• Callback is registered before poll_thread starts (orchestrator/routes/pipelines.py:11543-11605), and check_tripwires() (line 11567) → check_brc_progress() (line 886 in health_monitor.py) is the live path that actually fires it.
• Dedup is correct: brc_progress_escalated is set inside the lock before the callback fires; the callback fires outside the lock; the flag is reset only when the producer leaves the fully-acked set.
• Producer wakeup is sound: the producer's wait_loop filter — both pre-confirm [CONSENSUS_ACK, CONSENSUS_NACK, CONSENSUS_RE_REVIEW, OVERSEER_ALERT] and post-confirm [CONSENSUS_CONFIRMED, CONSENSUS_RE_REVIEW, OVERSEER_ALERT] — includes OVERSEER_ALERT, so a directed alert wakes the wait via the per-pipeline cv (in-memory) or XADD (Redis).
• Recommending mcp__brc__confirm is safe under a benign race (already-confirmed): tests/test_consensus_confirmed_idempotent.py confirms the signal is idempotent.
• Phase is read at fire time, not registration time, so the message records the phase the producer is actually in (the closure correctly closes over health_monitor_instance, not over the local pipeline.current_phase.value).
• Teardown ordering in _run_pipeline (health_monitor_timer.set() → poll_thread.join → health_monitor_instance.stop()) means the closure is still valid when the polling thread fires its last callback.
• 23 new + touched tests pass locally.

Non-blocking suggestions

1. Docstring framing — "post-confirm wait_loop filter" is misleading

orchestrator/routes/pipelines.py:523-525:

Uses OVERSEER_ALERT (not STATUS or NUDGE) because the producer's post-confirm wait_loop filter is CONSENSUS_CONFIRMED,CONSENSUS_RE_REVIEW,OVERSEER_ALERT…

The wedged producer is in the fully_acked but not confirmed set — i.e., pre-confirm. They are most likely blocked on the pre-confirm wait whose filter is CONSENSUS_ACK,CONSENSUS_NACK,CONSENSUS_RE_REVIEW,OVERSEER_ALERT, not the post-confirm one quoted. The wakeup logic is still correct (both filters include OVERSEER_ALERT), but the cited filter is wrong for the actual wedge case. The same wording is repeated in the test docstring at orchestrator/tests/test_brc_confirmation_nudge.py:89-90.

Suggested rephrase: "OVERSEER_ALERT is the only message type that appears in both the producer's pre-confirm and post-confirm wait_loop filters, so it will wake the producer regardless of which wait they are blocked on."

2. Redundant try: from message_store import … inside the function

orchestrator/routes/pipelines.py:546-549:

try:
    from message_store import Message, MessageType
except ImportError:
    return False

This runs after _get_message_store() (which itself imports from message_store). If the package is missing, _get_message_store() already returned None and we exited above. The defensive import only adds a per-call import cost without buying any new safety.

3. elapsed_seconds default of 0 is unreachable but renders nonsensical text if it ever fires

orchestrator/routes/pipelines.py:540:

elapsed = escalation.get("elapsed_seconds", 0)

check_brc_progress always sets elapsed_seconds, so the default is unreachable today. But if a future tripwire ever emits alert_type="brc_confirmation_timeout" without elapsed_seconds, the body would render "have not confirmed in 0s" — confusing and sets up agents to mistrust the alert. Consider if elapsed is None or elapsed <= 0: return False (treat as malformed) so the contract on the escalation dict is enforced explicitly.

4. MESSAGE_SENT event is not emitted for this path

The route POST /messages/send emits EventType.MESSAGE_SENT; the direct msg_store.add_message(...) call in this PR does not. Today the only consumer is HealthMonitor._on_message_sent (rate-limit + HEARTBEAT tracking), and skipping that for an orchestrator-internal nudge is correct behavior. But the asymmetry isn't documented in code — a future audit/observability subscriber could be silently missing this traffic. Worth a one-line comment on add_message explaining why this path bypasses the event.

5. Integration test for the wiring in _run_pipeline

The unit tests mock _get_message_store and exercise _send_brc_confirmation_nudge in isolation. The actual closure (_on_health_escalation) capture and the registration ordering against init_health_monitor are not asserted — a refactor that, e.g., reorders init_health_monitor and set_current_phase could leave get_current_phase() returning None for the first escalation without any test catching it. The PR description already flags a live-pipeline verification as TODO; an in-process test that drives monitor.check_brc_progress() from a mocked _run_pipeline would close that gap before the live run.

6. Body wording

global zero-proposal guard is a term of art that the producer agent prompts may or may not recognize. If the recovery path matters, consider linking to (or quoting) the specific check in the body, or simplifying to "if confirm returns pending_acks, treat the response's blocking field as the wait filter."

— Authored by egg

[egg-reviewer]

Verdict: Approve — no new blocking issues

All six non-blocking items from the prior review were addressed in d72f99a. Tests pass (11/11 in orchestrator/tests/test_brc_confirmation_nudge.py). Verified end-to-end: the new TestEscalationCallbackWiring integration test exercises monitor.check_brc_progress() through the actual on_escalation callback registry and asserts the closure reads get_current_phase() at fire time (not registration time), and that non-BRC escalations are rejected by the alert_type filter.

Items addressed

1. ✅ Docstring framing (orchestrator/routes/pipelines.py:523-534) — now correctly explains OVERSEER_ALERT is the only type in both the pre-confirm and post-confirm wait_loop filters, with the wedge case (pre-confirm) called out explicitly. Test docstring updated to match.
2. ✅ Redundant try/except removed (line 560) — bare from message_store import … with the comment explaining why _get_message_store() already verified importability.
3. ✅ elapsed_seconds validation (lines 546-551) — None or <= 0 returns False with a comment explaining the contract. Two new unit tests (test_missing_elapsed_seconds_rejected, test_zero_or_negative_elapsed_seconds_rejected) cover the rejection paths.
4. ✅ MESSAGE_SENT bypass documented (lines 573-577) — comment explains the intentional skip of the POST /messages/send route to avoid HealthMonitor._on_message_sent rate-limit/HEARTBEAT tracking on this orchestrator-internal nudge.
5. ⚠️ Body wording rewrite — see issue below.
6. ✅ Integration test — test_callback_records_current_phase_at_fire_time advances the phase between on_escalation() registration and the firing of check_brc_progress() and asserts the message records the post-registration phase, exactly the regression class flagged in the prior review.

Non-blocking issues

1. Body refers to a non-existent blocking field

orchestrator/routes/pipelines.py:562-569:

body = (
    f"You are PROPOSED and fully ACKed but have not confirmed in "
    f"{elapsed}s. Call `mcp__brc__confirm` now. If it returns "
    "status='pending_acks', the response's ``blocking`` field lists "
    "the prerequisite events to wait on (CONSENSUS_PROPOSE from "
    "missing producers, CONSENSUS_ACK from your reviewers, or "
    "CONSENSUS_RE_REVIEW) — wait on those instead, then retry confirm."
)

The actual mcp__brc__confirm response on pending_acks (verified at sandbox/egg_agent_tools/handlers/brc.py:223-252 and the orchestrator-side guard returns at orchestrator/peer_consensus.py:500-597) carries:

• ok: True
• status: "pending_acks"
• consensus_reached: bool
• message: str — the guard reason
• signal: <full envelope>

Inside signal.data you find one of zero_proposal_producers, stale_acks, unresolved_nacks, stale_nacks depending on which guard tripped — but there is no blocking field in the response. My prior suggestion #6 ("treat the response's blocking field as the wait filter") was based on a flawed assumption that I should have verified before suggesting it; the literal substitution into the body has now landed an incorrect instruction in agent-facing text.

A producer reading this body will look for blocking, not find it, and have to fall back to parsing message. Functionally the wakeup still works (the OVERSEER_ALERT is the load-bearing piece), so this is non-blocking, but worth fixing.

Suggested rewording (keeps the concrete event-type list, drops the false field reference):

"If it returns status='pending_acks', read message for the guard reason and wait on the prerequisite events instead: CONSENSUS_PROPOSE if a producer hasn't proposed (zero_proposal_producers), CONSENSUS_ACK / CONSENSUS_RE_REVIEW if a reviewer's ACK is stale or unresolved. Then retry confirm."

— Authored by egg

[egg-reviewer]

Verdict: Approve — no blocking issues

Re-review of eaaa64f on top of the prior approve-with-suggestions at d72f99a.

Delta

Five-line edit to the nudge body in orchestrator/routes/pipelines.py:562-570. Replaces the false blocking field reference with concrete guidance: read message for the guard reason, then wait on the appropriate event type.

Verification

The new wording is factually correct against the actual mcp__brc__confirm response. Traced peer_consensus.handle_confirmed (lines 490-597): every pending_acks return path includes status + message, and depending on which guard tripped, one of:

• zero_proposal_producers (global zero-proposal guard at L503-514, or zero-proposal at L534-545)
• stale_acks (L547-558)
• unresolved_nacks (L560-571)
• stale_nacks (L573-584)

There is no blocking field anywhere in the response — confirmed by grep across peer_consensus.py and sandbox/egg_agent_tools/handlers/brc.py. The prior body's "the response's blocking field" instruction would have sent producers chasing a key that doesn't exist; this fix removes that.

The replacement instruction ("CONSENSUS_PROPOSE if a producer hasn't proposed (zero_proposal_producers), CONSENSUS_ACK / CONSENSUS_RE_REVIEW if a reviewer's ACK is stale or unresolved") covers the four guard variants reasonably — zero_proposal_producers and global_zero_proposal both fall under "producer hasn't proposed," and stale_acks / unresolved_nacks / stale_nacks all resolve when the producer addresses the issue and the reviewer re-reviews (CONSENSUS_RE_REVIEW).

Tests

11/11 pass in orchestrator/tests/test_brc_confirmation_nudge.py. Existing body assertions ("mcp__brc__confirm" in message.body, "240s" in message.body) remain valid; no new assertion was needed because the wording change is internal to the human-readable body, not the contract.

No remaining issues

No stale blocking references in pipelines.py or the test file (the other matches are unrelated: "non-blocking suggestion," "blocking_agents" consensus state, etc.). All six items from the original review are resolved as of d72f99a; the sole carry-over from the second review is now closed.

— Authored by egg

[james-in-a-box]

egg review completed. View run logs

6 previous review(s) hidden.