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
27 changes: 27 additions & 0 deletions pm-stack/aggregator/src/feed.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import type { UnmatchedCharge } from "./charges-match.ts";
import type { CheckoutDrift } from "./checkout-drift.ts";
import { isProjectPoolItem } from "./projects.ts";
import { isRepoCensus, type RepoCensus } from "./coverage.ts";
import { isSessionRecord } from "./registry.ts";
import {
arrayOf,
isFiniteNumber,
Expand Down Expand Up @@ -60,6 +61,9 @@ export interface FleetFeed {
* only when at least one row (or a non-array section) was dropped;
* absent on raw parsed feeds and on clean loads. */
subscriptionsDropped?: number;
/** Registry rows loadFeed's sanitization dropped (#2539), set only
* when at least one row (or a non-array section) was dropped. */
registryDropped?: number;
/** Rows loadFeed's charges sanitization dropped (#2445), set only
* when drops occurred; absent on raw parsed feeds and clean loads. */
chargesDropped?: number;
Expand Down Expand Up @@ -637,6 +641,24 @@ export function sanitizeSubscriptions(v: unknown): SubscriptionSanitization {
return { rows, dropped: v.length - rows.length };
}

export interface RegistrySanitization {
/** Safe registry rows, or undefined when the optional section is absent. */
rows?: SessionRecord[];
dropped: number;
}

/** Drop malformed optional registry rows at the shared loadFeed boundary
* (#2539), so registry() and coverage() consume the same safe records.
* An absent section stays absent; a non-array section becomes an empty
* registry and counts as one dropped section; malformed array entries are
* dropped individually. */
export function sanitizeRegistry(v: unknown): RegistrySanitization {
if (v === undefined) return { dropped: 0 };
if (!Array.isArray(v)) return { rows: [], dropped: 1 };
const rows = v.filter(isSessionRecord);
return { rows, dropped: v.length - rows.length };
}

/** A PAYG line safe for both the attention derivation and the quota renderer. */
export function isPaygLine(v: unknown): v is PaygLine {
if (!isRecord(v)) return false;
Expand Down Expand Up @@ -699,6 +721,9 @@ export async function loadFeed(path: string): Promise<FleetFeed | null> {
const charges =
parsed.chargesAttention === undefined ? undefined : arrayOf(isUnmatchedCharge)(parsed.chargesAttention);
const subs = sanitizeSubscriptions(parsed.subscriptions);
// Registry rows feed both the session registry and coverage, so sanitize
// once here rather than letting the two consumers see different inputs.
const registry = sanitizeRegistry(parsed.registry);
const payg = sanitizePayg(parsed.payg);
// Charges rows get the same drop-and-count treatment at this same
// shared gate (#2445): the section is optional, so absence stays
Expand Down Expand Up @@ -730,6 +755,8 @@ export async function loadFeed(path: string): Promise<FleetFeed | null> {
...(charges && charges.dropped > 0 ? { chargesAttentionDropped: charges.dropped } : {}),
subscriptions: subs.rows,
...(subs.dropped > 0 ? { subscriptionsDropped: subs.dropped } : {}),
...(registry.rows !== undefined ? { registry: registry.rows } : {}),
...(registry.dropped > 0 ? { registryDropped: registry.dropped } : {}),
...(payg.rows !== undefined ? { payg: payg.rows } : {}),
...(payg.dropped > 0 ? { paygDropped: payg.dropped } : {}),
...chargesPart,
Expand Down
45 changes: 45 additions & 0 deletions pm-stack/aggregator/test/feed.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import {
sanitizeCollectionError,
sanitizeQuotaView,
sanitizeFeedMetadata,
sanitizeRegistry,
sanitizeSubscriptions,
} from "../src/feed.ts";
import { buildAttentionFlags } from "../src/derive.ts";
Expand Down Expand Up @@ -403,6 +404,50 @@ describe("feed subscriptions sanitization (#2446)", () => {
});
});

describe("feed registry sanitization (#2539)", () => {
const RECORD = { id: "ses-1", source: "claude-code", label: "Muse run" };
const feedWithRegistry = (registry: unknown) => ({
version: 1,
generatedAt: "2026-10-01T00:00:00Z",
quota: { pools: [], burn: [], choices: { hasHeadroom: [], burnBeforeReset: [], doNotStart: [] } },
attention: [],
registry,
});
const loadWithRegistry = async (registry: unknown) => {
const dir = mkdtempSync(join(tmpdir(), "agg-feed-registry-"));
const file = join(dir, "feed.json");
writeFileSync(file, JSON.stringify(feedWithRegistry(registry)));
return loadFeed(file);
};

test("preserves an absent section and a well-formed registry", async () => {
expect(sanitizeRegistry(undefined)).toEqual({ dropped: 0 });
const absent = await loadWithRegistry(undefined);
expect(absent?.registry).toBeUndefined();
expect(absent?.registryDropped).toBeUndefined();

expect(sanitizeRegistry([RECORD])).toEqual({ rows: [RECORD], dropped: 0 });
const clean = await loadWithRegistry([RECORD]);
expect(clean?.registry).toEqual([RECORD]);
expect(clean?.registryDropped).toBeUndefined();
expect(clean?.quota.pools).toEqual([]);
});

test("drops malformed containers and rows without rejecting the feed", async () => {
for (const malformed of ["bad", { registry: "bad" }, 42, [null, RECORD]]) {
const loaded = await loadWithRegistry(malformed);
expect(loaded).not.toBeNull();
expect(loaded?.registry).toEqual(Array.isArray(malformed) ? [RECORD] : []);
expect(loaded?.registryDropped).toBe(1);
expect(loaded?.quota.pools).toEqual([]);
}
expect(sanitizeRegistry("bad")).toEqual({ rows: [], dropped: 1 });
expect(sanitizeRegistry({ registry: "bad" })).toEqual({ rows: [], dropped: 1 });
expect(sanitizeRegistry(42)).toEqual({ rows: [], dropped: 1 });
expect(sanitizeRegistry([null])).toEqual({ rows: [], dropped: 1 });
});
});

describe("feed PAYG sanitization (#2450)", () => {
const PAYG: PaygLine = {
id: "meta-payg",
Expand Down
24 changes: 24 additions & 0 deletions pm-stack/aggregator/test/orchestration.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -508,4 +508,28 @@ describe("live source diagnostics + provenance detail (#1139/#1138)", () => {
expect(facet?.countLabel).toBe("session records");
expect(facet?.state).toBe("live");
});

test("registry and coverage share sanitized optional feed rows (#2539)", async () => {
for (const registry of ["bad", { registry: "bad" }, 42, [null]]) {
const src = await sourceWithFeed({ registry });
const records = await src.registry();
expect(Array.isArray(records)).toBe(true);

const coverage = await src.coverage();
const muse = coverage.find((row) => row.source === "claude-code");
expect(muse?.wired).toBe(false);
expect(muse?.detail).toContain("no launch records");
}

const clean = await sourceWithFeed({
registry: [
{ id: "ses-muse-1", source: "claude-code", label: "Muse launch" },
{ id: "ses-opencode-1", source: "opencode-sandbox", label: "OpenCode session" },
],
});
expect(await clean.registry()).toHaveLength(2);
const muse = (await clean.coverage()).find((row) => row.source === "claude-code");
expect(muse?.wired).toBe(true);
expect(muse?.detail).toContain("1 recorded");
});
});
Loading