Make Realtime voice follow-ups reliable

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.

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<void>((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,74 @@ export function createCallBridge(deps: CallBridgeDeps) {
       }
     }
 
-    await new Promise<void>((resolve) => ws.once("close", () => resolve()));
+    let poll: ReturnType<typeof setInterval> | 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 === "local"
+                      ? "agent"
+                      : 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.

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 ` +

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<string>;
@@ -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<void> {
     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) : {};

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");
   });

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");