fix(qa): use current voice call runtime generation (#120920)

Punchcard-Session: amber-workshop-workshop-36
2026-08-12 21:53:00 -06:00 · 2026-08-09 14:01:40 +08:00
parent b164a4222e
commit 13e9bf2317
2 changed files with 149 additions and 1 deletions
@@ -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;
@@ -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<void>;
+
+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<PropertyKey, unknown>)[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<void> {
+  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<PropertyKey, unknown>)[runtimeCoordinatorKey];
+  });
+
+  afterEach(() => {
+    delete (globalThis as Record<PropertyKey, unknown>)[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",
+    });
+  });
+});