fix(routing): health evidence counts physical sends and final terminal

src/routing/health.ts

@@ -94,7 +94,6 @@ interface HealthSample {
 }
 
 interface HealthRow extends HealthSample {
-  attemptCount?: number;
   rowJson?: string | null;
 }
 
@@ -103,33 +102,53 @@ interface HealthRow extends HealthSample {
  * Combo/failover requests store each upstream try in `entry.attempts` while
  * the top-level row records the final outcome; a provider/model that failed
  * as a non-final attempt must still contribute its own health samples.
+ *
+ * Only physical sends are evidence: an attempt answered locally or refused
+ * before it left the proxy says nothing about the target. The row's terminal
+ * facts belong to its last physical send. Returns `undefined` when the row
+ * itself is the sample (no attempts, a single attempt, or none named for this
+ * candidate) and `[]` when the row sent nothing upstream at all.
  */
 function attemptSamplesFor(
-  row: Pick<HealthRow, "timestamp" | "attemptCount" | "rowJson">,
+  row: Pick<HealthRow, "timestamp" | "rowJson">,
   provider: string,
   model: string,
-): HealthSample[] {
-  if (!row.rowJson || (row.attemptCount ?? 1) <= 1) return [];
+): HealthSample[] | undefined {
+  if (!row.rowJson) return undefined;
   try {
-    const parsed = JSON.parse(row.rowJson) as { attempts?: unknown };
-    if (!Array.isArray(parsed.attempts)) return [];
+    const parsed = JSON.parse(row.rowJson) as {
+      attempts?: unknown;
+      closeReason?: unknown;
+      terminalStatus?: unknown;
+      failureCause?: unknown;
+    };
+    // A local refusal logged before any attempt began never reached a target.
+    if (!Array.isArray(parsed.attempts)) return parsed.failureCause === "local-refusal" ? [] : undefined;
+    const attempts = parsed.attempts.filter(
+      (attempt): attempt is Record<string, unknown> =>
+        !!attempt && typeof attempt === "object" && !Array.isArray(attempt),
+    );
+    const sent = attempts.filter(attempt => attempt.locallyAnswered !== true && attempt.sendCount !== 0);
+    if (sent.length === 0) return [];
+    const mine = attempts.filter(attempt => attempt.provider === provider && attempt.model === model);
+    if (attempts.length <= 1 || mine.length === 0) return undefined;
+    const finalSend = sent.at(-1);
     const samples: HealthSample[] = [];
-    for (const attempt of parsed.attempts) {
-      if (!attempt || typeof attempt !== "object" || Array.isArray(attempt)) continue;
-      const record = attempt as Record<string, unknown>;
-      if (record.provider !== provider || record.model !== model) continue;
+    for (const record of mine) {
+      if (!sent.includes(record)) continue;
       if (typeof record.status !== "number" || typeof record.durationMs !== "number") continue;
+      const isFinal = record === finalSend;
       samples.push({
         status: record.status,
-        closeReason: null,
-        terminalStatus: null,
+        closeReason: isFinal && typeof parsed.closeReason === "string" ? parsed.closeReason : null,
+        terminalStatus: isFinal && typeof parsed.terminalStatus === "string" ? parsed.terminalStatus : null,
         durationMs: record.durationMs,
         timestamp: row.timestamp,
       });
     }
     return samples;
   } catch {
-    return [];
+    return undefined;
   }
 }
 
@@ -243,8 +262,7 @@ function computeHistoricalHealthEvidence(
     }
     const rows = handle.query(
       `SELECT status, close_reason AS closeReason, terminal_status AS terminalStatus,
-              duration_ms AS durationMs, timestamp,
-              attempt_count AS attemptCount, row_json AS rowJson
+              duration_ms AS durationMs, timestamp, row_json AS rowJson
        FROM requests WHERE ${where.join(" AND ")}
        ORDER BY timestamp DESC LIMIT ?`,
     ).all(...values, HEALTH_MAX_SAMPLES) as HealthRow[];
@@ -254,7 +272,7 @@ function computeHistoricalHealthEvidence(
     // rows that cannot contribute samples for this candidate.
     const escapeLike = (value: string): string => value.replace(/[\\%_]/g, match => `\\${match}`);
     const attemptRows = handle.query(
-      `SELECT timestamp, attempt_count AS attemptCount, row_json AS rowJson
+      `SELECT timestamp, row_json AS rowJson
        FROM requests WHERE timestamp >= ? AND attempt_count > 1
          AND row_json LIKE ? ESCAPE '\\'
          AND row_json LIKE ? ESCAPE '\\'
@@ -267,17 +285,14 @@ function computeHistoricalHealthEvidence(
       input.provider,
       input.model,
       HEALTH_MAX_SAMPLES,
-    ) as Array<
-      Pick<HealthRow, "timestamp" | "attemptCount" | "rowJson">
-    >;
+    ) as Array<Pick<HealthRow, "timestamp" | "rowJson">>;
 
     const samples: HealthSample[] = [];
     for (const row of rows) {
-      const attemptSamples = attemptSamplesFor(row, input.provider, input.model);
-      samples.push(...(attemptSamples.length > 0 ? attemptSamples : [row]));
+      samples.push(...(attemptSamplesFor(row, input.provider, input.model) ?? [row]));
     }
     for (const row of attemptRows) {
-      samples.push(...attemptSamplesFor(row, input.provider, input.model));
+      samples.push(...(attemptSamplesFor(row, input.provider, input.model) ?? []));
     }
     // Newest first for the consecutive-failure walk; attempt samples inherit
     // their row's timestamp.

structure/dashboard-and-usage.md

@@ -279,7 +279,7 @@ inheriting its bounds, so deleting a ledger row removes it from the grouping.
 `usageLedgerMaxBytes` is unset by default; when set, an append that crosses it publishes the newest
 whole rows byte for byte through the shared atomic writer, refuses the rename unless the source is
 the exact revision that was copied, and then discards the Logs ring, the retained aggregates and the
-request-history index so no surface serves rows the ledger no longer has. That index (`src/routing/history/`) rebuilds a ledger of at least `REQUEST_HISTORY_WORKER_REBUILD_MIN_BYTES` (1 MiB) on a worker thread (#6990): `rebuild-worker.ts`, driven by `rebuild-job.ts`, ingests the ledger snapshot the refresh recorded into a private `<index>.rebuild-<pid>-<n>` file through the same `ingest.ts` the inline path uses, switches it to WAL and closes it; the serving thread then closes the old index and renames the build over it. Async readers wait for the build; the routing-time sync entry never does and throws `RequestHistoryIndexBusyError`, which health scoring reads as unknown evidence. A close discards the build, rejects waiting readers with `RequestHistoryIndexClosedError`, and removes the build files only after the worker thread exits. For 30 s after a worker build fails (other than on a replaced ledger), async readers and manual rebuilds rebuild inline instead, while routing-time reads keep reporting busy.
+request-history index so no surface serves rows the ledger no longer has. That index (`src/routing/history/`) rebuilds a ledger of at least `REQUEST_HISTORY_WORKER_REBUILD_MIN_BYTES` (1 MiB) on a worker thread (#6990): `rebuild-worker.ts`, driven by `rebuild-job.ts`, ingests the ledger snapshot the refresh recorded into a private `<index>.rebuild-<pid>-<n>` file through the same `ingest.ts` the inline path uses, switches it to WAL and closes it; the serving thread then closes the old index and renames the build over it. Async readers wait for the build; the routing-time sync entry never does and throws `RequestHistoryIndexBusyError`, which health scoring reads as unknown evidence. Health scoring (`src/routing/health.ts`) counts physical sends only: a request answered locally or refused before any send is no evidence for any target, and a row's terminal status and close reason score its last physical send, so a 200 stream cut mid-flight after a failover counts against the target that dropped it (`tests/server/health-scoring.test.ts`). A close discards the build, rejects waiting readers with `RequestHistoryIndexClosedError`, and removes the build files only after the worker thread exits. For 30 s after a worker build fails (other than on a replaced ledger), async readers and manual rebuilds rebuild inline instead, while routing-time reads keep reporting busy.
 Ledger revisions in `src/usage/log.ts` capture device/inode identifiers from bigint descriptor stats as exact decimal strings; history metadata, ingestion, cooperative scans and retention compare them without narrowing to Numbers, preserving distinct NTFS identifiers above 2^53. An older index containing a rounded identifier fails the identity check and rebuilds once. `src/usage/summary.ts` turns that file into the `/api/usage` shape — totals, daily zero-filled
 grid, model and provider breakdowns, and `measured / reported / unreported / unsupported / estimated` counts. Each scope also aggregates end-to-end output throughput (#6309): measured output tokens and wall-clock `durationMs` are summed over attempts reporting both, exposing `throughputTokensPerSec` — a token-sum-over-duration-sum, never a mean of per-request rates. `throughputSamples` counts qualifying attempts (legacy rows without attempts contribute one); nonpositive or nonfinite tokens/timing are excluded. The Usage summary and Models/Providers tables display the rate and sample basis, or unavailable when none qualify.
 All three throughput displays use the same finite-number guard as their tooltip; malformed or nonfinite rates are unavailable rather than formatted.

tests/server/health-scoring.test.ts

@@ -247,6 +247,60 @@ describe("health-aware scoring (RI-06)", () => {
     expect(finalTarget.successRate).toBe(1);
   });
 
+  test("a stream that dies after its 200 head is a failure for the attempt that dropped it", () => {
+    for (let index = 0; index < 5; index++) appendUsageEntry(row(`ok-${index}`, 200, 800));
+    // b/m2 failed over to a/m1, whose 200 stream was then cut mid-flight.
+    appendUsageEntry({
+      ...row("combo-abort", 200, 1500, {
+        provider: "combo",
+        model: "c1",
+        terminalStatus: "incomplete",
+        timestamp: Date.now() - 1_000,
+      }),
+      attempts: [
+        { ordinal: 1, provider: "b", model: "m2", adapter: "openai-chat", status: 503, durationMs: 400, sendCount: 1, recoveryKinds: [], usageStatus: "reported" },
+        { ordinal: 2, provider: "a", model: "m1", adapter: "openai-chat", status: 200, durationMs: 1100, streamAborted: true, sendCount: 1, recoveryKinds: [], usageStatus: "unreported" },
+      ],
+    });
+    const evidence = healthEvidenceForCandidate({ provider: "a", model: "m1" });
+    expect(evidence.sampleCount).toBe(6);
+    expect(evidence.failures).toBe(1);
+    expect(evidence.incompleteStreamRate).toBeCloseTo(1 / 6, 5);
+  });
+
+  test("a request with no physical send to a target is not health evidence for it", () => {
+    for (let index = 0; index < 5; index++) appendUsageEntry(row(`ok-${index}`, 200, 800));
+    // Answered locally: the row names a/m1 but nothing left the proxy.
+    appendUsageEntry({
+      ...row("local", 200, 5, { localTerminalReason: "local-answer" }),
+      attempts: [
+        { ordinal: 1, provider: "a", model: "m1", adapter: "openai-chat", status: 200, durationMs: 5, locallyAnswered: true, sendCount: 0, recoveryKinds: [], usageStatus: "reported" },
+      ],
+    });
+    // Refused before send for a/m1, then served by b/m2.
+    appendUsageEntry({
+      ...row("refused", 200, 900, { provider: "b", model: "m2", timestamp: Date.now() - 1_000 }),
+      attempts: [
+        { ordinal: 1, provider: "a", model: "m1", adapter: "openai-chat", status: 503, durationMs: 1, sendCount: 0, recoveryKinds: [], usageStatus: "unreported" },
+        { ordinal: 2, provider: "b", model: "m2", adapter: "openai-chat", status: 200, durationMs: 900, sendCount: 1, recoveryKinds: [], usageStatus: "reported" },
+      ],
+    });
+    const evidence = healthEvidenceForCandidate({ provider: "a", model: "m1" });
+    expect(evidence.sampleCount).toBe(5);
+    expect(evidence.failures).toBeUndefined();
+    expect(healthEvidenceForCandidate({ provider: "b", model: "m2" }).sampleCount).toBe(1);
+  });
+
+  test("a local refusal logged before any attempt began is not health evidence", () => {
+    for (let index = 0; index < 5; index++) appendUsageEntry(row(`ok-${index}`, 200, 800));
+    // Refused by the proxy itself (e.g. server_busy admission): no attempts array, and the
+    // persisted row carries only the derived `local-refusal` cause, not `localTerminalReason`.
+    appendUsageEntry(row("busy", 503, 2, { failureCause: "local-refusal", timestamp: Date.now() - 1_000 }));
+    const evidence = healthEvidenceForCandidate({ provider: "a", model: "m1" });
+    expect(evidence.sampleCount).toBe(5);
+    expect(evidence.failures).toBeUndefined();
+  });
+
   test("execution path applies live codex account cooldown to openai candidates", async () => {
     const { clearCodexUpstreamHealth, recordCodexUpstreamOutcome } = await import("../../src/codex/routing");
     clearCodexUpstreamHealth();