fix(talk): await realtime tool result delivery (#103268)

* fix(talk): await realtime tool result delivery

* fix(talk): terminally cancel queued results

* chore: keep release changelog out of PR

---------

Co-authored-by: Peter Steinberger <peter@steipete.me>
This commit is contained in:
Peter Steinberger
2026-07-09 20:11:20 -07:00
committed by GitHub
parent 4c4609d42b
commit e8fcc93cd3
33 changed files with 3447 additions and 391 deletions

View File

@@ -400,7 +400,7 @@ methods. Treat this as feature discovery, not a full enumeration of
- `talk.session.appendAudio` appends base64 PCM input audio to gateway-owned realtime relay and transcription sessions.
- `talk.session.startTurn`, `talk.session.endTurn`, and `talk.session.cancelTurn` drive managed-room turn lifecycle with stale-turn rejection before state clears.
- `talk.session.cancelOutput` stops assistant audio output, primarily for VAD-gated barge-in in gateway relay sessions.
- `talk.session.submitToolResult` completes a provider tool call emitted by a gateway-owned realtime relay session. Pass `options: { willContinue: true }` for interim tool output when a final result follows, or `options: { suppressResponse: true }` when the tool result should satisfy the provider call without starting another realtime response.
- `talk.session.submitToolResult` completes a provider tool call emitted by a gateway-owned realtime relay session. The request waits for any asynchronous completion signal exposed by the provider bridge; failed submissions keep the linked run active and do not emit a successful tool-result event. Pass `options: { willContinue: true }` for interim tool output or `options: { suppressResponse: true }` when the provider bridge advertises suppression support and the result should not start another response.
- `talk.session.steer` sends active-run voice control into a gateway-owned agent-backed Talk session: `{ sessionId, text, mode? }`, where `mode` is `status`, `steer`, `cancel`, or `followup`; omitted mode is classified from the spoken text.
- `talk.session.close` closes a gateway-owned relay, transcription, or managed-room session and emits terminal Talk events.
- `talk.mode` sets/broadcasts the current Talk mode state for WebChat/Control UI clients.

View File

@@ -952,16 +952,16 @@ Method map for readers migrating from the older `talk.realtime.*` /
The unified control vocabulary is also deliberately narrow:
| Method | Applies to | Contract |
| ------------------------------- | ------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `talk.session.appendAudio` | `realtime/gateway-relay`, `transcription/gateway-relay` | Append a base64 PCM audio chunk to the provider session owned by the same Gateway connection. |
| `talk.session.startTurn` | `stt-tts/managed-room` | Start a managed-room user turn. |
| `talk.session.endTurn` | `stt-tts/managed-room` | End the active turn after stale-turn validation. |
| `talk.session.cancelTurn` | all Gateway-owned sessions | Cancel active capture/provider/agent/TTS work for a turn. |
| `talk.session.cancelOutput` | `realtime/gateway-relay` | Stop assistant audio output without necessarily ending the user turn. |
| `talk.session.submitToolResult` | `realtime/gateway-relay` | Complete a provider tool call emitted by the relay; pass `options.willContinue` for interim output or `options.suppressResponse` to satisfy the call without another assistant response. |
| `talk.session.steer` | agent-backed Talk sessions | Send spoken `status`, `steer`, `cancel`, or `followup` control to the active embedded run resolved from the Talk session. |
| `talk.session.close` | all unified sessions | Stop relay sessions or revoke managed-room state, then forget the unified session id. |
| Method | Applies to | Contract |
| ------------------------------- | ------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `talk.session.appendAudio` | `realtime/gateway-relay`, `transcription/gateway-relay` | Append a base64 PCM audio chunk to the provider session owned by the same Gateway connection. |
| `talk.session.startTurn` | `stt-tts/managed-room` | Start a managed-room user turn. |
| `talk.session.endTurn` | `stt-tts/managed-room` | End the active turn after stale-turn validation. |
| `talk.session.cancelTurn` | all Gateway-owned sessions | Cancel active capture/provider/agent/TTS work for a turn. |
| `talk.session.cancelOutput` | `realtime/gateway-relay` | Stop assistant audio output without necessarily ending the user turn. |
| `talk.session.submitToolResult` | `realtime/gateway-relay` | Complete a provider tool call after any asynchronous completion exposed by its bridge; pass `options.willContinue` for interim output or, when supported, `options.suppressResponse` to avoid another assistant response. |
| `talk.session.steer` | agent-backed Talk sessions | Send spoken `status`, `steer`, `cancel`, or `followup` control to the active embedded run resolved from the Talk session. |
| `talk.session.close` | all unified sessions | Stop relay sessions or revoke managed-room state, then forget the unified session id. |
Do not introduce provider or platform special cases in core to make this work.
Core owns Talk session semantics. Provider plugins own vendor session setup.

View File

@@ -834,6 +834,18 @@ catalog, API-key auth, and dynamic model resolution.
clients. Implement `handleBargeIn` when a transport can detect that a
human is interrupting assistant playback and the provider supports
truncating or clearing the active audio response.
`submitToolResult` may return `void` for synchronous submission, or a
`Promise<void>` for an asynchronous completion boundary the provider
bridge can expose. Gateway relay sessions wait for that promise before
confirming a final result or clearing the linked run; reject it when
submission fails.
Set `supportsToolResultSuppression: false` when the provider cannot
honor `options.suppressResponse`. OpenClaw then avoids suppression for
internal forced-consult and cancellation results, and rejects direct
suppressed-result requests instead of silently starting a response.
Consumers of `createRealtimeVoiceBridgeSession` may likewise return a
promise from `onToolCall`; synchronous throws and rejections are routed
to the session's `onError` callback.
Set `handlesInputAudioBargeIn` only when provider VAD confirms an
interruption by calling `onClearAudio("barge-in")`. Providers that omit
the flag use OpenClaw's local input-audio fallback detection.

View File

@@ -1,6 +1,9 @@
// Discord tests cover manager plugin behavior.
import { PassThrough, type Readable } from "node:stream";
import type { RealtimeVoiceAgentControlResult } from "openclaw/plugin-sdk/realtime-voice";
import type {
RealtimeVoiceAgentControlResult,
RealtimeVoiceForcedConsultCoordinator,
} from "openclaw/plugin-sdk/realtime-voice";
import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { ChannelType } from "../internal/discord.js";
import { createVoiceCaptureState } from "./capture-state.js";
@@ -102,7 +105,10 @@ const {
const getVoiceConnectionMockLocal = vi.fn((): MockConnection | undefined => undefined);
const realtimeSessionMockLocal = {
bridge: { supportsToolResultContinuation: true },
bridge: {
supportsToolResultContinuation: true,
supportsToolResultSuppression: true as boolean | undefined,
},
acknowledgeMark: vi.fn(),
close: vi.fn(),
connect: vi.fn(async () => undefined),
@@ -371,6 +377,7 @@ describe("DiscordVoiceManager", () => {
realtimeSessionMock.handleBargeIn.mockClear();
realtimeSessionMock.setMediaTimestamp.mockClear();
realtimeSessionMock.submitToolResult.mockClear();
realtimeSessionMock.bridge.supportsToolResultSuppression = true;
createRealtimeVoiceBridgeSessionMock.mockClear();
createRealtimeVoiceBridgeSessionMock.mockReturnValue(realtimeSessionMock);
controlRealtimeVoiceAgentRunMock.mockReset();
@@ -2417,6 +2424,104 @@ describe("DiscordVoiceManager", () => {
);
});
it("keeps the realtime tool callback pending until result delivery completes", async () => {
let acceptResult = () => {};
const accepted = new Promise<void>((resolve) => {
acceptResult = resolve;
});
realtimeSessionMock.submitToolResult.mockImplementationOnce(() => accepted);
const manager = createManager({
groupPolicy: "open",
voice: {
enabled: true,
mode: "agent-proxy",
realtime: { provider: "openai" },
},
});
await manager.join({ guildId: "g1", channelId: "1001" });
const bridgeParams = lastRealtimeBridgeParams() as
| {
onToolCall?: (
event: {
itemId: string;
callId: string;
name: string;
args: unknown;
},
session: typeof realtimeSessionMock,
) => Promise<void>;
}
| undefined;
const handled = bridgeParams?.onToolCall?.(
{
itemId: "item-unknown",
callId: "call-unknown",
name: "unknown_tool",
args: {},
},
realtimeSessionMock,
);
if (!handled) {
throw new Error("expected realtime tool callback promise");
}
let settled = false;
void handled.then(() => {
settled = true;
});
await Promise.resolve();
expect(realtimeSessionMock.submitToolResult).toHaveBeenCalledTimes(1);
expect(settled).toBe(false);
acceptResult();
await handled;
expect(settled).toBe(true);
});
it("does not retry a rejected control result submission as a tool error", async () => {
realtimeSessionMock.submitToolResult.mockRejectedValueOnce(new Error("result delivery failed"));
const manager = createManager({
groupPolicy: "open",
voice: {
enabled: true,
mode: "agent-proxy",
realtime: { provider: "openai" },
},
});
await manager.join({ guildId: "g1", channelId: "1001" });
const bridgeParams = lastRealtimeBridgeParams() as
| {
onToolCall?: (
event: {
itemId: string;
callId: string;
name: string;
args: unknown;
},
session: typeof realtimeSessionMock,
) => Promise<void>;
}
| undefined;
const handled = bridgeParams?.onToolCall?.(
{
itemId: "item-control",
callId: "call-control",
name: "openclaw_agent_control",
args: { text: "check this", mode: "steer" },
},
realtimeSessionMock,
);
if (!handled) {
throw new Error("expected realtime tool callback promise");
}
await expect(handled).rejects.toThrow("result delivery failed");
expect(realtimeSessionMock.submitToolResult).toHaveBeenCalledTimes(1);
});
it("rejects malformed realtime consult tool calls without crashing Discord voice", async () => {
const manager = createManager({
groupPolicy: "open",
@@ -4246,10 +4351,186 @@ describe("DiscordVoiceManager", () => {
"call-late",
{
status: "already_delivered",
message: "OpenClaw already delivered this answer to Discord voice.",
message: "OpenClaw already delivered this answer to Discord voice. Do not repeat it.",
},
{ suppressResponse: true },
);
realtimeSessionMock.bridge.supportsToolResultSuppression = false;
bridgeParams?.onToolCall?.(
{
itemId: "item-late-unsuppressed",
callId: "call-late-unsuppressed",
name: "openclaw_agent_consult",
args: { question: "late question" },
},
realtimeSessionMock,
);
await vi.waitFor(() => {
const call = realtimeSessionMock.submitToolResult.mock.calls.find(
([callId]) => callId === "call-late-unsuppressed",
);
expect(call).toEqual([
"call-late-unsuppressed",
{
status: "already_delivered",
message: "OpenClaw already delivered this answer to Discord voice. Do not repeat it.",
},
]);
});
});
it("terminally satisfies a late native call for a cancelled forced consult", async () => {
const manager = createManager({
groupPolicy: "open",
voice: {
enabled: true,
mode: "agent-proxy",
realtime: { provider: "openai" },
},
});
await manager.join({ guildId: "g1", channelId: "1001" });
const entry = getSessionEntry(manager) as {
realtime?: unknown;
};
const realtime = entry.realtime as {
forcedConsults: RealtimeVoiceForcedConsultCoordinator;
};
const cancelled = realtime.forcedConsults.prepare("cancelled question");
if (!cancelled) {
throw new Error("expected forced consult handle");
}
realtime.forcedConsults.markStarted(cancelled);
realtime.forcedConsults.markCancelled(cancelled);
const bridgeParams = lastRealtimeBridgeParams() as
| {
onToolCall?: (
event: {
itemId: string;
callId: string;
name: string;
args: unknown;
},
session: typeof realtimeSessionMock,
) => Promise<void>;
}
| undefined;
await bridgeParams?.onToolCall?.(
{
itemId: "item-cancelled",
callId: "call-cancelled",
name: "openclaw_agent_consult",
args: { question: "cancelled question" },
},
realtimeSessionMock,
);
expect(agentCommandMock).not.toHaveBeenCalled();
expect(realtimeSessionMock.submitToolResult).toHaveBeenCalledWith(
"call-cancelled",
{
status: "cancelled",
message: "OpenClaw cancelled this consult before completion. Do not restart it.",
},
{ suppressResponse: true },
);
});
it("lets an unsuppressed in-flight native result own forced consult delivery", async () => {
let resolveAgentTurn: ((result: { payloads: Array<{ text: string }> }) => void) | undefined;
agentCommandMock.mockReturnValueOnce(
new Promise((resolve) => {
resolveAgentTurn = resolve;
}),
);
const manager = createManager({
groupPolicy: "open",
voice: {
enabled: true,
mode: "agent-proxy",
realtime: { provider: "openai" },
},
});
await manager.join({ guildId: "g1", channelId: "1001" });
const entry = getSessionEntry(manager) as {
realtime?: {
beginSpeakerTurn: (
context: { extraSystemPrompt?: string; senderIsOwner: boolean; speakerLabel: string },
userId: string,
) => { sendInputAudio: (audio: Buffer) => void };
};
};
const bridgeParams = lastRealtimeBridgeParams() as
| {
onToolCall?: (
event: {
itemId: string;
callId: string;
name: string;
args: unknown;
},
session: typeof realtimeSessionMock,
) => Promise<void>;
onTranscript?: (role: "user" | "assistant", text: string, isFinal: boolean) => void;
}
| undefined;
const ownerTurn = entry.realtime?.beginSpeakerTurn(
{ extraSystemPrompt: undefined, senderIsOwner: true, speakerLabel: "Owner" },
"u-owner",
);
ownerTurn?.sendInputAudio(Buffer.alloc(8));
await emitFinalRealtimeUserTranscript(bridgeParams, "late question");
realtimeSessionMock.bridge.supportsToolResultSuppression = false;
const submission = bridgeParams?.onToolCall?.(
{
itemId: "item-late",
callId: "call-late",
name: "openclaw_agent_consult",
args: { question: "late question" },
},
realtimeSessionMock,
);
resolveAgentTurn?.({ payloads: [{ text: "forced answer" }] });
await submission;
expect(realtimeSessionMock.submitToolResult).toHaveBeenCalledWith("call-late", {
text: "forced answer",
});
expectUserMessageNotIncludes("forced answer");
expectUserMessageNotIncludes("I hit an error while checking that. Please try again.");
let resolveRetryTurn: ((result: { payloads: Array<{ text: string }> }) => void) | undefined;
agentCommandMock.mockReturnValueOnce(
new Promise((resolve) => {
resolveRetryTurn = resolve;
}),
);
const retryTurn = entry.realtime?.beginSpeakerTurn(
{ extraSystemPrompt: undefined, senderIsOwner: true, speakerLabel: "Owner" },
"u-owner",
);
retryTurn?.sendInputAudio(Buffer.alloc(8));
await emitFinalRealtimeUserTranscript(bridgeParams, "retry question");
realtimeSessionMock.submitToolResult.mockRejectedValueOnce(
new Error("native delivery rejected"),
);
const rejectedSubmission = bridgeParams?.onToolCall?.(
{
itemId: "item-retry",
callId: "call-retry",
name: "openclaw_agent_consult",
args: { question: "retry question" },
},
realtimeSessionMock,
);
resolveRetryTurn?.({ payloads: [{ text: "local retry answer" }] });
await expect(rejectedSubmission).rejects.toThrow("native delivery rejected");
await vi.waitFor(() => expectUserMessageIncludes("local retry answer"));
});
it("suppresses late forced agent-proxy tool calls when the forced consult rejects", async () => {
@@ -4315,7 +4596,7 @@ describe("DiscordVoiceManager", () => {
"call-late",
{
status: "already_delivered",
message: "OpenClaw already delivered this answer to Discord voice.",
message: "OpenClaw already delivered this answer to Discord voice. Do not repeat it.",
},
{ suppressResponse: true },
),
@@ -4485,7 +4766,7 @@ describe("DiscordVoiceManager", () => {
"call-new",
{
status: "already_delivered",
message: "OpenClaw already delivered this answer to Discord voice.",
message: "OpenClaw already delivered this answer to Discord voice. Do not repeat it.",
},
{ suppressResponse: true },
);

View File

@@ -133,6 +133,7 @@ type RecentAgentProxyConsultResult =
type AgentProxyConsultState = {
speaker: DiscordRealtimeSpeakerContext;
handledByForcedPlayback?: boolean;
providerDelivery?: Promise<boolean>;
promise?: Promise<string>;
result?: RecentAgentProxyConsultResult;
};
@@ -1107,21 +1108,21 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
);
}
private handleToolCall(
private async handleToolCall(
event: RealtimeVoiceToolCallEvent,
session: RealtimeVoiceBridgeSession,
): void {
): Promise<void> {
const callId = event.callId || event.itemId || "unknown";
if (event.name === REALTIME_VOICE_AGENT_CONTROL_TOOL_NAME) {
void this.handleAgentControlToolCall(event, session, callId);
await this.handleAgentControlToolCall(event, session, callId);
return;
}
if (event.name !== REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) {
session.submitToolResult(callId, { error: `Tool "${event.name}" not available` });
await session.submitToolResult(callId, { error: `Tool "${event.name}" not available` });
return;
}
if (this.consultToolPolicy === "none") {
session.submitToolResult(callId, { error: `Tool "${event.name}" not available` });
await session.submitToolResult(callId, { error: `Tool "${event.name}" not available` });
return;
}
const exactSpeechText = extractDiscordExactSpeechConsultText(event.args);
@@ -1129,7 +1130,7 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
logger.info(
`discord voice: realtime exact speech consult bypassed call=${callId || "unknown"} answerChars=${exactSpeechText.length}`,
);
session.submitToolResult(callId, { text: exactSpeechText });
await session.submitToolResult(callId, { text: exactSpeechText });
return;
}
let consultMessage: string;
@@ -1140,13 +1141,23 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
logger.warn(
`discord voice: realtime consult rejected malformed args call=${callId || "unknown"}: ${message}`,
);
session.submitToolResult(callId, { error: message });
await session.submitToolResult(callId, { error: message });
return;
}
logger.info(
`discord voice: realtime consult requested call=${callId || "unknown"} voiceSession=${this.params.entry.voiceSessionKey} supervisorSession=${this.params.entry.route.sessionKey} agent=${this.params.entry.route.agentId} question=${formatVoiceLogPreview(consultMessage)}`,
);
const nativeConsult = this.forcedConsults.recordNativeConsult(event.args, callId);
if (
nativeConsult.kind === "already_delivered" &&
this.forcedConsults.isCancelled(nativeConsult.handle)
) {
await this.submitTerminalRealtimeToolResult(callId, session, {
status: "cancelled",
message: "OpenClaw cancelled this consult before completion. Do not restart it.",
});
return;
}
const pendingConsult = nativeConsult.kind === "pending" ? nativeConsult.handle : undefined;
if (pendingConsult) {
this.forcedConsults.rememberQuestion(pendingConsult, consultMessage);
@@ -1164,12 +1175,12 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
logger.info(
`discord voice: realtime consult matched recent agent result but newer speaker audio is pending call=${callId} speaker=${recentSpeaker?.speakerLabel ?? "unknown"} owner=${recentSpeaker?.senderIsOwner ?? false}`,
);
session.submitToolResult(callId, {
await session.submitToolResult(callId, {
error: "Discord speaker context changed before this realtime consult completed",
});
return;
}
if (this.submitRecentAgentProxyConsultResult(callId, recentConsult, session)) {
if (await this.submitRecentAgentProxyConsultResult(callId, recentConsult, session)) {
return;
}
}
@@ -1187,7 +1198,7 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
logger.warn(
`discord voice: realtime consult has no speaker context call=${callId || "unknown"}`,
);
session.submitToolResult(callId, { error: "No Discord speaker context available" });
await session.submitToolResult(callId, { error: "No Discord speaker context available" });
return;
}
const promise = this.runAgentTurn({
@@ -1197,19 +1208,19 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
if (recent) {
this.setRecentAgentProxyConsultPromise(recent, promise);
}
void promise
.then((text) => {
logger.info(
`discord voice: realtime consult answer (${text.length} chars) voiceSession=${this.params.entry.voiceSessionKey} supervisorSession=${this.params.entry.route.sessionKey} agent=${this.params.entry.route.agentId} speaker=${context.speakerLabel} owner=${context.senderIsOwner}: ${formatVoiceLogPreview(text)}`,
);
session.submitToolResult(callId, { text });
})
.catch((error: unknown) => {
logger.warn(
`discord voice: realtime consult failed call=${callId || "unknown"}: ${formatErrorMessage(error)}`,
);
session.submitToolResult(callId, { error: formatErrorMessage(error) });
});
let text: string;
try {
text = await promise;
} catch (error) {
const message = formatErrorMessage(error);
logger.warn(`discord voice: realtime consult failed call=${callId || "unknown"}: ${message}`);
await session.submitToolResult(callId, { error: message });
return;
}
logger.info(
`discord voice: realtime consult answer (${text.length} chars) voiceSession=${this.params.entry.voiceSessionKey} supervisorSession=${this.params.entry.route.sessionKey} agent=${this.params.entry.route.agentId} speaker=${context.speakerLabel} owner=${context.senderIsOwner}: ${formatVoiceLogPreview(text)}`,
);
await session.submitToolResult(callId, { text });
}
private async handleAgentControlToolCall(
@@ -1217,18 +1228,20 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
session: RealtimeVoiceBridgeSession,
callId: string,
): Promise<void> {
let result: RealtimeVoiceAgentControlResult;
try {
const parsed = parseRealtimeVoiceAgentControlToolArgs(event.args);
const result = await controlRealtimeVoiceAgentRun({
result = await controlRealtimeVoiceAgentRun({
sessionKey: this.params.entry.route.sessionKey,
text: parsed.text,
mode: parsed.mode,
});
this.logAgentControlResult(result);
session.submitToolResult(callId, result);
} catch (error) {
session.submitToolResult(callId, { error: formatErrorMessage(error) });
await session.submitToolResult(callId, { error: formatErrorMessage(error) });
return;
}
this.logAgentControlResult(result);
await session.submitToolResult(callId, result);
}
private async runAgentTurn(params: {
@@ -1493,17 +1506,21 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
});
this.setRecentAgentProxyConsultPromise(pending, promise);
const text = await promise;
await state.providerDelivery;
logger.info(
`discord voice: realtime forced agent consult answer (${text.length} chars) elapsedMs=${Date.now() - startedAt} voiceSession=${this.params.entry.voiceSessionKey} supervisorSession=${this.params.entry.route.sessionKey} agent=${this.params.entry.route.agentId}: ${formatVoiceLogPreview(text)}`,
);
if (text.trim()) {
if (text.trim() && state.handledByForcedPlayback) {
this.enqueueExactSpeechMessage(text);
}
} catch (error) {
await state.providerDelivery;
logger.warn(
`discord voice: realtime forced agent consult failed elapsedMs=${Date.now() - startedAt}: ${formatErrorMessage(error)}`,
);
this.enqueueExactSpeechMessage(DISCORD_REALTIME_FALLBACK_TEXT);
if (state.handledByForcedPlayback) {
this.enqueueExactSpeechMessage(DISCORD_REALTIME_FALLBACK_TEXT);
}
}
}
@@ -1618,41 +1635,65 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
return this.forcedConsults.findRecent(consultMessage);
}
private submitRecentAgentProxyConsultResult(
private async submitTerminalRealtimeToolResult(
callId: string,
session: RealtimeVoiceBridgeSession,
result: Record<string, string>,
): Promise<void> {
// Providers without suppressed results still need a terminal result; the payload tells the
// model not to repeat audio that Discord already played or restart cancelled work.
if (session.bridge.supportsToolResultSuppression === false) {
await session.submitToolResult(callId, result);
return;
}
await session.submitToolResult(callId, result, { suppressResponse: true });
}
private async submitRecentAgentProxyConsultResult(
callId: string,
recent: AgentProxyConsultHandle,
session: RealtimeVoiceBridgeSession,
): boolean {
): Promise<boolean> {
const state = recent.context;
if (!state) {
return false;
}
const submitAlreadyDelivered = () => {
session.submitToolResult(
callId,
{
status: "already_delivered",
message: "OpenClaw already delivered this answer to Discord voice.",
},
{ suppressResponse: true },
);
const providerOwnsDelivery = Boolean(
state.handledByForcedPlayback &&
state.promise &&
!state.result &&
session.bridge.supportsToolResultSuppression === false,
);
let resolveProviderDelivery: ((accepted: boolean) => void) | undefined;
if (providerOwnsDelivery) {
// Forced playback waits for native acceptance so a failed delivery can restore
// the local success/fallback path instead of losing the answer entirely.
state.providerDelivery = new Promise<boolean>((resolve) => {
resolveProviderDelivery = resolve;
});
}
const submitAlreadyDelivered = async (): Promise<void> => {
await this.submitTerminalRealtimeToolResult(callId, session, {
status: "already_delivered",
message: "OpenClaw already delivered this answer to Discord voice. Do not repeat it.",
});
};
const submitResult = (result: RecentAgentProxyConsultResult) => {
if (state.handledByForcedPlayback) {
submitAlreadyDelivered();
const submitResult = async (result: RecentAgentProxyConsultResult): Promise<void> => {
if (state.handledByForcedPlayback && !providerOwnsDelivery) {
await submitAlreadyDelivered();
return;
}
if (result.status === "fulfilled") {
session.submitToolResult(callId, { text: result.text });
await session.submitToolResult(callId, { text: result.text });
return;
}
session.submitToolResult(callId, { error: result.error });
await session.submitToolResult(callId, { error: result.error });
};
if (state.result) {
logger.info(
`discord voice: realtime consult reused recent agent result call=${callId || "unknown"} speaker=${state.speaker.speakerLabel} owner=${state.speaker.senderIsOwner}`,
);
submitResult(state.result);
await submitResult(state.result);
return true;
}
if (!state.promise) {
@@ -1661,15 +1702,27 @@ export class DiscordRealtimeVoiceSession implements VoiceRealtimeSession {
logger.info(
`discord voice: realtime consult joined in-flight agent result call=${callId || "unknown"} speaker=${state.speaker.speakerLabel} owner=${state.speaker.senderIsOwner}`,
);
if (state.handledByForcedPlayback) {
void state.promise.then(submitAlreadyDelivered, submitAlreadyDelivered);
if (state.handledByForcedPlayback && !providerOwnsDelivery) {
await state.promise.catch(() => undefined);
await submitAlreadyDelivered();
return true;
}
void state.promise
.then((text) => session.submitToolResult(callId, { text }))
.catch((error: unknown) =>
session.submitToolResult(callId, { error: formatErrorMessage(error) }),
);
let result: RecentAgentProxyConsultResult;
try {
result = { status: "fulfilled", text: await state.promise };
} catch (error) {
result = { status: "rejected", error: formatErrorMessage(error) };
}
try {
await submitResult(result);
if (providerOwnsDelivery) {
state.handledByForcedPlayback = false;
resolveProviderDelivery?.(true);
}
} catch (error) {
resolveProviderDelivery?.(false);
throw error;
}
return true;
}
}

View File

@@ -0,0 +1,145 @@
import type { PluginRuntime, RuntimeLogger } from "openclaw/plugin-sdk/plugin-runtime";
import type {
RealtimeVoiceBridgeSession,
TalkEventInput,
} from "openclaw/plugin-sdk/realtime-voice";
import { describe, expect, it, vi } from "vitest";
import { handleGoogleMeetRealtimeConsultToolCall } from "./agent-consult.js";
import { resolveGoogleMeetConfig } from "./config.js";
function makeSession(
submitToolResult: RealtimeVoiceBridgeSession["submitToolResult"],
supportsToolResultContinuation = false,
): RealtimeVoiceBridgeSession {
return {
bridge: { supportsToolResultContinuation },
submitToolResult,
} as unknown as RealtimeVoiceBridgeSession;
}
function makeRuntime() {
const sessions: Record<string, Record<string, unknown>> = {};
const runEmbeddedAgent = vi.fn(async () => ({
payloads: [{ text: "Use the launch summary." }],
meta: {},
}));
const runtime = {
agent: {
resolveAgentDir: vi.fn(() => "/tmp/agent"),
resolveAgentWorkspaceDir: vi.fn(() => "/tmp/workspace"),
ensureAgentWorkspace: vi.fn(async () => {}),
resolveAgentTimeoutMs: vi.fn(() => 1_000),
session: {
resolveStorePath: vi.fn(() => "/tmp/sessions.json"),
loadSessionStore: vi.fn(() => sessions),
saveSessionStore: vi.fn(async () => {}),
updateSessionStore: vi.fn(
async (
_storePath: string,
update: (store: Record<string, Record<string, unknown>>) => unknown,
) => await update(sessions),
),
getSessionEntry: vi.fn(({ sessionKey }: { sessionKey: string }) => sessions[sessionKey]),
patchSessionEntry: vi.fn(
async ({
sessionKey,
fallbackEntry,
update,
}: {
sessionKey: string;
fallbackEntry: Record<string, unknown>;
update: (
entry: Record<string, unknown>,
) => Promise<Record<string, unknown>> | Record<string, unknown>;
}) => {
const current = sessions[sessionKey] ?? fallbackEntry;
const patch = await update(current);
const next = { ...current, ...patch };
sessions[sessionKey] = next;
return next;
},
),
resolveSessionFilePath: vi.fn(() => "/tmp/session.json"),
},
runEmbeddedAgent,
},
} as unknown as PluginRuntime;
return { runEmbeddedAgent, runtime };
}
const logger = {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
} as unknown as RuntimeLogger;
describe("handleGoogleMeetRealtimeConsultToolCall", () => {
it("emits a final tool event only after the bridge accepts the result", async () => {
let acceptResult = () => {};
const accepted = new Promise<void>((resolve) => {
acceptResult = resolve;
});
const submitToolResult = vi.fn(() => accepted);
const events: TalkEventInput[] = [];
const handled = handleGoogleMeetRealtimeConsultToolCall({
strategy: "agent",
session: makeSession(submitToolResult),
event: {
itemId: "item-1",
callId: "call-1",
name: "openclaw_agent_consult",
args: { question: "What should I say?" },
},
config: resolveGoogleMeetConfig({}),
fullConfig: {},
runtime: {} as PluginRuntime,
logger,
meetingSessionId: "meet-1",
transcript: [],
onTalkEvent: (event) => events.push(event),
});
expect(submitToolResult).toHaveBeenCalledTimes(1);
expect(events).toEqual([]);
acceptResult();
await handled;
expect(events).toEqual([
expect.objectContaining({ type: "tool.error", callId: "call-1", final: true }),
]);
});
it("does not retry a rejected result submission as a second tool error", async () => {
const deliveryError = new Error("result delivery failed");
const submitToolResult = vi.fn(async () => {
throw deliveryError;
});
const events: TalkEventInput[] = [];
const { runEmbeddedAgent, runtime } = makeRuntime();
await expect(
handleGoogleMeetRealtimeConsultToolCall({
strategy: "bidi",
session: makeSession(submitToolResult),
event: {
itemId: "item-1",
callId: "call-1",
name: "openclaw_agent_consult",
args: { question: "What should I say?" },
},
config: resolveGoogleMeetConfig({ realtime: { agentId: "jay" } }),
fullConfig: {},
runtime,
logger,
meetingSessionId: "meet-1",
transcript: [],
onTalkEvent: (event) => events.push(event),
}),
).rejects.toThrow("result delivery failed");
expect(runEmbeddedAgent).toHaveBeenCalledTimes(1);
expect(submitToolResult).toHaveBeenCalledTimes(1);
expect(events.map((event) => event.type)).toEqual(["tool.progress"]);
});
});

View File

@@ -29,16 +29,20 @@ export function resolveGoogleMeetRealtimeTools(policy: GoogleMeetToolPolicy): Re
return resolveRealtimeVoiceAgentConsultTools(policy);
}
function submitGoogleMeetConsultWorkingResponse(
async function submitGoogleMeetConsultWorkingResponse(
session: RealtimeVoiceBridgeSession,
callId: string,
): void {
): Promise<void> {
if (!session.bridge.supportsToolResultContinuation) {
return;
}
session.submitToolResult(callId, buildRealtimeVoiceAgentConsultWorkingResponse("participant"), {
willContinue: true,
});
await session.submitToolResult(
callId,
buildRealtimeVoiceAgentConsultWorkingResponse("participant"),
{
willContinue: true,
},
);
}
export async function consultOpenClawAgentForGoogleMeet(params: {
@@ -77,7 +81,7 @@ export async function consultOpenClawAgentForGoogleMeet(params: {
});
}
export function handleGoogleMeetRealtimeConsultToolCall(params: {
export async function handleGoogleMeetRealtimeConsultToolCall(params: {
strategy: string;
session: RealtimeVoiceBridgeSession;
event: RealtimeVoiceToolCallEvent;
@@ -89,69 +93,67 @@ export function handleGoogleMeetRealtimeConsultToolCall(params: {
requesterSessionKey?: string;
transcript: Array<{ role: "user" | "assistant"; text: string }>;
onTalkEvent?: (event: TalkEventInput) => void;
}): void {
}): Promise<void> {
const callId = params.event.callId || params.event.itemId;
if (params.strategy !== "bidi") {
const error = `Tool "${params.event.name}" is only available in bidi realtime strategy`;
await params.session.submitToolResult(callId, { error });
params.onTalkEvent?.({
type: "tool.error",
callId,
payload: {
name: params.event.name,
error: `Tool "${params.event.name}" is only available in bidi realtime strategy`,
error,
},
final: true,
});
params.session.submitToolResult(callId, {
error: `Tool "${params.event.name}" is only available in bidi realtime strategy`,
});
return;
}
if (params.event.name !== REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) {
const error = `Tool "${params.event.name}" not available`;
await params.session.submitToolResult(callId, { error });
params.onTalkEvent?.({
type: "tool.error",
callId,
payload: { name: params.event.name, error: `Tool "${params.event.name}" not available` },
payload: { name: params.event.name, error },
final: true,
});
params.session.submitToolResult(callId, {
error: `Tool "${params.event.name}" not available`,
});
return;
}
await submitGoogleMeetConsultWorkingResponse(params.session, callId);
params.onTalkEvent?.({
type: "tool.progress",
callId,
payload: { name: params.event.name, status: "working" },
});
submitGoogleMeetConsultWorkingResponse(params.session, callId);
void consultOpenClawAgentForGoogleMeet({
config: params.config,
fullConfig: params.fullConfig,
runtime: params.runtime,
logger: params.logger,
meetingSessionId: params.meetingSessionId,
requesterSessionKey: params.requesterSessionKey,
args: params.event.args,
transcript: params.transcript,
})
.then((result) => {
params.onTalkEvent?.({
type: "tool.result",
callId,
payload: { name: params.event.name, result },
final: true,
});
params.session.submitToolResult(callId, result);
})
.catch((error: unknown) => {
params.onTalkEvent?.({
type: "tool.error",
callId,
payload: { name: params.event.name, error: formatErrorMessage(error) },
final: true,
});
params.session.submitToolResult(callId, {
error: formatErrorMessage(error),
});
let result: { text: string };
try {
result = await consultOpenClawAgentForGoogleMeet({
config: params.config,
fullConfig: params.fullConfig,
runtime: params.runtime,
logger: params.logger,
meetingSessionId: params.meetingSessionId,
requesterSessionKey: params.requesterSessionKey,
args: params.event.args,
transcript: params.transcript,
});
} catch (error) {
const message = formatErrorMessage(error);
await params.session.submitToolResult(callId, { error: message });
params.onTalkEvent?.({
type: "tool.error",
callId,
payload: { name: params.event.name, error: message },
final: true,
});
return;
}
await params.session.submitToolResult(callId, result);
params.onTalkEvent?.({
type: "tool.result",
callId,
payload: { name: params.event.name, result },
final: true,
});
}

View File

@@ -677,7 +677,7 @@ export async function startNodeRealtimeAudioBridge(params: {
payload: { name: event.name, args: event.args },
});
const turnId = ensureTalkTurn();
handleGoogleMeetRealtimeConsultToolCall({
return handleGoogleMeetRealtimeConsultToolCall({
strategy,
session,
event,

View File

@@ -1260,7 +1260,7 @@ export async function startCommandRealtimeAudioBridge(params: {
payload: { name: event.name, args: event.args },
});
const turnId = ensureTalkTurn();
handleGoogleMeetRealtimeConsultToolCall({
return handleGoogleMeetRealtimeConsultToolCall({
strategy,
session,
event,

View File

@@ -405,6 +405,7 @@ describe("google provider plugin hooks", () => {
throw new Error("expected Google realtime bridge");
}
expect(bridge.supportsToolResultContinuation).toBe(false);
expect(bridge.supportsToolResultSuppression).toBe(false);
expect(bridge.sendAudio(Buffer.alloc(160))).toBeUndefined();
expect(bridge.setMediaTimestamp(20)).toBeUndefined();
expect(bridge.sendUserMessage?.("hello")).toBeUndefined();

View File

@@ -248,6 +248,7 @@ function createLazyGoogleRealtimeVoiceBridge(
get supportsToolResultContinuation() {
return bridge?.supportsToolResultContinuation ?? false;
},
supportsToolResultSuppression: false,
connect: async () => {
const loadedBridge = await loadBridge();
if (closed) {

View File

@@ -188,6 +188,7 @@ describe("buildGoogleRealtimeVoiceProvider", () => {
});
expect(bridge.supportsToolResultContinuation).toBe(false);
expect(bridge.supportsToolResultSuppression).toBe(false);
await bridge.connect();
const params = lastConnectParams();
@@ -1295,7 +1296,7 @@ describe("buildGoogleRealtimeVoiceProvider", () => {
args: { query: "hi" },
});
bridge.submitToolResult("call-1", { result: "ok" });
void bridge.submitToolResult("call-1", { result: "ok" });
expect(session.sendToolResponse).toHaveBeenCalledWith({
functionResponses: [
@@ -1330,12 +1331,12 @@ describe("buildGoogleRealtimeVoiceProvider", () => {
},
});
bridge.submitToolResult(
void bridge.submitToolResult(
"consult-call",
{ status: "working", message: "Tell the participant you are checking." },
{ willContinue: true },
);
bridge.submitToolResult("consult-call", { text: "The meeting starts at 3." });
void bridge.submitToolResult("consult-call", { text: "The meeting starts at 3." });
expect(session.sendToolResponse).toHaveBeenNthCalledWith(1, {
functionResponses: [
@@ -1381,13 +1382,13 @@ describe("buildGoogleRealtimeVoiceProvider", () => {
},
});
bridge.submitToolResult("consult-call", { status: "working" }, { willContinue: true });
void bridge.submitToolResult("consult-call", { status: "working" }, { willContinue: true });
expect(session.sendToolResponse).not.toHaveBeenCalled();
expect(requireFirstError(onError).message).toContain(
"does not support continuing tool responses",
);
bridge.submitToolResult("consult-call", { text: "The meeting starts at 3." });
void bridge.submitToolResult("consult-call", { text: "The meeting starts at 3." });
expect(session.sendToolResponse).toHaveBeenCalledWith({
functionResponses: [
@@ -1412,7 +1413,7 @@ describe("buildGoogleRealtimeVoiceProvider", () => {
await bridge.connect();
bridge.submitToolResult("missing-call", { result: "ok" });
void bridge.submitToolResult("missing-call", { result: "ok" });
expect(session.sendToolResponse).not.toHaveBeenCalled();
const error = requireFirstError(onError);
@@ -1444,11 +1445,11 @@ describe("buildGoogleRealtimeVoiceProvider", () => {
throw sendError;
});
bridge.submitToolResult("call-1", ["retryable"]);
void bridge.submitToolResult("call-1", ["retryable"]);
expect(onError).toHaveBeenCalledWith(sendError);
bridge.submitToolResult("call-1", { result: "ok" });
void bridge.submitToolResult("call-1", { result: "ok" });
expect(session.sendToolResponse).toHaveBeenLastCalledWith({
functionResponses: [

View File

@@ -494,6 +494,7 @@ function formatGoogleLiveCloseEvent(
class GoogleRealtimeVoiceBridge implements RealtimeVoiceBridge {
readonly supportsToolResultContinuation: boolean;
readonly supportsToolResultSuppression = false;
private session: GoogleLiveSession | null = null;
private connected = false;

View File

@@ -316,6 +316,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
});
expect(bridge.supportsToolResultContinuation).toBe(true);
expect(bridge.supportsToolResultSuppression).toBe(true);
});
it("adds OpenClaw attribution headers to native realtime websocket requests", () => {
@@ -1938,7 +1939,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })),
);
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
expect(parseSent(socket).slice(-1)).toEqual([
{
@@ -2154,7 +2155,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
socket.emit("message", Buffer.from(JSON.stringify({ type: "session.updated" })));
await connecting;
bridge.submitToolResult("call_1", { status: "working" }, { willContinue: true });
void bridge.submitToolResult("call_1", { status: "working" }, { willContinue: true });
expect(parseSent(socket).slice(-1)).toEqual([
{
@@ -2168,7 +2169,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
]);
expect(hasSentEventType(socket, "response.create")).toBe(false);
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
expect(parseSent(socket).slice(-3)).toEqual([
{
@@ -2209,7 +2210,11 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
socket.emit("message", Buffer.from(JSON.stringify({ type: "session.updated" })));
await connecting;
bridge.submitToolResult("call_1", { status: "already_delivered" }, { suppressResponse: true });
void bridge.submitToolResult(
"call_1",
{ status: "already_delivered" },
{ suppressResponse: true },
);
expect(parseSent(socket).slice(-1)).toEqual([
{
@@ -2248,13 +2253,13 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
"message",
Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })),
);
bridge.submitToolResult("call_1", { status: "working" }, { willContinue: true });
void bridge.submitToolResult("call_1", { status: "working" }, { willContinue: true });
socket.emit("message", Buffer.from(JSON.stringify({ type: "response.done" })));
expect(onError).not.toHaveBeenCalled();
expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([]);
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
expect(parseSent(socket).slice(-3)).toEqual([
{
@@ -2292,7 +2297,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })),
);
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
socket.emit("message", Buffer.from(JSON.stringify({ type: "response.cancelled" })));
expect(parseSent(socket).slice(-1)).toEqual([expectedResponseCreateEvent()]);
@@ -2530,7 +2535,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })),
);
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
bridge.handleBargeIn?.({ audioPlaybackActive: true });
const responseCancelEvent = parseSent(socket).findLast(
(event) => event.type === "response.cancel",
@@ -2598,7 +2603,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
if (!responseCancelEvent?.event_id) {
throw new Error("expected response.cancel event id");
}
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
socket.emit("message", Buffer.from(JSON.stringify({ type: "response.done" })));
const sessionUpdateCount = parseSent(socket).filter(
(event) => event.type === "session.update",
@@ -2656,7 +2661,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
"message",
Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })),
);
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
expect(parseSent(socket).slice(-1)[0]?.type).toBe("conversation.item.create");
@@ -2706,7 +2711,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => {
socket.emit("message", Buffer.from(JSON.stringify({ type: "session.updated" })));
await connecting;
bridge.submitToolResult("call_1", { text: "done" });
void bridge.submitToolResult("call_1", { text: "done" });
const responseCreateEvent = parseSent(socket).findLast(
(event) => event.type === "response.create",
);

View File

@@ -471,6 +471,7 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge {
private static readonly BASE_RECONNECT_DELAY_MS = 1000;
private static readonly CONNECT_TIMEOUT_MS = 10_000;
readonly supportsToolResultContinuation = true;
readonly supportsToolResultSuppression = true;
private ws: WebSocket | null = null;
private connected = false;

View File

@@ -2,6 +2,7 @@
import http from "node:http";
import type {
RealtimeVoiceBridge,
RealtimeVoiceForcedConsultCoordinator,
RealtimeVoiceProviderPlugin,
RealtimeVoiceToolCallEvent,
} from "openclaw/plugin-sdk/realtime-voice";
@@ -1036,8 +1037,45 @@ describe("RealtimeCallHandler path routing", () => {
}
| undefined;
let resolveConsult: ((value: unknown) => void) | undefined;
let resolveWorkingSubmission: (() => void) | undefined;
let rejectWorkingSubmission = false;
const resolveFinalSubmissions: Array<() => void> = [];
let receivedPartialTranscript: string | undefined;
const submitToolResult = vi.fn();
const submitToolResult = vi.fn(
(_callId: string, result: unknown, _options?: unknown): void | Promise<void> => {
if (
rejectWorkingSubmission &&
result &&
typeof result === "object" &&
"status" in result &&
result.status === "working"
) {
return Promise.reject(new Error("working result rejected"));
}
if (
_callId === "consult-call" &&
result &&
typeof result === "object" &&
"status" in result &&
result.status === "working"
) {
return new Promise<void>((resolve) => {
resolveWorkingSubmission = resolve;
});
}
if (
result &&
typeof result === "object" &&
"text" in result &&
result.text === "The basement lights are on."
) {
return new Promise<void>((resolve) => {
resolveFinalSubmissions.push(resolve);
});
}
return undefined;
},
);
const bridge = makeBridge({
supportsToolResultContinuation: true,
submitToolResult,
@@ -1048,33 +1086,35 @@ describe("RealtimeCallHandler path routing", () => {
return bridge;
},
);
const getCallByProviderCallId = vi.fn(
(): CallRecord => ({
callId: "call-1",
providerCallId: "CA-tool",
provider: "twilio",
direction: "inbound",
state: "ringing",
from: "+15550001234",
to: "+15550009999",
startedAt: Date.now(),
transcript: [],
processedEventIds: [],
metadata: {},
}),
);
const call: CallRecord = {
callId: "call-1",
providerCallId: "CA-tool",
provider: "twilio",
direction: "inbound",
state: "ringing",
from: "+15550001234",
to: "+15550009999",
startedAt: Date.now(),
transcript: [],
processedEventIds: [],
metadata: {},
};
const getCallByProviderCallId = vi.fn((): CallRecord => call);
const handler = makeHandler(undefined, {
manager: {
getCallByProviderCallId,
},
realtimeProvider: makeRealtimeProvider(createBridge),
});
handler.registerToolHandler("openclaw_agent_consult", (_args, _callId, context) => {
receivedPartialTranscript = context.partialUserTranscript;
return new Promise((resolve) => {
resolveConsult = resolve;
});
});
const consultHandler = vi.fn(
(_args: unknown, _callId: string, context: { partialUserTranscript?: string }) => {
receivedPartialTranscript = context.partialUserTranscript;
return new Promise((resolve) => {
resolveConsult = resolve;
});
},
);
handler.registerToolHandler("openclaw_agent_consult", consultHandler);
handler.registerToolHandler("custom_lookup", async () => ({ ok: true }));
const server = await startRealtimeServer(handler);
@@ -1099,6 +1139,14 @@ describe("RealtimeCallHandler path routing", () => {
name: "openclaw_agent_consult",
args: { question: "Are the basement lights on?" },
});
callbacks?.onToolCall?.({
itemId: "item-2",
callId: "consult-call-2",
name: "openclaw_agent_consult",
args: { question: "Are the basement lights on?" },
});
expect(receivedPartialTranscript).toBeUndefined();
resolveWorkingSubmission?.();
await vi.advanceTimersByTimeAsync(350);
await waitForRealtimeTest(() => {
expect(receivedPartialTranscript).toBe("Are the basement");
@@ -1117,19 +1165,35 @@ describe("RealtimeCallHandler path routing", () => {
expect(typeof payload?.message).toBe("string");
expect(workingCall[2]).toEqual({ willContinue: true });
});
expect(submitToolResult).toHaveBeenCalledTimes(1);
expect(
submitToolResult.mock.calls.filter(
([, result]) =>
result &&
typeof result === "object" &&
"status" in result &&
result.status === "working",
),
).toHaveLength(2);
resolveConsult?.({ text: "The basement lights are on." });
await waitForRealtimeTest(() => {
expect(submitToolResult).toHaveBeenLastCalledWith(
"consult-call",
"consult-call-2",
{
text: "The basement lights are on.",
},
undefined,
);
});
expect(recentTalkEvents(call).some((event) => event.type === "tool.result")).toBe(false);
for (const resolve of resolveFinalSubmissions) {
resolve();
}
await waitForRealtimeTest(() => {
expect(recentTalkEvents(call).some((event) => event.type === "tool.result")).toBe(true);
});
expect(consultHandler).toHaveBeenCalledTimes(1);
submitToolResult.mockClear();
callbacks?.onToolCall?.({
@@ -1147,6 +1211,25 @@ describe("RealtimeCallHandler path routing", () => {
);
expect(customCallResults).toHaveLength(1);
expect(customCallResults[0]?.[2]).toBeUndefined();
submitToolResult.mockClear();
rejectWorkingSubmission = true;
callbacks?.onToolCall?.({
itemId: "item-rejected",
callId: "consult-rejected",
name: "openclaw_agent_consult",
args: { question: "Do not run this twice" },
});
await waitForRealtimeTest(() => {
expect(submitToolResult).toHaveBeenCalledTimes(1);
});
await vi.advanceTimersByTimeAsync(0);
expect(consultHandler).toHaveBeenCalledTimes(1);
expect(submitToolResult).toHaveBeenCalledWith(
"consult-rejected",
expect.objectContaining({ status: "working" }),
{ willContinue: true },
);
} finally {
vi.useRealTimers();
if (ws.readyState !== WebSocket.CLOSED && ws.readyState !== WebSocket.CLOSING) {
@@ -1158,6 +1241,97 @@ describe("RealtimeCallHandler path routing", () => {
}
});
it("terminally satisfies a late native call for a cancelled forced consult", async () => {
let callbacks:
| {
onToolCall?: (event: {
itemId: string;
callId: string;
name: string;
args: unknown;
}) => void;
}
| undefined;
const submitToolResult = vi.fn();
const createBridge = vi.fn(
(request: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0]) => {
callbacks = request;
return makeBridge({ submitToolResult });
},
);
const call: CallRecord = {
callId: "call-1",
providerCallId: "CA-cancelled-consult",
provider: "twilio",
direction: "inbound",
state: "ringing",
from: "+15550001234",
to: "+15550009999",
startedAt: Date.now(),
transcript: [],
processedEventIds: [],
metadata: {},
};
const handler = makeHandler(undefined, {
manager: { getCallByProviderCallId: vi.fn(() => call) },
realtimeProvider: makeRealtimeProvider(createBridge),
});
const consult = vi.fn(async () => ({ text: "should not run" }));
handler.registerToolHandler("openclaw_agent_consult", consult);
const coordinator = (
handler as unknown as {
forcedConsultCoordinator(callId: string): RealtimeVoiceForcedConsultCoordinator;
}
).forcedConsultCoordinator(call.callId);
const cancelled = coordinator.prepare("cancelled question");
if (!cancelled) {
throw new Error("expected forced consult handle");
}
coordinator.markStarted(cancelled);
coordinator.markCancelled(cancelled);
const server = await startRealtimeServer(handler);
try {
const ws = await connectWs(server.url);
try {
ws.send(
JSON.stringify({
event: "start",
start: { streamSid: "MZ-cancelled-consult", callSid: call.providerCallId },
}),
);
await waitForRealtimeTest(() => {
expect(createBridge).toHaveBeenCalled();
});
callbacks?.onToolCall?.({
itemId: "item-cancelled",
callId: "native-cancelled",
name: "openclaw_agent_consult",
args: { question: "cancelled question" },
});
await waitForRealtimeTest(() => {
expect(submitToolResult).toHaveBeenCalledWith(
"native-cancelled",
{
status: "cancelled",
message: "OpenClaw cancelled this consult before completion. Do not restart it.",
},
undefined,
);
});
expect(consult).not.toHaveBeenCalled();
} finally {
if (ws.readyState !== WebSocket.CLOSED && ws.readyState !== WebSocket.CLOSING) {
ws.close();
}
}
} finally {
await server.close();
}
});
it("forces an agent consult from final user transcript when consult policy is always", async () => {
let callbacks:
| {

View File

@@ -849,7 +849,7 @@ export class RealtimeCallHandler {
console.log(
`[voice-call] realtime tool call received callId=${callId} providerCallId=${callSid} tool=${toolEvent.name}`,
);
void this.executeToolCall(
return this.executeToolCall(
sessionLocal,
callId,
toolEvent.callId || toolEvent.itemId,
@@ -1347,28 +1347,28 @@ export class RealtimeCallHandler {
final: true,
});
};
const submitFinalToolResult = (result: unknown): void => {
bridge.submitToolResult(bridgeCallId, result);
const submitFinalToolResult = async (result: unknown): Promise<void> => {
await bridge.submitToolResult(bridgeCallId, result);
emitFinalToolEvent(result);
};
const submitWorkingResponse = () => {
const submitWorkingResponse = async (): Promise<void> => {
if (
handler &&
name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME &&
bridge.bridge.supportsToolResultContinuation &&
!this.config.fastContext.enabled
) {
await bridge.submitToolResult(
bridgeCallId,
buildRealtimeVoiceAgentConsultWorkingResponse("caller"),
{ willContinue: true },
);
emitTalkEvent?.({
type: "tool.progress",
turnId,
callId: bridgeCallId,
payload: { name, status: "working" },
});
bridge.submitToolResult(
bridgeCallId,
buildRealtimeVoiceAgentConsultWorkingResponse("caller"),
{ willContinue: true },
);
}
};
if (name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) {
@@ -1381,9 +1381,19 @@ export class RealtimeCallHandler {
}
}
const forcedConsult = this.forcedConsultsByCallId.get(callId);
if (forcedMatch.kind === "already_delivered" && coordinator.isCancelled(forcedMatch.handle)) {
if (forcedConsult) {
forcedConsult.sendSpeechPrompt = false;
}
await submitFinalToolResult({
status: "cancelled",
message: "OpenClaw cancelled this consult before completion. Do not restart it.",
});
return;
}
if (forcedConsult) {
if (forcedConsult.completedAt || forcedMatch.kind === "already_delivered") {
submitFinalToolResult({
await submitFinalToolResult({
status: "already_delivered",
message: "OpenClaw already delivered this consult result internally. Do not repeat it.",
});
@@ -1393,7 +1403,7 @@ export class RealtimeCallHandler {
const result = await forcedConsult.promise.catch((error: unknown) => ({
error: formatErrorMessage(error),
}));
submitFinalToolResult(result);
await submitFinalToolResult(result);
return;
}
@@ -1402,32 +1412,36 @@ export class RealtimeCallHandler {
console.log(
`[voice-call] realtime tool call sharing in-flight agent consult callId=${callId} ageMs=${Date.now() - existingNativeConsult.startedAt}`,
);
submitWorkingResponse();
submitFinalToolResult(await existingNativeConsult.promise);
await submitWorkingResponse();
await submitFinalToolResult(await existingNativeConsult.promise);
return;
}
submitWorkingResponse();
const state: NativeConsultState = {
startedAt,
promise: Promise.resolve(),
};
state.promise = (async () => {
await this.waitForConsultTranscriptSettle(callId, startedAt);
const context = {
partialUserTranscript: this.resolveUserTranscriptContext(callId),
};
state.partialUserTranscript = context.partialUserTranscript;
const handlerArgs = withFallbackConsultQuestion(args, context.partialUserTranscript);
console.log(
`[voice-call] realtime tool call executing callId=${callId} tool=${name} hasHandler=${Boolean(handler)}`,
);
return !handler
? { error: `Tool "${name}" not available` }
: await handler(handlerArgs, callId, context);
})().catch((error: unknown) => ({
error: formatErrorMessage(error),
}));
const workingSubmission = submitWorkingResponse();
state.promise = workingSubmission.then(async () => {
try {
await this.waitForConsultTranscriptSettle(callId, startedAt);
const context = {
partialUserTranscript: this.resolveUserTranscriptContext(callId),
};
state.partialUserTranscript = context.partialUserTranscript;
const handlerArgs = withFallbackConsultQuestion(args, context.partialUserTranscript);
console.log(
`[voice-call] realtime tool call executing callId=${callId} tool=${name} hasHandler=${Boolean(handler)}`,
);
return !handler
? { error: `Tool "${name}" not available` }
: await handler(handlerArgs, callId, context);
} catch (error) {
return {
error: formatErrorMessage(error),
};
}
});
this.nativeConsultsInFlightByCallId.set(callId, state);
try {
const result = await state.promise;
@@ -1442,7 +1456,7 @@ export class RealtimeCallHandler {
console.log(
`[voice-call] realtime tool call completed callId=${callId} tool=${name} status=${status} elapsedMs=${Date.now() - startedAt}${error ? ` error=${error}` : ""}`,
);
submitFinalToolResult(result);
await submitFinalToolResult(result);
if (status === "ok") {
this.consumePartialUserTranscript(callId, state.partialUserTranscript);
}
@@ -1479,7 +1493,7 @@ export class RealtimeCallHandler {
console.log(
`[voice-call] realtime tool call completed callId=${callId} tool=${name} status=${status} elapsedMs=${Date.now() - startedAt}${error ? ` error=${error}` : ""}`,
);
submitFinalToolResult(result);
await submitFinalToolResult(result);
if (name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME && status === "ok") {
this.consumePartialUserTranscript(callId, context.partialUserTranscript);
}

View File

@@ -623,7 +623,7 @@ export const talkSessionHandlers: GatewayRequestHandlers = {
return;
}
const connId = requireUnifiedTalkSessionConn(session, client?.connId);
submitTalkRealtimeRelayToolResult({
await submitTalkRealtimeRelayToolResult({
relaySessionId: session.relaySessionId,
connId,
callId: params.callId,

View File

@@ -1521,8 +1521,14 @@ describe("talk.session unified handlers", () => {
reason: "barge-in",
});
let acceptToolResult!: () => void;
mocks.submitTalkRealtimeRelayToolResult.mockReturnValueOnce(
new Promise<void>((resolve) => {
acceptToolResult = resolve;
}),
);
const toolRespond = vi.fn();
await talkHandlers["talk.session.submitToolResult"]({
const toolRequest = talkHandlers["talk.session.submitToolResult"]({
req: { type: "req", id: "4", method: "talk.session.submitToolResult" },
params: {
sessionId: "relay-unified-1",
@@ -1535,6 +1541,9 @@ describe("talk.session unified handlers", () => {
respond: toolRespond as never,
context: {} as never,
});
expect(toolRespond).not.toHaveBeenCalled();
acceptToolResult();
await toolRequest;
expect(mocks.submitTalkRealtimeRelayToolResult).toHaveBeenCalledWith({
relaySessionId: "relay-unified-1",
connId: "conn-1",
@@ -1542,6 +1551,28 @@ describe("talk.session unified handlers", () => {
result: { status: "working" },
options: { suppressResponse: true, willContinue: true },
});
expectRespondOk(toolRespond, { ok: true });
mocks.submitTalkRealtimeRelayToolResult.mockRejectedValueOnce(
new Error("provider rejected tool result"),
);
const rejectedToolRespond = vi.fn();
await talkHandlers["talk.session.submitToolResult"]({
req: { type: "req", id: "4-rejected", method: "talk.session.submitToolResult" },
params: {
sessionId: "relay-unified-1",
callId: "call-rejected",
result: { ok: true },
},
client: { connId: "conn-1" } as never,
isWebchatConnect: () => false,
respond: rejectedToolRespond as never,
context: {} as never,
});
expectRespondError(rejectedToolRespond, {
code: ErrorCodes.UNAVAILABLE,
message: "Error: provider rejected tool result",
});
const steerRespond = vi.fn();
await talkHandlers["talk.session.steer"]({

File diff suppressed because it is too large Load Diff

View File

@@ -8,6 +8,7 @@ import {
REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME,
buildRealtimeVoiceAgentConsultWorkingResponse,
} from "../talk/agent-consult-tool.js";
import { buildRealtimeVoiceAgentCancelProviderResult } from "../talk/agent-run-control-shared.js";
import {
buildRealtimeVoiceAgentControlSpeechMessage,
controlRealtimeVoiceAgentRun,
@@ -18,6 +19,7 @@ import { readSpeakableRealtimeVoiceToolResult } from "../talk/consult-question.j
import {
createRealtimeVoiceForcedConsultCoordinator,
type RealtimeVoiceForcedConsultCoordinator,
type RealtimeVoiceForcedConsultHandle,
} from "../talk/forced-consult-coordinator.js";
import { recordTalkObservabilityEvent } from "../talk/observability.js";
import {
@@ -104,6 +106,18 @@ type TalkRealtimeRelayEventPayload =
type TalkRealtimeRelayEvent = TalkRealtimeRelayEventPayload & { talkEvent?: TalkEvent };
type ForcedTerminalProviderResult = {
result: unknown;
options?: RealtimeVoiceToolResultOptions;
turnId: string;
epoch: number;
};
type RelayAgentControlProviderSubmission = {
completion?: Promise<void>;
providerResponseStarted: boolean;
};
type RelaySession = {
id: string;
connId: string;
@@ -116,6 +130,22 @@ type RelaySession = {
activeAgentRuns: Map<string, string>;
activeAgentToolCalls: Map<string, string>;
completedAgentToolCalls: Set<string>;
// Cancelled calls retain their original turn long enough to terminally satisfy
// late browser results without creating a replacement turn or owner success event.
cancelledAgentToolCalls: Map<string, string>;
pendingFinalToolResults: Map<string, Promise<void>>;
// Provider acceptance survives partial retries independently from the owner-facing
// agent-call lifecycle, so accepted native ids are never submitted twice.
completedProviderToolResults: Set<string>;
pendingProviderToolResults: Map<string, Promise<void>>;
// A final result must wait until the provider accepts its continuation result;
// otherwise async bridges can observe final-before-working ordering.
pendingWorkingToolResults: Map<string, Promise<void>>;
// Keep a forced terminal result open while late matching native ids join it.
// Delivery/cancellation closes the state only after every current id accepts.
forcedTerminalProviderResults: Map<string, ForcedTerminalProviderResult>;
// Turn cancellation invalidates async acceptance callbacks from the prior turn.
toolResultEpoch: number;
forcedConsults: RealtimeVoiceForcedConsultCoordinator;
transcript: RealtimeVoiceTranscriptEntry[];
};
@@ -233,6 +263,14 @@ function buildAlreadyDeliveredToolResult(): Record<string, string> {
};
}
function suppressedToolResultOptions(
session: RelaySession,
): RealtimeVoiceToolResultOptions | undefined {
return session.bridge.bridge.supportsToolResultSuppression === false
? undefined
: { suppressResponse: true };
}
function cancelForcedConsults(session: RelaySession): void {
for (const handle of session.forcedConsults.handles()) {
session.forcedConsults.markCancelled(handle);
@@ -311,39 +349,223 @@ function broadcastToolResultToOwner(
});
}
function completeAfterToolResultSubmissions(
session: RelaySession,
submissions: Array<void | Promise<void>>,
onAccepted: () => void,
): void | Promise<void> {
const pending = submissions.filter(
(submission): submission is Promise<void> => submission !== undefined,
);
const complete = () => {
if (relaySessions.get(session.id) === session) {
onAccepted();
}
};
if (pending.length === 0) {
complete();
return;
}
return Promise.all(pending).then(complete);
}
function submitFinalProviderToolResult(params: {
session: RelaySession;
callId: string;
result: unknown;
options?: RealtimeVoiceToolResultOptions;
onAccepted?: () => void;
}): void | Promise<void> {
if (params.session.completedProviderToolResults.has(params.callId)) {
if (relaySessions.get(params.session.id) === params.session) {
params.onAccepted?.();
}
return;
}
const pending = params.session.pendingProviderToolResults.get(params.callId);
if (pending) {
return pending;
}
const submit = () =>
params.session.bridge.submitToolResult(params.callId, params.result, params.options);
const working = params.session.pendingWorkingToolResults.get(params.callId);
const epoch = params.session.toolResultEpoch;
const submitAfterWorking = async () => {
if (relaySessions.get(params.session.id) !== params.session) {
return false;
}
if (params.session.toolResultEpoch !== epoch) {
if (!params.session.cancelledAgentToolCalls.has(params.callId)) {
return false;
}
// The browser already considers this final submitted while it waits behind
// the provider's working-result acknowledgement. Finish the cancelled call
// here so the provider is not left waiting for a terminal result.
await params.session.bridge.submitToolResult(
params.callId,
buildRealtimeVoiceAgentCancelProviderResult(
"OpenClaw cancelled this consult before completion. Do not restart it.",
),
suppressedToolResultOptions(params.session),
);
params.session.completedProviderToolResults.add(params.callId);
params.session.cancelledAgentToolCalls.delete(params.callId);
params.session.completedAgentToolCalls.add(params.callId);
return false;
}
await submit();
return true;
};
const submission = working ? working.then(submitAfterWorking, submitAfterWorking) : submit();
const accept = () => {
params.session.completedProviderToolResults.add(params.callId);
if (relaySessions.get(params.session.id) === params.session) {
params.onAccepted?.();
}
};
if (!submission) {
accept();
return;
}
const tracked = submission
.then((submitted) => {
if (submitted !== false) {
accept();
}
})
.finally(() => {
if (params.session.pendingProviderToolResults.get(params.callId) === tracked) {
params.session.pendingProviderToolResults.delete(params.callId);
}
});
params.session.pendingProviderToolResults.set(params.callId, tracked);
return tracked;
}
function trackAgentFinalToolResult(
session: RelaySession,
callId: string,
completion: void | Promise<void>,
): void | Promise<void> {
if (!completion) {
return;
}
const tracked = completion.finally(() => {
if (session.pendingFinalToolResults.get(callId) === tracked) {
session.pendingFinalToolResults.delete(callId);
}
});
session.pendingFinalToolResults.set(callId, tracked);
return tracked;
}
function trackPendingWorkingToolResult(
session: RelaySession,
callId: string,
completion: void | Promise<void>,
): void | Promise<void> {
if (!completion) {
return;
}
const tracked = completion.finally(() => {
if (session.pendingWorkingToolResults.get(callId) === tracked) {
session.pendingWorkingToolResults.delete(callId);
}
});
session.pendingWorkingToolResults.set(callId, tracked);
return tracked;
}
function clearRelayAgentToolCall(session: RelaySession, callId: string): void {
const runId = session.activeAgentToolCalls.get(callId);
session.activeAgentToolCalls.delete(callId);
if (!runId) {
return;
}
const runStillActive = [...session.activeAgentToolCalls.values()].includes(runId);
if (!runStillActive) {
session.activeAgentRuns.delete(runId);
}
}
function submitRelayAgentControlProviderResults(
session: RelaySession,
result: RealtimeVoiceAgentControlResult,
turnId: string,
): void {
): RelayAgentControlProviderSubmission | undefined {
if (result.mode !== "cancel" || !result.ok || !result.providerResult) {
return;
return undefined;
}
const activeCallIds = [...session.activeAgentToolCalls.keys()];
for (const callId of activeCallIds) {
const forcedConsult = session.forcedConsults.handles().find((handle) => handle.id === callId);
const providerResult = result.providerResult;
const epoch = session.toolResultEpoch;
const callIds = [...session.activeAgentToolCalls.keys()];
const activeCallIds = callIds.filter((callId) => !session.pendingFinalToolResults.has(callId));
const submissions: Array<void | Promise<void>> = callIds
.map((callId) => session.pendingFinalToolResults.get(callId))
.filter((pending): pending is Promise<void> => pending !== undefined);
const toolResultOptions = suppressedToolResultOptions(session);
let providerResponseStarted = toolResultOptions === undefined && submissions.length > 0;
const finalizeAgentCall = (callId: string, forcedConsult?: RealtimeVoiceForcedConsultHandle) => {
if (session.toolResultEpoch !== epoch) {
return;
}
if (forcedConsult) {
// Forced consults may have both synthetic and provider-native call ids;
// cancelling must satisfy every native id or the provider keeps waiting.
session.forcedConsults.markCancelled(forcedConsult);
for (const nativeCallId of session.forcedConsults.nativeCallIds(forcedConsult)) {
session.bridge.submitToolResult(nativeCallId, result.providerResult, {
suppressResponse: true,
});
}
} else {
session.bridge.submitToolResult(callId, result.providerResult, { suppressResponse: true });
}
broadcastToolResultToOwner(session, {
callId,
turnId,
result: result.providerResult,
result: providerResult,
final: true,
});
session.activeAgentToolCalls.delete(callId);
clearRelayAgentToolCall(session, callId);
session.completedAgentToolCalls.add(callId);
};
for (const callId of activeCallIds) {
const forcedConsult = session.forcedConsults.handles().find((handle) => handle.id === callId);
if (forcedConsult) {
const nativeCallIds = session.forcedConsults.nativeCallIds(forcedConsult);
providerResponseStarted ||= toolResultOptions === undefined && nativeCallIds.length > 0;
const terminal: ForcedTerminalProviderResult = {
result: providerResult,
options: toolResultOptions,
turnId,
epoch,
};
session.forcedTerminalProviderResults.set(callId, terminal);
const clearTerminal = () => {
if (session.forcedTerminalProviderResults.get(callId) === terminal) {
session.forcedTerminalProviderResults.delete(callId);
}
};
const drained = drainForcedTerminalProviderResultsAfterPending(
session,
forcedConsult,
terminal,
);
const completed = completeAfterToolResultSubmissions(session, [drained], () => {
clearTerminal();
finalizeAgentCall(callId, forcedConsult);
});
const tracked = trackAgentFinalToolResult(session, callId, completed?.finally(clearTerminal));
submissions.push(tracked);
continue;
}
providerResponseStarted ||= toolResultOptions === undefined;
const submitted = submitFinalProviderToolResult({
session,
callId,
result: providerResult,
options: toolResultOptions,
onAccepted: () => finalizeAgentCall(callId),
});
submissions.push(trackAgentFinalToolResult(session, callId, submitted));
}
session.activeAgentRuns.clear();
const completion = completeAfterToolResultSubmissions(session, submissions, () => {});
return {
...(completion ? { completion } : {}),
providerResponseStarted,
};
}
function closeRelaySession(session: RelaySession, reason: "completed" | "error"): void {
@@ -582,7 +804,7 @@ export function createTalkRealtimeRelaySession(
},
onToolCall: (toolCall) => {
const relay = relayRef.current;
const turnId = relay ? ensureRelayTurn(relay) : undefined;
let shouldSubmitWorkingResult = false;
if (relay && toolCall.name === REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) {
const forcedConsult = relay.forcedConsults.recordNativeConsult(
toolCall.args,
@@ -590,14 +812,26 @@ export function createTalkRealtimeRelaySession(
);
if (forcedConsult.kind === "in_flight" || forcedConsult.kind === "already_delivered") {
if (forcedConsult.kind === "already_delivered") {
submitAlreadyDeliveredToolResult(relay, toolCall.callId, turnId);
} else {
submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId, turnId);
const result = relay.forcedConsults.isCancelled(forcedConsult.handle)
? buildRealtimeVoiceAgentCancelProviderResult(
"OpenClaw cancelled this consult before completion. Do not restart it.",
)
: buildAlreadyDeliveredToolResult();
return submitForcedConsultProviderResult(
relay,
toolCall.callId,
result,
suppressedToolResultOptions(relay),
);
}
return;
if (relay.forcedTerminalProviderResults.has(forcedConsult.handle.id)) {
return relay.pendingFinalToolResults.get(forcedConsult.handle.id);
}
return submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId);
}
submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId, turnId);
shouldSubmitWorkingResult = true;
}
const turnId = relay ? ensureRelayTurn(relay) : undefined;
emit(
{
relaySessionId,
@@ -615,6 +849,9 @@ export function createTalkRealtimeRelaySession(
payload: { name: toolCall.name, args: toolCall.args },
},
);
if (relay && shouldSubmitWorkingResult) {
return submitRealtimeAgentConsultWorkingResponse(relay, toolCall.callId, turnId);
}
},
onReady: () => {
ready = true;
@@ -680,6 +917,13 @@ export function createTalkRealtimeRelaySession(
activeAgentRuns: new Map(),
activeAgentToolCalls: new Map(),
completedAgentToolCalls: new Set(),
cancelledAgentToolCalls: new Map(),
pendingFinalToolResults: new Map(),
completedProviderToolResults: new Set(),
pendingProviderToolResults: new Map(),
pendingWorkingToolResults: new Map(),
forcedTerminalProviderResults: new Map(),
toolResultEpoch: 0,
forcedConsults: createRealtimeVoiceForcedConsultCoordinator(),
transcript: [],
};
@@ -770,49 +1014,97 @@ function scheduleForcedAgentConsult(session: RelaySession | undefined, question:
});
}
function submitAlreadyDeliveredToolResult(
function submitForcedConsultProviderResult(
session: RelaySession,
callId: string,
turnId = ensureRelayTurn(session),
): void {
const result = buildAlreadyDeliveredToolResult();
session.bridge.submitToolResult(callId, result, { suppressResponse: true });
broadcastToOwner(session.context, session.connId, {
relaySessionId: session.id,
type: "toolResult",
result: unknown,
options: RealtimeVoiceToolResultOptions | undefined,
): void | Promise<void> {
return submitFinalProviderToolResult({
session,
callId,
talkEvent: session.talk.emit({
type: "tool.result",
callId,
turnId,
payload: { name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, result },
final: true,
}),
result,
options,
});
}
function drainForcedTerminalProviderResults(
session: RelaySession,
handle: RealtimeVoiceForcedConsultHandle,
terminal: ForcedTerminalProviderResult,
): void | Promise<void> {
if (session.forcedTerminalProviderResults.get(handle.id) !== terminal) {
return;
}
const submissions = session.forcedConsults
.nativeCallIds(handle)
.map((callId) =>
submitForcedConsultProviderResult(session, callId, terminal.result, terminal.options),
);
const pending = submissions.filter(
(submission): submission is Promise<void> => submission !== undefined,
);
if (pending.length > 0) {
return Promise.all(pending).then(() =>
drainForcedTerminalProviderResults(session, handle, terminal),
);
}
const hasUnsubmittedCall = session.forcedConsults
.nativeCallIds(handle)
.some((callId) => !session.completedProviderToolResults.has(callId));
if (hasUnsubmittedCall) {
return drainForcedTerminalProviderResults(session, handle, terminal);
}
}
function drainForcedTerminalProviderResultsAfterPending(
session: RelaySession,
handle: RealtimeVoiceForcedConsultHandle,
terminal: ForcedTerminalProviderResult,
): void | Promise<void> {
const pending = session.forcedConsults
.nativeCallIds(handle)
.map((callId) => session.pendingProviderToolResults.get(callId))
.filter((submission): submission is Promise<void> => submission !== undefined);
if (pending.length === 0) {
return drainForcedTerminalProviderResults(session, handle, terminal);
}
return Promise.allSettled(pending).then(() =>
drainForcedTerminalProviderResults(session, handle, terminal),
);
}
function submitRealtimeAgentConsultWorkingResponse(
session: RelaySession,
callId: string,
turnId = ensureRelayTurn(session),
): void {
): void | Promise<void> {
if (!session.bridge.bridge.supportsToolResultContinuation) {
return;
}
session.bridge.submitToolResult(callId, buildRealtimeVoiceAgentConsultWorkingResponse("person"), {
willContinue: true,
});
broadcastToOwner(session.context, session.connId, {
relaySessionId: session.id,
type: "toolResult",
const epoch = session.toolResultEpoch;
const submission = session.bridge.submitToolResult(
callId,
talkEvent: session.talk.emit({
type: "tool.progress",
buildRealtimeVoiceAgentConsultWorkingResponse("person"),
{ willContinue: true },
);
const completion = completeAfterToolResultSubmissions(session, [submission], () => {
if (session.toolResultEpoch !== epoch) {
return;
}
broadcastToOwner(session.context, session.connId, {
relaySessionId: session.id,
type: "toolResult",
callId,
turnId,
payload: { name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, status: "working" },
}),
talkEvent: session.talk.emit({
type: "tool.progress",
callId,
turnId,
payload: { name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME, status: "working" },
}),
});
});
return trackPendingWorkingToolResult(session, callId, completion);
}
function ensureRelayTurn(session: RelaySession): string {
@@ -874,64 +1166,200 @@ export function submitTalkRealtimeRelayToolResult(params: {
callId: string;
result: unknown;
options?: RealtimeVoiceToolResultOptions;
}): void {
}): void | Promise<void> {
const session = getRelaySession(params.relaySessionId, params.connId);
if (session.completedAgentToolCalls.has(params.callId)) {
return;
}
const pendingFinal = session.pendingFinalToolResults.get(params.callId);
const cancelledAgentCall = session.cancelledAgentToolCalls.has(params.callId);
if (pendingFinal && !cancelledAgentCall) {
return pendingFinal;
}
const forcedConsult = session.forcedConsults
.handles()
.find((handle) => handle.id === params.callId);
if (forcedConsult) {
const turnId = ensureRelayTurn(session);
const cancelled = session.forcedConsults.isCancelled(forcedConsult);
const turnId = cancelled
? (session.cancelledAgentToolCalls.get(params.callId) ?? session.talk.activeTurnId)
: ensureRelayTurn(session);
if (!turnId) {
throw new Error("Cancelled realtime consult is missing its original turn");
}
if (cancelled) {
if (params.options?.willContinue !== true) {
const providerResult = buildRealtimeVoiceAgentCancelProviderResult(
"OpenClaw cancelled this consult before completion. Do not restart it.",
);
const terminal: ForcedTerminalProviderResult = {
result: providerResult,
options: suppressedToolResultOptions(session),
turnId,
epoch: session.toolResultEpoch,
};
session.forcedTerminalProviderResults.set(forcedConsult.id, terminal);
const clearTerminal = () => {
if (session.forcedTerminalProviderResults.get(forcedConsult.id) === terminal) {
session.forcedTerminalProviderResults.delete(forcedConsult.id);
}
};
const drained = drainForcedTerminalProviderResultsAfterPending(
session,
forcedConsult,
terminal,
);
const completion = completeAfterToolResultSubmissions(session, [drained], () => {
clearTerminal();
if (session.toolResultEpoch !== terminal.epoch) {
return;
}
session.forcedConsults.markCancelled(forcedConsult);
}
} else if (isWorkingToolResult(params.result)) {
session.bridge.sendUserMessage(buildForcedConsultCheckingPrompt());
} else {
session.forcedConsults.markDelivered(forcedConsult);
const text = readSpeakableRealtimeVoiceToolResult(params.result, {
maxChars: FORCED_CONSULT_RESULT_MAX_CHARS,
clearRelayAgentToolCall(session, params.callId);
session.cancelledAgentToolCalls.delete(params.callId);
session.completedAgentToolCalls.add(params.callId);
broadcastToolResultToOwner(session, {
callId: params.callId,
turnId,
result: providerResult,
forced: true,
final: true,
});
});
for (const nativeCallId of session.forcedConsults.nativeCallIds(forcedConsult)) {
submitAlreadyDeliveredToolResult(session, nativeCallId, turnId);
}
if (text) {
session.bridge.sendUserMessage(buildForcedConsultSpeechPrompt(text));
}
return trackAgentFinalToolResult(session, params.callId, completion?.finally(clearTerminal));
}
const final = params.options?.willContinue !== true;
if (final && !cancelled && !isWorkingToolResult(params.result)) {
if (!final) {
if (isWorkingToolResult(params.result)) {
session.bridge.sendUserMessage(buildForcedConsultCheckingPrompt());
}
broadcastToolResultToOwner(session, {
callId: params.callId,
turnId,
result: params.result,
forced: true,
final: false,
});
return;
}
const text = readSpeakableRealtimeVoiceToolResult(params.result, {
maxChars: FORCED_CONSULT_RESULT_MAX_CHARS,
});
const providerOptions = suppressedToolResultOptions(session);
const providerResult = providerOptions ? buildAlreadyDeliveredToolResult() : params.result;
const terminal: ForcedTerminalProviderResult = {
result: providerResult,
options: providerOptions,
turnId,
epoch: session.toolResultEpoch,
};
session.forcedTerminalProviderResults.set(forcedConsult.id, terminal);
const submission = drainForcedTerminalProviderResults(session, forcedConsult, terminal);
const clearTerminal = () => {
if (session.forcedTerminalProviderResults.get(forcedConsult.id) === terminal) {
session.forcedTerminalProviderResults.delete(forcedConsult.id);
}
};
const completion = completeAfterToolResultSubmissions(session, [submission], () => {
clearTerminal();
if (session.toolResultEpoch !== terminal.epoch) {
return;
}
session.forcedConsults.markDelivered(forcedConsult);
clearRelayAgentToolCall(session, params.callId);
session.completedAgentToolCalls.add(params.callId);
const hasNativeCalls = session.forcedConsults.nativeCallIds(forcedConsult).length > 0;
if (text && (!hasNativeCalls || providerOptions)) {
session.bridge.sendUserMessage(buildForcedConsultSpeechPrompt(text));
}
broadcastToolResultToOwner(session, {
callId: params.callId,
turnId,
result: params.result,
forced: true,
final: true,
});
});
const trackedCompletion = completion?.finally(clearTerminal);
return trackAgentFinalToolResult(session, params.callId, trackedCompletion);
}
if (cancelledAgentCall) {
const providerResult = buildRealtimeVoiceAgentCancelProviderResult(
"OpenClaw cancelled this consult before completion. Do not restart it.",
);
const submitCancellation = () =>
submitFinalProviderToolResult({
session,
callId: params.callId,
result: providerResult,
options: suppressedToolResultOptions(session),
onAccepted: () => {
session.cancelledAgentToolCalls.delete(params.callId);
session.completedAgentToolCalls.add(params.callId);
},
});
const pendingProvider = session.pendingProviderToolResults.get(params.callId);
const completion = pendingProvider
? pendingProvider.then(submitCancellation, submitCancellation)
: submitCancellation();
return trackAgentFinalToolResult(session, params.callId, completion);
}
if (
params.options?.suppressResponse === true &&
session.bridge.bridge.supportsToolResultSuppression === false
) {
throw new Error("Realtime provider does not support suppressed tool results");
}
// A final result owns provider completion for this call. Follow-up RPCs share it so
// only one accepted submission can clear the linked run and emit the success event.
const final = params.options?.willContinue !== true;
const turnId = ensureRelayTurn(session);
const epoch = session.toolResultEpoch;
const onAccepted = () => {
if (session.toolResultEpoch !== epoch) {
return;
}
if (final) {
clearRelayAgentToolCall(session, params.callId);
session.completedAgentToolCalls.add(params.callId);
}
broadcastToolResultToOwner(session, {
callId: params.callId,
turnId,
result: params.result,
forced: true,
final,
});
return;
}
session.bridge.submitToolResult(params.callId, params.result, params.options);
const turnId = ensureRelayTurn(session);
const final = params.options?.willContinue !== true;
};
if (final) {
const runId = session.activeAgentToolCalls.get(params.callId);
if (runId) {
session.activeAgentRuns.delete(runId);
session.activeAgentToolCalls.delete(params.callId);
}
const completion = submitFinalProviderToolResult({
session,
callId: params.callId,
result: params.result,
options: params.options,
onAccepted,
});
return trackAgentFinalToolResult(session, params.callId, completion);
}
broadcastToolResultToOwner(session, {
callId: params.callId,
turnId,
result: params.result,
final,
});
const submit = () =>
session.bridge.submitToolResult(params.callId, params.result, params.options);
const pendingWorking = session.pendingWorkingToolResults.get(params.callId);
if (pendingWorking) {
const submission = pendingWorking.then(async () => {
if (relaySessions.get(session.id) !== session || session.toolResultEpoch !== epoch) {
return false;
}
await submit();
return true;
});
const completion = submission.then((submitted) => {
if (submitted && relaySessions.get(session.id) === session) {
onAccepted();
}
});
return trackPendingWorkingToolResult(session, params.callId, completion);
}
const submission = submit();
const completion = completeAfterToolResultSubmissions(session, [submission], onAccepted);
return trackPendingWorkingToolResult(session, params.callId, completion);
}
/** Tracks the chat run started for a realtime agent-consult tool call. */
@@ -975,24 +1403,36 @@ export async function steerTalkRealtimeRelayAgentRun(params: {
mode: params.mode,
recentEvents: session.talk.recentEvents,
});
if (relaySessions.get(session.id) !== session) {
throw new Error("Realtime relay session closed while steering the agent run");
}
const turnId = ensureRelayTurn(session);
const providerSubmission = submitRelayAgentControlProviderResults(session, result, turnId);
if (providerSubmission?.completion) {
await providerSubmission.completion;
}
const finalResult = providerSubmission?.providerResponseStarted
? { ...result, suppress: true }
: result;
if (relaySessions.get(session.id) !== session) {
return finalResult;
}
broadcastToOwner(session.context, session.connId, {
relaySessionId: session.id,
type: "toolProgress",
result,
result: finalResult,
talkEvent: session.talk.emit({
type: "tool.progress",
turnId,
payload: {
name: "openclaw_agent_control",
phase: result.mode,
result,
phase: finalResult.mode,
result: finalResult,
},
final: result.mode === "cancel" || result.mode === "status",
final: finalResult.mode === "cancel" || finalResult.mode === "status",
}),
});
submitRelayAgentControlProviderResults(session, result, turnId);
return result;
return finalResult;
}
/** Cancels the active relay turn, aborts agent work, and clears provider audio. */
@@ -1002,9 +1442,22 @@ export function cancelTalkRealtimeRelayTurn(params: {
reason?: string;
}): void {
const session = getRelaySession(params.relaySessionId, params.connId);
session.toolResultEpoch += 1;
session.forcedTerminalProviderResults.clear();
const turnId = ensureRelayTurn(session);
const reason = params.reason ?? "client-cancelled";
cancelForcedConsults(session);
for (const callId of session.activeAgentToolCalls.keys()) {
session.cancelledAgentToolCalls.set(callId, turnId);
}
for (const forcedConsult of session.forcedConsults.handles()) {
if (session.forcedConsults.isCancelled(forcedConsult)) {
session.cancelledAgentToolCalls.set(forcedConsult.id, turnId);
for (const nativeCallId of session.forcedConsults.nativeCallIds(forcedConsult)) {
session.cancelledAgentToolCalls.set(nativeCallId, turnId);
}
}
}
session.bridge.handleBargeIn({ audioPlaybackActive: true });
abortRelayAgentRuns(session, reason);
const cancelled = session.talk.cancelTurn({

View File

@@ -47,6 +47,20 @@ describe("realtime voice forced consult coordinator", () => {
}
});
it("does not retroactively cancel a delivered handle", () => {
const coordinator = createRealtimeVoiceForcedConsultCoordinator();
const handle = coordinator.prepare("Can you check this?", { id: "forced-1" });
coordinator.markStarted(handle!);
coordinator.markDelivered(handle!);
coordinator.markCancelled(handle!);
expect(coordinator.isCancelled(handle!)).toBe(false);
expect(
coordinator.recordNativeConsult({ question: "Can you check this?" }, "native-call"),
).toMatchObject({ kind: "already_delivered", handle: { id: "forced-1" } });
});
it("tracks native-first consults during the dedupe window", () => {
vi.useFakeTimers();
try {
@@ -174,7 +188,7 @@ describe("realtime voice forced consult coordinator", () => {
expect(scheduledDelays).toEqual([MAX_TIMER_TIMEOUT_MS]);
});
it("reports cancelled handles until the dedupe window expires", () => {
it("matches cancelled handles until the dedupe window expires", () => {
vi.useFakeTimers();
try {
const coordinator = createRealtimeVoiceForcedConsultCoordinator();
@@ -183,8 +197,15 @@ describe("realtime voice forced consult coordinator", () => {
coordinator.markCancelled(pending!);
expect(coordinator.isCancelled(pending!)).toBe(true);
expect(
coordinator.recordNativeConsult({ question: "check status" }, "native-call"),
).toMatchObject({ kind: "already_delivered", handle: { id: "forced-1" } });
expect(coordinator.nativeCallIds(pending!)).toEqual(["native-call"]);
vi.advanceTimersByTime(2_001);
expect(coordinator.isCancelled(pending!)).toBe(false);
expect(coordinator.recordNativeConsult({ question: "check status" })).toMatchObject({
kind: "none",
});
} finally {
vi.useRealTimers();
}

View File

@@ -326,13 +326,16 @@ export function createRealtimeVoiceForcedConsultCoordinator<TContext = unknown>(
return { kind: "pending", question, handle: pending.handle };
}
const stored = findMatching(question);
if (!stored || stored.cancelled) {
if (!stored) {
return { kind: "none", question };
}
if (nativeCallId) {
stored.nativeCallIds.add(nativeCallId);
}
rememberStoredQuestion(stored, question);
if (stored.cancelled) {
return { kind: "already_delivered", question, handle: stored.handle };
}
if (stored.delivered) {
return { kind: "already_delivered", question, handle: stored.handle };
}
@@ -363,7 +366,7 @@ export function createRealtimeVoiceForcedConsultCoordinator<TContext = unknown>(
},
markCancelled(handle) {
const stored = getStored(handle);
if (!stored) {
if (!stored || stored.delivered) {
return;
}
clearTimer(stored);

View File

@@ -188,13 +188,23 @@ export type RealtimeVoiceBrowserSession =
export type RealtimeVoiceBridge = {
supportsToolResultContinuation?: boolean;
/** False when the provider cannot accept a tool result without starting a response. */
supportsToolResultSuppression?: boolean;
connect(): Promise<void>;
sendAudio(audio: Buffer): void;
setMediaTimestamp(ts: number): void;
sendUserMessage?(text: string): void;
triggerGreeting?(instructions?: string): void;
handleBargeIn?(options?: RealtimeVoiceBargeInOptions): void;
submitToolResult(callId: string, result: unknown, options?: RealtimeVoiceToolResultOptions): void;
/**
* Returns void when submission completes synchronously, or a Promise that resolves at the
* asynchronous completion boundary exposed by the provider and rejects on submission failure.
*/
submitToolResult(
callId: string,
result: unknown,
options?: RealtimeVoiceToolResultOptions,
): void | Promise<void>;
acknowledgeMark(): void;
close(): void;
isConnected(): boolean;

View File

@@ -227,8 +227,126 @@ describe("realtime voice bridge session runtime", () => {
expect(onToolCall).toHaveBeenCalledWith(event, session);
});
it("forwards tool result continuation options to the provider bridge", () => {
const bridge = makeBridge();
it("routes synchronous and asynchronous tool-call callback failures to onError", async () => {
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
const onError = vi.fn();
const syncFailure = new Error("sync callback failed");
const provider: RealtimeVoiceProviderPlugin = {
id: "test",
label: "Test",
isConfigured: () => true,
createBridge: (request) => {
callbacks = request;
return makeBridge();
},
};
const onToolCall = vi
.fn()
.mockImplementationOnce(() => {
throw syncFailure;
})
.mockImplementationOnce(() => Promise.reject(new Error("async callback failed")));
createRealtimeVoiceBridgeSession({
provider,
providerConfig: {},
audioSink: { sendAudio: vi.fn() },
onToolCall,
onError,
});
const event = {
itemId: "item-1",
callId: "call-1",
name: "lookup",
args: {},
};
callbacks?.onToolCall?.(event);
callbacks?.onToolCall?.(event);
await Promise.resolve();
expect(onError).toHaveBeenNthCalledWith(1, syncFailure);
expect(onError).toHaveBeenNthCalledWith(2, new Error("async callback failed"));
});
it("contains an onError exception after an asynchronous tool-call failure", async () => {
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
const onError = vi.fn(() => {
throw new Error("error callback failed");
});
const provider: RealtimeVoiceProviderPlugin = {
id: "test",
label: "Test",
isConfigured: () => true,
createBridge: (request) => {
callbacks = request;
return makeBridge();
},
};
createRealtimeVoiceBridgeSession({
provider,
providerConfig: {},
audioSink: { sendAudio: vi.fn() },
onToolCall: () => Promise.reject(new Error("tool callback failed")),
onError,
});
callbacks?.onToolCall?.({
itemId: "item-1",
callId: "call-1",
name: "lookup",
args: {},
});
await new Promise<void>((resolve) => {
setImmediate(resolve);
});
expect(onError).toHaveBeenCalledWith(new Error("tool callback failed"));
});
it("does not report an asynchronous tool-call failure after the session closes", async () => {
let callbacks: Parameters<RealtimeVoiceProviderPlugin["createBridge"]>[0] | undefined;
let rejectToolCall: ((error: Error) => void) | undefined;
const close = vi.fn();
const bridge = makeBridge({ close });
const onError = vi.fn();
const provider: RealtimeVoiceProviderPlugin = {
id: "test",
label: "Test",
isConfigured: () => true,
createBridge: (request) => {
callbacks = request;
return bridge;
},
};
const session = createRealtimeVoiceBridgeSession({
provider,
providerConfig: {},
audioSink: { sendAudio: vi.fn() },
onToolCall: () =>
new Promise<void>((_resolve, reject) => {
rejectToolCall = reject;
}),
onError,
});
callbacks?.onToolCall?.({
itemId: "item-1",
callId: "call-1",
name: "lookup",
args: {},
});
session.close();
rejectToolCall?.(new Error("late tool callback failure"));
await Promise.resolve();
expect(close).toHaveBeenCalledTimes(1);
expect(onError).not.toHaveBeenCalled();
});
it("forwards tool result continuation options and async acceptance to the provider bridge", () => {
const acceptance = Promise.resolve();
const submitToolResult = vi.fn(() => acceptance);
const bridge = makeBridge({ submitToolResult });
const provider: RealtimeVoiceProviderPlugin = {
id: "test",
label: "Test",
@@ -241,13 +359,42 @@ describe("realtime voice bridge session runtime", () => {
audioSink: { sendAudio: vi.fn() },
});
session.submitToolResult("call-1", { status: "working" }, { willContinue: true });
expect(bridge["submitToolResult"]).toHaveBeenCalledWith(
const submitted = session.submitToolResult(
"call-1",
{ status: "working" },
{ willContinue: true },
);
expect(submitted).toBe(acceptance);
expect(submitToolResult).toHaveBeenCalledWith(
"call-1",
{ status: "working" },
{ willContinue: true },
);
});
it("rejects suppressed results before calling an unsupported provider bridge", () => {
const submitToolResult = vi.fn();
const bridge = makeBridge({
submitToolResult,
supportsToolResultSuppression: false,
});
const provider: RealtimeVoiceProviderPlugin = {
id: "test",
label: "Test",
isConfigured: () => true,
createBridge: () => bridge,
};
const session = createRealtimeVoiceBridgeSession({
provider,
providerConfig: {},
audioSink: { sendAudio: vi.fn() },
});
expect(() =>
session.submitToolResult("call-1", { ok: true }, { suppressResponse: true }),
).toThrow("Realtime provider does not support suppressed tool results");
expect(submitToolResult).not.toHaveBeenCalled();
});
it("does not expose session callbacks until the provider returns its bridge", () => {

View File

@@ -42,7 +42,11 @@ export type RealtimeVoiceBridgeSession = {
sendUserMessage(text: string): void;
handleBargeIn(options?: RealtimeVoiceBargeInOptions): void;
setMediaTimestamp(ts: number): void;
submitToolResult(callId: string, result: unknown, options?: RealtimeVoiceToolResultOptions): void;
submitToolResult(
callId: string,
result: unknown,
options?: RealtimeVoiceToolResultOptions,
): void | Promise<void>;
triggerGreeting(instructions?: string): void;
};
@@ -64,7 +68,10 @@ export type RealtimeVoiceBridgeSessionParams = {
tools?: RealtimeVoiceTool[];
onTranscript?: (role: RealtimeVoiceRole, text: string, isFinal: boolean) => void;
onEvent?: (event: RealtimeVoiceBridgeEvent) => void;
onToolCall?: (event: RealtimeVoiceToolCallEvent, session: RealtimeVoiceBridgeSession) => void;
onToolCall?: (
event: RealtimeVoiceToolCallEvent,
session: RealtimeVoiceBridgeSession,
) => void | Promise<void>;
onReady?: (session: RealtimeVoiceBridgeSession) => void;
onError?: (error: Error) => void;
onClose?: (reason: RealtimeVoiceCloseReason) => void;
@@ -77,6 +84,7 @@ export function createRealtimeVoiceBridgeSession(
params: RealtimeVoiceBridgeSessionParams,
): RealtimeVoiceBridgeSession {
const bridgeRef: { current?: RealtimeVoiceBridge } = {};
let isActive = true;
const requireBridge = () => {
if (!bridgeRef.current) {
throw new Error("Realtime voice bridge is not ready");
@@ -90,17 +98,38 @@ export function createRealtimeVoiceBridgeSession(
return requireBridge();
},
acknowledgeMark: () => requireBridge().acknowledgeMark(),
close: () => requireBridge().close(),
close: () => {
const bridge = requireBridge();
isActive = false;
bridge.close();
},
connect: () => requireBridge().connect(),
sendAudio: (audio) => requireBridge().sendAudio(audio),
sendUserMessage: (text) => requireBridge().sendUserMessage?.(text),
handleBargeIn: (options) => requireBridge().handleBargeIn?.(options),
setMediaTimestamp: (ts) => requireBridge().setMediaTimestamp(ts),
submitToolResult: (callId, result, options) =>
requireBridge().submitToolResult(callId, result, options),
submitToolResult: (callId, result, options) => {
const bridge = requireBridge();
if (options?.suppressResponse && bridge.supportsToolResultSuppression === false) {
throw new Error("Realtime provider does not support suppressed tool results");
}
return bridge.submitToolResult(callId, result, options);
},
triggerGreeting: (instructions) => requireBridge().triggerGreeting?.(instructions),
};
const canSendAudio = () => params.audioSink.isOpen?.() ?? true;
const reportCallbackError = (error: unknown) => {
// Async tool handlers can settle after the provider closes. Once inactive, no
// callback may report stale failures into the next session lifecycle.
if (!isActive) {
return;
}
try {
params.onError?.(error instanceof Error ? error : new Error(String(error)));
} catch {
// An error callback is the terminal boundary for provider callback failures.
}
};
const bridge = params.provider.createBridge({
cfg: params.cfg,
providerConfig: params.providerConfig,
@@ -136,10 +165,17 @@ export function createRealtimeVoiceBridgeSession(
onTranscript: params.onTranscript,
onEvent: params.onEvent,
onToolCall: (event) => {
if (!bridgeRef.current) {
if (!bridgeRef.current || !isActive) {
return;
}
params.onToolCall?.(event, session);
try {
const pending = params.onToolCall?.(event, session);
if (pending) {
void pending.catch(reportCallbackError);
}
} catch (error) {
reportCallbackError(error);
}
},
onReady: () => {
if (!bridgeRef.current) {
@@ -151,7 +187,10 @@ export function createRealtimeVoiceBridgeSession(
params.onReady?.(session);
},
onError: params.onError,
onClose: params.onClose,
onClose: (reason) => {
isActive = false;
params.onClose?.(reason);
},
});
bridgeRef.current = bridge;

View File

@@ -489,6 +489,115 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
transport.stop();
});
it("waits for provider tool-result submission before returning to listening", async () => {
let resolveSubmission: () => void = () => undefined;
const submission = new Promise<void>((resolve) => {
resolveSubmission = resolve;
});
const onStatus = vi.fn();
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {
if (method === "talk.client.toolCall") {
return { runId: "run-1" };
}
if (method === "talk.session.submitToolResult") {
await submission;
}
return {};
});
const transport = new GatewayRelayRealtimeTalkTransport(createSession(), {
callbacks: { onStatus },
client,
sessionKey: "main",
});
await transport.start();
emitGatewayFrame({
event: "talk.event",
payload: {
relaySessionId: "relay-1",
type: "toolCall",
callId: "call-1",
name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME,
args: { question: "status?" },
},
});
await vi.waitFor(() => expect(requestCallsFor(client, "talk.client.toolCall")).toHaveLength(1));
emitGatewayFrame({
event: "chat",
payload: {
runId: "run-1",
state: "final",
message: { text: "All systems green." },
},
});
await vi.waitFor(() =>
expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(1),
);
emitGatewayFrame({
event: "talk.event",
payload: {
relaySessionId: "relay-1",
type: "toolResult",
callId: "call-1",
},
});
expect(onStatus).not.toHaveBeenCalledWith("listening");
expect(requestCallsFor(client, "chat.abort")).toHaveLength(0);
resolveSubmission();
await vi.waitFor(() => expect(onStatus).toHaveBeenCalledWith("listening"));
transport.stop();
});
it("surfaces rejected provider tool-result submission without an unhandled rejection", async () => {
const onStatus = vi.fn();
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {
if (method === "talk.client.toolCall") {
return { runId: "run-1" };
}
if (method === "talk.session.submitToolResult") {
throw new Error("Provider rejected the tool result");
}
return {};
});
const transport = new GatewayRelayRealtimeTalkTransport(createSession(), {
callbacks: { onStatus },
client,
sessionKey: "main",
});
await transport.start();
emitGatewayFrame({
event: "talk.event",
payload: {
relaySessionId: "relay-1",
type: "toolCall",
callId: "call-1",
name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME,
args: { question: "status?" },
},
});
await vi.waitFor(() => expect(requestCallsFor(client, "talk.client.toolCall")).toHaveLength(1));
emitGatewayFrame({
event: "chat",
payload: {
runId: "run-1",
state: "final",
message: { text: "All systems green." },
},
});
await vi.waitFor(() =>
expect(onStatus).toHaveBeenCalledWith("error", "Provider rejected the tool result"),
);
expect(onStatus).toHaveBeenLastCalledWith("error", "Provider rejected the tool result");
expect(onStatus).not.toHaveBeenCalledWith("listening");
expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(1);
transport.stop();
});
it("submits an interim working result for forced consult tool calls", async () => {
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {
@@ -532,6 +641,47 @@ describe("GatewayRelayRealtimeTalkTransport", () => {
transport.stop();
});
it("does not start a forced consult when the working result is terminally cancelled", async () => {
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {
if (method === "talk.session.submitToolResult") {
emitGatewayFrame({
event: "talk.event",
payload: {
relaySessionId: "relay-1",
type: "toolResult",
callId: "call-1",
},
});
}
return {};
});
const transport = new GatewayRelayRealtimeTalkTransport(createSession(), {
callbacks: {},
client,
sessionKey: "main",
});
await transport.start();
emitGatewayFrame({
event: "talk.event",
payload: {
relaySessionId: "relay-1",
type: "toolCall",
callId: "call-1",
name: REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME,
forced: true,
args: { question: "status?" },
},
});
await vi.waitFor(() =>
expect(requestCallsFor(client, "talk.session.submitToolResult")).toHaveLength(1),
);
expect(requestCallsFor(client, "talk.client.toolCall")).toHaveLength(0);
transport.stop();
});
it("treats server relay tool results as terminal for active consult calls", async () => {
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {

View File

@@ -61,6 +61,7 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
private readonly outputQueue = new RealtimeTalkPcmOutputQueue();
private readonly consultAbortControllers = new Map<string, AbortController>();
private readonly completedToolCalls = new Set<string>();
private readonly submittingToolCalls = new Set<string>();
private cancelRequestedForPlayback = false;
private speechFramesDuringPlayback = 0;
private lastRelayError: string | undefined;
@@ -201,6 +202,9 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
return;
case "clear":
this.stopOutput();
if (event.talkEvent?.type === "turn.cancelled") {
this.abortConsults();
}
return;
case "mark":
this.scheduleMarkAck();
@@ -215,7 +219,9 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
}
return;
case "toolCall":
void this.handleToolCall(event);
void this.handleToolCall(event).catch((error: unknown) => {
this.reportToolResultSubmissionError(error);
});
return;
case "toolResult":
if (this.isFinalToolResult(event)) {
@@ -278,14 +284,16 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
return;
}
if (name !== REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) {
this.submitToolResult(callId, { error: `Tool "${name}" not available in browser Talk` });
await this.submitToolResult(callId, {
error: `Tool "${name}" not available in browser Talk`,
});
return;
}
const abortController = new AbortController();
this.consultAbortControllers.set(callId, abortController);
try {
if (event.forced) {
this.submitToolResult(
await this.submitToolResult(
callId,
{
status: "working",
@@ -295,6 +303,13 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
},
{ willContinue: true },
);
if (this.completedToolCalls.has(callId)) {
return;
}
if (abortController.signal.aborted) {
await this.submitToolResult(callId, { status: "cancelled" });
return;
}
}
await submitRealtimeTalkConsult({
ctx: this.ctx,
@@ -309,20 +324,34 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
}
}
private submitToolResult(
private async submitToolResult(
callId: string,
result: unknown,
options?: { suppressResponse?: boolean; willContinue?: boolean },
): void {
): Promise<void> {
if (this.completedToolCalls.has(callId)) {
return;
}
void this.ctx.client.request("talk.session.submitToolResult", {
sessionId: this.session.relaySessionId,
callId,
result,
...(options ? { options } : {}),
});
this.submittingToolCalls.add(callId);
try {
await this.ctx.client.request("talk.session.submitToolResult", {
sessionId: this.session.relaySessionId,
callId,
result,
...(options ? { options } : {}),
});
} finally {
this.submittingToolCalls.delete(callId);
}
}
private reportToolResultSubmissionError(error: unknown): void {
if (this.closed) {
return;
}
const message = error instanceof Error ? error.message : String(error);
this.lastRelayError = message;
this.ctx.callbacks.onStatus?.("error", message);
}
private completeToolCall(callIdRaw: string | undefined): void {
@@ -331,6 +360,11 @@ export class GatewayRelayRealtimeTalkTransport implements RealtimeTalkTransport
return;
}
this.completedToolCalls.add(callId);
// The Gateway broadcasts acceptance before resolving the matching RPC.
// Do not turn our own accepted result into a late consult cancellation.
if (this.submittingToolCalls.has(callId)) {
return;
}
this.consultAbortControllers.get(callId)?.abort();
this.consultAbortControllers.delete(callId);
}

View File

@@ -4,7 +4,10 @@ import {
buildGoogleLiveUrl,
GoogleLiveRealtimeTalkTransport,
} from "./realtime-talk-google-live.ts";
import { REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME } from "./realtime-talk-shared.ts";
import {
REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME,
REALTIME_VOICE_AGENT_CONTROL_TOOL_NAME,
} from "./realtime-talk-shared.ts";
import type {
RealtimeTalkJsonPcmWebSocketSessionResult,
RealtimeTalkTransportContext,
@@ -495,6 +498,69 @@ describe("GoogleLiveRealtimeTalkTransport", () => {
transport.stop();
});
it("surfaces Google Live tool-result send failures without an unhandled rejection", async () => {
const onStatus = vi.fn();
const onTalkEvent = vi.fn();
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {
if (method === "talk.client.steer") {
return {
ok: true,
mode: "status",
sessionKey: "main",
active: true,
message: "Still working.",
};
}
throw new Error(`unexpected request: ${method}`);
});
const transport = createTransport({ onStatus, onTalkEvent }, client);
await transport.start();
const ws = latestWebSocket();
vi.spyOn(ws, "send").mockImplementation(() => {
throw new Error("Google Live socket rejected the tool result");
});
ws.emitMessage(
encodeJsonFrame({
toolCall: {
functionCalls: [
{
id: "call-control",
name: REALTIME_VOICE_AGENT_CONTROL_TOOL_NAME,
args: { text: "status", mode: "status" },
},
],
},
}),
);
await vi.waitFor(() =>
expect(onStatus).toHaveBeenCalledWith("error", "Google Live socket rejected the tool result"),
);
expect(
onTalkEvent.mock.calls.some(
([event]) =>
(event.type === "tool.progress" || event.type === "tool.error") && event.final === true,
),
).toBe(false);
expect(
(
transport as unknown as {
pendingCalls: Map<string, unknown>;
}
).pendingCalls.has("call-control"),
).toBe(true);
expect(() =>
(
transport as unknown as {
submitToolResult: (callId: string, result: unknown) => void;
}
).submitToolResult("missing-call", { ok: true }),
).toThrow("Google Live has no pending tool call for missing-call");
transport.stop();
});
it("sends spoken active-control acknowledgements through Google Live", async () => {
const client = createClient();
vi.mocked(client["request"]).mockImplementation(async (method) => {

View File

@@ -213,10 +213,12 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
this.inputProcessor.connect(this.inputContext.destination);
}
private send(message: unknown): void {
private send(message: unknown): boolean {
if (!this.closed && this.ws?.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(message));
return true;
}
return false;
}
private async handleMessage(data: unknown): Promise<void> {
@@ -310,7 +312,9 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
this.emitTalkEvent({ type: "turn.ended", final: true });
}
for (const call of message.toolCall?.functionCalls ?? []) {
void this.handleToolCall(call);
void this.handleToolCall(call).catch((error: unknown) => {
this.reportToolResultSubmissionError(error);
});
}
}
@@ -393,10 +397,9 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
private submitToolResult(callId: string, result: unknown): void {
const pending = this.pendingCalls.get(callId);
if (!pending) {
return;
throw new Error(`Google Live has no pending tool call for ${callId}`);
}
this.pendingCalls.delete(callId);
this.send({
const sent = this.send({
toolResponse: {
functionResponses: [
{
@@ -411,6 +414,18 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport {
],
},
});
if (!sent) {
throw new Error("Google Live socket is not open");
}
this.pendingCalls.delete(callId);
}
private reportToolResultSubmissionError(error: unknown): void {
if (this.closed) {
return;
}
const message = error instanceof Error ? error.message : String(error);
this.ctx.callbacks.onStatus?.("error", message);
}
private sendControlSpeechMessage(message: string): void {

View File

@@ -452,14 +452,16 @@ export async function steerRealtimeTalkActiveConsult(params: {
export async function submitRealtimeTalkAgentControl(params: {
ctx: RealtimeTalkTransportContext;
args: unknown;
submit: (callId: string, result: unknown) => void;
submit: (callId: string, result: unknown) => void | Promise<void>;
callId: string;
sessionId?: string;
emitTalkEvent?: (input: RealtimeTalkEventInput) => void;
}): Promise<void> {
let result: unknown;
let talkEvent: RealtimeTalkEventInput;
try {
const parsed = parseRealtimeVoiceAgentControlToolArgs(params.args);
const result =
result =
params.sessionId && params.sessionId.trim()
? await params.ctx.client.request("talk.session.steer", {
sessionId: params.sessionId,
@@ -472,7 +474,7 @@ export async function submitRealtimeTalkAgentControl(params: {
text: parsed.text,
mode: parsed.mode,
});
params.emitTalkEvent?.({
talkEvent = {
type: "tool.progress",
callId: params.callId,
payload: {
@@ -483,18 +485,19 @@ export async function submitRealtimeTalkAgentControl(params: {
result && typeof result === "object" && "mode" in result
? result.mode === "status" || result.mode === "cancel"
: undefined,
});
params.submit(params.callId, result);
};
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
params.emitTalkEvent?.({
talkEvent = {
type: "tool.error",
callId: params.callId,
payload: { message },
final: true,
});
params.submit(params.callId, { error: message });
};
result = { error: message };
}
await params.submit(params.callId, result);
params.emitTalkEvent?.(talkEvent);
}
function maybeSpeakRealtimeTalkControlResult(
@@ -523,7 +526,7 @@ function maybeSpeakRealtimeTalkControlResult(
export async function submitRealtimeTalkConsult(params: {
ctx: RealtimeTalkTransportContext;
args: unknown;
submit: (callId: string, result: unknown) => void;
submit: (callId: string, result: unknown) => void | Promise<void>;
callId: string;
relaySessionId?: string;
emitTalkEvent?: (input: RealtimeTalkEventInput) => void;
@@ -535,16 +538,18 @@ export async function submitRealtimeTalkConsult(params: {
let runId: string | undefined;
let aborted = false;
let submitted = false;
const submitOnce = (result: unknown) => {
let submissionCompleted = false;
const submitOnce = async (result: unknown): Promise<void> => {
if (submitted) {
return;
}
submitted = true;
submit(callId, result);
await submit(callId, result);
submissionCompleted = true;
};
const submitAbortResult = () => {
const submitAbortResult = async (): Promise<void> => {
if (params.submitAbortResult !== false) {
submitOnce(buildRealtimeVoiceAgentCancelProviderResult());
await submitOnce(buildRealtimeVoiceAgentCancelProviderResult());
}
};
const abortRun = () => {
@@ -554,7 +559,7 @@ export async function submitRealtimeTalkConsult(params: {
}
};
if (params.signal?.aborted) {
submitAbortResult();
await submitAbortResult();
return;
}
params.signal?.addEventListener("abort", abortRun, { once: true });
@@ -577,7 +582,7 @@ export async function submitRealtimeTalkConsult(params: {
}
if (params.signal?.aborted) {
abortRun();
submitAbortResult();
await submitAbortResult();
return;
}
const result = await waitForChatResult({
@@ -587,18 +592,21 @@ export async function submitRealtimeTalkConsult(params: {
emitTalkEvent: params.emitTalkEvent,
signal: params.signal,
});
submitOnce({ result });
await submitOnce({ result });
} catch (error) {
if (submitted) {
throw error;
}
if (aborted || params.signal?.aborted || isAbortError(error)) {
submitAbortResult();
await submitAbortResult();
return;
}
submitOnce({
await submitOnce({
error: error instanceof Error ? error.message : String(error),
});
} finally {
params.signal?.removeEventListener("abort", abortRun);
if (!aborted && !params.signal?.aborted) {
if (submissionCompleted && !aborted && !params.signal?.aborted) {
ctx.callbacks.onStatus?.("listening");
}
}

View File

@@ -896,4 +896,50 @@ describe("WebRtcSdpRealtimeTalkTransport", () => {
});
transport.stop();
});
it("surfaces OpenAI tool-result send failures without an unhandled rejection", async () => {
stubAnswerSdpFetch();
const onStatus = vi.fn();
const onTalkEvent = vi.fn();
const request = vi.fn(async (method: string) => {
if (method === "talk.client.steer") {
return {
ok: true,
mode: "status",
sessionKey: "main",
active: true,
message: "Still working.",
};
}
throw new Error(`unexpected request: ${method}`);
});
const transport = createOpenAiTransport({ request }, { onStatus, onTalkEvent });
await transport.start();
const peer = FakePeerConnection.instances[0];
peer?.channel.send.mockImplementation(() => {
throw new Error("OpenAI data channel rejected the tool result");
});
dispatchRealtimeEvent(peer, {
type: "response.function_call_arguments.done",
item_id: "item-control",
call_id: "call-control",
name: REALTIME_VOICE_AGENT_CONTROL_TOOL_NAME,
arguments: JSON.stringify({ text: "status", mode: "status" }),
});
await vi.waitFor(() =>
expect(onStatus).toHaveBeenCalledWith(
"error",
"OpenAI data channel rejected the tool result",
),
);
expect(
onTalkEvent.mock.calls.some(
([event]) =>
(event.type === "tool.progress" || event.type === "tool.error") && event.final === true,
),
).toBe(false);
transport.stop();
});
});

View File

@@ -261,7 +261,9 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
this.bufferToolDelta(event);
return;
case "response.function_call_arguments.done":
void this.handleToolCall(event);
void this.handleToolCall(event).catch((error: unknown) => {
this.reportToolResultSubmissionError(error);
});
return;
case "input_audio_buffer.speech_started":
this.ctx.callbacks.onStatus?.("listening", "Speech detected");
@@ -409,6 +411,14 @@ export class WebRtcSdpRealtimeTalkTransport implements RealtimeTalkTransport {
this.requestResponseCreate();
}
private reportToolResultSubmissionError(error: unknown): void {
if (this.closed) {
return;
}
const message = error instanceof Error ? error.message : String(error);
this.ctx.callbacks.onStatus?.("error", message);
}
private sendControlSpeechMessage(message: string): void {
if (this.responseActive) {
this.send({ type: "response.cancel" });