Fix BRC plan phase spawning multiple cycles instead of single consensus round

orchestrator/peer_consensus.py

@@ -353,7 +353,10 @@ def handle_confirmed(self, agent_role: str) -> dict[str, Any]:
             # Check if agent can confirm
             if self.graph.is_producer(agent_role):
                 if not self.matrix.is_fully_acked(agent_role):
-                    raise ValueError(f"Producer {agent_role} cannot confirm: not fully ACKed")
+                    return {
+                        "status": "pending_acks",
+                        "message": f"Producer {agent_role} waiting for reviewer re-ACKs",
+                    }
                 self._producer_phases[agent_role] = ConsensusPhase.CONFIRMED
 
             if self.graph.is_reviewer(agent_role):

orchestrator/routes/pipelines.py

@@ -4798,7 +4798,7 @@ def _update_agents_complete() -> None:
                 pip = store.load_pipeline(pipeline_id)
                 pe = pip.get_phase_execution(PipelinePhase(phase_str))
                 for agent in pe.agents:
-                    if agent.status == StateAgentStatus.RUNNING:
+                    if agent.status in (StateAgentStatus.RUNNING, StateAgentStatus.FAILED):
                         agent.status = StateAgentStatus.COMPLETE
                         agent.completed_at = datetime.utcnow()
                 store.save_pipeline(pip)
@@ -6409,6 +6409,13 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None:
                         phase_execution = pipeline.get_phase_execution(current_phase)
                         review_cycle = phase_execution.review_cycles
 
+                        # Reset status to RUNNING at cycle start so that a
+                        # previous cycle's FAILED status doesn't persist and
+                        # cause _derive_subphase_status() to misreport (see
+                        # issue #1178).
+                        phase_execution.status = PipelineStatus.RUNNING
+                        pipeline.status = PipelineStatus.RUNNING
+
                         # Record when actual agent work begins (excludes sandbox setup
                         # and HITL waiting time from the phase duration).
                         phase_execution.work_started_at = datetime.utcnow()
@@ -6788,8 +6795,11 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None:
                     reviewer_roles = _phase_reviewer_roles.get(current_phase.value, [])
                     if use_tier3:
                         reviewer_roles = [r for r in reviewer_roles if r != AgentRole.REVIEWER_CODE]
-                    if not reviewer_roles:
-                        break  # No reviewers for this phase — advance
+                    # BRC concurrent phases already include reviewers in the consensus
+                    # protocol. Spawning separate reviewers would duplicate the review
+                    # and trigger unnecessary retry cycles (see issue #1178).
+                    if use_concurrent or not reviewer_roles:
+                        break  # No separate reviewers needed — advance to next phase
 
                     # Clean stale verdict files
                     for role in reviewer_roles:

orchestrator/routes/signals.py

@@ -58,12 +58,13 @@ def make_error_response(
 def make_success_response(
     message: str,
     data: dict[str, Any] | None = None,
+    status_code: int = 200,
 ) -> tuple[Response, int]:
     """Create a success response."""
     response: dict[str, Any] = {"success": True, "message": message}
     if data:
         response["data"] = data
-    return jsonify(response), 200
+    return jsonify(response), status_code
 
 
 from routes import (  # noqa: E402 — shared helper
@@ -976,6 +977,15 @@ def handle_consensus_confirmed_signal(
     try:
         result = tracker.handle_confirmed(agent_role)
 
+        # If the producer is waiting for reviewer re-ACKs (e.g. after a
+        # re-proposal invalidated stale ACKs), return 202 so the agent
+        # knows to retry later instead of treating it as an error.
+        # Note: we intentionally skip writing a CONSENSUS_CONFIRMED message
+        # to the message store here — the agent hasn't actually confirmed,
+        # so peers polling for CONSENSUS_CONFIRMED won't see a premature one.
+        if result.get("status") == "pending_acks":
+            return make_success_response(result["message"], data=result, status_code=202)
+
         from message_store import Message, MessageType, get_message_store
 
         store = get_message_store()
@@ -1061,7 +1071,8 @@ def handle_batch_signals(pipeline_id: str) -> tuple[Response, int]:
                 results.append(
                     {
                         "signal_type": signal_type,
-                        "success": status == 200,
+                        "success": status in (200, 202),
+                        "pending": status == 202,
                         "response": response.get_json(),
                     }
                 )

orchestrator/tests/test_concurrent_integration.py

@@ -677,3 +677,115 @@ def test_wrapper_contains_restart_logic(self):
         assert "RESTART_COUNT" in script
         # Must NOT contain auto-READY
         assert "Auto-signaling READY" not in script
+
+
+class TestConcurrentPhaseSkipsReviewerSpawn:
+    """Test that concurrent (BRC) phases do not spawn separate reviewer
+    containers after consensus, preventing duplicate review cycles (issue #1178)."""
+
+    def test_is_concurrent_execution_true_for_concurrent_pipeline(self):
+        """is_concurrent_execution returns True for concurrent pipelines,
+        which causes the reviewer-spawn guard to break."""
+        from multi_agent import is_concurrent_execution
+
+        pipeline = _make_concurrent_pipeline()
+
+        # Concurrent execution is detected for all phases
+        assert is_concurrent_execution(pipeline, phase="implement") is True
+        assert is_concurrent_execution(pipeline, phase="plan") is True
+        assert is_concurrent_execution(pipeline, phase="refine") is True
+
+    def test_reviewer_roles_exist_for_phases(self):
+        """Reviewer roles ARE defined for phases — proving the concurrent
+        guard is needed to prevent redundant spawning."""
+        from egg_contracts.agent_roles import _PHASE_REVIEWERS
+
+        # Without the use_concurrent guard, these roles would be spawned
+        assert len(_PHASE_REVIEWERS.get("implement", [])) > 0
+        assert len(_PHASE_REVIEWERS.get("plan", [])) > 0
+
+    def test_non_concurrent_pipeline_allows_reviewers(self):
+        """When concurrent_execution is False, is_concurrent_execution is
+        False and the reviewer-spawn guard would NOT break."""
+        from multi_agent import is_concurrent_execution
+
+        pipeline = _make_concurrent_pipeline()
+        pipeline.config.concurrent_execution = False
+        pipeline.config.concurrent_phases = []
+
+        assert is_concurrent_execution(pipeline, phase="implement") is False
+
+
+class TestAgentsMarkedCompleteAfterConsensus:
+    """Test that _update_agents_complete transitions FAILED agents to COMPLETE
+    when BRC consensus succeeds (issue #1178, Bug 3)."""
+
+    def test_failed_agents_become_complete_via_store_roundtrip(self):
+        """Agents that exited non-zero should be marked COMPLETE after
+        consensus. Tested through a mock store load→update→save cycle
+        matching the _update_agents_complete code path."""
+        from models import (
+            AgentExecution,
+            AgentExecutionStatus,
+        )
+
+        # Build a pipeline with one RUNNING and one FAILED agent
+        pipeline = _make_concurrent_pipeline()
+        pe = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT)
+        pe.status = PipelineStatus.RUNNING
+        pe.agents = [
+            AgentExecution(role="coder", status=AgentExecutionStatus.RUNNING),
+            AgentExecution(
+                role="tester",
+                status=AgentExecutionStatus.FAILED,
+                error="Container exited with code 1",
+            ),
+        ]
+
+        # Mock store: load_pipeline returns our pipeline, save captures it
+        mock_store = MagicMock()
+        mock_store.load_pipeline.return_value = pipeline
+
+        # Simulate _update_agents_complete: load → update → save
+        pip = mock_store.load_pipeline(pipeline.id)
+        loaded_pe = pip.get_phase_execution(PipelinePhase.IMPLEMENT)
+        for agent in loaded_pe.agents:
+            if agent.status in (
+                AgentExecutionStatus.RUNNING,
+                AgentExecutionStatus.FAILED,
+            ):
+                agent.status = AgentExecutionStatus.COMPLETE
+                agent.completed_at = datetime.now(tz=UTC)
+        mock_store.save_pipeline(pip)
+
+        # Verify both agents are COMPLETE in the saved pipeline
+        saved_pipeline = mock_store.save_pipeline.call_args[0][0]
+        saved_pe = saved_pipeline.get_phase_execution(PipelinePhase.IMPLEMENT)
+        assert saved_pe.agents[0].status == AgentExecutionStatus.COMPLETE
+        assert saved_pe.agents[1].status == AgentExecutionStatus.COMPLETE
+        assert saved_pe.agents[1].completed_at is not None
+        # Error info should be preserved
+        assert saved_pe.agents[1].error == "Container exited with code 1"
+
+    def test_phase_status_reset_via_get_phase_execution(self):
+        """Phase and pipeline status reset to RUNNING at cycle start is
+        verified through the model accessor, not direct attribute assignment."""
+        pipeline = _make_concurrent_pipeline()
+        pe = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT)
+        pe.status = PipelineStatus.FAILED
+        pipeline.status = PipelineStatus.FAILED
+
+        # Simulate the status reset that happens at cycle start
+        # (pipelines.py:6416) — access through get_phase_execution to
+        # ensure the model's dict-based storage works correctly.
+        loaded_pe = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT)
+        assert loaded_pe.status == PipelineStatus.FAILED  # precondition
+
+        loaded_pe.status = PipelineStatus.RUNNING
+        pipeline.status = PipelineStatus.RUNNING
+
+        # Re-fetch via accessor — must reflect the update
+        assert (
+            pipeline.get_phase_execution(PipelinePhase.IMPLEMENT).status == PipelineStatus.RUNNING
+        )
+        assert pipeline.status == PipelineStatus.RUNNING

orchestrator/tests/test_peer_consensus_integration.py

@@ -931,8 +931,9 @@ def test_producer_confirm_fails_without_re_review(self, refine_tracker):
         t.handle_propose("refiner", {"summary": "v3", "artifacts": ["design.md"]})
 
         # Refiner should NOT be able to confirm (not fully ACKed on v3)
-        with pytest.raises(ValueError, match="not fully ACKed"):
-            t.handle_confirmed("refiner")
+        # Returns pending_acks instead of raising ValueError (issue #1178)
+        result = t.handle_confirmed("refiner")
+        assert result["status"] == "pending_acks"
 
         # After both reviewers re-ACK, refiner can confirm
         t.handle_ack("reviewer_agent_design", "refiner", {"artifact_references": ["design.md"]})
@@ -985,3 +986,71 @@ def test_nacked_then_confirmed_reviewer_is_unconfirmed_on_reproposal(self, refin
         t.handle_confirmed("reviewer_agent_design")
         result = t.handle_confirmed("reviewer_refine")
         assert result["consensus_reached"] is True
+
+
+class TestPrematureConfirmReturnsPending:
+    """Test that handle_confirmed returns pending_acks instead of raising
+    when a producer tries to confirm before being fully ACKed (issue #1178)."""
+
+    def test_confirm_before_acked_returns_pending(self, tracker):
+        """Producer confirming without full ACKs gets pending_acks, not ValueError."""
+        # Coder proposes
+        tracker.handle_propose(
+            "coder",
+            {
+                "summary": "Implemented feature",
+                "artifacts": ["src/feature.py"],
+            },
+        )
+
+        # Only reviewer_code ACKs, but checker hasn't ACKed yet
+        tracker.handle_ack(
+            "reviewer_code",
+            "coder",
+            {
+                "attestation": {
+                    "files_reviewed": ["src/feature.py"],
+                    "issues_found": 0,
+                    "issues_resolved": 0,
+                    "risk_considered": "None",
+                },
+                "artifact_references": ["src/feature.py"],
+            },
+        )
+
+        # Coder tries to confirm before checker ACKs — should return pending, not raise
+        result = tracker.handle_confirmed("coder")
+        assert result["status"] == "pending_acks"
+        assert "waiting" in result["message"].lower()
+
+        # Coder should NOT be in confirmed set
+        assert "coder" not in tracker._confirmed
+
+    def test_confirm_after_re_propose_invalidates_stale_acks(self, tracker):
+        """After re-proposal un-confirms stale reviewers, premature confirm returns pending."""
+        # Full happy path first: propose, ACK, but then re-propose
+        tracker.handle_propose(
+            "coder",
+            {"summary": "v1", "artifacts": ["src/auth.py"]},
+        )
+        tracker.handle_ack(
+            "reviewer_code",
+            "coder",
+            {"artifact_references": ["src/auth.py"]},
+        )
+        tracker.handle_ack(
+            "checker",
+            "coder",
+            {"artifact_references": ["src/auth.py"]},
+        )
+
+        # Coder re-proposes (invalidating stale ACKs)
+        tracker.handle_re_propose(
+            "coder",
+            {"summary": "v2", "artifacts": ["src/auth.py"]},
+            changed_artifacts=["src/auth.py"],
+        )
+
+        # Now coder tries to confirm — should get pending_acks
+        result = tracker.handle_confirmed("coder")
+        assert result["status"] == "pending_acks"

sandbox/egg_lib/orch_cli.py

@@ -1248,7 +1248,12 @@ def cmd_consensus_confirmed(args: argparse.Namespace) -> int:
         return 0
 
     if result.get("success"):
-        consensus_reached = result.get("data", {}).get("consensus_reached", False)
+        data = result.get("data", {})
+        # 202: producer is waiting for reviewer re-ACKs after re-proposal
+        if data.get("status") == "pending_acks":
+            print(f"Waiting for reviewer re-ACKs: {result.get('message')}")
+            return 2
+        consensus_reached = data.get("consensus_reached", False)
         print(f"Confirmation recorded for {role}")
         if consensus_reached:
             print("  Consensus reached!")