diff --git a/README.md b/README.md index 7d050779..fc5bb903 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,13 @@ dashboard API layer. dataset on this page), candidate-vs-current-release fit, and a hierarchical map of weighted target-error increases and reductions. Countries without a staging repository show an explicit - unavailable state. + unavailable state. A run's status, stage and times come from the hosted + telemetry collector when it has a record of the run, because a build's + Hugging Face copy can stop updating partway through; the page then names the + source and says what the Hugging Face copy last recorded. On both tabs, a + collector record that stopped before the run ended gives way to a Hugging + Face copy that recorded the end later. Runs only the + collector knows have no candidate files and appear on Build progress only. - **Build progress** (the second tab of Staging candidates, `/microcosm/staging?view=progress`) — follow a build run while it runs, from the staging repository or from the `runs//` telemetry folders a @@ -81,8 +87,8 @@ returns HTTP 404 with `{"error": "country_not_configured", "requested_country", | `GET /api/microcosm/compare?a=&b=` | Version-over-version diff | | `GET /api/microcosm/builds?country=&source=local\|staging` | Every build run from one source as a stage timeline | | `GET /api/microcosm/builds/run?id=&country=&source=local\|staging` | One run's timeline and completion forecast, with stage-time and gate-failure statistics for its pipeline | -| `GET /api/microcosm/staging/runs` | List staging build runs | -| `GET /api/microcosm/staging/run?id=` | One staging run's progress and its candidate diagnostics: an uploaded `calibration_diagnostics` telemetry artifact, or the `calibration_diagnostics.json` of the dataset bundle the run staged under `staged//` in the country's release repository (read at the commit the run recorded, digest-verified) | +| `GET /api/microcosm/staging/runs` | List staging build runs, each with the collector's state when it has a record of the run (`state_source`, Hugging Face values in `hugging_face_*`, and `collector_note` when the collector is unavailable) | +| `GET /api/microcosm/staging/run?id=` | One staging run's progress, its state as in the run list (`state`), and its candidate diagnostics: an uploaded `calibration_diagnostics` telemetry artifact, or the `calibration_diagnostics.json` of the dataset bundle the run staged under `staged//` in the country's release repository (read at the commit the run recorded, digest-verified) | | `GET /api/microcosm/staging/target-diagnostics?id=&...` | Faceted diagnostics for a staging candidate once diagnostics exist | | `GET /api/microcosm/staging/compare?run=&release=latest` | Diff a staging candidate against a published release: by default the current release, or the newest published local-area build for a local-area candidate (see below) | | `GET /api/microcosm/staging/target-change-tree?run=&release=&mode=reported\|shared&...` | One hierarchy level of weighted target-error changes for a staging candidate and an explicit current release | diff --git a/docs/build-monitor.md b/docs/build-monitor.md index 57741c34..7dbc26a5 100644 --- a/docs/build-monitor.md +++ b/docs/build-monitor.md @@ -29,7 +29,9 @@ checks stop builds. The run selected on either tab carries over to the other. - **Historical Hugging Face telemetry.** During migration, the hosted view also reads version 1, 2 and 3 run documents from each country's Hugging Face staging repository. A collector record replaces a historical record with the - same run id. + same run id, except that a collector record that stopped before the run + ended never hides an end the Hugging Face copy recorded after the record's + latest update. The Candidate tab applies the same rule to the runs it lists. - **Local run folders.** The dashboard can still inspect legacy or diagnostic run folders on the same machine. Point it at the directory that holds them: diff --git a/frontend/app/api/microcosm/staging/run/route.test.ts b/frontend/app/api/microcosm/staging/run/route.test.ts new file mode 100644 index 00000000..b104fdde --- /dev/null +++ b/frontend/app/api/microcosm/staging/run/route.test.ts @@ -0,0 +1,133 @@ +import { expect, test } from "bun:test"; + +import { buildTimeline } from "@/lib/microcosm/build-monitor"; +import type { CollectorTimelines } from "@/lib/microcosm/build-runs"; +import { + INVESTIGATION_NOW_MS, + US_FAILED_RUN_ID, + US_RELEASE_CHECK_FAILURE, + usFailedCollectorDocuments, +} from "@/lib/microcosm/test-support/collector-run-fixtures"; +import type { StagingRunDetail } from "@policyengine/calibration-tree/staging-artifact"; +import { createStagingRunHandler } from "./route"; + +// The failed US run's Hugging Face copy: its last upload, mid-calibration. +function huggingFaceDetail(): StagingRunDetail { + return { + available: true, + source_repo: "policyengine/populace-us-staging", + revision: "main", + run_id: US_FAILED_RUN_ID, + candidate_release_id: US_FAILED_RUN_ID, + release_id: null, + country_code: "US", + run_kind: null, + non_release: null, + schema_version: 1, + delivery: null, + progress: { + schema_version: 1, + run_id: US_FAILED_RUN_ID, + status: "running", + stage: "calibrating", + started_at: "2026-10-19T22:25:40+00:00", + updated_at: "2026-10-20T06:09:35+00:00", + calibration: { epoch: 273, epochs: 1500 }, + }, + run_manifest: null, + calibration_progress: null, + events: [], + has_calibration: false, + calibration: null, + reform_validation: null, + build_manifest: null, + release_manifest: null, + }; +} + +function handler(collector: () => Promise) { + const collectorRequests: (readonly string[])[] = []; + const get = createStagingRunHandler({ + loadRun: async () => huggingFaceDetail(), + loadCollectorTimelines: async (_country, runIds) => { + collectorRequests.push(runIds); + return collector(); + }, + now: () => INVESTIGATION_NOW_MS, + }); + return { get, collectorRequests }; +} + +async function loadRun(get: (request: Request) => Promise) { + const response = await get( + new Request( + `http://dashboard.test/api/microcosm/staging/run?country=us&id=${US_FAILED_RUN_ID}`, + ), + ); + expect(response.status).toBe(200); + return (await response.json()) as StagingRunDetail & { + state: Record; + collector_note: string | null; + }; +} + +test("the selected run's state comes from the collector, with the Hugging Face copy kept", async () => { + const { get, collectorRequests } = handler(async () => ({ + timelines: new Map([ + [ + US_FAILED_RUN_ID, + buildTimeline(usFailedCollectorDocuments(), INVESTIGATION_NOW_MS), + ], + ]), + note: null, + })); + + const body = await loadRun(get); + + expect(collectorRequests).toEqual([[US_FAILED_RUN_ID]]); + expect(body.collector_note).toBeNull(); + // The Hugging Face documents are returned unchanged. + expect(body.progress).toMatchObject({ status: "running", stage: "calibrating" }); + expect(body.state).toMatchObject({ + state_source: "collector", + status: "failed", + stage: "release_gates", + ended_at: "2026-10-20T06:37:45.000Z", + failure: { stage: "release_gates", message: US_RELEASE_CHECK_FAILURE }, + hugging_face_status: "running", + hugging_face_stage: "calibrating", + hugging_face_updated_at: "2026-10-20T06:09:35+00:00", + hugging_face_differs: true, + }); +}); + +test("a collector failure returns the Hugging Face state with a note", async () => { + const note = + "The telemetry collector could not be read (Telemetry collector request failed (503).), so run states come from the Hugging Face staging repository."; + const { get } = handler(async () => ({ timelines: new Map(), note })); + + const body = await loadRun(get); + + expect(body.collector_note).toBe(note); + expect(body.state).toMatchObject({ + state_source: "hugging_face", + status: "stalled", + stage: "calibrating", + updated_at: "2026-10-20T06:09:35+00:00", + calibration: { epoch: 273, epochs: 1500 }, + failure: null, + }); +}); + +test("a run the collector does not know keeps its Hugging Face state without a note", async () => { + const { get } = handler(async () => ({ timelines: new Map(), note: null })); + + const body = await loadRun(get); + + expect(body.collector_note).toBeNull(); + expect(body.state).toMatchObject({ + state_source: "hugging_face", + status: "stalled", + hugging_face_differs: false, + }); +}); diff --git a/frontend/app/api/microcosm/staging/run/route.ts b/frontend/app/api/microcosm/staging/run/route.ts index 0000cf99..934d2e27 100644 --- a/frontend/app/api/microcosm/staging/run/route.ts +++ b/frontend/app/api/microcosm/staging/run/route.ts @@ -1,31 +1,121 @@ import { NextResponse } from "next/server"; import { requestedCountry } from "@/lib/api/request-country"; -import { hasCapability } from "@policyengine/calibration-tree/countries"; +import { + type CollectorTimelines, + loadCollectorTimelines, +} from "@/lib/microcosm/build-runs"; +import { + type CalibrationEpoch, + candidateRunState, + type HuggingFaceRunState, +} from "@/lib/microcosm/staging-run-state"; +import { + hasCapability, + type MicrocosmCountry, +} from "@policyengine/calibration-tree/countries"; import { scrub } from "@policyengine/calibration-tree/latest-artifact"; -import { loadStagingRun } from "@policyengine/calibration-tree/staging-artifact"; +import { + loadStagingRun, + type StagingRunDetail, +} from "@policyengine/calibration-tree/staging-artifact"; export const dynamic = "force-dynamic"; export const revalidate = 0; export const runtime = "nodejs"; export const maxDuration = 300; -export async function GET(request: Request) { - const params = new URL(request.url).searchParams; - const country = requestedCountry(params); - if (country instanceof Response) return country; - const runId = params.get("id")?.trim(); - if (!runId && hasCapability(country, "staging")) { - return NextResponse.json({ detail: "Provide a staging run id via ?id=." }, { status: 400 }); - } - try { - return NextResponse.json(scrub(await loadStagingRun(runId ?? "", revalidate, country)), { - headers: { "Cache-Control": "no-store" }, - }); - } catch (error) { - return NextResponse.json( - { detail: error instanceof Error ? error.message : String(error) }, - { status: 502 }, - ); - } +export interface StagingRunRouteDependencies { + loadRun( + runId: string, + revalidate: number, + country: MicrocosmCountry, + ): Promise; + loadCollectorTimelines( + country: MicrocosmCountry, + runIds: readonly string[], + nowMs: number, + ): Promise; + now(): number; } + +function text(value: unknown): string | null { + return typeof value === "string" && value.trim() ? value : null; +} + +function finiteNumber(value: unknown): number | null { + return typeof value === "number" && Number.isFinite(value) ? value : null; +} + +function calibrationEpoch(value: unknown): CalibrationEpoch | null { + if (value == null || typeof value !== "object" || Array.isArray(value)) return null; + const calibration = value as Record; + return { + epoch: finiteNumber(calibration.epoch), + epochs: finiteNumber(calibration.epochs), + }; +} + +// The state the run's Hugging Face copy records: its progress document, or +// its run manifest where the progress document says nothing. +function huggingFaceRunState(detail: StagingRunDetail): HuggingFaceRunState { + const progress = detail.progress ?? {}; + const manifest = detail.run_manifest ?? {}; + return { + status: text(progress.status) ?? text(manifest.status), + stage: text(progress.stage) ?? text(manifest.stage), + started_at: text(progress.started_at) ?? text(manifest.started_at), + updated_at: text(progress.updated_at) ?? text(manifest.updated_at), + calibration: calibrationEpoch(progress.calibration), + }; +} + +// One Candidate tab run: its Hugging Face copy (candidate files, progress and +// events, unchanged) plus `state`, which is the collector's state when the +// collector has a record of the run. +export function createStagingRunHandler( + dependencies: StagingRunRouteDependencies = { + loadRun: loadStagingRun, + loadCollectorTimelines, + now: Date.now, + }, +) { + return async function GET(request: Request) { + const params = new URL(request.url).searchParams; + const country = requestedCountry(params); + if (country instanceof Response) return country; + const runId = params.get("id")?.trim(); + if (!runId && hasCapability(country, "staging")) { + return NextResponse.json({ detail: "Provide a staging run id via ?id=." }, { status: 400 }); + } + try { + const detail = await dependencies.loadRun(runId ?? "", revalidate, country); + if (!detail.available || !runId) { + return NextResponse.json(scrub({ ...detail, state: null, collector_note: null }), { + headers: { "Cache-Control": "no-store" }, + }); + } + const nowMs = dependencies.now(); + const collector = await dependencies.loadCollectorTimelines(country, [runId], nowMs); + return NextResponse.json( + scrub({ + ...detail, + state: candidateRunState( + huggingFaceRunState(detail), + collector.timelines.get(runId), + nowMs, + ), + collector_note: collector.note, + }), + { headers: { "Cache-Control": "no-store" } }, + ); + } catch (error) { + return NextResponse.json( + { detail: error instanceof Error ? error.message : String(error) }, + { status: 502 }, + ); + } + }; +} + +export const GET = createStagingRunHandler(); diff --git a/frontend/app/api/microcosm/staging/runs/route.test.ts b/frontend/app/api/microcosm/staging/runs/route.test.ts new file mode 100644 index 00000000..79483a38 --- /dev/null +++ b/frontend/app/api/microcosm/staging/runs/route.test.ts @@ -0,0 +1,216 @@ +import { expect, test } from "bun:test"; + +import { + type BuildTimeline, + buildTimeline, +} from "@/lib/microcosm/build-monitor"; +import type { CollectorTimelines } from "@/lib/microcosm/build-runs"; +import { + INVESTIGATION_NOW_MS, + UK_BLOCKED_RUN_ID, + UK_COMPLETED_HUGGING_FACE_RUN_ID, + US_ABANDONED_HUGGING_FACE_RUN_ID, + US_FAILED_RUN_ID, + US_FINISHED_HUGGING_FACE_RUN_ID, + US_RUNNING_RUN_ID, + ukBlockedCollectorDocuments, + ukHuggingFaceRuns, + usFailedCollectorDocuments, + usHuggingFaceRuns, + usRunningCollectorDocuments, +} from "@/lib/microcosm/test-support/collector-run-fixtures"; +import type { MicrocosmCountry } from "@policyengine/calibration-tree/countries"; +import type { StagingRunSummary } from "@policyengine/calibration-tree/staging-artifact"; +import { createStagingRunsHandler } from "./route"; + +type RunList = { + available: boolean; + source_repo: string; + revision: string; + truncated: boolean; + runs: StagingRunSummary[]; + incompatible_runs: never[]; +}; + +function huggingFaceList(runs: StagingRunSummary[]): RunList { + return { + available: true, + source_repo: "policyengine/populace-us-staging", + revision: "main", + truncated: false, + runs, + incompatible_runs: [], + }; +} + +function handler( + list: RunList, + collector: (runIds: readonly string[]) => Promise, +) { + const collectorRequests: { country: MicrocosmCountry; runIds: readonly string[] }[] = []; + const get = createStagingRunsHandler({ + loadRuns: async () => list, + loadCollectorTimelines: async (country, runIds) => { + collectorRequests.push({ country, runIds }); + return collector(runIds); + }, + now: () => INVESTIGATION_NOW_MS, + }); + return { get, collectorRequests }; +} + +function timelines(...entries: BuildTimeline[]): Map { + return new Map(entries.map((timeline) => [timeline.run_id, timeline])); +} + +async function listRuns(get: (request: Request) => Promise, country: string) { + const response = await get( + new Request(`http://dashboard.test/api/microcosm/staging/runs?country=${country}`), + ); + expect(response.status).toBe(200); + return (await response.json()) as { + available: boolean; + collector_note: string | null; + runs: (StagingRunSummary & Record)[]; + }; +} + +function byRunId(runs: Record[]) { + return Object.fromEntries(runs.map((run) => [run.run_id, run])); +} + +test("US runs take the collector's state when the collector has their record", async () => { + const { get, collectorRequests } = handler( + huggingFaceList(usHuggingFaceRuns()), + async () => ({ + timelines: timelines( + buildTimeline(usFailedCollectorDocuments(), INVESTIGATION_NOW_MS), + buildTimeline(usRunningCollectorDocuments(), INVESTIGATION_NOW_MS), + ), + note: null, + }), + ); + + const body = await listRuns(get, "us"); + const runs = byRunId(body.runs); + + expect(collectorRequests).toEqual([ + { + country: "us", + runIds: [ + US_RUNNING_RUN_ID, + US_FAILED_RUN_ID, + US_ABANDONED_HUGGING_FACE_RUN_ID, + US_FINISHED_HUGGING_FACE_RUN_ID, + ], + }, + ]); + expect(body.collector_note).toBeNull(); + // Hugging Face's order and run list are kept. + expect(body.runs.map((run) => run.run_id)).toEqual( + usHuggingFaceRuns().map((run) => run.run_id), + ); + expect(runs[US_FAILED_RUN_ID]).toMatchObject({ + state_source: "collector", + status: "failed", + stage: "release_gates", + ended_at: "2026-10-20T06:37:45.000Z", + hugging_face_status: "running", + hugging_face_stage: "calibrating", + hugging_face_updated_at: "2026-10-20T06:09:35+00:00", + }); + expect(runs[US_RUNNING_RUN_ID]).toMatchObject({ + state_source: "collector", + status: "running", + stage: "reform_validation", + heartbeat_at: "2026-10-22T17:16:45.000Z", + hugging_face_stage: "calibrating", + hugging_face_updated_at: "2026-10-21T18:59:03+00:00", + }); + expect(runs[US_FINISHED_HUGGING_FACE_RUN_ID]).toMatchObject({ + state_source: "hugging_face", + status: "passed", + stage: "complete", + }); +}); + +test("runs only the collector knows are not added to the Candidate list", async () => { + const collectorOnly = buildTimeline( + { ...usRunningCollectorDocuments(), run_id: "collector-only-run" }, + INVESTIGATION_NOW_MS, + ); + const { get } = handler(huggingFaceList(usHuggingFaceRuns()), async () => ({ + timelines: timelines(collectorOnly), + note: null, + })); + + const body = await listRuns(get, "us"); + + expect(body.runs.map((run) => run.run_id)).not.toContain("collector-only-run"); + expect(body.runs).toHaveLength(usHuggingFaceRuns().length); +}); + +test("a collector failure returns the Hugging Face states with a note", async () => { + const note = + "The telemetry collector could not be read (Telemetry collector request failed (503).), so run states come from the Hugging Face staging repository."; + const { get } = handler(huggingFaceList(usHuggingFaceRuns()), async () => ({ + timelines: new Map(), + note, + })); + + const body = await listRuns(get, "us"); + const runs = byRunId(body.runs); + + expect(body.available).toBe(true); + expect(body.collector_note).toBe(note); + expect(runs[US_FAILED_RUN_ID]).toMatchObject({ + state_source: "hugging_face", + status: "stalled", + stage: "calibrating", + updated_at: "2026-10-20T06:09:35+00:00", + hugging_face_status: "running", + }); +}); + +test("a UK list overlays the collector's record and keeps Hugging Face-only runs", async () => { + const { get, collectorRequests } = handler( + { ...huggingFaceList(ukHuggingFaceRuns()), source_repo: "policyengine/populace-uk-staging" }, + async () => ({ + timelines: timelines( + buildTimeline(ukBlockedCollectorDocuments(), INVESTIGATION_NOW_MS), + ), + note: null, + }), + ); + + const body = await listRuns(get, "uk"); + const runs = byRunId(body.runs); + + expect(collectorRequests.map((request) => request.country)).toEqual(["uk"]); + expect(runs[UK_BLOCKED_RUN_ID]).toMatchObject({ + state_source: "collector", + status: "blocked", + stage: "gate_battery", + hugging_face_differs: false, + }); + expect(runs[UK_COMPLETED_HUGGING_FACE_RUN_ID]).toMatchObject({ + state_source: "hugging_face", + status: "passed", + hugging_face_status: "completed", + }); +}); + +test("an unreadable staging repository asks the collector nothing", async () => { + const { get, collectorRequests } = handler( + { ...huggingFaceList([]), available: false }, + async () => { + throw new Error("The collector must not be asked."); + }, + ); + + const body = await listRuns(get, "us"); + + expect(body.available).toBe(false); + expect(body.collector_note).toBeNull(); + expect(collectorRequests).toEqual([]); +}); diff --git a/frontend/app/api/microcosm/staging/runs/route.ts b/frontend/app/api/microcosm/staging/runs/route.ts index 1df3a5fa..fdabce64 100644 --- a/frontend/app/api/microcosm/staging/runs/route.ts +++ b/frontend/app/api/microcosm/staging/runs/route.ts @@ -1,23 +1,75 @@ import { NextResponse } from "next/server"; import { requestedCountry } from "@/lib/api/request-country"; +import { + type CollectorTimelines, + loadCollectorTimelines, +} from "@/lib/microcosm/build-runs"; +import { overlayRunSummary } from "@/lib/microcosm/staging-run-state"; +import type { MicrocosmCountry } from "@policyengine/calibration-tree/countries"; import { loadStagingRuns } from "@policyengine/calibration-tree/staging-artifact"; import { scrub } from "@policyengine/calibration-tree/latest-artifact"; export const dynamic = "force-dynamic"; export const revalidate = 0; +export const runtime = "nodejs"; +// The collector lookup can add two 30-second request timeouts. +export const maxDuration = 120; -export async function GET(request: Request) { - const country = requestedCountry(new URL(request.url).searchParams); - if (country instanceof Response) return country; - try { - return NextResponse.json(scrub(await loadStagingRuns(revalidate, country)), { - headers: { "Cache-Control": "no-store" }, - }); - } catch (error) { - return NextResponse.json( - { detail: error instanceof Error ? error.message : String(error) }, - { status: 502 }, - ); - } +export interface StagingRunsRouteDependencies { + loadRuns( + revalidate: number, + country: MicrocosmCountry, + ): ReturnType; + loadCollectorTimelines( + country: MicrocosmCountry, + runIds: readonly string[], + nowMs: number, + ): Promise; + now(): number; } + +// The Candidate tab's run list: the runs in the country's Hugging Face staging +// repository, each with the collector's state when the collector has a record +// of it. Runs only the collector knows have no candidate files there and are +// listed on the Build progress tab instead. +export function createStagingRunsHandler( + dependencies: StagingRunsRouteDependencies = { + loadRuns: loadStagingRuns, + loadCollectorTimelines, + now: Date.now, + }, +) { + return async function GET(request: Request) { + const country = requestedCountry(new URL(request.url).searchParams); + if (country instanceof Response) return country; + try { + const list = await dependencies.loadRuns(revalidate, country); + const nowMs = dependencies.now(); + const collector: CollectorTimelines = list.runs.length + ? await dependencies.loadCollectorTimelines( + country, + list.runs.map((run) => run.run_id), + nowMs, + ) + : { timelines: new Map(), note: null }; + return NextResponse.json( + scrub({ + ...list, + runs: list.runs.map((run) => + overlayRunSummary(run, collector.timelines.get(run.run_id), nowMs), + ), + collector_note: collector.note, + }), + { headers: { "Cache-Control": "no-store" } }, + ); + } catch (error) { + return NextResponse.json( + { detail: error instanceof Error ? error.message : String(error) }, + { status: 502 }, + ); + } + }; +} + +export const GET = createStagingRunsHandler(); diff --git a/frontend/components/microcosm/microcosm-staging-view.test.tsx b/frontend/components/microcosm/microcosm-staging-view.test.tsx new file mode 100644 index 00000000..64896bd7 --- /dev/null +++ b/frontend/components/microcosm/microcosm-staging-view.test.tsx @@ -0,0 +1,119 @@ +import { expect, test } from "bun:test"; +import { createElement } from "react"; +import { renderToStaticMarkup } from "react-dom/server"; + +import { buildTimeline } from "@/lib/microcosm/build-monitor"; +import { + candidateRunState, + type CandidateRunState, +} from "@/lib/microcosm/staging-run-state"; +import { + INVESTIGATION_NOW_MS, + UK_BLOCKED_RUN_ID, + US_FAILED_RUN_ID, + US_RUNNING_RUN_ID, + ukBlockedCollectorDocuments, + usFailedCollectorDocuments, + usRunningCollectorDocuments, +} from "@/lib/microcosm/test-support/collector-run-fixtures"; +import { CandidateOverview } from "./microcosm-staging-view"; + +// The failed US run's Hugging Face progress document: its last upload. +const FAILED_RUN_PROGRESS = { + status: "running", + stage: "calibrating", + started_at: "2026-10-19T22:25:40+00:00", + updated_at: "2026-10-20T06:09:35+00:00", + calibration: { epoch: 273, epochs: 1500 }, +}; + +function renderOverview( + runId: string, + state: CandidateRunState, + progress: Record, + collectorNote: string | null = null, +): string { + return renderToStaticMarkup( + createElement(CandidateOverview, { + candidateReleaseId: runId, + state, + progress, + calibration: null, + collectorNote, + }), + ); +} + +test("the failed US run shows the collector's failure and its out-of-date Hugging Face copy", () => { + const state = candidateRunState( + FAILED_RUN_PROGRESS, + buildTimeline(usFailedCollectorDocuments(), INVESTIGATION_NOW_MS), + INVESTIGATION_NOW_MS, + ); + + const markup = renderOverview(US_FAILED_RUN_ID, state, FAILED_RUN_PROGRESS); + + expect(markup).toContain(">Failed<"); + expect(markup).toContain("Telemetry collector"); + expect(markup).toContain("Failed during Release gates"); + expect(markup).toContain("relative_error=-0.0641635"); + expect(markup).toContain("Hugging Face copy"); + expect(markup).toContain("Running, Calibration, epoch 273 of 1500 · updated"); + expect(markup).not.toContain("Stalled"); +}); + +test("the US run in reform validation shows that stage and its last heartbeat", () => { + const progress = { + status: "running", + stage: "calibrating", + updated_at: "2026-10-21T18:59:03+00:00", + calibration: { epoch: 935, epochs: 1500 }, + }; + const state = candidateRunState( + { ...progress, started_at: null }, + buildTimeline(usRunningCollectorDocuments(), INVESTIGATION_NOW_MS), + INVESTIGATION_NOW_MS, + ); + + const markup = renderOverview(US_RUNNING_RUN_ID, state, progress); + + expect(markup).toContain(">Running<"); + expect(markup).toContain("Reform validation"); + expect(markup).toContain("Last heartbeat"); + expect(markup).toContain("Running, Calibration, epoch 935 of 1500 · updated"); +}); + +test("a collector failure shows the Hugging Face state and the collector note", () => { + const note = + "The telemetry collector could not be read (Telemetry collector request failed (503).), so run states come from the Hugging Face staging repository."; + const state = candidateRunState(FAILED_RUN_PROGRESS, undefined, INVESTIGATION_NOW_MS); + + const markup = renderOverview(US_FAILED_RUN_ID, state, FAILED_RUN_PROGRESS, note); + + expect(markup).toContain(">Stalled<"); + expect(markup).toContain("Hugging Face staging repository"); + expect(markup).toContain("Calibration, epoch 273 of 1500"); + expect(markup).toContain("Telemetry collector request failed (503)."); + expect(markup).not.toContain("Hugging Face copy"); + expect(markup).not.toContain("Last heartbeat"); +}); + +test("a UK run refused by its acceptance checks shows where, without a copy warning", () => { + const progress = { + status: "blocked", + stage: "blocked", + updated_at: "2026-10-18T23:45:45+00:00", + }; + const state = candidateRunState( + { ...progress, started_at: null }, + buildTimeline(ukBlockedCollectorDocuments(), INVESTIGATION_NOW_MS), + INVESTIGATION_NOW_MS, + ); + + const markup = renderOverview(UK_BLOCKED_RUN_ID, state, progress); + + expect(markup).toContain(">Blocked<"); + expect(markup).toContain("Refused at Gate battery"); + expect(markup).toContain("Telemetry collector"); + expect(markup).not.toContain("Hugging Face copy"); +}); diff --git a/frontend/components/microcosm/microcosm-staging-view.tsx b/frontend/components/microcosm/microcosm-staging-view.tsx index 0a27d54d..d8601336 100644 --- a/frontend/components/microcosm/microcosm-staging-view.tsx +++ b/frontend/components/microcosm/microcosm-staging-view.tsx @@ -39,10 +39,13 @@ import { type MicrocosmStagingRunResponse, type MicrocosmStagingRunSummary, } from "@/lib/api/hooks/use-microcosm"; +import type { CandidateRunState } from "@/lib/microcosm/staging-run-state"; import { countryRegistration, hasCapability } from "@policyengine/calibration-tree/countries"; import { + formatCandidateCurrentStatus, formatStagingCurrentStatus, formatStagingStatus, + formatStateSource, } from "@/lib/microcosm/staging-status"; import { formatWeightedTargetError } from "@/lib/microcosm/target-change-visualization"; import { stagingComparisonBaseline } from "@/lib/microcosm/calibration-selection"; @@ -137,17 +140,9 @@ function runLabel(run: MicrocosmStagingRunSummary): string { )}`; } -// Display a running or queued run as stalled after six hours without an update. -const STALL_MS = 6 * 60 * 60 * 1000; - -function effectiveStatus( - status: string | null | undefined, - updatedAt: string | null | undefined, -): string | null { - if (status !== "running" && status !== "queued") return status ?? null; - const t = updatedAt ? new Date(updatedAt).valueOf() : NaN; - if (Number.isFinite(t) && Date.now() - t > STALL_MS) return "stalled"; - return status ?? null; +// Where the run picker's states come from; each option's tooltip names it. +function stateSourceTitle(run: MicrocosmStagingRunSummary): string { + return `State from the ${formatStateSource(run.state_source).toLowerCase()}`; } const STATUS_INDICATOR_CLASS: Record = { @@ -172,10 +167,7 @@ function RunSelect({ const [open, setOpen] = useState(false); const rootRef = useRef(null); const selectedRun = runs.find((run) => run.run_id === selected); - const selectedStatus = effectiveStatus( - selectedRun?.status, - selectedRun?.updated_at, - ); + const selectedStatus = selectedRun?.status ?? null; const selectedTone = statusTone(selectedStatus); const disabled = !runs.length; @@ -221,7 +213,10 @@ function RunSelect({ {selectedRun ? runLabel(selectedRun) : placeholder} {selectedRun ? ( - + {formatStagingStatus(selectedStatus)} ) : null} @@ -249,7 +244,7 @@ function RunSelect({
    {runs.map((run) => { const active = run.run_id === selected; - const shownStatus = effectiveStatus(run.status, run.updated_at); + const shownStatus = run.status; const tone = statusTone(shownStatus); return (
  • @@ -270,7 +265,10 @@ function RunSelect({ className={`h-2 w-2 shrink-0 rounded-full ${STATUS_INDICATOR_CLASS[tone]}`} /> {runLabel(run)} - + {formatStagingStatus(shownStatus)} @@ -278,6 +276,10 @@ function RunSelect({ ); })}
+

+ States come from the telemetry collector when it has a record of the run, otherwise + from the Hugging Face staging repository. +

) : null} @@ -293,6 +295,111 @@ function agoLabel(value: string | null | undefined): string { return `${Math.round(mins / 60 / 24)}d ago`; } +function timeWithAgo(value: string | null | undefined): string { + const ago = agoLabel(value); + return `${timeLabel(value)}${ago ? ` · ${ago}` : ""}`; +} + +// What the run's Hugging Face copy last recorded: "Running, Calibration, +// epoch 273 of 1500 · updated Oct 20, 06:09 · 2d ago". +function huggingFaceCopyLabel( + state: CandidateRunState, + progress: Record, +): string { + const recorded = [formatStagingStatus(state.hugging_face_status)]; + if (state.hugging_face_stage) { + recorded.push(formatStagingCurrentStatus({ ...progress, stage: state.hugging_face_stage })); + } + return `${recorded.join(", ")} · updated ${timeWithAgo(state.hugging_face_updated_at)}`; +} + +function OverviewRow({ + label, + value, + wrap = false, +}: { + label: string; + value: string; + // Long text, such as a failure message, wraps to three lines instead of one. + wrap?: boolean; +}) { + return ( + <> +
+ {label} +
+
+ {value} +
+ + ); +} + +// The selected candidate's identity and state. The state is the telemetry +// collector's when it has a record of the run, else the Hugging Face copy's; +// when the copy has fallen behind the collector, the card also says what the +// copy still records. +export function CandidateOverview({ + candidateReleaseId, + state, + progress, + calibration, + collectorNote, +}: { + candidateReleaseId: string; + state: CandidateRunState | null; + progress: Record; + calibration: MicrocosmStagingRunResponse["calibration"]; + collectorNote: string | null; +}) { + const currentStatus = state + ? formatCandidateCurrentStatus(state) + : formatStagingCurrentStatus(progress); + const huggingFaceCopy = state?.hugging_face_differs + ? huggingFaceCopyLabel(state, progress) + : null; + return ( + +
+ + + {state && } + + {state?.heartbeat_at && ( + + )} + + {state?.failure?.message && ( + + )} + {huggingFaceCopy && } + {calibration && ( + <> +
+ Dataset +
+
+ +
+ + )} +
+ {collectorNote && ( +

+ {collectorNote} +

+ )} + + ); +} + // Human rendering for the small details dicts events carry (n_targets, // learning_rate, gate results, repair factors…). function detailChips(details: unknown): [string, string][] { @@ -354,6 +461,7 @@ function RunInternalsPanel({ lossValues, status, stage, + huggingFaceCopyNote, buildManifest, artifacts, open, @@ -364,6 +472,9 @@ function RunInternalsPanel({ lossValues: number[]; status: string | null; stage: string | null; + // Set when the run's state comes from the collector and its Hugging Face + // copy, which every section here reads, has fallen behind. + huggingFaceCopyNote: string | null; buildManifest: Record | null; artifacts: Record; open: boolean; @@ -383,6 +494,11 @@ function RunInternalsPanel({ Optimizer progress, stage timeline with logged numbers, build manifest (versions, hashes, validation results), and uploaded artifacts. + {huggingFaceCopyNote ? ( +
+ {huggingFaceCopyNote} +
+ ) : null} ▾ @@ -519,9 +635,9 @@ function RunInternalsPanel({ {index === all.length - 1 && event.stage !== "complete" && !failed && !blocked - ? status === "stalled" - ? "⚠ last event" - : "…" + ? status === "running" || status == null + ? "…" + : "⚠ last event" : durationLabel(duration)} @@ -883,13 +999,18 @@ function MicrocosmStagingRunsView({ [calibrationEvents], ); const progress = runData?.progress ?? {}; - const rawStatus = typeof progress.status === "string" ? progress.status : null; - const stage = typeof progress.stage === "string" ? progress.stage : null; - const updatedAt = typeof progress.updated_at === "string" ? progress.updated_at : null; - const status = effectiveStatus(rawStatus, updatedAt); - const statusLabel = formatStagingStatus(status); - const lastUpdate = `${timeLabel(updatedAt)}${agoLabel(updatedAt) ? ` · ${agoLabel(updatedAt)}` : ""}`; - const currentStatus = formatStagingCurrentStatus(progress); + // The collector's state when it has a record of the run, else the Hugging + // Face copy's, derived on the server with the Build progress tab's rule. + const candidateState = runData?.state ?? null; + const status = candidateState?.status ?? null; + const stage = candidateState?.stage ?? null; + // Run internals read the Hugging Face copy; say so when it has fallen behind. + const internalsCopyNote = candidateState?.hugging_face_differs + ? `These sections read the run's Hugging Face copy (${huggingFaceCopyLabel(candidateState, progress)}); the status above comes from the telemetry collector.` + : null; + // The selected run's own note: the collector may have answered for it even + // when the run list's request failed, or the other way round. + const collectorNote = runData?.collector_note ?? null; const candidateReleaseId = runData?.candidate_release_id ?? selectedRun; const buildManifest = (runData?.build_manifest ?? null) as Record | null; const artifacts = ((runData?.run_manifest as Record | null)?.artifacts ?? @@ -963,70 +1084,13 @@ function MicrocosmStagingRunsView({ )}
- -
-
- Candidate -
-
- {candidateReleaseId} -
-
- Status -
-
- {statusLabel} -
-
- Last update -
-
- {lastUpdate} -
-
- Current status -
-
- {currentStatus} -
- {runData.calibration && ( - <> -
- Dataset -
-
- -
- - )} -
- + {candidateValidationPending && ( | null; release_manifest?: Record | null; + // The run's state: the collector's when it has a record of the run, else + // the Hugging Face copy's. Null when staging is unavailable. + state?: CandidateRunState | null; + collector_note?: string | null; } export interface MicrocosmVariableValue { diff --git a/frontend/lib/microcosm/build-monitor.ts b/frontend/lib/microcosm/build-monitor.ts index b864ff16..73e15628 100644 --- a/frontend/lib/microcosm/build-monitor.ts +++ b/frontend/lib/microcosm/build-monitor.ts @@ -281,7 +281,10 @@ function pipelineIdentity( }; } -function runState( +// A run's state from its recorded status and when it last reported. The +// Staging page's Candidate tab uses the same rule for runs the collector does +// not know. +export function runState( rawStatus: string | null, updatedMs: number | null, nowMs: number, @@ -296,6 +299,33 @@ function runState( return "running"; } +const FINAL_RUN_STATES: ReadonlySet = new Set(["passed", "blocked", "failed"]); + +export interface RecordedRunState { + state: BuildRunState | null; + // The record's latest update. + updated_ms: number | null; +} + +// Whether a run's collector record must give way to its Hugging Face copy: +// the record stopped before the run ended (it says running, or stalled once +// its heartbeats stopped), and the copy recorded the end (passed, blocked or +// failed) after the record's latest update. Otherwise the collector's record +// stands, final or newer. Both Staging tabs choose between the two sources +// with this rule. +export function collectorMissedTheEnd( + collector: RecordedRunState, + huggingFace: RecordedRunState, +): boolean { + return ( + (collector.state == null || !FINAL_RUN_STATES.has(collector.state)) && + huggingFace.state != null && + FINAL_RUN_STATES.has(huggingFace.state) && + huggingFace.updated_ms != null && + (collector.updated_ms == null || huggingFace.updated_ms > collector.updated_ms) + ); +} + function resourceSnapshot(value: unknown, at: number | null): ResourceSnapshot | null { const resources = obj(value); if (!resources || at == null) return null; diff --git a/frontend/lib/microcosm/build-runs.test.ts b/frontend/lib/microcosm/build-runs.test.ts index ed575fa3..1f505054 100644 --- a/frontend/lib/microcosm/build-runs.test.ts +++ b/frontend/lib/microcosm/build-runs.test.ts @@ -7,6 +7,7 @@ import { import { type HostedRunLoaders, loadBuildRuns, + loadCollectorTimelines, refreshHostedRun, } from "@/lib/microcosm/build-runs"; import { @@ -163,6 +164,34 @@ describe("refreshHostedRun", () => { expect(historyCalls).toBe(0); }); + test("keeps the list's later Hugging Face end over a collector record that stopped before it", async () => { + // The list holds the Hugging Face copy, which recorded the run completing + // at 11:00; the collector's record stops at 10:30, still running. + const running = documents("run-1", "running"); + const stoppedEarly: BuildRunDocuments = { + ...running, + progress: { ...running.progress, updated_at: "2026-10-05T10:30:00Z" }, + events: running.events.map((event) => ({ + ...event, + timestamp: "2026-10-05T10:30:00Z", + time: "2026-10-05T10:30:00Z", + })), + }; + const current = buildTimeline(documents("run-1", "completed"), NOW); + + const run = await refreshHostedRun( + current, + "us", + "run-1", + NOW, + loaders({ loadCollector: async () => stoppedEarly }), + ); + + expect(buildTimeline(stoppedEarly, NOW).state).toBe("running"); + expect(run?.state).toBe("passed"); + expect(run).toBe(current); + }); + test("surfaces collector failure when no staged history exists", async () => { await expect( refreshHostedRun( @@ -227,3 +256,130 @@ describe("loadBuildRuns from the collector", () => { } }); }); + +const READ_SECRET_VERSION = + "projects/test-project/secrets/collector-read/versions/latest"; + +// Runs `body` with the collector configured to answer through `collectorFetch`, +// or with the collector not configured when it is null. +async function withCollector( + collectorFetch: typeof fetch | null, + body: () => Promise, +): Promise { + const originalFetch = globalThis.fetch; + const originalEnvironment = { + MICROCOSM_TELEMETRY_COLLECTOR_URL: + process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL, + MICROCOSM_TELEMETRY_READ_SECRET_VERSION: + process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION, + }; + const readSecret = spyOn(secretManager, "readVersion").mockResolvedValue( + "secret-read-token", + ); + if (collectorFetch) { + process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL = FAKE_COLLECTOR_URL; + process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION = READ_SECRET_VERSION; + globalThis.fetch = collectorFetch; + } else { + delete process.env.MICROCOSM_TELEMETRY_COLLECTOR_URL; + delete process.env.MICROCOSM_TELEMETRY_READ_SECRET_VERSION; + globalThis.fetch = (async () => { + throw new Error("An unconfigured collector must not be requested."); + }) as unknown as typeof fetch; + } + try { + return await body(); + } finally { + globalThis.fetch = originalFetch; + readSecret.mockRestore(); + for (const [name, value] of Object.entries(originalEnvironment)) { + if (value == null) delete process.env[name]; + else process.env[name] = value; + } + } +} + +describe("loadCollectorTimelines for the Candidate tab", () => { + test("reads the collector's records only for the requested runs it knows", async () => { + // The am fixture country has a staging repository, like a country whose + // Candidate tab lists Hugging Face runs. + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 80 }); + const newest = collector.runs[0].run_id; + const finished = collector.runs[5].run_id; + const olderThanTheListedRuns = collector.runs[70].run_id; + + const loaded = await withCollector(collector.fetch, () => + loadCollectorTimelines( + "am", + [newest, finished, olderThanTheListedRuns, "hugging-face-only-run"], + NOW, + ), + ); + + expect(loaded.note).toBeNull(); + expect(collector.listRequests).toHaveLength(1); + expect(collector.listRequests[0].searchParams.get("country")).toBe("AM"); + expect(collector.listRequests[0].searchParams.get("limit")).toBe("60"); + expect(collector.documentRequests.toSorted()).toEqual( + [newest, finished].toSorted(), + ); + expect([...loaded.timelines.keys()].toSorted()).toEqual( + [newest, finished].toSorted(), + ); + expect(loaded.timelines.get(newest)?.state).toBe("running"); + expect(loaded.timelines.get(finished)?.state).toBe("passed"); + }); + + test("asks the collector nothing when no run is requested", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 3 }); + + const loaded = await withCollector(collector.fetch, () => + loadCollectorTimelines("am", [], NOW), + ); + + expect(loaded).toEqual({ timelines: new Map(), note: null }); + expect(collector.listRequests).toHaveLength(0); + }); + + test("an unconfigured collector leaves every run on Hugging Face, with a note", async () => { + const loaded = await withCollector(null, () => + loadCollectorTimelines("am", ["run-0001"], NOW), + ); + + expect(loaded.timelines.size).toBe(0); + expect(loaded.note).toContain("not configured"); + }); + + test("a failed collector request leaves every run on Hugging Face, with a note", async () => { + const unavailable = (async () => + Response.json({ detail: "Unavailable." }, { status: 503 })) as unknown as typeof fetch; + + const loaded = await withCollector(unavailable, () => + loadCollectorTimelines("am", ["run-0001"], NOW), + ); + + expect(loaded.timelines.size).toBe(0); + expect(loaded.note).toContain("could not be read"); + expect(loaded.note).toContain("503"); + }); + + test("a run whose collector record fails keeps its Hugging Face state, with a note", async () => { + const collector = fakeTelemetryCollector({ countryCode: "AM", runCount: 4 }); + const unreadable = collector.runs[1].run_id; + const readable = collector.runs[2].run_id; + const failingOneRecord = (async (input: RequestInfo | URL) => { + if (String(input).endsWith(`/v1/runs/${unreadable}`)) { + return Response.json({ detail: "Internal error." }, { status: 500 }); + } + return collector.fetch(input); + }) as typeof fetch; + + const loaded = await withCollector(failingOneRecord, () => + loadCollectorTimelines("am", [unreadable, readable], NOW), + ); + + expect([...loaded.timelines.keys()]).toEqual([readable]); + expect(loaded.note).toContain(unreadable); + expect(loaded.note).toContain("500"); + }); +}); diff --git a/frontend/lib/microcosm/build-runs.ts b/frontend/lib/microcosm/build-runs.ts index 5cc493c2..9a62e2de 100644 --- a/frontend/lib/microcosm/build-runs.ts +++ b/frontend/lib/microcosm/build-runs.ts @@ -7,6 +7,7 @@ import { type PhaseTotals, type StageStat, buildTimeline, + collectorMissedTheEnd, compactTimeline, type StopStat, stopStatistics, @@ -35,6 +36,7 @@ import { loadStagingRunTelemetry, } from "@policyengine/calibration-tree/staging-artifact"; import { + type CollectorRunSummary, collectorConfigured, loadCollectorRun, loadCollectorRuns, @@ -226,31 +228,38 @@ async function loadHuggingFaceStaging( return { available: true, detail: null, documents, problems }; } +// One collector run's documents. A finished run's documents are reused until +// the collector reports a newer update for it. +async function loadCollectorDocuments( + summary: CollectorRunSummary, + country: MicrocosmCountry, +): Promise { + const key = `collector:${country}:${summary.run_id}`; + const cached = stagingRunCache.get(key); + if ( + cached?.final && + FINAL_STATUSES.has(summary.status) && + cached.collector_version === summary.updated_at + ) { + return cached.documents; + } + const loaded = await loadCollectorRun(summary.run_id, country); + stagingRunCache.set(key, { + documents: loaded, + final: isFinal(loaded), + collector_version: summary.updated_at, + }); + return loaded; +} + async function loadCollector(country: MicrocosmCountry): Promise { const summaries = await loadCollectorRuns(country, STAGING_RUN_LIMIT); const documents: BuildRunDocuments[] = []; const problems: BuildRunProblem[] = []; await Promise.all( summaries.map(async (summary) => { - const key = `collector:${country}:${summary.run_id}`; - const cached = stagingRunCache.get(key); - const summaryFinal = FINAL_STATUSES.has(summary.status); - if ( - cached?.final && - summaryFinal && - cached.collector_version === summary.updated_at - ) { - documents.push(cached.documents); - return; - } try { - const loaded = await loadCollectorRun(summary.run_id, country); - stagingRunCache.set(key, { - documents: loaded, - final: isFinal(loaded), - collector_version: summary.updated_at, - }); - documents.push(loaded); + documents.push(await loadCollectorDocuments(summary, country)); } catch (error) { problems.push({ run_id: summary.run_id, detail: message(error) }); } @@ -259,12 +268,109 @@ async function loadCollector(country: MicrocosmCountry): Promise { return { available: true, detail: null, documents, problems }; } +export interface CollectorTimelines { + // The build monitor's timeline of each requested run the collector has a + // record of, by run id. + timelines: Map; + // Why some or all requested runs could not use the collector's record; + // null when the collector answered for every run it knows. + note: string | null; +} + +const HUGGING_FACE_FALLBACK = + "run states come from the Hugging Face staging repository"; + +// The collector's state for runs the Candidate tab lists from Hugging Face. +// It reads the same records the Build progress tab does (the country's 60 +// most recent collector runs) and derives each state with the same +// `buildTimeline`, so the two tabs agree on every run the collector knows. +export async function loadCollectorTimelines( + country: MicrocosmCountry, + runIds: readonly string[], + nowMs: number = Date.now(), +): Promise { + const timelines = new Map(); + if (!runIds.length) return { timelines, note: null }; + let summaries: CollectorRunSummary[]; + try { + // An invalid collector setting throws here; it must not hide the + // Hugging Face runs. + if (!collectorConfigured()) { + return { + timelines, + note: `The telemetry collector is not configured on this server, so ${HUGGING_FACE_FALLBACK}.`, + }; + } + summaries = await loadCollectorRuns(country, STAGING_RUN_LIMIT); + } catch (error) { + return { + timelines, + note: `The telemetry collector could not be read (${message(error)}), so ${HUGGING_FACE_FALLBACK}.`, + }; + } + const requested = new Set(runIds); + const problems: BuildRunProblem[] = []; + await Promise.all( + summaries + .filter((summary) => requested.has(summary.run_id)) + .map(async (summary) => { + try { + const documents = await loadCollectorDocuments(summary, country); + timelines.set(summary.run_id, buildTimeline(documents, nowMs)); + } catch (error) { + problems.push({ run_id: summary.run_id, detail: message(error) }); + } + }), + ); + if (!problems.length) return { timelines, note: null }; + const unreadable = problems + .map((problem) => `${problem.run_id} (${problem.detail})`) + .join(", "); + return { + timelines, + note: `The telemetry collector's record could not be read for ${unreadable}, so ${ + problems.length === 1 ? "that run's state comes" : "those runs' states come" + } from the Hugging Face staging repository.`, + }; +} + +// The hosted documents the build monitor shows, one per run id. During the +// migration period a collector record replaces the run's historical Hugging +// Face copy, unless the record stopped before an end the copy recorded later +// (`collectorMissedTheEnd`, the rule the Candidate tab uses too). +export function mergeHostedDocuments( + collector: readonly BuildRunDocuments[], + huggingFace: readonly BuildRunDocuments[], + nowMs: number, +): BuildRunDocuments[] { + const byId = new Map(); + for (const document of huggingFace) byId.set(document.run_id, document); + for (const document of collector) { + const copy = byId.get(document.run_id); + if ( + copy && + collectorMissedTheEnd( + buildTimeline(document, nowMs), + buildTimeline(copy, nowMs), + ) + ) { + continue; + } + byId.set(document.run_id, document); + } + return [...byId.values()]; +} + async function loadStaging(country: MicrocosmCountry): Promise { const hosted: LoadedRuns[] = []; const sourceProblems: BuildRunProblem[] = []; + let collectorRunDocuments: BuildRunDocuments[] = []; + let huggingFaceRunDocuments: BuildRunDocuments[] = []; if (collectorConfigured()) { try { - hosted.push(await loadCollector(country)); + const loaded = await loadCollector(country); + hosted.push(loaded); + collectorRunDocuments = loaded.documents; } catch (error) { sourceProblems.push({ run_id: "collector", @@ -273,7 +379,9 @@ async function loadStaging(country: MicrocosmCountry): Promise { } } try { - hosted.push(await loadHuggingFaceStaging(country)); + const loaded = await loadHuggingFaceStaging(country); + hosted.push(loaded); + huggingFaceRunDocuments = loaded.documents; } catch (error) { sourceProblems.push({ run_id: "hugging-face-history", @@ -281,13 +389,6 @@ async function loadStaging(country: MicrocosmCountry): Promise { }); } const available = hosted.some((source) => source.available); - const byId = new Map(); - // Historical Hugging Face documents enter first; collector documents are - // newer and replace the same run id during the migration period. - for (const source of [...hosted].reverse()) { - for (const document of source.documents) - byId.set(document.run_id, document); - } return { available, detail: available @@ -296,7 +397,11 @@ async function loadStaging(country: MicrocosmCountry): Promise { .map((source) => source.detail) .filter(Boolean) .join(" ") || "Hosted build telemetry is unavailable.", - documents: [...byId.values()], + documents: mergeHostedDocuments( + collectorRunDocuments, + huggingFaceRunDocuments, + Date.now(), + ), problems: [ ...hosted.flatMap((source) => source.problems), ...sourceProblems, @@ -339,11 +444,20 @@ export async function refreshHostedRun( ): Promise { let collectorError: unknown = null; if (loaders.collectorConfigured()) { + let collectorRun: BuildTimeline | null = null; try { - return buildTimeline(await loaders.loadCollector(runId, country), nowMs); + collectorRun = buildTimeline(await loaders.loadCollector(runId, country), nowMs); } catch (error) { collectorError = error; } + if (collectorRun) { + // The list kept the Hugging Face copy when the collector's record + // stopped before an end the copy recorded later; a fresh read of that + // record must not hide the end either. + return current && collectorMissedTheEnd(collectorRun, current) + ? current + : collectorRun; + } } if (!loaders.hasStagingHistory(country)) { if (current == null && collectorError != null) throw collectorError; @@ -386,7 +500,8 @@ export async function loadBuildRun( const all = timelines(loaded.documents, nowMs); let run = all.find((timeline) => timeline.run_id === runId) ?? null; // A running hosted run may be newer than the cached list. The live - // collector is authoritative when it responds; staged files are fallback. + // collector's record wins when it responds, unless it stopped before an end + // the Hugging Face copy recorded later; staged files are fallback. if (source === "staging") run = await refreshHostedRun(run, country, runId, nowMs); if (run == null) return null; diff --git a/frontend/lib/microcosm/staging-run-state.test.ts b/frontend/lib/microcosm/staging-run-state.test.ts new file mode 100644 index 00000000..a34d0078 --- /dev/null +++ b/frontend/lib/microcosm/staging-run-state.test.ts @@ -0,0 +1,308 @@ +import { describe, expect, test } from "bun:test"; + +import { buildTimeline } from "@/lib/microcosm/build-monitor"; +import { + candidateRunState, + overlayRunSummary, +} from "@/lib/microcosm/staging-run-state"; +import { + INVESTIGATION_NOW_MS, + UK_BLOCKED_RUN_ID, + UK_COMPLETED_HUGGING_FACE_RUN_ID, + US_ABANDONED_HUGGING_FACE_RUN_ID, + US_FAILED_RUN_ID, + US_FINISHED_HUGGING_FACE_RUN_ID, + US_RELEASE_CHECK_FAILURE, + US_RUNNING_RUN_ID, + ukBlockedCollectorDocuments, + ukHuggingFaceRuns, + usFailedCollectorDocuments, + usHuggingFaceRuns, + usRunningCollectorDocuments, +} from "@/lib/microcosm/test-support/collector-run-fixtures"; + +function usSummary(runId: string) { + return usHuggingFaceRuns().find((run) => run.run_id === runId)!; +} + +function ukSummary(runId: string) { + return ukHuggingFaceRuns().find((run) => run.run_id === runId)!; +} + +describe("a run the collector has a record of", () => { + test("the US run that failed its release acceptance checks shows as failed", () => { + const collectorTimeline = buildTimeline( + usFailedCollectorDocuments(), + INVESTIGATION_NOW_MS, + ); + + const overlaid = overlayRunSummary( + usSummary(US_FAILED_RUN_ID), + collectorTimeline, + INVESTIGATION_NOW_MS, + ); + + expect(overlaid).toMatchObject({ + run_id: US_FAILED_RUN_ID, + candidate_release_id: US_FAILED_RUN_ID, + state_source: "collector", + status: "failed", + stage: "release_gates", + started_at: "2026-10-19T22:23:36.000Z", + updated_at: "2026-10-20T06:37:45.000Z", + ended_at: "2026-10-20T06:37:45.000Z", + heartbeat_at: "2026-10-20T06:37:43.000Z", + failure: { stage: "release_gates", message: US_RELEASE_CHECK_FAILURE }, + hugging_face_status: "running", + hugging_face_stage: "calibrating", + hugging_face_updated_at: "2026-10-20T06:09:35+00:00", + hugging_face_differs: true, + }); + }); + + test("the US run in reform validation shows as running in that stage", () => { + const collectorTimeline = buildTimeline( + usRunningCollectorDocuments(), + INVESTIGATION_NOW_MS, + ); + + const overlaid = overlayRunSummary( + usSummary(US_RUNNING_RUN_ID), + collectorTimeline, + INVESTIGATION_NOW_MS, + ); + + expect(overlaid).toMatchObject({ + state_source: "collector", + status: "running", + stage: "reform_validation", + started_at: "2026-10-21T16:02:20.000Z", + updated_at: "2026-10-22T17:16:47.000Z", + ended_at: null, + heartbeat_at: "2026-10-22T17:16:45.000Z", + failure: null, + hugging_face_status: "running", + hugging_face_stage: "calibrating", + hugging_face_updated_at: "2026-10-21T18:59:03+00:00", + hugging_face_differs: true, + }); + }); + + test("the collector's 5-minute heartbeat rule marks a silent running run stalled", () => { + const sixMinutesAfterHeartbeat = Date.parse("2026-10-22T17:22:45Z"); + const collectorTimeline = buildTimeline( + usRunningCollectorDocuments(), + sixMinutesAfterHeartbeat, + ); + + const overlaid = overlayRunSummary( + usSummary(US_RUNNING_RUN_ID), + collectorTimeline, + sixMinutesAfterHeartbeat, + ); + + expect(overlaid.status).toBe("stalled"); + expect(overlaid.state_source).toBe("collector"); + }); + + test("a UK run whose Hugging Face copy is current keeps the same state", () => { + const collectorTimeline = buildTimeline( + ukBlockedCollectorDocuments(), + INVESTIGATION_NOW_MS, + ); + + const overlaid = overlayRunSummary( + ukSummary(UK_BLOCKED_RUN_ID), + collectorTimeline, + INVESTIGATION_NOW_MS, + ); + + expect(overlaid).toMatchObject({ + state_source: "collector", + status: "blocked", + stage: "gate_battery", + ended_at: "2026-10-18T23:45:50.000Z", + hugging_face_status: "blocked", + hugging_face_updated_at: "2026-10-18T23:45:45+00:00", + hugging_face_differs: false, + }); + expect(overlaid.failure?.stage).toBe("gate_battery"); + expect(overlaid.failure?.message).toContain("3 blocking failures"); + }); +}); + +describe("a collector record older than the Hugging Face copy", () => { + test("a record that never saw the run end does not override a later final Hugging Face state", () => { + // The collector last heard from the run at 17:16:47 and then nothing more, + // so by 20:10 its record says stalled; the Hugging Face copy recorded the + // run passing at 20:05. + const laterNowMs = Date.parse("2026-10-22T20:10:00Z"); + const collectorTimeline = buildTimeline(usRunningCollectorDocuments(), laterNowMs); + const finishedOnHuggingFace = { + ...usSummary(US_RUNNING_RUN_ID), + status: "passed", + stage: "complete", + updated_at: "2026-10-22T20:05:00+00:00", + }; + + const overlaid = overlayRunSummary(finishedOnHuggingFace, collectorTimeline, laterNowMs); + + expect(collectorTimeline.state).toBe("stalled"); + expect(overlaid).toMatchObject({ + state_source: "hugging_face", + status: "passed", + stage: "complete", + updated_at: "2026-10-22T20:05:00+00:00", + ended_at: null, + failure: null, + hugging_face_status: "passed", + hugging_face_differs: false, + }); + }); + + test("a final collector state is kept when the Hugging Face copy recorded the end later", () => { + // Both sources saw the run end; the collector's refusal, inferred from its + // acceptance check events, stays even though the copy was written after it. + const collectorTimeline = buildTimeline( + ukBlockedCollectorDocuments(), + INVESTIGATION_NOW_MS, + ); + const completedLaterOnHuggingFace = { + ...ukSummary(UK_BLOCKED_RUN_ID), + status: "completed", + stage: "complete", + updated_at: "2026-10-18T23:46:30+00:00", + }; + + const overlaid = overlayRunSummary( + completedLaterOnHuggingFace, + collectorTimeline, + INVESTIGATION_NOW_MS, + ); + + expect(overlaid).toMatchObject({ + state_source: "collector", + status: "blocked", + hugging_face_status: "completed", + hugging_face_differs: true, + }); + }); + + test("a Hugging Face final state older than the collector's record does not override it", () => { + // The copy recorded a failure before the collector's last heartbeat: the + // collector's later record wins, as it does for every other run it knows. + const collectorTimeline = buildTimeline( + usRunningCollectorDocuments(), + INVESTIGATION_NOW_MS, + ); + const failedEarlierOnHuggingFace = { + ...usSummary(US_RUNNING_RUN_ID), + status: "failed", + stage: "failed", + updated_at: "2026-10-22T17:00:00+00:00", + }; + + const overlaid = overlayRunSummary( + failedEarlierOnHuggingFace, + collectorTimeline, + INVESTIGATION_NOW_MS, + ); + + expect(overlaid).toMatchObject({ + state_source: "collector", + status: "running", + stage: "reform_validation", + }); + }); +}); + +describe("a run the collector has no record of", () => { + test("keeps its Hugging Face state, with the shared 6-hour rule for silent runs", () => { + const abandoned = overlayRunSummary( + usSummary(US_ABANDONED_HUGGING_FACE_RUN_ID), + undefined, + INVESTIGATION_NOW_MS, + ); + const finished = overlayRunSummary( + usSummary(US_FINISHED_HUGGING_FACE_RUN_ID), + undefined, + INVESTIGATION_NOW_MS, + ); + + expect(abandoned).toMatchObject({ + state_source: "hugging_face", + status: "stalled", + stage: "base_population_repair", + updated_at: "2026-07-02T13:48:44+00:00", + ended_at: null, + heartbeat_at: null, + failure: null, + hugging_face_status: "running", + hugging_face_stage: "base_population_repair", + hugging_face_updated_at: "2026-07-02T13:48:44+00:00", + hugging_face_differs: false, + }); + expect(finished).toMatchObject({ + state_source: "hugging_face", + status: "passed", + stage: "complete", + }); + }); + + test("a running Hugging Face run updated within 6 hours is still running", () => { + const threeHoursLater = Date.parse("2026-10-20T09:09:35Z"); + + const overlaid = overlayRunSummary( + usSummary(US_FAILED_RUN_ID), + undefined, + threeHoursLater, + ); + + expect(overlaid).toMatchObject({ + state_source: "hugging_face", + status: "running", + stage: "calibrating", + }); + }); + + test("a UK local-area candidate that closed as completed shows as finished", () => { + const overlaid = overlayRunSummary( + ukSummary(UK_COMPLETED_HUGGING_FACE_RUN_ID), + undefined, + INVESTIGATION_NOW_MS, + ); + + expect(overlaid).toMatchObject({ + state_source: "hugging_face", + status: "passed", + hugging_face_status: "completed", + }); + }); + + test("a run with no recorded status stays unknown", () => { + const state = candidateRunState( + { status: null, stage: null, started_at: null, updated_at: null }, + undefined, + INVESTIGATION_NOW_MS, + ); + + expect(state.status).toBeNull(); + expect(state.state_source).toBe("hugging_face"); + }); +}); + +test("the Hugging Face calibration epoch is kept for a run the collector does not know", () => { + const state = candidateRunState( + { + status: "running", + stage: "calibrating", + started_at: "2026-10-19T22:25:40+00:00", + updated_at: "2026-10-20T06:09:35+00:00", + calibration: { epoch: 273, epochs: 1500 }, + }, + undefined, + Date.parse("2026-10-20T06:10:00Z"), + ); + + expect(state.calibration).toEqual({ epoch: 273, epochs: 1500 }); +}); diff --git a/frontend/lib/microcosm/staging-run-state.ts b/frontend/lib/microcosm/staging-run-state.ts new file mode 100644 index 00000000..ffd43e6e --- /dev/null +++ b/frontend/lib/microcosm/staging-run-state.ts @@ -0,0 +1,175 @@ +// The run state the Staging page's Candidate tab shows. The tab lists runs +// from the Hugging Face staging repository, but a build's Hugging Face copy +// can stop changing partway through (a rate-limited writer switches its +// uploads off), while the telemetry collector keeps receiving events. So when +// the collector has a record of a run, its state replaces the Hugging Face +// one, derived exactly as the Build progress tab derives it; the Hugging Face +// values stay alongside for comparison. The one exception: a collector record +// that stopped before the run ended never hides an end the Hugging Face copy +// recorded after the record's latest update. +import { + type BuildRunState, + type BuildTimeline, + collectorMissedTheEnd, + runState, +} from "@/lib/microcosm/build-monitor"; +import type { StagingRunSummary } from "@policyengine/calibration-tree/staging-artifact"; + +export type RunStateSource = "collector" | "hugging_face"; + +export interface CalibrationEpoch { + epoch: number | null; + epochs: number | null; +} + +// What the run's Hugging Face copy records. +export interface HuggingFaceRunState { + status: string | null; + stage: string | null; + started_at: string | null; + updated_at: string | null; + calibration?: CalibrationEpoch | null; +} + +export interface CandidateRunFailure { + // The stage that failed, or the acceptance check stage that refused the + // candidate. + stage: string | null; + message: string | null; +} + +export interface CandidateRunState { + // "collector" when the collector has a record of the run, unless that + // record stopped before the run ended and the Hugging Face copy recorded the + // end later. + state_source: RunStateSource; + // Null only when the Hugging Face copy records no status. + status: BuildRunState | null; + // The current stage; for a failed or refused run, the stage it stopped at. + stage: string | null; + started_at: string | null; + // The run's latest event, heartbeat or upload. + updated_at: string | null; + ended_at: string | null; + heartbeat_at: string | null; + failure: CandidateRunFailure | null; + calibration: CalibrationEpoch | null; + hugging_face_status: string | null; + hugging_face_stage: string | null; + hugging_face_updated_at: string | null; + // The Hugging Face copy shows another state, or is well behind the + // collector's record. Always false when the state comes from Hugging Face. + hugging_face_differs: boolean; +} + +export type CandidateRunSummary = StagingRunSummary & CandidateRunState; + +// How far the Hugging Face copy may trail the collector's record before the +// page says so. Staging writers upload the copy only when the build records +// something, at most every 5 minutes for the UK writer, while the collector +// gets a heartbeat every minute. So a copy this far behind has either stopped +// changing or is in a long stage with nothing new to upload; either way the +// page shows what the copy still records. +export const HUGGING_FACE_LAG_MS = 15 * 60 * 1000; + +function timeMs(value: string | null): number | null { + if (!value) return null; + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? null : parsed; +} + +function isoTime(ms: number | null): string | null { + return ms == null ? null : new Date(ms).toISOString(); +} + +function huggingFaceState( + huggingFace: HuggingFaceRunState, + nowMs: number, +): BuildRunState | null { + return huggingFace.status == null + ? null + : runState(huggingFace.status, timeMs(huggingFace.updated_at), nowMs); +} + +function lastCalibrationEpoch(timeline: BuildTimeline): CalibrationEpoch | null { + const last = timeline.calibration.at(-1); + return last ? { epoch: last.epoch, epochs: last.epochs } : null; +} + +export function candidateRunState( + huggingFace: HuggingFaceRunState, + collector: BuildTimeline | undefined, + nowMs: number, +): CandidateRunState { + const huggingFaceValues = { + hugging_face_status: huggingFace.status, + hugging_face_stage: huggingFace.stage, + hugging_face_updated_at: huggingFace.updated_at, + }; + const huggingFaceStatus = huggingFaceState(huggingFace, nowMs); + const huggingFaceUpdatedMs = timeMs(huggingFace.updated_at); + if ( + !collector || + collectorMissedTheEnd(collector, { + state: huggingFaceStatus, + updated_ms: huggingFaceUpdatedMs, + }) + ) { + return { + state_source: "hugging_face", + status: huggingFaceStatus, + stage: huggingFace.stage, + started_at: huggingFace.started_at, + updated_at: huggingFace.updated_at, + ended_at: null, + heartbeat_at: null, + failure: null, + calibration: huggingFace.calibration ?? null, + ...huggingFaceValues, + hugging_face_differs: false, + }; + } + const huggingFaceLagMs = + collector.updated_ms != null && huggingFaceUpdatedMs != null + ? collector.updated_ms - huggingFaceUpdatedMs + : null; + return { + state_source: "collector", + status: collector.state, + stage: collector.failure?.stage ?? collector.current_stage, + started_at: isoTime(collector.started_ms), + updated_at: isoTime(collector.updated_ms), + ended_at: isoTime(collector.ended_ms), + heartbeat_at: isoTime(collector.heartbeat_ms), + failure: collector.failure + ? { stage: collector.failure.stage, message: collector.failure.message } + : null, + calibration: lastCalibrationEpoch(collector), + ...huggingFaceValues, + hugging_face_differs: + huggingFaceStatus !== collector.state || + huggingFaceLagMs == null || + huggingFaceLagMs > HUGGING_FACE_LAG_MS, + }; +} + +// A Hugging Face run summary with the collector's state laid over it. +export function overlayRunSummary( + summary: StagingRunSummary, + collector: BuildTimeline | undefined, + nowMs: number, +): CandidateRunSummary { + return { + ...summary, + ...candidateRunState( + { + status: summary.status, + stage: summary.stage, + started_at: summary.started_at, + updated_at: summary.updated_at, + }, + collector, + nowMs, + ), + }; +} diff --git a/frontend/lib/microcosm/staging-status.test.ts b/frontend/lib/microcosm/staging-status.test.ts index 8edc45e0..6ed501fd 100644 --- a/frontend/lib/microcosm/staging-status.test.ts +++ b/frontend/lib/microcosm/staging-status.test.ts @@ -1,6 +1,11 @@ import { expect, test } from "bun:test"; -import { formatStagingCurrentStatus, formatStagingStatus } from "./staging-status"; +import { + formatCandidateCurrentStatus, + formatStagingCurrentStatus, + formatStagingStatus, + formatStateSource, +} from "./staging-status"; test("sentence-cases status values", () => { expect(formatStagingStatus("running")).toBe("Running"); @@ -33,3 +38,54 @@ test("falls back safely when stage details are incomplete", () => { expect(formatStagingCurrentStatus({ status: "queued" })).toBe("Queued"); expect(formatStagingCurrentStatus({})).toBe("Unknown"); }); + +test("names the stage where a failed or refused candidate stopped", () => { + expect( + formatCandidateCurrentStatus({ + status: "failed", + stage: "release_gates", + failure: { stage: "release_gates", message: "Release gates failed" }, + calibration: null, + }), + ).toBe("Failed during Release gates"); + expect( + formatCandidateCurrentStatus({ + status: "blocked", + stage: "gate_battery", + failure: { stage: "gate_battery", message: null }, + calibration: null, + }), + ).toBe("Refused at Gate battery"); + expect( + formatCandidateCurrentStatus({ + status: "failed", + stage: "failed", + failure: { stage: null, message: "The build failed." }, + calibration: null, + }), + ).toBe("Failed"); +}); + +test("formats a running candidate's stage the way the Hugging Face progress did", () => { + expect( + formatCandidateCurrentStatus({ + status: "running", + stage: "reform_validation", + failure: null, + calibration: null, + }), + ).toBe("Reform validation"); + expect( + formatCandidateCurrentStatus({ + status: "running", + stage: "calibrating", + failure: null, + calibration: { epoch: 273, epochs: 1500 }, + }), + ).toBe("Calibration, epoch 273 of 1500"); +}); + +test("names where a candidate's state came from", () => { + expect(formatStateSource("collector")).toBe("Telemetry collector"); + expect(formatStateSource("hugging_face")).toBe("Hugging Face staging repository"); +}); diff --git a/frontend/lib/microcosm/staging-status.ts b/frontend/lib/microcosm/staging-status.ts index 205ac77f..7661ae27 100644 --- a/frontend/lib/microcosm/staging-status.ts +++ b/frontend/lib/microcosm/staging-status.ts @@ -39,3 +39,32 @@ export function formatStagingCurrentStatus(progress: JsonObject): string { return formatStagingStatus(progress.status); } + +export interface CandidateCurrentStatus { + status: string | null; + stage: string | null; + failure: { stage: string | null; message: string | null } | null; + calibration: { epoch: number | null; epochs: number | null } | null; +} + +// The Candidate tab's "Current status": the stage a failed or refused run +// stopped at, otherwise the current stage as the Hugging Face progress +// document presents it. +export function formatCandidateCurrentStatus(state: CandidateCurrentStatus): string { + const stoppedAt = stringValue(state.failure?.stage); + if (state.status === "failed") { + return stoppedAt ? `Failed during ${sentenceCaseIdentifier(stoppedAt)}` : "Failed"; + } + if (state.status === "blocked") { + return stoppedAt ? `Refused at ${sentenceCaseIdentifier(stoppedAt)}` : "Refused"; + } + return formatStagingCurrentStatus({ + status: state.status, + stage: state.stage, + calibration: state.calibration, + }); +} + +export function formatStateSource(source: "collector" | "hugging_face"): string { + return source === "collector" ? "Telemetry collector" : "Hugging Face staging repository"; +} diff --git a/frontend/lib/microcosm/staging-tabs-agree.test.ts b/frontend/lib/microcosm/staging-tabs-agree.test.ts new file mode 100644 index 00000000..2f942c09 --- /dev/null +++ b/frontend/lib/microcosm/staging-tabs-agree.test.ts @@ -0,0 +1,180 @@ +// One run seen through both Staging tabs, when the collector and the Hugging +// Face copy record different states. The Candidate tab overlays the collector's +// state on the Hugging Face summary; Build progress merges the two sources' +// documents for its list and refreshes the selected run from the collector. +// Both must apply the same rule to choose between them. +import { describe, expect, test } from "bun:test"; + +import { + type BuildRunDocuments, + type BuildRunState, + buildTimeline, +} from "@/lib/microcosm/build-monitor"; +import { + type HostedRunLoaders, + mergeHostedDocuments, + refreshHostedRun, +} from "@/lib/microcosm/build-runs"; +import { overlayRunSummary } from "@/lib/microcosm/staging-run-state"; +import { + INVESTIGATION_NOW_MS, + UK_BLOCKED_RUN_ID, + US_RUNNING_RUN_ID, + huggingFaceSummary, + ukBlockedCollectorDocuments, + usRunningCollectorDocuments, +} from "@/lib/microcosm/test-support/collector-run-fixtures"; +import type { MicrocosmCountry } from "@policyengine/calibration-tree/countries"; + +interface HuggingFaceEnd { + schemaVersion: 1 | 3; + status: string; + stage: string; + updatedAt: string; +} + +// The run's Hugging Face copy as Build progress reads it: its progress +// document and events, the last event recording `end`. +function huggingFaceDocuments( + runId: string, + country: MicrocosmCountry, + end: HuggingFaceEnd, +): BuildRunDocuments { + const startedAt = "2026-10-18T16:10:08+00:00"; + return { + run_id: runId, + source: "staging", + country, + progress: { + schema_version: end.schemaVersion, + run_id: runId, + status: end.status, + stage: end.stage, + started_at: startedAt, + updated_at: end.updatedAt, + }, + run_manifest: null, + calibration_progress: null, + events: [ + { time: startedAt, type: "stage", status: "running", stage: "created" }, + { time: end.updatedAt, type: "stage", status: end.status, stage: end.stage }, + ], + }; +} + +// The same copy as the Candidate tab lists it. +function huggingFaceListing(runId: string, end: HuggingFaceEnd) { + return huggingFaceSummary(runId, { + schema_version: end.schemaVersion, + status: end.status, + stage: end.stage, + started_at: "2026-10-18T16:10:08+00:00", + updated_at: end.updatedAt, + }); +} + +async function buildProgressStates( + collector: BuildRunDocuments, + huggingFace: BuildRunDocuments, + nowMs: number, +): Promise<{ listed: BuildRunState | undefined; selected: BuildRunState | undefined }> { + const listed = mergeHostedDocuments([collector], [huggingFace], nowMs) + .map((documents) => buildTimeline(documents, nowMs)) + .find((timeline) => timeline.run_id === collector.run_id); + const loaders: HostedRunLoaders = { + collectorConfigured: () => true, + hasStagingHistory: () => true, + loadCollector: async () => collector, + loadHistory: async () => huggingFace, + }; + const selected = await refreshHostedRun( + listed ?? null, + collector.country as MicrocosmCountry, + collector.run_id, + nowMs, + loaders, + ); + return { listed: listed?.state, selected: selected?.state }; +} + +describe("both Staging tabs show the same state for a run both sources know", () => { + test("a collector record that stopped before the run ended gives way to a later Hugging Face end", async () => { + // The collector last heard from the run at 17:16:47, so by 20:10 its record + // says stalled; the Hugging Face copy recorded the run passing at 20:05. + const laterNowMs = Date.parse("2026-10-22T20:10:00Z"); + const end: HuggingFaceEnd = { + schemaVersion: 1, + status: "passed", + stage: "complete", + updatedAt: "2026-10-22T20:05:00+00:00", + }; + const collector = usRunningCollectorDocuments(); + + const candidate = overlayRunSummary( + huggingFaceListing(US_RUNNING_RUN_ID, end), + buildTimeline(collector, laterNowMs), + laterNowMs, + ); + const buildProgress = await buildProgressStates( + collector, + huggingFaceDocuments(US_RUNNING_RUN_ID, "us", end), + laterNowMs, + ); + + expect(buildTimeline(collector, laterNowMs).state).toBe("stalled"); + expect(candidate.status).toBe("passed"); + expect(buildProgress).toEqual({ listed: "passed", selected: "passed" }); + }); + + test("a final collector state stands even when the Hugging Face copy recorded the end later", async () => { + // The collector infers the refusal from its acceptance check events; the + // copy closed the run as completed a little later. + const end: HuggingFaceEnd = { + schemaVersion: 3, + status: "completed", + stage: "complete", + updatedAt: "2026-10-18T23:46:30+00:00", + }; + const collector = ukBlockedCollectorDocuments(); + + const candidate = overlayRunSummary( + huggingFaceListing(UK_BLOCKED_RUN_ID, end), + buildTimeline(collector, INVESTIGATION_NOW_MS), + INVESTIGATION_NOW_MS, + ); + const buildProgress = await buildProgressStates( + collector, + huggingFaceDocuments(UK_BLOCKED_RUN_ID, "uk", end), + INVESTIGATION_NOW_MS, + ); + + expect(candidate.status).toBe("blocked"); + expect(buildProgress).toEqual({ listed: "blocked", selected: "blocked" }); + }); + + test("a collector record newer than a Hugging Face end stands", async () => { + // The copy recorded a failure before the collector's last heartbeat, so + // the collector's later record of a running run wins on both tabs. + const end: HuggingFaceEnd = { + schemaVersion: 1, + status: "failed", + stage: "failed", + updatedAt: "2026-10-22T17:00:00+00:00", + }; + const collector = usRunningCollectorDocuments(); + + const candidate = overlayRunSummary( + huggingFaceListing(US_RUNNING_RUN_ID, end), + buildTimeline(collector, INVESTIGATION_NOW_MS), + INVESTIGATION_NOW_MS, + ); + const buildProgress = await buildProgressStates( + collector, + huggingFaceDocuments(US_RUNNING_RUN_ID, "us", end), + INVESTIGATION_NOW_MS, + ); + + expect(candidate.status).toBe("running"); + expect(buildProgress).toEqual({ listed: "running", selected: "running" }); + }); +}); diff --git a/frontend/lib/microcosm/test-support/collector-run-fixtures.ts b/frontend/lib/microcosm/test-support/collector-run-fixtures.ts new file mode 100644 index 00000000..99c92f24 --- /dev/null +++ b/frontend/lib/microcosm/test-support/collector-run-fixtures.ts @@ -0,0 +1,249 @@ +// Test support: runs from the 2026-10-22 investigation of the Staging page, +// as the Hugging Face staging repository lists them and as the telemetry +// collector rebuilds their documents. The two US runs stopped uploading to +// Hugging Face partway through calibration, so their Hugging Face copies are +// out of date; the UK run's copy is current. +import type { BuildRunDocuments } from "@/lib/microcosm/build-monitor"; +import type { StagingRunSummary } from "@policyengine/calibration-tree/staging-artifact"; + +// Production time when the collector last heard from the running US run. +export const INVESTIGATION_NOW_MS = Date.parse("2026-10-22T17:18:00Z"); + +// Failed at its release acceptance checks on 10-20; Hugging Face still says +// it is calibrating. +export const US_FAILED_RUN_ID = + "populace-us-2024-b5db384-8b0c1950b65f-20261019T222143Z"; +// In reform validation with recent heartbeats; Hugging Face still says it is +// calibrating. +export const US_RUNNING_RUN_ID = + "populace-us-2024-b5db384-8b0c1950b65f-20261021T160022Z"; +// Runs the collector has no record of. +export const US_FINISHED_HUGGING_FACE_RUN_ID = + "populace-us-2024-f0af251-0ad74ed34493-20260619T181855Z"; +export const US_ABANDONED_HUGGING_FACE_RUN_ID = + "us-2024-immigration-266-b-20260702T134743Z"; +// Refused by its acceptance checks; both sources agree. +export const UK_BLOCKED_RUN_ID = + "uk-local-candidate-f100-s42-20261018T161007Z-35fb4c89"; +// One of the five schema 2 local-area candidates that closed as `completed` +// although their acceptance checks reported blocking failures. The collector +// has no record of it. +export const UK_COMPLETED_HUGGING_FACE_RUN_ID = + "uk-local-candidate-f100-s42-20261007T120029Z-2ee47c99"; + +export const US_RELEASE_CHECK_FAILURE = + "Release gates failed: Critical fiscal target 'ssa_supplement.cy2024.oasdi_ssi_payments.social_security_benefits.payment_amount@2024' (Social Security benefits) has relative_error=-0.0641635, exceeding 0.05"; + +export function huggingFaceSummary( + runId: string, + fields: Pick< + StagingRunSummary, + "schema_version" | "status" | "stage" | "started_at" | "updated_at" + >, +): StagingRunSummary { + return { + run_id: runId, + candidate_release_id: runId, + release_id: null, + country_code: null, + run_kind: null, + non_release: null, + progress_path: `runs/${runId}/progress.json`, + run_manifest_path: `runs/${runId}/run_manifest.json`, + ...fields, + }; +} + +// The US list as production served it, newest update first. +export function usHuggingFaceRuns(): StagingRunSummary[] { + return [ + huggingFaceSummary(US_RUNNING_RUN_ID, { + schema_version: 1, + status: "running", + stage: "calibrating", + started_at: "2026-10-21T16:06:14+00:00", + updated_at: "2026-10-21T18:59:03+00:00", + }), + huggingFaceSummary(US_FAILED_RUN_ID, { + schema_version: 1, + status: "running", + stage: "calibrating", + started_at: "2026-10-19T22:25:40+00:00", + updated_at: "2026-10-20T06:09:35+00:00", + }), + huggingFaceSummary(US_ABANDONED_HUGGING_FACE_RUN_ID, { + schema_version: 1, + status: "running", + stage: "base_population_repair", + started_at: null, + updated_at: "2026-07-02T13:48:44+00:00", + }), + huggingFaceSummary(US_FINISHED_HUGGING_FACE_RUN_ID, { + schema_version: 1, + status: "passed", + stage: "complete", + started_at: null, + updated_at: "2026-06-19T18:38:51+00:00", + }), + ]; +} + +export function ukHuggingFaceRuns(): StagingRunSummary[] { + return [ + huggingFaceSummary(UK_BLOCKED_RUN_ID, { + schema_version: 3, + status: "blocked", + stage: "blocked", + started_at: "2026-10-18T16:10:09+00:00", + updated_at: "2026-10-18T23:45:45+00:00", + }), + huggingFaceSummary(UK_COMPLETED_HUGGING_FACE_RUN_ID, { + schema_version: 2, + status: "completed", + stage: "complete", + started_at: "2026-10-07T12:00:29+00:00", + updated_at: "2026-10-07T20:43:34+00:00", + }), + ]; +} + +interface CollectorEvent { + stage: string; + status: string; + time: string; + message?: string; + details?: Record; +} + +// Documents in the shape the collector's `GET /v1/runs/{run_id}` returns: +// version 2 lifecycle events rebuilt from the events the build sent. +function collectorDocuments( + runId: string, + country: string, + progress: Record, + events: CollectorEvent[], +): BuildRunDocuments { + return { + run_id: runId, + source: "staging", + country, + progress: { schema_version: 2, run_id: runId, ...progress }, + run_manifest: null, + calibration_progress: null, + events: events.map((event, index) => ({ + schema_version: 2, + sequence: index + 1, + run_id: runId, + event_type: "stage", + stage_id: event.stage, + stage: event.stage, + status: event.status, + timestamp: event.time, + time: event.time, + message: event.message ?? null, + details: event.details ?? {}, + })), + }; +} + +export function usFailedCollectorDocuments(): BuildRunDocuments { + return collectorDocuments( + US_FAILED_RUN_ID, + "us", + { + country_code: "US", + pipeline: { id: "us-fiscal-refresh", version: "collector-v1" }, + started_at: "2026-10-19T22:23:36Z", + updated_at: "2026-10-20T06:37:45Z", + heartbeat_at: "2026-10-20T06:37:43Z", + status: "failed", + current_stage: "failed", + failure: { + message: US_RELEASE_CHECK_FAILURE, + error_type: "ReleaseGateError", + error_code: null, + failure_class: null, + failed_during: "release_gates", + }, + }, + [ + { stage: "created", status: "started", time: "2026-10-19T22:23:36Z" }, + { stage: "calibration", status: "started", time: "2026-10-20T05:48:41Z" }, + { stage: "calibration", status: "completed", time: "2026-10-20T06:27:42Z" }, + { stage: "export_dataset", status: "started", time: "2026-10-20T06:37:31Z" }, + { + stage: "release_gates", + status: "failed", + time: "2026-10-20T06:37:45Z", + message: US_RELEASE_CHECK_FAILURE, + }, + { + stage: "failed", + status: "failed", + time: "2026-10-20T06:37:45Z", + message: US_RELEASE_CHECK_FAILURE, + }, + ], + ); +} + +export function usRunningCollectorDocuments(): BuildRunDocuments { + return collectorDocuments( + US_RUNNING_RUN_ID, + "us", + { + country_code: "US", + pipeline: { id: "us-fiscal-refresh", version: "collector-v1" }, + started_at: "2026-10-21T16:02:20Z", + updated_at: "2026-10-22T17:16:47Z", + heartbeat_at: "2026-10-22T17:16:45Z", + status: "running", + current_stage: "reform_validation", + }, + [ + { stage: "created", status: "started", time: "2026-10-21T16:02:20Z" }, + { stage: "calibration", status: "started", time: "2026-10-21T18:22:41Z" }, + { stage: "calibration", status: "completed", time: "2026-10-21T19:20:23Z" }, + { stage: "release_gates", status: "started", time: "2026-10-21T19:40:50Z" }, + { stage: "release_gates", status: "completed", time: "2026-10-21T19:40:56Z" }, + { stage: "export_dataset", status: "started", time: "2026-10-21T19:40:57Z" }, + { stage: "export_dataset", status: "completed", time: "2026-10-21T20:38:21Z" }, + { stage: "reform_validation", status: "started", time: "2026-10-22T10:12:02Z" }, + ], + ); +} + +export function ukBlockedCollectorDocuments(): BuildRunDocuments { + const block = { + phase: "terminal", + blocking_failure_count: 3, + blocking_gate_ids: ["uk_target_fit", "uk_local_area_support", "uk_local_target_fit"], + }; + return collectorDocuments( + UK_BLOCKED_RUN_ID, + "uk", + { + country_code: "GB", + pipeline: { id: "uk-local-candidate", version: "collector-v1" }, + started_at: "2026-10-18T16:10:08Z", + updated_at: "2026-10-18T23:45:50Z", + heartbeat_at: "2026-10-18T23:45:02Z", + status: "blocked", + current_stage: "blocked", + block, + }, + [ + { stage: "created", status: "started", time: "2026-10-18T16:10:08Z" }, + { stage: "calibration", status: "started", time: "2026-10-18T18:00:00Z" }, + { stage: "calibration", status: "completed", time: "2026-10-18T23:30:00Z" }, + { stage: "gate_battery", status: "started", time: "2026-10-18T23:40:00Z" }, + { + stage: "gate_battery", + status: "completed", + time: "2026-10-18T23:45:49Z", + details: { blocking_failure_count: 3 }, + }, + { stage: "blocked", status: "blocked", time: "2026-10-18T23:45:50Z", details: block }, + ], + ); +}