mirror of
https://github.com/openclaw/openclaw.git
synced 2026-10-03 09:39:25 +00:00
feat(deepgram): support Flux voice-note transcription (#141836)
Closes #129452. Replaces #129442 and preserves Safzan Pirani's original Flux contribution. ## Problem and fix Selecting `flux-general-en` or `flux-general-multi` sends voice-note audio to Deepgram's prerecorded HTTP `/v1/listen` endpoint, which rejects Flux. This change routes those models through the streaming `/v2/listen` protocol. Operators use their existing media model settings; Nova remains the default. Flux requires `ffmpeg`. The Deepgram plugin owns bounded audio conversion, frame pacing, protocol parsing, and transcript assembly. The shared WebSocket connector applies resolved authentication, private-network policy, proxy routing, and TLS settings. One connection deadline covers DNS preparation, proxy CONNECT, and the opening handshake; Flux keeps its original transcription-attempt deadline after preparation. Cancellation and socket closure release pending connections. Proxy connections use the existing shared Node agent backed by `@openclaw/proxyline@0.3.12`. [Proxyline #34](https://github.com/openclaw/proxyline/pull/34), now merged, adds prepared proxy DNS/TLS connection options while preserving its existing pending-socket ownership. OpenClaw passes these through `createNodeProxyAgent(...)`; it does not retain a separate proxy-agent implementation or add `https-proxy-agent` as a direct dependency. Proxy TLS and target TLS remain separate. Configured proxies retain resolved target-address checks before connection. Applicable managed and ambient HTTP(S) proxies retain their existing DNS delegation. `NO_PROXY` bypasses and `ALL_PROXY` alone do not disable address checks. HTTP and WebSocket paths share the managed-proxy predicate. The query builder combines saved language and explicit query inputs before applying [Deepgram's model contract](https://developers.deepgram.com/reference/speech-to-text/listen-flux). Only `flux-general-multi` receives `language_hint`; explicit query values keep their precedence. The English-only model ignores both language inputs without requiring changes to saved settings. The documentation follows main's split-page structure: request policy is documented on [Custom providers](https://docs.openclaw.ai/gateway/config-tools/custom-providers), and connection ownership on [Provider voice capabilities](https://docs.openclaw.ai/plugins/sdk-provider-plugins/voice-and-audio). The Deepgram manifest keeps main's categories and the Flux description. No configuration keys or storage changes are added. ## Canonical transport evidence These results cover the retained compiled candidate with the canonical Proxyline implementation. They are not claims about a newly installed or rebuilt merge head. - Independent public CLI acceptance saved and read back all four combinations of the two Flux models with either top-level `language: "en"` or `providerOptions.deepgram.language_hint: "en"`. Each ran `openclaw infer audio transcribe --file sample.wav --json` without model or language overrides, returned the expected “Life moves pretty fast” transcript, and exited 0. The original model configuration was restored and verified. - A saved Nova configuration and a post-fault Flux control returned the expected transcript and exited 0. - A configured proxy with private-network access explicitly denied produced a visible target-address rejection and exit 1, with no CONNECT admissions, no target connections or bytes, and no remaining proxy connections. - A stalled CONNECT produced a visible transcription timeout and exit 1 without forced termination. Both admitted connections closed before the CLI exited; zero connections remained. This records two attempts, not a one-second deadline for the whole CLI invocation. - A successful real-provider transcription through an HTTPS proxy recorded certificate verification enabled, the explicit server name, an authorized client certificate, CONNECT to the intended provider, 570,755 bytes forwarded upstream and 47,863 downstream, and complete peer cleanup. Verification mode was observed from the operator configuration; this public acceptance did not independently inject an invalid server certificate. - Focused canonical tests passed: 136 WebSocket/HTTP address-policy tests, 3 shared Node-agent tests, and 37 Deepgram tests. Proxyline's 11 connection-control tests passed, including prepared lookup/TLS settings and Node certificate-verification defaults. The retained runtime build, formatting, lint, and documentation checks also passed. - Proxyline's upstream review and Linux/macOS/Windows, package, and CodeQL checks passed before merge. Its published `0.3.12` package has been inspected: `src` and `dist` are byte-identical to the tested package. Registry metadata identifies release commit `46a8aa2e3c4b8fbed5fc3ccc16a4631cb7ca3d24` and includes package provenance. ## Regression evidence retained - On baseline `8954f104fb`, the compiled public command failed with HTTP 400 `V2_MODEL_ON_V1_LISTEN_ENDPOINT` and exit 1. The repaired command returned the expected real transcript. - Five deadline/cancellation cases failed before repair and passed afterward, including socket termination during pending proxy CONNECT. Both English-only language-input cases also failed before their query repair and passed afterward. - Before the address-policy repair, the forbidden-target fixture received one connection and 1,600 TLS handshake bytes. After repair, it received zero connections and zero bytes. The canonical acceptance above repeats the repaired denial through the public CLI. - Earlier broad media validation passed 354 tests across 26 files. Earlier SDK surface and import-cycle checks passed. A missing-file public CLI control produced a visible error and exit 1. These are historical coverage, not fresh merge-head results. - An explicit `undefined` query value exposed by test-type CI was corrected by omitting absent fixture keys. The 11 Flux tests passed afterward; that correction did not change production code or live-proof inputs. ## Review and merge status The latest ClawSweeper review of `550a7f60d612b1f19efcaec9b94112cf76338931` found no actionable code defect and requested dependency authorization, conflict resolution, and branch readiness. Its suggested direct `https-proxy-agent` addition is superseded by the canonical Proxyline repair above. The existing Proxyline dependency is updated to the published `0.3.12` release; any repository-enforced dependency approval must cover the eventual head. The maintainer approved an exact-version release-age exception for `@openclaw/proxyline@0.3.12` through **2026-09-15 10:08 UTC**. This exception does not relax the policy for other packages or versions. The integration preserves main's documentation moves and both manifest fields, and regenerates the config-help digest from the combined inputs. Main leaves the Flux runtime, WebSocket connection owner, and proxy helpers unchanged. Its shared HTTP capture changes require current HTTP integration evidence, including the saved Nova control. The integrated tree passes a frozen install from the published registry, formatting, targeted lint, generated config and plugin inventory updates, and a fresh compiled CLI build. Fresh tests passed: 138 WebSocket/HTTP address-policy cases, 18 Deepgram cases, and 13 HTTP capture-release/shared-agent cases. A real saved `nova-3` CLI transcription returned the expected sample text with exit 0. The initial direct test configuration excluded the capture-release file; the canonical test router then ran all 13 cases successfully. Relative to pinned main `412755bd5c`: production TypeScript +559 net, plugin manifest +13, tests/support +714. The growth implements the missing Flux streaming protocol, bounded conversion/transcription, and generic guarded WebSocket transport. Proxyline itself removes 15 net production lines by unifying the proxy dialers. Final exact-head review and CI remain required. ## Separate follow-ups - The existing HTTP dispatcher maps explicit provider proxy TLS settings to the target TLS hop. The documented settings describe the proxy hop; this WebSocket implementation applies them there. The HTTP mismatch predates this change and needs its own reproduction and repair. - The CLI's existing `--model` argument does not override an explicit media-model list. Acceptance selects Nova through saved configuration; override semantics remain a separate follow-up. - Historical [CI run 34193953000](https://github.com/openclaw/openclaw/actions/runs/34193953000) passed test types, browser-extension end-to-end checks, and the other selected children, except [Control UI shard 3](https://github.com/openclaw/openclaw/actions/runs/34193953000/job/101957740234). Its image-handoff failure also reproduced on the exact main parent `64656c24fa` with the same 67 selected files; the standalone case passed. The mocked image scenario does not invoke Deepgram. This is historical evidence of an unrelated failure, not a result or blanket CI exception for the new head. Nine missing-video-encoder errors in that probe were excluded from the recurrence evidence. The earlier [selected-tab browser failure](https://github.com/openclaw/openclaw/actions/runs/34192553828/job/101953552659) remains a separate browser-lifecycle follow-up with its cause unproven. AI-assisted repair. Co-authored-by: Safzan Pirani <5602916+safzanpirani@users.noreply.github.com> Co-authored-by: Ayaan Zaidi <hi@obviy.us>
This commit is contained in:
Ayaan Zaidi
GitHub
parent
421245a606
commit
41aec3ca92
20 changed files with 1378 additions and 40 deletions
|
|
@ -1,4 +1,4 @@
|
|||
9fb9e35f29a8bfcaf6625e4d4500758a4521c558dd884a643b66af3f8500ecee config-baseline.json
|
||||
4b5bfc6cdb729bcbd4a4452232e5594ac181128edab14201b18102050c35a1ef config-baseline.json
|
||||
10a4d786887fafa09ac97fa4039863c96b5b313646f6591faa3857ec97a7857a config-baseline.core.json
|
||||
29d7abfcdefe9fbdf5c9e2eac31fefba0d6c8225e876ae03c9fedb62f4394e53 config-baseline.channel.json
|
||||
808f7fad87549a5267cbe568c178b9419e61dfbca9f415ed4f5f9db5da511160 config-baseline.plugin.json
|
||||
71ca3291bdedaeb2ff28776df25102febb373f00b16073e1b003330ed656d7cb config-baseline.plugin.json
|
||||
|
|
|
|||
|
|
@ -89,7 +89,7 @@ Configuring a custom/local provider `baseUrl` is also the narrow network trust d
|
|||
- `request.auth`: auth strategy override. Modes: `"provider-default"` (use provider's built-in auth), `"authorization-bearer"` (with `token`), `"header"` (with `headerName`, `value`, optional `prefix`).
|
||||
- `request.proxy`: HTTP proxy override. Modes: `"env-proxy"` (use `HTTP_PROXY`/`HTTPS_PROXY` env vars), `"explicit-proxy"` (with `url`). Both modes accept an optional `tls` sub-object.
|
||||
- `request.tls`: TLS override for direct connections. Fields: `ca`, `cert`, `key`, `passphrase` (all accept SecretRef), `serverName`, `insecureSkipVerify`.
|
||||
- `request.allowPrivateNetwork`: when `true`, allow model-provider HTTP requests to private, CGNAT, or similar ranges through the provider HTTP fetch guard. Custom/local provider base URLs already trust the exact configured origin, except metadata, link-local, and local-use NAT64 (`64:ff9b:1::/48`) origins, which remain blocked without explicit opt-in. Set this to `false` to opt out of exact-origin trust. WebSocket uses the same `request` for headers/TLS but not that fetch SSRF gate. Default `false`.
|
||||
- `request.allowPrivateNetwork`: when `true`, allow guarded model-provider HTTP and WebSocket requests to private, CGNAT, or similar ranges through the shared private-network policy. Custom/local provider base URLs already trust the exact configured origin, except metadata, link-local, and local-use NAT64 (`64:ff9b:1::/48`) origins, which remain blocked without explicit opt-in. Set this to `false` to opt out of exact-origin trust. Default `false`.
|
||||
|
||||
</Accordion>
|
||||
<Accordion title="Model catalog entries">
|
||||
|
|
|
|||
|
|
@ -80,7 +80,7 @@ Each entry lists the package, distribution route, and description.
|
|||
|
||||
- **[cua-computer](/plugins/reference/cua-computer)** (`@openclaw/cua-computer`) - included in OpenClaw. Experimental CUA Driver computer control for macOS, Windows, and Linux node hosts.
|
||||
|
||||
- **[deepgram](/plugins/reference/deepgram)** (`@openclaw/deepgram-provider`) - included in OpenClaw. Adds media understanding provider support. Adds realtime transcription provider support.
|
||||
- **[deepgram](/plugins/reference/deepgram)** (`@openclaw/deepgram-provider`) - included in OpenClaw. Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition.
|
||||
|
||||
- **[device-pair](/plugins/reference/device-pair)** (`openclaw`) - included in OpenClaw. Generate setup codes and approve device pairing requests.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,5 +1,5 @@
|
|||
---
|
||||
summary: "Adds media understanding provider support. Adds realtime transcription provider support."
|
||||
summary: "Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition."
|
||||
read_when:
|
||||
- You are installing, configuring, or auditing the deepgram plugin
|
||||
title: "Deepgram plugin reference"
|
||||
|
|
@ -10,7 +10,7 @@ Run `pnpm plugins:inventory:gen` to rebuild it. Hand-written text survives only
|
|||
between the openclaw-plugin-reference:manual-start and
|
||||
openclaw-plugin-reference:manual-end comment markers. -->
|
||||
|
||||
Adds media understanding provider support. Adds realtime transcription provider support.
|
||||
Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition.
|
||||
|
||||
## Distribution
|
||||
|
||||
|
|
|
|||
|
|
@ -307,6 +307,37 @@ Register each capability inside `register(api)` alongside your existing
|
|||
model in its result. `transcribeAudio` remains available for providers
|
||||
using host-owned API-key resolution and rotation.
|
||||
|
||||
Bundled media providers can use `openProviderWebSocket(...)` from the
|
||||
private-local `openclaw/plugin-sdk/provider-http` entrypoint. Resolve
|
||||
request settings with `resolveProviderHttpRequestConfigWithOriginTrust(...)`
|
||||
first, then pass its `baseUrl`, `headers`, `dispatcherPolicy`,
|
||||
`allowPrivateNetwork`, and `trustConfiguredBaseUrlOrigin` alongside the
|
||||
WebSocket `url`.
|
||||
|
||||
Configured proxy routes retain resolved target-address checks.
|
||||
Applicable ambient HTTP(S) proxies and OpenClaw-managed proxies retain
|
||||
their existing DNS delegation; `NO_PROXY` bypasses and `ALL_PROXY` alone
|
||||
do not disable target-address checks.
|
||||
|
||||
Proxy connections use the shared Proxyline-backed Node agent. Prepared
|
||||
proxy DNS lookups and proxy TLS settings go through the `proxyConnect`
|
||||
option on `createNodeProxyAgent(...)`; target TLS remains separate.
|
||||
Proxyline owns pending proxy sockets, including cleanup before CONNECT
|
||||
completes, so providers do not need a separate proxy-agent dependency.
|
||||
|
||||
- The promise resolves after network-policy and agent preparation,
|
||||
while the returned socket is still connecting. Attach `error`,
|
||||
`close`, and `open` handlers immediately; send frames after `open`.
|
||||
- `timeoutMs` covers DNS preparation, proxy CONNECT, and the WebSocket
|
||||
handshake as one connection deadline. The connection timer stops on
|
||||
`open`; the provider owns the remaining transcription deadline.
|
||||
- `signal` cancels preparation and remains active for the socket's
|
||||
lifetime. Closing or terminating the socket also cancels a pending
|
||||
proxy connection. Release the socket in the operation's cleanup path.
|
||||
- `maxPayloadBytes` limits each incoming message and defaults to 16 MiB.
|
||||
Compression is disabled. The provider owns audio buffering, frame
|
||||
pacing, protocol parsing, and transcript-size limits.
|
||||
|
||||
```typescript
|
||||
api.registerMediaUnderstandingProvider({
|
||||
id: "acme-ai",
|
||||
|
|
|
|||
|
|
@ -11,8 +11,9 @@ Deepgram is a speech-to-text API. OpenClaw uses it for inbound audio/voice-note
|
|||
transcription through `tools.media.audio` and for Voice Call streaming STT
|
||||
through `plugins.entries.voice-call.config.streaming`.
|
||||
|
||||
Batch transcription uploads the complete audio file to Deepgram and injects
|
||||
the transcript into the reply pipeline (`{{Transcript}}` + `[Audio]` block).
|
||||
Batch transcription uploads the complete audio file to Deepgram. Flux models
|
||||
use Deepgram's one-shot WebSocket instead. Both paths inject the transcript
|
||||
into the reply pipeline (`{{Transcript}}` + `[Audio]` block).
|
||||
Voice Call streaming forwards live G.711 u-law frames over Deepgram's
|
||||
WebSocket `listen` endpoint and emits partial/final transcripts as Deepgram
|
||||
returns them.
|
||||
|
|
@ -102,6 +103,22 @@ Deepgram `/listen` request, so any Deepgram-supported param name works
|
|||
</Tab>
|
||||
</Tabs>
|
||||
|
||||
### Flux models
|
||||
|
||||
Use `flux-general-en` or `flux-general-multi` for Deepgram Flux. OpenClaw
|
||||
converts the voice note to 16 kHz mono linear16 audio with `ffmpeg`, then sends
|
||||
it to Deepgram's `/v2/listen` WebSocket endpoint.
|
||||
Set `tools.media.models[].model` to either Flux model in the getting-started
|
||||
configuration above.
|
||||
|
||||
Flux supports `eager_eot_threshold`, `eot_threshold`, `eot_timeout_ms`,
|
||||
`keyterm`, `language_hint`, `mip_opt_out`, `numerals`, `profanity_filter`,
|
||||
`redact`, and `tag` in `providerOptions.deepgram`. OpenClaw ignores batch-only
|
||||
options such as `detect_language`, `punctuate`, and `smart_format` on Flux.
|
||||
Language hints apply only to `flux-general-multi`: OpenClaw maps the model entry's
|
||||
`language` setting to `language_hint`, with an explicit provider option taking
|
||||
precedence. Both settings are ignored for the English-only `flux-general-en` model.
|
||||
|
||||
## Voice Call streaming STT
|
||||
|
||||
The bundled `deepgram` plugin also registers a realtime transcription provider
|
||||
|
|
@ -169,6 +186,10 @@ Twilio media frames can be forwarded directly.
|
|||
Output follows the same audio rules as other providers (size caps, timeouts,
|
||||
transcript injection).
|
||||
</Accordion>
|
||||
<Accordion title="Flux requires ffmpeg">
|
||||
Install `ffmpeg` with the gateway host's package manager before selecting a
|
||||
Flux model.
|
||||
</Accordion>
|
||||
</AccordionGroup>
|
||||
|
||||
## Related
|
||||
|
|
|
|||
296
extensions/deepgram/audio-flux.test.ts
Normal file
296
extensions/deepgram/audio-flux.test.ts
Normal file
|
|
@ -0,0 +1,296 @@
|
|||
import { createServer } from "node:http";
|
||||
import type { AddressInfo } from "node:net";
|
||||
import { createDeferred } from "openclaw/plugin-sdk/extension-shared";
|
||||
import type { AudioTranscriptionRequest } from "openclaw/plugin-sdk/media-understanding";
|
||||
import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { RawData, WebSocket } from "ws";
|
||||
import { WebSocketServer } from "ws";
|
||||
import { isDeepgramFluxModel } from "./audio-flux.js";
|
||||
import { transcribeDeepgramAudio } from "./audio.js";
|
||||
|
||||
const runCommandBuffered = vi.hoisted(() => vi.fn());
|
||||
const prepareWebSocket = vi.hoisted(() => vi.fn<() => Promise<void>>());
|
||||
|
||||
vi.mock("openclaw/plugin-sdk/media-runtime", async (importOriginal) => ({
|
||||
...(await importOriginal<typeof import("openclaw/plugin-sdk/media-runtime")>()),
|
||||
resolveFfmpegBin: () => "/usr/bin/ffmpeg",
|
||||
}));
|
||||
vi.mock("openclaw/plugin-sdk/process-runtime", () => ({ runCommandBuffered }));
|
||||
vi.mock("openclaw/plugin-sdk/provider-http", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("openclaw/plugin-sdk/provider-http")>();
|
||||
return {
|
||||
...actual,
|
||||
openProviderWebSocket: async (params: Parameters<typeof actual.openProviderWebSocket>[0]) => {
|
||||
await prepareWebSocket();
|
||||
return await actual.openProviderWebSocket(params);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const cleanups: Array<() => Promise<void>> = [];
|
||||
|
||||
function parseClientMessage(data: RawData): Record<string, unknown> | undefined {
|
||||
if (typeof data !== "string" && !Buffer.isBuffer(data)) {
|
||||
return undefined;
|
||||
}
|
||||
const parsed: unknown = JSON.parse(data.toString());
|
||||
return asOptionalRecord(parsed);
|
||||
}
|
||||
|
||||
async function createFluxServer(params: {
|
||||
onCloseStream: (socket: WebSocket) => void;
|
||||
onRequest?: (url: URL, headers: Record<string, string | string[] | undefined>) => void;
|
||||
}) {
|
||||
const server = createServer();
|
||||
const websocketServer = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 });
|
||||
const audioFrames: Buffer[] = [];
|
||||
server.on("upgrade", (request, socket, head) => {
|
||||
params.onRequest?.(new URL(request.url ?? "/", "http://127.0.0.1"), request.headers);
|
||||
websocketServer.handleUpgrade(request, socket, head, (client) => {
|
||||
client.on("message", (data, isBinary) => {
|
||||
if (isBinary) {
|
||||
const bytes = Array.isArray(data)
|
||||
? Buffer.concat(data)
|
||||
: Buffer.isBuffer(data)
|
||||
? data
|
||||
: Buffer.from(data);
|
||||
audioFrames.push(bytes);
|
||||
return;
|
||||
}
|
||||
if (parseClientMessage(data)?.type === "CloseStream") {
|
||||
params.onCloseStream(client);
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const port = (server.address() as AddressInfo).port;
|
||||
cleanups.push(
|
||||
async () =>
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
for (const client of websocketServer.clients) {
|
||||
client.terminate();
|
||||
}
|
||||
websocketServer.close(() => server.close((error) => (error ? reject(error) : resolve())));
|
||||
}),
|
||||
);
|
||||
return { audioFrames, baseUrl: `http://127.0.0.1:${port}/v1` };
|
||||
}
|
||||
|
||||
function fluxRequest(
|
||||
baseUrl: string,
|
||||
extra: Partial<AudioTranscriptionRequest> = {},
|
||||
): AudioTranscriptionRequest {
|
||||
return {
|
||||
buffer: Buffer.from("source audio"),
|
||||
fileName: "note.ogg",
|
||||
apiKey: "default-key",
|
||||
baseUrl,
|
||||
model: "flux-general-multi",
|
||||
timeoutMs: 5000,
|
||||
request: { allowPrivateNetwork: true },
|
||||
...extra,
|
||||
};
|
||||
}
|
||||
|
||||
function mockDecodedPcm(pcm: Buffer): void {
|
||||
runCommandBuffered.mockResolvedValueOnce({
|
||||
stdout: pcm,
|
||||
stderr: Buffer.alloc(0),
|
||||
code: 0,
|
||||
signal: null,
|
||||
killed: false,
|
||||
termination: "exit",
|
||||
});
|
||||
}
|
||||
|
||||
describe("Deepgram Flux audio", () => {
|
||||
afterEach(async () => {
|
||||
runCommandBuffered.mockReset();
|
||||
prepareWebSocket.mockReset();
|
||||
vi.restoreAllMocks();
|
||||
vi.useRealTimers();
|
||||
await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()));
|
||||
});
|
||||
|
||||
it("keeps decoding, connection preparation, and transcription within one deadline", async () => {
|
||||
const decodeStarted = createDeferred<void>();
|
||||
const releaseDecode = createDeferred<void>();
|
||||
const preparationStarted = createDeferred<void>();
|
||||
const releasePreparation = createDeferred<void>();
|
||||
const flushed = createDeferred<void>();
|
||||
const server = await createFluxServer({ onCloseStream: () => flushed.resolve() });
|
||||
runCommandBuffered.mockImplementationOnce(async () => {
|
||||
decodeStarted.resolve();
|
||||
await releaseDecode.promise;
|
||||
return {
|
||||
stdout: Buffer.alloc(10, 1),
|
||||
stderr: Buffer.alloc(0),
|
||||
code: 0,
|
||||
signal: null,
|
||||
killed: false,
|
||||
termination: "exit",
|
||||
};
|
||||
});
|
||||
prepareWebSocket.mockImplementationOnce(async () => {
|
||||
preparationStarted.resolve();
|
||||
await releasePreparation.promise;
|
||||
});
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
let failure: unknown;
|
||||
const transcription = transcribeDeepgramAudio(
|
||||
fluxRequest(server.baseUrl, { timeoutMs: 1000 }),
|
||||
).catch((error: unknown) => {
|
||||
failure = error;
|
||||
});
|
||||
|
||||
await decodeStarted.promise;
|
||||
await vi.advanceTimersByTimeAsync(200);
|
||||
releaseDecode.resolve();
|
||||
await preparationStarted.promise;
|
||||
await vi.advanceTimersByTimeAsync(300);
|
||||
releasePreparation.resolve();
|
||||
await flushed.promise;
|
||||
await vi.advanceTimersByTimeAsync(499);
|
||||
expect(failure).toBeUndefined();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(failure).toBeInstanceOf(Error);
|
||||
expect(failure).toMatchObject({ message: expect.stringContaining("timed out") });
|
||||
await transcription;
|
||||
});
|
||||
|
||||
it("routes documented Flux models only", () => {
|
||||
expect(isDeepgramFluxModel("flux-general-en")).toBe(true);
|
||||
expect(isDeepgramFluxModel(" Flux-General-Multi ")).toBe(true);
|
||||
expect(isDeepgramFluxModel("flux")).toBe(false);
|
||||
expect(isDeepgramFluxModel("reflux-general-en")).toBe(false);
|
||||
expect(isDeepgramFluxModel("nova-3")).toBe(false);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ model: "flux-general-multi", language: " en ", queryLanguage: undefined, expectedHint: "en" },
|
||||
{ model: "flux-general-multi", language: "en", queryLanguage: "fr", expectedHint: "fr" },
|
||||
{ model: "flux-general-en", language: " en ", queryLanguage: undefined, expectedHint: null },
|
||||
{ model: "flux-general-en", language: undefined, queryLanguage: "en", expectedHint: null },
|
||||
])(
|
||||
"uses valid protocol fields for $model with language=$language and query=$queryLanguage",
|
||||
async ({ model, language, queryLanguage, expectedHint }) => {
|
||||
const pcm = Buffer.alloc(6000, 1);
|
||||
mockDecodedPcm(pcm);
|
||||
let requestUrl: URL | undefined;
|
||||
let authorization: string | string[] | undefined;
|
||||
const server = await createFluxServer({
|
||||
onRequest: (url, headers) => {
|
||||
requestUrl = url;
|
||||
authorization = headers.authorization;
|
||||
},
|
||||
onCloseStream: (socket) => {
|
||||
socket.send(
|
||||
JSON.stringify({ type: "TurnInfo", event: "EndOfTurn", transcript: "life moves" }),
|
||||
);
|
||||
socket.send(
|
||||
JSON.stringify({ type: "TurnInfo", event: "EndOfTurn", transcript: "pretty fast" }),
|
||||
);
|
||||
socket.close();
|
||||
},
|
||||
});
|
||||
|
||||
const result = await transcribeDeepgramAudio(
|
||||
fluxRequest(server.baseUrl, {
|
||||
model,
|
||||
language,
|
||||
query: {
|
||||
...(queryLanguage === undefined ? {} : { language_hint: queryLanguage }),
|
||||
eot_threshold: 0.7,
|
||||
numerals: true,
|
||||
profanity_filter: true,
|
||||
smart_format: true,
|
||||
},
|
||||
request: {
|
||||
allowPrivateNetwork: true,
|
||||
auth: {
|
||||
mode: "header",
|
||||
headerName: "authorization",
|
||||
value: "Token configured-key",
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
expect(result).toEqual({ model, text: "life moves pretty fast" });
|
||||
expect(authorization).toBe("Token configured-key");
|
||||
expect(requestUrl?.pathname).toBe("/v2/listen");
|
||||
expect(requestUrl?.searchParams.get("encoding")).toBe("linear16");
|
||||
expect(requestUrl?.searchParams.get("sample_rate")).toBe("16000");
|
||||
expect(requestUrl?.searchParams.get("language_hint")).toBe(expectedHint);
|
||||
expect(requestUrl?.searchParams.get("eot_threshold")).toBe("0.7");
|
||||
expect(requestUrl?.searchParams.get("numerals")).toBe("true");
|
||||
expect(requestUrl?.searchParams.get("profanity_filter")).toBe("true");
|
||||
expect(requestUrl?.searchParams.has("smart_format")).toBe(false);
|
||||
expect(server.audioFrames.map((frame) => frame.byteLength)).toEqual([2560, 2560, 880]);
|
||||
expect(Buffer.concat(server.audioFrames)).toEqual(pcm);
|
||||
expect(runCommandBuffered).toHaveBeenCalledWith(
|
||||
expect.arrayContaining([
|
||||
"/usr/bin/ffmpeg",
|
||||
"-t",
|
||||
"1200",
|
||||
"-c:a",
|
||||
"pcm_s16le",
|
||||
"-ar",
|
||||
"16000",
|
||||
]),
|
||||
expect.objectContaining({
|
||||
maxOutputBytes: { stdout: 38_400_000, stderr: 65_536 },
|
||||
terminateOnOutputError: true,
|
||||
}),
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["null", "[]", "42"])("rejects valid non-object server JSON: %s", async (payload) => {
|
||||
mockDecodedPcm(Buffer.alloc(10, 1));
|
||||
const server = await createFluxServer({
|
||||
onCloseStream: (socket) => socket.send(payload),
|
||||
});
|
||||
await expect(transcribeDeepgramAudio(fluxRequest(server.baseUrl))).rejects.toThrow(
|
||||
"malformed JSON response",
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects retained transcript growth above the provider limit", async () => {
|
||||
mockDecodedPcm(Buffer.alloc(10, 1));
|
||||
const server = await createFluxServer({
|
||||
onCloseStream: (socket) =>
|
||||
socket.send(
|
||||
JSON.stringify({
|
||||
type: "TurnInfo",
|
||||
event: "EndOfTurn",
|
||||
transcript: "x".repeat(256 * 1024 + 1),
|
||||
}),
|
||||
),
|
||||
});
|
||||
await expect(transcribeDeepgramAudio(fluxRequest(server.baseUrl))).rejects.toThrow(
|
||||
"transcript exceeds size limit",
|
||||
);
|
||||
});
|
||||
|
||||
it("does not open a private socket without the request-policy opt-in", async () => {
|
||||
mockDecodedPcm(Buffer.alloc(10, 1));
|
||||
let opened = false;
|
||||
const server = await createFluxServer({
|
||||
onRequest: () => {
|
||||
opened = true;
|
||||
},
|
||||
onCloseStream: () => undefined,
|
||||
});
|
||||
await expect(
|
||||
transcribeDeepgramAudio(
|
||||
fluxRequest(server.baseUrl, { request: { allowPrivateNetwork: false } }),
|
||||
),
|
||||
).rejects.toThrow(/private|loopback|blocked/iu);
|
||||
expect(opened).toBe(false);
|
||||
});
|
||||
});
|
||||
331
extensions/deepgram/audio-flux.ts
Normal file
331
extensions/deepgram/audio-flux.ts
Normal file
|
|
@ -0,0 +1,331 @@
|
|||
// Deepgram Flux voice-note transcription uses the provider's one-shot WebSocket protocol.
|
||||
import {
|
||||
MEDIA_FFMPEG_MAX_AUDIO_DURATION_SECS,
|
||||
resolveFfmpegBin,
|
||||
} from "openclaw/plugin-sdk/media-runtime";
|
||||
import type {
|
||||
AudioTranscriptionRequest,
|
||||
AudioTranscriptionResult,
|
||||
} from "openclaw/plugin-sdk/media-understanding";
|
||||
import { runCommandBuffered } from "openclaw/plugin-sdk/process-runtime";
|
||||
import {
|
||||
createProviderOperationDeadline,
|
||||
createProviderOperationTimeoutResolver,
|
||||
openProviderWebSocket,
|
||||
requireTranscriptionText,
|
||||
} from "openclaw/plugin-sdk/provider-http";
|
||||
import { asOptionalRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
|
||||
const DEEPGRAM_FLUX_SAMPLE_RATE = 16_000;
|
||||
// Deepgram recommends 80 ms chunks. 16 kHz mono linear16 contains 32 bytes per millisecond.
|
||||
const DEEPGRAM_FLUX_AUDIO_CHUNK_BYTES = 2_560;
|
||||
const DEEPGRAM_FLUX_MAX_MESSAGE_BYTES = 1024 * 1024;
|
||||
const DEEPGRAM_FLUX_MAX_TRANSCRIPT_BYTES = 256 * 1024;
|
||||
const DEEPGRAM_FLUX_MAX_PCM_BYTES =
|
||||
DEEPGRAM_FLUX_SAMPLE_RATE * 2 * MEDIA_FFMPEG_MAX_AUDIO_DURATION_SECS;
|
||||
const DEEPGRAM_FLUX_QUERY_KEYS = new Set([
|
||||
"eager_eot_threshold",
|
||||
"eot_threshold",
|
||||
"eot_timeout_ms",
|
||||
"keyterm",
|
||||
"language_hint",
|
||||
"mip_opt_out",
|
||||
"numerals",
|
||||
"profanity_filter",
|
||||
"redact",
|
||||
"tag",
|
||||
]);
|
||||
|
||||
type DeepgramFluxRequestConfig = {
|
||||
allowPrivateNetwork: boolean;
|
||||
baseUrl: string;
|
||||
dispatcherPolicy?: Parameters<typeof openProviderWebSocket>[0]["dispatcherPolicy"];
|
||||
headers: Headers;
|
||||
trustConfiguredBaseUrlOrigin: boolean;
|
||||
};
|
||||
|
||||
export function isDeepgramFluxModel(model?: string): boolean {
|
||||
return model?.trim().toLowerCase().startsWith("flux-") ?? false;
|
||||
}
|
||||
|
||||
function buildDeepgramFluxUrl(params: {
|
||||
baseUrl: string;
|
||||
language?: string;
|
||||
model: string;
|
||||
query?: Record<string, string | number | boolean | undefined>;
|
||||
}): string {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(params.baseUrl);
|
||||
} catch {
|
||||
throw new Error("Invalid Deepgram baseUrl: value is not a valid URL");
|
||||
}
|
||||
if (url.protocol === "http:") {
|
||||
url.protocol = "ws:";
|
||||
} else if (url.protocol === "https:") {
|
||||
url.protocol = "wss:";
|
||||
} else if (url.protocol !== "ws:" && url.protocol !== "wss:") {
|
||||
throw new Error(
|
||||
`Invalid Deepgram baseUrl: unsupported scheme "${url.protocol}" (expected http, https, ws, or wss)`,
|
||||
);
|
||||
}
|
||||
const basePath = url.pathname.replace(/\/+$/u, "");
|
||||
url.pathname = `${basePath ? basePath.replace(/\/v1$/u, "/v2") : "/v2"}/listen`;
|
||||
url.search = "";
|
||||
url.searchParams.set("model", params.model);
|
||||
url.searchParams.set("encoding", "linear16");
|
||||
url.searchParams.set("sample_rate", String(DEEPGRAM_FLUX_SAMPLE_RATE));
|
||||
const query = { ...params.query };
|
||||
// Deepgram's v2 contract permits language hints only on its multilingual model.
|
||||
query.language_hint =
|
||||
params.model === "flux-general-multi"
|
||||
? (query.language_hint ?? (params.language?.trim() || undefined))
|
||||
: undefined;
|
||||
for (const [key, value] of Object.entries(query)) {
|
||||
if (value !== undefined && DEEPGRAM_FLUX_QUERY_KEYS.has(key)) {
|
||||
url.searchParams.set(key, String(value));
|
||||
}
|
||||
}
|
||||
return url.toString();
|
||||
}
|
||||
|
||||
function readFluxEvent(data: Buffer | ArrayBuffer | Buffer[]): Record<string, unknown> {
|
||||
const bytes = Array.isArray(data)
|
||||
? Buffer.concat(data)
|
||||
: Buffer.isBuffer(data)
|
||||
? data
|
||||
: Buffer.from(data);
|
||||
let parsed: unknown;
|
||||
try {
|
||||
parsed = JSON.parse(bytes.toString("utf8"));
|
||||
} catch {
|
||||
throw new Error("Audio transcription failed: malformed JSON response");
|
||||
}
|
||||
const event = asOptionalRecord(parsed);
|
||||
if (!event) {
|
||||
throw new Error("Audio transcription failed: malformed JSON response");
|
||||
}
|
||||
return event;
|
||||
}
|
||||
|
||||
function readFluxErrorDetail(event: Record<string, unknown>): string {
|
||||
const nested = asOptionalRecord(event.error);
|
||||
for (const value of [event.description, event.message, nested?.message]) {
|
||||
if (typeof value === "string" && value.trim()) {
|
||||
return value.trim();
|
||||
}
|
||||
}
|
||||
return "Deepgram Flux transcription error";
|
||||
}
|
||||
|
||||
type DeepgramFluxSocket = Awaited<ReturnType<typeof openProviderWebSocket>>;
|
||||
|
||||
function sendSocketFrame(socket: DeepgramFluxSocket, data: Buffer | string): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
socket.send(data, (error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
|
||||
async function decodeDeepgramFluxAudio(params: {
|
||||
buffer: Buffer;
|
||||
signal?: AbortSignal;
|
||||
timeoutMs: number;
|
||||
}): Promise<Buffer> {
|
||||
const result = await runCommandBuffered(
|
||||
[
|
||||
resolveFfmpegBin(),
|
||||
"-hide_banner",
|
||||
"-loglevel",
|
||||
"error",
|
||||
"-i",
|
||||
"pipe:0",
|
||||
"-t",
|
||||
String(MEDIA_FFMPEG_MAX_AUDIO_DURATION_SECS),
|
||||
"-vn",
|
||||
"-sn",
|
||||
"-dn",
|
||||
"-c:a",
|
||||
"pcm_s16le",
|
||||
"-ar",
|
||||
String(DEEPGRAM_FLUX_SAMPLE_RATE),
|
||||
"-ac",
|
||||
"1",
|
||||
"-f",
|
||||
"s16le",
|
||||
"pipe:1",
|
||||
],
|
||||
{
|
||||
input: params.buffer,
|
||||
maxOutputBytes: { stdout: DEEPGRAM_FLUX_MAX_PCM_BYTES, stderr: 64 * 1024 },
|
||||
signal: params.signal,
|
||||
terminateOnOutputError: true,
|
||||
timeoutMs: params.timeoutMs,
|
||||
},
|
||||
);
|
||||
if (result.termination === "exit" && result.code === 0) {
|
||||
return result.stdout;
|
||||
}
|
||||
if (result.termination === "output-limit") {
|
||||
throw new Error("Audio transcription failed: decoded audio exceeds size limit");
|
||||
}
|
||||
const detail = result.stderr.toString("utf8").trim();
|
||||
throw new Error(
|
||||
`Audio transcription failed: ffmpeg ${result.termination}${detail ? `: ${detail}` : ""}`,
|
||||
{ cause: result.error },
|
||||
);
|
||||
}
|
||||
|
||||
export async function transcribeDeepgramFluxAudio(params: {
|
||||
request: AudioTranscriptionRequest;
|
||||
requestConfig: DeepgramFluxRequestConfig;
|
||||
model: string;
|
||||
}): Promise<AudioTranscriptionResult> {
|
||||
const deadline = createProviderOperationDeadline({
|
||||
timeoutMs: params.request.timeoutMs,
|
||||
label: "Deepgram Flux transcription",
|
||||
});
|
||||
const resolveTimeoutMs = createProviderOperationTimeoutResolver({
|
||||
deadline,
|
||||
defaultTimeoutMs: params.request.timeoutMs,
|
||||
});
|
||||
const pcm = await decodeDeepgramFluxAudio({
|
||||
buffer: params.request.buffer,
|
||||
signal: params.request.signal,
|
||||
timeoutMs: resolveTimeoutMs(),
|
||||
});
|
||||
if (pcm.byteLength === 0) {
|
||||
throw new Error("Audio transcription failed: decoded audio is empty");
|
||||
}
|
||||
|
||||
const url = buildDeepgramFluxUrl({
|
||||
baseUrl: params.requestConfig.baseUrl,
|
||||
language: params.request.language,
|
||||
model: params.model,
|
||||
query: params.request.query,
|
||||
});
|
||||
const timeoutMs = resolveTimeoutMs();
|
||||
const socket = await openProviderWebSocket({
|
||||
allowPrivateNetwork: params.requestConfig.allowPrivateNetwork,
|
||||
baseUrl: params.requestConfig.baseUrl,
|
||||
dispatcherPolicy: params.requestConfig.dispatcherPolicy,
|
||||
headers: params.requestConfig.headers,
|
||||
maxPayloadBytes: DEEPGRAM_FLUX_MAX_MESSAGE_BYTES,
|
||||
signal: params.request.signal,
|
||||
timeoutMs,
|
||||
trustConfiguredBaseUrlOrigin: params.requestConfig.trustConfiguredBaseUrlOrigin,
|
||||
url,
|
||||
});
|
||||
|
||||
const transcript = await new Promise<string>((resolve, reject) => {
|
||||
const finalizedTurns: string[] = [];
|
||||
let finalizedTranscriptBytes = 0;
|
||||
let lastPartial = "";
|
||||
let settled = false;
|
||||
let closeStreamSent = false;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
|
||||
const settle = (outcome: { error?: Error; text?: string }) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
timer = undefined;
|
||||
if (outcome.error) {
|
||||
reject(outcome.error);
|
||||
} else {
|
||||
resolve(outcome.text ?? "");
|
||||
}
|
||||
};
|
||||
socket.on("open", () => {
|
||||
void (async () => {
|
||||
try {
|
||||
for (let offset = 0; offset < pcm.byteLength; offset += DEEPGRAM_FLUX_AUDIO_CHUNK_BYTES) {
|
||||
await sendSocketFrame(
|
||||
socket,
|
||||
pcm.subarray(offset, offset + DEEPGRAM_FLUX_AUDIO_CHUNK_BYTES),
|
||||
);
|
||||
}
|
||||
await sendSocketFrame(socket, JSON.stringify({ type: "CloseStream" }));
|
||||
closeStreamSent = true;
|
||||
} catch (error) {
|
||||
settle({
|
||||
error: new Error(
|
||||
`Audio transcription failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
),
|
||||
});
|
||||
}
|
||||
})();
|
||||
});
|
||||
|
||||
socket.on("message", (data) => {
|
||||
try {
|
||||
const event = readFluxEvent(data);
|
||||
if (event.type === "Error" || event.type === "error") {
|
||||
settle({
|
||||
error: new Error(`Audio transcription failed: ${readFluxErrorDetail(event)}`),
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (event.type !== "TurnInfo" || typeof event.transcript !== "string") {
|
||||
return;
|
||||
}
|
||||
if (event.event === "EndOfTurn") {
|
||||
finalizedTranscriptBytes += Buffer.byteLength(event.transcript, "utf8");
|
||||
finalizedTurns.push(event.transcript);
|
||||
lastPartial = "";
|
||||
} else {
|
||||
lastPartial = event.transcript;
|
||||
}
|
||||
if (
|
||||
finalizedTranscriptBytes + Buffer.byteLength(lastPartial, "utf8") >
|
||||
DEEPGRAM_FLUX_MAX_TRANSCRIPT_BYTES
|
||||
) {
|
||||
settle({ error: new Error("Audio transcription failed: transcript exceeds size limit") });
|
||||
}
|
||||
} catch (error) {
|
||||
settle({ error: error instanceof Error ? error : new Error(String(error)) });
|
||||
}
|
||||
});
|
||||
|
||||
socket.on("error", (error) => {
|
||||
settle({ error: new Error(`Audio transcription failed: ${error.message}`) });
|
||||
});
|
||||
|
||||
socket.on("close", (code, reason) => {
|
||||
if (!closeStreamSent) {
|
||||
settle({ error: new Error("Audio transcription failed: Flux closed before flush") });
|
||||
return;
|
||||
}
|
||||
if (code !== 1000 && code !== 1005) {
|
||||
const detail = reason.toString().trim();
|
||||
settle({
|
||||
error: new Error(
|
||||
`Audio transcription failed: Flux closed abnormally (code ${code}${detail ? `: ${detail}` : ""})`,
|
||||
),
|
||||
});
|
||||
return;
|
||||
}
|
||||
settle({
|
||||
text: [...finalizedTurns, lastPartial]
|
||||
.map((part) => part.trim())
|
||||
.filter(Boolean)
|
||||
.join(" "),
|
||||
});
|
||||
});
|
||||
// Connection preparation consumed part of the original operation budget.
|
||||
// Register handlers first so an exhausted budget still closes the socket safely.
|
||||
timer = setTimeout(
|
||||
() => settle({ error: new Error("Deepgram Flux transcription timed out") }),
|
||||
resolveTimeoutMs(),
|
||||
);
|
||||
}).finally(() => socket.terminate());
|
||||
|
||||
return {
|
||||
text: requireTranscriptionText(
|
||||
transcript || undefined,
|
||||
"Audio transcription response missing transcript",
|
||||
),
|
||||
model: params.model,
|
||||
};
|
||||
}
|
||||
|
|
@ -72,7 +72,9 @@ describeLive("deepgram live", () => {
|
|||
baseUrl: DEEPGRAM_BASE_URL,
|
||||
timeoutMs: 20000,
|
||||
});
|
||||
expect(result.text.trim().length).toBeGreaterThan(0);
|
||||
expect(result.text.toLowerCase().replaceAll(/[^a-z0-9]/gu, "")).toContain(
|
||||
"lifemovesprettyfast",
|
||||
);
|
||||
}, 30000);
|
||||
|
||||
it("streams realtime STT through the registered transcription provider", async () => {
|
||||
|
|
|
|||
|
|
@ -45,25 +45,30 @@ export async function transcribeDeepgramAudio(
|
|||
assertOkOrThrowHttpError,
|
||||
postTranscriptionRequest,
|
||||
readProviderJsonObjectResponse,
|
||||
resolveProviderHttpRequestConfig,
|
||||
resolveProviderHttpRequestConfigWithOriginTrust,
|
||||
requireTranscriptionText,
|
||||
} = await import("openclaw/plugin-sdk/provider-http");
|
||||
const fetchFn = params.fetchFn ?? fetch;
|
||||
const { isDeepgramFluxModel, transcribeDeepgramFluxAudio } = await import("./audio-flux.js");
|
||||
const model = resolveModel(params.model);
|
||||
const { baseUrl, allowPrivateNetwork, headers, dispatcherPolicy } =
|
||||
resolveProviderHttpRequestConfig({
|
||||
baseUrl: params.baseUrl,
|
||||
defaultBaseUrl: DEFAULT_DEEPGRAM_AUDIO_BASE_URL,
|
||||
headers: params.headers,
|
||||
request: params.request,
|
||||
defaultHeaders: {
|
||||
authorization: `Token ${params.apiKey}`,
|
||||
"content-type": params.mime ?? "application/octet-stream",
|
||||
},
|
||||
provider: "deepgram",
|
||||
capability: "audio",
|
||||
transport: "media-understanding",
|
||||
});
|
||||
const flux = isDeepgramFluxModel(model);
|
||||
const requestConfig = resolveProviderHttpRequestConfigWithOriginTrust({
|
||||
baseUrl: params.baseUrl,
|
||||
defaultBaseUrl: DEFAULT_DEEPGRAM_AUDIO_BASE_URL,
|
||||
headers: params.headers,
|
||||
request: params.request,
|
||||
defaultHeaders: {
|
||||
authorization: `Token ${params.apiKey}`,
|
||||
...(flux ? {} : { "content-type": params.mime ?? "application/octet-stream" }),
|
||||
},
|
||||
provider: "deepgram",
|
||||
capability: "audio",
|
||||
transport: "media-understanding",
|
||||
});
|
||||
if (flux) {
|
||||
return await transcribeDeepgramFluxAudio({ request: params, requestConfig, model });
|
||||
}
|
||||
const fetchFn = params.fetchFn ?? fetch;
|
||||
const { baseUrl, allowPrivateNetwork, headers, dispatcherPolicy } = requestConfig;
|
||||
|
||||
const url = new URL(`${baseUrl}/listen`);
|
||||
url.searchParams.set("model", model);
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
{
|
||||
"id": "deepgram",
|
||||
"categories": ["models", "voice", "media"],
|
||||
"categories": [
|
||||
"models",
|
||||
"voice",
|
||||
"media"
|
||||
],
|
||||
"capabilityCatalogEntry": "./capability-catalog.ts",
|
||||
"activation": {
|
||||
"onStartup": false
|
||||
|
|
@ -10,17 +14,25 @@
|
|||
"providers": [
|
||||
{
|
||||
"id": "deepgram",
|
||||
"envVars": ["DEEPGRAM_API_KEY"]
|
||||
"envVars": [
|
||||
"DEEPGRAM_API_KEY"
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
"contracts": {
|
||||
"mediaUnderstandingProviders": ["deepgram"],
|
||||
"realtimeTranscriptionProviders": ["deepgram"]
|
||||
"mediaUnderstandingProviders": [
|
||||
"deepgram"
|
||||
],
|
||||
"realtimeTranscriptionProviders": [
|
||||
"deepgram"
|
||||
]
|
||||
},
|
||||
"mediaUnderstandingProviderMetadata": {
|
||||
"deepgram": {
|
||||
"capabilities": ["audio"],
|
||||
"capabilities": [
|
||||
"audio"
|
||||
],
|
||||
"defaultModels": {
|
||||
"audio": "nova-3"
|
||||
},
|
||||
|
|
@ -33,5 +45,6 @@
|
|||
"type": "object",
|
||||
"additionalProperties": false,
|
||||
"properties": {}
|
||||
}
|
||||
},
|
||||
"description": "Deepgram audio transcription with Nova and Flux models, plus realtime speech recognition."
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2159,7 +2159,7 @@
|
|||
"@mozilla/readability": "0.6.0",
|
||||
"@openclaw/ai": "workspace:*",
|
||||
"@openclaw/fs-safe": "0.8.5",
|
||||
"@openclaw/proxyline": "0.3.11",
|
||||
"@openclaw/proxyline": "0.3.12",
|
||||
"@silvia-odwyer/photon-node": "0.3.4",
|
||||
"@trycua/cua-driver": "0.22.2",
|
||||
"acorn": "8.18.0",
|
||||
|
|
|
|||
10
pnpm-lock.yaml
generated
10
pnpm-lock.yaml
generated
|
|
@ -205,8 +205,8 @@ importers:
|
|||
specifier: 0.8.5
|
||||
version: 0.8.5
|
||||
'@openclaw/proxyline':
|
||||
specifier: 0.3.11
|
||||
version: 0.3.11(undici@8.10.2)
|
||||
specifier: 0.3.12
|
||||
version: 0.3.12(undici@8.10.2)
|
||||
'@silvia-odwyer/photon-node':
|
||||
specifier: 0.3.4
|
||||
version: 0.3.4
|
||||
|
|
@ -4211,8 +4211,8 @@ packages:
|
|||
node-pty:
|
||||
optional: true
|
||||
|
||||
'@openclaw/proxyline@0.3.11':
|
||||
resolution: {integrity: sha512-XDlrrQE0fapfTpqCBK9WB+U4W+/0lsrRIXI2nZ74W/lWMF+X4T5t2dfj8Ms250Th4qjnNi6WuSX5OZJqyLoZ3g==}
|
||||
'@openclaw/proxyline@0.3.12':
|
||||
resolution: {integrity: sha512-KPUvtLgORF0NWxiuUPwHBuxYhUsFgxnOm8+XkDR0on236lnfMF44A5BnOvBkeDV8wrBaGHs/rQ8kt1tB4WNhmw==}
|
||||
engines: {node: '>=22.19.0'}
|
||||
peerDependencies:
|
||||
undici: '>=8.5.0 <9'
|
||||
|
|
@ -11543,7 +11543,7 @@ snapshots:
|
|||
dependencies:
|
||||
ghostty-web: 0.4.0
|
||||
|
||||
'@openclaw/proxyline@0.3.11(undici@8.10.2)':
|
||||
'@openclaw/proxyline@0.3.12(undici@8.10.2)':
|
||||
dependencies:
|
||||
undici: 8.10.2
|
||||
|
||||
|
|
|
|||
|
|
@ -10,8 +10,8 @@ minimumReleaseAgeStrict: true
|
|||
|
||||
# Trusted Codex runtimes are outside the dependency cooldown.
|
||||
minimumReleaseAgeExclude:
|
||||
# Reviewed Proxyline Bun runtime fix; remove after 2026-09-13 20:51 UTC.
|
||||
- "@openclaw/proxyline@0.3.11"
|
||||
# Reviewed proxy DNS/TLS controls; remove after 2026-09-15 10:08 UTC.
|
||||
- "@openclaw/proxyline@0.3.12"
|
||||
# GHSA-vp8m-p9jh-q5pm / GHSA-w293-vg96-wgc3 security fixes; remove after 2026-09-12.
|
||||
- "undici@8.10.2"
|
||||
- "undici@7.29.1"
|
||||
|
|
|
|||
|
|
@ -176,7 +176,7 @@ function resolveGuardedFetchMode(params: GuardedFetchOptions): GuardedFetchMode
|
|||
return GUARDED_FETCH_MODE.STRICT;
|
||||
}
|
||||
|
||||
function isManagedProxyActive(): boolean {
|
||||
export function isManagedProxyActive(): boolean {
|
||||
return process.env["OPENCLAW_PROXY_ACTIVE"] === "1";
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ type ProxylineCreateAmbientNodeProxyAgent =
|
|||
type ProxylineAgentOptions = NonNullable<Parameters<ProxylineCreateAmbientNodeProxyAgent>[0]>;
|
||||
type ProxylineEnvSnapshot = NonNullable<ProxylineAgentOptions["env"]>;
|
||||
type ProxylineTlsOptions = ProxylineAgentOptions["proxyTls"];
|
||||
type ProxylineProxyConnectOptions = import("@openclaw/proxyline").ProxyConnectOptions;
|
||||
type NodeProxyAgentOptions = HttpAgentOptions & HttpsAgentOptions;
|
||||
type NodeProxyAgentWithOptions = HttpAgent & {
|
||||
keepAlive: boolean;
|
||||
|
|
@ -36,12 +37,14 @@ export type CreateNodeProxyAgentOptions =
|
|||
targetUrl: string | URL;
|
||||
protocol?: NodeProxyProtocol;
|
||||
agentOptions?: NodeProxyAgentOptions;
|
||||
proxyConnect?: ProxylineProxyConnectOptions;
|
||||
}
|
||||
| {
|
||||
mode: "explicit";
|
||||
proxyUrl: string | URL;
|
||||
protocol?: NodeProxyProtocol;
|
||||
agentOptions?: NodeProxyAgentOptions;
|
||||
proxyConnect?: ProxylineProxyConnectOptions;
|
||||
};
|
||||
|
||||
function proxyUrlWithDefaultScheme(proxyUrl: string, protocol: NodeProxyProtocol): URL {
|
||||
|
|
@ -163,16 +166,19 @@ function createFixedNodeProxyAgent(
|
|||
protocol?: NodeProxyProtocol;
|
||||
proxyTls?: ProxylineTlsOptions;
|
||||
agentOptions?: NodeProxyAgentOptions;
|
||||
proxyConnect?: ProxylineProxyConnectOptions;
|
||||
} = {},
|
||||
): HttpAgent {
|
||||
const parsedProxyUrl =
|
||||
proxyUrl instanceof URL
|
||||
? proxyUrl
|
||||
: proxyUrlWithDefaultScheme(proxyUrl, options.protocol ?? "https");
|
||||
const proxyConnect = options.proxyConnect;
|
||||
const agent = loadCreateAmbientNodeProxyAgent()({
|
||||
env: fixedProxyEnv(parsedProxyUrl),
|
||||
protocol: options.protocol ?? "https",
|
||||
...(options.proxyTls !== undefined ? { proxyTls: options.proxyTls } : {}),
|
||||
...(proxyConnect !== undefined ? { resolveProxyConnectOptions: () => proxyConnect } : {}),
|
||||
});
|
||||
if (agent === undefined) {
|
||||
throw new Error(`${UNSUPPORTED_PROXY_PROTOCOL_MESSAGE} Got ${parsedProxyUrl.protocol}`);
|
||||
|
|
@ -194,11 +200,13 @@ export function createNodeProxyAgent(options: CreateNodeProxyAgentOptions): Http
|
|||
return createFixedNodeProxyAgent(options.proxyUrl, {
|
||||
protocol: options.protocol,
|
||||
agentOptions: options.agentOptions,
|
||||
proxyConnect: options.proxyConnect,
|
||||
});
|
||||
}
|
||||
return createEnvNodeProxyAgentForTarget(options.targetUrl, {
|
||||
protocol: options.protocol,
|
||||
agentOptions: options.agentOptions,
|
||||
proxyConnect: options.proxyConnect,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -207,6 +215,7 @@ function createEnvNodeProxyAgentForTarget(
|
|||
options: {
|
||||
protocol?: NodeProxyProtocol;
|
||||
agentOptions?: NodeProxyAgentOptions;
|
||||
proxyConnect?: ProxylineProxyConnectOptions;
|
||||
} = {},
|
||||
): HttpAgent | undefined {
|
||||
const target = resolveEnvNodeProxyTarget(targetUrl);
|
||||
|
|
@ -217,6 +226,7 @@ function createEnvNodeProxyAgentForTarget(
|
|||
protocol: options.protocol ?? target.protocol,
|
||||
proxyTls: resolveActiveManagedProxyTlsOptions({ proxyUrl: target.proxyUrl.href }),
|
||||
agentOptions: options.agentOptions,
|
||||
proxyConnect: options.proxyConnect,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
|
|||
402
src/infra/net/provider-websocket.test.ts
Normal file
402
src/infra/net/provider-websocket.test.ts
Normal file
|
|
@ -0,0 +1,402 @@
|
|||
import { once } from "node:events";
|
||||
import { createServer as createHttpServer } from "node:http";
|
||||
import { createServer as createHttpsServer } from "node:https";
|
||||
import net, { type AddressInfo } from "node:net";
|
||||
import type { Duplex } from "node:stream";
|
||||
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { WebSocketServer } from "ws";
|
||||
import { createDeferred, withTestTimeout } from "../../../test/helpers/promise.js";
|
||||
import { TEST_TLS_CERT_PEM, TEST_TLS_KEY_PEM } from "../../../test/helpers/tls-fixture.js";
|
||||
import { openProviderWebSocket } from "./provider-websocket.js";
|
||||
import * as ssrf from "./ssrf.js";
|
||||
|
||||
const cleanups: Array<() => Promise<void>> = [];
|
||||
|
||||
async function createStalledHandshakeServer(event: "upgrade" | "connect") {
|
||||
const server = createHttpServer();
|
||||
const received = createDeferred<Duplex>();
|
||||
const connections = new Set<net.Socket>();
|
||||
let connectionCount = 0;
|
||||
server.on("connection", (socket) => {
|
||||
connectionCount += 1;
|
||||
connections.add(socket);
|
||||
socket.once("close", () => connections.delete(socket));
|
||||
});
|
||||
server.on(event, (_request, socket) => {
|
||||
// HTTP hands upgraded sockets to us with allowHalfOpen enabled.
|
||||
// Complete our half only after the client ends its connection.
|
||||
socket.once("end", () => socket.end());
|
||||
socket.resume();
|
||||
received.resolve(socket);
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const port = (server.address() as AddressInfo).port;
|
||||
cleanups.push(async () => {
|
||||
for (const socket of connections) {
|
||||
socket.destroy();
|
||||
}
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
});
|
||||
return {
|
||||
url: `http://127.0.0.1:${port}`,
|
||||
received: received.promise,
|
||||
connectionCount: () => connectionCount,
|
||||
};
|
||||
}
|
||||
|
||||
function configureProxyEnvironment(proxyUrl: string, noProxy = "") {
|
||||
vi.stubEnv("OPENCLAW_PROXY_ACTIVE", undefined);
|
||||
for (const name of [
|
||||
"HTTP_PROXY",
|
||||
"HTTPS_PROXY",
|
||||
"ALL_PROXY",
|
||||
"http_proxy",
|
||||
"https_proxy",
|
||||
"all_proxy",
|
||||
]) {
|
||||
vi.stubEnv(name, proxyUrl);
|
||||
}
|
||||
vi.stubEnv("NO_PROXY", noProxy);
|
||||
vi.stubEnv("no_proxy", noProxy);
|
||||
}
|
||||
|
||||
function resolveProviderToLoopback() {
|
||||
const resolveHostname = ssrf.resolvePinnedHostnameWithPolicy;
|
||||
vi.spyOn(ssrf, "resolvePinnedHostnameWithPolicy").mockImplementation((hostname, params) =>
|
||||
resolveHostname(
|
||||
hostname,
|
||||
hostname === "provider.example"
|
||||
? { ...params, lookupFn: async () => [{ address: "127.0.0.1", family: 4 }] }
|
||||
: params,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
async function createLocalWebSocketServer(options: { tls?: boolean } = {}) {
|
||||
const server = options.tls
|
||||
? createHttpsServer({ cert: TEST_TLS_CERT_PEM, key: TEST_TLS_KEY_PEM })
|
||||
: createHttpServer();
|
||||
const websocketServer = new WebSocketServer({ noServer: true, maxPayload: 1024 * 1024 });
|
||||
const requestHeaders: Array<Record<string, string | string[] | undefined>> = [];
|
||||
server.on("upgrade", (request, socket, head) => {
|
||||
requestHeaders.push(request.headers);
|
||||
websocketServer.handleUpgrade(request, socket, head, (client) => {
|
||||
websocketServer.emit("connection", client, request);
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const port = (server.address() as AddressInfo).port;
|
||||
cleanups.push(
|
||||
async () =>
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
for (const client of websocketServer.clients) {
|
||||
client.terminate();
|
||||
}
|
||||
websocketServer.close(() => server.close((error) => (error ? reject(error) : resolve())));
|
||||
}),
|
||||
);
|
||||
return {
|
||||
requestHeaders,
|
||||
url: `${options.tls ? "wss" : "ws"}://127.0.0.1:${port}/listen`,
|
||||
};
|
||||
}
|
||||
|
||||
async function createConnectProxy(proxyHostname = "127.0.0.1") {
|
||||
const server = createHttpServer();
|
||||
let connectCount = 0;
|
||||
server.on("connect", (request, clientSocket, head) => {
|
||||
connectCount += 1;
|
||||
const [hostname, rawPort] = (request.url ?? "").split(":");
|
||||
const targetSocket = net.connect(Number(rawPort), hostname, () => {
|
||||
clientSocket.write("HTTP/1.1 200 Connection Established\r\n\r\n");
|
||||
if (head.byteLength > 0) {
|
||||
targetSocket.write(head);
|
||||
}
|
||||
targetSocket.pipe(clientSocket);
|
||||
clientSocket.pipe(targetSocket);
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve) => {
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const port = (server.address() as AddressInfo).port;
|
||||
cleanups.push(
|
||||
async () =>
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
}),
|
||||
);
|
||||
return { connectCount: () => connectCount, url: `http://${proxyHostname}:${port}` };
|
||||
}
|
||||
|
||||
describe("openProviderWebSocket", () => {
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
vi.unstubAllEnvs();
|
||||
vi.useRealTimers();
|
||||
await Promise.all(cleanups.splice(0).map((cleanup) => cleanup()));
|
||||
});
|
||||
|
||||
it.each(["explicit-proxy", "env-proxy"] as const)(
|
||||
"rejects private target DNS before connecting through configured %s",
|
||||
async (mode) => {
|
||||
const proxy = await createStalledHandshakeServer("connect");
|
||||
configureProxyEnvironment(proxy.url);
|
||||
resolveProviderToLoopback();
|
||||
const opening = openProviderWebSocket({
|
||||
allowPrivateNetwork: false,
|
||||
baseUrl: "wss://provider.example/audio",
|
||||
dispatcherPolicy:
|
||||
mode === "explicit-proxy"
|
||||
? { mode, proxyUrl: proxy.url, allowPrivateProxy: true }
|
||||
: { mode },
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: "wss://provider.example/audio",
|
||||
}).then((socket) => {
|
||||
socket.on("error", () => {});
|
||||
socket.terminate();
|
||||
return socket;
|
||||
});
|
||||
await expect(opening).rejects.toThrow(/private|loopback|blocked/iu);
|
||||
expect(proxy.connectionCount()).toBe(0);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
{ name: "an ambient proxy", managed: false, dispatcherPolicy: undefined },
|
||||
{
|
||||
name: "a managed proxy with configured env-proxy policy",
|
||||
managed: true,
|
||||
dispatcherPolicy: { mode: "env-proxy" },
|
||||
},
|
||||
{
|
||||
name: "a managed proxy with configured direct policy",
|
||||
managed: true,
|
||||
dispatcherPolicy: { mode: "direct" },
|
||||
},
|
||||
] as const)("leaves target DNS to $name", async ({ managed, dispatcherPolicy }) => {
|
||||
const proxy = await createStalledHandshakeServer("connect");
|
||||
configureProxyEnvironment(proxy.url);
|
||||
if (managed) {
|
||||
vi.stubEnv("OPENCLAW_PROXY_ACTIVE", "1");
|
||||
}
|
||||
resolveProviderToLoopback();
|
||||
const socket = await openProviderWebSocket({
|
||||
allowPrivateNetwork: false,
|
||||
baseUrl: "wss://provider.example/audio",
|
||||
dispatcherPolicy,
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: "wss://provider.example/audio",
|
||||
});
|
||||
socket.on("error", () => {});
|
||||
const peer = await withTestTimeout(proxy.received, 2000, "Proxy did not receive CONNECT");
|
||||
const closed = once(peer, "close", { signal: AbortSignal.timeout(2000) });
|
||||
socket.terminate();
|
||||
await closed;
|
||||
expect(proxy.connectionCount()).toBe(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: "ambient", managed: false, dispatcherPolicy: undefined },
|
||||
{ name: "managed", managed: true, dispatcherPolicy: { mode: "env-proxy" } },
|
||||
] as const)(
|
||||
"checks target DNS when NO_PROXY bypasses a $name proxy",
|
||||
async ({ managed, dispatcherPolicy }) => {
|
||||
const proxy = await createStalledHandshakeServer("connect");
|
||||
configureProxyEnvironment(proxy.url, "provider.example");
|
||||
if (managed) {
|
||||
vi.stubEnv("OPENCLAW_PROXY_ACTIVE", "1");
|
||||
}
|
||||
resolveProviderToLoopback();
|
||||
await expect(
|
||||
openProviderWebSocket({
|
||||
allowPrivateNetwork: false,
|
||||
baseUrl: "wss://provider.example/audio",
|
||||
dispatcherPolicy,
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: "wss://provider.example/audio",
|
||||
}),
|
||||
).rejects.toThrow(/private|loopback|blocked/iu);
|
||||
expect(proxy.connectionCount()).toBe(0);
|
||||
},
|
||||
);
|
||||
|
||||
it("checks target DNS when only ALL_PROXY is configured", async () => {
|
||||
const proxy = await createStalledHandshakeServer("connect");
|
||||
configureProxyEnvironment(proxy.url);
|
||||
for (const name of ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]) {
|
||||
vi.stubEnv(name, undefined);
|
||||
}
|
||||
resolveProviderToLoopback();
|
||||
const opening = openProviderWebSocket({
|
||||
allowPrivateNetwork: false,
|
||||
baseUrl: "wss://provider.example/audio",
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: "wss://provider.example/audio",
|
||||
}).then((socket) => {
|
||||
socket.on("error", () => {});
|
||||
socket.terminate();
|
||||
return socket;
|
||||
});
|
||||
await expect(opening).rejects.toThrow(/private|loopback|blocked/iu);
|
||||
expect(proxy.connectionCount()).toBe(0);
|
||||
});
|
||||
|
||||
it("keeps DNS preparation and the handshake within one connection deadline", async () => {
|
||||
const server = await createStalledHandshakeServer("upgrade");
|
||||
const lookupStarted = createDeferred();
|
||||
const releaseLookup = createDeferred();
|
||||
const resolveHostname = ssrf.resolvePinnedHostnameWithPolicy;
|
||||
vi.spyOn(ssrf, "resolvePinnedHostnameWithPolicy").mockImplementationOnce(async (...args) => {
|
||||
lookupStarted.resolve();
|
||||
await releaseLookup.promise;
|
||||
return await resolveHostname(...args);
|
||||
});
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
const url = server.url.replace("http:", "ws:");
|
||||
const opening = openProviderWebSocket({
|
||||
allowPrivateNetwork: true,
|
||||
baseUrl: url,
|
||||
dispatcherPolicy: { mode: "direct" },
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url,
|
||||
});
|
||||
await lookupStarted.promise;
|
||||
await vi.advanceTimersByTimeAsync(600);
|
||||
releaseLookup.resolve();
|
||||
const socket = await opening;
|
||||
const errors: Error[] = [];
|
||||
socket.on("error", (error) => errors.push(error));
|
||||
const peer = await server.received;
|
||||
const closed = once(peer, "close", { signal: AbortSignal.timeout(2000) }).then(
|
||||
() => "closed",
|
||||
(error: unknown) => error,
|
||||
);
|
||||
await vi.advanceTimersByTimeAsync(399);
|
||||
expect(socket.readyState).toBe(socket.CONNECTING);
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(errors).toHaveLength(1);
|
||||
expect(await closed).toBe("closed");
|
||||
});
|
||||
|
||||
it.each(["deadline", "caller cancellation", "socket termination"] as const)(
|
||||
"closes a stalled proxy CONNECT on %s",
|
||||
async (stop) => {
|
||||
const proxy = await createStalledHandshakeServer("connect");
|
||||
const controller = new AbortController();
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
const socket = await openProviderWebSocket({
|
||||
allowPrivateNetwork: true,
|
||||
baseUrl: "wss://127.0.0.1/audio",
|
||||
dispatcherPolicy: { mode: "explicit-proxy", proxyUrl: proxy.url },
|
||||
signal: controller.signal,
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: "wss://127.0.0.1/audio",
|
||||
});
|
||||
socket.on("error", () => {});
|
||||
const peer = await proxy.received;
|
||||
const closed = once(peer, "close", { signal: AbortSignal.timeout(2000) }).then(
|
||||
() => "closed",
|
||||
(error: unknown) => error,
|
||||
);
|
||||
if (stop === "deadline") {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
} else if (stop === "caller cancellation") {
|
||||
controller.abort();
|
||||
} else {
|
||||
socket.terminate();
|
||||
}
|
||||
expect(socket.readyState).not.toBe(socket.CONNECTING);
|
||||
expect(await closed).toBe("closed");
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
{ name: "private", allowPrivateNetwork: false },
|
||||
{ name: "pre-aborted", allowPrivateNetwork: true, signal: AbortSignal.abort() },
|
||||
])("does not open a $name socket", async (testCase) => {
|
||||
const server = await createLocalWebSocketServer();
|
||||
await expect(
|
||||
openProviderWebSocket({
|
||||
allowPrivateNetwork: testCase.allowPrivateNetwork,
|
||||
baseUrl: server.url,
|
||||
headers: { authorization: "Token test" },
|
||||
...(testCase.signal ? { signal: testCase.signal } : {}),
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: server.url,
|
||||
}),
|
||||
).rejects.toThrow(/private|loopback|blocked|aborted/iu);
|
||||
expect(server.requestHeaders).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("opens an allowed socket with resolved request headers", async () => {
|
||||
const server = await createLocalWebSocketServer();
|
||||
const socket = await openProviderWebSocket({
|
||||
allowPrivateNetwork: true,
|
||||
baseUrl: server.url,
|
||||
headers: { authorization: "Token configured", "x-provider": "deepgram" },
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: server.url,
|
||||
});
|
||||
await once(socket, "open");
|
||||
expect(server.requestHeaders[0]?.authorization).toBe("Token configured");
|
||||
expect(server.requestHeaders[0]?.["x-provider"]).toBe("deepgram");
|
||||
socket.close();
|
||||
await once(socket, "close");
|
||||
});
|
||||
|
||||
it("applies target TLS settings to the WebSocket handshake", async () => {
|
||||
const server = await createLocalWebSocketServer({ tls: true });
|
||||
const socket = await openProviderWebSocket({
|
||||
allowPrivateNetwork: true,
|
||||
baseUrl: server.url,
|
||||
dispatcherPolicy: { mode: "direct", connect: { rejectUnauthorized: false } },
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: server.url,
|
||||
});
|
||||
await once(socket, "open");
|
||||
expect(server.requestHeaders).toHaveLength(1);
|
||||
socket.close();
|
||||
await once(socket, "close");
|
||||
});
|
||||
|
||||
it("routes the WebSocket handshake through an explicit proxy", async () => {
|
||||
const target = await createLocalWebSocketServer();
|
||||
const connectSpy = vi.spyOn(net, "connect");
|
||||
const proxy = await createConnectProxy("localhost");
|
||||
const socket = await openProviderWebSocket({
|
||||
allowPrivateNetwork: true,
|
||||
baseUrl: target.url,
|
||||
dispatcherPolicy: { mode: "explicit-proxy", proxyUrl: proxy.url },
|
||||
timeoutMs: 1000,
|
||||
trustConfiguredBaseUrlOrigin: false,
|
||||
url: target.url,
|
||||
});
|
||||
await once(socket, "open");
|
||||
expect(proxy.connectCount()).toBe(1);
|
||||
expect(target.requestHeaders).toHaveLength(1);
|
||||
expect(
|
||||
connectSpy.mock.calls.some(
|
||||
([options]) => typeof asOptionalRecord(options)?.lookup === "function",
|
||||
),
|
||||
).toBe(true);
|
||||
socket.close();
|
||||
await once(socket, "close");
|
||||
});
|
||||
});
|
||||
211
src/infra/net/provider-websocket.ts
Normal file
211
src/infra/net/provider-websocket.ts
Normal file
|
|
@ -0,0 +1,211 @@
|
|||
// Provider WebSocket connector applies the same auth, proxy, TLS, and SSRF policy as provider HTTP.
|
||||
import http from "node:http";
|
||||
import type { Agent as HttpAgent } from "node:http";
|
||||
import https from "node:https";
|
||||
import WebSocket from "ws";
|
||||
import { resolveProviderTransportSsrFPolicy } from "../../agents/provider-transport-fetch.js";
|
||||
import { buildTimeoutAbortSignal } from "../../utils/fetch-timeout.js";
|
||||
import { racePromiseWithAbortSignal } from "../abort-signal.js";
|
||||
import { isManagedProxyActive } from "./fetch-guard.js";
|
||||
import { createNodeProxyAgent, resolveEnvNodeProxyUrlForTarget } from "./node-proxy-agent.js";
|
||||
import { shouldUseEnvHttpProxyForUrl } from "./proxy-env.js";
|
||||
import { resolveActiveManagedProxyTlsOptions } from "./proxy/active-managed-proxy-tls.js";
|
||||
import {
|
||||
assertHostnameAllowedWithPolicy,
|
||||
resolvePinnedHostnameWithPolicy,
|
||||
type PinnedDispatcherPolicy,
|
||||
type SsrFPolicy,
|
||||
} from "./ssrf.js";
|
||||
|
||||
const DEFAULT_PROVIDER_WEBSOCKET_MAX_PAYLOAD_BYTES = 16 * 1024 * 1024;
|
||||
|
||||
type OpenProviderWebSocketParams = {
|
||||
allowPrivateNetwork: boolean;
|
||||
baseUrl: string;
|
||||
dispatcherPolicy?: PinnedDispatcherPolicy;
|
||||
headers?: HeadersInit;
|
||||
maxPayloadBytes?: number;
|
||||
signal?: AbortSignal;
|
||||
timeoutMs: number;
|
||||
trustConfiguredBaseUrlOrigin: boolean;
|
||||
url: string;
|
||||
};
|
||||
|
||||
function toHttpUrl(value: string): string {
|
||||
const url = new URL(value);
|
||||
if (url.protocol === "ws:") {
|
||||
url.protocol = "http:";
|
||||
} else if (url.protocol === "wss:") {
|
||||
url.protocol = "https:";
|
||||
}
|
||||
return url.toString();
|
||||
}
|
||||
|
||||
function targetTlsOptions(policy: PinnedDispatcherPolicy | undefined): Record<string, unknown> {
|
||||
return policy?.mode === "direct" || policy?.mode === "env-proxy" ? { ...policy.connect } : {};
|
||||
}
|
||||
|
||||
function proxyPolicy(
|
||||
policy: SsrFPolicy | undefined,
|
||||
allowPrivateProxy: boolean,
|
||||
): SsrFPolicy | undefined {
|
||||
if (!policy && !allowPrivateProxy) {
|
||||
return undefined;
|
||||
}
|
||||
return {
|
||||
...policy,
|
||||
hostnameAllowlist: undefined,
|
||||
...(allowPrivateProxy ? { allowPrivateNetwork: true } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
async function createProxyAgent(params: {
|
||||
policy: SsrFPolicy | undefined;
|
||||
proxyUrl: URL;
|
||||
proxyTls?: Record<string, unknown>;
|
||||
allowPrivateProxy: boolean;
|
||||
signal?: AbortSignal;
|
||||
}): Promise<HttpAgent> {
|
||||
const pinnedProxy = await resolvePinnedHostnameWithPolicy(params.proxyUrl.hostname, {
|
||||
policy: proxyPolicy(params.policy, params.allowPrivateProxy),
|
||||
signal: params.signal,
|
||||
});
|
||||
return createNodeProxyAgent({
|
||||
mode: "explicit",
|
||||
proxyUrl: params.proxyUrl,
|
||||
proxyConnect: { ...params.proxyTls, lookup: pinnedProxy.lookup },
|
||||
});
|
||||
}
|
||||
|
||||
async function createProviderWebSocketAgent(params: {
|
||||
dispatcherPolicy?: PinnedDispatcherPolicy;
|
||||
policy: SsrFPolicy | undefined;
|
||||
url: URL;
|
||||
signal?: AbortSignal;
|
||||
}): Promise<HttpAgent> {
|
||||
const { dispatcherPolicy, policy, url, signal } = params;
|
||||
const canDelegateEnvDns = shouldUseEnvHttpProxyForUrl(toHttpUrl(url.href));
|
||||
const useManagedProxy = isManagedProxyActive() && canDelegateEnvDns;
|
||||
const envProxyUrl =
|
||||
useManagedProxy || dispatcherPolicy?.mode !== "direct"
|
||||
? resolveEnvNodeProxyUrlForTarget(url)
|
||||
: undefined;
|
||||
let proxyUrl: URL | undefined;
|
||||
if (dispatcherPolicy?.mode === "explicit-proxy") {
|
||||
try {
|
||||
proxyUrl = new URL(dispatcherPolicy.proxyUrl);
|
||||
} catch {
|
||||
throw new Error("Invalid explicit proxy URL");
|
||||
}
|
||||
if (proxyUrl.protocol !== "http:" && proxyUrl.protocol !== "https:") {
|
||||
throw new Error("Explicit proxy URL must use http or https");
|
||||
}
|
||||
} else if (dispatcherPolicy?.mode !== "direct") {
|
||||
proxyUrl = envProxyUrl;
|
||||
}
|
||||
if (useManagedProxy) {
|
||||
proxyUrl = envProxyUrl;
|
||||
}
|
||||
|
||||
if (!proxyUrl) {
|
||||
const pinned = await resolvePinnedHostnameWithPolicy(url.hostname, { policy, signal });
|
||||
const options = {
|
||||
keepAlive: false,
|
||||
...targetTlsOptions(dispatcherPolicy),
|
||||
lookup: pinned.lookup,
|
||||
};
|
||||
return url.protocol === "wss:" ? new https.Agent(options) : new http.Agent(options);
|
||||
}
|
||||
|
||||
// Match guarded HTTP: configured proxies stay strict, while applicable
|
||||
// managed/ambient HTTP proxy routes own DNS. ALL_PROXY alone grants no trust.
|
||||
if (!useManagedProxy && (dispatcherPolicy || !canDelegateEnvDns)) {
|
||||
await resolvePinnedHostnameWithPolicy(url.hostname, { policy, signal });
|
||||
} else {
|
||||
assertHostnameAllowedWithPolicy(url.hostname, policy);
|
||||
}
|
||||
return await createProxyAgent({
|
||||
policy,
|
||||
proxyUrl,
|
||||
proxyTls:
|
||||
!useManagedProxy &&
|
||||
(dispatcherPolicy?.mode === "explicit-proxy" || dispatcherPolicy?.mode === "env-proxy")
|
||||
? dispatcherPolicy.proxyTls
|
||||
: resolveActiveManagedProxyTlsOptions({ proxyUrl: proxyUrl.href }),
|
||||
allowPrivateProxy:
|
||||
useManagedProxy ||
|
||||
dispatcherPolicy?.mode !== "explicit-proxy" ||
|
||||
dispatcherPolicy.allowPrivateProxy === true,
|
||||
signal,
|
||||
});
|
||||
}
|
||||
|
||||
/** Opens a provider WebSocket through the resolved request and network policy. */
|
||||
export async function openProviderWebSocket(
|
||||
params: OpenProviderWebSocketParams,
|
||||
): Promise<WebSocket> {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(params.url);
|
||||
} catch {
|
||||
throw new Error("Invalid provider WebSocket URL");
|
||||
}
|
||||
if (url.protocol !== "ws:" && url.protocol !== "wss:") {
|
||||
throw new Error("Provider WebSocket URL must use ws or wss");
|
||||
}
|
||||
const policy = resolveProviderTransportSsrFPolicy({
|
||||
baseUrl: toHttpUrl(params.baseUrl),
|
||||
url: toHttpUrl(url.toString()),
|
||||
allowPrivateNetwork: params.allowPrivateNetwork,
|
||||
trustConfiguredBaseUrlOrigin: params.trustConfiguredBaseUrlOrigin,
|
||||
});
|
||||
// DNS preparation and the opening handshake share one deadline. Proxyline
|
||||
// owns pending proxy sockets and closes them when the request or agent ends.
|
||||
const { signal, cleanup } = buildTimeoutAbortSignal({
|
||||
signal: params.signal,
|
||||
timeoutMs: Math.max(1, params.timeoutMs),
|
||||
operation: "Provider WebSocket connection",
|
||||
});
|
||||
let agent: HttpAgent;
|
||||
try {
|
||||
signal?.throwIfAborted();
|
||||
const pending = createProviderWebSocketAgent({
|
||||
dispatcherPolicy: params.dispatcherPolicy,
|
||||
policy,
|
||||
url,
|
||||
signal,
|
||||
});
|
||||
void pending.then(
|
||||
(resolved) => signal?.aborted && resolved.destroy(),
|
||||
() => undefined,
|
||||
);
|
||||
agent = await racePromiseWithAbortSignal(pending, signal);
|
||||
} catch (error) {
|
||||
cleanup();
|
||||
throw error;
|
||||
}
|
||||
let socket: WebSocket;
|
||||
try {
|
||||
signal?.throwIfAborted();
|
||||
socket = new WebSocket(url, {
|
||||
agent,
|
||||
headers: Object.fromEntries(new Headers(params.headers).entries()),
|
||||
maxPayload: params.maxPayloadBytes ?? DEFAULT_PROVIDER_WEBSOCKET_MAX_PAYLOAD_BYTES,
|
||||
perMessageDeflate: false,
|
||||
...targetTlsOptions(params.dispatcherPolicy),
|
||||
});
|
||||
} catch (error) {
|
||||
cleanup();
|
||||
agent.destroy();
|
||||
throw error;
|
||||
}
|
||||
const onAbort = () => socket.terminate();
|
||||
signal?.addEventListener("abort", onAbort, { once: true });
|
||||
socket.once("open", cleanup);
|
||||
socket.once("close", () => {
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
cleanup();
|
||||
agent.destroy();
|
||||
});
|
||||
return socket;
|
||||
}
|
||||
|
|
@ -42,11 +42,13 @@ export {
|
|||
postTranscriptionRequest,
|
||||
resolveProviderOperationTimeoutMs,
|
||||
resolveProviderHttpRequestConfig,
|
||||
resolveProviderHttpRequestConfigWithOriginTrust,
|
||||
resolveAudioTranscriptionUploadFileName,
|
||||
requireTranscriptionText,
|
||||
sanitizeConfiguredModelProviderRequest,
|
||||
waitProviderOperationPollInterval,
|
||||
} from "../media-understanding/shared.js";
|
||||
export { openProviderWebSocket } from "../infra/net/provider-websocket.js";
|
||||
export type {
|
||||
ProviderOperationDeadline,
|
||||
ProviderOperationTimeoutMs,
|
||||
|
|
|
|||
|
|
@ -328,14 +328,28 @@ vi.mock("openclaw/plugin-sdk/provider-http", async (importActual) => ({
|
|||
pollProviderOperationJson: providerHttpMocks.pollProviderOperationJsonMock,
|
||||
postJsonRequest: providerHttpMocks.postJsonRequestMock,
|
||||
postMultipartRequest: providerHttpMocks.postMultipartRequestMock,
|
||||
postTranscriptionRequest: (
|
||||
await importActual<typeof import("openclaw/plugin-sdk/provider-http")>()
|
||||
).postTranscriptionRequest,
|
||||
providerOperationRetryConfig: (_stage: string) => true,
|
||||
readProviderBinaryResponse: (
|
||||
await importActual<typeof import("openclaw/plugin-sdk/provider-http")>()
|
||||
).readProviderBinaryResponse,
|
||||
readProviderJsonResponse: providerHttpMocks.readProviderJsonResponseMock,
|
||||
readProviderJsonObjectResponse: (
|
||||
await importActual<typeof import("openclaw/plugin-sdk/provider-http")>()
|
||||
).readProviderJsonObjectResponse,
|
||||
requireTranscriptionText: (
|
||||
await importActual<typeof import("openclaw/plugin-sdk/provider-http")>()
|
||||
).requireTranscriptionText,
|
||||
resolveProviderOperationTimeoutMs: ({ defaultTimeoutMs }: { defaultTimeoutMs: number }) =>
|
||||
defaultTimeoutMs,
|
||||
resolveProviderHttpRequestConfig: providerHttpMocks.resolveProviderHttpRequestConfigMock,
|
||||
resolveProviderHttpRequestConfigWithOriginTrust: (
|
||||
await importActual<typeof import("openclaw/plugin-sdk/provider-http")>()
|
||||
).resolveProviderHttpRequestConfigWithOriginTrust,
|
||||
openProviderWebSocket: (await importActual<typeof import("openclaw/plugin-sdk/provider-http")>())
|
||||
.openProviderWebSocket,
|
||||
resolveProviderRequestHeaders: providerHttpMocks.resolveProviderRequestHeadersMock,
|
||||
[providerHttpMockKeys.sanitizeConfiguredModelProviderRequest]:
|
||||
providerHttpMocks.sanitizeConfiguredModelProviderRequestMock,
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue