From ce33549d4f7167d02c5c0e7e8386aee3f2d92799 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 14 Mar 2026 01:23:52 -0700 Subject: [PATCH 01/10] Fix orchestrator crash: add missing egg_agent module to Dockerfile The egg_agent shared module was added in #1088 but the orchestrator Dockerfile was not updated to copy it into the container image, causing an ImportError crash loop on startup. Co-Authored-By: Claude Opus 4.6 (1M context) --- orchestrator/Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/orchestrator/Dockerfile b/orchestrator/Dockerfile index 78d2b0d7fd..7882ed7e9b 100644 --- a/orchestrator/Dockerfile +++ b/orchestrator/Dockerfile @@ -21,6 +21,7 @@ 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/ From 9f3fe9405ed73f33fd24aa413704b276432d283c Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 14 Mar 2026 08:32:08 +0000 Subject: [PATCH 02/10] Update Dockerfile comment to include egg_agent in shared modules list --- orchestrator/Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/orchestrator/Dockerfile b/orchestrator/Dockerfile index 7882ed7e9b..f592000f6a 100644 --- a/orchestrator/Dockerfile +++ b/orchestrator/Dockerfile @@ -16,7 +16,7 @@ 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/ From a97d400221847933c8784370b302fdb2f99a4f9d Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 14 Mar 2026 01:32:38 -0700 Subject: [PATCH 03/10] Remove spawn dependency enforcement to allow concurrent agents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dependency blocking added in #1085 prevented agents from spawning until their declared dependencies completed. This is wrong — agents should run concurrently and communicate via the messaging system. Co-Authored-By: Claude Opus 4.6 (1M context) --- orchestrator/routes/coordinator.py | 35 +------- .../test_coordinator_routes_functional.py | 79 ------------------- 2 files changed, 1 insertion(+), 113 deletions(-) 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/tests/test_coordinator_routes_functional.py b/orchestrator/tests/test_coordinator_routes_functional.py index 07a4c117ab..cc60c999da 100644 --- a/orchestrator/tests/test_coordinator_routes_functional.py +++ b/orchestrator/tests/test_coordinator_routes_functional.py @@ -1452,85 +1452,6 @@ def test_spawn_in_coordinator_phase_without_role_mapping_allowed( # ── 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 ────────────────────────────── From fa090185f95f5f4a9d6390fc3b36a2407efdb0c6 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 14 Mar 2026 01:38:32 -0700 Subject: [PATCH 04/10] Fix stale container/agent status after pipeline cancellation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a pipeline is cancelled, cleanup_pipeline() removes Docker containers but the pipeline state store still shows them as "running". This adds _mark_pipeline_records_cancelled() which syncs the persisted state after container cleanup — marking containers as REMOVED, agents as FAILED, and coordinator spawn records as cancelled. Co-Authored-By: Claude Opus 4.6 (1M context) --- orchestrator/routes/pipelines.py | 66 +++++++++ orchestrator/tests/test_pipelines_api.py | 166 ++++++++++++++++++++++- 2 files changed, 231 insertions(+), 1 deletion(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 91c0bcd52e..32544f5928 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -554,6 +554,68 @@ def create_pipeline() -> tuple[Response, int]: return make_error_response(f"Failed to create pipeline: {e}", status_code=500) +def _mark_pipeline_records_cancelled( + pipeline: "Pipeline", + store: "StateStore", + pipeline_id: str, +) -> None: + """Mark all running containers and agents as stopped after pipeline cancellation. + + After Docker containers are force-removed, the pipeline state still shows + them as "running". This function syncs the persisted state to reflect that + containers have been removed and agents are no longer running. + """ + try: + from models import AgentExecutionStatus, ContainerStatus # type: ignore + except ImportError: + from ..models import AgentExecutionStatus, ContainerStatus + + 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 = "Pipeline cancelled" + 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: + try: + store.save_pipeline(pipeline) + logger.info( + "Synced pipeline state after cancellation", + pipeline_id=pipeline_id, + ) + except Exception as e: + logger.warning( + "Failed to sync pipeline state after cancellation", + pipeline_id=pipeline_id, + error=str(e), + ) + + @pipelines_bp.route("/", methods=["PATCH"]) def update_pipeline(pipeline_id: str) -> tuple[Response, int]: """ @@ -632,6 +694,10 @@ def update_pipeline(pipeline_id: str) -> tuple[Response, int]: exc_info=True, ) + # Sync pipeline state: mark all running containers/agents as + # stopped so the persisted state reflects reality after cleanup. + _mark_pipeline_records_cancelled(pipeline, store, pipeline_id) + logger.info("Pipeline updated", pipeline_id=pipeline_id) return make_success_response( diff --git a/orchestrator/tests/test_pipelines_api.py b/orchestrator/tests/test_pipelines_api.py index a6262fb43e..72236efb3a 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,154 @@ 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 TestCancelPipelineSyncsState: + """Tests that cancelling 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 + # 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 + for agent in pipeline.phases["refine"].agents: + assert agent.status == AgentExecutionStatus.FAILED + assert agent.completed_at is not None + assert agent.error == "Pipeline cancelled" + + # 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 + 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 From fa6577a790746419aa0c97b8c9fc2b23b393cffd Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 14 Mar 2026 01:51:49 -0700 Subject: [PATCH 05/10] Fix cancel state sync: reload pipeline before marking records The previous approach mutated the stale pipeline object returned by update_pipeline(), which could overwrite coordinator state changes made between the status update and container cleanup. Now reloads the latest state from the store after containers are killed, ensuring we mark the most up-to-date records and don't clobber concurrent writes. Co-Authored-By: Claude Opus 4.6 (1M context) --- orchestrator/routes/pipelines.py | 44 ++++++++++++++---------- orchestrator/tests/test_pipelines_api.py | 2 ++ 2 files changed, 27 insertions(+), 19 deletions(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 32544f5928..8512f161f0 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -555,21 +555,24 @@ def create_pipeline() -> tuple[Response, int]: def _mark_pipeline_records_cancelled( - pipeline: "Pipeline", store: "StateStore", pipeline_id: str, -) -> None: +) -> "Pipeline": """Mark all running containers and agents as stopped after pipeline cancellation. After Docker containers are force-removed, the pipeline state still shows - them as "running". This function syncs the persisted state to reflect that - containers have been removed and agents are no longer running. + 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. """ try: from models import AgentExecutionStatus, ContainerStatus # type: ignore except ImportError: from ..models import AgentExecutionStatus, ContainerStatus + pipeline = store.load_pipeline(pipeline_id) now = datetime.utcnow() changed = False @@ -602,18 +605,13 @@ def _mark_pipeline_records_cancelled( changed = True if changed: - try: - store.save_pipeline(pipeline) - logger.info( - "Synced pipeline state after cancellation", - pipeline_id=pipeline_id, - ) - except Exception as e: - logger.warning( - "Failed to sync pipeline state after cancellation", - pipeline_id=pipeline_id, - error=str(e), - ) + store.save_pipeline(pipeline) + logger.info( + "Synced pipeline state after cancellation", + pipeline_id=pipeline_id, + ) + + return pipeline @pipelines_bp.route("/", methods=["PATCH"]) @@ -694,9 +692,17 @@ def update_pipeline(pipeline_id: str) -> tuple[Response, int]: exc_info=True, ) - # Sync pipeline state: mark all running containers/agents as - # stopped so the persisted state reflects reality after cleanup. - _mark_pipeline_records_cancelled(pipeline, store, pipeline_id) + # 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_cancelled(store, pipeline_id) + except Exception as e: + logger.warning( + "Failed to sync pipeline state after cancellation", + pipeline_id=pipeline_id, + error=str(e), + ) logger.info("Pipeline updated", pipeline_id=pipeline_id) diff --git a/orchestrator/tests/test_pipelines_api.py b/orchestrator/tests/test_pipelines_api.py index 72236efb3a..49ff8f9b5d 100644 --- a/orchestrator/tests/test_pipelines_api.py +++ b/orchestrator/tests/test_pipelines_api.py @@ -316,6 +316,7 @@ def test_cancel_marks_running_containers_as_removed( 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) @@ -369,6 +370,7 @@ def test_cancel_skips_already_completed_agents( 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) From c193e39007946acff05643206b512f1e1f8ae10e Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 14 Mar 2026 08:57:14 +0000 Subject: [PATCH 06/10] Address review feedback: fix error message, rename function, clean up docs --- docs/guides/coordinator.md | 5 --- orchestrator/routes/pipelines.py | 22 +++++----- .../test_coordinator_routes_functional.py | 3 -- orchestrator/tests/test_pipelines_api.py | 43 +++++++++++++++++-- 4 files changed, 50 insertions(+), 23 deletions(-) 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/routes/pipelines.py b/orchestrator/routes/pipelines.py index 32544f5928..884fd2e129 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,22 +558,18 @@ def create_pipeline() -> tuple[Response, int]: return make_error_response(f"Failed to create pipeline: {e}", status_code=500) -def _mark_pipeline_records_cancelled( +def _mark_pipeline_records_terminated( pipeline: "Pipeline", store: "StateStore", pipeline_id: str, ) -> None: - """Mark all running containers and agents as stopped after pipeline cancellation. + """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 function syncs the persisted state to reflect that containers have been removed and agents are no longer running. """ - try: - from models import AgentExecutionStatus, ContainerStatus # type: ignore - except ImportError: - from ..models import AgentExecutionStatus, ContainerStatus - now = datetime.utcnow() changed = False @@ -591,7 +591,7 @@ def _mark_pipeline_records_cancelled( ): agent.status = AgentExecutionStatus.FAILED agent.completed_at = now - agent.error = "Pipeline cancelled" + agent.error = f"Pipeline {pipeline.status.value}" changed = True if pipeline.coordinator_state: @@ -605,12 +605,12 @@ def _mark_pipeline_records_cancelled( try: store.save_pipeline(pipeline) logger.info( - "Synced pipeline state after cancellation", + "Synced pipeline state after termination", pipeline_id=pipeline_id, ) except Exception as e: logger.warning( - "Failed to sync pipeline state after cancellation", + "Failed to sync pipeline state after termination", pipeline_id=pipeline_id, error=str(e), ) @@ -696,7 +696,7 @@ def update_pipeline(pipeline_id: str) -> tuple[Response, int]: # Sync pipeline state: mark all running containers/agents as # stopped so the persisted state reflects reality after cleanup. - _mark_pipeline_records_cancelled(pipeline, store, pipeline_id) + _mark_pipeline_records_terminated(pipeline, store, pipeline_id) logger.info("Pipeline updated", pipeline_id=pipeline_id) diff --git a/orchestrator/tests/test_coordinator_routes_functional.py b/orchestrator/tests/test_coordinator_routes_functional.py index cc60c999da..c45c427326 100644 --- a/orchestrator/tests/test_coordinator_routes_functional.py +++ b/orchestrator/tests/test_coordinator_routes_functional.py @@ -1449,9 +1449,6 @@ def test_spawn_in_coordinator_phase_without_role_mapping_allowed( assert response.status_code == 200 -# ── Dependency validation tests ──────────────────────────────────── - - # ── Spawn contract enforcement tests ────────────────────────────── diff --git a/orchestrator/tests/test_pipelines_api.py b/orchestrator/tests/test_pipelines_api.py index 72236efb3a..71c64de343 100644 --- a/orchestrator/tests/test_pipelines_api.py +++ b/orchestrator/tests/test_pipelines_api.py @@ -301,8 +301,8 @@ def _make_cancellable_pipeline(pipeline_id="test-pipeline"): return pipeline -class TestCancelPipelineSyncsState: - """Tests that cancelling a pipeline marks running containers/agents as stopped.""" +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") @@ -339,11 +339,11 @@ def test_cancel_marks_running_containers_as_removed( assert container.status == ContainerStatus.REMOVED assert container.exited_at is not None - # All running agents should be marked FAILED + # 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" + assert agent.error == "Pipeline cancelled" # status-specific message # Coordinator spawn records should be marked cancelled for record in pipeline.coordinator_state.agents_spawned: @@ -394,3 +394,38 @@ def test_cancel_skips_already_completed_agents( # 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 + 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" From 153583fa09480372c0f9894c05c1f0c850d8024c Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 14 Mar 2026 01:58:31 -0700 Subject: [PATCH 07/10] Fix spawned agents missing EGG_ORCHESTRATOR_URL env var MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit container_spawner.py set EGG_PIPELINE_ID but never set EGG_ORCHESTRATOR_URL, causing the sandbox entrypoint to bail out with "Orchestrator mode enabled but missing URL or pipeline_id" (exit 124). The other two spawn paths (sandbox_template.py and routes/pipelines.py) both set it correctly — this was the only one missing it. Co-Authored-By: Claude Opus 4.6 (1M context) --- orchestrator/container_spawner.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/orchestrator/container_spawner.py b/orchestrator/container_spawner.py index 1846763b7a..1b827d4258 100644 --- a/orchestrator/container_spawner.py +++ b/orchestrator/container_spawner.py @@ -28,6 +28,12 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] return logging.getLogger(name) +from sandbox_template import ( + ORCHESTRATOR_CONTAINER_NAME, + ORCHESTRATOR_ISOLATED_IP, + ORCHESTRATOR_PORT, +) + try: from egg_config import ( EGG_CONTAINER_IP, @@ -371,11 +377,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_CONTAINER_NAME + ) + 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) From d13887136d44e59eda39ef1bdacc533fe7b6f26b Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 14 Mar 2026 02:00:24 -0700 Subject: [PATCH 08/10] Increase cancel_task MCP endpoint timeout from 30s to 120s Cancellation involves stopping containers which regularly exceeds the default 30s request timeout, causing spurious "timed out" errors. Co-Authored-By: Claude Opus 4.6 (1M context) --- orchestrator/mcp_tools.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) 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 From d86961e19813231edec2d19f6081f60faa319b5a Mon Sep 17 00:00:00 2001 From: jwbron <8340608+jwbron@users.noreply.github.com> Date: Sat, 14 Mar 2026 09:06:55 +0000 Subject: [PATCH 09/10] Fix test to expect timeout=120 in cancel_task --- orchestrator/tests/test_coordinator_gaps.py | 1 + 1 file changed, 1 insertion(+) 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): From 4c2b4c587a44834f49b16cd0ba9ac0fe346a0b68 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 14 Mar 2026 09:21:31 +0000 Subject: [PATCH 10/10] Align orchestrator URL construction across spawn paths Use ORCHESTRATOR_EXTERNAL_IP (static IP) instead of ORCHESTRATOR_CONTAINER_NAME (hostname) in container_spawner.py for public mode, matching the pattern in routes/pipelines.py. --- orchestrator/container_spawner.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/orchestrator/container_spawner.py b/orchestrator/container_spawner.py index 1b827d4258..abb4156764 100644 --- a/orchestrator/container_spawner.py +++ b/orchestrator/container_spawner.py @@ -29,7 +29,6 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from sandbox_template import ( - ORCHESTRATOR_CONTAINER_NAME, ORCHESTRATOR_ISOLATED_IP, ORCHESTRATOR_PORT, ) @@ -41,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, @@ -56,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) @@ -378,7 +379,7 @@ def spawn_agent_container( # 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_CONTAINER_NAME + ORCHESTRATOR_ISOLATED_IP if mode == "private" else ORCHESTRATOR_EXTERNAL_IP ) orchestrator_url = f"http://{orchestrator_host}:{ORCHESTRATOR_PORT}" spawner_env: dict[str, str] = {