fix(server): schedule inbound voice cleanup from durable obligations

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

server/src/services/chat-channels/company-deletion.ts

@@ -1,9 +1,14 @@
+import { assertVoiceInboundCleanupSettled } from "../voice/voice-inbound-cleanup.js";
 import { chatEndpoints } from "@paperclipai/db";
 import { eq } from "drizzle-orm";
 import type { CompanyDeletionParticipant } from "../../lib/company-deletion.js";
 
 export const chatCompanyDeletion: CompanyDeletionParticipant = {
   async deleteCompanyData(tx, companyId) {
+    // Ingress locks the endpoint before admitting a provider call. Hold that
+    // lock through the cleanup proof and delete so a new call cannot slip in.
+    await tx.select({ id: chatEndpoints.id }).from(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)).for("update");
+    await assertVoiceInboundCleanupSettled(tx, companyId);
     await tx.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId));
   },
 };

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",
+  voiceInboundCleanup: "voice-inbound-cleanup",
   voiceReplies: "voice-reply-delivery",
   chatCommandRegistration: "chat-command-registration",
   chatTelegramSubscriptions: "chat-telegram-subscriptions",

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

@@ -0,0 +1,123 @@
+import { and, eq, gt, isNull } from "drizzle-orm";
+import { chatEndpoints, chatVoiceInboundCalls, chatVoiceSessions, type Db } from "@paperclipai/db";
+import { conflict } from "../../errors.js";
+import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import { notifyVoiceInbound, voiceInboundScope, type VoiceInboundScope } from "./voice-inbound-notifications.js";
+import { notifyVoiceRecovery } from "./voice-recovery-notifications.js";
+
+type Admission = typeof chatVoiceInboundCalls.$inferSelect;
+type Session = typeof chatVoiceSessions.$inferSelect;
+export type VoiceInboundEntry = { id: string; sessionId: string | null; companyId: string; endpointId: string; connectionId: string; dueAt: number };
+const terminal = (session: Session) => ["ended", "failed", "expired"].includes(session.state) && session.endedAt !== null;
+const rejected = (call: Admission) => ["denied", "expired", "ended"].includes(call.state);
+
+/** A bound call belongs to live-session recovery. Admission cleanup may only
+ * hand off a rejection or copy confirmed termination; it never ends that leg. */
+export function inboundCleanupAt(call: Admission, session: Session | null): number | null {
+  if (call.cleanupConfirmedAt) return null;
+  if (session) {
+    if (terminal(session) || (call.state === "approving" && session.callerAuthority !== "guest_intake")) return 0;
+    return rejected(call) && session.state !== "ending" ? 0 : null;
+  }
+  if (call.cleanupNextAttemptAt) return call.cleanupNextAttemptAt.getTime();
+  return ["awaiting_approval", "guest_intake", "approving"].includes(call.state) ? call.expiresAt.getTime() : 0;
+}
+
+/** Caller locks company endpoints before this check, matching admission ingress. */
+export async function assertVoiceInboundCleanupSettled(db: Pick<Db, "select">, companyId: string) {
+  const [pending] = await db.select({ id: chatVoiceInboundCalls.id }).from(chatVoiceInboundCalls)
+    .where(and(eq(chatVoiceInboundCalls.companyId, companyId), isNull(chatVoiceInboundCalls.cleanupConfirmedAt))).limit(1);
+  if (pending) throw conflict("Incoming phone calls must finish cleanup before deleting this company");
+}
+
+export function voiceInboundCleanup(db: Db, terminate: (call: Admission) => Promise<boolean>) {
+  async function load(scope?: VoiceInboundScope, signal?: AbortSignal): Promise<VoiceInboundEntry[]> {
+    const entries: VoiceInboundEntry[] = [];
+    let cursor: string | undefined;
+    for (;;) {
+      signal?.throwIfAborted();
+      const rows = await db.select({ call: chatVoiceInboundCalls, session: chatVoiceSessions, connectionId: chatEndpoints.connectionId })
+        .from(chatVoiceInboundCalls)
+        .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceInboundCalls.endpointId), eq(chatEndpoints.companyId, chatVoiceInboundCalls.companyId)))
+        .leftJoin(chatVoiceSessions, and(eq(chatVoiceSessions.id, chatVoiceInboundCalls.sessionId), eq(chatVoiceSessions.companyId, chatVoiceInboundCalls.companyId)))
+        .where(and(isNull(chatVoiceInboundCalls.cleanupConfirmedAt),
+          scope?.kind === "admission" ? eq(chatVoiceInboundCalls.id, scope.id) : undefined,
+          scope?.kind === "session" ? eq(chatVoiceInboundCalls.sessionId, 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);
+      for (const { call, session, connectionId } of rows) {
+        const dueAt = inboundCleanupAt(call, session);
+        if (dueAt !== null) entries.push({ id: call.id, sessionId: call.sessionId, companyId: call.companyId, endpointId: call.endpointId, connectionId, dueAt });
+      }
+      if (rows.length < 100) return entries;
+      cursor = rows[rows.length - 1].call.id;
+    }
+  }
+  async function check(id: string, signal?: AbortSignal): Promise<VoiceInboundEntry | null> {
+    signal?.throwIfAborted();
+    const claimed = await db.transaction(async tx => {
+      // Discovery grants no ownership. Approval and lifecycle use this same lock,
+      // always before the session lock; no provider request runs in this TX.
+      const [call] = await tx.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, id)).for("update");
+      if (!call || call.cleanupConfirmedAt) return null;
+      const [session] = await tx.select().from(chatVoiceSessions).where(and(
+        eq(chatVoiceSessions.companyId, call.companyId), eq(chatVoiceSessions.endpointId, call.endpointId),
+        eq(chatVoiceSessions.providerSessionId, call.providerSessionId))).for("update");
+      if (session) {
+        if (call.sessionId !== session.id || (call.state === "approving" && session.callerAuthority !== "guest_intake")) {
+          // Repair older commits that inserted/promoted an authorized session
+          // before the separate admission binding write. Never infer approval
+          // from a matching provider ID alone.
+          if (!["approving", "approved"].includes(call.state) || !call.approvedByUserId
+            || session.idempotencyKey !== `inbound:${call.id}` || session.callerId !== call.approvedByUserId
+            || session.approvedByUserId !== call.approvedByUserId || session.callerAuthority !== call.callerAuthority
+            || session.generation !== call.generation || session.credentialFingerprint !== call.credentialFingerprint
+            || session.toolTokenHash !== call.toolTokenHash || (call.requestedIssueId !== null && session.issueId !== call.requestedIssueId))
+            throw conflict("Incoming call cleanup has conflicting session ownership");
+          await notifyVoiceInbound(tx, { kind: "admission", id });
+          await tx.update(chatVoiceInboundCalls).set({ state: "approved", sessionId: session.id, updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, id));
+          call.state = "approved"; call.sessionId = session.id;
+        }
+        if (terminal(session)) {
+          await notifyVoiceInbound(tx, { kind: "admission", id });
+          await tx.update(chatVoiceInboundCalls).set({ ...(!rejected(call) ? { state: "ended" as const } : {}), cleanupConfirmedAt: session.endedAt, cleanupNextAttemptAt: null, updatedAt: new Date() }).where(eq(chatVoiceInboundCalls.id, id));
+        } else if (rejected(call) && session.state !== "ending") {
+          await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
+          await tx.update(chatVoiceSessions).set({ state: "ending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
+        }
+        return null;
+      }
+      if (call.sessionId) throw conflict("Incoming call cleanup owner is missing");
+      const dueAt = inboundCleanupAt(call, null);
+      if (dueAt === null || dueAt > Date.now()) return null;
+      await notifyVoiceInbound(tx, { kind: "admission", id });
+      const [claim] = await tx.update(chatVoiceInboundCalls).set({
+        ...(!rejected(call) ? { state: "expired" as const } : {}),
+        cleanupNextAttemptAt: new Date(Date.now() + 30_000), updatedAt: new Date(),
+      }).where(eq(chatVoiceInboundCalls.id, id)).returning();
+      return claim;
+    });
+    if (claimed) {
+      signal?.throwIfAborted();
+      if (await terminate(claimed)) await db.transaction(async tx => {
+        await notifyVoiceInbound(tx, { kind: "admission", id });
+        await tx.update(chatVoiceInboundCalls).set({ cleanupConfirmedAt: new Date(), cleanupNextAttemptAt: null, updatedAt: new Date() })
+          .where(and(eq(chatVoiceInboundCalls.id, id), isNull(chatVoiceInboundCalls.cleanupConfirmedAt), eq(chatVoiceInboundCalls.cleanupNextAttemptAt, claimed.cleanupNextAttemptAt!)));
+      });
+    }
+    return (await load({ kind: "admission", id }, signal))[0] ?? null;
+  }
+  function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+    return registerScopedDeadlineWork(coordinator, {
+      queue: DELIVERY_QUEUES.voiceInboundCleanup, decodeScope: voiceInboundScope,
+      matches: (entry, scope) => !scope || (scope.kind === "admission" ? entry.id : scope.kind === "session" ? entry.sessionId : 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, register };
+}

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

@@ -0,0 +1,13 @@
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
+
+export type VoiceInboundScope = { kind: "admission" | "session" | "endpoint" | "connection" | "company"; id: string };
+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");
+  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}` : "*");
+}

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

@@ -1,7 +1,9 @@
+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, sql } from "drizzle-orm";
+import { and, asc, desc, eq, gte, inArray, lte, isNull, isNotNull, 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";
@@ -120,7 +122,8 @@ export function voiceInboundService(db: Db, options: {
         if (prior) return prior;
         const recent = await tx.select({state: chatVoiceInboundCalls.state, expiresAt: chatVoiceInboundCalls.expiresAt}).from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, row.companyId), eq(chatVoiceInboundCalls.endpointId, row.id), gte(chatVoiceInboundCalls.createdAt, new Date(Date.now() - 3_600_000)))).limit(60);
         if (recent.length >= 60 || recent.filter(call => ["guest_intake", "awaiting_approval", "approving"].includes(call.state) && call.expiresAt.getTime() > Date.now()).length >= 10) throw conflict("This phone line has reached its incoming call limit. Try again later.");
-        const [admission] = await tx.insert(chatVoiceInboundCalls).values({companyId: row.companyId, endpointId: row.id, providerSessionId, state: line.guestIntake ? "guest_intake" : "awaiting_approval", approvalCode: String(randomInt(100000, 1000000)), generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(capability), requestFingerprint: fingerprint, callerAuthority: line.guestIntake ? "guest_intake" : null, expiresAt: new Date(Date.now() + (line.guestIntake ? 600_000 : 120_000))}).onConflictDoNothing().returning();
+        await notifyVoiceInbound(tx, { kind: "endpoint", id: row.id });
+        const [admission] = await tx.insert(chatVoiceInboundCalls).values({companyId: row.companyId, endpointId: row.id, providerSessionId, providerAccountId: event.organization_id, state: line.guestIntake ? "guest_intake" : "awaiting_approval", approvalCode: String(randomInt(100000, 1000000)), generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(capability), requestFingerprint: fingerprint, callerAuthority: line.guestIntake ? "guest_intake" : null, expiresAt: new Date(Date.now() + (line.guestIntake ? 600_000 : 120_000))}).onConflictDoNothing().returning();
         return admission;
       });
       if (!existing || existing.requestFingerprint !== fingerprint || existing.toolTokenHash !== voiceTokenHash(capability) || !["awaiting_approval", "guest_intake"].includes(existing.state) || existing.expiresAt.getTime() <= Date.now()) throw forbidden("Incoming call is no longer pending");
@@ -155,7 +158,10 @@ export function voiceInboundService(db: Db, options: {
           await tx.update(chatVoiceSessions).set({state: event.status === "failed" ? "failed" : "ended", endedAt: session.endedAt ?? new Date(), updatedAt: new Date()}).where(eq(chatVoiceSessions.id, session.id));
           await tx.insert(chatVoiceReports).values({companyId: row.companyId, sessionId: session.id}).onConflictDoUpdate({target: chatVoiceReports.sessionId, set: {nextCheckAt: new Date()}});
         }
-        if (admission) await tx.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, admission.id));
+        if (admission) {
+          await notifyVoiceInbound(tx, { kind: "admission", id: admission.id });
+          await tx.update(chatVoiceInboundCalls).set({state: "ended", cleanupConfirmedAt: new Date(), cleanupNextAttemptAt: null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, admission.id));
+        }
         await tx.insert(activityLog).values({companyId: row.companyId, actorType: "system", actorId: "speko-lifecycle", action: "voice.call.ended", entityType: session ? "voice_session" : "voice_inbound_call", entityId: session?.id ?? admission!.id, details: {endpointId: row.id}});
       }
       return true;
@@ -174,7 +180,7 @@ export function voiceInboundService(db: Db, options: {
       const [call] = await tx.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, row.companyId), eq(chatVoiceInboundCalls.endpointId, row.id), eq(chatVoiceInboundCalls.providerSessionId, envelope.session_id))).for("update");
       const hash = voiceTokenHash(capability);
       if (!call || !timingSafeEqual(Buffer.from(hash), Buffer.from(call.toolTokenHash)) || context.generation !== call.generation || context.fingerprint !== call.credentialFingerprint) throw forbidden("Invalid incoming call credential");
-      if (call.expiresAt.getTime() <= Date.now() && ["guest_intake", "awaiting_approval", "approving"].includes(call.state)) { await tx.update(chatVoiceInboundCalls).set({state: "expired", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); return {authorization: "expired", status: "unavailable", updates: [], cursor: 0}; }
+      if (call.expiresAt.getTime() <= Date.now() && ["guest_intake", "awaiting_approval", "approving"].includes(call.state)) { await notifyVoiceInbound(tx, { kind: "admission", id: call.id }); await tx.update(chatVoiceInboundCalls).set({state: "expired", cleanupNextAttemptAt: null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); return {authorization: "expired", status: "unavailable", updates: [], cursor: 0}; }
       if (call.state === "guest_intake") {
         const [line] = await tx.select().from(chatVoicePhoneLines).where(and(eq(chatVoicePhoneLines.companyId, row.companyId), eq(chatVoicePhoneLines.endpointId, row.id), eq(chatVoicePhoneLines.enabled, true), eq(chatVoicePhoneLines.guestIntake, true))).for("share");
         if (!line) throw forbidden("Guest intake is no longer enabled");
@@ -230,6 +236,7 @@ export function voiceInboundService(db: Db, options: {
       if (!["guest_intake", "awaiting_approval", "approving"].includes(row.state)) throw conflict("This incoming call is no longer waiting for approval");
       if (row.state === "approving" && (!input.approve || row.requestedIssueId !== (input.issueId ?? row.intakeIssueId ?? null))) throw conflict("This call is already being approved");
       if (input.issueId) await store.authorizeTask(tx, companyId, input.issueId, caller);
+      await notifyVoiceInbound(tx, { kind: "admission", id: row.id });
       const [changed] = await tx.update(chatVoiceInboundCalls).set({state: input.approve ? "approving" : row.state === "guest_intake" ? "ended" : "denied", approvedByUserId: caller.id, callerAuthority: caller.authority, requestedIssueId: input.issueId ?? row.intakeIssueId ?? null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, row.id)).returning();
       await tx.insert(activityLog).values({companyId, actorType: "user", actorId: caller.id, action: input.approve ? "voice.inbound.approval_started" : row.state === "guest_intake" ? "voice.low_trust_call.ended" : "voice.inbound.denied", entityType: "voice_inbound_call", entityId: row.id});
       if (!input.approve && row.state === "guest_intake" && row.sessionId) {
@@ -239,10 +246,21 @@ export function voiceInboundService(db: Db, options: {
       return changed;
     });
     if (admission.state === "approved") return projection(admission);
+    if (!input.approve) {
+      // Bound calls have already been handed to live-session recovery. Unbound
+      // rejection uses the same durable claim as scheduled cleanup.
+      await cleanup.check(admission.id).catch(error => logger.warn({ err: error, callId: admission.id }, "Incoming call termination remains pending"));
+      return projection(admission);
+    }
     const row = await endpoint(companyId, endpointId), credentials = await options.credentials(row, admission.credentialFingerprint);
     const client = provider(credentials.apiKey);
-    if (!input.approve) { await client.endSession(admission.providerSessionId).catch(() => undefined); return projection(admission); }
-    if ((await client.inspectSession(admission.providerSessionId)).endedAt) { await db.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, callId)); throw conflict("The caller already hung up"); }
+    if ((await client.inspectSession(admission.providerSessionId)).endedAt) {
+      await db.transaction(async tx => {
+        await notifyVoiceInbound(tx, { kind: "admission", id: callId });
+        await tx.update(chatVoiceInboundCalls).set({state: "ended", cleanupConfirmedAt: new Date(), cleanupNextAttemptAt: null, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, callId));
+      });
+      throw conflict("The caller already hung up");
+    }
     const result = await options.start({companyId, endpointId, issueId: admission.requestedIssueId ?? undefined, caller, idempotencyKey: `inbound:${admission.id}`, maxDurationSeconds: 600, mode: "inbound_phone", providerSessionId: admission.providerSessionId, toolToken: token(row, admission.providerSessionId, admission.generation, credentials.signingSecret)});
     [admission] = await db.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.id, callId), eq(chatVoiceInboundCalls.state, "approved"), eq(chatVoiceInboundCalls.sessionId, result.session.id)));
     if (!admission) throw conflict("This call ended during approval");
@@ -254,17 +272,26 @@ export function voiceInboundService(db: Db, options: {
     try { await db.transaction(tx => store.authorizeSession(tx, session.companyId, sessionId)); return true; }
     catch (error) { if (error instanceof Error && "status" in error && [403, 404].includes(Number(error.status))) return false; throw error; }
   }
-  async function reconcile(limit: number) {
-    await db.delete(chatVoiceInboundCalls).where(and(isNull(chatVoiceInboundCalls.sessionId), isNull(chatVoiceInboundCalls.intakeIssueId), 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))));
-    const rows = await db.select().from(chatVoiceInboundCalls).where(and(inArray(chatVoiceInboundCalls.state, ["guest_intake", "awaiting_approval", "approving", "denied", "expired"]), lte(chatVoiceInboundCalls.expiresAt, new Date()))).orderBy(asc(chatVoiceInboundCalls.updatedAt)).limit(limit);
-    for (const call of rows) {
-      try {
-        const row = await endpoint(call.companyId, call.endpointId), credentials = await options.credentials(row);
-        const client = provider(credentials.apiKey), state = await client.inspectSession(call.providerSessionId);
-        if (state.endedAt) await db.update(chatVoiceInboundCalls).set({state: "ended", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id));
-        else { await db.update(chatVoiceInboundCalls).set({state: "expired", updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); await client.endSession(call.providerSessionId); }
-      } catch { await db.update(chatVoiceInboundCalls).set({updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id)); }
-    }
+  const cleanup = voiceInboundCleanup(db, async call => {
+    const row = await endpoint(call.companyId, call.endpointId);
+    // New admissions bind the signed provider account, allowing key rotation
+    // within that account. Older rows require the original credential version.
+    const credentials = await options.credentials(row, call.providerAccountId ? undefined : call.credentialFingerprint);
+    const client = provider(credentials.apiKey);
+    if (call.providerAccountId && (await client.verifyAgent(credentials.agentId)).organizationId !== call.providerAccountId)
+      throw conflict("Incoming call belongs to a different voice provider account");
+    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))));
   }
-  return {configuration, saveConfiguration, lifecycle, pendingTool, list, decide, reconcile, authorizeGuest, history};
+  /** Explicit one-shot compatibility path. Recurring maintenance only owns retention. */
+  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};
 }

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

@@ -1,3 +1,4 @@
+import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
 import { notifyVoiceReplies } from "./voice-reply-notifications.js";
 import type { Db } from "@paperclipai/db";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
@@ -18,6 +19,7 @@ export function notifyVoiceRecovery(tx: object, scope?: VoiceRecoveryScope) {
 export async function notifyVoiceChange(tx: object, scope?: VoiceRecoveryScope) {
   await notifyVoiceRecovery(tx, scope);
   await notifyVoiceReplies(tx, scope);
+  await notifyVoiceInbound(tx, scope);
 }
 export function withVoiceSessionChange<T>(db: Db, id: string, write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>, options: { replyDelivery?: boolean } = {}): Promise<T> {
   return db.transaction(async tx => {

server/src/services/voice/voice-reply-push.ts

@@ -1,3 +1,4 @@
+import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
 import { notifyVoiceReplies } from "./voice-reply-notifications.js";
 import { recoverVoiceReplyDispatches } from "./voice-reply-dispatch-recovery.js";
 import { notifyVoiceRecovery } from "./voice-recovery-notifications.js";
@@ -83,6 +84,7 @@ export async function pushVoiceReplies(db: Db, options: {
           await notifyVoiceReplies(tx, { kind: "session", id: candidate.id });
           await tx.update(chatActions).set({status, result: {errorCode: providerError?.code ?? "voice_authority_unavailable", ...(status === "retry" ? {retryAt: Date.now() + 10_000} : {})}, updatedAt: new Date()}).where(eq(chatActions.id, claim.action.id));
           if (status === "call_ended" || status === "unknown") await notifyVoiceRecovery(tx, { kind: "session", id: candidate.id });
+          if (status === "call_ended") await notifyVoiceInbound(tx, { kind: "session", id: candidate.id });
           if (status === "call_ended") await tx.update(chatVoiceSessions).set({state: "ended", endedAt: new Date(), updatedAt: new Date()}).where(and(eq(chatVoiceSessions.id, candidate.id), eq(chatVoiceSessions.companyId, candidate.companyId)));
           if (status === "unknown") await tx.update(chatVoiceSessions).set({errorCode: "message_delivery_unknown", updatedAt: new Date()}).where(eq(chatVoiceSessions.id, candidate.id));
         });

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

@@ -230,7 +230,7 @@ export function voiceSessionService(db: Db, options: {
     return registerVoiceReplyWork(coordinator, replies, canRun);
   }
   async function reconcileMaintenance(limit = 25) {
-    await inbound.reconcile(limit);
+    await inbound.pruneHistory();
     await reconcileReports(limit);
   }
   /** Explicit one-shot recovery for diagnostics/tests; never register on a recurring lane. */
@@ -238,7 +238,8 @@ export function voiceSessionService(db: Db, options: {
     const due = (await recovery.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 due) await recovery.check(row.id).catch(() => undefined);
     await pushReplies(limit);
-    await reconcileMaintenance(limit);
+    await inbound.reconcile(limit);
+    await reconcileReports(limit);
     return due.length;
   }
   async function pushReplies(limit = 25, companyId?: string, endpointId?: string) {

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

@@ -1,3 +1,4 @@
+import { notifyVoiceInbound } from "./voice-inbound-notifications.js";
 import { notifyVoiceReplies } from "./voice-reply-notifications.js";
 import { notifyVoiceChange } from "./voice-recovery-notifications.js";
 import { notifyChatDeliveryWork } from "../chat-work-notifications.js";
@@ -180,6 +181,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
         if (!inboundAdmissionId) return;
         // Admission expiry must relinquish cleanup in the same commit that
         // establishes an authorized owner for the existing provider call.
+        await notifyVoiceInbound(tx, { kind: "admission", id: inboundAdmissionId });
         await tx.update(chatVoiceInboundCalls).set({ state: "approved", sessionId, updatedAt: new Date() })
           .where(eq(chatVoiceInboundCalls.id, inboundAdmissionId));
       }