diff --git a/packages/app/src/voice/call.ts b/packages/app/src/voice/call.ts index b303400f83d..dea298593b6 100644 --- a/packages/app/src/voice/call.ts +++ b/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() - const completed = new Set() + let delegations = createVoiceDelegationState() + // Timers and tool aborts of open delegations; the machine decides when each ends. + const handles = new Map; abort?: AbortController }>() const dirty = new Set() 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, 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)]) }, } diff --git a/packages/app/src/voice/machine.test.ts b/packages/app/src/voice/machine.test.ts new file mode 100644 index 00000000000..5a7f08f9bf4 --- /dev/null +++ b/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) +}) diff --git a/packages/app/src/voice/machine.ts b/packages/app/src/voice/machine.ts new file mode 100644 index 00000000000..c7a5dc46473 --- /dev/null +++ b/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, +): 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 } { + 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: [] } +} diff --git a/packages/app/src/voice/protocol.ts b/packages/app/src/voice/protocol.ts index 5c88e133ac3..9db48128580 100644 --- a/packages/app/src/voice/protocol.ts +++ b/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) })), }), diff --git a/packages/app/src/voice/tools.ts b/packages/app/src/voice/tools.ts index 1b9e2de2afd..8bd082974d5 100644 --- a/packages/app/src/voice/tools.ts +++ b/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>() private observed = new Map>() /** When each session last received a delegation's full header in this call. */ private explained = new Map() @@ -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[] = []