From 75fcaba9191e2a28fe9e9ef45f3f2a183a79b749 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 17 Aug 2026 18:59:19 -0700 Subject: [PATCH] fix(voice-call): survive gateway in-process restart and stop CLI dead-ends (#125458) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(voice-call): survive gateway in-process restart and stop CLI dead-ends The gateway's in-process restart (SIGUSR1 config reload) reuses the cached plugin registry, so service stop/start run on the same retained voice-call registration. Generation fencing from #120289 treated that restart as a stale actor: stop retired the generation forever, the next start silently bailed, and every voicecall.* RPC answered UNAVAILABLE "runtime generation is retired" while the webhook never rebound. - Registrations now hold a replaceable generation: service start after stop mints a fresh generation, takes over a running slot owned by a retired predecessor, and reports start failures to service health instead of silently returning. - The voicecall CLI classifies gateway failures with typed guards instead of message substrings: standalone/store fallback only when the gateway is genuinely absent; reachable-but-failed (request errors, auth, timeout) exits with actionable text; a standalone webhook port collision explains that a running Gateway probably owns the port instead of raw EADDRINUSE. - Plugin SDK gateway-runtime exports structural isGatewayTransportError / isGatewayClientRequestError guards (+2 documented surface budget). - Regression coverage: same-registration stop/start restart, retired-owner takeover, typed CLI fallback classification, and a real token-auth gateway server routing voicecall.status through callGatewayFromCli. * refactor(voice-call): split CLI modules and dedupe gateway fallbacks Collapse the four duplicated gateway-or-runtime command blocks (speak, dtmf, end, continue fallback) into one generic runGatewayManagerCommand helper — the continue command owns its legacy-method fallback and operation polling via a gatewayCall closure, so the helper carries no per-command policy. Smoke reuses the shared initiateVoiceCall path instead of a bespoke fallback. Split the 988-line cli.ts into concept modules (cli-gateway-call, cli-call-log, cli-command-io) and drop its grandfathered max-lines suppression plus the now-stale max-lines and assertion-safety baseline entries (shrink-only ratchet maintenance). Behavior-frozen: stdout/exit semantics unchanged; net -2 production LOC. * fix(voice-call): redact gateway URLs in CLI operational errors ClawSweeper P1: the operational-error formatter interpolated the raw connectionDetails.url, so a configured gateway URL with userinfo or query tokens would print credentials into terminal output. Redact the composed message once with the canonical net-policy redactor (also covers remote-controlled close-reason text), exported through the plugin SDK gateway-runtime subpath (+1 documented surface budget). Regression test covers a credential-bearing URL in both the URL and message fields. --- config/assertion-safety-baseline.txt | 1 - config/max-lines-baseline.txt | 1 - docs/cli/voicecall.md | 10 +- extensions/voice-call/index.lifecycle.test.ts | 71 +- extensions/voice-call/index.test.ts | 34 +- extensions/voice-call/index.ts | 153 ++-- extensions/voice-call/src/cli-call-log.ts | 192 +++++ extensions/voice-call/src/cli-command-io.ts | 24 + extensions/voice-call/src/cli-gateway-call.ts | 321 ++++++++ extensions/voice-call/src/cli.test.ts | 114 ++- extensions/voice-call/src/cli.ts | 683 +++--------------- scripts/plugin-sdk-surface-report.mts | 8 +- src/gateway/call.ts | 54 +- .../gateway-cli-plugin-routing.test.ts | 76 ++ src/plugin-sdk/gateway-runtime.ts | 4 + 15 files changed, 1030 insertions(+), 716 deletions(-) create mode 100644 extensions/voice-call/src/cli-call-log.ts create mode 100644 extensions/voice-call/src/cli-command-io.ts create mode 100644 extensions/voice-call/src/cli-gateway-call.ts create mode 100644 src/gateway/gateway-cli-plugin-routing.test.ts diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index 0632d524244b..5841ec107e5e 100644 --- a/config/assertion-safety-baseline.txt +++ b/config/assertion-safety-baseline.txt @@ -1382,7 +1382,6 @@ extensions/vllm/stream.ts 1 extensions/voice-call/doctor-contract-api.ts 1 extensions/voice-call/index.ts 2 extensions/voice-call/setup-api.ts 1 -extensions/voice-call/src/cli.ts 2 extensions/voice-call/src/config.ts 2 extensions/voice-call/src/manager.ts 2 extensions/voice-call/src/manager/outbound.ts 3 diff --git a/config/max-lines-baseline.txt b/config/max-lines-baseline.txt index b10f28a8d9b1..34fdb6c20358 100644 --- a/config/max-lines-baseline.txt +++ b/config/max-lines-baseline.txt @@ -256,7 +256,6 @@ extensions/telegram/src/thread-bindings.ts extensions/telegram/src/webhook.test.ts extensions/tlon/src/monitor/index.ts extensions/voice-call/index.test.ts -extensions/voice-call/src/cli.ts extensions/voice-call/src/media-stream.ts extensions/voice-call/src/webhook.test.ts extensions/voice-call/src/webhook.ts diff --git a/docs/cli/voicecall.md b/docs/cli/voicecall.md index 34a17e161e55..6012b08cb7cb 100644 --- a/docs/cli/voicecall.md +++ b/docs/cli/voicecall.md @@ -14,7 +14,15 @@ plugin is installed and enabled. When the Gateway is running, operational commands (`call`, `start`, `continue`, `speak`, `dtmf`, `end`, `status`) route to that Gateway's voice-call runtime. If no Gateway is reachable, they fall back to a standalone -CLI runtime. +CLI runtime. `status` uses the persisted call store instead of starting that +runtime. + +Fallback is limited to transport-level absence. If the Gateway responds with a +request or authentication error, or does not answer before the timeout, the +command exits nonzero and points to `openclaw gateway status`; it does not start +a second webhook server. If standalone fallback cannot bind the configured +`serve.port`, the error identifies the likely running Gateway instead of +printing a raw `EADDRINUSE` failure. ## Subcommands diff --git a/extensions/voice-call/index.lifecycle.test.ts b/extensions/voice-call/index.lifecycle.test.ts index 358ebfaa6c87..e51c250f2138 100644 --- a/extensions/voice-call/index.lifecycle.test.ts +++ b/extensions/voice-call/index.lifecycle.test.ts @@ -13,6 +13,7 @@ import plugin from "./index.js"; import { createVoiceCallRuntime } from "./runtime-entry.js"; type VoiceCallService = Parameters[0]; +type VoiceCallGatewayHandler = Parameters[1]; type VoiceCallTool = { execute: (toolCallId: string, params: unknown) => Promise; }; @@ -66,6 +67,7 @@ function registerVoiceCall(params: { }) { let service: VoiceCallService | undefined; let toolFactory: VoiceCallToolFactory | undefined; + const gatewayHandlers = new Map(); const api = createTestPluginApi({ id: "voice-call", name: "Voice Call", @@ -77,7 +79,9 @@ function registerVoiceCall(params: { pluginConfig: { provider: "mock", ...params.config }, runtime: { tts: { textToSpeechTelephony: vi.fn() } } as unknown as OpenClawPluginApi["runtime"], logger: params.logger ?? createLogger(), - registerGatewayMethod: () => {}, + registerGatewayMethod: (method, handler) => { + gatewayHandlers.set(method, handler); + }, registerTool: (registration) => { toolFactory = typeof registration === "function" @@ -95,13 +99,27 @@ function registerVoiceCall(params: { throw new Error("expected voice-call service and tool registrations"); } const registeredToolFactory = toolFactory; - return { service, toolFactory: registeredToolFactory, tool: () => registeredToolFactory({}) }; + return { + gatewayHandlers, + service, + toolFactory: registeredToolFactory, + tool: () => registeredToolFactory({}), + }; } function executeCall(tool: VoiceCallTool): Promise { return tool.execute("call", { action: "initiate_call", message: "hello" }); } +async function executeGatewayCall(registration: ReturnType) { + const respond = vi.fn(); + await registration.gatewayHandlers.get("voicecall.initiate")?.({ + params: { message: "hello" }, + respond, + } as never); + return respond; +} + function expectLifecycleError(result: VoiceCallToolResult, text: string): void { const detail = result.details?.error; const error = typeof detail === "string" ? detail : JSON.stringify(detail ?? ""); @@ -162,6 +180,29 @@ describe("voice-call runtime lifecycle", () => { expect(createVoiceCallRuntime).toHaveBeenCalledTimes(1); }); + it("restarts tools and gateway commands on the same service registration", async () => { + const runtimeA = createRuntime("call-a", "+15550000001"); + const runtimeB = createRuntime("call-b", "+15550000002"); + vi.mocked(createVoiceCallRuntime) + .mockResolvedValueOnce(runtimeA.runtime) + .mockResolvedValueOnce(runtimeB.runtime); + const registration = registerVoiceCall({}); + const retainedTool = registration.tool(); + + expect(registration.service.start(serviceContext)).toBeUndefined(); + await executeCall(retainedTool); + await registration.service.stop?.(serviceContext); + expect(registration.service.start(serviceContext)).toBeUndefined(); + + await executeCall(retainedTool); + const respond = await executeGatewayCall(registration); + + expect(respond).toHaveBeenCalledWith(true, { callId: "call-b", initiated: true }); + expect(createVoiceCallRuntime).toHaveBeenCalledTimes(2); + expect(runtimeA.stop).toHaveBeenCalledTimes(1); + expect(runtimeB.initiateCall).toHaveBeenCalledTimes(2); + }); + it("waits for A stopping before creating B with B config", async () => { const aStopEntered = createDeferred(); const releaseAStop = createDeferred(); @@ -242,32 +283,44 @@ describe("voice-call runtime lifecycle", () => { }); it("does not revive an older staged A after activated B stops", async () => { + const logged = createDeferred(); const runtimeB = createRuntime("call-b", "+15550000002"); vi.mocked(createVoiceCallRuntime).mockResolvedValue(runtimeB.runtime); - const stagedA = registerVoiceCall({ registrationMode: "full" }); + const stagedA = registerVoiceCall({ + logger: createLogger(logged.resolve), + registrationMode: "full", + }); const retainedToolA = stagedA.tool(); const generationB = registerVoiceCall({ registrationMode: "full" }); await executeCall(generationB.tool()); await generationB.service.stop?.(serviceContext); expectLifecycleError(await executeCall(retainedToolA), "superseded"); + expect(stagedA.service.start(serviceContext)).toBeUndefined(); + await expect(logged.promise).resolves.toContain("superseded"); + expect(serviceHealth.reportFailure).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining("superseded") }), + ); expect(createVoiceCallRuntime).toHaveBeenCalledTimes(1); expect(runtimeB.stop).toHaveBeenCalledTimes(1); }); - it("logs when successor startup is blocked by an active predecessor", async () => { - const logged = createDeferred(); + it("takes over a running slot owned by a retired predecessor", async () => { const runtimeA = createRuntime("call-a", "+15550000001"); - vi.mocked(createVoiceCallRuntime).mockResolvedValue(runtimeA.runtime); + const runtimeB = createRuntime("call-b", "+15550000002"); + vi.mocked(createVoiceCallRuntime) + .mockResolvedValueOnce(runtimeA.runtime) + .mockResolvedValueOnce(runtimeB.runtime); const generationA = registerVoiceCall({}); await executeCall(generationA.tool()); - const generationB = registerVoiceCall({ logger: createLogger(logged.resolve) }); + const generationB = registerVoiceCall({}); expect(generationB.service.start(serviceContext)).toBeUndefined(); - await expect(logged.promise).resolves.toContain("previous voice call runtime generation"); + await executeCall(generationB.tool()); - await generationA.service.stop?.(serviceContext); + expect(runtimeA.stop).toHaveBeenCalledTimes(1); + expect(runtimeB.initiateCall).toHaveBeenCalledTimes(1); await generationB.service.stop?.(serviceContext); }); diff --git a/extensions/voice-call/index.test.ts b/extensions/voice-call/index.test.ts index 0a3965a8450e..a327bd5f49af 100644 --- a/extensions/voice-call/index.test.ts +++ b/extensions/voice-call/index.test.ts @@ -187,10 +187,6 @@ function firstRuntimeConfig(): VoiceCallRuntime["config"] | undefined { return options?.config; } -function expectWarningIncludes(text: string): void { - expect(noopLogger.warn.mock.calls.map(([message]) => String(message)).join("\n")).toContain(text); -} - function expectRedactedVoiceCallStatus(value: unknown): void { expect(value).toEqual({ callId: "call-1", @@ -247,7 +243,13 @@ describe("voice-call plugin", () => { noopLogger.debug.mockClear(); runtimeStub = createRuntimeStub(); callGatewayFromCliMock.mockReset(); - callGatewayFromCliMock.mockRejectedValue(new Error("connect ECONNREFUSED 127.0.0.1:18789")); + callGatewayFromCliMock.mockRejectedValue( + Object.assign(new Error("gateway transport failed"), { + name: "GatewayTransportError", + kind: "closed", + connectionDetails: { url: "ws://127.0.0.1:18789" }, + }), + ); vi.mocked(createVoiceCallRuntime).mockReset(); vi.mocked(createVoiceCallRuntime).mockImplementation(async () => runtimeStub); }); @@ -337,22 +339,26 @@ describe("voice-call plugin", () => { } }); - it("does not log a startup error when provider setup is incomplete", async () => { + it("reports degraded service health when provider setup is incomplete", async () => { vi.stubEnv("TWILIO_ACCOUNT_SID", ""); vi.stubEnv("TWILIO_AUTH_TOKEN", ""); vi.stubEnv("TWILIO_FROM_NUMBER", ""); const { service } = setup({ provider: "twilio" }); + const reportFailure = vi.fn(); - await service?.start(createServiceContext()); + await service?.start({ + ...createServiceContext(), + serviceHealth: { reportFailure, clearFailure: vi.fn() }, + }); expect(createVoiceCallRuntime).not.toHaveBeenCalled(); - expect( - noopLogger.error.mock.calls.some(([message]) => - String(message).includes("Failed to start runtime"), - ), - ).toBe(false); - expectWarningIncludes("Runtime not started; setup incomplete"); - expectWarningIncludes("TWILIO_ACCOUNT_SID"); + expect(noopLogger.error.mock.calls.flat().join("\n")).toContain( + "Runtime not started: setup incomplete", + ); + expect(noopLogger.error.mock.calls.flat().join("\n")).toContain("TWILIO_ACCOUNT_SID"); + expect(reportFailure).toHaveBeenCalledWith( + expect.objectContaining({ message: expect.stringContaining("TWILIO_ACCOUNT_SID") }), + ); }); it("registers Twilio configs with SecretRef auth tokens", async () => { diff --git a/extensions/voice-call/index.ts b/extensions/voice-call/index.ts index 35c3c66889fc..4cac55e23ea9 100644 --- a/extensions/voice-call/index.ts +++ b/extensions/voice-call/index.ts @@ -95,12 +95,15 @@ function isCliOnlyProcess(): boolean { const VOICE_CALL_RUNTIME_COORDINATOR_KEY = Symbol.for("openclaw.voice-call.runtimeCoordinator"); type VoiceCallRuntimeGeneration = { - epoch: number; retired: boolean; serviceHealth?: Parameters< Parameters[0]["start"] >[0]["serviceHealth"]; - stopPromise?: Promise; +}; + +type VoiceCallRuntimeRegistration = { + epoch: number; + generation: VoiceCallRuntimeGeneration; }; type VoiceCallRuntimeSlot = @@ -121,10 +124,8 @@ type VoiceCallRuntimeSlot = }; type VoiceCallRuntimeCoordinator = { - current?: VoiceCallRuntimeGeneration; - // Activation advances this fence; construction alone stays rollback-safe. + current?: VoiceCallRuntimeRegistration; epochCounter: number; - registeredEpoch: number; slot?: VoiceCallRuntimeSlot; }; @@ -133,15 +134,18 @@ class VoiceCallRuntimeLifecycleError extends Error {} function getVoiceCallRuntimeCoordinator(): VoiceCallRuntimeCoordinator { return resolveGlobalSingleton(VOICE_CALL_RUNTIME_COORDINATOR_KEY, () => ({ epochCounter: 0, - registeredEpoch: 0, })); } function activateVoiceCallRuntimeGeneration( coordinator: VoiceCallRuntimeCoordinator, + registration: VoiceCallRuntimeRegistration, generation: VoiceCallRuntimeGeneration, ): void { - if (generation.epoch < coordinator.registeredEpoch) { + if ( + registration.epoch < (coordinator.current?.epoch ?? 0) || + registration.generation !== generation + ) { throw new VoiceCallRuntimeLifecycleError( "Voice call runtime generation was superseded; use the current plugin registration", ); @@ -151,15 +155,37 @@ function activateVoiceCallRuntimeGeneration( "Voice call runtime generation is retired; use the current plugin registration", ); } - if (coordinator.current !== generation) { + if (coordinator.current !== registration) { if (coordinator.current) { - coordinator.current.retired = true; + coordinator.current.generation.retired = true; } - coordinator.current = generation; - coordinator.registeredEpoch = generation.epoch; + coordinator.current = registration; } } +function stopVoiceCallRuntimeGeneration( + coordinator: VoiceCallRuntimeCoordinator, + generation: VoiceCallRuntimeGeneration, +): Promise { + const ownedSlot = coordinator.slot?.owner === generation ? coordinator.slot : undefined; + if (!ownedSlot || ownedSlot.state === "stopping") { + return ownedSlot?.promise ?? Promise.resolve(); + } + const stopPromise = Promise.resolve().then(async () => { + const runtime = ownedSlot.state === "running" ? ownedSlot.runtime : await ownedSlot.promise; + await runtime.stop(); + }); + const stoppingSlot = { state: "stopping" as const, owner: generation, promise: stopPromise }; + if (coordinator.slot === ownedSlot) { + coordinator.slot = stoppingSlot; + } + return stopPromise.finally(() => { + if (coordinator.slot === stoppingSlot) { + coordinator.slot = undefined; + } + }); +} + export default definePluginEntry({ id: "voice-call", name: "Voice Call", @@ -170,20 +196,24 @@ export default definePluginEntry({ const validation = validateProviderConfig(config); const runtimeCoordinator = getVoiceCallRuntimeCoordinator(); - const runtimeGeneration: VoiceCallRuntimeGeneration = + const runtimeRegistration: VoiceCallRuntimeRegistration = api.registrationMode !== "full" && runtimeCoordinator.current ? runtimeCoordinator.current : { epoch: ++runtimeCoordinator.epochCounter, - retired: false, + generation: { retired: false }, }; const continueOperationStore = createVoiceCallContinueOperationStore({ config, coreConfig: api.config as OpenClawConfig, }); + const activateRuntimeGeneration = (generation: VoiceCallRuntimeGeneration) => + activateVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeRegistration, generation); - const ensureRuntimeForGeneration = async (): Promise => { - activateVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeGeneration); + const ensureRuntimeForGeneration = async ( + runtimeGeneration: VoiceCallRuntimeGeneration, + ): Promise => { + activateRuntimeGeneration(runtimeGeneration); if (!config.enabled) { throw new Error("Voice call disabled in plugin config"); } @@ -192,12 +222,12 @@ export default definePluginEntry({ } while (true) { - activateVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeGeneration); + activateRuntimeGeneration(runtimeGeneration); const slot = runtimeCoordinator.slot; if (slot) { if (slot.owner !== runtimeGeneration) { - if (slot.state === "stopping") { - await slot.promise; + if (slot.owner.retired) { + await stopVoiceCallRuntimeGeneration(runtimeCoordinator, slot.owner); continue; } throw new VoiceCallRuntimeLifecycleError( @@ -222,7 +252,7 @@ export default definePluginEntry({ } throw err; } - activateVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeGeneration); + activateRuntimeGeneration(runtimeGeneration); if (runtimeCoordinator.slot !== slot) { continue; } @@ -251,15 +281,15 @@ export default definePluginEntry({ runtimeCoordinator.slot = startingSlot; } }; - const ensureRuntime = async (): Promise => { + const ensureRuntime = async ( + runtimeGeneration = runtimeRegistration.generation, + ): Promise => { try { - const runtime = await ensureRuntimeForGeneration(); + const runtime = await ensureRuntimeForGeneration(runtimeGeneration); runtimeGeneration.serviceHealth?.clearFailure(); return runtime; } catch (err) { - const staleGeneration = - runtimeGeneration.retired || runtimeGeneration.epoch < runtimeCoordinator.registeredEpoch; - if (!(err instanceof VoiceCallRuntimeLifecycleError && staleGeneration)) { + if (!(err instanceof VoiceCallRuntimeLifecycleError)) { runtimeGeneration.serviceHealth?.reportFailure(err); } throw err; @@ -531,80 +561,49 @@ export default definePluginEntry({ api.registerService({ id: "voicecall", start: (ctx) => { - runtimeGeneration.serviceHealth = ctx.serviceHealth; if (isCliOnlyProcess()) { return; } try { - activateVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeGeneration); - } catch (err) { - if (err instanceof VoiceCallRuntimeLifecycleError) { - return; + if (runtimeRegistration.generation.retired) { + if (runtimeCoordinator.current !== runtimeRegistration) { + throw new VoiceCallRuntimeLifecycleError( + "Voice call runtime generation was superseded; use the current plugin registration", + ); + } + runtimeRegistration.generation = { retired: false }; } - throw err; + runtimeRegistration.generation.serviceHealth = ctx.serviceHealth; + activateRuntimeGeneration(runtimeRegistration.generation); + } catch (err) { + ctx.serviceHealth?.reportFailure(err); + api.logger.error(`[voice-call] Failed to start runtime: ${formatErrorMessage(err)}`); + return; } if (!config.enabled) { return; } if (!validation.valid) { - api.logger.warn( - `[voice-call] Runtime not started; setup incomplete: ${validation.errors.join("; ")}`, - ); + const error = new Error(`setup incomplete: ${validation.errors.join("; ")}`); + ctx.serviceHealth?.reportFailure(error); + api.logger.error(`[voice-call] Runtime not started: ${error.message}`); return; } - void ensureRuntime().catch((err: unknown) => { - const staleGeneration = - runtimeGeneration.retired || - runtimeGeneration.epoch < runtimeCoordinator.registeredEpoch; - if (err instanceof VoiceCallRuntimeLifecycleError && staleGeneration) { + const startingGeneration = runtimeRegistration.generation; + void ensureRuntime(startingGeneration).catch((err: unknown) => { + if (err instanceof VoiceCallRuntimeLifecycleError) { return; } + ctx.serviceHealth?.reportFailure(err); api.logger.error(`[voice-call] Failed to start runtime: ${formatErrorMessage(err)}`); }); }, stop: async () => { - if (runtimeGeneration.stopPromise) { - return await runtimeGeneration.stopPromise; - } + const runtimeGeneration = runtimeRegistration.generation; runtimeGeneration.retired = true; - const ownedSlot = - runtimeCoordinator.slot?.owner === runtimeGeneration - ? runtimeCoordinator.slot - : undefined; - let stoppingSlot: VoiceCallRuntimeSlot | undefined; - const stopPromise = - ownedSlot?.state === "stopping" - ? ownedSlot.promise - : Promise.resolve().then(async () => { - if (!ownedSlot) { - return; - } - const runtime = - ownedSlot.state === "running" ? ownedSlot.runtime : await ownedSlot.promise; - await runtime.stop(); - }); - runtimeGeneration.stopPromise = stopPromise; - if (ownedSlot && ownedSlot.state !== "stopping") { - stoppingSlot = { - state: "stopping", - owner: runtimeGeneration, - promise: stopPromise, - }; - if (runtimeCoordinator.slot === ownedSlot) { - runtimeCoordinator.slot = stoppingSlot; - } - } else if (ownedSlot) { - stoppingSlot = ownedSlot; - } try { - await stopPromise; + await stopVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeGeneration); } finally { - if (stoppingSlot && runtimeCoordinator.slot === stoppingSlot) { - runtimeCoordinator.slot = undefined; - } - if (runtimeCoordinator.current === runtimeGeneration) { - runtimeCoordinator.current = undefined; - } runtimeGeneration.serviceHealth = undefined; } }, diff --git a/extensions/voice-call/src/cli-call-log.ts b/extensions/voice-call/src/cli-call-log.ts new file mode 100644 index 000000000000..da6b0eebd360 --- /dev/null +++ b/extensions/voice-call/src/cli-call-log.ts @@ -0,0 +1,192 @@ +// Voice Call plugin module implements cli call log commands. +import fs from "node:fs"; +import path from "node:path"; +import { StringDecoder } from "node:string_decoder"; +import type { Command } from "commander"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { sleep } from "../api.js"; +import { parseCliInteger, writeCliJson, writeCliLine } from "./cli-command-io.js"; +import { getCallHistoryFromStore } from "./manager/store.js"; + +function percentile(values: number[], p: number): number { + if (values.length === 0) { + return 0; + } + const sorted = [...values].toSorted((a, b) => a - b); + const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)); + return sorted[idx] ?? 0; +} + +function summarizeSeries(values: number[]): { + count: number; + minMs: number; + maxMs: number; + avgMs: number; + p50Ms: number; + p95Ms: number; +} { + if (values.length === 0) { + return { count: 0, minMs: 0, maxMs: 0, avgMs: 0, p50Ms: 0, p95Ms: 0 }; + } + + // Reduce instead of Math.min(...values): spread throws past V8's argument + // cap, and `latency --last ` can scan an unbounded JSONL history. + const minMs = values.reduce((min, value) => (value < min ? value : min)); + const maxMs = values.reduce((max, value) => (value > max ? value : max)); + const avgMs = values.reduce((sum, value) => sum + value, 0) / values.length; + return { + count: values.length, + minMs, + maxMs, + avgMs, + p50Ms: percentile(values, 50), + p95Ms: percentile(values, 95), + }; +} + +function writeVoiceCallLatencySummary(calls: unknown[]) { + const turnLatencyMs: number[] = []; + const listenWaitMs: number[] = []; + + for (const call of calls) { + const metadata = isRecord(call) && isRecord(call.metadata) ? call.metadata : undefined; + const latency = metadata?.lastTurnLatencyMs; + const listenWait = metadata?.lastTurnListenWaitMs; + if (typeof latency === "number" && Number.isFinite(latency)) { + turnLatencyMs.push(latency); + } + if (typeof listenWait === "number" && Number.isFinite(listenWait)) { + listenWaitMs.push(listenWait); + } + } + + writeCliJson({ + recordsScanned: calls.length, + turnLatency: summarizeSeries(turnLatencyMs), + listenWait: summarizeSeries(listenWaitMs), + }); +} + +export function registerVoiceCallLogs(params: { + root: Command; + defaultFile: string; + ensureHistoryStateRuntime: () => void; +}): void { + params.root + .command("tail") + .description("Tail voice-call JSONL logs (prints new lines; useful during provider tests)") + .option("--file ", "Path to calls.jsonl", params.defaultFile) + .option("--since ", "Print last N lines first", "25") + .option("--poll ", "Poll interval in ms", "250") + .action(async (options: { file: string; since?: string; poll?: string }) => { + const file = options.file; + const since = parseCliInteger(options.since, "--since", { min: 0 }); + const pollMs = parseCliInteger(options.poll, "--poll", { min: 50 }); + + const tailSqliteHistory = async (initialLimit: number): Promise => { + params.ensureHistoryStateRuntime(); + const seen = new Set(); + const printCall = (call: unknown): void => { + const line = JSON.stringify(call); + if (!seen.has(line)) { + seen.add(line); + writeCliLine(line); + } + }; + if (initialLimit > 0) { + for (const call of await getCallHistoryFromStore(path.dirname(file), initialLimit)) { + printCall(call); + } + } + for (;;) { + try { + for (const call of await getCallHistoryFromStore(path.dirname(file), 1000)) { + printCall(call); + } + } catch { + // ignore and retry + } + await sleep(pollMs); + } + }; + + if (fs.existsSync(file) && path.basename(file) !== "calls.jsonl") { + const initial = fs.readFileSync(file); + let decoder = new StringDecoder("utf8"); + const initialLines = decoder.write(initial).split("\n"); + let pendingLine = initialLines.pop() ?? ""; + const lines = initialLines.filter(Boolean); + for (const line of lines.slice(Math.max(0, lines.length - since))) { + writeCliLine(line); + } + + let offset = initial.length; + let lastObservedSize = initial.length; + for (;;) { + try { + const stat = fs.statSync(file); + // A short read can leave the cursor behind the observed file size; + // compare observed sizes so copytruncate also clears buffered text. + if (stat.size < lastObservedSize) { + offset = 0; + decoder = new StringDecoder("utf8"); + pendingLine = ""; + } + lastObservedSize = stat.size; + if (stat.size > offset) { + const fd = fs.openSync(file, "r"); + try { + const buf = Buffer.alloc(stat.size - offset); + const bytesRead = fs.readSync(fd, buf, 0, buf.length, offset); + offset += bytesRead; + const text = decoder.write(buf.subarray(0, bytesRead)); + const completeLines = `${pendingLine}${text}`.split("\n"); + pendingLine = completeLines.pop() ?? ""; + for (const line of completeLines.filter(Boolean)) { + writeCliLine(line); + } + } finally { + fs.closeSync(fd); + } + } + } catch { + // ignore and retry + } + await sleep(pollMs); + } + } else { + await tailSqliteHistory(since); + } + }); + + params.root + .command("latency") + .description("Summarize turn latency metrics from voice-call JSONL logs") + .option("--file ", "Path to calls.jsonl", params.defaultFile) + .option("--last ", "Analyze last N records", "200") + .action(async (options: { file: string; last?: string }) => { + const file = options.file; + const last = parseCliInteger(options.last, "--last", { min: 1 }); + + if (fs.existsSync(file) && path.basename(file) !== "calls.jsonl") { + const content = fs.readFileSync(file, "utf8"); + const calls = content + .split("\n") + .filter(Boolean) + .slice(-last) + .map((line) => { + try { + const parsed: unknown = JSON.parse(line); + return (isRecord(parsed) ? parsed.call : undefined) ?? parsed; + } catch { + return null; + } + }) + .filter((call) => call !== null); + writeVoiceCallLatencySummary(calls); + } else { + params.ensureHistoryStateRuntime(); + writeVoiceCallLatencySummary(await getCallHistoryFromStore(path.dirname(file), last)); + } + }); +} diff --git a/extensions/voice-call/src/cli-command-io.ts b/extensions/voice-call/src/cli-command-io.ts new file mode 100644 index 000000000000..992cc281f453 --- /dev/null +++ b/extensions/voice-call/src/cli-command-io.ts @@ -0,0 +1,24 @@ +// Voice Call plugin module handles cli command input and output. +import { format } from "node:util"; +import { parseStrictNonNegativeInteger } from "openclaw/plugin-sdk/number-runtime"; + +export function writeCliLine(...values: unknown[]): void { + process.stdout.write(`${format(...values)}\n`); +} + +export function writeCliJson(value: unknown): void { + process.stdout.write(`${JSON.stringify(value, null, 2)}\n`); +} + +export function parseCliInteger( + raw: string | undefined, + optionName: string, + opts?: { min?: number; max?: number }, +): number { + const min = opts?.min ?? 0; + const parsed = parseStrictNonNegativeInteger(raw?.trim() ?? ""); + if (parsed === undefined || parsed < min || (opts?.max !== undefined && parsed > opts.max)) { + throw new Error(`Invalid numeric value for ${optionName}: ${raw ?? ""}`); + } + return parsed; +} diff --git a/extensions/voice-call/src/cli-gateway-call.ts b/extensions/voice-call/src/cli-gateway-call.ts new file mode 100644 index 000000000000..18efd04530e8 --- /dev/null +++ b/extensions/voice-call/src/cli-gateway-call.ts @@ -0,0 +1,321 @@ +// Voice Call plugin module implements cli gateway calls. +import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import { + callGatewayFromCli, + isGatewayClientRequestError, + isGatewayTransportError, + redactSensitiveUrlLikeString, +} from "openclaw/plugin-sdk/gateway-runtime"; +import { + addTimerTimeoutGraceMs, + clampTimerTimeoutMs, + MAX_TIMER_TIMEOUT_MS, +} from "openclaw/plugin-sdk/number-runtime"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import { sleep } from "../api.js"; +import { writeCliJson } from "./cli-command-io.js"; +import type { VoiceCallConfig } from "./config.js"; +import type { VoiceCallRuntime } from "./runtime.js"; + +type VoiceCallGatewayMethod = + | "voicecall.initiate" + | "voicecall.start" + | "voicecall.continue" + | "voicecall.continue.start" + | "voicecall.continue.result" + | "voicecall.speak" + | "voicecall.dtmf" + | "voicecall.end" + | "voicecall.status"; + +type VoiceCallGatewayCallResult = { ok: true; payload: unknown } | { ok: false; error: unknown }; + +type GatewayCallOptions = { timeoutMs?: number }; + +const VOICE_CALL_GATEWAY_DEFAULT_TIMEOUT_MS = 5000; +const VOICE_CALL_GATEWAY_OPERATION_TIMEOUT_MS = 30000; +const VOICE_CALL_GATEWAY_TRANSCRIPT_BUFFER_MS = 10000; +const VOICE_CALL_GATEWAY_POLL_INTERVAL_MS = 1000; + +function isGatewayUnavailableForLocalFallback(err: unknown): boolean { + return ( + isGatewayTransportError(err) && + err.kind === "closed" && + (err.code === undefined || err.code === 1006) + ); +} + +function isGatewayCredentialFailure(err: unknown): err is Error { + return ( + err instanceof Error && + (err.name === "GatewayCredentialsRequiredError" || + err.name === "GatewayExplicitAuthRequiredError" || + err.name === "GatewaySecretRefUnavailableError") + ); +} + +function gatewayOperationalError(err: unknown): Error { + const message = formatErrorMessage(err); + const detail = (() => { + if (isGatewayClientRequestError(err)) { + return `Gateway responded but voicecall failed: ${message}\nThe running Gateway owns the voice-call runtime; check \`openclaw gateway status\` or restart it.`; + } + if (isGatewayCredentialFailure(err)) { + return `Gateway requires credentials: ${message}\nConfigure gateway.auth or pair this device with \`openclaw devices approve --latest\`.`; + } + if (isGatewayTransportError(err)) { + const url = err.connectionDetails.url; + if (err.kind === "timeout") { + const timeout = + err.timeoutMs === undefined ? "the configured timeout" : `${err.timeoutMs}ms`; + return `Gateway at ${url} did not answer within ${timeout}: ${message}\nIt may be starting or wedged; check \`openclaw gateway status\`.`; + } + return `Gateway connection at ${url} failed: ${message}\nCheck gateway.auth and \`openclaw gateway status\`, then retry.`; + } + return `Gateway voicecall request failed: ${message}\nCheck \`openclaw gateway status\`, then retry.`; + })(); + // Configured gateway URLs may embed userinfo/tokens, and close reasons are + // remote-controlled text; redact once where the text becomes operator-visible. + return new Error(redactSensitiveUrlLikeString(detail)); +} + +export function isUnknownMethod(err: unknown, method: VoiceCallGatewayMethod): boolean { + return formatErrorMessage(err).includes(`unknown method: ${method}`); +} + +export async function callVoiceCallGateway( + method: VoiceCallGatewayMethod, + params?: Record, + opts?: GatewayCallOptions, +): Promise { + try { + const timeoutMs = + typeof opts?.timeoutMs === "number" && Number.isFinite(opts.timeoutMs) + ? Math.max(1, Math.ceil(opts.timeoutMs)) + : VOICE_CALL_GATEWAY_DEFAULT_TIMEOUT_MS; + const payload = await callGatewayFromCli( + method, + { json: true, timeout: String(timeoutMs) }, + params, + { progress: false }, + ); + return { ok: true, payload }; + } catch (err) { + if (isGatewayUnavailableForLocalFallback(err)) { + return { ok: false, error: err }; + } + throw gatewayOperationalError(err); + } +} + +export function resolveOperationTimeout(config: VoiceCallConfig): number { + return Math.max( + VOICE_CALL_GATEWAY_OPERATION_TIMEOUT_MS, + addTimerTimeoutGraceMs(config.ringTimeoutMs) ?? 1, + ); +} + +export function resolveContinueTimeout(config: VoiceCallConfig): number { + return ( + clampTimerTimeoutMs( + config.transcriptTimeoutMs + + VOICE_CALL_GATEWAY_OPERATION_TIMEOUT_MS + + VOICE_CALL_GATEWAY_TRANSCRIPT_BUFFER_MS, + ) ?? 1 + ); +} + +function resolveVoiceCallDeadlineMs(timeoutMs: number, nowMs = Date.now()): number { + return nowMs + (clampTimerTimeoutMs(timeoutMs) ?? MAX_TIMER_TIMEOUT_MS); +} + +function readGatewayOperationId(payload: unknown): string { + if (isRecord(payload) && typeof payload.operationId === "string" && payload.operationId) { + return payload.operationId; + } + throw new Error("voicecall gateway response missing operationId"); +} + +function readGatewayPollTimeoutMs(payload: unknown, fallbackTimeoutMs: number): number { + if (isRecord(payload) && typeof payload.pollTimeoutMs === "number") { + return clampTimerTimeoutMs(payload.pollTimeoutMs) ?? fallbackTimeoutMs; + } + return fallbackTimeoutMs; +} + +function readCompletedContinueResult( + payload: unknown, +): + | { status: "pending" } + | { status: "completed"; result: unknown } + | { status: "failed"; error: string } { + if (!isRecord(payload)) { + throw new Error("voicecall gateway response missing operation status"); + } + if (payload.status === "pending") { + return { status: "pending" }; + } + if (payload.status === "failed") { + return { + status: "failed", + error: typeof payload.error === "string" ? payload.error : "continue failed", + }; + } + if (payload.status === "completed") { + return { status: "completed", result: payload.result }; + } + throw new Error("voicecall gateway response has unknown operation status"); +} + +export async function pollContinueGateway( + payload: unknown, + fallbackTimeoutMs: number, +): Promise { + if (!isRecord(payload) || typeof payload.operationId !== "string") { + return payload; + } + const params = { + operationId: readGatewayOperationId(payload), + timeoutMs: readGatewayPollTimeoutMs(payload, fallbackTimeoutMs), + }; + const deadlineMs = resolveVoiceCallDeadlineMs(params.timeoutMs); + + for (;;) { + // Sleep already clamps to remaining budget; the gateway RPC must too. + // Otherwise the final poll can overrun the continue deadline by a full RPC timeout. + const remainingMs = deadlineMs - Date.now(); + if (remainingMs <= 0) { + break; + } + const gateway = await callVoiceCallGateway( + "voicecall.continue.result", + { operationId: params.operationId }, + { timeoutMs: Math.min(VOICE_CALL_GATEWAY_DEFAULT_TIMEOUT_MS, remainingMs) }, + ); + if (!gateway.ok) { + throw new Error( + `gateway unavailable while waiting for voicecall continue result: ${formatErrorMessage( + gateway.error, + )}`, + ); + } + const result = readCompletedContinueResult(gateway.payload); + if (result.status === "completed") { + return result.result; + } + if (result.status === "failed") { + throw new Error(result.error); + } + const sleepMs = Math.min(VOICE_CALL_GATEWAY_POLL_INTERVAL_MS, deadlineMs - Date.now()); + if (sleepMs <= 0) { + break; + } + await sleep(sleepMs); + } + + throw new Error("voicecall continue timed out waiting for gateway operation"); +} + +async function ensureStandaloneRuntime(params: { + config: VoiceCallConfig; + ensureRuntime: () => Promise; +}): Promise { + try { + return await params.ensureRuntime(); + } catch (err) { + if (err instanceof Error && "code" in err && err.code === "EADDRINUSE") { + throw new Error( + `Voice-call webhook port ${params.config.serve.port} is already in use. A running Gateway probably already serves it; operational commands route through that Gateway. Check \`openclaw gateway status\` and retry.`, + { cause: err }, + ); + } + throw err; + } +} + +export async function runGatewayManagerCommand(params: { + config: VoiceCallConfig; + ensureRuntime: () => Promise; + gatewayCall: () => Promise; + resolveGatewayPayload?: (payload: unknown) => Promise; + managerFallback: ( + manager: VoiceCallRuntime["manager"], + ) => Promise<{ success: boolean; error?: string }>; + failureLabel: string; +}): Promise { + const gateway = await params.gatewayCall(); + if (gateway.ok) { + const payload = params.resolveGatewayPayload + ? await params.resolveGatewayPayload(gateway.payload) + : gateway.payload; + writeCliJson(payload); + return; + } + + const runtime = await ensureStandaloneRuntime(params); + const result = await params.managerFallback(runtime.manager); + if (!result.success) { + throw new Error(result.error || `${params.failureLabel} failed`); + } + writeCliJson(result); +} + +function readGatewayCallId(payload: unknown, invalidCallIdMessage?: string): string { + if (isRecord(payload) && typeof payload.callId === "string") { + if (!invalidCallIdMessage || payload.callId) { + return payload.callId; + } + } + if (invalidCallIdMessage) { + throw new Error(invalidCallIdMessage); + } + if (isRecord(payload) && typeof payload.error === "string") { + throw new Error(payload.error); + } + throw new Error("voicecall gateway response missing callId"); +} + +export async function initiateVoiceCall(params: { + ensureRuntime: () => Promise; + config: VoiceCallConfig; + method: "voicecall.initiate" | "voicecall.start"; + to?: string; + message?: string; + mode?: string; + defaultMode?: "notify" | "conversation"; + failureMessage?: string; +}): Promise { + const mode = + params.mode === "notify" || params.mode === "conversation" ? params.mode : params.defaultMode; + const gateway = await callVoiceCallGateway( + params.method, + { + ...(params.to ? { to: params.to } : {}), + ...(params.message ? { message: params.message } : {}), + ...(mode ? { mode } : {}), + }, + { + timeoutMs: resolveOperationTimeout(params.config), + }, + ); + if (gateway.ok) { + return readGatewayCallId(gateway.payload, params.failureMessage); + } + + const runtime = await ensureStandaloneRuntime(params); + const to = params.to ?? runtime.config.toNumber; + if (!to) { + throw new Error("Missing --to and no toNumber configured"); + } + const result = await runtime.manager.initiateCall(to, undefined, { + message: params.message, + mode, + }); + if (!result.success) { + throw new Error(result.error || params.failureMessage || "initiate failed"); + } + if (params.failureMessage && !result.callId) { + throw new Error(params.failureMessage); + } + return result.callId; +} diff --git a/extensions/voice-call/src/cli.test.ts b/extensions/voice-call/src/cli.test.ts index 8666ad197c47..7393bb76995d 100644 --- a/extensions/voice-call/src/cli.test.ts +++ b/extensions/voice-call/src/cli.test.ts @@ -56,6 +56,31 @@ function captureStdout() { }; } +function gatewayTransportError(code?: number): Error { + return Object.assign(new Error("gateway transport failed"), { + name: "GatewayTransportError", + kind: "closed", + connectionDetails: { url: "ws://127.0.0.1:18789" }, + ...(code === undefined ? {} : { code }), + }); +} + +function gatewayRequestError(message: string, gatewayCode = "UNAVAILABLE"): Error { + return Object.assign(new Error(message), { + name: "GatewayClientRequestError", + gatewayCode, + retryable: false, + }); +} + +function gatewayCredentialsError(message: string): Error { + return Object.assign(new Error(message), { + name: "GatewayCredentialsRequiredError", + method: "voicecall.status", + configPath: "/tmp/openclaw.json", + }); +} + describe("voice-call CLI status fallback", () => { afterEach(() => { callGatewayFromCliMock.mockReset(); @@ -78,12 +103,13 @@ describe("voice-call CLI status fallback", () => { function buildProgram( manager: Record, config: Record = {}, + ensureRuntime = async () => ({ manager }) as never, ): Command { const program = new Command(); registerVoiceCallCli({ program, config: config as never, - ensureRuntime: async () => ({ manager }) as never, + ensureRuntime, logger: { info() {}, warn() {}, error() {}, debug() {} } as never, }); return program; @@ -94,9 +120,7 @@ describe("voice-call CLI status fallback", () => { error?: Error; args?: string[]; }): Promise { - callGatewayFromCliMock.mockRejectedValue( - params.error ?? new Error("connect ECONNREFUSED 127.0.0.1:18789"), - ); + callGatewayFromCliMock.mockRejectedValue(params.error ?? gatewayTransportError()); findCallMatchesInStoreMock.mockResolvedValue({ byCallId: params.persisted }); const ensureRuntime = vi.fn(async () => { throw new Error("status fallback must not initialize the telephony runtime"); @@ -153,11 +177,89 @@ describe("voice-call CLI status fallback", () => { it("falls back after an abnormal local gateway close", async () => { const result = await runStatusWithUnavailableGateway({ persisted: { callId: "call-1", state: "completed" }, - error: new Error("gateway closed (1006 abnormal closure (no close frame)): no close reason"), + error: gatewayTransportError(1006), }); expect(result).toMatchObject({ callId: "call-1", state: "completed" }); }); + it("keeps reachable gateway request failures out of the standalone runtime", async () => { + callGatewayFromCliMock.mockRejectedValue( + gatewayRequestError("Voice call runtime generation is retired; use the current registration"), + ); + const ensureRuntime = vi.fn(); + const program = buildProgram({}, {}, ensureRuntime); + + await expect( + program.parseAsync(["voicecall", "call", "--message", "hello"], { from: "user" }), + ).rejects.toThrow( + "Gateway responded but voicecall failed: Voice call runtime generation is retired; use the current registration", + ); + expect(ensureRuntime).not.toHaveBeenCalled(); + }); + + it("explains a standalone webhook port collision", async () => { + callGatewayFromCliMock.mockRejectedValue(gatewayTransportError()); + const ensureRuntime = vi.fn(async () => { + throw Object.assign(new Error("listen failed"), { code: "EADDRINUSE" }); + }); + const program = buildProgram({}, { serve: { port: 3334 } }, ensureRuntime); + + await expect( + program.parseAsync(["voicecall", "call", "--message", "hello"], { from: "user" }), + ).rejects.toThrow( + "Voice-call webhook port 3334 is already in use. A running Gateway probably already serves it", + ); + }); + + it("keeps gateway credential failures out of the standalone runtime", async () => { + callGatewayFromCliMock.mockRejectedValue( + gatewayCredentialsError("gateway voicecall.status requires credentials"), + ); + const ensureRuntime = vi.fn(); + const program = buildProgram({}, {}, ensureRuntime); + + await expect( + program.parseAsync(["voicecall", "status", "--json"], { from: "user" }), + ).rejects.toThrow( + "Gateway requires credentials: gateway voicecall.status requires credentials", + ); + expect(ensureRuntime).not.toHaveBeenCalled(); + expect(loadActiveCallsFromStoreMock).not.toHaveBeenCalled(); + }); + + it("redacts credential-bearing gateway URLs from operational errors", async () => { + callGatewayFromCliMock.mockRejectedValue( + Object.assign( + new Error( + "gateway closed (1008): policy wss://operator:hunter2secret@gw.example.ts.net:18789", + ), + { + name: "GatewayTransportError", + kind: "closed", + code: 1008, + connectionDetails: { + url: "wss://operator:hunter2secret@gw.example.ts.net:18789?token=tok123", + }, + }, + ), + ); + const ensureRuntime = vi.fn(); + const program = buildProgram({}, {}, ensureRuntime); + + let thrown: unknown; + await program + .parseAsync(["voicecall", "status", "--json"], { from: "user" }) + .catch((err: unknown) => { + thrown = err; + }); + + const text = thrown instanceof Error ? thrown.message : String(thrown); + expect(text).toContain("Gateway connection at wss://***:***@gw.example.ts.net:18789"); + expect(text).not.toContain("hunter2secret"); + expect(text).not.toContain("tok123"); + expect(ensureRuntime).not.toHaveBeenCalled(); + }); + it("rejects non-decimal tail options through the registered command", async () => { const program = buildProgram({}); await expect( @@ -431,7 +533,7 @@ describe("voice-call CLI status fallback", () => { it("caps oversized legacy continue timeouts through the command", async () => { callGatewayFromCliMock - .mockRejectedValueOnce(new Error("unknown method: voicecall.continue.start")) + .mockRejectedValueOnce(gatewayRequestError("unknown method: voicecall.continue.start")) .mockResolvedValueOnce({ success: true, transcript: "done" }); const program = buildProgram({}, { transcriptTimeoutMs: Number.MAX_SAFE_INTEGER }); await program.parseAsync( diff --git a/extensions/voice-call/src/cli.ts b/extensions/voice-call/src/cli.ts index e28b2c638acc..95b170c608ad 100644 --- a/extensions/voice-call/src/cli.ts +++ b/extensions/voice-call/src/cli.ts @@ -1,33 +1,28 @@ // Voice Call plugin module implements cli behavior. -import fs from "node:fs"; import path from "node:path"; -import { StringDecoder } from "node:string_decoder"; -import { format } from "node:util"; import type { Command } from "commander"; -import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; -import { callGatewayFromCli } from "openclaw/plugin-sdk/gateway-runtime"; -import { - addTimerTimeoutGraceMs, - clampTimerTimeoutMs, - MAX_TIMER_TIMEOUT_MS, - MAX_TCP_PORT, - parseStrictNonNegativeInteger, -} from "openclaw/plugin-sdk/number-runtime"; +import { MAX_TCP_PORT } from "openclaw/plugin-sdk/number-runtime"; import { isRecord, normalizeOptionalLowercaseString, } from "openclaw/plugin-sdk/string-coerce-runtime"; -import { sleep } from "../api.js"; +import { registerVoiceCallLogs } from "./cli-call-log.js"; +import { parseCliInteger, writeCliJson, writeCliLine } from "./cli-command-io.js"; +import { + callVoiceCallGateway, + initiateVoiceCall, + isUnknownMethod, + pollContinueGateway, + resolveContinueTimeout, + resolveOperationTimeout, + runGatewayManagerCommand, +} from "./cli-gateway-call.js"; import { resolveVoiceCallStreamExposurePaths, validateProviderConfig, type VoiceCallConfig, } from "./config.js"; -import { - findCallMatchesInStore, - getCallHistoryFromStore, - loadActiveCallsFromStore, -} from "./manager/store.js"; +import { findCallMatchesInStore, loadActiveCallsFromStore } from "./manager/store.js"; import { setVoiceCallStateRuntime, type VoiceCallStateRuntime } from "./runtime-state.js"; import type { VoiceCallRuntime } from "./runtime.js"; import { resolveDefaultVoiceCallStoreDir } from "./store-path.js"; @@ -56,188 +51,6 @@ type SetupStatus = { checks: SetupCheck[]; }; -type VoiceCallGatewayMethod = - | "voicecall.initiate" - | "voicecall.start" - | "voicecall.continue" - | "voicecall.continue.start" - | "voicecall.continue.result" - | "voicecall.speak" - | "voicecall.dtmf" - | "voicecall.end" - | "voicecall.status"; - -type VoiceCallGatewayCallResult = { ok: true; payload: unknown } | { ok: false; error: unknown }; - -const VOICE_CALL_GATEWAY_DEFAULT_TIMEOUT_MS = 5000; -const VOICE_CALL_GATEWAY_OPERATION_TIMEOUT_MS = 30000; -const VOICE_CALL_GATEWAY_TRANSCRIPT_BUFFER_MS = 10000; -const VOICE_CALL_GATEWAY_POLL_INTERVAL_MS = 1000; - -function writeStdoutLine(...values: unknown[]): void { - process.stdout.write(`${format(...values)}\n`); -} - -function writeStdoutJson(value: unknown): void { - process.stdout.write(`${JSON.stringify(value, null, 2)}\n`); -} - -function parseVoiceCallIntOption( - raw: string | undefined, - optionName: string, - opts?: { min?: number; max?: number }, -): number { - const min = opts?.min ?? 0; - const value = raw?.trim() ?? ""; - const parsed = parseStrictNonNegativeInteger(value); - if (parsed === undefined || parsed < min || (opts?.max !== undefined && parsed > opts.max)) { - throw new Error(`Invalid numeric value for ${optionName}: ${raw ?? ""}`); - } - return parsed; -} - -function isGatewayUnavailableForLocalFallback(err: unknown): boolean { - const message = formatErrorMessage(err); - return ( - message.includes("ECONNREFUSED") || - message.includes("ECONNRESET") || - message.includes("EHOSTUNREACH") || - message.includes("ENOTFOUND") || - message.includes("gateway closed (1006") || - message.includes("gateway not connected") - ); -} - -async function callVoiceCallGateway( - method: VoiceCallGatewayMethod, - params?: Record, - opts?: { timeoutMs?: number }, -): Promise { - try { - const timeoutMs = - typeof opts?.timeoutMs === "number" && Number.isFinite(opts.timeoutMs) - ? Math.max(1, Math.ceil(opts.timeoutMs)) - : VOICE_CALL_GATEWAY_DEFAULT_TIMEOUT_MS; - const payload = await callGatewayFromCli( - method, - { json: true, timeout: String(timeoutMs) }, - params, - { progress: false }, - ); - return { ok: true, payload }; - } catch (err) { - if (isGatewayUnavailableForLocalFallback(err)) { - return { ok: false, error: err }; - } - throw err; - } -} - -function resolveGatewayOperationTimeoutMs(config: VoiceCallConfig): number { - return Math.max( - VOICE_CALL_GATEWAY_OPERATION_TIMEOUT_MS, - addTimerTimeoutGraceMs(config.ringTimeoutMs) ?? 1, - ); -} - -function resolveGatewayContinueTimeoutMs(config: VoiceCallConfig): number { - return ( - clampTimerTimeoutMs( - config.transcriptTimeoutMs + - VOICE_CALL_GATEWAY_OPERATION_TIMEOUT_MS + - VOICE_CALL_GATEWAY_TRANSCRIPT_BUFFER_MS, - ) ?? 1 - ); -} - -function resolveVoiceCallDeadlineMs(timeoutMs: number, nowMs = Date.now()): number { - return nowMs + (clampTimerTimeoutMs(timeoutMs) ?? MAX_TIMER_TIMEOUT_MS); -} - -function isUnknownGatewayMethod(err: unknown, method: VoiceCallGatewayMethod): boolean { - return formatErrorMessage(err).includes(`unknown method: ${method}`); -} - -function readGatewayOperationId(payload: unknown): string { - if (isRecord(payload) && typeof payload.operationId === "string" && payload.operationId) { - return payload.operationId; - } - throw new Error("voicecall gateway response missing operationId"); -} - -function readGatewayPollTimeoutMs(payload: unknown, fallbackTimeoutMs: number): number { - if (isRecord(payload) && typeof payload.pollTimeoutMs === "number") { - return clampTimerTimeoutMs(payload.pollTimeoutMs) ?? fallbackTimeoutMs; - } - return fallbackTimeoutMs; -} - -function readCompletedContinueResult( - payload: unknown, -): - | { status: "pending" } - | { status: "completed"; result: unknown } - | { status: "failed"; error: string } { - if (!isRecord(payload)) { - throw new Error("voicecall gateway response missing operation status"); - } - if (payload.status === "pending") { - return { status: "pending" }; - } - if (payload.status === "failed") { - return { - status: "failed", - error: typeof payload.error === "string" ? payload.error : "continue failed", - }; - } - if (payload.status === "completed") { - return { status: "completed", result: payload.result }; - } - throw new Error("voicecall gateway response has unknown operation status"); -} - -async function pollVoiceCallContinueGateway(params: { - operationId: string; - timeoutMs: number; -}): Promise { - const deadlineMs = resolveVoiceCallDeadlineMs(params.timeoutMs); - - for (;;) { - // Sleep already clamps to remaining budget; the gateway RPC must too. - // Otherwise the final poll can overrun the continue deadline by a full RPC timeout. - const remainingMs = deadlineMs - Date.now(); - if (remainingMs <= 0) { - break; - } - const gateway = await callVoiceCallGateway( - "voicecall.continue.result", - { operationId: params.operationId }, - { timeoutMs: Math.min(VOICE_CALL_GATEWAY_DEFAULT_TIMEOUT_MS, remainingMs) }, - ); - if (!gateway.ok) { - throw new Error( - `gateway unavailable while waiting for voicecall continue result: ${formatErrorMessage( - gateway.error, - )}`, - ); - } - const result = readCompletedContinueResult(gateway.payload); - if (result.status === "completed") { - return result.result; - } - if (result.status === "failed") { - throw new Error(result.error); - } - const sleepMs = Math.min(VOICE_CALL_GATEWAY_POLL_INTERVAL_MS, deadlineMs - Date.now()); - if (sleepMs <= 0) { - break; - } - await sleep(sleepMs); - } - - throw new Error("voicecall continue timed out waiting for gateway operation"); -} - function resolveMode(input: string): "off" | "serve" | "funnel" { const raw = normalizeOptionalLowercaseString(input) ?? ""; if (raw === "serve" || raw === "off") { @@ -253,50 +66,6 @@ function resolveDefaultStorePath(config: VoiceCallConfig): string { return path.join(base, "calls.jsonl"); } -function percentile(values: number[], p: number): number { - if (values.length === 0) { - return 0; - } - const sorted = [...values].toSorted((a, b) => a - b); - const idx = Math.min(sorted.length - 1, Math.max(0, Math.ceil((p / 100) * sorted.length) - 1)); - return sorted[idx] ?? 0; -} - -function summarizeSeries(values: number[]): { - count: number; - minMs: number; - maxMs: number; - avgMs: number; - p50Ms: number; - p95Ms: number; -} { - if (values.length === 0) { - return { count: 0, minMs: 0, maxMs: 0, avgMs: 0, p50Ms: 0, p95Ms: 0 }; - } - - const minMs = values.reduce( - (min, value) => (value < min ? value : min), - Number.POSITIVE_INFINITY, - ); - const maxMs = values.reduce( - (max, value) => (value > max ? value : max), - Number.NEGATIVE_INFINITY, - ); - const avgMs = values.reduce((sum, value) => sum + value, 0) / values.length; - return { - count: values.length, - minMs, - maxMs, - avgMs, - p50Ms: percentile(values, 50), - p95Ms: percentile(values, 95), - }; -} - -function resolveCallMode(mode?: string): "notify" | "conversation" | undefined { - return mode === "notify" || mode === "conversation" ? mode : undefined; -} - function buildSetupStatus(config: VoiceCallConfig): SetupStatus { const validation = validateProviderConfig(config); const webhookExposure = resolveWebhookExposureStatus(config); @@ -347,77 +116,12 @@ function buildSetupStatus(config: VoiceCallConfig): SetupStatus { } function writeSetupStatus(status: SetupStatus): void { - writeStdoutLine("Voice Call setup: %s", status.ok ? "OK" : "needs attention"); + writeCliLine("Voice Call setup: %s", status.ok ? "OK" : "needs attention"); for (const check of status.checks) { - writeStdoutLine("%s %s: %s", check.ok ? "OK" : "FAIL", check.id, check.message); + writeCliLine("%s %s: %s", check.ok ? "OK" : "FAIL", check.id, check.message); } } -async function initiateCallAndPrintId(params: { - runtime: VoiceCallRuntime; - to: string; - message?: string; - mode?: string; -}) { - const result = await params.runtime.manager.initiateCall(params.to, undefined, { - message: params.message, - mode: resolveCallMode(params.mode), - }); - if (!result.success) { - throw new Error(result.error || "initiate failed"); - } - writeStdoutJson({ callId: result.callId }); -} - -function writeGatewayCallId(payload: unknown): void { - if (isRecord(payload) && typeof payload.callId === "string") { - writeStdoutJson({ callId: payload.callId }); - return; - } - if (isRecord(payload) && typeof payload.error === "string") { - throw new Error(payload.error); - } - throw new Error("voicecall gateway response missing callId"); -} - -async function initiateCallViaGatewayOrRuntime(params: { - ensureRuntime: () => Promise; - config: VoiceCallConfig; - method: "voicecall.initiate" | "voicecall.start"; - to?: string; - message?: string; - mode?: string; -}) { - const mode = resolveCallMode(params.mode); - const gateway = await callVoiceCallGateway( - params.method, - { - ...(params.to ? { to: params.to } : {}), - ...(params.message ? { message: params.message } : {}), - ...(mode ? { mode } : {}), - }, - { - timeoutMs: resolveGatewayOperationTimeoutMs(params.config), - }, - ); - if (gateway.ok) { - writeGatewayCallId(gateway.payload); - return; - } - - const rt = await params.ensureRuntime(); - const to = params.to ?? rt.config.toNumber; - if (!to) { - throw new Error("Missing --to and no toNumber configured"); - } - await initiateCallAndPrintId({ - runtime: rt, - to, - message: params.message, - mode: params.mode, - }); -} - export function registerVoiceCallCli(params: { program: Command; config: VoiceCallConfig; @@ -443,7 +147,7 @@ export function registerVoiceCallCli(params: { .action((options: { json?: boolean }) => { const status = buildSetupStatus(config); if (options.json) { - writeStdoutJson(status); + writeCliJson(status); return; } writeSetupStatus(status); @@ -472,7 +176,7 @@ export function registerVoiceCallCli(params: { const setup = buildSetupStatus(config); if (!setup.ok) { if (options.json) { - writeStdoutJson({ ok: false, setup }); + writeCliJson({ ok: false, setup }); } else { writeSetupStatus(setup); } @@ -481,57 +185,38 @@ export function registerVoiceCallCli(params: { } if (!options.to) { if (options.json) { - writeStdoutJson({ ok: true, setup, liveCall: false }); + writeCliJson({ ok: true, setup, liveCall: false }); } else { writeSetupStatus(setup); - writeStdoutLine("live-call: skipped (pass --to and --yes to place one)"); + writeCliLine("live-call: skipped (pass --to and --yes to place one)"); } return; } if (!options.yes) { if (options.json) { - writeStdoutJson({ ok: true, setup, liveCall: false, wouldCall: options.to }); + writeCliJson({ ok: true, setup, liveCall: false, wouldCall: options.to }); } else { writeSetupStatus(setup); - writeStdoutLine("live-call: dry run for %s (add --yes to place it)", options.to); + writeCliLine("live-call: dry run for %s (add --yes to place it)", options.to); } return; } - const mode = resolveCallMode(options.mode) ?? "notify"; - const gateway = await callVoiceCallGateway( - "voicecall.start", - { - to: options.to, - ...(options.message ? { message: options.message } : {}), - mode, - }, - { - timeoutMs: resolveGatewayOperationTimeoutMs(config), - }, - ); - let callId: unknown; - if (gateway.ok) { - callId = isRecord(gateway.payload) ? gateway.payload.callId : undefined; - } else { - const rt = await ensureRuntime(); - const result = await rt.manager.initiateCall(options.to, undefined, { - message: options.message, - mode, - }); - if (!result.success) { - throw new Error(result.error || "smoke call failed"); - } - callId = result.callId; - } - if (typeof callId !== "string" || !callId) { - throw new Error("smoke call failed"); - } + const callId = await initiateVoiceCall({ + ensureRuntime, + config, + method: "voicecall.start", + to: options.to, + message: options.message, + mode: options.mode, + defaultMode: "notify", + failureMessage: "smoke call failed", + }); if (options.json) { - writeStdoutJson({ ok: true, setup, liveCall: true, callId }); + writeCliJson({ ok: true, setup, liveCall: true, callId }); return; } writeSetupStatus(setup); - writeStdoutLine("live-call: started %s", callId); + writeCliLine("live-call: started %s", callId); }, ); @@ -549,7 +234,7 @@ export function registerVoiceCallCli(params: { "conversation", ) .action(async (options: { message: string; to?: string; mode?: string }) => { - await initiateCallViaGatewayOrRuntime({ + const callId = await initiateVoiceCall({ ensureRuntime, config, method: "voicecall.initiate", @@ -557,6 +242,7 @@ export function registerVoiceCallCli(params: { message: options.message, mode: options.mode, }); + writeCliJson({ callId }); }); root @@ -570,7 +256,7 @@ export function registerVoiceCallCli(params: { "conversation", ) .action(async (options: { to: string; message?: string; mode?: string }) => { - await initiateCallViaGatewayOrRuntime({ + const callId = await initiateVoiceCall({ ensureRuntime, config, method: "voicecall.start", @@ -578,6 +264,7 @@ export function registerVoiceCallCli(params: { message: options.message, mode: options.mode, }); + writeCliJson({ callId }); }); root @@ -586,54 +273,29 @@ export function registerVoiceCallCli(params: { .requiredOption("--call-id ", "Call ID") .requiredOption("--message ", "Message to speak") .action(async (options: { callId: string; message: string }) => { - let gateway: VoiceCallGatewayCallResult; - try { - gateway = await callVoiceCallGateway( - "voicecall.continue.start", - { - callId: options.callId, - message: options.message, - }, - { - timeoutMs: resolveGatewayOperationTimeoutMs(config), - }, - ); - } catch (err) { - if (!isUnknownGatewayMethod(err, "voicecall.continue.start")) { - throw err; - } - gateway = await callVoiceCallGateway( - "voicecall.continue", - { - callId: options.callId, - message: options.message, - }, - { - timeoutMs: resolveGatewayContinueTimeoutMs(config), - }, - ); - } - if (gateway.ok) { - if (isRecord(gateway.payload) && typeof gateway.payload.operationId === "string") { - const result = await pollVoiceCallContinueGateway({ - operationId: readGatewayOperationId(gateway.payload), - timeoutMs: readGatewayPollTimeoutMs( - gateway.payload, - resolveGatewayContinueTimeoutMs(config), - ), - }); - writeStdoutJson(result); - return; - } - writeStdoutJson(gateway.payload); - return; - } - const rt = await ensureRuntime(); - const result = await rt.manager.continueCall(options.callId, options.message); - if (!result.success) { - throw new Error(result.error || "continue failed"); - } - writeStdoutJson(result); + const gatewayParams = { callId: options.callId, message: options.message }; + const continueTimeoutMs = resolveContinueTimeout(config); + await runGatewayManagerCommand({ + config, + ensureRuntime, + gatewayCall: async () => { + try { + return await callVoiceCallGateway("voicecall.continue.start", gatewayParams, { + timeoutMs: resolveOperationTimeout(config), + }); + } catch (err) { + if (!isUnknownMethod(err, "voicecall.continue.start")) { + throw err; + } + return callVoiceCallGateway("voicecall.continue", gatewayParams, { + timeoutMs: continueTimeoutMs, + }); + } + }, + resolveGatewayPayload: (payload) => pollContinueGateway(payload, continueTimeoutMs), + managerFallback: (manager) => manager.continueCall(options.callId, options.message), + failureLabel: "continue", + }); }); root @@ -642,20 +304,17 @@ export function registerVoiceCallCli(params: { .requiredOption("--call-id ", "Call ID") .requiredOption("--message ", "Message to speak") .action(async (options: { callId: string; message: string }) => { - const gateway = await callVoiceCallGateway("voicecall.speak", { - callId: options.callId, - message: options.message, + await runGatewayManagerCommand({ + config, + ensureRuntime, + gatewayCall: () => + callVoiceCallGateway("voicecall.speak", { + callId: options.callId, + message: options.message, + }), + managerFallback: (manager) => manager.speak(options.callId, options.message), + failureLabel: "speak", }); - if (gateway.ok) { - writeStdoutJson(gateway.payload); - return; - } - const rt = await ensureRuntime(); - const result = await rt.manager.speak(options.callId, options.message); - if (!result.success) { - throw new Error(result.error || "speak failed"); - } - writeStdoutJson(result); }); root @@ -664,20 +323,17 @@ export function registerVoiceCallCli(params: { .requiredOption("--call-id ", "Call ID") .requiredOption("--digits ", "DTMF digits") .action(async (options: { callId: string; digits: string }) => { - const gateway = await callVoiceCallGateway("voicecall.dtmf", { - callId: options.callId, - digits: options.digits, + await runGatewayManagerCommand({ + config, + ensureRuntime, + gatewayCall: () => + callVoiceCallGateway("voicecall.dtmf", { + callId: options.callId, + digits: options.digits, + }), + managerFallback: (manager) => manager.sendDtmf(options.callId, options.digits), + failureLabel: "dtmf", }); - if (gateway.ok) { - writeStdoutJson(gateway.payload); - return; - } - const rt = await ensureRuntime(); - const result = await rt.manager.sendDtmf(options.callId, options.digits); - if (!result.success) { - throw new Error(result.error || "dtmf failed"); - } - writeStdoutJson(result); }); root @@ -685,19 +341,13 @@ export function registerVoiceCallCli(params: { .description("Hang up an active call") .requiredOption("--call-id ", "Call ID") .action(async (options: { callId: string }) => { - const gateway = await callVoiceCallGateway("voicecall.end", { - callId: options.callId, + await runGatewayManagerCommand({ + config, + ensureRuntime, + gatewayCall: () => callVoiceCallGateway("voicecall.end", { callId: options.callId }), + managerFallback: (manager) => manager.endCall(options.callId), + failureLabel: "end", }); - if (gateway.ok) { - writeStdoutJson(gateway.payload); - return; - } - const rt = await ensureRuntime(); - const result = await rt.manager.endCall(options.callId); - if (!result.success) { - throw new Error(result.error || "end failed"); - } - writeStdoutJson(result); }); root @@ -713,15 +363,15 @@ export function registerVoiceCallCli(params: { if (gateway.ok) { if (options.callId && isRecord(gateway.payload)) { if (gateway.payload.found === true && "call" in gateway.payload) { - writeStdoutJson(gateway.payload.call); + writeCliJson(gateway.payload.call); return; } if (gateway.payload.found === false) { - writeStdoutJson({ found: false }); + writeCliJson({ found: false }); return; } } - writeStdoutJson(gateway.payload); + writeCliJson(gateway.payload); return; } // Status is a read-only command. Starting the telephony runtime here would @@ -731,154 +381,20 @@ export function registerVoiceCallCli(params: { if (options.callId) { const persisted = await findCallMatchesInStore(storePath, options.callId); const call = persisted.byCallId ?? persisted.byProviderCallId; - writeStdoutJson(call ?? { found: false }); + writeCliJson(call ?? { found: false }); return; } - writeStdoutJson({ + writeCliJson({ found: true, calls: Array.from(loadActiveCallsFromStore(storePath).activeCalls.values()), }); }); - root - .command("tail") - .description("Tail voice-call JSONL logs (prints new lines; useful during provider tests)") - .option("--file ", "Path to calls.jsonl", resolveDefaultStorePath(config)) - .option("--since ", "Print last N lines first", "25") - .option("--poll ", "Poll interval in ms", "250") - .action(async (options: { file: string; since?: string; poll?: string }) => { - const file = options.file; - const since = parseVoiceCallIntOption(options.since, "--since", { min: 0 }); - const pollMs = parseVoiceCallIntOption(options.poll, "--poll", { min: 50 }); - - const tailSqliteHistory = async (initialLimit: number): Promise => { - ensureHistoryStateRuntime(); - const seen = new Set(); - const printCall = (call: unknown): void => { - const line = JSON.stringify(call); - if (!seen.has(line)) { - seen.add(line); - writeStdoutLine(line); - } - }; - if (initialLimit > 0) { - for (const call of await getCallHistoryFromStore(path.dirname(file), initialLimit)) { - printCall(call); - } - } - for (;;) { - try { - for (const call of await getCallHistoryFromStore(path.dirname(file), 1000)) { - printCall(call); - } - } catch { - // ignore and retry - } - await sleep(pollMs); - } - }; - - if (fs.existsSync(file) && path.basename(file) !== "calls.jsonl") { - const initial = fs.readFileSync(file); - let decoder = new StringDecoder("utf8"); - const initialLines = decoder.write(initial).split("\n"); - let pendingLine = initialLines.pop() ?? ""; - const lines = initialLines.filter(Boolean); - for (const line of lines.slice(Math.max(0, lines.length - since))) { - writeStdoutLine(line); - } - - let offset = initial.length; - let lastObservedSize = initial.length; - for (;;) { - try { - const stat = fs.statSync(file); - // A short read can leave the cursor behind the observed file size; - // compare observed sizes so copytruncate also clears buffered text. - if (stat.size < lastObservedSize) { - offset = 0; - decoder = new StringDecoder("utf8"); - pendingLine = ""; - } - lastObservedSize = stat.size; - if (stat.size > offset) { - const fd = fs.openSync(file, "r"); - try { - const buf = Buffer.alloc(stat.size - offset); - const bytesRead = fs.readSync(fd, buf, 0, buf.length, offset); - offset += bytesRead; - const text = decoder.write(buf.subarray(0, bytesRead)); - const completeLines = `${pendingLine}${text}`.split("\n"); - pendingLine = completeLines.pop() ?? ""; - for (const line of completeLines.filter(Boolean)) { - writeStdoutLine(line); - } - } finally { - fs.closeSync(fd); - } - } - } catch { - // ignore and retry - } - await sleep(pollMs); - } - } else { - await tailSqliteHistory(since); - } - }); - - root - .command("latency") - .description("Summarize turn latency metrics from voice-call JSONL logs") - .option("--file ", "Path to calls.jsonl", resolveDefaultStorePath(config)) - .option("--last ", "Analyze last N records", "200") - .action(async (options: { file: string; last?: string }) => { - const file = options.file; - const last = parseVoiceCallIntOption(options.last, "--last", { min: 1 }); - - if (fs.existsSync(file) && path.basename(file) !== "calls.jsonl") { - const content = fs.readFileSync(file, "utf8"); - const calls = content - .split("\n") - .filter(Boolean) - .slice(-last) - .map((line) => { - try { - const parsed = JSON.parse(line) as { call?: unknown }; - return (parsed.call ?? parsed) as { metadata?: Record }; - } catch { - return null; - } - }) - .filter((call): call is { metadata?: Record } => call !== null); - writeVoiceCallLatencySummary(calls); - } else { - ensureHistoryStateRuntime(); - writeVoiceCallLatencySummary(await getCallHistoryFromStore(path.dirname(file), last)); - } - }); - - function writeVoiceCallLatencySummary(calls: Array<{ metadata?: Record }>) { - const turnLatencyMs: number[] = []; - const listenWaitMs: number[] = []; - - for (const call of calls) { - const latency = call.metadata?.lastTurnLatencyMs; - const listenWait = call.metadata?.lastTurnListenWaitMs; - if (typeof latency === "number" && Number.isFinite(latency)) { - turnLatencyMs.push(latency); - } - if (typeof listenWait === "number" && Number.isFinite(listenWait)) { - listenWaitMs.push(listenWait); - } - } - - writeStdoutJson({ - recordsScanned: calls.length, - turnLatency: summarizeSeries(turnLatencyMs), - listenWait: summarizeSeries(listenWaitMs), - }); - } + registerVoiceCallLogs({ + root, + defaultFile: resolveDefaultStorePath(config), + ensureHistoryStateRuntime, + }); root .command("expose") @@ -890,7 +406,7 @@ export function registerVoiceCallCli(params: { .action( async (options: { mode?: string; port?: string; path?: string; servePath?: string }) => { const mode = resolveMode(options.mode ?? "funnel"); - const servePort = parseVoiceCallIntOption( + const servePort = parseCliInteger( options.port ?? String(config.serve.port ?? 3334), "--port", { min: 1, max: MAX_TCP_PORT }, @@ -909,7 +425,7 @@ export function registerVoiceCallCli(params: { await cleanupTailscaleExposureRoute({ mode: "serve", path: exposurePath }); await cleanupTailscaleExposureRoute({ mode: "funnel", path: exposurePath }); } - writeStdoutJson({ ok: true, mode: "off", path: tsPath, streamPaths }); + writeCliJson({ ok: true, mode: "off", path: tsPath, streamPaths }); return; } @@ -929,7 +445,7 @@ export function registerVoiceCallCli(params: { ? `https://login.tailscale.com/f/${mode}?node=${tsInfo.nodeId}` : null; - writeStdoutJson({ + writeCliJson({ ok: Boolean(publicUrl), mode, path: tsPath, @@ -946,4 +462,3 @@ export function registerVoiceCallCli(params: { }, ); } -/* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/scripts/plugin-sdk-surface-report.mts b/scripts/plugin-sdk-surface-report.mts index 6c1942a305ca..2f2256e4cf64 100644 --- a/scripts/plugin-sdk-surface-report.mts +++ b/scripts/plugin-sdk-surface-report.mts @@ -295,7 +295,9 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env // -2: remove obsolete transcript display helper exports. // +2: lightweight agent config resolution and nonthrowing default-agent lookup. // +1: focused media-store URL/path ingestion (saveMediaSource) off the deprecated barrel. - 4328, + // +2: structural Gateway transport and request-error guards for plugin CLI routing. + // +1: canonical sensitive-URL redactor so plugin CLI errors never print URL userinfo. + 4331, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( @@ -376,7 +378,9 @@ export function readPluginSdkSurfaceBudgets(env: NodeJS.ProcessEnv = process.env // -1: remove the obsolete transcript tool-call predicate. // +2: lightweight agent config resolution and nonthrowing default-agent lookup. // +1: focused media-store URL/path ingestion (saveMediaSource) off the deprecated barrel. - 2572, + // +2: structural Gateway transport and request-error guards for plugin CLI routing. + // +1: canonical sensitive-URL redactor so plugin CLI errors never print URL userinfo. + 2575, env, ), publicDeprecatedExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/gateway/call.ts b/src/gateway/call.ts index 721720dee9a5..c5245a4c968d 100644 --- a/src/gateway/call.ts +++ b/src/gateway/call.ts @@ -307,27 +307,10 @@ export function formatGatewayTransportErrorJson(value: unknown): GatewayTranspor export function formatGatewayClientRequestErrorJson( value: unknown, ): GatewayClientRequestErrorJson | null { - if (!(value instanceof Error) || value.name !== "GatewayClientRequestError") { - return null; - } - const requestError = value as Error & { - gatewayCode?: unknown; - details?: unknown; - retryable?: unknown; - retryAfterMs?: unknown; - }; - if ( - typeof requestError.gatewayCode !== "string" || - requestError.gatewayCode.length === 0 || - requestError.message.length === 0 || - typeof requestError.retryable !== "boolean" || - (requestError.retryAfterMs !== undefined && - (typeof requestError.retryAfterMs !== "number" || - !Number.isInteger(requestError.retryAfterMs) || - requestError.retryAfterMs < 0)) - ) { + if (!isGatewayClientRequestError(value)) { return null; } + const requestError = value; return { ok: false, error: { @@ -343,6 +326,35 @@ export function formatGatewayClientRequestErrorJson( }; } +export function isGatewayClientRequestError(value: unknown): value is Error & { + gatewayCode: string; + details?: unknown; + retryable: boolean; + retryAfterMs?: number; +} { + if (!(value instanceof Error) || value.name !== "GatewayClientRequestError") { + return false; + } + const requestError = value as Error & { + gatewayCode?: unknown; + retryable?: unknown; + retryAfterMs?: unknown; + }; + if ( + typeof requestError.gatewayCode !== "string" || + requestError.gatewayCode.length === 0 || + requestError.message.length === 0 || + typeof requestError.retryable !== "boolean" || + (requestError.retryAfterMs !== undefined && + (typeof requestError.retryAfterMs !== "number" || + !Number.isInteger(requestError.retryAfterMs) || + requestError.retryAfterMs < 0)) + ) { + return false; + } + return true; +} + /** Preserve machine-readable output for auth failures raised before transport startup. */ export function formatGatewayAuthErrorJson(value: unknown): GatewayAuthErrorJson | null { if ( @@ -1035,7 +1047,7 @@ async function executeGatewayRequestWithScopes(params: { ); }, onConnectError: (err) => { - const isGatewayClientRequestError = err.name === "GatewayClientRequestError"; + const gatewayClientRequestError = err.name === "GatewayClientRequestError"; const isAgentRuntimeIdentityConnectError = Boolean(opts.agentRuntimeIdentityToken) && isRequiredAgentRuntimeIdentityConnectError(err); @@ -1043,7 +1055,7 @@ async function executeGatewayRequestWithScopes(params: { isGatewayConnectAssemblyError(err) || isAgentRuntimeIdentityConnectError || isAllowlistedGatewayConnectRequestError(err) || - (surfaceGatewayClientRequestErrors && isGatewayClientRequestError); + (surfaceGatewayClientRequestErrors && gatewayClientRequestError); if (settled || !shouldSurface) { return; } diff --git a/src/gateway/gateway-cli-plugin-routing.test.ts b/src/gateway/gateway-cli-plugin-routing.test.ts new file mode 100644 index 000000000000..5fc9591e3344 --- /dev/null +++ b/src/gateway/gateway-cli-plugin-routing.test.ts @@ -0,0 +1,76 @@ +import { describe, expect, it, vi } from "vitest"; +import { callGatewayFromCli, isGatewayClientRequestError } from "../plugin-sdk/gateway-runtime.js"; +import { createEmptyPluginRegistry } from "../plugins/registry-empty.js"; +import { createPluginGatewayMethodDescriptor } from "./methods/registry.js"; +import type { GatewayRequestHandler } from "./server-methods/types.js"; +import { + getGatewayTestPort, + installGatewayTestHooks, + setTestPluginRegistry, + startTestGatewayServer, +} from "./test-helpers.js"; + +installGatewayTestHooks(); + +describe("gateway-routed plugin CLI calls", () => { + it("routes voicecall status with token auth and preserves wrong-token request errors", async () => { + const handler = vi.fn(({ params, respond }) => { + respond(true, { found: true, call: { callId: params?.callId, state: "active" } }); + }); + const registry = createEmptyPluginRegistry(); + registry.gatewayHandlers["voicecall.status"] = handler; + registry.gatewayMethodDescriptors.push( + createPluginGatewayMethodDescriptor({ + pluginId: "voice-call", + name: "voicecall.status", + handler, + scope: "operator.read", + }), + ); + setTestPluginRegistry(registry); + + const port = await getGatewayTestPort(); + const token = "voice-call-cli-routing-token"; + const url = `ws://127.0.0.1:${port}`; + const server = await startTestGatewayServer(port, { + bind: "loopback", + auth: { mode: "token", token }, + controlUiEnabled: false, + }); + try { + await expect( + callGatewayFromCli( + "voicecall.status", + { url, token, json: true, timeout: "5000" }, + { callId: "call-1" }, + { progress: false }, + ), + ).resolves.toEqual({ + found: true, + call: { callId: "call-1", state: "active" }, + }); + expect(handler).toHaveBeenCalledOnce(); + + let wrongTokenError: unknown; + await callGatewayFromCli( + "voicecall.status", + { url, token: "wrong-token", json: true, timeout: "5000" }, + { callId: "call-1" }, + { progress: false }, + ).catch((error: unknown) => { + wrongTokenError = error; + }); + + expect(isGatewayClientRequestError(wrongTokenError)).toBe(true); + expect(wrongTokenError).toMatchObject({ + name: "GatewayClientRequestError", + gatewayCode: "INVALID_REQUEST", + details: { code: "AUTH_TOKEN_MISMATCH" }, + retryable: false, + }); + expect(handler).toHaveBeenCalledOnce(); + } finally { + await server.close({ reason: "voice-call CLI routing test complete" }); + } + }, 30_000); +}); diff --git a/src/plugin-sdk/gateway-runtime.ts b/src/plugin-sdk/gateway-runtime.ts index ae2b2f957251..7d0791a58b87 100644 --- a/src/plugin-sdk/gateway-runtime.ts +++ b/src/plugin-sdk/gateway-runtime.ts @@ -2,6 +2,10 @@ export { addGatewayClientOptions, callGatewayFromCli } from "../cli/gateway-rpc.js"; export type { GatewayRpcOpts } from "../cli/gateway-rpc.js"; +export { isGatewayClientRequestError, isGatewayTransportError } from "../gateway/call.js"; +// Plugin CLIs echo gateway URLs/close reasons into operator-visible errors; +// they must use the canonical redactor so URL userinfo/tokens never print. +export { redactSensitiveUrlLikeString } from "@openclaw/net-policy/redact-sensitive-url"; export { isLoopbackHost } from "../gateway/net.js"; export async function resolveAdvertisedLanHost(): Promise { const runtime = await import("../infra/advertised-lan-host.js");