Make a stranded agent fleet say so, and give 2-BM's three roles a slot each

apps/api/src/cora/agent/fleet_readiness.py

@@ -0,0 +1,169 @@
+"""Whether the shipped fleet can actually act, as a value and as a verdict.
+
+`_seeded_fleet` made the fleet a value so a gesture could be made over
+it. This asks the question that value exists for, and says the answer out
+loud at boot.
+
+## Why this is a warning and not a metric
+
+A `Defined` Agent is registered and inert: the subscribers' lifecycle
+gate fires on `Versioned` only, and refuses anything less WITHOUT
+SAYING SO. On the 2-BM pilot that stranded seventeen agents for three
+months behind a log that looked clean the entire time. The remedy
+(`promote_seeded_fleet`) has existed since; what did not exist was
+anything that told an operator the remedy was needed.
+
+So the point here is not the number. It is that a deployment whose fleet
+cannot act now says so on every boot, at `warning`, naming the members.
+An absence has to be its own loud verdict, because the alternative is
+what already happened: a correct system, a clean log, and nothing
+running.
+
+## Four not-ready reasons, kept apart
+
+They are not interchangeable and collapsing them would restore the
+silence at one remove:
+
+  `not_ready`  `Defined`. Never promoted. THE silent case, and the only
+               one this treats as a fault, because it is the only one
+               nobody chose.
+  `held`       `Suspended`. A live operator decision. Reported, never
+               warned about; a deployment pausing an agent on purpose
+               should not be nagged for it.
+  `retired`    `Deprecated`. Terminal and deliberate.
+  `absent`     Not in the record at all. Worth seeing on a partly-seeded
+               deployment rather than counting as a silent zero, the same
+               reasoning `promote_seeded_fleet` applies to it.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from cora.agent._seeded_fleet import SEEDED_FLEET
+from cora.agent.aggregates.agent import AgentStatus, load_agent
+from cora.infrastructure.logging import get_logger
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping, Sequence
+    from uuid import UUID
+
+    from cora.agent._seeded_fleet import SeededAgent
+    from cora.infrastructure.ports import EventStore
+
+_log = get_logger(__name__)
+
+
+@dataclass(frozen=True)
+class FleetReadiness:
+    """Fleet members by whether they can act, named rather than counted.
+
+    Names and not ids: this is read by an operator deciding whether to
+    run a promotion, and `RunWitness` answers that question where
+    `01900000-0000-7000-8000-0000aaaa0010` does not. The ids stay
+    reachable through `SEEDED_FLEET` for anything that needs to act on a
+    member.
+    """
+
+    ready: tuple[str, ...]
+    not_ready: tuple[str, ...]
+    held: tuple[str, ...]
+    retired: tuple[str, ...]
+    absent: tuple[str, ...]
+
+    @property
+    def total(self) -> int:
+        return (
+            len(self.ready)
+            + len(self.not_ready)
+            + len(self.held)
+            + len(self.retired)
+            + len(self.absent)
+        )
+
+    @property
+    def stranded(self) -> bool:
+        """Whether any member is inert for a reason nobody chose."""
+        return bool(self.not_ready)
+
+
+def fleet_readiness(
+    statuses: Mapping[UUID, AgentStatus | None],
+    fleet: Sequence[SeededAgent] = SEEDED_FLEET,
+) -> FleetReadiness:
+    """Sort the fleet by readiness. Pure, so the unit tier drives every
+    branch without a record.
+
+    Ranges over `fleet`, never over `statuses`: a member the caller
+    failed to look up has to land in `absent`, and a map missing that key
+    would otherwise drop it from the count entirely. The total is then a
+    property of the shipped fleet, which is the only thing that makes
+    "4 of 20" mean anything.
+    """
+    buckets: dict[str, list[str]] = {
+        "ready": [],
+        "not_ready": [],
+        "held": [],
+        "retired": [],
+        "absent": [],
+    }
+    for member in fleet:
+        match statuses.get(member.agent_id):
+            case AgentStatus.VERSIONED:
+                buckets["ready"].append(member.name)
+            case AgentStatus.DEFINED:
+                buckets["not_ready"].append(member.name)
+            case AgentStatus.SUSPENDED:
+                buckets["held"].append(member.name)
+            case AgentStatus.DEPRECATED:
+                buckets["retired"].append(member.name)
+            case None:
+                buckets["absent"].append(member.name)
+    return FleetReadiness(
+        ready=tuple(buckets["ready"]),
+        not_ready=tuple(buckets["not_ready"]),
+        held=tuple(buckets["held"]),
+        retired=tuple(buckets["retired"]),
+        absent=tuple(buckets["absent"]),
+    )
+
+
+async def read_fleet_readiness(event_store: EventStore) -> FleetReadiness:
+    """Load every shipped member and sort it. Read-only."""
+    statuses: dict[UUID, AgentStatus | None] = {}
+    for member in SEEDED_FLEET:
+        agent = await load_agent(event_store, member.agent_id)
+        statuses[member.agent_id] = None if agent is None else agent.status
+    return fleet_readiness(statuses)
+
+
+def log_fleet_readiness(readiness: FleetReadiness) -> None:
+    """Say it out loud, at a level that matches whether anything is wrong.
+
+    `warning` when a member is stranded, because that is a deployment
+    that will quietly do nothing, and the whole reason this function
+    exists is that the previous behaviour was indistinguishable from a
+    healthy boot. `info` otherwise: a fleet nobody needs to act on should
+    not train an operator to scroll past this line.
+    """
+    if readiness.stranded:
+        _log.warning(
+            "agent_fleet.stranded",
+            ready=len(readiness.ready),
+            total=readiness.total,
+            not_ready=list(readiness.not_ready),
+            remedy="promote_seeded_fleet",
+        )
+    else:
+        _log.info("agent_fleet.ready", ready=len(readiness.ready), total=readiness.total)
+    if readiness.absent:
+        _log.warning("agent_fleet.absent", absent=list(readiness.absent))
+
+
+__all__ = [
+    "FleetReadiness",
+    "fleet_readiness",
+    "log_fleet_readiness",
+    "read_fleet_readiness",
+]

apps/api/src/cora/api/_status_push.py

@@ -177,6 +177,7 @@
 from websockets.asyncio.client import connect
 from websockets.exceptions import ConnectionClosed, InvalidStatus
 
+from cora.agent.fleet_readiness import read_fleet_readiness
 from cora.agent.seed_status_publisher import STATUS_PUBLISHER_AGENT_ID
 from cora.campaign.errors import UnauthorizedError as _CampaignUnauthorizedError
 from cora.campaign.features.list_campaigns import ListCampaigns
@@ -273,6 +274,20 @@
 REWIND timeline refreshed at the same 2Hz cadence; the terminal push (see
 `_RunHistoryTail.poll`) guarantees the final version is always complete
 regardless of where this clock happened to be when the run closed."""
+_FLEET_READINESS_REFRESH_TICKS = 150
+"""How often (in ticks) the agent fleet's readiness is re-read: 5 minutes at
+the 2.0s default tick.
+
+Reading it costs one event-store load per shipped agent, and the answer only
+changes when an operator promotes, suspends or deprecates one, which is a rare
+deliberate gesture rather than beamline traffic. Re-reading twenty streams at
+2Hz to watch a value that moves monthly would be the most expensive thing in
+the tick loop by a wide margin.
+
+Not cached for the process's lifetime either, which is the other obvious
+choice and the wrong one: an operator who has just run `promote_seeded_fleet`
+is looking at this page to see whether it worked, and "restart the API to find
+out" is not an answer."""
 _ACTIVITY_PAGE_LIMIT = 500
 """Per-`read_since` call cap for `_ActivityTail`. Measured against the real
 2-BM deployment (2026-08-28): 13,822 events total, ever, across 25 stream
@@ -681,6 +696,47 @@ async def _drain_active_enclosures(
     return rows, raw_enclosure_ids
 
 
+class _FleetReadinessTail:
+    """Holds the fleet's readiness across ticks, re-reading it rarely.
+
+    Unlike every other domain here this is not beamline traffic: it is a
+    standing fact about the deployment, true between operator gestures
+    and expensive to ask (one stream load per shipped agent). So it is
+    read on the first tick and then only every
+    `_FLEET_READINESS_REFRESH_TICKS`, and the held value rides every
+    snapshot in between.
+
+    Deliberately NOT skipped when the fleet is healthy. A page that shows
+    the row only when something is wrong teaches its reader that an
+    absent row means nothing to see, which is the same lesson that let a
+    stranded fleet sit unnoticed for three months. "20 of 20 ready" is
+    the row earning trust for the one day it says 4.
+    """
+
+    def __init__(self) -> None:
+        self._held: dict[str, Any] | None = None
+        self._ticks_since_read = 0
+
+    async def poll(self, deps: Kernel) -> dict[str, Any]:
+        due = self._held is None or self._ticks_since_read >= _FLEET_READINESS_REFRESH_TICKS
+        if not due:
+            self._ticks_since_read += 1
+            return self._held if self._held is not None else {}
+        readiness = await read_fleet_readiness(deps.event_store)
+        self._ticks_since_read = 0
+        # Names, never ids: this row exists to tell an operator which
+        # agents will not act, and `RunWitness` answers that where a uuid
+        # does not. Nothing on the page can resolve an agent id anyway.
+        self._held = {
+            "ready": len(readiness.ready),
+            "total": readiness.total,
+            "not_ready": list(readiness.not_ready),
+            "held": list(readiness.held),
+            "absent": list(readiness.absent),
+        }
+        return self._held
+
+
 class _DecisionTail:
     """Tail-follows `list_decisions` since this instance was created,
     keeping only the most recent `_DECISION_RING_SIZE`.
@@ -1214,6 +1270,7 @@ def build_snapshot(
     clearances: list[dict[str, Any]],
     enclosures: list[dict[str, Any]],
     decisions: list[dict[str, Any]],
+    agents: dict[str, Any],
     sequence: int,
     generated_at: str,
     producer_id: str,
@@ -1234,6 +1291,7 @@ def build_snapshot(
         "clearances": clearances,
         "enclosures": enclosures,
         "decisions": decisions,
+        "agents": agents,
     }
 
 
@@ -1475,6 +1533,7 @@ async def _build_payload_fields(
     list_plans: ListPlansHandler,
     list_enclosures: ListEnclosuresHandler,
     decision_tail: _DecisionTail,
+    fleet_tail: _FleetReadinessTail,
     list_decisions: ListDecisionsHandler,
     run_history_tail: _RunHistoryTail,
     get_run_history: GetRunHistoryHandler,
@@ -1485,14 +1544,21 @@ async def _build_payload_fields(
     witness_recorder: RunWitnessRecorder | None,
     generated_at: str,
     producer_id: str,
-) -> tuple[dict[str, list[dict[str, Any]]], list[dict[str, Any]]]:
+) -> tuple[dict[str, Any], list[dict[str, Any]]]:
     """Every domain's rows for one tick, plus any extra messages (run
     history, enclosure timelines, activity) that are new this tick. Each
     drain is independently guarded by the caller's
     `except _UNAUTHORIZED_ERRORS` (per-domain, not caught here) so a
     missing grant on one command blinds only that section of the page,
     never the whole tick.
 
+    `fields` is `dict[str, Any]` rather than a dict of row lists because
+    `agents` is not rows. Every other field answers "which instances are
+    open"; that one answers "can this deployment's fleet act", which is a
+    single standing fact and would be a lie as a list (a healthy fleet
+    would have to be an empty one, and an empty list is exactly how every
+    other field spells "none").
+
     Returns `(fields, extra_messages)`, not one dict: none of run
     history, enclosure timelines, or activity is a snapshot field (see
     `_RunHistoryTail`'s, `_EnclosureTimelineTail`'s, and `_ActivityTail`'s
@@ -1514,6 +1580,7 @@ async def _build_payload_fields(
         "clearances": await _drain_active_clearances(list_clearances, deps),
         "enclosures": enclosures,
         "decisions": await decision_tail.poll(list_decisions, deps),
+        "agents": await fleet_tail.poll(deps),
     }
     extra_messages = await run_history_tail.poll(
         get_run_history,
@@ -1584,6 +1651,7 @@ async def _push_loop(
     token = deps.settings.status_push_token
     headers = {"Authorization": f"Bearer {token.get_secret_value()}"} if token is not None else {}
 
+    fleet_tail = _FleetReadinessTail()
     decision_tail = _DecisionTail(
         started_at_cursor=encode_cursor(created_at=deps.clock.now(), item_id=_MIN_UUID)
     )
@@ -1627,6 +1695,7 @@ async def _push_loop(
                             list_plans=list_plans,
                             list_enclosures=list_enclosures,
                             decision_tail=decision_tail,
+                            fleet_tail=fleet_tail,
                             list_decisions=list_decisions,
                             run_history_tail=run_history_tail,
                             get_run_history=get_run_history,