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

fix(meeting-bot): negotiate realtime output generations

Browse Source
This commit is contained in:
Vincent Koc
2026-07-31 10:39:48 +08:00
parent 4bb0563757
commit 1bd2f93214
10 changed files with 316 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
extensions/google-meet/index.test.ts
View File

@@ -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,

2
extensions/google-meet/node-host.test.ts
View File

@@ -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,
});

42
extensions/google-meet/src/node-invoke-policy.test.ts
View File

@@ -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({

5
extensions/google-meet/src/transports/chrome.ts
View File

@@ -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,

84
src/meeting-bot/chrome-transport.test.ts
View File

@@ -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 }),
);
});
});

3
src/meeting-bot/chrome-transport.ts
View File

@@ -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,

7
src/meeting-bot/node-host.ts
View File

@@ -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 };
}
}

30
src/meeting-bot/node-invoke-policy.ts
View File

@@ -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);

109
src/meeting-bot/realtime-node-audio-transport.test.ts
View File

@@ -1,7 +1,10 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { createNodeMeetingRealtimeAudioTransport } from "./realtime-node-audio-transport.js";
function createTransport(invoke: ReturnType<typeof vi.fn>) {
function createTransport(
invoke: ReturnType<typeof vi.fn>,
options: { outputGenerationSupported?: boolean } = {},
) {
return createNodeMeetingRealtimeAudioTransport({
runtime: { nodes: { invoke } } as never,
nodeId: "node-1",
@@ -10,6 +13,7 @@ function createTransport(invoke: ReturnType<typeof vi.fn>) {
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<void>((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<void>((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<void>((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"]);
});
});

61
src/meeting-bot/realtime-node-audio-transport.ts
View File

@@ -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 = <T>(task: () => Promise<T>): Promise<T> => {
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 () => {