Fix BRC consensus deadlock after tracker loss (#1259)

orchestrator/approval_matrix.py

@@ -156,12 +156,16 @@ def is_context_change_nack(
         return not bool(prev_refs & new_refs)
 
     def is_fully_acked(self, producer: str) -> bool:
-        """Check if all assigned reviewers have ACKed the producer's latest proposal."""
+        """Check if all critical reviewers have ACKed the producer's latest proposal.
+
+        Advisory reviewers are excluded from the check — their ACK is
+        informational but does not block consensus.
+        """
         latest_version = self._proposal_versions.get(producer, 0)
         if latest_version == 0:
             return False
 
-        reviewers = self._graph.reviewers_for(producer)
+        reviewers = self._graph.critical_reviewers_for(producer)
         for reviewer in reviewers:
             key = (reviewer, producer)
             entry = self._entries.get(key)

orchestrator/concurrent_executor.py

@@ -354,7 +354,36 @@ def check_consensus(self) -> dict[str, Any]:
                     pipeline_id=self.pipeline.id,
                 )
         if tracker:
-            return tracker.evaluate()
+            result = tracker.evaluate()
+            # Message-bus fallback: if reconstruction produced a tracker but
+            # evaluate() says not complete, check the message store directly.
+            # This handles the case where reconstruction replayed into an empty
+            # tracker state (RC1/RC5) but all roles have CONFIRMED messages.
+            if not result.get("is_complete"):
+                try:
+                    from message_store import get_message_store
+
+                    store = get_message_store()
+                    messages = store.get_messages(self.pipeline.id, limit=10000)
+                    confirmed_roles = {
+                        m.from_role for m in messages if m.message_type == "CONSENSUS_CONFIRMED"
+                    }
+                    all_roles = tracker.graph.all_roles()
+                    if all_roles and all_roles.issubset(confirmed_roles):
+                        logger.info(
+                            "All roles confirmed via message bus fallback",
+                            pipeline_id=self.pipeline.id,
+                            confirmed_roles=sorted(confirmed_roles),
+                        )
+                        result["is_complete"] = True
+                        result["fallback"] = "message_bus"
+                except Exception as e:
+                    logger.warning(
+                        "Message-bus fallback in check_consensus failed",
+                        pipeline_id=self.pipeline.id,
+                        error=str(e),
+                    )
+            return result
         return {"is_complete": False, "blocking_agents": [], "has_objections": False, "agents": {}}
 
 

orchestrator/consensus_wrapper.py

@@ -36,14 +36,23 @@
     "## Current BRC state\n\n"
     "{brc_state}\n\n"
     "{nack_feedback}"
+    "## Empty state recovery\n\n"
+    "If BRC state is empty (`{{}}`), the in-memory tracker was likely lost "
+    "(e.g. orchestrator restart). In this case:\n"
+    "1. Run `egg-orch consensus status` to check if state was reconstructed.\n"
+    "2. If you are already fully ACKed, call `egg-orch consensus confirmed` "
+    "to re-confirm.\n"
+    "3. If already confirmed, stay alive and poll — do NOT re-propose.\n\n"
     "## Required actions\n\n"
     "1. Check consensus status: `egg-orch consensus status`\n"
     "2. Poll for messages: `egg-orch message poll --wait 30`\n"
     "3. Based on your role type:\n"
     "   - **Producer**: If you received NACKs, address the reviewer feedback, "
     "revise your work, and re-propose (`egg-orch consensus propose`). "
     "If WORKING, complete work and propose. "
-    "If PROPOSED, check for ACKs/NACKs and respond. If all ACKed, confirm.\n"
+    "If PROPOSED, check for ACKs/NACKs and respond. If all ACKed, confirm "
+    "(`egg-orch consensus confirmed`). "
+    "**Do NOT re-propose if already fully ACKed** — call confirmed instead.\n"
     "   - **Reviewer**: Check for proposals from assigned producers. Review "
     "artifacts in git, then ACK (`egg-orch consensus ack <role>`) or "
     'NACK (`egg-orch consensus nack <role> --reason "..."`).\n'
@@ -156,20 +165,30 @@
 
 # Check if this agent already reached CONFIRMED state (BRC protocol)
 AGENT_ROLE="${{EGG_AGENT_ROLE:-}}"
-if [ -n "$AGENT_ROLE" ]; then
-    AGENT_CONFIRMED=$(get_agent_confirmed "$RESPONSE" "$AGENT_ROLE")
+
+# Shell function: check if agent is confirmed (via tracker or message bus)
+# and wait for global consensus. Exits 0 if consensus reached.
+# Returns 0 if agent is confirmed (caller should not restart).
+# Returns 1 if agent is NOT confirmed (caller should continue to restart loop).
+check_confirmed_and_wait() {{
+    local response="$1"
+    local agent_role="$2"
+    local agent_confirmed
+    agent_confirmed=$(get_agent_confirmed "$response" "$agent_role")
 
     # Message bus fallback: if pipeline status returned empty consensus state
     # (e.g. after orchestrator restart lost in-memory tracker), check the
     # message store directly for our own CONSENSUS_CONFIRMED message.
-    if [ "$AGENT_CONFIRMED" != "True" ]; then
-        AGENTS_EMPTY=$(echo "$RESPONSE" | python3 -c \
+    if [ "$agent_confirmed" != "True" ]; then
+        local agents_empty
+        agents_empty=$(echo "$response" | python3 -c \
             "import sys,json; d=json.load(sys.stdin); agents=d.get('data',{{}}).get('concurrent',{{}}).get('consensus',{{}}).get('agents',{{}}); print('True' if not agents else 'False')" \
             2>/dev/null || echo "False")
-        if [ "$AGENTS_EMPTY" = "True" ]; then
+        if [ "$agents_empty" = "True" ]; then
             cw_log "Consensus state empty (tracker lost?). Checking message bus..."
-            MSG_RESPONSE=$(egg-orch message poll --json --limit 1000 2>/dev/null || echo "[]")
-            CONFIRMED_VIA_MSG=$(echo "$MSG_RESPONSE" | python3 -c "
+            local msg_response confirmed_via_msg
+            msg_response=$(egg-orch message poll --json --limit 1000 2>/dev/null || echo "[]")
+            confirmed_via_msg=$(echo "$msg_response" | python3 -c "
 import sys, json
 role = sys.argv[1]
 try:
@@ -183,33 +202,41 @@
     print('True' if found else 'False')
 except Exception:
     print('False')
-" "$AGENT_ROLE" 2>/dev/null || echo "False")
-            if [ "$CONFIRMED_VIA_MSG" = "True" ]; then
+" "$agent_role" 2>/dev/null || echo "False")
+            if [ "$confirmed_via_msg" = "True" ]; then
                 cw_log "Found own CONSENSUS_CONFIRMED in message bus. Already confirmed."
-                AGENT_CONFIRMED="True"
+                agent_confirmed="True"
             fi
         fi
     fi
 
-    if [ "$AGENT_CONFIRMED" = "True" ]; then
+    if [ "$agent_confirmed" = "True" ]; then
         cw_log "Agent already CONFIRMED in BRC protocol. 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 \
+        local poll_interval wait_count
+        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"
+            local resp is_complete
+            resp=$(egg-orch pipeline status --json 2>/dev/null || echo "{{}}")
+            is_complete=$(echo "$resp" | 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
+            if [ "$is_complete" = "True" ]; then
                 cw_log "Consensus reached. Exiting."
                 exit 0
             fi
         done
         cw_log "Agent was CONFIRMED but consensus not reached. Exiting cleanly."
         exit 0
     fi
+
+    return 1
+}}
+
+if [ -n "$AGENT_ROLE" ]; then
+    check_confirmed_and_wait "$RESPONSE" "$AGENT_ROLE" || true
 fi
 
 # --- Restart loop for clean exits without BRC consensus ---
@@ -223,6 +250,12 @@
     NACK_FEEDBACK=""
     if [ -n "$AGENT_ROLE" ]; then
         BRC_STATE=$(get_brc_state "$RESPONSE" "$AGENT_ROLE")
+        # RC1: When BRC state is empty (tracker lost), query consensus status
+        # directly for better recovery context.
+        if [ "$BRC_STATE" = "{{}}" ]; then
+            CONSENSUS_STATUS=$(egg-orch consensus status --json 2>/dev/null || echo "{{}}")
+            BRC_STATE="Empty (tracker likely lost). Consensus status: $CONSENSUS_STATUS"
+        fi
         NACK_FEEDBACK=$(get_nack_feedback "$RESPONSE" "$AGENT_ROLE")
     fi
 
@@ -263,6 +296,13 @@
         cw_log "Consensus reached after restart $RESTART_COUNT. Exiting."
         exit 0
     fi
+
+    # RC4: After restart, check if this agent reached CONFIRMED state.
+    # If so, enter the wait-for-consensus polling loop instead of
+    # burning another restart on a pointless re-run.
+    if [ -n "$AGENT_ROLE" ]; then
+        check_confirmed_and_wait "$RESPONSE" "$AGENT_ROLE" || true
+    fi
 done
 
 # --- Max restarts exhausted: shut down with failure ---

orchestrator/peer_consensus.py

@@ -140,7 +140,9 @@ def _handle_propose_inner(
             raise ValueError(
                 f"Producer {agent_role} is already fully ACKed "
                 f"(v{self.matrix.get_proposal_version(agent_role)}). "
-                f"Call confirmed instead of re-proposing."
+                f"Call `egg-orch consensus confirmed` instead of re-proposing. "
+                f"Re-proposing when fully ACKed is not allowed — confirm to "
+                f"complete the BRC protocol."
             )
 
         # Validate payload
@@ -524,6 +526,56 @@ def handle_agent_crash(self, role: str) -> dict[str, Any]:
 
             return {"action": "continue", "crashed_role": role}
 
+    def handle_stall_demotion(self, role: str, reason: str) -> dict[str, Any]:
+        """Demote a stalled dual-role agent's review edges to ADVISORY.
+
+        When a dual-role agent (e.g. tester) stalls, its pending reviewer
+        assignments should not block other agents from reaching consensus.
+        This demotes all CRITICAL edges where the stalled agent is a reviewer
+        to ADVISORY, allowing consensus to proceed without its ACK.
+
+        Args:
+            role: The stalled agent's role.
+            reason: Why the agent is being demoted (e.g. "missed heartbeats for 5+ minutes").
+
+        Returns:
+            Dict with action taken and affected producers.
+
+        Raises:
+            ValueError: If the role is not a reviewer in the review graph.
+        """
+        with self._lock:
+            if not self.graph.is_reviewer(role):
+                raise ValueError(f"Cannot demote '{role}': not a reviewer in the review graph")
+
+            demoted_edges = self.graph.demote_edges_for_reviewer(role)
+
+            if demoted_edges:
+                emit_event(
+                    EventType.CONSENSUS_FAILURE,
+                    self.pipeline_id,
+                    data={
+                        "type": "stall_demotion",
+                        "role": role,
+                        "reason": reason,
+                        "demoted_producers": demoted_edges,
+                    },
+                )
+                logger.info(
+                    "Demoted stalled reviewer edges to advisory",
+                    role=role,
+                    reason=reason,
+                    demoted_producers=demoted_edges,
+                    pipeline_id=self.pipeline_id,
+                )
+
+            return {
+                "action": "demoted",
+                "role": role,
+                "reason": reason,
+                "demoted_producers": demoted_edges,
+            }
+
     def excuse_reviewer(self, role: str) -> dict[str, Any]:
         """Remove a reviewer from the review graph (HITL-gated).
 

orchestrator/review_graph.py

@@ -105,6 +105,33 @@ def get_edge(self, reviewer: str, producer: str) -> ReviewEdge | None:
                 return e
         return None
 
+    def demote_edges_for_reviewer(
+        self,
+        reviewer: str,
+        new_criticality: "ReviewCriticality | None" = None,
+    ) -> list[str]:
+        """Demote all CRITICAL edges for a reviewer to a new criticality.
+
+        Args:
+            reviewer: Reviewer role whose edges should be demoted.
+            new_criticality: Target criticality (defaults to ADVISORY).
+
+        Returns:
+            List of producer roles whose edges were demoted.
+        """
+        if new_criticality is None:
+            new_criticality = ReviewCriticality.ADVISORY
+        demoted: list[str] = []
+        new_edges: list[ReviewEdge] = []
+        for e in self._edges:
+            if e.reviewer_role == reviewer and e.criticality == ReviewCriticality.CRITICAL:
+                new_edges.append(ReviewEdge(e.reviewer_role, e.producer_role, new_criticality))
+                demoted.append(e.producer_role)
+            else:
+                new_edges.append(e)
+        self._edges = new_edges
+        return demoted
+
     def remove_edge(self, reviewer: str, producer: str) -> bool:
         """Remove a review edge. Returns True if edge was found and removed."""
         for i, e in enumerate(self._edges):

orchestrator/routes/pipelines.py

@@ -3935,6 +3935,8 @@ def _update_agents_complete() -> None:
                 error=str(track_err),
             )
 
+    _demoted_agents: set[str] = set()
+
     while True:
         elapsed = time.monotonic() - start_time
 
@@ -4003,6 +4005,46 @@ def _update_agents_complete() -> None:
                     error=str(e),
                 )
 
+        # 3b. RC3: Stall demotion for dual-role agents.
+        # If a dual-role agent has missed heartbeats for 5+ minutes,
+        # demote its reviewer edges to ADVISORY so other agents can proceed.
+        try:
+            from health_monitor import get_health_monitor
+
+            _hm = get_health_monitor()
+            if _hm is not None:
+                from ..peer_consensus import get_peer_consensus_tracker  # type: ignore[import-not-found]  # noqa: I001
+
+                _brc_tracker = get_peer_consensus_tracker(pipeline_id)
+                if _brc_tracker is not None:
+                    heartbeat_actions = _hm.check_heartbeats()
+                    for hb_action in heartbeat_actions:
+                        stalled_agent = hb_action.get("agent_id", "")
+                        stall_elapsed = hb_action.get("elapsed_seconds", 0)
+                        if (
+                            stall_elapsed >= 300
+                            and stalled_agent not in _demoted_agents
+                            and _brc_tracker.graph.is_dual_role(stalled_agent)
+                        ):
+                            try:
+                                _brc_tracker.handle_stall_demotion(
+                                    stalled_agent,
+                                    reason=f"Missed heartbeats for {stall_elapsed}s",
+                                )
+                                _demoted_agents.add(stalled_agent)
+                            except Exception as demote_err:
+                                logger.debug(
+                                    "Stall demotion skipped",
+                                    agent=stalled_agent,
+                                    error=str(demote_err),
+                                )
+        except Exception as stall_err:
+            logger.debug(
+                "Stall demotion check failed",
+                pipeline_id=pipeline_id,
+                error=str(stall_err),
+            )
+
         # 4. Non-blocking check for exited containers
         for exec_info in active_executions:
             if exec_info.container_id in exited_containers: