Enforce spawn dependency ordering and fix coordinator bugs

orchestrator/concurrent_executor.py

@@ -1,6 +1,6 @@
 """Concurrent phase executor for running multiple agents simultaneously.
 
-Spawns all agents at phase start, each with its own worktree branch.
+Spawns all agents at phase start, all sharing the pipeline branch.
 Monitors agent health, collects completion signals, and manages
 consensus-based phase completion.
 """
@@ -49,9 +49,9 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 class ConcurrentPhaseExecutor:
     """Executes a pipeline phase with all agents running concurrently.
 
-    Each agent gets its own worktree branch (egg/issue-{N}/{role}) and
-    communicates via the orchestrator message bus. Phase completion
-    requires consensus from all agents.
+    All agents share the pipeline branch and communicate via the
+    orchestrator message bus. Phase completion requires consensus
+    from all agents.
 
     Container failure behavior:
     - Single failure: Log, notify other agents, create HITL decision
@@ -88,9 +88,16 @@ def get_agent_roles(self) -> list[AgentRole]:
         ]
 
     def get_worktree_branch(self, role: AgentRole) -> str:
-        """Get the worktree branch name for an agent role."""
+        """Get the worktree branch name for an agent role.
+
+        Returns the pipeline's shared branch when set, falling back to
+        an issue-based branch name.  All agents share the same branch
+        so their commits land on a single history.
+        """
+        if self.pipeline.branch:
+            return self.pipeline.branch
         issue = self.pipeline.issue_number or self.pipeline.id
-        return f"egg/issue-{issue}/{role.value}"
+        return f"egg/issue-{issue}"
 
     def get_agent_env(self, role: AgentRole) -> dict[str, str]:
         """Get additional environment variables for concurrent mode."""

orchestrator/mcp_tools.py

@@ -275,7 +275,7 @@ def _handle_provide_input(self, args: dict[str, Any]) -> dict[str, Any]:
         decision_id = quote(args["decision_id"], safe="")
         data = {"resolution": args["response"]}
         result = self._make_request(
-            f"/api/v1/pipelines/{task_id}/decisions/{decision_id}",
+            f"/api/v1/pipelines/{task_id}/decisions/{decision_id}/resolve",
             method="POST",
             data=data,
         )

orchestrator/routes/coordinator.py

@@ -34,7 +34,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 
 from container_spawner import ContainerSpawnError, get_container_spawner
 from decision_queue import get_decision_queue
-from egg_contracts.agent_roles import get_roles_for_phase
+from egg_contracts.agent_roles import get_role_definition, get_roles_for_phase
 from events import EventType, emit_event
 from gateway_client import GatewayError, get_gateway_client
 from models import (
@@ -209,6 +209,56 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]:
                     },
                 )
 
+            # Block spawns in implement/PR phases if no contract exists
+            if (
+                pipeline.current_phase in (PipelinePhase.IMPLEMENT, PipelinePhase.PR)
+                and not pipeline.contract_synced
+            ):
+                return make_error_response(
+                    f"Cannot spawn agent in '{pipeline.current_phase.value}' phase: "
+                    "no contract exists for this pipeline. A contract must be "
+                    "created before implementation can begin.",
+                    status_code=409,
+                    details={
+                        "pipeline_id": pipeline_id,
+                        "phase": pipeline.current_phase.value,
+                        "contract_synced": False,
+                    },
+                )
+
+            # 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:
@@ -283,6 +333,17 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]:
                 f"Execute your role for the {pipeline.current_phase.value} phase. "
                 f"Follow the instructions in your CLAUDE.md."
             )
+
+            # Append consensus protocol reminder so agents signal
+            # readiness and stay alive for the orchestrator to collect.
+            agent_prompt += (
+                "\n\nIMPORTANT: When your work is complete, signal readiness:\n"
+                '  egg-orch signal readiness --state READY --reason "Work complete"\n'
+                "Then stay alive polling for messages. Do NOT exit.\n"
+                "  while true; do egg-orch message poll; "
+                'sleep "${EGG_MESSAGE_POLL_INTERVAL:-30}"; done'
+            )
+
             agent_command = [
                 "claude",
                 "--dangerously-skip-permissions",

orchestrator/routes/pipelines.py

@@ -4379,7 +4379,7 @@ def _run_concurrent_phase(
     """Run a phase using concurrent all-agents-at-once execution.
 
     Creates a ConcurrentPhaseExecutor that spawns all agents simultaneously,
-    each with its own worktree branch. Each container receives a role-specific
+    all sharing the pipeline branch. Each container receives a role-specific
     prompt built via ``_build_agent_prompt``. After spawning, waits for all
     containers to exit and records their state in the pipeline store.
 

orchestrator/tests/test_concurrent_integration.py

@@ -485,3 +485,35 @@ def test_returns_six_concurrent_roles(self):
         assert AgentRole.CHECKER in roles
         assert AgentRole.REVIEWER_CODE in roles
         assert AgentRole.REVIEWER_CONTRACT in roles
+
+
+class TestGetWorktreeBranch:
+    """Tests for ConcurrentPhaseExecutor.get_worktree_branch()."""
+
+    def test_get_worktree_branch_returns_pipeline_branch(self):
+        """When pipeline.branch is set, all roles share it."""
+        from concurrent_executor import ConcurrentPhaseExecutor
+
+        pipeline = _make_concurrent_pipeline()
+        assert pipeline.branch == "egg/issue-999"
+
+        executor = ConcurrentPhaseExecutor(pipeline=pipeline, spawn_fn=MagicMock())
+
+        for role in executor.get_agent_roles():
+            assert executor.get_worktree_branch(role) == "egg/issue-999"
+
+    def test_get_worktree_branch_fallback(self):
+        """When pipeline.branch is None, falls back to issue-based name."""
+        from concurrent_executor import ConcurrentPhaseExecutor
+        from models import AgentRole
+
+        pipeline = _make_concurrent_pipeline()
+        pipeline.branch = None  # Clear branch
+        pipeline.issue_number = 777  # Distinct from default to prove fallback computes
+
+        executor = ConcurrentPhaseExecutor(pipeline=pipeline, spawn_fn=MagicMock())
+
+        branch = executor.get_worktree_branch(AgentRole.CODER)
+        assert branch == "egg/issue-777"
+        # Confirm no role suffix
+        assert "coder" not in branch

orchestrator/tests/test_coordinator_mcp_functional.py

@@ -555,7 +555,7 @@ def test_provide_input(self, mock_req):
         assert result["success"] is True
         # Verify correct endpoint
         mock_req.assert_called_once_with(
-            "/api/v1/pipelines/issue-42/decisions/d-1",
+            "/api/v1/pipelines/issue-42/decisions/d-1/resolve",
             method="POST",
             data={"resolution": "REST"},
         )

orchestrator/tests/test_coordinator_routes_functional.py

@@ -301,9 +301,13 @@ def test_spawn_increments_guardrail_counters(
         mock_lock.return_value.__enter__ = MagicMock()
         mock_lock.return_value.__exit__ = MagicMock(return_value=False)
 
+        # tester depends on coder — provide a completed coder record
         pipeline = _make_pipeline(
             coordinator_state=CoordinatorState(
                 guardrail_counters=GuardrailCounters(total_agents_spawned=1),
+                agents_spawned=[
+                    AgentSpawnRecord(role=AgentRole.CODER, status="complete"),
+                ],
             ),
         )
         store = MagicMock()
@@ -1289,12 +1293,22 @@ def test_spawn_correct_role_for_refine_phase(
     def test_spawn_reviewer_role_allowed_for_phase(
         self, mock_repo, mock_lock, mock_store_fn, mock_spawner_fn, mock_emit, client
     ):
-        """Reviewer roles should be allowed for their corresponding phase."""
+        """Reviewer roles should be allowed for their corresponding phase
+        when all dependencies have completed."""
         mock_repo.return_value = Path("/tmp/repo")
         mock_lock.return_value.__enter__ = MagicMock()
         mock_lock.return_value.__exit__ = MagicMock(return_value=False)
 
-        pipeline = _make_pipeline(phase=PipelinePhase.IMPLEMENT)
+        # reviewer_code depends on integrator, task_planner, risk_analyst
+        completed_deps = [
+            AgentSpawnRecord(role=AgentRole.INTEGRATOR, status="complete"),
+            AgentSpawnRecord(role=AgentRole.TASK_PLANNER, status="complete"),
+            AgentSpawnRecord(role=AgentRole.RISK_ANALYST, status="complete"),
+        ]
+        pipeline = _make_pipeline(
+            phase=PipelinePhase.IMPLEMENT,
+            coordinator_state=CoordinatorState(agents_spawned=completed_deps),
+        )
         store = MagicMock()
         store.load_pipeline.return_value = pipeline
         mock_store_fn.return_value = store
@@ -1348,7 +1362,17 @@ def test_spawn_wrong_reviewer_for_phase_rejected(
         mock_lock.return_value.__enter__ = MagicMock()
         mock_lock.return_value.__exit__ = MagicMock(return_value=False)
 
-        pipeline = _make_pipeline(phase=PipelinePhase.IMPLEMENT)
+        # reviewer_plan depends on task_planner and risk_analyst — satisfy
+        # those so the dependency check passes and phase-role check fires
+        pipeline = _make_pipeline(
+            phase=PipelinePhase.IMPLEMENT,
+            coordinator_state=CoordinatorState(
+                agents_spawned=[
+                    AgentSpawnRecord(role=AgentRole.TASK_PLANNER, status="complete"),
+                    AgentSpawnRecord(role=AgentRole.RISK_ANALYST, status="complete"),
+                ],
+            ),
+        )
         store = MagicMock()
         store.load_pipeline.return_value = pipeline
         mock_store_fn.return_value = store
@@ -1423,3 +1447,149 @@ def test_spawn_in_coordinator_phase_without_role_mapping_allowed(
             json={"role": "coder"},
         )
         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 ──────────────────────────────
+
+
+class TestSpawnContractEnforcement:
+    """Spawn must be blocked in implement/PR phases without a contract."""
+
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_spawn_blocked_without_contract_in_implement_phase(
+        self, mock_repo, mock_lock, mock_store_fn, client
+    ):
+        """Spawning in implement phase without contract_synced returns 409."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.IMPLEMENT)
+        pipeline.contract_synced = False
+        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": "coder"},
+        )
+        assert response.status_code == 409
+        assert "contract" in response.get_json()["message"].lower()
+
+    @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_without_contract_in_refine_phase(
+        self, mock_repo, mock_lock, mock_store_fn, mock_spawner_fn, mock_emit, client
+    ):
+        """Spawning in refine phase is allowed even without a contract."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.REFINE)
+        pipeline.contract_synced = False
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        spawner = MagicMock()
+        spawned = MagicMock()
+        spawned.container_info = ContainerInfo(
+            container_id="ref123", container_name="egg-test-refiner"
+        )
+        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": "refiner"},
+        )
+        assert response.status_code == 200