diff --git a/docs/guides/concurrent-execution.md b/docs/guides/concurrent-execution.md index c398d29e8c..de83783330 100644 --- a/docs/guides/concurrent-execution.md +++ b/docs/guides/concurrent-execution.md @@ -23,8 +23,6 @@ Relevant `PipelineConfig` fields: | `consensus_timeout_minutes` | `30` | Timeout before HITL escalation | | `agent_idle_timeout_minutes` | `60` | Idle agent timeout before termination | -Agent containers also respect `EGG_CONSENSUS_WRAPPER_TIMEOUT` (default: `300` seconds) — see [Consensus Wrapper](#consensus-wrapper) below. - ## Agent Startup Protocol When concurrent execution starts for the implement phase, the `ConcurrentPhaseExecutor` (in `orchestrator/concurrent_executor.py`) spawns the following roles simultaneously using a `ThreadPoolExecutor`: @@ -49,28 +47,24 @@ Each agent is registered in the consensus evaluator before spawning begins. ## Consensus Wrapper -All concurrent agent containers are wrapped with a shell safety net defined in `orchestrator/consensus_wrapper.py`. The wrapper runs after the Claude process exits and enforces lifecycle compliance for agents that exit before the orchestrator stops them. +All concurrent agent containers are wrapped with a shell script defined in `orchestrator/consensus_wrapper.py`. The wrapper detects when Claude exits without the orchestrator confirming consensus and restarts the agent with a recovery prompt instead of silently marking it as ready. **How it works:** -1. Claude runs inside the wrapper script. -2. When Claude exits cleanly (code 0), the wrapper auto-signals `READY` via `egg-orch signal readiness` (a no-op if the agent already signaled). -3. The wrapper then polls the consensus endpoint in a loop, sleeping `EGG_MESSAGE_POLL_INTERVAL` seconds between checks, until consensus is reached or `EGG_CONSENSUS_WRAPPER_TIMEOUT` expires. -4. When consensus is reached (`is_complete: true`), the wrapper exits with Claude's original exit code. -5. If Claude exits non-zero (crashed), the wrapper does **not** signal `READY` and exits immediately with the same code. - -**Environment variables:** - -| Variable | Default | Description | -|----------|---------|-------------| -| `EGG_CONSENSUS_WRAPPER_TIMEOUT` | `300` | Seconds the wrapper polls for consensus before giving up | -| `EGG_MESSAGE_POLL_INTERVAL` | `30` | Seconds between wrapper poll iterations | +1. Claude runs inside the wrapper script with the original task prompt. +2. If Claude exits non-zero (crashed), the wrapper exits immediately with the same code — no restart. +3. If Claude exits cleanly (code 0), the wrapper restarts Claude with a **recovery prompt** that explains the agent was restarted because it exited without signaling `READY`. The recovery prompt instructs the agent to poll for messages, assess state, and explicitly signal `READY` or continue working. +4. Restarts are capped at `MAX_CONSENSUS_RESTARTS` (default: 2). After each restart, the wrapper checks if consensus was reached. If so, it exits cleanly. +5. After exhausting all restarts, the wrapper exits with code 1, triggering the orchestrator's agent failure path (HITL decision with retry/abort/continue options). -**Implicit READY on clean exit (orchestrator-side):** +**Key design principle:** Agents must **explicitly** participate in consensus. The wrapper never auto-signals `READY` on behalf of an agent — it restarts the agent so it can assess state and signal for itself. -In addition to the wrapper, the orchestrator's `_run_concurrent_phase()` monitors container exit codes. When a container exits with code 0 and has not yet signaled `READY`, the orchestrator auto-registers it as `READY` with the reason `"Container exited cleanly (implicit READY)"`. This ensures agents that complete their work and exit without explicitly signaling do not block consensus indefinitely. +**Configuration:** -**Agents should still follow the protocol.** The wrapper and implicit READY are safety nets for cases where Claude's exit is unavoidable (e.g., max turns reached, context exhausted). Well-behaved agents explicitly signal `READY` and enter a polling loop — this allows them to react to late-arriving messages before the orchestrator stops the container. +| Parameter | Default | Description | +|-----------|---------|-------------| +| `max_restarts` | `2` | Maximum restart attempts (passed to `build_consensus_wrapped_command()`) | +| `EGG_MESSAGE_POLL_INTERVAL` | `30` | Seconds between message polls during restarts | ## Message Bus diff --git a/orchestrator/consensus_wrapper.py b/orchestrator/consensus_wrapper.py index 8b8f4fdcec..90993d773a 100644 --- a/orchestrator/consensus_wrapper.py +++ b/orchestrator/consensus_wrapper.py @@ -2,23 +2,73 @@ 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. +that detects early Claude exits and restarts the agent with a recovery prompt +instead of blindly marking consensus as approved. + +If the agent exits without signaling READY, the wrapper restarts Claude with +a prompt that explains what happened and instructs it to assess state, then +either signal READY or continue working. Restarts are capped at +``MAX_CONSENSUS_RESTARTS`` (default 2). After exhausting restarts the wrapper +exits with code 1 so the orchestrator's failure path handles escalation. """ 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. +# Default maximum number of times the wrapper will restart Claude after a +# clean exit without consensus being reached. +MAX_CONSENSUS_RESTARTS = 2 + +# Default maximum number of poll cycles to wait for consensus when the agent +# already signaled READY. With a default poll interval of 30s, this gives +# 10 * 30 = 300 seconds (5 minutes) for other agents to finish. +MAX_READY_POLL_CYCLES = 10 + +# Recovery prompt given to Claude when it is restarted by the wrapper. +# Placeholders: {restart_number}, {max_restarts} +_RECOVERY_PROMPT = ( + "## CONSENSUS RECOVERY — You were restarted by the consensus wrapper\n\n" + "You exited your previous session without the orchestrator confirming " + "consensus. This is restart {restart_number} of {max_restarts}.\n\n" + "**What happened**: Your agent process exited cleanly, but the consensus " + "protocol requires you to remain alive until ALL agents signal READY and " + "the orchestrator stops your container. Because you exited early, the " + "wrapper restarted you so you can finish the protocol.\n\n" + "**What you must do now**:\n" + "1. Poll for messages: `egg-orch message poll`\n" + "2. Check if your work is complete — review any new commits or feedback " + "from other agents.\n" + "3. If your work is done, signal READY: " + '`egg-orch signal readiness --state READY --reason "Work complete"`\n' + "4. If there is new feedback or work to address, handle it first, then " + "signal READY.\n" + "5. **Stay alive** — keep polling with `egg-orch message poll` in a loop. " + "Do NOT exit. The orchestrator will send SIGTERM when consensus is reached.\n\n" + "**If you exit again without signaling READY, you will be restarted again " + "(up to the maximum). After that, your role will be left without consensus " + "and the orchestrator will need to handle it.**\n" +) + +# Shell script that wraps the Claude CLI invocation. After Claude exits: +# - Non-concurrent mode: exit normally. +# - Non-zero exit: treat as failure, no restart. +# - Clean exit (code 0): restart Claude with a recovery prompt (up to +# MAX_RESTARTS times). After max restarts, exit 1 to trigger the +# orchestrator's agent failure path (HITL decision). _CONSENSUS_WRAPPER_TEMPLATE = r""" #!/bin/bash set -uo pipefail -# Run the Claude agent -{claude_command} +MAX_RESTARTS={max_restarts} +RESTART_COUNT=0 + +run_claude() {{ + local prompt="$1" + {claude_command_prefix} "$prompt" + return $? +}} + +# --- Initial run --- +run_claude {initial_prompt} CLAUDE_EXIT=$? # If not in concurrent mode, exit normally @@ -26,49 +76,88 @@ 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. +# Non-zero exit means the agent crashed — do not restart or signal READY. if [ "$CLAUDE_EXIT" -ne 0 ]; then - echo "[consensus-wrapper] Agent failed (code $CLAUDE_EXIT). NOT signaling READY." + echo "[consensus-wrapper] Agent failed (code $CLAUDE_EXIT). NOT restarting." 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 +# --- Check if consensus is already complete or agent already signaled READY --- +# If the agent signaled READY but then exited (e.g., context exhaustion), +# restarting is unnecessary. Query pipeline status for consensus state. +MAX_READY_POLLS={max_ready_polls} +RESPONSE=$(egg-orch pipeline status --json 2>/dev/null || echo "{{}}") +IS_COMPLETE=$(echo "$RESPONSE" | python3 -c \ + "import sys,json; d=json.load(sys.stdin); print(d.get('data',{{}}).get('concurrent',{{}}).get('consensus',{{}}).get('is_complete',False))" \ + 2>/dev/null || echo "False") +if [ "$IS_COMPLETE" = "True" ]; then + echo "[consensus-wrapper] Consensus already reached. Exiting." + exit 0 +fi -# Stay alive polling for consensus until reached or timeout. -POLL_INTERVAL="${{EGG_MESSAGE_POLL_INTERVAL:-30}}" -TIMEOUT="${{EGG_CONSENSUS_WRAPPER_TIMEOUT:-300}}" -ELAPSED=0 +# Check if this agent already signaled READY (uses EGG_AGENT_ROLE env var) +AGENT_ROLE="${{EGG_AGENT_ROLE:-}}" +if [ -n "$AGENT_ROLE" ]; then + AGENT_STATE=$(echo "$RESPONSE" | python3 -c \ + "import sys,json; d=json.load(sys.stdin); agents=d.get('data',{{}}).get('concurrent',{{}}).get('consensus',{{}}).get('agents',{{}}); print(agents.get('$AGENT_ROLE',{{}}).get('state',''))" \ + 2>/dev/null || echo "") + if [ "$AGENT_STATE" = "READY" ]; then + echo "[consensus-wrapper] Agent already signaled READY. Skipping restart, waiting for consensus..." + POLL_INTERVAL="${{EGG_MESSAGE_POLL_INTERVAL:-30}}" + WAIT_COUNT=0 + while [ "$WAIT_COUNT" -lt "$MAX_READY_POLLS" ]; do + WAIT_COUNT=$((WAIT_COUNT + 1)) + sleep "$POLL_INTERVAL" + RESPONSE=$(egg-orch pipeline status --json 2>/dev/null || echo "{{}}") + IS_COMPLETE=$(echo "$RESPONSE" | python3 -c \ + "import sys,json; d=json.load(sys.stdin); print(d.get('data',{{}}).get('concurrent',{{}}).get('consensus',{{}}).get('is_complete',False))" \ + 2>/dev/null || echo "False") + if [ "$IS_COMPLETE" = "True" ]; then + echo "[consensus-wrapper] Consensus reached. Exiting." + exit 0 + fi + done + echo "[consensus-wrapper] Agent was READY but consensus not reached. Exiting cleanly." + exit 0 + fi +fi -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 "{{}}") +# --- Restart loop for clean exits without consensus --- +while [ "$RESTART_COUNT" -lt "$MAX_RESTARTS" ]; do + RESTART_COUNT=$((RESTART_COUNT + 1)) + echo "[consensus-wrapper] Agent exited without consensus. Restarting ($RESTART_COUNT/$MAX_RESTARTS)..." - 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") + # Build recovery prompt with restart context + RECOVERY_PROMPT=$(cat <<'RECOVERY_EOF' +{recovery_prompt_template} +RECOVERY_EOF +) + # Substitute restart number into the prompt + RECOVERY_PROMPT=$(echo "$RECOVERY_PROMPT" | sed "s/{{restart_number}}/$RESTART_COUNT/g; s/{{max_restarts}}/$MAX_RESTARTS/g") - if [ "$IS_COMPLETE" = "True" ]; then - echo "[consensus-wrapper] Consensus reached. Exiting." + run_claude "$RECOVERY_PROMPT" + CLAUDE_EXIT=$? + + if [ "$CLAUDE_EXIT" -ne 0 ]; then + echo "[consensus-wrapper] Agent failed on restart $RESTART_COUNT (code $CLAUDE_EXIT). Stopping." exit $CLAUDE_EXIT fi - # Poll for messages (may contain work that should have been handled) - egg-orch message poll 2>/dev/null || true + # Check if consensus was reached during the restart + RESPONSE=$(egg-orch pipeline status --json 2>/dev/null || echo "{{}}") + IS_COMPLETE=$(echo "$RESPONSE" | python3 -c \ + "import sys,json; d=json.load(sys.stdin); print(d.get('data',{{}}).get('concurrent',{{}}).get('consensus',{{}}).get('is_complete',False))" \ + 2>/dev/null || echo "False") - sleep "$POLL_INTERVAL" - ELAPSED=$((ELAPSED + POLL_INTERVAL)) + if [ "$IS_COMPLETE" = "True" ]; then + echo "[consensus-wrapper] Consensus reached after restart $RESTART_COUNT. Exiting." + exit 0 + fi done -echo "[consensus-wrapper] Consensus not reached within ${{TIMEOUT}}s. Exiting." -exit $CLAUDE_EXIT +# --- Max restarts exhausted: shut down with failure --- +echo "[consensus-wrapper] Max restarts ($MAX_RESTARTS) exhausted. Agent never signaled READY. Exiting with failure." +exit 1 """ @@ -76,22 +165,28 @@ def build_consensus_wrapped_command( prompt_text: str, model: str = "opus", max_turns: int = 200, + max_restarts: int = MAX_CONSENSUS_RESTARTS, + max_ready_polls: int = MAX_READY_POLL_CYCLES, ) -> list[str]: - """Build a shell command that runs Claude with a consensus wait wrapper. + """Build a shell command that runs Claude with a consensus restart 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. + The wrapper detects when Claude exits without consensus and restarts it + with a recovery prompt instead of auto-signaling READY. This ensures + agents explicitly participate in consensus rather than having it faked. Args: prompt_text: The prompt to pass to the Claude CLI. model: Claude model to use. max_turns: Maximum number of tool-call turns. + max_restarts: Maximum restart attempts before exiting with failure. + max_ready_polls: Maximum poll cycles to wait when agent already + signaled READY (avoids unnecessary restarts). Returns: Command list suitable for container spawning (bash -c "..."). """ - claude_parts = [ + # Build the claude command prefix (everything except the prompt argument) + claude_prefix_parts = [ "claude", "--dangerously-skip-permissions", "--print", @@ -102,9 +197,16 @@ def build_consensus_wrapped_command( 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) + claude_command_prefix = " ".join(shlex.quote(p) for p in claude_prefix_parts) + initial_prompt = shlex.quote(prompt_text) + + script = _CONSENSUS_WRAPPER_TEMPLATE.format( + claude_command_prefix=claude_command_prefix, + initial_prompt=initial_prompt, + max_restarts=max_restarts, + max_ready_polls=max_ready_polls, + recovery_prompt_template=_RECOVERY_PROMPT, + ) return ["bash", "-c", script] diff --git a/orchestrator/routes/coordinator.py b/orchestrator/routes/coordinator.py index 40b9098bec..d2a74f92d0 100644 --- a/orchestrator/routes/coordinator.py +++ b/orchestrator/routes/coordinator.py @@ -303,8 +303,8 @@ def spawn_agent(pipeline_id: str) -> tuple[Response, int]: ) # 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. + # restarts the agent with a recovery prompt if Claude exits + # before the orchestrator confirms consensus. agent_command = build_consensus_wrapped_command(agent_prompt) # Spawn the container diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 7c6c8b53a8..745a0aa396 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -703,6 +703,12 @@ def update_pipeline(pipeline_id: str) -> tuple[Response, int]: pipeline_id=pipeline_id, error=str(e), ) + # Reload pipeline so the response reflects current state + # rather than the stale pre-cleanup object. + try: + pipeline = store.load_pipeline(pipeline_id) + except Exception: + pass # Use stale pipeline if reload also fails logger.info("Pipeline updated", pipeline_id=pipeline_id) @@ -4779,33 +4785,15 @@ def _update_agents_complete() -> None: 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), - ) + # Clean exit (code 0): the consensus wrapper inside the + # container handles restarts if the agent didn't signal + # READY. We do NOT auto-register READY here — agents + # must explicitly participate in consensus. + logger.info( + "Container exited cleanly, wrapper handles consensus", + pipeline_id=pipeline_id, + role=exec_info.role.value, + ) # 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 afe4b5a2e0..70145a6b6f 100644 --- a/orchestrator/tests/test_concurrent_integration.py +++ b/orchestrator/tests/test_concurrent_integration.py @@ -601,18 +601,23 @@ def test_spawn_agent_uses_wrapped_command(self): assert command[0] == "bash" assert command[1] == "-c" assert "claude" in command[2] - assert "egg-orch signal readiness" in command[2] + assert "RESTART_COUNT" in command[2] + assert "CONSENSUS RECOVERY" in command[2] -class TestImplicitReadyOnCleanExit: - """Tests for auto-registering READY when containers exit cleanly.""" +class TestNoImplicitReadyOnCleanExit: + """Verify that clean container exits do NOT auto-register READY. - def test_clean_exit_registers_ready(self): - """Container exiting with code 0 should auto-register as READY.""" + The consensus wrapper restarts the agent instead. The orchestrator + must not fake consensus on behalf of agents. + """ + + def test_clean_exit_does_not_register_ready(self): + """Container exiting with code 0 should NOT auto-register as READY.""" from consensus import ReadinessState, get_consensus_evaluator evaluator = get_consensus_evaluator() - pipeline_id = "test-implicit-ready" + pipeline_id = "test-no-implicit-ready" # Register an agent as WORKING evaluator.register_agent(pipeline_id, "tester") @@ -620,42 +625,33 @@ def test_clean_exit_registers_ready(self): 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)", - ) - + # The orchestrator should NOT auto-register READY on clean exit. + # The agent must remain blocking until it explicitly signals. 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" + assert not state["is_complete"] + assert "tester" in state["blocking_agents"] - evaluator.register_agent(pipeline_id, "coder") + # Only explicit READY from the agent should complete consensus evaluator.update_readiness( - pipeline_id, "coder", ReadinessState.READY, reason="Explicit READY" + pipeline_id, + "tester", + ReadinessState.READY, + reason="Agent explicitly signaled 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 + state = evaluator.evaluate(pipeline_id) + assert state["is_complete"] # Cleanup evaluator.clear(pipeline_id) + + def test_wrapper_contains_restart_logic(self): + """The consensus wrapper should restart agents, not auto-signal READY.""" + from consensus_wrapper import build_consensus_wrapped_command + + cmd = build_consensus_wrapped_command("Do work") + script = cmd[2] + # Must contain restart logic + assert "Restarting" in script + assert "RESTART_COUNT" in script + # Must NOT contain auto-READY + assert "Auto-signaling READY" not in script diff --git a/orchestrator/tests/test_consensus_wrapper.py b/orchestrator/tests/test_consensus_wrapper.py index b3108be9a0..3eb1933710 100644 --- a/orchestrator/tests/test_consensus_wrapper.py +++ b/orchestrator/tests/test_consensus_wrapper.py @@ -5,7 +5,12 @@ import subprocess import tempfile -from consensus_wrapper import _CONSENSUS_WRAPPER_TEMPLATE, build_consensus_wrapped_command +from consensus_wrapper import ( + _RECOVERY_PROMPT, + MAX_CONSENSUS_RESTARTS, + MAX_READY_POLL_CYCLES, + build_consensus_wrapped_command, +) class TestBuildConsensusWrappedCommand: @@ -37,20 +42,26 @@ def test_prompt_is_shell_escaped(self): escaped = shlex.quote(prompt) assert escaped in script - def test_contains_consensus_wait_loop(self): - """The wrapper should include the consensus polling loop.""" + def test_contains_restart_logic(self): + """The wrapper should include restart logic, not auto-READY.""" cmd = build_consensus_wrapped_command("Do something") script = cmd[2] - assert "egg-orch signal readiness" in script - assert "READY" in script + assert "Restarting" in script + assert "RESTART_COUNT" in script + assert "MAX_RESTARTS" in script assert "egg-orch message poll" in script assert "EGG_CONCURRENT_MODE" in script + def test_does_not_auto_signal_ready(self): + """The wrapper must NOT auto-signal READY on clean exit.""" + cmd = build_consensus_wrapped_command("Do something") + script = cmd[2] + assert "Auto-signaling READY" not 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 @@ -63,38 +74,101 @@ def test_custom_model_and_max_turns(self): assert shlex.quote("50") in script def test_consensus_check_parses_json(self): - """The script should parse JSON response to check is_complete.""" + """The script should use pipeline status and parse nested consensus JSON.""" cmd = build_consensus_wrapped_command("Prompt") script = cmd[2] + assert "egg-orch pipeline status --json" in script assert "is_complete" in script assert "python3" in script + # Must use the correct nested path: data.concurrent.consensus + assert "concurrent" in script - def test_has_timeout(self): - """The consensus wait loop should have a timeout.""" + def test_has_max_restarts(self): + """The wrapper should cap restart attempts via MAX_RESTARTS.""" cmd = build_consensus_wrapped_command("Prompt") script = cmd[2] - assert "TIMEOUT" in script + assert "MAX_RESTARTS" in script - def test_nonzero_exit_does_not_signal_ready(self): - """On non-zero Claude exit, wrapper must NOT signal READY.""" + def test_nonzero_exit_does_not_restart(self): + """On non-zero Claude exit, wrapper must NOT restart.""" 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 + assert "NOT restarting" in script - def test_clean_exit_signals_ready(self): - """On zero Claude exit, wrapper should signal READY.""" + def test_contains_recovery_prompt(self): + """The wrapper should contain the recovery prompt text.""" cmd = build_consensus_wrapped_command("Prompt") script = cmd[2] - assert "Agent exited cleanly" in script - assert "auto-signaling READY" in script + assert "CONSENSUS RECOVERY" in script + assert "You were restarted" in script - def test_timeout_configurable_via_env_var(self): - """Wrapper timeout should be configurable via EGG_CONSENSUS_WRAPPER_TIMEOUT.""" + def test_exits_with_failure_after_max_restarts(self): + """After exhausting restarts, wrapper should exit 1 (not wait passively).""" cmd = build_consensus_wrapped_command("Prompt") script = cmd[2] - assert "EGG_CONSENSUS_WRAPPER_TIMEOUT" in script + assert "Exiting with failure" in script + assert "exit 1" in script + + def test_custom_max_restarts(self): + """Should support custom max_restarts parameter.""" + cmd = build_consensus_wrapped_command("Prompt", max_restarts=5) + script = cmd[2] + assert "MAX_RESTARTS=5" in script + + def test_default_max_restarts(self): + """Default max_restarts should match module constant.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert f"MAX_RESTARTS={MAX_CONSENSUS_RESTARTS}" in script + + def test_recovery_prompt_has_placeholders(self): + """Recovery prompt should contain restart number placeholders.""" + assert "{restart_number}" in _RECOVERY_PROMPT + assert "{max_restarts}" in _RECOVERY_PROMPT + # {role} was removed — it is not used in the prompt + assert "{role}" not in _RECOVERY_PROMPT + + def test_contains_ready_check_before_restart(self): + """Wrapper should check if agent already signaled READY before restarting.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert "already signaled READY" in script + assert "EGG_AGENT_ROLE" in script + + def test_ready_polling_uses_separate_constant(self): + """READY polling loop should use MAX_READY_POLLS, not MAX_RESTARTS.""" + cmd = build_consensus_wrapped_command("Prompt", max_ready_polls=15) + script = cmd[2] + assert "MAX_READY_POLLS=15" in script + + def test_default_max_ready_polls(self): + """Default max_ready_polls should match module constant.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + assert f"MAX_READY_POLLS={MAX_READY_POLL_CYCLES}" in script + + +def _make_mock_claude(tmpdir: str, claude_log_file: str | None = None, exit_code: int = 0) -> None: + """Create a mock claude script. + + Args: + tmpdir: Directory to create the mock in (must be on PATH). + claude_log_file: File to log calls to. If None, logs to tmpdir/claude.log. + exit_code: Exit code for the mock. When 0, logs call details; + when non-zero, exits immediately with that code. + """ + mock_claude = os.path.join(tmpdir, "claude") + claude_log = claude_log_file or os.path.join(tmpdir, "claude.log") + with open(mock_claude, "w") as f: + f.write("#!/bin/bash\n") + if exit_code != 0: + f.write(f"exit {exit_code}\n") + else: + f.write(f'echo "---CLAUDE_CALL_START---" >> {shlex.quote(claude_log)}\n') + f.write(f'echo "${{@: -1}}" >> {shlex.quote(claude_log)}\n') + f.write(f'echo "---CLAUDE_CALL_END---" >> {shlex.quote(claude_log)}\n') + os.chmod(mock_claude, 0o755) # nosec B103 class TestConsensusWrapperBehavior: @@ -105,100 +179,293 @@ class TestConsensusWrapperBehavior: """ @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: + def _make_mock_tools(tmpdir: str, log_file: str, claude_log_file: str | None = None) -> None: + """Create mock egg-orch and claude scripts. + + The mock claude script logs a delimiter + its prompt arg and exits 0. + The mock egg-orch logs calls and returns consensus-complete JSON + matching the real ``egg-orch pipeline status`` response structure. + """ + # Mock egg-orch + mock_orch = os.path.join(tmpdir, "egg-orch") + with open(mock_orch, "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 + f.write('echo \'{"data": {"concurrent": {"consensus": {"is_complete": true}}}}\'\n') + os.chmod(mock_orch, 0o755) # nosec B103 + + _make_mock_claude(tmpdir, claude_log_file) @staticmethod - def _run_wrapper( - claude_command: str, tmpdir: str, timeout: int = 10 + def _make_failing_claude(tmpdir: str, exit_code: int = 1) -> None: + """Create a mock claude that exits with a non-zero code.""" + _make_mock_claude(tmpdir, exit_code=exit_code) + + @staticmethod + def _run_wrapper_command( + cmd: list[str], + tmpdir: str, + timeout: int = 15, + concurrent: bool = True, + agent_role: str | None = None, ) -> subprocess.CompletedProcess: - """Run the wrapper script with a substituted claude command.""" - script = _CONSENSUS_WRAPPER_TEMPLATE.format(claude_command=claude_command) + """Run a wrapper command with test environment.""" env = os.environ.copy() env["PATH"] = f"{tmpdir}:{env.get('PATH', '')}" - env["EGG_CONCURRENT_MODE"] = "true" + if concurrent: + env["EGG_CONCURRENT_MODE"] = "true" + else: + env.pop("EGG_CONCURRENT_MODE", None) + if agent_role: + env["EGG_AGENT_ROLE"] = agent_role + else: + env.pop("EGG_AGENT_ROLE", None) env["EGG_CONSENSUS_WRAPPER_TIMEOUT"] = "2" env["EGG_MESSAGE_POLL_INTERVAL"] = "1" return subprocess.run( - ["bash", "-c", script], + cmd, 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.""" + def test_nonzero_exit_does_not_restart(self): + """A non-zero Claude exit must not trigger restart or egg-orch calls.""" with tempfile.TemporaryDirectory() as tmpdir: log_file = os.path.join(tmpdir, "egg-orch.log") - self._make_mock_egg_orch(tmpdir, log_file) + self._make_mock_tools(tmpdir, log_file) + self._make_failing_claude(tmpdir, exit_code=1) - # Use 'false' (returns 1) instead of 'exit 1' which would exit the shell - result = self._run_wrapper("false", tmpdir) + cmd = build_consensus_wrapped_command("Do the work", max_restarts=2) + result = self._run_wrapper_command(cmd, 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 restarting" in result.stdout + + @staticmethod + def _make_mock_tools_with_delayed_consensus( + tmpdir: str, + log_file: str, + claude_log_file: str | None = None, + consensus_after: int = 2, + ) -> None: + """Create mock tools where egg-orch returns is_complete=false initially. + + The mock egg-orch uses a counter file to track calls to 'pipeline status'. + It returns is_complete=false until the Nth 'pipeline status' call, then true. + Response structure matches real ``egg-orch pipeline status --json`` output. + """ + counter_file = os.path.join(tmpdir, "orch_status_count") + mock_orch = os.path.join(tmpdir, "egg-orch") + with open(mock_orch, "w") as f: + f.write("#!/bin/bash\n") + f.write(f'echo "$@" >> {shlex.quote(log_file)}\n') + # Only track 'pipeline status' calls for consensus gating + f.write('if echo "$@" | grep -q "pipeline status"; then\n') + f.write(" COUNT=0\n") + f.write(f" if [ -f {shlex.quote(counter_file)} ]; then\n") + f.write(f" COUNT=$(cat {shlex.quote(counter_file)})\n") + f.write(" fi\n") + f.write(" COUNT=$((COUNT + 1))\n") + f.write(f' echo "$COUNT" > {shlex.quote(counter_file)}\n') + f.write(f' if [ "$COUNT" -ge {consensus_after} ]; then\n') + f.write(' echo \'{"data": {"concurrent": {"consensus": {"is_complete": true}}}}\'\n') + f.write(" else\n") + f.write( + ' echo \'{"data": {"concurrent": {"consensus": {"is_complete": false}}}}\'\n' ) - assert "NOT signaling READY" in result.stdout + f.write(" fi\n") + f.write("else\n") + f.write(' echo \'{"data": {"concurrent": {"consensus": {"is_complete": false}}}}\'\n') + f.write("fi\n") + os.chmod(mock_orch, 0o755) # nosec B103 - def test_clean_exit_calls_readiness(self): - """A zero Claude exit must invoke egg-orch signal readiness.""" + _make_mock_claude(tmpdir, claude_log_file) + + def test_clean_exit_triggers_restart(self): + """A zero Claude exit should trigger a restart, not auto-signal READY.""" with tempfile.TemporaryDirectory() as tmpdir: log_file = os.path.join(tmpdir, "egg-orch.log") - self._make_mock_egg_orch(tmpdir, log_file) + claude_log = os.path.join(tmpdir, "claude.log") + # Use delayed consensus: false on first status check, true on second + self._make_mock_tools_with_delayed_consensus( + tmpdir, + log_file, + claude_log, + consensus_after=2, + ) - # Use 'true' (returns 0) instead of 'exit 0' which would exit the shell - result = self._run_wrapper("true", tmpdir) + cmd = build_consensus_wrapped_command("Do the work", max_restarts=1) + result = self._run_wrapper_command(cmd, tmpdir) assert result.returncode == 0 - assert os.path.exists(log_file) - with open(log_file) as f: + assert "Restarting" in result.stdout + # Claude should have been called at least twice (initial + 1 restart) + with open(claude_log) as f: log_content = f.read() - assert "signal readiness --state READY" in log_content - assert "Auto-signaling READY" in result.stdout + call_count = log_content.count("---CLAUDE_CALL_START---") + assert call_count >= 2 + # Second call should contain recovery prompt content + assert "CONSENSUS RECOVERY" in log_content 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) + self._make_mock_tools(tmpdir, log_file) + self._make_failing_claude(tmpdir, exit_code=42) - # Use a subshell to produce a specific exit code - result = self._run_wrapper("(exit 42)", tmpdir) + cmd = build_consensus_wrapped_command("Prompt", max_restarts=2) + result = self._run_wrapper_command(cmd, tmpdir) assert result.returncode == 42 def test_non_concurrent_mode_skips_consensus(self): - """Without EGG_CONCURRENT_MODE=true, wrapper exits without consensus logic.""" + """Without EGG_CONCURRENT_MODE=true, wrapper exits without restart logic.""" with tempfile.TemporaryDirectory() as tmpdir: log_file = os.path.join(tmpdir, "egg-orch.log") - self._make_mock_egg_orch(tmpdir, log_file) + claude_log = os.path.join(tmpdir, "claude.log") + self._make_mock_tools(tmpdir, log_file, claude_log) - # 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) + cmd = build_consensus_wrapped_command("Do the work", max_restarts=2) + result = self._run_wrapper_command(cmd, tmpdir, concurrent=False) - result = subprocess.run( - ["bash", "-c", script], - env=env, - capture_output=True, - text=True, - timeout=10, + assert result.returncode == 0 + # Claude should only have been called once (no restart) + with open(claude_log) as f: + call_count = f.read().count("---CLAUDE_CALL_START---") + assert call_count == 1 + # No egg-orch calls + assert not os.path.exists(log_file) + + def test_max_restarts_respected(self): + """Wrapper should not restart more than max_restarts times.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + claude_log = os.path.join(tmpdir, "claude.log") + + # Mock egg-orch that never reports consensus complete + mock_orch = os.path.join(tmpdir, "egg-orch") + with open(mock_orch, "w") as f: + f.write("#!/bin/bash\n") + f.write(f'echo "$@" >> {shlex.quote(log_file)}\n') + f.write( + 'echo \'{"data": {"concurrent": {"consensus": {"is_complete": false}}}}\'\n' + ) + os.chmod(mock_orch, 0o755) # nosec B103 + + # Mock claude that always exits cleanly — uses a delimiter to count calls + mock_claude = os.path.join(tmpdir, "claude") + with open(mock_claude, "w") as f: + f.write("#!/bin/bash\n") + f.write(f'echo "---CLAUDE_CALL---" >> {shlex.quote(claude_log)}\n') + os.chmod(mock_claude, 0o755) # nosec B103 + + cmd = build_consensus_wrapped_command("Do the work", max_restarts=2) + result = self._run_wrapper_command(cmd, tmpdir, timeout=30) + + # Claude should have been called 3 times: initial + 2 restarts + with open(claude_log) as f: + call_count = f.read().count("---CLAUDE_CALL---") + assert call_count == 3 + assert "Max restarts (2) exhausted" in result.stdout + # Should exit with failure code after exhausting restarts + assert result.returncode == 1 + + @staticmethod + def _make_mock_tools_with_agent_ready_state( + tmpdir: str, + log_file: str, + claude_log_file: str | None = None, + agent_role: str = "coder", + consensus_after: int = 2, + ) -> None: + """Create mock tools where the agent is already READY but consensus is pending. + + The mock egg-orch returns per-agent state showing the agent as READY + with ``is_complete=false`` initially. After ``consensus_after`` calls + to ``pipeline status``, it returns ``is_complete=true``. This exercises + the READY polling path (skip restart, wait for consensus). + """ + counter_file = os.path.join(tmpdir, "orch_status_count") + mock_orch = os.path.join(tmpdir, "egg-orch") + # Build JSON strings with agent state — use string concatenation to + # avoid f-string brace escaping confusion. + json_incomplete = ( + '{"data": {"concurrent": {"consensus": {"is_complete": false, ' + '"agents": {"' + agent_role + '": {"state": "READY"}}}}}}' + ) + json_complete = ( + '{"data": {"concurrent": {"consensus": {"is_complete": true, ' + '"agents": {"' + agent_role + '": {"state": "READY"}}}}}}' + ) + with open(mock_orch, "w") as f: + f.write("#!/bin/bash\n") + f.write(f'echo "$@" >> {shlex.quote(log_file)}\n') + f.write('if echo "$@" | grep -q "pipeline status"; then\n') + f.write(" COUNT=0\n") + f.write(f" if [ -f {shlex.quote(counter_file)} ]; then\n") + f.write(f" COUNT=$(cat {shlex.quote(counter_file)})\n") + f.write(" fi\n") + f.write(" COUNT=$((COUNT + 1))\n") + f.write(f' echo "$COUNT" > {shlex.quote(counter_file)}\n') + f.write(f' if [ "$COUNT" -ge {consensus_after} ]; then\n') + f.write(f" echo '{json_complete}'\n") + f.write(" else\n") + f.write(f" echo '{json_incomplete}'\n") + f.write(" fi\n") + f.write("else\n") + f.write(' echo \'{"data": {"concurrent": {"consensus": {"is_complete": false}}}}\'\n') + f.write("fi\n") + os.chmod(mock_orch, 0o755) # nosec B103 + + _make_mock_claude(tmpdir, claude_log_file) + + def test_ready_agent_skips_restart_and_polls(self): + """Agent already READY should skip restart and poll for consensus.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + claude_log = os.path.join(tmpdir, "claude.log") + # Mock returns agent as READY, consensus false then true on 3rd call + self._make_mock_tools_with_agent_ready_state( + tmpdir, + log_file, + claude_log, + agent_role="coder", + consensus_after=3, ) + cmd = build_consensus_wrapped_command("Do the work", max_restarts=2, max_ready_polls=5) + result = self._run_wrapper_command(cmd, tmpdir, timeout=30, agent_role="coder") + + # Should exit cleanly assert result.returncode == 0 - # No egg-orch calls should have been made - assert not os.path.exists(log_file) + # Should detect agent is already READY and skip restart + assert "already signaled READY" in result.stdout + # Should eventually detect consensus + assert "Consensus reached" in result.stdout + # Claude should only be called once (no restart) + with open(claude_log) as f: + call_count = f.read().count("---CLAUDE_CALL_START---") + assert call_count == 1 + # Should NOT show any restart messages + assert "Restarting" not in result.stdout + + def test_no_auto_ready_on_clean_exit(self): + """Wrapper must NOT auto-signal READY — only restarts are allowed.""" + with tempfile.TemporaryDirectory() as tmpdir: + log_file = os.path.join(tmpdir, "egg-orch.log") + claude_log = os.path.join(tmpdir, "claude.log") + self._make_mock_tools(tmpdir, log_file, claude_log) + + cmd = build_consensus_wrapped_command("Do the work", max_restarts=1) + self._run_wrapper_command(cmd, tmpdir) + + # Check egg-orch calls — should not contain "signal readiness --state READY" + # from the wrapper itself (only the agent inside Claude should signal READY) + if os.path.exists(log_file): + with open(log_file) as f: + log_content = f.read() + # The wrapper should only call pipeline status, not signal READY + assert "signal readiness --state READY" not in log_content