Show the collector's run state on the Staging page's Candidate tab

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/<run_id>/` telemetry folders a
@@ -81,8 +87,8 @@ returns HTTP 404 with `{"error": "country_not_configured", "requested_country",
 | `GET /api/microcosm/compare?a=<id>&b=<id>` | Version-over-version diff |
 | `GET /api/microcosm/builds?country=<code>&source=local\|staging` | Every build run from one source as a stage timeline |
 | `GET /api/microcosm/builds/run?id=<run_id>&country=<code>&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=<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/<run_id>/` 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=<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/<run_id>/` in the country's release repository (read at the commit the run recorded, digest-verified) |
 | `GET /api/microcosm/staging/target-diagnostics?id=<run_id>&...` | Faceted diagnostics for a staging candidate once diagnostics exist |
 | `GET /api/microcosm/staging/compare?run=<run_id>&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=<run_id>&release=<resolved_id>&mode=reported\|shared&...` | One hierarchy level of weighted target-error changes for a staging candidate and an explicit current release |

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:
 

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<CollectorTimelines>) {
+  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<Response>) {
+  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<string, unknown>;
+    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,
+  });
+});

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<StagingRunDetail>;
+  loadCollectorTimelines(
+    country: MicrocosmCountry,
+    runIds: readonly string[],
+    nowMs: number,
+  ): Promise<CollectorTimelines>;
+  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<string, unknown>;
+  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();