From f2a682a4a2909d47d9901cb74ef02aeef8cd4486 Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Wed, 29 Apr 2026 10:25:49 -0700 Subject: [PATCH 1/3] Fix #2243: progress gate before BRC consensus-failure HITL decision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit At ``consensus_timeout_minutes`` (default 30) the orchestrator opened a ``choice`` decision unconditionally — including in cases where producers were minutes from their first commit (decision-15/decision-17 on ``issue-1557-v2``). "Continue waiting" was the only correct answer in every recorded misfire. Add a progress gate in the polling loop in ``_run_concurrent_phase``: before opening the decision, defer if any of the following has fired within ``brc_consensus_progress_gate_seconds`` (default 300s): * ``CONSENSUS_PROPOSE`` (latest proposal timestamp on the tracker) * ACK / NACK on the approval matrix (new ``get_latest_entry_timestamp``) * container heartbeat (filtered to active roles to avoid cross-phase pollution in the singleton ``HealthMonitor``) The polling loop keeps polling on defer; the decision opens only once the bus and containers have both gone quiet for the gate window. ``brc_consensus_progress_gate_seconds=0`` disables the gate. Failures in either signal collector are logged at WARNING and treated as "no signal from that source" — never as a defer — so a crashed collector can't silently keep us off the HITL surface. --- orchestrator/approval_matrix.py | 16 ++ orchestrator/models.py | 9 + orchestrator/peer_consensus.py | 15 ++ orchestrator/routes/pipelines.py | 133 ++++++++++++++ .../tests/test_peer_consensus_integration.py | 35 ++++ orchestrator/tests/test_pipelines_routes.py | 166 +++++++++++++++++- 6 files changed, 373 insertions(+), 1 deletion(-) diff --git a/orchestrator/approval_matrix.py b/orchestrator/approval_matrix.py index 22e54262d7..5d958dae55 100644 --- a/orchestrator/approval_matrix.py +++ b/orchestrator/approval_matrix.py @@ -344,6 +344,22 @@ def get_pre_merge_conditions(self) -> list[dict[str, Any]]: ) return conditions + def get_latest_entry_timestamp(self) -> datetime | None: + """Return the timestamp of the most recent ACK or NACK across all edges. + + Used by the BRC progress gate (#2243) to detect reviewer activity in + the window before opening an HITL consensus-failure decision. Returns + None if no edge has ever transitioned out of PENDING. + """ + latest: datetime | None = None + for entry in self._entries.values(): + ts = entry.timestamp + if ts is None: + continue + if latest is None or ts > latest: + latest = ts + return latest + def get_latest_review_versions(self, reviewer: str) -> dict[str, int]: """Get the version of each review the reviewer has submitted. diff --git a/orchestrator/models.py b/orchestrator/models.py index 1fba8ed297..659e548626 100644 --- a/orchestrator/models.py +++ b/orchestrator/models.py @@ -377,6 +377,15 @@ class PipelineConfig(BaseModel): consensus_timeout_minutes: int = Field( default=30, ge=1, description="Timeout for consensus before HITL escalation" ) + brc_consensus_progress_gate_seconds: int = Field( + default=300, + ge=0, + description=( + "Defer the consensus-timeout HITL decision while any BRC progress signal " + "(CONSENSUS_PROPOSE/ACK/NACK or container heartbeat) has fired within this " + "many seconds. 0 disables the gate. (#2243)" + ), + ) agent_idle_timeout_minutes: int = Field( default=60, ge=1, description="Timeout for idle agents before termination" ) diff --git a/orchestrator/peer_consensus.py b/orchestrator/peer_consensus.py index 88307602f8..3867fe8a42 100644 --- a/orchestrator/peer_consensus.py +++ b/orchestrator/peer_consensus.py @@ -1444,6 +1444,21 @@ def get_latest_proposal_timestamp(self) -> datetime | None: return None return max(self._proposal_timestamps.values()) + def get_latest_progress_timestamp(self) -> datetime | None: + """Return the most recent BRC-bus activity timestamp, or None. + + Aggregates the latest CONSENSUS_PROPOSE timestamp with the latest + ACK/NACK timestamp from the approval matrix. Used by the BRC + progress gate (#2243) to defer the auto consensus-failure HITL + decision while the bus is still moving. + """ + with self._lock: + latest = self.get_latest_proposal_timestamp() + entry_ts = self.matrix.get_latest_entry_timestamp() + if entry_ts is not None and (latest is None or entry_ts > latest): + latest = entry_ts + return latest + def evaluate(self) -> dict[str, Any]: """Evaluate current consensus state. diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 7412162f6d..a98a4cbde1 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -9560,6 +9560,97 @@ def _persist_hitl_decision( return None +def _check_brc_progress_gate( + pipeline_id: str, + slice_id: str | None, + active_role_names: list[str], + gate_seconds: float, +) -> tuple[bool, str | None]: + """Return (defer, reason) for the BRC consensus-timeout progress gate (#2243). + + Defers the auto-HITL consensus-failure decision when *any* of the + following has fired within ``gate_seconds``: + + * The BRC tracker's most recent ``CONSENSUS_PROPOSE`` (producer + proposal) timestamp. + * The most recent ACK/NACK timestamp on the approval matrix. + * The most recent container heartbeat for any role in + ``active_role_names`` (filters out cross-phase pollution in the + shared :class:`HealthMonitor` singleton). + + The gate is the operator-friendly half of the issue-2243 fix: at + :data:`consensus_timeout_minutes` we previously opened a `choice` + decision unconditionally, even when producers were minutes from + their first commit. With the gate, the polling loop keeps polling + while signals are alive; the decision only opens once the bus and + containers have both gone quiet for ``gate_seconds``. + + ``gate_seconds <= 0`` disables the gate (returns ``(False, None)``). + Failures in any signal source are logged at WARNING and treated as + "no signal from that source" — never as a gate defer, since a + crashed signal collector must not silently keep us off the HITL + surface. + """ + if gate_seconds <= 0: + return False, None + + now_dt = datetime.now(UTC) + now_mono = time.time() + + # 1. BRC bus signals (proposal + ACK/NACK timestamps). + try: + try: + from peer_consensus import get_peer_consensus_tracker + except ImportError: + from ..peer_consensus import ( + get_peer_consensus_tracker, # type: ignore[no-redef] + ) + try: + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) + except TypeError: + tracker = get_peer_consensus_tracker(pipeline_id) + if tracker is not None: + ts = tracker.get_latest_progress_timestamp() + if ts is not None and (now_dt - ts).total_seconds() < gate_seconds: + age = (now_dt - ts).total_seconds() + return True, f"BRC bus active {age:.0f}s ago" + except Exception as e: + logger.warning( + "BRC progress-gate tracker check failed", + pipeline_id=pipeline_id, + error=str(e), + ) + + # 2. Container heartbeats. Filter by active roles so a stale + # heartbeat from a prior phase in the singleton HealthMonitor + # doesn't keep us out of the HITL surface forever. + try: + from health_monitor import get_health_monitor + + hm = get_health_monitor() + if hm is not None: + active_set = set(active_role_names) + latest_hb: float | None = None + with hm._lock: # noqa: SLF001 — read-only snapshot + hb_snapshot = dict(hm._last_heartbeat) # noqa: SLF001 + for agent_id, hb_time in hb_snapshot.items(): + if active_set and agent_id not in active_set: + continue + if latest_hb is None or hb_time > latest_hb: + latest_hb = hb_time + if latest_hb is not None and (now_mono - latest_hb) < gate_seconds: + age = now_mono - latest_hb + return True, f"container heartbeat {age:.0f}s ago" + except Exception as e: + logger.warning( + "BRC progress-gate heartbeat check failed", + pipeline_id=pipeline_id, + error=str(e), + ) + + return False, None + + def _handle_brc_consensus_timeout( pipeline: Pipeline, pipeline_id: str, @@ -10762,6 +10853,11 @@ def _update_agents_complete() -> None: _demoted_agents: set[str] = set() + # #2243 progress-gate state: log on first defer + first un-defer only + # so the polling loop doesn't spam at every iteration once we cross + # ``consensus_timeout``. + _progress_gate_deferring = False + while True: elapsed = time.monotonic() - start_time @@ -11186,6 +11282,43 @@ def _update_agents_complete() -> None: # 6. Consensus timeout if elapsed >= consensus_timeout: + # #2243 progress gate: keep polling instead of opening the + # auto-HITL decision while producer/reviewer activity is + # still live on the BRC bus or in container heartbeats. + # Without this gate, decision-15 / decision-17 on + # ``issue-1557-v2`` fired minutes before the next commit + # landed; "Continue waiting" was the only correct answer. + _gate_seconds = max( + 0, + int(getattr(pipeline.config, "brc_consensus_progress_gate_seconds", 300)), + ) + _gate_defer, _gate_reason = _check_brc_progress_gate( + pipeline_id, + slice_id, + [e.role.value for e in active_executions], + _gate_seconds, + ) + if _gate_defer: + if not _progress_gate_deferring: + logger.info( + "Consensus timeout deferred by progress gate", + pipeline_id=pipeline_id, + elapsed_seconds=round(elapsed, 1), + gate_seconds=_gate_seconds, + reason=_gate_reason, + ) + _progress_gate_deferring = True + time.sleep(poll_interval) + continue + if _progress_gate_deferring: + logger.info( + "Consensus timeout proceeding — progress gate window elapsed", + pipeline_id=pipeline_id, + elapsed_seconds=round(elapsed, 1), + gate_seconds=_gate_seconds, + ) + _progress_gate_deferring = False + logger.warning( "Consensus timeout reached, falling back to container exit", pipeline_id=pipeline_id, diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index afb59f2339..359d90b6c3 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -293,6 +293,41 @@ def test_timeout_advisory_only_proceeds(self): assert result["action"] == "proceed_with_notification" +class TestProgressTimestamps: + """Aggregate progress-timestamp accessor (#2243). + + The BRC progress gate in routes/pipelines.py uses + ``get_latest_progress_timestamp`` to defer the auto consensus-failure + HITL decision while the bus is still moving. The accessor unifies + proposal timestamps and approval-matrix ACK/NACK timestamps so the + gate has a single signal to read. + """ + + def test_returns_none_when_no_activity(self, tracker): + assert tracker.get_latest_progress_timestamp() is None + + def test_returns_proposal_timestamp_when_only_proposals(self, tracker): + tracker.handle_propose( + "coder", {"summary": "v1", "artifacts": ["a.py"], "commit_sha": "abc123"} + ) + proposal_ts = tracker.get_latest_proposal_timestamp() + progress_ts = tracker.get_latest_progress_timestamp() + assert proposal_ts is not None + assert progress_ts == proposal_ts + + def test_ack_advances_progress_timestamp_past_proposal(self, tracker): + tracker.handle_propose( + "coder", {"summary": "v1", "artifacts": ["a.py"], "commit_sha": "abc123"} + ) + proposal_ts = tracker.get_latest_proposal_timestamp() + # ACK happens after proposal — progress should advance to the ACK ts. + tracker.handle_ack("reviewer_code", "coder", {"artifact_references": ["a.py"]}) + progress_ts = tracker.get_latest_progress_timestamp() + assert proposal_ts is not None + assert progress_ts is not None + assert progress_ts >= proposal_ts + + class TestAgentCrash: """Test agent crash handling.""" diff --git a/orchestrator/tests/test_pipelines_routes.py b/orchestrator/tests/test_pipelines_routes.py index 13d1381b92..63ad230490 100644 --- a/orchestrator/tests/test_pipelines_routes.py +++ b/orchestrator/tests/test_pipelines_routes.py @@ -3,14 +3,20 @@ Covers issue #1783: the BRC/consensus timeout path used bare relative imports that crashed under k3s's top-level-module layout, and silently swallowed add_decision failures so the stall had no visible HITL decision. + +Also covers issue #2243: the BRC progress gate must defer the auto +consensus-failure HITL decision while the bus or container heartbeats +have fired within the gate window. """ +import time +from datetime import UTC, datetime, timedelta from unittest.mock import MagicMock, patch import pytest from events import EventType from models import Pipeline, PipelinePhase -from routes.pipelines import _handle_brc_consensus_timeout +from routes.pipelines import _check_brc_progress_gate, _handle_brc_consensus_timeout @pytest.fixture @@ -172,3 +178,161 @@ def test_add_decision_failure_is_logged_not_swallowed(self, mock_emit, mock_logg _, kwargs = warning_calls[0] assert kwargs.get("pipeline_id") == pipeline.id assert kwargs.get("exc_info") is True + + +class TestBrcProgressGate: + """Issue #2243 — defer the auto consensus-failure HITL decision while + BRC bus or container heartbeats have advanced within the gate window. + + The gate is the operator-friendly half of the fix: previously, at + ``consensus_timeout_minutes`` the orchestrator opened a `choice` + decision unconditionally, even when producers were minutes away + from their first commit (decision-15 / decision-17 on + ``issue-1557-v2``). The gate keeps the polling loop polling while + signals are alive, and only opens the decision once the bus and + containers have both gone quiet for ``gate_seconds``. + """ + + PIPELINE_ID = "issue-2243-test" + + def _patch_tracker(self, latest_progress: datetime | None): + tracker = MagicMock() + tracker.get_latest_progress_timestamp.return_value = latest_progress + return patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker) + + def _patch_health_monitor(self, last_heartbeats: dict[str, float] | None): + if last_heartbeats is None: + return patch("health_monitor.get_health_monitor", return_value=None) + hm = MagicMock() + hm._lock = MagicMock() + hm._lock.__enter__ = MagicMock(return_value=hm._lock) + hm._lock.__exit__ = MagicMock(return_value=False) + hm._last_heartbeat = dict(last_heartbeats) + return patch("health_monitor.get_health_monitor", return_value=hm) + + def test_disabled_gate_returns_no_defer(self): + defer, reason = _check_brc_progress_gate(self.PIPELINE_ID, None, ["coder"], 0) + assert defer is False + assert reason is None + + def test_recent_proposal_defers(self): + recent = datetime.now(UTC) - timedelta(seconds=30) + with ( + self._patch_tracker(recent), + self._patch_health_monitor(None), + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder"], gate_seconds=300 + ) + assert defer is True + assert reason is not None and "BRC bus" in reason + + def test_stale_proposal_does_not_defer(self): + stale = datetime.now(UTC) - timedelta(seconds=600) + with ( + self._patch_tracker(stale), + self._patch_health_monitor(None), + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder"], gate_seconds=300 + ) + assert defer is False + assert reason is None + + def test_recent_heartbeat_defers_when_bus_silent(self): + # Bus completely silent (decision-17 shape: coder mid-merge-conflict + # before its first CONSENSUS_PROPOSE), but the container is still + # emitting heartbeats. The gate should still defer. + recent_hb = time.time() - 30 + with ( + self._patch_tracker(None), + self._patch_health_monitor({"coder": recent_hb}), + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder", "tester"], gate_seconds=300 + ) + assert defer is True + assert reason is not None and "heartbeat" in reason + + def test_stale_heartbeat_does_not_defer(self): + stale_hb = time.time() - 600 + with ( + self._patch_tracker(None), + self._patch_health_monitor({"coder": stale_hb}), + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder"], gate_seconds=300 + ) + assert defer is False + assert reason is None + + def test_heartbeat_for_inactive_role_is_ignored(self): + # Cross-phase pollution: the singleton HealthMonitor's + # ``_last_heartbeat`` may carry a stale entry for a role that + # isn't part of the current phase. The gate must filter so a + # ghost heartbeat from a finished phase doesn't keep the gate + # deferring forever. + recent_hb = time.time() - 30 + with ( + self._patch_tracker(None), + self._patch_health_monitor({"refiner": recent_hb}), + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder", "tester"], gate_seconds=300 + ) + assert defer is False + assert reason is None + + def test_no_signals_returns_no_defer(self): + with ( + self._patch_tracker(None), + self._patch_health_monitor({}), + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder"], gate_seconds=300 + ) + assert defer is False + assert reason is None + + def test_tracker_failure_logged_and_not_treated_as_defer(self): + # If the tracker collector raises, treat it as "no signal" rather + # than as a defer — a crashed signal source must never silently + # keep us off the HITL surface. + with ( + patch( + "peer_consensus.get_peer_consensus_tracker", + side_effect=RuntimeError("simulated tracker failure"), + ), + self._patch_health_monitor({}), + patch("routes.pipelines.logger") as mock_logger, + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder"], gate_seconds=300 + ) + assert defer is False + assert reason is None + assert any( + "tracker check failed" in (call.args[0] if call.args else "") + for call in mock_logger.warning.call_args_list + ) + + def test_heartbeat_failure_logged_and_not_treated_as_defer(self): + # Same as above for the heartbeat collector. + bad_hm = MagicMock() + bad_hm._lock = MagicMock() + bad_hm._lock.__enter__ = MagicMock(side_effect=RuntimeError("hm boom")) + bad_hm._lock.__exit__ = MagicMock(return_value=False) + with ( + self._patch_tracker(None), + patch("health_monitor.get_health_monitor", return_value=bad_hm), + patch("routes.pipelines.logger") as mock_logger, + ): + defer, reason = _check_brc_progress_gate( + self.PIPELINE_ID, None, ["coder"], gate_seconds=300 + ) + assert defer is False + assert reason is None + assert any( + "heartbeat check failed" in (call.args[0] if call.args else "") + for call in mock_logger.warning.call_args_list + ) From 292a5a658327b02455b6a836d0456220de73132c Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Wed, 29 Apr 2026 17:47:35 +0000 Subject: [PATCH 2/3] Address review feedback on #2254 progress gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rename now_mono → now_wallclock and document the two deliberate clocks (datetime UTC vs time.time epoch); both ends of the heartbeat comparison are wall-clock, not monotonic, so an NTP step is a known narrow gap (callout for the per-phase-timeout follow-up). - Empty active_role_names now early-returns False, matching the contract in the comment ("filters out cross-phase pollution"); previously the `if active_set and ...` short-circuited and accepted every stale heartbeat. - Drop the dead TypeError fallback around get_peer_consensus_tracker in both _check_brc_progress_gate and _handle_brc_consensus_timeout; the function already accepts (pipeline_id, slice_id=None). - Add a TODO calling out same-role cross-phase pollution (coder reappearing in implement / implement-fix / fix-on-PR) — the role-name filter handles different-role ghosts but not same-role ones, tracked under #2243 step 2. - Document the heartbeat-cadence contract the decision-17 path depends on (gate falls open if heartbeats stop firing within gate_seconds). - Tighten test_ack_advances_progress_timestamp_past_proposal from >= to > so the test catches a regression where ACKs stop advancing progress (back-to-back datetime.now(UTC) calls are reliably strictly increasing). - Add test_empty_active_roles_does_not_defer_on_heartbeat to lock in the empty-list contract. --- orchestrator/routes/pipelines.py | 51 ++++++++++++++----- .../tests/test_peer_consensus_integration.py | 8 ++- orchestrator/tests/test_pipelines_routes.py | 16 ++++++ 3 files changed, 60 insertions(+), 15 deletions(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index a98a4cbde1..9cac615542 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -9590,12 +9590,39 @@ def _check_brc_progress_gate( "no signal from that source" — never as a gate defer, since a crashed signal collector must not silently keep us off the HITL surface. + + Heartbeat-cadence contract: the decision-17 path (coder mid-merge- + conflict before any ``CONSENSUS_PROPOSE``) relies on container + heartbeats firing at least every ``gate_seconds``. Sandbox + heartbeats (see ``shared/egg_agent`` heartbeat scheduler and + ``orchestrator/health_monitor.py``) cadence today is well under + 300s, but a long uninterruptible subprocess (e.g. ``git rebase`` + blocked on a merge driver) could starve them; once that happens + the gate falls open and the pre-fix behaviour returns. Tracked as + a follow-up under #2243. + + TODO(#2243 step 2): same-role cross-phase pollution. The role-name + filter handles different-role ghosts (refiner heartbeat lingering + during a coder phase) but not same-role ghosts: ``coder`` reappears + across implement / implement-fix / fix-on-PR phases and + ``HealthMonitor._last_heartbeat['coder']`` is only popped on + ``clear_agent_state``. A phase boundary clear (or stamping the + heartbeat key with the phase) would close it; per-phase timeouts + in step 2 of the issue plan will likely subsume it. """ if gate_seconds <= 0: return False, None + # Two clocks, deliberately. ``now_dt`` is used for tracker + # timestamps (datetime in UTC). ``now_wallclock`` is the float + # epoch ``time.time()`` returns, matching the wall-clock values + # ``HealthMonitor._last_heartbeat`` is populated with. Despite the + # earlier name ``now_mono``, these are NOT monotonic — an NTP step + # on the orchestrator host can make ``(now - latest_hb)`` negative + # or skip the gate window. Acceptable today; revisit alongside the + # per-phase-timeout follow-up. now_dt = datetime.now(UTC) - now_mono = time.time() + now_wallclock = time.time() # 1. BRC bus signals (proposal + ACK/NACK timestamps). try: @@ -9605,10 +9632,7 @@ def _check_brc_progress_gate( from ..peer_consensus import ( get_peer_consensus_tracker, # type: ignore[no-redef] ) - try: - tracker = get_peer_consensus_tracker(pipeline_id, slice_id) - except TypeError: - tracker = get_peer_consensus_tracker(pipeline_id) + tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if tracker is not None: ts = tracker.get_latest_progress_timestamp() if ts is not None and (now_dt - ts).total_seconds() < gate_seconds: @@ -9623,7 +9647,11 @@ def _check_brc_progress_gate( # 2. Container heartbeats. Filter by active roles so a stale # heartbeat from a prior phase in the singleton HealthMonitor - # doesn't keep us out of the HITL surface forever. + # doesn't keep us out of the HITL surface forever. An empty + # ``active_role_names`` means the caller has no live containers + # to gate on, so match nothing rather than every stale heartbeat. + if not active_role_names: + return False, None try: from health_monitor import get_health_monitor @@ -9634,12 +9662,12 @@ def _check_brc_progress_gate( with hm._lock: # noqa: SLF001 — read-only snapshot hb_snapshot = dict(hm._last_heartbeat) # noqa: SLF001 for agent_id, hb_time in hb_snapshot.items(): - if active_set and agent_id not in active_set: + if agent_id not in active_set: continue if latest_hb is None or hb_time > latest_hb: latest_hb = hb_time - if latest_hb is not None and (now_mono - latest_hb) < gate_seconds: - age = now_mono - latest_hb + if latest_hb is not None and (now_wallclock - latest_hb) < gate_seconds: + age = now_wallclock - latest_hb return True, f"container heartbeat {age:.0f}s ago" except Exception as e: logger.warning( @@ -9673,10 +9701,7 @@ def _handle_brc_consensus_timeout( get_peer_consensus_tracker, # type: ignore[no-redef] ) - try: - _brc_tracker = get_peer_consensus_tracker(pipeline_id, slice_id) - except TypeError: - _brc_tracker = get_peer_consensus_tracker(pipeline_id) + _brc_tracker = get_peer_consensus_tracker(pipeline_id, slice_id) if _brc_tracker is not None: _brc_timeout_result = _brc_tracker.handle_timeout() _brc_handled = _brc_tracker.is_timeout_handled() diff --git a/orchestrator/tests/test_peer_consensus_integration.py b/orchestrator/tests/test_peer_consensus_integration.py index 359d90b6c3..3e29a74ce0 100644 --- a/orchestrator/tests/test_peer_consensus_integration.py +++ b/orchestrator/tests/test_peer_consensus_integration.py @@ -320,12 +320,16 @@ def test_ack_advances_progress_timestamp_past_proposal(self, tracker): "coder", {"summary": "v1", "artifacts": ["a.py"], "commit_sha": "abc123"} ) proposal_ts = tracker.get_latest_proposal_timestamp() - # ACK happens after proposal — progress should advance to the ACK ts. + # ACK happens after proposal — progress should advance strictly + # past the proposal ts. ``datetime.now(UTC)`` has microsecond + # resolution so back-to-back calls are reliably increasing; + # using ``>`` (not ``>=``) catches a regression where ACKs stop + # advancing progress. tracker.handle_ack("reviewer_code", "coder", {"artifact_references": ["a.py"]}) progress_ts = tracker.get_latest_progress_timestamp() assert proposal_ts is not None assert progress_ts is not None - assert progress_ts >= proposal_ts + assert progress_ts > proposal_ts class TestAgentCrash: diff --git a/orchestrator/tests/test_pipelines_routes.py b/orchestrator/tests/test_pipelines_routes.py index 63ad230490..a651977438 100644 --- a/orchestrator/tests/test_pipelines_routes.py +++ b/orchestrator/tests/test_pipelines_routes.py @@ -283,6 +283,22 @@ def test_heartbeat_for_inactive_role_is_ignored(self): assert defer is False assert reason is None + def test_empty_active_roles_does_not_defer_on_heartbeat(self): + # Contract: an empty ``active_role_names`` means the caller has + # no live containers to gate on, so match nothing rather than + # accept every stale heartbeat in the singleton HealthMonitor. + # ``_run_concurrent_phase`` exits before reaching the gate when + # there are no live containers, but the contract is explicit so + # a future caller can't accidentally widen the gate. + recent_hb = time.time() - 30 + with ( + self._patch_tracker(None), + self._patch_health_monitor({"coder": recent_hb, "refiner": recent_hb}), + ): + defer, reason = _check_brc_progress_gate(self.PIPELINE_ID, None, [], gate_seconds=300) + assert defer is False + assert reason is None + def test_no_signals_returns_no_defer(self): with ( self._patch_tracker(None), From a15ccf34ec3c2c098391f98b86fb031703d5837a Mon Sep 17 00:00:00 2001 From: "james-in-a-box[bot]" <246424927+james-in-a-box[bot]@users.noreply.github.com> Date: Wed, 29 Apr 2026 17:57:48 +0000 Subject: [PATCH 3/3] Fix checks: remove stale TypeError-fallback test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The TypeError fallback in _handle_brc_consensus_timeout was deliberately removed in 292a5a6 ('Drop the dead TypeError fallback... the function already accepts (pipeline_id, slice_id=None)'), but the corresponding test still expected two get_peer_consensus_tracker calls (raise + retry). Production code now makes a single call, catches TypeError, and falls back to the HITL escalation path — exactly what the warning log line in the failing run shows. Drop the obsolete test. --- .../tests/test_slice_run_loop_integration.py | 33 ------------------- 1 file changed, 33 deletions(-) diff --git a/orchestrator/tests/test_slice_run_loop_integration.py b/orchestrator/tests/test_slice_run_loop_integration.py index d892762a6b..9de8f7163f 100644 --- a/orchestrator/tests/test_slice_run_loop_integration.py +++ b/orchestrator/tests/test_slice_run_loop_integration.py @@ -1031,36 +1031,3 @@ def test_no_slice_id_uses_pipeline_scope(self) -> None: if slice_passed is None and len(args) >= 2: slice_passed = args[1] assert slice_passed is None - - def test_typeerror_falls_back_to_pipeline_scope(self) -> None: - """Older import-shim trackers without slice_id fall back gracefully.""" - pipeline = _make_pipeline() - - call_history: list[tuple] = [] - - def _shim_get(*args: Any, **kwargs: Any) -> MagicMock: - call_history.append((args, kwargs)) - if len(args) > 1 or "slice_id" in kwargs: - raise TypeError("legacy shim — no slice_id support") - tracker = MagicMock() - tracker.handle_timeout.return_value = {"action": "noop"} - tracker.is_timeout_handled.return_value = False - return tracker - - with patch("peer_consensus.get_peer_consensus_tracker", side_effect=_shim_get): - _handle_brc_consensus_timeout( - pipeline=pipeline, - pipeline_id=pipeline.id, - consensus_timeout=1800.0, - blocking_agents=["coder"], - store=MagicMock(), - slice_id="slice-7", - ) - # Two calls: first with slice_id (raises TypeError), second - # without slice_id (succeeds). - assert len(call_history) == 2 - # Second call has only the pipeline_id positionally. - second_args, second_kwargs = call_history[1] - assert second_args == (pipeline.id,) or ( - second_args == (pipeline.id, None) and "slice_id" not in second_kwargs - )