fix(aggregator): consume feed charges outside the saas gate (#3124)

pm-stack/aggregator/src/sources/live.ts

@@ -23,6 +23,7 @@ import { isSessionRecord, mergeRecords, recordFromSessionEntry } from "../regist
 import { buildAttentionFlags } from "../derive.ts";
 import { summarizeDns } from "../dns.ts";
 import { applyCloudflareExport, applySupabaseExport, extractCharges, extractPaygLines, extractSaasServices, extractSubscriptions } from "../saas-extract.ts";
+import { mergeChargeRows } from "../charges-collect.ts";
 import type { CloudflareExportDoc, SupabaseExportDoc } from "../saas-extract.ts";
 import { poolItemToGhItem } from "../projects.ts";
 import { INITIAL_STANDING_AUTO, sanitizeTierItemExtensions, withUserFallbacks } from "../tier-items.ts";
@@ -35,6 +36,7 @@ import {
   type AaFeedSection,
   type AttentionFlag,
   type BoardItem,
+  type ChargeRow,
   type Comment,
   type DnsSection,
   type GhItem,
@@ -866,6 +868,24 @@ export function liveSource(cfg: Config, opts?: { standingAuto?: () => Promise<bo
       const supabaseExport = await readFile(cfg.supabaseExportFile, "utf8")
         .then((text) => JSON.parse(text) as SupabaseExportDoc)
         .catch(() => null);
+      // Charges evidence + unmatched-charge flags (#3124): computed
+      // once, before the branch. The feed's charge data is not a
+      // byproduct of the saas section, so every return path below
+      // consumes it — previously both lived only inside the saas
+      // gate, and a feed with charges but no saas (or a quota-view
+      // fallback) silently dropped every auto-matched charge.
+      const feedCharges: ChargeRow[] = feed?.charges ?? [];
+      // Charges auto-matching (#1057): the feed carries only
+      // first-time unmatched SaaS/AI charges (the collector's
+      // surfaced-state guarantees exactly-once), so each becomes
+      // an attention flag here and never repeats on later feeds.
+      const feedChargeFlags: AttentionFlag[] = (feed?.chargesAttention ?? []).map((u) => ({
+        kind: "unmatched-charge",
+        label: `${u.merchant} — unrecognized SaaS/AI charge`,
+        detail: `$${u.amountUsd.toFixed(2)} on ${u.at} via ${u.source === "plaid" ? "Plaid" : "Gmail alert"} — no hub service matched; classify it or add the service`,
+        href: "/quota",
+        severity: u.amountUsd,
+      }));
       if (feed?.saas && feed.saas.length) {
         provenance.usage = `live (fleet feed · ${feedAgeText(feed.generatedAt)} · collected ${feed.generatedAt})`;
         // The Cloudflare live export overlays tracker prose on BOTH
@@ -877,27 +897,16 @@ export function liveSource(cfg: Config, opts?: { standingAuto?: () => Promise<bo
         const partial = {
           services: applySupabaseExport(applyCloudflareExport(feed.saas, cfExport), supabaseExport),
           payg: feed.payg ?? [],
-          charges: feed.charges ?? [],
+          charges: feedCharges,
           subscriptions: feed.subscriptions ?? [],
         };
-        // Charges auto-matching (#1057): the feed carries only
-        // first-time unmatched SaaS/AI charges (the collector's
-        // surfaced-state guarantees exactly-once), so each becomes
-        // an attention flag here and never repeats on later feeds.
         partial.services = await tursoOverlay(cfg, partial.services);
-        const chargeFlags: AttentionFlag[] = (feed.chargesAttention ?? []).map((u) => ({
-          kind: "unmatched-charge",
-          label: `${u.merchant} — unrecognized SaaS/AI charge`,
-          detail: `$${u.amountUsd.toFixed(2)} on ${u.at} via ${u.source === "plaid" ? "Plaid" : "Gmail alert"} — no hub service matched; classify it or add the service`,
-          href: "/quota",
-          severity: u.amountUsd,
-        }));
         return {
           ...partial,
           chargesNote: partial.charges.length
             ? undefined
             : "No charge evidence in the feed's records yet — rows appear as Plaid/Gmail receipt evidence is recorded (§2.4).",
-          attention: [...buildAttentionFlags(feed.quota, partial), ...chargeFlags],
+          attention: [...buildAttentionFlags(feed.quota, partial), ...feedChargeFlags],
           domains,
           history,
         };
@@ -910,10 +919,17 @@ export function liveSource(cfg: Config, opts?: { standingAuto?: () => Promise<bo
         const cfExport = await readFile(cfg.cloudflareExportFile, "utf8")
           .then((text) => JSON.parse(text) as CloudflareExportDoc)
           .catch(() => null);
+        // #3124: the feed's charges win/merge here too — prose
+        // extraction is the fallback for when the feed carries none,
+        // never a reason to drop the ones it does carry. The merge is
+        // charge-level (mergeChargeRows): the same charge recorded in
+        // both places differs in label/date/source strings, so
+        // full-row equality would display it twice; automatic
+        // evidence (the feed rows) takes precedence.
         const partial = {
           services: applySupabaseExport(applyCloudflareExport(extractSaasServices(doc), cfExport), supabaseExport),
           payg: extractPaygLines(doc, registryText),
-          charges: extractCharges(doc, registryText),
+          charges: mergeChargeRows(feedCharges, extractCharges(doc, registryText)),
           subscriptions: extractSubscriptions(doc),
         };
         partial.services = await tursoOverlay(cfg, partial.services);
@@ -923,14 +939,23 @@ export function liveSource(cfg: Config, opts?: { standingAuto?: () => Promise<bo
           chargesNote: partial.charges.length
             ? undefined
             : "No charge evidence in the local records yet — rows appear as Plaid/Gmail receipt evidence is recorded (§2.4).",
-          attention: buildAttentionFlags(feed?.quota ?? (await fix.quota()), partial),
+          attention: [...buildAttentionFlags(feed?.quota ?? (await fix.quota()), partial), ...feedChargeFlags],
           domains,
           history,
         };
       } catch {
         provenance.usage = "fixtures (no fleet feed hub sections and quota view unreadable)";
         const fallback = await fix.usage();
-        return { ...fallback, history: Object.keys(history).length ? history : fallback.history };
+        // #3124: even the fixtures fallback keeps the feed's charge
+        // evidence when a feed exists — fixtures stand in for the
+        // sections that are missing, not for the ones that arrived.
+        return {
+          ...fallback,
+          charges: feedCharges.length ? feedCharges : fallback.charges,
+          chargesNote: feedCharges.length ? undefined : fallback.chargesNote,
+          attention: [...fallback.attention, ...feedChargeFlags],
+          history: Object.keys(history).length ? history : fallback.history,
+        };
       }
     },
     async attention() {

pm-stack/aggregator/test/charges-feed-gate.test.ts

@@ -0,0 +1,134 @@
+// #3124 (queue task 4590): live source usage() must consume the feed's
+// charges + chargesAttention in EVERY return path, not only inside the
+// saas gate. A feed carrying charges but no saas section — e.g. when
+// the quota-view fallback serves usage — must still surface the charge
+// rows and the unmatched-charge attention flags.
+import { describe, expect, test } from "bun:test";
+import { mkdtemp, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { loadConfig } from "../src/config.ts";
+import { extractSaasServices } from "../src/saas-extract.ts";
+import type { ChargeRow } from "../src/model.ts";
+import type { UnmatchedCharge } from "../src/charges-match.ts";
+import { liveSource } from "../src/sources/live.ts";
+
+const FEED_CHARGE: ChargeRow = {
+  at: "2026-10-30",
+  label: "Mystery AI Inc",
+  amountUsd: 42.5,
+  method: "Visa •7535",
+  source: "plaid",
+};
+const FEED_UNMATCHED: UnmatchedCharge = {
+  merchant: "Mystery AI Inc",
+  normalized: "mystery ai",
+  amountUsd: 42.5,
+  at: "2026-10-30",
+  source: "plaid",
+};
+
+const VIEW = `
+## Summary (b) — subscriptions, infrastructure, account tooling
+
+| Service | Plan / price | Allowance sketch | Usage / status | Next date recorded |
+|---|---|---|---|---|
+| Supabase | Free plan (muse-dev) | DB + egress + storage | manual dashboard read needed | unknown |
+`;
+
+async function sourceWith(opts: {
+  feedOver: Record<string, unknown>;
+  quotaView?: string;
+}): Promise<ReturnType<typeof liveSource>> {
+  const dir = await mkdtemp(join(tmpdir(), "pm-charges-gate-"));
+  const feedFile = join(dir, "fleet-feed.json");
+  await writeFile(
+    feedFile,
+    JSON.stringify({
+      version: 1,
+      generatedAt: new Date().toISOString(),
+      quota: { pools: [], burn: [], choices: { hasHeadroom: [], burnBeforeReset: [], doNotStart: [] } },
+      attention: [],
+      ...opts.feedOver,
+    }),
+  );
+  const quotaViewFile = join(dir, "quota-view.md");
+  if (opts.quotaView !== undefined) await writeFile(quotaViewFile, opts.quotaView);
+  const cfg = loadConfig({
+    PM_AGG_SOURCE: "live",
+    PM_AGG_FEED_FILE: feedFile,
+    PM_AGG_QUOTA_VIEW: quotaViewFile,
+    PM_AGG_SUPABASE_EXPORT: join(dir, "no-supabase-export.json"),
+    PM_AGG_CLOUDFLARE_EXPORT: join(dir, "no-cloudflare-export.json"),
+    PM_AGG_REGISTRY_FILE: join(dir, "registry.json"),
+    PM_AGG_COMMENTS_FILE: join(dir, "comments.json"),
+    PM_AGG_SNAPSHOTS_DIR: join(dir, "snapshots"),
+    PM_AGG_OPENCHAMBER_STATE: join(dir, "no-state.json"),
+    PM_PROJECTOR_URL: "off",
+    KANEO_ADMIN_EMAIL: "",
+    KANEO_ADMIN_PASSWORD: "",
+    GRIST_API_KEY: "",
+  });
+  return liveSource(cfg);
+}
+
+describe("live source usage(): feed charges outside the saas gate (#3124)", () => {
+  test("quota-view fallback: feed charges + unmatched-charge flags survive with no saas section", async () => {
+    const src = await sourceWith({
+      feedOver: { charges: [FEED_CHARGE], chargesAttention: [FEED_UNMATCHED] },
+      quotaView: VIEW,
+    });
+    const usage = await src.usage();
+    expect(usage.charges).toContainEqual(FEED_CHARGE);
+    expect(usage.chargesNote).toBeUndefined();
+    const flags = usage.attention.filter((a) => a.kind === "unmatched-charge");
+    expect(flags).toHaveLength(1);
+    expect(flags[0]!.label).toContain("Mystery AI Inc");
+  });
+
+  test("fixtures fallback: feed charges are kept when the quota view is unreadable", async () => {
+    const src = await sourceWith({
+      feedOver: { charges: [FEED_CHARGE], chargesAttention: [FEED_UNMATCHED] },
+      // no quotaView file written -> the direct read throws -> fixtures fallback
+    });
+    const usage = await src.usage();
+    expect(usage.charges).toContainEqual(FEED_CHARGE);
+    expect(usage.chargesNote).toBeUndefined();
+    expect(usage.attention.filter((a) => a.kind === "unmatched-charge")).toHaveLength(1);
+  });
+
+  test("quota-view fallback: a charge recorded in both the feed and the prose appears once (feed row wins)", async () => {
+    // The prose extractor builds its own row for this charge (label
+    // "Meta PAYG spill", date 2026-10-25, descriptive source string);
+    // the feed row is the same charge with different strings. The
+    // charge-level merge must keep exactly one row — the feed's.
+    const feedRow: ChargeRow = {
+      at: "2026-10-27",
+      label: "Meta PAYG spill",
+      amountUsd: 42.5,
+      method: "Visa •7535",
+      source: "plaid",
+    };
+    const src = await sourceWith({
+      feedOver: { charges: [feedRow] },
+      quotaView: `${VIEW}\n\nVisa •7535 — $42.50 charged + $42.50 charged\n`,
+    });
+    const usage = await src.usage();
+    expect(usage.charges).toEqual([feedRow]);
+  });
+
+  test("feed-saas path unchanged: charges + flags come from the feed as before", async () => {
+    const src = await sourceWith({
+      feedOver: {
+        saas: extractSaasServices(VIEW),
+        charges: [FEED_CHARGE],
+        chargesAttention: [FEED_UNMATCHED],
+      },
+      quotaView: VIEW,
+    });
+    const usage = await src.usage();
+    expect(usage.charges).toEqual([FEED_CHARGE]);
+    expect(usage.services.some((s) => s.id === "supabase")).toBe(true);
+    expect(usage.attention.filter((a) => a.kind === "unmatched-charge")).toHaveLength(1);
+  });
+});