diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index cb54c5de90..27b13773cb 100644 --- a/orchestrator/concurrent_executor.py +++ b/orchestrator/concurrent_executor.py @@ -28,7 +28,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] from consensus import get_consensus_evaluator -from egg_agent import build_agent_command +from consensus_wrapper import build_consensus_wrapped_command from events import EventType, emit_event from message_store import Message, MessageType, get_message_store from models import ( @@ -177,7 +177,7 @@ def _spawn_agent(self, role: AgentRole, prompt_text: str = "") -> AgentExecution command: list[str] | None = None if prompt_text: - command = build_agent_command(prompt_text) + command = build_consensus_wrapped_command(prompt_text) result = self.spawn_fn( role=role, diff --git a/orchestrator/consensus_wrapper.py b/orchestrator/consensus_wrapper.py new file mode 100644 index 0000000000..8b8f4fdcec --- /dev/null +++ b/orchestrator/consensus_wrapper.py @@ -0,0 +1,110 @@ +"""Build consensus-wrapped commands for concurrent agent containers. + +When agents run in concurrent mode, they must stay alive after completing +their work to participate in consensus. This module provides a shell wrapper +that catches early Claude exits and keeps the container alive polling for +consensus, as a safety net for agents that exit without following the +stay-alive protocol. +""" + +import shlex + +# Shell script that wraps the Claude CLI invocation. After Claude exits, +# if EGG_CONCURRENT_MODE is set and exit was clean (code 0), it auto-signals +# READY (if the agent didn't) and polls until consensus is reached or a +# timeout expires. Non-zero exits are treated as failures — no READY signal. +_CONSENSUS_WRAPPER_TEMPLATE = r""" +#!/bin/bash +set -uo pipefail + +# Run the Claude agent +{claude_command} +CLAUDE_EXIT=$? + +# If not in concurrent mode, exit normally +if [ "${{EGG_CONCURRENT_MODE:-}}" != "true" ]; then + exit $CLAUDE_EXIT +fi + +# Only signal READY on clean exit. A non-zero exit means the agent crashed +# or errored — its work may be incomplete, so we must NOT claim readiness. +if [ "$CLAUDE_EXIT" -ne 0 ]; then + echo "[consensus-wrapper] Agent failed (code $CLAUDE_EXIT). NOT signaling READY." + exit $CLAUDE_EXIT +fi + +# Clean exit — auto-signal READY as a safety net. +# If the agent already signaled READY, this is a no-op update. +echo "[consensus-wrapper] Agent exited cleanly. Auto-signaling READY..." +egg-orch signal readiness --state READY \ + --reason "Agent process exited cleanly, auto-signaling READY" \ + 2>/dev/null || true + +# Stay alive polling for consensus until reached or timeout. +POLL_INTERVAL="${{EGG_MESSAGE_POLL_INTERVAL:-30}}" +TIMEOUT="${{EGG_CONSENSUS_WRAPPER_TIMEOUT:-300}}" +ELAPSED=0 + +echo "[consensus-wrapper] Entering consensus wait loop (timeout=${{TIMEOUT}}s)..." +while [ "$ELAPSED" -lt "$TIMEOUT" ]; do + # Check consensus via readiness signal response + RESPONSE=$(egg-orch signal readiness --state READY \ + --reason "Waiting for consensus" --json 2>/dev/null || echo "{{}}") + + IS_COMPLETE=$(echo "$RESPONSE" | python3 -c \ + "import sys,json; d=json.load(sys.stdin); print(d.get('data',{{}}).get('consensus',{{}}).get('is_complete',False))" \ + 2>/dev/null || echo "False") + + if [ "$IS_COMPLETE" = "True" ]; then + echo "[consensus-wrapper] Consensus reached. Exiting." + exit $CLAUDE_EXIT + fi + + # Poll for messages (may contain work that should have been handled) + egg-orch message poll 2>/dev/null || true + + sleep "$POLL_INTERVAL" + ELAPSED=$((ELAPSED + POLL_INTERVAL)) +done + +echo "[consensus-wrapper] Consensus not reached within ${{TIMEOUT}}s. Exiting." +exit $CLAUDE_EXIT +""" + + +def build_consensus_wrapped_command( + prompt_text: str, + model: str = "opus", + max_turns: int = 200, +) -> list[str]: + """Build a shell command that runs Claude with a consensus wait wrapper. + + The wrapper ensures that after Claude exits, the container stays alive + polling for consensus rather than disappearing and triggering the + orchestrator's fallback path. + + Args: + prompt_text: The prompt to pass to the Claude CLI. + model: Claude model to use. + max_turns: Maximum number of tool-call turns. + + Returns: + Command list suitable for container spawning (bash -c "..."). + """ + claude_parts = [ + "claude", + "--dangerously-skip-permissions", + "--print", + "--verbose", + "--output-format", + "stream-json", + "--model", + model, + "--max-turns", + str(max_turns), + prompt_text, + ] + claude_command = " ".join(shlex.quote(p) for p in claude_parts) + script = _CONSENSUS_WRAPPER_TEMPLATE.format(claude_command=claude_command) + + return ["bash", "-c", script] diff --git a/orchestrator/routes/coordinator.py b/orchestrator/routes/coordinator.py index d1fa4e976a..4ece6dd2ae 100644 --- a/orchestrator/routes/coordinator.py +++ b/orchestrator/routes/coordinator.py @@ -32,9 +32,9 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] return logging.getLogger(name) +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_agent import build_agent_command 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 @@ -335,17 +335,10 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]: 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 = build_agent_command(agent_prompt) + # Wrap the Claude invocation in a consensus shell wrapper that + # keeps the container alive polling for consensus if Claude exits + # before the orchestrator stops the container. + agent_command = build_consensus_wrapped_command(agent_prompt) # Spawn the container spawner = get_container_spawner() diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 64486b3d77..5f2b8a3b57 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -2880,6 +2880,7 @@ def _build_agent_prompt( short_circuit: bool = False, phase_obj=None, all_phases=None, + concurrent: bool = False, ) -> str: """Build a role-specific prompt for multi-agent execution. @@ -2911,6 +2912,9 @@ def _build_agent_prompt( short_circuit: Whether short-circuit mode is enabled phase_obj: Current plan phase object (Tier 3 context, optional) all_phases: All contract phases (Tier 3 context, optional) + concurrent: Whether agent runs in concurrent multi-agent mode. + When True, adds consensus lifecycle preamble instructing the + agent to stay alive, poll messages, and participate in consensus. Returns: Complete prompt string for the agent @@ -2947,6 +2951,32 @@ def _build_agent_prompt( lines.append(f"Issue: #{issue_number}") lines.append("") + # Concurrent mode: add consensus lifecycle preamble so agents understand + # they must stay alive and participate in consensus, not just do their task. + if concurrent: + lines.extend( + [ + "## CRITICAL: Concurrent Consensus Protocol\n", + "You are running in CONCURRENT mode alongside other agents. " + "Your job is NOT just your task — it is the **full lifecycle**:\n", + "1. **BOOTSTRAP**: Check if the agents you depend on have produced work yet. " + "If not, signal BLOCKED and poll every 30s until their work appears.", + "2. **EXECUTE**: Do your assigned work (see Your Task below).", + "3. **SIGNAL READY**: When your work is complete, run: " + '`egg-orch signal readiness --state READY --reason "Work complete"`', + "4. **STAY ALIVE & REACT**: Continue polling for messages with " + "`egg-orch message poll`. If new commits land or another agent sends " + "feedback, transition back to WORKING, address it, then signal READY again.", + "5. **WAIT FOR STOP**: The orchestrator sends SIGTERM when consensus is " + "reached. **You do NOT decide when to exit.** Use your remaining turns " + "to poll and react.\n", + "**If you exit before the orchestrator stops you, you have FAILED your role.** " + "Completing your task is necessary but NOT sufficient — you must remain " + "available to react to other agents' work until consensus.\n", + "", + ] + ) + # Include role-appropriate context instead of the raw issue body. # Analysis roles (architect, task_planner, risk_analyst) receive the full # issue body. Execution roles (tester, documenter, integrator) receive a @@ -3222,9 +3252,29 @@ def _build_agent_prompt( ) lines.append("## Phase Completion\n") - lines.append( - "When you have completed your work, ensure everything is committed and exit successfully." - ) + if concurrent: + lines.extend( + [ + "When you have completed your primary work:\n", + "1. Commit all changes", + '2. Run: `egg-orch signal readiness --state READY --reason "Work complete"`', + "3. Enter a stay-alive polling loop:", + "```bash", + "while true; do", + " egg-orch message poll", + ' sleep "${EGG_MESSAGE_POLL_INTERVAL:-30}"', + "done", + "```", + "4. If a message arrives that affects your work, transition back to WORKING, " + "address it, then signal READY again.", + "5. **Do NOT exit.** The orchestrator will stop your container when consensus " + "is reached.", + ] + ) + else: + lines.append( + "When you have completed your work, ensure everything is committed and exit successfully." + ) return "\n".join(lines) @@ -4351,6 +4401,7 @@ def _run_concurrent_phase( branch=pipeline.branch, repo_path=str(worktree_repo_path), short_circuit=pipeline.short_circuit, + concurrent=True, ) agent_prompts[role] = prompt @@ -4641,6 +4692,34 @@ def _update_agents_complete() -> None: role=exec_info.role.value, error=str(e), ) + else: + # Clean exit (code 0): treat as implicit READY if the + # agent didn't explicitly signal. This prevents early + # exits from blocking consensus indefinitely. + try: + from consensus import ReadinessState, get_consensus_evaluator + + evaluator = get_consensus_evaluator() + current = evaluator.evaluate(pipeline_id) + agent_state = current.get("agents", {}).get(exec_info.role.value) + if agent_state is None or agent_state.state != ReadinessState.READY: + evaluator.update_readiness( + pipeline_id, + exec_info.role.value, + ReadinessState.READY, + reason="Container exited cleanly (implicit READY)", + ) + logger.info( + "Auto-registered READY for cleanly exited container", + pipeline_id=pipeline_id, + role=exec_info.role.value, + ) + except Exception as e: + logger.warning( + "Failed to auto-register READY for exited container", + role=exec_info.role.value, + error=str(e), + ) # 5. All containers exited — fall back to exit-code-based result if len(exited_containers) >= len(active_executions): diff --git a/orchestrator/tests/test_concurrent_integration.py b/orchestrator/tests/test_concurrent_integration.py index c636f3c252..afe4b5a2e0 100644 --- a/orchestrator/tests/test_concurrent_integration.py +++ b/orchestrator/tests/test_concurrent_integration.py @@ -517,3 +517,145 @@ def test_get_worktree_branch_fallback(self): assert branch == "egg/issue-777" # Confirm no role suffix assert "coder" not in branch + + +class TestConcurrentPromptLifecycle: + """Tests for consensus lifecycle preamble in agent prompts.""" + + def test_concurrent_prompt_includes_lifecycle_preamble(self): + """When concurrent=True, prompt includes consensus protocol section.""" + from routes.pipelines import _build_agent_prompt + + prompt = _build_agent_prompt( + role_value="tester", + phase="implement", + pipeline_id="issue-123", + pipeline_mode="issue", + concurrent=True, + ) + assert "Concurrent Consensus Protocol" in prompt + assert "BOOTSTRAP" in prompt + assert "STAY ALIVE" in prompt + assert "WAIT FOR STOP" in prompt + assert "FAILED your role" in prompt + + def test_non_concurrent_prompt_omits_lifecycle_preamble(self): + """When concurrent=False (default), prompt has no consensus section.""" + from routes.pipelines import _build_agent_prompt + + prompt = _build_agent_prompt( + role_value="tester", + phase="implement", + pipeline_id="issue-123", + pipeline_mode="issue", + ) + assert "Concurrent Consensus Protocol" not in prompt + + def test_concurrent_phase_completion_includes_polling_loop(self): + """Concurrent prompts should have stay-alive instructions in Phase Completion.""" + from routes.pipelines import _build_agent_prompt + + prompt = _build_agent_prompt( + role_value="documenter", + phase="implement", + pipeline_id="issue-123", + pipeline_mode="issue", + concurrent=True, + ) + assert "egg-orch signal readiness --state READY" in prompt + assert "egg-orch message poll" in prompt + assert "Do NOT exit" in prompt + + def test_non_concurrent_phase_completion_says_exit(self): + """Non-concurrent prompts should tell agents to exit normally.""" + from routes.pipelines import _build_agent_prompt + + prompt = _build_agent_prompt( + role_value="documenter", + phase="implement", + pipeline_id="issue-123", + pipeline_mode="issue", + concurrent=False, + ) + assert "exit successfully" in prompt + + +class TestSpawnUsesConsensusWrapper: + """Tests that concurrent spawns use the consensus shell wrapper.""" + + def test_spawn_agent_uses_wrapped_command(self): + """_spawn_agent should produce a bash -c wrapper, not raw claude args.""" + from concurrent_executor import ConcurrentPhaseExecutor + from models import AgentRole + + pipeline = _make_concurrent_pipeline() + mock_spawn = MagicMock() + mock_spawn.return_value = MagicMock(container_info=MagicMock(container_id="abc123")) + + executor = ConcurrentPhaseExecutor(pipeline=pipeline, spawn_fn=mock_spawn) + executor._spawn_agent(AgentRole.CODER, prompt_text="Do the work") + + mock_spawn.assert_called_once() + call_kwargs = mock_spawn.call_args + command = call_kwargs.kwargs.get("command") or call_kwargs[1].get("command") + assert command[0] == "bash" + assert command[1] == "-c" + assert "claude" in command[2] + assert "egg-orch signal readiness" in command[2] + + +class TestImplicitReadyOnCleanExit: + """Tests for auto-registering READY when containers exit cleanly.""" + + def test_clean_exit_registers_ready(self): + """Container exiting with code 0 should auto-register as READY.""" + from consensus import ReadinessState, get_consensus_evaluator + + evaluator = get_consensus_evaluator() + pipeline_id = "test-implicit-ready" + + # Register an agent as WORKING + evaluator.register_agent(pipeline_id, "tester") + state = evaluator.evaluate(pipeline_id) + assert not state["is_complete"] + assert "tester" in state["blocking_agents"] + + # Simulate what the orchestrator does on clean exit: + # check state and auto-register READY + current = evaluator.evaluate(pipeline_id) + agent_state = current.get("agents", {}).get("tester") + if agent_state and agent_state.state != ReadinessState.READY: + evaluator.update_readiness( + pipeline_id, + "tester", + ReadinessState.READY, + reason="Container exited cleanly (implicit READY)", + ) + + state = evaluator.evaluate(pipeline_id) + assert state["is_complete"] + assert "tester" not in state["blocking_agents"] + + # Cleanup + evaluator.clear(pipeline_id) + + def test_already_ready_agent_not_overwritten(self): + """If agent already signaled READY, implicit READY is a no-op.""" + from consensus import ReadinessState, get_consensus_evaluator + + evaluator = get_consensus_evaluator() + pipeline_id = "test-implicit-noop" + + evaluator.register_agent(pipeline_id, "coder") + evaluator.update_readiness( + pipeline_id, "coder", ReadinessState.READY, reason="Explicit READY" + ) + + # Simulate implicit READY logic + current = evaluator.evaluate(pipeline_id) + agent_state = current.get("agents", {}).get("coder") + # Should skip because already READY + assert agent_state.state == ReadinessState.READY + + # Cleanup + evaluator.clear(pipeline_id) diff --git a/orchestrator/tests/test_consensus_wrapper.py b/orchestrator/tests/test_consensus_wrapper.py new file mode 100644 index 0000000000..b3108be9a0 --- /dev/null +++ b/orchestrator/tests/test_consensus_wrapper.py @@ -0,0 +1,204 @@ +"""Tests for the consensus wrapper module.""" + +import os +import shlex +import subprocess +import tempfile + +from consensus_wrapper import _CONSENSUS_WRAPPER_TEMPLATE, build_consensus_wrapped_command + + +class TestBuildConsensusWrappedCommand: + """Tests for build_consensus_wrapped_command().""" + + def test_returns_bash_command(self): + """Command should be a bash -c invocation.""" + cmd = build_consensus_wrapped_command("Do something") + assert cmd[0] == "bash" + assert cmd[1] == "-c" + assert len(cmd) == 3 + + def test_contains_claude_invocation(self): + """The wrapper script should contain the full claude command.""" + cmd = build_consensus_wrapped_command("Test prompt") + script = cmd[2] + assert "claude" in script + assert "--dangerously-skip-permissions" in script + assert "--print" in script + assert "--max-turns" in script + assert "200" in script + + def test_prompt_is_shell_escaped(self): + """Prompts with special characters should be properly escaped.""" + prompt = 'Test "quotes" and $variables and $(commands)' + cmd = build_consensus_wrapped_command(prompt) + script = cmd[2] + # The prompt should appear shell-quoted in the script + escaped = shlex.quote(prompt) + assert escaped in script + + def test_contains_consensus_wait_loop(self): + """The wrapper should include the consensus polling loop.""" + cmd = build_consensus_wrapped_command("Do something") + script = cmd[2] + assert "egg-orch signal readiness" in script + assert "READY" in script + assert "egg-orch message poll" in script + assert "EGG_CONCURRENT_MODE" in script + + def test_skips_consensus_when_not_concurrent(self): + """Script should exit normally when EGG_CONCURRENT_MODE is not set.""" + cmd = build_consensus_wrapped_command("Do something") + script = cmd[2] + # Should check EGG_CONCURRENT_MODE and exit early if not set + assert "EGG_CONCURRENT_MODE" in script + assert "exit $CLAUDE_EXIT" in script + + def test_custom_model_and_max_turns(self): + """Should support custom model and max_turns.""" + cmd = build_consensus_wrapped_command("Prompt", model="sonnet", max_turns=50) + script = cmd[2] + assert "--model" in script + assert shlex.quote("sonnet") in script + assert shlex.quote("50") in script + + def test_consensus_check_parses_json(self): + """The script should parse JSON response to check is_complete.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert "is_complete" in script + assert "python3" in script + + def test_has_timeout(self): + """The consensus wait loop should have a timeout.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert "TIMEOUT" in script + + def test_nonzero_exit_does_not_signal_ready(self): + """On non-zero Claude exit, wrapper must NOT signal READY.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + # The script should check CLAUDE_EXIT != 0 and exit early + assert 'if [ "$CLAUDE_EXIT" -ne 0 ]' in script + assert "NOT signaling READY" in script + + def test_clean_exit_signals_ready(self): + """On zero Claude exit, wrapper should signal READY.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert "Agent exited cleanly" in script + assert "auto-signaling READY" in script + + def test_timeout_configurable_via_env_var(self): + """Wrapper timeout should be configurable via EGG_CONSENSUS_WRAPPER_TIMEOUT.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert "EGG_CONSENSUS_WRAPPER_TIMEOUT" in script + + +class TestConsensusWrapperBehavior: + """Behavioral tests that run the wrapper script in a subprocess. + + These exercise the actual bash logic rather than just checking for + string patterns in the generated script. + """ + + @staticmethod + def _make_mock_egg_orch(tmpdir: str, log_file: str) -> str: + """Create a mock egg-orch script that logs calls and returns consensus JSON.""" + mock_path = os.path.join(tmpdir, "egg-orch") + with open(mock_path, "w") as f: + f.write("#!/bin/bash\n") + f.write(f'echo "$@" >> {shlex.quote(log_file)}\n') + # Return JSON with is_complete=true so the polling loop exits fast + f.write('echo \'{"data": {"consensus": {"is_complete": true}}}\'\n') + os.chmod(mock_path, 0o755) # nosec B103 - test helper needs executable permissions + return mock_path + + @staticmethod + def _run_wrapper( + claude_command: str, tmpdir: str, timeout: int = 10 + ) -> subprocess.CompletedProcess: + """Run the wrapper script with a substituted claude command.""" + script = _CONSENSUS_WRAPPER_TEMPLATE.format(claude_command=claude_command) + env = os.environ.copy() + env["PATH"] = f"{tmpdir}:{env.get('PATH', '')}" + env["EGG_CONCURRENT_MODE"] = "true" + env["EGG_CONSENSUS_WRAPPER_TIMEOUT"] = "2" + env["EGG_MESSAGE_POLL_INTERVAL"] = "1" + return subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + timeout=timeout, + ) + + def test_nonzero_exit_does_not_call_readiness(self): + """A non-zero Claude exit must not invoke egg-orch signal readiness.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + self._make_mock_egg_orch(tmpdir, log_file) + + # Use 'false' (returns 1) instead of 'exit 1' which would exit the shell + result = self._run_wrapper("false", tmpdir) + + assert result.returncode == 1 + # egg-orch should never have been called at all + assert not os.path.exists(log_file), ( + f"egg-orch was called on non-zero exit: " + f"{open(log_file).read() if os.path.exists(log_file) else ''}" + ) + assert "NOT signaling READY" in result.stdout + + def test_clean_exit_calls_readiness(self): + """A zero Claude exit must invoke egg-orch signal readiness.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + self._make_mock_egg_orch(tmpdir, log_file) + + # Use 'true' (returns 0) instead of 'exit 0' which would exit the shell + result = self._run_wrapper("true", tmpdir) + + assert result.returncode == 0 + assert os.path.exists(log_file) + with open(log_file) as f: + log_content = f.read() + assert "signal readiness --state READY" in log_content + assert "Auto-signaling READY" in result.stdout + + def test_nonzero_exit_propagates_exit_code(self): + """Wrapper must propagate the original non-zero exit code.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + self._make_mock_egg_orch(tmpdir, log_file) + + # Use a subshell to produce a specific exit code + result = self._run_wrapper("(exit 42)", tmpdir) + + assert result.returncode == 42 + + def test_non_concurrent_mode_skips_consensus(self): + """Without EGG_CONCURRENT_MODE=true, wrapper exits without consensus logic.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + self._make_mock_egg_orch(tmpdir, log_file) + + # Use 'true' instead of 'exit 0' which would exit the shell + script = _CONSENSUS_WRAPPER_TEMPLATE.format(claude_command="true") + env = os.environ.copy() + env["PATH"] = f"{tmpdir}:{env.get('PATH', '')}" + env.pop("EGG_CONCURRENT_MODE", None) + + result = subprocess.run( + ["bash", "-c", script], + env=env, + capture_output=True, + text=True, + timeout=10, + ) + + assert result.returncode == 0 + # No egg-orch calls should have been made + assert not os.path.exists(log_file)