Fix live voice leg correlation

.github/workflows/live-voice.yml

@@ -5,8 +5,8 @@ name: Live — voice calls (Voice AI + Realtime + Inkbox TTS/STT)
 #   inbound_inkbox    — driver calls the agent; agent answers Inkbox STT/TTS.
 #   outbound_realtime — driver texts "call me"; agent calls back on Realtime.
 #   outbound_hosted   — driver requests a Voice AI call and one exact post-call SMS.
-# Each leg verifies the stored call transcript shows the agent spoke to the
-# caller. Real model + real calls — the priciest suite — so it runs only on
+# Each leg verifies driver-local speech and a two-way paired AUT transcript.
+# Real model + real calls — the priciest suite — so it runs only on
 # ready (non-draft) PRs + manual dispatch, and shares the AUT tunnel lock.
 on:
   workflow_call:
@@ -99,7 +99,9 @@ jobs:
           if [ "${{ matrix.scenario }}" = "outbound_hosted" ]; then
             HOSTED_MARKER="$(node scripts/nato-marker.mjs "$GITHUB_RUN_ID" "$GITHUB_RUN_ATTEMPT")"
             echo "HOSTED_POST_CALL_MARKER=$HOSTED_MARKER" >> "$GITHUB_ENV"
-            export VOICE_DRIVER_LINE="After we hang up, send me one SMS. Create one post-call action now with the title Send SMS and put this exact five-word SMS body in the action details: $HOSTED_MARKER. Wait for the action tool to succeed, then read all five words back to me. Do not paraphrase, omit a word, or send the SMS during the call."
+            export VOICE_DRIVER_LINE="After we hang up, send me one SMS containing exactly $HOSTED_MARKER. Before we hang up, record one post-call action whose action title and details are both exactly Send SMS $HOSTED_MARKER. Read the five-word SMS body back after recording it. Do not send the SMS during the call."
+            export VOICE_DRIVER_FOLLOWUP_LINE="Verify the post-call action you just recorded. If its action title or details are not exactly Send SMS $HOSTED_MARKER, edit that same action now so both match exactly. Then read back exactly $HOSTED_MARKER."
+            export VOICE_DRIVER_FOLLOWUP_AFTER=45
             export VOICE_DRIVER_LISTEN=180
             export VOICE_DRIVER_AUTO_STOP=false
           fi

scripts/nato-marker.mjs

@@ -1,44 +1,38 @@
 import { pathToFileURL } from "node:url";
 
-const RADIO_WORDS = [
-  "alpha",
-  "bravo",
-  "charlie",
-  "delta",
-  "echo",
-  "foxtrot",
-  "golf",
-  "hotel",
-  "india",
-  "juliet",
-  "kilo",
-  "lima",
-  "mike",
-  "november",
-  "oscar",
-  "papa",
-  "quebec",
-  "romeo",
-  "sierra",
-  "tango",
-  "uniform",
-  "victor",
-  "whiskey",
-  "xray",
-  "yankee",
-  "zulu",
+const SPEECH_WORDS = [
+  "banana",
+  "elephant",
+  "pineapple",
+  "alligator",
+  "motorcycle",
+  "umbrella",
+  "dinosaur",
+  "potato",
+  "computer",
+  "volcano",
+  "airplane",
+  "butterfly",
+  "kangaroo",
+  "octopus",
+  "calendar",
+  "chocolate",
+  "hospital",
+  "library",
+  "sandwich",
+  "telescope",
 ];
 
 export function natoMarker(runId, runAttempt) {
   let value = BigInt(runId) * 10n + BigInt(runAttempt);
   const used = new Set();
   const marker = [];
   for (let count = 0; count < 5; count += 1) {
-    let index = Number(value % BigInt(RADIO_WORDS.length));
-    value /= BigInt(RADIO_WORDS.length);
-    while (used.has(index)) index = (index + 1) % RADIO_WORDS.length;
+    let index = Number(value % BigInt(SPEECH_WORDS.length));
+    value /= BigInt(SPEECH_WORDS.length);
+    while (used.has(index)) index = (index + 1) % SPEECH_WORDS.length;
     used.add(index);
-    marker.push(RADIO_WORDS[index]);
+    marker.push(SPEECH_WORDS[index]);
   }
   return marker.join(" ");
 }

src/a2a-context.ts

@@ -74,6 +74,12 @@ export function activeA2ATurn(sessionID: string): ActiveA2ATurn | undefined {
   return turns.get(sessionID) ?? readTurn(sessionID);
 }
 
+export function a2aReplyIntentCommitted(sessionID: string, turn: ActiveA2ATurn): boolean {
+  if (turn.replyIntentCommitted) return true;
+  const persisted = readTurn(sessionID);
+  return Boolean(persisted && sameTurn(persisted, turn) && persisted.replyIntentCommitted);
+}
+
 export function commitActiveA2ATurn(sessionID: string, turn: ActiveA2ATurn): void {
   turn.replyIntentCommitted = true;
   writeTurn(sessionID, turn);

src/client.ts

@@ -7,6 +7,15 @@ import {
 } from "@inkbox/sdk";
 
 const USER_AGENT_NAME = "inkbox-opencode";
+const IDENTITY_RETRY_DELAYS_MS = [250, 750] as const;
+const TRANSIENT_NETWORK_CODES = new Set([
+  "ECONNRESET",
+  "ECONNREFUSED",
+  "ETIMEDOUT",
+  "EAI_AGAIN",
+  "ENETDOWN",
+  "ENETUNREACH",
+]);
 
 // Read at runtime rather than hardcoded so the token can't drift from the
 // package version. `tsc` emits to dist/, so ../package.json resolves from
@@ -68,6 +77,50 @@ function runtimeCacheKey(cfg: InkboxCredentials): string {
   });
 }
 
+function isTransientIdentityError(error: unknown): boolean {
+  let current: unknown = error;
+  for (let depth = 0; depth < 4 && current; depth += 1) {
+    const value = current as {
+      code?: unknown;
+      status?: unknown;
+      statusCode?: unknown;
+      cause?: unknown;
+      message?: unknown;
+    };
+    const code = typeof value.code === "string" ? value.code.toUpperCase() : "";
+    if (TRANSIENT_NETWORK_CODES.has(code)) return true;
+
+    const status = Number(value.status ?? value.statusCode);
+    if (status === 408 || status === 425 || status === 429 || status >= 500) return true;
+
+    const message = typeof value.message === "string" ? value.message : String(current);
+    if (
+      /\b(?:ECONNRESET|ECONNREFUSED|ETIMEDOUT|EAI_AGAIN|ENETDOWN|ENETUNREACH)\b/i.test(message) ||
+      /(?:socket hang up|network connection was lost|fetch failed)/i.test(message)
+    ) {
+      return true;
+    }
+    current = value.cause;
+  }
+  return false;
+}
+
+export async function resolveIdentityWithRetry<T>(
+  resolveIdentity: () => Promise<T>,
+  wait: (delayMs: number) => Promise<void> = (delayMs) =>
+    new Promise((resolve) => setTimeout(resolve, delayMs)),
+): Promise<T> {
+  for (let attempt = 0; ; attempt += 1) {
+    try {
+      return await resolveIdentity();
+    } catch (error) {
+      const delayMs = IDENTITY_RETRY_DELAYS_MS[attempt];
+      if (delayMs === undefined || !isTransientIdentityError(error)) throw error;
+      await wait(delayMs);
+    }
+  }
+}
+
 // Build a lazy-cached runtime. The Inkbox SDK client and the identity
 // resolution happen on first tool call, not at plugin load. This keeps
 // startup cheap when a session never invokes an Inkbox tool.
@@ -118,7 +171,7 @@ export function createInkboxRuntime(source: ConfigSource, logger?: PluginLogger)
             `Inkbox plugin: whoami() failed during init: ${e instanceof Error ? e.message : String(e)}`,
           );
         }
-        const identity = await inkbox.getIdentity(identityHandle);
+        const identity = await resolveIdentityWithRetry(() => inkbox.getIdentity(identityHandle));
         return { inkbox, identity };
       })();
       const entry = {

src/gateway/sessions.ts

@@ -1,6 +1,11 @@
 import { randomBytes, randomUUID } from "node:crypto";
 import type { OpencodeClient } from "@opencode-ai/sdk";
-import { type ActiveA2ATurn, clearActiveA2ATurn, setActiveA2ATurn } from "../a2a-context.js";
+import {
+  type ActiveA2ATurn,
+  a2aReplyIntentCommitted,
+  clearActiveA2ATurn,
+  setActiveA2ATurn,
+} from "../a2a-context.js";
 import type { InkboxRuntime } from "../client.js";
 import type { ResolvedConfig } from "../config.js";
 import {
@@ -206,6 +211,15 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
     );
   }
 
+  async function stopToolCompletedSession(sessionID: string): Promise<void> {
+    const aborted = await deps.opencode.session.abort({
+      path: { id: sessionID },
+      query: { directory: deps.directory },
+    });
+    const error = (aborted as any)?.error;
+    if (error) throw new Error(`session.abort failed: ${JSON.stringify(error).slice(0, 300)}`);
+  }
+
   async function submit(turn: DurableTurn): Promise<DurableTurn> {
     const sessionID = turn.sessionID ?? (await ensureSession(turn.chatKey));
     const next = deps.state.transitionTurn(turn.id, ["queued"], {
@@ -262,6 +276,27 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
         throw new Error("Durable turn lease was lost.");
       }
       if (deps.state.getTurn(turn.id)?.state === "interrupted") return undefined;
+      if (turn.hostedCapture) {
+        const entry = getHostedCall(turn.hostedCapture.identityId, turn.hostedCapture.callId);
+        const attempt = entry?.smsAttempts.find(
+          (candidate) => candidate.phase === turn.hostedCapture?.phase,
+        );
+        if (attempt && attempt.state !== "pending") {
+          await stopToolCompletedSession(turn.sessionID);
+          deps.logger.info("hosted_call.turn_stopped_after_sms_attempt", {
+            callId: turn.hostedCapture.callId,
+            state: attempt.state,
+          });
+          return undefined;
+        }
+      }
+      if (turn.a2aContext && a2aReplyIntentCommitted(turn.sessionID, turn.a2aContext)) {
+        await stopToolCompletedSession(turn.sessionID);
+        deps.logger.info("a2a.turn_stopped_after_reply", {
+          taskId: turn.a2aContext.taskId,
+        });
+        return undefined;
+      }
       const messages = await listMessages(turn.sessionID);
       const userIndex = messages.findIndex((message) => message?.info?.id === turn.messageID);
       if (userIndex < 0) {

tests/contract/live-harness.test.ts

@@ -4,6 +4,7 @@ import { describe, expect, it } from "vitest";
 const liveAut = readFileSync("scripts/live-aut.sh", "utf8");
 const liveChannels = readFileSync(".github/workflows/live-channels.yml", "utf8");
 const liveVoice = readFileSync(".github/workflows/live-voice.yml", "utf8");
+const voiceDriver = readFileSync("tests/live/voice-driver.mjs", "utf8");
 
 function shellCommands(source: string): string[] {
   return source.replace(/\\\n\s*/g, " ").split("\n");
@@ -18,6 +19,10 @@ function yamlJob(source: string, name: string): string {
 }
 
 describe("live harness readiness bounds", () => {
+  it("allows the realtime model to answer before the media peer hangs up", () => {
+    expect(voiceDriver).toContain('VOICE_DRIVER_LISTEN || "30"');
+  });
+
   it("bounds both opencode /config readiness probes", () => {
     const configProbes = shellCommands(liveAut).filter(
       (command) => /\bcurl\b/.test(command) && command.includes("/config"),
@@ -56,10 +61,15 @@ describe("live harness readiness bounds", () => {
     expect(voiceJob).toMatch(/^ {4}timeout-minutes: 15$/m);
   });
 
-  it("requires the hosted caller to persist and read back the exact SMS body", () => {
+  it("requires the hosted caller to persist, correct, and read back the exact SMS body", () => {
+    expect(liveVoice).toContain(
+      'export VOICE_DRIVER_LINE="After we hang up, send me one SMS containing exactly $HOSTED_MARKER. Before we hang up, record one post-call action whose action title and details are both exactly Send SMS $HOSTED_MARKER. Read the five-word SMS body back after recording it. Do not send the SMS during the call."',
+    );
     expect(liveVoice).toContain(
-      'export VOICE_DRIVER_LINE="After we hang up, send me one SMS. Create one post-call action now with the title Send SMS and put this exact five-word SMS body in the action details: $HOSTED_MARKER. Wait for the action tool to succeed, then read all five words back to me. Do not paraphrase, omit a word, or send the SMS during the call."',
+      'export VOICE_DRIVER_FOLLOWUP_LINE="Verify the post-call action you just recorded. If its action title or details are not exactly Send SMS $HOSTED_MARKER, edit that same action now so both match exactly. Then read back exactly $HOSTED_MARKER."',
     );
+    expect(liveVoice).not.toContain("list the actions");
+    expect(liveVoice).toContain("export VOICE_DRIVER_FOLLOWUP_AFTER=45");
   });
 
   it("preserves diagnostics when the voice job is cancelled by its timeout", () => {

tests/gateway/sessions.test.ts

@@ -2,9 +2,15 @@ import * as fs from "node:fs";
 import * as os from "node:os";
 import * as path from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
+import { a2aTurnContextPath } from "../../src/a2a-context.js";
 import type { ResolvedConfig } from "../../src/config.js";
 import { defaultGatewayConfig } from "../../src/config.js";
-import { getHostedCall, saveHostedCall } from "../../src/gateway/hosted-call-registry.js";
+import {
+  beginHostedSmsAttempt,
+  getHostedCall,
+  saveHostedCall,
+  settleHostedSmsAttempt,
+} from "../../src/gateway/hosted-call-registry.js";
 import { createSessionManager, extractText } from "../../src/gateway/sessions.js";
 import { createStateStore, type DurableTurn } from "../../src/gateway/state.js";
 import type { InboundMessage } from "../../src/gateway/types.js";
@@ -416,6 +422,36 @@ describe("capture turns", () => {
     expect(d.opencode.session.promptAsync).toHaveBeenCalledTimes(submitted);
   });
 
+  it("releases a hosted session after its SMS attempt is durably settled", async () => {
+    const d = makeManager();
+    prepareHostedCall(d.dir);
+    d.setAutoComplete(false);
+    const capture = {
+      identityId: "ident-1",
+      callId: "call-1",
+      phase: "initial" as const,
+      expectedTarget: "+14155550123",
+    };
+
+    const pending = d.mgr.runHostedCapture?.("ck", "call", capture);
+    await vi.waitFor(() =>
+      expect(getHostedCall("ident-1", "call-1")?.active?.sessionID).toBe("sess-1"),
+    );
+    const guard = beginHostedSmsAttempt({
+      sessionID: "sess-1",
+      target: "+14155550123",
+      hasConversationId: false,
+    });
+    if (!guard) throw new Error("Expected the hosted SMS attempt to start.");
+    settleHostedSmsAttempt(guard, "success");
+
+    await expect(pending).resolves.toMatchObject({ attempt: { state: "success" } });
+    expect(d.opencode.session.abort).toHaveBeenCalledWith({
+      path: { id: "sess-1" },
+      query: { directory: "/proj" },
+    });
+  });
+
   it("restores the hosted SMS guard before monitoring a submitted turn", async () => {
     const d = makeManager();
     prepareHostedCall(d.dir);
@@ -467,6 +503,32 @@ describe("capture turns", () => {
 
     expect(d.opencode.session.promptAsync).toHaveBeenCalledTimes(submitted);
   });
+
+  it("releases an A2A session after a tool commits the remote reply", async () => {
+    const d = makeManager();
+    process.env.INKBOX_OPENCODE_HOME = d.dir;
+    d.setAutoComplete(false);
+    const context = {
+      taskId: "task-1",
+      messageId: "message-1",
+      contextId: "context-1",
+      replyIntentCommitted: false,
+    };
+
+    const pending = d.mgr.runA2A("a2a:context-1", "task", context);
+    await vi.waitFor(() => expect(d.opencode.session.promptAsync).toHaveBeenCalledOnce());
+    fs.writeFileSync(
+      a2aTurnContextPath("sess-1"),
+      `${JSON.stringify({ ...context, replyIntentCommitted: true })}\n`,
+      { mode: 0o600 },
+    );
+
+    await expect(pending).resolves.toBeUndefined();
+    expect(d.opencode.session.abort).toHaveBeenCalledWith({
+      path: { id: "sess-1" },
+      query: { directory: "/proj" },
+    });
+  });
 });
 
 describe("control", () => {