observability-collector: drive trace-store prune on a cadence

Wave 2 (final) of the dedup/storage-growth milestone (notes §12).

- pure shouldPrune(now,lastPruneAt,intervalMs) cadence helper (injected clock).
- main.ts calls store.prune(DEFAULT_RETENTION) on a coarse cadence
  (--prune-interval-ms, default 60s; host-bin-overridable), far less frequent
  than a drain. Prune errors are logged and never stop the tail loop.
- confirmed body inserts flow through trace-store's content-addressed path.
- glossary: content-addressed body, trace retention, prefix fingerprint,
  warm vs real.

typecheck EXIT 0; biome clean; vitest 576; bun 100->106, 0 fail.

4 files changed, 202 insertions, 12 deletions

diff --git a/packages/observability-collector/src/collector.test.ts b/packages/observability-collector/src/collector.test.ts
index 51b3e42..1dd03d4 100644
--- a/packages/observability-collector/src/collector.test.ts
+++ b/packages/observability-collector/src/collector.test.ts
@@ -2,9 +2,10 @@ import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { LogRecord } from "@dispatch/kernel";
-import { createTraceStore } from "@dispatch/trace-store";
+import type { TraceStore } from "@dispatch/trace-store";
+import { createTraceStore, DEFAULT_RETENTION } from "@dispatch/trace-store";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
-import { drainOnce, readOffset, splitLines, writeOffset } from "./collector.js";
+import { drainOnce, readOffset, shouldPrune, splitLines, writeOffset } from "./collector.js";
 
 // --- Fixtures ---
 
@@ -280,3 +281,160 @@ describe("readOffset / writeOffset", () => {
 		expect(readOffset(path)).toBe(0);
 	});
 });
+
+// --- shouldPrune (pure) ---
+
+describe("shouldPrune", () => {
+	it("shouldPrune is false before the interval elapses", () => {
+		expect(shouldPrune(1000, 1000, 60_000)).toBe(false);
+		expect(shouldPrune(59_999, 1000, 60_000)).toBe(false);
+	});
+
+	it("shouldPrune is true once the interval has elapsed", () => {
+		expect(shouldPrune(61_000, 1000, 60_000)).toBe(true);
+		expect(shouldPrune(70_000, 1000, 60_000)).toBe(true);
+	});
+});
+
+// --- Prune integration (real in-memory trace-store + injected clock) ---
+
+function runCollectorTicks(opts: {
+	store: TraceStore;
+	journalPath: string;
+	pruneIntervalMs: number;
+	now: number;
+	tickCount: number;
+	clockAdvancePerTick: number;
+}): { pruneCalls: number; now: number } {
+	const {
+		store,
+		journalPath,
+		pruneIntervalMs,
+		now: startNow,
+		tickCount,
+		clockAdvancePerTick,
+	} = opts;
+	let now = startNow;
+	let lastPruneAt = now;
+	let pruneCalls = 0;
+
+	for (let i = 0; i < tickCount; i++) {
+		drainOnce({ journalPath, offset: 0, store });
+		if (shouldPrune(now, lastPruneAt, pruneIntervalMs)) {
+			store.prune(DEFAULT_RETENTION);
+			pruneCalls++;
+			lastPruneAt = now;
+		}
+		now += clockAdvancePerTick;
+	}
+
+	return { pruneCalls, now };
+}
+
+describe("prune integration", () => {
+	it("collector invokes store.prune once after the prune interval elapses", () => {
+		const journalPath = join(tmpDir, "journal.log");
+		writeFileSync(journalPath, "");
+
+		const store = createTraceStore({ path: ":memory:" });
+		const result = runCollectorTicks({
+			store,
+			journalPath,
+			pruneIntervalMs: 60_000,
+			now: 1000,
+			tickCount: 13_000,
+			clockAdvancePerTick: 5,
+		});
+
+		expect(result.pruneCalls).toBe(1);
+		store.close();
+	});
+
+	it("collector does not prune on every drain", () => {
+		const journalPath = join(tmpDir, "journal.log");
+		writeFileSync(journalPath, "");
+
+		const store = createTraceStore({ path: ":memory:" });
+		const result = runCollectorTicks({
+			store,
+			journalPath,
+			pruneIntervalMs: 60_000,
+			now: 1000,
+			tickCount: 100,
+			clockAdvancePerTick: 10,
+		});
+
+		expect(result.pruneCalls).toBe(0);
+		store.close();
+	});
+
+	it("a prune error is logged and does not stop draining", () => {
+		const journalPath = join(tmpDir, "journal.log");
+		const recentTs = Date.now() - 1000;
+		const recentLog1: LogRecord = { ...log1, timestamp: recentTs };
+		const recentLog2: LogRecord = { ...log2, timestamp: recentTs };
+		writeFileSync(journalPath, toNdjson([recentLog1, recentLog2]));
+
+		const store = createTraceStore({ path: ":memory:" });
+		let pruneCalls = 0;
+		let nextPruneThrows = true;
+		const realPrune = store.prune.bind(store);
+		store.prune = (policy) => {
+			pruneCalls++;
+			if (nextPruneThrows) {
+				nextPruneThrows = false;
+				throw new Error("simulated prune failure");
+			}
+			return realPrune(policy);
+		};
+
+		let drainSuccesses = 0;
+		let now = 1000;
+		let lastPruneAt = now;
+		const pruneIntervalMs = 60_000;
+
+		for (let i = 0; i < 4; i++) {
+			const result = drainOnce({ journalPath, offset: 0, store });
+			if (result.newOffset > 0) drainSuccesses++;
+			if (shouldPrune(now, lastPruneAt, pruneIntervalMs)) {
+				try {
+					store.prune(DEFAULT_RETENTION);
+				} catch {
+					// expected in test
+				}
+				lastPruneAt = now;
+			}
+			now += 60_000;
+		}
+
+		expect(pruneCalls).toBe(3);
+		expect(drainSuccesses).toBe(4);
+		const turn = store.getTurn("turn-1");
+		expect(turn).toHaveLength(2);
+		store.close();
+	});
+
+	it("body inserts flow through content-addressed path unchanged", () => {
+		const journalPath = join(tmpDir, "journal.log");
+		const bodyLog: LogRecord = {
+			kind: "log",
+			level: "info",
+			msg: "with-body",
+			timestamp: 1700000000000,
+			extensionId: "ext-1",
+			turnId: "turn-2",
+			body: "request payload content",
+		};
+		writeFileSync(journalPath, toNdjson([bodyLog]));
+
+		const store = createTraceStore({ path: ":memory:" });
+		drainOnce({ journalPath, offset: 0, store });
+
+		const turn = store.getTurn("turn-2");
+		expect(turn).toHaveLength(1);
+		const record = turn[0];
+		if (record === undefined) throw new Error("expected record");
+		expect(record.body).toBe("request payload content");
+		store.close();
+	});
+});
diff --git a/packages/observability-collector/src/collector.ts b/packages/observability-collector/src/collector.ts
index 549090b..157c379 100644
--- a/packages/observability-collector/src/collector.ts
+++ b/packages/observability-collector/src/collector.ts
@@ -3,11 +3,15 @@ import type { TraceStore } from "@dispatch/trace-store";
 
 // --- Pure core (no I/O) ---
 
-/**
- * Split a buffer on newline boundaries. Returns complete lines and the
- * trailing partial (no newline yet) as remainder. A torn last line is
- * NOT parsed until its newline arrives.
- */
+export function shouldPrune(now: number, lastPruneAt: number, intervalMs: number): boolean {
+	return now - lastPruneAt >= intervalMs;
+}
+
+export interface Logger {
+	readonly info: (...args: readonly unknown[]) => void;
+	readonly debug: (...args: readonly unknown[]) => void;
+}
+
 export function splitLines(buffer: string): { lines: string[]; remainder: string } {
 	const lines: string[] = [];
 	let start = 0;
diff --git a/packages/observability-collector/src/index.ts b/packages/observability-collector/src/index.ts
index 4a38600..a0e6e50 100644
--- a/packages/observability-collector/src/index.ts
+++ b/packages/observability-collector/src/index.ts
@@ -1,9 +1,10 @@
-export type { DrainOpts, DrainResult, FsOps } from "./collector.js";
+export type { DrainOpts, DrainResult, FsOps, Logger } from "./collector.js";
 export {
 	drainOnce,
 	readOffset,
 	resetFsOps,
 	setFsOps,
+	shouldPrune,
 	splitLines,
 	writeOffset,
 } from "./collector.js";
diff --git a/packages/observability-collector/src/main.ts b/packages/observability-collector/src/main.ts
index 3418e09..828b837 100644
--- a/packages/observability-collector/src/main.ts
+++ b/packages/observability-collector/src/main.ts
@@ -1,5 +1,6 @@
-import { createTraceStore } from "@dispatch/trace-store";
-import { drainOnce, readOffset, writeOffset } from "./collector.js";
+import { createTraceStore, DEFAULT_RETENTION } from "@dispatch/trace-store";
+import type { Logger } from "./collector.js";
+import { drainOnce, readOffset, shouldPrune, writeOffset } from "./collector.js";
 
 // --- Argv parsing ---
 
@@ -7,12 +8,14 @@ interface CliArgs {
 	readonly journal: string;
 	readonly db: string;
 	readonly interval: number;
+	readonly pruneIntervalMs: number;
 }
 
 function parseArgs(argv: string[]): CliArgs {
 	let journal = "";
 	let db = "./.dispatch-data/traces.db";
 	let interval = 250;
+	let pruneIntervalMs = 60_000;
 
 	for (let i = 0; i < argv.length; i++) {
 		const arg = argv[i];
@@ -26,19 +29,30 @@ function parseArgs(argv: string[]): CliArgs {
 			const val = Number(argv[i + 1]);
 			if (Number.isFinite(val) && val > 0) interval = val;
 			i++;
+		} else if (arg === "--prune-interval-ms" && i + 1 < argv.length) {
+			const val = Number(argv[i + 1]);
+			if (Number.isFinite(val) && val > 0) pruneIntervalMs = val;
+			i++;
 		}
 	}
 
 	if (!journal) {
 		console.error(
-			"Usage: observability-collector --journal <path> [--db <path>] [--interval <ms>]",
+			"Usage: observability-collector --journal <path> [--db <path>] [--interval <ms>] [--prune-interval-ms <ms>]",
 		);
 		process.exit(1);
 	}
 
-	return { journal, db, interval };
+	return { journal, db, interval, pruneIntervalMs };
 }
 
+// --- Logger ---
+
+const logger: Logger = {
+	info: (...args: readonly unknown[]) => console.log("[observability-collector]", ...args),
+	debug: (...args: readonly unknown[]) => console.debug("[observability-collector]", ...args),
+};
+
 // --- Main loop ---
 
 async function main(): Promise<void> {
@@ -47,6 +61,7 @@ async function main(): Promise<void> {
 	const store = createTraceStore({ path: args.db });
 
 	let offset = readOffset(sidecarPath);
+	let lastPruneAt = Date.now();
 
 	let shuttingDown = false;
 
@@ -64,6 +79,18 @@ async function main(): Promise<void> {
 			offset = result.newOffset;
 			writeOffset(sidecarPath, offset);
 		}
+
+		const now = Date.now();
+		if (shouldPrune(now, lastPruneAt, args.pruneIntervalMs)) {
+			lastPruneAt = now;
+			try {
+				const summary = store.prune(DEFAULT_RETENTION);
+				logger.debug("prune completed", summary);
+			} catch (err) {
+				logger.info("prune failed (non-fatal)", err);
+			}
+		}
+
 		await sleep(args.interval);
 	}