diff --git a/apps/api/src/cora/agent/seed_status_publisher.py b/apps/api/src/cora/agent/seed_status_publisher.py index 7f5c24475f7..52bae134240 100644 --- a/apps/api/src/cora/agent/seed_status_publisher.py +++ b/apps/api/src/cora/agent/seed_status_publisher.py @@ -15,17 +15,15 @@ (`prompt_template_id=None`) and a Rule brain (`BrainRef.for_rule("StatusPublisher:v1")`). Never used to build an LLM: the runtime is a read-and-relay loop, not an LLM subscriber. - - Authorization: `_status_push.py` reads across nine BCs to assemble the + - Authorization: `_status_push.py` reads across ten BCs to assemble the snapshot it relays: `ListPlans` (Recipe), `ListRuns` and `GetRunHistory` (Run), `ListSubjects` (Subject), `ListCampaigns` (Campaign), `ListDatasets` (Data), `ListProcedures` (Operation), `ListClearances` (Safety), `ListEnclosures` and `GetEnclosureHistory` - (Enclosure), and `ListDecisions` (Decision). This identity only seeds - the Agent record; `_status_push.py` still issues every one of those - reads as `SYSTEM_PRINCIPAL_ID` and does not yet act as this agent. - Switching it over, and granting this principal the eleven commands - above under a real `TrustAuthorize` Policy, is separate follow-up - work, not part of this seed. + (Enclosure), `ListSupplies` (Supply), and `ListDecisions` (Decision). + Every one of those reads is issued as this principal. Granting it the + twelve commands above under a real `TrustAuthorize` Policy is + separate follow-up work, not part of this seed. """ from __future__ import annotations diff --git a/apps/api/src/cora/api/_status_push.py b/apps/api/src/cora/api/_status_push.py index b3fe30c3d61..717e75d6245 100644 --- a/apps/api/src/cora/api/_status_push.py +++ b/apps/api/src/cora/api/_status_push.py @@ -42,6 +42,11 @@ something required to start a run currently in force) - Active Enclosures (permit status: the most direct "is it safe right now" answer CORA records) + - every Supply a run can draw on, with the status CORA holds and the + reason it last moved. Deliberately NOT narrowed to the healthy ones: + `Unavailable` after an interlock trip, `Recovering` while it waits + for an operator, and never-observed `Unknown` are precisely the + states worth showing (see `_drain_supplies`) - the most recent Decisions since this process started, tail-followed (see `_DecisionTail`) rather than paged from the beginning, since Decisions have no "open" status to filter on and the table is @@ -209,6 +214,8 @@ from cora.safety.features.list_clearances import ListClearances from cora.subject.errors import UnauthorizedError as _SubjectUnauthorizedError from cora.subject.features.list_subjects import ListSubjects +from cora.supply.errors import UnauthorizedError as _SupplyUnauthorizedError +from cora.supply.features.list_supplies import ListSupplies if TYPE_CHECKING: from collections.abc import AsyncGenerator, Awaitable, Callable @@ -240,6 +247,7 @@ from cora.run.features.list_runs.handler import Handler as ListRunsHandler from cora.safety.features.list_clearances.handler import Handler as ListClearancesHandler from cora.subject.features.list_subjects.handler import Handler as ListSubjectsHandler + from cora.supply.features.list_supplies.handler import Handler as ListSuppliesHandler _log = get_logger(__name__) @@ -258,6 +266,7 @@ _OPEN_CAMPAIGN_STATUSES: list[CampaignStatusFilter] = ["Planned", "Active", "Held"] _ACTIVE_CLEARANCE_STATUS = "Active" _ACTIVE_ENCLOSURE_LIFECYCLE = "Active" +_DECOMMISSIONED_SUPPLY_STATUS = "Decommissioned" _DECISION_RING_SIZE = 20 _PROGRESS_TRAIL_POINTS = 30 """Cap on how many trail points ride the wire per (run, role), independent @@ -333,6 +342,7 @@ _SafetyUnauthorizedError, _EnclosureUnauthorizedError, _OperationUnauthorizedError, + _SupplyUnauthorizedError, ) @@ -696,6 +706,45 @@ async def _drain_active_enclosures( return rows, raw_enclosure_ids +async def _drain_supplies(list_supplies: ListSuppliesHandler, deps: Kernel) -> list[dict[str, Any]]: + """Every Supply a run can draw on, with the status CORA currently + holds and the reason it last moved. + + Unlike every other drain here, this one does NOT narrow to the open + or active rows. A Supply's whole value to a viewer is the state it is + resting in, and the states worth seeing are exactly the bad ones: + `Unavailable` after a BLEPS trip, `Recovering` while it waits for an + operator to accept it back, `Unknown` for one nothing has ever + observed. Filtering to "active" would leave the panel empty on the + days it matters most. + + `Decommissioned` is dropped in Python rather than by the query + because `ListSupplies.status` takes a single value, so "everything + except one terminal state" is not expressible as a filter. + """ + items = await _drain_all( + lambda cursor: list_supplies( + ListSupplies(cursor=cursor, limit=_PAGE_LIMIT), + principal_id=STATUS_PUBLISHER_AGENT_ID, + correlation_id=deps.id_generator.new_id(), + surface_id=SYSTEM_IN_PROCESS_SURFACE_ID, + ) + ) + return [ + { + "supply_id": render_value(item.supply_id), + "name": item.name, + "kind": item.kind, + "status": item.status, + "last_status_reason": item.last_status_reason, + "last_status_changed_at": render_value(item.last_status_changed_at), + "last_trigger": item.last_trigger, + } + for item in items + if item.status != _DECOMMISSIONED_SUPPLY_STATUS + ] + + class _FleetReadinessTail: """Holds the fleet's readiness across ticks, re-reading it rarely. @@ -1269,6 +1318,7 @@ def build_snapshot( procedures: list[dict[str, Any]], clearances: list[dict[str, Any]], enclosures: list[dict[str, Any]], + supplies: list[dict[str, Any]], decisions: list[dict[str, Any]], agents: dict[str, Any], sequence: int, @@ -1290,6 +1340,7 @@ def build_snapshot( "procedures": procedures, "clearances": clearances, "enclosures": enclosures, + "supplies": supplies, "decisions": decisions, "agents": agents, } @@ -1532,6 +1583,7 @@ async def _build_payload_fields( list_clearances: ListClearancesHandler, list_plans: ListPlansHandler, list_enclosures: ListEnclosuresHandler, + list_supplies: ListSuppliesHandler, decision_tail: _DecisionTail, fleet_tail: _FleetReadinessTail, list_decisions: ListDecisionsHandler, @@ -1579,6 +1631,7 @@ async def _build_payload_fields( "procedures": await _drain_procedures_for_runs(list_procedures, deps, run_ids=raw_run_ids), "clearances": await _drain_active_clearances(list_clearances, deps), "enclosures": enclosures, + "supplies": await _drain_supplies(list_supplies, deps), "decisions": await decision_tail.poll(list_decisions, deps), "agents": await fleet_tail.poll(deps), } @@ -1619,6 +1672,7 @@ async def _push_loop( list_clearances: ListClearancesHandler, list_plans: ListPlansHandler, list_enclosures: ListEnclosuresHandler, + list_supplies: ListSuppliesHandler, list_decisions: ListDecisionsHandler, get_run_history: GetRunHistoryHandler, get_enclosure_history: GetEnclosureHistoryHandler, @@ -1694,6 +1748,7 @@ async def _push_loop( list_clearances=list_clearances, list_plans=list_plans, list_enclosures=list_enclosures, + list_supplies=list_supplies, decision_tail=decision_tail, fleet_tail=fleet_tail, list_decisions=list_decisions, @@ -1772,6 +1827,7 @@ async def status_push_lifespan( list_clearances: ListClearancesHandler, list_plans: ListPlansHandler, list_enclosures: ListEnclosuresHandler, + list_supplies: ListSuppliesHandler, list_decisions: ListDecisionsHandler, get_run_history: GetRunHistoryHandler, get_enclosure_history: GetEnclosureHistoryHandler, @@ -1837,6 +1893,7 @@ async def status_push_lifespan( list_clearances=list_clearances, list_plans=list_plans, list_enclosures=list_enclosures, + list_supplies=list_supplies, list_decisions=list_decisions, get_run_history=get_run_history, get_enclosure_history=get_enclosure_history, diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 381032fb4c6..b844db05723 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -1557,6 +1557,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: list_clearances=app.state.safety.list_clearances, list_plans=app.state.recipe.list_plans, list_enclosures=app.state.enclosure.list_enclosures, + list_supplies=app.state.supply.list_supplies, list_decisions=app.state.decision.list_decisions, get_run_history=app.state.run.get_run_history, get_enclosure_history=app.state.enclosure.get_enclosure_history, diff --git a/apps/api/tests/unit/api/test_status_push.py b/apps/api/tests/unit/api/test_status_push.py index 24b37721aa2..a00da6f6853 100644 --- a/apps/api/tests/unit/api/test_status_push.py +++ b/apps/api/tests/unit/api/test_status_push.py @@ -92,6 +92,7 @@ ListClearances, ) from cora.subject.features.list_subjects import ListSubjects, SubjectListPage, SubjectSummaryItem +from cora.supply.features.list_supplies import ListSupplies, SupplyListPage, SupplySummaryItem _NOW = datetime(2026, 6, 22, 12, 0, 0, tzinfo=UTC) @@ -109,6 +110,7 @@ def test_build_snapshot_shape() -> None: procedures=[], clearances=[], enclosures=[], + supplies=[], decisions=[], agents={"ready": 2, "total": 2, "not_ready": [], "held": [], "absent": []}, sequence=3, @@ -128,6 +130,7 @@ def test_build_snapshot_shape() -> None: "procedures": [], "clearances": [], "enclosures": [], + "supplies": [], "decisions": [], "agents": {"ready": 2, "total": 2, "not_ready": [], "held": [], "absent": []}, } @@ -440,6 +443,20 @@ async def list_enclosures( return list_enclosures +def _make_list_supplies(items: list[SupplySummaryItem]): + async def list_supplies( + query: ListSupplies, + *, + principal_id: UUID, + correlation_id: UUID, + surface_id: UUID = NIL_SENTINEL_ID, + ) -> SupplyListPage: + matching = [i for i in items if query.status is None or i.status == query.status] + return SupplyListPage(items=matching, next_cursor=None) + + return list_supplies + + def _make_list_decisions(items: list[DecisionSummaryItem]): async def list_decisions( query: ListDecisions, @@ -509,6 +526,7 @@ def _default_handlers(**overrides: Any) -> dict[str, Any]: "list_clearances": _make_list_clearances([]), "list_plans": _make_list_plans([]), "list_enclosures": _make_list_enclosures([]), + "list_supplies": _make_list_supplies([]), "list_decisions": _make_list_decisions([]), "get_run_history": _make_get_run_history(), "get_enclosure_history": _make_get_enclosure_history(), @@ -2400,6 +2418,68 @@ async def handler(ws: ServerConnection) -> None: assert snapshot["enclosures"][0]["name"] == "2-BM-B" +def _supply(name: str, status: str, reason: str | None = None) -> SupplySummaryItem: + return SupplySummaryItem( + supply_id=uuid4(), + kind="CoolingWater", + name=name, + facility_code="cora", + containing_asset_id=None, + status=status, + registered_at=_NOW, + last_status_changed_at=_NOW if reason else None, + last_status_reason=reason, + last_trigger="Monitor" if reason else None, + ) + + +@pytest.mark.unit +async def test_lifespan_pushes_unhealthy_supplies_and_drops_only_decommissioned() -> None: + """The supplies drain deliberately does NOT narrow to the healthy + rows, unlike every other drain here. + + A Supply is worth showing precisely when it is in a bad state, so + `Unavailable` (tripped), `Recovering` (waiting on an operator) and + `Unknown` (nothing has ever observed it) must all survive to the + wire. Narrowing this drain the way the enclosure one narrows to + Active would empty the panel on exactly the days it earns its place, + and that regression would otherwise look like a passing suite. + Only `Decommissioned`, which is terminal, is dropped. + """ + received: asyncio.Queue[str] = asyncio.Queue() + + async def handler(ws: ServerConnection) -> None: + async for message in ws: + await received.put(message if isinstance(message, str) else message.decode()) + + async with serve(handler, "127.0.0.1", 0) as server: + port = next(iter(server.sockets)).getsockname()[1] + url = f"ws://127.0.0.1:{port}/ingest" + kernel = _kernel( + status_push_enabled=True, status_push_url=url, status_push_tick_seconds=0.1 + ) + supplies = [ + _supply("cooling water", "Unavailable", "BLEPS trip: Flow4"), + _supply("vacuum", "Recovering", "trips clear; awaiting operator"), + _supply("durable-tier", "Unknown"), + _supply("staging", "Available", "seeded"), + _supply("retired dewar", "Decommissioned"), + ] + async with status_push_lifespan( + kernel, + **_default_handlers(list_supplies=_make_list_supplies(supplies)), + ): + raw = await asyncio.wait_for(received.get(), timeout=5) + + snapshot = json.loads(raw) + by_name = {s["name"]: s for s in snapshot["supplies"]} + assert set(by_name) == {"cooling water", "vacuum", "durable-tier", "staging"} + assert by_name["cooling water"]["status"] == "Unavailable" + assert by_name["cooling water"]["last_status_reason"] == "BLEPS trip: Flow4" + assert by_name["cooling water"]["last_trigger"] == "Monitor" + assert by_name["durable-tier"]["last_status_changed_at"] is None + + @pytest.mark.unit async def test_lifespan_pushes_enclosure_timeline_alongside_the_snapshot() -> None: """End to end against a real socket: an Active enclosure with a diff --git a/infra/status-relay/design/fixtures.js b/infra/status-relay/design/fixtures.js index 94b01828e94..f85aae9d23f 100644 --- a/infra/status-relay/design/fixtures.js +++ b/infra/status-relay/design/fixtures.js @@ -29,14 +29,19 @@ Clearances: [["ClearanceReviewStepAppended", 3], ["ClearanceApproved", 2], ["ClearanceExpired", 1], ["ClearanceRejected", 1]], Cautions: [["CautionRegistered", 2], ["CautionRetired", 1]], + // Deliberately the sparsest mix here. A Supply transitions a handful of + // times a run cycle, and the lane is drawn even when it holds nothing, + // so the harness has to show what a near-empty always-on lane looks like. + Supplies: [["SupplyMarkedUnavailable", 1], ["SupplyMarkedRecovering", 1]], Enclosures: [["EnclosurePermitObserved", 8], ["EnclosureDecommissioned", 1]], Decisions: [["DecisionRegistered", 40], ["DecisionRated", 8], ["DecisionLogbookOpened", 3], ["DecisionDebriefRequested", 1]], Other: [["ActorRegistered", 3], ["CalibrationRecorded", 4], ["AllocationGranted", 2]], }; var LANE_WEIGHT = { Runs: 47, Decisions: 44, Procedures: 2.4, Datasets: 2.2, Subjects: 1.2, - Enclosures: 1.1, Campaigns: 0.8, Clearances: 0.6, Other: 0.5, Cautions: 0.2 }; - var LANE_ORDER = ["Runs", "Procedures", "Subjects", "Campaigns", "Datasets", + Enclosures: 1.1, Campaigns: 0.8, Clearances: 0.6, Other: 0.5, Cautions: 0.2, + Supplies: 0.1 }; + var LANE_ORDER = ["Runs", "Procedures", "Subjects", "Supplies", "Campaigns", "Datasets", "Clearances", "Cautions", "Enclosures", "Decisions", "Other"]; // Mirrors page.html's own EVENT_TIER. Duplicated deliberately: the harness @@ -48,6 +53,8 @@ ClearanceRejected: 2, DatasetDiscarded: 2, EnclosureDecommissioned: 2, RunStopped: 1, RunTruncated: 1, RunResumed: 1, CampaignHeld: 1, SubjectDiscarded: 1, DatasetDemoted: 1, CautionRetired: 1, DecisionDebriefRequested: 1, + SupplyMarkedUnavailable: 2, SupplyDegraded: 1, SupplyMarkedRecovering: 1, + SupplyDeregistered: 1, }; function mulberry32(a) { diff --git a/infra/status-relay/page.html b/infra/status-relay/page.html index cf19c88b8a5..505d9ac0f18 100644 --- a/infra/status-relay/page.html +++ b/infra/status-relay/page.html @@ -701,6 +701,15 @@