From ddcc8efdc6333315d1c78fa4596cbc750b33b86f Mon Sep 17 00:00:00 2001 From: alex-w-99 Date: Thu, 16 Jul 2026 04:39:27 +0000 Subject: [PATCH 1/2] Make Realtime follow-ups reliable --- CHANGELOG.md | 3 ++ src/gateway/voice/bridge.ts | 86 +++++++++++++++++++++++++++++- src/gateway/voice/instructions.ts | 2 + src/gateway/voice/realtime.ts | 14 +++++ tests/gateway/instructions.test.ts | 1 + tests/gateway/realtime.test.ts | 30 +++++++++++ 6 files changed, 135 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c858ee..5223f20 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,3 +27,6 @@ Initial release. - Local-file media on `inkbox_send_email` (`attachmentPaths`), `inkbox_send_sms`/`inkbox_send_imessage` (`mediaPaths`); `inkbox_place_call` carries a call purpose and opening message. +- Realtime voice follow-ups recover from completed calls even when the media + socket stays open, using the persisted call transcript to execute promised + post-call work. diff --git a/src/gateway/voice/bridge.ts b/src/gateway/voice/bridge.ts index 4d12a14..6eb2718 100644 --- a/src/gateway/voice/bridge.ts +++ b/src/gateway/voice/bridge.ts @@ -37,6 +37,10 @@ const GREETING = "Hi, you've reached the assistant. How can I help?"; // How long to wait for the OpenAI session (connect + session.update round // trip) before falling back to Inkbox speech. Cold TLS/DNS can eat seconds. const REALTIME_READY_TIMEOUT_MS = 10_000; +const CALL_STATUS_POLL_MS = 2_000; +const TRANSCRIPT_RETRY_MS = 500; +const TRANSCRIPT_ATTEMPTS = 5; +const ENDED_CALL_STATUSES = new Set(["completed", "failed", "canceled"]); // Owns the /phone/media/ws endpoint. The upgrade is authenticated with the // Inkbox webhook signature over the X-Call-Context header, and the caller @@ -205,6 +209,9 @@ export function createCallBridge(deps: CallBridgeDeps) { onAudio: (b64) => callWs && sendMedia(callWs, b64), onAudioDone: () => callWs && sendAudioDone(callWs), onBargeIn: () => callWs && sendClear(callWs), + onTranscript: (party, text) => { + ctx.transcript.push(`${party}: ${text}`); + }, onConsult: async (query) => { ctx.transcript.push(`caller: ${query}`); const answer = await deps.sessions.runText(ctx.chatKey, `${voiceTag(ctx)}\n${query}`); @@ -277,6 +284,21 @@ export function createCallBridge(deps: CallBridgeDeps) { realtime.start(buildVoiceGreeting(meta)); } + let finishCall: () => void = () => {}; + const callEnded = new Promise((resolve) => { + let finished = false; + finishCall = () => { + if (finished) return; + finished = true; + resolve(); + }; + ws.once("close", finishCall); + ws.once("error", (err) => { + deps.logger.warn("call.socket_error", { callId: ctx.callId, error: String(err) }); + finishCall(); + }); + }); + ws.on("message", (data) => { const frame = parseFrame(data); if (!frame) return; @@ -316,6 +338,7 @@ export function createCallBridge(deps: CallBridgeDeps) { return; } if (frame.event === "stop" || frame.event === "closed" || frame.event === "hangup") { + finishCall(); try { ws.close(); } catch { @@ -324,8 +347,69 @@ export function createCallBridge(deps: CallBridgeDeps) { } } - await new Promise((resolve) => ws.once("close", () => resolve())); + let poll: ReturnType | undefined; + if (ctx.callId) { + let checking = false; + const client = await deps.inkbox.getClient(); + poll = setInterval(() => { + if (checking) return; + checking = true; + void client.calls + .get(ctx.callId as string) + .then((call) => { + if (ENDED_CALL_STATUSES.has(String(call.status).toLowerCase())) finishCall(); + }) + .catch((err) => { + deps.logger.warn("call.status_check_failed", { + callId: ctx.callId, + error: String(err), + }); + }) + .finally(() => { + checking = false; + }); + }, CALL_STATUS_POLL_MS); + poll.unref?.(); + } + + await callEnded; + if (poll) clearInterval(poll); + try { + ws.close(); + } catch { + /* already closing */ + } await realtime?.close(); + + if (ctx.callId) { + const client = await deps.inkbox.getClient(); + for (let attempt = 1; attempt <= TRANSCRIPT_ATTEMPTS; attempt++) { + try { + const stored = await client.calls.transcripts(ctx.callId); + if (stored.length > 0) { + ctx.transcript.splice( + 0, + ctx.transcript.length, + ...stored.map((segment) => { + const party = segment.party === "remote" ? "caller" : segment.party; + return `${party}: ${segment.text}`; + }), + ); + break; + } + } catch (err) { + if (attempt === TRANSCRIPT_ATTEMPTS) { + deps.logger.warn("call.transcript_fetch_failed", { + callId: ctx.callId, + error: String(err), + }); + } + } + if (attempt < TRANSCRIPT_ATTEMPTS) { + await new Promise((resolve) => setTimeout(resolve, TRANSCRIPT_RETRY_MS)); + } + } + } deps.logger.info("call.ended", { chatKey: ctx.chatKey }); // Post-call: run queued actions, or a reflection turn, off the closed call. diff --git a/src/gateway/voice/instructions.ts b/src/gateway/voice/instructions.ts index e0961a6..5b36ca0 100644 --- a/src/gateway/voice/instructions.ts +++ b/src/gateway/voice/instructions.ts @@ -118,6 +118,8 @@ export function buildVoiceInstructions(meta: CallMeta): string { `If the caller explicitly asks for work to happen after the call, or accepts an ` + `after-call deferral, call ${REGISTER_ACTION_TOOL}. Tell the caller the action is ` + `queued for after the call; do not claim it has already been completed.`, + `Never say work is queued, in progress, or will run later unless ${CONSULT_TOOL} or ` + + `${REGISTER_ACTION_TOOL} returned success. There is no implicit background queue.`, `If the caller changes or cancels previously queued after-call work, call ` + `${EDIT_ACTION_TOOL} or ${DELETE_ACTION_TOOL} with the id returned when it was queued.`, `If ${CONSULT_TOOL} completes or queues work that matches a previously registered ` + diff --git a/src/gateway/voice/realtime.ts b/src/gateway/voice/realtime.ts index a94ebb9..4e86e5f 100644 --- a/src/gateway/voice/realtime.ts +++ b/src/gateway/voice/realtime.ts @@ -26,6 +26,8 @@ export interface RealtimeCallbacks { onAudioDone?(): void; // The caller started talking over the model — clear queued playback. onBargeIn?(): void; + // Completed Realtime transcripts used for post-call recovery. + onTranscript?(party: "caller" | "agent", text: string): void; // Run a full agent turn in the caller's session; the returned text is // spoken back. Runs off the audio pump so speech never freezes. onConsult(query: string): Promise; @@ -229,6 +231,17 @@ export function openRealtimeBridge( case "response.audio.done": cb.onAudioDone?.(); break; + case "conversation.item.input_audio_transcription.completed": { + const text = String(evt.transcript ?? "").trim(); + if (text) cb.onTranscript?.("caller", text); + break; + } + case "response.output_audio_transcript.done": + case "response.audio_transcript.done": { + const text = String(evt.transcript ?? "").trim(); + if (text) cb.onTranscript?.("agent", text); + break; + } case "response.output_item.added": { const item = evt.item ?? {}; if (item.type === "function_call") { @@ -295,6 +308,7 @@ export function openRealtimeBridge( }): Promise { const name = evt.name; const callId = evt.call_id; + cb.logger.info("realtime.function_call", { name }); let args: any = {}; try { args = evt.arguments ? JSON.parse(evt.arguments) : {}; diff --git a/tests/gateway/instructions.test.ts b/tests/gateway/instructions.test.ts index ba6432a..7d04b6f 100644 --- a/tests/gateway/instructions.test.ts +++ b/tests/gateway/instructions.test.ts @@ -73,6 +73,7 @@ describe("buildVoiceInstructions", () => { expect(out).toContain("consult_agent"); expect(out).toContain("Do not promise work outside that list"); expect(out).toContain("register_post_call_action"); + expect(out).toContain("There is no implicit background queue"); expect(out).toContain("hang_up_call"); expect(out).toContain("third parties"); }); diff --git a/tests/gateway/realtime.test.ts b/tests/gateway/realtime.test.ts index 09b9d32..87f433d 100644 --- a/tests/gateway/realtime.test.ts +++ b/tests/gateway/realtime.test.ts @@ -86,6 +86,36 @@ describe("realtime session configuration", () => { }); describe("realtime function-call lifecycle", () => { + it("captures completed caller and agent transcripts", () => { + const fake = fakeSocket(); + const onTranscript = vi.fn(); + openRealtimeBridge( + { apiKey: "k", model: "m", voice: "v", instructions: "hi" }, + createPostCallRegistry(), + { + onAudio: vi.fn(), + onTranscript, + onConsult: vi.fn(async () => ""), + onHangup: vi.fn(), + logger, + }, + () => 0, + () => fake.ws as never, + ); + + emitMessage(fake, { + type: "conversation.item.input_audio_transcription.completed", + transcript: "do this after the call", + }); + emitMessage(fake, { + type: "response.output_audio_transcript.done", + transcript: "I queued it", + }); + + expect(onTranscript).toHaveBeenNthCalledWith(1, "caller", "do this after the call"); + expect(onTranscript).toHaveBeenNthCalledWith(2, "agent", "I queued it"); + }); + it("accumulates name + call id + args across the three events before dispatching", async () => { const fake = fakeSocket(); const onConsult = vi.fn(async () => "the answer"); From 8a1018d9ccd221005facc5fd4b9eeb0f4b1b3e3e Mon Sep 17 00:00:00 2001 From: alex-w-99 Date: Thu, 16 Jul 2026 04:40:06 +0000 Subject: [PATCH 2/2] Normalize persisted call transcript roles --- src/gateway/voice/bridge.ts | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/gateway/voice/bridge.ts b/src/gateway/voice/bridge.ts index 6eb2718..6005ff6 100644 --- a/src/gateway/voice/bridge.ts +++ b/src/gateway/voice/bridge.ts @@ -391,7 +391,12 @@ export function createCallBridge(deps: CallBridgeDeps) { 0, ctx.transcript.length, ...stored.map((segment) => { - const party = segment.party === "remote" ? "caller" : segment.party; + const party = + segment.party === "remote" + ? "caller" + : segment.party === "local" + ? "agent" + : segment.party; return `${party}: ${segment.text}`; }), );