From edd052989b5da1f2408a62c8d20d42e443fcda3f Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 17:39:53 -0700 Subject: [PATCH] refactor(google-meet): merge host and node realtime engines behind one audio transport (#109413) * refactor(google-meet): merge host and node realtime engines behind one audio transport * refactor(google-meet): internalize engine helpers left test-only after realtime-node removal --- config/max-lines-baseline.txt | 1 - extensions/google-meet/index.test.ts | 387 ++++++--- .../google-meet/src/realtime-audio-format.ts | 108 +++ .../src/realtime-audio-transport.ts | 13 + .../src/realtime-local-audio-transport.ts | 283 +++++++ .../src/realtime-node-audio-transport.ts | 140 ++++ extensions/google-meet/src/realtime-node.ts | 772 ------------------ .../google-meet/src/realtime.process.test.ts | 168 +++- extensions/google-meet/src/realtime.ts | 622 ++++---------- .../google-meet/src/transports/chrome.ts | 103 ++- 10 files changed, 1214 insertions(+), 1383 deletions(-) create mode 100644 extensions/google-meet/src/realtime-audio-format.ts create mode 100644 extensions/google-meet/src/realtime-audio-transport.ts create mode 100644 extensions/google-meet/src/realtime-local-audio-transport.ts create mode 100644 extensions/google-meet/src/realtime-node-audio-transport.ts delete mode 100644 extensions/google-meet/src/realtime-node.ts diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index 5f081336af01..a3aad5421d46 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -134,7 +134,6 @@ extensions/google-meet/index.ts extensions/google-meet/src/cli.test.ts extensions/google-meet/src/cli.ts extensions/google-meet/src/meet.ts -extensions/google-meet/src/realtime-node.ts extensions/google-meet/src/realtime.ts extensions/google-meet/src/runtime.ts extensions/google-meet/src/transports/chrome.ts diff --git a/extensions/google-meet/index.test.ts b/extensions/google-meet/index.test.ts index daad76343061..0f55859e141c 100644 --- a/extensions/google-meet/index.test.ts +++ b/extensions/google-meet/index.test.ts @@ -25,17 +25,11 @@ import { fetchGoogleMeetSpace, } from "./src/meet.js"; import { handleGoogleMeetNodeHostCommand } from "./src/node-host.js"; -import { startNodeRealtimeAudioBridge } from "./src/realtime-node.js"; -import { - convertGoogleMeetTtsAudioForBridge, - isGoogleMeetLikelyAssistantEchoTranscript, - GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS, - recordGoogleMeetOutputActivity, - resolveGoogleMeetRealtimeProvider, - resolveGoogleMeetRealtimeTranscriptionProvider, - startCommandAgentAudioBridge, - startCommandRealtimeAudioBridge, -} from "./src/realtime.js"; +import { convertGoogleMeetTtsAudioForBridge } from "./src/realtime-audio-format.js"; +import type { MeetRealtimeAudioTransport } from "./src/realtime-audio-transport.js"; +import { createLocalMeetRealtimeAudioTransport } from "./src/realtime-local-audio-transport.js"; +import { createNodeMeetRealtimeAudioTransport } from "./src/realtime-node-audio-transport.js"; +import { startMeetAgentRealtimeEngine, startMeetRealtimeEngine } from "./src/realtime.js"; import { GoogleMeetRuntime } from "./src/runtime.js"; import { invokeGoogleMeetGatewayMethodForTest, @@ -55,6 +49,117 @@ type GoogleMeetManifestConfigSchema = JsonSchemaObject & { properties?: Record }>; }; +const TEST_TALKBACK_DEBOUNCE_MS = 900; + +type MeetRealtimeAudioSpawn = NonNullable< + Parameters[0]["spawn"] +>; + +function createTestMeetRealtimeAudioTransport(): { + transport: MeetRealtimeAudioTransport; + deliverInput: (audio: Buffer) => void; +} { + let inputHandler: ((audio: Buffer) => void) | undefined; + const transport: MeetRealtimeAudioTransport = { + onFatal: vi.fn(), + startInput: vi.fn((handler) => { + inputHandler = handler; + }), + stop: vi.fn(async () => {}), + writeOutput: vi.fn(async () => {}), + clearOutput: vi.fn(async () => {}), + dispose: vi.fn(async () => {}), + }; + return { + transport, + deliverInput: (audio) => { + if (!inputHandler) { + throw new Error("Expected Google Meet realtime input to be started"); + } + inputHandler(audio); + }, + }; +} + +type TestLocalAgentEngineParams = Omit< + Parameters[0], + "transport" +> & { + inputCommand: string[]; + outputCommand: string[]; + spawn?: MeetRealtimeAudioSpawn; +}; + +async function startTestLocalAgentAudioBridge(params: TestLocalAgentEngineParams) { + const { inputCommand, outputCommand, spawn, ...engineParams } = params; + const transport = createLocalMeetRealtimeAudioTransport({ + inputCommand, + outputCommand, + bargeInInputCommand: params.config.chrome.bargeInInputCommand, + bargeInRmsThreshold: params.config.chrome.bargeInRmsThreshold, + bargeInPeakThreshold: params.config.chrome.bargeInPeakThreshold, + bargeInCooldownMs: params.config.chrome.bargeInCooldownMs, + logger: params.logger, + spawn, + }); + return await startMeetAgentRealtimeEngine({ ...engineParams, transport }); +} + +type TestLocalRealtimeEngineParams = Omit< + Parameters[0], + "transport" +> & { + inputCommand: string[]; + outputCommand: string[]; + spawn?: MeetRealtimeAudioSpawn; +}; + +async function startTestLocalRealtimeAudioBridge(params: TestLocalRealtimeEngineParams) { + const { inputCommand, outputCommand, spawn, ...engineParams } = params; + const transport = createLocalMeetRealtimeAudioTransport({ + inputCommand, + outputCommand, + bargeInInputCommand: params.config.chrome.bargeInInputCommand, + bargeInRmsThreshold: params.config.chrome.bargeInRmsThreshold, + bargeInPeakThreshold: params.config.chrome.bargeInPeakThreshold, + bargeInCooldownMs: params.config.chrome.bargeInCooldownMs, + logger: params.logger, + spawn, + }); + return await startMeetRealtimeEngine({ ...engineParams, transport }); +} + +type TestNodeRealtimeEngineParams = Omit< + Parameters[0], + "transport" +> & { + nodeId: string; + bridgeId: string; +}; + +async function startTestNodeRealtimeAudioBridge(params: TestNodeRealtimeEngineParams) { + const { nodeId, bridgeId, ...engineParams } = params; + const transport = createNodeMeetRealtimeAudioTransport({ + runtime: params.runtime, + nodeId, + bridgeId, + logger: params.logger, + logPrefix: "node", + }); + return { + type: "node-command-pair" as const, + nodeId, + bridgeId, + ...(await startMeetRealtimeEngine({ + ...engineParams, + logPrefix: "node", + talkSessionId: `google-meet:${params.meetingSessionId}:${bridgeId}:node-realtime`, + talkContext: { nodeId, bridgeId }, + transport, + })), + }; +} + const voiceCallMocks = vi.hoisted(() => ({ createVoiceCallGateway: vi.fn( ({ runtime }: { runtime: { gateway: unknown } }) => runtime.gateway, @@ -779,7 +884,19 @@ describe("google-meet plugin", () => { expect(google.realtime.transcriptionProvider).toBe("openai"); }); - it("uses voiceProvider for bidi and transcriptionProvider for agent mode resolution", () => { + it("uses voiceProvider for bidi and transcriptionProvider for agent mode resolution", async () => { + let voiceRequest: Parameters[0] | undefined; + const voiceBridge = { + connect: vi.fn(async () => {}), + sendAudio: vi.fn(), + sendUserMessage: vi.fn(), + setMediaTimestamp: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + triggerGreeting: vi.fn(), + isConnected: vi.fn(() => true), + }; const voiceProviders: RealtimeVoiceProviderPlugin[] = [ { id: "openai", @@ -796,19 +913,30 @@ describe("google-meet plugin", () => { autoSelectOrder: 2, resolveConfig: ({ rawConfig }) => rawConfig, isConfigured: () => true, - createBridge: () => { - throw new Error("unused"); + createBridge: (request) => { + voiceRequest = request; + return voiceBridge; }, }, ]; + let transcriptionRequest: + | Parameters[0] + | undefined; + const transcriptionSession = { + connect: vi.fn(async () => {}), + sendAudio: vi.fn(), + close: vi.fn(), + isConnected: vi.fn(() => true), + }; const transcriptionProviders: RealtimeTranscriptionProviderPlugin[] = [ { id: "openai", label: "OpenAI", autoSelectOrder: 1, isConfigured: () => true, - createSession: () => { - throw new Error("unused"); + createSession: (request) => { + transcriptionRequest = request; + return transcriptionSession; }, }, ]; @@ -821,21 +949,35 @@ describe("google-meet plugin", () => { }, }); - const voiceProvider = resolveGoogleMeetRealtimeProvider({ + const voiceTransport = createTestMeetRealtimeAudioTransport(); + const voiceHandle = await startMeetRealtimeEngine({ config, fullConfig: {} as never, + runtime: {} as never, + meetingSessionId: "provider-resolution-voice", + logger: noopLogger, providers: voiceProviders, + transport: voiceTransport.transport, }); - expect(voiceProvider.provider.id).toBe("google"); - expect(voiceProvider.providerConfig).toEqual({ + expect(voiceHandle.providerId).toBe("google"); + expect(voiceRequest?.providerConfig).toEqual({ model: "gemini-2.5-flash-native-audio-preview-12-2025", }); - const transcriptionProvider = resolveGoogleMeetRealtimeTranscriptionProvider({ + await voiceHandle.stop(); + + const transcriptionTransport = createTestMeetRealtimeAudioTransport(); + const transcriptionHandle = await startMeetAgentRealtimeEngine({ config, fullConfig: {} as never, + runtime: {} as never, + meetingSessionId: "provider-resolution-transcription", + logger: noopLogger, providers: transcriptionProviders, + transport: transcriptionTransport.transport, }); - expect(transcriptionProvider.provider.id).toBe("openai"); + expect(transcriptionHandle.providerId).toBe("openai"); + expect(transcriptionRequest).toBeDefined(); + await transcriptionHandle.stop(); }); it("declares advanced config metadata in the plugin entry and manifest", () => { @@ -6710,7 +6852,7 @@ describe("google-meet plugin", () => { }, }; - const handle = await startCommandAgentAudioBridge({ + const handle = await startTestLocalAgentAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { provider: "openai", agentId: "jay", introMessage: "" }, }), @@ -6729,7 +6871,7 @@ describe("google-meet plugin", () => { ); inputStdout.write(Buffer.from([1, 0, 2, 0, 3, 0, 4, 0])); callbacks?.onTranscript?.("Please summarize the launch."); - await vi.advanceTimersByTimeAsync(GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS); + await vi.advanceTimersByTimeAsync(TEST_TALKBACK_DEBOUNCE_MS); expect(sendAudio).toHaveBeenCalledTimes(1); const audioChunk = mockCallArg(sendAudio, 0) as Buffer; @@ -6770,6 +6912,43 @@ describe("google-meet plugin", () => { await handle.stop(); }); + it("aborts agent engine startup when the transport already failed", async () => { + const createSession = vi.fn(); + const provider: RealtimeTranscriptionProviderPlugin = { + id: "openai", + label: "OpenAI", + defaultModel: "gpt-4o-transcribe", + autoSelectOrder: 1, + resolveConfig: ({ rawConfig }) => rawConfig, + isConfigured: () => true, + createSession, + }; + const stop = vi.fn(async () => {}); + await expect( + startMeetAgentRealtimeEngine({ + config: resolveGoogleMeetConfig({ + realtime: { provider: "openai", agentId: "jay", introMessage: "" }, + }), + fullConfig: {} as never, + runtime: {} as never, + meetingSessionId: "meet-fatal", + logger: noopLogger, + providers: [provider], + transport: { + // Pre-registration fatal replays synchronously; startup must abort before createSession. + onFatal: (handler) => handler(), + startInput: vi.fn(), + stop, + writeOutput: vi.fn(async () => {}), + clearOutput: vi.fn(async () => {}), + dispose: vi.fn(async () => {}), + }, + }), + ).rejects.toThrow("transport failed before transcription provider setup"); + expect(createSession).not.toHaveBeenCalled(); + expect(stop).toHaveBeenCalled(); + }); + it("stops the Chrome agent audio bridge when child stdio streams error", async () => { const sttSession = { connect: vi.fn(async () => {}), @@ -6822,7 +7001,7 @@ describe("google-meet plugin", () => { }); const spawnMock = vi.fn().mockReturnValueOnce(outputProcess).mockReturnValueOnce(inputProcess); - const handle = await startCommandAgentAudioBridge({ + const handle = await startTestLocalAgentAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { provider: "openai", agentId: "jay", introMessage: "" }, }), @@ -6957,7 +7136,7 @@ describe("google-meet plugin", () => { }, }; - const handle = await startCommandRealtimeAudioBridge({ + const handle = await startTestLocalRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { strategy: "bidi", provider: "openai", model: "gpt-realtime", agentId: "jay" }, }), @@ -7154,7 +7333,7 @@ describe("google-meet plugin", () => { const inputProcess = makeProcess({ stdin: null, stdout: inputStdout }); const spawnMock = vi.fn().mockReturnValueOnce(outputProcess).mockReturnValueOnce(inputProcess); - const handle = await startCommandRealtimeAudioBridge({ + const handle = await startTestLocalRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { strategy: "bidi", provider: "openai", model: "gpt-realtime" }, }), @@ -7250,7 +7429,7 @@ describe("google-meet plugin", () => { }, }; - const handle = await startCommandRealtimeAudioBridge({ + const handle = await startTestLocalRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { provider: "openai", agentId: "jay" } }), fullConfig: {} as never, runtime: runtime as never, @@ -7267,10 +7446,24 @@ describe("google-meet plugin", () => { } expect(callbacks.autoRespondToAudio).toBe(false); expect(callbacks.tools).toStrictEqual([]); + callbacks.onTranscript?.( + "assistant", + "Hi Molty, glad to have you here. Let me know if there's anything specific you'd like to cover or if you need any support during the meeting.", + true, + ); + callbacks.onTranscript?.( + "user", + "Let me know if there's anything specific you'd like to cover or if you need any support during the", + true, + ); + await vi.advanceTimersByTimeAsync(TEST_TALKBACK_DEBOUNCE_MS); + expect(runtime.agent.runEmbeddedAgent).not.toHaveBeenCalled(); + + callbacks.onTranscript?.("user", "yes yes yes yes", true); callbacks?.onTranscript?.("user", "Are we still on track?", true); callbacks?.onTranscript?.("user", "Please include launch blockers.", true); - await vi.advanceTimersByTimeAsync(GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS); + await vi.advanceTimersByTimeAsync(TEST_TALKBACK_DEBOUNCE_MS); await vi.waitFor(() => { expect(runtime.agent.runEmbeddedAgent).toHaveBeenCalledTimes(1); }); @@ -7282,6 +7475,7 @@ describe("google-meet plugin", () => { expect(consultArgs.spawnedBy).toBe("agent:jay:main"); expect(consultArgs.sessionKey).toBe("agent:jay:subagent:google-meet:meet-1"); expect(consultArgs.sandboxSessionKey).toBe("agent:jay:subagent:google-meet:meet-1"); + expect(JSON.stringify(consultArgs)).toContain("yes yes yes yes"); expect(JSON.stringify(consultArgs)).toContain( "Are we still on track?\\nPlease include launch blockers.", ); @@ -7297,75 +7491,68 @@ describe("google-meet plugin", () => { } }); - it("tracks queued playback time when suppressing realtime input echo", () => { - const markPlaybackStarted = vi.fn(); - const markAudio = vi.fn(); - const tracker = { - markPlaybackStarted, - markAudio, - } as unknown as Parameters[0]["tracker"]; - const first = recordGoogleMeetOutputActivity({ - tracker, - audio: Buffer.alloc(48_000), - audioFormat: "pcm16-24khz", - nowMs: 1_000, - lastOutputPlayableUntilMs: 0, - suppressInputUntilMs: 0, - }); - const second = recordGoogleMeetOutputActivity({ - tracker, - audio: Buffer.alloc(48_000), - audioFormat: "pcm16-24khz", - nowMs: 1_100, - lastOutputPlayableUntilMs: first.lastOutputPlayableUntilMs, - suppressInputUntilMs: first.suppressInputUntilMs, - }); + it("extends realtime input suppression across queued playback", async () => { + vi.useFakeTimers(); + try { + vi.setSystemTime(1_000); + let callbacks: Parameters[0] | undefined; + const bridge = { + connect: vi.fn(async () => {}), + sendAudio: vi.fn(), + sendUserMessage: vi.fn(), + setMediaTimestamp: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + triggerGreeting: vi.fn(), + isConnected: vi.fn(() => true), + }; + const provider: RealtimeVoiceProviderPlugin = { + id: "openai", + label: "OpenAI", + autoSelectOrder: 1, + resolveConfig: ({ rawConfig }) => rawConfig, + isConfigured: () => true, + createBridge: (request) => { + callbacks = request; + return bridge; + }, + }; + const audioTransport = createTestMeetRealtimeAudioTransport(); + const handle = await startMeetRealtimeEngine({ + config: resolveGoogleMeetConfig({ + realtime: { strategy: "bidi", provider: "openai", model: "gpt-realtime" }, + }), + fullConfig: {} as never, + runtime: {} as never, + meetingSessionId: "queued-playback", + logger: noopLogger, + providers: [provider], + transport: audioTransport.transport, + }); + if (!callbacks) { + throw new Error("Expected realtime bridge callbacks"); + } - expect(first.durationMs).toBe(1_000); - expect(first.lastOutputPlayableUntilMs).toBe(2_000); - expect(first.suppressInputUntilMs).toBe(5_000); - expect(second.durationMs).toBe(1_000); - expect(second.lastOutputPlayableUntilMs).toBe(3_000); - expect(second.suppressInputUntilMs).toBe(6_000); - expect(markPlaybackStarted).toHaveBeenCalledTimes(2); - expect(markAudio).toHaveBeenNthCalledWith(1, { - audioMs: 1_000, - sourceAudioBytes: 48_000, - sinkAudioBytes: 48_000, - }); - }); + callbacks.onAudio(Buffer.alloc(48_000)); + vi.setSystemTime(1_100); + callbacks.onAudio(Buffer.alloc(48_000)); + vi.setSystemTime(5_500); + audioTransport.deliverInput(Buffer.from([1, 2, 3, 4])); + expect(bridge.sendAudio).not.toHaveBeenCalled(); - it("detects assistant transcript echoes before agent consult", () => { - const nowMs = Date.parse("2026-05-04T01:00:00.000Z"); - const transcript = [ - { - at: new Date(nowMs - 1_000).toISOString(), - role: "assistant" as const, - text: "Hi Molty, glad to have you here. Let me know if there's anything specific you'd like to cover or if you need any support during the meeting.", - }, - ]; - - expect( - isGoogleMeetLikelyAssistantEchoTranscript({ - transcript, - text: "Let me know if there's anything specific you'd like to cover or if you need any support during the", - nowMs, - }), - ).toBe(true); - expect( - isGoogleMeetLikelyAssistantEchoTranscript({ - transcript, - text: "Tell me a story.", - nowMs, - }), - ).toBe(false); - expect( - isGoogleMeetLikelyAssistantEchoTranscript({ - transcript, - text: "yes yes yes yes", - nowMs, - }), - ).toBe(false); + vi.setSystemTime(6_001); + audioTransport.deliverInput(Buffer.from([5, 6, 7])); + expect(bridge.sendAudio).toHaveBeenCalledWith(Buffer.from([5, 6, 7])); + expect(handle.getHealth()).toMatchObject({ + lastInputBytes: 3, + lastOutputBytes: 96_000, + suppressedInputBytes: 4, + }); + await handle.stop(); + } finally { + vi.useRealTimers(); + } }); it("uses a local barge-in input command to clear active Chrome playback", async () => { @@ -7434,7 +7621,7 @@ describe("google-meet plugin", () => { .mockReturnValueOnce(bargeInProcess) .mockReturnValueOnce(replacementOutputProcess); - const handle = await startCommandRealtimeAudioBridge({ + const handle = await startTestLocalRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ chrome: { bargeInInputCommand: ["capture-human"], @@ -7536,7 +7723,7 @@ describe("google-meet plugin", () => { }, }; - const handle = await startNodeRealtimeAudioBridge({ + const handle = await startTestNodeRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { strategy: "bidi", provider: "openai", model: "gpt-realtime" }, }), @@ -7720,10 +7907,10 @@ describe("google-meet plugin", () => { }), }, }; - let handle: Awaited> | undefined; + let handle: Awaited> | undefined; try { - handle = await startNodeRealtimeAudioBridge({ + handle = await startTestNodeRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { provider: "openai", model: "gpt-realtime" }, }), @@ -7788,7 +7975,7 @@ describe("google-meet plugin", () => { }; try { - const handle = await startNodeRealtimeAudioBridge({ + const handle = await startTestNodeRealtimeAudioBridge({ config: resolveGoogleMeetConfig({ realtime: { provider: "openai", model: "gpt-realtime" }, }), diff --git a/extensions/google-meet/src/realtime-audio-format.ts b/extensions/google-meet/src/realtime-audio-format.ts new file mode 100644 index 000000000000..99d6e98c1f7a --- /dev/null +++ b/extensions/google-meet/src/realtime-audio-format.ts @@ -0,0 +1,108 @@ +import { + convertPcmToMulaw8k, + mulawToPcm, + REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ, + REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, + resamplePcm, +} from "openclaw/plugin-sdk/realtime-voice"; +import type { GoogleMeetConfig } from "./config.js"; + +export function resolveGoogleMeetRealtimeAudioFormat(config: GoogleMeetConfig) { + return config.chrome.audioFormat === "g711-ulaw-8khz" + ? REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ + : REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ; +} + +export function convertGoogleMeetBridgeAudioForStt( + audio: Buffer, + config: GoogleMeetConfig, +): Buffer { + if (config.chrome.audioFormat === "g711-ulaw-8khz") { + return audio; + } + return convertPcmToMulaw8k(audio, 24_000); +} + +export function convertGoogleMeetTtsAudioForBridge( + audio: Buffer, + sampleRate: number, + config: GoogleMeetConfig, + outputFormat?: string, +): Buffer { + const sourceFormat = sourceTelephonyTtsFormat(outputFormat); + if ( + config.chrome.audioFormat === "g711-ulaw-8khz" && + sourceFormat === "mulaw" && + sampleRate === 8_000 + ) { + return audio; + } + const pcm = decodeGoogleMeetTelephonyTtsAudio(audio, sourceFormat); + return config.chrome.audioFormat === "g711-ulaw-8khz" + ? convertPcmToMulaw8k(pcm, sampleRate) + : resamplePcm(pcm, sampleRate, 24_000); +} + +type GoogleMeetTelephonyTtsFormat = "pcm" | "mulaw" | "alaw"; + +function sourceTelephonyTtsFormat(outputFormat: string | undefined): GoogleMeetTelephonyTtsFormat { + const normalized = outputFormat?.trim().toLowerCase().replaceAll("_", "-") ?? ""; + if ( + !normalized || + normalized === "pcm" || + normalized.startsWith("pcm-") || + normalized.includes("pcm16") || + normalized.includes("16bit-mono-pcm") + ) { + return "pcm"; + } + if ( + normalized === "mulaw" || + normalized === "ulaw" || + normalized.includes("mu-law") || + normalized.includes("mulaw") || + normalized.includes("ulaw") + ) { + return "mulaw"; + } + if (normalized === "alaw" || normalized.includes("a-law") || normalized.includes("alaw")) { + return "alaw"; + } + throw new Error(`Unsupported telephony TTS output format for Google Meet: ${outputFormat}`); +} + +function decodeGoogleMeetTelephonyTtsAudio( + audio: Buffer, + sourceFormat: GoogleMeetTelephonyTtsFormat, +): Buffer { + switch (sourceFormat) { + case "pcm": + return audio; + case "mulaw": + return mulawToPcm(audio); + case "alaw": + return alawToPcm(audio); + } + return unsupportedGoogleMeetTelephonyTtsFormat(sourceFormat); +} + +function unsupportedGoogleMeetTelephonyTtsFormat(_format: never): never { + throw new Error("Unsupported telephony TTS output format for Google Meet"); +} + +function alawToPcm(alaw: Buffer): Buffer { + const pcm = Buffer.alloc(alaw.length * 2); + for (let index = 0; index < alaw.length; index += 1) { + pcm.writeInt16LE(alawByteToLinear(alaw[index] ?? 0), index * 2); + } + return pcm; +} + +function alawByteToLinear(value: number): number { + const aLaw = value ^ 0x55; + const sign = aLaw & 0x80; + const exponent = (aLaw & 0x70) >> 4; + const mantissa = aLaw & 0x0f; + const sample = exponent === 0 ? (mantissa << 4) + 8 : ((mantissa << 4) + 0x108) << (exponent - 1); + return sign ? sample : -sample; +} diff --git a/extensions/google-meet/src/realtime-audio-transport.ts b/extensions/google-meet/src/realtime-audio-transport.ts new file mode 100644 index 000000000000..af3d91cb37a3 --- /dev/null +++ b/extensions/google-meet/src/realtime-audio-transport.ts @@ -0,0 +1,13 @@ +import type { GoogleMeetChromeHealth } from "./transports/types.js"; + +export interface MeetRealtimeAudioTransport { + /** Delivers a prior failure immediately so provider setup cannot outrun transport teardown. */ + onFatal(handler: () => void): void; + startInput(onAudio: (audio: Buffer) => void): void; + stop(): Promise; + writeOutput(audio: Buffer): Promise; + clearOutput(): Promise; + dispose(): Promise; + getHealth?(): Pick; + startBargeInMonitor?(onBargeIn: (audio: Buffer) => boolean): void; +} diff --git a/extensions/google-meet/src/realtime-local-audio-transport.ts b/extensions/google-meet/src/realtime-local-audio-transport.ts new file mode 100644 index 000000000000..6bf2db8c9816 --- /dev/null +++ b/extensions/google-meet/src/realtime-local-audio-transport.ts @@ -0,0 +1,283 @@ +import { spawn } from "node:child_process"; +import type { Writable } from "node:stream"; +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import type { RuntimeLogger } from "openclaw/plugin-sdk/plugin-runtime"; +import type { MeetRealtimeAudioTransport } from "./realtime-audio-transport.js"; + +type BridgeProcess = { + pid?: number; + killed?: boolean; + stdin?: Writable | null; + stdout?: { + on(event: "data", listener: (chunk: Buffer | string) => void): unknown; + on(event: "error", listener: (error: Error) => void): unknown; + } | null; + stderr?: { + on(event: "data", listener: (chunk: Buffer | string) => void): unknown; + on(event: "error", listener: (error: Error) => void): unknown; + } | null; + kill(signal?: NodeJS.Signals): boolean; + on( + event: "exit", + listener: (code: number | null, signal: NodeJS.Signals | null) => void, + ): unknown; + on(event: "error", listener: (error: Error) => void): unknown; +}; + +type MeetRealtimeAudioSpawn = ( + command: string, + args: string[], + options: { stdio: ["pipe" | "ignore", "pipe" | "ignore", "pipe" | "ignore"] }, +) => BridgeProcess; + +function splitCommand(argv: string[]): { command: string; args: string[] } { + const [command, ...args] = argv; + if (!command) { + throw new Error("audio bridge command must not be empty"); + } + return { command, args }; +} + +function terminateBridgeProcess(proc: BridgeProcess, signal: NodeJS.Signals = "SIGTERM"): void { + if (proc.killed && signal !== "SIGKILL") { + return; + } + let exited = false; + proc.on("exit", () => { + exited = true; + }); + try { + proc.kill(signal); + } catch { + return; + } + if (signal === "SIGKILL") { + return; + } + const timer = setTimeout(() => { + if (!exited) { + try { + proc.kill("SIGKILL"); + } catch { + // The process may exit between the grace check and signal. + } + } + }, 1_000); + timer.unref?.(); +} + +function readPcm16Stats(audio: Buffer): { rms: number; peak: number } { + let sumSquares = 0; + let peak = 0; + let samples = 0; + for (let offset = 0; offset + 1 < audio.byteLength; offset += 2) { + const sample = audio.readInt16LE(offset); + const abs = Math.abs(sample); + peak = Math.max(peak, abs); + sumSquares += sample * sample; + samples += 1; + } + return { + rms: samples > 0 ? Math.sqrt(sumSquares / samples) : 0, + peak, + }; +} + +export function createLocalMeetRealtimeAudioTransport(params: { + inputCommand: string[]; + outputCommand: string[]; + bargeInInputCommand?: string[]; + bargeInRmsThreshold: number; + bargeInPeakThreshold: number; + bargeInCooldownMs: number; + logger: RuntimeLogger; + spawn?: MeetRealtimeAudioSpawn; +}): MeetRealtimeAudioTransport { + const input = splitCommand(params.inputCommand); + const output = splitCommand(params.outputCommand); + const spawnFn: MeetRealtimeAudioSpawn = + params.spawn ?? + ((command, args, options) => spawn(command, args, options) as unknown as BridgeProcess); + const spawnOutputProcess = () => + spawnFn(output.command, output.args, { stdio: ["pipe", "ignore", "pipe"] }); + let outputProcess = spawnOutputProcess(); + const inputProcess = spawnFn(input.command, input.args, { + stdio: ["ignore", "pipe", "pipe"], + }); + let bargeInInputProcess: BridgeProcess | undefined; + let stopped = false; + let inputStarted = false; + let fatalSignaled = false; + let fatalHandler: (() => void) | undefined; + + const signalFatal = () => { + if (!fatalSignaled) { + fatalSignaled = true; + fatalHandler?.(); + } + }; + const fail = (label: string) => (error: Error) => { + params.logger.warn(`[google-meet] ${label} failed: ${formatErrorMessage(error)}`); + signalFatal(); + }; + const attachOutputProcessHandlers = (proc: BridgeProcess) => { + proc.on("error", (error) => { + if (proc === outputProcess) { + fail("audio output command")(error); + } + }); + proc.stdin?.on?.("error", (error: Error) => { + if (proc === outputProcess) { + fail("audio output command")(error); + } + }); + proc.on("exit", (code, signal) => { + if (proc === outputProcess && !stopped) { + params.logger.warn( + `[google-meet] audio output command exited (${code ?? signal ?? "done"})`, + ); + signalFatal(); + } + }); + proc.stderr?.on("data", (chunk) => { + params.logger.debug?.(`[google-meet] audio output: ${String(chunk).trim()}`); + }); + proc.stderr?.on("error", (error: Error) => { + if (proc === outputProcess) { + fail("audio output command stderr")(error); + } + }); + }; + attachOutputProcessHandlers(outputProcess); + inputProcess.on("error", fail("audio input command")); + inputProcess.on("exit", (code, signal) => { + if (!stopped) { + params.logger.warn(`[google-meet] audio input command exited (${code ?? signal ?? "done"})`); + signalFatal(); + } + }); + inputProcess.stderr?.on("data", (chunk) => { + params.logger.debug?.(`[google-meet] audio input: ${String(chunk).trim()}`); + }); + inputProcess.stdout?.on("error", fail("audio input command stdout")); + inputProcess.stderr?.on("error", fail("audio input command stderr")); + + const transport: MeetRealtimeAudioTransport = { + onFatal: (handler) => { + fatalHandler = handler; + if (fatalSignaled) { + handler(); + } + }, + startInput: (onAudio) => { + if (inputStarted) { + throw new Error("audio input transport already started"); + } + inputStarted = true; + inputProcess.stdout?.on("data", (chunk) => { + if (!stopped) { + onAudio(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + }); + }, + stop: async () => { + if (stopped) { + return; + } + stopped = true; + terminateBridgeProcess(inputProcess); + terminateBridgeProcess(outputProcess); + if (bargeInInputProcess) { + terminateBridgeProcess(bargeInInputProcess); + } + }, + writeOutput: async (audio) => { + if (stopped) { + return; + } + try { + outputProcess.stdin?.write(audio); + } catch (error) { + fail("audio output command")(error as Error); + } + }, + clearOutput: async () => { + if (stopped) { + return; + } + const previousOutput = outputProcess; + outputProcess = spawnOutputProcess(); + attachOutputProcessHandlers(outputProcess); + params.logger.debug?.( + "[google-meet] cleared realtime audio output buffer by restarting playback command", + ); + terminateBridgeProcess(previousOutput, "SIGKILL"); + }, + dispose: async () => { + await transport.stop(); + }, + }; + + if (!params.bargeInInputCommand) { + return transport; + } + + return { + ...transport, + startBargeInMonitor: (onBargeIn) => { + if (bargeInInputProcess || stopped) { + return; + } + const command = splitCommand(params.bargeInInputCommand ?? []); + let lastBargeInAt = 0; + bargeInInputProcess = spawnFn(command.command, command.args, { + stdio: ["ignore", "pipe", "pipe"], + }); + bargeInInputProcess.stdout?.on("data", (chunk) => { + const audio = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + const now = Date.now(); + if (stopped || now - lastBargeInAt < params.bargeInCooldownMs) { + return; + } + const stats = readPcm16Stats(audio); + if (stats.rms < params.bargeInRmsThreshold && stats.peak < params.bargeInPeakThreshold) { + return; + } + if (!onBargeIn(audio)) { + return; + } + lastBargeInAt = now; + params.logger.debug?.( + `[google-meet] human barge-in detected by local input (rms=${Math.round( + stats.rms, + )}, peak=${stats.peak})`, + ); + }); + bargeInInputProcess.stdout?.on("error", (error: Error) => { + params.logger.warn( + `[google-meet] human barge-in input stdout failed: ${formatErrorMessage(error)}`, + ); + }); + bargeInInputProcess.stderr?.on("data", (chunk) => { + params.logger.debug?.(`[google-meet] barge-in input: ${String(chunk).trim()}`); + }); + bargeInInputProcess.stderr?.on("error", (error: Error) => { + params.logger.warn( + `[google-meet] human barge-in input stderr failed: ${formatErrorMessage(error)}`, + ); + }); + bargeInInputProcess.on("error", (error) => { + params.logger.warn( + `[google-meet] human barge-in input failed: ${formatErrorMessage(error)}`, + ); + }); + bargeInInputProcess.on("exit", (code, signal) => { + if (!stopped) { + params.logger.debug?.( + `[google-meet] human barge-in input exited (${code ?? signal ?? "done"})`, + ); + } + }); + }, + }; +} diff --git a/extensions/google-meet/src/realtime-node-audio-transport.ts b/extensions/google-meet/src/realtime-node-audio-transport.ts new file mode 100644 index 000000000000..7bff85f119b5 --- /dev/null +++ b/extensions/google-meet/src/realtime-node-audio-transport.ts @@ -0,0 +1,140 @@ +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import type { PluginRuntime, RuntimeLogger } from "openclaw/plugin-sdk/plugin-runtime"; +import type { MeetRealtimeAudioTransport } from "./realtime-audio-transport.js"; + +function asRecord(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + +function readString(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value : undefined; +} + +export function createNodeMeetRealtimeAudioTransport(params: { + runtime: PluginRuntime; + nodeId: string; + bridgeId: string; + logger: RuntimeLogger; + logPrefix: string; +}): MeetRealtimeAudioTransport { + let stopped = false; + let inputStarted = false; + let consecutiveInputErrors = 0; + let lastInputError: string | undefined; + let fatalSignaled = false; + let fatalHandler: (() => void) | undefined; + const signalFatal = () => { + if (!fatalSignaled) { + fatalSignaled = true; + fatalHandler?.(); + } + }; + + const transport: MeetRealtimeAudioTransport = { + onFatal: (handler) => { + fatalHandler = handler; + if (fatalSignaled) { + handler(); + } + }, + startInput: (onAudio) => { + if (inputStarted) { + throw new Error("audio input transport already started"); + } + inputStarted = true; + void (async () => { + for (;;) { + if (stopped) { + break; + } + try { + // Long-poll cadence bounds both normal input latency and transient-error retries. + const raw = await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: "googlemeet.chrome", + params: { action: "pullAudio", bridgeId: params.bridgeId, timeoutMs: 250 }, + timeoutMs: 2_000, + }); + const result = asRecord(asRecord(raw).payload ?? raw); + consecutiveInputErrors = 0; + lastInputError = undefined; + const base64 = readString(result.base64); + if (base64) { + onAudio(Buffer.from(base64, "base64")); + } + if (result.closed === true) { + signalFatal(); + break; + } + } catch (error) { + if (stopped) { + break; + } + const message = formatErrorMessage(error); + consecutiveInputErrors += 1; + lastInputError = message; + params.logger.warn( + `[google-meet] ${params.logPrefix} audio input failed (${consecutiveInputErrors}/5): ${message}`, + ); + if ( + consecutiveInputErrors >= 5 || + /unknown bridgeId|bridge is not open/i.test(message) + ) { + signalFatal(); + break; + } + await new Promise((resolve) => { + setTimeout(resolve, 250); + }); + } + } + })(); + }, + stop: async () => { + if (stopped) { + return; + } + stopped = true; + try { + await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: "googlemeet.chrome", + params: { action: "stop", bridgeId: params.bridgeId }, + timeoutMs: 5_000, + }); + } catch (error) { + params.logger.debug?.( + `[google-meet] node audio bridge stop ignored: ${formatErrorMessage(error)}`, + ); + } + }, + writeOutput: async (audio) => { + await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: "googlemeet.chrome", + params: { + action: "pushAudio", + bridgeId: params.bridgeId, + base64: audio.toString("base64"), + }, + timeoutMs: 5_000, + }); + }, + clearOutput: async () => { + await params.runtime.nodes.invoke({ + nodeId: params.nodeId, + command: "googlemeet.chrome", + params: { action: "clearAudio", bridgeId: params.bridgeId }, + timeoutMs: 5_000, + }); + }, + dispose: async () => { + await transport.stop(); + }, + getHealth: () => ({ consecutiveInputErrors, lastInputError }), + }; + + return transport; +} diff --git a/extensions/google-meet/src/realtime-node.ts b/extensions/google-meet/src/realtime-node.ts deleted file mode 100644 index 8ddd21607af9..000000000000 --- a/extensions/google-meet/src/realtime-node.ts +++ /dev/null @@ -1,772 +0,0 @@ -// Google Meet plugin module implements realtime node behavior. -import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; -import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; -import type { PluginRuntime, RuntimeLogger } from "openclaw/plugin-sdk/plugin-runtime"; -import type { - RealtimeTranscriptionProviderPlugin, - RealtimeTranscriptionSession, -} from "openclaw/plugin-sdk/realtime-transcription"; -import { - createRealtimeVoiceAgentTalkbackQueue, - createTalkSessionController, - createRealtimeVoiceBridgeSession, - createRealtimeVoiceOutputActivityTracker, - recordTalkObservabilityEvent, - type RealtimeVoiceAgentTalkbackQueue, - type RealtimeVoiceBridgeSession, - type RealtimeVoiceProviderPlugin, - type TalkEvent, - type TalkEventInput, - type TalkSessionController, -} from "openclaw/plugin-sdk/realtime-voice"; -import { - consultOpenClawAgentForGoogleMeet, - handleGoogleMeetRealtimeConsultToolCall, - resolveGoogleMeetRealtimeTools, -} from "./agent-consult.js"; -import type { GoogleMeetConfig } from "./config.js"; -import { - getGoogleMeetRealtimeTranscriptHealth, - buildGoogleMeetSpeakExactUserMessage, - GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS, - recordGoogleMeetOutputActivity, - getGoogleMeetRealtimeEventHealth, - recordGoogleMeetRealtimeTranscript, - recordGoogleMeetRealtimeEvent, - resolveGoogleMeetRealtimeAudioFormat, - resolveGoogleMeetRealtimeProvider, - resolveGoogleMeetRealtimeTranscriptionProvider, - isGoogleMeetLikelyAssistantEchoTranscript, - pushGoogleMeetTalkEvent, - summarizeGoogleMeetTalkEvents, - convertGoogleMeetBridgeAudioForStt, - convertGoogleMeetTtsAudioForBridge, - formatGoogleMeetAgentAudioModelLog, - formatGoogleMeetAgentTtsResultLog, - formatGoogleMeetTranscriptSummaryLog, - formatGoogleMeetRealtimeVoiceModelLog, - type GoogleMeetRealtimeEventEntry, - type GoogleMeetRealtimeTranscriptEntry, -} from "./realtime.js"; -import type { GoogleMeetChromeHealth } from "./transports/types.js"; - -export type ChromeNodeRealtimeAudioBridgeHandle = { - type: "node-command-pair"; - providerId: string; - nodeId: string; - bridgeId: string; - speak: (instructions?: string) => void; - getHealth: () => GoogleMeetChromeHealth; - stop: () => Promise; -}; - -function asRecord(value: unknown): Record { - return value && typeof value === "object" && !Array.isArray(value) - ? (value as Record) - : {}; -} - -function readString(value: unknown): string | undefined { - return typeof value === "string" && value.trim() ? value : undefined; -} - -function normalizeGoogleMeetTtsPromptText(text: string | undefined): string | undefined { - const trimmed = text?.trim(); - if (!trimmed) { - return undefined; - } - const sayExactly = trimmed.match(/^say exactly:\s*(?.+)$/is)?.groups?.text?.trim(); - if (sayExactly) { - return sayExactly.replace(/^["']|["']$/g, "").trim() || trimmed; - } - return trimmed; -} - -function startGoogleMeetNodeAudioInputLoop(params: { - runtime: PluginRuntime; - nodeId: string; - bridgeId: string; - logger: RuntimeLogger; - logPrefix: string; - isStopped: () => boolean; - stop: () => Promise; - isInputSuppressed: () => boolean; - onAudio: (audio: Buffer) => void; -}) { - let lastInputAt: string | undefined; - let lastInputBytes = 0; - let suppressedInputBytes = 0; - let lastSuppressedInputAt: string | undefined; - let consecutiveInputErrors = 0; - let lastInputError: string | undefined; - - void (async () => { - for (;;) { - if (params.isStopped()) { - break; - } - try { - const raw = await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: "googlemeet.chrome", - params: { action: "pullAudio", bridgeId: params.bridgeId, timeoutMs: 250 }, - timeoutMs: 2_000, - }); - const result = asRecord(asRecord(raw).payload ?? raw); - consecutiveInputErrors = 0; - lastInputError = undefined; - const base64 = readString(result.base64); - if (base64) { - const audio = Buffer.from(base64, "base64"); - if (params.isInputSuppressed()) { - lastSuppressedInputAt = new Date().toISOString(); - suppressedInputBytes += audio.byteLength; - continue; - } - lastInputAt = new Date().toISOString(); - lastInputBytes += audio.byteLength; - params.onAudio(audio); - } - if (result.closed === true) { - await params.stop(); - } - } catch (error) { - if (!params.isStopped()) { - const message = formatErrorMessage(error); - consecutiveInputErrors += 1; - lastInputError = message; - params.logger.warn( - `[google-meet] ${params.logPrefix} audio input failed (${consecutiveInputErrors}/5): ${message}`, - ); - if (consecutiveInputErrors >= 5 || /unknown bridgeId|bridge is not open/i.test(message)) { - await params.stop(); - } else { - await new Promise((resolve) => { - setTimeout(resolve, 250); - }); - } - } - } - } - })(); - - return { - getHealth: () => ({ - audioInputActive: lastInputBytes > 0, - lastInputAt, - lastSuppressedInputAt, - lastInputBytes, - suppressedInputBytes, - consecutiveInputErrors, - lastInputError, - }), - }; -} - -export async function startNodeAgentAudioBridge(params: { - config: GoogleMeetConfig; - fullConfig: OpenClawConfig; - runtime: PluginRuntime; - meetingSessionId: string; - requesterSessionKey?: string; - nodeId: string; - bridgeId: string; - logger: RuntimeLogger; - providers?: RealtimeTranscriptionProviderPlugin[]; -}): Promise { - let stopped = false; - let sttSession: RealtimeTranscriptionSession | null = null; - let realtimeReady = false; - let lastOutputAt: string | undefined; - const outputActivity = createRealtimeVoiceOutputActivityTracker(); - let suppressInputUntil = 0; - let lastOutputPlayableUntilMs = 0; - const resolved = resolveGoogleMeetRealtimeTranscriptionProvider({ - config: params.config, - fullConfig: params.fullConfig, - providers: params.providers, - }); - params.logger.info( - formatGoogleMeetAgentAudioModelLog({ - provider: resolved.provider, - providerConfig: resolved.providerConfig, - audioFormat: params.config.chrome.audioFormat, - }), - ); - const transcript: GoogleMeetRealtimeTranscriptEntry[] = []; - let ttsQueue = Promise.resolve(); - - const stop = async () => { - if (stopped) { - return; - } - stopped = true; - agentTalkback?.close(); - try { - sttSession?.close(); - } catch (error) { - params.logger.debug?.( - `[google-meet] node agent transcription bridge close ignored: ${formatErrorMessage(error)}`, - ); - } - try { - await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: "googlemeet.chrome", - params: { action: "stop", bridgeId: params.bridgeId }, - timeoutMs: 5_000, - }); - } catch (error) { - params.logger.debug?.( - `[google-meet] node audio bridge stop ignored: ${formatErrorMessage(error)}`, - ); - } - }; - - const pushOutputAudio = async (audio: Buffer) => { - const suppression = recordGoogleMeetOutputActivity({ - tracker: outputActivity, - audio, - audioFormat: params.config.chrome.audioFormat, - nowMs: Date.now(), - lastOutputPlayableUntilMs, - suppressInputUntilMs: suppressInputUntil, - }); - suppressInputUntil = suppression.suppressInputUntilMs; - lastOutputPlayableUntilMs = suppression.lastOutputPlayableUntilMs; - lastOutputAt = new Date().toISOString(); - await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: "googlemeet.chrome", - params: { - action: "pushAudio", - bridgeId: params.bridgeId, - base64: Buffer.from(audio).toString("base64"), - }, - timeoutMs: 5_000, - }); - }; - - const enqueueSpeakText = (text: string | undefined) => { - const normalized = normalizeGoogleMeetTtsPromptText(text); - if (!normalized || stopped) { - return; - } - ttsQueue = ttsQueue - .then(async () => { - if (stopped) { - return; - } - recordGoogleMeetRealtimeTranscript(transcript, "assistant", normalized); - params.logger.info( - formatGoogleMeetTranscriptSummaryLog("node agent assistant", normalized), - ); - const result = await params.runtime.tts.textToSpeechTelephony({ - text: normalized, - cfg: params.fullConfig, - }); - if (!result.success || !result.audioBuffer || !result.sampleRate) { - throw new Error(result.error ?? "TTS conversion failed"); - } - params.logger.info(formatGoogleMeetAgentTtsResultLog("node agent", result)); - await pushOutputAudio( - convertGoogleMeetTtsAudioForBridge( - result.audioBuffer, - result.sampleRate, - params.config, - result.outputFormat, - ), - ); - }) - .catch((error: unknown) => { - params.logger.warn(`[google-meet] node agent TTS failed: ${formatErrorMessage(error)}`); - }); - }; - - const agentTalkback: RealtimeVoiceAgentTalkbackQueue | undefined = - createRealtimeVoiceAgentTalkbackQueue({ - debounceMs: GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS, - isStopped: () => stopped, - logger: params.logger, - logPrefix: "[google-meet] node agent", - responseStyle: "Brief, natural spoken answer for a live meeting.", - fallbackText: "I hit an error while checking that. Please try again.", - consult: ({ question, responseStyle }) => - consultOpenClawAgentForGoogleMeet({ - config: params.config, - fullConfig: params.fullConfig, - runtime: params.runtime, - logger: params.logger, - meetingSessionId: params.meetingSessionId, - requesterSessionKey: params.requesterSessionKey, - args: { question, responseStyle }, - transcript, - }), - deliver: enqueueSpeakText, - }); - - sttSession = resolved.provider.createSession({ - cfg: params.fullConfig, - providerConfig: resolved.providerConfig, - onTranscript: (text) => { - const trimmed = text.trim(); - if (!trimmed || stopped) { - return; - } - recordGoogleMeetRealtimeTranscript(transcript, "user", trimmed); - params.logger.info(formatGoogleMeetTranscriptSummaryLog("node agent user", trimmed)); - if (isGoogleMeetLikelyAssistantEchoTranscript({ transcript, text: trimmed })) { - params.logger.info( - formatGoogleMeetTranscriptSummaryLog( - "node agent ignored assistant echo transcript", - trimmed, - ), - ); - return; - } - agentTalkback?.enqueue(trimmed); - }, - onError: (error) => { - params.logger.warn( - `[google-meet] node agent transcription bridge failed: ${formatErrorMessage(error)}`, - ); - void stop(); - }, - }); - await sttSession.connect(); - realtimeReady = true; - - const audioInputLoop = startGoogleMeetNodeAudioInputLoop({ - runtime: params.runtime, - nodeId: params.nodeId, - bridgeId: params.bridgeId, - logger: params.logger, - logPrefix: "node agent", - isStopped: () => stopped, - stop, - isInputSuppressed: () => Date.now() < suppressInputUntil, - onAudio: (audio) => { - sttSession?.sendAudio(convertGoogleMeetBridgeAudioForStt(audio, params.config)); - }, - }); - - return { - type: "node-command-pair", - providerId: resolved.provider.id, - nodeId: params.nodeId, - bridgeId: params.bridgeId, - speak: enqueueSpeakText, - getHealth: () => ({ - providerConnected: sttSession?.isConnected() ?? false, - realtimeReady, - ...audioInputLoop.getHealth(), - audioOutputActive: outputActivity.isActive(), - lastOutputAt, - lastOutputBytes: outputActivity.snapshot().sinkAudioBytes, - ...getGoogleMeetRealtimeTranscriptHealth(transcript), - bridgeClosed: stopped, - }), - stop, - }; -} - -export async function startNodeRealtimeAudioBridge(params: { - config: GoogleMeetConfig; - fullConfig: OpenClawConfig; - runtime: PluginRuntime; - meetingSessionId: string; - requesterSessionKey?: string; - nodeId: string; - bridgeId: string; - logger: RuntimeLogger; - providers?: RealtimeVoiceProviderPlugin[]; -}): Promise { - let stopped = false; - let bridge: RealtimeVoiceBridgeSession | null = null; - let realtimeReady = false; - let lastOutputAt: string | undefined; - let lastClearAt: string | undefined; - const outputActivity = createRealtimeVoiceOutputActivityTracker(); - let suppressInputUntil = 0; - let lastOutputPlayableUntilMs = 0; - let clearCount = 0; - const resolved = resolveGoogleMeetRealtimeProvider({ - config: params.config, - fullConfig: params.fullConfig, - providers: params.providers, - }); - const transcript: GoogleMeetRealtimeTranscriptEntry[] = []; - const realtimeEvents: GoogleMeetRealtimeEventEntry[] = []; - const strategy = params.config.realtime.strategy; - const talk: TalkSessionController = createTalkSessionController( - { - sessionId: `google-meet:${params.meetingSessionId}:${params.bridgeId}:node-realtime`, - mode: "realtime", - transport: "gateway-relay", - brain: strategy === "bidi" ? "direct-tools" : "agent-consult", - provider: resolved.provider.id, - }, - { onEvent: recordTalkObservabilityEvent }, - ); - const recentTalkEvents: TalkEvent[] = []; - const rememberTalkEvent = (event: TalkEvent | undefined): void => { - if (event) { - pushGoogleMeetTalkEvent(recentTalkEvents, event); - } - }; - const emitTalkEvent = (input: TalkEventInput): void => { - rememberTalkEvent(talk.emit(input)); - }; - const ensureTalkTurn = (): string => { - const turn = talk.ensureTurn({ - payload: { bridgeId: params.bridgeId, meetingSessionId: params.meetingSessionId }, - }); - if (turn.event) { - rememberTalkEvent(turn.event); - } - return turn.turnId; - }; - const finishOutputAudio = (reason: string): void => { - rememberTalkEvent( - talk.finishOutputAudio({ - payload: { bridgeId: params.bridgeId, reason }, - }), - ); - }; - const endTalkTurn = (reason = "completed"): void => { - const ended = talk.endTurn({ - payload: { bridgeId: params.bridgeId, reason }, - }); - if (ended.ok) { - rememberTalkEvent(ended.event); - } - }; - emitTalkEvent({ - type: "session.started", - payload: { - bridgeId: params.bridgeId, - meetingSessionId: params.meetingSessionId, - nodeId: params.nodeId, - }, - }); - params.logger.info( - formatGoogleMeetRealtimeVoiceModelLog({ - strategy, - provider: resolved.provider, - providerConfig: resolved.providerConfig, - fallbackModel: params.config.realtime.model, - audioFormat: params.config.chrome.audioFormat, - }), - ); - const agentTalkback: RealtimeVoiceAgentTalkbackQueue | undefined = - createRealtimeVoiceAgentTalkbackQueue({ - debounceMs: GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS, - isStopped: () => stopped, - logger: params.logger, - logPrefix: "[google-meet] node realtime agent", - responseStyle: "Brief, natural spoken answer for a live meeting.", - fallbackText: "I hit an error while checking that. Please try again.", - consult: ({ question, responseStyle }) => - consultOpenClawAgentForGoogleMeet({ - config: params.config, - fullConfig: params.fullConfig, - runtime: params.runtime, - logger: params.logger, - meetingSessionId: params.meetingSessionId, - requesterSessionKey: params.requesterSessionKey, - args: { question, responseStyle }, - transcript, - }), - deliver: (text) => { - bridge?.sendUserMessage(buildGoogleMeetSpeakExactUserMessage(text)); - }, - }); - - const stop = async () => { - if (stopped) { - return; - } - stopped = true; - agentTalkback?.close(); - try { - bridge?.close(); - } catch (error) { - params.logger.debug?.( - `[google-meet] node realtime bridge close ignored: ${formatErrorMessage(error)}`, - ); - } - try { - await params.runtime.nodes.invoke({ - nodeId: params.nodeId, - command: "googlemeet.chrome", - params: { action: "stop", bridgeId: params.bridgeId }, - timeoutMs: 5_000, - }); - } catch (error) { - params.logger.debug?.( - `[google-meet] node audio bridge stop ignored: ${formatErrorMessage(error)}`, - ); - } - }; - - bridge = createRealtimeVoiceBridgeSession({ - provider: resolved.provider, - cfg: params.fullConfig, - providerConfig: resolved.providerConfig, - audioFormat: resolveGoogleMeetRealtimeAudioFormat(params.config), - instructions: params.config.realtime.instructions, - initialGreetingInstructions: params.config.realtime.introMessage, - autoRespondToAudio: strategy === "bidi", - triggerGreetingOnReady: false, - markStrategy: "ack-immediately", - tools: - strategy === "bidi" ? resolveGoogleMeetRealtimeTools(params.config.realtime.toolPolicy) : [], - audioSink: { - isOpen: () => !stopped, - sendAudio: (audio) => { - const turnId = ensureTalkTurn(); - rememberTalkEvent( - talk.startOutputAudio({ - turnId, - payload: { bridgeId: params.bridgeId }, - }).event, - ); - emitTalkEvent({ - type: "output.audio.delta", - turnId, - payload: { byteLength: audio.byteLength }, - }); - const suppression = recordGoogleMeetOutputActivity({ - tracker: outputActivity, - audio, - audioFormat: params.config.chrome.audioFormat, - nowMs: Date.now(), - lastOutputPlayableUntilMs, - suppressInputUntilMs: suppressInputUntil, - }); - suppressInputUntil = suppression.suppressInputUntilMs; - lastOutputPlayableUntilMs = suppression.lastOutputPlayableUntilMs; - lastOutputAt = new Date().toISOString(); - void params.runtime.nodes - .invoke({ - nodeId: params.nodeId, - command: "googlemeet.chrome", - params: { - action: "pushAudio", - bridgeId: params.bridgeId, - base64: Buffer.from(audio).toString("base64"), - }, - timeoutMs: 5_000, - }) - .catch((error: unknown) => { - params.logger.warn( - `[google-meet] node audio output failed: ${formatErrorMessage(error)}`, - ); - void stop(); - }); - }, - clearAudio: () => { - lastClearAt = new Date().toISOString(); - clearCount += 1; - finishOutputAudio("clear"); - suppressInputUntil = 0; - lastOutputPlayableUntilMs = 0; - void params.runtime.nodes - .invoke({ - nodeId: params.nodeId, - command: "googlemeet.chrome", - params: { - action: "clearAudio", - bridgeId: params.bridgeId, - }, - timeoutMs: 5_000, - }) - .catch((error: unknown) => { - params.logger.warn( - `[google-meet] node audio clear failed: ${formatErrorMessage(error)}`, - ); - void stop(); - }); - }, - }, - onTranscript: (role, text, isFinal) => { - const turnId = ensureTalkTurn(); - const eventType = - role === "assistant" - ? isFinal - ? "output.text.done" - : "output.text.delta" - : isFinal - ? "transcript.done" - : "transcript.delta"; - const payload = role === "assistant" ? { text } : { role, text }; - emitTalkEvent({ - type: eventType, - turnId, - payload, - final: isFinal, - }); - if (role === "user" && isFinal) { - emitTalkEvent({ - type: "input.audio.committed", - turnId, - payload: { bridgeId: params.bridgeId }, - final: true, - }); - } - if (isFinal) { - recordGoogleMeetRealtimeTranscript(transcript, role, text); - params.logger.info(formatGoogleMeetTranscriptSummaryLog(`node realtime ${role}`, text)); - if (role === "user" && strategy === "agent") { - if (isGoogleMeetLikelyAssistantEchoTranscript({ transcript, text })) { - params.logger.info( - formatGoogleMeetTranscriptSummaryLog( - "node realtime ignored assistant echo transcript", - text, - ), - ); - return; - } - agentTalkback?.enqueue(text); - } - } - }, - onEvent: (event) => { - recordGoogleMeetRealtimeEvent(realtimeEvents, event); - if (event.type === "input_audio_buffer.speech_started") { - ensureTalkTurn(); - } else if (event.type === "input_audio_buffer.speech_stopped") { - const turnId = talk.activeTurnId; - if (!turnId) { - return; - } - emitTalkEvent({ - type: "input.audio.committed", - turnId, - payload: { bridgeId: params.bridgeId, source: event.type }, - final: true, - }); - } else if (event.type === "response.done") { - finishOutputAudio("response.done"); - endTalkTurn("response.done"); - } else if (event.type === "error") { - emitTalkEvent({ - type: "session.error", - payload: { message: event.detail ?? "Realtime provider error" }, - final: true, - }); - } - if ( - event.type === "error" || - event.type === "response.done" || - event.type === "input_audio_buffer.speech_started" || - event.type === "input_audio_buffer.speech_stopped" || - event.type === "conversation.item.input_audio_transcription.completed" || - event.type === "conversation.item.input_audio_transcription.failed" - ) { - const detail = event.detail ? ` ${event.detail}` : ""; - params.logger.info(`[google-meet] node realtime ${event.direction}:${event.type}${detail}`); - } - }, - onToolCall: (event, session) => { - emitTalkEvent({ - type: "tool.call", - turnId: ensureTalkTurn(), - itemId: event.itemId, - callId: event.callId, - payload: { name: event.name, args: event.args }, - }); - const turnId = ensureTalkTurn(); - return handleGoogleMeetRealtimeConsultToolCall({ - strategy, - session, - event, - config: params.config, - fullConfig: params.fullConfig, - runtime: params.runtime, - logger: params.logger, - meetingSessionId: params.meetingSessionId, - requesterSessionKey: params.requesterSessionKey, - transcript, - onTalkEvent: (input) => emitTalkEvent({ ...input, turnId: input.turnId ?? turnId }), - }); - }, - onError: (error) => { - params.logger.warn( - `[google-meet] node realtime voice bridge failed: ${formatErrorMessage(error)}`, - ); - emitTalkEvent({ - type: "session.error", - payload: { message: formatErrorMessage(error) }, - final: true, - }); - void stop(); - }, - onClose: (reason) => { - realtimeReady = false; - finishOutputAudio(reason); - emitTalkEvent({ - type: "session.closed", - payload: { reason }, - final: true, - }); - if (reason === "error") { - void stop(); - } - }, - onReady: () => { - realtimeReady = true; - emitTalkEvent({ - type: "session.ready", - payload: { bridgeId: params.bridgeId }, - }); - }, - }); - - await bridge.connect(); - - const audioInputLoop = startGoogleMeetNodeAudioInputLoop({ - runtime: params.runtime, - nodeId: params.nodeId, - bridgeId: params.bridgeId, - logger: params.logger, - logPrefix: "node", - isStopped: () => stopped, - stop, - isInputSuppressed: () => Date.now() < suppressInputUntil, - onAudio: (audio) => { - emitTalkEvent({ - type: "input.audio.delta", - turnId: ensureTalkTurn(), - payload: { byteLength: audio.byteLength }, - }); - bridge?.sendAudio(audio); - }, - }); - - return { - type: "node-command-pair", - providerId: resolved.provider.id, - nodeId: params.nodeId, - bridgeId: params.bridgeId, - speak: (instructions) => { - bridge?.triggerGreeting(instructions); - }, - getHealth: () => ({ - providerConnected: bridge?.bridge.isConnected() ?? false, - realtimeReady, - ...audioInputLoop.getHealth(), - audioOutputActive: outputActivity.isActive(), - lastOutputAt, - lastClearAt, - lastOutputBytes: outputActivity.snapshot().sinkAudioBytes, - ...getGoogleMeetRealtimeTranscriptHealth(transcript), - ...getGoogleMeetRealtimeEventHealth(realtimeEvents), - recentTalkEvents: summarizeGoogleMeetTalkEvents(recentTalkEvents), - clearCount, - bridgeClosed: stopped, - }), - stop, - }; -} -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/extensions/google-meet/src/realtime.process.test.ts b/extensions/google-meet/src/realtime.process.test.ts index 7e2572720872..466843b214e7 100644 --- a/extensions/google-meet/src/realtime.process.test.ts +++ b/extensions/google-meet/src/realtime.process.test.ts @@ -8,11 +8,17 @@ import type { RealtimeTranscriptionProviderPlugin } from "openclaw/plugin-sdk/re import type { RealtimeVoiceProviderPlugin } from "openclaw/plugin-sdk/realtime-voice"; import { afterEach, describe, expect, it, vi } from "vitest"; import { resolveGoogleMeetConfig } from "./config.js"; -import { formatGoogleMeetRealtimeVoiceModelLog, startCommandAgentAudioBridge } from "./realtime.js"; +import type { MeetRealtimeAudioTransport } from "./realtime-audio-transport.js"; +import { createLocalMeetRealtimeAudioTransport } from "./realtime-local-audio-transport.js"; +import { startMeetAgentRealtimeEngine, startMeetRealtimeEngine } from "./realtime.js"; const tempDirs: string[] = []; const spawnedChildren: ChildProcess[] = []; +type MeetRealtimeAudioSpawn = NonNullable< + Parameters[0]["spawn"] +>; + function writeBridgeCommand(): string { const dir = mkdtempSync(path.join(tmpdir(), "openclaw-google-meet-bridge-")); tempDirs.push(dir); @@ -32,15 +38,11 @@ function writeBridgeCommand(): string { return scriptPath; } -function makeRecordingSpawn(): NonNullable< - Parameters[0]["spawn"] -> { +function makeRecordingSpawn(): MeetRealtimeAudioSpawn { return (command, args, options) => { const child = spawnChildProcess(command, args, options); spawnedChildren.push(child); - return child as ReturnType< - NonNullable[0]["spawn"]> - >; + return child as ReturnType; }; } @@ -56,7 +58,79 @@ afterEach(() => { vi.restoreAllMocks(); }); -describe("startCommandAgentAudioBridge real process stream errors", () => { +describe("local Meet realtime transport process stream errors", () => { + it("stops the engine when input fails during provider setup", async () => { + const bridgeScript = writeBridgeCommand(); + let finishConnect = () => {}; + const connectGate = new Promise((resolve) => { + finishConnect = resolve; + }); + const sttSession = { + connect: vi.fn(() => connectGate), + sendAudio: vi.fn(), + close: vi.fn(), + isConnected: vi.fn(() => false), + }; + const provider: RealtimeTranscriptionProviderPlugin = { + id: "openai", + label: "OpenAI", + defaultModel: "gpt-4o-transcribe", + autoSelectOrder: 1, + resolveConfig: ({ rawConfig }) => rawConfig, + isConfigured: () => true, + createSession: () => sttSession, + }; + const logger = { + debug: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + }; + const transport = createLocalMeetRealtimeAudioTransport({ + inputCommand: [path.join(path.dirname(bridgeScript), "missing-capture")], + outputCommand: [process.execPath, bridgeScript, "play"], + bargeInRmsThreshold: 10, + bargeInPeakThreshold: 10, + bargeInCooldownMs: 1, + logger, + spawn: makeRecordingSpawn(), + }); + const config = resolveGoogleMeetConfig({ + chrome: { audioFormat: "pcm16-24khz" }, + realtime: { provider: "openai", agentId: "jay", introMessage: "" }, + }); + const engineResult = startMeetAgentRealtimeEngine({ + config, + fullConfig: {} as never, + runtime: {} as never, + meetingSessionId: "meet-startup-failure", + logger, + providers: [provider], + transport, + }).then( + () => new Error("Expected Google Meet engine startup to fail"), + (error: unknown) => error, + ); + const inputProcess = spawnedChildren[1]; + if (!inputProcess) { + throw new Error("Expected Google Meet transport to spawn an input child process"); + } + + await once(inputProcess, "error"); + await vi.waitFor(() => { + expect(sttSession.close).toHaveBeenCalledTimes(1); + }); + finishConnect(); + + await expect(engineResult).resolves.toEqual( + new Error("Google Meet audio transport stopped during transcription provider setup"), + ); + expect(logger.warn).toHaveBeenCalledWith( + expect.stringContaining("[google-meet] audio input command failed:"), + ); + await transport.dispose(); + }); + it("contains a forced local command-pair stdout stream error through bridge stop", async () => { const bridgeScript = writeBridgeCommand(); const sttSession = { @@ -80,19 +154,27 @@ describe("startCommandAgentAudioBridge real process stream errors", () => { warn: vi.fn(), }; - const handle = await startCommandAgentAudioBridge({ - config: resolveGoogleMeetConfig({ - chrome: { audioFormat: "pcm16-24khz" }, - realtime: { provider: "openai", agentId: "jay", introMessage: "" }, - }), + const config = resolveGoogleMeetConfig({ + chrome: { audioFormat: "pcm16-24khz" }, + realtime: { provider: "openai", agentId: "jay", introMessage: "" }, + }); + const transport = createLocalMeetRealtimeAudioTransport({ + inputCommand: [process.execPath, bridgeScript, "capture"], + outputCommand: [process.execPath, bridgeScript, "play"], + bargeInRmsThreshold: config.chrome.bargeInRmsThreshold, + bargeInPeakThreshold: config.chrome.bargeInPeakThreshold, + bargeInCooldownMs: config.chrome.bargeInCooldownMs, + logger: logger as never, + spawn: makeRecordingSpawn(), + }); + const handle = await startMeetAgentRealtimeEngine({ + config, fullConfig: {} as never, runtime: {} as never, meetingSessionId: "meet-1", - inputCommand: [process.execPath, bridgeScript, "capture"], - outputCommand: [process.execPath, bridgeScript, "play"], logger: logger as never, providers: [provider], - spawn: makeRecordingSpawn(), + transport, }); const [outputProcess, inputProcess] = spawnedChildren; if (!inputProcess || !outputProcess) { @@ -132,15 +214,55 @@ describe("startCommandAgentAudioBridge real process stream errors", () => { }); describe("Google Meet realtime model logs", () => { - it("keeps a whole code point when a provider id crosses the log boundary", () => { + it("keeps a whole code point when a provider id crosses the log boundary", async () => { const prefix = "a".repeat(179); - const log = formatGoogleMeetRealtimeVoiceModelLog({ - strategy: "native", - provider: { id: `${prefix}😀tail` } as RealtimeVoiceProviderPlugin, - providerConfig: {}, - audioFormat: "pcm16-24khz", + const providerId = `${prefix}😀tail`; + const bridge = { + connect: vi.fn(async () => {}), + sendAudio: vi.fn(), + sendUserMessage: vi.fn(), + setMediaTimestamp: vi.fn(), + submitToolResult: vi.fn(), + acknowledgeMark: vi.fn(), + close: vi.fn(), + triggerGreeting: vi.fn(), + isConnected: vi.fn(() => true), + }; + const provider: RealtimeVoiceProviderPlugin = { + id: providerId, + label: "Long identifier provider", + isConfigured: () => true, + createBridge: () => bridge, + }; + const transport: MeetRealtimeAudioTransport = { + onFatal: vi.fn(), + startInput: vi.fn(), + stop: vi.fn(async () => {}), + writeOutput: vi.fn(async () => {}), + clearOutput: vi.fn(async () => {}), + dispose: vi.fn(async () => {}), + }; + const logger = { + debug: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + }; + const handle = await startMeetRealtimeEngine({ + config: resolveGoogleMeetConfig({ + realtime: { strategy: "native", provider: providerId }, + }), + fullConfig: {} as never, + runtime: {} as never, + meetingSessionId: "long-provider-log", + logger, + providers: [provider], + transport, }); - expect(log).toContain(`provider=${prefix} model=provider-default`); + expect(logger.info).toHaveBeenCalledWith( + expect.stringContaining(`provider=${prefix} model=provider-default`), + ); + await handle.stop(); }); }); diff --git a/extensions/google-meet/src/realtime.ts b/extensions/google-meet/src/realtime.ts index 6505aac7d385..66ec2d381801 100644 --- a/extensions/google-meet/src/realtime.ts +++ b/extensions/google-meet/src/realtime.ts @@ -1,6 +1,4 @@ // Google Meet plugin module implements realtime behavior. -import { spawn } from "node:child_process"; -import type { Writable } from "node:stream"; import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import type { PluginRuntime, RuntimeLogger } from "openclaw/plugin-sdk/plugin-runtime"; @@ -16,18 +14,13 @@ import { createRealtimeVoiceBridgeSession, createRealtimeVoiceOutputActivityTracker, createTalkSessionController, - convertPcmToMulaw8k, extendRealtimeVoiceOutputEchoSuppression, getRealtimeVoiceBridgeEventHealth, getRealtimeVoiceTranscriptHealth, isLikelyRealtimeVoiceAssistantEchoTranscript, - mulawToPcm, - REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ, - REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ, recordRealtimeVoiceBridgeEvent, recordTalkObservabilityEvent, recordRealtimeVoiceTranscript, - resamplePcm, resolveConfiguredRealtimeVoiceProvider, type RealtimeVoiceAgentTalkbackQueue, type RealtimeVoiceBridgeEventLogEntry, @@ -47,66 +40,16 @@ import { resolveGoogleMeetRealtimeTools, } from "./agent-consult.js"; import type { GoogleMeetConfig } from "./config.js"; +import { + convertGoogleMeetBridgeAudioForStt, + convertGoogleMeetTtsAudioForBridge, + resolveGoogleMeetRealtimeAudioFormat, +} from "./realtime-audio-format.js"; +import type { MeetRealtimeAudioTransport } from "./realtime-audio-transport.js"; import type { GoogleMeetChromeHealth } from "./transports/types.js"; -type BridgeProcess = { - pid?: number; - killed?: boolean; - stdin?: Writable | null; - stdout?: { - on(event: "data", listener: (chunk: Buffer | string) => void): unknown; - on(event: "error", listener: (error: Error) => void): unknown; - } | null; - stderr?: { - on(event: "data", listener: (chunk: Buffer | string) => void): unknown; - on(event: "error", listener: (error: Error) => void): unknown; - } | null; - kill(signal?: NodeJS.Signals): boolean; - on( - event: "exit", - listener: (code: number | null, signal: NodeJS.Signals | null) => void, - ): unknown; - on(event: "error", listener: (error: Error) => void): unknown; -}; - -type SpawnFn = ( - command: string, - args: string[], - options: { stdio: ["pipe" | "ignore", "pipe" | "ignore", "pipe" | "ignore"] }, -) => BridgeProcess; - -function terminateBridgeProcess(proc: BridgeProcess, signal: NodeJS.Signals = "SIGTERM"): void { - if (proc.killed && signal !== "SIGKILL") { - return; - } - let exited = false; - proc.on("exit", () => { - exited = true; - }); - try { - proc.kill(signal); - } catch { - return; - } - if (signal === "SIGKILL") { - return; - } - const timer = setTimeout(() => { - if (!exited) { - try { - proc.kill("SIGKILL"); - } catch { - // Process may have exited after the grace check. - } - } - }, 1000); - timer.unref?.(); -} - -export type ChromeRealtimeAudioBridgeHandle = { +export type MeetRealtimeAudioEngineHandle = { providerId: string; - inputCommand: string[]; - outputCommand: string[]; speak: (instructions?: string) => void; getHealth: () => GoogleMeetChromeHealth; stop: () => Promise; @@ -122,60 +65,36 @@ type ResolvedRealtimeTranscriptionProvider = { providerConfig: RealtimeTranscriptionProviderConfig; }; -export type GoogleMeetRealtimeTranscriptEntry = RealtimeVoiceTranscriptEntry; -export const recordGoogleMeetRealtimeTranscript = recordRealtimeVoiceTranscript; +type GoogleMeetRealtimeTranscriptEntry = RealtimeVoiceTranscriptEntry; +const recordGoogleMeetRealtimeTranscript = recordRealtimeVoiceTranscript; -export function getGoogleMeetRealtimeTranscriptHealth( +function getGoogleMeetRealtimeTranscriptHealth( transcript: GoogleMeetRealtimeTranscriptEntry[], ): Pick> { return getRealtimeVoiceTranscriptHealth(transcript); } -export type GoogleMeetRealtimeEventEntry = RealtimeVoiceBridgeEventLogEntry; +type GoogleMeetRealtimeEventEntry = RealtimeVoiceBridgeEventLogEntry; -export const GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS = 900; +const GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS = 900; +// Playback duration plus a tail blocks live loopback; transcript lookback catches delayed echo. const GOOGLE_MEET_OUTPUT_ECHO_SUPPRESSION_TAIL_MS = 3_000; const GOOGLE_MEET_TRANSCRIPT_ECHO_LOOKBACK_MS = 45_000; -export function recordGoogleMeetRealtimeEvent( +function recordGoogleMeetRealtimeEvent( events: GoogleMeetRealtimeEventEntry[], event: Parameters[1], ): void { recordRealtimeVoiceBridgeEvent(events, event); } -export function getGoogleMeetRealtimeEventHealth( +function getGoogleMeetRealtimeEventHealth( events: GoogleMeetRealtimeEventEntry[], ): Pick> { return getRealtimeVoiceBridgeEventHealth(events); } -function splitCommand(argv: string[]): { command: string; args: string[] } { - const [command, ...args] = argv; - if (!command) { - throw new Error("audio bridge command must not be empty"); - } - return { command, args }; -} - -function readPcm16Stats(audio: Buffer): { rms: number; peak: number } { - let sumSquares = 0; - let peak = 0; - let samples = 0; - for (let offset = 0; offset + 1 < audio.byteLength; offset += 2) { - const sample = audio.readInt16LE(offset); - const abs = Math.abs(sample); - peak = Math.max(peak, abs); - sumSquares += sample * sample; - samples += 1; - } - return { - rms: samples > 0 ? Math.sqrt(sumSquares / samples) : 0, - peak, - }; -} - -export function isGoogleMeetLikelyAssistantEchoTranscript(params: { +function isGoogleMeetLikelyAssistantEchoTranscript(params: { transcript: GoogleMeetRealtimeTranscriptEntry[]; text: string; nowMs?: number; @@ -201,7 +120,7 @@ function extendGoogleMeetOutputEchoSuppression(params: { }); } -export function recordGoogleMeetOutputActivity(params: { +function recordGoogleMeetOutputActivity(params: { tracker: RealtimeVoiceOutputActivityTracker; audio: Buffer; audioFormat: GoogleMeetConfig["chrome"]["audioFormat"]; @@ -219,107 +138,7 @@ export function recordGoogleMeetOutputActivity(params: { return suppression; } -export function resolveGoogleMeetRealtimeAudioFormat(config: GoogleMeetConfig) { - return config.chrome.audioFormat === "g711-ulaw-8khz" - ? REALTIME_VOICE_AUDIO_FORMAT_G711_ULAW_8KHZ - : REALTIME_VOICE_AUDIO_FORMAT_PCM16_24KHZ; -} - -export function convertGoogleMeetBridgeAudioForStt( - audio: Buffer, - config: GoogleMeetConfig, -): Buffer { - if (config.chrome.audioFormat === "g711-ulaw-8khz") { - return audio; - } - return convertPcmToMulaw8k(audio, 24_000); -} - -export function convertGoogleMeetTtsAudioForBridge( - audio: Buffer, - sampleRate: number, - config: GoogleMeetConfig, - outputFormat?: string, -): Buffer { - const sourceFormat = sourceTelephonyTtsFormat(outputFormat); - if ( - config.chrome.audioFormat === "g711-ulaw-8khz" && - sourceFormat === "mulaw" && - sampleRate === 8_000 - ) { - return audio; - } - const pcm = decodeGoogleMeetTelephonyTtsAudio(audio, sourceFormat); - return config.chrome.audioFormat === "g711-ulaw-8khz" - ? convertPcmToMulaw8k(pcm, sampleRate) - : resamplePcm(pcm, sampleRate, 24_000); -} - -type GoogleMeetTelephonyTtsFormat = "pcm" | "mulaw" | "alaw"; - -function sourceTelephonyTtsFormat(outputFormat: string | undefined): GoogleMeetTelephonyTtsFormat { - const normalized = outputFormat?.trim().toLowerCase().replaceAll("_", "-") ?? ""; - if ( - !normalized || - normalized === "pcm" || - normalized.startsWith("pcm-") || - normalized.includes("pcm16") || - normalized.includes("16bit-mono-pcm") - ) { - return "pcm"; - } - if ( - normalized === "mulaw" || - normalized === "ulaw" || - normalized.includes("mu-law") || - normalized.includes("mulaw") || - normalized.includes("ulaw") - ) { - return "mulaw"; - } - if (normalized === "alaw" || normalized.includes("a-law") || normalized.includes("alaw")) { - return "alaw"; - } - throw new Error(`Unsupported telephony TTS output format for Google Meet: ${outputFormat}`); -} - -function decodeGoogleMeetTelephonyTtsAudio( - audio: Buffer, - sourceFormat: GoogleMeetTelephonyTtsFormat, -): Buffer { - switch (sourceFormat) { - case "pcm": - return audio; - case "mulaw": - return mulawToPcm(audio); - case "alaw": - return alawToPcm(audio); - } - return unsupportedGoogleMeetTelephonyTtsFormat(sourceFormat); -} - -function unsupportedGoogleMeetTelephonyTtsFormat(_format: never): never { - throw new Error("Unsupported telephony TTS output format for Google Meet"); -} - -function alawToPcm(alaw: Buffer): Buffer { - const pcm = Buffer.alloc(alaw.length * 2); - for (let index = 0; index < alaw.length; index += 1) { - pcm.writeInt16LE(alawByteToLinear(alaw[index] ?? 0), index * 2); - } - return pcm; -} - -function alawByteToLinear(value: number): number { - const aLaw = value ^ 0x55; - const sign = aLaw & 0x80; - const exponent = (aLaw & 0x70) >> 4; - const mantissa = aLaw & 0x0f; - const sample = exponent === 0 ? (mantissa << 4) + 8 : ((mantissa << 4) + 0x108) << (exponent - 1); - return sign ? sample : -sample; -} - -export function resolveGoogleMeetRealtimeProvider(params: { +function resolveGoogleMeetRealtimeProvider(params: { config: GoogleMeetConfig; fullConfig: OpenClawConfig; providers?: RealtimeVoiceProviderPlugin[]; @@ -335,7 +154,7 @@ export function resolveGoogleMeetRealtimeProvider(params: { }); } -export function resolveGoogleMeetRealtimeTranscriptionProvider(params: { +function resolveGoogleMeetRealtimeTranscriptionProvider(params: { config: GoogleMeetConfig; fullConfig: OpenClawConfig; providers?: RealtimeTranscriptionProviderPlugin[]; @@ -369,7 +188,7 @@ export function resolveGoogleMeetRealtimeTranscriptionProvider(params: { return { provider, providerConfig }; } -export function buildGoogleMeetSpeakExactUserMessage(text: string): string { +function buildGoogleMeetSpeakExactUserMessage(text: string): string { return [ "Speak this exact OpenClaw answer to the meeting, without adding, removing, or rephrasing words.", `Answer: ${JSON.stringify(text)}`, @@ -399,7 +218,7 @@ function resolveProviderModelForLog(params: { ); } -export function formatGoogleMeetRealtimeVoiceModelLog(params: { +function formatGoogleMeetRealtimeVoiceModelLog(params: { strategy: string; provider: RealtimeVoiceProviderPlugin; providerConfig: RealtimeVoiceProviderConfig; @@ -420,7 +239,7 @@ export function formatGoogleMeetRealtimeVoiceModelLog(params: { ].join(" "); } -export function formatGoogleMeetAgentAudioModelLog(params: { +function formatGoogleMeetAgentAudioModelLog(params: { provider: RealtimeTranscriptionProviderPlugin; providerConfig: RealtimeTranscriptionProviderConfig; audioFormat: GoogleMeetConfig["chrome"]["audioFormat"]; @@ -449,7 +268,7 @@ type GoogleMeetTtsResultLogFields = { fallbackFrom?: string; }; -export function formatGoogleMeetAgentTtsResultLog( +function formatGoogleMeetAgentTtsResultLog( prefix: string, result: GoogleMeetTtsResultLogFields, ): string { @@ -463,7 +282,7 @@ export function formatGoogleMeetAgentTtsResultLog( ].join(" "); } -export function formatGoogleMeetTranscriptSummaryLog(prefix: string, text: string): string { +function formatGoogleMeetTranscriptSummaryLog(prefix: string, text: string): string { return `[google-meet] ${prefix}: chars=${text.length}`; } @@ -479,18 +298,14 @@ function normalizeGoogleMeetTtsPromptText(text: string | undefined): string | un return trimmed; } -export function pushGoogleMeetTalkEvent( - events: TalkEvent[], - event: TalkEvent, - maxEntries = 40, -): void { +function pushGoogleMeetTalkEvent(events: TalkEvent[], event: TalkEvent, maxEntries = 40): void { events.push(event); if (events.length > maxEntries) { events.splice(0, events.length - maxEntries); } } -export function summarizeGoogleMeetTalkEvents( +function summarizeGoogleMeetTalkEvents( events: TalkEvent[], ): NonNullable { return events.slice(-20).map((event) => ({ @@ -504,29 +319,17 @@ export function summarizeGoogleMeetTalkEvents( })); } -export async function startCommandAgentAudioBridge(params: { +export async function startMeetAgentRealtimeEngine(params: { config: GoogleMeetConfig; fullConfig: OpenClawConfig; runtime: PluginRuntime; meetingSessionId: string; requesterSessionKey?: string; - inputCommand: string[]; - outputCommand: string[]; + logPrefix?: "node"; + transport: MeetRealtimeAudioTransport; logger: RuntimeLogger; providers?: RealtimeTranscriptionProviderPlugin[]; - spawn?: SpawnFn; -}): Promise { - const input = splitCommand(params.inputCommand); - const output = splitCommand(params.outputCommand); - const spawnFn: SpawnFn = - params.spawn ?? - ((command, args, options) => spawn(command, args, options) as unknown as BridgeProcess); - const outputProcess = spawnFn(output.command, output.args, { - stdio: ["pipe", "ignore", "pipe"], - }); - const inputProcess = spawnFn(input.command, input.args, { - stdio: ["ignore", "pipe", "pipe"], - }); +}): Promise { let stopped = false; let sttSession: RealtimeTranscriptionSession | null = null; let realtimeReady = false; @@ -540,6 +343,7 @@ export async function startCommandAgentAudioBridge(params: { let lastOutputPlayableUntilMs = 0; let ttsQueue = Promise.resolve(); const transcript: GoogleMeetRealtimeTranscriptEntry[] = []; + const agentLogScope = params.logPrefix ? `${params.logPrefix} agent` : "agent"; const resolved = resolveGoogleMeetRealtimeTranscriptionProvider({ config: params.config, fullConfig: params.fullConfig, @@ -594,7 +398,7 @@ export async function startCommandAgentAudioBridge(params: { sttSession?.close(); } catch (error) { params.logger.debug?.( - `[google-meet] agent transcription bridge close ignored: ${formatErrorMessage(error)}`, + `[google-meet] ${agentLogScope} transcription bridge close ignored: ${formatErrorMessage(error)}`, ); } emitTalkEvent({ @@ -602,40 +406,11 @@ export async function startCommandAgentAudioBridge(params: { final: true, payload: { meetingSessionId: params.meetingSessionId }, }); - terminateBridgeProcess(inputProcess); - terminateBridgeProcess(outputProcess); + await params.transport.stop(); + await params.transport.dispose(); }; - const fail = (label: string) => (error: Error) => { - params.logger.warn(`[google-meet] ${label} failed: ${formatErrorMessage(error)}`); - void stop(); - }; - inputProcess.on("error", fail("audio input command")); - inputProcess.on("exit", (code, signal) => { - if (!stopped) { - params.logger.warn(`[google-meet] audio input command exited (${code ?? signal ?? "done"})`); - void stop(); - } - }); - inputProcess.stderr?.on("data", (chunk) => { - params.logger.debug?.(`[google-meet] audio input: ${String(chunk).trim()}`); - }); - inputProcess.stdout?.on("error", fail("audio input command stdout")); - inputProcess.stderr?.on("error", fail("audio input command stderr")); - outputProcess.on("error", fail("audio output command")); - outputProcess.stdin?.on?.("error", fail("audio output command")); - outputProcess.on("exit", (code, signal) => { - if (!stopped) { - params.logger.warn(`[google-meet] audio output command exited (${code ?? signal ?? "done"})`); - void stop(); - } - }); - outputProcess.stderr?.on("data", (chunk) => { - params.logger.debug?.(`[google-meet] audio output: ${String(chunk).trim()}`); - }); - outputProcess.stderr?.on("error", fail("audio output command stderr")); - - const writeOutputAudio = (audio: Buffer) => { + const writeOutputAudio = async (audio: Buffer) => { const suppression = recordGoogleMeetOutputActivity({ tracker: outputActivity, audio, @@ -652,11 +427,7 @@ export async function startCommandAgentAudioBridge(params: { turnId: ensureTalkTurn(), payload: { meetingSessionId: params.meetingSessionId, bytes: audio.byteLength }, }); - try { - outputProcess.stdin?.write(audio); - } catch (error) { - fail("audio output command")(error as Error); - } + await params.transport.writeOutput(audio); }; const enqueueSpeakText = (text: string | undefined) => { @@ -670,7 +441,9 @@ export async function startCommandAgentAudioBridge(params: { return; } recordGoogleMeetRealtimeTranscript(transcript, "assistant", normalized); - params.logger.info(formatGoogleMeetTranscriptSummaryLog("agent assistant", normalized)); + params.logger.info( + formatGoogleMeetTranscriptSummaryLog(`${agentLogScope} assistant`, normalized), + ); const turnId = ensureTalkTurn(); emitTalkEvent({ type: "output.text.done", @@ -685,13 +458,13 @@ export async function startCommandAgentAudioBridge(params: { if (!result.success || !result.audioBuffer || !result.sampleRate) { throw new Error(result.error ?? "TTS conversion failed"); } - params.logger.info(formatGoogleMeetAgentTtsResultLog("agent", result)); + params.logger.info(formatGoogleMeetAgentTtsResultLog(agentLogScope, result)); emitTalkEvent({ type: "output.audio.started", turnId, payload: { meetingSessionId: params.meetingSessionId }, }); - writeOutputAudio( + await writeOutputAudio( convertGoogleMeetTtsAudioForBridge( result.audioBuffer, result.sampleRate, @@ -708,7 +481,9 @@ export async function startCommandAgentAudioBridge(params: { endTalkTurn(); }) .catch((error: unknown) => { - params.logger.warn(`[google-meet] agent TTS failed: ${formatErrorMessage(error)}`); + params.logger.warn( + `[google-meet] ${agentLogScope} TTS failed: ${formatErrorMessage(error)}`, + ); }); }; @@ -717,7 +492,7 @@ export async function startCommandAgentAudioBridge(params: { debounceMs: GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS, isStopped: () => stopped, logger: params.logger, - logPrefix: "[google-meet] agent", + logPrefix: `[google-meet] ${agentLogScope}`, responseStyle: "Brief, natural spoken answer for a live meeting.", fallbackText: "I hit an error while checking that. Please try again.", consult: ({ question, responseStyle }) => @@ -734,6 +509,15 @@ export async function startCommandAgentAudioBridge(params: { deliver: enqueueSpeakText, }); + params.transport.onFatal(() => { + void stop(); + }); + // onFatal replays a pre-registration failure synchronously; abort before creating a + // provider session that the already-completed stop() could never close. + if (stopped) { + throw new Error("Google Meet audio transport failed before transcription provider setup"); + } + sttSession = resolved.provider.createSession({ cfg: params.fullConfig, providerConfig: resolved.providerConfig, @@ -756,10 +540,13 @@ export async function startCommandAgentAudioBridge(params: { payload: { meetingSessionId: params.meetingSessionId, text: trimmed, role: "user" }, }); recordGoogleMeetRealtimeTranscript(transcript, "user", trimmed); - params.logger.info(formatGoogleMeetTranscriptSummaryLog("agent user", trimmed)); + params.logger.info(formatGoogleMeetTranscriptSummaryLog(`${agentLogScope} user`, trimmed)); if (isGoogleMeetLikelyAssistantEchoTranscript({ transcript, text: trimmed })) { params.logger.info( - formatGoogleMeetTranscriptSummaryLog("agent ignored assistant echo transcript", trimmed), + formatGoogleMeetTranscriptSummaryLog( + `${agentLogScope} ignored assistant echo transcript`, + trimmed, + ), ); return; } @@ -767,7 +554,7 @@ export async function startCommandAgentAudioBridge(params: { }, onError: (error) => { params.logger.warn( - `[google-meet] agent transcription bridge failed: ${formatErrorMessage(error)}`, + `[google-meet] ${agentLogScope} transcription bridge failed: ${formatErrorMessage(error)}`, ); emitTalkEvent({ type: "session.error", @@ -782,18 +569,12 @@ export async function startCommandAgentAudioBridge(params: { type: "session.started", payload: { meetingSessionId: params.meetingSessionId, provider: resolved.provider.id }, }); - await sttSession.connect(); - realtimeReady = true; - emitTalkEvent({ - type: "session.ready", - payload: { meetingSessionId: params.meetingSessionId }, - }); - - inputProcess.stdout?.on("data", (chunk) => { - if (stopped) { + // Drain transport input while connect() is pending so the capture pipe never backpressures; + // chunks before session.ready are dropped instead of arriving later as a stale burst. + params.transport.startInput((audio) => { + if (stopped || !realtimeReady || audio.byteLength === 0) { return; } - const audio = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); if (Date.now() < suppressInputUntil) { lastSuppressedInputAt = new Date().toISOString(); suppressedInputBytes += audio.byteLength; @@ -809,10 +590,18 @@ export async function startCommandAgentAudioBridge(params: { sttSession?.sendAudio(convertGoogleMeetBridgeAudioForStt(audio, params.config)); }); + await sttSession.connect(); + if (stopped) { + throw new Error("Google Meet audio transport stopped during transcription provider setup"); + } + realtimeReady = true; + emitTalkEvent({ + type: "session.ready", + payload: { meetingSessionId: params.meetingSessionId }, + }); + return { providerId: resolved.provider.id, - inputCommand: params.inputCommand, - outputCommand: params.outputCommand, speak: enqueueSpeakText, getHealth: () => ({ providerConnected: sttSession?.isConnected() ?? false, @@ -825,6 +614,7 @@ export async function startCommandAgentAudioBridge(params: { lastInputBytes, lastOutputBytes: outputActivity.snapshot().sinkAudioBytes, suppressedInputBytes, + ...params.transport.getHealth?.(), ...getGoogleMeetRealtimeTranscriptHealth(transcript), recentTalkEvents: summarizeGoogleMeetTalkEvents(recentTalkEvents), bridgeClosed: stopped, @@ -833,31 +623,19 @@ export async function startCommandAgentAudioBridge(params: { }; } -export async function startCommandRealtimeAudioBridge(params: { +export async function startMeetRealtimeEngine(params: { config: GoogleMeetConfig; fullConfig: OpenClawConfig; runtime: PluginRuntime; meetingSessionId: string; requesterSessionKey?: string; - inputCommand: string[]; - outputCommand: string[]; + logPrefix?: "node"; + talkSessionId?: string; + talkContext?: { nodeId: string; bridgeId: string }; + transport: MeetRealtimeAudioTransport; logger: RuntimeLogger; providers?: RealtimeVoiceProviderPlugin[]; - spawn?: SpawnFn; -}): Promise { - const input = splitCommand(params.inputCommand); - const output = splitCommand(params.outputCommand); - const spawnFn: SpawnFn = - params.spawn ?? - ((command, args, options) => spawn(command, args, options) as unknown as BridgeProcess); - const spawnOutputProcess = () => - spawnFn(output.command, output.args, { - stdio: ["pipe", "ignore", "pipe"], - }); - let outputProcess = spawnOutputProcess(); - const inputProcess = spawnFn(input.command, input.args, { - stdio: ["ignore", "pipe", "pipe"], - }); +}): Promise { let stopped = false; let bridge: RealtimeVoiceBridgeSession | null = null; let realtimeReady = false; @@ -871,7 +649,7 @@ export async function startCommandRealtimeAudioBridge(params: { let lastSuppressedInputAt: string | undefined; let suppressInputUntil = 0; let lastOutputPlayableUntilMs = 0; - let bargeInInputProcess: BridgeProcess | undefined; + const realtimeLogScope = params.logPrefix ? `${params.logPrefix} realtime` : "realtime"; const suppressInputForOutput = (audio: Buffer) => { const suppression = recordGoogleMeetOutputActivity({ @@ -896,158 +674,59 @@ export async function startCommandRealtimeAudioBridge(params: { bridge?.close(); } catch (error) { params.logger.debug?.( - `[google-meet] realtime voice bridge close ignored: ${formatErrorMessage(error)}`, + `[google-meet] ${realtimeLogScope}${params.logPrefix ? "" : " voice"} bridge close ignored: ${formatErrorMessage(error)}`, ); } - terminateBridgeProcess(inputProcess); - terminateBridgeProcess(outputProcess); - if (bargeInInputProcess) { - terminateBridgeProcess(bargeInInputProcess); - } - }; - - const fail = (label: string) => (error: Error) => { - params.logger.warn(`[google-meet] ${label} failed: ${formatErrorMessage(error)}`); - void stop(); - }; - const attachOutputProcessHandlers = (proc: BridgeProcess) => { - proc.on("error", (error) => { - if (proc !== outputProcess) { - return; - } - fail("audio output command")(error); - }); - proc.stdin?.on?.("error", (error: Error) => { - if (proc !== outputProcess) { - return; - } - fail("audio output command")(error); - }); - proc.on("exit", (code, signal) => { - if (proc !== outputProcess) { - return; - } - if (!stopped) { - params.logger.warn( - `[google-meet] audio output command exited (${code ?? signal ?? "done"})`, - ); - void stop(); - } - }); - proc.stderr?.on("data", (chunk) => { - params.logger.debug?.(`[google-meet] audio output: ${String(chunk).trim()}`); - }); - proc.stderr?.on("error", (error: Error) => { - if (proc !== outputProcess) { - return; - } - fail("audio output command stderr")(error); - }); + await params.transport.stop(); + await params.transport.dispose(); }; const clearOutputPlayback = () => { if (stopped) { return; } - const previousOutput = outputProcess; - outputProcess = spawnOutputProcess(); - attachOutputProcessHandlers(outputProcess); clearCount += 1; lastClearAt = new Date().toISOString(); suppressInputUntil = 0; lastOutputPlayableUntilMs = 0; - params.logger.debug?.( - `[google-meet] cleared realtime audio output buffer by restarting playback command`, - ); - terminateBridgeProcess(previousOutput, "SIGKILL"); + void params.transport.clearOutput().catch((error: unknown) => { + params.logger.warn( + `[google-meet] ${params.logPrefix ? `${params.logPrefix} audio clear` : "audio output clear"} failed: ${formatErrorMessage(error)}`, + ); + void stop(); + }); }; const writeOutputAudio = (audio: Buffer) => { - try { - outputProcess.stdin?.write(audio); - } catch (error) { - fail("audio output command")(error as Error); - } + void params.transport.writeOutput(audio).catch((error: unknown) => { + params.logger.warn( + `[google-meet] ${params.logPrefix ? `${params.logPrefix} audio output` : "audio output"} failed: ${formatErrorMessage(error)}`, + ); + void stop(); + }); }; const startHumanBargeInMonitor = () => { - const commandArgv = params.config.chrome.bargeInInputCommand; - if (!commandArgv) { + if (!params.transport.startBargeInMonitor) { return; } - const command = splitCommand(commandArgv); - let lastBargeInAt = 0; - bargeInInputProcess = spawnFn(command.command, command.args, { - stdio: ["ignore", "pipe", "pipe"], - }); - bargeInInputProcess.stdout?.on("data", (chunk) => { + params.transport.startBargeInMonitor(() => { if (stopped || !outputActivity.isInterruptible()) { - return; + return false; } const now = Date.now(); const playbackActive = now <= Math.max(lastOutputPlayableUntilMs, suppressInputUntil); const lastOutputAudioAt = outputActivity.snapshot().lastAudioAt; if (!playbackActive && (lastOutputAudioAt === undefined || now - lastOutputAudioAt > 1_000)) { - return; + return false; } - if (now - lastBargeInAt < params.config.chrome.bargeInCooldownMs) { - return; - } - const audio = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - const stats = readPcm16Stats(audio); - if ( - stats.rms < params.config.chrome.bargeInRmsThreshold && - stats.peak < params.config.chrome.bargeInPeakThreshold - ) { - return; - } - lastBargeInAt = now; suppressInputUntil = 0; const beforeClearCount = clearCount; bridge?.handleBargeIn({ audioPlaybackActive: true }); + // Provider clear callbacks normally flush; this fallback keeps barge-in and playback coupled. if (beforeClearCount === clearCount) { clearOutputPlayback(); } - params.logger.debug?.( - `[google-meet] human barge-in detected by local input (rms=${Math.round( - stats.rms, - )}, peak=${stats.peak})`, - ); - }); - bargeInInputProcess.stdout?.on("error", (error: Error) => { - params.logger.warn( - `[google-meet] human barge-in input stdout failed: ${formatErrorMessage(error)}`, - ); - }); - bargeInInputProcess.stderr?.on("data", (chunk) => { - params.logger.debug?.(`[google-meet] barge-in input: ${String(chunk).trim()}`); - }); - bargeInInputProcess.stderr?.on("error", (error: Error) => { - params.logger.warn( - `[google-meet] human barge-in input stderr failed: ${formatErrorMessage(error)}`, - ); - }); - bargeInInputProcess.on("error", (error) => { - params.logger.warn(`[google-meet] human barge-in input failed: ${formatErrorMessage(error)}`); - }); - bargeInInputProcess.on("exit", (code, signal) => { - if (!stopped) { - params.logger.debug?.( - `[google-meet] human barge-in input exited (${code ?? signal ?? "done"})`, - ); - } + return true; }); }; - inputProcess.on("error", fail("audio input command")); - inputProcess.on("exit", (code, signal) => { - if (!stopped) { - params.logger.warn(`[google-meet] audio input command exited (${code ?? signal ?? "done"})`); - void stop(); - } - }); - attachOutputProcessHandlers(outputProcess); - inputProcess.stderr?.on("data", (chunk) => { - params.logger.debug?.(`[google-meet] audio input: ${String(chunk).trim()}`); - }); - inputProcess.stdout?.on("error", fail("audio input command stdout")); - inputProcess.stderr?.on("error", fail("audio input command stderr")); const resolved = resolveGoogleMeetRealtimeProvider({ config: params.config, @@ -1068,7 +747,7 @@ export async function startCommandRealtimeAudioBridge(params: { const realtimeEvents: GoogleMeetRealtimeEventEntry[] = []; const talk: TalkSessionController = createTalkSessionController( { - sessionId: `google-meet:${params.meetingSessionId}:command-realtime`, + sessionId: params.talkSessionId ?? `google-meet:${params.meetingSessionId}:command-realtime`, mode: "realtime", transport: "gateway-relay", brain: strategy === "bidi" ? "direct-tools" : "agent-consult", @@ -1077,6 +756,14 @@ export async function startCommandRealtimeAudioBridge(params: { { onEvent: recordTalkObservabilityEvent }, ); const recentTalkEvents: TalkEvent[] = []; + const meetingTalkPayload = params.talkContext + ? { bridgeId: params.talkContext.bridgeId, meetingSessionId: params.meetingSessionId } + : { meetingSessionId: params.meetingSessionId }; + const outputTalkPayload = params.talkContext + ? { bridgeId: params.talkContext.bridgeId } + : { meetingSessionId: params.meetingSessionId }; + const reasonTalkPayload = (reason: string) => + params.talkContext ? { bridgeId: params.talkContext.bridgeId, reason } : { reason }; const rememberTalkEvent = (event: TalkEvent | undefined): void => { if (event) { pushGoogleMeetTalkEvent(recentTalkEvents, event); @@ -1087,7 +774,7 @@ export async function startCommandRealtimeAudioBridge(params: { }; const ensureTalkTurn = (): string => { const turn = talk.ensureTurn({ - payload: { meetingSessionId: params.meetingSessionId }, + payload: meetingTalkPayload, }); if (turn.event) { rememberTalkEvent(turn.event); @@ -1097,13 +784,13 @@ export async function startCommandRealtimeAudioBridge(params: { const finishOutputAudio = (reason: string): void => { rememberTalkEvent( talk.finishOutputAudio({ - payload: { reason }, + payload: reasonTalkPayload(reason), }), ); }; const endTalkTurn = (reason = "completed"): void => { const ended = talk.endTurn({ - payload: { reason }, + payload: reasonTalkPayload(reason), }); if (ended.ok) { rememberTalkEvent(ended.event); @@ -1111,14 +798,16 @@ export async function startCommandRealtimeAudioBridge(params: { }; emitTalkEvent({ type: "session.started", - payload: { meetingSessionId: params.meetingSessionId }, + payload: params.talkContext + ? { ...meetingTalkPayload, nodeId: params.talkContext.nodeId } + : meetingTalkPayload, }); const agentTalkback: RealtimeVoiceAgentTalkbackQueue | undefined = createRealtimeVoiceAgentTalkbackQueue({ debounceMs: GOOGLE_MEET_AGENT_TRANSCRIPT_DEBOUNCE_MS, isStopped: () => stopped, logger: params.logger, - logPrefix: "[google-meet] realtime agent", + logPrefix: `[google-meet] ${realtimeLogScope} agent`, responseStyle: "Brief, natural spoken answer for a live meeting.", fallbackText: "I hit an error while checking that. Please try again.", consult: ({ question, responseStyle }) => @@ -1136,6 +825,14 @@ export async function startCommandRealtimeAudioBridge(params: { bridge?.sendUserMessage(buildGoogleMeetSpeakExactUserMessage(text)); }, }); + params.transport.onFatal(() => { + void stop(); + }); + // onFatal replays a pre-registration failure synchronously; abort before creating a + // voice bridge that the already-completed stop() could never close. + if (stopped) { + throw new Error("Google Meet audio transport failed before realtime provider setup"); + } bridge = createRealtimeVoiceBridgeSession({ provider: resolved.provider, cfg: params.fullConfig, @@ -1155,7 +852,7 @@ export async function startCommandRealtimeAudioBridge(params: { rememberTalkEvent( talk.startOutputAudio({ turnId, - payload: { meetingSessionId: params.meetingSessionId }, + payload: outputTalkPayload, }).event, ); emitTalkEvent({ @@ -1193,18 +890,20 @@ export async function startCommandRealtimeAudioBridge(params: { emitTalkEvent({ type: "input.audio.committed", turnId, - payload: { meetingSessionId: params.meetingSessionId }, + payload: outputTalkPayload, final: true, }); } if (isFinal) { recordGoogleMeetRealtimeTranscript(transcript, role, text); - params.logger.info(formatGoogleMeetTranscriptSummaryLog(`realtime ${role}`, text)); + params.logger.info( + formatGoogleMeetTranscriptSummaryLog(`${realtimeLogScope} ${role}`, text), + ); if (role === "user" && strategy === "agent") { if (isGoogleMeetLikelyAssistantEchoTranscript({ transcript, text })) { params.logger.info( formatGoogleMeetTranscriptSummaryLog( - "realtime ignored assistant echo transcript", + `${realtimeLogScope} ignored assistant echo transcript`, text, ), ); @@ -1226,7 +925,7 @@ export async function startCommandRealtimeAudioBridge(params: { emitTalkEvent({ type: "input.audio.committed", turnId, - payload: { meetingSessionId: params.meetingSessionId, source: event.type }, + payload: { ...outputTalkPayload, source: event.type }, final: true, }); } else if (event.type === "response.done") { @@ -1248,7 +947,9 @@ export async function startCommandRealtimeAudioBridge(params: { event.type === "conversation.item.input_audio_transcription.failed" ) { const detail = event.detail ? ` ${event.detail}` : ""; - params.logger.info(`[google-meet] realtime ${event.direction}:${event.type}${detail}`); + params.logger.info( + `[google-meet] ${realtimeLogScope} ${event.direction}:${event.type}${detail}`, + ); } }, onToolCall: (event, session) => { @@ -1281,7 +982,10 @@ export async function startCommandRealtimeAudioBridge(params: { payload: { message: formatErrorMessage(error) }, final: true, }); - fail("realtime voice bridge")(error); + params.logger.warn( + `[google-meet] ${realtimeLogScope} voice bridge failed: ${formatErrorMessage(error)}`, + ); + void stop(); }, onClose: (reason) => { realtimeReady = false; @@ -1299,36 +1003,41 @@ export async function startCommandRealtimeAudioBridge(params: { realtimeReady = true; emitTalkEvent({ type: "session.ready", - payload: { meetingSessionId: params.meetingSessionId }, + payload: outputTalkPayload, }); }, }); startHumanBargeInMonitor(); - inputProcess.stdout?.on("data", (chunk) => { - const audio = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - if (!stopped && audio.byteLength > 0) { - if (Date.now() < suppressInputUntil) { - lastSuppressedInputAt = new Date().toISOString(); - suppressedInputBytes += audio.byteLength; - return; - } - lastInputAt = new Date().toISOString(); - lastInputBytes += audio.byteLength; - emitTalkEvent({ - type: "input.audio.delta", - turnId: ensureTalkTurn(), - payload: { byteLength: audio.byteLength }, - }); - bridge?.sendAudio(Buffer.from(audio)); + // Drain transport input while connect() is pending so the capture pipe never backpressures. + // Pre-connect audio is forwarded; the voice bridge owns buffering, matching the previous + // local command-pair behavior. + params.transport.startInput((audio) => { + if (stopped || audio.byteLength === 0) { + return; } + if (Date.now() < suppressInputUntil) { + lastSuppressedInputAt = new Date().toISOString(); + suppressedInputBytes += audio.byteLength; + return; + } + lastInputAt = new Date().toISOString(); + lastInputBytes += audio.byteLength; + emitTalkEvent({ + type: "input.audio.delta", + turnId: ensureTalkTurn(), + payload: { byteLength: audio.byteLength }, + }); + bridge?.sendAudio(audio); }); await bridge.connect(); + if (stopped) { + throw new Error("Google Meet audio transport stopped during realtime provider setup"); + } + return { providerId: resolved.provider.id, - inputCommand: params.inputCommand, - outputCommand: params.outputCommand, speak: (instructions) => { bridge?.triggerGreeting(instructions); }, @@ -1343,6 +1052,7 @@ export async function startCommandRealtimeAudioBridge(params: { lastInputBytes, lastOutputBytes: outputActivity.snapshot().sinkAudioBytes, suppressedInputBytes, + ...params.transport.getHealth?.(), ...getGoogleMeetRealtimeTranscriptHealth(transcript), ...getGoogleMeetRealtimeEventHealth(realtimeEvents), recentTalkEvents: summarizeGoogleMeetTalkEvents(recentTalkEvents), diff --git a/extensions/google-meet/src/transports/chrome.ts b/extensions/google-meet/src/transports/chrome.ts index bad04701e7e1..f6deeef87184 100644 --- a/extensions/google-meet/src/transports/chrome.ts +++ b/extensions/google-meet/src/transports/chrome.ts @@ -6,15 +6,12 @@ import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; import type { RuntimeLogger } from "openclaw/plugin-sdk/plugin-runtime"; import { uniqueStrings } from "openclaw/plugin-sdk/string-coerce-runtime"; import type { GoogleMeetConfig, GoogleMeetMode } from "../config.js"; +import { createLocalMeetRealtimeAudioTransport } from "../realtime-local-audio-transport.js"; +import { createNodeMeetRealtimeAudioTransport } from "../realtime-node-audio-transport.js"; import { - startNodeAgentAudioBridge, - startNodeRealtimeAudioBridge, - type ChromeNodeRealtimeAudioBridgeHandle, -} from "../realtime-node.js"; -import { - startCommandAgentAudioBridge, - startCommandRealtimeAudioBridge, - type ChromeRealtimeAudioBridgeHandle, + startMeetAgentRealtimeEngine, + startMeetRealtimeEngine, + type MeetRealtimeAudioEngineHandle, } from "../realtime.js"; import { GOOGLE_MEET_SYSTEM_PROFILER_COMMAND, @@ -49,6 +46,17 @@ type BrowserRequestCaller = (params: BrowserRequestParams) => Promise; const GOOGLE_MEET_CAPTION_SETTLE_MS = 1_000; +type ChromeRealtimeAudioBridgeHandle = MeetRealtimeAudioEngineHandle & { + inputCommand: string[]; + outputCommand: string[]; +}; + +type ChromeNodeRealtimeAudioBridgeHandle = MeetRealtimeAudioEngineHandle & { + type: "node-command-pair"; + nodeId: string; + bridgeId: string; +}; + function isGoogleMeetTalkBackMode(mode: GoogleMeetMode): boolean { return mode === "agent" || mode === "bidi"; } @@ -164,20 +172,27 @@ export async function launchChromeMeet(params: { "Chrome talk-back mode requires chrome.audioInputCommand and chrome.audioOutputCommand, or chrome.audioBridgeCommand for an external bridge.", ); } - return { - type: "command-pair", - ...(params.mode === "agent" - ? await startCommandAgentAudioBridge({ + const transport = createLocalMeetRealtimeAudioTransport({ + inputCommand: params.config.chrome.audioInputCommand, + outputCommand: params.config.chrome.audioOutputCommand, + bargeInInputCommand: params.config.chrome.bargeInInputCommand, + bargeInRmsThreshold: params.config.chrome.bargeInRmsThreshold, + bargeInPeakThreshold: params.config.chrome.bargeInPeakThreshold, + bargeInCooldownMs: params.config.chrome.bargeInCooldownMs, + logger: params.logger, + }); + const engine = + params.mode === "agent" + ? await startMeetAgentRealtimeEngine({ config: params.config, fullConfig: params.fullConfig, runtime: params.runtime, meetingSessionId: params.meetingSessionId, requesterSessionKey: params.requesterSessionKey, - inputCommand: params.config.chrome.audioInputCommand, - outputCommand: params.config.chrome.audioOutputCommand, + transport, logger: params.logger, }) - : await startCommandRealtimeAudioBridge({ + : await startMeetRealtimeEngine({ config: { ...params.config, realtime: { ...params.config.realtime, strategy: "bidi" }, @@ -186,10 +201,14 @@ export async function launchChromeMeet(params: { runtime: params.runtime, meetingSessionId: params.meetingSessionId, requesterSessionKey: params.requesterSessionKey, - inputCommand: params.config.chrome.audioInputCommand, - outputCommand: params.config.chrome.audioOutputCommand, + transport, logger: params.logger, - })), + }); + return { + type: "command-pair", + inputCommand: params.config.chrome.audioInputCommand, + outputCommand: params.config.chrome.audioOutputCommand, + ...engine, }; }; @@ -1667,24 +1686,46 @@ export async function launchChromeMeetOnNode(params: { if (!result.bridgeId) { throw new Error("Google Meet node did not return an audio bridge id."); } - const bridge = await ( - params.mode === "agent" ? startNodeAgentAudioBridge : startNodeRealtimeAudioBridge - )({ - config: - params.mode === "agent" - ? params.config - : { - ...params.config, - realtime: { ...params.config.realtime, strategy: "bidi" }, - }, - fullConfig: params.fullConfig, + const transport = createNodeMeetRealtimeAudioTransport({ runtime: params.runtime, - meetingSessionId: params.meetingSessionId, - requesterSessionKey: params.requesterSessionKey, nodeId, bridgeId: result.bridgeId, logger: params.logger, + logPrefix: params.mode === "agent" ? "node agent" : "node", }); + const engine = + params.mode === "agent" + ? await startMeetAgentRealtimeEngine({ + config: params.config, + fullConfig: params.fullConfig, + runtime: params.runtime, + meetingSessionId: params.meetingSessionId, + requesterSessionKey: params.requesterSessionKey, + logPrefix: "node", + transport, + logger: params.logger, + }) + : await startMeetRealtimeEngine({ + config: { + ...params.config, + realtime: { ...params.config.realtime, strategy: "bidi" }, + }, + fullConfig: params.fullConfig, + runtime: params.runtime, + meetingSessionId: params.meetingSessionId, + requesterSessionKey: params.requesterSessionKey, + logPrefix: "node", + talkSessionId: `google-meet:${params.meetingSessionId}:${result.bridgeId}:node-realtime`, + talkContext: { nodeId, bridgeId: result.bridgeId }, + transport, + logger: params.logger, + }); + const bridge: ChromeNodeRealtimeAudioBridgeHandle = { + type: "node-command-pair", + nodeId, + bridgeId: result.bridgeId, + ...engine, + }; return { nodeId, launched: browserControl.launched || result.launched === true,