fix(app): end voice delegations on their own instead of stalling the call (#61608)

packages/app/src/voice/call.ts

@@ -6,7 +6,16 @@ import { TranscriptTail } from "./tail"
 import { VoiceTools } from "./tools"
 import { VoiceFailure, VoiceTransport } from "./transport"
 import { delegatorInstructions } from "./spec"
-import type { ToolCall, VoiceEvent } from "./protocol"
+import type { VoiceEvent } from "./protocol"
+import {
+  createVoiceDelegationState,
+  delegating,
+  hasDelegation,
+  responseEvent,
+  transitionVoiceDelegation,
+  type VoiceDelegationEffect,
+  type VoiceDelegationEvent,
+} from "./machine"
 
 export type CallState = {
   status: "idle" | "connecting" | "connected" | "dropped"
@@ -19,13 +28,14 @@ export type CallState = {
   title: string
 }
 
-type Delegation = {
-  id: string
-  sessionID: string
-  transcript: string[]
-  calls: ToolCall[]
-  rounds: number
-  responseID?: string
+type CallRuntime = {
+  transport?: VoiceTransport
+  viewing: SessionInfo
+  muted: boolean
+  ended: boolean
+  watching: boolean
+  ready: boolean
+  joining: boolean
 }
 
 export function createVoiceCall(input: {
@@ -51,15 +61,7 @@ export function createVoiceCall(input: {
     tabs: input.tabs,
   })
 
-  const state: {
-    transport?: VoiceTransport
-    viewing: SessionInfo
-    muted: boolean
-    ended: boolean
-    watching: boolean
-    ready: boolean
-    joining: boolean
-  } = {
+  const state: CallRuntime = {
     viewing: input.session,
     muted: false,
     ended: false,
@@ -68,8 +70,9 @@ export function createVoiceCall(input: {
     joining: false,
   }
 
-  const delegations = new Map<string, Delegation>()
-  const completed = new Set<string>()
+  let delegations = createVoiceDelegationState()
+  // Timers and tool aborts of open delegations; the machine decides when each ends.
+  const handles = new Map<string, { timer?: ReturnType<typeof setTimeout>; abort?: AbortController }>()
   const dirty = new Set<string>()
   const unlisten = input.ctx.sdk.event.on("session.idle", (event) => dirty.add(event.data.sessionID))
 
@@ -98,8 +101,7 @@ export function createVoiceCall(input: {
   function drop(failure: CallState["failure"] = "dropped") {
     if (state.ended) return
     state.ready = false
-    delegations.forEach((delegation) => tail.release(delegation.id))
-    delegations.clear()
+    dispatch({ type: "call.reset" })
     input.change({ status: "dropped", failure, level: 0 })
     void state.transport?.close()
     void input.platform.setVoiceCallActive?.(false).catch(() => undefined)
@@ -110,7 +112,7 @@ export function createVoiceCall(input: {
     state.joining = true
     input.change({ status: "connecting", failure: undefined })
     await connect()
-      .catch((error: unknown) => {
+      .catch((error) => {
         if (!state.ended) drop(error instanceof VoiceFailure ? error.reason : "connection")
       })
       .finally(() => {
@@ -128,7 +130,7 @@ export function createVoiceCall(input: {
     const transport = new VoiceTransport({
       api: input.ctx.sdk.api,
       event: (event) => {
-        if (state.transport === transport && !transport.abort.signal.aborted) eventReceived(event, transport)
+        if (state.transport === transport && !transport.abort.signal.aborted) eventReceived(event)
       },
       levels: (mic, speaker) => {
         gates.user(!state.muted && mic > 0.015)
@@ -157,7 +159,7 @@ export function createVoiceCall(input: {
     input.change({ status: "connected", muted: state.muted })
   }
 
-  function eventReceived(event: VoiceEvent, transport: VoiceTransport) {
+  function eventReceived(event: VoiceEvent) {
     if (event.type === "session.input_transcript.delta" || event.type === "session.output_transcript.delta") {
       const speaker = event.type === "session.input_transcript.delta" ? "user" : "assistant"
       const caption = tail.append(speaker, event.delta ?? "")
@@ -173,83 +175,119 @@ export function createVoiceCall(input: {
     if (event.type === "session.delegation.created" && event.delegation) {
       const id = event.delegation.id
 
-      if (delegations.has(id)) return
-      delegations.set(id, { id, sessionID: state.viewing.id, transcript: tail.capture(id), calls: [], rounds: 0 })
+      // A repeated event must not capture the transcript again: that would clear this delegation's reservation.
+      if (hasDelegation(delegations, id)) return
+      dispatch({ type: "delegation.created", id, sessionID: state.viewing.id, transcript: tail.capture(id) })
 
       return
     }
 
-    if (event.type !== "response.event" || !event.event) return
-    const nested = event.event
-    // Like iOS, only infer the owner when there is exactly one request.
-    const id = event.delegation_id ?? (delegations.size === 1 ? delegations.keys().next().value : undefined)
-    const delegation = id ? delegations.get(id) : undefined
+    const next = responseEvent(event)
 
-    if (!delegation) return
-
-    if (nested.type === "response.created") delegation.responseID = nested.response?.id
-
-    if (nested.type === "response.output_item.done" && nested.item?.type === "function_call") {
-      const item = nested.item
-
-      if (item.call_id && item.name && !delegation.calls.some((call) => call.call_id === item.call_id)) {
-        delegation.calls.push({ call_id: item.call_id, name: item.name, arguments: item.arguments ?? "{}" })
-      }
-
-      return
-    }
-
-    if (!["response.completed", "response.failed", "response.incomplete", "response.cancelled"].includes(nested.type))
-      return
-    const key = nested.response?.id ?? delegation.responseID
-
-    if (!key || completed.has(key)) return
-    completed.add(key)
-
-    if (nested.type !== "response.completed") {
-      drop("connection")
-
-      return
-    }
-
-    if (!delegation.calls.length) {
-      // Responses final text goes to Live directly; never append it again.
-      tools.answered(delegation.id)
-      tail.accept(delegation.id)
-      delegations.delete(delegation.id)
-      gates.answer()
-
-      return
-    }
-
-    const calls = delegation.calls.splice(0)
-    void runRound(delegation, calls, transport).catch(() => {
-      if (state.transport === transport && !transport.abort.signal.aborted) drop("connection")
-    })
+    if (next) dispatch(next)
   }
 
-  async function runRound(delegation: Delegation, calls: ToolCall[], transport: VoiceTransport) {
-    if (++delegation.rounds > 50) throw new VoiceFailure("connection")
-    // Serial execution preserves dependencies between tools in one round.
-    const outputs: { call_id: string; output: string }[] = []
+  function dispatch(event: VoiceDelegationEvent) {
+    const next = transitionVoiceDelegation(delegations, event)
+    delegations = next.state
+    next.effects.forEach(perform)
+  }
 
-    for (const call of calls) {
-      transport.abort.signal.throwIfAborted()
-      outputs.push({ call_id: call.call_id, output: await tools.run(call, delegation, transport.abort.signal) })
+  function perform(effect: VoiceDelegationEffect) {
+    if (effect.type === "deadline") {
+      const handle = handles.get(effect.id) ?? {}
+      clearTimeout(handle.timer)
+      handle.timer = setTimeout(
+        () =>
+          dispatch({
+            type: "delegation.timeout",
+            id: effect.id,
+            attempt: effect.attempt,
+            // The transport aborts as soon as its data channel closes.
+            channelOpen: !!state.transport && !state.transport.abort.signal.aborted,
+          }),
+        effect.ms,
+      )
+      handles.set(effect.id, handle)
+
+      return
     }
 
-    transport.abort.signal.throwIfAborted()
-    outputs.forEach((output) =>
-      transport.send({ type: "response.item.create", item: { type: "function_call_output", ...output } }),
-    )
-    transport.send({ type: "response.create" })
-    tools.replies.sessions.forEach((id) => dirty.add(id))
+    if (effect.type === "round") {
+      const transport = state.transport
+
+      if (!transport) return
+      const abort = new AbortController()
+      handles.set(effect.id, { ...handles.get(effect.id), abort })
+      void runRound(effect, AbortSignal.any([abort.signal, transport.abort.signal]))
+
+      return
+    }
+
+    if (effect.type === "abort") {
+      handles.get(effect.id)?.abort?.abort()
+
+      return
+    }
+
+    if (effect.type === "outputs") {
+      const transport = state.transport
+
+      if (!transport) return
+      void Promise.resolve()
+        .then(() => {
+          effect.outputs.forEach((output) =>
+            transport.send({ type: "response.item.create", item: { type: "function_call_output", ...output } }),
+          )
+          transport.send({ type: "response.create" })
+          tools.replies.sessions.forEach((id) => dirty.add(id))
+        })
+        .catch(() => {
+          if (state.transport === transport && !transport.abort.signal.aborted) drop("connection")
+        })
+
+      return
+    }
+
+    const handle = handles.get(effect.id)
+    clearTimeout(handle?.timer)
+    handle?.abort?.abort()
+    handles.delete(effect.id)
+
+    if (effect.outcome === "discarded") {
+      tools.discard(effect.id)
+      tail.release(effect.id)
+
+      return
+    }
+
+    // Responses final text goes to Live directly; never append it again.
+    tools.answered(effect.id)
+    tail.accept(effect.id)
+    gates.answer()
+  }
+
+  async function runRound(effect: Extract<VoiceDelegationEffect, { type: "round" }>, signal: AbortSignal) {
+    // open_session moves the rest of this request onto the session it opened.
+    const request = { ...effect.request }
+
+    // Serial execution preserves dependencies between tools in one round.
+    for (const call of effect.calls) {
+      const output = await tools.run(call, request, signal)
+      dispatch({
+        type: "tool.completed",
+        id: effect.id,
+        attempt: effect.attempt,
+        sessionID: request.sessionID,
+        output: { call_id: call.call_id, output },
+      })
+    }
   }
 
   async function watch() {
     const transport = state.transport
 
-    if (!state.ready || !transport || state.watching || delegations.size || !dirty.size) return
+    if (!state.ready || !transport || state.watching || delegating(delegations) || !dirty.size) return
     state.watching = true
     await Promise.resolve()
       .then(async () => {
@@ -266,7 +304,12 @@ export function createVoiceCall(input: {
           if (!replies.length) continue
 
           for (const reply of replies) {
-            if (delegations.size || gates.delay() || transport !== state.transport || transport.abort.signal.aborted) {
+            if (
+              delegating(delegations) ||
+              gates.delay() ||
+              transport !== state.transport ||
+              transport.abort.signal.aborted
+            ) {
               dirty.add(sessionID)
 
               return
@@ -308,6 +351,7 @@ export function createVoiceCall(input: {
       state.ready = false
       unlisten()
       clearInterval(timer)
+      dispatch({ type: "call.reset" })
       await Promise.all([state.transport?.close(), input.platform.setVoiceCallActive?.(false).catch(() => undefined)])
     },
   }

packages/app/src/voice/machine.test.ts

@@ -0,0 +1,264 @@
+import { expect, test } from "bun:test"
+import { decodeEvent } from "./protocol"
+import {
+  createVoiceDelegationState,
+  delegating,
+  MAX_ROUNDS,
+  responseEvent,
+  RESPONSE_TIMEOUT,
+  TIMED_OUT,
+  transitionVoiceDelegation,
+  type ResponseOutcome,
+  type VoiceDelegationEffect,
+  type VoiceDelegationEvent,
+  type VoiceDelegationState,
+} from "./machine"
+
+function replay(events: VoiceDelegationEvent[], start = createVoiceDelegationState()) {
+  return events.reduce(
+    (result: { state: VoiceDelegationState; effects: VoiceDelegationEffect[] }, event) => {
+      const next = transitionVoiceDelegation(result.state, event)
+
+      return { state: next.state, effects: [...result.effects, ...next.effects] }
+    },
+    { state: start, effects: [] },
+  )
+}
+
+type LiveResponseEvent = {
+  type: string
+  response_id?: string
+  response?: { id: string }
+  item?: { type: string; call_id: string; name: string; arguments: string }
+}
+
+// Response events go through the real decoder, so they carry only the fields Live sends.
+function live(delegationID: string | null, nested: LiveResponseEvent) {
+  const event = decodeEvent(JSON.stringify({ type: "response.event", delegation_id: delegationID, event: nested }))
+  const next = event && responseEvent(event)
+
+  if (!next) throw new Error("not a delegation event")
+
+  return next
+}
+
+const opened = (id: string): VoiceDelegationEvent => ({
+  type: "delegation.created",
+  id,
+  sessionID: "ses_here",
+  transcript: [`user: ${id}`],
+})
+
+const started = (responseID: string, delegationID: string | null = null) =>
+  live(delegationID, { type: "response.created", response: { id: responseID } })
+
+// Function call items name their response only through `response_id`, when Live sends it.
+const called = (callID: string, responseID?: string, delegationID: string | null = null) =>
+  live(delegationID, {
+    type: "response.output_item.done",
+    response_id: responseID,
+    item: { type: "function_call", call_id: callID, name: "send_message", arguments: "{}" },
+  })
+
+const ended = (responseID: string, outcome: ResponseOutcome, delegationID: string | null = null) =>
+  live(delegationID, { type: `response.${outcome}`, response: { id: responseID } })
+
+const attempt = (state: VoiceDelegationState, id: string) =>
+  state.delegations.find((delegation) => delegation.id === id)?.phase.attempt ?? -1
+
+const ran = (state: VoiceDelegationState, id: string, callID: string, output = "{}"): VoiceDelegationEvent => ({
+  type: "tool.completed",
+  id,
+  attempt: attempt(state, id),
+  sessionID: "ses_here",
+  output: { call_id: callID, output },
+})
+
+const timeout = (state: VoiceDelegationState, id: string, channelOpen: boolean): VoiceDelegationEvent => ({
+  type: "delegation.timeout",
+  id,
+  attempt: attempt(state, id),
+  channelOpen,
+})
+
+const finishes = (effects: VoiceDelegationEffect[]) => effects.filter((effect) => effect.type === "finish")
+
+const rounds = (effects: VoiceDelegationEffect[]) =>
+  effects.flatMap((effect) => (effect.type === "round" ? [[effect.id, effect.calls.map((call) => call.call_id)]] : []))
+
+test("an untagged response reaches the open delegation while another runs tools", () => {
+  // del_1 runs send_message when del_2 starts; Live then omits delegation_id on del_2's events.
+  const running = replay([
+    opened("del_1"),
+    started("resp_1", "del_1"),
+    called("call_1", "resp_1", "del_1"),
+    ended("resp_1", "completed", "del_1"),
+    opened("del_2"),
+  ])
+
+  expect(rounds(running.effects)).toEqual([["del_1", ["call_1"]]])
+
+  const second = replay([started("resp_2"), ended("resp_2", "completed")], running.state)
+
+  expect(finishes(second.effects)).toEqual([{ type: "finish", id: "del_2", outcome: "answered" }])
+
+  const first = replay(
+    [ran(second.state, "del_1", "call_1"), started("resp_3"), ended("resp_3", "completed")],
+    second.state,
+  )
+
+  expect(first.effects).toContainEqual({ type: "outputs", outputs: [{ call_id: "call_1", output: "{}" }] })
+  expect(finishes(first.effects)).toEqual([{ type: "finish", id: "del_1", outcome: "answered" }])
+  // Session replies wait on this; before, del_2 never finished and replies were never announced.
+  expect(delegating(first.state)).toBe(false)
+})
+
+test("an untagged function call follows its response ID to its own delegation", () => {
+  const result = replay([
+    opened("del_1"),
+    opened("del_2"),
+    started("resp_1"),
+    started("resp_2"),
+    called("call_2", "resp_2"),
+    ended("resp_2", "completed"),
+  ])
+
+  // Attached to del_1, the call would leave del_2 answered while Live waits for call_2's output.
+  expect(rounds(result.effects)).toEqual([["del_2", ["call_2"]]])
+  expect(finishes(result.effects)).toEqual([])
+})
+
+test.each(["cancelled", "failed", "incomplete"] as const)("response.%s ends only its own delegation", (outcome) => {
+  const result = replay([
+    opened("del_1"),
+    started("resp_1"),
+    opened("del_2"),
+    ended("resp_1", outcome),
+    started("resp_2"),
+    ended("resp_2", "completed"),
+  ])
+
+  expect(finishes(result.effects)).toEqual([
+    { type: "finish", id: "del_1", outcome: "discarded" },
+    { type: "finish", id: "del_2", outcome: "answered" },
+  ])
+  expect(delegating(result.state)).toBe(false)
+})
+
+test("a timeout with the data channel closed discards only its delegation and ignores earlier attempts", () => {
+  const start = replay([opened("del_1"), opened("del_2")])
+
+  expect(start.effects).toContainEqual({
+    type: "deadline",
+    id: "del_1",
+    attempt: attempt(start.state, "del_1"),
+    ms: RESPONSE_TIMEOUT,
+  })
+
+  const running = replay(
+    [called("call_1"), ended("resp_1", "completed", "del_1"), timeout(start.state, "del_1", true)],
+    start.state,
+  )
+
+  expect(finishes(running.effects)).toEqual([])
+
+  const timedOut = replay(
+    [timeout(running.state, "del_1", false), ran(running.state, "del_1", "call_1")],
+    running.state,
+  )
+
+  expect(timedOut.effects).toEqual([{ type: "finish", id: "del_1", outcome: "discarded" }])
+  expect(timedOut.state.delegations.map((delegation) => delegation.id)).toEqual(["del_2"])
+
+  // A response that never finishes has no outputs to send, so it ends even with the channel open.
+  const silent = replay([timeout(timedOut.state, "del_2", true)], timedOut.state)
+
+  expect(silent.effects).toEqual([{ type: "finish", id: "del_2", outcome: "discarded" }])
+})
+
+test("a delegation that ends early leaves its response unable to finish a newer one", () => {
+  const bound = replay([opened("del_1"), started("resp_1")])
+  const late = [called("call_1", "resp_1"), ended("resp_1", "completed")]
+  const timedOut = replay([timeout(bound.state, "del_1", true), opened("del_2")], bound.state)
+  const reset = replay([{ type: "call.reset" }, opened("del_2")], bound.state)
+
+  expect(replay(late, timedOut.state)).toEqual({ state: timedOut.state, effects: [] })
+  expect(replay(late, reset.state)).toEqual({ state: reset.state, effects: [] })
+})
+
+test("a tool round that times out keeps finished outputs and reports only unfinished calls", () => {
+  const running = replay([opened("del_1"), called("call_1"), called("call_2"), ended("resp_1", "completed", "del_1")])
+  const admitted = JSON.stringify({ admitted: true })
+  const progressed = replay([ran(running.state, "del_1", "call_1", admitted)], running.state)
+
+  expect(progressed.effects).toEqual([])
+
+  const timedOut = transitionVoiceDelegation(progressed.state, timeout(progressed.state, "del_1", true))
+
+  expect(timedOut.effects).toEqual([
+    { type: "abort", id: "del_1" },
+    {
+      type: "outputs",
+      outputs: [
+        { call_id: "call_1", output: admitted },
+        { call_id: "call_2", output: TIMED_OUT },
+      ],
+    },
+    { type: "deadline", id: "del_1", attempt: attempt(timedOut.state, "del_1"), ms: RESPONSE_TIMEOUT },
+  ])
+
+  const answered = replay(
+    [ran(progressed.state, "del_1", "call_2"), started("resp_2"), ended("resp_2", "completed")],
+    timedOut.state,
+  )
+
+  expect(answered.effects).toEqual([{ type: "finish", id: "del_1", outcome: "answered" }])
+})
+
+test("a runaway delegation is discarded after the last allowed round", () => {
+  const result = Array.from({ length: MAX_ROUNDS + 1 }, (_, index) => index).reduce(
+    (current, index) => {
+      const finished = replay([called(`call_${index}`), ended(`resp_${index}`, "completed", "del_1")], current.state)
+
+      const done = finished.state.delegations.length
+        ? replay([ran(finished.state, "del_1", `call_${index}`)], finished.state)
+        : { state: finished.state, effects: [] }
+
+      return { state: done.state, effects: [...current.effects, ...finished.effects, ...done.effects] }
+    },
+    replay([opened("del_1")]),
+  )
+
+  expect(rounds(result.effects)).toHaveLength(MAX_ROUNDS)
+  expect(finishes(result.effects)).toEqual([{ type: "finish", id: "del_1", outcome: "discarded" }])
+})
+
+test("replayed function calls and terminal events are ignored", () => {
+  const start = replay([opened("del_1"), opened("del_2")])
+
+  const result = replay(
+    [
+      started("resp_1", "del_1"),
+      called("call_1", undefined, "del_1"),
+      ended("resp_1", "completed", "del_1"),
+      // Untagged repeats would otherwise run call_1 again or finish del_2 with del_1's response.
+      called("call_1", "resp_2"),
+      ended("resp_1", "completed"),
+    ],
+    start.state,
+  )
+
+  expect(rounds(result.effects)).toHaveLength(1)
+  expect(finishes(result.effects)).toEqual([])
+  expect(result.state.delegations[1]).toEqual(start.state.delegations[1])
+})
+
+test("a dropped call discards every open delegation", () => {
+  const result = replay([opened("del_1"), opened("del_2"), { type: "call.reset" }])
+
+  expect(finishes(result.effects)).toEqual([
+    { type: "finish", id: "del_1", outcome: "discarded" },
+    { type: "finish", id: "del_2", outcome: "discarded" },
+  ])
+  expect(delegating(result.state)).toBe(false)
+})

packages/app/src/voice/machine.ts

@@ -0,0 +1,324 @@
+import type { ToolCall, VoiceEvent } from "./protocol"
+
+export const MAX_ROUNDS = 50
+
+/** How long the delegator has to finish a response, from the delegation or from the tool outputs that resume it. */
+export const RESPONSE_TIMEOUT = 60_000
+
+/** Each tool call in a round gets this long; above wait_for_session's 90 s server wait. */
+export const TOOL_TIMEOUT = 100_000
+
+/** The output of each unfinished call in a round that ran past its deadline. */
+export const TIMED_OUT = JSON.stringify({
+  error: "Tool request timed out. Do not claim it succeeded or resend an uncertain admission.",
+})
+
+export type ToolOutput = { call_id: string; output: string }
+
+export type DelegationPhase =
+  /** Waiting for the delegator's response to finish, collecting the function calls it makes. */
+  | { type: "responding"; attempt: number; responseID?: string; calls: readonly ToolCall[] }
+  /** Running the previous response's function calls locally, in order; Live waits for an output for each. */
+  | { type: "running"; attempt: number; outputs: readonly ToolOutput[]; pending: readonly string[] }
+
+export type Delegation = {
+  id: string
+  /** The session tools default to; open_session moves it. */
+  sessionID: string
+  transcript: string[]
+  rounds: number
+  /** Every function call this delegation has received, so a replayed item never runs twice. */
+  callIDs: readonly string[]
+  phase: DelegationPhase
+}
+
+export type VoiceDelegationState = {
+  /** The last attempt ID. Each phase takes a new one, so a late timer or round for an earlier phase is ignored. */
+  seq: number
+  /** Open delegations, oldest first. */
+  delegations: readonly Delegation[]
+  /** Responses that already finished; Live may repeat a terminal event. */
+  finished: readonly string[]
+}
+
+const outcomes = ["completed", "failed", "incomplete", "cancelled"] as const
+
+export type ResponseOutcome = (typeof outcomes)[number]
+
+/** Response events name their delegation, or leave `delegationID` empty when Live does not know it. */
+type Routed = { delegationID?: string | null; responseID?: string }
+
+export type VoiceDelegationEvent =
+  | { type: "delegation.created"; id: string; sessionID: string; transcript: string[] }
+  | ({ type: "response.created" } & Routed)
+  | ({ type: "response.call"; call: ToolCall } & Routed)
+  | ({ type: "response.finished"; outcome: ResponseOutcome } & Routed)
+  /** One call of a round finished; `sessionID` is where later tools default to after it. */
+  | { type: "tool.completed"; id: string; attempt: number; sessionID: string; output: ToolOutput }
+  /** `channelOpen` is whether Live's data channel can still take tool outputs when the timer fires. */
+  | { type: "delegation.timeout"; id: string; attempt: number; channelOpen: boolean }
+  | { type: "call.reset" }
+
+export type VoiceDelegationEffect =
+  /** Replaces the delegation's timer; when it fires, dispatch `delegation.timeout` with this attempt. */
+  | { type: "deadline"; id: string; attempt: number; ms: number }
+  /** Runs the calls in order and dispatches `tool.completed` with this attempt as each one finishes. */
+  | {
+      type: "round"
+      id: string
+      attempt: number
+      request: { id: string; sessionID: string; transcript: string[] }
+      calls: readonly ToolCall[]
+    }
+  /** Aborts the delegation's running tools; the machine already ignores their late results. */
+  | { type: "abort"; id: string }
+  /** Sends the outputs to Live and asks the delegator to continue. */
+  | { type: "outputs"; outputs: readonly ToolOutput[] }
+  /** Clears the delegation's timer and tools. An answered transcript is spent; a discarded one is offered again. */
+  | { type: "finish"; id: string; outcome: "answered" | "discarded" }
+
+export type VoiceDelegationTransition = {
+  state: VoiceDelegationState
+  effects: VoiceDelegationEffect[]
+}
+
+export function createVoiceDelegationState(): VoiceDelegationState {
+  return { seq: 0, delegations: [], finished: [] }
+}
+
+export function hasDelegation(state: VoiceDelegationState, id: string) {
+  return state.delegations.some((delegation) => delegation.id === id)
+}
+
+/** Session replies wait while any delegation is open, so they never talk over its answer. */
+export function delegating(state: VoiceDelegationState) {
+  return state.delegations.length > 0
+}
+
+export function transitionVoiceDelegation(
+  state: VoiceDelegationState,
+  event: VoiceDelegationEvent,
+): VoiceDelegationTransition {
+  if (event.type === "delegation.created") {
+    if (hasDelegation(state, event.id)) return unchanged(state)
+    const attempt = state.seq + 1
+
+    return {
+      state: {
+        ...state,
+        seq: attempt,
+        delegations: [
+          ...state.delegations,
+          {
+            id: event.id,
+            sessionID: event.sessionID,
+            transcript: event.transcript,
+            rounds: 0,
+            callIDs: [],
+            phase: { type: "responding", attempt, calls: [] },
+          },
+        ],
+      },
+      effects: [{ type: "deadline", id: event.id, attempt, ms: RESPONSE_TIMEOUT }],
+    }
+  }
+
+  if (event.type === "call.reset")
+    return {
+      state: { ...state, delegations: [], finished: [...state.finished, ...state.delegations.flatMap(abandoned)] },
+      effects: state.delegations.map(
+        (delegation): VoiceDelegationEffect => ({ type: "finish", id: delegation.id, outcome: "discarded" }),
+      ),
+    }
+
+  if (event.type === "delegation.timeout") {
+    const delegation = state.delegations.find((item) => item.id === event.id)
+
+    if (delegation?.phase.attempt !== event.attempt) return unchanged(state)
+
+    if (delegation.phase.type !== "running" || !event.channelOpen)
+      return finish({ ...state, finished: [...state.finished, ...abandoned(delegation)] }, delegation, "discarded")
+
+    // Live waits for an output per call; an error lets the delegator tell the user instead of waiting forever.
+    // Finished calls keep their real output, so a delivered message is never reported as uncertain.
+    const resumed = resume(state, delegation, [
+      ...delegation.phase.outputs,
+      ...delegation.phase.pending.map((id) => ({ call_id: id, output: TIMED_OUT })),
+    ])
+
+    return { state: resumed.state, effects: [{ type: "abort", id: delegation.id }, ...resumed.effects] }
+  }
+
+  if (event.type === "tool.completed") {
+    const found = state.delegations.find((item) => item.id === event.id)
+
+    if (
+      found?.phase.type !== "running" ||
+      found.phase.attempt !== event.attempt ||
+      !found.phase.pending.includes(event.output.call_id)
+    )
+      return unchanged(state)
+
+    const delegation = {
+      ...found,
+      sessionID: event.sessionID,
+      phase: {
+        ...found.phase,
+        outputs: [...found.phase.outputs, event.output],
+        pending: found.phase.pending.filter((id) => id !== event.output.call_id),
+      },
+    }
+
+    if (!delegation.phase.pending.length) return resume(state, delegation, delegation.phase.outputs)
+
+    return { state: replace(state, delegation), effects: [] }
+  }
+
+  return responded(state, event)
+}
+
+/** Maps a Live event to the machine, reading the response ID wherever the decoder found one. */
+export function responseEvent(event: VoiceEvent): VoiceDelegationEvent | undefined {
+  if (event.type !== "response.event" || !event.event) return undefined
+  const nested = event.event
+  const routed = { delegationID: event.delegation_id, responseID: nested.response?.id ?? nested.response_id }
+
+  if (nested.type === "response.created") return { type: "response.created", ...routed }
+
+  if (nested.type === "response.output_item.done" && nested.item?.type === "function_call") {
+    const item = nested.item
+
+    if (!item.call_id || !item.name) return undefined
+
+    return {
+      type: "response.call",
+      ...routed,
+      call: { call_id: item.call_id, name: item.name, arguments: item.arguments ?? "{}" },
+    }
+  }
+
+  const outcome = outcomes.find((item) => nested.type === `response.${item}`)
+
+  if (!outcome) return undefined
+
+  return { type: "response.finished", ...routed, outcome }
+}
+
+// A delegation that ends before its response does must not leave that response free to bind to a newer one.
+function abandoned(delegation: Delegation) {
+  return delegation.phase.type === "responding" && delegation.phase.responseID ? [delegation.phase.responseID] : []
+}
+
+function resume(
+  state: VoiceDelegationState,
+  delegation: Delegation,
+  outputs: readonly ToolOutput[],
+): VoiceDelegationTransition {
+  const attempt = state.seq + 1
+
+  return {
+    state: replace({ ...state, seq: attempt }, { ...delegation, phase: { type: "responding", attempt, calls: [] } }),
+    effects: [
+      { type: "outputs", outputs },
+      { type: "deadline", id: delegation.id, attempt, ms: RESPONSE_TIMEOUT },
+    ],
+  }
+}
+
+function responded(
+  state: VoiceDelegationState,
+  event: Extract<VoiceDelegationEvent, { type: "response.created" | "response.call" | "response.finished" }>,
+): VoiceDelegationTransition {
+  const delegation = owner(state, event)
+
+  if (!delegation) return unchanged(state)
+  const phase = { ...delegation.phase, responseID: delegation.phase.responseID ?? event.responseID }
+
+  if (event.type === "response.created") return { state: replace(state, { ...delegation, phase }), effects: [] }
+
+  if (event.type === "response.call") {
+    if (state.delegations.some((item) => item.callIDs.includes(event.call.call_id))) return unchanged(state)
+
+    return {
+      state: replace(state, {
+        ...delegation,
+        callIDs: [...delegation.callIDs, event.call.call_id],
+        phase: { ...phase, calls: [...phase.calls, event.call] },
+      }),
+      effects: [],
+    }
+  }
+
+  const key = event.responseID ?? delegation.phase.responseID
+
+  if (!key || state.finished.includes(key)) return unchanged(state)
+  const next = { ...state, finished: [...state.finished, key] }
+
+  // A cancelled, failed or runaway response ends only its own delegation; the call goes on.
+  if (event.outcome !== "completed" || (phase.calls.length > 0 && delegation.rounds >= MAX_ROUNDS))
+    return finish(next, delegation, "discarded")
+
+  if (!phase.calls.length) return finish(next, delegation, "answered")
+  const attempt = state.seq + 1
+
+  return {
+    state: replace(
+      { ...next, seq: attempt },
+      {
+        ...delegation,
+        rounds: delegation.rounds + 1,
+        phase: { type: "running", attempt, outputs: [], pending: phase.calls.map((call) => call.call_id) },
+      },
+    ),
+    effects: [
+      { type: "deadline", id: delegation.id, attempt, ms: phase.calls.length * TOOL_TIMEOUT },
+      {
+        type: "round",
+        id: delegation.id,
+        attempt,
+        request: { id: delegation.id, sessionID: delegation.sessionID, transcript: delegation.transcript },
+        calls: phase.calls,
+      },
+    ],
+  }
+}
+
+// Without a delegation ID, a response belongs to the delegation already bound to it, else to the oldest one still
+// waiting for a response. Delegations running tools have not asked for one yet.
+function owner(state: VoiceDelegationState, event: Routed) {
+  if (event.delegationID) return state.delegations.filter(responding).find((item) => item.id === event.delegationID)
+
+  if (event.responseID && state.finished.includes(event.responseID)) return undefined
+  const waiting = state.delegations.filter(responding)
+
+  if (!event.responseID) return waiting[0]
+
+  return (
+    waiting.find((item) => item.phase.responseID === event.responseID) ?? waiting.find((item) => !item.phase.responseID)
+  )
+}
+
+function responding(
+  delegation: Delegation,
+): delegation is Delegation & { phase: Extract<DelegationPhase, { type: "responding" }> } {
+  return delegation.phase.type === "responding"
+}
+
+function replace(state: VoiceDelegationState, delegation: Delegation): VoiceDelegationState {
+  return { ...state, delegations: state.delegations.map((item) => (item.id === delegation.id ? delegation : item)) }
+}
+
+function finish(
+  state: VoiceDelegationState,
+  delegation: Delegation,
+  outcome: "answered" | "discarded",
+): VoiceDelegationTransition {
+  return {
+    state: { ...state, delegations: state.delegations.filter((item) => item.id !== delegation.id) },
+    effects: [{ type: "finish", id: delegation.id, outcome }],
+  }
+}
+
+function unchanged(state: VoiceDelegationState): VoiceDelegationTransition {
+  return { state, effects: [] }
+}

packages/app/src/voice/protocol.ts

@@ -17,6 +17,8 @@ const Envelope = Schema.Struct({
   event: Schema.optional(
     Schema.Struct({
       type: Schema.String,
+      // Output item events name their response here instead of carrying a `response` object.
+      response_id: Schema.optional(Schema.String),
       item: Schema.optional(Item),
       response: Schema.optional(Schema.Struct({ id: Schema.String, status: Schema.optional(Schema.String) })),
     }),

packages/app/src/voice/tools.ts

@@ -11,7 +11,6 @@ type Request = { id: string; sessionID: string; transcript: string[] }
 
 export class VoiceTools {
   readonly replies = new ReplyWatch()
-  private calls = new Map<string, Promise<string>>()
   private observed = new Map<string, Map<string, string[]>>()
   /** When each session last received a delegation's full header in this call. */
   private explained = new Map<string, number>()
@@ -27,19 +26,11 @@ export class VoiceTools {
   ) {}
 
   run(call: ToolCall, request: Request, signal: AbortSignal) {
-    const existing = this.calls.get(call.call_id)
-
-    if (existing) return existing
-
-    const result = this.execute(call, request, signal)
+    return this.execute(call, request, signal)
       .then(JSON.stringify)
       .catch(() =>
         JSON.stringify({ error: "Tool request failed. Do not claim it succeeded or resend an uncertain admission." }),
       )
-
-    this.calls.set(call.call_id, result)
-
-    return result
   }
 
   answered(id: string) {
@@ -47,6 +38,11 @@ export class VoiceTools {
     this.observed.delete(id)
   }
 
+  /** A delegation that ended without an answer read nothing aloud, so its replies are still announced. */
+  discard(id: string) {
+    this.observed.delete(id)
+  }
+
   async history(sessionID: string, signal: AbortSignal) {
     const messages: SessionMessageInfo[] = []