diff --git a/doc/idle-voice-inbound-cleanup.md b/doc/idle-voice-inbound-cleanup.md new file mode 100644 index 00000000000..a0830f19912 --- /dev/null +++ b/doc/idle-voice-inbound-cleanup.md @@ -0,0 +1,101 @@ +# Inbound voice cleanup without idle scans + +Inbound admission cleanup runs from committed changes and saved deadlines. It +uses the shared delivery coordinator and scoped deadline worker. The recurring +voice maintenance lane now owns only history retention and reports, which remain +separate migrations. Manual reconciliation remains available for diagnostics. + +## Ownership and durable state + +An unbound admission owns the existing provider call until approval or guest +intake binds a voice session. The live-session recovery service then owns physical +termination. Admission cleanup never also ends that bound call. It can request +ending through the session row and copy the owner's confirmed terminal result. + +Approval now commits the admission binding with the authorized session insert or +guest promotion. Expiry cannot see an old approving admission after that session +commits. Cleanup also locks and rechecks the admission before claiming it; a +candidate from an earlier scan is not permission to terminate a call. Both paths +lock admission before session. Provider requests run outside SQL transactions. +Older interrupted handoffs are repaired only when the existing session matches +the approval actor, authority, request key, task, token, generation and credential +identity. Conflicting ownership retains an error/retry instead of terminating a +call or inventing authority. + +The migration adds three nullable fields to `chat_voice_inbound_calls`, not a new +queue table: + +- `cleanupConfirmedAt` separates admission status from physical termination. + An ended or denied admission can still have an unconfirmed provider leg. +- `cleanupNextAttemptAt` survives crashes and restart. The claim saves the retry + before contacting the provider. +- `providerAccountId` records the account authenticated by the signed pre-call. + Cleanup can use a rotated key after verifying its account. A key for another + account cannot inspect or terminate this call. Older admissions lack that + evidence and require their original credential fingerprint. + +The migration is replayable. It backfills only historical ended rows whose old +writer or signed lifecycle audit proves termination. It leaves old guest-denial +rows uncertain when their ended label did not prove physical cleanup. Those rows +receive a real check rather than disappearing as completed history. The partial +index covers unconfirmed admission identities; startup inventory is paged. + +## Deadlines and quiet state + +Pending unbound admissions retain their original two-minute approval or ten-minute +guest-intake expiry. Rejection is immediately due. A cleanup claim saves a +30-second retry before provider I/O. An unconfirmed end receipt keeps that retry; +inspection, a confirmed end receipt, or a signed terminal callback clears it. +Unexpected failures also receive the shared worker's bounded failure backoff. +Neither a timeout nor an end request by itself proves termination. + +Bound live calls leave no admission cleanup deadline. Their existing live-session +worker owns expiry, authorization checks and provider cleanup. Its terminal write +signals admission settlement. Healthy passive provider inspections do not wake +this admission worker. Confirmed history and configured-but-idle endpoints leave +no recurring admission SQL or provider requests. + +Feature disable does not discard an outstanding call. Cleanup continues with its +saved identity. Missing credentials or account mismatches retain debt; they do not +permit use of a replacement account. No recovery path creates or redials a call. + +## Producer audit + +| Producer | Commit boundary and wake | +| --- | --- | +| Signed pre-call admission | Endpoint-scoped intent before insert in the quota transaction | +| Approval or rejection | Admission-scoped intent before the locked state write | +| Authorized session insert or guest promotion | Admission binding and intent inside the reservation transaction | +| Guest intake task/session binding | Session-scoped voice change before insert; admission binding shares the transaction | +| Pending tool detects expiry | Admission intent before the expired-state write | +| Signed terminal lifecycle/report callback | Session and admission intents before confirmed termination writes | +| Live-session termination, inspection or no-resource expiry | Existing voice-change boundary also signals inbound settlement | +| Reply provider confirms the call ended | Inbound session intent before the terminal session write | +| Configuration, authority or connection changes | Existing scoped voice-change composition | +| Cleanup claim, legacy handoff repair and confirmation | Admission intent in the owning transaction | + +Notifications are process-local hints. Existing database rows remain truth. The +shared transaction layer delays hints until outer commit and retains uncertainty. +Other replicas and independent SQL writers require an explicit integration. + +History retention cannot delete unconfirmed cleanup. Company deletion locks its +endpoints, matching ingress, then asks the voice owner to prove inbound cleanup is +settled. It rejects deletion while a provider obligation remains. This prevents +cascade deletion from erasing the evidence needed to finish cleanup. Existing +voice-history foreign-key restrictions otherwise remain unchanged. + +## Verification and limits + +Real PostgreSQL tests cover confirmed-history quiet, exact expiry, restart, +transaction rollback/outer commit, stale discovery, old interrupted handoffs, +bound-session ownership, unconfirmed requests, failures, fair paged backlog, +drain/resume, shutdown, retention protection, and deletion protection. Production +service tests exercise signed ingress, rejection, account mismatch and rotation, +feature disable, and atomic ordinary/guest approval. See the dated +[inbound cleanup audit](plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md) for +commands, counts and preserved failures. + +These changes do not migrate history retention or call reports, extend Cloud's +external wake coverage, relax sleep eligibility for live calls, or prove actual +host/database suspension. A registered in-process deadline alone cannot wake +stopped compute. 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 6c79515d99f..a85eb03abce 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1286,3 +1286,14 @@ locked cleanup claims, legacy handoff recovery, bound-guest ownership, terminati confirmation, retention protection and scheduler/producer migration remain required. The full original inventory, older #17490 E2E failure, and stack/master reconciliation remain outstanding. No stack PR has been merged by this work. + + +Slice 23 inbound cleanup now uses committed admission/session changes and actual +expiry/retry deadlines. The admission row separately records physical confirmation +and saved provider account. Locked claims, legacy handoff repair, bound-call owner +handoff, retention/deletion protection and safe same-account rotation are verified. +All 205 selected tests, workspace types and build pass; publication and remote +acceptance remain pending. History retention and call reports remain separate next +slices. Parent #17578 at 90caaf3fa9 has no pending/failed checks and exact-head +Greptile 5/5; the second Paperclip reviewer stopped before completion. The original +remaining inventory, #17490 E2E failure and final reconciliation remain in scope. diff --git a/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md b/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md index 2110c23b6e5..296dc367d82 100644 --- a/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-inbound-cleanup-sleep-audit.md @@ -2,7 +2,7 @@ Baseline: voice reply PR #17578, head 439d0a86eb75c129c30742a8cc24bf18e2a6ebf3. Branch: codex/sleep-23-voice-inbound-cleanup. -Status: atomic approval handoff implemented and verified; cleanup worker, remaining ownership cases, and publication remain required. +Status: cleanup implementation and local verification complete; publication and remote acceptance pending. ## One reason in this slice @@ -148,3 +148,53 @@ This is a prerequisite, not the complete cleanup migration. Still required: - Wire all producer notifications, actual deadlines, scoped worker registration, lifecycle tests and remove only the cleanup portion of periodic reconciliation. - Keep history retention and call reports as separate forthcoming PRs. + + +## Cleanup implementation and local acceptance + +The worker is now registered in app startup. `reconcileMaintenance` no longer +runs inbound cleanup; it retains history pruning and reports for separate PRs. +See [the implementation contract](../idle-voice-inbound-cleanup.md) for the complete +producer table, ownership, credential rotation and remaining host-sleep limits. + +The previous checkpoint's remaining implementation items are now handled: +locked current-state claims; matching legacy handoff repair; bound-session cleanup +ownership; durable termination confirmation/retry; account isolation; retention +protection; notifications and shared scheduler registration. A company-delete +participant check also prevents cascades from erasing unfinished cleanup and locks +endpoints to serialize against signed admission ingress. + +The existing admission row needs separate authorization and cleanup state. Three +nullable fields record confirmation, retry and the signed provider account. +Migration 0388_lying_trish_tilby.sql was generated with `pnpm db:generate`, made +idempotent, and given a concurrent partial index. It backfills only provable old +termination. The initial two-field unpublished generation was replaced by this +single migration after account-rotation analysis. Five newest snapshots remain; +the generated deletion of 0383 is intentional. + +Local evidence: +- `voice-inbound-cleanup.test.ts`: 13 tests pass using a separate PostgreSQL DB. + Includes migration backfill/replay, company deletion and shutdown joining. +- Full voice integration and app wiring: 129 tests pass on a fresh DB (113 voice, + 16 app wiring). Signed ingress wakes an idle worker; a feature-disabled instance + retains an unconfirmed end; a replacement account cannot issue cleanup; a key + for the saved account can. Rejection failure survives history pruning. +- Company deletion and server startup consumers: 52 tests pass on their isolated DB. +- Parent live-session recovery: 11 tests pass on its separate DB. +- Total selected coverage: 205 passing cases across these nonoverlapping suites. +- `pnpm -r typecheck` and `pnpm build` pass on the current implementation. +- Migration safety/numbering checks passed for the initial generated migration; + rerun on the final regenerated name before publication. Initial application and + replay succeeded on real PostgreSQL. Final generation also applied cleanly; + its backfill is exercised twice by the lifecycle test. + +Earlier handoff regression and fixture assertion failures remain recorded above. +The repository-wide local test command still has the previously recorded embedded +PostgreSQL host limitation; this is not a claim that that command completed. Fresh +remote CI and review are required before accepting the PR. + +Parent #17578 at 90caaf3fa9 has all checks terminal with no failures and Greptile +5/5 on that exact head. Both requested findings are fixed/resolved. The fresh +Paperclip reviewer stopped before completing its turn; do not report a new 5/5 +assessment from that reviewer. The full original migration ledger, #17490 failure, +and final stack/master reconciliation remain outstanding. diff --git a/packages/db/src/migrations/0388_lying_trish_tilby.sql b/packages/db/src/migrations/0388_lying_trish_tilby.sql new file mode 100644 index 00000000000..d7c77372313 --- /dev/null +++ b/packages/db/src/migrations/0388_lying_trish_tilby.sql @@ -0,0 +1,15 @@ +ALTER TABLE "chat_voice_inbound_calls" ADD COLUMN IF NOT EXISTS "provider_account_id" text;--> statement-breakpoint +ALTER TABLE "chat_voice_inbound_calls" ADD COLUMN IF NOT EXISTS "cleanup_confirmed_at" timestamp with time zone;--> statement-breakpoint +ALTER TABLE "chat_voice_inbound_calls" ADD COLUMN IF NOT EXISTS "cleanup_next_attempt_at" timestamp with time zone;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_voice_inbound_cleanup_idx" ON "chat_voice_inbound_calls" USING btree ("id") WHERE "chat_voice_inbound_calls"."cleanup_confirmed_at" is null;--> statement-breakpoint +-- Historical ended admissions without an approving/denying actor came from +-- provider confirmation. Guest denial used ended before confirming termination; +-- retain those rows unless the signed lifecycle audit supplies separate proof. +UPDATE chat_voice_inbound_calls c +SET cleanup_confirmed_at = c.updated_at +WHERE c.cleanup_confirmed_at IS NULL AND c.state = 'ended' + AND (c.approved_by_user_id IS NULL OR EXISTS ( + SELECT 1 FROM activity_log a + WHERE a.company_id = c.company_id AND a.entity_type = 'voice_inbound_call' + AND a.entity_id = c.id::text AND a.action = 'voice.call.ended' + )); diff --git a/packages/db/src/migrations/meta/0383_snapshot.json b/packages/db/src/migrations/meta/0388_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0383_snapshot.json rename to packages/db/src/migrations/meta/0388_snapshot.json index 98f34247932..c887637d520 100644 --- a/packages/db/src/migrations/meta/0383_snapshot.json +++ b/packages/db/src/migrations/meta/0388_snapshot.json @@ -1,6 +1,6 @@ { - "id": "75e7a97b-132d-4326-b683-003c1a2b7048", - "prevId": "5e8d042e-d697-4753-9582-7d0c042ef893", + "id": "f83c3343-8665-48bc-919b-bc9fe7513b87", + "prevId": "4c86af49-02a0-496f-b028-b468d9e4e960", "version": "7", "dialect": "postgresql", "tables": { @@ -5721,6 +5721,28 @@ } }, "indexes": { + "approvals_pending_retention_idx": { + "name": "approvals_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"approvals\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "approvals_company_status_type_idx": { "name": "approvals_company_status_type_idx", "columns": [ @@ -14138,6 +14160,24 @@ "primaryKey": false, "notNull": false }, + "provider_account_id": { + "name": "provider_account_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "cleanup_confirmed_at": { + "name": "cleanup_confirmed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "cleanup_next_attempt_at": { + "name": "cleanup_next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "expires_at": { "name": "expires_at", "type": "timestamp with time zone", @@ -14160,6 +14200,22 @@ } }, "indexes": { + "chat_voice_inbound_cleanup_idx": { + "name": "chat_voice_inbound_cleanup_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_voice_inbound_expiry_idx": { "name": "chat_voice_inbound_expiry_idx", "columns": [ @@ -24030,6 +24086,28 @@ "method": "btree", "with": {} }, + "decisions_running_recovery_idx": { + "name": "decisions_running_recovery_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'", + "concurrently": false, + "method": "btree", + "with": {} + }, "decisions_open_expiry_idx": { "name": "decisions_open_expiry_idx", "columns": [ @@ -32057,6 +32135,28 @@ "method": "btree", "with": {} }, + "grok_dispatch_prepared_idx": { + "name": "grok_dispatch_prepared_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"grok_bot_dispatches\".\"status\" = 'prepared'", + "concurrently": false, + "method": "btree", + "with": {} + }, "grok_dispatch_company_status_idx": { "name": "grok_dispatch_company_status_idx", "columns": [ @@ -41851,6 +41951,28 @@ } }, "indexes": { + "join_requests_pending_retention_idx": { + "name": "join_requests_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"join_requests\".\"status\" = 'pending_approval'", + "concurrently": false, + "method": "btree", + "with": {} + }, "join_requests_invite_unique_idx": { "name": "join_requests_invite_unique_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 6979bbe3e1d..779a28f7114 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2703,6 +2703,13 @@ "when": 1792880971454, "tag": "0387_huge_skrulls", "breakpoints": true + }, + { + "idx": 388, + "version": "7", + "when": 1792921047529, + "tag": "0388_lying_trish_tilby", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_voice_inbound.ts b/packages/db/src/schema/chat_voice_inbound.ts index c50ccf39a60..de95907e3eb 100644 --- a/packages/db/src/schema/chat_voice_inbound.ts +++ b/packages/db/src/schema/chat_voice_inbound.ts @@ -27,6 +27,9 @@ export const chatVoiceInboundCalls = pgTable("chat_voice_inbound_calls", { approvedByUserId: text("approved_by_user_id"), callerAuthority: text("caller_authority"), intakeIssueId: uuid("intake_issue_id"), requestedIssueId: uuid("requested_issue_id"), sessionId: uuid("session_id"), + providerAccountId: text("provider_account_id"), + cleanupConfirmedAt: timestamp("cleanup_confirmed_at", {withTimezone: true}), + cleanupNextAttemptAt: timestamp("cleanup_next_attempt_at", {withTimezone: true}), expiresAt: timestamp("expires_at", {withTimezone: true}).notNull(), createdAt: timestamp("created_at", {withTimezone: true}).notNull().defaultNow(), updatedAt: timestamp("updated_at", {withTimezone: true}).notNull().defaultNow(), @@ -36,6 +39,7 @@ export const chatVoiceInboundCalls = pgTable("chat_voice_inbound_calls", { foreignKey({columns: [t.companyId, t.sessionId], foreignColumns: [chatVoiceSessions.companyId, chatVoiceSessions.id], name: "chat_voice_inbound_session_fk"}), foreignKey({columns: [t.companyId, t.intakeIssueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_inbound_intake_fk"}), foreignKey({columns: [t.companyId, t.requestedIssueId], foreignColumns: [issues.companyId, issues.id], name: "chat_voice_inbound_requested_issue_fk"}), + index("chat_voice_inbound_cleanup_idx").on(t.id).where(sql`${t.cleanupConfirmedAt} is null`), index("chat_voice_inbound_expiry_idx").on(t.state, t.expiresAt), check("chat_voice_inbound_state_check", sql`${t.state} in ('guest_intake', 'awaiting_approval', 'approving', 'approved', 'denied', 'expired', 'ended')`), check("chat_voice_inbound_token_check", sql`${t.toolTokenHash} ~ '^[a-f0-9]{64}$'`), diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 98e7fc44f01..f1fdef98fa8 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -86,6 +86,7 @@ describe("createChatReconciliationCoordinator", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); 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.reconcileMaintenance()"); expect(source).not.toContain("chatChannels.voice.reconcile()"); }); diff --git a/server/src/__tests__/voice-inbound-cleanup.test.ts b/server/src/__tests__/voice-inbound-cleanup.test.ts new file mode 100644 index 00000000000..2a7940a3cb7 --- /dev/null +++ b/server/src/__tests__/voice-inbound-cleanup.test.ts @@ -0,0 +1,194 @@ +import { readFile } from "node:fs/promises"; +import { chatCompanyDeletion } from "../services/chat-channels/company-deletion.js"; +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 { voiceInboundCleanup } from "../services/voice/voice-inbound-cleanup.js"; +import { notifyVoiceInbound } from "../services/voice/voice-inbound-notifications.js"; +import { notifyVoiceChange } from "../services/voice/voice-recovery-notifications.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_INBOUND_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("inbound voice cleanup 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 terminate = vi.fn(async (_call: typeof chatVoiceInboundCalls.$inferSelect) => true); + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("voice-inbound-cleanup-"); + 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; terminate.mockReset().mockResolvedValue(true); + ({ 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: "awaiting_approval", approvalCode: "123456", generation: 1, credentialFingerprint: "test", toolTokenHash: "a".repeat(64), requestFingerprint: id, expiresAt: new Date(Date.now() - 1), ...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 = voiceInboundCleanup(db, terminate)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled, canReconcile: () => true, onError: error => errors.push(error) }); + await service.register(coordinator, () => enabled).ready; + return service; + } + async function settled() { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); }, { 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(); } + } + it("leaves configured idle and confirmed history quiet", async () => { + await admission({ state: "ended", cleanupConfirmedAt: new Date() }); + await start(); await settled(); await quiet(); expect(terminate).not.toHaveBeenCalled(); + }); + it("prevents company deletion from erasing an unfinished provider cleanup", async () => { + const call = await admission(); + await expect(db.transaction(tx => chatCompanyDeletion.deleteCompanyData(tx, companyId))).rejects.toMatchObject({ status: 409 }); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id))).toHaveLength(1); + await start(); await settled(); + await db.transaction(tx => chatCompanyDeletion.deleteCompanyData(tx, companyId)); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id))).toHaveLength(0); + }); + it("backfills only provable legacy termination and preserves rejected cleanup debt", async () => { + const confirmed = await admission({ state: "ended" }); + const uncertain = await admission({ state: "ended", approvedByUserId: randomUUID(), callerAuthority: "member" }); + const migration = await readFile(new URL("../../../packages/db/src/migrations/0388_lying_trish_tilby.sql", import.meta.url), "utf8"); + const backfill = migration.split("--> statement-breakpoint").at(-1)!; + await db.execute(sql.raw(backfill)); await db.execute(sql.raw(backfill)); + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, confirmed.id)))[0].cleanupConfirmedAt).not.toBeNull(); + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, uncertain.id)))[0].cleanupConfirmedAt).toBeNull(); + await start(); await settled(); expect(terminate).toHaveBeenCalledOnce(); + expect(terminate.mock.calls[0][0].id).toBe(uncertain.id); await quiet(); + }); + it("joins provider cleanup at shutdown and saves a restart deadline for uncertainty", async () => { + const call = await admission(); let release!: (confirmed: boolean) => void; + terminate.mockImplementationOnce(() => new Promise(resolve => { release = resolve; })); + const service = voiceInboundCleanup(db, terminate); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = service.register(coordinator, () => true); + await vi.waitFor(() => expect(terminate).toHaveBeenCalledOnce()); + let stopped = false; + const stopping = coordinator.stop().then(() => { stopped = true; }); + try { + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release(false); await stopping; await worker.ready; + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0].cleanupNextAttemptAt).not.toBeNull(); + await start(); await settled(); expect(terminate).toHaveBeenCalledTimes(2); + } finally { release(false); await stopping; } + }); + it("waits for the original expiry and reconstructs it on restart", async () => { + const due = Date.now() + 60_000; await admission({ expiresAt: new Date(due) }); + await start(); expect(coordinator!.nextWakeAt()).toBe(due); await quiet(10_000); + await coordinator!.stop(); await start(); expect(coordinator!.nextWakeAt()).toBe(due); + await vi.advanceTimersByTimeAsync(due - Date.now()); await settled(); expect(terminate).toHaveBeenCalledOnce(); await quiet(); + }); + it("waits for outer commit and ignores a rolled-back nested producer", async () => { + await start(); await settled(); + await expect(db.transaction(async tx => { await admission({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(2000); expect(terminate).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback"); + await settled(); await quiet(); + await db.transaction(async tx => { await admission({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(2000); expect(terminate).not.toHaveBeenCalled(); }); + await settled(); expect(terminate).toHaveBeenCalledOnce(); await quiet(); + }); + it("retains unconfirmed termination across restart without early requests", async () => { + terminate.mockResolvedValueOnce(false); + const call = await admission(); await start(); + const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)); + expect(pending.cleanupConfirmedAt).toBeNull(); expect(pending.cleanupNextAttemptAt).not.toBeNull(); + await coordinator!.stop(); await start(); await quiet(10_000); expect(terminate).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(pending.cleanupNextAttemptAt!.getTime() - Date.now()); await settled(); + expect(terminate).toHaveBeenCalledTimes(2); await quiet(); + }); + it("rechecks ownership after discovery rather than ending a newly bound call", async () => { + const call = await admission({ state: "approving", approvedByUserId: randomUUID(), callerAuthority: "member", requestedIssueId: issueId }); + const service = voiceInboundCleanup(db, terminate), [candidate] = await service.load(); + const session = await owner(call); + await db.update(chatVoiceInboundCalls).set({ state: "approved", sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, call.id)); + expect(await service.check(candidate.id)).toBeNull(); expect(terminate).not.toHaveBeenCalled(); + }); + it.each([false, true])("repairs older interrupted handoffs (existing guest binding=%s)", async bound => { + const call = await admission({ state: "approving", approvedByUserId: randomUUID(), callerAuthority: "member", requestedIssueId: issueId }); + const session = await owner(call); + if (bound) await db.update(chatVoiceInboundCalls).set({ sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, call.id)); + await start(); await settled(); expect(terminate).not.toHaveBeenCalled(); + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0]).toMatchObject({ state: "approved", sessionId: session.id }); + await quiet(); + }); + it("hands rejected bound calls to live recovery and settles from its confirmation", async () => { + const call = await admission({ state: "denied" }), session = await owner(call); + await db.update(chatVoiceInboundCalls).set({ sessionId: session.id }).where(eq(chatVoiceInboundCalls.id, call.id)); + await start(); await settled(); expect(terminate).not.toHaveBeenCalled(); + expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, session.id)))[0].state).toBe("ending"); + await quiet(); + await db.transaction(async tx => { + await notifyVoiceChange(tx, { kind: "session", id: session.id }); + await tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)); + }); + await settled(); + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)))[0].cleanupConfirmedAt).not.toBeNull(); + await quiet(); expect(terminate).not.toHaveBeenCalled(); + }); + it("retains first work committed during drain without passive SQL", async () => { + await start(); await settled(); enabled = false; + await admission(); await quiet(60_000); expect(terminate).not.toHaveBeenCalled(); + enabled = true; await settled(); expect(terminate).toHaveBeenCalledOnce(); await quiet(); + }); + it("pages backlog fairly while a failed provider retains its own retry", async () => { + const broken = await admission(); + for (let i = 0; i < 104; i++) await admission(); + const other = await fixture(), last = await admission({ companyId: other.companyId, endpointId: other.endpointId }); + let failed = true; + terminate.mockImplementation(async call => { if (call.id === broken.id && failed) throw new Error("provider unavailable"); return true; }); + await start(); + expect(terminate.mock.calls.slice(0, 25).some(([call]) => call.id === last.id)).toBe(true); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(terminate.mock.calls.filter(([call]) => call.id !== broken.id)).toHaveLength(105); + }, { timeout: 5000 }); + expect(errors).toHaveLength(1); failed = false; + await settled(); errors.length = 0; 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 16af29cb463..434008cfcbf 100644 --- a/server/src/__tests__/voice-session-store.integration.test.ts +++ b/server/src/__tests__/voice-session-store.integration.test.ts @@ -78,6 +78,91 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication }; } + 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" }); + await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers); + 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: number) => { + 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.registerCleanupWork(coordinator, () => true).ready; + await settle(); expect(coordinator.nextWakeAt()).toBeNull(); await quiet(10_000); + const providerSessionId = randomUUID(), incoming = f.sign({ ...f.event, call_id: providerSessionId, session_id: providerSessionId }); + await f.service.inbound.lifecycle(f.endpoint.publicId, incoming.body, incoming.headers); + await settle(); + const [call] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.providerSessionId, providerSessionId)); + expect(coordinator.nextWakeAt()).toBe(call.expiresAt.getTime()); + await quiet(60_000); expect(f.transport.inspectSession).not.toHaveBeenCalled(); + // Disabling new voice work must not erase an existing physical cleanup. + await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false }); + await settle(); + f.credentials.mockResolvedValue({ apiKey: "rotated-key-in-original-account", agentId: "agent_test" }); + f.transport.endSession.mockResolvedValueOnce({ confirmed: false }); + await vi.advanceTimersByTimeAsync(call.expiresAt.getTime() - Date.now()); await settle(); + expect(f.transport.endSession).toHaveBeenCalledWith(providerSessionId); + const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, call.id)); + expect(pending.cleanupConfirmedAt).toBeNull(); + expect(coordinator.nextWakeAt()).toBe(pending.cleanupNextAttemptAt!.getTime()); + await quiet(10_000); expect(f.transport.endSession).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(pending.cleanupNextAttemptAt!.getTime() - Date.now()); await settle(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(coordinator.nextWakeAt()).toBeNull(); }); + expect(f.transport.endSession).toHaveBeenCalledTimes(2); await quiet(120_000); + expect(onError).not.toHaveBeenCalled(); + expect(f.transport.createPhoneSession).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); vi.useRealTimers(); await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true }); } + }); + + it("retains rejected inbound cleanup across provider failure and history retention", async () => { + const f = await inboundFixture(true); + f.transport.endSession.mockRejectedValueOnce(new Error("provider unavailable")); + const result = await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, + { approve: false, approvalCode: f.call.approvalCode }); + expect(result.state).toBe("ended"); + const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)); + expect(pending.cleanupConfirmedAt).toBeNull(); expect(pending.cleanupNextAttemptAt).not.toBeNull(); + await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id)); + await f.service.inbound.pruneHistory(); + expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(1); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await f.service.inbound.registerCleanupWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBe(pending.cleanupNextAttemptAt!.getTime()); + await vi.advanceTimersByTimeAsync(pending.cleanupNextAttemptAt!.getTime() - Date.now()); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); }); + expect(f.transport.endSession).toHaveBeenCalledTimes(2); + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)))[0].cleanupConfirmedAt).not.toBeNull(); + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("does not clean up an inbound call through a different provider account", async () => { + const f = await inboundFixture(); + f.credentials.mockResolvedValue({ apiKey: "key-for-another-account", agentId: "other-agent" }); + f.transport.verifyAgent.mockResolvedValue({ id: "other-agent", organizationId: "another-account", name: "Other voice" }); + await f.service.inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller, + { approve: false, approvalCode: f.call.approvalCode }); + expect(f.transport.inspectSession).not.toHaveBeenCalled(); expect(f.transport.endSession).not.toHaveBeenCalled(); + const [pending] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)); + expect(pending.cleanupConfirmedAt).toBeNull(); expect(pending.cleanupNextAttemptAt).not.toBeNull(); + // A usable key for the saved account restores cleanup without redialing. + f.credentials.mockResolvedValue({ apiKey: "restored-account-key", agentId: "agent_test" }); + f.transport.verifyAgent.mockResolvedValue({ id: "agent_test", organizationId: `org_${f.companyId}`, name: "Test voice" }); + await db.update(chatVoiceInboundCalls).set({ cleanupNextAttemptAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id)); + await f.service.inbound.reconcile(1000); + expect(f.transport.endSession).toHaveBeenCalledWith(f.call.providerSessionId); + expect((await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id)))[0].cleanupConfirmedAt).not.toBeNull(); + expect(f.transport.createPhoneSession).not.toHaveBeenCalled(); + }); + it("restores blocked replies through the production live-call recovery wiring", async () => { const f = await pushFixture(); await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.principalId, f.callerId)); @@ -1085,6 +1170,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await instanceSettingsService(db).updateExperimental({enableChatConnectors: true, enableIsolatedWorkspaces: guestIntake}); const signingSecret = `whsec_${Buffer.alloc(32, 7).toString("base64")}`; f.credentials.mockResolvedValue({apiKey: "server-only-key", agentId: "agent_test", signingSecret} as never); + f.transport.verifyAgent.mockResolvedValue({ id: "agent_test", organizationId: `org_${f.companyId}`, name: "Test voice" }); await db.update(chatEndpoints).set({providerAccountId: `org_${f.companyId}`, botExternalId: "agent_test"}).where(eq(chatEndpoints.id, f.endpointId)); await db.insert(chatVoicePhoneLines).values({companyId: f.companyId, endpointId: f.endpointId, providerNumberId: randomUUID(), phoneNumber: "+12015550123", enabled: true, guestIntake, lowTrustEnvironmentId: sandboxId ?? null}); const [line] = await db.select().from(chatVoicePhoneLines).where(eq(chatVoicePhoneLines.endpointId, f.endpointId)); diff --git a/server/src/app.ts b/server/src/app.ts index 0a1249f9ea7..ada93020bf6 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1300,6 +1300,7 @@ export async function createApp( registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.voice.registerReplyWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.voice.inbound.registerCleanupWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); diff --git a/server/src/services/chat-channels/company-deletion.ts b/server/src/services/chat-channels/company-deletion.ts index a9a3575a6da..710b4d21cb1 100644 --- a/server/src/services/chat-channels/company-deletion.ts +++ b/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)); }, }; diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 058664fe9d8..d7c878fa61d 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", + voiceInboundCleanup: "voice-inbound-cleanup", voiceReplies: "voice-reply-delivery", chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", diff --git a/server/src/services/voice/voice-inbound-cleanup.ts b/server/src/services/voice/voice-inbound-cleanup.ts new file mode 100644 index 00000000000..64f82bf2e72 --- /dev/null +++ b/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, 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) { + async function load(scope?: VoiceInboundScope, signal?: AbortSignal): Promise { + 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 { + 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, 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 }; +} diff --git a/server/src/services/voice/voice-inbound-notifications.ts b/server/src/services/voice/voice-inbound-notifications.ts new file mode 100644 index 00000000000..0cfe67d3d8f --- /dev/null +++ b/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}` : "*"); +} diff --git a/server/src/services/voice/voice-inbound-service.ts b/server/src/services/voice/voice-inbound-service.ts index 36b60dbccb1..aa8f584570d 100644 --- a/server/src/services/voice/voice-inbound-service.ts +++ b/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}; } diff --git a/server/src/services/voice/voice-recovery-notifications.ts b/server/src/services/voice/voice-recovery-notifications.ts index 8476546ab1a..f95e3627cdb 100644 --- a/server/src/services/voice/voice-recovery-notifications.ts +++ b/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(db: Db, id: string, write: (tx: Parameters[0]>[0]) => PromiseLike, options: { replyDelivery?: boolean } = {}): Promise { return db.transaction(async tx => { diff --git a/server/src/services/voice/voice-reply-push.ts b/server/src/services/voice/voice-reply-push.ts index 2316056849a..31b1d257ef3 100644 --- a/server/src/services/voice/voice-reply-push.ts +++ b/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)); }); diff --git a/server/src/services/voice/voice-session-service.ts b/server/src/services/voice/voice-session-service.ts index 16884007dfb..e262900772e 100644 --- a/server/src/services/voice/voice-session-service.ts +++ b/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) { diff --git a/server/src/services/voice/voice-session-store.ts b/server/src/services/voice/voice-session-store.ts index 05d54e5856a..bfd1ee9feda 100644 --- a/server/src/services/voice/voice-session-store.ts +++ b/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)); }