diff --git a/orchestrator/api.py b/orchestrator/api.py index db6c189605..9d5a77d068 100644 --- a/orchestrator/api.py +++ b/orchestrator/api.py @@ -189,6 +189,23 @@ def main() -> None: error=str(reconcile_err), ) + # Start runtime container liveness monitor + try: + from container_monitor import ( + create_pipeline_reconciliation_handler, + get_container_monitor, + ) + + monitor = get_container_monitor() + monitor.add_handler(create_pipeline_reconciliation_handler(repo_path)) + monitor.start() + logger.info("Container monitor started for runtime liveness checks") + except Exception as monitor_err: + logger.warning( + "Container monitor startup failed", + error=str(monitor_err), + ) + if debug: # Use Flask's built-in server for development app.run(host=host, port=port, debug=True) diff --git a/orchestrator/container_monitor.py b/orchestrator/container_monitor.py index 9a836c3520..1df3c76105 100644 --- a/orchestrator/container_monitor.py +++ b/orchestrator/container_monitor.py @@ -293,3 +293,144 @@ def get_container_monitor() -> ContainerMonitor: if _container_monitor is None: _container_monitor = ContainerMonitor() return _container_monitor + + +def _reconcile_container_state(store: Any, container_info: ContainerInfo) -> bool: + """Update pipeline state for a single container that has exited. + + Scans all RUNNING pipelines for a container matching the given + container_info and marks the container and its agent as FAILED. + If any changes are made, the pipeline itself is marked FAILED. + + Uses per-pipeline locking (via ``get_pipeline_state_lock``) and + optimistic version checks (``expected_version``) to prevent race + conditions with concurrent state writers (e.g. agent signal handlers). + + A container belongs to exactly one pipeline, so the function returns + after updating the first matching pipeline. + + Args: + store: StateStore instance + container_info: Info about the exited/failed container + + Returns: + True if any pipeline state was updated + """ + from models import AgentExecutionStatus, PipelineStatus + from state_store import VersionConflictError, get_pipeline_state_lock + + try: + pipeline_ids: list[str] = store.list_pipelines() + except Exception as e: + logger.warning( + "Runtime reconciliation: could not list pipelines", + error=str(e), + ) + return False + + for pipeline_id in pipeline_ids: + with get_pipeline_state_lock(pipeline_id): + try: + pipeline = store.load_pipeline(pipeline_id) + except Exception: + continue + + if pipeline.status != PipelineStatus.RUNNING: + continue + + changed = False + + for phase_execution in pipeline.phases.values(): + if phase_execution.status != PipelineStatus.RUNNING: + continue + + for ci in phase_execution.containers: + if ci.container_id == container_info.container_id and ci.status == ContainerStatus.RUNNING: + logger.warning( + "Runtime reconciliation: container exited, marking FAILED", + pipeline_id=pipeline_id, + container_id=container_info.container_id[:12], + ) + ci.status = ContainerStatus.FAILED + ci.exit_code = container_info.exit_code if container_info.exit_code is not None else -1 + ci.exited_at = container_info.exited_at or datetime.utcnow() + changed = True + + for agent in phase_execution.agents: + if ( + agent.status == AgentExecutionStatus.RUNNING + and agent.container_id == container_info.container_id + ): + logger.warning( + "Runtime reconciliation: agent container exited, marking FAILED", + pipeline_id=pipeline_id, + agent_role=str(agent.role), + container_id=container_info.container_id[:12], + ) + agent.status = AgentExecutionStatus.FAILED + agent.completed_at = datetime.utcnow() + agent.error = ( + "Container exited unexpectedly during execution — " + "detected by runtime container monitor" + ) + changed = True + + if changed: + pipeline.status = PipelineStatus.FAILED + pipeline.error = ( + "Pipeline marked FAILED: agent container exited unexpectedly " + "during execution. Restart via POST /pipelines/{id}/start." + ) + try: + store.save_pipeline( + pipeline, + expected_version=pipeline.version, + ) + logger.warning( + "Runtime reconciliation: pipeline marked FAILED", + pipeline_id=pipeline_id, + ) + return True + except VersionConflictError: + logger.warning( + "Runtime reconciliation: version conflict, skipping " + "(concurrent writer updated pipeline)", + pipeline_id=pipeline_id, + ) + return False + except Exception as e: + logger.error( + "Runtime reconciliation: could not save pipeline", + pipeline_id=pipeline_id, + error=str(e), + ) + return False + + return False + + +def create_pipeline_reconciliation_handler(repo_path: str) -> EventHandler: + """Create handler that updates pipeline state when containers exit. + + The handler is invoked by the ContainerMonitor whenever a container + state change is detected. Only FAILED events (non-zero exit) trigger + reconciliation — STOPPED (exit code 0) represents a graceful exit + and should not mark pipelines as failed. + + Args: + repo_path: Path to the repository (for StateStore access) + + Returns: + Event handler function + """ + + def handler(event: ContainerEvent) -> None: + if event.event_type != ContainerEvent.FAILED: + return + + from state_store import get_state_store + + store = get_state_store(repo_path) + _reconcile_container_state(store, event.container_info) + + return handler diff --git a/orchestrator/multi_agent.py b/orchestrator/multi_agent.py index 2060f13300..25deb3ccd0 100644 --- a/orchestrator/multi_agent.py +++ b/orchestrator/multi_agent.py @@ -510,17 +510,28 @@ def execute_all_waves( self, on_wave_complete: Callable[[AgentWave], None] | None = None, agent_prompts: dict[AgentRole, str] | None = None, + max_waves: int = 5, ) -> list[AgentWave]: """Execute all waves until completion or failure. Args: on_wave_complete: Optional callback after each wave agent_prompts: Role-to-prompt mapping (required when using spawn_fn) + max_waves: Safety cap on number of wave iterations (default: 5) Returns: List of completed waves """ + waves_executed = 0 while True: + if waves_executed >= max_waves: + logger.warning( + "Max waves reached, stopping execution", + pipeline_id=self.pipeline.id, + max_waves=max_waves, + ) + break + wave = self.get_next_wave() if not wave: break @@ -529,6 +540,7 @@ def execute_all_waves( wave, agent_prompts=agent_prompts, ) + waves_executed += 1 if on_wave_complete: on_wave_complete(completed) diff --git a/orchestrator/tests/test_container_monitor.py b/orchestrator/tests/test_container_monitor.py new file mode 100644 index 0000000000..4a1223bc99 --- /dev/null +++ b/orchestrator/tests/test_container_monitor.py @@ -0,0 +1,366 @@ +"""Tests for container_monitor runtime reconciliation handler.""" + +import sys +from datetime import datetime +from pathlib import Path +from unittest.mock import MagicMock, patch + +_orchestrator_path = Path(__file__).parent.parent +if str(_orchestrator_path) not in sys.path: + sys.path.insert(0, str(_orchestrator_path)) + +_shared_path = Path(__file__).parent.parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +# Mock docker before importing modules that depend on it +sys.modules.setdefault("docker", MagicMock()) +sys.modules.setdefault("docker.errors", MagicMock()) +sys.modules.setdefault("docker.types", MagicMock()) + +from container_monitor import ( + ContainerEvent, + ContainerMonitor, + _reconcile_container_state, + create_pipeline_reconciliation_handler, +) +from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + ContainerInfo, + ContainerStatus, + Pipeline, + PipelinePhase, + PipelineStatus, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_pipeline_with_running_agent(container_id: str = "abc123") -> Pipeline: + """Return a RUNNING pipeline with one RUNNING coder agent.""" + pipeline = Pipeline( + id="issue-99", + issue_number=99, + repo="owner/repo", + branch="egg/issue-99", + mode="issue", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + phase = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT) + phase.status = PipelineStatus.RUNNING + phase.started_at = datetime.utcnow() + + phase.containers.append( + ContainerInfo( + container_id=container_id, + container_name="egg-coder-issue-99", + status=ContainerStatus.RUNNING, + started_at=datetime.utcnow(), + ) + ) + phase.agents.append( + AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.RUNNING, + container_id=container_id, + started_at=datetime.utcnow(), + ) + ) + return pipeline + + +def _make_store(pipeline: Pipeline) -> MagicMock: + store = MagicMock() + store.list_pipelines.return_value = [pipeline.id] + store.load_pipeline.return_value = pipeline + return store + + +def _make_container_info(container_id: str, exit_code: int = 1) -> ContainerInfo: + """Build a ContainerInfo for a container that has exited.""" + return ContainerInfo( + container_id=container_id, + container_name=f"egg-container-{container_id[:8]}", + status=ContainerStatus.EXITED, + exit_code=exit_code, + exited_at=datetime.utcnow(), + ) + + +# --------------------------------------------------------------------------- +# Tests: _reconcile_container_state +# --------------------------------------------------------------------------- + + +class TestReconcileContainerState: + """Tests for the _reconcile_container_state helper.""" + + def test_marks_pipeline_failed_when_container_exits(self): + """A RUNNING pipeline whose container exits is marked FAILED.""" + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent(container_id) + store = _make_store(pipeline) + exited_info = _make_container_info(container_id) + + result = _reconcile_container_state(store, exited_info) + + assert result is True + assert pipeline.status == PipelineStatus.FAILED + assert pipeline.error is not None + store.save_pipeline.assert_called_once_with( + pipeline, expected_version=pipeline.version, + ) + + def test_marks_agent_failed_when_container_exits(self): + """The agent whose container exited is marked FAILED with an error.""" + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent(container_id) + store = _make_store(pipeline) + exited_info = _make_container_info(container_id) + + _reconcile_container_state(store, exited_info) + + phase = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT) + agent = phase.agents[0] + assert agent.status == AgentExecutionStatus.FAILED + assert agent.error is not None + assert "runtime container monitor" in agent.error + assert agent.completed_at is not None + + def test_marks_container_info_failed(self): + """The ContainerInfo entry is marked FAILED with exit_code from event.""" + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent(container_id) + store = _make_store(pipeline) + exited_info = _make_container_info(container_id, exit_code=137) + + _reconcile_container_state(store, exited_info) + + phase = pipeline.get_phase_execution(PipelinePhase.IMPLEMENT) + ci = phase.containers[0] + assert ci.status == ContainerStatus.FAILED + assert ci.exit_code == 137 + + def test_ignores_untracked_containers(self): + """A container not tracked by any pipeline is silently ignored.""" + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent("other_container") + store = _make_store(pipeline) + exited_info = _make_container_info(container_id) + + result = _reconcile_container_state(store, exited_info) + + assert result is False + assert pipeline.status == PipelineStatus.RUNNING + store.save_pipeline.assert_not_called() + + def test_ignores_non_running_pipelines(self): + """A COMPLETE pipeline is not affected by container exits.""" + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent(container_id) + pipeline.status = PipelineStatus.COMPLETE + store = _make_store(pipeline) + exited_info = _make_container_info(container_id) + + result = _reconcile_container_state(store, exited_info) + + assert result is False + store.save_pipeline.assert_not_called() + + def test_handles_store_list_error(self): + """Returns False without crashing when store.list_pipelines fails.""" + store = MagicMock() + store.list_pipelines.side_effect = Exception("Store unavailable") + exited_info = _make_container_info("some_id") + + result = _reconcile_container_state(store, exited_info) + + assert result is False + + def test_handles_store_load_error(self): + """Skips pipelines that fail to load.""" + store = MagicMock() + store.list_pipelines.return_value = ["bad-pipeline"] + store.load_pipeline.side_effect = Exception("corrupt state") + exited_info = _make_container_info("some_id") + + result = _reconcile_container_state(store, exited_info) + + assert result is False + + @patch("state_store.get_pipeline_state_lock") + def test_acquires_pipeline_lock(self, mock_get_lock): + """Reconciliation acquires the per-pipeline lock during load-modify-save.""" + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent(container_id) + store = _make_store(pipeline) + exited_info = _make_container_info(container_id) + + mock_lock = MagicMock() + mock_get_lock.return_value = mock_lock + + _reconcile_container_state(store, exited_info) + + mock_get_lock.assert_called_once_with(pipeline.id) + mock_lock.__enter__.assert_called_once() + mock_lock.__exit__.assert_called_once() + + def test_handles_version_conflict(self): + """Returns False on VersionConflictError (concurrent writer won).""" + from state_store import VersionConflictError + + container_id = "dead_container_xyz" + pipeline = _make_pipeline_with_running_agent(container_id) + store = _make_store(pipeline) + store.save_pipeline.side_effect = VersionConflictError("conflict") + exited_info = _make_container_info(container_id) + + result = _reconcile_container_state(store, exited_info) + + assert result is False + + +# --------------------------------------------------------------------------- +# Tests: create_pipeline_reconciliation_handler +# --------------------------------------------------------------------------- + + +class TestCreatePipelineReconciliationHandler: + """Tests for the handler factory function.""" + + @patch("state_store.get_state_store") + def test_handler_calls_reconcile_on_failed_event(self, mock_get_store): + """Handler processes FAILED events (non-zero exit).""" + container_id = "dead_container" + pipeline = _make_pipeline_with_running_agent(container_id) + store = _make_store(pipeline) + mock_get_store.return_value = store + + handler = create_pipeline_reconciliation_handler("/repo") + event = ContainerEvent( + ContainerEvent.FAILED, + _make_container_info(container_id), + ) + handler(event) + + assert pipeline.status == PipelineStatus.FAILED + + @patch("state_store.get_state_store") + def test_handler_ignores_started_event(self, mock_get_store): + """Handler does NOT process STARTED events.""" + handler = create_pipeline_reconciliation_handler("/repo") + event = ContainerEvent( + ContainerEvent.STARTED, + _make_container_info("some_id"), + ) + handler(event) + + mock_get_store.assert_not_called() + + @patch("state_store.get_state_store") + def test_handler_ignores_stopped_event(self, mock_get_store): + """Handler does NOT process STOPPED events (graceful exit code 0).""" + handler = create_pipeline_reconciliation_handler("/repo") + event = ContainerEvent( + ContainerEvent.STOPPED, + _make_container_info("some_id", exit_code=0), + ) + handler(event) + + mock_get_store.assert_not_called() + + @patch("state_store.get_state_store") + def test_handler_ignores_exited_event(self, mock_get_store): + """Handler does NOT process EXITED events (never emitted by monitor).""" + handler = create_pipeline_reconciliation_handler("/repo") + event = ContainerEvent( + ContainerEvent.EXITED, + _make_container_info("some_id"), + ) + handler(event) + + mock_get_store.assert_not_called() + + +# --------------------------------------------------------------------------- +# Tests: ContainerMonitor integration +# --------------------------------------------------------------------------- + + +class TestContainerMonitorDetection: + """Tests that the monitor detects container state changes.""" + + def test_monitor_detects_exited_container(self): + """Monitor emits FAILED event when a running container exits with non-zero.""" + mock_docker = MagicMock() + container_id = "test_container_123" + + # First call: container is running + running_info = ContainerInfo( + container_id=container_id, + container_name="egg-test", + status=ContainerStatus.RUNNING, + started_at=datetime.utcnow(), + ) + # Second call: container has exited + exited_info = ContainerInfo( + container_id=container_id, + container_name="egg-test", + status=ContainerStatus.EXITED, + exit_code=1, + exited_at=datetime.utcnow(), + ) + mock_docker.list_containers.side_effect = [ + [running_info], + [exited_info], + ] + + monitor = ContainerMonitor(docker_client=mock_docker, check_interval=1) + events_received: list[ContainerEvent] = [] + monitor.add_handler(lambda e: events_received.append(e)) + + # Simulate two check cycles + monitor._check_all_containers() # First: STARTED + monitor._check_all_containers() # Second: FAILED (non-zero exit) + + event_types = [e.event_type for e in events_received] + assert ContainerEvent.STARTED in event_types + assert ContainerEvent.FAILED in event_types + + def test_monitor_emits_stopped_for_zero_exit(self): + """Monitor emits STOPPED event when a running container exits with code 0.""" + mock_docker = MagicMock() + container_id = "test_container_456" + + running_info = ContainerInfo( + container_id=container_id, + container_name="egg-test", + status=ContainerStatus.RUNNING, + started_at=datetime.utcnow(), + ) + exited_info = ContainerInfo( + container_id=container_id, + container_name="egg-test", + status=ContainerStatus.EXITED, + exit_code=0, + exited_at=datetime.utcnow(), + ) + mock_docker.list_containers.side_effect = [ + [running_info], + [exited_info], + ] + + monitor = ContainerMonitor(docker_client=mock_docker, check_interval=1) + events_received: list[ContainerEvent] = [] + monitor.add_handler(lambda e: events_received.append(e)) + + monitor._check_all_containers() + monitor._check_all_containers() + + event_types = [e.event_type for e in events_received] + assert ContainerEvent.STOPPED in event_types diff --git a/orchestrator/tests/test_multi_agent.py b/orchestrator/tests/test_multi_agent.py new file mode 100644 index 0000000000..04b70ecc92 --- /dev/null +++ b/orchestrator/tests/test_multi_agent.py @@ -0,0 +1,152 @@ +"""Tests for multi_agent.MultiAgentExecutor.execute_all_waves() max_waves cap.""" + +import sys +from pathlib import Path +from unittest.mock import MagicMock + +_orchestrator_path = Path(__file__).parent.parent +if str(_orchestrator_path) not in sys.path: + sys.path.insert(0, str(_orchestrator_path)) + +_shared_path = Path(__file__).parent.parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +# Mock docker before importing modules that depend on it +sys.modules.setdefault("docker", MagicMock()) +sys.modules.setdefault("docker.errors", MagicMock()) +sys.modules.setdefault("docker.types", MagicMock()) + +from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + Pipeline, + PipelinePhase, + PipelineStatus, +) +from multi_agent import MultiAgentExecutor + + +def _make_pipeline() -> Pipeline: + """Create a minimal RUNNING pipeline.""" + return Pipeline( + id="issue-99", + issue_number=99, + repo="owner/repo", + branch="egg/issue-99", + mode="issue", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.IMPLEMENT, + ) + + +class TestExecuteAllWavesMaxCap: + """Tests for the max_waves safety cap on execute_all_waves().""" + + def test_stops_after_max_waves(self): + """execute_all_waves stops after max_waves even if dispatcher keeps returning agents.""" + pipeline = _make_pipeline() + mock_dispatcher = MagicMock() + mock_dispatcher.save_contract = MagicMock() + + # get_agents_to_run always returns something -> infinite loop without cap + mock_dispatcher.get_agents_to_run.return_value = [AgentRole.CODER] + mock_dispatcher.get_next_dispatch.return_value = MagicMock(wave_number=1) + + # Use a spawn_fn to avoid needing Docker + call_count = 0 + + def fake_spawn(role, prompt, extra_env): + nonlocal call_count + call_count += 1 + return (0, "ok") + + executor = MultiAgentExecutor( + pipeline=pipeline, + repo_path=Path("/repo"), + dispatcher=mock_dispatcher, + spawn_fn=fake_spawn, + ) + + max_waves = 5 + waves = executor.execute_all_waves( + agent_prompts={AgentRole.CODER: "do work"}, + max_waves=max_waves, + ) + + assert len(waves) == max_waves + mock_dispatcher.save_contract.assert_called_once() + + def test_default_max_waves_is_5(self): + """Default max_waves is 5 (verify signature default).""" + import inspect + + sig = inspect.signature(MultiAgentExecutor.execute_all_waves) + default = sig.parameters["max_waves"].default + assert default == 5 + + def test_stops_before_max_when_no_more_waves(self): + """Stops normally when dispatcher returns no more agents before max_waves.""" + pipeline = _make_pipeline() + mock_dispatcher = MagicMock() + mock_dispatcher.save_contract = MagicMock() + + # Return agents for 2 waves, then None + mock_dispatcher.get_agents_to_run.side_effect = [ + [AgentRole.CODER], + [AgentRole.CODER], + [], # No more agents + ] + mock_dispatcher.get_next_dispatch.return_value = MagicMock(wave_number=1) + + def fake_spawn(role, prompt, extra_env): + return (0, "ok") + + executor = MultiAgentExecutor( + pipeline=pipeline, + repo_path=Path("/repo"), + dispatcher=mock_dispatcher, + spawn_fn=fake_spawn, + ) + + waves = executor.execute_all_waves( + agent_prompts={AgentRole.CODER: "do work"}, + max_waves=10, + ) + + assert len(waves) == 2 + + def test_stops_on_failure_before_max_waves(self): + """Stops on wave failure before reaching max_waves.""" + pipeline = _make_pipeline() + mock_dispatcher = MagicMock() + mock_dispatcher.save_contract = MagicMock() + + mock_dispatcher.get_agents_to_run.return_value = [AgentRole.CODER] + mock_dispatcher.get_next_dispatch.return_value = MagicMock(wave_number=1) + + # fail_agent returns a FAILED execution + mock_dispatcher.fail_agent.return_value = AgentExecution( + role=AgentRole.CODER, + status=AgentExecutionStatus.FAILED, + error="Exit code: 1", + ) + + def fail_spawn(role, prompt, extra_env): + return (1, "error") + + executor = MultiAgentExecutor( + pipeline=pipeline, + repo_path=Path("/repo"), + dispatcher=mock_dispatcher, + spawn_fn=fail_spawn, + ) + + waves = executor.execute_all_waves( + agent_prompts={AgentRole.CODER: "do work"}, + max_waves=10, + ) + + # Should stop after first wave due to failure + assert len(waves) == 1