vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-02 08:51:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(meeting): reset realtime provider generations

Browse Source
This commit is contained in:
Vincent Koc
2026-08-02 01:10:53 +08:00
parent e1cc00f9b6
commit e81dc9c613
3 changed files with 196 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
src/meeting-bot/agent-consult.ts
View File

@@ -92,10 +92,11 @@ export function createMeetingRealtimeEngineBindings(params: {
async function submitMeetingConsultWorkingResponse(params: {
session: RealtimeVoiceBridgeSession;
abortSignal?: AbortSignal;
callId: string;
label: string;
}): Promise<void> {
if (!params.session.bridge.supportsToolResultContinuation) {
if (params.abortSignal?.aborted || !params.session.bridge.supportsToolResultContinuation) {
return;
}
await params.session.submitToolResult(
@@ -116,6 +117,7 @@ async function consultMeetingAgent(params: {
requesterSessionKey?: string;
args: unknown;
transcript: Array<{ role: "user" | "assistant"; text: string }>;
abortSignal?: AbortSignal;
}): Promise<{ text: string }> {
const agentId = params.agentId
? normalizeAgentId(params.agentId)
@@ -142,6 +144,7 @@ async function consultMeetingAgent(params: {
questionSourceLabel: params.surface.questionSourceLabel,
toolsAllow: resolveRealtimeVoiceAgentConsultToolsAllow(params.toolPolicy),
extraSystemPrompt: params.surface.extraSystemPrompt,
abortSignal: params.abortSignal,
});
}
@@ -158,12 +161,19 @@ async function handleMeetingRealtimeConsultToolCall(params: {
meetingSessionId: string;
requesterSessionKey?: string;
transcript: Array<{ role: "user" | "assistant"; text: string }>;
abortSignal?: AbortSignal;
onTalkEvent?: (event: TalkEventInput) => void;
}): Promise<void> {
const callId = params.event.callId || params.event.itemId;
if (params.abortSignal?.aborted) {
return;
}
if (params.strategy !== "bidi") {
const error = `Tool "${params.event.name}" is only available in bidi realtime strategy`;
await params.session.submitToolResult(callId, { error });
if (params.abortSignal?.aborted) {
return;
}
params.onTalkEvent?.({
type: "tool.error",
callId,
@@ -175,6 +185,9 @@ async function handleMeetingRealtimeConsultToolCall(params: {
if (params.event.name !== REALTIME_VOICE_AGENT_CONSULT_TOOL_NAME) {
const error = `Tool "${params.event.name}" not available`;
await params.session.submitToolResult(callId, { error });
if (params.abortSignal?.aborted) {
return;
}
params.onTalkEvent?.({
type: "tool.error",
callId,
@@ -185,9 +198,13 @@ async function handleMeetingRealtimeConsultToolCall(params: {
}
await submitMeetingConsultWorkingResponse({
session: params.session,
abortSignal: params.abortSignal,
callId,
label: params.surface.workingResponseLabel,
});
if (params.abortSignal?.aborted) {
return;
}
params.onTalkEvent?.({
type: "tool.progress",
callId,
@@ -206,10 +223,17 @@ async function handleMeetingRealtimeConsultToolCall(params: {
requesterSessionKey: params.requesterSessionKey,
args: params.event.args,
transcript: params.transcript,
abortSignal: params.abortSignal,
});
} catch (error) {
if (params.abortSignal?.aborted) {
return;
}
const message = formatErrorMessage(error);
await params.session.submitToolResult(callId, { error: message });
if (params.abortSignal?.aborted) {
return;
}
params.onTalkEvent?.({
type: "tool.error",
callId,
@@ -218,7 +242,13 @@ async function handleMeetingRealtimeConsultToolCall(params: {
});
return;
}
if (params.abortSignal?.aborted) {
return;
}
await params.session.submitToolResult(callId, result);
if (params.abortSignal?.aborted) {
return;
}
params.onTalkEvent?.({
type: "tool.result",
callId,

99
src/meeting-bot/realtime-engine.test.ts
View File

@@ -5,16 +5,22 @@ import type {
RealtimeVoiceBridgeCreateRequest,
} from "../talk/provider-types.js";
import type { MeetingRealtimeAudioTransport } from "./realtime-audio-transport.js";
import { startMeetingRealtimeEngine } from "./realtime-engine.js";
import {
startMeetingRealtimeEngine,
type MeetingRealtimeToolCallParams,
} from "./realtime-engine.js";
type PendingWrite = {
resolve: () => void;
};
async function createEngineFixture() {
async function createEngineFixture(options?: {
handleToolCall?: (params: MeetingRealtimeToolCallParams) => Promise<void>;
}) {
let callbacks: RealtimeVoiceBridgeCreateRequest | undefined;
let onHumanBargeIn: ((audio: Buffer) => boolean) | undefined;
const handleBargeIn = vi.fn();
const submitToolResult = vi.fn();
const bridge: RealtimeVoiceBridge = {
acknowledgeMark: vi.fn(),
close: vi.fn(),
@@ -23,7 +29,7 @@ async function createEngineFixture() {
isConnected: vi.fn(() => true),
sendAudio: vi.fn(),
setMediaTimestamp: vi.fn(),
submitToolResult: vi.fn(),
submitToolResult,
};
const provider: RealtimeVoiceProviderPlugin = {
id: "test",
@@ -66,7 +72,7 @@ async function createEngineFixture() {
},
consultAgent: vi.fn(async () => ({ text: "unused" })),
fullConfig: {} as never,
handleToolCall: vi.fn(async () => {}),
handleToolCall: options?.handleToolCall ?? vi.fn(async () => {}),
logger: {
debug: vi.fn(),
error: vi.fn(),
@@ -94,6 +100,7 @@ async function createEngineFixture() {
clearOutput,
handle,
handleBargeIn,
submitToolResult,
releaseWrite(index: number) {
const pending = pendingWrites[index];
if (!pending) {
@@ -127,6 +134,90 @@ async function createEngineFixture() {
}
describe("meeting realtime engine output ownership", () => {
it("resets provider continuity without replaying old output or tool work", async () => {
let releaseTool: (() => void) | undefined;
const toolGate = new Promise<void>((resolve) => {
releaseTool = resolve;
});
const fixture = await createEngineFixture({
handleToolCall: async ({ session, event, onTalkEvent }) => {
await toolGate;
await session.submitToolResult(event.callId, { text: "stale result" });
onTalkEvent({
type: "tool.result",
callId: event.callId,
payload: { name: event.name },
final: true,
});
},
});
try {
const active = Buffer.from([1]);
const stale = Buffer.from([2]);
const fresh = Buffer.from([3]);
fixture.callbacks.onReady?.();
fixture.callbacks.onTranscript?.("user", "old turn", true);
fixture.sendOutputAudio(active, "response-1");
await vi.waitFor(() => {
expect(fixture.writeOutput).toHaveBeenCalledOnce();
});
fixture.sendOutputAudio(stale, "response-1");
fixture.callbacks.onToolCall?.({
itemId: "item-old",
callId: "call-old",
name: "openclaw_agent_consult",
args: { question: "old work" },
});
fixture.callbacks.onEvent?.({
direction: "client",
type: "session.continuity.reset",
});
fixture.callbacks.onEvent?.({
direction: "client",
type: "session.continuity.reset",
});
await vi.waitFor(() => {
expect(fixture.clearOutput).toHaveBeenCalledOnce();
});
expect(fixture.handleBargeIn).not.toHaveBeenCalled();
expect(
fixture.handle
.getHealth()
.recentTalkEvents.filter((event) => event.type === "turn.cancelled"),
).toHaveLength(1);
releaseTool?.();
await Promise.resolve();
await Promise.resolve();
expect(fixture.submitToolResult).not.toHaveBeenCalled();
expect(
fixture.handle
.getHealth()
.recentTalkEvents.some(
(event) => event.type === "tool.result" && event.callId === "call-old",
),
).toBe(false);
fixture.releaseWrite(0);
await vi.waitFor(() => {
expect(fixture.clearOutput).toHaveBeenCalledTimes(2);
});
expect(fixture.writeOutput).not.toHaveBeenCalledWith(stale);
fixture.callbacks.onReady?.();
fixture.sendOutputAudio(fresh, "response-1");
await vi.waitFor(() => {
expect(fixture.writeOutput).toHaveBeenCalledTimes(2);
});
expect(fixture.writeOutput).toHaveBeenLastCalledWith(fresh);
fixture.releaseWrite(1);
} finally {
await fixture.handle.stop();
}
});
it("serializes transport writes and coalesces queued 20 ms frames", async () => {
const fixture = await createEngineFixture();
try {

80
src/meeting-bot/realtime-engine.ts
View File

@@ -67,6 +67,7 @@ export type MeetingAgentConsultParams = {
export type MeetingRealtimeToolCallParams = {
strategy: string;
session: RealtimeVoiceBridgeSession;
abortSignal?: AbortSignal;
event: RealtimeVoiceToolCallEvent;
meetingSessionId: string;
requesterSessionKey?: string;
@@ -135,7 +136,10 @@ export async function startMeetingRealtimeEngine(params: {
let outputPendingBytes = 0;
let outputPendingFrames = 0;
let outputGenerationActive = false;
let continuityResetActive = false;
let toolContinuityEpoch = 0;
let outputClearTail = Promise.resolve();
const activeToolCalls = new Set<AbortController>();
const outputQueue: Array<{ audio: Buffer; generation: number }> = [];
const outputOwner = createMeetingRealtimeOutputOwner();
const outputMaxPendingBytes =
@@ -160,6 +164,11 @@ export async function startMeetingRealtimeEngine(params: {
outputOwner.reset();
outputClearAfterActive = false;
invalidateOutputQueue();
toolContinuityEpoch += 1;
for (const controller of activeToolCalls) {
controller.abort("meeting realtime stopped");
}
activeToolCalls.clear();
}
if (stopPromise) {
await stopPromise;
@@ -549,6 +558,31 @@ export async function startMeetingRealtimeEngine(params: {
}
},
onEvent: (event) => {
if (event.direction === "client" && event.type === "session.continuity.reset") {
if (continuityResetActive) {
return;
}
continuityResetActive = true;
realtimeReady = false;
outputOwner.reset();
outputGenerationActive = false;
toolContinuityEpoch += 1;
for (const controller of activeToolCalls) {
controller.abort(event.type);
}
activeToolCalls.clear();
const turnId = harness.talk.activeTurnId;
invalidateOutputPlayback();
harness.flushOutput(clearOutputPlayback);
harness.finishOutputAudio(event.type);
if (turnId) {
harness.talk.cancelTurn({
turnId,
payload: { ...outputTalkPayload, reason: event.type },
});
}
return;
}
outputOwner.noteEvent(event);
if (event.type === "input_audio_buffer.speech_started") {
harness.ensureTurn();
@@ -601,6 +635,9 @@ export async function startMeetingRealtimeEngine(params: {
}
},
onToolCall: (event, session) => {
const epoch = toolContinuityEpoch;
const controller = new AbortController();
activeToolCalls.add(controller);
harness.emit({
type: "tool.call",
turnId: harness.ensureTurn(),
@@ -609,16 +646,38 @@ export async function startMeetingRealtimeEngine(params: {
payload: { name: event.name, args: event.args },
});
const turnId = harness.ensureTurn();
return params.handleToolCall({
strategy,
session,
event,
meetingSessionId: params.meetingSessionId,
requesterSessionKey: params.requesterSessionKey,
transcript: harness.transcript,
onTalkEvent: (inputLocal) =>
harness.emit({ ...inputLocal, turnId: inputLocal.turnId ?? turnId }),
});
const guardedSession = Object.create(session) as RealtimeVoiceBridgeSession;
guardedSession.submitToolResult = (callId, result, options) => {
if (controller.signal.aborted || epoch !== toolContinuityEpoch) {
return;
}
return session.submitToolResult(callId, result, options);
};
return params
.handleToolCall({
strategy,
session: guardedSession,
abortSignal: controller.signal,
event,
meetingSessionId: params.meetingSessionId,
requesterSessionKey: params.requesterSessionKey,
transcript: harness.transcript,
onTalkEvent: (inputLocal) => {
if (controller.signal.aborted || epoch !== toolContinuityEpoch) {
return;
}
harness.emit({ ...inputLocal, turnId: inputLocal.turnId ?? turnId });
},
})
.catch((error: unknown) => {
if (controller.signal.aborted || epoch !== toolContinuityEpoch) {
return;
}
throw error;
})
.finally(() => {
activeToolCalls.delete(controller);
});
},
onError: (error) => {
harness.emit({
@@ -644,6 +703,7 @@ export async function startMeetingRealtimeEngine(params: {
},
onReady: () => {
realtimeReady = true;
continuityResetActive = false;
harness.emit({
type: "session.ready",
payload: outputTalkPayload,