fix(server): schedule inbound voice history retention by deadline

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

server/src/__tests__/app-private-hostname-gate.test.ts

@@ -87,6 +87,7 @@ describe("createChatReconciliationCoordinator", () => {
     expect(source).toContain("chatChannels.voice.registerRecoveryWork(deliveryWork,");
     expect(source).toContain("chatChannels.voice.registerReplyWork(deliveryWork,");
     expect(source).toContain("chatChannels.voice.inbound.registerCleanupWork(deliveryWork,");
+    expect(source).toContain("chatChannels.voice.inbound.registerRetentionWork(deliveryWork,");
     expect(source).toContain("chatChannels.voice.reconcileMaintenance()");
     expect(source).not.toContain("chatChannels.voice.reconcile()");
   });

server/src/__tests__/voice-inbound-retention.test.ts

@@ -0,0 +1,180 @@
+import { randomUUID } from "node:crypto";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { eq, sql } from "drizzle-orm";
+import { agents, chatConversations, chatEndpoints, chatVoiceInboundCalls, chatVoiceSessions, companies, createDb, issues, toolApplications, toolConnections } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { voiceInboundRetention } from "../services/voice/voice-inbound-retention.js";
+import { notifyVoiceInbound } from "../services/voice/voice-inbound-notifications.js";
+
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_RETENTION_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("inbound voice retention lifecycle", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let enabled = true;
+  const errors: unknown[] = [], fixtureCompanies: string[] = [];
+  const week = 7 * 86_400_000;
+  beforeAll(async () => {
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-inbound-retention-");
+    db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 60_000);
+  async function fixture() {
+    const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID();
+    fixtureCompanies.push(companyId);
+    const applicationId = randomUUID(), connectionId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Inbound cleanup", issuePrefix: `V${companyId.slice(0, 7)}` });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Voice", adapterType: "paperclip_runner" });
+    await db.insert(issues).values({ id: issueId, companyId, title: "Voice", assigneeAgentId: agentId, status: "in_progress" });
+    await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Voice", type: "chat" });
+    await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Voice", uid: connectionId, transport: "voice", connectionPurpose: "channel", enabled: true, status: "active" });
+    await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, publicId: endpointId, provider: "speko", status: "active", setup: { step: "complete", runtimeGeneration: 1 } });
+    await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: conversationId, externalThreadId: conversationId, externalLabel: "Voice", isDirectMessage: true });
+    return { companyId, endpointId, agentId, issueId, conversationId };
+  }
+  beforeEach(async () => {
+    enabled = true; errors.length = 0;
+    ({ companyId, endpointId, agentId, issueId, conversationId } = await fixture());
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+  });
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks();
+    for (const companyId of fixtureCompanies.splice(0)) {
+      await db.delete(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, companyId));
+      await db.delete(chatVoiceSessions).where(eq(chatVoiceSessions.companyId, companyId));
+      await db.delete(chatConversations).where(eq(chatConversations.companyId, companyId));
+      await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId));
+      await db.delete(issues).where(eq(issues.companyId, companyId));
+      await db.delete(agents).where(eq(agents.companyId, companyId));
+      await db.delete(companies).where(eq(companies.id, companyId));
+    }
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  async function admission(patch: Partial<typeof chatVoiceInboundCalls.$inferInsert> = {}, writer = db) {
+    const id = randomUUID();
+    return writer.transaction(async tx => {
+      await notifyVoiceInbound(tx, { kind: "admission", id });
+      return (await tx.insert(chatVoiceInboundCalls).values({ id, companyId, endpointId, providerSessionId: randomUUID(), state: "ended", cleanupConfirmedAt: new Date(), updatedAt: new Date(Date.now() - week - 1000), approvalCode: "123456", generation: 1, credentialFingerprint: "test", toolTokenHash: "a".repeat(64), requestFingerprint: id, expiresAt: new Date(Date.now() - week - 1000), ...patch }).returning())[0];
+    });
+  }
+  async function owner(call: typeof chatVoiceInboundCalls.$inferSelect, patch: Partial<typeof chatVoiceSessions.$inferInsert> = {}) {
+    return (await db.insert(chatVoiceSessions).values({ companyId, endpointId, conversationId, issueId, assignedAgentId: agentId, callerId: call.approvedByUserId ?? `guest:${call.id}`, callerAuthority: call.callerAuthority === "member" ? "member" : "guest_intake", approvedByUserId: call.approvedByUserId,
+      mode: "inbound_phone", state: "active", generation: call.generation, credentialFingerprint: call.credentialFingerprint, providerSessionId: call.providerSessionId, toolTokenHash: call.toolTokenHash, idempotencyKey: `inbound:${call.id}`, requestFingerprint: call.requestFingerprint, expiresAt: new Date(Date.now() + 600_000), ...patch }).returning())[0];
+  }
+  async function start(service = voiceInboundRetention(db)) {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) });
+    await service.register(coordinator, () => enabled).ready;
+    return service;
+  }
+  async function flush() {
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 5000 });
+  }
+  async function quiet(ms = 120_000) {
+    const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+    try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); }
+    finally { reads.mockRestore(); writes.mockRestore(); }
+  }
+  async function row(id: string) { return (await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, id)))[0]; }
+  it("leaves configured idle quiet and reacts to the first committed history", async () => {
+    await start(); await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet();
+    const call = await admission(); await flush(); expect(await row(call.id)).toBeUndefined();
+    expect(coordinator!.nextWakeAt()).toBeNull(); await quiet();
+  });
+  it("uses the later timestamp, stays quiet before its deadline, and reconstructs it on restart", async () => {
+    const now = Date.now(), later = now + 400_000;
+    const call = await admission({ expiresAt: new Date(now - week + 200_000), updatedAt: new Date(later - week) });
+    await start(); await flush(); expect(coordinator!.nextWakeAt()).toBe(later); await quiet();
+    await coordinator!.stop(); await start(); await flush(); expect(coordinator!.nextWakeAt()).toBe(later);
+    expect(await row(call.id)).toBeDefined();
+    await vi.advanceTimersByTimeAsync(later - Date.now() + 1000); await flush();
+    expect(await row(call.id)).toBeUndefined(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet();
+  });
+  it("rounds PostgreSQL submillisecond deadlines up and never deletes early", async () => {
+    const call = await admission(); const dueAt = Math.trunc(Date.now()) + 60_000;
+    await db.execute(sql`UPDATE chat_voice_inbound_calls SET updated_at = to_timestamp(${dueAt - week}::double precision / 1000) + interval '0.0005 seconds' WHERE id = ${call.id}`);
+    const service = voiceInboundRetention(db), [entry] = await service.load({ kind: "admission", id: call.id });
+    expect(entry.dueAt).toBe(dueAt + 1);
+    vi.setSystemTime(dueAt); await service.check(call.id); expect(await row(call.id)).toBeDefined();
+    vi.setSystemTime(dueAt + 1); await service.check(call.id); expect(await row(call.id)).toBeUndefined();
+  });
+  it("never prunes unconfirmed, live, task-bound or session-bound history", async () => {
+    const unconfirmed = await admission({ cleanupConfirmedAt: null });
+    const live = await admission({ state: "awaiting_approval" });
+    const task = await admission({ intakeIssueId: issueId });
+    const bound = await admission(), session = await owner(bound);
+    await db.update(chatVoiceInboundCalls).set({ sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, bound.id));
+    const service = await start(); await flush(); await service.prune();
+    for (const call of [unconfirmed, live, task, bound]) expect(await row(call.id)).toBeDefined();
+    expect(coordinator!.nextWakeAt()).toBeNull(); await quiet();
+  });
+  it("rechecks current dates and bindings after stale discovery", async () => {
+    const service = voiceInboundRetention(db), changed = await admission(), bound = await admission();
+    expect(await service.load()).toHaveLength(2);
+    await db.update(chatVoiceInboundCalls).set({ updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, changed.id));
+    await db.update(chatVoiceInboundCalls).set({ intakeIssueId: issueId }).where(eq(chatVoiceInboundCalls.id, bound.id));
+    expect((await service.check(changed.id))!.dueAt).toBeGreaterThan(Date.now());
+    expect(await service.check(bound.id)).toBeNull();
+    expect(await row(changed.id)).toBeDefined(); expect(await row(bound.id)).toBeDefined();
+  });
+  it("waits for a concurrent binding transaction and rechecks after its commit", async () => {
+    const call = await admission(), service = voiceInboundRetention(db);
+    let release!: () => void, acquired!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const ready = new Promise<void>(resolve => { acquired = resolve; });
+    const binding = db.transaction(async tx => {
+      await tx.update(chatVoiceInboundCalls).set({ intakeIssueId: issueId }).where(eq(chatVoiceInboundCalls.id, call.id));
+      acquired(); await gate;
+    });
+    await ready;
+    let finished = false;
+    const deleting = service.check(call.id).then(result => { finished = true; return result; });
+    try {
+      await vi.advanceTimersByTimeAsync(100); expect(finished).toBe(false);
+    } finally { release(); await binding; }
+    expect(await deleting).toBeNull(); expect((await row(call.id)).intakeIssueId).toBe(issueId);
+  });
+  it("defers nested writes until outer commit and discards rollback", async () => {
+    await start(); await flush(); let committed!: string;
+    await db.transaction(async tx => {
+      committed = (await admission({}, tx as typeof db)).id;
+      await vi.advanceTimersByTimeAsync(5000);
+      expect(await row(committed)).toBeUndefined();
+      expect(await tx.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, committed))).toHaveLength(1);
+    });
+    await flush(); expect(await row(committed)).toBeUndefined(); await quiet();
+    await expect(db.transaction(async tx => { await admission({}, tx as typeof db); throw new Error("rollback"); })).rejects.toThrow("rollback");
+    await flush(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet();
+  });
+  it("keeps the first write during drain and runs it on resume", async () => {
+    await start(); await flush(); enabled = false;
+    const call = await admission(); await quiet(60_000); expect(await row(call.id)).toBeDefined();
+    enabled = true; await flush(); expect(await row(call.id)).toBeUndefined(); await quiet();
+  });
+  it("pages through more than one batch of history across companies", async () => {
+    for (let i = 0; i < 104; i++) await admission();
+    const other = await fixture(); const last = await admission({ companyId: other.companyId, endpointId: other.endpointId });
+    const service = await start();
+    expect(await row(last.id)).toBeUndefined();
+    await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(await service.load()).toHaveLength(0); }, { timeout: 5000 });
+    await flush(); await quiet();
+  });
+  it("retains failed deletion for a bounded retry and then becomes quiet", async () => {
+    const call = await admission();
+    const transaction = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("database unavailable"));
+    await start(); transaction.mockRestore();
+    expect(errors).toHaveLength(1); errors.length = 0;
+    expect(await row(call.id)).toBeDefined();
+    await quiet(10_000);
+    await vi.advanceTimersByTimeAsync(31_000); await flush(); expect(await row(call.id)).toBeUndefined(); await quiet();
+  });
+  it("stops future deletion at shutdown and resumes its durable deadline after restart", async () => {
+    const call = await admission({ updatedAt: new Date(Date.now() - week + 60_000) });
+    await start(); await flush(); await coordinator!.stop();
+    await quiet(120_000); expect(await row(call.id)).toBeDefined();
+    await start(); await flush(); expect(await row(call.id)).toBeUndefined(); await quiet();
+  });
+});

server/src/__tests__/voice-session-store.integration.test.ts

@@ -78,6 +78,40 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication };
   }
 
+  it.each(["signed callback", "cleanup receipt"])("schedules retained history from a real %s", async confirmation => {
+    const f = await inboundFixture();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    const settle = () => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0);
+    }, { timeout: 5000 });
+    const quiet = async (ms = 120_000) => {
+      const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction");
+      try { await vi.advanceTimersByTimeAsync(ms); expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); }
+      finally { reads.mockRestore(); transactions.mockRestore(); }
+    };
+    try {
+      await f.service.inbound.registerRetentionWork(coordinator, () => true).ready;
+      await settle(); expect(coordinator.nextWakeAt()).toBeNull(); await quiet(10_000);
+      if (confirmation === "signed callback") {
+        const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" });
+        await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers);
+      } else {
+        await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
+          { approve: false, approvalCode: f.call.approvalCode });
+        expect(f.transport.endSession).toHaveBeenCalledWith(f.call.providerSessionId);
+      }
+      await settle();
+      const [saved] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
+      expect(saved.cleanupConfirmedAt).not.toBeNull();
+      const dueAt = Math.max(saved.updatedAt.getTime(), saved.expiresAt.getTime()) + 7 * 86_400_000;
+      expect(coordinator.nextWakeAt()).toBe(dueAt); await quiet();
+      await vi.advanceTimersByTimeAsync(dueAt - Date.now() + 1000); await settle();
+      expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(0);
+      expect(coordinator.nextWakeAt()).toBeNull(); await quiet(); expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); vi.useRealTimers(); }
+  });
+
   it("wakes inbound cleanup from a real signed pre-call and retains an unconfirmed end", async () => {
     const f = await inboundFixture();
     const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" });

server/src/app.ts

@@ -1301,6 +1301,7 @@ export async function createApp(
   chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.voice.registerReplyWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
+  chatChannels.voice.inbound.registerRetentionWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   emailChannels.start();

server/src/services/delivery-work-notifications.ts

@@ -42,6 +42,7 @@ export const DELIVERY_QUEUES = {
   chatTelegramMaintenance: "chat-telegram-maintenance",
   chatReceivers: "chat-receivers",
   voiceSessionRecovery: "voice-session-recovery",
+  voiceInboundRetention: "voice-inbound-retention",
   voiceInboundCleanup: "voice-inbound-cleanup",
   voiceReplies: "voice-reply-delivery",
   chatCommandRegistration: "chat-command-registration",

server/src/services/voice/voice-inbound-notifications.ts

@@ -5,9 +5,11 @@ export function voiceInboundScope(hint: string): VoiceInboundScope | undefined {
   if (hint === "*") return undefined;
   const [kind, id, extra] = hint.split(":");
   if (!id || extra !== undefined || !["admission", "session", "endpoint", "connection", "company"].includes(kind))
-    throw new Error("Invalid inbound voice cleanup scope");
+    throw new Error("Invalid inbound voice scope");
   return { kind: kind as VoiceInboundScope["kind"], id };
 }
-export function notifyVoiceInbound(tx: object, scope?: VoiceInboundScope) {
-  return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceInboundCleanup, scope ? `${scope.kind}:${scope.id}` : "*");
+export async function notifyVoiceInbound(tx: object, scope?: VoiceInboundScope) {
+  const hint = scope ? `${scope.kind}:${scope.id}` : "*";
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceInboundCleanup, hint);
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceInboundRetention, hint);
 }

server/src/services/voice/voice-inbound-retention.ts

@@ -0,0 +1,65 @@
+import { and, eq, gt, inArray, isNotNull, isNull, lte, sql } from "drizzle-orm";
+import { chatEndpoints, chatVoiceInboundCalls, type Db } from "@paperclipai/db";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import { voiceInboundScope, type VoiceInboundScope } from "./voice-inbound-notifications.js";
+
+type RetentionEntry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number };
+const RETENTION_MS = 7 * 86_400_000;
+const eligible = () => and(isNull(chatVoiceInboundCalls.sessionId), isNull(chatVoiceInboundCalls.intakeIssueId),
+  isNotNull(chatVoiceInboundCalls.cleanupConfirmedAt), inArray(chatVoiceInboundCalls.state, ["denied", "expired", "ended"]));
+const due = (now: number) => and(eligible(), lte(chatVoiceInboundCalls.expiresAt, new Date(now - RETENTION_MS)),
+  lte(chatVoiceInboundCalls.updatedAt, new Date(now - RETENTION_MS)));
+
+/** Retain the existing seven-day policy for both timestamps. SQL rounds the wake
+ * up to a millisecond so PostgreSQL submillisecond timestamps cannot fire early. */
+export function voiceInboundRetention(db: Db) {
+  async function load(scope?: VoiceInboundScope, signal?: AbortSignal): Promise<RetentionEntry[]> {
+    if (scope?.kind === "session") return []; // Bound call history is never ours.
+    const entries: RetentionEntry[] = [];
+    let cursor: string | undefined;
+    for (;;) {
+      signal?.throwIfAborted();
+      const rows = await db.select({ id: chatVoiceInboundCalls.id, companyId: chatVoiceInboundCalls.companyId,
+        endpointId: chatVoiceInboundCalls.endpointId, connectionId: chatEndpoints.connectionId,
+        dueAt: sql<string>`ceil(extract(epoch from greatest(${chatVoiceInboundCalls.expiresAt}, ${chatVoiceInboundCalls.updatedAt})) * 1000) + ${RETENTION_MS}`,
+      }).from(chatVoiceInboundCalls)
+        .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceInboundCalls.endpointId), eq(chatEndpoints.companyId, chatVoiceInboundCalls.companyId)))
+        .where(and(eligible(),
+          scope?.kind === "admission" ? eq(chatVoiceInboundCalls.id, scope.id) : undefined,
+          scope?.kind === "endpoint" ? eq(chatVoiceInboundCalls.endpointId, scope.id) : undefined,
+          scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined,
+          scope?.kind === "company" ? eq(chatVoiceInboundCalls.companyId, scope.id) : undefined,
+          cursor ? gt(chatVoiceInboundCalls.id, cursor) : undefined)).orderBy(chatVoiceInboundCalls.id).limit(100);
+      entries.push(...rows.map(row => ({ ...row, dueAt: Number(row.dueAt) })));
+      if (rows.length < 100) return entries;
+      cursor = rows[rows.length - 1].id;
+    }
+  }
+  async function check(id: string, signal?: AbortSignal): Promise<RetentionEntry | null> {
+    signal?.throwIfAborted();
+    await db.transaction(async tx => {
+      // Eligibility and both age conditions are current-state predicates under
+      // the admission lock. Cached discovery cannot delete a changed binding.
+      const [row] = await tx.select({ id: chatVoiceInboundCalls.id }).from(chatVoiceInboundCalls)
+        .where(and(eq(chatVoiceInboundCalls.id, id), due(Date.now()))).for("update");
+      if (!row) return;
+      signal?.throwIfAborted();
+      await tx.delete(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.id, id), due(Date.now())));
+    });
+    return (await load({ kind: "admission", id }, signal))[0] ?? null;
+  }
+  /** Explicit compatibility path; never register this bulk operation on a tick. */
+  async function prune() {
+    await db.delete(chatVoiceInboundCalls).where(due(Date.now()));
+  }
+  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.voiceInboundRetention, decodeScope: voiceInboundScope,
+      matches: (entry, scope) => !scope || scope.kind !== "session" && (scope.kind === "admission" ? entry.id : scope.kind === "endpoint" ? entry.endpointId : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id,
+      load, check: (entry, signal) => check(entry.id, signal),
+    }, canRun);
+  }
+  return { load, check, prune, register };
+}

server/src/services/voice/voice-inbound-service.ts

@@ -1,9 +1,10 @@
+import { voiceInboundRetention } from "./voice-inbound-retention.js";
 import { voiceInboundCleanup } from "./voice-inbound-cleanup.js";
 import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
 import { notifyVoiceChange } from "./voice-recovery-notifications.js";
 import { logger } from "../../middleware/logger.js";
 import { createHash, createHmac, randomInt, randomUUID, timingSafeEqual } from "node:crypto";
-import { and, asc, desc, eq, gte, inArray, lte, isNull, isNotNull, sql } from "drizzle-orm";
+import { and, desc, eq, gte, inArray, isNull, sql } from "drizzle-orm";
 import { z } from "zod";
 import { chatEndpointResources, chatConversations, chatActions, chatEndpoints, chatVoiceInboundCalls, chatVoicePhoneLines, chatVoiceReports, chatVoiceSessions, activityLog, type Db } from "@paperclipai/db";
 import { LOW_TRUST_REVIEW_PRESET, LOW_TRUST_REVIEW_PRESET_VERSION, LOW_TRUST_REVIEW_RAW_OUTPUT_DISPOSITION } from "@paperclipai/shared";
@@ -283,15 +284,13 @@ export function voiceInboundService(db: Db, options: {
     if ((await client.inspectSession(call.providerSessionId)).endedAt) return true;
     return (await client.endSession(call.providerSessionId)).confirmed;
   });
-  async function pruneHistory() {
-    // Retention cannot erase an unconfirmed physical cleanup obligation.
-    await db.delete(chatVoiceInboundCalls).where(and(isNull(chatVoiceInboundCalls.sessionId), isNull(chatVoiceInboundCalls.intakeIssueId), isNotNull(chatVoiceInboundCalls.cleanupConfirmedAt), inArray(chatVoiceInboundCalls.state, ["denied", "expired", "ended"]), lte(chatVoiceInboundCalls.expiresAt, new Date(Date.now() - 7 * 86_400_000)), lte(chatVoiceInboundCalls.updatedAt, new Date(Date.now() - 7 * 86_400_000))));
-  }
-  /** Explicit one-shot compatibility path. Recurring maintenance only owns retention. */
+  const retention = voiceInboundRetention(db);
+  const pruneHistory = retention.prune;
+  /** Explicit one-shot compatibility path; never used by periodic maintenance. */
   async function reconcile(limit: number) {
     const rows = (await cleanup.load()).filter(row => row.dueAt <= Date.now()).sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, limit);
     for (const row of rows) await cleanup.check(row.id).catch(error => logger.warn({ err: error, callId: row.id }, "Incoming call cleanup remains pending"));
     await pruneHistory();
   }
-  return {configuration, saveConfiguration, lifecycle, pendingTool, list, decide, reconcile, pruneHistory, registerCleanupWork: cleanup.register, authorizeGuest, history};
+  return {configuration, saveConfiguration, lifecycle, pendingTool, list, decide, reconcile, pruneHistory, registerCleanupWork: cleanup.register, registerRetentionWork: retention.register, authorizeGuest, history};
 }

server/src/services/voice/voice-session-service.ts

@@ -230,7 +230,6 @@ export function voiceSessionService(db: Db, options: {
     return registerVoiceReplyWork(coordinator, replies, canRun);
   }
   async function reconcileMaintenance(limit = 25) {
-    await inbound.pruneHistory();
     await reconcileReports(limit);
   }
   /** Explicit one-shot recovery for diagnostics/tests; never register on a recurring lane. */