dezren39 · Nov 1, 2026 · Nov 1, 2026 · Nov 1, 2026
@@ -46,7 +46,11 @@
 // store cannot be read, the tick writes a heartbeat carrying the
 // error and exits non-zero WITHOUT touching quota-current.json. A
 // single provider failing or exceeding its cap degrades only its own
-// rows; the tick still succeeds.
+// rows; the tick still succeeds. A provider whose run errors keeps
+// its last-known ok entries from the previous tick's
+// quota-current.json (#1268) — retained verbatim, so their original
+// measuredAt makes the hub classify them stale instead of the fresh
+// error rows replacing the last good data.
 //
 // Usage: bun run scripts/gather-quota.ts
 // Env: OPENCHAMBER_LIB_DIR (default: first of ~/git/devtoday-
@@ -55,7 +59,8 @@
 //      QUOTA_CURRENT_PATH, QUOTA_HEARTBEAT_PATH,
 //      QUOTA_GATHERER_STATE_PATH, QUOTA_SNAPSHOTS_DIR (./snapshots),
 //      USAGE_SNAPSHOT_OUTBOX_PATH, META_QUOTA_EXPORT_PATH,
-//      QUOTA_GATHER_TIMEOUT_MS (15000), TURSO_API_TOKEN (else the
+//      QUOTA_GATHER_TIMEOUT_MS (15000),
+//      QUOTA_GATHER_CONCURRENCY (6), TURSO_API_TOKEN (else the
 //      Bitwarden secret turso.platform-api-token-muse-cli, resolved
 //      by name via ~/tools/bws-env conventions — BWS_BIN and
 //      BWS_ACCESS_TOKEN_FILE override its plumbing),
@@ -73,6 +78,9 @@ import {
   entryFromResult,
   entryFromFileExport,
   entryFromTurso,
+  lastKnownEntries,
+  mapWithConcurrency,
+  markRetained,
   recordsFromEntries,
   type FileExportDoc,
   type QuotaProviderResult,
@@ -104,6 +112,7 @@ const statePath = process.env.QUOTA_GATHERER_STATE_PATH ?? join(PKG, "quota-gath
 const snapshotsDir = process.env.QUOTA_SNAPSHOTS_DIR ?? join(PKG, "snapshots");
 const outboxPath = process.env.USAGE_SNAPSHOT_OUTBOX_PATH ?? null;
 const timeoutMs = Number(process.env.QUOTA_GATHER_TIMEOUT_MS ?? 15_000);
+const gatherConcurrency = Number(process.env.QUOTA_GATHER_CONCURRENCY ?? 6);
 const OBSERVED_BY = "quota-gatherer";
 
 /** File-export producers ingested each tick (the #1301 convention). */
@@ -305,6 +314,8 @@ interface ProviderRun {
   ms: number;
   entries: QuotaCurrentEntry[];
   error?: string;
+  /** #1268: entries are last-known retentions, not fresh reads. */
+  retained?: boolean;
 }
 
 async function runProvider(quota: QuotaModule, id: string, tickIso: string): Promise<ProviderRun> {
@@ -363,6 +374,20 @@ async function runFileExport(fx: (typeof FILE_EXPORTS)[number]): Promise<Provide
   };
 }
 
+/** The previous tick's quota-current entries, for #1268 last-known
+ *  retention. Any read/parse failure means no retention this tick. */
+async function readPreviousEntries(): Promise<QuotaCurrentEntry[]> {
+  try {
+    const parsed: unknown = JSON.parse(await readFile(currentPath, "utf8"));
+    if (typeof parsed === "object" && parsed !== null && Array.isArray((parsed as QuotaCurrentFile).providers)) {
+      return (parsed as QuotaCurrentFile).providers;
+    }
+  } catch {
+    // no previous tick yet, or an unreadable file — nothing to retain
+  }
+  return [];
+}
+
 async function readState(): Promise<Map<string, string>> {
   try {
     const parsed: unknown = JSON.parse(await readFile(statePath, "utf8"));
@@ -399,9 +424,28 @@ async function main(): Promise<number> {
     );
   }
 
-  const runs = await Promise.all(providerIds.map((id) => runProvider(quota, id, tickAt)));
+  // #1268: bounded provider concurrency (default 6, matching the
+  // OpenChamber CLI) instead of one Promise.all over the catalogue.
+  const runs = await mapWithConcurrency(providerIds, gatherConcurrency, (id) => runProvider(quota, id, tickAt));
   for (const fx of FILE_EXPORTS) runs.push(await runFileExport(fx));
   runs.push(await runTurso(tickAt));
+
+  // #1268: an errored run with no ok row of its own retains the
+  // provider's last-known ok entries from the previous tick instead
+  // of replacing them with fresh error rows (or with nothing, on a
+  // whole-fetch timeout). The heartbeat below still reports the run
+  // as an error, with its reason and the retained row count.
+  const prevEntries = await readPreviousEntries();
+  if (prevEntries.length) {
+    for (const run of runs) {
+      if (run.state !== "error" || run.entries.some((e) => e.state === "ok")) continue;
+      const retained = lastKnownEntries(prevEntries, run.id);
+      if (retained.length) {
+        run.entries = markRetained(retained, tickAt);
+        run.retained = true;
+      }
+    }
+  }
   const entries = runs.flatMap((r) => r.entries);
 
   const perProvider: QuotaGatherHeartbeat["perProvider"] = {};
@@ -413,7 +457,7 @@ async function main(): Promise<number> {
       ...(run.error ? { error: run.error } : {}),
     };
     console.log(
-      `quota-gatherer: ${run.id} ${run.state} ${run.ms}ms rows=${run.entries.length}${run.error ? ` error=${run.error.slice(0, 140)}` : ""}`,
+      `quota-gatherer: ${run.id} ${run.state} ${run.ms}ms rows=${run.entries.length}${run.retained ? " retained-last-known" : ""}${run.error ? ` error=${run.error.slice(0, 140)}` : ""}`,
     );
   }
 

@@ -1304,6 +1304,13 @@ export interface QuotaCurrentEntry {
   planLabel?: string;
   state: "ok" | "error" | "unconfigured";
   measuredAt: string;
+  /** #1268: ISO time of the tick that retained this entry from the
+   *  previous tick after the provider's run errored. The data is
+   *  last-known, never a fresh read: readers classify it stale
+   *  regardless of measuredAt freshness (a retained entry is usually
+   *  only one tick old, well inside the freshness bound). Absent on
+   *  fresh reads. */
+  retainedFromFailedTick?: string;
   source: string;
   error?: string;
   windows: QuotaCurrentWindow[];

@@ -68,6 +68,10 @@ type FreshState = PoolFreshness["state"];
 
 function entryState(entry: QuotaCurrentEntry, nowMs: number): FreshState {
   if (entry.state === "ok") {
+    // #1268: retained last-known entries are stale by definition —
+    // the provider's latest run failed, however recent the retained
+    // measuredAt is (one tick is 15 min, inside the 20-min bound).
+    if (entry.retainedFromFailedTick) return "stale";
     const t = Date.parse(entry.measuredAt);
     return Number.isFinite(t) && t - nowMs <= QUOTA_CURRENT_FUTURE_SKEW_MS && nowMs - t <= QUOTA_CURRENT_FRESH_MS
       ? "ok"

@@ -528,3 +528,57 @@ export function entryFromTurso(
   entry.usage = usage;
   return entry;
 }
+
+/**
+ * #1268: map over items with at most `limit` invocations in flight,
+ * preserving input order in the results. The gatherer fetches ~30
+ * catalogue providers per tick; an unbounded Promise.all fires all
+ * of them at once. A non-positive or non-finite limit degrades to
+ * sequential (1) rather than to unbounded. Errors propagate like
+ * Promise.all: the first rejection rejects the whole map (callers
+ * that fail soft per item catch inside their worker, as runProvider
+ * does).
+ */
+export async function mapWithConcurrency<T, R>(
+  items: readonly T[],
+  limit: number,
+  fn: (item: T, index: number) => Promise<R>,
+): Promise<R[]> {
+  const cap = Number.isFinite(limit) && limit >= 1 ? Math.floor(limit) : 1;
+  const results = new Array<R>(items.length);
+  let next = 0;
+  const workers = Array.from({ length: Math.min(cap, items.length) }, async () => {
+    while (next < items.length) {
+      const i = next++;
+      results[i] = await fn(items[i], i);
+    }
+  });
+  await Promise.all(workers);
+  return results;
+}
+
+/**
+ * #1268: the last-known entries for a provider — the previous tick's
+ * `ok` entries, verbatim. When this tick's run for a provider errors
+ * out (timeout, fetch failure), the gatherer retains these instead
+ * of writing fresh error rows over them: the entries keep their
+ * original measuredAt, so the hub's freshness classification
+ * (quota-current.ts) surfaces them as stale — last-known data with
+ * visible staleness, never fresh-looking zeros and never a vanished
+ * provider. Error/unconfigured previous entries are not data and are
+ * never retained.
+ */
+export function lastKnownEntries(prev: readonly QuotaCurrentEntry[], providerId: string): QuotaCurrentEntry[] {
+  return prev.filter((e) => e.providerId === providerId && e.state === "ok");
+}
+
+/**
+ * #1268: stamp retained entries with the failed tick that kept them.
+ * Copies, never the previous tick's objects; measuredAt is preserved
+ * (it is the honest age of the data) — the stamp is the explicit
+ * signal readers use to classify the entry stale immediately,
+ * instead of it reading as fresh until the freshness bound lapses.
+ */
+export function markRetained(entries: readonly QuotaCurrentEntry[], failedTickIso: string): QuotaCurrentEntry[] {
+  return entries.map((e) => ({ ...e, retainedFromFailedTick: failedTickIso }));
+}
@@ -171,6 +171,30 @@ describe("quota-current overlay", () => {
     expect(cline.windows[0].percent).toBe(69); // dated read, not the aged gatherer value
   });
 
+  test("a retained entry is stale immediately, even inside the freshness bound (#1268)", () => {
+    // One tick is 15 min and the freshness bound is 20, so a retained
+    // entry's measuredAt alone would read fresh for ~5 min after the
+    // failed tick; the retention stamp is what surfaces the failure.
+    const at = iso(5 * 60_000);
+    const out = overlayQuotaCurrent(
+      baseView(),
+      file([
+        entry({
+          serviceId: "cline-pass:cline-pass",
+          measuredAt: at,
+          windows: [win("monthly", 71, at)],
+          retainedFromFailedTick: iso(1 * 60_000),
+        }),
+      ]),
+      NOW,
+    );
+    const cline = out.pools.find((p) => p.id === "cline-pass:cline-pass")!;
+    expect(cline.state).toBe("stale");
+    expect(cline.freshness).toEqual({ measuredAt: at, source: "openchamber-quota", state: "stale" });
+    expect(cline.windows[0].percent).toBe(69); // dated read, not the retained gatherer value
+    expect(out.gatherer?.providers.find((p) => p.serviceId === "cline-pass:cline-pass")?.state).toBe("stale");
+  });
+
   test("an ok entry dated past the future-skew tolerance is stale, windows untouched (#1541)", () => {
     for (const futureMs of [60 * 60_000, 30 * 24 * 60 * 60_000]) {
       const futureAt = iso(-futureMs);

@@ -3,6 +3,9 @@ import {
   entryFromFileExport,
   entryFromTurso,
   entryFromResult,
+  lastKnownEntries,
+  mapWithConcurrency,
+  markRetained,
   normalizeResetAt,
   recordsFromEntries,
   serviceIdFor,
@@ -105,6 +108,75 @@ describe("entryFromResult", () => {
   });
 });
 
+describe("mapWithConcurrency (#1268)", () => {
+  test("preserves order and never exceeds the in-flight bound", async () => {
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const out = await mapWithConcurrency([1, 2, 3, 4, 5, 6, 7], 3, async (n) => {
+      inFlight++;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await Bun.sleep(n % 3);
+      inFlight--;
+      return n * 10;
+    });
+    expect(out).toEqual([10, 20, 30, 40, 50, 60, 70]);
+    expect(maxInFlight).toBe(3);
+  });
+  test("a limit above the item count still completes; empty input maps to empty", async () => {
+    expect(await mapWithConcurrency([1, 2], 6, async (n) => n + 1)).toEqual([2, 3]);
+    expect(await mapWithConcurrency([], 6, async (n: number) => n)).toEqual([]);
+  });
+  test("a non-positive limit degrades to sequential, never unbounded", async () => {
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const out = await mapWithConcurrency([1, 2, 3], 0, async (n) => {
+      inFlight++;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await Bun.sleep(1);
+      inFlight--;
+      return n;
+    });
+    expect(out).toEqual([1, 2, 3]);
+    expect(maxInFlight).toBe(1);
+  });
+  test("a worker rejection rejects the map", async () => {
+    await expect(
+      mapWithConcurrency([1, 2], 2, async (n) => {
+        if (n === 2) throw new Error("boom");
+        return n;
+      }),
+    ).rejects.toThrow("boom");
+  });
+});
+
+describe("lastKnownEntries (#1268)", () => {
+  const tick = "2026-10-27T05:00:00.000Z";
+  const okEntry = entryFromResult(OK_RESULT, tick);
+  const errEntry = entryFromResult(
+    { providerId: "cline-pass", ok: false, configured: true, error: "timed out", usage: null },
+    tick,
+  );
+  const otherOk = entryFromResult({ ...OK_RESULT, providerId: "claude" }, tick);
+  test("returns the provider's ok entries verbatim, original measuredAt kept", () => {
+    const kept = lastKnownEntries([okEntry, errEntry, otherOk], "cline-pass");
+    expect(kept).toEqual([okEntry]);
+    expect(kept[0].measuredAt).toBe(okEntry.measuredAt);
+  });
+  test("error-only history retains nothing; unknown providers retain nothing", () => {
+    expect(lastKnownEntries([errEntry], "cline-pass")).toEqual([]);
+    expect(lastKnownEntries([okEntry], "nope")).toEqual([]);
+    expect(lastKnownEntries([], "cline-pass")).toEqual([]);
+  });
+  test("markRetained stamps copies with the failed tick; originals and measuredAt untouched", () => {
+    const stamped = markRetained([okEntry], "2026-10-27T05:15:00.000Z");
+    expect(stamped).toHaveLength(1);
+    expect(stamped[0].retainedFromFailedTick).toBe("2026-10-27T05:15:00.000Z");
+    expect(stamped[0].measuredAt).toBe(okEntry.measuredAt);
+    expect(stamped[0]).not.toBe(okEntry);
+    expect(okEntry.retainedFromFailedTick).toBeUndefined();
+  });
+});
+
 describe("recordsFromEntries", () => {
   test("a pct record per changed window; label-only windows yield none", () => {
     const entry = entryFromResult(OK_RESULT, "2026-10-27T05:00:00.000Z");