From a8a9f64a6ff5fd23c969e94d688bc9332849ee97 Mon Sep 17 00:00:00 2001 From: egg Date: Thu, 12 Mar 2026 20:36:18 +0000 Subject: [PATCH 1/4] Add consensus-driven phase advancement for concurrent execution --- orchestrator/events.py | 2 + orchestrator/routes/pipelines.py | 254 ++++-- orchestrator/tests/test_concurrent_wait.py | 53 +- orchestrator/tests/test_consensus_polling.py | 770 +++++++++++++++++++ 4 files changed, 1021 insertions(+), 58 deletions(-) create mode 100644 orchestrator/tests/test_consensus_polling.py diff --git a/orchestrator/events.py b/orchestrator/events.py index db1d879bab..09fc1ac259 100644 --- a/orchestrator/events.py +++ b/orchestrator/events.py @@ -64,6 +64,8 @@ class EventType(StrEnum): # Consensus / readiness READINESS_CHANGED = "readiness.changed" + CONSENSUS_REACHED = "consensus.reached" + CONSENSUS_TIMEOUT = "consensus.timeout" # HITL events DECISION_CREATED = "decision.created" diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index d91b683cc7..bf15700df5 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -8,6 +8,7 @@ import subprocess import sys import threading +import time from datetime import datetime from pathlib import Path from typing import TYPE_CHECKING, Any @@ -4549,43 +4550,27 @@ def _run_concurrent_phase( ) return 1, logs - # Wait for all containers to exit concurrently. + # Consensus-driven polling loop with container-exit fallback. # - # NOTE: The ConcurrentPhaseExecutor exposes check_consensus() and - # handle_agent_failure() for consensus-driven phase advancement, but - # they are not used here. For V1, phase completion is determined by - # container exit codes (same model as sequential/wave paths). - # Consensus-driven advancement — where agents signal READY/BLOCKED/ - # OBJECTING and the orchestrator evaluates consensus mid-execution — - # will be integrated in a follow-up once the polling loop is added. + # The loop periodically checks consensus via executor.check_consensus(). + # When all agents signal READY, the phase completes immediately without + # waiting for containers to exit. If consensus is never reached (timeout + # or all containers exit first), fall back to exit-code-based completion. active_executions = [e for e in executions if e.container_id] docker_client = spawner.docker all_logs: list[str] = [] - logs_lock = threading.Lock() has_failures = [False] # Mutable container for closure access - def _wait_and_record(exec_info: "StateAgentExecution") -> None: - """Wait for one container, capture logs, update pipeline state.""" - try: - final_info = docker_client.wait_for_container( - exec_info.container_id, - timeout=3600, - ) - except (ContainerNotFoundError, ContainerOperationError) as e: - logger.warning( - "Container lost during wait", - container_id=exec_info.container_id, - role=exec_info.role.value, - error=str(e), - ) - final_info = ContainerInfo( - container_id=exec_info.container_id, - container_name=f"{pipeline_id}-{exec_info.role.value}", - status=ContainerStatus.FAILED, - exit_code=-1, - exited_at=datetime.utcnow(), - ) + poll_interval = 5 # seconds + consensus_timeout = getattr(pipeline.config, "consensus_timeout_minutes", 30) * 60 + start_time = time.monotonic() + objection_decision_created = False + + # Track which containers have exited and their results. + exited_containers: dict[str, ContainerInfo] = {} + def _record_container_exit(exec_info: "StateAgentExecution", final_info: ContainerInfo) -> None: + """Capture logs and update pipeline state for an exited container.""" container_logs = "" if final_info.exit_code != 0: has_failures[0] = True @@ -4597,12 +4582,10 @@ def _wait_and_record(exec_info: "StateAgentExecution") -> None: except Exception: pass - with logs_lock: - all_logs.append( - f"--- {exec_info.role.value} (exit={final_info.exit_code}) ---\n{container_logs}" - ) + all_logs.append( + f"--- {exec_info.role.value} (exit={final_info.exit_code}) ---\n{container_logs}" + ) - # Update container and agent status in pipeline state. if store is not None: try: with get_pipeline_state_lock(pipeline_id): @@ -4634,23 +4617,196 @@ def _wait_and_record(exec_info: "StateAgentExecution") -> None: error=str(track_err), ) - with ThreadPoolExecutor(max_workers=len(active_executions) or 1) as pool: - futures = {pool.submit(_wait_and_record, e): e for e in active_executions} - for future in as_completed(futures): - exc = future.exception() - if exc: - logger.error( - "Error waiting for concurrent agent", - role=futures[future].role.value, - error=str(exc), + def _stop_running_containers() -> None: + """Gracefully stop all containers that haven't exited yet.""" + for e in active_executions: + if e.container_id not in exited_containers: + try: + docker_client.stop_container(e.container_id, timeout=30) + except Exception: + pass + + def _update_agents_complete() -> None: + """Mark all running agents as COMPLETE in pipeline state (consensus path).""" + if store is None: + return + try: + with get_pipeline_state_lock(pipeline_id): + pip = store.load_pipeline(pipeline_id) + pe = pip.get_phase_execution(PipelinePhase(phase_str)) + for agent in pe.agents: + if agent.status == StateAgentStatus.RUNNING: + agent.status = StateAgentStatus.COMPLETE + agent.completed_at = datetime.utcnow() + store.save_pipeline(pip) + except Exception as track_err: + logger.warning( + "Failed to update agents to COMPLETE after consensus", + pipeline_id=pipeline_id, + error=str(track_err), + ) + + while True: + elapsed = time.monotonic() - start_time + + # 1. Check consensus + try: + consensus = executor.check_consensus() + except Exception as e: + logger.warning( + "Consensus check failed, continuing poll", + pipeline_id=pipeline_id, + error=str(e), + ) + consensus = {"is_complete": False, "has_objections": False, "blocking_agents": []} + + # 2. Consensus reached — stop containers and return success + if consensus.get("is_complete"): + if _emit_event is not None: + _emit_event( + EventType.CONSENSUS_REACHED, + pipeline_id, + data={"elapsed_seconds": elapsed}, ) - has_failures[0] = True + logger.info( + "Consensus reached, stopping containers", + pipeline_id=pipeline_id, + elapsed_seconds=round(elapsed, 1), + ) + _update_agents_complete() + _stop_running_containers() + combined_logs = "\n".join(all_logs) if all_logs else "Consensus reached; phase complete." + return 0, combined_logs - combined_logs = "\n".join(all_logs) - if has_failures[0]: - return 1, combined_logs + # 3. Handle objections (create HITL decision once) + if consensus.get("has_objections") and not objection_decision_created: + try: + pipeline.add_decision( + question="Agent(s) objecting to phase completion. How to proceed?", + options=["Override objections", "Wait for resolution", "Abort phase"], + phase=pipeline.current_phase, + ) + objection_decision_created = True + logger.info( + "Objection detected, HITL decision created", + pipeline_id=pipeline_id, + blocking_agents=consensus.get("blocking_agents", []), + ) + except Exception as e: + logger.warning( + "Failed to create objection HITL decision", + pipeline_id=pipeline_id, + error=str(e), + ) - return 0, combined_logs + # 4. Non-blocking check for exited containers + for exec_info in active_executions: + if exec_info.container_id in exited_containers: + continue + try: + info = docker_client.get_container_info(exec_info.container_id) + except (ContainerNotFoundError, ContainerOperationError) as e: + logger.warning( + "Container lost during poll", + container_id=exec_info.container_id, + role=exec_info.role.value, + error=str(e), + ) + info = ContainerInfo( + container_id=exec_info.container_id, + container_name=f"{pipeline_id}-{exec_info.role.value}", + status=ContainerStatus.FAILED, + exit_code=-1, + exited_at=datetime.utcnow(), + ) + + if info.status in (ContainerStatus.EXITED, ContainerStatus.FAILED): + exited_containers[exec_info.container_id] = info + _record_container_exit(exec_info, info) + + # Handle non-zero exit as agent failure + if info.exit_code != 0: + try: + executor.handle_agent_failure( + role=exec_info.role.value, + error=f"Container exited with code {info.exit_code}", + ) + except Exception as e: + logger.warning( + "handle_agent_failure error", + role=exec_info.role.value, + error=str(e), + ) + + # 5. All containers exited — fall back to exit-code-based result + if len(exited_containers) >= len(active_executions): + combined_logs = "\n".join(all_logs) + if has_failures[0]: + return 1, combined_logs + return 0, combined_logs + + # 6. Consensus timeout + if elapsed >= consensus_timeout: + if _emit_event is not None: + _emit_event( + EventType.CONSENSUS_TIMEOUT, + pipeline_id, + data={ + "timeout_minutes": consensus_timeout / 60, + "blocking_agents": consensus.get("blocking_agents", []), + }, + ) + logger.warning( + "Consensus timeout reached, falling back to container exit", + pipeline_id=pipeline_id, + timeout_minutes=consensus_timeout / 60, + ) + try: + pipeline.add_decision( + question=f"Consensus not reached after {int(consensus_timeout / 60)} minutes. How to proceed?", + options=["Continue waiting", "Accept current state", "Abort phase"], + phase=pipeline.current_phase, + ) + except Exception: + pass + + # Fall back: wait for remaining containers with ThreadPoolExecutor + remaining = [e for e in active_executions if e.container_id not in exited_containers] + if remaining: + with ThreadPoolExecutor(max_workers=len(remaining)) as pool: + def _wait_remaining(exec_info): + try: + final_info = docker_client.wait_for_container( + exec_info.container_id, timeout=3600, + ) + except (ContainerNotFoundError, ContainerOperationError): + final_info = ContainerInfo( + container_id=exec_info.container_id, + container_name=f"{pipeline_id}-{exec_info.role.value}", + status=ContainerStatus.FAILED, + exit_code=-1, + exited_at=datetime.utcnow(), + ) + _record_container_exit(exec_info, final_info) + + futures = {pool.submit(_wait_remaining, e): e for e in remaining} + for future in as_completed(futures): + exc = future.exception() + if exc: + logger.error( + "Error waiting for container after timeout", + role=futures[future].role.value, + error=str(exc), + ) + has_failures[0] = True + + combined_logs = "\n".join(all_logs) + if has_failures[0]: + return 1, combined_logs + return 0, combined_logs + + # 7. Sleep before next poll + time.sleep(poll_interval) def _spawn_and_wait( diff --git a/orchestrator/tests/test_concurrent_wait.py b/orchestrator/tests/test_concurrent_wait.py index 4bc531ae4a..c7b2822697 100644 --- a/orchestrator/tests/test_concurrent_wait.py +++ b/orchestrator/tests/test_concurrent_wait.py @@ -21,6 +21,9 @@ PipelineStatus, ) +# Common consensus result for tests that rely on container-exit fallback. +_NO_CONSENSUS = {"is_complete": False, "has_objections": False, "blocking_agents": []} + def _make_concurrent_pipeline(pipeline_id: str = "issue-999") -> Pipeline: """Create a pipeline with concurrent_execution enabled.""" @@ -88,7 +91,8 @@ def _make_mocks(self, executions, wait_results=None): Args: executions: List of AgentExecution returned by spawn_all. wait_results: Dict mapping container_id to ContainerInfo returned - by wait_for_container. Defaults to exit_code=0 for all. + by wait_for_container / get_container_info. + Defaults to exit_code=0 for all. """ pipeline = _make_concurrent_pipeline() phase_exec = _make_phase_execution() @@ -116,7 +120,11 @@ def _make_mocks(self, executions, wait_results=None): def _wait_side_effect(container_id, timeout=3600): return wait_results[container_id] + def _info_side_effect(container_id): + return wait_results[container_id] + mock_docker.wait_for_container.side_effect = _wait_side_effect + mock_docker.get_container_info.side_effect = _info_side_effect # Spawner mock mock_spawner = MagicMock() @@ -126,13 +134,17 @@ def _wait_side_effect(container_id, timeout=3600): return pipeline, mock_store, mock_spawner, mock_docker, phase_exec + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) def test_all_containers_exit_successfully( - self, MockExecutor, mock_build_prompt, mock_state_lock + self, MockExecutor, mock_build_prompt, mock_state_lock, mock_monotonic, mock_sleep ): """When all containers exit with code 0, returns (0, logs).""" + mock_monotonic.return_value = 0.0 + executions = [ _make_execution(AgentRole.CODER, "coder-abc"), _make_execution(AgentRole.TESTER, "tester-abc"), @@ -142,6 +154,7 @@ def test_all_containers_exit_successfully( mock_executor_instance = MagicMock() mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = _NO_CONSENSUS MockExecutor.return_value = mock_executor_instance mock_state_lock.return_value.__enter__ = MagicMock(return_value=None) @@ -162,15 +175,19 @@ def test_all_containers_exit_successfully( ) assert exit_code == 0 - assert mock_docker.wait_for_container.call_count == 3 + assert mock_docker.get_container_info.call_count == 3 + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) def test_container_failure_returns_nonzero( - self, MockExecutor, mock_build_prompt, mock_state_lock + self, MockExecutor, mock_build_prompt, mock_state_lock, mock_monotonic, mock_sleep ): """When a container exits with non-zero code, returns (1, logs).""" + mock_monotonic.return_value = 0.0 + executions = [ _make_execution(AgentRole.CODER, "coder-abc"), _make_execution(AgentRole.TESTER, "tester-abc"), @@ -199,6 +216,7 @@ def test_container_failure_returns_nonzero( mock_executor_instance = MagicMock() mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = _NO_CONSENSUS MockExecutor.return_value = mock_executor_instance mock_state_lock.return_value.__enter__ = MagicMock(return_value=None) @@ -221,13 +239,17 @@ def test_container_failure_returns_nonzero( assert exit_code == 1 assert "tester" in logs + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) def test_container_not_found_during_wait( - self, MockExecutor, mock_build_prompt, mock_state_lock + self, MockExecutor, mock_build_prompt, mock_state_lock, mock_monotonic, mock_sleep ): - """When a container disappears during wait, returns failure.""" + """When a container disappears during poll, returns failure.""" + mock_monotonic.return_value = 0.0 + from docker_client import ContainerNotFoundError executions = [ @@ -235,10 +257,11 @@ def test_container_not_found_during_wait( ] pipeline, mock_store, mock_spawner, mock_docker, _ = self._make_mocks(executions) - mock_docker.wait_for_container.side_effect = ContainerNotFoundError("coder-abc") + mock_docker.get_container_info.side_effect = ContainerNotFoundError("coder-abc") mock_executor_instance = MagicMock() mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = _NO_CONSENSUS MockExecutor.return_value = mock_executor_instance mock_state_lock.return_value.__enter__ = MagicMock(return_value=None) @@ -261,13 +284,17 @@ def test_container_not_found_during_wait( assert exit_code == 1 assert "coder" in logs + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) def test_state_store_records_containers_and_agents( - self, MockExecutor, mock_build_prompt, mock_state_lock + self, MockExecutor, mock_build_prompt, mock_state_lock, mock_monotonic, mock_sleep ): """Pipeline state is updated with container/agent info after spawn and wait.""" + mock_monotonic.return_value = 0.0 + executions = [ _make_execution(AgentRole.CODER, "coder-abc"), ] @@ -276,6 +303,7 @@ def test_state_store_records_containers_and_agents( mock_executor_instance = MagicMock() mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = _NO_CONSENSUS MockExecutor.return_value = mock_executor_instance mock_state_lock.return_value.__enter__ = MagicMock(return_value=None) @@ -299,11 +327,17 @@ def test_state_store_records_containers_and_agents( # once after wait/status update assert mock_store.save_pipeline.call_count >= 2 + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") @patch("routes.pipelines.get_pipeline_state_lock") @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) - def test_store_none_does_not_crash(self, MockExecutor, mock_build_prompt, mock_state_lock): + def test_store_none_does_not_crash( + self, MockExecutor, mock_build_prompt, mock_state_lock, mock_monotonic, mock_sleep + ): """When store=None, state recording is skipped gracefully.""" + mock_monotonic.return_value = 0.0 + executions = [ _make_execution(AgentRole.CODER, "coder-abc"), ] @@ -311,6 +345,7 @@ def test_store_none_does_not_crash(self, MockExecutor, mock_build_prompt, mock_s mock_executor_instance = MagicMock() mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = _NO_CONSENSUS MockExecutor.return_value = mock_executor_instance exit_code, logs = _run_concurrent_phase( diff --git a/orchestrator/tests/test_consensus_polling.py b/orchestrator/tests/test_consensus_polling.py new file mode 100644 index 0000000000..a91594f9be --- /dev/null +++ b/orchestrator/tests/test_consensus_polling.py @@ -0,0 +1,770 @@ +"""Tests for consensus-driven phase advancement in _run_concurrent_phase. + +Covers the polling loop that checks consensus, handles objections and timeouts, +and falls back to container-exit-based completion. +""" + +from datetime import datetime +from pathlib import Path +from unittest.mock import MagicMock, call, patch, PropertyMock + +from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + ContainerInfo, + ContainerStatus, + PhaseExecution, + Pipeline, + PipelineConfig, + PipelinePhase, + PipelineStatus, +) +from routes.pipelines import _run_concurrent_phase + + +def _make_concurrent_pipeline(pipeline_id: str = "issue-999") -> Pipeline: + """Create a pipeline with concurrent_execution enabled.""" + config = PipelineConfig() + for key, val in { + "concurrent_execution": True, + "max_concurrent_agents": 4, + "message_poll_hint_seconds": 30, + "consensus_timeout_minutes": 30, + }.items(): + try: + setattr(config, key, val) + except (AttributeError, ValueError): + config.__dict__[key] = val + + return Pipeline( + id=pipeline_id, + issue_number=999, + repo="owner/repo", + branch="egg/issue-999", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + config=config, + ) + + +def _make_execution(role: AgentRole, container_id: str, status=AgentExecutionStatus.RUNNING): + return AgentExecution( + role=role, + status=status, + container_id=container_id, + started_at=datetime.utcnow(), + ) + + +def _make_phase_execution(): + return PhaseExecution( + phase=PipelinePhase.IMPLEMENT, + status=PipelineStatus.RUNNING, + ) + + +def _base_mocks(executions, container_infos=None): + """Create common mocks for the consensus polling tests. + + Args: + executions: AgentExecution list returned by spawn_all. + container_infos: Dict of container_id -> ContainerInfo for get_container_info. + Defaults to RUNNING status for all containers. + """ + pipeline = _make_concurrent_pipeline() + phase_exec = _make_phase_execution() + + mock_store = MagicMock() + mock_pipeline_state = MagicMock() + mock_pipeline_state.get_phase_execution.return_value = phase_exec + mock_store.load_pipeline.return_value = mock_pipeline_state + + mock_docker = MagicMock() + + if container_infos is None: + container_infos = {} + for e in executions: + if e.container_id: + container_infos[e.container_id] = ContainerInfo( + container_id=e.container_id, + container_name=f"issue-999-{e.role.value}", + status=ContainerStatus.RUNNING, + exit_code=None, + ) + + mock_docker.get_container_info.side_effect = lambda cid: container_infos[cid] + + mock_spawner = MagicMock() + mock_spawner.docker = mock_docker + mock_spawner.create_concurrent_spawn_fn.return_value = MagicMock() + + return pipeline, mock_store, mock_spawner, mock_docker + + +_CALL_ARGS = dict( + repo_volumes={}, + gateway_mode="public", + repos=["owner/repo"], + sandbox_env={}, + certs_volume=None, + worktree_repo_path=Path("/tmp/test-repo"), +) + + +class TestConsensusReached: + """Consensus is reached before timeout or container exit.""" + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_consensus_reached_returns_zero( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """When check_consensus returns is_complete=True, returns (0, ...) immediately.""" + mock_monotonic.return_value = 10.0 + + executions = [ + _make_execution(AgentRole.CODER, "coder-1"), + _make_execution(AgentRole.TESTER, "tester-1"), + ] + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": True, + "has_objections": False, + "blocking_agents": [], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 0 + assert "Consensus reached" in logs + # Containers should be stopped on consensus + assert mock_docker.stop_container.call_count == 2 + # No sleep needed — consensus on first poll + mock_sleep.assert_not_called() + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_consensus_reached_after_n_polls( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """Consensus reached after several polls — sleep is called between polls.""" + poll_count = [0] + + def _monotonic(): + return poll_count[0] * 5.0 + + mock_monotonic.side_effect = _monotonic + + executions = [ + _make_execution(AgentRole.CODER, "coder-1"), + ] + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + + def _check_consensus(): + poll_count[0] += 1 + if poll_count[0] >= 3: + return {"is_complete": True, "has_objections": False, "blocking_agents": []} + return {"is_complete": False, "has_objections": False, "blocking_agents": ["coder"]} + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.side_effect = _check_consensus + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 0 + # sleep called twice (polls 1 and 2; poll 3 returns consensus) + assert mock_sleep.call_count == 2 + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_consensus_emits_event( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """CONSENSUS_REACHED event is emitted when consensus completes.""" + from events import EventType + + mock_monotonic.return_value = 42.0 + + executions = [_make_execution(AgentRole.CODER, "coder-1")] + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": True, + "has_objections": False, + "blocking_agents": [], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + mock_emit.assert_any_call( + EventType.CONSENSUS_REACHED, + "issue-999", + data={"elapsed_seconds": 0.0}, + ) + + +class TestConsensusTimeout: + """Consensus not reached within timeout window.""" + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_timeout_creates_hitl_decision( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """When consensus times out, a HITL decision is created.""" + # Start past the timeout (30 min = 1800s) + mock_monotonic.side_effect = [0.0, 1801.0] + + executions = [_make_execution(AgentRole.CODER, "coder-1")] + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + + # Container exits cleanly during the fallback wait + mock_docker.wait_for_container.return_value = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": False, + "has_objections": False, + "blocking_agents": ["coder"], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + mock_add_decision = MagicMock(return_value=MagicMock(id="dec-1")) + with patch.object(type(pipeline), "add_decision", mock_add_decision): + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 0 + # HITL decision should have been created on the pipeline + mock_add_decision.assert_called_once() + call_args = mock_add_decision.call_args + question = call_args[1].get("question", call_args[0][0] if call_args[0] else "") + assert "30 minutes" in question + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_timeout_emits_event( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """CONSENSUS_TIMEOUT event is emitted on timeout.""" + from events import EventType + + mock_monotonic.side_effect = [0.0, 1801.0] + + executions = [_make_execution(AgentRole.CODER, "coder-1")] + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + mock_docker.wait_for_container.return_value = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": False, + "has_objections": False, + "blocking_agents": ["coder"], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + mock_emit.assert_any_call( + EventType.CONSENSUS_TIMEOUT, + "issue-999", + data={"timeout_minutes": 30.0, "blocking_agents": ["coder"]}, + ) + + +class TestObjectionHandling: + """Objections trigger HITL decisions.""" + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_objection_creates_hitl_decision_once( + self, MockExecutor, mock_prompt, mock_lock, mock_monotonic, mock_sleep + ): + """When has_objections=True, a HITL decision is created only once.""" + poll_count = [0] + + def _monotonic(): + return poll_count[0] * 5.0 + + mock_monotonic.side_effect = _monotonic + + executions = [_make_execution(AgentRole.CODER, "coder-1")] + + # Container exits on third poll + running_info = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.RUNNING, + exit_code=None, + ) + exited_info = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ) + + def _get_info(cid): + if poll_count[0] >= 3: + return exited_info + return running_info + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + mock_docker.get_container_info.side_effect = _get_info + + def _check_consensus(): + poll_count[0] += 1 + return { + "is_complete": False, + "has_objections": True, + "blocking_agents": ["coder"], + } + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.side_effect = _check_consensus + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + mock_add_decision = MagicMock(return_value=MagicMock(id="dec-1")) + with patch.object(type(pipeline), "add_decision", mock_add_decision): + _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + # add_decision called exactly once despite multiple polls with objections + mock_add_decision.assert_called_once() + call_args = mock_add_decision.call_args + question = call_args[1].get("question", call_args[0][0] if call_args[0] else "") + assert "objecting" in question.lower() + + +class TestContainerExitFallback: + """All containers exit before consensus — fallback to exit codes.""" + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_all_containers_exit_success( + self, MockExecutor, mock_prompt, mock_lock, mock_monotonic, mock_sleep + ): + """When all containers exit code 0 without consensus, returns (0, ...).""" + mock_monotonic.return_value = 0.0 + + executions = [ + _make_execution(AgentRole.CODER, "coder-1"), + _make_execution(AgentRole.TESTER, "tester-1"), + ] + + container_infos = { + "coder-1": ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ), + "tester-1": ContainerInfo( + container_id="tester-1", + container_name="issue-999-tester", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ), + } + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks( + executions, container_infos=container_infos + ) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": False, + "has_objections": False, + "blocking_agents": ["coder", "tester"], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 0 + assert "coder" in logs + assert "tester" in logs + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_container_exit_failure_returns_nonzero( + self, MockExecutor, mock_prompt, mock_lock, mock_monotonic, mock_sleep + ): + """When a container exits non-zero, returns (1, ...) via fallback.""" + mock_monotonic.return_value = 0.0 + + executions = [_make_execution(AgentRole.CODER, "coder-1")] + + container_infos = { + "coder-1": ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.FAILED, + exit_code=1, + exited_at=datetime.utcnow(), + ), + } + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks( + executions, container_infos=container_infos + ) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": False, + "has_objections": False, + "blocking_agents": ["coder"], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 1 + # handle_agent_failure should have been called + mock_executor_instance.handle_agent_failure.assert_called_once_with( + role="coder", + error="Container exited with code 1", + ) + + +class TestMixedScenarios: + """Container exits and consensus interact correctly.""" + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_some_containers_exit_then_consensus( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """One container exits early, then consensus reached — still returns success.""" + poll_count = [0] + + def _monotonic(): + return poll_count[0] * 5.0 + + mock_monotonic.side_effect = _monotonic + + executions = [ + _make_execution(AgentRole.CODER, "coder-1"), + _make_execution(AgentRole.TESTER, "tester-1"), + ] + + exited_coder = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ) + running_tester = ContainerInfo( + container_id="tester-1", + container_name="issue-999-tester", + status=ContainerStatus.RUNNING, + exit_code=None, + ) + + def _get_info(cid): + if cid == "coder-1": + return exited_coder + return running_tester + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + mock_docker.get_container_info.side_effect = _get_info + + def _check_consensus(): + poll_count[0] += 1 + if poll_count[0] >= 2: + return {"is_complete": True, "has_objections": False, "blocking_agents": []} + return {"is_complete": False, "has_objections": False, "blocking_agents": ["tester"]} + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.side_effect = _check_consensus + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 0 + # Tester container should be stopped on consensus + stopped_ids = {c.args[0] for c in mock_docker.stop_container.call_args_list} + assert "tester-1" in stopped_ids + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_agent_failure_calls_handle_agent_failure( + self, MockExecutor, mock_prompt, mock_lock, mock_monotonic, mock_sleep + ): + """When a container crashes, handle_agent_failure() is called.""" + mock_monotonic.return_value = 0.0 + + executions = [ + _make_execution(AgentRole.CODER, "coder-1"), + _make_execution(AgentRole.TESTER, "tester-1"), + ] + + container_infos = { + "coder-1": ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.FAILED, + exit_code=137, + exited_at=datetime.utcnow(), + ), + "tester-1": ContainerInfo( + container_id="tester-1", + container_name="issue-999-tester", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ), + } + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks( + executions, container_infos=container_infos + ) + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.return_value = { + "is_complete": False, + "has_objections": False, + "blocking_agents": ["coder"], + } + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + assert exit_code == 1 + mock_executor_instance.handle_agent_failure.assert_called_once_with( + role="coder", + error="Container exited with code 137", + ) + + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_consensus_check_error_continues_polling( + self, MockExecutor, mock_prompt, mock_lock, mock_monotonic, mock_sleep + ): + """If check_consensus raises, the loop continues polling.""" + poll_count = [0] + + def _monotonic(): + return poll_count[0] * 5.0 + + mock_monotonic.side_effect = _monotonic + + executions = [_make_execution(AgentRole.CODER, "coder-1")] + + running_info = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.RUNNING, + exit_code=None, + ) + exited_info = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ) + + def _get_info(cid): + if poll_count[0] >= 3: + return exited_info + return running_info + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + mock_docker.get_container_info.side_effect = _get_info + + def _check_consensus(): + poll_count[0] += 1 + if poll_count[0] == 1: + raise RuntimeError("evaluator error") + return {"is_complete": False, "has_objections": False, "blocking_agents": ["coder"]} + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.side_effect = _check_consensus + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + # Should succeed via container-exit fallback despite first consensus error + assert exit_code == 0 From c0d448584fc76a6094d39074e136aaf8e95deee7 Mon Sep 17 00:00:00 2001 From: egg Date: Thu, 12 Mar 2026 20:39:03 +0000 Subject: [PATCH 2/4] Fix checks: apply automated formatting fixes --- orchestrator/routes/pipelines.py | 9 ++++++--- orchestrator/tests/test_consensus_polling.py | 18 +++++++++--------- 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index bf15700df5..b04473314d 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -4436,7 +4436,6 @@ def _run_concurrent_phase( Returns: (exit_code, logs) — 0 on success. """ - import threading from concurrent.futures import ThreadPoolExecutor, as_completed from models import ( @@ -4675,7 +4674,9 @@ def _update_agents_complete() -> None: ) _update_agents_complete() _stop_running_containers() - combined_logs = "\n".join(all_logs) if all_logs else "Consensus reached; phase complete." + combined_logs = ( + "\n".join(all_logs) if all_logs else "Consensus reached; phase complete." + ) return 0, combined_logs # 3. Handle objections (create HITL decision once) @@ -4774,10 +4775,12 @@ def _update_agents_complete() -> None: remaining = [e for e in active_executions if e.container_id not in exited_containers] if remaining: with ThreadPoolExecutor(max_workers=len(remaining)) as pool: + def _wait_remaining(exec_info): try: final_info = docker_client.wait_for_container( - exec_info.container_id, timeout=3600, + exec_info.container_id, + timeout=3600, ) except (ContainerNotFoundError, ContainerOperationError): final_info = ContainerInfo( diff --git a/orchestrator/tests/test_consensus_polling.py b/orchestrator/tests/test_consensus_polling.py index a91594f9be..2f5ae97074 100644 --- a/orchestrator/tests/test_consensus_polling.py +++ b/orchestrator/tests/test_consensus_polling.py @@ -6,7 +6,7 @@ from datetime import datetime from pathlib import Path -from unittest.mock import MagicMock, call, patch, PropertyMock +from unittest.mock import MagicMock, patch from models import ( AgentExecution, @@ -102,14 +102,14 @@ def _base_mocks(executions, container_infos=None): return pipeline, mock_store, mock_spawner, mock_docker -_CALL_ARGS = dict( - repo_volumes={}, - gateway_mode="public", - repos=["owner/repo"], - sandbox_env={}, - certs_volume=None, - worktree_repo_path=Path("/tmp/test-repo"), -) +_CALL_ARGS = { + "repo_volumes": {}, + "gateway_mode": "public", + "repos": ["owner/repo"], + "sandbox_env": {}, + "certs_volume": None, + "worktree_repo_path": Path("/tmp/test-repo"), +} class TestConsensusReached: From add79c3c805ca2d89611ac3189fce63dd2a51562 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 12 Mar 2026 20:56:44 +0000 Subject: [PATCH 3/4] Address review feedback on consensus polling loop - Add threading lock for all_logs/has_failures mutations in the timeout fallback path (ThreadPoolExecutor threads) - Return exit code 1 when consensus is reached but containers have already failed (e.g. OOM kill) - Add clarifying comments for fire-and-forget HITL decisions - Guard consensus_timeout_minutes with min 1 minute to prevent immediate timeout on 0/negative config values - Include ContainerStatus.REMOVED in container exit check to avoid one-cycle delay and spurious warning log - Add explanatory comment on test_consensus_emits_event elapsed_seconds assertion --- orchestrator/routes/pipelines.py | 47 ++++++++++++++++---- orchestrator/tests/test_consensus_polling.py | 2 + 2 files changed, 40 insertions(+), 9 deletions(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index b04473314d..55f632e3fc 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -4559,9 +4559,15 @@ def _run_concurrent_phase( docker_client = spawner.docker all_logs: list[str] = [] has_failures = [False] # Mutable container for closure access + # Lock protects all_logs and has_failures mutations from the + # ThreadPoolExecutor threads in the timeout fallback path (step 6). + # The main polling loop is single-threaded, but the lock is cheap + # and makes the code safe regardless of GIL guarantees. + _logs_lock = threading.Lock() poll_interval = 5 # seconds - consensus_timeout = getattr(pipeline.config, "consensus_timeout_minutes", 30) * 60 + raw_timeout = getattr(pipeline.config, "consensus_timeout_minutes", 30) + consensus_timeout = max(raw_timeout, 1) * 60 # minimum 1 minute start_time = time.monotonic() objection_decision_created = False @@ -4572,7 +4578,6 @@ def _record_container_exit(exec_info: "StateAgentExecution", final_info: Contain """Capture logs and update pipeline state for an exited container.""" container_logs = "" if final_info.exit_code != 0: - has_failures[0] = True try: container_logs = docker_client.get_container_logs( exec_info.container_id, @@ -4581,9 +4586,12 @@ def _record_container_exit(exec_info: "StateAgentExecution", final_info: Contain except Exception: pass - all_logs.append( - f"--- {exec_info.role.value} (exit={final_info.exit_code}) ---\n{container_logs}" - ) + with _logs_lock: + if final_info.exit_code != 0: + has_failures[0] = True + all_logs.append( + f"--- {exec_info.role.value} (exit={final_info.exit_code}) ---\n{container_logs}" + ) if store is not None: try: @@ -4659,7 +4667,7 @@ def _update_agents_complete() -> None: ) consensus = {"is_complete": False, "has_objections": False, "blocking_agents": []} - # 2. Consensus reached — stop containers and return success + # 2. Consensus reached — stop containers and return if consensus.get("is_complete"): if _emit_event is not None: _emit_event( @@ -4671,15 +4679,28 @@ def _update_agents_complete() -> None: "Consensus reached, stopping containers", pipeline_id=pipeline_id, elapsed_seconds=round(elapsed, 1), + has_failures=has_failures[0], ) _update_agents_complete() _stop_running_containers() combined_logs = ( "\n".join(all_logs) if all_logs else "Consensus reached; phase complete." ) + # If any container failed before consensus was reached (e.g. OOM + # kill), propagate the failure even though remaining agents agreed. + # The HITL decision from handle_agent_failure is still pending but + # callers need a non-zero exit to trigger failure handling. + if has_failures[0]: + return 1, combined_logs return 0, combined_logs - # 3. Handle objections (create HITL decision once) + # 3. Handle objections (create HITL decision once). + # The decision is fire-and-forget: resolution is processed by the + # orchestrator's decision queue (outside this function). If the + # human selects "Override objections", the orchestrator updates + # agent readiness, which is picked up by check_consensus() on + # the next poll iteration. "Abort phase" triggers pipeline + # cancellation via a separate control path. if consensus.get("has_objections") and not objection_decision_created: try: pipeline.add_decision( @@ -4721,7 +4742,11 @@ def _update_agents_complete() -> None: exited_at=datetime.utcnow(), ) - if info.status in (ContainerStatus.EXITED, ContainerStatus.FAILED): + if info.status in ( + ContainerStatus.EXITED, + ContainerStatus.FAILED, + ContainerStatus.REMOVED, + ): exited_containers[exec_info.container_id] = info _record_container_exit(exec_info, info) @@ -4762,6 +4787,9 @@ def _update_agents_complete() -> None: pipeline_id=pipeline_id, timeout_minutes=consensus_timeout / 60, ) + # Fire-and-forget HITL decision: the orchestrator's decision + # queue handles resolution asynchronously. This function falls + # through to wait for remaining containers regardless. try: pipeline.add_decision( question=f"Consensus not reached after {int(consensus_timeout / 60)} minutes. How to proceed?", @@ -4801,7 +4829,8 @@ def _wait_remaining(exec_info): role=futures[future].role.value, error=str(exc), ) - has_failures[0] = True + with _logs_lock: + has_failures[0] = True combined_logs = "\n".join(all_logs) if has_failures[0]: diff --git a/orchestrator/tests/test_consensus_polling.py b/orchestrator/tests/test_consensus_polling.py index 2f5ae97074..73c19aa64a 100644 --- a/orchestrator/tests/test_consensus_polling.py +++ b/orchestrator/tests/test_consensus_polling.py @@ -248,6 +248,8 @@ def test_consensus_emits_event( **_CALL_ARGS, ) + # elapsed_seconds is 0.0 because mock_monotonic always returns 42.0, + # so start_time and the loop's time.monotonic() are identical. mock_emit.assert_any_call( EventType.CONSENSUS_REACHED, "issue-999", From 6195a489cf76820fa64a2e3e71226fbfbdc8fd07 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 12 Mar 2026 21:10:24 +0000 Subject: [PATCH 4/4] Add test for consensus-reached-with-failures path --- orchestrator/tests/test_consensus_polling.py | 77 ++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/orchestrator/tests/test_consensus_polling.py b/orchestrator/tests/test_consensus_polling.py index 73c19aa64a..231b353de7 100644 --- a/orchestrator/tests/test_consensus_polling.py +++ b/orchestrator/tests/test_consensus_polling.py @@ -705,6 +705,83 @@ def test_agent_failure_calls_handle_agent_failure( error="Container exited with code 137", ) + @patch("routes.pipelines.time.sleep") + @patch("routes.pipelines.time.monotonic") + @patch("routes.pipelines._emit_event") + @patch("routes.pipelines.get_pipeline_state_lock") + @patch("routes.pipelines._build_agent_prompt", return_value="test prompt") + @patch("concurrent_executor.ConcurrentPhaseExecutor", autospec=False) + def test_consensus_with_prior_failure_returns_nonzero( + self, MockExecutor, mock_prompt, mock_lock, mock_emit, mock_monotonic, mock_sleep + ): + """When a container fails but remaining agents reach consensus, returns (1, ...).""" + poll_count = [0] + + def _monotonic(): + return poll_count[0] * 5.0 + + mock_monotonic.side_effect = _monotonic + + executions = [ + _make_execution(AgentRole.CODER, "coder-1"), + _make_execution(AgentRole.TESTER, "tester-1"), + ] + + # Coder exits 137 (OOM kill) immediately; tester stays running + failed_coder = ContainerInfo( + container_id="coder-1", + container_name="issue-999-coder", + status=ContainerStatus.FAILED, + exit_code=137, + exited_at=datetime.utcnow(), + ) + running_tester = ContainerInfo( + container_id="tester-1", + container_name="issue-999-tester", + status=ContainerStatus.RUNNING, + exit_code=None, + ) + + def _get_info(cid): + if cid == "coder-1": + return failed_coder + return running_tester + + pipeline, mock_store, mock_spawner, mock_docker = _base_mocks(executions) + mock_docker.get_container_info.side_effect = _get_info + + def _check_consensus(): + poll_count[0] += 1 + # After handle_agent_failure removes coder, tester alone reaches consensus + if poll_count[0] >= 2: + return {"is_complete": True, "has_objections": False, "blocking_agents": []} + return {"is_complete": False, "has_objections": False, "blocking_agents": ["tester"]} + + mock_executor_instance = MagicMock() + mock_executor_instance.spawn_all.return_value = executions + mock_executor_instance.check_consensus.side_effect = _check_consensus + MockExecutor.return_value = mock_executor_instance + + mock_lock.return_value.__enter__ = MagicMock(return_value=None) + mock_lock.return_value.__exit__ = MagicMock(return_value=False) + + exit_code, logs = _run_concurrent_phase( + pipeline_id="issue-999", + pipeline=pipeline, + phase="implement", + spawner=mock_spawner, + store=mock_store, + **_CALL_ARGS, + ) + + # Consensus was reached, but a prior failure means the phase should fail + assert exit_code == 1 + # handle_agent_failure should have been called for the crashed coder + mock_executor_instance.handle_agent_failure.assert_called_once_with( + role="coder", + error="Container exited with code 137", + ) + @patch("routes.pipelines.time.sleep") @patch("routes.pipelines.time.monotonic") @patch("routes.pipelines.get_pipeline_state_lock")