From eeb10f4c42595ce36ca942c4b9d1f77ceab8df4b Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 25 Apr 2026 13:43:47 -0700 Subject: [PATCH 1/3] Fix #2064: reject wait_loop on CONSENSUS_CONFIRMED for non-confirmed producers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A producer's own consensus_confirmed is part of what generates the global CONSENSUS_CONFIRMED signal — waiting on it before the producer has reached CONFIRMED state is a circular dependency that deadlocks the pipeline until the overseer's heartbeat-stall band-aid eventually intervenes. Observed in pipeline issue-1965 (PR #2061): the documenter proposed, called consensus confirmed but got status='pending_acks' (because coder/tester hadn't yet proposed and the global_zero_proposal guard fired), then entered the post-confirm STAY ALIVE wait_loop on CONSENSUS_CONFIRMED as if confirm had succeeded. The agent sat there for ~36 minutes waking briefly on each peer's confirm but never re-attempting its own. Only the agent-heartbeat-stall OVERSEER_ALERT broke it out. This change adds a server-side guard in /messages/wait that returns HTTP 400 with an actionable error when the caller is a producer in WORKING/PROPOSED state and CONSENSUS_CONFIRMED appears in for_types. The error tells the agent to call mcp__brc__confirm and, if it returns pending_acks, to wait on the prerequisite events instead (CONSENSUS_PROPOSE, CONSENSUS_ACK, CONSENSUS_RE_REVIEW) before retrying confirm. The fix is structural — matching the project's preference for infrastructure constraints over prompt-based rules — and applies generically to every producer role (coder, tester, documenter, and any future producer), not just the documenter case that surfaced it. - orchestrator/peer_consensus.py: add is_producer_pending_confirm helper - orchestrator/routes/messages.py: add guard at the wait endpoint - orchestrator/tests/test_messages.py: 7 cases covering the documenter scenario, dual-role agents, reviewer-only roles, missing tracker, missing role, and other for_types passthrough - orchestrator/tests/test_peer_consensus_integration.py: 5 cases for the helper itself across the producer state machine - docs/reference/agent-wait-patterns.md: anti-pattern 5 documenting the deadlock and the recovery idiom Co-Authored-By: Claude Opus 4.7 (1M context) --- docs/reference/agent-wait-patterns.md | 40 +++ orchestrator/peer_consensus.py | 15 ++ orchestrator/routes/messages.py | 62 +++++ orchestrator/tests/test_messages.py | 246 ++++++++++++++++++ .../tests/test_peer_consensus_integration.py | 63 +++++ 5 files changed, 426 insertions(+) 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..c6cdb49ba5 100644 --- a/orchestrator/routes/messages.py +++ b/orchestrator/routes/messages.py @@ -353,6 +353,64 @@ 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. + + 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: # pragma: no cover - import-shim parity with other routes + try: + from ..peer_consensus import ( # type: ignore[no-redef,import-not-found] + get_peer_consensus_tracker, + ) + except ImportError: + 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 +463,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..8a2604465e 100644 --- a/orchestrator/tests/test_messages.py +++ b/orchestrator/tests/test_messages.py @@ -1552,6 +1552,252 @@ 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), + ] + ) + 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_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..6b82bbc3d9 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -3029,3 +3029,66 @@ 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"]}) + # Advisory ACK still required for reviewer to satisfy "must have + # reviewed" guard, even though it doesn't gate is_fully_acked. + t.handle_ack("reviewer_code", "documenter", {"artifact_references": ["d.md"]}) + # Reviewer must confirm before producers can confirm cleanly + result = t.handle_confirmed("reviewer_code") + assert result["status"] in ("confirmed", "partially_confirmed") + 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") From 31d32db9134bcee6de410c8b20d34e591a9a0cb1 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 25 Apr 2026 21:09:10 +0000 Subject: [PATCH 2/3] Address review feedback on PR #2077 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Drop misleading 'reviewer must confirm before producers' comment in test_producer_in_confirmed_is_not_pending and remove the unnecessary reviewer handle_confirmed call. check_confirm_guard only gates a producer's confirm on global_zero_proposal + producer_not_fully_acked, not on reviewer confirms. - Add test_dual_role_tester_in_proposed_blocked locking the helper's contract for the tester (dual-role) case the implement_tracker fixture exercises. - Simplify the peer_consensus import-shim in _check_producer_pending_confirm_guard to match the one-tier pattern already used by _apply_delphi_filter — drops the unused package-relative fallback. - Add an inline comment documenting the intentional broadness of the guard wrt the from_role query parameter. --- orchestrator/routes/messages.py | 18 +++++---- orchestrator/tests/test_messages.py | 39 +++++++++++++++++++ .../tests/test_peer_consensus_integration.py | 7 ++-- 3 files changed, 54 insertions(+), 10 deletions(-) diff --git a/orchestrator/routes/messages.py b/orchestrator/routes/messages.py index c6cdb49ba5..186de5291e 100644 --- a/orchestrator/routes/messages.py +++ b/orchestrator/routes/messages.py @@ -384,15 +384,19 @@ def _check_producer_pending_confirm_guard( blocking = _PRODUCER_PENDING_CONFIRM_REJECTED_FOR_TYPES.intersection(wait_for_types) if not blocking: return None + # The guard intentionally ignores ``from_role`` — 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. try: from peer_consensus import get_peer_consensus_tracker - except ImportError: # pragma: no cover - import-shim parity with other routes - try: - from ..peer_consensus import ( # type: ignore[no-redef,import-not-found] - get_peer_consensus_tracker, - ) - except ImportError: - return None + 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 diff --git a/orchestrator/tests/test_messages.py b/orchestrator/tests/test_messages.py index 8a2604465e..011780c037 100644 --- a/orchestrator/tests/test_messages.py +++ b/orchestrator/tests/test_messages.py @@ -1706,6 +1706,45 @@ def test_producer_in_confirmed_passes(self, client, app, implement_tracker): 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 + implicit reviewer surface in some + graphs) is still blocked by the guard while in PROPOSED — its + producer phase has not yet transitioned to CONFIRMED, so the + deadlock condition still holds. Locks the helper's contract for + the tester-specific 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 diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index 6b82bbc3d9..9ac961e4b9 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -3077,9 +3077,10 @@ def test_producer_in_confirmed_is_not_pending(self): # Advisory ACK still required for reviewer to satisfy "must have # reviewed" guard, even though it doesn't gate is_fully_acked. t.handle_ack("reviewer_code", "documenter", {"artifact_references": ["d.md"]}) - # Reviewer must confirm before producers can confirm cleanly - result = t.handle_confirmed("reviewer_code") - assert result["status"] in ("confirmed", "partially_confirmed") + # 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") From c78a4be614bb542d0f7ffe81249cce7ec02d9c3d Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 25 Apr 2026 21:25:18 +0000 Subject: [PATCH 3/3] Address second-pass review feedback on PR #2077 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three non-blocking suggestions from the re-review of commit 31d32db: 1. Drop the redundant advisory ACK and stale comment in test_producer_in_confirmed_is_not_pending — documenter has zero critical reviewers in this fixture, so it is already fully ACKed without the reviewer_code/documenter advisory ACK, and the 'must have reviewed' guard the comment cited only applies when the reviewer (not the producer) tries to confirm. 2. Add ReviewEdge(tester, coder, CRITICAL) to implement_tracker so the fixture genuinely makes tester dual-role (producer of its own artifacts + reviewer of coder, matching get_default_implement_graph). test_dual_role_tester_in_proposed_blocked now locks the dual-role contract instead of just tester-as-producer, and the docstring is updated to reflect that. All other tests using the fixture continue to pass — the new edge only activates if coder confirms (it doesn't in any of these tests). 3. Move the from_role-broadness comment from the body of _check_producer_pending_confirm_guard up into its docstring so future readers see the policy when they look at the function contract, not when they wonder why the import shim is shaped the way it is. --- orchestrator/routes/messages.py | 15 ++++++++------- orchestrator/tests/test_messages.py | 16 +++++++++++----- .../tests/test_peer_consensus_integration.py | 3 --- 3 files changed, 19 insertions(+), 15 deletions(-) diff --git a/orchestrator/routes/messages.py b/orchestrator/routes/messages.py index 186de5291e..cdfd9e59f8 100644 --- a/orchestrator/routes/messages.py +++ b/orchestrator/routes/messages.py @@ -376,6 +376,14 @@ def _check_producer_pending_confirm_guard( 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. """ @@ -384,13 +392,6 @@ def _check_producer_pending_confirm_guard( blocking = _PRODUCER_PENDING_CONFIRM_REJECTED_FOR_TYPES.intersection(wait_for_types) if not blocking: return None - # The guard intentionally ignores ``from_role`` — 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. try: from peer_consensus import get_peer_consensus_tracker except ImportError: diff --git a/orchestrator/tests/test_messages.py b/orchestrator/tests/test_messages.py index 011780c037..5b156b0eb0 100644 --- a/orchestrator/tests/test_messages.py +++ b/orchestrator/tests/test_messages.py @@ -1577,6 +1577,11 @@ def implement_tracker(self): 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) @@ -1707,11 +1712,12 @@ def test_producer_in_confirmed_passes(self, client, app, implement_tracker): assert resp.status_code == 200 def test_dual_role_tester_in_proposed_blocked(self, client, app, implement_tracker): - """Dual-role tester (producer + implicit reviewer surface in some - graphs) is still blocked by the guard while in PROPOSED — its - producer phase has not yet transitioned to CONFIRMED, so the - deadlock condition still holds. Locks the helper's contract for - the tester-specific case.""" + """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", { diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index 9ac961e4b9..f3db220381 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -3074,9 +3074,6 @@ def test_producer_in_confirmed_is_not_pending(self): {"summary": "y" * 60, "artifacts": ["d.md"], "commit_sha": "def5678"}, ) t.handle_ack("reviewer_code", "coder", {"artifact_references": ["a.py"]}) - # Advisory ACK still required for reviewer to satisfy "must have - # reviewed" guard, even though it doesn't gate is_fully_acked. - t.handle_ack("reviewer_code", "documenter", {"artifact_references": ["d.md"]}) # 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