diff --git a/.github/workflows/openclaw-performance.yml b/.github/workflows/openclaw-performance.yml index 0185f411b367..66aac720f846 100644 --- a/.github/workflows/openclaw-performance.yml +++ b/.github/workflows/openclaw-performance.yml @@ -807,7 +807,8 @@ jobs: OPENCLAW_HOME="$gateway_home" OPENCLAW_STATE_DIR="$gateway_state" OPENCLAW_CONFIG_PATH="$gateway_config" OPENCLAW_GATEWAY_PORT="$gateway_port" \ node --import tsx scripts/bench-cli-startup.ts \ - --case gatewayHealthJson \ + --case gatewayHealthJsonConnected \ + --case gatewayHealthJsonFirstDevice \ --case configGetGatewayPort \ --runs "$source_runs" \ --warmup 1 \ diff --git a/extensions/openai/realtime-audio-buffer-ownership.test.ts b/extensions/openai/realtime-audio-buffer-ownership.test.ts new file mode 100644 index 000000000000..dd26c050f524 --- /dev/null +++ b/extensions/openai/realtime-audio-buffer-ownership.test.ts @@ -0,0 +1,132 @@ +import { once } from "node:events"; +import type { RealtimeVoiceBridge } from "openclaw/plugin-sdk/realtime-voice"; +import { describe, expect, it, vi } from "vitest"; +import WebSocket, { type RawData, WebSocketServer } from "ws"; +import { OpenAIQuicksilverVoiceBridge } from "./realtime-quicksilver-bridge.js"; +import { buildOpenAIRealtimeVoiceProvider } from "./realtime-voice-provider.js"; + +type RealtimeProviderKind = "native" | "gpt-live"; + +function parseWebSocketMessage(data: RawData): Record { + const bytes = Buffer.isBuffer(data) + ? data + : Array.isArray(data) + ? Buffer.concat(data) + : Buffer.from(data); + return JSON.parse(bytes.toString("utf8")) as Record; +} + +async function withRealtimeProvider( + kind: RealtimeProviderKind, + prepareAudio: (bridge: RealtimeVoiceBridge) => void, +): Promise>> { + const audioEventType = kind === "native" ? "input_audio_buffer.append" : "input_audio.append"; + const server = new WebSocketServer({ host: "127.0.0.1", port: 0 }); + await once(server, "listening"); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("expected an available local realtime WebSocket address"); + } + const received: Array> = []; + server.once("connection", (socket) => { + socket.on("message", (payload) => { + const event = parseWebSocketMessage(payload); + received.push(event); + if (event.type === "session.update") { + socket.send( + JSON.stringify( + kind === "native" + ? { type: "session.updated" } + : { + type: "session.started", + session: { id: "fixture-live", expires_at: Math.floor(Date.now() / 1000) + 60 }, + }, + ), + ); + } + }); + }); + + const endpoint = `http://127.0.0.1:${address.port}`; + const bridge = + kind === "native" + ? buildOpenAIRealtimeVoiceProvider().createBridge({ + providerConfig: { + apiKey: "fixture-local", // pragma: allowlist secret + azureEndpoint: endpoint, + azureDeployment: "fixture-realtime", + }, + onAudio: vi.fn(), + onClearAudio: vi.fn(), + }) + : new OpenAIQuicksilverVoiceBridge({ + providerConfig: {}, + model: "gpt-live-1-codex", + audioFormat: { encoding: "pcm16", sampleRateHz: 24000, channels: 1 }, + resolveAuth: async () => ({ type: "api-key", token: "fixture-local" }), + webSocketFactory: (_url, options) => new WebSocket(endpoint, options), + onAudio: vi.fn(), + onClearAudio: vi.fn(), + }); + + try { + prepareAudio(bridge); + await bridge.connect(); + await vi.waitFor(() => { + expect(received.some((event) => event.type === audioEventType)).toBe(true); + }); + return received; + } finally { + bridge.close(); + for (const client of server.clients) { + client.terminate(); + } + await new Promise((resolve, reject) => { + server.close((error) => (error ? reject(error) : resolve())); + }); + } +} + +describe("OpenAI realtime queued audio buffer ownership", () => { + it.each(["native", "gpt-live"])( + "%s preserves each reusable producer frame until the real WebSocket is ready", + async (kind) => { + const audioEventType = kind === "native" ? "input_audio_buffer.append" : "input_audio.append"; + const received = await withRealtimeProvider(kind, (bridge) => { + const producerAllocation = Buffer.alloc(2 * 1024 * 1024, 0x7f); + const producerView = producerAllocation.subarray(0, 1); + bridge.sendAudio(producerView); + producerAllocation[0] = 0x41; + bridge.sendAudio(producerView); + producerAllocation[0] = 0; + }); + + expect(received.filter((event) => event.type === audioEventType)).toEqual([ + { type: audioEventType, audio: "fw==" }, + { type: audioEventType, audio: "QQ==" }, + ]); + }, + ); + + it.each(["native", "gpt-live"])( + "%s rejects oversized producer frames before allocating a queued copy", + async (kind) => { + const audioEventType = kind === "native" ? "input_audio_buffer.append" : "input_audio.append"; + const received = await withRealtimeProvider(kind, (bridge) => { + const oversized = Buffer.alloc(1024 * 1024 + 1); + const copyBuffer = vi.spyOn(Buffer, "from"); + try { + bridge.sendAudio(oversized); + expect(copyBuffer).not.toHaveBeenCalled(); + } finally { + copyBuffer.mockRestore(); + } + bridge.sendAudio(Buffer.from([0x7f])); + }); + + expect(received.filter((event) => event.type === audioEventType)).toEqual([ + { type: audioEventType, audio: "fw==" }, + ]); + }, + ); +}); diff --git a/extensions/openai/realtime-quicksilver-bridge.ts b/extensions/openai/realtime-quicksilver-bridge.ts index de8583cfcca1..6d43551f56df 100644 --- a/extensions/openai/realtime-quicksilver-bridge.ts +++ b/extensions/openai/realtime-quicksilver-bridge.ts @@ -568,8 +568,10 @@ export class OpenAIQuicksilverVoiceBridge implements RealtimeVoiceBridge { ) { return; } - this.pendingAudio.push(audio); - this.pendingAudioBytes += audio.byteLength; + // Capture transports can recycle caller-owned views before the provider becomes ready. + const queuedAudio = Buffer.from(audio); + this.pendingAudio.push(queuedAudio); + this.pendingAudioBytes += queuedAudio.byteLength; } private resetTerminalState(): void { diff --git a/extensions/openai/realtime-voice-provider.ts b/extensions/openai/realtime-voice-provider.ts index 9e8e448c65ab..907783437518 100644 --- a/extensions/openai/realtime-voice-provider.ts +++ b/extensions/openai/realtime-voice-provider.ts @@ -1684,8 +1684,10 @@ class OpenAIRealtimeVoiceBridge implements RealtimeVoiceBridge { ) { return; } - this.pendingAudio.push(audio); - this.pendingAudioBytes += audio.byteLength; + // Capture transports can recycle caller-owned views before the provider becomes ready. + const queuedAudio = Buffer.from(audio); + this.pendingAudio.push(queuedAudio); + this.pendingAudioBytes += queuedAudio.byteLength; } private clearPendingAudio(): void { diff --git a/extensions/whatsapp/src/qa-driver.runtime.test.ts b/extensions/whatsapp/src/qa-driver.runtime.test.ts index f4395916ff73..45d0e2b08561 100644 --- a/extensions/whatsapp/src/qa-driver.runtime.test.ts +++ b/extensions/whatsapp/src/qa-driver.runtime.test.ts @@ -462,6 +462,100 @@ describe("startWhatsAppQaDriverSession", () => { await session.close(); }); + it.each([ + ...[ + { name: "captionless video note", message: { ptvMessage: {} } }, + { + name: "ephemeral captionless video note", + message: { ephemeralMessage: { message: { ptvMessage: {} } } }, + }, + { + name: "edited captionless video note", + message: { editedMessage: { message: { ptvMessage: {} } } }, + }, + ].map(({ name, message }) => ({ + name, + message, + expected: { kind: "media", mediaType: "video/mp4", text: "" }, + })), + ...[ + "pollCreationMessage", + "pollCreationMessageV2", + "pollCreationMessageV3", + "pollCreationMessageV5", + ].flatMap((pollKey) => { + const poll = { + [pollKey]: { + name: "Choose a time", + options: [{ optionName: "Morning" }, { optionName: "Afternoon" }], + }, + }; + const expected = { + kind: "poll", + poll: { question: "Choose a time", options: ["Morning", "Afternoon"] }, + }; + return [ + { name: pollKey, message: poll, expected }, + { + name: `ephemeral ${pollKey}`, + message: { ephemeralMessage: { message: poll } }, + expected, + }, + ]; + }), + ...["pollCreationMessageV3", "pollCreationMessageV5"].flatMap((pollKey) => { + const poll = { + [pollKey]: { + name: "Choose a time", + options: [{ optionName: "Morning" }, { optionName: "Afternoon" }], + }, + }; + const wrappedPoll = { pollCreationMessageV4: { message: poll } }; + const expected = { + kind: "poll", + poll: { question: "Choose a time", options: ["Morning", "Afternoon"] }, + }; + return [ + { name: `future-proof version-4 ${pollKey}`, message: wrappedPoll, expected }, + { + name: `ephemeral future-proof version-4 ${pollKey}`, + message: { ephemeralMessage: { message: wrappedPoll } }, + expected, + }, + { name: `edited ${pollKey}`, message: { editedMessage: { message: poll } }, expected }, + ]; + }), + ])("resolves live ingress waiters for $name", async ({ message, expected }) => { + const sock = createMockSocket(); + mocks.createWaSocket.mockResolvedValue(sock); + mocks.waitForWaConnection.mockResolvedValue(undefined); + mocks.jidToE164.mockReturnValue("+15551234567"); + + const session = await startWhatsAppQaDriverSession({ + authDir: "/tmp/openclaw-whatsapp-auth", + }); + + try { + const observed = session.waitForMessage({ + timeoutMs: 150, + match: (candidate) => candidate.kind === expected.kind, + }); + sock.ev.emit("messages.upsert", { + messages: [ + { + key: { fromMe: false, id: "observed-message", remoteJid: "12345@lid" }, + message, + } as WAMessage, + ], + }); + + await expect(observed).resolves.toMatchObject(expected); + expect(session.getObservedMessages()).toHaveLength(1); + } finally { + await session.close(); + } + }); + it("uses canonical WhatsApp media MIME defaults when Baileys omits MIME", async () => { const sock = createMockSocket(); mocks.createWaSocket.mockResolvedValue(sock); diff --git a/extensions/whatsapp/src/qa-driver.runtime.ts b/extensions/whatsapp/src/qa-driver.runtime.ts index dab09b25ad8b..43462d160449 100644 --- a/extensions/whatsapp/src/qa-driver.runtime.ts +++ b/extensions/whatsapp/src/qa-driver.runtime.ts @@ -1,5 +1,5 @@ // Whatsapp plugin module implements qa driver behavior. -import type { ConnectionState, proto, WAMessage } from "baileys"; +import { getContentType, type ConnectionState, type proto, type WAMessage } from "baileys"; import { formatLocationText } from "openclaw/plugin-sdk/channel-inbound"; import { describeReplyContext, @@ -180,19 +180,10 @@ function findMessageSection( if (current.depth >= 4) { continue; } - for (const wrapperName of [ - "botInvokeMessage", - "documentWithCaptionMessage", - "ephemeralMessage", - "groupMentionedMessage", - "viewOnceMessage", - "viewOnceMessageV2", - "viewOnceMessageV2Extension", - ]) { - const wrapper = current.value[wrapperName]; - if (isRecord(wrapper) && isRecord(wrapper.message)) { - queue.push({ depth: current.depth + 1, value: wrapper.message }); - } + const contentType = getContentType(current.value as proto.IMessage); + const wrapper = contentType ? current.value[contentType] : undefined; + if (isRecord(wrapper) && isRecord(wrapper.message)) { + queue.push({ depth: current.depth + 1, value: wrapper.message }); } } return undefined; @@ -218,6 +209,7 @@ function readPoll(message: unknown): WhatsAppQaDriverObservedPoll | undefined { "pollCreationMessage", "pollCreationMessageV2", "pollCreationMessageV3", + "pollCreationMessageV5", ]); if (!poll) { return undefined; @@ -244,6 +236,7 @@ function readMedia(message: unknown): const mediaSections = [ "imageMessage", "videoMessage", + "ptvMessage", "audioMessage", "documentMessage", "stickerMessage", diff --git a/scripts/bench-cli-startup.ts b/scripts/bench-cli-startup.ts index affb00029acd..349849d8b149 100644 --- a/scripts/bench-cli-startup.ts +++ b/scripts/bench-cli-startup.ts @@ -12,6 +12,7 @@ type CommandCase = { name: string; args: string[]; presets: readonly string[]; + stateScope?: "case" | "sample"; expectedExitCodes?: readonly number[]; expectedNonzeroOutputIncludes?: readonly string[]; firstOutputBudgetMs?: number; @@ -444,6 +445,19 @@ const COMMAND_CASES: readonly CommandCase[] = [ expectedExitCodes: [0, 1], expectedNonzeroOutputIncludes: ['"ok"', '"gateway_transport_error"'], }, + { + id: "gatewayHealthJsonConnected", + name: "gateway health --json (connected)", + args: ["gateway", "health", "--json"], + presets: [], + stateScope: "case", + }, + { + id: "gatewayHealthJsonFirstDevice", + name: "gateway health --json (first device)", + args: ["gateway", "health", "--json"], + presets: [], + }, { id: "configGetGatewayPort", name: "config get gateway.port", @@ -649,6 +663,8 @@ function buildConfigFixture(commandCase: CommandCase): Record | if ( commandCase.id !== "configGetGatewayPort" && commandCase.id !== "gatewayHealthJson" && + commandCase.id !== "gatewayHealthJsonConnected" && + commandCase.id !== "gatewayHealthJsonFirstDevice" && commandCase.id !== "health" && commandCase.id !== "healthJson" ) { @@ -717,8 +733,10 @@ async function runSample(params: { cpuProfDir?: string; heapProfDir?: string; rssHookPath: string; + runRoot?: string; }): Promise { - const runRoot = mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-bench-home-")); + const runRoot = params.runRoot ?? mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-bench-home-")); + const ownsRunRoot = params.runRoot == null; const stateDir = path.join(runRoot, ".openclaw"); const configPath = path.join(stateDir, "openclaw.json"); const configFixture = buildConfigFixture(params.commandCase); @@ -849,7 +867,9 @@ async function runSample(params: { }); }); } finally { - rmSync(runRoot, { recursive: true, force: true }); + if (ownsRunRoot) { + rmSync(runRoot, { recursive: true, force: true }); + } } } @@ -939,14 +959,24 @@ async function runCase(params: { }): Promise { const samples: Sample[] = []; const totalRuns = params.warmup + params.runs; - for (let i = 0; i < totalRuns; i += 1) { - const sample = await runSample(params); - if (i < params.warmup) { - continue; + const caseRunRoot = + params.commandCase.stateScope === "case" + ? mkdtempSync(path.join(os.tmpdir(), "openclaw-cli-bench-home-")) + : undefined; + try { + for (let i = 0; i < totalRuns; i += 1) { + const sample = await runSample({ ...params, runRoot: caseRunRoot }); + if (i < params.warmup) { + continue; + } + samples.push(sample); + } + return samples; + } finally { + if (caseRunRoot) { + rmSync(caseRunRoot, { recursive: true, force: true }); } - samples.push(sample); } - return samples; } function tailLines(value: string, maxLines: number): string { diff --git a/scripts/lib/state-schema-inline-plugin.d.mts b/scripts/lib/state-schema-inline-plugin.d.mts new file mode 100644 index 000000000000..329eaf70ac1a --- /dev/null +++ b/scripts/lib/state-schema-inline-plugin.d.mts @@ -0,0 +1,9 @@ +export const STATE_SCHEMA_INLINE_PLUGIN_NAME: string; + +export function createStateSchemaInlinePlugin(rootDir?: string): { + name: string; + load( + this: { addWatchFile(id: string): void }, + id: string, + ): { code: string; moduleType: "js" } | null; +}; diff --git a/scripts/lib/state-schema-inline-plugin.mjs b/scripts/lib/state-schema-inline-plugin.mjs new file mode 100644 index 000000000000..fb48b55d8f0d --- /dev/null +++ b/scripts/lib/state-schema-inline-plugin.mjs @@ -0,0 +1,40 @@ +import fs from "node:fs"; +import path from "node:path"; + +export const STATE_SCHEMA_INLINE_PLUGIN_NAME = "openclaw:inline-state-schemas"; + +const STATE_SCHEMA_MODULES = [ + { + modulePath: "src/state/openclaw-state-schema.ts", + schemaPath: "src/state/openclaw-state-schema.sql", + exportName: "OPENCLAW_STATE_SCHEMA_SQL", + }, + { + modulePath: "src/state/openclaw-agent-schema.ts", + schemaPath: "src/state/openclaw-agent-schema.sql", + exportName: "OPENCLAW_AGENT_SCHEMA_SQL", + }, +]; + +/** Inline canonical schema bytes so bundled consumers need no SQL asset. */ +export function createStateSchemaInlinePlugin(rootDir = process.cwd()) { + const schemasByModulePath = new Map( + STATE_SCHEMA_MODULES.map((schema) => [path.resolve(rootDir, schema.modulePath), schema]), + ); + + return { + name: STATE_SCHEMA_INLINE_PLUGIN_NAME, + load(id) { + const schema = schemasByModulePath.get(path.resolve(id)); + if (!schema) { + return null; + } + const schemaPath = path.resolve(rootDir, schema.schemaPath); + this.addWatchFile(schemaPath); + return { + code: `export const ${schema.exportName} = ${JSON.stringify(fs.readFileSync(schemaPath, "utf8"))};\n`, + moduleType: "js", + }; + }, + }; +} diff --git a/src/plugins/capability-provider-runtime.test.ts b/src/plugins/capability-provider-runtime.test.ts index 7f9ddd94cba3..9aacbec76877 100644 --- a/src/plugins/capability-provider-runtime.test.ts +++ b/src/plugins/capability-provider-runtime.test.ts @@ -80,8 +80,8 @@ vi.mock("./manifest-registry.js", async (importOriginal) => { }; }); -vi.mock("./plugin-registry.js", async (importOriginal) => { - const actual = await importOriginal(); +vi.mock("./plugin-registry-snapshot.js", async (importOriginal) => { + const actual = await importOriginal(); return { ...actual, loadPluginRegistrySnapshot: mocks.loadPluginRegistrySnapshot, diff --git a/src/plugins/migration-provider-runtime.test.ts b/src/plugins/migration-provider-runtime.test.ts index ab91bc697aa0..fa710f244440 100644 --- a/src/plugins/migration-provider-runtime.test.ts +++ b/src/plugins/migration-provider-runtime.test.ts @@ -61,7 +61,7 @@ vi.mock("./active-runtime-registry.js", () => ({ }, })); -vi.mock("./plugin-registry.js", () => ({ +vi.mock("./plugin-registry-snapshot.js", () => ({ loadPluginRegistrySnapshot: mocks.loadPluginRegistrySnapshot, loadPluginRegistrySnapshotWithMetadata: mocks.loadPluginRegistrySnapshotWithMetadata, })); diff --git a/src/plugins/providers.runtime.consult-current-snapshot.test.ts b/src/plugins/providers.runtime.consult-current-snapshot.test.ts index da2fa927c1f0..b08058ccfe5c 100644 --- a/src/plugins/providers.runtime.consult-current-snapshot.test.ts +++ b/src/plugins/providers.runtime.consult-current-snapshot.test.ts @@ -16,8 +16,8 @@ import { resetPluginRuntimeStateForTest } from "./runtime.js"; const loadPluginRegistrySnapshotWithMetadata = vi.hoisted(() => vi.fn()); const loadPluginManifestRegistryForInstalledIndex = vi.hoisted(() => vi.fn()); -vi.mock("./plugin-registry.js", async (importOriginal) => { - const actual = await importOriginal(); +vi.mock("./plugin-registry-snapshot.js", async (importOriginal) => { + const actual = await importOriginal(); return { ...actual, loadPluginRegistrySnapshotWithMetadata: (params: unknown) => diff --git a/src/plugins/web-fetch-providers.runtime.test.ts b/src/plugins/web-fetch-providers.runtime.test.ts index 5b010654b66d..9ded1713f780 100644 --- a/src/plugins/web-fetch-providers.runtime.test.ts +++ b/src/plugins/web-fetch-providers.runtime.test.ts @@ -104,9 +104,10 @@ function createRuntimeWebFetchProvider() { describe("resolvePluginWebFetchProviders", () => { beforeAll(async () => { - vi.doMock("./plugin-registry.js", async () => { - const actual = - await vi.importActual("./plugin-registry.js"); + vi.doMock("./plugin-registry-snapshot.js", async () => { + const actual = await vi.importActual( + "./plugin-registry-snapshot.js", + ); return { ...actual, loadPluginRegistrySnapshotWithMetadata: () => ({ diff --git a/test/scripts/bench-cli-startup.test.ts b/test/scripts/bench-cli-startup.test.ts index a4792df914de..506d76812331 100644 --- a/test/scripts/bench-cli-startup.test.ts +++ b/test/scripts/bench-cli-startup.test.ts @@ -462,6 +462,18 @@ describe("bench-cli-startup", () => { args: ["gateway", "health", "--json"], presets: ["real"], }, + { + id: "gatewayHealthJsonConnected", + name: "gateway health --json (connected)", + args: ["gateway", "health", "--json"], + presets: [], + }, + { + id: "gatewayHealthJsonFirstDevice", + name: "gateway health --json (first device)", + args: ["gateway", "health", "--json"], + presets: [], + }, { id: "health", name: "health", args: ["health"], presets: ["startup", "real"] }, { id: "healthJson", @@ -485,16 +497,22 @@ describe("bench-cli-startup", () => { expect(testing.parseGatewayPortEnv("::1")).toBe(32123); expect(testing.parseGatewayPortEnv("[::1]")).toBe(32123); - expect( - withEnv({ OPENCLAW_GATEWAY_PORT: "45678" }, () => - testing.buildConfigFixture({ - id: "gatewayHealthJson", - name: "gateway health --json", - args: ["gateway", "health", "--json"], - presets: ["real"], - }), - ), - ).toMatchObject({ gateway: { port: 45678 } }); + for (const id of [ + "gatewayHealthJson", + "gatewayHealthJsonConnected", + "gatewayHealthJsonFirstDevice", + ]) { + expect( + withEnv({ OPENCLAW_GATEWAY_PORT: "45678" }, () => + testing.buildConfigFixture({ + id, + name: "gateway health --json", + args: ["gateway", "health", "--json"], + presets: [], + }), + ), + ).toMatchObject({ gateway: { port: 45678 } }); + } for (const invalid of ["45678abc", "127.0.0.1:45678abc"]) { expect(() => diff --git a/test/scripts/cli-startup-bench-spawner.test.ts b/test/scripts/cli-startup-bench-spawner.test.ts index 14c6f0f43d9b..50a84d997b61 100644 --- a/test/scripts/cli-startup-bench-spawner.test.ts +++ b/test/scripts/cli-startup-bench-spawner.test.ts @@ -34,6 +34,109 @@ describe("CLI startup benchmark script spawners", () => { ); }); + it("reuses warmed state for gateway health while isolating first-device samples", () => { + const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-bench-state-scope-test-")); + try { + const fixturePath = path.join(tmpDir, "record-home.mjs"); + const homeLogPath = path.join(tmpDir, "homes.log"); + fs.writeFileSync( + fixturePath, + [ + 'import { appendFileSync } from "node:fs";', + "appendFileSync(process.env.OPENCLAW_BENCH_HOME_LOG, `${process.env.HOME}\\n`);", + "console.log('{\"ok\":true}');", + "", + ].join("\n"), + ); + + const runCase = (caseId: string) => { + fs.rmSync(homeLogPath, { force: true }); + execFileSync( + process.execPath, + [ + "--import", + "tsx", + "scripts/bench-cli-startup.ts", + "--entry", + fixturePath, + "--case", + caseId, + "--runs", + "2", + "--warmup", + "1", + ], + { + cwd: process.cwd(), + env: { + ...process.env, + OPENCLAW_BENCH_HOME_LOG: homeLogPath, + }, + stdio: "pipe", + }, + ); + return fs.readFileSync(homeLogPath, "utf8").trim().split("\n"); + }; + + const warmedHomes = runCase("gatewayHealthJsonConnected"); + expect(warmedHomes).toHaveLength(3); + expect(new Set(warmedHomes).size).toBe(1); + expect(warmedHomes.every((home) => !fs.existsSync(home))).toBe(true); + + for (const caseId of ["gatewayHealthJson", "gatewayHealthJsonFirstDevice"]) { + const sampleHomes = runCase(caseId); + expect(sampleHomes).toHaveLength(3); + expect(new Set(sampleHomes).size).toBe(3); + expect(sampleHomes.every((home) => !fs.existsSync(home))).toBe(true); + } + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); + + it("requires connected gateway health probes to exit successfully", () => { + const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-bench-connected-test-")); + try { + const fixturePath = path.join(tmpDir, "transport-error.mjs"); + fs.writeFileSync( + fixturePath, + [ + 'console.log(\'{"ok":false,"gateway_transport_error":"closed"}\');', + "process.exitCode = 1;", + "", + ].join("\n"), + ); + + const runCase = (caseId: string) => + spawnSync( + process.execPath, + [ + "--import", + "tsx", + "scripts/bench-cli-startup.ts", + "--entry", + fixturePath, + "--case", + caseId, + "--runs", + "1", + "--warmup", + "0", + ], + { cwd: process.cwd(), encoding: "utf8" }, + ); + + expect(runCase("gatewayHealthJson").status).toBe(0); + for (const caseId of ["gatewayHealthJsonConnected", "gatewayHealthJsonFirstDevice"]) { + const result = runCase(caseId); + expect(result.status).toBe(1); + expect(result.stderr).toContain(`${caseId} sample 1: exited with code 1`); + } + } finally { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } + }); + it("does not require unrelated fixture cases for a narrowed preset", () => { const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-bench-budget-test-")); try { diff --git a/test/scripts/openclaw-performance-workflow.test.ts b/test/scripts/openclaw-performance-workflow.test.ts index 0f1129dde2f9..522df47e675e 100644 --- a/test/scripts/openclaw-performance-workflow.test.ts +++ b/test/scripts/openclaw-performance-workflow.test.ts @@ -254,6 +254,13 @@ describe("OpenClaw performance workflow", () => { expect(run.indexOf(probeCap)).toBeLessThan(run.indexOf(boundedProbe)); }); + it("measures warmed and first-device gateway health separately", () => { + const run = findStep("Run OpenClaw source performance probes", "source_performance").run ?? ""; + + expect(run).toContain("--case gatewayHealthJsonConnected \\"); + expect(run).toContain("--case gatewayHealthJsonFirstDevice \\"); + }); + it("isolates required publication in a fresh artifact-consuming job", () => { const workflow = readWorkflow(); const publisher = workflow.jobs?.publish; diff --git a/test/vitest/vitest.shared.config.ts b/test/vitest/vitest.shared.config.ts index 5da7b85e5c85..33d2fbe5e696 100644 --- a/test/vitest/vitest.shared.config.ts +++ b/test/vitest/vitest.shared.config.ts @@ -5,6 +5,7 @@ import { fileURLToPath } from "node:url"; import acpCorePackageJson from "../../packages/acp-core/package.json" with { type: "json" }; import { pluginSdkSubpaths } from "../../scripts/lib/plugin-sdk-entries.mjs"; import privateLocalOnlyPluginSdkSubpaths from "../../scripts/lib/plugin-sdk-private-local-only-subpaths.json" with { type: "json" }; +import { createStateSchemaInlinePlugin } from "../../scripts/lib/state-schema-inline-plugin.mjs"; import { detectVitestHostInfo as detectVitestHostInfoImpl, isCiLikeEnv, @@ -158,6 +159,7 @@ if (!isCI && localScheduling.throttledBySystem && shouldPrintVitestThrottle(proc export const sharedVitestConfig = { root: repoRoot, envDir: false as const, + plugins: [createStateSchemaInlinePlugin(repoRoot)], resolve: { alias: [ { diff --git a/tsdown.config.ts b/tsdown.config.ts index 1ca9275f7644..2dcafbe5956d 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -11,6 +11,10 @@ import { pluginSdkEntrypoints, productionPluginSdkEntrypoints, } from "./scripts/lib/plugin-sdk-entries.mjs"; +import { + createStateSchemaInlinePlugin, + STATE_SCHEMA_INLINE_PLUGIN_NAME, +} from "./scripts/lib/state-schema-inline-plugin.mjs"; import { TSDOWN_PACKAGE_CONFIG_GROUP, TSDOWN_UNIFIED_CONFIG_GROUP, @@ -46,43 +50,7 @@ const env = { const OUTPUT_SOURCE_MAPS = process.env.OUTPUT_SOURCE_MAPS === "1"; const RUN_NODE_SKIP_DTS_BUILD = process.env.OPENCLAW_RUN_NODE_SKIP_DTS_BUILD === "1"; const TSDOWN_DECLARATIONS = !RUN_NODE_SKIP_DTS_BUILD; -export const STATE_SCHEMA_INLINE_PLUGIN_NAME = "openclaw:inline-state-schemas"; - -const STATE_SCHEMA_MODULES = [ - { - modulePath: "src/state/openclaw-state-schema.ts", - schemaPath: "src/state/openclaw-state-schema.sql", - exportName: "OPENCLAW_STATE_SCHEMA_SQL", - }, - { - modulePath: "src/state/openclaw-agent-schema.ts", - schemaPath: "src/state/openclaw-agent-schema.sql", - exportName: "OPENCLAW_AGENT_SCHEMA_SQL", - }, -] as const; - -/** Inline canonical schema bytes so packaged database opens need no SQL asset. */ -export function createStateSchemaInlinePlugin(rootDir: string = process.cwd()) { - const schemasByModulePath = new Map( - STATE_SCHEMA_MODULES.map((schema) => [path.resolve(rootDir, schema.modulePath), schema]), - ); - - return { - name: STATE_SCHEMA_INLINE_PLUGIN_NAME, - load(this: { addWatchFile(id: string): void }, id: string) { - const schema = schemasByModulePath.get(path.resolve(id)); - if (!schema) { - return null; - } - const schemaPath = path.resolve(rootDir, schema.schemaPath); - this.addWatchFile(schemaPath); - return { - code: `export const ${schema.exportName} = ${JSON.stringify(fs.readFileSync(schemaPath, "utf8"))};\n`, - moduleType: "js" as const, - }; - }, - }; -} +export { createStateSchemaInlinePlugin, STATE_SCHEMA_INLINE_PLUGIN_NAME }; const SUPPRESSED_EVAL_WARNING_PATHS = [ "@protobufjs/inquire/index.js", diff --git a/ui/src/pages/chat/realtime-talk-audio.test.ts b/ui/src/pages/chat/realtime-talk-audio.test.ts index b0db4684e9b6..bb771a9e6263 100644 --- a/ui/src/pages/chat/realtime-talk-audio.test.ts +++ b/ui/src/pages/chat/realtime-talk-audio.test.ts @@ -95,6 +95,44 @@ describe("RealtimeTalkMediaStreamMeter", () => { expect(close).toHaveBeenCalledOnce(); }); + it("reclaims its interval when the initial level callback stops it", () => { + vi.useFakeTimers(); + const close = vi.fn(async () => undefined); + const disconnectSource = vi.fn(); + const disconnectAnalyser = vi.fn(); + class MockAudioContext { + readonly close = close; + createMediaStreamSource() { + return { connect: vi.fn(), disconnect: disconnectSource }; + } + createAnalyser() { + return { + fftSize: 0, + smoothingTimeConstant: 0, + disconnect: disconnectAnalyser, + getFloatTimeDomainData: (samples: Float32Array) => samples.fill(0.25), + }; + } + } + vi.stubGlobal("AudioContext", MockAudioContext); + const onLevel = vi.fn((level: number) => { + if (level > 0) { + meter.stop(); + } + }); + const meter = new RealtimeTalkMediaStreamMeter(onLevel); + + meter.start({} as MediaStream); + meter.stop(); + meter.stop(); + vi.advanceTimersByTime(1_000); + + expect(vi.getTimerCount()).toBe(0); + expect(disconnectSource).toHaveBeenCalledOnce(); + expect(disconnectAnalyser).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + }); + it("closes an owned AudioContext when analyser setup fails", () => { const close = vi.fn(async () => undefined); class MockAudioContext { diff --git a/ui/src/pages/chat/realtime-talk-audio.ts b/ui/src/pages/chat/realtime-talk-audio.ts index 8b4511d69ecd..e4a264ee1f41 100644 --- a/ui/src/pages/chat/realtime-talk-audio.ts +++ b/ui/src/pages/chat/realtime-talk-audio.ts @@ -160,8 +160,10 @@ export class RealtimeTalkMediaStreamMeter { analyser.fftSize = this.samples.length; analyser.smoothingTimeConstant = 0; source.connect(analyser); - this.publishCurrentLevel(); this.timer = globalThis.setInterval(() => this.publishCurrentLevel(), 100); + // The initial level callback can synchronously stop its owning transport. + // Own the interval first so that reentrant cleanup cannot leave it behind. + this.publishCurrentLevel(); } catch { // Metering is feedback only; capture must still work if Web Audio analysis // is unavailable in an otherwise functional WebRTC browser. diff --git a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts index aac2ba302117..a729faa6a7e8 100644 --- a/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts +++ b/ui/src/pages/chat/realtime-talk-gateway-relay.test.ts @@ -614,6 +614,25 @@ describe("GatewayRelayRealtimeTalkTransport", () => { expect(onInputLevel).toHaveBeenLastCalledWith(0); }); + it("reclaims the input meter when its first level update stops the transport", async () => { + vi.useFakeTimers(); + const client = createClient(); + const onInputLevel = vi.fn((level: number) => { + if (level > 0) { + transport.stop(); + } + }); + const transport = createTransport({ client, callbacks: { onInputLevel } }); + + await expect(transport.start()).resolves.toBe("ready"); + transport.stop(); + transport.stop(); + vi.advanceTimersByTime(1_000); + + expect(vi.getTimerCount()).toBe(0); + expect(requestCallsFor(client, "talk.session.close")).toHaveLength(1); + }); + it("bounds stalled microphone appends and aborts every owner on stop", async () => { const onStatus = vi.fn(); const client = createClient(); diff --git a/ui/src/pages/chat/realtime-talk-google-live.ts b/ui/src/pages/chat/realtime-talk-google-live.ts index d507c45ec251..2dbc4f909723 100644 --- a/ui/src/pages/chat/realtime-talk-google-live.ts +++ b/ui/src/pages/chat/realtime-talk-google-live.ts @@ -216,11 +216,6 @@ export class GoogleLiveRealtimeTalkTransport implements RealtimeTalkTransport { const inputMeter = new RealtimeTalkMediaStreamMeter(this.ctx.callbacks.onInputLevel); this.inputMeter = inputMeter; inputMeter.start(this.media, this.inputContext); - if (this.closed || !this.lifecycle.isActive || this.inputMeter !== inputMeter) { - // start() publishes synchronously before installing its interval. A - // reentrant stop must reclaim the interval that start() installs next. - inputMeter.stop(false); - } this.assertActivationCurrent(); } this.startMicrophonePump(); diff --git a/ui/src/pages/chat/realtime-talk-webrtc.test.ts b/ui/src/pages/chat/realtime-talk-webrtc.test.ts index 6bef1ac0e370..8a662e211f2d 100644 --- a/ui/src/pages/chat/realtime-talk-webrtc.test.ts +++ b/ui/src/pages/chat/realtime-talk-webrtc.test.ts @@ -223,6 +223,42 @@ describe("WebRtcSdpRealtimeTalkTransport", () => { expect(close).toHaveBeenCalledOnce(); }); + it("reclaims the input meter when its first level update stops the transport", async () => { + vi.useFakeTimers(); + stubAnswerSdpFetch(); + const close = vi.fn(async () => undefined); + class MockAudioContext { + readonly close = close; + createMediaStreamSource() { + return { connect: vi.fn(), disconnect: vi.fn() }; + } + createAnalyser() { + return { + fftSize: 0, + smoothingTimeConstant: 0, + disconnect: vi.fn(), + getFloatTimeDomainData: (samples: Float32Array) => samples.fill(0.25), + }; + } + } + vi.stubGlobal("AudioContext", MockAudioContext); + const onInputLevel = vi.fn((level: number) => { + if (level > 0) { + transport.stop(); + } + }); + const transport = createOpenAiTransport({}, { onInputLevel }); + + await expect(transport.start()).resolves.toBe("cancelled"); + transport.stop(); + transport.stop(); + vi.advanceTimersByTime(1_000); + + expect(vi.getTimerCount()).toBe(0); + expect(stopInputTrack).toHaveBeenCalledOnce(); + expect(close).toHaveBeenCalledOnce(); + }); + it("does not continue WebRTC setup when stopped while microphone access is pending", async () => { const fetchMock = vi.fn(async () => new Response("answer-sdp")); vi.stubGlobal("fetch", fetchMock as unknown as typeof fetch);