fix(orchestrator): surface live event-loop pods in running-agent views (#3230)

orchestrator/mcp_tools.py

@@ -1536,6 +1536,33 @@ 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 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"]
+
     def _build_status_snapshot(self, raw_task_id: str) -> dict[str, Any]:
         """Build the full enriched status snapshot for a pipeline.
 
@@ -1601,6 +1628,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)
 

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

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

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