fix(db): dedupe signal_snapshots to bounded latest-per-key retention

src/api/routes.ts

@@ -107,7 +107,7 @@ import {
   getGlobalAgentFrozenState,
   setGlobalAgentFrozen,
 } from "../db/repositories";
-import { pruneExpiredRecords, RETENTION_POLICY } from "../db/retention";
+import { dedupeSignalSnapshots, pruneExpiredRecords, RETENTION_POLICY } from "../db/retention";
 import {
   backfillOpenPullRequestDetails,
   backfillRegisteredRepositories,
@@ -3808,11 +3808,19 @@ export function createApp() {
     return c.json(await getRepositoryAiKeyStatus(c.env, fullName));
   });
 
-  // Read-only retention preview: counts the rows the daily prune cron would delete, per table. Does NOT
-  // delete anything (dry-run); the actual prune runs on the schedule via the prune-retention job.
+  // Read-only retention preview: counts the rows the daily prune cron would delete, per table, plus the
+  // duplicate signal_snapshots rows the dedup pass would remove. Does NOT delete anything (dry-run); the
+  // actual prune + dedup runs on the schedule via the prune-retention job.
   app.get("/v1/internal/retention/preview", async (c) => {
     const results = await pruneExpiredRecords(c.env, { dryRun: true });
-    return c.json({ policy: RETENTION_POLICY, eligible: results, totalEligible: results.reduce((sum, r) => sum + r.deleted, 0) });
+    const dedupeResults = await dedupeSignalSnapshots(c.env, { dryRun: true });
+    return c.json({
+      policy: RETENTION_POLICY,
+      eligible: results,
+      totalEligible: results.reduce((sum, r) => sum + r.deleted, 0),
+      signalSnapshotDuplicates: dedupeResults,
+      totalSignalSnapshotDuplicates: dedupeResults.reduce((sum, r) => sum + r.deleted, 0),
+    });
   });
 
   app.post("/v1/internal/repos/:owner/:repo/ai-key", async (c) => {

src/db/retention.ts

@@ -88,3 +88,50 @@ export async function pruneExpiredRecords(
 
   return results;
 }
+
+export type SignalSnapshotDedupeResult = { signalType: string; deleted: number };
+
+/**
+ * signal_snapshots has no dedup: `generate-signal-snapshots` inserts a NEW row per (signal_type,
+ * target_key) on every run rather than replacing the prior one, so within RETENTION_POLICY's 90-day
+ * age window a key can accumulate hundreds of superseded snapshots (#3810 -- 342,243 rows for 2,183
+ * distinct keys contributed to hitting D1's size cap). This keeps only the latest row per
+ * (signal_type, target_key), batched PER signal_type (not one table-wide window-function delete) so
+ * each statement stays within D1's per-statement CPU budget -- the same batching split used during
+ * the incident's manual remediation. "Latest" is the highest rowid per key: signal_snapshots is
+ * populated by a single sequential batch job, so insertion order and generated_at agree, and rowid
+ * (unlike generated_at) can never tie.
+ */
+export async function dedupeSignalSnapshots(
+  env: Env,
+  options: { dryRun?: boolean; batchSize?: number; maxPerType?: number } = {},
+): Promise<SignalSnapshotDedupeResult[]> {
+  const dryRun = options.dryRun ?? false;
+  const batchSize = options.batchSize ?? BATCH_SIZE;
+  const maxPerType = options.maxPerType ?? MAX_DELETED_PER_TABLE;
+  const results: SignalSnapshotDedupeResult[] = [];
+
+  const types = await env.DB.prepare("SELECT DISTINCT signal_type FROM signal_snapshots").all<{ signal_type: string }>();
+  for (const { signal_type: signalType } of types.results) {
+    const staleCondition = `signal_type = ?1 AND rowid NOT IN (SELECT MAX(rowid) FROM signal_snapshots WHERE signal_type = ?1 GROUP BY target_key)`;
+
+    if (dryRun) {
+      const row = await env.DB.prepare(`SELECT count(*) AS n FROM signal_snapshots WHERE ${staleCondition}`).bind(signalType).first<{ n: number }>();
+      results.push({ signalType, deleted: Number(row?.n ?? 0) });
+      continue;
+    }
+
+    let deleted = 0;
+    for (;;) {
+      const result = await env.DB.prepare(`DELETE FROM signal_snapshots WHERE rowid IN (SELECT rowid FROM signal_snapshots WHERE ${staleCondition} LIMIT ${batchSize})`)
+        .bind(signalType)
+        .run();
+      const changes = Number(result.meta?.changes ?? 0);
+      deleted += changes;
+      if (changes < batchSize || deleted >= maxPerType) break;
+    }
+    results.push({ signalType, deleted });
+  }
+
+  return results;
+}

src/queue/processors.ts

@@ -83,7 +83,7 @@ import {
   upsertPullRequestFromGitHub,
   upsertRepositoryFromGitHub,
 } from "../db/repositories";
-import { pruneExpiredRecords } from "../db/retention";
+import { dedupeSignalSnapshots, pruneExpiredRecords } from "../db/retention";
 import {
   effectiveIssueCapForAccountAge,
   isBelowAccountAgeThreshold,
@@ -858,27 +858,33 @@ function refreshLiveMergeState(
 }
 
 /**
- * Run (or dry-run) the data-retention prune across the configured log/snapshot tables and audit the
- * outcome. The per-table windows live in RETENTION_POLICY; only append-only/superseded tables are pruned.
+ * Run (or dry-run) the data-retention prune across the configured log/snapshot tables, plus the
+ * signal_snapshots dedup pass (#3810 -- signal_snapshots has no natural dedup, so within its own
+ * retention window a key can still accumulate many superseded rows), and audit the combined outcome.
+ * The per-table windows live in RETENTION_POLICY; only append-only/superseded tables are pruned.
  */
 export async function runRetentionPrune(
   env: Env,
   requestedBy: string,
   dryRun: boolean,
 ): Promise<void> {
   const results = await pruneExpiredRecords(env, { dryRun });
+  const dedupeResults = await dedupeSignalSnapshots(env, { dryRun });
   const totalDeleted = results.reduce((sum, result) => sum + result.deleted, 0);
+  const totalDeduped = dedupeResults.reduce((sum, result) => sum + result.deleted, 0);
   await recordAuditEvent(env, {
     eventType: "retention.prune",
     actor: requestedBy,
     outcome: dryRun ? "completed" : "success",
     detail: dryRun
-      ? `dry-run: ${totalDeleted} row(s) eligible`
-      : `pruned ${totalDeleted} row(s)`,
+      ? `dry-run: ${totalDeleted} row(s) eligible, ${totalDeduped} duplicate signal_snapshots row(s) eligible`
+      : `pruned ${totalDeleted} row(s), deduped ${totalDeduped} signal_snapshots row(s)`,
     metadata: {
       dryRun,
       totalDeleted,
       perTable: Object.fromEntries(results.map((r) => [r.table, r.deleted])),
+      totalDeduped,
+      perSignalType: Object.fromEntries(dedupeResults.map((r) => [r.signalType, r.deleted])),
     },
   });
 }

test/unit/retention.test.ts

@@ -1,7 +1,7 @@
 import { describe, expect, it } from "vitest";
 import { createApp } from "../../src/api/routes";
 import { getDb } from "../../src/db/client";
-import { pruneExpiredRecords, RETENTION_POLICY } from "../../src/db/retention";
+import { dedupeSignalSnapshots, pruneExpiredRecords, RETENTION_POLICY } from "../../src/db/retention";
 import { aiUsageEvents, webhookEvents } from "../../src/db/schema";
 import { processJob, runRetentionPrune } from "../../src/queue/processors";
 import { createTestEnv } from "../helpers/d1";
@@ -26,6 +26,21 @@ async function seed(env: Env) {
 
 const countWebhook = async (env: Env) => (await env.DB.prepare("SELECT count(*) AS n FROM webhook_events").first<{ n: number }>())?.n ?? 0;
 
+async function insertSignalSnapshot(env: Env, id: string, signalType: string, targetKey: string, generatedAt: string) {
+  await env.DB.prepare(
+    "INSERT INTO signal_snapshots (id, signal_type, target_key, repo_full_name, payload_json, generated_at) VALUES (?,?,?,?,?,?)",
+  )
+    .bind(id, signalType, targetKey, "JSONbored/gittensory", "{}", generatedAt)
+    .run();
+}
+
+const countSignalSnapshots = async (env: Env, signalType?: string) =>
+  (
+    await env.DB.prepare(signalType ? "SELECT count(*) AS n FROM signal_snapshots WHERE signal_type = ?" : "SELECT count(*) AS n FROM signal_snapshots")
+      .bind(...(signalType ? [signalType] : []))
+      .first<{ n: number }>()
+  )?.n ?? 0;
+
 describe("pruneExpiredRecords", () => {
   it("dry-run reports eligible rows per table without deleting anything", async () => {
     const env = createTestEnv();
@@ -95,6 +110,83 @@ describe("pruneExpiredRecords", () => {
   });
 });
 
+describe("dedupeSignalSnapshots", () => {
+  it("returns no results when the table is empty", async () => {
+    const env = createTestEnv();
+    const results = await dedupeSignalSnapshots(env);
+    expect(results).toEqual([]);
+  });
+
+  it("dry-run counts duplicates per signal_type without deleting anything", async () => {
+    const env = createTestEnv();
+    await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-3", "repo_culture", "other/repo", "2026-06-01T00:00:00.000Z"); // distinct key, not a duplicate
+    const results = await dedupeSignalSnapshots(env, { dryRun: true });
+    expect(results).toEqual([{ signalType: "repo_culture", deleted: 1 }]);
+    expect(await countSignalSnapshots(env)).toBe(3); // nothing actually deleted
+  });
+
+  it("keeps only the highest-rowid row per (signal_type, target_key) and leaves other signal_types untouched", async () => {
+    const env = createTestEnv();
+    await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-3", "repo_culture", "JSONbored/gittensory", "2026-06-03T00:00:00.000Z"); // latest, kept
+    await insertSignalSnapshot(env, "s-4", "burden_forecast", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z"); // sole row, kept
+
+    const results = await dedupeSignalSnapshots(env);
+    expect(results.find((r) => r.signalType === "repo_culture")?.deleted).toBe(2);
+    expect(results.find((r) => r.signalType === "burden_forecast")?.deleted).toBe(0);
+    expect(await countSignalSnapshots(env, "repo_culture")).toBe(1);
+    expect(await countSignalSnapshots(env, "burden_forecast")).toBe(1);
+    const remaining = await env.DB.prepare("SELECT id FROM signal_snapshots WHERE signal_type = ?").bind("repo_culture").first<{ id: string }>();
+    expect(remaining?.id).toBe("s-3");
+  });
+
+  it("deletes across multiple batches per signal_type and stops at the per-type cap", async () => {
+    const env = createTestEnv();
+    for (let i = 0; i < 6; i++) {
+      await insertSignalSnapshot(env, `s-${i}`, "repo_culture", "JSONbored/gittensory", `2026-06-0${i + 1}T00:00:00.000Z`);
+    }
+    // The 6th insert (highest generated_at, inserted last so it also has the highest rowid) is kept, leaving 5
+    // duplicates; batchSize 2 forces multiple full (changes === batchSize) delete iterations before maxPerType 4
+    // is reached, so the loop continues past its first batch instead of stopping there.
+    const results = await dedupeSignalSnapshots(env, { batchSize: 2, maxPerType: 4 });
+    expect(results).toEqual([{ signalType: "repo_culture", deleted: 4 }]); // 2 + 2, then cap reached
+    expect(await countSignalSnapshots(env, "repo_culture")).toBe(2); // 1 kept + 1 duplicate left for the next run
+  });
+
+  it("dry-run falls back to 0 when the count query returns no row (defensive ?? 0 arm)", async () => {
+    const noRowEnv = {
+      DB: {
+        prepare: (sql: string) => ({
+          all: async () => ({ results: [{ signal_type: "repo_culture" }] }), // the DISTINCT signal_type query
+          bind: (..._binds: unknown[]) => ({
+            first: async () => undefined, // count query returns no row → `row?.n ?? 0` fallback fires
+          }),
+        }),
+      },
+    } as unknown as Env;
+    const results = await dedupeSignalSnapshots(noRowEnv, { dryRun: true });
+    expect(results).toEqual([{ signalType: "repo_culture", deleted: 0 }]);
+  });
+
+  it("falls back to 0 changes when a delete run() result lacks meta (defensive ?? 0 arm)", async () => {
+    const noMetaEnv = {
+      DB: {
+        prepare: (sql: string) => ({
+          all: async () => ({ results: [{ signal_type: "repo_culture" }] }), // the DISTINCT signal_type query
+          bind: (..._binds: unknown[]) => ({
+            run: async () => ({}), // no meta → `result.meta?.changes ?? 0` fallback fires, so changes = 0 < batchSize
+          }),
+        }),
+      },
+    } as unknown as Env;
+    const results = await dedupeSignalSnapshots(noMetaEnv);
+    expect(results).toEqual([{ signalType: "repo_culture", deleted: 0 }]);
+  });
+});
+
 describe("runRetentionPrune + processJob", () => {
   it("audits a dry-run without deleting", async () => {
     const env = createTestEnv();
@@ -106,13 +198,17 @@ describe("runRetentionPrune + processJob", () => {
     expect(audit?.detail).toMatch(/dry-run/);
   });
 
-  it("processJob prune-retention deletes and audits", async () => {
+  it("processJob prune-retention deletes, dedupes signal_snapshots, and audits both", async () => {
     const env = createTestEnv();
     await seed(env);
+    await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z");
     await processJob(env, { type: "prune-retention", requestedBy: "schedule" });
     expect(await countWebhook(env)).toBe(3);
-    const audit = await env.DB.prepare("SELECT outcome FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string }>();
+    expect(await countSignalSnapshots(env, "repo_culture")).toBe(1);
+    const audit = await env.DB.prepare("SELECT outcome, detail FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string; detail: string }>();
     expect(audit?.outcome).toBe("success");
+    expect(audit?.detail).toMatch(/deduped 1 signal_snapshots row/);
   });
 });
 
@@ -121,11 +217,21 @@ describe("retention preview route", () => {
     const app = createApp();
     const env = createTestEnv();
     await seed(env);
+    await insertSignalSnapshot(env, "s-1", "repo_culture", "JSONbored/gittensory", "2026-06-01T00:00:00.000Z");
+    await insertSignalSnapshot(env, "s-2", "repo_culture", "JSONbored/gittensory", "2026-06-02T00:00:00.000Z");
     const res = await app.request("/v1/internal/retention/preview", { headers: { authorization: `Bearer ${env.INTERNAL_JOB_TOKEN}` } }, env);
     expect(res.status).toBe(200);
-    const body = (await res.json()) as { totalEligible: number; eligible: Array<{ table: string; deleted: number }> };
+    const body = (await res.json()) as {
+      totalEligible: number;
+      eligible: Array<{ table: string; deleted: number }>;
+      totalSignalSnapshotDuplicates: number;
+      signalSnapshotDuplicates: Array<{ signalType: string; deleted: number }>;
+    };
     expect(body.totalEligible).toBeGreaterThanOrEqual(1);
     expect(body.eligible.find((r) => r.table === "webhook_events")).toBeUndefined();
+    expect(body.totalSignalSnapshotDuplicates).toBe(1);
+    expect(body.signalSnapshotDuplicates).toEqual([{ signalType: "repo_culture", deleted: 1 }]);
     expect(await countWebhook(env)).toBe(3); // preview is read-only
+    expect(await countSignalSnapshots(env)).toBe(2); // preview is read-only
   });
 });