feat(server): establish voice session recovery deadlines and producer signals

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

doc/plans/2026-10-25-voice-maintenance-sleep-audit.md

@@ -87,3 +87,28 @@ and is published/attached. CI run 40173962833 is live on its exact head; Greptil
 security and Paperclip review are pending. This is not stack completion or merge
 permission. The full original ledger, #17490 failure and final master
 reconciliation remain required.
+
+
+## Initial implementation checkpoint
+
+Live-session recovery now has a voice-owned paged deadline reader and a scoped
+coordinator worker. The periodic wrapper retains reply, inbound, and report
+maintenance. Session writes signal their existing transactions; standalone service
+updates use one small transaction helper. Configuration, feature-disable and
+voice tool/access removal paths signal relevant scopes. Active sessions retain
+30-second inspection, unknown creation retains the saved expiry plus two-minute
+grace, and no call-creation action exists in recovery. Authority lookup failures
+retry instead of treating an unavailable database as confirmed revocation.
+
+Four initial actual-service cases pass: first start after quiet and retirement
+on missed hangup, membership revocation, and feature disable; unknown creation
+expires at its saved grace deadline without another provider call. Server types
+pass after fixing the database projection of the internal runtime generation.
+The startup wiring regression failed before removal. This is a checkpoint only:
+full producer, lifecycle, race, backlog, and consumer evidence is still required.
+
+Receiver #17572 has Greptile 5/5 and no failed checks, but Paperclip Review 4/5
+requests suppressing rediscovery after discovery-initiated stops. Its optional
+full-refresh coalescing observation will also be addressed in the owning parent
+before this branch is published. Preserve all unfinished voice changes in this
+commit while repairing that parent.

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

@@ -78,10 +78,17 @@ describe("createChatReconciliationCoordinator", () => {
     const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
     const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals"));
     expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()");
-    expect(maintenance).toContain("chatChannels.voice.reconcile()");
+    expect(maintenance).toContain("chatChannels.voice.reconcileMaintenance()");
     expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,");
   });
 
+  it("separates live voice recovery from the remaining periodic maintenance", () => {
+    const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
+    expect(source).toContain("chatChannels.voice.registerRecoveryWork(deliveryWork,");
+    expect(source).toContain("chatChannels.voice.reconcileMaintenance()");
+    expect(source).not.toContain("chatChannels.voice.reconcile()");
+  });
+
   it("wires GitHub recovery into its independent lane", () => {
     const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
     expect(source).toContain(

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

@@ -1,3 +1,6 @@
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import express from "express";
 import request from "supertest";
 import { errorHandler } from "../middleware/error-handler.js";
@@ -28,15 +31,16 @@ import { buildPaperclipRuntimeMcpServers } from "../services/heartbeat.js";
 import { createToolGatewayService } from "../services/tool-gateway.js";
 import { initializeRunIdentity } from "../services/run-identity.js";
 import { syncSpekoVoiceTools, spekoToolsForSession, executeSpekoVoiceTool } from "../services/voice/speko-agent-tools.js";
-const support = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_VOICE_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 (support.supported ? describe : describe.skip)("durable Speko voice sessions", () => {
   let db: ReturnType<typeof createDb>;
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("paperclip-speko-store-");
-    db = createDb(database.connectionString);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-speko-store-");
+    db = createDb(external ?? database!.connectionString, { idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
   }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
 
   async function fixture() {
     const companyId = randomUUID(), endpointId = randomUUID(), agentId = randomUUID(), applicationId = randomUUID(), connectionId = randomUUID(), issueId = randomUUID(), conversationId = randomUUID(), callerId = randomUUID();
@@ -67,6 +71,63 @@ const support = await getEmbeddedPostgresTestSupport();
     return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication };
   }
 
+  it.each(["hangup", "revocation", "feature disable"] as const)("wakes live recovery from a real start and settles after %s", async reason => {
+    await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
+    const f = await nativeServiceFixture();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    const settle = async (terminal = false) => vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(idleWorkSnapshot().active).toBe(0);
+      if (terminal) expect(coordinator.nextWakeAt()).toBeNull();
+      else expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now());
+    });
+    const quiet = async () => {
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
+      try { await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); }
+      finally { reads.mockRestore(); writes.mockRestore(); }
+    };
+    try {
+      await f.service.registerRecoveryWork(coordinator, () => true).ready;
+      await settle(true); await quiet();
+      const started = await f.service.start(f.input); await settle();
+      expect(f.transport.createBrowserSession).toHaveBeenCalledOnce();
+      expect(f.transport.inspectSession).not.toHaveBeenCalled();
+      if (reason === "hangup") f.transport.inspectSession.mockResolvedValue({ status: "ended", endedAt: new Date().toISOString() });
+      if (reason === "revocation") await db.update(companyMemberships).set({ status: "removed" }).where(eq(companyMemberships.principalId, f.callerId));
+      if (reason === "feature disable") await instanceSettingsService(db).updateExperimental({ enableChatConnectors: false });
+      else await vi.advanceTimersByTimeAsync(31_000);
+      await settle(true);
+      const [row] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, started.session.id));
+      expect(row.state).toBe("ended");
+      if (reason !== "hangup") expect(f.transport.endSession).toHaveBeenCalledOnce();
+      await quiet(); expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); }
+  });
+
+  it("retires uncertain creation at its saved grace deadline without creating another provider call", async () => {
+    await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
+    const f = await nativeServiceFixture();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
+    const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
+    try {
+      await f.service.registerRecoveryWork(coordinator, () => true).ready;
+      f.transport.createBrowserSession.mockRejectedValue(new SpekoProviderError("provider_unavailable", true));
+      await expect(f.service.start({ ...f.input, maxDurationSeconds: 30 })).rejects.toMatchObject({ status: 409 });
+      const row = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.idempotencyKey, f.input.idempotencyKey)).then(rows => rows[0]);
+      const deadline = row.expiresAt.getTime() + 120_001;
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBe(deadline); });
+      const reads = vi.spyOn(db, "select");
+      try { await vi.advanceTimersByTimeAsync(100_000); expect(reads).not.toHaveBeenCalled(); }
+      finally { reads.mockRestore(); }
+      await vi.advanceTimersByTimeAsync(deadline - Date.now());
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); });
+      expect((await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, row.id)))[0].state).toBe("expired");
+      expect(f.transport.createBrowserSession).toHaveBeenCalledOnce(); expect(f.transport.endSession).not.toHaveBeenCalled();
+      expect(onError).not.toHaveBeenCalled();
+    } finally { await coordinator.stop(); f.service.unregisterAgentTools(); vi.useRealTimers(); }
+  });
+
   it("reports signing-secret presence for interrupted setup without returning credential refs", async () => {
     const f = await fixture();
     const chat = chatChannelService(db, {heartbeat: {wakeup: vi.fn()}});

server/src/app.ts

@@ -1298,6 +1298,7 @@ export async function createApp(
   }
   registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
   registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
+  chatChannels.voice.registerRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
   emailChannels.start();
@@ -1305,7 +1306,7 @@ export async function createApp(
     await chatChannels.processPublicationMaintenance();
   };
   const chatReconciliation = createChatReconciliationCoordinator({
-    reconcileVoiceSessions: () => chatChannels.voice.reconcile(),
+    reconcileVoiceSessions: () => chatChannels.voice.reconcileMaintenance(),
     // Chat delivery, action, and Telegram discovery workers now own these
     // duties. Keep the coordinator's other maintenance lanes independent.
     processPendingDeliveries: async () => undefined,

server/src/services/access.ts

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } from "./voice/voice-recovery-notifications.js";
 import { notifyChatEndpointWork } from "./chat-work-notifications.js";
 import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
 import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm";
@@ -328,6 +329,7 @@ export function accessService(db: Db) {
             && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)),
           );
           if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id);
+          if (connection.transport === "voice") await notifyVoiceRecovery(tx, { kind: "connection", id: connection.id });
           if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id });
           await tx.update(toolConnections).set({
             credentialRefs,

server/src/services/chat-channels/configuration-work-notifications.ts

@@ -1,8 +1,10 @@
+import { notifyVoiceRecovery } from "../voice/voice-recovery-notifications.js";
 import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js";
 import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared";
 import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js";
 
 export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) {
+  if (endpoint.provider === "speko") await notifyVoiceRecovery(transaction, { kind: "endpoint", id: endpoint.id });
   await notifyChatReceiver(transaction, endpoint);
   if (endpoint.provider === "discord") {
     await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id });
@@ -10,6 +12,7 @@ export async function notifyChatEndpointConfiguration(transaction: object, endpo
 }
 
 export async function notifyChatConnectionConfiguration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") {
+  if (transport === "chat_sdk" || transport === "voice") await notifyVoiceRecovery(transaction, { kind: "connection", id: connectionId });
   await notifyChatConnectionReceiver(transaction, connectionId, transport);
   if (transport === "chat_sdk") {
     await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId });
@@ -17,6 +20,7 @@ export async function notifyChatConnectionConfiguration(transaction: object, con
 }
 
 export async function notifyChatCompanyConfiguration(transaction: object, companyId: string) {
+  await notifyVoiceRecovery(transaction, { kind: "company", id: companyId });
   await notifyChatCompanyReceivers(transaction, companyId);
   await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId });
 }

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

@@ -41,6 +41,7 @@ export const DELIVERY_QUEUES = {
   chatFailedRunRetries: "chat-failed-run-retries",
   chatTelegramMaintenance: "chat-telegram-maintenance",
   chatReceivers: "chat-receivers",
+  voiceSessionRecovery: "voice-session-recovery",
   chatCommandRegistration: "chat-command-registration",
   chatTelegramSubscriptions: "chat-telegram-subscriptions",
   chatSessionSyncs: "chat-session-syncs",

server/src/services/instance-settings.ts

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } from "./voice/voice-recovery-notifications.js";
 import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 import type { Db } from "@paperclipai/db";
@@ -614,6 +615,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
         await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
+        if (patch.enableChatConnectors !== undefined) await notifyVoiceRecovery(tx);
         await notifyToolConnectionHealth(tx, "*");
         return tx
           .update(instanceSettings)

server/src/services/tool-access.ts

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } from "./voice/voice-recovery-notifications.js";
 import { notifyToolConnectionHealth, writeToolConnectionHealthChange } from "./tool-connection-health-notifications.js";
 import { connectionHealthSchedule } from "./tool-access/catalog/health-schedule.js";
 import { restConnectorFor } from "./connectors/rest-registry.js";
@@ -4649,6 +4650,7 @@ export function toolAccessService(
             return rows;
           })
         : await writeToolConnectionHealthChange(db, existing, async tx => {
+            if (existing.transport === "voice") await notifyVoiceRecovery(tx, { kind: "connection", id: existing.id });
             const rows = await write(tx);
             await syncCredentialBindings(rows[0]!, [], tx);
             await createToolRuntimeSlots(tx, runtimeSupervisor).ensureRuntimeSlot(rows[0]!);

server/src/services/tool-access/connections/removal.ts

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } from "../../voice/voice-recovery-notifications.js";
 import { notifyChatEndpointWork } from "../../chat-work-notifications.js";
 import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js";
 import { toConnection } from "./projection.js";
@@ -539,6 +540,7 @@ async function removeConnection(
 
     const archived = await db.transaction(async (tx) => {
       if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id);
+          if (connection.transport === "voice") await notifyVoiceRecovery(tx, { kind: "connection", id: connection.id });
       await notifyToolConnectionHealth(tx, connection);
       await tx.select({ id: toolApplications.id }).from(toolApplications)
         .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId)))

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

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } 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";
@@ -150,6 +151,7 @@ export function voiceInboundService(db: Db, options: {
       }
       if (event.type === "call.report" || event.status && ["ended", "failed"].includes(event.status)) {
         if (session) {
+          await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
           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()}});
         }
@@ -184,7 +186,9 @@ export function voiceInboundService(db: Db, options: {
           const task = await issueSvc.create(row.companyId, {id: taskId, title: "Incoming phone conversation", description: "Conversation with an unverified phone caller. The agent runs under a task-scoped low-trust policy. You are on a live phone call and must attempt to answer as quickly as possible. Immediately post a brief, caller-safe task comment explaining what you are doing, then find the answer and post the result on this task so it can be spoken to the caller. Continue working after the initial update; do not treat it as the final answer. Preserve the task-scoped low-trust policy and ordinary permissions.", status: "todo", priority: "medium", assigneeAgentId: row.assignedAgentId, executionWorkspaceSettings: {mode: "isolated_workspace", workspaceStrategy: {type: "cloud_sandbox"}, ...(line.lowTrustEnvironmentId ? {environmentId: line.lowTrustEnvironmentId} : {})}, originKind: "chat_channel", originId: row.id, idempotencyKey: `guest-intake:${call.id}`, sourceTrust: {preset: LOW_TRUST_REVIEW_PRESET, disposition: "quarantined", sourceIssueId: taskId}, executionPolicy: {mode: "normal", commentRequired: true, stages: [], reviewPreset, authorizationPolicy: {trustPreset: LOW_TRUST_REVIEW_PRESET, reviewPreset, trustBoundary: {mode: LOW_TRUST_REVIEW_PRESET, companyId: row.companyId, rootIssueId: taskId, issueIds: [taskId], allowedAgentIds: [row.assignedAgentId], allowedToolClasses: [], allowedSecretBindingIds: []}}}}, tx);
           const [resource] = await tx.insert(chatEndpointResources).values({companyId: row.companyId, endpointId: row.id, type: "direct_message", providerResourceId: threadId, label: "Incoming phone conversation", enabled: true, availability: "available"}).returning();
           await tx.insert(chatConversations).values({id: conversationId, companyId: row.companyId, endpointId: row.id, resourceId: resource.id, issueId: task.id, externalConversationId: threadId, externalThreadId: threadId, externalLabel: "Incoming phone conversation", isDirectMessage: true, communicationGuidance: [buildChatCommunicationGuidance({ provider: "speko", isDirectMessage: true, communicationInstructions: row.communicationInstructions }), "This conversation is its own low-trust task. Ask clarifications as brief task comments; the caller answers through submit_request follow-ups. Do not use ask_user_questions or create a protected human-input wait. Use only context supplied in this conversation; do not browse other company tasks, secrets or private tools. Spoken consent does not grant governed authority."].join("\n\n")});
-          const [session] = await tx.insert(chatVoiceSessions).values({companyId: row.companyId, endpointId: row.id, conversationId, issueId: task.id, assignedAgentId: row.assignedAgentId, callerId: `guest:${call.id}`, callerAuthority: "guest_intake", mode: "inbound_phone", state: "active", generation: call.generation, credentialFingerprint: call.credentialFingerprint, providerSessionId: call.providerSessionId, toolTokenHash: call.toolTokenHash, idempotencyKey: `guest:${call.id}`, requestFingerprint: call.requestFingerprint, createdAt: call.createdAt, expiresAt: call.expiresAt}).returning({id: chatVoiceSessions.id});
+          const sessionId = randomUUID();
+          await notifyVoiceRecovery(tx, { kind: "session", id: sessionId });
+          const [session] = await tx.insert(chatVoiceSessions).values({id: sessionId, companyId: row.companyId, endpointId: row.id, conversationId, issueId: task.id, assignedAgentId: row.assignedAgentId, callerId: `guest:${call.id}`, callerAuthority: "guest_intake", mode: "inbound_phone", state: "active", generation: call.generation, credentialFingerprint: call.credentialFingerprint, providerSessionId: call.providerSessionId, toolTokenHash: call.toolTokenHash, idempotencyKey: `guest:${call.id}`, requestFingerprint: call.requestFingerprint, createdAt: call.createdAt, expiresAt: call.expiresAt}).returning({id: chatVoiceSessions.id});
           await tx.update(chatVoiceInboundCalls).set({intakeIssueId: task.id, sessionId: session.id, updatedAt: new Date()}).where(eq(chatVoiceInboundCalls.id, call.id));
           await tx.insert(activityLog).values({companyId: row.companyId, actorType: "system", actorId: "speko-inbound", action: "voice.low_trust_call.started", entityType: "issue", entityId: task.id, details: {callId: call.id, sessionId: session.id}});
         }
@@ -228,7 +232,10 @@ export function voiceInboundService(db: Db, options: {
       if (input.issueId) await store.authorizeTask(tx, companyId, input.issueId, caller);
       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) await tx.update(chatVoiceSessions).set({state: "ending", updatedAt: new Date()}).where(and(eq(chatVoiceSessions.companyId, companyId), eq(chatVoiceSessions.id, row.sessionId)));
+      if (!input.approve && row.state === "guest_intake" && row.sessionId) {
+        await notifyVoiceRecovery(tx, { kind: "session", id: row.sessionId });
+        await tx.update(chatVoiceSessions).set({state: "ending", updatedAt: new Date()}).where(and(eq(chatVoiceSessions.companyId, companyId), eq(chatVoiceSessions.id, row.sessionId)));
+      }
       return changed;
     });
     if (admission.state === "approved") return projection(admission);

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

@@ -0,0 +1,20 @@
+import type { Db } from "@paperclipai/db";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
+
+export type VoiceRecoveryScope = { kind: "session" | "endpoint" | "connection" | "company"; id: string };
+export function voiceRecoveryScope(hint: string): VoiceRecoveryScope | undefined {
+  if (hint === "*") return undefined;
+  const [kind, id, extra] = hint.split(":");
+  if (!id || extra !== undefined || !["session", "endpoint", "connection", "company"].includes(kind))
+    throw new Error("Invalid voice recovery scope");
+  return { kind: kind as VoiceRecoveryScope["kind"], id };
+}
+export function notifyVoiceRecovery(tx: object, scope?: VoiceRecoveryScope) {
+  return notifyDeliveryWork(tx, DELIVERY_QUEUES.voiceSessionRecovery, scope ? `${scope.kind}:${scope.id}` : "*");
+}
+export function withVoiceSessionChange<T>(db: Db, id: string, write: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => PromiseLike<T>): Promise<T> {
+  return db.transaction(async tx => {
+    await notifyVoiceRecovery(tx, { kind: "session", id });
+    return write(tx);
+  });
+}

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

@@ -0,0 +1,84 @@
+import { takeCompanyWorkBatch } from "../company-work-batch.js";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
+import type { VoiceRecoveryEntry, voiceSessionRecovery } from "./voice-session-recovery.js";
+import { voiceRecoveryScope, type VoiceRecoveryScope } from "./voice-recovery-notifications.js";
+
+function matches(entry: VoiceRecoveryEntry, scope?: VoiceRecoveryScope) {
+  return !scope || (scope.kind === "session" ? entry.id : scope.kind === "endpoint" ? entry.endpointId : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id;
+}
+export function registerVoiceRecoveryWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: ReturnType<typeof voiceSessionRecovery>,
+  canRun: () => boolean,
+) {
+  const dirty = new Set<string>(["*"]);
+  const entries = new Map<string, VoiceRecoveryEntry>();
+  const failures = new Map<string, { count: number; retryAt: number }>();
+  const scopeFailures = new Map<string, { count: number; retryAt: number }>();
+  let earliest = 0;
+  let lastCompany: string | null = null;
+  function failed(map: typeof failures, id: string) {
+    const count = Math.min((map.get(id)?.count ?? 0) + 1, 5);
+    map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) });
+  }
+  function dirtyScopes() {
+    return [...dirty].map(hint => { try { return voiceRecoveryScope(hint); } catch { return undefined; } });
+  }
+  function next() {
+    let nextAt: number | null = null;
+    const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); };
+    for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0));
+    const unavailable = dirtyScopes();
+    for (const entry of entries.values()) if (entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope)))
+      take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0));
+    return nextAt;
+  }
+  const worker = coordinator.register(DELIVERY_QUEUES.voiceSessionRecovery, {
+    retryMs: 30_000,
+    pausePassiveMaintenance: true,
+    onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); },
+    run: async signal => {
+      if (!canRun() || Date.now() < earliest) return;
+      earliest = Date.now() + 1000;
+      const errors: unknown[] = [];
+      const requested = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
+      for (const hint of requested) dirty.delete(hint);
+      for (let index = 0; index < requested.length; index++) {
+        if (signal.aborted || !canRun()) { for (const hint of requested.slice(index)) dirty.add(hint); return; }
+        const hint = requested[index];
+        let scope: VoiceRecoveryScope | undefined;
+        let decoded = false;
+        try {
+          scope = voiceRecoveryScope(hint); decoded = true;
+          const rows = await service.load(scope, signal), retained = new Set(rows.map(row => row.id));
+          for (const entry of entries.values()) if (matches(entry, scope) && !retained.has(entry.id)) {
+            entries.delete(entry.id); failures.delete(entry.id);
+          }
+          for (const entry of rows) entries.set(entry.id, entry);
+          scopeFailures.delete(hint);
+        } catch (error) {
+          const retryHint = decoded ? hint : "*";
+          dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error);
+        }
+      }
+      const unavailable = dirtyScopes();
+      const due = [...entries.values()].filter(entry => entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope))
+        && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())
+        .sort((a, b) => a.dueAt! - b.dueAt! || a.id.localeCompare(b.id));
+      const batch = takeCompanyWorkBatch(due, 25, lastCompany); lastCompany = batch.lastCompany;
+      for (const entry of batch.entries) {
+        if (signal.aborted || !canRun()) break;
+        try {
+          const current = await service.check(entry.id, signal);
+          if (current) entries.set(entry.id, current); else entries.delete(entry.id);
+          failures.delete(entry.id);
+        } catch (error) { failed(failures, entry.id); errors.push(error); }
+      }
+      if (errors.length) throw new AggregateError(errors, "Voice session recovery could not settle");
+    },
+    hasPending: async () => dirty.size > 0,
+    nextRunAt: next,
+  });
+  return worker;
+}

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

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } from "./voice-recovery-notifications.js";
 import { createHash } from "node:crypto";
 import { and, asc, eq, gt, inArray, isNull, lt, sql } from "drizzle-orm";
 import { chatActions, chatEndpoints, chatPublications, chatVoiceReplies, chatVoiceSessions, type Db } from "@paperclipai/db";
@@ -114,6 +115,7 @@ export async function pushVoiceReplies(db: Db, options: {
           : providerError?.httpStatus === 429 ? "retry" : "failed";
         await db.transaction(async tx => {
           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 tx.update(chatVoiceSessions).set({state: "ended", endedAt: new Date(), updatedAt: new Date()}).where(and(eq(chatVoiceSessions.id, candidate.id), eq(chatVoiceSessions.companyId, candidate.companyId)));
           if (status === "unknown") await tx.update(chatVoiceSessions).set({errorCode: "message_delivery_unknown", updatedAt: new Date()}).where(eq(chatVoiceSessions.id, candidate.id));
         });

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

@@ -0,0 +1,89 @@
+import { and, eq, gt, inArray, sql } from "drizzle-orm";
+import { chatEndpoints, chatVoiceSessions, toolConnections, type Db } from "@paperclipai/db";
+import { HttpError } from "../../errors.js";
+import { instanceSettingsService } from "../instance-settings.js";
+import { withVoiceSessionChange, type VoiceRecoveryScope } from "./voice-recovery-notifications.js";
+
+export const LIVE_VOICE_STATES = ["reserved", "creating", "creation_unknown", "connecting", "active", "ending"] as const;
+type Session = typeof chatVoiceSessions.$inferSelect;
+export type VoiceRecoveryEntry = { id: string; companyId: string; endpointId: string; connectionId: string; dueAt: number };
+type Candidate = { session: Session; connectionId: string; endpointStatus: string; enabled: boolean; generation: number };
+
+export function voiceSessionRecoveryAt(row: Candidate, featureEnabled: boolean): number {
+  const session = row.session;
+  if (!session.providerSessionId) return session.expiresAt.getTime() + 120_001;
+  const changed = !featureEnabled || !row.enabled || !["active", "verifying"].includes(row.endpointStatus)
+    || session.generation !== row.generation || session.state === "ending";
+  if (changed) return session.errorCode === "cleanup_pending" ? session.updatedAt.getTime() + 30_000 : 0;
+  if (["connecting", "active"].includes(session.state)) return Math.min(session.expiresAt.getTime(), session.updatedAt.getTime() + 30_000);
+  return session.expiresAt.getTime();
+}
+
+export function voiceSessionRecovery(db: Db, operations: {
+  authorize(session: Session): Promise<void>;
+  inspect(session: Session): Promise<{ endedAt?: unknown } | null>;
+  close(session: Session): Promise<unknown>;
+}) {
+  async function candidates(scope?: VoiceRecoveryScope, signal?: AbortSignal) {
+    const rows: Candidate[] = [];
+    let cursor: string | undefined;
+    for (;;) {
+      signal?.throwIfAborted();
+      const page = await db.select({ session: chatVoiceSessions, connectionId: toolConnections.id,
+        endpointStatus: chatEndpoints.status, enabled: toolConnections.enabled, generation: sql<number>`coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0)` })
+        .from(chatVoiceSessions)
+        .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatVoiceSessions.endpointId), eq(chatEndpoints.companyId, chatVoiceSessions.companyId)))
+        .innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatVoiceSessions.companyId)))
+        .where(and(inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES]),
+          scope?.kind === "session" ? eq(chatVoiceSessions.id, scope.id) : undefined,
+          scope?.kind === "endpoint" ? eq(chatVoiceSessions.endpointId, scope.id) : undefined,
+          scope?.kind === "connection" ? eq(toolConnections.id, scope.id) : undefined,
+          scope?.kind === "company" ? eq(chatVoiceSessions.companyId, scope.id) : undefined,
+          cursor ? gt(chatVoiceSessions.id, cursor) : undefined))
+        .orderBy(chatVoiceSessions.id).limit(100);
+      rows.push(...page);
+      if (page.length < 100) return rows;
+      cursor = page[page.length - 1].session.id;
+    }
+  }
+  async function load(scope?: VoiceRecoveryScope, signal?: AbortSignal): Promise<VoiceRecoveryEntry[]> {
+    const rows = await candidates(scope, signal);
+    if (!rows.length) return [];
+    const enabled = (await instanceSettingsService(db).getExperimental()).enableChatConnectors;
+    return rows.map(row => ({ id: row.session.id, companyId: row.session.companyId, endpointId: row.session.endpointId,
+      connectionId: row.connectionId, dueAt: voiceSessionRecoveryAt(row, enabled) }));
+  }
+  async function check(id: string, signal?: AbortSignal): Promise<VoiceRecoveryEntry | null> {
+    const scope = { kind: "session", id } as const;
+    const [row] = await candidates(scope, signal);
+    if (!row) return null;
+    const enabled = (await instanceSettingsService(db).getExperimental()).enableChatConnectors;
+    if (voiceSessionRecoveryAt(row, enabled) <= Date.now()) {
+      signal?.throwIfAborted();
+      const session = row.session;
+      try {
+        let authorized = false;
+        if (enabled && ["connecting", "active"].includes(session.state) && session.expiresAt.getTime() > Date.now()) {
+          try { await operations.authorize(session); authorized = true; }
+          catch (error) { if (!(error instanceof HttpError && [403, 404].includes(error.status))) throw error; }
+        }
+        if (authorized) {
+          const state = await operations.inspect(session);
+          await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({
+            ...(state?.endedAt ? { state: "ended" as const, endedAt: new Date(), errorCode: null } : {}), updatedAt: new Date(),
+          }).where(and(eq(chatVoiceSessions.id, id), inArray(chatVoiceSessions.state, ["connecting", "active"]))));
+        } else if (session.providerSessionId) await operations.close(session);
+        else if (session.expiresAt.getTime() + 120_000 < Date.now()) {
+          await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ state: "expired", endedAt: new Date(), updatedAt: new Date() })
+            .where(and(eq(chatVoiceSessions.id, id), inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES]))));
+        }
+      } catch (error) {
+        await withVoiceSessionChange(db, id, tx => tx.update(chatVoiceSessions).set({ errorCode: "cleanup_pending", updatedAt: new Date() })
+          .where(and(eq(chatVoiceSessions.id, id), inArray(chatVoiceSessions.state, [...LIVE_VOICE_STATES]))));
+        throw error;
+      }
+    }
+    return (await load(scope, signal))[0] ?? null;
+  }
+  return { load, check };
+}

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

@@ -1,3 +1,7 @@
+import { voiceSessionRecovery } from "./voice-session-recovery.js";
+import { registerVoiceRecoveryWork } from "./voice-recovery-work.js";
+import { notifyVoiceRecovery, withVoiceSessionChange } from "./voice-recovery-notifications.js";
+import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { logger } from "../../middleware/logger.js";
 import { captureVoiceDeliveryDiagnostics } from "./voice-delivery-diagnostics.js";
 import { pushVoiceReplies } from "./voice-reply-push.js";
@@ -92,7 +96,7 @@ export function voiceSessionService(db: Db, options: {
     // browser response must inspect/end this attempt, never blindly create again.
     if (!reserved.created || input.mode === "inbound_phone") return { session: serializeVoiceSession(reserved.session) };
     const session = reserved.session;
-    const [claim] = await db.update(chatVoiceSessions).set({ state: "creating", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "reserved"))).returning({ id: chatVoiceSessions.id });
+    const [claim] = await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: "creating", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "reserved"))).returning({ id: chatVoiceSessions.id }));
     if (!claim) throw conflict("The call was ended before Speko connected");
     let client: ReturnType<typeof createSpekoProvider> | undefined;
     let created: { sessionId: string; transportToken?: string; transportUrl?: string };
@@ -107,13 +111,13 @@ export function voiceSessionService(db: Db, options: {
         : await client.createBrowserSession({ ...shared, systemPrompt: VOICE_PROMPT });
     } catch (error) {
       const known = error instanceof SpekoProviderError && !error.outcomeUnknown || error instanceof HttpError && [403, 404].includes(error.status);
-      await db.update(chatVoiceSessions).set({ state: known ? "failed" : "creation_unknown", errorCode: error instanceof SpekoProviderError ? error.code : "provider_unavailable", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating")));
+      await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: known ? "failed" : "creation_unknown", errorCode: error instanceof SpekoProviderError ? error.code : "provider_unavailable", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating"))));
       if (error instanceof SpekoProviderError && error.code === "insufficient_credits") throw conflict("Speko needs credits before starting a call. Add credits in Speko, then try again.", {code: "voice_credits_required", sessionId: session.id});
       throw conflict(known ? "Speko could not start the call. Check connection status before trying again." : "Speko may have started this call. Inspect this attempt before starting another.", { code: known ? "voice_creation_failed" : "voice_creation_unknown", sessionId: session.id });
     }
     let bound: VoiceSessionRow | undefined;
     try {
-    [bound] = await db.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "connecting", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating"))).returning();
+    [bound] = await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "connecting", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), eq(chatVoiceSessions.state, "creating"))).returning());
     } catch (error) {
       // If the provider started but persistence failed, stop that exact call.
       // Never return media or retry creation on an uncertain database outcome.
@@ -122,7 +126,7 @@ export function voiceSessionService(db: Db, options: {
     }
     if (!bound) {
       // A concurrent end must still learn the provider identity for reconciliation.
-      await db.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "ending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
+      await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ providerSessionId: created.sessionId, state: "ending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)));
       await closeProvider({ ...session, providerSessionId: created.sessionId, state: "ending" });
       throw conflict("The call was ended while Speko was connecting");
     }
@@ -141,6 +145,7 @@ export function voiceSessionService(db: Db, options: {
   async function closeProviderOnce(session: VoiceSessionRow) {
     if (terminalStates.some(state => state === session.state)) return session;
     await db.transaction(async (tx) => {
+      await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
       const [changed] = await tx.update(chatVoiceSessions).set({ state: "ending", updatedAt: new Date() }).where(and(eq(chatVoiceSessions.id, session.id), inArray(chatVoiceSessions.state, [...liveStates].filter((state) => state !== "ending")))).returning({ id: chatVoiceSessions.id });
       if (changed) await tx.insert(activityLog).values({ companyId: session.companyId, actorType: "system", actorId: "voice-session-service", action: "voice.session.ending", entityType: "voice_session", entityId: session.id, details: { issueId: session.issueId } });
     });
@@ -156,11 +161,11 @@ export function voiceSessionService(db: Db, options: {
         try { confirmed = Boolean((await client.inspectSession(session.providerSessionId)).endedAt); }
         catch (error) { if (!(error instanceof SpekoProviderError)) throw error; }
       }
-      await db.update(chatVoiceSessions).set(confirmed
+      await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set(confirmed
         ? { state: "ended", endedAt: new Date(), errorCode: null, updatedAt: new Date() }
-        : { errorCode: "cleanup_pending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
+        : { errorCode: "cleanup_pending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)));
     } else if (session.state === "reserved") {
-      await db.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
+      await withVoiceSessionChange(db, session.id, tx => tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id)));
     }
     return db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, session.id)).then((rows) => rows[0]);
   }
@@ -170,7 +175,7 @@ export function voiceSessionService(db: Db, options: {
       const endpoint = await endpointFor(companyId, session.endpointId), credentials = await options.credentials(endpoint);
       const state = await provider(credentials.apiKey).inspectSession(session.providerSessionId);
       if (state.endedAt) {
-        [session] = await db.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, sessionId)).returning();
+        [session] = await withVoiceSessionChange(db, sessionId, tx => tx.update(chatVoiceSessions).set({ state: "ended", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, sessionId)).returning());
       }
     }
     return serializeVoiceSession(session);
@@ -199,51 +204,30 @@ export function voiceSessionService(db: Db, options: {
     if (verified.envelope.tool === "answer_question") options.onQuestionAnswered(verified.envelope.args.interactionId);
     return session.mode === "inbound_phone" ? {...result, authorization: session.authority === "guest_intake" ? "guest_intake" : "approved"} : result;
   }
-  async function reconcile(limit = 25) {
-    // Provider maximum duration is the final bound even if creation response was
-    // lost. Unknown attempts remain reserved until that bound plus teardown grace.
-    const enabled = (await instanceSettingsService(db).getExperimental()).enableChatConnectors;
-    const rows = await db.select({ session: chatVoiceSessions }).from(chatVoiceSessions)
-      .innerJoin(chatEndpoints, eq(chatEndpoints.id, chatVoiceSessions.endpointId))
-      .innerJoin(toolConnections, eq(toolConnections.id, chatEndpoints.connectionId))
-      .where(and(inArray(chatVoiceSessions.state, [...liveStates]), or(
-        enabled ? undefined : sql`true`,
-        lte(chatVoiceSessions.expiresAt, new Date()), eq(chatVoiceSessions.state, "ending"),
-        notInArray(chatEndpoints.status, ["active", "verifying"]), eq(toolConnections.enabled, false),
-        sql`${chatVoiceSessions.generation} <> coalesce((${chatEndpoints.setup}->>'runtimeGeneration')::integer, 0)`,
-        // Recheck membership and bindings periodically even without callbacks.
-        and(inArray(chatVoiceSessions.state, ["connecting", "active"]), lte(chatVoiceSessions.updatedAt, new Date(Date.now() - 30_000))),
-      ))).orderBy(asc(chatVoiceSessions.updatedAt)).limit(limit);
-    for (const { session } of rows) {
-      try {
-        let authorized = false;
-        if (enabled && ["connecting", "active"].includes(session.state) && session.callerAuthority === "guest_intake" && session.expiresAt.getTime() > Date.now()) {
-          authorized = await inbound.authorizeGuest(session.id);
-        } else if (enabled && ["connecting", "active"].includes(session.state)) {
-          try { await db.transaction((tx) => store.authorizeSession(tx, session.companyId, session.id)); authorized = true; } catch { /* revoked, changed binding, or expired */ }
-        }
-        if (authorized) {
-          // A caller can hang up without a final browser request. Poll provider
-          // state as well as permissions so a missed callback releases the task.
-          const endpoint = await endpointFor(session.companyId, session.endpointId);
-          const credentials = await options.credentials(endpoint);
-          const state = session.providerSessionId ? await provider(credentials.apiKey).inspectSession(session.providerSessionId) : null;
-          await db.update(chatVoiceSessions).set({
-            ...(state?.endedAt ? { state: "ended", endedAt: new Date(), errorCode: null } : {}),
-            updatedAt: new Date(),
-          }).where(and(eq(chatVoiceSessions.id, session.id), inArray(chatVoiceSessions.state, ["connecting", "active"])));
-          continue;
-        }
-        if (session.providerSessionId) await closeProvider(session);
-        else if (session.expiresAt.getTime() + 120_000 < Date.now()) await db.update(chatVoiceSessions).set({ state: "expired", endedAt: new Date(), updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
-      } catch {
-        await db.update(chatVoiceSessions).set({ errorCode: "cleanup_pending", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
-      }
-    }
+  const recovery = voiceSessionRecovery(db, {
+    authorize: async session => {
+      await db.transaction(tx => store.authorizeSession(tx, session.companyId, session.id));
+    },
+    inspect: async session => {
+      const endpoint = await endpointFor(session.companyId, session.endpointId);
+      const credentials = await options.credentials(endpoint);
+      return session.providerSessionId ? provider(credentials.apiKey).inspectSession(session.providerSessionId) : null;
+    },
+    close: closeProvider,
+  });
+  function registerRecoveryWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
+    return registerVoiceRecoveryWork(coordinator, recovery, canRun);
+  }
+  async function reconcileMaintenance(limit = 25) {
     await pushReplies(limit);
     await inbound.reconcile(limit);
     await reconcileReports(limit);
-    return rows.length;
+  }
+  async function reconcile(limit = 25) {
+    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 reconcileMaintenance(limit);
+    return due.length;
   }
   async function pushReplies(limit = 25, companyId?: string, endpointId?: string) {
     return pushVoiceReplies(db, {store, credentials: options.credentials, provider, limit, companyId, endpointId});
@@ -355,7 +339,7 @@ export function voiceSessionService(db: Db, options: {
     const result = await start({ ...binding, mode: "outbound_phone", phoneNumber: preference.phoneNumber, idempotencyKey: invocationId, maxDurationSeconds: 600 });
     return { session: result.session };
   });
-  return { start, inspect, end, tool, reconcile, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools };
+  return { start, inspect, end, tool, reconcile, reconcileMaintenance, registerRecoveryWork, pushReplies, inbound, notification: store.notification, callbackPreference, saveCallbackPreference, history, callDetail, unregisterAgentTools };
 
 }
 export type VoiceSessionService = ReturnType<typeof voiceSessionService>;

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

@@ -1,3 +1,4 @@
+import { notifyVoiceRecovery } from "./voice-recovery-notifications.js";
 import { notifyChatDeliveryWork } from "../chat-work-notifications.js";
 import { environmentService } from "../environments.js";
 import { logger } from "../../middleware/logger.js";
@@ -192,13 +193,16 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
           if (!["connecting", "active"].includes(guest.state) || guest.expiresAt.getTime() <= Date.now() || guest.generation !== context.generation || guest.toolTokenHash !== voiceTokenHash(input.toolToken ?? "")) throw forbidden("Guest call is no longer available for approval");
           const task = await authorizeTask(tx, input.companyId, binding.issueId, input.caller);
           if (task.assigneeAgentId !== context.endpoint.assignedAgentId) throw conflict("The selected agent must be assigned to this task");
+          await notifyVoiceRecovery(tx, { kind: "session", id: guest.id });
           const [promoted] = await tx.update(chatVoiceSessions).set({...binding, callerId: input.caller.id, callerAuthority: input.caller.authority, approvedByUserId: input.caller.id, idempotencyKey: input.idempotencyKey, requestFingerprint: input.requestFingerprint, updatedAt: new Date()}).where(eq(chatVoiceSessions.id, guest.id)).returning();
           await tx.insert(activityLog).values({companyId: input.companyId, actorType: "user", actorId: input.caller.id, action: "voice.guest.private_call_approved", entityType: "voice_session", entityId: guest.id, details: {issueId: binding.issueId}});
           return {session: promoted, created: false as const};
         }
       }
       const token = input.toolToken ?? randomBytes(32).toString("base64url"), createdAt = new Date();
-      const [session] = await tx.insert(chatVoiceSessions).values({ id: randomUUID(), companyId: input.companyId, endpointId: input.endpointId, ...binding, assignedAgentId: context.endpoint.assignedAgentId,
+      const sessionId = randomUUID();
+      await notifyVoiceRecovery(tx, { kind: "session", id: sessionId });
+      const [session] = await tx.insert(chatVoiceSessions).values({ id: sessionId, companyId: input.companyId, endpointId: input.endpointId, ...binding, assignedAgentId: context.endpoint.assignedAgentId,
         callerId: input.caller.id, callerAuthority: input.caller.authority, mode: input.mode ?? "browser", state: input.providerSessionId ? "connecting" : "reserved", providerSessionId: input.providerSessionId ?? null, approvedByUserId: input.mode === "inbound_phone" ? input.caller.id : null, generation: context.generation, credentialFingerprint: context.fingerprint,
         toolTokenHash: voiceTokenHash(token), idempotencyKey: input.idempotencyKey, requestFingerprint: input.requestFingerprint, createdAt, expiresAt: new Date(createdAt.getTime() + input.maxDurationSeconds * 1000) }).returning();
       await tx.insert(activityLog).values({ companyId: input.companyId, actorType: "user", actorId: input.caller.id, action: "voice.session.requested", entityType: "voice_session", entityId: session.id, details: { endpointId: input.endpointId, issueId: session.issueId, mode: input.mode ?? "browser" } });
@@ -269,6 +273,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
       await tx.insert(chatVoiceReplies).values({ companyId: session.companyId, sessionId: session.id, publicationId: publication.id, cursor: ++cursor });
     }
     if (cursor !== session.replyCursor) {
+      await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
       await tx.update(chatVoiceSessions).set({ replyCursor: cursor, updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
       session.replyCursor = cursor;
     }
@@ -290,6 +295,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
     if (existing) return;
     const cursor = session.replyCursor + 1;
     await tx.insert(chatVoiceReplies).values({ companyId: session.companyId, sessionId: session.id, publicationId: previous.publicationId, cursor, deliveredAt: new Date() });
+    await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
     await tx.update(chatVoiceSessions).set({ replyCursor: cursor, updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
     session.replyCursor = cursor;
   }
@@ -327,6 +333,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
       const context = await authorizeSession(tx, input.companyId, input.sessionId), { session } = context;
       if (session.endpointId !== input.endpointId || session.providerSessionId !== input.envelope.session_id || !timingSafeEqual(Buffer.from(session.toolTokenHash, "hex"), Buffer.from(voiceTokenHash(input.token), "hex"))) throw forbidden("Voice tool scope does not match this session");
       if (session.state === "connecting") {
+        await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
         await tx.update(chatVoiceSessions).set({ state: "active", updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
         await tx.insert(activityLog).values({ companyId: session.companyId, actorType: "system", actorId: "voice-session-service", action: "voice.session.connected", entityType: "voice_session", entityId: session.id, details: { endpointId: session.endpointId, issueId: session.issueId } });
       }
@@ -463,6 +470,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
           if (priorCall?.blocked) return;
           const cursor = session.replyCursor + 1;
           await tx.insert(chatVoiceReplies).values({ companyId, sessionId: session.id, publicationId: publication.id, cursor });
+          await notifyVoiceRecovery(tx, { kind: "session", id: session.id });
           await tx.update(chatVoiceSessions).set({ replyCursor: cursor, updatedAt: new Date() }).where(eq(chatVoiceSessions.id, session.id));
           return {sessionId: session.id, providerSessionId: session.providerSessionId, replyCursor: cursor, mode: session.mode};
         }).then(queued => {