diff --git a/docs/guides/coordinator.md b/docs/guides/coordinator.md index 7f985b4f00..969b9b1881 100644 --- a/docs/guides/coordinator.md +++ b/docs/guides/coordinator.md @@ -143,7 +143,6 @@ Common error codes: - **409** — Conflict. Possible causes: - Phase advancement blocked because no contract exists before implement/pr phase - Agent spawn rejected because the pipeline has no contract in implement/pr phase - - Role's dependencies have not yet completed (e.g., spawning `tester` before `coder` is done) - **429** — Guardrail limit exceeded (max agents or max retries per role) - **500** — Internal error (container spawn failure, etc.) @@ -167,8 +166,6 @@ The `role` must be a valid `AgentRole` **and** must be appropriate for the curre | `implement` | `coder`, `tester`, `documenter`, `integrator`, `reviewer_code`, `reviewer_contract` | | `pr`, `coordinator` | Any role (no phase-role restriction) | -**Dependency enforcement**: The orchestrator checks that the role's declared dependencies are complete before spawning. For example, `tester` depends on `coder` — spawning `tester` before `coder` has a `"complete"` status returns HTTP 409 with a `missing_dependencies` list. Dependencies are defined per-role in `shared/egg_contracts/agent_roles.py`. Spawn dependencies across the coordinator's full agent history are checked (including agents from prior phases). - **Contract enforcement**: Spawning any agent in the `implement` or `pr` phase when the pipeline has no contract (`contract_synced: false`) returns HTTP 409. Contracts are auto-created at pipeline startup; a 409 here indicates that creation failed — check orchestrator logs. Returns 429 if guardrail limits are exceeded. The response includes the `spawn_record` with the assigned `retry_number` (0 for the first spawn of a given role, incremented for each subsequent spawn of the same role). @@ -397,8 +394,6 @@ The coordinator runs with `phase="coordinator"` — a special phase value distin **Agent spawn rejected (HTTP 400 — invalid phase-role)**: The role is not valid for the current pipeline phase. Check the current phase via `egg-orch coordinator state ` and spawn a role that matches. For example, `coder` is only valid in the `implement` phase; `refiner` is only valid in the `refine` phase. Phases `pr` and `coordinator` have no restriction. -**Agent spawn rejected (HTTP 409 — missing dependencies)**: The role's declared dependencies have not yet completed. Check `egg-orch coordinator state ` and look at `completed_agents` to see which roles have finished. Spawn the dependency roles first (e.g., spawn `coder` before `tester`). The error response includes a `missing_dependencies` field listing which roles still need to complete. - **Agent spawn rejected (HTTP 409 — no contract)**: The pipeline is in the implement or pr phase but has no contract. Run `egg-orch pipeline get ` and check `contract_synced`. If false, contract creation at startup failed — see the troubleshooting entry for "Phase advance blocked (HTTP 409 — no contract)" below. **Agent spawn rejected (HTTP 429)**: Check guardrail limits via `egg-orch coordinator state `. The `guardrail_counters` section shows current counts vs. configured limits. diff --git a/orchestrator/Dockerfile b/orchestrator/Dockerfile index 78d2b0d7fd..f592000f6a 100644 --- a/orchestrator/Dockerfile +++ b/orchestrator/Dockerfile @@ -16,11 +16,12 @@ COPY orchestrator/*.py ./ COPY orchestrator/routes/ ./routes/ COPY orchestrator/health_checks/ ./health_checks/ -# Copy shared modules (egg_logging, egg_config, egg_contracts, egg_container) +# Copy shared modules (egg_logging, egg_config, egg_contracts, egg_container, egg_agent) COPY shared/egg_logging/ ./egg_logging/ COPY shared/egg_config/ ./egg_config/ COPY shared/egg_contracts/ ./egg_contracts/ COPY shared/egg_container/ ./egg_container/ +COPY shared/egg_agent/ ./egg_agent/ # Copy shared prompt criteria files COPY shared/prompts/ ./prompts/ diff --git a/orchestrator/container_spawner.py b/orchestrator/container_spawner.py index 1846763b7a..abb4156764 100644 --- a/orchestrator/container_spawner.py +++ b/orchestrator/container_spawner.py @@ -28,6 +28,11 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] return logging.getLogger(name) +from sandbox_template import ( + ORCHESTRATOR_ISOLATED_IP, + ORCHESTRATOR_PORT, +) + try: from egg_config import ( EGG_CONTAINER_IP, @@ -35,6 +40,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] GATEWAY_EXTERNAL_IP, GATEWAY_ISOLATED_IP, GATEWAY_PORT, + ORCHESTRATOR_EXTERNAL_IP, ) from egg_config import ( EGG_EXTERNAL_NETWORK as _DEFAULT_EXTERNAL_NETWORK, @@ -50,6 +56,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] GATEWAY_PORT = 9848 # noqa: EGG002 GATEWAY_ISOLATED_IP = "172.32.0.2" GATEWAY_EXTERNAL_IP = "172.33.0.2" + ORCHESTRATOR_EXTERNAL_IP = "172.33.0.3" # Allow override via environment for test stacks with non-standard network names EGG_ISOLATED_NETWORK = os.environ.get("EGG_ISOLATED_NETWORK", _DEFAULT_ISOLATED_NETWORK) @@ -371,11 +378,16 @@ def spawn_agent_container( # CONTAINER_ID must match the worktree container_id so the gateway # git proxy can map /home/egg/repos/ to the correct worktree # at /home/egg/.egg-worktrees//. + orchestrator_host = ( + ORCHESTRATOR_ISOLATED_IP if mode == "private" else ORCHESTRATOR_EXTERNAL_IP + ) + orchestrator_url = f"http://{orchestrator_host}:{ORCHESTRATOR_PORT}" spawner_env: dict[str, str] = { "CONTAINER_ID": pipeline_id, "EGG_REPO_PATH": "/home/egg/repos", "EGG_AGENT_ROLE": agent_role.value, "EGG_PIPELINE_ID": pipeline_id, + "EGG_ORCHESTRATOR_URL": orchestrator_url, } if issue_number is not None: spawner_env["EGG_ISSUE_NUMBER"] = str(issue_number) diff --git a/orchestrator/mcp_tools.py b/orchestrator/mcp_tools.py index e02b4e2bda..2fa70dfee6 100644 --- a/orchestrator/mcp_tools.py +++ b/orchestrator/mcp_tools.py @@ -173,7 +173,11 @@ def handle_tool_call(self, tool_name: str, arguments: dict[str, Any]) -> dict[st return {"error": str(e)} def _make_request( - self, endpoint: str, method: str = "GET", data: dict[str, Any] | None = None + self, + endpoint: str, + method: str = "GET", + data: dict[str, Any] | None = None, + timeout: int = 30, ) -> dict[str, Any]: """Make HTTP request to orchestrator.""" import json @@ -186,7 +190,7 @@ def _make_request( opener = build_opener(ProxyHandler({})) req = Request(url, data=body, headers=headers, method=method) - with opener.open(req, timeout=30) as response: + with opener.open(req, timeout=timeout) as response: return json.loads(response.read().decode()) def _handle_submit_task(self, args: dict[str, Any]) -> dict[str, Any]: @@ -322,5 +326,6 @@ def _handle_cancel_task(self, args: dict[str, Any]) -> dict[str, Any]: f"/api/v1/pipelines/{task_id}", method="PATCH", data=data, + timeout=120, ) return result diff --git a/orchestrator/routes/coordinator.py b/orchestrator/routes/coordinator.py index 4ece6dd2ae..40b9098bec 100644 --- a/orchestrator/routes/coordinator.py +++ b/orchestrator/routes/coordinator.py @@ -35,7 +35,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from consensus_wrapper import build_consensus_wrapped_command from container_spawner import ContainerSpawnError, get_container_spawner from decision_queue import get_decision_queue -from egg_contracts.agent_roles import get_role_definition, get_roles_for_phase +from egg_contracts.agent_roles import get_roles_for_phase from events import EventType, emit_event from gateway_client import GatewayError, get_gateway_client from models import ( @@ -227,39 +227,6 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]: }, ) - # Check role dependencies — reviewer roles must wait for - # their primary agents to complete - try: - role_def = get_role_definition(role_str) - if role_def.dependencies: - coord_state = pipeline.coordinator_state or CoordinatorState() - completed_roles = { - s.role.value for s in coord_state.agents_spawned if s.status == "complete" - } - missing = [ - dep.value - for dep in role_def.dependencies - if dep.value not in completed_roles - ] - if missing: - return make_error_response( - f"Cannot spawn '{role_str}': dependencies not yet complete: " - f"{missing}. These roles must finish before '{role_str}' can start.", - status_code=409, - details={ - "role": role_str, - "missing_dependencies": missing, - "completed_roles": sorted(completed_roles), - }, - ) - except (ValueError, KeyError): - # Role not found in egg_contracts definitions — allow spawn - # but warn since this bypasses a safety check - logger.warning( - "No role definition found for dependency check, allowing spawn", - role=role_str, - ) - # Validate role is appropriate for the current phase current_phase_str = pipeline.current_phase.value try: diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 91c0bcd52e..d8d8e35052 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -37,8 +37,10 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from ..decision_queue import get_decision_queue from ..docker_client import ContainerNotFoundError, ContainerOperationError, DockerClientError from ..models import ( + AgentExecutionStatus, AgentRole, AggregatedReviewResult, + ContainerStatus, CycleTiming, Pipeline, PipelinePhase, @@ -63,9 +65,11 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] DockerClientError, ) from models import ( # type: ignore + AgentExecutionStatus, AgentRole, AggregatedReviewResult, ComplexityTier, + ContainerStatus, CycleTiming, DecisionStatus, Pipeline, @@ -554,6 +558,62 @@ def create_pipeline() -> tuple[Response, int]: return make_error_response(f"Failed to create pipeline: {e}", status_code=500) +def _mark_pipeline_records_terminated( + store: "StateStore", + pipeline_id: str, +) -> "Pipeline": + """Mark all running containers and agents as stopped after pipeline termination. + + Called when a pipeline transitions to a terminal state (cancelled or failed). + After Docker containers are force-removed, the pipeline state still shows + them as "running". This reloads the latest state from the store (to avoid + overwriting coordinator updates made between the status change and container + cleanup), marks running records as stopped, and saves. + + Returns the updated pipeline so the caller can use it in the response. + """ + pipeline = store.load_pipeline(pipeline_id) + now = datetime.utcnow() + changed = False + + for phase_exec in pipeline.phases.values(): + for container in phase_exec.containers: + if container.status in ( + ContainerStatus.PENDING, + ContainerStatus.CREATING, + ContainerStatus.RUNNING, + ): + container.status = ContainerStatus.REMOVED + container.exited_at = now + changed = True + + for agent in phase_exec.agents: + if agent.status in ( + AgentExecutionStatus.PENDING, + AgentExecutionStatus.RUNNING, + ): + agent.status = AgentExecutionStatus.FAILED + agent.completed_at = now + agent.error = f"Pipeline {pipeline.status.value}" + changed = True + + if pipeline.coordinator_state: + for spawn_record in pipeline.coordinator_state.agents_spawned: + if spawn_record.status == "running": + spawn_record.status = "cancelled" + spawn_record.completed_at = now + changed = True + + if changed: + store.save_pipeline(pipeline) + logger.info( + "Synced pipeline state after termination", + pipeline_id=pipeline_id, + ) + + return pipeline + + @pipelines_bp.route("/", methods=["PATCH"]) def update_pipeline(pipeline_id: str) -> tuple[Response, int]: """ @@ -632,6 +692,18 @@ def update_pipeline(pipeline_id: str) -> tuple[Response, int]: exc_info=True, ) + # Sync pipeline state: reload latest state (coordinator may have + # written updates between status change and container cleanup), + # mark all running records as stopped, and re-save. + try: + pipeline = _mark_pipeline_records_terminated(store, pipeline_id) + except Exception as e: + logger.warning( + "Failed to sync pipeline state after termination", + pipeline_id=pipeline_id, + error=str(e), + ) + logger.info("Pipeline updated", pipeline_id=pipeline_id) return make_success_response( diff --git a/orchestrator/tests/test_coordinator_gaps.py b/orchestrator/tests/test_coordinator_gaps.py index f14002ed11..05b684f5d4 100644 --- a/orchestrator/tests/test_coordinator_gaps.py +++ b/orchestrator/tests/test_coordinator_gaps.py @@ -827,6 +827,7 @@ def test_cancel_task_passes_reason(self): "/api/v1/pipelines/issue-42", method="PATCH", data={"status": "cancelled", "reason": "No longer needed"}, + timeout=120, ) def test_provide_input_calls_correct_endpoint(self): diff --git a/orchestrator/tests/test_coordinator_routes_functional.py b/orchestrator/tests/test_coordinator_routes_functional.py index 07a4c117ab..c45c427326 100644 --- a/orchestrator/tests/test_coordinator_routes_functional.py +++ b/orchestrator/tests/test_coordinator_routes_functional.py @@ -1449,88 +1449,6 @@ def test_spawn_in_coordinator_phase_without_role_mapping_allowed( assert response.status_code == 200 -# ── Dependency validation tests ──────────────────────────────────── - - -class TestSpawnDependencyValidation: - """Spawn must be blocked when role dependencies are not complete.""" - - @patch("routes.coordinator.get_state_store") - @patch("routes.coordinator.get_pipeline_state_lock") - @patch("routes.coordinator.get_repo_path") - def test_spawn_blocked_when_dependency_not_complete( - self, mock_repo, mock_lock, mock_store_fn, client - ): - """Spawning reviewer_code without its dependencies complete returns 409.""" - mock_repo.return_value = Path("/tmp/repo") - mock_lock.return_value.__enter__ = MagicMock() - mock_lock.return_value.__exit__ = MagicMock(return_value=False) - - # Only integrator completed — task_planner and risk_analyst missing - pipeline = _make_pipeline( - phase=PipelinePhase.IMPLEMENT, - coordinator_state=CoordinatorState( - agents_spawned=[ - AgentSpawnRecord(role=AgentRole.INTEGRATOR, status="complete"), - AgentSpawnRecord(role=AgentRole.TASK_PLANNER, status="running"), - ], - ), - ) - store = MagicMock() - store.load_pipeline.return_value = pipeline - mock_store_fn.return_value = store - - response = client.post( - "/api/v1/pipelines/test-pipeline/coordinator/spawn", - json={"role": "reviewer_code"}, - ) - assert response.status_code == 409 - body = response.get_json() - assert "dependencies not yet complete" in body["message"] - assert "missing_dependencies" in body["details"] - - @patch("routes.coordinator.emit_event") - @patch("routes.coordinator.get_container_spawner") - @patch("routes.coordinator.get_state_store") - @patch("routes.coordinator.get_pipeline_state_lock") - @patch("routes.coordinator.get_repo_path") - def test_spawn_allowed_when_dependencies_complete( - self, mock_repo, mock_lock, mock_store_fn, mock_spawner_fn, mock_emit, client - ): - """Spawning tester succeeds when coder has completed.""" - mock_repo.return_value = Path("/tmp/repo") - mock_lock.return_value.__enter__ = MagicMock() - mock_lock.return_value.__exit__ = MagicMock(return_value=False) - - # tester depends on coder — coder is complete - pipeline = _make_pipeline( - phase=PipelinePhase.IMPLEMENT, - coordinator_state=CoordinatorState( - agents_spawned=[ - AgentSpawnRecord(role=AgentRole.CODER, status="complete"), - ], - ), - ) - store = MagicMock() - store.load_pipeline.return_value = pipeline - mock_store_fn.return_value = store - - spawner = MagicMock() - spawned = MagicMock() - spawned.container_info = ContainerInfo( - container_id="tst123", container_name="egg-test-tester" - ) - spawner.spawn_agent_container.return_value = spawned - mock_spawner_fn.return_value = spawner - - response = client.post( - "/api/v1/pipelines/test-pipeline/coordinator/spawn", - json={"role": "tester"}, - ) - assert response.status_code == 200 - assert response.get_json()["data"]["role"] == "tester" - - # ── Spawn contract enforcement tests ────────────────────────────── diff --git a/orchestrator/tests/test_pipelines_api.py b/orchestrator/tests/test_pipelines_api.py index a6262fb43e..ee3baba300 100644 --- a/orchestrator/tests/test_pipelines_api.py +++ b/orchestrator/tests/test_pipelines_api.py @@ -8,7 +8,20 @@ import pytest from flask import Flask -from models import ContainerInfo, PhaseExecution, Pipeline, PipelinePhase +from models import ( + AgentExecution, + AgentExecutionStatus, + AgentRole, + AgentSpawnRecord, + ContainerInfo, + ContainerStatus, + CoordinatorState, + GuardrailCounters, + PhaseExecution, + Pipeline, + PipelinePhase, + PipelineStatus, +) from routes.pipelines import pipelines_bp @@ -230,3 +243,192 @@ def test_delete_pipeline_succeeds_when_gateway_client_raises( # Pipeline should still be deleted despite branch cleanup exception mock_store.delete_pipeline.assert_called_once_with("test-pipeline") + + +def _make_cancellable_pipeline(pipeline_id="test-pipeline"): + """Create a pipeline with running containers and agents for cancellation tests.""" + pipeline = Pipeline( + id=pipeline_id, + issue_number=42, + repo="owner/repo", + branch="egg/test", + status=PipelineStatus.RUNNING, + current_phase=PipelinePhase.REFINE, + ) + pipeline.phases = { + "refine": PhaseExecution( + phase=PipelinePhase.REFINE, + status=PipelineStatus.RUNNING, + containers=[ + ContainerInfo( + container_id="coordinator-aaa", + container_name="egg-test-coordinator", + agent_role=AgentRole.COORDINATOR, + status=ContainerStatus.RUNNING, + ), + ContainerInfo( + container_id="refiner-bbb", + container_name="egg-test-refiner", + agent_role=AgentRole.REFINER, + status=ContainerStatus.RUNNING, + ), + ], + agents=[ + AgentExecution( + role=AgentRole.COORDINATOR, + status=AgentExecutionStatus.RUNNING, + container_id="coordinator-aaa", + ), + AgentExecution( + role=AgentRole.REFINER, + status=AgentExecutionStatus.RUNNING, + container_id="refiner-bbb", + ), + ], + ), + } + pipeline.coordinator_state = CoordinatorState( + agents_spawned=[ + AgentSpawnRecord( + role=AgentRole.REFINER, + status="running", + container_id="refiner-bbb", + task_context="Refine issue", + ), + ], + guardrail_counters=GuardrailCounters(), + ) + return pipeline + + +class TestTerminatedPipelineSyncsState: + """Tests that terminating a pipeline marks running containers/agents as stopped.""" + + @patch("routes.pipelines.get_decision_queue") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_cancel_marks_running_containers_as_removed( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_dq_fn, client + ): + mock_repo.return_value = "/repo" + pipeline = _make_cancellable_pipeline() + + mock_store = MagicMock() + mock_store.update_pipeline.return_value = pipeline + mock_store.load_pipeline.return_value = pipeline + # Simulate update_pipeline setting status to cancelled + pipeline.status = PipelineStatus.CANCELLED + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + mock_spawner.cleanup_pipeline.return_value = 2 + mock_spawner_fn.return_value = mock_spawner + + mock_dq = MagicMock() + mock_dq.get_pending_decisions.return_value = [] + mock_dq_fn.return_value = mock_dq + + response = client.patch( + "/api/v1/pipelines/test-pipeline", + json={"status": "cancelled"}, + ) + assert response.status_code == 200 + + # All running containers should be marked REMOVED + for container in pipeline.phases["refine"].containers: + assert container.status == ContainerStatus.REMOVED + assert container.exited_at is not None + + # All running agents should be marked FAILED with correct error + for agent in pipeline.phases["refine"].agents: + assert agent.status == AgentExecutionStatus.FAILED + assert agent.completed_at is not None + assert agent.error == "Pipeline cancelled" # status-specific message + + # Coordinator spawn records should be marked cancelled + for record in pipeline.coordinator_state.agents_spawned: + assert record.status == "cancelled" + assert record.completed_at is not None + + # Pipeline state should have been saved + mock_store.save_pipeline.assert_called_once_with(pipeline) + + @patch("routes.pipelines.get_decision_queue") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_cancel_skips_already_completed_agents( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_dq_fn, client + ): + mock_repo.return_value = "/repo" + pipeline = _make_cancellable_pipeline() + # Mark one agent as already complete + pipeline.phases["refine"].agents[0].status = AgentExecutionStatus.COMPLETE + pipeline.phases["refine"].containers[0].status = ContainerStatus.EXITED + pipeline.coordinator_state.agents_spawned[0].status = "complete" + + mock_store = MagicMock() + mock_store.update_pipeline.return_value = pipeline + mock_store.load_pipeline.return_value = pipeline + pipeline.status = PipelineStatus.CANCELLED + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + mock_spawner.cleanup_pipeline.return_value = 1 + mock_spawner_fn.return_value = mock_spawner + + mock_dq = MagicMock() + mock_dq.get_pending_decisions.return_value = [] + mock_dq_fn.return_value = mock_dq + + response = client.patch( + "/api/v1/pipelines/test-pipeline", + json={"status": "cancelled"}, + ) + assert response.status_code == 200 + + # Already-complete agent/container should NOT be overwritten + assert pipeline.phases["refine"].agents[0].status == AgentExecutionStatus.COMPLETE + assert pipeline.phases["refine"].containers[0].status == ContainerStatus.EXITED + assert pipeline.coordinator_state.agents_spawned[0].status == "complete" + + # Still-running agent/container should be updated + assert pipeline.phases["refine"].agents[1].status == AgentExecutionStatus.FAILED + assert pipeline.phases["refine"].containers[1].status == ContainerStatus.REMOVED + + @patch("routes.pipelines.get_decision_queue") + @patch("routes.pipelines.get_container_spawner") + @patch("routes.pipelines._resolve_pipeline") + @patch("routes.pipelines.get_repo_path") + def test_failed_pipeline_uses_correct_error_message( + self, mock_repo, mock_resolve, mock_spawner_fn, mock_dq_fn, client + ): + """When a pipeline transitions to FAILED, agent errors say 'Pipeline failed'.""" + mock_repo.return_value = "/repo" + pipeline = _make_cancellable_pipeline() + + mock_store = MagicMock() + mock_store.update_pipeline.return_value = pipeline + mock_store.load_pipeline.return_value = pipeline + pipeline.status = PipelineStatus.FAILED + mock_resolve.return_value = (mock_store, pipeline) + + mock_spawner = MagicMock() + mock_spawner.cleanup_pipeline.return_value = 2 + mock_spawner_fn.return_value = mock_spawner + + mock_dq = MagicMock() + mock_dq.get_pending_decisions.return_value = [] + mock_dq_fn.return_value = mock_dq + + response = client.patch( + "/api/v1/pipelines/test-pipeline", + json={"status": "failed"}, + ) + assert response.status_code == 200 + + # Agent errors should reflect the actual terminal status + for agent in pipeline.phases["refine"].agents: + assert agent.status == AgentExecutionStatus.FAILED + assert agent.error == "Pipeline failed" # not "Pipeline cancelled"