From 85ec0082cb2a5cc5191a4662db27f4440ab5a5e4 Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Wed, 5 Aug 2026 23:12:05 -0400 Subject: [PATCH 1/9] feat(cli): add experimental voice gateway Signed-off-by: Julie Yaunches --- ci/source-architecture-budget.json | 2 +- docs/reference/commands.mdx | 78 +++- src/commands/internal/voice-gateway/serve.ts | 73 ++++ src/lib/actions/voice-gateway/serve.test.ts | 131 +++++++ src/lib/actions/voice-gateway/serve.ts | 194 ++++++++++ src/lib/adapters/http/voice-gateway-server.ts | 311 ++++++++++++++++ src/lib/voice-gateway/contracts.ts | 99 +++++ src/lib/voice-gateway/credential-file.test.ts | 98 +++++ src/lib/voice-gateway/credential-file.ts | 68 ++++ src/lib/voice-gateway/openclaw-client.test.ts | 185 ++++++++++ src/lib/voice-gateway/openclaw-client.ts | 286 ++++++++++++++ src/lib/voice-gateway/session-service.test.ts | 349 ++++++++++++++++++ src/lib/voice-gateway/session-service.ts | 346 +++++++++++++++++ .../voice-gateway/pinned-runtime-adapter.ts | 112 ++++++ test/internal-cli.test.ts | 31 ++ .../cli/oclif-metadata.test.ts | 3 + test/voice-gateway-integration.test.ts | 259 +++++++++++++ 17 files changed, 2623 insertions(+), 2 deletions(-) create mode 100644 src/commands/internal/voice-gateway/serve.ts create mode 100644 src/lib/actions/voice-gateway/serve.test.ts create mode 100644 src/lib/actions/voice-gateway/serve.ts create mode 100644 src/lib/adapters/http/voice-gateway-server.ts create mode 100644 src/lib/voice-gateway/contracts.ts create mode 100644 src/lib/voice-gateway/credential-file.test.ts create mode 100644 src/lib/voice-gateway/credential-file.ts create mode 100644 src/lib/voice-gateway/openclaw-client.test.ts create mode 100644 src/lib/voice-gateway/openclaw-client.ts create mode 100644 src/lib/voice-gateway/session-service.test.ts create mode 100644 src/lib/voice-gateway/session-service.ts create mode 100644 test/fixtures/voice-gateway/pinned-runtime-adapter.ts create mode 100644 test/voice-gateway-integration.test.ts diff --git a/ci/source-architecture-budget.json b/ci/source-architecture-budget.json index fbeabf1555b..df880bd8a2e 100644 --- a/ci/source-architecture-budget.json +++ b/ci/source-architecture-budget.json @@ -12,7 +12,7 @@ "src/lib/adapters/openshell/timeouts.ts": 37, "src/lib/agent/defs.ts": 32, "src/lib/cli/branding.ts": 87, - "src/lib/cli/nemoclaw-oclif-command.ts": 104, + "src/lib/cli/nemoclaw-oclif-command.ts": 105, "src/lib/cli/terminal-style.ts": 45, "src/lib/core/json-types.ts": 37, "src/lib/core/ports.ts": 88, diff --git a/docs/reference/commands.mdx b/docs/reference/commands.mdx index bf789ea109c..399b6dfe688 100644 --- a/docs/reference/commands.mdx +++ b/docs/reference/commands.mdx @@ -3818,7 +3818,83 @@ For contributor guidance on how these command files are structured, refer to These commands do not appear in the command-level parity check, which compares `$$nemoclaw --help` against the public command headings in this reference; hidden commands are excluded from both. The table above is the canonical reference for -the family. +the script-backed family. +The experimental adapter is documented separately because it has no owning script. + +`$$nemoclaw internal voice-gateway serve` is registered for the OpenClaw-only experimental adapter described below. +Hermes and Deep Agents Code do not have an equivalent adapter. + + + +#### $$nemoclaw internal voice-gateway serve + + + +This hidden command is an experimental implementation detail, not a supported NemoClaw product surface or public protocol. +Do not expose its listener through a port forward, proxy, or public ingress. + + + +This foreground command runs a private HTTP adapter that streams newline-delimited JSON (NDJSON) responses. +It accepts one authenticated runtime deployment and at most one active voice session. +Each voice session accepts one committed text turn. +NemoClaw selects the runtime profile, sandbox, and OpenClaw agent from command-line configuration. +The runtime cannot select an agent, OpenClaw session key, upstream URL, or forwarding destination. + +The command requires the exact feature gate `NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY=1`. +Any other value stops the command before argument parsing, credential reads, or listener creation. +Other experimental feature gates do not enable this command. + +Before you start the adapter, provision these two credential files outside the repository: + +- `--deployment-credential-file` authenticates the configured runtime deployment during session admission. +- `--openclaw-credential-file` authenticates NemoClaw to the configured OpenClaw agent gateway with `operator.read` and `operator.write` scopes. + +Both paths must be absolute paths to owner-only regular files. +The command rejects symbolic links, access by group or other users, malformed values, and oversized values. +The deployment owns each file location, lifetime, rotation, and removal. +Stopping the command does not remove either file. +The command reads both files only at startup and does not reload them. +To rotate either credential, stop the command, replace the file, and start the command again. +Removing a file does not revoke the credential in a running process. +NemoClaw does not send the OpenClaw credential to the runtime. + +Start the adapter with trusted operator-selected values: + +```bash +NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY=1 $$nemoclaw internal voice-gateway serve \ + --deployment-credential-file /absolute/private/path/deployment-bearer \ + --openclaw-credential-file /absolute/private/path/openclaw-bearer \ + --gateway-url ws://127.0.0.1:18789/ws \ + --runtime-identity \ + --runtime-profile \ + --sandbox \ + --agent \ + --listen-port 18800 +``` + +The `--gateway-url` value must use `ws://`, an explicit port, the `/ws` path, and a loopback IP address literal. +The URL must not contain credentials, a query string, or a fragment. +The adapter binds only to `127.0.0.1` and uses port `18800` when you omit `--listen-port`. +The JSON diagnostic with `"event":"voice_gateway"` and `"state":"listening"` confirms that the adapter acquired the configured loopback listener. + +The runtime must keep the raw session grant in process memory. +NemoClaw keeps only the digest needed for constant-time grant validation. +When the session closes or expires, the runtime must discard the raw grant, and NemoClaw removes its validation digest from active session state. +The process owns the OpenClaw session binding, turn state, and response correlation. +It clears that state when the session closes or expires and when the foreground process stops. +Normal onboarding and managed startup do not start or supervise this command. + +The adapter exchanges committed text and normalized response events only. +It does not implement audio, WebRTC, RTVI, voice activity detection, speech recognition, speech synthesis, playback, or runtime-specific UI behavior. +It also does not establish VoiceClaw, ElevenLabs, Hermes, or general voice support. + + + + +The experimental voice gateway has no Hermes or Deep Agents Code equivalent. + + ## Environment Variables diff --git a/src/commands/internal/voice-gateway/serve.ts b/src/commands/internal/voice-gateway/serve.ts new file mode 100644 index 00000000000..113414531e5 --- /dev/null +++ b/src/commands/internal/voice-gateway/serve.ts @@ -0,0 +1,73 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { Flags } from "@oclif/core"; + +import { + assertVoiceGatewayEnabled, + runVoiceGatewayServe, +} from "../../../lib/actions/voice-gateway/serve"; +import { DEFAULT_VOICE_GATEWAY_LISTEN_PORT } from "../../../lib/voice-gateway/contracts"; +import { NemoClawCommand } from "../../../lib/cli/nemoclaw-oclif-command"; + +export default class InternalVoiceGatewayServeCommand extends NemoClawCommand { + static hidden = true; + static strict = true; + static summary = "Internal: serve the experimental voice gateway"; + static description = + "Serve one authenticated runtime session through a private loopback HTTP and NDJSON adapter."; + static usage = [ + "internal voice-gateway serve --deployment-credential-file --openclaw-credential-file --gateway-url --runtime-identity --runtime-profile --sandbox --agent [--listen-port ]", + ]; + static flags = { + "deployment-credential-file": Flags.string({ + description: "Absolute path to the owner-only deployment bearer file", + required: true, + }), + "openclaw-credential-file": Flags.string({ + description: "Absolute path to the owner-only OpenClaw credential file", + required: true, + }), + "gateway-url": Flags.string({ + description: "Fixed loopback OpenClaw WebSocket URL", + required: true, + }), + "runtime-identity": Flags.string({ + description: "Trusted local runtime deployment identity", + required: true, + }), + "runtime-profile": Flags.string({ + description: "Operator-selected runtime profile", + required: true, + }), + sandbox: Flags.string({ + description: "Operator-selected sandbox", + required: true, + }), + agent: Flags.string({ + description: "Operator-selected OpenClaw agent", + required: true, + }), + "listen-port": Flags.integer({ + default: DEFAULT_VOICE_GATEWAY_LISTEN_PORT, + description: "Loopback port for the private runtime adapter", + min: 1024, + max: 65_535, + }), + }; + + public async run(): Promise { + assertVoiceGatewayEnabled(); + const { flags } = await this.parse(InternalVoiceGatewayServeCommand); + await runVoiceGatewayServe({ + deploymentCredentialFile: flags["deployment-credential-file"], + openClawCredentialFile: flags["openclaw-credential-file"], + gatewayUrl: flags["gateway-url"], + runtimeIdentity: flags["runtime-identity"], + runtimeProfile: flags["runtime-profile"], + sandbox: flags.sandbox, + agent: flags.agent, + listenPort: flags["listen-port"], + }); + } +} diff --git a/src/lib/actions/voice-gateway/serve.test.ts b/src/lib/actions/voice-gateway/serve.test.ts new file mode 100644 index 00000000000..d308d80b738 --- /dev/null +++ b/src/lib/actions/voice-gateway/serve.test.ts @@ -0,0 +1,131 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { EventEmitter } from "node:events"; +import type { Server } from "node:http"; + +import { describe, expect, it, vi } from "vitest"; + +import { + assertVoiceGatewayEnabled, + runVoiceGatewayServe, + validateOpenClawGatewayUrl, +} from "./serve"; + +const OPTIONS = { + deploymentCredentialFile: "/run/voice/deployment", + openClawCredentialFile: "/run/voice/openclaw", + gatewayUrl: "ws://127.0.0.1:18789/ws", + runtimeIdentity: "voiceclaw-local", + runtimeProfile: "voiceclaw-pinned", + sandbox: "demo-sandbox", + agent: "main", +}; + +describe("experimental voice gateway service gate", () => { + it.each([undefined, "", "0", "true", "01"])("rejects feature value %s", (value) => { + expect(() => + assertVoiceGatewayEnabled({ + NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY: value, + }), + ).toThrow("disabled"); + }); + + it("checks the exact feature gate before reading either credential (#8378)", async () => { + const readBearerFile = vi.fn(); + const createServer = vi.fn(); + + await expect( + runVoiceGatewayServe(OPTIONS, { + env: { + NEMOCLAW_EXPERIMENTAL_OTHER_CAPABILITY: "1", + NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY: "0", + }, + readBearerFile, + createServer, + }), + ).rejects.toThrow("disabled"); + expect(readBearerFile).not.toHaveBeenCalled(); + expect(createServer).not.toHaveBeenCalled(); + }); +}); + +describe("voice gateway destination validation", () => { + it("accepts only a fixed credential-free loopback WebSocket URL", () => { + expect(validateOpenClawGatewayUrl("ws://127.0.0.1:18789/ws")).toBe("ws://127.0.0.1:18789/ws"); + }); + + it.each([ + "wss://127.0.0.1:18789/ws", + "ws://localhost:18789/ws", + "ws://10.0.0.2:18789/ws", + "ws://user:secret@127.0.0.1:18789/ws", + "ws://127.0.0.1:18789/other", + "ws://127.0.0.1:18789/ws?target=other", + "ws://127.0.0.1/ws", + ])("rejects untrusted or ambiguous destination %s (#8378)", (value) => { + expect(() => validateOpenClawGatewayUrl(value)).toThrow("must be"); + }); +}); + +describe("voice gateway listener lifetime", () => { + it("binds loopback, logs only trusted labels, and closes on SIGTERM (#8378)", async () => { + class FakeServer extends EventEmitter { + listening = false; + listenArgs: unknown[] = []; + + listen(...args: unknown[]): this { + this.listenArgs = args.slice(0, 2); + this.listening = true; + const callback = args.at(-1); + if (typeof callback === "function") callback(); + return this; + } + + close(callback?: (error?: Error) => void): this { + this.listening = false; + callback?.(); + return this; + } + } + + const server = new FakeServer(); + const processEvents = new EventEmitter(); + const log = vi.fn(); + const readBearerFile = vi + .fn() + .mockReturnValueOnce("deployment-secret") + .mockReturnValueOnce("openclaw-secret"); + const createServer = vi.fn(() => server as unknown as Server); + + const running = runVoiceGatewayServe(OPTIONS, { + env: { NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY: "1" }, + readBearerFile, + createServer, + processEvents, + log, + }); + await vi.waitFor(() => expect(log).toHaveBeenCalledTimes(1)); + + expect(server.listenArgs).toEqual([18800, "127.0.0.1"]); + expect(createServer).toHaveBeenCalledWith({ + deploymentCredential: "deployment-secret", + service: expect.anything(), + }); + expect(log).toHaveBeenCalledWith({ + event: "voice_gateway", + state: "listening", + runtimeIdentity: "voiceclaw-local", + runtimeProfile: "voiceclaw-pinned", + sandbox: "demo-sandbox", + agent: "main", + }); + + processEvents.emit("SIGTERM"); + await running; + + expect(server.listening).toBe(false); + expect(log).toHaveBeenLastCalledWith(expect.objectContaining({ state: "stopped" })); + expect(JSON.stringify(log.mock.calls)).not.toContain("secret"); + }); +}); diff --git a/src/lib/actions/voice-gateway/serve.ts b/src/lib/actions/voice-gateway/serve.ts new file mode 100644 index 00000000000..a6cdc8c33b2 --- /dev/null +++ b/src/lib/actions/voice-gateway/serve.ts @@ -0,0 +1,194 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import type { Server } from "node:http"; + +import { createVoiceGatewayServer } from "../../adapters/http/voice-gateway-server"; +import { + DEFAULT_VOICE_GATEWAY_LISTEN_PORT, + VOICE_GATEWAY_FEATURE_ENV, + VOICE_GATEWAY_LISTEN_ADDRESS, + type VoiceGatewayDiagnostic, +} from "../../voice-gateway/contracts"; +import { readPrivateBearerFile } from "../../voice-gateway/credential-file"; +import { OpenClawVoiceClient } from "../../voice-gateway/openclaw-client"; +import { VoiceSessionService } from "../../voice-gateway/session-service"; + +const MIN_SERVICE_PORT = 1024; +const MAX_SERVICE_PORT = 65_535; + +type ServiceSignal = "SIGINT" | "SIGTERM"; + +interface ProcessEvents { + once(event: ServiceSignal, listener: () => void): unknown; + removeListener(event: ServiceSignal, listener: () => void): unknown; +} + +export interface VoiceGatewayServeOptions { + readonly deploymentCredentialFile: string; + readonly openClawCredentialFile: string; + readonly gatewayUrl: string; + readonly runtimeIdentity: string; + readonly runtimeProfile: string; + readonly sandbox: string; + readonly agent: string; + readonly listenPort?: number; +} + +export interface VoiceGatewayServeDeps { + readonly env?: NodeJS.ProcessEnv; + readonly readBearerFile?: typeof readPrivateBearerFile; + readonly createServer?: typeof createVoiceGatewayServer; + readonly processEvents?: ProcessEvents; + readonly log?: (entry: VoiceGatewayDiagnostic) => void; +} + +export function assertVoiceGatewayEnabled(env: NodeJS.ProcessEnv = process.env): void { + if (env[VOICE_GATEWAY_FEATURE_ENV] !== "1") { + throw new Error( + `Experimental voice gateway is disabled; set ${VOICE_GATEWAY_FEATURE_ENV}=1 to enable this command.`, + ); + } +} + +function validatePort(port: number): void { + if (!Number.isInteger(port) || port < MIN_SERVICE_PORT || port > MAX_SERVICE_PORT) { + throw new Error( + `Voice gateway listen port must be an integer between ${MIN_SERVICE_PORT} and ${MAX_SERVICE_PORT}.`, + ); + } +} + +export function validateOpenClawGatewayUrl(value: string): string { + let url: URL; + try { + url = new URL(value); + } catch { + throw new Error("OpenClaw gateway URL is invalid."); + } + if ( + url.protocol !== "ws:" || + (url.hostname !== "127.0.0.1" && url.hostname !== "[::1]") || + url.port === "" || + url.pathname !== "/ws" || + url.username !== "" || + url.password !== "" || + url.search !== "" || + url.hash !== "" + ) { + throw new Error( + "OpenClaw gateway URL must be a credential-free ws:// loopback IP literal with an explicit port and /ws path.", + ); + } + validatePort(Number(url.port)); + return url.href; +} + +function listen(server: Server, port: number): Promise { + return new Promise((resolve, reject) => { + const onError = (error: Error) => { + server.removeListener("error", onError); + reject(error); + }; + server.once("error", onError); + server.listen(port, VOICE_GATEWAY_LISTEN_ADDRESS, () => { + server.removeListener("error", onError); + resolve(); + }); + }); +} + +function closeServer(server: Server): Promise { + if (!server.listening) return Promise.resolve(); + return new Promise((resolve, reject) => { + server.close((error) => { + if (error) reject(error); + else resolve(); + }); + }); +} + +/** Runs the experimental loopback voice gateway until interrupted. */ +export async function runVoiceGatewayServe( + options: VoiceGatewayServeOptions, + deps: VoiceGatewayServeDeps = {}, +): Promise { + const env = deps.env ?? process.env; + assertVoiceGatewayEnabled(env); + const listenPort = options.listenPort ?? DEFAULT_VOICE_GATEWAY_LISTEN_PORT; + validatePort(listenPort); + const gatewayUrl = validateOpenClawGatewayUrl(options.gatewayUrl); + + const readBearerFile = deps.readBearerFile ?? readPrivateBearerFile; + const deploymentCredential = readBearerFile( + options.deploymentCredentialFile, + "Voice gateway deployment credential", + ); + const openClawCredential = readBearerFile( + options.openClawCredentialFile, + "Voice gateway OpenClaw credential", + ); + const log = + deps.log ?? + ((entry: VoiceGatewayDiagnostic) => { + console.log(JSON.stringify(entry)); + }); + const service = new VoiceSessionService({ + runtimeIdentity: options.runtimeIdentity, + runtimeProfile: options.runtimeProfile, + sandbox: options.sandbox, + agent: options.agent, + createClient: () => + new OpenClawVoiceClient({ + gatewayUrl, + credential: openClawCredential, + }), + diagnostic: log, + }); + const createServer = deps.createServer ?? createVoiceGatewayServer; + const server = createServer({ deploymentCredential, service }); + const processEvents = deps.processEvents ?? process; + + let resolveShutdown: () => void = () => {}; + let rejectShutdown: (error: Error) => void = () => {}; + const shutdown = new Promise((resolve, reject) => { + resolveShutdown = resolve; + rejectShutdown = reject; + }); + const onSigint = () => resolveShutdown(); + const onSigterm = () => resolveShutdown(); + const onClose = () => resolveShutdown(); + const onError = (_error: Error) => rejectShutdown(new Error("Voice gateway listener failed.")); + processEvents.once("SIGINT", onSigint); + processEvents.once("SIGTERM", onSigterm); + + try { + await listen(server, listenPort); + server.once("close", onClose); + server.once("error", onError); + log({ + event: "voice_gateway", + state: "listening", + runtimeIdentity: options.runtimeIdentity, + runtimeProfile: options.runtimeProfile, + sandbox: options.sandbox, + agent: options.agent, + }); + await shutdown; + } finally { + processEvents.removeListener("SIGINT", onSigint); + processEvents.removeListener("SIGTERM", onSigterm); + server.removeListener("close", onClose); + server.removeListener("error", onError); + service.closeAll(); + await closeServer(server); + log({ + event: "voice_gateway", + state: "stopped", + runtimeIdentity: options.runtimeIdentity, + runtimeProfile: options.runtimeProfile, + sandbox: options.sandbox, + agent: options.agent, + }); + } +} diff --git a/src/lib/adapters/http/voice-gateway-server.ts b/src/lib/adapters/http/voice-gateway-server.ts new file mode 100644 index 00000000000..803b9f45a28 --- /dev/null +++ b/src/lib/adapters/http/voice-gateway-server.ts @@ -0,0 +1,311 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { createHash, timingSafeEqual } from "node:crypto"; +import http from "node:http"; + +import { + VOICE_GATEWAY_MAX_REQUEST_BYTES, + VoiceGatewayRequestError, + type VoiceResponseEvent, +} from "../../voice-gateway/contracts"; +import type { VoiceSessionService } from "../../voice-gateway/session-service"; + +const MAX_HEADER_BYTES = 16 * 1024; +const BODY_TIMEOUT_MS = 10_000; +const SESSION_ROUTE = "/v1/voice/sessions"; +const TURN_ROUTE_PATTERN = /^\/v1\/voice\/sessions\/([^/]+)\/turns$/u; +const CLOSE_ROUTE_PATTERN = /^\/v1\/voice\/sessions\/([^/]+)$/u; + +class BodyError extends Error { + constructor(readonly status: number) { + super("invalid_request"); + } +} + +function parseBearer(header: string | string[] | undefined): string { + if (typeof header !== "string") return ""; + return /^Bearer ([\x21-\x7e]+)$/u.exec(header)?.[1] ?? ""; +} + +function bearerMatches(header: string | string[] | undefined, expectedHash: Buffer): boolean { + const actualHash = createHash("sha256").update(parseBearer(header)).digest(); + return timingSafeEqual(actualHash, expectedHash); +} + +function writeJson(response: http.ServerResponse, status: number, value: object): void { + if (response.destroyed) return; + const body = Buffer.from(JSON.stringify(value)); + response.writeHead(status, { + "cache-control": "no-store", + "content-length": String(body.length), + "content-type": "application/json", + ...(status >= 400 ? { connection: "close" } : {}), + }); + response.end(body); +} + +function writeEmpty(response: http.ServerResponse, status: number): void { + if (response.destroyed) return; + response.writeHead(status, { + "cache-control": "no-store", + "content-length": "0", + ...(status >= 400 ? { connection: "close" } : {}), + }); + response.end(); +} + +function errorStatus(code: VoiceGatewayRequestError["code"]): number { + if (code === "authentication_failed") return 401; + if (code === "session_not_found" || code === "session_expired") return 404; + if ( + code === "duplicate_turn" || + code === "session_in_progress" || + code === "turn_in_progress" || + code === "turn_limit_reached" + ) { + return 409; + } + return 400; +} + +function sendRequestError(response: http.ServerResponse, error: unknown): void { + if (response.headersSent) { + response.destroy(); + return; + } + if (error instanceof VoiceGatewayRequestError) { + writeJson(response, errorStatus(error.code), { error: error.code }); + return; + } + if (error instanceof BodyError) { + writeJson(response, error.status, { error: "invalid_request" }); + return; + } + writeJson(response, 500, { error: "internal_error" }); +} + +function readJsonBody(request: http.IncomingMessage): Promise> { + return new Promise((resolve, reject) => { + if (request.headers["content-type"] !== "application/json") { + reject(new BodyError(415)); + return; + } + const rawLength = request.headers["content-length"]; + if (Array.isArray(rawLength)) { + reject(new BodyError(400)); + return; + } + const declaredLength = rawLength === undefined ? null : Number(rawLength); + if (declaredLength !== null && (!Number.isSafeInteger(declaredLength) || declaredLength < 0)) { + reject(new BodyError(400)); + return; + } + if (declaredLength !== null && declaredLength > VOICE_GATEWAY_MAX_REQUEST_BYTES) { + reject(new BodyError(413)); + return; + } + + const chunks: Buffer[] = []; + let size = 0; + let settled = false; + const finish = (error?: Error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + if (error) { + reject(error); + return; + } + try { + const parsed = JSON.parse(Buffer.concat(chunks).toString("utf8")); + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) { + reject(new BodyError(400)); + } else { + resolve(parsed as Record); + } + } catch { + reject(new BodyError(400)); + } + }; + const timer = setTimeout(() => finish(new BodyError(408)), BODY_TIMEOUT_MS); + request.on("data", (chunk: Buffer | string) => { + if (settled) return; + const value = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + size += value.length; + if (size > VOICE_GATEWAY_MAX_REQUEST_BYTES) { + finish(new BodyError(413)); + return; + } + chunks.push(value); + }); + request.once("aborted", () => finish(new BodyError(400))); + request.once("error", () => finish(new BodyError(400))); + request.once("end", () => finish()); + }); +} + +function exactBody( + body: Record, + fields: readonly string[], +): body is Record { + const keys = Object.keys(body).sort(); + return ( + keys.length === fields.length && + keys.every((key, index) => key === [...fields].sort()[index]) && + fields.every((field) => typeof body[field] === "string") + ); +} + +function decodeRouteValue(value: string): string | null { + try { + return decodeURIComponent(value); + } catch { + return null; + } +} + +async function handleCreateSession( + request: http.IncomingMessage, + response: http.ServerResponse, + service: VoiceSessionService, + deploymentCredentialHash: Buffer, +): Promise { + if (!bearerMatches(request.headers.authorization, deploymentCredentialHash)) { + writeJson(response, 401, { error: "authentication_failed" }); + return; + } + const body = await readJsonBody(request); + if (!exactBody(body, ["runtimeConversationId"])) { + throw new VoiceGatewayRequestError("invalid_request"); + } + writeJson(response, 201, service.createSession(body.runtimeConversationId)); +} + +async function handleTurn( + request: http.IncomingMessage, + response: http.ServerResponse, + service: VoiceSessionService, + voiceSessionId: string, +): Promise { + const grant = parseBearer(request.headers.authorization); + service.authorizeSession(voiceSessionId, grant); + const body = await readJsonBody(request); + if (!exactBody(body, ["commitId", "text"])) { + throw new VoiceGatewayRequestError("invalid_request"); + } + let deliveryOpen = true; + let streamStarted = false; + response.once("close", () => { + deliveryOpen = false; + }); + const deliver = (event: VoiceResponseEvent) => { + if (!deliveryOpen) return; + if (!streamStarted) { + streamStarted = true; + response.writeHead(200, { + "cache-control": "no-store", + "content-type": "application/x-ndjson", + "x-content-type-options": "nosniff", + }); + } + response.write(`${JSON.stringify(event)}\n`); + }; + await service.commitTurn({ + voiceSessionId, + grant, + commitId: body.commitId, + text: body.text, + deliver, + deliveryOpen: () => deliveryOpen, + }); + if (deliveryOpen) { + if (streamStarted) response.end(); + else writeJson(response, 502, { error: "agent_gateway_unavailable" }); + } +} + +async function handleRequest(options: { + readonly request: http.IncomingMessage; + readonly response: http.ServerResponse; + readonly service: VoiceSessionService; + readonly deploymentCredentialHash: Buffer; +}): Promise { + const { request, response } = options; + if (!request.url?.startsWith("/") || request.url.startsWith("//")) { + writeEmpty(response, 404); + return; + } + let url: URL; + try { + url = new URL(request.url, "http://127.0.0.1"); + } catch { + writeEmpty(response, 400); + return; + } + if (url.search !== "" || request.url !== url.pathname) { + writeEmpty(response, 404); + return; + } + if (request.method === "GET" && url.pathname === "/healthz") { + if (!bearerMatches(request.headers.authorization, options.deploymentCredentialHash)) { + writeEmpty(response, 401); + return; + } + writeEmpty(response, 204); + return; + } + if (request.method === "POST" && url.pathname === SESSION_ROUTE) { + await handleCreateSession(request, response, options.service, options.deploymentCredentialHash); + return; + } + const turn = request.method === "POST" ? TURN_ROUTE_PATTERN.exec(url.pathname) : null; + if (turn) { + const voiceSessionId = decodeRouteValue(turn[1]); + if (voiceSessionId === null) { + writeEmpty(response, 400); + return; + } + await handleTurn(request, response, options.service, voiceSessionId); + return; + } + const close = request.method === "DELETE" ? CLOSE_ROUTE_PATTERN.exec(url.pathname) : null; + if (close) { + const voiceSessionId = decodeRouteValue(close[1]); + if (voiceSessionId === null) { + writeEmpty(response, 400); + return; + } + options.service.closeSession(voiceSessionId, parseBearer(request.headers.authorization)); + writeEmpty(response, 204); + return; + } + writeEmpty(response, 404); +} + +export function createVoiceGatewayServer(options: { + readonly deploymentCredential: string; + readonly service: VoiceSessionService; +}): http.Server { + const deploymentCredentialHash = createHash("sha256") + .update(options.deploymentCredential) + .digest(); + const server = http.createServer({ maxHeaderSize: MAX_HEADER_BYTES }, (request, response) => { + void handleRequest({ + request, + response, + service: options.service, + deploymentCredentialHash, + }).catch((error) => sendRequestError(response, error)); + }); + server.headersTimeout = 5_000; + server.requestTimeout = 15_000; + server.keepAliveTimeout = 5_000; + server.on("connect", (_request, socket) => socket.destroy()); + server.on("upgrade", (_request, socket) => socket.destroy()); + server.on("clientError", (_error, socket) => { + if (!socket.writable) return; + socket.end("HTTP/1.1 400 Bad Request\r\nConnection: close\r\nContent-Length: 0\r\n\r\n"); + }); + return server; +} diff --git a/src/lib/voice-gateway/contracts.ts b/src/lib/voice-gateway/contracts.ts new file mode 100644 index 00000000000..9afe5f35d4a --- /dev/null +++ b/src/lib/voice-gateway/contracts.ts @@ -0,0 +1,99 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +export const VOICE_GATEWAY_FEATURE_ENV = "NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY"; +export const VOICE_GATEWAY_LISTEN_ADDRESS = "127.0.0.1"; +export const DEFAULT_VOICE_GATEWAY_LISTEN_PORT = 18_800; +export const VOICE_GATEWAY_SESSION_LIFETIME_MS = 5 * 60_000; +export const VOICE_GATEWAY_TURN_TIMEOUT_MS = 2 * 60_000; +export const VOICE_GATEWAY_MAX_REQUEST_BYTES = 64 * 1024; +export const VOICE_GATEWAY_MAX_RESPONSE_BYTES = 2 * 1024 * 1024; + +export type VoiceGatewayFailureReason = + | "agent_failed" + | "agent_gateway_unavailable" + | "agent_protocol_error" + | "response_too_large" + | "session_closed" + | "session_expired" + | "turn_timeout"; + +export type VoiceResponseEvent = + | { + readonly type: "response.started"; + readonly voiceSessionId: string; + readonly turnId: string; + readonly responseId: string; + } + | { + readonly type: "response.text.delta"; + readonly voiceSessionId: string; + readonly turnId: string; + readonly responseId: string; + readonly sequence: number; + readonly text: string; + } + | { + readonly type: "response.completed"; + readonly voiceSessionId: string; + readonly turnId: string; + readonly responseId: string; + } + | { + readonly type: "response.failed"; + readonly voiceSessionId: string; + readonly turnId: string; + readonly responseId: string; + readonly reason: VoiceGatewayFailureReason; + }; + +export type AgentTurnEvent = + | { readonly type: "started" } + | { readonly type: "text"; readonly text: string }; + +export interface AgentTurnClient { + runTurn(options: { + readonly idempotencyKey: string; + readonly message: string; + readonly onEvent: (event: AgentTurnEvent) => void; + readonly sessionKey: string; + }): Promise< + | { readonly outcome: "completed" } + | { + readonly outcome: "failed"; + readonly reason: "agent_failed" | "agent_gateway_unavailable" | "agent_protocol_error"; + } + >; + close(): void; +} + +export interface VoiceGatewayDiagnostic { + readonly event: string; + readonly state: string; + readonly runtimeIdentity?: string; + readonly runtimeProfile?: string; + readonly sandbox?: string; + readonly agent?: string; + readonly voiceSessionId?: string; + readonly runtimeConversationId?: string; + readonly turnId?: string; + readonly responseId?: string; + readonly reason?: VoiceGatewayFailureReason | string; + readonly durationMs?: number; +} + +export class VoiceGatewayRequestError extends Error { + constructor( + readonly code: + | "authentication_failed" + | "duplicate_turn" + | "invalid_request" + | "session_expired" + | "session_in_progress" + | "session_not_found" + | "turn_in_progress" + | "turn_limit_reached", + ) { + super(code); + } +} diff --git a/src/lib/voice-gateway/credential-file.test.ts b/src/lib/voice-gateway/credential-file.test.ts new file mode 100644 index 00000000000..ac016770add --- /dev/null +++ b/src/lib/voice-gateway/credential-file.test.ts @@ -0,0 +1,98 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { readPrivateBearerFile } from "./credential-file"; + +const CREDENTIAL = "voice-gateway-test-credential-0123456789"; +const directories: string[] = []; + +function temporaryDirectory(): string { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-voice-credential-")); + directories.push(directory); + return directory; +} + +afterEach(() => { + for (const directory of directories.splice(0)) { + fs.rmSync(directory, { force: true, recursive: true }); + } +}); + +describe("voice gateway credential file", () => { + it("reads one owner-only regular file and removes one trailing newline (#8378)", () => { + const file = path.join(temporaryDirectory(), "credential"); + fs.writeFileSync(file, `${CREDENTIAL}\n`, { mode: 0o600 }); + + expect(readPrivateBearerFile(file, "Test credential")).toBe(CREDENTIAL); + }); + + it.each([ + { + name: "relative path", + arrange: () => "credential", + message: "must be absolute", + }, + { + name: "group-readable file", + arrange: () => { + const file = path.join(temporaryDirectory(), "credential"); + fs.writeFileSync(file, CREDENTIAL, { mode: 0o640 }); + fs.chmodSync(file, 0o640); + return file; + }, + message: "must not be accessible by group or others", + }, + { + name: "directory", + arrange: () => temporaryDirectory(), + message: "not a regular file", + }, + { + name: "symbolic link", + arrange: () => { + const directory = temporaryDirectory(); + const target = path.join(directory, "target"); + const link = path.join(directory, "credential"); + fs.writeFileSync(target, CREDENTIAL, { mode: 0o600 }); + fs.symlinkSync(target, link); + return link; + }, + message: "symbolic-link", + }, + { + name: "short value", + arrange: () => { + const file = path.join(temporaryDirectory(), "credential"); + fs.writeFileSync(file, "short", { mode: 0o600 }); + return file; + }, + message: "invalid size", + }, + { + name: "value with whitespace", + arrange: () => { + const file = path.join(temporaryDirectory(), "credential"); + fs.writeFileSync(file, `${CREDENTIAL} extra`, { mode: 0o600 }); + return file; + }, + message: "malformed", + }, + { + name: "oversized value", + arrange: () => { + const file = path.join(temporaryDirectory(), "credential"); + fs.writeFileSync(file, "a".repeat(4098), { mode: 0o600 }); + return file; + }, + message: "invalid size", + }, + ])("rejects a $name before returning credential material (#8378)", ({ arrange, message }) => { + expect(() => readPrivateBearerFile(arrange(), "Test credential")).toThrow(message); + }); +}); diff --git a/src/lib/voice-gateway/credential-file.ts b/src/lib/voice-gateway/credential-file.ts new file mode 100644 index 00000000000..b5d7a566100 --- /dev/null +++ b/src/lib/voice-gateway/credential-file.ts @@ -0,0 +1,68 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import fs from "node:fs"; +import path from "node:path"; + +const MAX_CREDENTIAL_BYTES = 4096; +const MIN_CREDENTIAL_BYTES = 32; + +function validatePrivateRegularFile(stat: fs.Stats, filePath: string, label: string): void { + if (!stat.isFile()) throw new Error(`${label} path is not a regular file: ${filePath}`); + if ((stat.mode & 0o077) !== 0) { + throw new Error(`${label} file must not be accessible by group or others: ${filePath}`); + } + if (typeof process.getuid === "function" && stat.uid !== process.getuid()) { + throw new Error(`${label} file is not owned by the current user: ${filePath}`); + } + if (stat.size < MIN_CREDENTIAL_BYTES || stat.size > MAX_CREDENTIAL_BYTES + 1) { + throw new Error(`${label} file has an invalid size: ${filePath}`); + } +} + +function validateBearer(value: string, label: string): void { + const bytes = Buffer.byteLength(value); + if ( + bytes < MIN_CREDENTIAL_BYTES || + bytes > MAX_CREDENTIAL_BYTES || + !/^[\x21-\x7e]+$/u.test(value) + ) { + throw new Error(`${label} is malformed.`); + } +} + +/** Read one owner-only bearer without following the final path component. */ +export function readPrivateBearerFile(filePath: string, label: string): string { + if (!path.isAbsolute(filePath)) throw new Error(`${label} file path must be absolute.`); + if (typeof fs.constants.O_NOFOLLOW !== "number") { + throw new Error("Secure no-follow file opens are unavailable on this platform."); + } + + let descriptor: number | undefined; + try { + try { + descriptor = fs.openSync( + filePath, + fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | (fs.constants.O_NONBLOCK ?? 0), + ); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ELOOP") { + throw new Error(`Refusing to read a symbolic-link ${label} file: ${filePath}`); + } + throw error; + } + + validatePrivateRegularFile(fs.fstatSync(descriptor), filePath, label); + const buffer = Buffer.alloc(MAX_CREDENTIAL_BYTES + 2); + const bytesRead = fs.readSync(descriptor, buffer, 0, buffer.length, 0); + if (bytesRead > MAX_CREDENTIAL_BYTES + 1) { + throw new Error(`${label} file has an invalid size: ${filePath}`); + } + const contents = buffer.subarray(0, bytesRead).toString("utf8"); + const value = contents.endsWith("\n") ? contents.slice(0, -1) : contents; + validateBearer(value, label); + return value; + } finally { + if (descriptor !== undefined) fs.closeSync(descriptor); + } +} diff --git a/src/lib/voice-gateway/openclaw-client.test.ts b/src/lib/voice-gateway/openclaw-client.test.ts new file mode 100644 index 00000000000..1482dc7d6f4 --- /dev/null +++ b/src/lib/voice-gateway/openclaw-client.test.ts @@ -0,0 +1,185 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { describe, expect, it } from "vitest"; + +import type { AgentTurnEvent } from "./contracts"; +import { OpenClawVoiceClient } from "./openclaw-client"; + +interface SentRequest { + readonly type: string; + readonly id: string; + readonly method: string; + readonly params: Record; +} + +class FakeWebSocket { + onopen: (() => void) | null = null; + onmessage: ((event: { readonly data: unknown }) => void) | null = null; + onerror: (() => void) | null = null; + onclose: (() => void) | null = null; + readonly sent: SentRequest[] = []; + closed = false; + readonly firstReply: string; + readonly finalReply: string; + readonly oversizedFrameAfterConnect: boolean; + + constructor(firstReply = "hel", finalReply = "hello", oversizedFrameAfterConnect = false) { + this.firstReply = firstReply; + this.finalReply = finalReply; + this.oversizedFrameAfterConnect = oversizedFrameAfterConnect; + queueMicrotask(() => this.onopen?.()); + } + + send(data: string): void { + const request = JSON.parse(data) as SentRequest; + this.sent.push(request); + if (request.method === "connect") { + queueMicrotask(() => { + this.respond(request.id, {}); + if (this.oversizedFrameAfterConnect) { + this.onmessage?.({ data: `{"padding":"${"x".repeat(3 * 1024 * 1024)}"}` }); + } + }); + return; + } + if (request.method === "chat.send") { + const sessionKey = String(request.params.sessionKey); + queueMicrotask(() => { + this.respond(request.id, { runId: "expected-run" }); + queueMicrotask(() => { + this.event(sessionKey, "other-run", "delta", "discarded run"); + this.event("other-session", "expected-run", "delta", "discarded session"); + this.event(sessionKey, "expected-run", "delta", this.firstReply); + this.event(sessionKey, "expected-run", "final", this.finalReply); + }); + }); + } + } + + close(): void { + this.closed = true; + } + + private respond(id: string, payload: Record): void { + this.onmessage?.({ + data: JSON.stringify({ type: "res", id, ok: true, payload }), + }); + } + + private event(sessionKey: string, runId: string, state: string, text: string): void { + this.onmessage?.({ + data: JSON.stringify({ + type: "event", + event: "chat", + payload: { + sessionKey, + runId, + state, + message: { content: [{ type: "text", text }] }, + nativeSecret: "must-not-cross", + }, + }), + }); + } +} + +describe("OpenClaw voice gateway client", () => { + it("uses bounded operator scopes and emits only ordered normalized text for the expected session and run (#8378)", async () => { + const socket = new FakeWebSocket(); + const events: AgentTurnEvent[] = []; + const client = new OpenClawVoiceClient({ + gatewayUrl: "ws://127.0.0.1:18789/ws", + credential: "openclaw-credential-must-not-cross", + webSocketFactory: () => socket, + }); + + const result = await client.runTurn({ + sessionKey: "agent:main:nemoclaw-voice:session", + idempotencyKey: "generated-turn-id", + message: "repository status", + onEvent: (event) => events.push(event), + }); + + expect(result).toEqual({ outcome: "completed" }); + expect(events).toEqual([ + { type: "started" }, + { type: "text", text: "hel" }, + { type: "text", text: "lo" }, + ]); + const connect = socket.sent.find((request) => request.method === "connect"); + expect(connect?.params).toMatchObject({ + client: { id: "openclaw-cli", mode: "cli" }, + scopes: ["operator.read", "operator.write"], + auth: { token: "openclaw-credential-must-not-cross" }, + }); + const send = socket.sent.find((request) => request.method === "chat.send"); + expect(send?.params).toMatchObject({ + sessionKey: "agent:main:nemoclaw-voice:session", + message: "repository status", + idempotencyKey: "generated-turn-id", + deliver: false, + }); + expect(JSON.stringify(events)).not.toContain("openclaw-credential"); + expect(JSON.stringify(events)).not.toContain("expected-run"); + expect(JSON.stringify(events)).not.toContain("must-not-cross"); + }); + + it("fails closed when ordered response text changes its prior prefix (#8378)", async () => { + const socket = new FakeWebSocket("first", "different"); + const client = new OpenClawVoiceClient({ + gatewayUrl: "ws://127.0.0.1:18789/ws", + credential: "openclaw-credential-must-not-cross", + webSocketFactory: () => socket, + }); + + await expect( + client.runTurn({ + sessionKey: "agent:main:nemoclaw-voice:session", + idempotencyKey: "generated-turn-id", + message: "repository status", + onEvent: () => {}, + }), + ).resolves.toEqual({ outcome: "failed", reason: "agent_protocol_error" }); + }); + + it("closes the direct WebSocket connection when the session owner revokes it (#8378)", () => { + const socket = new FakeWebSocket(); + const client = new OpenClawVoiceClient({ + gatewayUrl: "ws://127.0.0.1:18789/ws", + credential: "openclaw-credential-must-not-cross", + webSocketFactory: () => socket, + }); + void client.runTurn({ + sessionKey: "agent:main:nemoclaw-voice:session", + idempotencyKey: "generated-turn-id", + message: "repository status", + onEvent: () => {}, + }); + + client.close(); + + expect(socket.closed).toBe(true); + }); + + it("rejects an oversized native frame before sending agent work (#8378)", async () => { + const socket = new FakeWebSocket("hel", "hello", true); + const client = new OpenClawVoiceClient({ + gatewayUrl: "ws://127.0.0.1:18789/ws", + credential: "openclaw-credential-must-not-cross", + webSocketFactory: () => socket, + }); + + await expect( + client.runTurn({ + sessionKey: "agent:main:nemoclaw-voice:session", + idempotencyKey: "generated-turn-id", + message: "must-not-send", + onEvent: () => {}, + }), + ).resolves.toEqual({ outcome: "failed", reason: "agent_protocol_error" }); + + expect(socket.sent.some((request) => request.method === "chat.send")).toBe(false); + expect(socket.closed).toBe(true); + }); +}); diff --git a/src/lib/voice-gateway/openclaw-client.ts b/src/lib/voice-gateway/openclaw-client.ts new file mode 100644 index 00000000000..6c0105b1d85 --- /dev/null +++ b/src/lib/voice-gateway/openclaw-client.ts @@ -0,0 +1,286 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { randomUUID } from "node:crypto"; + +import { + type AgentTurnClient, + type AgentTurnEvent, + VOICE_GATEWAY_MAX_RESPONSE_BYTES, +} from "./contracts"; + +const OPENCLAW_PROTOCOL_VERSION = 4; +const REQUEST_TIMEOUT_MS = 10_000; +const MAX_NATIVE_FRAME_BYTES = VOICE_GATEWAY_MAX_RESPONSE_BYTES + 64 * 1024; +const MAX_QUEUED_CHAT_EVENTS = 128; + +interface WebSocketLike { + onopen: (() => void) | null; + onmessage: ((event: { readonly data: unknown }) => void) | null; + onerror: (() => void) | null; + onclose: (() => void) | null; + send(data: string): void; + close(): void; +} + +type WebSocketFactory = (url: string) => WebSocketLike; + +interface PendingRequest { + readonly resolve: (value: Record) => void; + readonly reject: () => void; + readonly timer: NodeJS.Timeout; +} + +interface NativeChatEvent { + readonly sessionKey: string; + readonly runId: string; + readonly state: string; + readonly text: string; +} + +function record(value: unknown): Record | null { + return value !== null && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : null; +} + +function nonemptyString(value: unknown): string | null { + return typeof value === "string" && value.length > 0 ? value : null; +} + +function textFromMessage(message: unknown): string { + const value = record(message); + if (!value) return ""; + if (typeof value.text === "string") return value.text; + if (typeof value.content === "string") return value.content; + if (!Array.isArray(value.content)) return ""; + return value.content + .map((part) => { + const item = record(part); + return typeof item?.text === "string" ? item.text : ""; + }) + .filter(Boolean) + .join("\n"); +} + +function parseNativeChatEvent(frame: Record): NativeChatEvent | null { + if (frame.type !== "event" || frame.event !== "chat") return null; + const payload = record(frame.payload); + if (!payload) return null; + const sessionKey = nonemptyString(payload.sessionKey); + const runId = nonemptyString(payload.runId); + const state = nonemptyString(payload.state); + if (!sessionKey || !runId || !state) return null; + return { sessionKey, runId, state, text: textFromMessage(payload.message) }; +} + +function defaultWebSocketFactory(url: string): WebSocketLike { + if (typeof globalThis.WebSocket !== "function") { + throw new Error("WebSocket support is unavailable."); + } + return new globalThis.WebSocket(url) as unknown as WebSocketLike; +} + +export interface OpenClawVoiceClientOptions { + readonly gatewayUrl: string; + readonly credential: string; + readonly webSocketFactory?: WebSocketFactory; +} + +type TurnResult = Awaited>; + +/** Confines OpenClaw authentication, native frames, and run IDs to one client. */ +export class OpenClawVoiceClient implements AgentTurnClient { + private readonly gatewayUrl: string; + private readonly credential: string; + private readonly webSocketFactory: WebSocketFactory; + private socket: WebSocketLike | null = null; + private closed = false; + private cancelCurrent: (() => void) | null = null; + + constructor(options: OpenClawVoiceClientOptions) { + this.gatewayUrl = options.gatewayUrl; + this.credential = options.credential; + this.webSocketFactory = options.webSocketFactory ?? defaultWebSocketFactory; + } + + close(): void { + this.closed = true; + this.cancelCurrent?.(); + this.socket?.close(); + this.socket = null; + } + + async runTurn(options: { + readonly idempotencyKey: string; + readonly message: string; + readonly onEvent: (event: AgentTurnEvent) => void; + readonly sessionKey: string; + }): Promise { + if (this.closed) return { outcome: "failed", reason: "agent_gateway_unavailable" }; + + let socket: WebSocketLike; + try { + socket = this.webSocketFactory(this.gatewayUrl); + this.socket = socket; + } catch { + return { outcome: "failed", reason: "agent_gateway_unavailable" }; + } + + const pending = new Map(); + const queuedChatEvents: NativeChatEvent[] = []; + let requestCounter = 0; + let activeRunId: string | null = null; + let previousText = ""; + let terminal = false; + + const clearPending = () => { + for (const entry of pending.values()) { + clearTimeout(entry.timer); + entry.reject(); + } + pending.clear(); + }; + const request = (method: string, params: Record) => { + const id = `voice-${++requestCounter}`; + return new Promise>((resolve, reject) => { + const timer = setTimeout(() => { + pending.delete(id); + reject(new Error("request timeout")); + }, REQUEST_TIMEOUT_MS); + pending.set(id, { resolve, reject: () => reject(new Error("request failed")), timer }); + socket.send(JSON.stringify({ type: "req", id, method, params })); + }); + }; + + let resolveTerminal: (value: TurnResult) => void = () => {}; + const terminalPromise = new Promise((resolve) => { + resolveTerminal = resolve; + }); + const finish = (value: TurnResult) => { + if (terminal) return; + terminal = true; + this.cancelCurrent = null; + clearPending(); + resolveTerminal(value); + }; + this.cancelCurrent = () => finish({ outcome: "failed", reason: "agent_gateway_unavailable" }); + const handleChat = (event: NativeChatEvent) => { + if ( + terminal || + activeRunId === null || + event.sessionKey !== options.sessionKey || + event.runId !== activeRunId + ) { + return; + } + if (event.text.length > 0) { + if (!event.text.startsWith(previousText)) { + finish({ outcome: "failed", reason: "agent_protocol_error" }); + return; + } + const delta = event.text.slice(previousText.length); + previousText = event.text; + if (delta.length > 0) options.onEvent({ type: "text", text: delta }); + } + if (event.state === "final") finish({ outcome: "completed" }); + else if (event.state === "error" || event.state === "aborted") { + finish({ outcome: "failed", reason: "agent_failed" }); + } + }; + + socket.onmessage = (event) => { + let frame: Record | null = null; + try { + const raw = String(event.data); + if (Buffer.byteLength(raw) > MAX_NATIVE_FRAME_BYTES) { + finish({ outcome: "failed", reason: "agent_protocol_error" }); + socket.close(); + return; + } + frame = record(JSON.parse(raw)); + } catch { + finish({ outcome: "failed", reason: "agent_protocol_error" }); + return; + } + if (!frame) return; + if (frame.type === "res" && typeof frame.id === "string") { + const entry = pending.get(frame.id); + if (!entry) return; + pending.delete(frame.id); + clearTimeout(entry.timer); + if (frame.ok === false || frame.error !== undefined) entry.reject(); + else entry.resolve(record(frame.payload) ?? record(frame.result) ?? frame); + return; + } + const chat = parseNativeChatEvent(frame); + if (!chat) return; + if (activeRunId === null) { + if (queuedChatEvents.length >= MAX_QUEUED_CHAT_EVENTS) { + finish({ outcome: "failed", reason: "agent_protocol_error" }); + socket.close(); + return; + } + queuedChatEvents.push(chat); + } else handleChat(chat); + }; + let rejectOpen: (() => void) | null = null; + socket.onerror = () => { + rejectOpen?.(); + finish({ outcome: "failed", reason: "agent_gateway_unavailable" }); + }; + socket.onclose = () => { + rejectOpen?.(); + finish({ outcome: "failed", reason: "agent_gateway_unavailable" }); + }; + + try { + await new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("open timeout")), REQUEST_TIMEOUT_MS); + rejectOpen = () => { + clearTimeout(timer); + reject(new Error("open failed")); + }; + socket.onopen = () => { + clearTimeout(timer); + rejectOpen = null; + resolve(); + }; + }); + await request("connect", { + minProtocol: OPENCLAW_PROTOCOL_VERSION, + maxProtocol: OPENCLAW_PROTOCOL_VERSION, + client: { + id: "openclaw-cli", + displayName: "NemoClaw voice gateway", + version: "1", + platform: process.platform, + mode: "cli", + instanceId: randomUUID(), + }, + caps: [], + scopes: ["operator.read", "operator.write"], + auth: { token: this.credential }, + }); + if (terminal) return terminalPromise; + const sent = await request("chat.send", { + sessionKey: options.sessionKey, + message: options.message, + deliver: false, + timeoutMs: 90_000, + idempotencyKey: options.idempotencyKey, + }); + activeRunId = nonemptyString(sent.runId); + if (!activeRunId) { + finish({ outcome: "failed", reason: "agent_protocol_error" }); + } else { + options.onEvent({ type: "started" }); + for (const event of queuedChatEvents.splice(0)) handleChat(event); + } + } catch { + finish({ outcome: "failed", reason: "agent_gateway_unavailable" }); + } + + return terminalPromise; + } +} diff --git a/src/lib/voice-gateway/session-service.test.ts b/src/lib/voice-gateway/session-service.test.ts new file mode 100644 index 00000000000..71632405eae --- /dev/null +++ b/src/lib/voice-gateway/session-service.test.ts @@ -0,0 +1,349 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { describe, expect, it, vi } from "vitest"; + +import type { AgentTurnClient, AgentTurnEvent, VoiceResponseEvent } from "./contracts"; +import { VoiceGatewayRequestError } from "./contracts"; +import { VoiceSessionService } from "./session-service"; + +class FakeAgentClient implements AgentTurnClient { + readonly calls: Array<{ + idempotencyKey: string; + message: string; + sessionKey: string; + }> = []; + closed = false; + run: (onEvent: (event: AgentTurnEvent) => void) => ReturnType = + async (onEvent) => { + onEvent({ type: "started" }); + onEvent({ type: "text", text: "hello" }); + return { outcome: "completed" }; + }; + + close(): void { + this.closed = true; + } + + runTurn(options: { + readonly idempotencyKey: string; + readonly message: string; + readonly onEvent: (event: AgentTurnEvent) => void; + readonly sessionKey: string; + }): ReturnType { + this.calls.push({ + idempotencyKey: options.idempotencyKey, + message: options.message, + sessionKey: options.sessionKey, + }); + return this.run(options.onEvent); + } +} + +function serviceFixture( + overrides: { + client?: FakeAgentClient; + now?: () => number; + randomIds?: string[]; + sessionLifetimeMs?: number; + turnTimeoutMs?: number; + maxResponseBytes?: number; + } = {}, +) { + const client = overrides.client ?? new FakeAgentClient(); + const diagnostics: unknown[] = []; + const ids = [...(overrides.randomIds ?? ["voice-session", "agent-session", "turn", "response"])]; + const service = new VoiceSessionService({ + runtimeIdentity: "voiceclaw-local", + runtimeProfile: "voiceclaw-pinned", + sandbox: "demo-sandbox", + agent: "main", + createClient: () => client, + diagnostic: (entry) => diagnostics.push(entry), + randomId: () => ids.shift() ?? "extra-id", + randomGrant: () => Buffer.alloc(32, 7), + ...(overrides.now ? { now: overrides.now } : {}), + ...(overrides.sessionLifetimeMs ? { sessionLifetimeMs: overrides.sessionLifetimeMs } : {}), + ...(overrides.turnTimeoutMs ? { turnTimeoutMs: overrides.turnTimeoutMs } : {}), + ...(overrides.maxResponseBytes ? { maxResponseBytes: overrides.maxResponseBytes } : {}), + }); + return { service, client, diagnostics }; +} + +describe("voice session and committed turn boundary", () => { + it("binds trusted configuration and generates internal agent, turn, and response identities (#8378)", async () => { + const { service, client } = serviceFixture(); + const created = service.createSession("runtime-conversation"); + const events: VoiceResponseEvent[] = []; + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "repository status", + deliver: (event) => events.push(event), + deliveryOpen: () => true, + }); + + expect(created).toMatchObject({ voiceSessionId: "voice-session" }); + expect(created.grant).not.toContain("openclaw"); + expect(client.calls).toEqual([ + { + idempotencyKey: "turn", + message: "repository status", + sessionKey: "agent:main:nemoclaw-voice:agent-session", + }, + ]); + expect(events).toEqual([ + { + type: "response.started", + voiceSessionId: "voice-session", + turnId: "turn", + responseId: "response", + }, + { + type: "response.text.delta", + voiceSessionId: "voice-session", + turnId: "turn", + responseId: "response", + sequence: 0, + text: "hello", + }, + { + type: "response.completed", + voiceSessionId: "voice-session", + turnId: "turn", + responseId: "response", + }, + ]); + service.closeAll(); + }); + + it("rejects duplicate and overlapping runtime commit IDs without another invocation (#8378)", async () => { + const client = new FakeAgentClient(); + let resolveRun: (value: { outcome: "completed" }) => void = () => {}; + client.run = async (onEvent) => { + onEvent({ type: "started" }); + return new Promise((resolve) => { + resolveRun = resolve; + }); + }; + const { service } = serviceFixture({ + client, + randomIds: ["voice-session", "agent-session", "turn", "response"], + }); + const created = service.createSession("runtime-conversation"); + const first = service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "commit-one", + text: "first", + deliver: () => {}, + deliveryOpen: () => true, + }); + + await vi.waitFor(() => expect(client.calls).toHaveLength(1)); + await expect( + service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "commit-one", + text: "duplicate", + deliver: () => {}, + deliveryOpen: () => true, + }), + ).rejects.toMatchObject({ code: "duplicate_turn" }); + await expect( + service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "commit-two", + text: "overlap", + deliver: () => {}, + deliveryOpen: () => true, + }), + ).rejects.toMatchObject({ code: "turn_in_progress" }); + expect(client.calls).toHaveLength(1); + resolveRun({ outcome: "completed" }); + await first; + service.closeAll(); + }); + + it("revokes the grant and closes the direct agent connection on close (#8378)", () => { + const { service, client } = serviceFixture(); + const created = service.createSession("runtime-conversation"); + + expect(() => service.closeSession(created.voiceSessionId, "wrong-grant")).toThrow( + VoiceGatewayRequestError, + ); + expect(client.closed).toBe(false); + service.closeSession(created.voiceSessionId, created.grant); + expect(client.closed).toBe(true); + expect(() => service.closeSession(created.voiceSessionId, created.grant)).toThrow( + "session_not_found", + ); + }); + + it("stops delivery after disconnect while allowing bounded agent work to finish (#8378)", async () => { + const { service } = serviceFixture(); + const created = service.createSession("runtime-conversation"); + const events: VoiceResponseEvent[] = []; + let open = true; + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "repository status", + deliver: (event) => { + events.push(event); + open = false; + }, + deliveryOpen: () => open, + }); + + expect(events.map((event) => event.type)).toEqual(["response.started"]); + service.closeAll(); + }); + + it("returns one content-free failure when the response exceeds its bound (#8378)", async () => { + const { service } = serviceFixture({ maxResponseBytes: 4 }); + const created = service.createSession("runtime-conversation"); + const events: VoiceResponseEvent[] = []; + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "repository status", + deliver: (event) => events.push(event), + deliveryOpen: () => true, + }); + + expect(events.at(-1)).toMatchObject({ + type: "response.failed", + reason: "response_too_large", + }); + expect(events.filter((event) => event.type.endsWith("completed"))).toHaveLength(0); + service.closeAll(); + }); + + it("fails once when an agent client duplicates response.started (#8378)", async () => { + const client = new FakeAgentClient(); + client.run = async (onEvent) => { + onEvent({ type: "started" }); + onEvent({ type: "started" }); + return { outcome: "completed" }; + }; + const { service } = serviceFixture({ client }); + const created = service.createSession("runtime-conversation"); + const events: VoiceResponseEvent[] = []; + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "repository status", + deliver: (event) => events.push(event), + deliveryOpen: () => true, + }); + + expect(events.filter((event) => event.type === "response.started")).toHaveLength(1); + expect(events.filter((event) => event.type === "response.failed")).toEqual([ + expect.objectContaining({ reason: "agent_protocol_error" }), + ]); + expect(events.some((event) => event.type === "response.completed")).toBe(false); + service.closeAll(); + }); + + it("bounds a disconnected or stalled agent turn with one timeout outcome (#8378)", async () => { + const client = new FakeAgentClient(); + client.run = async (onEvent) => { + onEvent({ type: "started" }); + return new Promise(() => {}); + }; + const { service } = serviceFixture({ client, turnTimeoutMs: 5 }); + const created = service.createSession("runtime-conversation"); + const events: VoiceResponseEvent[] = []; + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "repository status", + deliver: (event) => events.push(event), + deliveryOpen: () => true, + }); + + expect(client.closed).toBe(true); + expect(events.filter((event) => event.type === "response.failed")).toEqual([ + expect.objectContaining({ reason: "turn_timeout" }), + ]); + expect(events.some((event) => event.type === "response.completed")).toBe(false); + service.closeAll(); + }); + + it("normalizes a thrown agent client into exactly one terminal failure (#8378)", async () => { + const client = new FakeAgentClient(); + client.run = async () => { + throw new Error("native error with private details"); + }; + const { service } = serviceFixture({ client }); + const created = service.createSession("runtime-conversation"); + const events: VoiceResponseEvent[] = []; + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "repository status", + deliver: (event) => events.push(event), + deliveryOpen: () => true, + }); + + expect(events).toEqual([ + expect.objectContaining({ + type: "response.failed", + reason: "agent_gateway_unavailable", + }), + ]); + expect(JSON.stringify(events)).not.toContain("private details"); + service.closeAll(); + }); + + it("expires the grant, removes the binding, and closes the agent connection (#8378)", () => { + let now = 0; + const { service, client } = serviceFixture({ + now: () => now, + sessionLifetimeMs: 1_000, + }); + const created = service.createSession("runtime-conversation"); + now = 1_000; + + expect(() => service.closeSession(created.voiceSessionId, created.grant)).toThrow( + "session_not_found", + ); + expect(client.closed).toBe(true); + }); + + it("keeps credentials and conversational content out of lifecycle diagnostics (#8378)", async () => { + const { service, diagnostics } = serviceFixture(); + const created = service.createSession("runtime-conversation"); + + await service.commitTurn({ + voiceSessionId: created.voiceSessionId, + grant: created.grant, + commitId: "runtime-commit", + text: "private transcript and prompt", + deliver: () => {}, + deliveryOpen: () => true, + }); + + const output = JSON.stringify(diagnostics); + expect(output).not.toContain(created.grant); + expect(output).not.toContain("private transcript"); + expect(output).not.toContain("hello"); + expect(output).not.toContain("agent:main:nemoclaw-voice"); + service.closeAll(); + }); +}); diff --git a/src/lib/voice-gateway/session-service.ts b/src/lib/voice-gateway/session-service.ts new file mode 100644 index 00000000000..393af688656 --- /dev/null +++ b/src/lib/voice-gateway/session-service.ts @@ -0,0 +1,346 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto"; + +import { + type AgentTurnClient, + VOICE_GATEWAY_MAX_RESPONSE_BYTES, + VOICE_GATEWAY_SESSION_LIFETIME_MS, + VOICE_GATEWAY_TURN_TIMEOUT_MS, + type VoiceGatewayDiagnostic, + type VoiceGatewayFailureReason, + VoiceGatewayRequestError, + type VoiceResponseEvent, +} from "./contracts"; + +const RUNTIME_VALUE_PATTERN = /^[A-Za-z0-9](?:[A-Za-z0-9._:-]{0,126}[A-Za-z0-9])?$/u; +const MAX_TURN_TEXT_BYTES = 48 * 1024; + +interface ActiveSession { + readonly voiceSessionId: string; + readonly runtimeConversationId: string; + readonly agentSessionKey: string; + readonly grantHash: Buffer; + readonly expiresAt: number; + readonly client: AgentTurnClient; + expiryTimer: NodeJS.Timeout; + commitId: string | null; + turnState: "idle" | "running" | "terminal"; + revokedReason: "session_closed" | "session_expired" | null; +} + +export interface VoiceSessionServiceOptions { + readonly runtimeIdentity: string; + readonly runtimeProfile: string; + readonly sandbox: string; + readonly agent: string; + readonly createClient: () => AgentTurnClient; + readonly diagnostic?: (entry: VoiceGatewayDiagnostic) => void; + readonly now?: () => number; + readonly randomId?: () => string; + readonly randomGrant?: () => Buffer; + readonly sessionLifetimeMs?: number; + readonly turnTimeoutMs?: number; + readonly maxResponseBytes?: number; +} + +export interface CreatedVoiceSession { + readonly voiceSessionId: string; + readonly grant: string; + readonly expiresAt: string; +} + +function validateRuntimeValue(value: string, label: string): void { + if (!RUNTIME_VALUE_PATTERN.test(value)) { + throw new VoiceGatewayRequestError("invalid_request"); + } + if (label.length === 0) throw new VoiceGatewayRequestError("invalid_request"); +} + +function hashBearer(value: string): Buffer { + return createHash("sha256").update(value).digest(); +} + +function grantMatches(value: string, expectedHash: Buffer): boolean { + return timingSafeEqual(hashBearer(value), expectedHash); +} + +/** Owns one runtime-neutral voice session and its single committed turn. */ +export class VoiceSessionService { + private readonly options: Required< + Pick< + VoiceSessionServiceOptions, + | "now" + | "randomId" + | "randomGrant" + | "sessionLifetimeMs" + | "turnTimeoutMs" + | "maxResponseBytes" + > + > & + Omit< + VoiceSessionServiceOptions, + | "now" + | "randomId" + | "randomGrant" + | "sessionLifetimeMs" + | "turnTimeoutMs" + | "maxResponseBytes" + >; + private session: ActiveSession | null = null; + + constructor(options: VoiceSessionServiceOptions) { + this.options = { + ...options, + now: options.now ?? Date.now, + randomId: options.randomId ?? randomUUID, + randomGrant: options.randomGrant ?? (() => randomBytes(32)), + sessionLifetimeMs: options.sessionLifetimeMs ?? VOICE_GATEWAY_SESSION_LIFETIME_MS, + turnTimeoutMs: options.turnTimeoutMs ?? VOICE_GATEWAY_TURN_TIMEOUT_MS, + maxResponseBytes: options.maxResponseBytes ?? VOICE_GATEWAY_MAX_RESPONSE_BYTES, + }; + validateRuntimeValue(options.runtimeIdentity, "runtime identity"); + validateRuntimeValue(options.runtimeProfile, "runtime profile"); + validateRuntimeValue(options.sandbox, "sandbox"); + validateRuntimeValue(options.agent, "agent"); + } + + createSession(runtimeConversationId: string): CreatedVoiceSession { + validateRuntimeValue(runtimeConversationId, "runtime conversation ID"); + this.expireIfNeeded(); + if (this.session !== null) throw new VoiceGatewayRequestError("session_in_progress"); + + const now = this.options.now(); + const voiceSessionId = this.options.randomId(); + const agentSessionKey = `agent:${this.options.agent}:nemoclaw-voice:${this.options.randomId()}`; + const grant = this.options.randomGrant().toString("base64url"); + const expiresAt = now + this.options.sessionLifetimeMs; + const session: ActiveSession = { + voiceSessionId, + runtimeConversationId, + agentSessionKey, + grantHash: hashBearer(grant), + expiresAt, + client: this.options.createClient(), + expiryTimer: setTimeout( + () => this.revoke(session, "session_expired"), + this.options.sessionLifetimeMs, + ), + commitId: null, + turnState: "idle", + revokedReason: null, + }; + session.expiryTimer.unref?.(); + this.session = session; + this.options.diagnostic?.({ + event: "voice_session", + state: "created", + runtimeIdentity: this.options.runtimeIdentity, + runtimeProfile: this.options.runtimeProfile, + sandbox: this.options.sandbox, + agent: this.options.agent, + voiceSessionId, + runtimeConversationId, + }); + return { voiceSessionId, grant, expiresAt: new Date(expiresAt).toISOString() }; + } + + async commitTurn(options: { + readonly voiceSessionId: string; + readonly grant: string; + readonly commitId: string; + readonly text: string; + readonly deliver: (event: VoiceResponseEvent) => void; + readonly deliveryOpen: () => boolean; + }): Promise { + const session = this.authorize(options.voiceSessionId, options.grant); + validateRuntimeValue(options.commitId, "commit ID"); + if ( + options.text.length === 0 || + Buffer.byteLength(options.text) > MAX_TURN_TEXT_BYTES || + options.text.includes("\u0000") + ) { + throw new VoiceGatewayRequestError("invalid_request"); + } + if (session.commitId === options.commitId) { + throw new VoiceGatewayRequestError("duplicate_turn"); + } + if (session.turnState === "running") throw new VoiceGatewayRequestError("turn_in_progress"); + if (session.turnState === "terminal") throw new VoiceGatewayRequestError("turn_limit_reached"); + + session.commitId = options.commitId; + session.turnState = "running"; + const turnId = this.options.randomId(); + const responseId = this.options.randomId(); + const startedAt = this.options.now(); + let sequence = 0; + let responseBytes = 0; + let forcedReason: VoiceGatewayFailureReason | null = null; + let terminalSent = false; + let startedSent = false; + + const deliver = (event: VoiceResponseEvent) => { + if (!options.deliveryOpen()) return; + options.deliver(event); + }; + const finish = (reason: VoiceGatewayFailureReason | null) => { + if (terminalSent) return; + terminalSent = true; + session.turnState = "terminal"; + if (reason) { + deliver({ + type: "response.failed", + voiceSessionId: session.voiceSessionId, + turnId, + responseId, + reason, + }); + } else { + deliver({ + type: "response.completed", + voiceSessionId: session.voiceSessionId, + turnId, + responseId, + }); + } + this.options.diagnostic?.({ + event: "voice_turn", + state: reason ? "failed" : "completed", + runtimeProfile: this.options.runtimeProfile, + runtimeIdentity: this.options.runtimeIdentity, + sandbox: this.options.sandbox, + agent: this.options.agent, + voiceSessionId: session.voiceSessionId, + runtimeConversationId: session.runtimeConversationId, + turnId, + responseId, + ...(reason ? { reason } : {}), + durationMs: Math.max(0, this.options.now() - startedAt), + }); + }; + + let timeout: NodeJS.Timeout | undefined; + const timeoutResult = new Promise<"timeout">((resolve) => { + timeout = setTimeout(() => resolve("timeout"), this.options.turnTimeoutMs); + timeout.unref?.(); + }); + const runResult = Promise.resolve() + .then(() => + session.client.runTurn({ + sessionKey: session.agentSessionKey, + idempotencyKey: turnId, + message: options.text, + onEvent: (event) => { + if (terminalSent || forcedReason) return; + if (event.type === "started") { + if (startedSent) { + forcedReason = "agent_protocol_error"; + session.client.close(); + return; + } + startedSent = true; + deliver({ + type: "response.started", + voiceSessionId: session.voiceSessionId, + turnId, + responseId, + }); + return; + } + if (!startedSent) { + forcedReason = "agent_protocol_error"; + session.client.close(); + return; + } + responseBytes += Buffer.byteLength(event.text); + if (responseBytes > this.options.maxResponseBytes) { + forcedReason = "response_too_large"; + session.client.close(); + return; + } + deliver({ + type: "response.text.delta", + voiceSessionId: session.voiceSessionId, + turnId, + responseId, + sequence: sequence++, + text: event.text, + }); + }, + }), + ) + .catch( + (): Awaited> => ({ + outcome: "failed", + reason: "agent_gateway_unavailable", + }), + ); + + const result = await Promise.race([runResult, timeoutResult]); + if (timeout) clearTimeout(timeout); + if (result === "timeout") { + session.client.close(); + finish(session.revokedReason ?? "turn_timeout"); + } else if (forcedReason) { + finish(forcedReason); + } else if (session.revokedReason) { + finish(session.revokedReason); + } else if (result.outcome === "failed") { + finish(result.reason); + } else { + finish(null); + } + } + + closeSession(voiceSessionId: string, grant: string): void { + const session = this.authorize(voiceSessionId, grant); + this.revoke(session, "session_closed"); + } + + closeAll(): void { + if (this.session) this.revoke(this.session, "session_closed"); + } + + authorizeSession(voiceSessionId: string, grant: string): void { + this.authorize(voiceSessionId, grant); + } + + private authorize(voiceSessionId: string, grant: string): ActiveSession { + this.expireIfNeeded(); + const session = this.session; + if (!session || session.voiceSessionId !== voiceSessionId) { + throw new VoiceGatewayRequestError("session_not_found"); + } + if (!grantMatches(grant, session.grantHash)) { + throw new VoiceGatewayRequestError("authentication_failed"); + } + return session; + } + + private expireIfNeeded(): void { + if (this.session && this.options.now() >= this.session.expiresAt) { + this.revoke(this.session, "session_expired"); + } + } + + private revoke(session: ActiveSession, reason: "session_closed" | "session_expired"): void { + if (session.revokedReason) return; + session.revokedReason = reason; + clearTimeout(session.expiryTimer); + session.client.close(); + session.grantHash.fill(0); + if (this.session === session) this.session = null; + this.options.diagnostic?.({ + event: "voice_session", + state: reason === "session_expired" ? "expired" : "closed", + runtimeIdentity: this.options.runtimeIdentity, + runtimeProfile: this.options.runtimeProfile, + sandbox: this.options.sandbox, + agent: this.options.agent, + voiceSessionId: session.voiceSessionId, + runtimeConversationId: session.runtimeConversationId, + reason, + }); + } +} diff --git a/test/fixtures/voice-gateway/pinned-runtime-adapter.ts b/test/fixtures/voice-gateway/pinned-runtime-adapter.ts new file mode 100644 index 00000000000..2c00b2808b6 --- /dev/null +++ b/test/fixtures/voice-gateway/pinned-runtime-adapter.ts @@ -0,0 +1,112 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import http from "node:http"; + +export interface RuntimeSession { + readonly voiceSessionId: string; + readonly grant: string; + readonly expiresAt: string; +} + +async function request(options: { + readonly port: number; + readonly method: string; + readonly path: string; + readonly bearer: string; + readonly body?: object; +}): Promise<{ readonly status: number; readonly body: string; readonly contentType: string }> { + const body = options.body ? JSON.stringify(options.body) : ""; + return new Promise((resolve, reject) => { + const client = http.request( + { + host: "127.0.0.1", + port: options.port, + method: options.method, + path: options.path, + headers: { + authorization: `Bearer ${options.bearer}`, + ...(body + ? { + "content-length": String(Buffer.byteLength(body)), + "content-type": "application/json", + } + : {}), + }, + }, + (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk) => chunks.push(Buffer.from(chunk))); + response.on("end", () => + resolve({ + status: response.statusCode ?? 0, + body: Buffer.concat(chunks).toString("utf8"), + contentType: String(response.headers["content-type"] ?? ""), + }), + ); + }, + ); + client.once("error", reject); + client.end(body); + }); +} + +/** + * Deterministic fixture for the pinned runtime integration seam. + * + * It knows only the deployment bearer, voice-session grant, committed text, + * normalized response events, and the runtime's existing text output callback. + */ +export class PinnedVoiceRuntimeAdapter { + constructor( + private readonly port: number, + private readonly deploymentBearer: string, + private readonly outputText: (text: string) => void, + ) {} + + async createSession(runtimeConversationId: string): Promise { + const result = await request({ + port: this.port, + method: "POST", + path: "/v1/voice/sessions", + bearer: this.deploymentBearer, + body: { runtimeConversationId }, + }); + if (result.status !== 201) throw new Error(`session admission failed: ${result.status}`); + return JSON.parse(result.body) as RuntimeSession; + } + + async commitTurn(session: RuntimeSession, commitId: string, text: string): Promise { + const result = await request({ + port: this.port, + method: "POST", + path: `/v1/voice/sessions/${encodeURIComponent(session.voiceSessionId)}/turns`, + bearer: session.grant, + body: { commitId, text }, + }); + if (result.status !== 200 || !result.contentType.startsWith("application/x-ndjson")) { + throw new Error(`turn failed: ${result.status}`); + } + const events = result.body + .trim() + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line) as Record); + for (const event of events) { + if (event.type === "response.text.delta" && typeof event.text === "string") { + this.outputText(event.text); + } + } + return events; + } + + async closeSession(session: RuntimeSession): Promise { + const result = await request({ + port: this.port, + method: "DELETE", + path: `/v1/voice/sessions/${encodeURIComponent(session.voiceSessionId)}`, + bearer: session.grant, + }); + if (result.status !== 204) throw new Error(`session close failed: ${result.status}`); + } +} diff --git a/test/internal-cli.test.ts b/test/internal-cli.test.ts index 8027c5d8685..14f12e1e56e 100644 --- a/test/internal-cli.test.ts +++ b/test/internal-cli.test.ts @@ -117,4 +117,35 @@ describe("internal oclif namespace", () => { provider: { normalized: "nim-local", raw: "nim", valid: true }, }); }); + + it("fails the experimental voice gateway gate before parsing credential flags (#8378)", () => { + const env = { ...process.env }; + delete env.NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY; + + const result = spawnSync(process.execPath, [CLI, "internal", "voice-gateway", "serve"], { + encoding: "utf-8", + env, + }); + + expect(result.status).not.toBe(0); + expect(result.stderr).toContain("Experimental voice gateway is disabled"); + expect(result.stderr).not.toContain("Missing required flag"); + }); + + it("ships hidden help for the feature-gated voice gateway command (#8378)", () => { + const result = spawnSync( + process.execPath, + [CLI, "internal", "voice-gateway", "serve", "--help"], + { + encoding: "utf-8", + env: { ...process.env, NEMOCLAW_EXPERIMENTAL_VOICE_GATEWAY: "1" }, + }, + ); + + expect(result.status).toBe(0); + expect(result.stdout).toContain("Internal: serve the experimental voice gateway"); + expect(result.stdout).toContain("--deployment-credential-file"); + expect(result.stdout).toContain("--openclaw-credential-file"); + expect(result.stdout).toContain("--runtime-identity"); + }); }); diff --git a/test/package-contract/cli/oclif-metadata.test.ts b/test/package-contract/cli/oclif-metadata.test.ts index 6995df7c53c..e8a47ccdb5d 100644 --- a/test/package-contract/cli/oclif-metadata.test.ts +++ b/test/package-contract/cli/oclif-metadata.test.ts @@ -24,6 +24,9 @@ describe("oclif metadata lookup", () => { expect(getRegisteredOclifCommandSummary("internal:uninstall:plan")).toBe( "Internal: build the NemoClaw uninstall plan", ); + expect(getRegisteredOclifCommandSummary("internal:voice-gateway:serve")).toBe( + "Internal: serve the experimental voice gateway", + ); }); it("keeps generated manifest command IDs aligned with oclif Config", async () => { diff --git a/test/voice-gateway-integration.test.ts b/test/voice-gateway-integration.test.ts new file mode 100644 index 00000000000..c5d589934a4 --- /dev/null +++ b/test/voice-gateway-integration.test.ts @@ -0,0 +1,259 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import http, { type Server } from "node:http"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { createVoiceGatewayServer } from "../src/lib/adapters/http/voice-gateway-server"; +import type { AgentTurnClient, AgentTurnEvent } from "../src/lib/voice-gateway/contracts"; +import { VoiceSessionService } from "../src/lib/voice-gateway/session-service"; +import { PinnedVoiceRuntimeAdapter } from "./fixtures/voice-gateway/pinned-runtime-adapter"; + +const DEPLOYMENT_BEARER = "deployment-bearer-for-voice-gateway-tests"; +const OPENCLAW_CREDENTIAL = "openclaw-credential-stays-in-nemoclaw"; +const servers = new Set(); + +class FakeOpenClawGatewayClient implements AgentTurnClient { + readonly calls: Array<{ + idempotencyKey: string; + message: string; + sessionKey: string; + credential: string; + }> = []; + closed = false; + + constructor(private readonly credential: string) {} + + close(): void { + this.closed = true; + } + + async runTurn(options: { + readonly idempotencyKey: string; + readonly message: string; + readonly onEvent: (event: AgentTurnEvent) => void; + readonly sessionKey: string; + }): ReturnType { + this.calls.push({ + idempotencyKey: options.idempotencyKey, + message: options.message, + sessionKey: options.sessionKey, + credential: this.credential, + }); + options.onEvent({ type: "started" }); + options.onEvent({ type: "text", text: "working tree " }); + options.onEvent({ type: "text", text: "is clean" }); + return { outcome: "completed" }; + } +} + +async function listen(server: Server): Promise { + servers.add(server); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + server.removeListener("error", reject); + resolve(); + }); + }); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("server did not bind"); + return address.port; +} + +async function requestJson(options: { + readonly port: number; + readonly method: string; + readonly path: string; + readonly bearer?: string; + readonly body?: object; +}): Promise<{ readonly status: number; readonly body: string }> { + const body = options.body ? JSON.stringify(options.body) : ""; + return new Promise((resolve, reject) => { + const client = http.request( + { + host: "127.0.0.1", + port: options.port, + method: options.method, + path: options.path, + headers: { + ...(options.bearer ? { authorization: `Bearer ${options.bearer}` } : {}), + ...(body + ? { + "content-length": String(Buffer.byteLength(body)), + "content-type": "application/json", + } + : {}), + }, + }, + (response) => { + const chunks: Buffer[] = []; + response.on("data", (chunk) => chunks.push(Buffer.from(chunk))); + response.on("end", () => + resolve({ + status: response.statusCode ?? 0, + body: Buffer.concat(chunks).toString("utf8"), + }), + ); + }, + ); + client.once("error", reject); + client.end(body); + }); +} + +afterEach(async () => { + await Promise.all( + [...servers].map( + (server) => + new Promise((resolve) => { + servers.delete(server); + if (server.listening) server.close(() => resolve()); + else resolve(); + }), + ), + ); +}); + +describe("experimental voice gateway composed boundary", () => { + it("routes one committed turn into the pinned runtime output without exposing OpenClaw authority (#8378)", async () => { + const fakeOpenClaw = new FakeOpenClawGatewayClient(OPENCLAW_CREDENTIAL); + const ids = ["voice-session", "agent-session", "turn", "response"]; + const service = new VoiceSessionService({ + runtimeIdentity: "voiceclaw-local", + runtimeProfile: "voiceclaw-pinned", + sandbox: "repository-fixture", + agent: "main", + createClient: () => fakeOpenClaw, + randomId: () => ids.shift() ?? "extra", + randomGrant: () => Buffer.alloc(32, 9), + }); + const server = createVoiceGatewayServer({ + deploymentCredential: DEPLOYMENT_BEARER, + service, + }); + const port = await listen(server); + const output: string[] = []; + const runtime = new PinnedVoiceRuntimeAdapter(port, DEPLOYMENT_BEARER, (text) => + output.push(text), + ); + + const session = await runtime.createSession("runtime-conversation"); + const events = await runtime.commitTurn(session, "runtime-commit", "repository status"); + await runtime.closeSession(session); + + expect(output.join("")).toBe("working tree is clean"); + expect(fakeOpenClaw.calls).toEqual([ + { + idempotencyKey: "turn", + message: "repository status", + sessionKey: "agent:main:nemoclaw-voice:agent-session", + credential: OPENCLAW_CREDENTIAL, + }, + ]); + const runtimeVisible = JSON.stringify({ session, events, output }); + expect(runtimeVisible).not.toContain(OPENCLAW_CREDENTIAL); + expect(runtimeVisible).not.toContain("agent:main:nemoclaw-voice"); + expect(runtimeVisible).not.toContain("runId"); + expect(events.map((event) => (event as { type: string }).type)).toEqual([ + "response.started", + "response.text.delta", + "response.text.delta", + "response.completed", + ]); + expect(fakeOpenClaw.closed).toBe(true); + }); + + it("authenticates before admission or turn parsing and rejects runtime-selected authority (#8378)", async () => { + const fakeOpenClaw = new FakeOpenClawGatewayClient(OPENCLAW_CREDENTIAL); + let clientsCreated = 0; + const service = new VoiceSessionService({ + runtimeIdentity: "voiceclaw-local", + runtimeProfile: "voiceclaw-pinned", + sandbox: "repository-fixture", + agent: "main", + createClient: () => { + clientsCreated += 1; + return fakeOpenClaw; + }, + randomGrant: () => Buffer.alloc(32, 9), + }); + const port = await listen( + createVoiceGatewayServer({ + deploymentCredential: DEPLOYMENT_BEARER, + service, + }), + ); + + const missingAdmission = await requestJson({ + port, + method: "POST", + path: "/v1/voice/sessions", + body: { runtimeConversationId: "runtime-conversation" }, + }); + expect(missingAdmission).toEqual({ + status: 401, + body: '{"error":"authentication_failed"}', + }); + + const override = await requestJson({ + port, + method: "POST", + path: "/v1/voice/sessions", + bearer: DEPLOYMENT_BEARER, + body: { + runtimeConversationId: "runtime-conversation", + agent: "runtime-selected", + gatewayUrl: "ws://attacker.invalid/ws", + }, + }); + expect(override.status).toBe(400); + expect(clientsCreated).toBe(0); + + const runtime = new PinnedVoiceRuntimeAdapter(port, DEPLOYMENT_BEARER, () => {}); + const session = await runtime.createSession("runtime-conversation"); + expect(clientsCreated).toBe(1); + + const wrongGrant = await requestJson({ + port, + method: "POST", + path: `/v1/voice/sessions/${session.voiceSessionId}/turns`, + bearer: "wrong-session-grant", + body: { commitId: "commit-one", text: "must not parse into an invocation" }, + }); + expect(wrongGrant.status).toBe(401); + expect(fakeOpenClaw.calls).toHaveLength(0); + + const otherSession = await requestJson({ + port, + method: "POST", + path: "/v1/voice/sessions/other-session/turns", + bearer: session.grant, + body: { commitId: "commit-one", text: "must not invoke" }, + }); + expect(otherSession.status).toBe(404); + expect(fakeOpenClaw.calls).toHaveLength(0); + + const oversized = await requestJson({ + port, + method: "POST", + path: `/v1/voice/sessions/${session.voiceSessionId}/turns`, + bearer: session.grant, + body: { commitId: "commit-one", text: "x".repeat(70 * 1024) }, + }); + expect(oversized.status).toBe(413); + expect(fakeOpenClaw.calls).toHaveLength(0); + + const malformedRoute = await requestJson({ + port, + method: "POST", + path: "/v1/voice/sessions/%ZZ/turns", + bearer: session.grant, + body: { commitId: "commit-one", text: "must not invoke" }, + }); + expect(malformedRoute).toEqual({ status: 400, body: "" }); + expect(fakeOpenClaw.calls).toHaveLength(0); + await runtime.closeSession(session); + }); +}); From 19db569b84fe32404512b32663a89100631a3c6b Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Wed, 5 Aug 2026 23:21:48 -0400 Subject: [PATCH 2/9] test(cli): keep voice gateway tests linear Signed-off-by: Julie Yaunches --- src/lib/actions/voice-gateway/serve.test.ts | 2 +- src/lib/voice-gateway/openclaw-client.test.ts | 40 ++++++++++--------- test/voice-gateway-integration.test.ts | 8 ++-- 3 files changed, 27 insertions(+), 23 deletions(-) diff --git a/src/lib/actions/voice-gateway/serve.test.ts b/src/lib/actions/voice-gateway/serve.test.ts index d308d80b738..32326a1e5b5 100644 --- a/src/lib/actions/voice-gateway/serve.test.ts +++ b/src/lib/actions/voice-gateway/serve.test.ts @@ -78,7 +78,7 @@ describe("voice gateway listener lifetime", () => { this.listenArgs = args.slice(0, 2); this.listening = true; const callback = args.at(-1); - if (typeof callback === "function") callback(); + typeof callback === "function" ? callback() : undefined; return this; } diff --git a/src/lib/voice-gateway/openclaw-client.test.ts b/src/lib/voice-gateway/openclaw-client.test.ts index 1482dc7d6f4..dfe1dc75641 100644 --- a/src/lib/voice-gateway/openclaw-client.test.ts +++ b/src/lib/voice-gateway/openclaw-client.test.ts @@ -34,26 +34,30 @@ class FakeWebSocket { send(data: string): void { const request = JSON.parse(data) as SentRequest; this.sent.push(request); - if (request.method === "connect") { - queueMicrotask(() => { - this.respond(request.id, {}); - if (this.oversizedFrameAfterConnect) { - this.onmessage?.({ data: `{"padding":"${"x".repeat(3 * 1024 * 1024)}"}` }); - } - }); - return; - } - if (request.method === "chat.send") { - const sessionKey = String(request.params.sessionKey); - queueMicrotask(() => { - this.respond(request.id, { runId: "expected-run" }); + switch (request.method) { + case "connect": + queueMicrotask(() => { + this.respond(request.id, {}); + this.oversizedFrameAfterConnect + ? this.onmessage?.({ data: `{"padding":"${"x".repeat(3 * 1024 * 1024)}"}` }) + : undefined; + }); + return; + case "chat.send": { + const sessionKey = String(request.params.sessionKey); queueMicrotask(() => { - this.event(sessionKey, "other-run", "delta", "discarded run"); - this.event("other-session", "expected-run", "delta", "discarded session"); - this.event(sessionKey, "expected-run", "delta", this.firstReply); - this.event(sessionKey, "expected-run", "final", this.finalReply); + this.respond(request.id, { runId: "expected-run" }); + queueMicrotask(() => { + this.event(sessionKey, "other-run", "delta", "discarded run"); + this.event("other-session", "expected-run", "delta", "discarded session"); + this.event(sessionKey, "expected-run", "delta", this.firstReply); + this.event(sessionKey, "expected-run", "final", this.finalReply); + }); }); - }); + return; + } + default: + return; } } diff --git a/test/voice-gateway-integration.test.ts b/test/voice-gateway-integration.test.ts index c5d589934a4..6802f00a9fa 100644 --- a/test/voice-gateway-integration.test.ts +++ b/test/voice-gateway-integration.test.ts @@ -58,8 +58,9 @@ async function listen(server: Server): Promise { }); }); const address = server.address(); - if (!address || typeof address === "string") throw new Error("server did not bind"); - return address.port; + expect(address).toBeTruthy(); + expect(typeof address).not.toBe("string"); + return (address as { readonly port: number }).port; } async function requestJson(options: { @@ -109,8 +110,7 @@ afterEach(async () => { (server) => new Promise((resolve) => { servers.delete(server); - if (server.listening) server.close(() => resolve()); - else resolve(); + server.listening ? server.close(() => resolve()) : resolve(); }), ), ); From b84e049fe69e732ee316017cc45d83a7908798bb Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 08:40:15 -0400 Subject: [PATCH 3/9] test(voice-gateway): keep listener fake linear Signed-off-by: Julie Yaunches --- src/lib/actions/voice-gateway/serve.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/lib/actions/voice-gateway/serve.test.ts b/src/lib/actions/voice-gateway/serve.test.ts index 32326a1e5b5..fed7a4d88d4 100644 --- a/src/lib/actions/voice-gateway/serve.test.ts +++ b/src/lib/actions/voice-gateway/serve.test.ts @@ -77,8 +77,8 @@ describe("voice gateway listener lifetime", () => { listen(...args: unknown[]): this { this.listenArgs = args.slice(0, 2); this.listening = true; - const callback = args.at(-1); - typeof callback === "function" ? callback() : undefined; + const callback = args.at(-1) as () => void; + callback(); return this; } From 8c5759b623ee366a10a78016036db7f4ba86a0ae Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 09:21:23 -0400 Subject: [PATCH 4/9] chore(architecture): reconcile voice command fan-in budget Signed-off-by: Julie Yaunches --- ci/source-architecture-budget.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ci/source-architecture-budget.json b/ci/source-architecture-budget.json index df880bd8a2e..4f44303b5c2 100644 --- a/ci/source-architecture-budget.json +++ b/ci/source-architecture-budget.json @@ -12,7 +12,7 @@ "src/lib/adapters/openshell/timeouts.ts": 37, "src/lib/agent/defs.ts": 32, "src/lib/cli/branding.ts": 87, - "src/lib/cli/nemoclaw-oclif-command.ts": 105, + "src/lib/cli/nemoclaw-oclif-command.ts": 106, "src/lib/cli/terminal-style.ts": 45, "src/lib/core/json-types.ts": 37, "src/lib/core/ports.ts": 88, From 402cdd225a9408253e3e3d01a21075ae6a4d7e76 Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 09:46:32 -0400 Subject: [PATCH 5/9] fix(voice-gateway): guard premature request close Signed-off-by: Julie Yaunches --- src/lib/adapters/http/voice-gateway-server.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/lib/adapters/http/voice-gateway-server.ts b/src/lib/adapters/http/voice-gateway-server.ts index 803b9f45a28..15649f8f41d 100644 --- a/src/lib/adapters/http/voice-gateway-server.ts +++ b/src/lib/adapters/http/voice-gateway-server.ts @@ -139,7 +139,9 @@ function readJsonBody(request: http.IncomingMessage): Promise finish(new BodyError(400))); + request.once("close", () => { + if (!request.readableEnded) finish(new BodyError(400)); + }); request.once("error", () => finish(new BodyError(400))); request.once("end", () => finish()); }); From c71a4fae8a5446fcdefbca57c50ae87cc5464884 Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 10:01:46 -0400 Subject: [PATCH 6/9] fix(voice-gateway): reject BSD credential symlinks Signed-off-by: Julie Yaunches --- src/lib/voice-gateway/credential-file.test.ts | 13 ++++++++++++- src/lib/voice-gateway/credential-file.ts | 3 ++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/lib/voice-gateway/credential-file.test.ts b/src/lib/voice-gateway/credential-file.test.ts index ac016770add..cfdb9b66c67 100644 --- a/src/lib/voice-gateway/credential-file.test.ts +++ b/src/lib/voice-gateway/credential-file.test.ts @@ -5,7 +5,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { readPrivateBearerFile } from "./credential-file"; @@ -32,6 +32,17 @@ describe("voice gateway credential file", () => { expect(readPrivateBearerFile(file, "Test credential")).toBe(CREDENTIAL); }); + it("treats macOS EMLINK as a rejected symbolic link (#8378)", () => { + const error = Object.assign(new Error("symbolic link"), { code: "EMLINK" }); + vi.spyOn(fs, "openSync").mockImplementationOnce(() => { + throw error; + }); + + expect(() => readPrivateBearerFile("/absolute/credential", "Test credential")).toThrow( + "symbolic-link", + ); + }); + it.each([ { name: "relative path", diff --git a/src/lib/voice-gateway/credential-file.ts b/src/lib/voice-gateway/credential-file.ts index b5d7a566100..5f7bdcdabc4 100644 --- a/src/lib/voice-gateway/credential-file.ts +++ b/src/lib/voice-gateway/credential-file.ts @@ -46,7 +46,8 @@ export function readPrivateBearerFile(filePath: string, label: string): string { fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW | (fs.constants.O_NONBLOCK ?? 0), ); } catch (error) { - if ((error as NodeJS.ErrnoException).code === "ELOOP") { + const code = (error as NodeJS.ErrnoException).code; + if (code === "ELOOP" || code === "EMLINK") { throw new Error(`Refusing to read a symbolic-link ${label} file: ${filePath}`); } throw error; From 5cb4bb06a2bab2d7ad08c9880b6536f5e7c505c3 Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 10:27:04 -0400 Subject: [PATCH 7/9] test(voice-gateway): make WebSocket fake linear Signed-off-by: Julie Yaunches --- src/lib/voice-gateway/openclaw-client.test.ts | 75 +++++++++---------- 1 file changed, 36 insertions(+), 39 deletions(-) diff --git a/src/lib/voice-gateway/openclaw-client.test.ts b/src/lib/voice-gateway/openclaw-client.test.ts index dfe1dc75641..95413310524 100644 --- a/src/lib/voice-gateway/openclaw-client.test.ts +++ b/src/lib/voice-gateway/openclaw-client.test.ts @@ -13,6 +13,8 @@ interface SentRequest { readonly params: Record; } +type Handler = (request: SentRequest, socket: FakeWebSocket) => void; + class FakeWebSocket { onopen: (() => void) | null = null; onmessage: ((event: { readonly data: unknown }) => void) | null = null; @@ -20,58 +22,28 @@ class FakeWebSocket { onclose: (() => void) | null = null; readonly sent: SentRequest[] = []; closed = false; - readonly firstReply: string; - readonly finalReply: string; - readonly oversizedFrameAfterConnect: boolean; - - constructor(firstReply = "hel", finalReply = "hello", oversizedFrameAfterConnect = false) { - this.firstReply = firstReply; - this.finalReply = finalReply; - this.oversizedFrameAfterConnect = oversizedFrameAfterConnect; + + constructor(private readonly handlers: Record) { queueMicrotask(() => this.onopen?.()); } send(data: string): void { const request = JSON.parse(data) as SentRequest; this.sent.push(request); - switch (request.method) { - case "connect": - queueMicrotask(() => { - this.respond(request.id, {}); - this.oversizedFrameAfterConnect - ? this.onmessage?.({ data: `{"padding":"${"x".repeat(3 * 1024 * 1024)}"}` }) - : undefined; - }); - return; - case "chat.send": { - const sessionKey = String(request.params.sessionKey); - queueMicrotask(() => { - this.respond(request.id, { runId: "expected-run" }); - queueMicrotask(() => { - this.event(sessionKey, "other-run", "delta", "discarded run"); - this.event("other-session", "expected-run", "delta", "discarded session"); - this.event(sessionKey, "expected-run", "delta", this.firstReply); - this.event(sessionKey, "expected-run", "final", this.finalReply); - }); - }); - return; - } - default: - return; - } + queueMicrotask(() => this.handlers[request.method]?.(request, this)); } close(): void { this.closed = true; } - private respond(id: string, payload: Record): void { + respond(id: string, payload: Record): void { this.onmessage?.({ data: JSON.stringify({ type: "res", id, ok: true, payload }), }); } - private event(sessionKey: string, runId: string, state: string, text: string): void { + event(sessionKey: string, runId: string, state: string, text: string): void { this.onmessage?.({ data: JSON.stringify({ type: "event", @@ -88,9 +60,34 @@ class FakeWebSocket { } } +function orderedReplyHandlers(firstReply = "hel", finalReply = "hello"): Record { + return { + connect: (request, socket) => socket.respond(request.id, {}), + "chat.send": (request, socket) => { + const sessionKey = String(request.params.sessionKey); + socket.respond(request.id, { runId: "expected-run" }); + queueMicrotask(() => { + socket.event(sessionKey, "other-run", "delta", "discarded run"); + socket.event("other-session", "expected-run", "delta", "discarded session"); + socket.event(sessionKey, "expected-run", "delta", firstReply); + socket.event(sessionKey, "expected-run", "final", finalReply); + }); + }, + }; +} + +function oversizedFrameHandlers(): Record { + return { + connect: (request, socket) => { + socket.respond(request.id, {}); + socket.onmessage?.({ data: `{\"padding\":\"${"x".repeat(3 * 1024 * 1024)}\"}` }); + }, + }; +} + describe("OpenClaw voice gateway client", () => { it("uses bounded operator scopes and emits only ordered normalized text for the expected session and run (#8378)", async () => { - const socket = new FakeWebSocket(); + const socket = new FakeWebSocket(orderedReplyHandlers()); const events: AgentTurnEvent[] = []; const client = new OpenClawVoiceClient({ gatewayUrl: "ws://127.0.0.1:18789/ws", @@ -130,7 +127,7 @@ describe("OpenClaw voice gateway client", () => { }); it("fails closed when ordered response text changes its prior prefix (#8378)", async () => { - const socket = new FakeWebSocket("first", "different"); + const socket = new FakeWebSocket(orderedReplyHandlers("first", "different")); const client = new OpenClawVoiceClient({ gatewayUrl: "ws://127.0.0.1:18789/ws", credential: "openclaw-credential-must-not-cross", @@ -148,7 +145,7 @@ describe("OpenClaw voice gateway client", () => { }); it("closes the direct WebSocket connection when the session owner revokes it (#8378)", () => { - const socket = new FakeWebSocket(); + const socket = new FakeWebSocket(orderedReplyHandlers()); const client = new OpenClawVoiceClient({ gatewayUrl: "ws://127.0.0.1:18789/ws", credential: "openclaw-credential-must-not-cross", @@ -167,7 +164,7 @@ describe("OpenClaw voice gateway client", () => { }); it("rejects an oversized native frame before sending agent work (#8378)", async () => { - const socket = new FakeWebSocket("hel", "hello", true); + const socket = new FakeWebSocket(oversizedFrameHandlers()); const client = new OpenClawVoiceClient({ gatewayUrl: "ws://127.0.0.1:18789/ws", credential: "openclaw-credential-must-not-cross", From 6f5b73a2b8c2a3f0d9829faf5d5f4e2a2f5d72f2 Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 10:52:55 -0400 Subject: [PATCH 8/9] test(voice-gateway): assert active turn revocation Signed-off-by: Julie Yaunches --- src/lib/voice-gateway/openclaw-client.test.ts | 22 +++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/src/lib/voice-gateway/openclaw-client.test.ts b/src/lib/voice-gateway/openclaw-client.test.ts index 95413310524..c02c22f9ce5 100644 --- a/src/lib/voice-gateway/openclaw-client.test.ts +++ b/src/lib/voice-gateway/openclaw-client.test.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import type { AgentTurnEvent } from "./contracts"; import { OpenClawVoiceClient } from "./openclaw-client"; @@ -76,6 +76,13 @@ function orderedReplyHandlers(firstReply = "hel", finalReply = "hello"): Record< }; } +function activeTurnHandlers(): Record { + return { + connect: (request, socket) => socket.respond(request.id, {}), + "chat.send": (request, socket) => socket.respond(request.id, { runId: "expected-run" }), + }; +} + function oversizedFrameHandlers(): Record { return { connect: (request, socket) => { @@ -144,22 +151,29 @@ describe("OpenClaw voice gateway client", () => { ).resolves.toEqual({ outcome: "failed", reason: "agent_protocol_error" }); }); - it("closes the direct WebSocket connection when the session owner revokes it (#8378)", () => { - const socket = new FakeWebSocket(orderedReplyHandlers()); + it("closes the direct WebSocket connection when the session owner revokes it (#8378)", async () => { + const socket = new FakeWebSocket(activeTurnHandlers()); const client = new OpenClawVoiceClient({ gatewayUrl: "ws://127.0.0.1:18789/ws", credential: "openclaw-credential-must-not-cross", webSocketFactory: () => socket, }); - void client.runTurn({ + const turn = client.runTurn({ sessionKey: "agent:main:nemoclaw-voice:session", idempotencyKey: "generated-turn-id", message: "repository status", onEvent: () => {}, }); + await vi.waitFor(() => + expect(socket.sent.some((request) => request.method === "chat.send")).toBe(true), + ); client.close(); + await expect(turn).resolves.toEqual({ + outcome: "failed", + reason: "agent_gateway_unavailable", + }); expect(socket.closed).toBe(true); }); From 79b1425f69e4d8f06fb4d9a85b90d863b4b0ffcf Mon Sep 17 00:00:00 2001 From: Julie Yaunches Date: Thu, 6 Aug 2026 11:11:55 -0400 Subject: [PATCH 9/9] fix(voice-gateway): centralize turn transport cleanup Signed-off-by: Julie Yaunches --- src/lib/voice-gateway/openclaw-client.ts | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/lib/voice-gateway/openclaw-client.ts b/src/lib/voice-gateway/openclaw-client.ts index 6c0105b1d85..b08631c400e 100644 --- a/src/lib/voice-gateway/openclaw-client.ts +++ b/src/lib/voice-gateway/openclaw-client.ts @@ -157,11 +157,14 @@ export class OpenClawVoiceClient implements AgentTurnClient { const terminalPromise = new Promise((resolve) => { resolveTerminal = resolve; }); + let settleOpen: (() => void) | null = null; const finish = (value: TurnResult) => { if (terminal) return; terminal = true; this.cancelCurrent = null; clearPending(); + settleOpen?.(); + socket.close(); resolveTerminal(value); }; this.cancelCurrent = () => finish({ outcome: "failed", reason: "agent_gateway_unavailable" }); @@ -195,7 +198,6 @@ export class OpenClawVoiceClient implements AgentTurnClient { const raw = String(event.data); if (Buffer.byteLength(raw) > MAX_NATIVE_FRAME_BYTES) { finish({ outcome: "failed", reason: "agent_protocol_error" }); - socket.close(); return; } frame = record(JSON.parse(raw)); @@ -218,7 +220,6 @@ export class OpenClawVoiceClient implements AgentTurnClient { if (activeRunId === null) { if (queuedChatEvents.length >= MAX_QUEUED_CHAT_EVENTS) { finish({ outcome: "failed", reason: "agent_protocol_error" }); - socket.close(); return; } queuedChatEvents.push(chat); @@ -237,6 +238,10 @@ export class OpenClawVoiceClient implements AgentTurnClient { try { await new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error("open timeout")), REQUEST_TIMEOUT_MS); + settleOpen = () => { + clearTimeout(timer); + reject(new Error("turn terminal")); + }; rejectOpen = () => { clearTimeout(timer); reject(new Error("open failed")); @@ -244,6 +249,7 @@ export class OpenClawVoiceClient implements AgentTurnClient { socket.onopen = () => { clearTimeout(timer); rejectOpen = null; + settleOpen = null; resolve(); }; });