Enforce consensus protocol for concurrent agents

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,

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]

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

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