diff --git a/doc/idle-voice-inbound-retention.md b/doc/idle-voice-inbound-retention.md new file mode 100644 index 00000000000..4e09180ccc9 --- /dev/null +++ b/doc/idle-voice-inbound-retention.md @@ -0,0 +1,36 @@ +# Inbound voice history retention + +Inbound history retention uses the shared work scheduler. It does not scan on +the chat maintenance tick. Call-report reconciliation remains a separate duty. + +An admission is eligible only when it is denied, expired or ended, provider +cleanup is confirmed, and neither a session nor an intake task owns it. The +existing policy retains it until both expiry and last update are seven days old: +`max(expiresAt, updatedAt) + 7 days`. No new table or persisted state is added. + +The voice owner loads eligible deadlines on startup. The existing +`notifyVoiceInbound` boundary now notifies cleanup and retention before writes. +Outer transaction settlement wakes the workers. Rolled-back changes do not +create retention work. Signed callbacks, rejection cleanup, approval/session +handoff and cleanup recovery retain their existing transaction ownership. +Company and endpoint changes use the existing composed notification boundary. + +At the deadline, retention locks the admission and rechecks current eligibility +and both timestamps before deleting. PostgreSQL submillisecond timestamps round +up for scheduling. Cached candidates cannot authorize deleting a new binding or +history whose retention deadline moved. Batches page through history and use the +shared worker's company fairness, coalescing, bounded backoff and drain behavior. +Deletion failure retains a retry; a paused first write remains pending for resume. +Shutdown cancels or joins work, and restart reloads remaining durable deadlines. + +Configured idle endpoints, ineligible history, and future retained history do not +cause recurring SQL between changes or actual deadlines. Unconfirmed cleanup +remains owned by the cleanup worker. Session/task history is not deleted here. +Explicit one-shot `pruneHistory` and diagnostic reconciliation remain available; +startup does not register those bulk operations as periodic work. + +Supported writers must participate in the shared transaction notification +boundary. Independent SQL writers or additional processes require their own +notification/recovery contract. This migration does not extend Cloud alarms or +sleep eligibility: an in-process deadline cannot wake stopped compute. The +external wake owner must cover retention before that deadline can permit sleep. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 024f4cc533e..049785da5a6 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1306,3 +1306,23 @@ The final migration applies/replays and its backfill is tested twice. Slice 24 starts the separate inbound-history retention audit. Its deadline preserves both seven-day age conditions and requires confirmed cleanup plus no task/session binding. Call reports follow separately. No original remaining scope is removed. + +## Slice 24: inbound voice history retention + +Branch `codex/sleep-24-voice-inbound-retention` follows inbound cleanup #17580 +including its migration runner repair `4f0364d269`. It removes only periodic +history pruning. The existing admission row owns the seven-day deadline and +confirmation/binding conditions. The existing notification helper wakes retention +on outer transaction settlement; no new table or caller wrapper is added. + +218 selected tests pass: 131 full voice/app wiring, 24 retention/cleanup lifecycle, +52 deletion/startup, and 11 parent recovery. Workspace typecheck/build and final +server typecheck pass. Tests prove real signed-callback and provider-cleanup +reactivation, pre-deadline quiet, exact deadlines, concurrent binding protection, +restart, drain, rollback, failure and shutdown. Initial fixture failures remain +recorded in the slice audit. Remote checks and review are pending publication. + +Next separate owner: voice call-report reconciliation. Preserve the existing +24-hour post-call refresh obligation, transcript ordering, task projection and +provider diagnostics while removing scans with no eligible calls. Inspect +completion semantics before treating a report marked available as fully settled. diff --git a/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md b/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md index bfa4e9e1f43..3a0d6242841 100644 --- a/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-inbound-retention-sleep-audit.md @@ -64,3 +64,58 @@ its latest Paperclip reviewer stopped before completion. The full original recurring-server ledger, older #17490 E2E failure and final stack/master reconciliation remain required. + +## Implementation and initial evidence + +The dedicated voice retention owner now loads eligible deadlines, rounds SQL +submillisecond dates up, and locks/rechecks before deletion. The existing producer +helper signals both voice admission cleanup and retention. App wiring registers +the owner; voice maintenance no longer calls `pruneHistory`. Reports remain in +that lane, and manual diagnostic reconciliation remains compatible. + +The initial ten PostgreSQL lifecycle cases had one invalid assertion: an open +transaction was expected to have no coordinator deadline. Intent reconciliation +correctly retains a deadline until settlement. The test now checks invisible +uncommitted state and the row still present within its owning transaction; it +requires deletion only after commit and quiet after rollback. All ten then passed. +An additional real concurrent binding-lock regression and production service +confirmation tests are being verified. Full voice acceptance uses a fresh database +created through the repaired application migrator, not direct SQL replay. + +Parent #17580 required repair 4f0364d269: register migration 0388 as top-level +because of its concurrent index. A new migration-runner regression reproduced the +failure, then proved fresh setup, invalid-index repair and concurrent/repeat +replay after the fix. This parent repair was merged into this branch; its fresh +CI is running. See the parent's audit for preserved failure evidence. + +## Acceptance evidence + +- 11 retention lifecycle tests and 13 parent cleanup tests pass (24 total). + The added concurrent-binding test holds the real row lock, commits a task + binding while deletion waits, and proves retention rechecks and keeps the row. +- Full voice integration and app wiring pass 131 cases on a fresh database. + Two new service tests start retention while the admission is unconfirmed, then + use a signed terminal callback or real rejection/provider cleanup to establish + confirmation. Each observes its saved seven-day deadline, SQL quiet before it, + deletion at it, and quiet afterwards. +- The first acceptance run waited 120 seconds before rejection, exceeding the + original call's approval window. That correctly rejected the test operation and + left an unconfirmed fixture affecting two later global cleanup assertions. The + corrected test checks pre-confirmation quiet for ten seconds and the full suite + passes on a newly migrated database. Production expiry policy did not change. +- Workspace typecheck and build pass. A final server typecheck covers the last + added concurrent-binding test. Shared deletion/startup and parent recovery + checks are tracked below when their current processes finish. +- No repository-wide `pnpm test:run` success is claimed: the local embedded + PostgreSQL initialization limitation from earlier slices remains. The selected + database suites use isolated real PostgreSQL; fresh remote CI is required. + +Logs: `/tmp/sleep24-retention-cleanup-final.log`, +`/tmp/sleep24-voice-acceptance.log` (preserved initial failure), +`/tmp/sleep24-voice-acceptance-fixed.log`, `/tmp/sleep24-workspace-types.log`, +`/tmp/sleep24-workspace-build.log`, `/tmp/sleep24-final-server-types.log`. + +Final local consumer checks pass: 52 company-deletion/startup tests, 11 parent +voice recovery tests, and final server typecheck. Selected nonoverlapping test +coverage totals 218 passing cases. The implementation and local verification are +complete; remote PR CI/review and the rest of the original stack remain pending. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index f1fdef98fa8..8e7fbd20d38 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/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()"); }); diff --git a/server/src/__tests__/voice-inbound-retention.test.ts b/server/src/__tests__/voice-inbound-retention.test.ts new file mode 100644 index 00000000000..ebafa59093c --- /dev/null +++ b/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> | undefined; + let db: ReturnType, companyId: string, endpointId: string, agentId: string, issueId: string, conversationId: string; + let coordinator: ReturnType | 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 = {}, 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 = {}) { + 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(resolve => { release = resolve; }); + const ready = new Promise(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(); + }); +}); diff --git a/server/src/__tests__/voice-session-store.integration.test.ts b/server/src/__tests__/voice-session-store.integration.test.ts index 434008cfcbf..42ddfbec05f 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/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" }); diff --git a/server/src/app.ts b/server/src/app.ts index ada93020bf6..995df660d5d 100644 --- a/server/src/app.ts +++ b/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(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index d7c878fa61d..354de7bd1da 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/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", diff --git a/server/src/services/voice/voice-inbound-notifications.ts b/server/src/services/voice/voice-inbound-notifications.ts index 0cfe67d3d8f..f5f34b10e9c 100644 --- a/server/src/services/voice/voice-inbound-notifications.ts +++ b/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); } diff --git a/server/src/services/voice/voice-inbound-retention.ts b/server/src/services/voice/voice-inbound-retention.ts new file mode 100644 index 00000000000..3e73b9ef151 --- /dev/null +++ b/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 { + 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`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 { + 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, 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 }; +} diff --git a/server/src/services/voice/voice-inbound-service.ts b/server/src/services/voice/voice-inbound-service.ts index aa8f584570d..ed86554ae60 100644 --- a/server/src/services/voice/voice-inbound-service.ts +++ b/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}; } diff --git a/server/src/services/voice/voice-session-service.ts b/server/src/services/voice/voice-session-service.ts index e262900772e..49a2cc796d5 100644 --- a/server/src/services/voice/voice-session-service.ts +++ b/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. */