fix(aggregator): confirm surfaced charges after delivery

pm-stack/aggregator/scripts/collect-fleet-feed.ts

@@ -445,13 +445,9 @@ async function main(): Promise<void> {
       } catch {
         /* no state yet: everything unmatched is new */
       }
-      const { fresh, keys } = filterNewUnmatched(auto.unmatchedSaas, surfaced);
+      const { fresh } = filterNewUnmatched(auto.unmatchedSaas, surfaced);
       if (fresh.length) {
         chargesAttention = fresh;
-        await writeFile(
-          chargesSurfacedStatePath,
-          JSON.stringify({ version: 1, surfaced: keys, updatedAt: now.toISOString() }, null, 2),
-        );
       }
       console.log(
         `collector: charges auto-match — ${auto.rows.length} rows from ${candidates.length} candidates, ${fresh.length} new unmatched surfaced`,

pm-stack/aggregator/scripts/confirm-charges-surfaced.ts

@@ -0,0 +1,32 @@
+#!/usr/bin/env bun
+// Confirm only the charges in the exact feed successfully installed remotely.
+// The push script holds the state-specific flock across collection and confirmation.
+import { readFile, rename, unlink, writeFile } from "node:fs/promises";
+import { homedir } from "node:os";
+import { join } from "node:path";
+import { unmatchedKey } from "../src/charges-collect.ts";
+
+const feedPath = process.argv[2];
+if (!feedPath) throw new Error("usage: confirm-charges-surfaced.ts <delivered-feed>");
+const statePath = process.env.CHARGES_SURFACED_STATE_PATH ?? join(homedir(), "workspace/goals/keep-the-four-repos-healthy/hidden_files/charges-surfaced-state.json");
+const feed = JSON.parse(await readFile(feedPath, "utf8"));
+if (feed.chargesAttention !== undefined && !Array.isArray(feed.chargesAttention)) throw new Error("invalid chargesAttention");
+const attention = feed.chargesAttention ?? [];
+if (attention.length) {
+  let state: { surfaced?: unknown } = {};
+  try {
+    state = JSON.parse(await readFile(statePath, "utf8"));
+    if (!state || !Array.isArray(state.surfaced) || !state.surfaced.every((key) => typeof key === "string")) throw new Error("invalid surfaced state");
+  } catch (err) {
+    if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err;
+  }
+  const keys = new Set<string>((state.surfaced as string[] | undefined) ?? []);
+  for (const charge of attention) keys.add(unmatchedKey(charge));
+  const temp = `${statePath}.${process.pid}.${crypto.randomUUID()}.tmp`;
+  try {
+    await writeFile(temp, JSON.stringify({ version: 1, surfaced: [...keys], updatedAt: new Date().toISOString() }, null, 2), { flag: "wx" });
+    await rename(temp, statePath);
+  } finally {
+    await unlink(temp).catch(() => {});
+  }
+}

pm-stack/aggregator/scripts/push-fleet-feed.sh

@@ -32,7 +32,6 @@ export PATH
 
 cd "$(dirname "$0")/.."
 PKG="$PWD"
-FEED="$PKG/fleet-feed.json"
 LAPTOP=user@100.99.113.103
 DEST_DIR="~/pm-stack/aggregator"
 KNOWN_HOSTS="${PUSH_FLEET_FEED_KNOWN_HOSTS:-$PKG/scripts/laptop_known_hosts}"
@@ -59,9 +58,18 @@ print(m.group(1) if m else "")' 2>/dev/null || true)"
 fi
 export NO_PROXY="${NO_PROXY:-127.0.0.1,localhost,100.99.113.103}" no_proxy="${no_proxy:-127.0.0.1,localhost,100.99.113.103}"
 
+# Serialize the read/publish/confirm transaction for this state file. A
+# standalone collection never confirms anything. Each push owns its feed, so
+# another collector cannot replace the bytes being sent or acknowledged.
+STATE="${CHARGES_SURFACED_STATE_PATH:-$HOME/workspace/goals/keep-the-four-repos-healthy/hidden_files/charges-surfaced-state.json}"
+mkdir -p "$(dirname "$STATE")"
+exec 9>"$STATE.lock"
+flock 9
+FEED="$(mktemp "$PKG/.fleet-feed.XXXXXXXX.json")"
+trap 'rm -f "$FEED"' EXIT
 bun run scripts/collect-fleet-feed.ts --out "$FEED"
 
-exec python3 - "$FEED" "$LAPTOP" "$DEST_DIR" "$KNOWN_HOSTS" <<'PYEOF'
+python3 - "$FEED" "$LAPTOP" "$DEST_DIR" "$KNOWN_HOSTS" <<'PYEOF'
 import base64, os, re, socket, subprocess, sys, threading, urllib.parse
 
 feed, laptop, dest_dir, known_hosts = sys.argv[1], sys.argv[2], sys.argv[3], sys.argv[4]
@@ -182,18 +190,20 @@ ssh_opts = [
     "-o", "ControlMaster=no",
 ]
 dest = f"{laptop.split('@', 1)[0]}@127.0.0.1"
+remote_tmp = f"{dest_dir}/{os.path.basename(feed)}.tmp"
 try:
     subprocess.run(
         ["scp", "-P", str(local_port), *ssh_opts, feed,
-         f"{dest}:{dest_dir}/fleet-feed.json.tmp"],
+         f"{dest}:{remote_tmp}"],
         check=True,
     )
     subprocess.run(
         ["ssh", "-p", str(local_port), *ssh_opts, dest,
-         f"mv {dest_dir}/fleet-feed.json.tmp {dest_dir}/fleet-feed.json"],
+         f"mv {remote_tmp} {dest_dir}/fleet-feed.json"],
         check=True,
     )
 finally:
     listener.close()
 print("push-fleet-feed: feed pushed to laptop")
 PYEOF
+bun run scripts/confirm-charges-surfaced.ts "$FEED"

pm-stack/aggregator/test/charges-push.test.ts

@@ -0,0 +1,50 @@
+import { test, expect } from "bun:test";
+import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+// Explicit timeout: five subprocess push invocations can exceed the bun default 5s per-test timeout on slower hosts (Codex P2 on PR #3167).
+test("only a successfully installed feed confirms its charges", async () => {
+  const dir = await mkdtemp(join(tmpdir(), "charges-push-"));
+  const bin = join(dir, "bin");
+  const state = join(dir, "state.json");
+  const known = join(dir, "known-hosts");
+  const realBun = process.execPath;
+  // Resolve the real interpreter to an absolute path BEFORE the wrapper
+  // directory leads the child PATH; a bare "python3" wrapper would exec
+  // itself in a loop (Codex P2 on PR #3167).
+  const pythonOverride = process.env.TEST_PYTHON3;
+  const python = pythonOverride?.startsWith("/")
+    ? pythonOverride
+    : (Bun.which(pythonOverride ?? "python3") ?? pythonOverride ?? "python3");
+  const charge = { normalized: "example", amountUsd: 12, at: "2026-10-01" };
+  const script = join(import.meta.dir, "../scripts/push-fleet-feed.sh");
+  try {
+    await mkdir(bin);
+    await writeFile(known, "pinned\n");
+    // The fake collector selects from confirmed state, like the real collector.
+    await writeFile(join(bin, "bun"), `#!/usr/bin/env bash\nif [[ "$2" == scripts/collect-fleet-feed.ts ]]; then\n  exec '${realBun}' '${join(dir, "collect.ts")}' "$4"\nfi\nexec '${realBun}' "$@"\n`, { mode: 0o755 });
+    await writeFile(join(dir, "collect.ts"), `import {readFile,writeFile,appendFile} from 'node:fs/promises';\nlet keys=[]; try { keys=JSON.parse(await readFile(process.env.CHARGES_SURFACED_STATE_PATH,'utf8')).surfaced } catch {}\nconst fresh=!keys.includes('example|12.00|2026-10-01');\nawait appendFile(process.env.COLLECTION_LOG, fresh?'fresh\\n':'seen\\n');\nawait writeFile(process.argv[2],JSON.stringify({chargesAttention:fresh?[${JSON.stringify(charge)}]:[]}));\n`);
+    await writeFile(join(bin, "python3"), `#!/usr/bin/env bash\nexec '${python}' "$@"\n`, { mode: 0o755 });
+    await writeFile(join(bin, "scp"), `#!/usr/bin/env bash\n[[ "$FAIL_STEP" != scp ]]\n`, { mode: 0o755 });
+    await writeFile(join(bin, "ssh"), `#!/usr/bin/env bash\n[[ "$FAIL_STEP" != ssh ]]\n`, { mode: 0o755 });
+    const run = async (fail: string) => {
+      const proc = Bun.spawn(["bash", script], { env: { ...process.env, PATH: `${bin}:${process.env.PATH ?? "/usr/bin:/bin"}`, HOME: dir, HTTPS_PROXY: "http://user:pass@localhost:3128", CHARGES_SURFACED_STATE_PATH: state, COLLECTION_LOG: join(dir, "collections"), PUSH_FLEET_FEED_KNOWN_HOSTS: known, FAIL_STEP: fail }, stdout: "pipe", stderr: "pipe" });
+      const output = await new Response(proc.stderr).text();
+      return { code: await proc.exited, output };
+    };
+    for (const step of ["scp", "ssh"]) {
+      expect((await run(step)).code).not.toBe(0);
+      expect(await Bun.file(state).exists()).toBe(false);
+    }
+    await writeFile(state, JSON.stringify({ version: 1, surfaced: ["previous"] }));
+    const successes = await Promise.all([run(""), run("")]);
+    expect(successes).toEqual([{ code: 0, output: "" }, { code: 0, output: "" }]);
+    expect((await readFile(join(dir, "collections"), "utf8")).trim().split("\n")).toEqual(["fresh", "fresh", "fresh", "seen"]);
+    expect(JSON.parse(await readFile(state, "utf8")).surfaced).toEqual(["previous", "example|12.00|2026-10-01"]);
+    expect((await run("")).code).toBe(0);
+    expect(JSON.parse(await readFile(state, "utf8")).surfaced).toEqual(["previous", "example|12.00|2026-10-01"]);
+  } finally {
+    await rm(dir, { recursive: true, force: true });
+  }
+}, 30000);