Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 142 additions & 5 deletions pm-stack/aggregator/src/feed.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,10 @@ import { readFile } from "node:fs/promises";
import type { UnmatchedCharge } from "./charges-match.ts";
import type { CheckoutDrift } from "./checkout-drift.ts";
import { isProjectPoolItem } from "./projects.ts";
import { isIsoDate, isTimestampNotFuture } from "./validate.ts";
import { arrayOf, isFiniteNumber, isIsoDate, isRecord, isStringArray, isTimestampNotFuture } from "./validate.ts";
import type {
AaFeedSection,
BurnSegment,
ChargeRow,
DnsSection,
GoalMeasurement,
Expand All @@ -18,7 +19,9 @@ import type {
OrchestrationSection,
PaygLine,
ProjectPoolItem,
QuotaPool,
QuotaView,
QuotaWindow,
SaasService,
SessionFeedEntry,
SessionRecord,
Expand All @@ -34,6 +37,9 @@ export interface FleetFeed {
* in feeds written before the drift check existed. */
checkout?: CheckoutDrift;
quota: QuotaView;
/** Element drops loadFeed's quota sanitization made (#2429), set
* only on feeds returned by loadFeed; absent on raw parsed feeds. */
quotaDropped?: { pools: number; windows: number; burn: number };
/** Sandbox-server + Jules sessions, pre-classification (render classifies). */
attention: SessionFeedEntry[];
/** Jules keeper totals (slice 2); absent when the keeper state is unreadable. */
Expand Down Expand Up @@ -247,25 +253,156 @@ export function poolFromFeed(feed: Partial<FleetFeed> | null | undefined): Proje
return raw.filter(isProjectPoolItem);
}

/** One quota window as the renderers dereference it (#2429): `limit`
* is read arithmetically by windowPercent, `used` is rendered, and a
* non-object element crashes both renderQuota and buildAttentionFlags. */
export function isQuotaWindow(v: unknown): v is QuotaWindow {
if (!isRecord(v)) return false;
return (
typeof v.name === "string" &&
(v.unit === "requests" || v.unit === "tokens" || v.unit === "usd" || v.unit === "messages" || v.unit === "other") &&
(v.limit === null || isFiniteNumber(v.limit)) &&
isFiniteNumber(v.used) &&
(v.resetsAt === null || typeof v.resetsAt === "string") &&
(v.percent === undefined || isFiniteNumber(v.percent))
);
}

const QUOTA_STATES = ["healthy", "near", "walled", "error", "stale"];

/** One quota pool whose own shape is safe to hand to the renderers
* (#2429): an object with a `windows` ARRAY — a non-array windows value
* must never be iterated as a collection of windows (it fabricated
* attention flags). Window ELEMENTS are filtered by sanitizeQuotaView,
* so a pool survives one bad window. Credential rows, when present,
* must at least be records holding a windows array; their elements are
* filtered the same way. */
export function isQuotaPool(v: unknown): v is QuotaPool {
if (!isRecord(v)) return false;
if (typeof v.id !== "string" || typeof v.provider !== "string" || typeof v.plan !== "string") return false;
if (typeof v.state !== "string" || !QUOTA_STATES.includes(v.state)) return false;
if (!Array.isArray(v.windows)) return false;
if (v.credentials !== undefined) {
if (!Array.isArray(v.credentials)) return false;
for (const row of v.credentials) {
if (!isRecord(row) || !Array.isArray(row.windows)) return false;
}
}
return true;
}

/** One historical burn segment as renderQuota maps it (#2429). */
export function isBurnSegment(v: unknown): v is BurnSegment {
if (!isRecord(v)) return false;
return (
typeof v.provider === "string" &&
typeof v.pool === "string" &&
typeof v.from === "string" &&
typeof v.to === "string" &&
isFiniteNumber(v.tokens)
);
}

/** The choices block renderQuota dereferences (hasHeadroom et al.). */
function isQuotaChoices(v: unknown): boolean {
if (!isRecord(v)) return false;
return isStringArray(v.hasHeadroom) && isStringArray(v.burnBeforeReset) && isStringArray(v.doNotStart);
}

export interface QuotaSanitization {
view: QuotaView;
droppedPools: number;
droppedWindows: number;
droppedBurn: number;
}

/** Drop-and-count sanitization of a gate-passed quota section (#2429,
* following poolFromFeed's per-element pattern, built on the shared
* arrayOf primitive from #2577): malformed pools and burn segments
* drop individually, malformed windows drop out of their pool, and the
* well-formed remainder renders. The counts ride out on the result so
* the caller surfaces them, never swallows them. */
export function sanitizeQuotaView(quota: QuotaView): QuotaSanitization {
const pools = arrayOf(isQuotaPool)(quota.pools);
let droppedWindows = 0;
const keptPools: QuotaPool[] = (pools?.values ?? []).map((pool) => {
const windows = arrayOf(isQuotaWindow)(pool.windows);
droppedWindows += windows?.dropped ?? 0;
let out: QuotaPool = { ...pool, windows: windows?.values ?? [] };
if (pool.credentials !== undefined) {
const rows = pool.credentials.map((row) => {
const rowWindows = arrayOf(isQuotaWindow)(row.windows);
droppedWindows += rowWindows?.dropped ?? 0;
return { ...row, windows: rowWindows?.values ?? [] };
});
out = { ...out, credentials: rows };
}
return out;
});
const burn = arrayOf(isBurnSegment)(quota.burn);
const droppedPools = pools?.dropped ?? 0;
const droppedBurn = burn?.dropped ?? 0;
// Guidance must describe the surviving usable pools only: the
// retained choices were computed over the pre-sanitization pools,
// and computeChoices() (quota-guidance.ts) gives no guidance for a
// pool with no windows — so filter every list to kept pools that
// still have windows. A pool whose data was discarded must not
// stay recommended under hasHeadroom or burnBeforeReset.
const usable = new Set(keptPools.filter((p) => p.windows.length > 0).map((p) => p.id));
const choices = {
hasHeadroom: quota.choices.hasHeadroom.filter((id) => usable.has(id)),
burnBeforeReset: quota.choices.burnBeforeReset.filter((id) => usable.has(id)),
doNotStart: quota.choices.doNotStart.filter((id) => usable.has(id)),
};
const view: QuotaView = { ...quota, pools: keptPools, burn: burn?.values ?? [], choices };
// Surface the loss on the view itself (its consumers project the
// view, e.g. /api/quota.json), not only on the feed wrapper.
if (droppedPools + droppedWindows + droppedBurn > 0) {
view.dropped = { pools: droppedPools, windows: droppedWindows, burn: droppedBurn };
}
return {
view,
droppedPools,
droppedWindows,
droppedBurn,
};
}

export function isFleetFeed(v: unknown): v is FleetFeed {
if (typeof v !== "object" || v === null) return false;
const f = v as Partial<FleetFeed>;
if (!(typeof f.quota === "object" && f.quota !== null)) return false;
const q = f.quota as QuotaView;
return (
f.version === 1 &&
// A generatedAt implausibly in the future is invalid, not
// "maximally fresh" (#2242, via the shared primitive).
isTimestampNotFuture(f.generatedAt) &&
typeof f.quota === "object" &&
f.quota !== null &&
Array.isArray((f.quota as QuotaView).pools) &&
Array.isArray(q.pools) &&
// burn and choices are mandatory in QuotaView and dereferenced
// unguarded by renderQuota (#2429): a feed without them is an old
// collector's shape and is rejected whole, like other invalid
// feeds, rather than crashing /quota.
Array.isArray(q.burn) &&
isQuotaChoices(q.choices) &&
Array.isArray(f.attention)
);
}

export async function loadFeed(path: string): Promise<FleetFeed | null> {
try {
const parsed: unknown = JSON.parse(await readFile(path, "utf8"));
return isFleetFeed(parsed) ? parsed : null;
if (!isFleetFeed(parsed)) return null;
// The gate pins the section-level shape; element-level damage is
// dropped here so every loadFeed consumer (live source, server)
// sees only renderer-safe pools/windows (#2429). Drop counts are
// attached to the feed so they are visible, not swallowed.
const san = sanitizeQuotaView(parsed.quota);
return {
...parsed,
quota: san.view,
quotaDropped: { pools: san.droppedPools, windows: san.droppedWindows, burn: san.droppedBurn },
Comment thread
dezren39 marked this conversation as resolved.
Resolved
};
} catch {
return null;
}
Expand Down
6 changes: 6 additions & 0 deletions pm-stack/aggregator/src/model.ts
Original file line number Diff line number Diff line change
Expand Up @@ -325,6 +325,12 @@ export interface QuotaView {
/** Gatherer overlay provenance (#1151); absent when no
* quota-current.json was read. Additive for existing consumers. */
gatherer?: GathererSummary;
/** Feed-load sanitization counts (#2429): present only when the
* loader dropped malformed pools/windows/burn segments, so an
* incomplete read is distinguishable from a successful empty one
* in the public payload (/api/quota.json projects this view).
* Additive for existing consumers. */
dropped?: { pools: number; windows: number; burn: number };
burn: BurnSegment[];
/** Guidance for choosing the next pool (burn order + reset rules). */
choices: {
Expand Down
4 changes: 4 additions & 0 deletions pm-stack/aggregator/src/sources/live.ts
Original file line number Diff line number Diff line change
Expand Up @@ -816,6 +816,10 @@ export function liveSource(cfg: Config, opts?: { standingAuto?: () => Promise<bo
if (feed) {
provenance.quota = `live (fleet feed · ${feedAgeText(feed.generatedAt)} · collected ${feed.generatedAt})`;
view = feed.quota;
const dropped = feed.quotaDropped;
if (dropped && dropped.pools + dropped.windows + dropped.burn > 0) {
provenance.quota += ` · sanitized at load: dropped ${dropped.pools} pool(s), ${dropped.windows} window(s), ${dropped.burn} burn segment(s) (#2429)`;
}
} else {
provenance.quota = "fixtures (no fleet feed file yet)";
view = await fix.quota();
Expand Down
102 changes: 100 additions & 2 deletions pm-stack/aggregator/test/feed.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,10 @@
import { describe, expect, test } from "bun:test";
import { isFleetFeed, isProjectPoolSection, poolFromFeed, projectPoolSectionProblem } from "../src/feed.ts";
import { isFleetFeed, isProjectPoolSection, loadFeed, poolFromFeed, projectPoolSectionProblem, sanitizeQuotaView } from "../src/feed.ts";
import { mkdtempSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { isProjectPoolItem, parseProjectItems } from "../src/projects.ts";
import type { QuotaPool, QuotaView, QuotaWindow } from "../src/model.ts";

// Minimal valid pool item: edges entirely absent (the missing-edge
// shape must validate cleanly — absence is data, not an error).
Expand Down Expand Up @@ -125,7 +129,7 @@ describe("isFleetFeed generatedAt stamps (#2578 Phase 2, #2242)", () => {
const feed = (generatedAt: string) => ({
version: 1,
generatedAt,
quota: { pools: [] },
quota: { pools: [], burn: [], choices: { hasHeadroom: [], burnBeforeReset: [], doNotStart: [] } },
attention: [],
});

Expand All @@ -139,3 +143,97 @@ describe("isFleetFeed generatedAt stamps (#2578 Phase 2, #2242)", () => {
expect(isFleetFeed(feed(future))).toBe(false);
});
});

describe("feed quota section gate + sanitization (#2429)", () => {
const WINDOW: QuotaWindow = { name: "5h", unit: "tokens", limit: 1000, used: 250, resetsAt: "2026-10-29T05:00:00Z", percent: 25 };
const POOL: QuotaPool = { id: "opencode-go", provider: "opencode", plan: "go", state: "healthy", windows: [WINDOW] };
const CHOICES = { hasHeadroom: ["opencode-go"], burnBeforeReset: [], doNotStart: [] };
const feedWith = (quota: unknown) => ({
version: 1,
generatedAt: "2026-10-01T00:00:00Z",
quota,
attention: [],
});
const quotaWith = (pools: unknown, extra: Record<string, unknown> = {}) => ({
pools,
burn: [],
choices: CHOICES,
...extra,
});

test("a feed missing burn or choices is rejected whole (old-collector shape)", () => {
expect(isFleetFeed(feedWith({ pools: [] }))).toBe(false);
expect(isFleetFeed(feedWith({ pools: [], burn: [] }))).toBe(false);
expect(isFleetFeed(feedWith(quotaWith([])))).toBe(true);
});

test("pools: [null] drops the pool, keeps the feed", () => {
const out = sanitizeQuotaView(quotaWith([null, POOL]) as never);
expect(out.droppedPools).toBe(1);
expect(out.view.pools).toEqual([POOL]);
});

test("a pool missing windows drops; a non-array windows is never iterated", () => {
const noWindows = { id: "a", provider: "p", plan: "x", state: "near" };
const stringWindows = { ...POOL, id: "b", windows: "soon" };
const out = sanitizeQuotaView(quotaWith([noWindows, stringWindows, POOL]) as never);
expect(out.droppedPools).toBe(2);
expect(out.view.pools).toEqual([POOL]);
});

test("windows: [null] drops the bad window, keeps pool and good windows", () => {
const pool = { ...POOL, windows: [null, WINDOW, { bogus: true }] };
const out = sanitizeQuotaView(quotaWith([pool]) as never);
expect(out.droppedPools).toBe(0);
expect(out.droppedWindows).toBe(2);
expect(out.view.pools[0]?.windows).toEqual([WINDOW]);
});

test("malformed burn segments drop individually", () => {
const seg = { provider: "opencode", pool: "go", from: "2026-10-01T00:00:00Z", to: "2026-10-01T01:00:00Z", tokens: 42 };
const out = sanitizeQuotaView(quotaWith([POOL], { burn: [seg, null, { provider: "x" }] }) as never);
expect(out.droppedBurn).toBe(2);
expect(out.view.burn).toEqual([seg]);
});

test("a fully well-formed quota view is unchanged", () => {
const q = quotaWith([POOL], { burn: [] }) as unknown as QuotaView;
const out = sanitizeQuotaView(q);
expect(out.droppedPools + out.droppedWindows + out.droppedBurn).toBe(0);
expect(out.view).toEqual(q);
});

test("choices are reconciled to surviving pools with usable windows", () => {
const noWindows = { ...POOL, windows: [null] };
const out = sanitizeQuotaView(quotaWith([noWindows]) as never);
expect(out.view.pools.length).toBe(1);
expect(out.view.choices).toEqual({ hasHeadroom: [], burnBeforeReset: [], doNotStart: [] });
expect(out.view.dropped).toEqual({ pools: 0, windows: 1, burn: 0 });
});

test("a dropped pool leaves the retained choices and surfaces dropped on the view", () => {
const out = sanitizeQuotaView(quotaWith([null, POOL]) as never);
expect(out.view.choices.hasHeadroom).toEqual(["opencode-go"]);
expect(out.view.dropped).toEqual({ pools: 1, windows: 0, burn: 0 });
});

test("loadFeed returns the sanitized remainder with drop counts attached", async () => {
const dir = mkdtempSync(join(tmpdir(), "agg-feed-quota-"));
const file = join(dir, "feed.json");
const pool = { ...POOL, windows: [WINDOW, null] };
writeFileSync(file, JSON.stringify(feedWith(quotaWith([null, pool]))));
const loaded = await loadFeed(file);
expect(loaded).not.toBeNull();
expect(loaded?.quota.pools.length).toBe(1);
expect(loaded?.quota.pools[0]?.windows).toEqual([WINDOW]);
expect(loaded?.quotaDropped).toEqual({ pools: 1, windows: 1, burn: 0 });
expect(loaded?.quota.dropped).toEqual({ pools: 1, windows: 1, burn: 0 });
});

test("loadFeed rejects a feed missing burn/choices (fixture-fallback path)", async () => {
const dir = mkdtempSync(join(tmpdir(), "agg-feed-quota-"));
const file = join(dir, "feed.json");
writeFileSync(file, JSON.stringify(feedWith({ pools: [POOL] })));
expect(await loadFeed(file)).toBeNull();
});
});
2 changes: 1 addition & 1 deletion pm-stack/aggregator/test/validate.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,7 @@ describe("feed gate: isFleetFeed uses the strict date (#2577 proof consumer)", (
const feed = (generatedAt: unknown) => ({
version: 1,
generatedAt,
quota: { pools: [] },
quota: { pools: [], burn: [], choices: { hasHeadroom: [], burnBeforeReset: [], doNotStart: [] } },
attention: [],
});
test("a genuine feed passes", () => {
Expand Down