Add bounded database response size diagnostics (#17504)

Count database response value sizes by public caller and static table for one explicit UTC window of at most 60 seconds. Keep diagnostics off by default and exclude SQL, parameters, returned values, errors, and credentials from output. Preserve lazy execution, transaction boundaries, and no-replay behavior.

Verification: full build and recursive typecheck; 63 focused database and PostgreSQL tests; eight physical disconnect tests with capture enabled.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -320,6 +320,31 @@ DATABASE_MAX_LIFETIME_SECONDS=1800   # recycle a pooled connection after this lo
 DATABASE_APPLICATION_NAME=paperclip  # application_name in pg_stat_activity; default: paperclip
 ```
 
+### Bounded response diagnostics
+
+`PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW=<start>/<end>` enables one explicit UTC
+window. Both timestamps use ISO format ending in `Z`. The window lasts at most
+60 seconds and starts within 30 minutes. Unset, invalid, and expired windows
+leave database construction unchanged. An expired window stays inactive after a
+restart; remove the setting after the investigation.
+
+The diagnostic measures awaited Postgres.js `unsafe` responses before Drizzle
+column decoding. It retains the Drizzle builder caller for normal, prepared,
+transaction, and dedicated-pool execution. It issues no queries.
+It logs the public server caller locations, static schema table, operation,
+query/row counts, and response value sizes. It emits at most 128 aggregate
+groups and a summary at the deadline. It never logs SQL, parameters, result
+values, connection strings, or query errors. In-flight queries continue normally.
+
+Look for `paperclip.db_response_capture.entry.v1` and
+`paperclip.db_response_capture.summary.v1` in the process logs. Tagged raw-driver calls, streams, unawaited raw-driver
+dispatch, and protocol overhead are not measured. Relational queries use a driver-level caller fallback.
+Queries still in flight at the deadline are reported separately. Sizes describe
+driver values before column decoding, not billed network bytes. Complex parsed
+values use JSON size estimates, counted in `estimatedValues`. Reconcile them with
+provider metering and socket measurements. Serialization adds temporary CPU and memory work;
+measure the final transfer rate after the window expires.
+
 ### Push the schema
 
 ```sh

packages/db/src/client.ts

@@ -4,6 +4,7 @@ import { readFile, readdir } from "node:fs/promises";
 import { fileURLToPath } from "node:url";
 import postgres from "postgres";
 import { installDatabaseWorkSignals } from "./work-signals.js";
+import { installDatabaseResponseCapture } from "./response-capture.js";
 import * as schema from "./schema/index.js";
 
 const MIGRATIONS_FOLDER = fileURLToPath(new URL("./migrations", import.meta.url));
@@ -307,6 +308,7 @@ function createDbClient(url: string, resolved: DatabaseClientOptions, owner?: ob
   // message cannot establish that replay is safe. Leave retries to callers
   // that know the complete operation is idempotent.
   const db = installDatabaseWorkSignals(drizzlePg(sql, { schema }), owner);
+  installDatabaseResponseCapture(db, schema);
   dedicatedDbFactories.set(db, () => createDbClient(url, {
     ...resolved, maxConnections: 1, applicationName: "paperclip-workspace-finalization-lock",
   }, db));

packages/db/src/response-capture-postgres.test.ts

@@ -0,0 +1,190 @@
+import { randomUUID } from "node:crypto";
+import { compileFunction } from "node:vm";
+import { eq, sql } from "drizzle-orm";
+import { pgTable, text } from "drizzle-orm/pg-core";
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import { createDb, withDedicatedDbConnection } from "./client.js";
+import { createDatabaseResponseCapture, databaseResponseCaptureFromEnv, installDatabaseResponseCapture } from "./response-capture.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js";
+import { encodeRunEventPayload, runEventPayload } from "./run-event-payload.js";
+import { databaseWorkPending, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js";
+
+const table = pgTable("response_capture_fixture", { id: text("id").primaryKey(), payload: text("payload").notNull() });
+const encodedTable = pgTable("response_capture_encoded_fixture", { id: text("id").primaryKey(), payload: runEventPayload("payload") });
+const support = await getEmbeddedPostgresTestSupport();
+describe.skipIf(!support.supported)("database response capture PostgreSQL behavior", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  const clients: Array<ReturnType<typeof createDb>> = [];
+  const captures: Array<ReturnType<typeof createDatabaseResponseCapture>> = [];
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("response-capture-");
+    const db = createDb(temporary.connectionString);
+    try {
+      await db.execute(sql`create table response_capture_fixture (id text primary key, payload text not null)`);
+      await db.execute(sql`create table response_capture_encoded_fixture (id text primary key, payload jsonb not null)`);
+    }
+    finally { await db.$client.end(); }
+  }, 90_000);
+  afterEach(async () => {
+    for (const capture of captures.splice(0)) capture.flush();
+    for (const client of clients.splice(0)) await client.$client.end();
+    vi.restoreAllMocks(); vi.unstubAllEnvs();
+  });
+  afterAll(async () => { await temporary?.cleanup(); });
+  function fixture() {
+    const events: Array<Record<string, unknown>> = [];
+    const now = Date.now();
+    const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
+    const db = createDb(temporary.connectionString);
+    clients.push(db); captures.push(capture);
+    expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db);
+    expect(installDatabaseResponseCapture(db, { table, encodedTable }, capture)).toBe(db);
+    return { db, capture, events };
+  }
+  function summary(events: Array<Record<string, unknown>>) { return events.find(event => event.event === "paperclip.db_response_capture.summary.v1")!; }
+
+  it("keeps rows and builder identity through lazy select and write returning", async () => {
+    const f = fixture(), id = randomUUID(), payload = "sentinel-result-and-parameter-value🙂\n\"";
+    const insert = f.db.insert(table).values({ id, payload });
+    expect(insert.returning()).toBe(insert);
+    expect(await insert).toEqual([{ id, payload }]);
+    const query = f.db.select().from(table);
+    expect(query.where(eq(table.id, id))).toBe(query);
+    expect(await query).toEqual([{ id, payload }]);
+    const updated = await f.db.update(table).set({ payload: "changed" }).where(eq(table.id, id)).returning();
+    expect(updated).toEqual([{ id, payload: "changed" }]);
+    expect(await f.db.delete(table).where(eq(table.id, id)).returning()).toEqual([{ id, payload: "changed" }]);
+    f.capture.flush();
+    expect(summary(f.events)).toMatchObject({ startedQueries: 4, completedQueries: 4 });
+    expect(f.events.filter(event => event.event === "paperclip.db_response_capture.entry.v1").map(event => event.operation).sort())
+      .toEqual(["delete", "insert", "select", "update"]);
+    expect(JSON.stringify(f.events)).not.toContain(payload);
+    expect(JSON.stringify(f.events)).not.toContain(id);
+  });
+  it("counts the stored encoding before a custom column decoder changes the value", async () => {
+    const setup = createDb(temporary.connectionString); clients.push(setup);
+    const id = randomUUID(), payload = { text: "codec-payload-sentinel\u0000value".repeat(512) };
+    await setup.insert(encodedTable).values({ id, payload });
+    const f = fixture();
+    expect(await f.db.select().from(encodedTable).where(eq(encodedTable.id, id))).toEqual([{ id, payload }]);
+    f.capture.flush();
+    const entry = f.events.find(event => event.table === "response_capture_encoded_fixture")!;
+    expect(entry.responseValueBytes).toBe(Buffer.byteLength(id) + Buffer.byteLength(encodeRunEventPayload(payload)));
+    expect(entry.responseValueBytes).toBeGreaterThan(Buffer.byteLength(JSON.stringify({ id, payload })));
+    expect(JSON.stringify(f.events)).not.toContain("codec-payload-sentinel");
+  });
+  it("keeps a raw driver query lazy and counts repeated awaits without replay", async () => {
+    const f = fixture(), id = randomUUID();
+    const observer = createDb(temporary.connectionString); clients.push(observer);
+    const pending = f.db.$client.unsafe("insert into response_capture_fixture (id, payload) values ($1, $2) returning id", [id, "raw-lazy-sentinel"]);
+    expect(await observer.select().from(table).where(eq(table.id, id))).toEqual([]);
+    const first = await pending;
+    expect(await pending).toBe(first);
+    expect(first).toEqual([{ id }]);
+    expect(await observer.select().from(table).where(eq(table.id, id))).toEqual([{ id, payload: "raw-lazy-sentinel" }]);
+    f.capture.flush();
+    expect(summary(f.events)).toMatchObject({ startedQueries: 1, completedQueries: 1 });
+    expect(JSON.stringify(f.events)).not.toContain("raw-lazy-sentinel");
+    expect(JSON.stringify(f.events)).not.toContain(id);
+  });
+  it("counts prepared executions once and retains their parameter bindings", async () => {
+    const f = fixture(), id = randomUUID();
+    await f.db.insert(table).values({ id, payload: "prepared-value" });
+    const prepared = f.db.select().from(table).where(eq(table.id, sql.placeholder("id"))).prepare("response_capture_prepared");
+    expect(await prepared.execute({ id })).toEqual([{ id, payload: "prepared-value" }]);
+    expect(await prepared.execute({ id: randomUUID() })).toEqual([]);
+    f.capture.flush();
+    expect(summary(f.events)).toMatchObject({ startedQueries: 3, completedQueries: 3 });
+    expect(JSON.stringify(f.events)).not.toContain("prepared-value");
+  });
+  it("retains distinct builder callers through normal, prepared, and transaction execution", async () => {
+    const f = fixture(), id = randomUUID();
+    const setup = createDb(temporary.connectionString); clients.push(setup);
+    await setup.insert(table).values({ id, payload: "caller-sentinel" });
+    const shape = f.db.select().from(table);
+    const caller = (name: string) => compileFunction(
+      "return db.select().from(table).where(eq(table.id, id))", ["db", "id"],
+      { filename: `/fixture/server/src/services/${name}.ts`, contextExtensions: [{ table, eq }] },
+    ) as (db: Pick<typeof f.db, "select">, id: string) => typeof shape;
+    const fromA = caller("capture-caller-a"), fromB = caller("capture-caller-b");
+    const normalA = fromA(f.db, id), normalB = fromB(f.db, id);
+    expect(await normalA).toEqual([{ id, payload: "caller-sentinel" }]);
+    expect(await normalB).toEqual([{ id, payload: "caller-sentinel" }]);
+    const prepared = fromA(f.db, id).prepare("capture_caller_prepared");
+    expect(await prepared.execute()).toEqual([{ id, payload: "caller-sentinel" }]);
+    await f.db.transaction(async tx => {
+      expect(await fromB(tx, id)).toEqual([{ id, payload: "caller-sentinel" }]);
+    });
+    f.capture.flush();
+    const entries = f.events.filter(event => event.event === "paperclip.db_response_capture.entry.v1");
+    expect(entries).toHaveLength(2);
+    for (const name of ["a", "b"]) {
+      expect(entries.find(event => (event.callers as string[]).some(location => location.startsWith(`services/capture-caller-${name}.ts:`))))
+        .toMatchObject({ operation: "select", table: "response_capture_fixture", calls: 2, rows: 2 });
+    }
+    expect(summary(f.events)).toMatchObject({ startedQueries: 4, completedQueries: 4 });
+    expect(JSON.stringify(f.events)).not.toContain("/fixture/");
+    expect(JSON.stringify(f.events)).not.toContain("caller-sentinel");
+    expect(JSON.stringify(f.events)).not.toContain(id);
+  });
+  it("records the executed CTE rather than its unexecuted inner select", async () => {
+    const f = fixture(), id = randomUUID();
+    await f.db.insert(table).values({ id, payload: "cte-value" });
+    const cte = f.db.$with("capture_cte").as(f.db.select().from(table).where(eq(table.id, id)));
+    expect(await f.db.with(cte).select().from(cte)).toEqual([{ id, payload: "cte-value" }]);
+    f.capture.flush();
+    expect(summary(f.events)).toMatchObject({ startedQueries: 2, completedQueries: 2 });
+  });
+  it("keeps raw execute results and rejection behavior without logging SQL or errors", async () => {
+    const f = fixture();
+    expect(await f.db.execute(sql`select ${"sentinel-sql-parameter"}::text as value`)).toEqual([{ value: "sentinel-sql-parameter" }]);
+    await expect(f.db.execute(sql`select 1/0`)).rejects.toThrow();
+    f.capture.flush();
+    expect(summary(f.events)).toMatchObject({ startedQueries: 2, completedQueries: 2 });
+    expect(f.events.find(event => event.operation === "execute")).toMatchObject({ queryErrors: 1 });
+    expect(JSON.stringify(f.events)).not.toContain("sentinel-sql-parameter");
+    expect(JSON.stringify(f.events)).not.toContain("select 1/0");
+  });
+  it("preserves nested commit, rollback, and durable work-signal boundaries", async () => {
+    const f = fixture(), id = randomUUID(), events: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(f.db, "capture-topic", event => events.push(event));
+    try {
+      expect(await f.db.transaction(async tx => {
+        await tx.transaction(async nested => {
+          await signalDatabaseWork(nested, "capture-topic");
+          await nested.insert(table).values({ id, payload: "committed" });
+        });
+        expect(events).toEqual(["intent"]); expect(databaseWorkPending(f.db, "capture-topic")).toBe(true);
+        return 42;
+      })).toBe(42);
+      expect(events).toEqual(["intent", "settled"]);
+      expect(databaseWorkPending(f.db, "capture-topic")).toBe(false);
+      const rolledBack = randomUUID();
+      await expect(f.db.transaction(async tx => {
+        await signalDatabaseWork(tx, "capture-topic");
+        await tx.insert(table).values({ id: rolledBack, payload: "rolled-back" });
+        throw new Error("rollback fixture");
+      })).rejects.toThrow("rollback fixture");
+      await reconcileDatabaseWork(f.db, "capture-topic");
+      expect(databaseWorkPending(f.db, "capture-topic")).toBe(false);
+      expect(await f.db.select().from(table).where(eq(table.id, rolledBack))).toEqual([]);
+      expect(await f.db.select().from(table).where(eq(table.id, id))).toEqual([{ id, payload: "committed" }]);
+    } finally { unsubscribe(); }
+    f.capture.flush();
+    expect(summary(f.events).completedQueries).toBeGreaterThan(4);
+  });
+  it("covers automatically created dedicated pools and retains owner identity", async () => {
+    const logs: string[] = [], now = Date.now();
+    vi.spyOn(console, "info").mockImplementation(value => logs.push(String(value)));
+    vi.stubEnv("PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW", `${new Date(now - 100).toISOString()}/${new Date(now + 59_000).toISOString()}`);
+    const db = createDb(temporary.connectionString); clients.push(db);
+    const capture = databaseResponseCaptureFromEnv()!; captures.push(capture);
+    expect(await withDedicatedDbConnection(db, async dedicated => dedicated.execute(sql`select ${"dedicated-value"}::text as value`)))
+      .toEqual([{ value: "dedicated-value" }]);
+    capture.flush();
+    const events = logs.map(line => JSON.parse(line) as Record<string, unknown>);
+    expect(summary(events)).toMatchObject({ startedQueries: 1, completedQueries: 1 });
+    expect(logs.join("\n")).not.toContain("dedicated-value");
+    expect(logs.join("\n")).not.toContain(temporary.connectionString);
+  });
+});

packages/db/src/response-capture.test.ts

@@ -0,0 +1,133 @@
+import { integer, pgTable } from "drizzle-orm/pg-core";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createDatabaseResponseCapture, installDatabaseResponseCapture, publicDatabaseCallers, responseCaptureWindow } from "./response-capture.js";
+
+const epoch = Date.parse("2020-01-01T00:00:00Z");
+const metadata = { operation: "select", table: "capture_fixture", callers: ["services/fixture.ts:4:2"] };
+const table = pgTable("capture_fixture", { id: integer("id") });
+const captures: Array<ReturnType<typeof createDatabaseResponseCapture>> = [];
+afterEach(() => { for (const capture of captures.splice(0)) capture.flush(); vi.unstubAllEnvs(); });
+function fixture(emitOverride?: (event: Record<string, unknown>) => void) {
+  let clock = epoch;
+  const events: Array<Record<string, unknown>> = [];
+  const capture = createDatabaseResponseCapture({ start: epoch, end: epoch + 60_000 }, emitOverride ?? (event => events.push(event)), () => clock);
+  captures.push(capture);
+  return { capture, events, now: (value: number) => { clock = value; } };
+}
+
+describe("bounded database response capture", () => {
+  it.each([undefined, "", "sentinel-invalid-input", "2020-01-01T00:00:00Z", "2020-01-01T00:00:00Z/2020-01-01T00:00:00Z",
+    "2020-01-01T00:00:00Z/2020-01-01T00:01:01Z", "2020-01-01T01:00:00Z/2020-01-01T01:01:00Z",
+    "2020-01-01T00:00:00Z/2020-01-01T00:01:00Z/extra",
+  ])("disables absent, malformed, expired, or unbounded windows: %s", value => {
+    expect(responseCaptureWindow(value, epoch + 60_000)).toBeUndefined();
+  });
+  it("accepts a fixed future window of at most one minute", () => {
+    expect(responseCaptureWindow("2020-01-01T00:01:00Z/2020-01-01T00:02:00Z", epoch))
+      .toEqual({ start: epoch + 60_000, end: epoch + 120_000 });
+    expect(() => createDatabaseResponseCapture({ start: epoch, end: epoch + 60_001 }, () => {})).toThrow("Invalid");
+  });
+  it("keeps only public server file locations from stacks", () => {
+    const stack = ["Error: sentinel-message", "at f (/private/fixture/server/src/services/example.ts:4:2)",
+      "at f (/private/fixture/server/src/services/example.ts:4:2)", "at f (/private/fixture/server/dist/routes/example.js:5:3)",
+      "at f (/private/fixture/customer-file.js:7:9)", "at f (/app/server/src/services/../../customer-file.ts:8:1)"].join("\n");
+    expect(publicDatabaseCallers(stack)).toEqual(["services/example.ts:4:2", "routes/example.js:5:3"]);
+  });
+  it("leaves methods and identity unchanged when disabled", () => {
+    vi.stubEnv("PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW", undefined);
+    const db = { select: vi.fn(), transaction: vi.fn() }, select = db.select, transaction = db.transaction;
+    expect(installDatabaseResponseCapture(db, { table })).toBe(db);
+    expect(db.select).toBe(select); expect(db.transaction).toBe(transaction);
+  });
+  it("preserves exact values and only emits counts, sizes, and caller metadata", async () => {
+    const f = fixture(), rows = [{ text: "sentinel-result-value🙂\n\"", n: 4n }];
+    expect(await f.capture.execute(() => Promise.resolve(rows), metadata)).toBe(rows);
+    f.capture.flush();
+    const entry = f.events.find(event => event.event === "paperclip.db_response_capture.entry.v1")!;
+    expect(entry).toMatchObject({ calls: 1, rows: 1, queryErrors: 0, serializationErrors: 0,
+      responseValueBytes: Buffer.byteLength(JSON.stringify(rows, (_, value) => typeof value === "bigint" ? value.toString() : value)) });
+    expect(JSON.stringify(f.events)).not.toContain("sentinel-result-value");
+    expect(f.events.at(-1)).toMatchObject({ startedQueries: 1, completedQueries: 1, inFlightAtEnd: 0,
+      rowValuesLogged: false, sqlOrParametersLogged: false, wireBytesMeasured: false });
+  });
+  it("counts driver text values without adding JSON escaping or column names", async () => {
+    const f = fixture(), text = "sentinel-driver-text\n\"";
+    const rows = Object.assign([[text, true, 42n, null]], { columns: [
+      { name: "sentinel-column-name" }, { name: "b" }, { name: "n" }, { name: "empty" } ] });
+    expect(await f.capture.execute(() => Promise.resolve(rows), metadata)).toBe(rows);
+    f.capture.flush();
+    expect(f.events[0]).toMatchObject({ responseValueBytes: Buffer.byteLength(text) + 1 + 2, estimatedValues: 0 });
+    expect(JSON.stringify(f.events)).not.toContain("sentinel-driver-text");
+    expect(JSON.stringify(f.events)).not.toContain("sentinel-column-name");
+  });
+  it("does not alter synchronous or asynchronous errors or retry them", async () => {
+    const f = fixture(), error = new Error("sentinel-private-error");
+    const sync = vi.fn(() => { throw error; }), asyncFailure = vi.fn(() => Promise.reject(error));
+    expect(() => f.capture.execute(sync, metadata)).toThrow(error);
+    await expect(f.capture.execute(asyncFailure, metadata)).rejects.toBe(error);
+    expect(sync).toHaveBeenCalledTimes(1); expect(asyncFailure).toHaveBeenCalledTimes(1);
+    f.capture.flush();
+    expect(f.events[0]).toMatchObject({ calls: 2, queryErrors: 2, responseValueBytes: 0 });
+    expect(JSON.stringify(f.events)).not.toContain("sentinel-private-error");
+  });
+  it("keeps unserializable results and logging failures out of the query outcome", () => {
+    const f = fixture(), circular: Record<string, unknown> = {}; circular.self = circular;
+    expect(f.capture.execute(() => circular, metadata)).toBe(circular);
+    f.capture.flush();
+    expect(f.events[0]).toMatchObject({ calls: 1, serializationErrors: 1 });
+    const brokenLogger = fixture(() => { throw new Error("logger fixture"); });
+    expect(brokenLogger.capture.execute(() => circular, metadata)).toBe(circular);
+    expect(() => brokenLogger.capture.flush()).not.toThrow();
+  });
+  it("stops accounting at the deadline and leaves in-flight work untouched", async () => {
+    const f = fixture();
+    let resolve!: (rows: unknown[]) => void;
+    const pending = new Promise<unknown[]>(done => { resolve = done; });
+    const result = f.capture.execute(() => pending, metadata);
+    f.now(epoch + 60_000); f.capture.flush();
+    expect(f.events.at(-1)).toMatchObject({ startedQueries: 1, completedQueries: 0, inFlightAtEnd: 1 });
+    const rows = [{ marker: "late-result-value" }]; resolve(rows);
+    expect(await result).toBe(rows);
+    const direct = Promise.resolve(rows);
+    expect(f.capture.execute(() => direct, metadata)).toBe(direct);
+    expect(f.events).toHaveLength(1);
+  });
+  it("retains the deadline count when a query resolves before a delayed flush", async () => {
+    const f = fixture();
+    let resolve!: (rows: unknown[]) => void;
+    const pending = new Promise<unknown[]>(done => { resolve = done; });
+    const result = f.capture.execute(() => pending, metadata);
+    f.now(epoch + 60_001);
+    const rows = [{ marker: "late-before-flush-sentinel" }]; resolve(rows);
+    expect(await result).toBe(rows);
+    f.capture.flush();
+    expect(f.events).toHaveLength(1);
+    expect(f.events[0]).toMatchObject({ startedQueries: 1, completedQueries: 0, inFlightAtEnd: 1, responseValueBytes: 0 });
+    expect(JSON.stringify(f.events)).not.toContain("late-before-flush-sentinel");
+  });
+  it("bounds cardinality without dropping response-byte totals", () => {
+    const f = fixture();
+    for (let i = 0; i < 200; i += 1) f.capture.execute(() => [i], { ...metadata, callers: [`services/f${i}.ts:1:1`] });
+    f.capture.flush();
+    const summary = f.events.at(-1)!;
+    expect(summary).toMatchObject({ startedQueries: 200, completedQueries: 200, groups: 128, overflowCalls: 73 });
+    expect(summary.responseValueBytes).toBe(Array.from({ length: 200 }, (_, i) => Buffer.byteLength(JSON.stringify([i]))).reduce((a, b) => a + b, 0));
+  });
+  it("does not execute builders or prepared statements until requested", async () => {
+    const f = fixture(), rows = [{ id: 1 }], execute = vi.fn(async () => rows);
+    const query = { execute, prepare: () => ({ execute }) };
+    const builder = { from: () => query };
+    const db = { select: () => builder };
+    expect(installDatabaseResponseCapture(db, { table }, f.capture)).toBe(db);
+    expect(db.select()).toBe(builder);
+    expect(db.select().from()).toBe(query);
+    expect(execute).not.toHaveBeenCalled();
+    const prepared = query.prepare();
+    expect(execute).not.toHaveBeenCalled();
+    expect(await prepared.execute()).toBe(rows);
+    expect(await query.execute()).toBe(rows);
+    expect(execute).toHaveBeenCalledTimes(2);
+    f.capture.flush();
+    expect(f.events.at(-1)).toMatchObject({ completedQueries: 0 });
+  });
+});

packages/db/src/response-capture.ts

@@ -0,0 +1,273 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+import { randomUUID } from "node:crypto";
+import { getTableName, is, Table } from "drizzle-orm";
+
+type Metadata = { operation: string; table: string; callers: string[] };
+type Entry = Metadata & { calls: number; rows: number; bytes: number; maxBytes: number; queryErrors: number; serializationErrors: number; estimatedValues: number };
+type Event = Record<string, unknown>;
+type Method = (this: unknown, ...args: unknown[]) => unknown;
+export type ResponseCaptureWindow = { start: number; end: number };
+
+export function responseCaptureWindow(value: string | undefined, now = Date.now()): ResponseCaptureWindow | undefined {
+  if (!value) return;
+  const parts = value.split("/");
+  if (parts.length !== 2 || parts.some(part => !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/.test(part))) return;
+  const start = Date.parse(parts[0]!), end = Date.parse(parts[1]!);
+  if (!Number.isFinite(start) || !Number.isFinite(end) || end <= start || end - start > 60_000 || end <= now || start > now + 30 * 60_000) return;
+  return { start, end };
+}
+
+export function publicDatabaseCallers(stack: string | undefined): string[] {
+  const callers: string[] = [];
+  for (const line of (stack ?? "").split("\n")) {
+    const match = line.match(/[/\\]server[/\\](?:src|dist)[/\\]((?:services|routes|modules|middleware)[/\\][A-Za-z0-9_./\\-]+\.[cm]?[jt]s):(\d+):(\d+)/);
+    if (!match) continue;
+    const relative = match[1]!.replaceAll("\\", "/");
+    if (relative.split("/").includes("..")) continue;
+    const caller = `${relative}:${match[2]}:${match[3]}`;
+    if (!callers.includes(caller)) callers.push(caller);
+    if (callers.length === 4) break;
+  }
+  return callers;
+}
+
+function responseValueSize(value: unknown): { bytes: number; estimatedValues: number } {
+  const columns = value && typeof value === "object" ? (value as { columns?: unknown }).columns : undefined;
+  if (!Array.isArray(value) || !Array.isArray(columns)) {
+    const serialized = JSON.stringify(value, (_, item: unknown) => typeof item === "bigint" ? item.toString() : item);
+    return { bytes: serialized === undefined ? 0 : Buffer.byteLength(serialized), estimatedValues: 1 };
+  }
+  let bytes = 0, estimatedValues = 0;
+  for (const row of value) {
+    const cells = Array.isArray(row) ? row : columns.map(column => row[column.name]);
+    for (const cell of cells) {
+      if (cell === null || cell === undefined) continue;
+      if (typeof cell === "string") bytes += Buffer.byteLength(cell);
+      else if (typeof cell === "boolean") bytes += 1;
+      else if (typeof cell === "number" || typeof cell === "bigint") bytes += Buffer.byteLength(String(cell));
+      else {
+        estimatedValues += 1;
+        const serialized = JSON.stringify(cell, (_, item: unknown) => typeof item === "bigint" ? item.toString() : item);
+        if (serialized !== undefined) bytes += Buffer.byteLength(serialized);
+      }
+    }
+  }
+  return { bytes, estimatedValues };
+}
+
+export function createDatabaseResponseCapture(
+  window: ResponseCaptureWindow,
+  emit: (event: Event) => void,
+  now: () => number = Date.now,
+) {
+  if (!Number.isFinite(new Date(window.start).getTime()) || !Number.isFinite(new Date(window.end).getTime()) ||
+      window.end <= window.start || window.end - window.start > 60_000) throw new Error("Invalid database response capture window");
+  const id = randomUUID(), entries = new Map<string, Entry>();
+  const roots = new WeakSet<object>(), queries = new WeakSet<object>(), factories = new WeakSet<object>();
+  const clients = new WeakSet<object>(), driverQueries = new WeakSet<object>();
+  const scope = new AsyncLocalStorage<Metadata>();
+  let closed = false, inFlight = 0, started = 0, completed = 0, overflowCalls = 0;
+  const send = (event: Event) => { try { emit(event); } catch {} };
+  const expired = () => closed || now() >= window.end;
+  const active = () => !expired() && now() >= window.start;
+  const flush = () => {
+    if (closed) return;
+    closed = true;
+    clearTimeout(timer);
+    scope.disable();
+    const ordered = [...entries.values()].sort((a, b) => b.bytes - a.bytes);
+    for (const entry of ordered) send({ event: "paperclip.db_response_capture.entry.v1", captureId: id,
+      operation: entry.operation, table: entry.table, callers: entry.callers, calls: entry.calls, rows: entry.rows,
+      responseValueBytes: entry.bytes, maxResponseValueBytes: entry.maxBytes, estimatedValues: entry.estimatedValues,
+      queryErrors: entry.queryErrors, serializationErrors: entry.serializationErrors });
+    send({ event: "paperclip.db_response_capture.summary.v1", captureId: id,
+      windowStart: new Date(window.start).toISOString(), windowEnd: new Date(window.end).toISOString(),
+      closedAt: new Date(now()).toISOString(),
+      startedQueries: started, completedQueries: completed, inFlightAtEnd: inFlight, groups: entries.size,
+      overflowCalls, responseValueBytes: ordered.reduce((sum, entry) => sum + entry.bytes, 0),
+      rowValuesLogged: false, sqlOrParametersLogged: false, wireBytesMeasured: false,
+      measurementBoundary: "postgres-js results before Drizzle column decoding",
+      relationalQueriesMeasured: true, unsafeDriverQueriesMeasured: true,
+      taggedDriverQueriesMeasured: false, streamingQueriesMeasured: false });
+    entries.clear();
+  };
+  const timer = setTimeout(flush, Math.max(0, window.end - now()));
+  timer.unref();
+  function finish(metadata: Metadata, value: unknown, failed: boolean) {
+    if (!active()) return;
+    inFlight -= 1;
+    completed += 1;
+    try {
+      let key = JSON.stringify(metadata);
+      if (!entries.has(key) && entries.size >= 127) {
+        overflowCalls += 1;
+        metadata = { operation: "other", table: "other", callers: [] };
+        key = "overflow";
+      }
+      let entry = entries.get(key);
+      if (!entry) {
+        entry = { ...metadata, calls: 0, rows: 0, bytes: 0, maxBytes: 0, queryErrors: 0, serializationErrors: 0, estimatedValues: 0 };
+        entries.set(key, entry);
+      }
+      entry.calls += 1;
+      if (failed) { entry.queryErrors += 1; return; }
+      entry.rows += Array.isArray(value) ? value.length : 0;
+      try {
+        const size = responseValueSize(value);
+        entry.bytes += size.bytes;
+        entry.maxBytes = Math.max(entry.maxBytes, size.bytes);
+        entry.estimatedValues += size.estimatedValues;
+      } catch { entry.serializationErrors += 1; }
+    } catch {}
+  }
+  function execute(action: () => unknown, metadata: Metadata) {
+    if (!active()) return action();
+    inFlight += 1; started += 1;
+    let result: unknown;
+    try { result = action(); }
+    catch (error) { finish(metadata, undefined, true); throw error; }
+    let asynchronous = false;
+    try { asynchronous = Boolean(result && typeof (result as { then?: unknown }).then === "function"); } catch {}
+    if (asynchronous) {
+      return Promise.resolve(result).then(value => { finish(metadata, value, false); return value; },
+        error => { finish(metadata, undefined, true); throw error; });
+    }
+    finish(metadata, result, false);
+    return result;
+  }
+  return { window, roots, queries, factories, clients, driverQueries, scope, active, expired, execute, flush };
+}
+
+type Capture = ReturnType<typeof createDatabaseResponseCapture>;
+let environmentCapture: { key: string; capture: Capture } | undefined;
+export function databaseResponseCaptureFromEnv(env: NodeJS.ProcessEnv = process.env): Capture | undefined {
+  const key = env.PAPERCLIP_DB_RESPONSE_CAPTURE_WINDOW;
+  const window = responseCaptureWindow(key);
+  if (!key || !window) return;
+  if (environmentCapture?.key === key) return environmentCapture.capture;
+  const capture = createDatabaseResponseCapture(window, event => console.info(JSON.stringify(event)));
+  environmentCapture = { key, capture };
+  return capture;
+}
+
+export function installDatabaseResponseCapture<T extends object>(db: T, schema: Record<string, unknown>, capture = databaseResponseCaptureFromEnv()): T {
+  if (!capture || capture.expired()) return db;
+  const tables = new Set(Object.values(schema).filter(value => is(value, Table)).map(table => getTableName(table)));
+  const object = (value: unknown): value is Record<string, unknown> => value !== null && typeof value === "object";
+  function replace(target: Record<string, unknown>, name: string, replacement: (original: Method) => Method) {
+    const original = target[name];
+    if (typeof original !== "function") return;
+    try { target[name] = replacement(original as Method); } catch {}
+  }
+  function tableName(value: unknown) {
+    try { if (is(value, Table)) { const name = getTableName(value); if (tables.has(name)) return name; } } catch {}
+    return "other";
+  }
+  function query(value: unknown, metadata: Metadata) {
+    if (!object(value) || capture!.queries.has(value)) return;
+    capture!.queries.add(value);
+    replace(value, "execute", original => function (this: unknown, ...args: unknown[]) {
+      if (capture!.expired()) return Reflect.apply(original, this, args);
+      return capture!.scope.run(metadata, () => Reflect.apply(original, this, args));
+    });
+    replace(value, "prepare", original => function (this: unknown, ...args: unknown[]) {
+      const prepared = Reflect.apply(original, this, args);
+      try { query(prepared, metadata); } catch {}
+      return prepared;
+    });
+  }
+  function factory(value: unknown, step: string, metadata: Metadata, from = false) {
+    if (!object(value) || capture!.factories.has(value)) return;
+    capture!.factories.add(value);
+    replace(value, step, original => function (this: unknown, ...args: unknown[]) {
+      const result = Reflect.apply(original, this, args);
+      try { query(result, { ...metadata, table: from ? tableName(args[0]) : metadata.table }); } catch {}
+      return result;
+    });
+  }
+  function driverQuery(value: unknown, metadata: Metadata) {
+    if (!object(value) || capture!.driverQueries.has(value)) return;
+    capture!.driverQueries.add(value);
+    let consumed = false;
+    replace(value, "then", original => function (this: unknown, ...handlers: unknown[]) {
+      if (consumed || !capture!.active()) {
+        consumed = true;
+        return Reflect.apply(original, this, handlers);
+      }
+      consumed = true;
+      const result = capture!.execute(() => Reflect.apply(original, this, []), metadata) as Promise<unknown>;
+      return Reflect.apply(result.then, result, handlers);
+    });
+    for (const method of ["execute", "cursor", "forEach", "readable", "writable"]) {
+      replace(value, method, original => function (this: unknown, ...args: unknown[]) {
+        consumed = true;
+        return Reflect.apply(original, this, args);
+      });
+    }
+  }
+  function sqlMetadata(value: unknown): Metadata {
+    const scoped = capture!.scope.getStore();
+    const text = typeof value === "string" ? value : "";
+    let table = "other";
+    for (const match of text.matchAll(/\b(?:from|update|into)\s+"([A-Za-z0-9_]+)"/gi)) {
+      if (tables.has(match[1]!)) { table = match[1]!; break; }
+    }
+    return scoped ? { ...scoped, table: scoped.table === "other" ? table : scoped.table }
+      : { operation: text.match(/^\s*(select|insert|update|delete)\b/i)?.[1]?.toLowerCase() ?? "execute",
+        table, callers: publicDatabaseCallers(new Error().stack) };
+  }
+  function client(value: unknown) {
+    if ((!object(value) && typeof value !== "function") || capture!.clients.has(value)) return;
+    capture!.clients.add(value);
+    const target = value as Record<string, unknown>;
+    replace(target, "unsafe", original => function (this: unknown, ...args: unknown[]) {
+      const result = Reflect.apply(original, this, args);
+      if (!capture!.expired()) { try { driverQuery(result, sqlMetadata(args[0])); } catch {} }
+      return result;
+    });
+    for (const method of ["begin", "savepoint"]) {
+      replace(target, method, original => function (this: unknown, ...args: unknown[]) {
+        const wrapped = args.map(arg => typeof arg !== "function" ? arg : function (this: unknown, ...inner: unknown[]) {
+          try { client(inner[0]); } catch {}
+          return Reflect.apply(arg, this, inner);
+        });
+        return Reflect.apply(original, this, wrapped);
+      });
+    }
+    replace(target, "reserve", original => function (this: unknown, ...args: unknown[]) {
+      return Promise.resolve(Reflect.apply(original, this, args)).then(reserved => {
+        try { client(reserved); } catch {}
+        return reserved;
+      });
+    });
+  }
+  function root(value: unknown) {
+    if (!object(value) || capture!.roots.has(value)) return;
+    capture!.roots.add(value);
+    for (const method of ["select", "selectDistinct", "selectDistinctOn", "insert", "update", "delete", "execute", "with"] as const) {
+      replace(value, method, original => function (this: unknown, ...args: unknown[]) {
+        const result = Reflect.apply(original, this, args);
+        if (capture!.expired()) return result;
+        try {
+          if (method === "with") { root(result); return result; }
+          const operation = method.startsWith("select") ? "select" : method;
+          const metadata = { operation, table: operation === "select" || method === "execute" ? "other" : tableName(args[0]), callers: publicDatabaseCallers(new Error().stack) };
+          if (operation === "select") factory(result, "from", metadata, true);
+          else if (method === "insert") factory(result, "values", metadata);
+          else if (method === "update") factory(result, "set", metadata);
+          else query(result, metadata);
+        } catch {}
+        return result;
+      });
+    }
+    replace(value, "transaction", original => function (this: unknown, callback: unknown, ...options: unknown[]) {
+      if (typeof callback !== "function" || capture!.expired()) return Reflect.apply(original, this, [callback, ...options]);
+      return Reflect.apply(original, this, [function (this: unknown, ...args: unknown[]) {
+        try { root(args[0]); } catch {}
+        return Reflect.apply(callback, this, args);
+      }, ...options]);
+    });
+  }
+  try { client((db as Record<string, unknown>).$client); root(db); } catch {}
+  return db;
+}