diff --git a/orchestrator/peer_consensus.py b/orchestrator/peer_consensus.py index eca1216c5c..33cf279260 100644 --- a/orchestrator/peer_consensus.py +++ b/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): diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 87c2874bad..a4da0ad5c3 100644 --- a/orchestrator/routes/pipelines.py +++ b/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: diff --git a/orchestrator/routes/signals.py b/orchestrator/routes/signals.py index 72e58d83a3..f54f4d7346 100644 --- a/orchestrator/routes/signals.py +++ b/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(), } ) diff --git a/orchestrator/tests/test_concurrent_integration.py b/orchestrator/tests/test_concurrent_integration.py index 1524c0dea5..cab58c9552 100644 --- a/orchestrator/tests/test_concurrent_integration.py +++ b/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 diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index de61705799..5d7c276c76 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/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" diff --git a/sandbox/egg_lib/orch_cli.py b/sandbox/egg_lib/orch_cli.py index 901ac0db4f..063ae5c6b7 100755 --- a/sandbox/egg_lib/orch_cli.py +++ b/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!")