From cc38ccf67a6ae0c711ed6615a9923fec8736f96b Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Thu, 25 Jun 2026 11:31:04 -0700 Subject: [PATCH 1/2] fix(orchestrator): surface live event-loop pods in running-agent views (#3230) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under the orchestrator-owned BRC event loop (#3164, now unconditional) role pods are on-demand one-shots that the loop deliberately never persists into phase_exec.agents — the consensus tracker plus live-Job labels are its only sources of truth. So the persisted agent list is empty even while role pods are Running, which both running-agent consumers read as "0 running agents": * get_status.running_agents → a blind dashboard. * concurrent.agents (the overseer's stall-duration source) → the overseer sees the split "3 BRC agents blocking, 0 running agents" against the populated tracker block and composes a false `phase stalled` alert. Backfill both views from the labels that ARE authoritative when the persisted list is empty: * routes.pipelines._live_event_agents() reconstructs the running-pod cohort from live Job labels (LIVE_POD_STATUSES, slice-scoped when a slice_id is supplied). _get_concurrent_status falls back to it. * mcp_tools._build_status_snapshot backfills running_agents from the /status endpoint's live concurrent.agents block — one server-side source of truth. Empty stays empty when no pod is live, so legitimate between-spawn quiescence is not misreported as a running cohort (or a stall). --- orchestrator/mcp_tools.py | 32 +++++ orchestrator/routes/pipelines.py | 80 +++++++++++- orchestrator/tests/test_concurrent_status.py | 122 +++++++++++++++++++ orchestrator/tests/test_mcp_tools.py | 84 +++++++++++++ 4 files changed, 317 insertions(+), 1 deletion(-) diff --git a/orchestrator/mcp_tools.py b/orchestrator/mcp_tools.py index 79970dd9e3..01da7616a8 100644 --- a/orchestrator/mcp_tools.py +++ b/orchestrator/mcp_tools.py @@ -1536,6 +1536,26 @@ def _handle_get_status(self, args: dict[str, Any]) -> dict[str, Any]: """ return self._build_status_snapshot(args["task_id"]) + def _live_running_agents_fallback(self, task_id: str) -> list[dict[str, Any]]: + """Live running-agent view from the ``/status`` concurrent block (#3230). + + Used to backfill ``get_status.running_agents`` when the persisted + phase-agent list is empty under the orchestrator-owned event loop + (#3164). The ``/status`` endpoint's ``concurrent.agents`` is + server-side reconstructed from live Job labels (see + ``routes.pipelines._live_event_agents``), so it reflects role pods + that are actually ``Running``. ``task_id`` must already be URL-quoted + (the caller quotes it once for both requests). Best-effort: any + failure or a non-concurrent phase yields ``[]``. + """ + try: + status_result = self._make_request(f"/api/v1/pipelines/{task_id}/status") + concurrent = status_result.get("data", {}).get("concurrent", {}) or {} + except Exception: + return [] + agents = concurrent.get("agents", []) or [] + return [a for a in agents if a.get("status") == "running"] + def _build_status_snapshot(self, raw_task_id: str) -> dict[str, Any]: """Build the full enriched status snapshot for a pipeline. @@ -1601,6 +1621,18 @@ def _build_status_snapshot(self, raw_task_id: str) -> dict[str, Any]: status["running_agents"] = [a for a in agents if a.get("status") == "running"] status["completed_agents"] = [a for a in agents if a.get("status") == "complete"] + # Under the orchestrator-owned BRC event loop (#3164, unconditional) + # role pods are on-demand one-shots that are never persisted into the + # phase's agent list, so the persisted view above is empty even while + # role pods are Running — a blind dashboard (#3230). Backfill + # ``running_agents`` from the ``/status`` endpoint's live + # ``concurrent.agents`` view (server-side reconstructed from live Job + # labels) so the dashboard reflects the live cohort during + # event-loop-owned phases. Stays empty when no pod is live, so + # between-spawn quiescence is not misreported as running. + if not status["running_agents"]: + status["running_agents"] = self._live_running_agents_fallback(task_id) + # Server-computed timing (#1702) now = datetime.now(UTC) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index ec75168cae..c23aa0708f 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -1125,6 +1125,73 @@ def _count_live_pods_for_pipeline(pipeline_id: str, *, quiet: bool = False) -> i return None +def _live_event_agents(pipeline_id: str, slice_id: str | None) -> list[dict[str, Any]]: + """Running-agent view reconstructed from live Job labels (#3230). + + Under the orchestrator-owned BRC event loop (#3164, now unconditional) + each role's pod is an on-demand one-shot the loop deliberately does NOT + persist into ``phase_exec.agents`` — ``event_loop.py`` treats the + consensus tracker plus live-Job labels as the only sources of truth. So + the persisted agent list is empty even while role pods are ``Running``, + which the dashboard (``get_status.running_agents``) and the overseer + (``concurrent.agents`` stall-duration math) both read as "0 running + agents" — a blind dashboard and false ``phase stalled`` alerts. + + This reconstructs the running-pod cohort from the labels that ARE + authoritative. Live = ``status`` in :data:`_LIVE_POD_STATUSES` + (Pending / Creating / Running); terminal pods lingering in the + ``ttlSecondsAfterFinished`` window are excluded so between-spawn + quiescence reads as "no running agents" (the normal idle state, not a + stall). Scoped to ``slice_id`` when supplied so a slice-DAG implement + phase reports its own slice's pods rather than a cross-slice union; + refine/plan phases are unsliced and query by pipeline label alone. + + Entry shape mirrors the persisted ``agents`` entries (``role`` / + ``status`` / ``started_at`` / ``elapsed_seconds`` / ``container_id``) + so consumers need no special-casing. ``status`` is reported as + ``"running"`` for every live pod — Pending/Creating pods are agents + spinning up, and the dashboard's running-agent filter keys on that + literal. + + Best-effort: an absent/failed label query yields ``[]`` (callers treat + that identically to "no persisted agents", so there is no regression + versus the pre-fix behavior). + """ + try: + spawner = _get_spawner() + labels = {LABEL_PIPELINE_ID: pipeline_id} + if slice_id: + labels[LABEL_SLICE_ID] = slice_id + pods = spawner.backend.list_containers(labels=labels) + except Exception as e: # noqa: BLE001 — observability backfill is best-effort + logger.debug( + "Live event-agent backfill query failed (#3230)", + pipeline_id=pipeline_id, + slice_id=slice_id, + error=str(e), + ) + return [] + + now = datetime.now(UTC) + entries: list[dict[str, Any]] = [] + for pod in pods: + if pod.status not in _LIVE_POD_STATUSES: + continue + role = pod.agent_role.value if pod.agent_role is not None else None + if not role: + continue + entry: dict[str, Any] = {"role": role, "status": "running"} + if isinstance(pod.container_id, str) and pod.container_id: + entry["container_id"] = pod.container_id + started_at = pod.started_at + if isinstance(started_at, datetime): + started_dt = started_at if started_at.tzinfo else started_at.replace(tzinfo=UTC) + entry["started_at"] = started_dt.isoformat() + entry["elapsed_seconds"] = max(0, int((now - started_dt).total_seconds())) + entries.append(entry) + return entries + + def _slice_agents_alive(spawner: Any, pipeline_id: str, slice_id: str) -> bool: """Check if any live agents exist for a slice (#2914). @@ -4516,9 +4583,9 @@ def _get_concurrent_status(pipeline: Pipeline, slice_id: str | None = None) -> d # ``started_at`` rather than pre-restart message-bus events (issue #2084). current_phase_name = pipeline.current_phase.value phase_exec = pipeline.phases.get(current_phase_name) + agents_info: list[dict[str, Any]] = [] if phase_exec and hasattr(phase_exec, "agents"): now = datetime.now(UTC) - agents_info = [] for agent in phase_exec.agents: if hasattr(agent, "role"): role = agent.role.value if hasattr(agent.role, "value") else str(agent.role) @@ -4551,6 +4618,17 @@ def _get_concurrent_status(pipeline: Pipeline, slice_id: str | None = None) -> d entry["elapsed_seconds"] = max(0, int((now - started_dt).total_seconds())) agents_info.append(entry) + + # When the persisted phase-agent list is empty, backfill the + # running-pod view from live Job labels (#3230). Under the + # orchestrator-owned event loop (#3164) on-demand one-shot pods are + # never persisted into ``phase_exec.agents``, so without this the + # overseer's stall-duration math and the dashboard see "0 running + # agents" while role pods are demonstrably ``Running``. Empty stays + # empty when no pod is live, so legitimate between-spawn quiescence is + # not misreported as a cohort. + if not agents_info: + agents_info = _live_event_agents(pipeline.id, slice_id) result["agents"] = agents_info return result diff --git a/orchestrator/tests/test_concurrent_status.py b/orchestrator/tests/test_concurrent_status.py index 2d0fb9c2e1..c81b494192 100644 --- a/orchestrator/tests/test_concurrent_status.py +++ b/orchestrator/tests/test_concurrent_status.py @@ -221,6 +221,128 @@ def test_agents_without_role_attribute(self): assert result["agents"][0]["status"] == "unknown" +class TestLiveEventAgentBackfill: + """Live running-agent backfill under the orchestrator-owned event loop (#3230). + + Under the unconditional event loop (#3164) role pods are on-demand + one-shots that are never persisted into ``phase_exec.agents``, so the + persisted agent list is empty even while role pods are ``Running``. + ``_get_concurrent_status`` backfills the running-pod view from live Job + labels so the dashboard and the overseer's stall-duration math are not + blind. These tests pin that backfill: it fires only when the persisted + list is empty, filters terminal pods, maps role / elapsed, and scopes to + the slice when one is supplied. + """ + + @staticmethod + def _pod(role, status, *, started_offset_s=None, container_id="cid"): + from datetime import UTC, datetime, timedelta + + from models import AgentRole, ContainerInfo, ContainerStatus + + started_at = None + if started_offset_s is not None: + started_at = datetime.now(UTC) - timedelta(seconds=started_offset_s) + return ContainerInfo( + container_id=container_id, + container_name=f"job-{role}", + status=ContainerStatus(status), + started_at=started_at, + agent_role=AgentRole(role), + ) + + def test_backfill_populates_from_live_pods_when_persisted_empty(self): + """An empty persisted agent list is backfilled from live Running pods.""" + pipeline = _make_concurrent_pipeline() + + mock_phase_exec = MagicMock() + mock_phase_exec.agents = [] # event loop persists nothing + pipeline.phases["implement"] = mock_phase_exec + + spawner = MagicMock() + spawner.backend.list_containers.return_value = [ + self._pod("refiner", "running", started_offset_s=152), + self._pod("reviewer_refine", "pending"), + ] + with patch("routes.pipelines._get_spawner", return_value=spawner): + result = _get_concurrent_status(pipeline) + + roles = {a["role"] for a in result["agents"]} + assert roles == {"refiner", "reviewer_refine"} + # All live pods (incl. Pending startup) report as "running" so the + # dashboard's running-agent filter surfaces them. + assert all(a["status"] == "running" for a in result["agents"]) + refiner = next(a for a in result["agents"] if a["role"] == "refiner") + assert 150 <= refiner["elapsed_seconds"] <= 160 + + def test_backfill_excludes_terminal_pods(self): + """Pods lingering in the TTL window after exit must not count as live.""" + pipeline = _make_concurrent_pipeline() + mock_phase_exec = MagicMock() + mock_phase_exec.agents = [] + pipeline.phases["implement"] = mock_phase_exec + + spawner = MagicMock() + spawner.backend.list_containers.return_value = [ + self._pod("refiner", "exited"), + self._pod("reviewer_refine", "failed"), + ] + with patch("routes.pipelines._get_spawner", return_value=spawner): + result = _get_concurrent_status(pipeline) + + # Terminal pods filtered → no live cohort (quiescence, not a stall). + assert result["agents"] == [] + + def test_persisted_agents_take_precedence_over_backfill(self): + """When the phase already records agents, the live query is not run.""" + pipeline = _make_concurrent_pipeline() + mock_phase_exec = MagicMock() + mock_agent = MagicMock(spec=["role", "status"]) + mock_agent.role = "coder" + mock_agent.status.value = "running" + mock_phase_exec.agents = [mock_agent] + pipeline.phases["implement"] = mock_phase_exec + + spawner = MagicMock() + with patch("routes.pipelines._get_spawner", return_value=spawner): + result = _get_concurrent_status(pipeline) + + assert [a["role"] for a in result["agents"]] == ["coder"] + spawner.backend.list_containers.assert_not_called() + + def test_backfill_is_slice_scoped_when_slice_given(self): + """A slice-scoped query filters live pods to that slice's label.""" + from kubernetes_client import LABEL_PIPELINE_ID, LABEL_SLICE_ID + + pipeline = _make_concurrent_pipeline() + mock_phase_exec = MagicMock() + mock_phase_exec.agents = [] + pipeline.phases["implement"] = mock_phase_exec + + spawner = MagicMock() + spawner.backend.list_containers.return_value = [] + with patch("routes.pipelines._get_spawner", return_value=spawner): + _get_concurrent_status(pipeline, slice_id="slice-2") + + spawner.backend.list_containers.assert_called_once_with( + labels={LABEL_PIPELINE_ID: pipeline.id, LABEL_SLICE_ID: "slice-2"} + ) + + def test_backfill_best_effort_on_query_failure(self): + """A failed label query degrades to an empty cohort, never raises.""" + pipeline = _make_concurrent_pipeline() + mock_phase_exec = MagicMock() + mock_phase_exec.agents = [] + pipeline.phases["implement"] = mock_phase_exec + + spawner = MagicMock() + spawner.backend.list_containers.side_effect = RuntimeError("k8s unreachable") + with patch("routes.pipelines._get_spawner", return_value=spawner): + result = _get_concurrent_status(pipeline) + + assert result["agents"] == [] + + class TestGetConcurrentStatusSliceAware: """_get_concurrent_status resolves the per-slice BRC tracker (#2761). diff --git a/orchestrator/tests/test_mcp_tools.py b/orchestrator/tests/test_mcp_tools.py index cd617753a3..c64361815f 100644 --- a/orchestrator/tests/test_mcp_tools.py +++ b/orchestrator/tests/test_mcp_tools.py @@ -1273,6 +1273,34 @@ def _pipeline_response_with_pr(self, pr_url: str): resp["data"]["pipeline"]["pr_url"] = pr_url return resp + def _pipeline_response_no_persisted_agents(self): + """Pipeline fixture whose phase records NO persisted agents (#3230). + + This is the steady state under the orchestrator-owned event loop + (#3164): role pods are on-demand one-shots that are never written + into ``phase_exec.agents``, so the persisted list is empty. + """ + resp = self._pipeline_response() + resp["data"]["pipeline"]["phases"]["implement"]["agents"] = [] + return resp + + def _status_response_with_live_agents(self): + """``/status`` fixture whose ``concurrent.agents`` carries live pods. + + Mirrors what ``routes.pipelines._get_concurrent_status`` returns once + it backfills the running-pod cohort from live Job labels. + """ + return { + "data": { + "concurrent": { + "agents": [ + {"role": "refiner", "status": "running", "elapsed_seconds": 152}, + {"role": "reviewer_refine", "status": "running"}, + ] + } + } + } + def _messages_response(self): return {"data": {"messages": []}} @@ -1564,6 +1592,62 @@ def test_phase_started_at_preserved_as_iso_string(self, handler): parsed = datetime.fromisoformat(result["phase_started_at"]) assert parsed.tzinfo is not None # timezone-aware + def test_running_agents_backfilled_from_live_status_when_persisted_empty(self, handler): + """Empty persisted agents are backfilled from /status live cohort (#3230). + + Under the orchestrator-owned event loop the persisted agent list is + empty even while role pods are Running; ``get_status`` must fall back + to the ``/status`` endpoint's live ``concurrent.agents`` so the + dashboard is not blind. Request order: pipeline → /status → messages. + """ + with patch.object( + handler, + "_make_request", + side_effect=[ + self._pipeline_response_no_persisted_agents(), + self._status_response_with_live_agents(), + self._messages_response(), + ], + ): + result = handler.handle_tool_call("get_status", {"task_id": "issue-42"}) + + roles = {a["role"] for a in result["running_agents"]} + assert roles == {"refiner", "reviewer_refine"} + + def test_running_agents_no_fallback_when_persisted_present(self, handler): + """Persisted running agents short-circuit the /status fallback (#3230). + + Only two requests (pipeline + messages) are issued — no /status call — + when the persisted list already has a running agent. + """ + mock_make = MagicMock(side_effect=[self._pipeline_response(), self._messages_response()]) + with patch.object(handler, "_make_request", mock_make): + result = handler.handle_tool_call("get_status", {"task_id": "issue-42"}) + + assert [a["role"] for a in result["running_agents"]] == ["coder"] + # No /status request was made — exactly pipeline + messages. + assert mock_make.call_count == 2 + + def test_running_agents_empty_when_no_live_pods(self, handler): + """Quiescence (no live pods) stays empty — not misreported (#3230). + + A backfill that finds no live pods must leave ``running_agents`` + empty so legitimate between-spawn idle is not dressed up as a cohort. + """ + empty_status = {"data": {"concurrent": {"agents": []}}} + with patch.object( + handler, + "_make_request", + side_effect=[ + self._pipeline_response_no_persisted_agents(), + empty_status, + self._messages_response(), + ], + ): + result = handler.handle_tool_call("get_status", {"task_id": "issue-42"}) + + assert result["running_agents"] == [] + class TestGetStatusWedgedNoSuccessor: """Tests for the ``wedged_no_successor`` watchdog field (#2166). From cfe67a4b20ff86cb1dacf861ba2eb056dea116e2 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Thu, 25 Jun 2026 18:53:11 +0000 Subject: [PATCH 2/2] Log debug line when live running-agent backfill query fails Mirror routes.pipelines._live_event_agents, which logs at debug on a failed label query. The mcp_tools-side _live_running_agents_fallback previously swallowed exceptions silently, making a repeatedly failing /status round-trip invisible. Add a matching logger.debug for symmetry. --- orchestrator/mcp_tools.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/orchestrator/mcp_tools.py b/orchestrator/mcp_tools.py index 01da7616a8..c47c350fd8 100644 --- a/orchestrator/mcp_tools.py +++ b/orchestrator/mcp_tools.py @@ -1551,7 +1551,14 @@ def _live_running_agents_fallback(self, task_id: str) -> list[dict[str, Any]]: try: status_result = self._make_request(f"/api/v1/pipelines/{task_id}/status") concurrent = status_result.get("data", {}).get("concurrent", {}) or {} - except Exception: + except Exception as e: + # Symmetric with routes.pipelines._live_event_agents: a repeatedly + # failing /status round-trip is otherwise invisible (#3230). + logger.debug( + "Live running-agent backfill query failed (#3230)", + task_id=task_id, + error=str(e), + ) return [] agents = concurrent.get("agents", []) or [] return [a for a in agents if a.get("status") == "running"]