fix(showcase/harness): persist partial rollup when a probe run is aborted mid-flight

An aborted D6 run (orchestrator process killed during a pool-churn burst)
left its `probe_runs` row orphaned in `running` state with a null summary.
The boot-time `sweepStaleRuns` then stamped it `state:failed` with
`{total:0,passed:0,failed:0}` and `duration_ms:null`, discarding the
partial per-service results the run had actually computed — a 578s run with
dozens of green features surfaced as `failed / total:0`.

Two minimal, success-path-consistent changes:

- probe-invoker: incrementally persist the running partial tally onto the
  probe_runs row via a new `runWriter.update()` as each fan-out target
  completes, so an orphaned row already carries real partial progress.
  Best-effort — never tanks the tick; `finish()` remains the authoritative
  final write.
- run-history: `sweepStaleRuns` now PRESERVES an existing partial summary
  (and derives a real duration from the persisted started_at) instead of
  clobbering it to zeros. Rows that died before any target completed still
  fall back to an explicit empty rollup.

Purely the result-aggregation-on-abort path; pool sizing and
launch/recycle logic are untouched (churn root cause handled separately).

Author: jpr5

showcase/harness/src/http/probes.test.ts

@@ -97,6 +97,7 @@ function makeFakeWriter(): ProbeRunWriter & {
   const recentMap = new Map<string, ProbeRunRecord[]>();
   return {
     start: async () => ({ id: "row1" }),
+    update: async () => {},
     finish: async () => {},
     recent: async (probeId, _limit) => recentMap.get(probeId) ?? [],
     setRecent: (probeId, runs) => recentMap.set(probeId, runs),
@@ -896,6 +897,7 @@ describe("GET /api/probes/:id — R2-A.9 graceful degradation", () => {
     const sched = makeFakeScheduler();
     const writer: ProbeRunWriter = {
       start: async () => ({ id: "x" }),
+      update: async () => {},
       finish: async () => {},
       recent: async () => {
         throw new Error("PB transient outage");

showcase/harness/src/probes/loader/probe-invoker.test.ts

@@ -2428,6 +2428,10 @@ describe("buildProbeInvoker", () => {
   function fakeRunWriter(): {
     writer: ProbeRunWriter;
     starts: Array<{ probeId: string; startedAt: number; triggered: boolean }>;
+    updates: Array<{
+      id: string;
+      summary: { total: number; passed: number; failed: number } | null;
+    }>;
     finishes: Array<{
       id: string;
       finishedAt: number;
@@ -2440,6 +2444,10 @@ describe("buildProbeInvoker", () => {
       startedAt: number;
       triggered: boolean;
     }> = [];
+    const updates: Array<{
+      id: string;
+      summary: { total: number; passed: number; failed: number } | null;
+    }> = [];
     const finishes: Array<{
       id: string;
       finishedAt: number;
@@ -2452,6 +2460,19 @@ describe("buildProbeInvoker", () => {
         starts.push(opts);
         return { id: `run-${nextId++}` };
       },
+      async update(opts) {
+        updates.push({
+          id: opts.id,
+          summary:
+            opts.summary === null
+              ? null
+              : {
+                  total: opts.summary.total,
+                  passed: opts.summary.passed,
+                  failed: opts.summary.failed,
+                },
+        });
+      },
       async finish(opts) {
         finishes.push({
           id: opts.id,
@@ -2471,7 +2492,7 @@ describe("buildProbeInvoker", () => {
         return [];
       },
     };
-    return { writer, starts, finishes };
+    return { writer, starts, updates, finishes };
   }
 
   it("registers a ProbeRunTracker on the scheduler entry while the handler runs and clears it after", async () => {
@@ -2734,6 +2755,65 @@ describe("buildProbeInvoker", () => {
     });
   });
 
+  // ---------------------------------------------------------------------
+  // Partial-rollup-on-abort: incrementally persist the partial summary to
+  // the probe_runs row as each target completes. Without this, a run that
+  // dies mid-flight (orchestrator process killed during a pool-churn
+  // burst) leaves a bare `running` row carrying NO partial counts — the
+  // boot-time sweep then has nothing to preserve and the dashboard shows
+  // `failed / total:0` even though dozens of features actually passed.
+  // ---------------------------------------------------------------------
+  it("incrementally persists partial summary via runWriter.update as targets complete", async () => {
+    const inputSchema = z.object({ key: z.string() }).passthrough();
+    const driver: ProbeDriver = {
+      kind: "smoke",
+      inputSchema,
+      async run(ctx, input) {
+        const key = (input as { key: string }).key;
+        return {
+          key,
+          state: key.endsWith("bad") ? "red" : "green",
+          signal: {},
+          observedAt: ctx.now().toISOString(),
+        };
+      },
+    };
+    const cfg: ProbeConfig = {
+      kind: "smoke",
+      id: "smoke",
+      schedule: "*/15 * * * *",
+      // Serialize so the update sequence is deterministic.
+      max_concurrency: 1,
+      targets: [{ key: "smoke:a" }, { key: "smoke:b" }, { key: "smoke:bad" }],
+    };
+    const { writer } = mkWriter();
+    const sched = fakeScheduler();
+    const rw = fakeRunWriter();
+    await buildProbeInvoker(cfg, {
+      driver,
+      schedulerId: cfg.id,
+      discoveryRegistry: createDiscoveryRegistry(),
+      writer,
+      scheduler: sched.scheduler,
+      runWriter: rw.writer,
+      ...BASE_DEPS,
+    })();
+    // One update per completed target, each carrying the running partial
+    // tally so an orphaned row reflects real progress.
+    expect(rw.updates.map((u) => u.summary)).toEqual([
+      { total: 3, passed: 1, failed: 0 },
+      { total: 3, passed: 2, failed: 0 },
+      { total: 3, passed: 2, failed: 1 },
+    ]);
+    // All updates target the same row created by start().
+    expect(rw.updates.every((u) => u.id === "run-1")).toBe(true);
+    // The final summary still lands via finish().
+    expect(rw.finishes[0]).toMatchObject({
+      state: "completed",
+      summary: { total: 3, passed: 2, failed: 1 },
+    });
+  });
+
   // ---------------------------------------------------------------------
   // R3-A.3: orphan-risk warn log when runWriter.start fails
   // ---------------------------------------------------------------------
@@ -2769,6 +2849,7 @@ describe("buildProbeInvoker", () => {
     const sched = fakeScheduler();
     const failingWriter: ProbeRunWriter = {
       start: vi.fn().mockRejectedValue(new Error("network blip")),
+      update: vi.fn().mockResolvedValue(undefined),
       finish: vi.fn().mockResolvedValue(undefined),
       recent: vi.fn().mockResolvedValue([]),
     };
@@ -2829,6 +2910,7 @@ describe("buildProbeInvoker", () => {
     const sched = fakeScheduler();
     const failingWriter: ProbeRunWriter = {
       start: vi.fn().mockRejectedValue(new Error("PB down")),
+      update: vi.fn().mockResolvedValue(undefined),
       finish: vi.fn().mockResolvedValue(undefined),
       recent: vi.fn().mockResolvedValue([]),
     };
@@ -2872,6 +2954,7 @@ describe("buildProbeInvoker", () => {
     const sched = fakeScheduler();
     const failingFinish: ProbeRunWriter = {
       start: vi.fn().mockResolvedValue({ id: "run-x" }),
+      update: vi.fn().mockResolvedValue(undefined),
       finish: vi.fn().mockRejectedValue(new Error("PB down")),
       recent: vi.fn().mockResolvedValue([]),
     };

showcase/harness/src/probes/loader/probe-invoker.ts

@@ -634,6 +634,27 @@ export function buildProbeInvoker(
           state: result.state,
           durationMs: Date.now() - targetStart,
         });
+        // Partial-rollup persistence: stamp the running partial tally onto
+        // the probe_runs row as each target completes so an orphaned row
+        // (process killed mid-run during a pool-churn burst) reflects real
+        // progress instead of a null summary. The boot-time sweep then
+        // preserves this partial rollup rather than zeroing it. Best-effort:
+        // a runWriter hiccup must never tank the probe tick — finish() in
+        // the finally block is still the authoritative final write.
+        if (runWriter && runRowId !== null) {
+          try {
+            await runWriter.update({
+              id: runRowId,
+              summary: { total: inputs.length, passed, failed },
+            });
+          } catch (err) {
+            logger.error("probe.run-writer-update-failed", {
+              probeId: cfg.id,
+              runId: runRowId,
+              err: err instanceof Error ? err.message : String(err),
+            });
+          }
+        }
         try {
           await writer.write(result);
         } catch (err) {

showcase/harness/src/probes/run-history.test.ts

@@ -2,8 +2,9 @@ import { describe, it, expect, beforeEach } from "vitest";
 import {
   PROBE_RUNS_COLLECTION,
   createProbeRunWriter,
-  type ProbeRunRecord,
+  sweepStaleRuns,
 } from "./run-history.js";
+import type { ProbeRunRecord } from "./run-history.js";
 import type { PbClient, ListOpts, ListResult } from "../storage/pb-client.js";
 
 /**
@@ -339,4 +340,77 @@ describe("run-history", () => {
       expect(fake.updateCalls).toHaveLength(0);
     });
   });
+
+  // ---------------------------------------------------------------------
+  // Partial-rollup-on-abort: when a probe run is aborted mid-flight (the
+  // orchestrator process dies during a pool-churn burst, leaving a
+  // `running` row orphaned), the boot-time sweep must PRESERVE whatever
+  // partial per-service rollup the row already carries rather than
+  // clobbering it with `{ total: 0, passed: 0, failed: 0 }`. Pre-fix the
+  // sweep discarded the real partial counts a 578s D6 run had computed
+  // (41 assertions passed, dozens of features green) — the dashboard /
+  // probe_runs then showed `failed / total:0` instead of reality.
+  // ---------------------------------------------------------------------
+  describe("sweepStaleRuns() — partial-rollup preservation", () => {
+    it("preserves an existing partial summary instead of zeroing it", async () => {
+      const fake = fakePb();
+      const stale = new Date(Date.now() - 20 * 60 * 1000).toISOString();
+      // An orphaned `running` row that DID accumulate partial per-service
+      // results before the process died.
+      const partialSummary = {
+        total: 38,
+        passed: 31,
+        failed: 2,
+        services: [
+          { slug: "d6:starter-lg-react", state: "completed", result: "green" },
+          { slug: "d6:starter-lg-py", state: "failed", error: "boom" },
+        ],
+      };
+      await fake.pb.create(PROBE_RUNS_COLLECTION, {
+        probe_id: "d6-all-pills-e2e",
+        started_at: stale,
+        finished_at: null,
+        duration_ms: null,
+        triggered: false,
+        state: "running",
+        summary: partialSummary,
+      });
+
+      const swept = await sweepStaleRuns(fake.pb);
+      expect(swept).toBe(1);
+
+      const updated = fake.updateCalls.at(-1)!;
+      // The run is terminal + failed (it was aborted), but the partial
+      // rollup MUST survive — not be overwritten with zeros.
+      expect(updated.record.state).toBe("failed");
+      expect(updated.record.summary).toEqual(partialSummary);
+    });
+
+    it("still zeroes the summary for a row that never accumulated partial results", async () => {
+      const fake = fakePb();
+      const stale = new Date(Date.now() - 20 * 60 * 1000).toISOString();
+      // Orphaned row that died before any target completed — summary is
+      // still null. The sweep should fall back to an explicit empty rollup.
+      await fake.pb.create(PROBE_RUNS_COLLECTION, {
+        probe_id: "smoke",
+        started_at: stale,
+        finished_at: null,
+        duration_ms: null,
+        triggered: false,
+        state: "running",
+        summary: null,
+      });
+
+      const swept = await sweepStaleRuns(fake.pb);
+      expect(swept).toBe(1);
+
+      const updated = fake.updateCalls.at(-1)!;
+      expect(updated.record.state).toBe("failed");
+      expect(updated.record.summary).toEqual({
+        total: 0,
+        passed: 0,
+        failed: 0,
+      });
+    });
+  });
 });

showcase/harness/src/probes/run-history.ts

@@ -63,6 +63,18 @@ export interface ProbeRunWriter {
     startedAt: number;
     triggered: boolean;
   }): Promise<{ id: string }>;
+  /**
+   * Persist a partial rollup onto a still-`running` row WITHOUT marking it
+   * terminal. Called incrementally as each fan-out target completes so an
+   * orphaned row (orchestrator process killed mid-run — e.g. a pool-churn
+   * burst) already carries the partial per-service results the run had
+   * computed. Without this, a `running` row that never reaches `finish()`
+   * has a null summary, and the boot-time `sweepStaleRuns` has nothing to
+   * preserve — the dashboard then shows `failed / total:0` even though
+   * dozens of features actually passed. State stays `running`; only
+   * `summary` is updated. Best-effort, like `finish()`.
+   */
+  update(opts: { id: string; summary: ProbeRunSummary }): Promise<void>;
   /**
    * Mark a row finished. `duration_ms` is computed from
    * `finishedAt - row.started_at` (read off the persisted row, not from a
@@ -126,6 +138,27 @@ export function createProbeRunWriter(pb: PbClient): ProbeRunWriter {
       return { id: created.id };
     },
 
+    async update(opts) {
+      // Refresh only the partial summary on a still-running row. We leave
+      // `state`, `finished_at`, and `duration_ms` untouched so this can be
+      // called repeatedly mid-run without prematurely marking the row
+      // terminal. The row must exist (start() created it); a missing row
+      // means the caller is updating an id it never created — surface it
+      // rather than silently writing junk, mirroring finish()'s guard.
+      const existing = await pb.getOne<ProbeRunRow>(
+        PROBE_RUNS_COLLECTION,
+        opts.id,
+      );
+      if (!existing) {
+        // eslint-disable-next-line no-console
+        console.warn("run-history.update: row missing", { runId: opts.id });
+        return;
+      }
+      await pb.update<ProbeRunRow>(PROBE_RUNS_COLLECTION, opts.id, {
+        summary: opts.summary,
+      });
+    },
+
     async finish(opts) {
       // Read the persisted started_at so duration is always defined off
       // the row — see the JSDoc on `finish()` for why we don't trust a
@@ -202,11 +235,29 @@ export async function sweepStaleRuns(
   let swept = 0;
   for (const row of stale.items) {
     try {
+      // Partial-rollup preservation: an orphaned `running` row may already
+      // carry the partial per-service rollup the invoker persisted
+      // incrementally (`runWriter.update`) before the process died. Marking
+      // the run `failed` is correct (it WAS aborted), but clobbering the
+      // summary to `{0,0,0}` discards real work — a 578s D6 run with dozens
+      // of green features would surface as `failed / total:0`. Keep the
+      // existing summary when present; fall back to an explicit empty
+      // rollup only for rows that died before any target completed.
+      const summary = row.summary ?? { total: 0, passed: 0, failed: 0 };
+      // Derive a real duration from the persisted started_at when possible
+      // so a preserved partial run still reports how long it actually ran.
+      const finishedAt = Date.now();
+      const startedAtMs = row.started_at
+        ? Date.parse(row.started_at)
+        : Number.NaN;
+      const durationMs = Number.isFinite(startedAtMs)
+        ? finishedAt - startedAtMs
+        : null;
       await pb.update<ProbeRunRow>(PROBE_RUNS_COLLECTION, row.id, {
         state: "failed",
-        finished_at: new Date().toISOString(),
-        duration_ms: null,
-        summary: { total: 0, passed: 0, failed: 0 },
+        finished_at: new Date(finishedAt).toISOString(),
+        duration_ms: durationMs,
+        summary,
       });
       swept++;
     } catch {