mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-23 10:55:31 -06:00
fix(voice-call): survive gateway in-process restart and stop CLI dead-ends (#125458)
* 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.
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
b82d07f466
commit
75fcaba919
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@ import plugin from "./index.js";
|
||||
import { createVoiceCallRuntime } from "./runtime-entry.js";
|
||||
|
||||
type VoiceCallService = Parameters<OpenClawPluginApi["registerService"]>[0];
|
||||
type VoiceCallGatewayHandler = Parameters<OpenClawPluginApi["registerGatewayMethod"]>[1];
|
||||
type VoiceCallTool = {
|
||||
execute: (toolCallId: string, params: unknown) => Promise<VoiceCallToolResult>;
|
||||
};
|
||||
@@ -66,6 +67,7 @@ function registerVoiceCall(params: {
|
||||
}) {
|
||||
let service: VoiceCallService | undefined;
|
||||
let toolFactory: VoiceCallToolFactory | undefined;
|
||||
const gatewayHandlers = new Map<string, VoiceCallGatewayHandler>();
|
||||
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<VoiceCallToolResult> {
|
||||
return tool.execute("call", { action: "initiate_call", message: "hello" });
|
||||
}
|
||||
|
||||
async function executeGatewayCall(registration: ReturnType<typeof registerVoiceCall>) {
|
||||
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<void>();
|
||||
const releaseAStop = createDeferred<void>();
|
||||
@@ -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<string>();
|
||||
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<string>();
|
||||
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);
|
||||
});
|
||||
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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<OpenClawPluginApi["registerService"]>[0]["start"]
|
||||
>[0]["serviceHealth"];
|
||||
stopPromise?: Promise<void>;
|
||||
};
|
||||
|
||||
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<void> {
|
||||
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<VoiceCallRuntime> => {
|
||||
activateVoiceCallRuntimeGeneration(runtimeCoordinator, runtimeGeneration);
|
||||
const ensureRuntimeForGeneration = async (
|
||||
runtimeGeneration: VoiceCallRuntimeGeneration,
|
||||
): Promise<VoiceCallRuntime> => {
|
||||
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<VoiceCallRuntime> => {
|
||||
const ensureRuntime = async (
|
||||
runtimeGeneration = runtimeRegistration.generation,
|
||||
): Promise<VoiceCallRuntime> => {
|
||||
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;
|
||||
}
|
||||
},
|
||||
|
||||
@@ -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 <n>` 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>", "Path to calls.jsonl", params.defaultFile)
|
||||
.option("--since <n>", "Print last N lines first", "25")
|
||||
.option("--poll <ms>", "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<never> => {
|
||||
params.ensureHistoryStateRuntime();
|
||||
const seen = new Set<string>();
|
||||
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>", "Path to calls.jsonl", params.defaultFile)
|
||||
.option("--last <n>", "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));
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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<string, unknown>,
|
||||
opts?: GatewayCallOptions,
|
||||
): Promise<VoiceCallGatewayCallResult> {
|
||||
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<unknown> {
|
||||
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<VoiceCallRuntime>;
|
||||
}): Promise<VoiceCallRuntime> {
|
||||
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<VoiceCallRuntime>;
|
||||
gatewayCall: () => Promise<VoiceCallGatewayCallResult>;
|
||||
resolveGatewayPayload?: (payload: unknown) => Promise<unknown>;
|
||||
managerFallback: (
|
||||
manager: VoiceCallRuntime["manager"],
|
||||
) => Promise<{ success: boolean; error?: string }>;
|
||||
failureLabel: string;
|
||||
}): Promise<void> {
|
||||
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<VoiceCallRuntime>;
|
||||
config: VoiceCallConfig;
|
||||
method: "voicecall.initiate" | "voicecall.start";
|
||||
to?: string;
|
||||
message?: string;
|
||||
mode?: string;
|
||||
defaultMode?: "notify" | "conversation";
|
||||
failureMessage?: string;
|
||||
}): Promise<string> {
|
||||
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;
|
||||
}
|
||||
@@ -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<string, unknown>,
|
||||
config: Record<string, unknown> = {},
|
||||
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<unknown> {
|
||||
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(
|
||||
|
||||
@@ -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<string, unknown>,
|
||||
opts?: { timeoutMs?: number },
|
||||
): Promise<VoiceCallGatewayCallResult> {
|
||||
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<unknown> {
|
||||
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<VoiceCallRuntime>;
|
||||
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 <id>", "Call ID")
|
||||
.requiredOption("--message <text>", "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 <id>", "Call ID")
|
||||
.requiredOption("--message <text>", "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 <id>", "Call ID")
|
||||
.requiredOption("--digits <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 <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>", "Path to calls.jsonl", resolveDefaultStorePath(config))
|
||||
.option("--since <n>", "Print last N lines first", "25")
|
||||
.option("--poll <ms>", "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<never> => {
|
||||
ensureHistoryStateRuntime();
|
||||
const seen = new Set<string>();
|
||||
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>", "Path to calls.jsonl", resolveDefaultStorePath(config))
|
||||
.option("--last <n>", "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<string, unknown> };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
})
|
||||
.filter((call): call is { metadata?: Record<string, unknown> } => call !== null);
|
||||
writeVoiceCallLatencySummary(calls);
|
||||
} else {
|
||||
ensureHistoryStateRuntime();
|
||||
writeVoiceCallLatencySummary(await getCallHistoryFromStore(path.dirname(file), last));
|
||||
}
|
||||
});
|
||||
|
||||
function writeVoiceCallLatencySummary(calls: Array<{ metadata?: Record<string, unknown> }>) {
|
||||
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. */
|
||||
|
||||
@@ -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(
|
||||
|
||||
+33
-21
@@ -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<T>(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<T>(params: {
|
||||
isGatewayConnectAssemblyError(err) ||
|
||||
isAgentRuntimeIdentityConnectError ||
|
||||
isAllowlistedGatewayConnectRequestError(err) ||
|
||||
(surfaceGatewayClientRequestErrors && isGatewayClientRequestError);
|
||||
(surfaceGatewayClientRequestErrors && gatewayClientRequestError);
|
||||
if (settled || !shouldSurface) {
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -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<GatewayRequestHandler>(({ 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);
|
||||
});
|
||||
@@ -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<string | null> {
|
||||
const runtime = await import("../infra/advertised-lan-host.js");
|
||||
|
||||
Reference in New Issue
Block a user