/** Tests ACP server startup readiness, Gateway bootstrap, and shutdown wiring. */ import { ReadableStream as NodeReadableStream } from "node:stream/web"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import type { resolveGatewayClientBootstrap } from "../gateway/client-bootstrap.js"; type GatewayClientCallbacks = { onEvent?: (evt: { event: string; payload?: unknown }) => void; onHelloOk?: () => void; onConnectError?: (err: Error) => void; onClose?: (code: number, reason: string) => void; }; type GatewayClientAuth = { token?: string; password?: string; }; type ResolveGatewayClientBootstrap = typeof resolveGatewayClientBootstrap; type GatewayClientOptions = GatewayClientCallbacks & GatewayClientAuth & { caps?: string[]; tlsFingerprint?: string; url?: string; }; type MockAcpStream = { writable: WritableStream; readable: ReadableStream; }; const mockState = vi.hoisted(() => ({ acpProtocolVersion: 1, acpInputMessages: [] as unknown[], rawInputChunks: [] as Uint8Array[], gateways: [] as MockGatewayClient[], gatewayAuth: [] as GatewayClientAuth[], gatewayOptions: [] as GatewayClientOptions[], sqliteEventLedgers: [] as unknown[], agentOptions: [] as unknown[], agentSideConnectionCtor: vi.fn(), closeAgentSideConnection: null as (() => void) | null, closeAcpInput: null as (() => void) | null, agentHandleGatewayEvent: vi.fn(async (_evt: unknown) => {}), agentStart: vi.fn(), agentShutdown: vi.fn(), routeLogsToStderr: vi.fn(), startProxy: vi.fn(async (_configForTest: unknown) => null as unknown), stopProxy: vi.fn(async (_handle: unknown) => {}), closeOpenClawStateDatabaseAsync: vi.fn<() => Promise>(async () => {}), gatewayStopDeferred: null as { resolve: () => void; promise: Promise; } | null, resolveGatewayClientBootstrap: vi.fn(), })); vi.mock("node:stream", async (importOriginal) => { const actual = await importOriginal(); vi.spyOn(actual.Readable, "toWeb").mockImplementation( () => new NodeReadableStream({ start(controller) { for (const chunk of mockState.rawInputChunks) { controller.enqueue(chunk); } }, }), ); vi.spyOn(actual.Writable, "toWeb").mockImplementation(() => new WritableStream()); return actual; }); class MockGatewayClient { private callbacks: GatewayClientCallbacks; constructor(opts: GatewayClientOptions) { this.callbacks = opts; mockState.gatewayOptions.push(opts); mockState.gatewayAuth.push({ token: opts.token, password: opts.password }); mockState.gateways.push(this); } start(): void {} stop(): void { this.callbacks.onClose?.(1000, "gateway stopped"); } async stopAndWait(): Promise { if (mockState.gatewayStopDeferred) { await mockState.gatewayStopDeferred.promise; } this.stop(); } emitHello(): void { this.callbacks.onHelloOk?.(); } emitConnectError(message: string): void { this.callbacks.onConnectError?.(new Error(message)); } emitEvent(event: { event: string; payload?: unknown }): void { this.callbacks.onEvent?.(event); } } vi.mock("@agentclientprotocol/sdk", () => ({ AGENT_METHODS: { initialize: "initialize", }, AgentSideConnection: function AgentSideConnection( factory: (conn: unknown) => unknown, stream: unknown, ) { mockState.agentSideConnectionCtor(factory, stream); factory({}); return { closed: new Promise((resolve) => { mockState.closeAgentSideConnection = resolve; }), }; }, PROTOCOL_VERSION: mockState.acpProtocolVersion, ndJsonStream: vi.fn(() => ({ writable: new WritableStream(), readable: new ReadableStream({ start(controller) { for (const message of mockState.acpInputMessages) { controller.enqueue(message); } mockState.closeAcpInput = () => controller.close(); }, }), })), })); vi.mock("../config/config.js", () => { const loadConfig = () => ({ gateway: { mode: "local", }, }); return { getRuntimeConfig: loadConfig, loadConfig, resolveGatewayPort: vi.fn(() => 18_789), }; }); vi.mock("../gateway/call.js", () => ({ callGateway: vi.fn(), buildGatewayConnectionDetails: ({ url }: { url?: string }) => { if (typeof url === "string" && url.trim().length > 0) { return { url: url.trim(), urlSource: "cli --url", message: `Gateway target: ${url.trim()}`, }; } return { url: "ws://127.0.0.1:18789", urlSource: "local loopback", message: "Gateway target: ws://127.0.0.1:18789", }; }, })); vi.mock("../gateway/client-bootstrap.js", () => ({ resolveGatewayClientBootstrap: mockState.resolveGatewayClientBootstrap, })); vi.mock("../gateway/client.js", () => ({ GatewayClient: MockGatewayClient, })); vi.mock("../../packages/gateway-client/src/readiness.js", () => ({ startGatewayClientWhenEventLoopReady: vi.fn(async (client: MockGatewayClient) => { client.start(); return { ready: true, elapsedMs: 0, maxDriftMs: 0, checks: 2, aborted: false, }; }), })); vi.mock("../infra/is-main.js", () => ({ isMainModule: () => false, })); vi.mock("../logging/console.js", async (importOriginal) => { const actual = await importOriginal(); return { ...actual, routeLogsToStderr: () => mockState.routeLogsToStderr(), }; }); vi.mock("../state/openclaw-state-db.js", () => ({ closeOpenClawStateDatabaseAsync: () => mockState.closeOpenClawStateDatabaseAsync(), })); vi.mock("./event-ledger.js", () => ({ createSqliteAcpEventLedger: vi.fn(() => { const ledger = { kind: "sqlite-acp-event-ledger" }; mockState.sqliteEventLedgers.push(ledger); return ledger; }), })); vi.mock("../infra/net/proxy/proxy-lifecycle.js", () => ({ startProxy: (config: unknown) => mockState.startProxy(config), stopProxy: (handle: unknown) => mockState.stopProxy(handle), })); vi.mock("./translator.js", () => ({ AcpGatewayAgent: class { constructor(_connection: unknown, _gateway: unknown, opts: unknown) { mockState.agentOptions.push(opts); } start(): void { mockState.agentStart(); } shutdown(): void { mockState.agentShutdown(); } handleGatewayReconnect(): void {} handleGatewayDisconnect(): void {} async handleGatewayEvent(event: unknown): Promise { await mockState.agentHandleGatewayEvent(event); } }, })); describe("serveAcpGateway startup", () => { let serveAcpGateway: typeof import("./server.js").serveAcpGateway; function getMockGateway() { const gateway = mockState.gateways[0]; if (!gateway) { throw new Error("Expected mocked gateway instance"); } return gateway; } function getGatewayBootstrapParams(): { env?: unknown; gatewayUrl?: unknown } { const firstCall = mockState.resolveGatewayClientBootstrap.mock.calls[0]; if (!firstCall) { throw new Error("Expected gateway bootstrap resolution call"); } const params = firstCall[0]; if (!params || typeof params !== "object") { throw new Error("Expected gateway bootstrap params"); } return params; } function captureProcessSignalHandlers() { const signalHandlers = new Map void>(); const onceSpy = vi.spyOn(process, "once").mockImplementation((( signal: NodeJS.Signals, handler: () => void, ) => { signalHandlers.set(signal, handler); return process; }) as typeof process.once); return { signalHandlers, onceSpy }; } async function emitHelloAndWaitForAgentSideConnection() { await vi.waitFor(() => { expect(mockState.gateways).toHaveLength(1); }); const gateway = getMockGateway(); gateway.emitHello(); await vi.waitFor(() => { expect(mockState.agentSideConnectionCtor).toHaveBeenCalledTimes(1); }); } function getCapturedAcpStream(): MockAcpStream { const stream = mockState.agentSideConnectionCtor.mock.calls[0]?.[1]; if ( !stream || typeof stream !== "object" || !(stream as MockAcpStream).readable || !(stream as MockAcpStream).writable ) { throw new Error("Expected AgentSideConnection stream"); } return stream as MockAcpStream; } async function readCapturedAcpMessages(): Promise { const reader = getCapturedAcpStream().readable.getReader(); const messages: unknown[] = []; try { while (true) { const { done, value } = await reader.read(); if (done) { return messages; } messages.push(value); } } finally { reader.releaseLock(); } } async function captureAcpMessagesAfterStartup(inputMessages: unknown[]): Promise { mockState.acpInputMessages.push(...inputMessages); const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); const servePromise = serveAcpGateway({}); try { await emitHelloAndWaitForAgentSideConnection(); mockState.closeAcpInput?.(); return await readCapturedAcpMessages(); } finally { signalHandlers.get("SIGINT")?.(); await servePromise; onceSpy.mockRestore(); } } async function stopServeWithSigint( signalHandlers: Map void>, servePromise: Promise, ) { signalHandlers.get("SIGINT")?.(); await servePromise; } beforeAll(async () => { // Vitest workers have closed stdin; model the open ACP transport used by // these startup tests. Closed-stdin behavior has process-level coverage. Object.defineProperty(process.stdin, "readableEnded", { configurable: true, value: false, }); Object.defineProperty(process.stdin, "readableLength", { configurable: true, value: 1, }); ({ serveAcpGateway } = await import("./server.js")); }); afterAll(() => { const testStdin = process.stdin as unknown as { readableEnded?: boolean; readableLength?: number; }; delete testStdin.readableEnded; delete testStdin.readableLength; }); beforeEach(async () => { mockState.acpInputMessages.length = 0; mockState.rawInputChunks.length = 0; mockState.gateways.length = 0; mockState.gatewayAuth.length = 0; mockState.gatewayOptions.length = 0; mockState.sqliteEventLedgers.length = 0; mockState.agentOptions.length = 0; mockState.agentSideConnectionCtor.mockReset(); mockState.closeAgentSideConnection = null; mockState.closeAcpInput = null; mockState.agentHandleGatewayEvent.mockReset(); mockState.agentStart.mockReset(); mockState.agentShutdown.mockReset(); mockState.routeLogsToStderr.mockReset(); mockState.startProxy.mockReset(); mockState.stopProxy.mockReset(); mockState.closeOpenClawStateDatabaseAsync.mockReset(); mockState.gatewayStopDeferred = null; mockState.startProxy.mockResolvedValue(null); mockState.stopProxy.mockResolvedValue(undefined); mockState.resolveGatewayClientBootstrap.mockReset(); mockState.resolveGatewayClientBootstrap.mockResolvedValue({ url: "ws://127.0.0.1:18789", urlSource: "local loopback", connectionDetails: { url: "ws://127.0.0.1:18789", urlSource: "local loopback", message: "Gateway target: ws://127.0.0.1:18789", }, auth: { token: undefined, password: undefined, }, }); }); it("preserves console exports for a co-sharded subsystem logger", async () => { const { createSubsystemLogger } = await import("../logging/subsystem.js"); expect(() => createSubsystemLogger("test/acp-startup").isEnabled("info", "console"), ).not.toThrow(); }); it("waits for gateway hello before creating AgentSideConnection", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await Promise.resolve(); expect(mockState.agentSideConnectionCtor).not.toHaveBeenCalled(); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("injects the server-owned SQLite event ledger into the ACP agent", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); expect(mockState.sqliteEventLedgers).toHaveLength(1); expect((mockState.agentOptions[0] as { eventLedger?: unknown }).eventLedger).toBe( mockState.sqliteEventLedgers[0], ); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("advertises approval handling and subscribes to run-scoped tool events", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); expect(mockState.gatewayOptions[0]?.caps).toEqual(["exec-approvals", "tool-events"]); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("passes the resolved TLS fingerprint into the ACP gateway client", async () => { mockState.resolveGatewayClientBootstrap.mockResolvedValue({ url: "wss://127.0.0.1:18789", urlSource: "local loopback", connectionDetails: { url: "wss://127.0.0.1:18789", urlSource: "local loopback", message: "Gateway target: wss://127.0.0.1:18789", }, tlsFingerprint: "sha256:local", auth: { token: undefined, password: undefined, }, }); const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); expect(mockState.gatewayOptions[0]?.tlsFingerprint).toBe("sha256:local"); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it.each([ { name: "default logging", opts: {}, expected: ["openclaw acp: gateway event chat failed\n"], }, { name: "verbose logging", opts: { verbose: true }, expected: [ "openclaw acp: gateway event chat failed\n", "openclaw acp: gateway event chat error: handler boom\n", ], }, ])("contains rejected gateway event handling with $name", async ({ opts, expected }) => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); const writes: string[] = []; const writeSpy = vi .spyOn(process.stderr, "write") .mockImplementation((chunk: string | Uint8Array): boolean => { writes.push(String(chunk)); return true; }); mockState.agentHandleGatewayEvent.mockRejectedValueOnce(new Error("handler boom")); try { const servePromise = serveAcpGateway(opts); await emitHelloAndWaitForAgentSideConnection(); getMockGateway().emitEvent({ event: "chat" }); await vi.waitFor(() => { expect(writes).toEqual(expected); }); await stopServeWithSigint(signalHandlers, servePromise); } finally { writeSpy.mockRestore(); onceSpy.mockRestore(); } }); it("routes logs to stderr before loading gateway config", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await Promise.resolve(); expect(mockState.routeLogsToStderr).toHaveBeenCalledTimes(1); expect(mockState.routeLogsToStderr.mock.invocationCallOrder[0]).toBeLessThan( mockState.resolveGatewayClientBootstrap.mock.invocationCallOrder[0] ?? Number.MAX_SAFE_INTEGER, ); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("rejects startup when gateway connect fails before hello", async () => { const onceSpy = vi .spyOn(process, "once") .mockImplementation( ((_signal: NodeJS.Signals, _handler: () => void) => process) as typeof process.once, ); try { const servePromise = serveAcpGateway({}); await Promise.resolve(); const gateway = getMockGateway(); gateway.emitConnectError("connect failed"); await expect(servePromise).rejects.toThrow("connect failed"); expect(mockState.agentSideConnectionCtor).not.toHaveBeenCalled(); } finally { onceSpy.mockRestore(); } }); it("shuts down when buffered pre-hello ACP input exceeds its limit", async () => { mockState.rawInputChunks.push(new Uint8Array(1024 * 1024 + 1)); const onceSpy = vi .spyOn(process, "once") .mockImplementation( ((_signal: NodeJS.Signals, _handler: () => void) => process) as typeof process.once, ); try { await serveAcpGateway({}); expect(mockState.agentSideConnectionCtor).not.toHaveBeenCalled(); expect(mockState.closeOpenClawStateDatabaseAsync).toHaveBeenCalledOnce(); } finally { onceSpy.mockRestore(); } }); it("passes resolved SecretInput gateway credentials to the ACP gateway client", async () => { mockState.resolveGatewayClientBootstrap.mockResolvedValue({ url: "ws://127.0.0.1:18789", urlSource: "local loopback", connectionDetails: { url: "ws://127.0.0.1:18789", urlSource: "local loopback", message: "Gateway target: ws://127.0.0.1:18789", }, auth: { token: undefined, password: "resolved-secret-password", // pragma: allowlist secret }, }); const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await Promise.resolve(); const bootstrapParams = getGatewayBootstrapParams(); expect(bootstrapParams.env).toBe(process.env); expect(mockState.gatewayAuth[0]).toEqual({ token: undefined, password: "resolved-secret-password", // pragma: allowlist secret }); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("passes CLI URL override context into shared gateway auth resolution", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({ gatewayUrl: "wss://override.example/ws", }); await Promise.resolve(); const bootstrapParams = getGatewayBootstrapParams(); expect(bootstrapParams.env).toBe(process.env); expect(bootstrapParams.gatewayUrl).toBe("wss://override.example/ws"); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("passes the configured Gateway URL into the ACP gateway client", async () => { mockState.resolveGatewayClientBootstrap.mockResolvedValue({ url: "ws://127.0.0.1:19999", urlSource: "cli --url", connectionDetails: { url: "ws://127.0.0.1:19999", urlSource: "cli --url", message: "Gateway target: ws://127.0.0.1:19999", }, auth: { token: undefined, password: undefined, }, }); const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({ gatewayUrl: "ws://127.0.0.1:19999", }); await Promise.resolve(); expect(mockState.gatewayOptions[0]?.url).toBe("ws://127.0.0.1:19999"); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); } finally { onceSpy.mockRestore(); } }); it("does not proxy the standalone ACP control-plane Gateway connection", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await vi.waitFor(() => { expect(mockState.gateways).toHaveLength(1); }); expect(mockState.startProxy).not.toHaveBeenCalled(); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); expect(mockState.stopProxy).not.toHaveBeenCalled(); } finally { onceSpy.mockRestore(); } }); it("closes the shared state database on shutdown", async () => { const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); expect(mockState.closeOpenClawStateDatabaseAsync).not.toHaveBeenCalled(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); expect(mockState.agentShutdown).toHaveBeenCalledOnce(); expect(mockState.closeOpenClawStateDatabaseAsync).toHaveBeenCalledOnce(); } finally { onceSpy.mockRestore(); } }); it("shuts down when the ACP client closes its stdio stream", async () => { const { onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); const closeConnection = mockState.closeAgentSideConnection; if (!closeConnection) { throw new Error("Expected mocked ACP connection close handler"); } closeConnection(); await servePromise; expect(mockState.agentShutdown).toHaveBeenCalledOnce(); expect(mockState.closeOpenClawStateDatabaseAsync).toHaveBeenCalledOnce(); } finally { onceSpy.mockRestore(); } }); it("waits for Gateway transport teardown before closing the shared state database", async () => { let resolveStop!: () => void; const stopPromise = new Promise((resolve) => { resolveStop = resolve; }); mockState.gatewayStopDeferred = { resolve: resolveStop, promise: stopPromise }; const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); signalHandlers.get("SIGTERM")?.(); await vi.waitFor(() => { expect(mockState.agentShutdown).toHaveBeenCalledOnce(); }); expect(mockState.closeOpenClawStateDatabaseAsync).not.toHaveBeenCalled(); resolveStop(); await servePromise; expect(mockState.closeOpenClawStateDatabaseAsync).toHaveBeenCalledOnce(); } finally { onceSpy.mockRestore(); } }); it("closes a real node:sqlite DatabaseSync handle through serveAcpGateway shutdown", async () => { // Use the real state-db module to open and verify a DatabaseSync handle — // this proves the full serveAcpGateway → shutdown → close path, not just // the closeOpenClawStateDatabaseAsync helper in isolation. const actualStateDb = await vi.importActual( "../state/openclaw-state-db.js", ); const realDb = actualStateDb.openOpenClawStateDatabase(); expect(realDb.db.isOpen).toBe(true); expect(actualStateDb.isOpenClawStateDatabaseOpen()).toBe(true); // Wire the test mock so serveAcpGateway's shutdown handler calls the // real closeOpenClawStateDatabaseAsync, which closes the handle we opened above. mockState.closeOpenClawStateDatabaseAsync.mockImplementation(() => actualStateDb.closeOpenClawStateDatabaseAsync(), ); const { signalHandlers, onceSpy } = captureProcessSignalHandlers(); try { const servePromise = serveAcpGateway({}); await emitHelloAndWaitForAgentSideConnection(); await stopServeWithSigint(signalHandlers, servePromise); // After serveAcpGateway shutdown completes, the real DatabaseSync // handle must be closed — proving the ACP shutdown fix works // end-to-end, not just in the helper function. expect(realDb.db.isOpen).toBe(false); expect(actualStateDb.isOpenClawStateDatabaseOpen()).toBe(false); } finally { await actualStateDb.closeOpenClawStateDatabaseAsync(); onceSpy.mockRestore(); } }); it("replays an ACP frame read before Gateway hello to AgentSideConnection", async () => { const initializeRequest = { jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: mockState.acpProtocolVersion, clientCapabilities: {}, }, }; const [message] = await captureAcpMessagesAfterStartup([initializeRequest]); expect(message).toBe(initializeRequest); }); it("coerces MCP date-string initialize protocol versions", async () => { const initializeRequest = { jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: "2025-11-25", clientCapabilities: {}, }, }; await expect(captureAcpMessagesAfterStartup([initializeRequest])).resolves.toEqual([ { ...initializeRequest, params: { ...initializeRequest.params, protocolVersion: mockState.acpProtocolVersion, }, }, ]); }); it("coerces non-integer numeric initialize protocol versions", async () => { const initializeRequest = { jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: 1.5, clientCapabilities: {}, }, }; await expect(captureAcpMessagesAfterStartup([initializeRequest])).resolves.toEqual([ { ...initializeRequest, params: { ...initializeRequest.params, protocolVersion: mockState.acpProtocolVersion, }, }, ]); }); it("passes uint16 numeric initialize protocol versions through unchanged", async () => { const initializeRequest = { jsonrpc: "2.0", id: 1, method: "initialize", params: { protocolVersion: 42, clientCapabilities: {}, }, }; const [message] = await captureAcpMessagesAfterStartup([initializeRequest]); expect(message).toBe(initializeRequest); }); it("passes non-initialize JSON-RPC messages through unchanged", async () => { const sessionRequest = { jsonrpc: "2.0", id: 2, method: "session/new", params: { protocolVersion: "2025-11-25", cwd: "/tmp/openclaw", }, }; const [message] = await captureAcpMessagesAfterStartup([sessionRequest]); expect(message).toBe(sessionRequest); }); });