diff --git a/extensions/signal/src/daemon.test.ts b/extensions/signal/src/daemon.test.ts index f903ba76e5de..637d6cf7517c 100644 --- a/extensions/signal/src/daemon.test.ts +++ b/extensions/signal/src/daemon.test.ts @@ -4,7 +4,7 @@ import os from "node:os"; import path from "node:path"; import { PassThrough } from "node:stream"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { spawnSignalDaemon } from "./daemon.js"; +import { spawnSignalDaemon, spawnSignalJsonRpcProcess } from "./daemon.js"; const spawnMock = vi.hoisted(() => vi.fn()); @@ -14,12 +14,14 @@ function createMockChild() { const child = new EventEmitter() as EventEmitter & { pid: number; killed: boolean; + stdin: PassThrough; stdout: PassThrough; stderr: PassThrough; kill: ReturnType; }; child.pid = 1234; child.killed = false; + child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.kill = vi.fn(() => true); @@ -36,12 +38,34 @@ beforeEach(() => { afterEach(() => { vi.useRealTimers(); + child.stdin.end(); child.stdout.end(); child.stderr.end(); child.removeAllListeners(); }); describe("spawnSignalDaemon", () => { + it("owns setup JSON-RPC through the child pipes", () => { + const process = spawnSignalJsonRpcProcess({ + cliPath: "signal-cli", + configPath: "~/.openclaw/signal-cli", + }); + + expect(spawnMock).toHaveBeenCalledWith( + "signal-cli", + [ + "--config", + path.join(os.homedir(), ".openclaw/signal-cli"), + "jsonRpc", + "--receive-mode", + "manual", + ], + { stdio: ["pipe", "pipe", "pipe"] }, + ); + expect(process.stdin).toBe(child.stdin); + expect(process.stdout).toBe(child.stdout); + }); + it("expands home-relative configPath before passing it to signal-cli", () => { spawnSignalDaemon({ cliPath: "signal-cli", diff --git a/extensions/signal/src/daemon.ts b/extensions/signal/src/daemon.ts index 44d7029fe018..28096fe6fdd0 100644 --- a/extensions/signal/src/daemon.ts +++ b/extensions/signal/src/daemon.ts @@ -27,12 +27,17 @@ export type SignalDaemonHandle = { const SIGNAL_DAEMON_STOP_KILL_TIMEOUT_MS = 1_500; -type SignalDaemonExitEvent = { +export type SignalDaemonExitEvent = { source: "process" | "spawn-error"; code: number | null; signal: NodeJS.Signals | null; }; +export type SignalJsonRpcProcess = SignalDaemonHandle & { + stdin: NodeJS.WritableStream; + stdout: NodeJS.ReadableStream; +}; + export function formatSignalDaemonExit(exit: SignalDaemonExitEvent): string { return `signal daemon exited (source=${exit.source} code=${exit.code ?? "null"} signal=${exit.signal ?? "null"})`; } @@ -121,15 +126,14 @@ function buildDaemonArgs(opts: SignalDaemonOpts): string[] { return args; } -export function spawnSignalDaemon(opts: SignalDaemonOpts): SignalDaemonHandle { - const args = buildDaemonArgs(opts); - // The executable is operator-selected or setup-discovered signal-cli. - // Runtime message content only flows through the daemon HTTP API, not argv. - const child = spawn(opts.cliPath, args, { - stdio: ["ignore", "pipe", "pipe"], - }); - const log = opts.runtime?.log ?? (() => {}); - const error = opts.runtime?.error ?? (() => {}); +function bindSignalProcess(params: { + child: ReturnType; + runtime?: RuntimeEnv; + bindStdout: boolean; +}): SignalDaemonHandle { + const { child } = params; + const log = params.runtime?.log ?? (() => {}); + const error = params.runtime?.error ?? (() => {}); let exited = false; let settledExit = false; let stopPromise: Promise | undefined; @@ -146,7 +150,9 @@ export function spawnSignalDaemon(opts: SignalDaemonOpts): SignalDaemonHandle { resolveExit(value); }; - bindSignalCliOutput({ stream: child.stdout, log, error }); + if (params.bindStdout) { + bindSignalCliOutput({ stream: child.stdout, log, error }); + } bindSignalCliOutput({ stream: child.stderr, log, error }); child.once("exit", (code, signal) => { settleExit({ @@ -216,3 +222,30 @@ export function spawnSignalDaemon(opts: SignalDaemonOpts): SignalDaemonHandle { }, }; } + +export function spawnSignalDaemon(opts: SignalDaemonOpts): SignalDaemonHandle { + // The executable is operator-selected or setup-discovered signal-cli. + // Runtime message content only flows through the daemon HTTP API, not argv. + const child = spawn(opts.cliPath, buildDaemonArgs(opts), { + stdio: ["ignore", "pipe", "pipe"], + }); + return bindSignalProcess({ child, runtime: opts.runtime, bindStdout: true }); +} + +export function spawnSignalJsonRpcProcess(opts: { + cliPath: string; + configPath?: string; + runtime?: RuntimeEnv; +}): SignalJsonRpcProcess { + const args: string[] = []; + if (opts.configPath?.trim()) { + args.push("--config", resolveSignalCliConfigPath(opts.configPath)); + } + args.push("jsonRpc", "--receive-mode", "manual"); + const child = spawn(opts.cliPath, args, { stdio: ["pipe", "pipe", "pipe"] }); + const handle = bindSignalProcess({ child, runtime: opts.runtime, bindStdout: false }); + if (!child.stdin || !child.stdout) { + throw new Error("signal-cli jsonRpc pipes unavailable"); + } + return { ...handle, stdin: child.stdin, stdout: child.stdout }; +} diff --git a/extensions/signal/src/link-rpc.test.ts b/extensions/signal/src/link-rpc.test.ts new file mode 100644 index 000000000000..81b75747efc1 --- /dev/null +++ b/extensions/signal/src/link-rpc.test.ts @@ -0,0 +1,104 @@ +// Signal link RPC tests cover child-owned JSON-line request routing. +import { once } from "node:events"; +import { PassThrough } from "node:stream"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { SignalDaemonExitEvent, SignalJsonRpcProcess } from "./daemon.js"; +import { createSignalLinkRpcClient } from "./link-rpc.js"; + +const mocks = vi.hoisted(() => ({ spawn: vi.fn() })); + +vi.mock("./daemon.js", async (importOriginal) => ({ + ...(await importOriginal()), + spawnSignalJsonRpcProcess: mocks.spawn, +})); + +function createProcess() { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + let resolveExit!: (exit: SignalDaemonExitEvent) => void; + const exited = new Promise((resolve) => { + resolveExit = resolve; + }); + const process: SignalJsonRpcProcess = { + stdin, + stdout, + exited, + isExited: () => false, + stop: vi.fn(async () => {}), + }; + return { process, resolveExit }; +} + +let fixture: ReturnType; + +beforeEach(() => { + fixture = createProcess(); + mocks.spawn.mockReset().mockReturnValue(fixture.process); +}); + +afterEach(() => { + (fixture.process.stdin as PassThrough).end(); + (fixture.process.stdout as PassThrough).end(); +}); + +describe("SignalLinkRpcClient", () => { + it("routes a response through the owned child pipes", async () => { + const client = createSignalLinkRpcClient({ + cliPath: "signal-cli", + configPath: "/tmp/signal", + }); + const written = once(fixture.process.stdin, "data"); + const response = client.request("startLink", undefined, { maxResponseBytes: 1024 }); + const [chunk] = await written; + const request = JSON.parse(String(chunk)) as { id: string; method: string }; + + expect(request.method).toBe("startLink"); + fixture.process.stdout.emit( + "data", + `${JSON.stringify({ jsonrpc: "2.0", id: request.id, result: { deviceLinkUri: "sgnl://linkdevice?uuid=test" } })}\n`, + ); + + await expect(response).resolves.toEqual({ + deviceLinkUri: "sgnl://linkdevice?uuid=test", + }); + expect(mocks.spawn).toHaveBeenCalledWith({ + cliPath: "signal-cli", + configPath: "/tmp/signal", + }); + await client.stop(); + expect(fixture.process.stop).toHaveBeenCalledOnce(); + }); + + it("rejects a pending request when the owned child exits", async () => { + const client = createSignalLinkRpcClient({ cliPath: "signal-cli" }); + const response = client.request("listAccounts"); + + fixture.resolveExit({ source: "process", code: 1, signal: null }); + + await expect(response).rejects.toThrow("signal daemon exited"); + }); + + it("rejects a pending request when the child closes stdin", async () => { + const client = createSignalLinkRpcClient({ cliPath: "signal-cli" }); + const response = client.request("listAccounts"); + + fixture.process.stdin.emit("error", new Error("write EPIPE")); + + await expect(response).rejects.toThrow("write EPIPE"); + }); + + it("rejects an oversized response before parsing it", async () => { + const client = createSignalLinkRpcClient({ cliPath: "signal-cli" }); + const written = once(fixture.process.stdin, "data"); + const response = client.request("listAccounts", undefined, { maxResponseBytes: 64 }); + const [chunk] = await written; + const request = JSON.parse(String(chunk)) as { id: string }; + + fixture.process.stdout.emit( + "data", + `${JSON.stringify({ id: request.id, result: "x".repeat(128) })}\n`, + ); + + await expect(response).rejects.toThrow("response exceeded size limit"); + }); +}); diff --git a/extensions/signal/src/link-rpc.ts b/extensions/signal/src/link-rpc.ts new file mode 100644 index 000000000000..2bffe47e3e42 --- /dev/null +++ b/extensions/signal/src/link-rpc.ts @@ -0,0 +1,165 @@ +// Setup-only signal-cli JSON-RPC transport bound to one child process. +import { createInterface } from "node:readline"; +import { generateSecureUuid } from "openclaw/plugin-sdk/core"; +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; +import type { SignalRpcOptions } from "./client.js"; +import { + formatSignalDaemonExit, + spawnSignalJsonRpcProcess, + type SignalJsonRpcProcess, +} from "./daemon.js"; + +type PendingRequest = { + id: string; + maxResponseBytes: number; + resolve: (value: unknown) => void; + reject: (error: Error) => void; + timer: ReturnType; +}; + +export type SignalLinkRpcClient = { + request: ( + method: string, + params?: Record, + options?: Pick, + ) => Promise; + stop: () => Promise; +}; + +const DEFAULT_TIMEOUT_MS = 10_000; +const DEFAULT_MAX_RESPONSE_BYTES = 16 * 1024; + +class SignalLinkRpcProcessClient implements SignalLinkRpcClient { + private readonly lines; + private pending: PendingRequest | undefined; + private terminalError: Error | undefined; + + constructor( + private readonly process: SignalJsonRpcProcess, + private readonly abortSignal?: AbortSignal, + ) { + this.lines = createInterface({ input: process.stdout }); + this.lines.on("line", this.handleLine); + process.stdin.on("error", this.fail); + process.stdout.on("error", this.fail); + void process.exited.then((exit) => this.fail(new Error(formatSignalDaemonExit(exit)))); + abortSignal?.addEventListener("abort", this.onAbort, { once: true }); + } + + async request( + method: string, + params?: Record, + options?: Pick, + ): Promise { + if (this.terminalError) { + throw this.terminalError; + } + if (this.pending) { + throw new Error("signal-cli link RPC request already in flight"); + } + this.abortSignal?.throwIfAborted(); + const id = generateSecureUuid(); + const timeoutMs = this.positiveInteger(options?.timeoutMs, DEFAULT_TIMEOUT_MS); + const maxResponseBytes = this.positiveInteger( + options?.maxResponseBytes, + DEFAULT_MAX_RESPONSE_BYTES, + ); + const response = new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending = undefined; + reject(new Error(`signal-cli jsonRpc timeout (${method})`)); + }, timeoutMs); + timer.unref?.(); + this.pending = { + id, + maxResponseBytes, + resolve: (value) => resolve(value as T), + reject, + timer, + }; + }); + try { + this.process.stdin.write( + `${JSON.stringify({ jsonrpc: "2.0", method, params, id })}\n`, + (error?: Error | null) => { + if (error) { + this.fail(error); + } + }, + ); + } catch (error) { + this.fail(error instanceof Error ? error : new Error(String(error))); + } + return await response; + } + + async stop(): Promise { + this.abortSignal?.removeEventListener("abort", this.onAbort); + this.lines.close(); + await this.process.stop(); + } + + private positiveInteger(value: number | undefined, fallback: number): number { + return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : fallback; + } + + private readonly onAbort = () => void this.process.stop(); + + private readonly handleLine = (line: string) => { + const pending = this.pending; + if (!pending || !line.trim()) { + return; + } + if (Buffer.byteLength(line) > pending.maxResponseBytes) { + this.fail(new Error("signal-cli jsonRpc response exceeded size limit")); + return; + } + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + this.fail(new Error("signal-cli jsonRpc returned malformed JSON")); + return; + } + if (!isRecord(parsed) || String(parsed.id ?? "") !== pending.id) { + return; + } + this.pending = undefined; + clearTimeout(pending.timer); + if (isRecord(parsed.error)) { + const code = typeof parsed.error.code === "number" ? ` ${parsed.error.code}` : ""; + const message = + typeof parsed.error.message === "string" + ? parsed.error.message.slice(0, 512) + : "request failed"; + pending.reject(new Error(`signal-cli jsonRpc${code}: ${message}`)); + } else if (Object.hasOwn(parsed, "result")) { + pending.resolve(parsed.result); + } else { + pending.reject(new Error("signal-cli jsonRpc returned an invalid response")); + } + }; + + private readonly fail = (error: Error) => { + if (this.terminalError) { + return; + } + this.terminalError = error; + const pending = this.pending; + this.pending = undefined; + if (pending) { + clearTimeout(pending.timer); + pending.reject(error); + } + void this.process.stop(); + }; +} + +export function createSignalLinkRpcClient(options: { + cliPath: string; + configPath?: string; + abortSignal?: AbortSignal; +}): SignalLinkRpcClient { + const { abortSignal, ...processOptions } = options; + return new SignalLinkRpcProcessClient(spawnSignalJsonRpcProcess(processOptions), abortSignal); +} diff --git a/extensions/signal/src/setup-surface.test.ts b/extensions/signal/src/setup-surface.test.ts index c9e50cc23532..f9a9c3ea3ba6 100644 --- a/extensions/signal/src/setup-surface.test.ts +++ b/extensions/signal/src/setup-surface.test.ts @@ -8,22 +8,17 @@ import { signalSetupWizard } from "./setup-surface.js"; const mocks = vi.hoisted(() => ({ detectBinary: vi.fn(), installSignalCli: vi.fn(), + createLinkClient: vi.fn(), + linkStop: vi.fn(async () => {}), rpc: vi.fn(), - spawnDaemon: vi.fn(), - waitReady: vi.fn(), })); vi.mock("openclaw/plugin-sdk/setup-tools", async (importOriginal) => ({ ...(await importOriginal()), detectBinary: mocks.detectBinary, })); -vi.mock("./client.js", () => ({ signalRpcRequest: mocks.rpc })); -vi.mock("./daemon.js", () => ({ spawnSignalDaemon: mocks.spawnDaemon })); -vi.mock("./daemon-lifecycle.js", async (importOriginal) => ({ - ...(await importOriginal()), - waitForSignalDaemonReady: mocks.waitReady, -})); vi.mock("./install-signal-cli.js", () => ({ installSignalCli: mocks.installSignalCli })); +vi.mock("./link-rpc.js", () => ({ createSignalLinkRpcClient: mocks.createLinkClient })); type PrepareParams = Parameters>[0]; @@ -52,14 +47,6 @@ function createPrompter( }; } -function createDaemonHandle() { - return { - exited: new Promise(() => {}), - isExited: () => false, - stop: vi.fn(async () => {}), - }; -} - async function prepareSignal(params: { cfg?: OpenClawConfig; prompter?: WizardPrompter; @@ -94,8 +81,10 @@ beforeEach(() => { vi.clearAllMocks(); mocks.detectBinary.mockResolvedValue(true); mocks.installSignalCli.mockResolvedValue({ ok: true, cliPath: "/tools/signal-cli" }); - mocks.spawnDaemon.mockImplementation(() => createDaemonHandle()); - mocks.waitReady.mockResolvedValue(undefined); + mocks.createLinkClient.mockImplementation(() => ({ + request: mocks.rpc, + stop: mocks.linkStop, + })); }); describe("Signal hosted setup linking", () => { @@ -126,13 +115,10 @@ describe("Signal hosted setup linking", () => { prompter: prompt.prompter, }); - expect(mocks.spawnDaemon).toHaveBeenCalledWith({ + expect(mocks.createLinkClient).toHaveBeenCalledWith({ cliPath: "signal-cli", - httpHost: "127.0.0.1", - httpPort: 8080, - receiveMode: "manual", + abortSignal: expect.any(AbortSignal), }); - expect(mocks.spawnDaemon.mock.calls[0]?.[0]).not.toHaveProperty("account"); expect(events).toEqual(["listAccounts", "startLink", "finishLink", "qrCode"]); expect(prompt.qrCode).toHaveBeenCalledWith({ title: "Link Signal", @@ -145,7 +131,7 @@ describe("Signal hosted setup linking", () => { signalNumber: "+15555550123", [SIGNAL_LINK_COMPLETED_CREDENTIAL]: "true", }); - expect(mocks.spawnDaemon.mock.results[0]?.value.stop).toHaveBeenCalledOnce(); + expect(mocks.linkStop).toHaveBeenCalledOnce(); const numberInput = signalSetupWizard.textInputs?.find( (input) => input.inputKey === "signalNumber", @@ -201,7 +187,7 @@ describe("Signal hosted setup linking", () => { ).rejects.toBe(guardError); expect(beforePersistentEffect).toHaveBeenCalledOnce(); - expect(mocks.spawnDaemon).not.toHaveBeenCalled(); + expect(mocks.createLinkClient).not.toHaveBeenCalled(); expect(mocks.rpc).not.toHaveBeenCalled(); expect(prompt.qrCode).not.toHaveBeenCalled(); }); @@ -239,7 +225,7 @@ describe("Signal hosted setup linking", () => { includeSignal, }); - expect(mocks.spawnDaemon).not.toHaveBeenCalled(); + expect(mocks.createLinkClient).not.toHaveBeenCalled(); expect(result?.credentialValues?.signalNumber).toBeUndefined(); expect( await signalSetupWizard.completionNote?.shouldShow?.({ @@ -276,7 +262,7 @@ describe("Signal hosted setup linking", () => { expect(notes).toContain("Automatic Signal linking could not complete"); expect(notes).not.toContain("private-token"); expect(notes).not.toContain("private-number"); - expect(mocks.spawnDaemon.mock.results[0]?.value.stop).toHaveBeenCalledOnce(); + expect(mocks.linkStop).toHaveBeenCalledOnce(); }); it("aborts linking and reaps its daemon without showing a dependency failure", async () => { @@ -301,6 +287,6 @@ describe("Signal hosted setup linking", () => { ).rejects.toBeInstanceOf(WizardCancelledError); expect(prompt.note).not.toHaveBeenCalled(); - expect(mocks.spawnDaemon.mock.results[0]?.value.stop).toHaveBeenCalledOnce(); + expect(mocks.linkStop).toHaveBeenCalledOnce(); }); }); diff --git a/extensions/signal/src/setup-surface.ts b/extensions/signal/src/setup-surface.ts index e39b8338482b..3372f6d3bc3f 100644 --- a/extensions/signal/src/setup-surface.ts +++ b/extensions/signal/src/setup-surface.ts @@ -12,10 +12,8 @@ import { import { detectBinary, formatCliCommand } from "openclaw/plugin-sdk/setup-tools"; import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; import { listSignalAccountIds, resolveSignalAccount } from "./accounts.js"; -import { signalRpcRequest } from "./client.js"; -import { createSignalDaemonLifecycle, waitForSignalDaemonReady } from "./daemon-lifecycle.js"; -import { spawnSignalDaemon } from "./daemon.js"; import { installSignalCli } from "./install-signal-cli.js"; +import { createSignalLinkRpcClient, type SignalLinkRpcClient } from "./link-rpc.js"; import { createSignalCliPathTextInput, normalizeSignalAccountInput, @@ -135,38 +133,25 @@ async function prepareManagedSignalLink(params: { await params.options?.beforePersistentEffect?.(); signal.throwIfAborted(); - const lifecycle = createSignalDaemonLifecycle({ abortSignal: signal }); - const onAbort = () => void lifecycle.stop(); - signal.addEventListener("abort", onAbort, { once: true }); + let linkClient: SignalLinkRpcClient | undefined; try { let accounts: string[]; let deviceLinkUri: string | undefined; try { - const daemon = spawnSignalDaemon({ + linkClient = createSignalLinkRpcClient({ cliPath: params.cliPath, ...(transport.configPath ? { configPath: transport.configPath } : {}), - httpHost: transport.httpHost, - httpPort: transport.httpPort, - receiveMode: "manual", - }); - lifecycle.attach(daemon); - await waitForSignalDaemonReady({ - baseUrl: transport.baseUrl, - abortSignal: lifecycle.abortSignal, - timeoutMs: transport.startupTimeoutMs, - logAfterMs: transport.startupTimeoutMs, - runtime: { ...params.runtime, log: () => {}, error: () => {} }, + abortSignal: signal, }); accounts = parseSignalAccounts( - await signalRpcRequest("listAccounts", undefined, { - baseUrl: transport.baseUrl, + await linkClient.request("listAccounts", undefined, { + timeoutMs: transport.startupTimeoutMs, maxResponseBytes: SIGNAL_LINK_RPC_MAX_BYTES, }), ); if (accounts.length === 0) { deviceLinkUri = parseSignalLinkUri( - await signalRpcRequest("startLink", undefined, { - baseUrl: transport.baseUrl, + await linkClient.request("startLink", undefined, { timeoutMs: 35_000, maxResponseBytes: SIGNAL_LINK_RPC_MAX_BYTES, }), @@ -195,15 +180,16 @@ async function prepareManagedSignalLink(params: { signal.throwIfAborted(); try { - const settled = signalRpcRequest( - "finishLink", - { deviceLinkUri, deviceName: "OpenClaw" }, - { - baseUrl: transport.baseUrl, - timeoutMs: SIGNAL_LINK_EXPIRES_IN_MS + 5_000, - maxResponseBytes: SIGNAL_LINK_RPC_MAX_BYTES, - }, - ).then(parseLinkedSignalNumber); + const settled = linkClient + .request( + "finishLink", + { deviceLinkUri, deviceName: "OpenClaw" }, + { + timeoutMs: SIGNAL_LINK_EXPIRES_IN_MS + 5_000, + maxResponseBytes: SIGNAL_LINK_RPC_MAX_BYTES, + }, + ) + .then(parseLinkedSignalNumber); return await params.prompter.qrCode({ title: "Link Signal", message: "In Signal, open Settings → Linked devices and scan this QR code.", @@ -219,8 +205,7 @@ async function prepareManagedSignalLink(params: { return undefined; } } finally { - signal.removeEventListener("abort", onAbort); - await lifecycle.stop(); + await linkClient?.stop(); } }