Fix orchestrator startup crash, spawn blocking, and cancel state sync

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 <id>` 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 <id>` 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 <id>` 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 <id>`. The `guardrail_counters` section shows current counts vs. configured limits.

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/

orchestrator/container_spawner.py

@@ -28,13 +28,19 @@ 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,
         GATEWAY_CONTAINER_NAME,
         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/<name> to the correct worktree
             # at /home/egg/.egg-worktrees/<id>/<name>.
+            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)

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

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:

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("/<pipeline_id>", 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(

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):

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 ──────────────────────────────