From b6cc65ed73302bbb174e16522c20ef81667b87c3 Mon Sep 17 00:00:00 2001 From: egg Date: Sat, 14 Mar 2026 06:41:01 +0000 Subject: [PATCH 1/3] Fix pipeline collaboration failures (#1081) Fix five coordination issues observed during the issue-1059 pipeline run: provide_input returning 405 due to missing /resolve in URL, reviewers finishing before primary agents due to no dependency enforcement, missing contract validation at spawn time, no consensus protocol reminder in agent prompts, and ConcurrentPhaseExecutor giving each agent a separate branch instead of sharing the pipeline branch. Closes #1081 --- orchestrator/concurrent_executor.py | 19 +- orchestrator/mcp_tools.py | 2 +- orchestrator/routes/coordinator.py | 63 ++++++- orchestrator/routes/pipelines.py | 2 +- .../tests/test_concurrent_integration.py | 32 ++++ .../tests/test_coordinator_mcp_functional.py | 2 +- .../test_coordinator_routes_functional.py | 176 +++++++++++++++++- 7 files changed, 283 insertions(+), 13 deletions(-) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index 50cf4d6f77..300cba600a 100644 --- a/orchestrator/concurrent_executor.py +++ b/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.""" diff --git a/orchestrator/mcp_tools.py b/orchestrator/mcp_tools.py index 058c54f775..e02b4e2bda 100644 --- a/orchestrator/mcp_tools.py +++ b/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, ) diff --git a/orchestrator/routes/coordinator.py b/orchestrator/routes/coordinator.py index 9b0e6af4cd..dadfa60498 100644 --- a/orchestrator/routes/coordinator.py +++ b/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 + logger.debug( + "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", diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 706640aa43..864232d9a1 100644 --- a/orchestrator/routes/pipelines.py +++ b/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. diff --git a/orchestrator/tests/test_concurrent_integration.py b/orchestrator/tests/test_concurrent_integration.py index 6c2e567dc3..624e5f3dbf 100644 --- a/orchestrator/tests/test_concurrent_integration.py +++ b/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 + from models import AgentRole + + 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 empty, falls back to issue-based name.""" + from concurrent_executor import ConcurrentPhaseExecutor + from models import AgentRole + + pipeline = _make_concurrent_pipeline() + pipeline.branch = "" # Clear branch + + executor = ConcurrentPhaseExecutor(pipeline=pipeline, spawn_fn=MagicMock()) + + branch = executor.get_worktree_branch(AgentRole.CODER) + assert branch == "egg/issue-999" + # Confirm no role suffix + assert "coder" not in branch diff --git a/orchestrator/tests/test_coordinator_mcp_functional.py b/orchestrator/tests/test_coordinator_mcp_functional.py index a7d6b8d81f..c1baf4ac7b 100644 --- a/orchestrator/tests/test_coordinator_mcp_functional.py +++ b/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"}, ) diff --git a/orchestrator/tests/test_coordinator_routes_functional.py b/orchestrator/tests/test_coordinator_routes_functional.py index 6b550cb7e6..07a4c117ab 100644 --- a/orchestrator/tests/test_coordinator_routes_functional.py +++ b/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 From 5ec697eee464980394f89c59870e0273126c8f01 Mon Sep 17 00:00:00 2001 From: egg Date: Sat, 14 Mar 2026 06:42:21 +0000 Subject: [PATCH 2/3] Fix checks: apply automated formatting fixes --- orchestrator/routes/coordinator.py | 11 +++++------ orchestrator/tests/test_concurrent_integration.py | 1 - 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/orchestrator/routes/coordinator.py b/orchestrator/routes/coordinator.py index dadfa60498..b4efade96b 100644 --- a/orchestrator/routes/coordinator.py +++ b/orchestrator/routes/coordinator.py @@ -233,12 +233,11 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]: 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" + s.role.value for s in coord_state.agents_spawned if s.status == "complete" } missing = [ - dep.value for dep in role_def.dependencies + dep.value + for dep in role_def.dependencies if dep.value not in completed_roles ] if missing: @@ -338,10 +337,10 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]: # 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" + ' 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" + 'sleep "${EGG_MESSAGE_POLL_INTERVAL:-30}"; done' ) agent_command = [ diff --git a/orchestrator/tests/test_concurrent_integration.py b/orchestrator/tests/test_concurrent_integration.py index 624e5f3dbf..ca4bb8362f 100644 --- a/orchestrator/tests/test_concurrent_integration.py +++ b/orchestrator/tests/test_concurrent_integration.py @@ -493,7 +493,6 @@ class TestGetWorktreeBranch: def test_get_worktree_branch_returns_pipeline_branch(self): """When pipeline.branch is set, all roles share it.""" from concurrent_executor import ConcurrentPhaseExecutor - from models import AgentRole pipeline = _make_concurrent_pipeline() assert pipeline.branch == "egg/issue-999" From c078126f5cfa5807c21c6404bed98016272aedc3 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 14 Mar 2026 06:56:17 +0000 Subject: [PATCH 3/3] Address review feedback: raise log level, improve fallback test --- orchestrator/routes/coordinator.py | 3 ++- orchestrator/tests/test_concurrent_integration.py | 7 ++++--- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/orchestrator/routes/coordinator.py b/orchestrator/routes/coordinator.py index b4efade96b..2e09d2507c 100644 --- a/orchestrator/routes/coordinator.py +++ b/orchestrator/routes/coordinator.py @@ -253,7 +253,8 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]: ) except (ValueError, KeyError): # Role not found in egg_contracts definitions — allow spawn - logger.debug( + # but warn since this bypasses a safety check + logger.warning( "No role definition found for dependency check, allowing spawn", role=role_str, ) diff --git a/orchestrator/tests/test_concurrent_integration.py b/orchestrator/tests/test_concurrent_integration.py index ca4bb8362f..c636f3c252 100644 --- a/orchestrator/tests/test_concurrent_integration.py +++ b/orchestrator/tests/test_concurrent_integration.py @@ -503,16 +503,17 @@ def test_get_worktree_branch_returns_pipeline_branch(self): assert executor.get_worktree_branch(role) == "egg/issue-999" def test_get_worktree_branch_fallback(self): - """When pipeline.branch is empty, falls back to issue-based name.""" + """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 = "" # Clear branch + 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-999" + assert branch == "egg/issue-777" # Confirm no role suffix assert "coder" not in branch