diff --git a/src/gateway/voice/bridge.ts b/src/gateway/voice/bridge.ts index 3fdc3de..f10ff9b 100644 --- a/src/gateway/voice/bridge.ts +++ b/src/gateway/voice/bridge.ts @@ -58,6 +58,16 @@ export function createCallBridge( ) { const wss = new WebSocketServer({ noServer: true }); const extraHeaders = new WeakMap(); + const closedSockets = new WeakSet(); + const closeSocketOnce = (ws: WebSocket | undefined) => { + if (!ws || closedSockets.has(ws)) return; + closedSockets.add(ws); + try { + ws.close(); + } catch { + /* already closing */ + } + }; wss.on("headers", (headers, req) => { for (const h of extraHeaders.get(req) ?? []) headers.push(h); }); @@ -129,11 +139,7 @@ export function createCallBridge( void runCall(ws, ctx, meta, realtime).catch((err) => { deps.logger.error("call.failed", { error: String(err) }); void realtime?.close().catch(() => {}); - try { - ws.close(); - } catch { - /* already closing */ - } + closeSocketOnce(ws); }); }); } @@ -258,11 +264,7 @@ export function createCallBridge( return describeContacts(await client.contacts.lookup(filters)); }, onHangup: () => { - try { - callWs?.close(); - } catch { - /* already closing */ - } + closeSocketOnce(callWs); }, logger: deps.logger, }, @@ -359,11 +361,7 @@ export function createCallBridge( } if (frame.event === "stop" || frame.event === "closed" || frame.event === "hangup") { finishCall(); - try { - ws.close(); - } catch { - /* already closing */ - } + closeSocketOnce(ws); } } @@ -394,11 +392,7 @@ export function createCallBridge( await callEnded; if (poll) clearInterval(poll); - try { - ws.close(); - } catch { - /* already closing */ - } + closeSocketOnce(ws); await realtime?.close(); if (ctx.callId) { diff --git a/src/gateway/voice/realtime.ts b/src/gateway/voice/realtime.ts index 4e86e5f..eaebfd5 100644 --- a/src/gateway/voice/realtime.ts +++ b/src/gateway/voice/realtime.ts @@ -11,6 +11,9 @@ export const HANG_UP_TOOL = "hang_up_call"; export const CONTACT_LOOKUP_TOOL = "inkbox_lookup_contact"; export const CONTACT_LIST_TOOL = "inkbox_list_contacts"; const HANGUP_WINDOW_MS = 10_000; +const CONTACT_READ_TIMEOUT_MS = 30_000; +const CONSULT_TIMEOUT_MS = 300_000; +const RESULT_RESPONSE_TIMEOUT_MS = 30_000; export interface RealtimeConfig { apiKey: string; @@ -158,6 +161,49 @@ export function openRealtimeBridge( ); const hangup = createHangupArmer(HANGUP_WINDOW_MS, now); const consults = new Set>(); + const pendingWork = new Set(); + const responseOwners: Array = []; + const ownedResponses = new Map< + string, + { callId: string; done: boolean; audioDone: boolean; transcriptDone: boolean } + >(); + const responseTimeouts = new Map>(); + let hangupRequested = false; + let hangupDelivered = false; + let closed = false; + + const finishWork = (callId: string) => { + pendingWork.delete(callId); + const timeout = responseTimeouts.get(callId); + if (timeout) clearTimeout(timeout); + responseTimeouts.delete(callId); + maybeHangup(); + }; + const maybeFinishResponse = (responseId: string) => { + const owned = ownedResponses.get(responseId); + if (!owned?.done || !owned.audioDone || !owned.transcriptDone) return; + ownedResponses.delete(responseId); + finishWork(owned.callId); + }; + const maybeHangup = () => { + if (!hangupRequested || hangupDelivered || pendingWork.size > 0) return; + hangupDelivered = true; + cb.onHangup(); + }; + const withTimeout = async (work: Promise, timeoutMs: number): Promise => { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + work, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error("realtime tool timed out")), timeoutMs); + timer.unref?.(); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } + }; // Function calls arrive across three events: output_item.added carries the // name + call id, arguments.delta streams the JSON, arguments.done fires the // dispatch. Accumulate by item/call id. @@ -223,14 +269,47 @@ export function openRealtimeBridge( return; } switch (evt.type) { + case "response.created": { + const owner = responseOwners.shift(); + const responseId = String(evt.response?.id ?? evt.response_id ?? ""); + if (owner && responseId) { + ownedResponses.set(responseId, { + callId: owner, + done: false, + audioDone: false, + transcriptDone: false, + }); + } + break; + } case "response.output_audio.delta": case "response.audio.delta": if (typeof evt.delta === "string") cb.onAudio(evt.delta); break; case "response.output_audio.done": - case "response.audio.done": + case "response.audio.done": { cb.onAudioDone?.(); + const responseId = String(evt.response_id ?? evt.response?.id ?? ""); + const owned = ownedResponses.get(responseId); + if (owned) { + owned.audioDone = true; + maybeFinishResponse(responseId); + } + break; + } + case "response.done": { + const responseId = String(evt.response?.id ?? evt.response_id ?? ""); + const owned = ownedResponses.get(responseId); + if (owned) { + owned.done = true; + if (String(evt.response?.status ?? "completed") !== "completed") { + owned.audioDone = true; + owned.transcriptDone = true; + } + maybeFinishResponse(responseId); + } break; + } case "conversation.item.input_audio_transcription.completed": { const text = String(evt.transcript ?? "").trim(); if (text) cb.onTranscript?.("caller", text); @@ -240,6 +319,12 @@ export function openRealtimeBridge( case "response.audio_transcript.done": { const text = String(evt.transcript ?? "").trim(); if (text) cb.onTranscript?.("agent", text); + const responseId = String(evt.response_id ?? evt.response?.id ?? ""); + const owned = ownedResponses.get(responseId); + if (owned && text) { + owned.transcriptDone = true; + maybeFinishResponse(responseId); + } break; } case "response.output_item.added": { @@ -317,8 +402,8 @@ export function openRealtimeBridge( } if (name === HANG_UP_TOOL) { if (hangup.press()) { - respond(callId, "Ending the call now."); - cb.onHangup(); + hangupRequested = true; + maybeHangup(); } else { respond(callId, "Armed. Say goodbye, then call hang_up_call once more to end."); } @@ -337,15 +422,16 @@ export function openRealtimeBridge( } if (name === CONTACT_LOOKUP_TOOL || name === CONTACT_LIST_TOOL) { const kind = name === CONTACT_LOOKUP_TOOL ? "lookup" : "list"; + pendingWork.add(callId); const task = (async () => { try { const summary = cb.onContactRead - ? await cb.onContactRead(kind, args) + ? await withTimeout(cb.onContactRead(kind, args), CONTACT_READ_TIMEOUT_MS) : "Contact reads are not available on this call."; - respond(callId, summary); + respond(callId, summary, callId); } catch (err) { cb.logger.warn("realtime.contact_read_failed", { error: String(err) }); - respond(callId, "The contact lookup failed."); + respond(callId, "The contact lookup failed.", callId); } })(); consults.add(task); @@ -354,13 +440,17 @@ export function openRealtimeBridge( } if (name === CONSULT_TOOL) { // Run the agent turn off the audio pump so speech keeps flowing. + pendingWork.add(callId); const task = (async () => { try { - const answer = await cb.onConsult(String(args.query ?? "")); - respond(callId, answer || "Done."); + const answer = await withTimeout( + cb.onConsult(String(args.query ?? "")), + CONSULT_TIMEOUT_MS, + ); + respond(callId, answer || "Done.", callId); } catch (err) { cb.logger.warn("realtime.consult_failed", { error: String(err) }); - respond(callId, "I hit a problem doing that."); + respond(callId, "I hit a problem doing that.", callId); } })(); consults.add(task); @@ -369,24 +459,34 @@ export function openRealtimeBridge( } // Return a function-call result to the model and ask it to speak. - function respond(callId: string, output: string): void { + function respond(callId: string, output: string, workOwner: string | null = null): void { + if (closed || ws.readyState !== WebSocket.OPEN) { + if (workOwner) finishWork(workOwner); + return; + } ws.send( JSON.stringify({ type: "conversation.item.create", item: { type: "function_call_output", call_id: callId, output }, }), ); + responseOwners.push(workOwner); ws.send(JSON.stringify({ type: "response.create" })); + if (workOwner) { + const timeout = setTimeout(() => finishWork(workOwner), RESULT_RESPONSE_TIMEOUT_MS); + timeout.unref?.(); + responseTimeouts.set(workOwner, timeout); + } } return { pushAudio(base64Ulaw) { - if (ws.readyState === WebSocket.OPEN) { + if (!closed && ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify({ type: "input_audio_buffer.append", audio: base64Ulaw })); } }, start(greetingInstructions) { - if (ws.readyState === WebSocket.OPEN) { + if (!closed && ws.readyState === WebSocket.OPEN) { ws.send( JSON.stringify({ type: "response.create", @@ -397,7 +497,13 @@ export function openRealtimeBridge( }, ready, async close() { - await Promise.allSettled([...consults]); + if (closed) return; + closed = true; + for (const timeout of responseTimeouts.values()) clearTimeout(timeout); + responseTimeouts.clear(); + pendingWork.clear(); + responseOwners.length = 0; + ownedResponses.clear(); try { ws.close(); } catch { diff --git a/tests/gateway/bridge.test.ts b/tests/gateway/bridge.test.ts index 91daba1..257a031 100644 --- a/tests/gateway/bridge.test.ts +++ b/tests/gateway/bridge.test.ts @@ -97,6 +97,33 @@ async function waitForCalls(mock: ReturnType, count: number): Prom } describe("call bridge signed context", () => { + it("lets plugin hangup close the call and realtime bridge exactly once", async () => { + const { bridgeDeps, runText } = deps(true); + let callbacks: RealtimeCallbacks | undefined; + const closeRealtime = vi.fn(async () => {}); + const openRealtime = vi.fn((_cfg: RealtimeConfig, _registry, cb: RealtimeCallbacks) => { + callbacks = cb; + return { + ready: Promise.resolve(), + start: vi.fn(), + pushAudio: vi.fn(), + close: closeRealtime, + }; + }); + const closeSpy = vi.spyOn(WebSocket.prototype, "close"); + process.env.INKBOX_REALTIME_API_KEY = "test-key"; + + const ws = await connect(createCallBridge(bridgeDeps, openRealtime as never)); + callbacks?.onHangup(); + callbacks?.onHangup(); + expect(closeSpy).toHaveBeenCalledTimes(1); + await new Promise((resolve) => ws.once("close", () => resolve())); + await waitForCalls(runText, 1); + + expect(closeRealtime).toHaveBeenCalledTimes(1); + closeSpy.mockRestore(); + }); + it("uses top-level call fields and contacts in realtime, consult, and post-call prompts", async () => { const { bridgeDeps, runText } = deps(true); let callbacks: RealtimeCallbacks | undefined; diff --git a/tests/gateway/realtime.test.ts b/tests/gateway/realtime.test.ts index 87f433d..7506120 100644 --- a/tests/gateway/realtime.test.ts +++ b/tests/gateway/realtime.test.ts @@ -16,7 +16,7 @@ function fakeSocket() { send(s: string) { sent.push(JSON.parse(s)); }, - close() {}, + close: vi.fn(), }; return { ws, sent, emit: (e: string, a?: unknown) => handlers[e]?.(a), handlers }; } @@ -86,6 +86,25 @@ describe("realtime session configuration", () => { }); describe("realtime function-call lifecycle", () => { + function toolCall( + fake: ReturnType, + itemId: string, + callId: string, + name: string, + args = "{}", + ) { + emitMessage(fake, { + type: "response.output_item.added", + item_id: itemId, + item: { type: "function_call", call_id: callId, name }, + }); + emitMessage(fake, { + type: "response.function_call_arguments.done", + item_id: itemId, + arguments: args, + }); + } + it("captures completed caller and agent transcripts", () => { const fake = fakeSocket(); const onTranscript = vi.fn(); @@ -184,6 +203,98 @@ describe("realtime function-call lifecycle", () => { expect(onHangup).toHaveBeenCalledTimes(1); // second press within window ends the call }); + it("defers hangup until a pending contact result response is complete and flushed", async () => { + const fake = fakeSocket(); + const onHangup = vi.fn(); + let resolveContact!: (value: string) => void; + const onContactRead = vi.fn(() => new Promise((resolve) => (resolveContact = resolve))); + let clock = 0; + const bridge = openRealtimeBridge( + { apiKey: "k", model: "m", voice: "v", instructions: "hi" }, + createPostCallRegistry(), + { + onAudio: vi.fn(), + onAudioDone: vi.fn(), + onConsult: vi.fn(async () => ""), + onContactRead, + onHangup, + logger, + }, + () => clock, + () => fake.ws as never, + ); + fake.emit("open"); + emitMessage(fake, { type: "session.updated" }); + await bridge.ready; + + toolCall(fake, "contact-item", "contact-call", "inkbox_list_contacts", '{"q":"ada"}'); + toolCall(fake, "hangup-item-1", "hangup-call-1", "hang_up_call"); + clock = 1000; + toolCall(fake, "hangup-item-2", "hangup-call-2", "hang_up_call"); + expect(onHangup).not.toHaveBeenCalled(); + + // The first response belongs to the armed-goodbye result. The second is + // the contact result and is the one that must drain before hangup. + emitMessage(fake, { type: "response.created", response: { id: "goodbye-response" } }); + resolveContact("Ada's email is ada@example.com"); + await new Promise((resolve) => setTimeout(resolve, 0)); + emitMessage(fake, { type: "response.created", response: { id: "contact-response" } }); + emitMessage(fake, { + type: "response.output_audio_transcript.done", + response_id: "contact-response", + transcript: "Ada's email is ada@example.com", + }); + emitMessage(fake, { + type: "response.output_audio.done", + response_id: "contact-response", + }); + expect(onHangup).not.toHaveBeenCalled(); + emitMessage(fake, { + type: "response.done", + response: { id: "contact-response", status: "completed" }, + }); + expect(onHangup).toHaveBeenCalledTimes(1); + + // Duplicate completion events cannot end the call twice. + emitMessage(fake, { + type: "response.done", + response: { id: "contact-response", status: "completed" }, + }); + expect(onHangup).toHaveBeenCalledTimes(1); + await bridge.close(); + expect(fake.ws.close).toHaveBeenCalledTimes(1); + }); + + it("closes immediately and idempotently when the call ends during async work", async () => { + const fake = fakeSocket(); + let resolveConsult!: (value: string) => void; + const bridge = openRealtimeBridge( + { apiKey: "k", model: "m", voice: "v", instructions: "hi" }, + createPostCallRegistry(), + { + onAudio: vi.fn(), + onConsult: vi.fn(() => new Promise((resolve) => (resolveConsult = resolve))), + onHangup: vi.fn(), + logger, + }, + () => 0, + () => fake.ws as never, + ); + fake.emit("open"); + emitMessage(fake, { type: "session.updated" }); + await bridge.ready; + toolCall(fake, "consult-item", "consult-call", "consult_agent", '{"query":"wait"}'); + + await bridge.close(); + await bridge.close(); + expect(fake.ws.close).toHaveBeenCalledTimes(1); + + const sentBeforeLateResult = fake.sent.length; + resolveConsult("late result"); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(fake.sent).toHaveLength(sentBeforeLateResult); + }); + it("answers contact-read tools directly via onContactRead", async () => { const fake = fakeSocket(); const onContactRead = vi.fn(async (kind: string) => `cards for ${kind}`);