From 1bd2f93214ae23b7f4ea6465f09cac9eec2ef92f Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Fri, 31 Jul 2026 10:39:48 +0800 Subject: [PATCH] fix(meeting-bot): negotiate realtime output generations --- extensions/google-meet/index.test.ts | 1 + extensions/google-meet/node-host.test.ts | 2 +- .../src/node-invoke-policy.test.ts | 42 +++++++ .../google-meet/src/transports/chrome.ts | 5 +- src/meeting-bot/chrome-transport.test.ts | 84 ++++++++++++++ src/meeting-bot/chrome-transport.ts | 3 +- src/meeting-bot/node-host.ts | 7 +- src/meeting-bot/node-invoke-policy.ts | 30 ++++- .../realtime-node-audio-transport.test.ts | 109 +++++++++++++++++- .../realtime-node-audio-transport.ts | 61 +++++++--- 10 files changed, 316 insertions(+), 28 deletions(-) diff --git a/extensions/google-meet/index.test.ts b/extensions/google-meet/index.test.ts index cf64882e7e57..69ecf7985e9d 100644 --- a/extensions/google-meet/index.test.ts +++ b/extensions/google-meet/index.test.ts @@ -267,6 +267,7 @@ async function startTestNodeRealtimeAudioBridge(params: TestNodeRealtimeEnginePa commandName: "googlemeet.chrome", logScope: "[google-meet]", logPrefix: "node", + outputGenerationSupported: true, }); return { type: "node-command-pair" as const, diff --git a/extensions/google-meet/node-host.test.ts b/extensions/google-meet/node-host.test.ts index b78723d5280e..594ad5f39e9d 100644 --- a/extensions/google-meet/node-host.test.ts +++ b/extensions/google-meet/node-host.test.ts @@ -553,7 +553,7 @@ describe("google-meet node host bridge sessions", () => { expect(typeof start.bridgeId).toBe("string"); expect(start.bridgeId.length).toBeGreaterThan(0); expect(start).toEqual({ - audioBridge: { type: "node-command-pair" }, + audioBridge: { type: "node-command-pair", outputGeneration: true }, bridgeId: start.bridgeId, launched: false, }); diff --git a/extensions/google-meet/src/node-invoke-policy.test.ts b/extensions/google-meet/src/node-invoke-policy.test.ts index c0b1690c55a2..801c6f690e99 100644 --- a/extensions/google-meet/src/node-invoke-policy.test.ts +++ b/extensions/google-meet/src/node-invoke-policy.test.ts @@ -128,7 +128,14 @@ describe("Google Meet node invoke policy", () => { { action: "pullAudio" }, { action: "pushAudio", bridgeId: "bridge-1" }, { action: "pushAudio", bridgeId: "bridge-1", base64: "not-base64!" }, + { + action: "pushAudio", + bridgeId: "bridge-1", + base64: Buffer.from([1]).toString("base64"), + outputGeneration: -1, + }, { action: "clearAudio", bridgeId: "" }, + { action: "clearAudio", bridgeId: "bridge-1", outputGeneration: 1.5 }, { action: "stopByUrl" }, { action: "stopByUrl", url: "https://example.com/not-meet" }, ]) { @@ -142,6 +149,41 @@ describe("Google Meet node invoke policy", () => { } }); + it("forwards validated output generations", async () => { + const policy = createGoogleMeetChromeNodeInvokePolicy(resolveGoogleMeetConfig({})); + const base64 = Buffer.from([1, 2, 3]).toString("base64"); + const push = createContext({ + action: "pushAudio", + bridgeId: "bridge-1", + base64, + outputGeneration: 3, + }); + const clear = createContext({ + action: "clearAudio", + bridgeId: "bridge-1", + outputGeneration: 4, + }); + + await policy.handle(push.ctx); + await policy.handle(clear.ctx); + + expect(push.invokeNode).toHaveBeenCalledWith({ + params: { + action: "pushAudio", + bridgeId: "bridge-1", + base64, + outputGeneration: 3, + }, + }); + expect(clear.invokeNode).toHaveBeenCalledWith({ + params: { + action: "clearAudio", + bridgeId: "bridge-1", + outputGeneration: 4, + }, + }); + }); + it("normalizes forwarded Meet URL filters before node dispatch", async () => { const policy = createGoogleMeetChromeNodeInvokePolicy(resolveGoogleMeetConfig({})); const { ctx, invokeNode } = createContext({ diff --git a/extensions/google-meet/src/transports/chrome.ts b/extensions/google-meet/src/transports/chrome.ts index 7729823f6868..d3026a38eb9d 100644 --- a/extensions/google-meet/src/transports/chrome.ts +++ b/extensions/google-meet/src/transports/chrome.ts @@ -233,7 +233,7 @@ export async function launchChromeMeet(params: { function parseNodeStartResult(raw: unknown): { launched?: boolean; bridgeId?: string; - audioBridge?: { type?: string }; + audioBridge?: { type?: string; outputGeneration?: boolean }; browser?: GoogleMeetChromeHealth; } { const value = @@ -246,7 +246,7 @@ function parseNodeStartResult(raw: unknown): { return value as { launched?: boolean; bridgeId?: string; - audioBridge?: { type?: string }; + audioBridge?: { type?: string; outputGeneration?: boolean }; browser?: GoogleMeetChromeHealth; }; } @@ -565,6 +565,7 @@ export async function launchChromeMeetOnNode(params: { commandName: GOOGLE_MEET_NODE_COMMAND, logScope: GOOGLE_MEET_PLATFORM_ADAPTER.logScope, logPrefix: params.mode === "agent" ? "node agent" : "node", + outputGenerationSupported: result.audioBridge.outputGeneration === true, }); const bindings = createMeetingRealtimeEngineBindings({ platform: GOOGLE_MEET_PLATFORM_ADAPTER, diff --git a/src/meeting-bot/chrome-transport.test.ts b/src/meeting-bot/chrome-transport.test.ts index b2523a9ffe69..00d2296d07ce 100644 --- a/src/meeting-bot/chrome-transport.test.ts +++ b/src/meeting-bot/chrome-transport.test.ts @@ -8,11 +8,17 @@ import type { createLocalMeetingRealtimeAudioTransport } from "./realtime-local- import type { createNodeMeetingRealtimeAudioTransport } from "./realtime-node-audio-transport.js"; const browserMocks = vi.hoisted(() => ({ + callNode: vi.fn(), leave: vi.fn(), open: vi.fn(), resolveLocal: vi.fn(), + resolveNode: vi.fn(), })); +vi.mock("./browser-node.js", () => ({ + callMeetingBrowserProxyOnNode: browserMocks.callNode, + resolveMeetingBrowserNode: browserMocks.resolveNode, +})); vi.mock("./browser-controller.js", () => ({ openMeetingWithBrowser: browserMocks.open, recoverMeetingBrowserTab: vi.fn(), @@ -120,6 +126,7 @@ describe.each(cases)("$name Chrome transport parity", (testCase) => { vi.clearAllMocks(); vi.spyOn(process, "platform", "get").mockReturnValue("darwin"); browserMocks.resolveLocal.mockResolvedValue(vi.fn()); + browserMocks.resolveNode.mockResolvedValue("node-1"); browserMocks.open.mockResolvedValue({ launched: true, browser: { inCall: true }, @@ -203,4 +210,81 @@ describe.each(cases)("$name Chrome transport parity", (testCase) => { ); expect(browserMocks.leave).toHaveBeenCalledTimes(testCase.expectedLeaves); }); + + it("enables output generations when the node host advertises support", async () => { + const nodeAudioTransport = { + clearOutput: vi.fn(async () => {}), + dispose: vi.fn(async () => {}), + onFatal: vi.fn(), + startInput: vi.fn(), + stop: vi.fn(async () => {}), + writeOutput: vi.fn(async () => {}), + }; + const createNodeAudioTransport = vi.fn(() => nodeAudioTransport); + const transport = createMeetingChromeTransport< + TestConfig, + TestMode, + MeetingBrowserHealth, + MeetingTranscriptSnapshot + >({ + browserNodeAdapter: platform, + isRealtimeRouteReady: () => true, + isTalkBackMode: () => true, + meetingLabel: `${testCase.name} meeting`, + nodeCommandName: platform.nodeCommandName, + outputMentionsAudioDevice: () => true, + platform, + preserveTrackedBrowserOnEngineFailure: testCase.preserveTrackedBrowserOnEngineFailure, + runtime: { + createBindings: vi.fn(() => ({ + platform: { displayName: "Test", logScope: "[test]", sessionIdPrefix: "test" }, + consultAgent: vi.fn(), + tools: [], + handleToolCall: vi.fn(), + })) as unknown as typeof createMeetingRealtimeEngineBindings, + createLocalAudioTransport: + vi.fn() as unknown as typeof createLocalMeetingRealtimeAudioTransport, + createNodeAudioTransport: + createNodeAudioTransport as unknown as typeof createNodeMeetingRealtimeAudioTransport, + startAgentRealtimeEngine: vi.fn(async () => ({ + providerId: "openai", + speak: vi.fn(), + getHealth: vi.fn(), + stop: vi.fn(async () => {}), + })) as unknown as typeof startMeetingAgentRealtimeEngine, + startRealtimeEngine: vi.fn() as unknown as typeof startMeetingRealtimeEngine, + }, + systemProfilerCommand: "/usr/sbin/system_profiler", + }); + const runtime = { + nodes: { + invoke: vi.fn(async ({ params }: { params: { action: string } }) => + params.action === "start" + ? { + payload: { + audioBridge: { type: "node-command-pair", outputGeneration: true }, + bridgeId: "bridge-1", + launched: true, + }, + } + : { payload: { ok: true } }, + ), + }, + } as unknown as PluginRuntime; + + const result = await transport.launchOnNode({ + config, + fullConfig: { transcripts: { enabled: false } } as OpenClawConfig, + logger, + meetingSessionId: "session-1", + mode: "agent", + runtime, + url: "https://example.test/meeting", + }); + + expect(result.audioBridge?.type).toBe("node-command-pair"); + expect(createNodeAudioTransport).toHaveBeenCalledWith( + expect.objectContaining({ outputGenerationSupported: true }), + ); + }); }); diff --git a/src/meeting-bot/chrome-transport.ts b/src/meeting-bot/chrome-transport.ts index 04082aff5226..c932a568ddb2 100644 --- a/src/meeting-bot/chrome-transport.ts +++ b/src/meeting-bot/chrome-transport.ts @@ -344,7 +344,7 @@ export function createMeetingChromeTransport< type MeetingNodeStartResult = { launched?: boolean; bridgeId?: string; - audioBridge?: { type?: string }; + audioBridge?: { type?: string; outputGeneration?: boolean }; browser?: Health; }; @@ -458,6 +458,7 @@ export function createMeetingChromeTransport< commandName: options.nodeCommandName, logScope: options.platform.logScope, logPrefix: params.mode === "agent" ? "node agent" : "node", + outputGenerationSupported: result.audioBridge.outputGeneration === true, }); const bindings = options.runtime.createBindings({ platform: options.platform, diff --git a/src/meeting-bot/node-host.ts b/src/meeting-bot/node-host.ts index 2de9f9d044b7..4c46685663ce 100644 --- a/src/meeting-bot/node-host.ts +++ b/src/meeting-bot/node-host.ts @@ -383,7 +383,10 @@ export function createMeetingNodeHost(options: MeetingNodeHostOptions): { const timeoutMs = readNumber(params.joinTimeoutMs, 30_000); const mode = readString(params.mode); let bridgeId: string | undefined; - let audioBridge: { type: "external-command" | "node-command-pair" } | undefined; + let audioBridge: + | { type: "external-command" } + | { type: "node-command-pair"; outputGeneration: true } + | undefined; if (mode && options.talkBackModes.has(mode)) { options.assertAudioAvailable(Math.min(timeoutMs, 10_000)); const healthCommand = readStringArray(params.audioBridgeHealthCommand); @@ -421,7 +424,7 @@ export function createMeetingNodeHost(options: MeetingNodeHostOptions): { mode, }); bridgeId = session.id; - audioBridge = { type: "node-command-pair" }; + audioBridge = { type: "node-command-pair", outputGeneration: true }; } } diff --git a/src/meeting-bot/node-invoke-policy.ts b/src/meeting-bot/node-invoke-policy.ts index 50905e79d59a..cfcf8b6a4da3 100644 --- a/src/meeting-bot/node-invoke-policy.ts +++ b/src/meeting-bot/node-invoke-policy.ts @@ -43,6 +43,10 @@ function readPositiveNumber(value: unknown): number | undefined { return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : undefined; } +function readOutputGeneration(value: unknown): number | undefined { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 ? value : undefined; +} + function copyCommand(command: string[] | undefined): string[] | undefined { return command && command.length > 0 ? [...command] : undefined; } @@ -205,11 +209,35 @@ function buildForwardParams( } forwarded.bridgeId = bridgeId; forwarded.base64 = base64; + const outputGeneration = readOutputGeneration(params.outputGeneration); + if (params.outputGeneration !== undefined && outputGeneration === undefined) { + return { + approved: false, + result: denied(options, "outputGeneration must be a non-negative safe integer"), + }; + } + if (outputGeneration !== undefined) { + forwarded.outputGeneration = outputGeneration; + } return approved(forwarded); } case "clearAudio": { const bridgeId = readString(params.bridgeId); - return bridgeId ? approved({ action, bridgeId }) : denyMissing(options, action, "bridgeId"); + if (!bridgeId) { + return denyMissing(options, action, "bridgeId"); + } + const outputGeneration = readOutputGeneration(params.outputGeneration); + if (params.outputGeneration !== undefined && outputGeneration === undefined) { + return { + approved: false, + result: denied(options, "outputGeneration must be a non-negative safe integer"), + }; + } + return approved({ + action, + bridgeId, + ...(outputGeneration !== undefined ? { outputGeneration } : {}), + }); } case "stop": { const bridgeId = readString(params.bridgeId); diff --git a/src/meeting-bot/realtime-node-audio-transport.test.ts b/src/meeting-bot/realtime-node-audio-transport.test.ts index f862ee84433d..4460bb1b9b31 100644 --- a/src/meeting-bot/realtime-node-audio-transport.test.ts +++ b/src/meeting-bot/realtime-node-audio-transport.test.ts @@ -1,7 +1,10 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { createNodeMeetingRealtimeAudioTransport } from "./realtime-node-audio-transport.js"; -function createTransport(invoke: ReturnType) { +function createTransport( + invoke: ReturnType, + options: { outputGenerationSupported?: boolean } = {}, +) { return createNodeMeetingRealtimeAudioTransport({ runtime: { nodes: { invoke } } as never, nodeId: "node-1", @@ -10,6 +13,7 @@ function createTransport(invoke: ReturnType) { commandName: "meeting.chrome", logScope: "[meeting]", logPrefix: "node", + ...options, }); } @@ -95,7 +99,7 @@ describe("node meeting realtime audio transport", () => { it("fences output writes across clear and stop", async () => { const invoke = vi.fn(async () => ({ ok: true })); - const transport = createTransport(invoke); + const transport = createTransport(invoke, { outputGenerationSupported: true }); await transport.writeOutput(Buffer.from([1, 2, 3])); await transport.clearOutput(); @@ -136,4 +140,105 @@ describe("node meeting realtime audio transport", () => { await transport.clearOutput(); expect(invoke).toHaveBeenCalledTimes(invokeCountAfterStop); }); + + it("serializes output commands for legacy node hosts", async () => { + let releasePush: (() => void) | undefined; + const invoke = vi.fn(async ({ params }: { params: { action: string } }) => { + if (params.action === "pushAudio") { + await new Promise((resolve) => { + releasePush = resolve; + }); + } + return { ok: true }; + }); + const transport = createTransport(invoke); + + const pushing = transport.writeOutput(Buffer.from([1, 2, 3])); + const clearing = transport.clearOutput(); + await vi.waitFor(() => { + expect(invoke).toHaveBeenCalledTimes(1); + }); + expect(invoke).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ + params: { + action: "pushAudio", + bridgeId: "bridge-1", + base64: Buffer.from([1, 2, 3]).toString("base64"), + }, + }), + ); + + releasePush?.(); + await Promise.all([pushing, clearing]); + expect(invoke).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + params: { + action: "clearAudio", + bridgeId: "bridge-1", + }, + }), + ); + + await transport.stop(); + }); + + it("clears immediately when the node host supports output generations", async () => { + let releasePush: (() => void) | undefined; + const invoke = vi.fn(async ({ params }: { params: { action: string } }) => { + if (params.action === "pushAudio") { + await new Promise((resolve) => { + releasePush = resolve; + }); + } + return { ok: true }; + }); + const transport = createTransport(invoke, { outputGenerationSupported: true }); + + const pushing = transport.writeOutput(Buffer.from([1, 2, 3])); + await vi.waitFor(() => { + expect(invoke).toHaveBeenCalledTimes(1); + }); + await transport.clearOutput(); + expect(invoke).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + params: { + action: "clearAudio", + bridgeId: "bridge-1", + outputGeneration: 1, + }, + }), + ); + + releasePush?.(); + await pushing; + await transport.stop(); + }); + + it("stops a legacy node host without waiting for blocked output", async () => { + let releasePush: (() => void) | undefined; + const invoke = vi.fn(async ({ params }: { params: { action: string } }) => { + if (params.action === "pushAudio") { + await new Promise((resolve) => { + releasePush = resolve; + }); + } else if (params.action === "stop") { + releasePush?.(); + } + return { ok: true }; + }); + const transport = createTransport(invoke); + + const pushing = transport.writeOutput(Buffer.from([1, 2, 3])); + const clearing = transport.clearOutput(); + await vi.waitFor(() => { + expect(invoke).toHaveBeenCalledTimes(1); + }); + await transport.stop(); + await Promise.all([pushing, clearing]); + + expect(invoke.mock.calls.map(([call]) => call.params.action)).toEqual(["pushAudio", "stop"]); + }); }); diff --git a/src/meeting-bot/realtime-node-audio-transport.ts b/src/meeting-bot/realtime-node-audio-transport.ts index 5ece650dbb41..af6ec04999ef 100644 --- a/src/meeting-bot/realtime-node-audio-transport.ts +++ b/src/meeting-bot/realtime-node-audio-transport.ts @@ -25,6 +25,7 @@ export function createNodeMeetingRealtimeAudioTransport(params: { logScope: string; logPrefix: string; audioFormat?: MeetingRealtimeAudioFormat; + outputGenerationSupported?: boolean; }): MeetingRealtimeAudioTransport { let stopped = false; let inputStarted = false; @@ -33,9 +34,21 @@ export function createNodeMeetingRealtimeAudioTransport(params: { let fatalSignaled = false; let fatalHandler: (() => void) | undefined; let outputGeneration = 0; + let legacyOutputTail = Promise.resolve(); const outputLoopbackVerifier = createMeetingOutputLoopbackVerifier({ audioFormat: params.audioFormat ?? "pcm16-24khz", }); + const runOutputCommand = (task: () => Promise): Promise => { + if (params.outputGenerationSupported) { + return task(); + } + const result = legacyOutputTail.then(task, task); + legacyOutputTail = result.then( + () => undefined, + () => undefined, + ); + return result; + }; const signalFatal = () => { if (!fatalSignaled) { fatalSignaled = true; @@ -131,16 +144,21 @@ export function createNodeMeetingRealtimeAudioTransport(params: { const generation = outputGeneration; outputLoopbackVerifier.recordOutput(audio); try { - await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: params.commandName, - params: { - action: "pushAudio", - bridgeId: params.bridgeId, - base64: audio.toString("base64"), - outputGeneration: generation, - }, - timeoutMs: 5_000, + await runOutputCommand(async () => { + if (stopped) { + return; + } + await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: params.commandName, + params: { + action: "pushAudio", + bridgeId: params.bridgeId, + base64: audio.toString("base64"), + ...(params.outputGenerationSupported ? { outputGeneration: generation } : {}), + }, + timeoutMs: 5_000, + }); }); } catch (error) { if (!stopped && generation === outputGeneration) { @@ -155,15 +173,20 @@ export function createNodeMeetingRealtimeAudioTransport(params: { } outputGeneration += 1; outputLoopbackVerifier.cancelOutput(); - await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: params.commandName, - params: { - action: "clearAudio", - bridgeId: params.bridgeId, - outputGeneration, - }, - timeoutMs: 5_000, + await runOutputCommand(async () => { + if (stopped) { + return; + } + await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: params.commandName, + params: { + action: "clearAudio", + bridgeId: params.bridgeId, + ...(params.outputGenerationSupported ? { outputGeneration } : {}), + }, + timeoutMs: 5_000, + }); }); }, dispose: async () => {