fix(server): bind inbound approval with its authorized voice session

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

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

@@ -24,6 +24,7 @@ import { chatActions, instanceUserRoles, principalPermissionGrants, environments
 import { configureSpekoSessionTools } from "../services/voice/speko-tool-setup.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
 import { chatChannelService } from "../services/chat-channels.js";
+import { voiceInboundService } from "../services/voice/voice-inbound-service.js";
 import { voiceSessionService } from "../services/voice/voice-session-service.js";
 import { captureVoiceDeliveryDiagnostics } from "../services/voice/voice-delivery-diagnostics.js";
 import { parseSpekoDeliveryDiagnostics } from "../services/voice/speko-delivery-diagnostics.js";
@@ -1256,6 +1257,40 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     expect(repeated?.latestRevisionId).toBe(first?.latestRevisionId);
     expect(await docs.listIssueDocuments(f.call.intakeIssueId!)).toHaveLength(1);
   });
+  it.each([false, true])("hands inbound cleanup ownership to the authorized session atomically (guest=%s)", async guest => {
+    const f = await inboundFixture(guest);
+    if (guest) await f.tool("submit_request", { text: "A public request before private approval" });
+    let release!: () => void, bound!: () => void;
+    const bindingReached = new Promise<void>(resolve => { bound = resolve; });
+    const releaseApproval = new Promise<void>(resolve => { release = resolve; });
+    const inbound = voiceInboundService(db, {
+      allowLocalBoard: false, credentials: f.credentials, provider: () => f.transport,
+      voicePrompt: "Test voice", lowTrustVoicePrompt: "Test guest voice",
+      start: async input => {
+        const result = await f.service.start(input);
+        bound(); await releaseApproval;
+        return result;
+      },
+    });
+    vi.useFakeTimers({ toFake: ["Date"] });
+    const approval = inbound.decide(f.companyId, f.endpointId, f.call.id, f.input.caller,
+      { approve: true, approvalCode: f.call.approvalCode, issueId: f.issueId });
+    try {
+      await Promise.race([bindingReached, approval.then(() => { throw new Error("Approval did not reach session binding"); })]);
+      // The session has committed but its caller has not received the result.
+      // Expiry or process death here must not leave two cleanup owners.
+      vi.setSystemTime(f.call.expiresAt.getTime() + 1);
+      await inbound.reconcile(1000);
+      expect(f.transport.endSession).not.toHaveBeenCalledWith(f.call.providerSessionId);
+      const [call] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id));
+      expect(call).toMatchObject({ state: "approved" });
+      expect(call.sessionId).not.toBeNull();
+      const [session] = await db.select().from(chatVoiceSessions).where(eq(chatVoiceSessions.id, call.sessionId!));
+      expect(session).toMatchObject({ providerSessionId: f.call.providerSessionId, callerId: f.callerId, issueId: f.issueId });
+      release();
+      await expect(approval).resolves.toMatchObject({ state: "approved", sessionId: session.id });
+    } finally { release(); await approval.catch(() => undefined); vi.useRealTimers(); }
+  });
   it("requires approval of the specific incoming call before delivering any private work", async () => {
     const f = await inboundFixture();
     expect(await f.service.inbound.lifecycle(f.endpoint.publicId, f.signed.body, f.signed.headers)).toEqual(f.response);

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

@@ -244,7 +244,7 @@ export function voiceInboundService(db: Db, options: {
     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"); }
     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.update(chatVoiceInboundCalls).set({state: "approved", sessionId: result.session.id, updatedAt: new Date()}).where(and(eq(chatVoiceInboundCalls.id, callId), eq(chatVoiceInboundCalls.state, "approving"))).returning();
+    [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");
     return projection(admission);
   }

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

@@ -175,9 +175,18 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
       const context = await endpointContext(tx, input.companyId, input.endpointId);
       await authorizeConnection(tx, context, input.caller);
       if (context.generation < 1 || input.maxDurationSeconds < 30 || input.maxDurationSeconds > 1800 || !Number.isInteger(input.maxDurationSeconds)) throw conflict("Voice setup or duration is invalid");
+      let inboundAdmissionId: string | undefined;
+      async function bindInboundSession(sessionId: string) {
+        if (!inboundAdmissionId) return;
+        // Admission expiry must relinquish cleanup in the same commit that
+        // establishes an authorized owner for the existing provider call.
+        await tx.update(chatVoiceInboundCalls).set({ state: "approved", sessionId, updatedAt: new Date() })
+          .where(eq(chatVoiceInboundCalls.id, inboundAdmissionId));
+      }
       if (input.mode === "inbound_phone") {
         const [admission] = await tx.select().from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, input.companyId), eq(chatVoiceInboundCalls.endpointId, input.endpointId), eq(chatVoiceInboundCalls.providerSessionId, input.providerSessionId ?? ""))).for("update");
         if (!admission || admission.state !== "approving" || admission.approvedByUserId !== input.caller.id || admission.callerAuthority !== input.caller.authority || admission.expiresAt.getTime() <= Date.now() || admission.generation !== context.generation || admission.credentialFingerprint !== context.fingerprint || admission.toolTokenHash !== voiceTokenHash(input.toolToken ?? "") || input.idempotencyKey !== `inbound:${admission.id}`) throw forbidden("This live call has not been approved for private work");
+        inboundAdmissionId = admission.id;
       }
       const binding = await input.prepareBinding(tx);
       const task = await authorizeTask(tx, input.companyId, binding.issueId, input.caller);
@@ -197,6 +206,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
           await notifyVoiceChange(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}});
+          await bindInboundSession(promoted.id);
           return {session: promoted, created: false as const};
         }
       }
@@ -207,6 +217,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
         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" } });
+      await bindInboundSession(session.id);
       return { session, created: true as const, token };
     });
   }