diff --git a/test/e2e/qa-lab/runtime/fixtures/voice-call-runtime-plugin/index.js b/test/e2e/qa-lab/runtime/fixtures/voice-call-runtime-plugin/index.js index c00688dee4a5..eae15b64d067 100644 --- a/test/e2e/qa-lab/runtime/fixtures/voice-call-runtime-plugin/index.js +++ b/test/e2e/qa-lab/runtime/fixtures/voice-call-runtime-plugin/index.js @@ -55,7 +55,10 @@ export default { }, }); api.registerGatewayMethod("qa.voiceCall.streamSession", async ({ params, respond }) => { - const runtime = globalThis[Symbol.for("openclaw.voice-call.runtime")]; + const coordinator = globalThis[Symbol.for("openclaw.voice-call.runtimeCoordinator")]; + const slot = coordinator?.slot; + const runtime = + slot?.state === "running" && slot.owner === coordinator.current ? slot.runtime : undefined; const callId = typeof params?.callId === "string" ? params.callId : ""; const call = runtime?.manager?.getCall?.(callId); const issue = runtime?.manager?.streamSessionIssuer; diff --git a/test/e2e/qa-lab/runtime/voice-call-runtime-plugin.test.ts b/test/e2e/qa-lab/runtime/voice-call-runtime-plugin.test.ts new file mode 100644 index 000000000000..c134fb58b2c2 --- /dev/null +++ b/test/e2e/qa-lab/runtime/voice-call-runtime-plugin.test.ts @@ -0,0 +1,145 @@ +import path from "node:path"; +import { pathToFileURL } from "node:url"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +type GatewayMethod = (options: { + params?: { callId?: unknown }; + respond: (ok: boolean, result?: unknown, error?: unknown) => void; +}) => Promise; + +type FixturePlugin = { + register(api: { + registerGatewayMethod(method: string, handler: GatewayMethod): void; + registerRealtimeVoiceProvider(provider: unknown): void; + }): void; +}; + +const runtimeCoordinatorKey = Symbol.for("openclaw.voice-call.runtimeCoordinator"); +const unavailableError = { + code: "UNAVAILABLE", + message: "Voice Call runtime stream issuer unavailable", +}; +const fixtureUrl = pathToFileURL( + path.resolve("test/e2e/qa-lab/runtime/fixtures/voice-call-runtime-plugin/index.js"), +).href; +const { default: fixturePlugin } = (await import(fixtureUrl)) as { default: FixturePlugin }; + +function setRuntimeCoordinator(coordinator: unknown): void { + (globalThis as Record)[runtimeCoordinatorKey] = coordinator; +} + +function registerStreamSessionMethod(): GatewayMethod { + let streamSessionMethod: GatewayMethod | undefined; + fixturePlugin.register({ + registerGatewayMethod(method, handler) { + if (method === "qa.voiceCall.streamSession") { + streamSessionMethod = handler; + } + }, + registerRealtimeVoiceProvider() {}, + }); + if (!streamSessionMethod) { + throw new Error("Voice Call fixture did not register qa.voiceCall.streamSession"); + } + return streamSessionMethod; +} + +async function expectUnavailable(streamSessionMethod: GatewayMethod): Promise { + const respond = vi.fn(); + await streamSessionMethod({ params: { callId: "call-123" }, respond }); + expect(respond).toHaveBeenCalledOnce(); + expect(respond).toHaveBeenCalledWith(false, undefined, unavailableError); +} + +describe("Voice Call runtime fixture", () => { + beforeEach(() => { + delete (globalThis as Record)[runtimeCoordinatorKey]; + }); + + afterEach(() => { + delete (globalThis as Record)[runtimeCoordinatorKey]; + vi.restoreAllMocks(); + }); + + it("returns unavailable without a runtime coordinator", async () => { + await expectUnavailable(registerStreamSessionMethod()); + }); + + it.each(["starting", "stopping"] as const)( + "returns unavailable while the runtime is %s", + async (state) => { + const owner = {}; + setRuntimeCoordinator({ + current: owner, + slot: { state, owner, promise: Promise.resolve() }, + }); + + await expectUnavailable(registerStreamSessionMethod()); + }, + ); + + it("returns unavailable for a running slot owned by a stale generation", async () => { + const getCall = vi.fn(); + const streamSessionIssuer = vi.fn(); + setRuntimeCoordinator({ + current: {}, + slot: { + state: "running", + owner: {}, + runtime: { manager: { getCall, streamSessionIssuer } }, + }, + }); + + await expectUnavailable(registerStreamSessionMethod()); + expect(getCall).not.toHaveBeenCalled(); + expect(streamSessionIssuer).not.toHaveBeenCalled(); + }); + + it("issues a stream session from the current running generation", async () => { + const callId = "call-123"; + const owner = {}; + const call = { + providerCallId: "provider-call-456", + from: "+15550000001", + to: "+15550000002", + direction: "outbound", + }; + const getCall = vi.fn(() => call); + const streamSessionIssuer = vi.fn(() => ({ + streamUrl: "wss://voice.example.test/stream", + token: "stream-token", + })); + setRuntimeCoordinator({ + current: owner, + slot: { + state: "running", + owner, + runtime: { + manager: { getCall, streamSessionIssuer }, + webhookUrl: "https://voice.example.test/webhook", + }, + }, + }); + const respond = vi.fn(); + + await registerStreamSessionMethod()({ params: { callId }, respond }); + + expect(getCall).toHaveBeenCalledOnce(); + expect(getCall).toHaveBeenCalledWith(callId); + expect(streamSessionIssuer).toHaveBeenCalledOnce(); + expect(streamSessionIssuer).toHaveBeenCalledWith({ + providerName: "twilio", + callId, + from: call.from, + to: call.to, + direction: call.direction, + }); + expect(respond).toHaveBeenCalledOnce(); + expect(respond).toHaveBeenCalledWith(true, { + streamUrl: "wss://voice.example.test/stream", + token: "stream-token", + providerCallId: call.providerCallId, + webhookUrl: "https://voice.example.test/webhook", + }); + }); +});