diff --git a/docs/reference/agent-wait-patterns.md b/docs/reference/agent-wait-patterns.md index abcda4367c..fe8badc037 100644 --- a/docs/reference/agent-wait-patterns.md +++ b/docs/reference/agent-wait-patterns.md @@ -159,6 +159,46 @@ respondent. Messages like this are bus noise. - `HANDOFF` — "I need you to act on this artifact" - `STATUS` / `PROGRESS` — informational, no reply expected +### Anti-pattern 5 — Producer waits on `CONSENSUS_CONFIRMED` before its own confirm has succeeded (#2064) + +```bash +# ❌ DO NOT DO THIS — happens when a producer treats the post-confirm +# STAY ALIVE wait_loop as the recovery path for a `pending_acks` confirm. +egg-orch consensus propose ... +egg-orch consensus confirmed # returns status='pending_acks' + # because another producer + # hasn't proposed yet +egg-orch message wait-loop \ + --for CONSENSUS_CONFIRMED \ # ← circular: own confirm + --for CONSENSUS_RE_REVIEW \ # is part of what generates + --for OVERSEER_ALERT --timeout 60 # this signal globally +``` + +**Why it's wrong:** the global `CONSENSUS_CONFIRMED` signal only fires +when **every** agent — including this producer — has confirmed. Waiting +on it before the producer's own confirm has been accepted by the +tracker is a self-deadlock. Observed in pipeline `issue-1965`: the +documenter sat in this wait for ~36 minutes, woken only by the +overseer's `agent-heartbeat-stall` band-aid. + +The orchestrator's `/messages/wait` endpoint now rejects this pattern +with **HTTP 400** when the caller's role is in producer state +`WORKING` or `PROPOSED` and the wait includes `CONSENSUS_CONFIRMED` — +the wrapper surfaces this as exit code 3 (permanent error). Read the +error: it tells you what to wait for instead. + +**Fix:** the post-confirm STAY ALIVE wait is only legitimate **after** +your own confirm has succeeded (status `confirmed`, not `pending_acks`). +For a `pending_acks` recovery loop: + +- **Global zero-proposal** (another producer hasn't proposed): wait on + `CONSENSUS_PROPOSE` (and `OVERSEER_ALERT`), then re-issue + `egg-orch consensus confirmed`. +- **Your reviewers haven't ACKed yet**: wait on `CONSENSUS_ACK,CONSENSUS_NACK` + per the producer-lifecycle Step 4 idiom, then re-issue confirm when + the orchestrator's directed STATUS nudge ("ready to confirm") + arrives. + ## 3. Exit-Code Contract for `egg-orch message wait` `egg-orch message wait` returns a deterministic exit code so the wrapper diff --git a/orchestrator/peer_consensus.py b/orchestrator/peer_consensus.py index 5ea3e8d15e..697bf6e047 100644 --- a/orchestrator/peer_consensus.py +++ b/orchestrator/peer_consensus.py @@ -1245,6 +1245,21 @@ def get_state(self) -> dict[str, Any]: """Alias for evaluate() -- compatibility with ConsensusEvaluator.""" return self.evaluate() + def is_producer_pending_confirm(self, role: str) -> bool: + """True if ``role`` is a producer that has not yet reached CONFIRMED. + + Used by the ``/messages/wait`` endpoint to reject incoherent + ``wait_loop --for CONSENSUS_CONFIRMED`` calls from producers + whose own confirm hasn't succeeded — their confirm is part of + what generates global consensus, so the wait would deadlock + (#2064). Reviewer-only roles return False (they may legitimately + wait on other agents' confirms). + """ + with self._lock: + if not self.graph.is_producer(role): + return False + return self._producer_phases.get(role) != ConsensusPhase.CONFIRMED + def are_all_producers_working(self, reviewer: str) -> bool: """Check if all upstream producers for a reviewer are still in WORKING phase. diff --git a/orchestrator/routes/messages.py b/orchestrator/routes/messages.py index 221f9d5cda..cdfd9e59f8 100644 --- a/orchestrator/routes/messages.py +++ b/orchestrator/routes/messages.py @@ -353,6 +353,69 @@ def _apply_delphi_filter( return filtered_messages +# Message types that are produced *as a side effect* of a producer's +# own confirm reaching global consensus. A producer in WORKING/PROPOSED +# that waits on these would be waiting on itself — its own confirm is +# part of what generates the signal — and would deadlock until the +# overseer's stall detector intervened (#2064). +_PRODUCER_PENDING_CONFIRM_REJECTED_FOR_TYPES: frozenset[str] = frozenset({"CONSENSUS_CONFIRMED"}) + + +def _check_producer_pending_confirm_guard( + pipeline_id: str, + role: str | None, + wait_for_types: list[str], +) -> tuple[Response, int] | None: + """Reject ``wait`` calls where a non-confirmed producer waits on + ``CONSENSUS_CONFIRMED``. + + A producer's own ``mcp__brc__confirm`` is part of what generates + the global ``CONSENSUS_CONFIRMED`` signal, so a producer in + ``WORKING`` or ``PROPOSED`` that blocks on it would wait on + itself (#2064). Rather than letting the overseer's heartbeat-stall + detector bail the pipeline out minutes later, we surface the bug + immediately with an actionable error. + + The guard intentionally ignores the route's ``from`` filter — even + a wait narrowly scoped to a peer's per-agent ``CONSENSUS_CONFIRMED`` + is still part of a chain that requires this producer's own confirm + to fire first. No documented producer pattern waits this way while + in ``WORKING``/``PROPOSED``, so the over-rejection is harmless; any + future cross-producer sync that wants to bypass it should update + both this guard and the wait_loop client contract. + + Returns ``None`` when the wait should proceed; otherwise an error + response tuple ready to return from the route. + """ + if not role: + return None + blocking = _PRODUCER_PENDING_CONFIRM_REJECTED_FOR_TYPES.intersection(wait_for_types) + if not blocking: + return None + try: + from peer_consensus import get_peer_consensus_tracker + except ImportError: + get_peer_consensus_tracker = None # type: ignore[assignment] + if not get_peer_consensus_tracker: + return None + tracker = get_peer_consensus_tracker(pipeline_id) + if tracker is None: + return None + if not tracker.is_producer_pending_confirm(role): + return None + sorted_blocking = sorted(blocking) + return _make_error( + f"Producer '{role}' cannot wait on {sorted_blocking} before its own " + "consensus_confirmed has succeeded — its own confirm is part of " + "what generates that signal, so the wait would deadlock (#2064). " + "Call mcp__brc__confirm first; if it returns status='pending_acks' " + "(e.g. another producer hasn't proposed yet, or your reviewers " + "haven't ACKed), wait on the prerequisite events instead — " + "CONSENSUS_PROPOSE from missing producers, CONSENSUS_ACK from " + "your reviewers, or CONSENSUS_RE_REVIEW — then retry confirm." + ) + + @messages_bp.route("//messages/wait", methods=["GET"]) def wait_messages(pipeline_id: str) -> tuple[Response, int]: """Block on a typed message event. @@ -405,6 +468,10 @@ def wait_messages(pipeline_id: str) -> tuple[Response, int]: # so the caller actually observes blocking semantics. timeout = 1 + guard_response = _check_producer_pending_confirm_guard(pipeline_id, role, wait_for_types) + if guard_response is not None: + return guard_response + message_store = get_message_store() _track_long_poll_start() diff --git a/orchestrator/tests/test_messages.py b/orchestrator/tests/test_messages.py index 0eea831229..5b156b0eb0 100644 --- a/orchestrator/tests/test_messages.py +++ b/orchestrator/tests/test_messages.py @@ -1552,6 +1552,297 @@ def test_wait_timeout_clamped_to_env_cap(self, client, app, monkeypatch): assert elapsed < 5 # would be 999s without clamp +class TestProducerPendingConfirmGuard: + """Reject ``wait --for CONSENSUS_CONFIRMED`` from producers in + WORKING/PROPOSED state (#2064). + + A producer's own confirm is part of what generates the global + CONSENSUS_CONFIRMED signal — waiting on it before having confirmed + is a circular dependency that would deadlock until the overseer's + heartbeat-stall detector intervened minutes later. The guard turns + that silent deadlock into an immediate, actionable 400. + """ + + @pytest.fixture + def implement_tracker(self): + """Build a tracker matching the implement-phase shape that + triggered #2064: documenter as producer with one ADVISORY + reviewer, plus coder/tester/reviewers.""" + from peer_consensus import PeerConsensusTracker + from review_graph import ReviewCriticality, ReviewEdge, ReviewGraph + + graph = ReviewGraph( + [ + ReviewEdge("reviewer_code", "coder", ReviewCriticality.CRITICAL), + ReviewEdge("reviewer_code", "tester", ReviewCriticality.CRITICAL), + ReviewEdge("reviewer_code", "documenter", ReviewCriticality.ADVISORY), + ReviewEdge("reviewer_contract", "coder", ReviewCriticality.CRITICAL), + # tester reviews coder in the default implement graph; this + # edge makes tester genuinely dual-role (producer + reviewer) + # so test_dual_role_tester_in_proposed_blocked locks the + # dual-role contract, not just the tester-as-producer case. + ReviewEdge("tester", "coder", ReviewCriticality.CRITICAL), + ] + ) + tracker = PeerConsensusTracker("test-pipeline", graph, cooldown_seconds=0) + for role in ("coder", "tester", "documenter", "reviewer_code", "reviewer_contract"): + tracker.register_agent(role) + return tracker + + def _wait(self, client, *, role: str, for_types: list[str] | None = None): + types = for_types or ["CONSENSUS_CONFIRMED"] + qs = "&".join(f"for={t}" for t in types) + f"&role={role}&timeout=1" + return client.get(f"/api/v1/pipelines/test-pipeline/messages/wait?{qs}") + + def test_producer_in_working_blocked_on_consensus_confirmed( + self, client, app, implement_tracker + ): + """Producer in WORKING (never proposed) cannot wait on + CONSENSUS_CONFIRMED — it must propose and confirm first.""" + with app.test_request_context(): + with ( + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait(client, role="documenter") + assert resp.status_code == 400 + msg = json.loads(resp.data)["message"] + assert "documenter" in msg + assert "CONSENSUS_CONFIRMED" in msg + assert "mcp__brc__confirm" in msg + assert "#2064" in msg + + def test_producer_in_proposed_blocked_on_consensus_confirmed( + self, client, app, implement_tracker + ): + """Reproduces the issue-1965 documenter case: PROPOSED but + not CONFIRMED, attempting to STAY ALIVE on CONSENSUS_CONFIRMED.""" + implement_tracker.handle_propose( + "documenter", + { + "summary": "Wrote docs for the new fan-out feature, covering " + "thresholds, partitioning, parent cross-partition consistency, " + "and the parallelism config knob.", + "artifacts": ["docs/guides/concurrent-execution.md"], + "commit_sha": "abc1234", + }, + ) + with app.test_request_context(): + with ( + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait( + client, + role="documenter", + for_types=["CONSENSUS_CONFIRMED", "CONSENSUS_RE_REVIEW", "OVERSEER_ALERT"], + ) + assert resp.status_code == 400 + msg = json.loads(resp.data)["message"] + assert "pending_acks" in msg + + def test_producer_in_confirmed_passes(self, client, app, implement_tracker): + """A producer that has actually confirmed may legitimately wait + on CONSENSUS_CONFIRMED — that's the post-confirm STAY ALIVE + pattern the producer-lifecycle prompt prescribes.""" + # Set up: every producer proposes, every reviewer ACKs the + # critical edges, then documenter (advisory-only) confirms. + for producer in ("coder", "tester", "documenter"): + implement_tracker.handle_propose( + producer, + { + "summary": ( + f"Stub proposal from {producer} so the global guard " + "passes — every producer must propose before any " + "agent can confirm consensus." + ), + "artifacts": [f"path/{producer}.py"], + "commit_sha": "abc1234", + }, + ) + for producer in ("coder", "tester"): + implement_tracker.handle_ack( + "reviewer_code", + producer, + {"artifact_references": [f"path/{producer}.py"]}, + ) + implement_tracker.handle_ack( + "reviewer_contract", + "coder", + {"artifact_references": ["path/coder.py"]}, + ) + # documenter's only reviewer is advisory, so it's already fully ACKed + result = implement_tracker.handle_confirmed("documenter") + assert result["status"] == "confirmed" + + with app.test_request_context(): + with ( + patch("routes.messages.get_message_store", return_value=MessageStore()), + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait(client, role="documenter") + assert resp.status_code == 200 + + def test_dual_role_tester_in_proposed_blocked(self, client, app, implement_tracker): + """Dual-role tester (producer of its own artifacts + reviewer of + coder, per the implement graph) is still blocked by the guard + while in PROPOSED — its producer phase has not yet transitioned + to CONFIRMED, so the deadlock condition still holds even though + the agent also has a reviewer phase. Locks the helper's contract + for the dual-role case.""" + implement_tracker.handle_propose( + "tester", + { + "summary": ( + "Added integration tests covering the new fan-out " + "thresholds, partition boundaries, and parent " + "cross-partition consistency invariants." + ), + "artifacts": ["orchestrator/tests/test_fan_out.py"], + "commit_sha": "abc1234", + }, + ) + with app.test_request_context(): + with ( + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait(client, role="tester") + assert resp.status_code == 400 + msg = json.loads(resp.data)["message"] + assert "tester" in msg + assert "CONSENSUS_CONFIRMED" in msg + assert "#2064" in msg + + def test_reviewer_only_role_passes(self, client, app, implement_tracker): + """Reviewer-only roles may wait on CONSENSUS_CONFIRMED at any + time — they have no producer-side confirm of their own to + block consensus on.""" + with app.test_request_context(): + with ( + patch("routes.messages.get_message_store", return_value=MessageStore()), + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait(client, role="reviewer_contract") + assert resp.status_code == 200 + + def test_other_for_types_pass_for_unconfirmed_producer(self, client, app, implement_tracker): + """Producers in WORKING/PROPOSED can still wait on the events + they legitimately need — CONSENSUS_ACK from reviewers, + CONSENSUS_PROPOSE from peers, OVERSEER_ALERT, etc.""" + with app.test_request_context(): + with ( + patch("routes.messages.get_message_store", return_value=MessageStore()), + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait( + client, + role="documenter", + for_types=["CONSENSUS_ACK", "CONSENSUS_NACK", "OVERSEER_ALERT"], + ) + assert resp.status_code == 200 + + def test_no_tracker_passes(self, client, app): + """When no consensus tracker is registered (e.g. test fixtures, + non-BRC pipelines, post-orchestrator-restart before reconstruction), + the guard short-circuits rather than wedging the wait endpoint.""" + with app.test_request_context(): + with ( + patch("routes.messages.get_message_store", return_value=MessageStore()), + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch("peer_consensus.get_peer_consensus_tracker", return_value=None), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = self._wait(client, role="documenter") + assert resp.status_code == 200 + + def test_no_role_passes(self, client, app, implement_tracker): + """Calls without a role parameter (e.g. broadcast snapshots) cannot + be evaluated by the guard and must pass through.""" + with app.test_request_context(): + with ( + patch("routes.messages.get_message_store", return_value=MessageStore()), + patch( + "routes.messages.get_state_store_for_pipeline" + ) as mock_get_store_for_pipeline, + patch( + "peer_consensus.get_peer_consensus_tracker", + return_value=implement_tracker, + ), + ): + mock_get_store_for_pipeline.return_value = ( + MagicMock(), + _make_pipeline_mock(), + ) + resp = client.get( + "/api/v1/pipelines/test-pipeline/messages/wait" + "?for=CONSENSUS_CONFIRMED&timeout=1" + ) + assert resp.status_code == 200 + + class TestEnvCapConfig: """`EGG_MESSAGE_POLL_MAX_WAIT` plumbing (issue #1897).""" diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index f9c4072ad8..f3db220381 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -3029,3 +3029,64 @@ def test_advisory_ack_not_needed(self): result = t.get_fully_acked_producers() assert "coder" in result, "Advisory reviewer ACK should not be required" + + +class TestIsProducerPendingConfirm: + """Backs the wait_loop guard added in #2064 — every producer state + machine transition is exercised so the guard's input never lies.""" + + def _build_tracker(self): + graph = ReviewGraph( + [ + ReviewEdge("reviewer_code", "coder", ReviewCriticality.CRITICAL), + ReviewEdge("reviewer_code", "documenter", ReviewCriticality.ADVISORY), + ] + ) + t = PeerConsensusTracker("test", graph, cooldown_seconds=0) + t.register_agent("coder") + t.register_agent("documenter") + t.register_agent("reviewer_code") + return t + + def test_producer_in_working_is_pending(self): + t = self._build_tracker() + assert t.is_producer_pending_confirm("coder") + assert t.is_producer_pending_confirm("documenter") + + def test_producer_in_proposed_is_pending(self): + t = self._build_tracker() + t.handle_propose( + "coder", + {"summary": "x" * 60, "artifacts": ["a.py"], "commit_sha": "abc1234"}, + ) + assert t.is_producer_pending_confirm("coder") + + def test_producer_in_confirmed_is_not_pending(self): + t = self._build_tracker() + # documenter has only an advisory reviewer, so confirm only + # needs every producer to have proposed (the global guard). + t.handle_propose( + "coder", + {"summary": "x" * 60, "artifacts": ["a.py"], "commit_sha": "abc1234"}, + ) + t.handle_propose( + "documenter", + {"summary": "y" * 60, "artifacts": ["d.md"], "commit_sha": "def5678"}, + ) + t.handle_ack("reviewer_code", "coder", {"artifact_references": ["a.py"]}) + # documenter's confirm doesn't depend on the reviewer having + # confirmed first — check_confirm_guard only gates on + # global_zero_proposal + producer_not_fully_acked. We jump + # straight to the producer's own confirm. + result = t.handle_confirmed("documenter") + assert result["status"] == "confirmed" + assert not t.is_producer_pending_confirm("documenter") + + def test_reviewer_only_role_is_not_pending(self): + t = self._build_tracker() + # reviewer_code is only a reviewer here, never a producer + assert not t.is_producer_pending_confirm("reviewer_code") + + def test_unknown_role_is_not_pending(self): + t = self._build_tracker() + assert not t.is_producer_pending_confirm("not_a_real_role")