diff --git a/orchestrator/approval_matrix.py b/orchestrator/approval_matrix.py index 4380222774..65aa9e81c2 100644 --- a/orchestrator/approval_matrix.py +++ b/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) diff --git a/orchestrator/concurrent_executor.py b/orchestrator/concurrent_executor.py index 35d533c179..e881d1a8a7 100644 --- a/orchestrator/concurrent_executor.py +++ b/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": {}} diff --git a/orchestrator/consensus_wrapper.py b/orchestrator/consensus_wrapper.py index 0256252c21..aa01b55076 100644 --- a/orchestrator/consensus_wrapper.py +++ b/orchestrator/consensus_wrapper.py @@ -36,6 +36,13 @@ "## 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" @@ -43,7 +50,9 @@ " - **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 `) or " 'NACK (`egg-orch consensus nack --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,26 +202,28 @@ 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 @@ -210,6 +231,12 @@ 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 --- diff --git a/orchestrator/peer_consensus.py b/orchestrator/peer_consensus.py index 1a9ee5388d..a96ebf826e 100644 --- a/orchestrator/peer_consensus.py +++ b/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). diff --git a/orchestrator/review_graph.py b/orchestrator/review_graph.py index d2f64fd2e4..631be28523 100644 --- a/orchestrator/review_graph.py +++ b/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): diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 068fcc311d..119a52229d 100644 --- a/orchestrator/routes/pipelines.py +++ b/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: diff --git a/orchestrator/routes/signals.py b/orchestrator/routes/signals.py index 818eb475dc..2a7f30b93d 100644 --- a/orchestrator/routes/signals.py +++ b/orchestrator/routes/signals.py @@ -999,7 +999,88 @@ def handle_consensus_confirmed_signal( tracker = get_peer_consensus_tracker(pipeline_id) if not tracker: - return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) + # Defaults must be outside the try block so the message-bus fallback + # (second try block) can reference them even if reconstruction fails. + _phase = "implement" + _repo = None + + # Attempt reconstruction from message store before returning 404 + try: + from peer_consensus import reconstruct_tracker_from_messages + from review_graph import get_review_graph_for_phase + + # Determine phase and repo from pipeline state if available + try: + from pipeline_state import get_pipeline_state_store + + _store = get_pipeline_state_store() + _pip = _store.load_pipeline(pipeline_id) + _phase = _pip.current_phase.value + _repo = getattr(_pip.config, "repo", None) + except Exception: + pass + + graph = get_review_graph_for_phase(_phase, repo=_repo) + tracker = reconstruct_tracker_from_messages(pipeline_id, graph) + except Exception as recon_err: + logger.warning( + "Tracker reconstruction failed in confirmed handler", + pipeline_id=pipeline_id, + error=str(recon_err), + ) + + if not tracker: + # Message-bus authoritative fallback: if all expected roles have + # CONSENSUS_CONFIRMED messages, accept the confirmation directly. + try: + from message_store import Message, MessageType, get_message_store + from review_graph import get_review_graph_for_phase + + store = get_message_store() + messages = store.get_messages(pipeline_id, limit=10000) + confirmed_roles = { + m.from_role for m in messages if m.message_type == "CONSENSUS_CONFIRMED" + } + # Agent sending this signal is also confirming + confirmed_roles.add(agent_role) + + graph = get_review_graph_for_phase(_phase, repo=_repo) + all_roles = graph.all_roles() + + if all_roles and all_roles.issubset(confirmed_roles): + logger.info( + "All roles confirmed via message bus (tracker lost)", + pipeline_id=pipeline_id, + confirmed_roles=sorted(confirmed_roles), + ) + # Write the CONSENSUS_CONFIRMED message + store.add_message( + Message( + pipeline_id=pipeline_id, + from_role=agent_role, + to_role="all", + message_type=MessageType.CONSENSUS_CONFIRMED, + subject=f"Confirmed by {agent_role}", + body="", + metadata={"consensus_reached": True, "fallback": "message_bus"}, + ) + ) + return make_success_response( + f"Confirmation recorded for {agent_role} (message-bus fallback)", + data={ + "status": "confirmed", + "consensus_reached": True, + "fallback": "message_bus", + }, + ) + except Exception as fallback_err: + logger.warning( + "Message-bus fallback failed in confirmed handler", + pipeline_id=pipeline_id, + error=str(fallback_err), + ) + + return make_error_response(f"No consensus tracker for pipeline {pipeline_id}", 404) try: result = tracker.handle_confirmed(agent_role) @@ -1032,9 +1113,12 @@ def handle_consensus_confirmed_signal( f"Confirmation recorded for {agent_role}", data=result, ) - except (ValueError, Exception) as e: + except ValueError as e: logger.error("Failed to process consensus confirmed", pipeline_id=pipeline_id, error=str(e)) - return make_error_response(str(e), 400 if isinstance(e, ValueError) else 500) + return make_error_response(str(e), 400) + except Exception as e: + logger.error("Failed to process consensus confirmed", pipeline_id=pipeline_id, error=str(e)) + return make_error_response(str(e), 500) @signals_bp.route("//signal/batch", methods=["POST"]) diff --git a/orchestrator/tests/test_consensus_wrapper.py b/orchestrator/tests/test_consensus_wrapper.py index 19e9c87465..a093ad7fd9 100644 --- a/orchestrator/tests/test_consensus_wrapper.py +++ b/orchestrator/tests/test_consensus_wrapper.py @@ -564,3 +564,36 @@ def test_message_bus_fallback_enters_confirmed_wait(self): assert "Already confirmed" in result.stderr # Should NOT restart assert "Restarting" not in result.stderr + + def test_post_restart_confirmed_detection(self): + """After restart, if agent reached CONFIRMED, should enter wait loop (RC4).""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + # The wrapper should call check_confirmed_and_wait after each restart + assert "check_confirmed_and_wait" in script + # Should be called both before restart loop and after restart + assert script.count("check_confirmed_and_wait") >= 2 + + def test_empty_state_recovery_prompt(self): + """Recovery prompt should include empty state recovery guidance (RC1).""" + assert "Empty state recovery" in _RECOVERY_SYSTEM_PROMPT + assert "egg-orch consensus confirmed" in _RECOVERY_SYSTEM_PROMPT + assert "Do NOT re-propose if already fully ACKed" in _RECOVERY_SYSTEM_PROMPT + + def test_wrapper_queries_consensus_status_on_empty_state(self): + """When BRC state is empty, wrapper should query consensus status for context (RC1).""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + # Should check for empty BRC state and query consensus status + assert "consensus status" in script + assert "tracker likely lost" in script + + def test_check_confirmed_and_wait_is_shell_function(self): + """check_confirmed_and_wait should be defined as a reusable shell function.""" + cmd = build_consensus_wrapped_command("Prompt") + script = cmd[2] + # Should define the function + assert "check_confirmed_and_wait()" in script + # Should contain the full logic + assert "CONSENSUS_CONFIRMED" in script + assert "message poll" in script diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index 050af3046e..8cfa7125ab 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -1596,3 +1596,164 @@ def test_skips_invalid_messages_gracefully(self, simple_graph): assert tracker is not None state = tracker.evaluate() assert state["agents"]["coder"]["producer_phase"] == "PROPOSED" + + +class TestACKGuardErrorMessage: + """Test that ACK guard error includes explicit guidance (RC2).""" + + def test_ack_guard_includes_confirmed_guidance(self, tracker): + """Error message should tell the agent to call confirmed.""" + tracker.handle_propose("coder", {"summary": "v1", "artifacts": ["a.py"]}) + + tracker.handle_ack("reviewer_code", "coder", {"artifact_references": ["a.py"]}) + tracker.handle_ack("reviewer_contract", "coder", {"artifact_references": ["a.py"]}) + + # Re-proposing when fully ACKed should raise with clear guidance + with pytest.raises(ValueError, match="egg-orch consensus confirmed"): + tracker.handle_propose("coder", {"summary": "v2", "artifacts": ["a.py"]}) + + +class TestStallDemotion: + """Test stall demotion for dual-role agents (RC3).""" + + @pytest.fixture + def dual_role_graph(self): + """Graph where tester is both producer and reviewer (dual-role).""" + return ReviewGraph( + [ + ReviewEdge("reviewer_code", "coder", ReviewCriticality.CRITICAL), + ReviewEdge("tester", "coder", ReviewCriticality.CRITICAL), + ReviewEdge("reviewer_code", "tester", ReviewCriticality.ADVISORY), + ] + ) + + @pytest.fixture + def dual_tracker(self, dual_role_graph): + """Tracker with dual-role tester.""" + t = PeerConsensusTracker("test-stall", dual_role_graph, cooldown_seconds=0) + t.register_agent("coder") + t.register_agent("tester") + t.register_agent("reviewer_code") + return t + + def test_stall_demotion_changes_edge_to_advisory(self, dual_tracker): + """Demoting a stalled dual-role agent should make its edges advisory.""" + assert dual_tracker.graph.is_dual_role("tester") + + result = dual_tracker.handle_stall_demotion("tester", reason="Missed heartbeats for 300s") + + assert result["action"] == "demoted" + assert "coder" in result["demoted_producers"] + + # Edge should now be advisory, not critical + critical = dual_tracker.graph.critical_reviewers_for("coder") + assert "tester" not in critical + advisory = dual_tracker.graph.advisory_reviewers_for("coder") + assert "tester" in advisory + + def test_stall_demotion_non_reviewer_raises(self, dual_tracker): + """Demoting a non-reviewer should raise.""" + with pytest.raises(ValueError, match="not a reviewer"): + dual_tracker.handle_stall_demotion("nonexistent", reason="test") + + def test_stall_demotion_allows_consensus_without_stalled_ack(self, dual_tracker): + """After demotion, consensus should proceed without the stalled agent's ACK.""" + dual_tracker.handle_propose("coder", {"summary": "v1", "artifacts": ["a.py"]}) + + # reviewer_code ACKs coder, but tester (stalled) does not ACK + dual_tracker.handle_ack("reviewer_code", "coder", {"artifact_references": ["a.py"]}) + + # Without demotion, coder is NOT fully acked (tester is critical) + assert not dual_tracker.matrix.is_fully_acked("coder") + + # Demote tester + dual_tracker.handle_stall_demotion("tester", reason="stalled") + + # Now coder should be fully acked (tester is advisory) + assert dual_tracker.matrix.is_fully_acked("coder") + + +class TestReconstructTrackerConfirmedReplay: + """Test tracker reconstruction replays CONFIRMED messages correctly (RC5).""" + + def test_reconstruct_with_confirmed_replay(self, simple_graph): + """Reconstructed tracker should mark agents as confirmed.""" + from datetime import datetime, timedelta + + base = datetime(2024, 1, 1) + + messages = [ + _FakeMessage( + message_type="CONSENSUS_PROPOSE", + from_role="coder", + body="proposal", + metadata={"payload": {"summary": "impl", "artifacts": ["a.py"]}}, + timestamp=base + timedelta(seconds=1), + ), + _FakeMessage( + message_type="CONSENSUS_ACK", + from_role="reviewer_code", + to_role="coder", + body="lgtm", + metadata={"payload": {"reason": "good", "artifact_references": ["a.py"]}}, + timestamp=base + timedelta(seconds=2), + ), + _FakeMessage( + message_type="CONSENSUS_ACK", + from_role="reviewer_contract", + to_role="coder", + body="lgtm", + metadata={"payload": {"reason": "good", "artifact_references": ["a.py"]}}, + timestamp=base + timedelta(seconds=3), + ), + _FakeMessage( + message_type="CONSENSUS_PROPOSE", + from_role="tester", + body="test results", + metadata={"payload": {"summary": "tests", "artifacts": ["test.py"]}}, + timestamp=base + timedelta(seconds=4), + ), + _FakeMessage( + message_type="CONSENSUS_ACK", + from_role="reviewer_code", + to_role="tester", + body="lgtm", + metadata={"payload": {"reason": "good", "artifact_references": ["test.py"]}}, + timestamp=base + timedelta(seconds=5), + ), + # Confirmed messages + _FakeMessage( + message_type="CONSENSUS_CONFIRMED", + from_role="coder", + timestamp=base + timedelta(seconds=6), + ), + _FakeMessage( + message_type="CONSENSUS_CONFIRMED", + from_role="reviewer_code", + timestamp=base + timedelta(seconds=7), + ), + _FakeMessage( + message_type="CONSENSUS_CONFIRMED", + from_role="reviewer_contract", + timestamp=base + timedelta(seconds=8), + ), + _FakeMessage( + message_type="CONSENSUS_CONFIRMED", + from_role="tester", + timestamp=base + timedelta(seconds=9), + ), + ] + + store = _FakeMessageStore(messages) + try: + tracker = reconstruct_tracker_from_messages( + "test-rc5", simple_graph, message_store=store + ) + assert tracker is not None + state = tracker.evaluate() + # All agents confirmed — consensus should be complete + assert state["is_complete"] is True + assert len(state["blocking_agents"]) == 0 + finally: + with _trackers_lock: + _trackers.pop("test-rc5", None)