src/acp/server.startup.test.ts

/** 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<unknown>;
  readable: ReadableStream<unknown>;
};

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<void>>(async () => {}),
  gatewayStopDeferred: null as {
    resolve: () => void;
    promise: Promise<void>;
  } | null,
  resolveGatewayClientBootstrap: vi.fn<ResolveGatewayClientBootstrap>(),
}));

vi.mock("node:stream", async (importOriginal) => {
  const actual = await importOriginal<typeof import("node:stream")>();
  vi.spyOn(actual.Readable, "toWeb").mockImplementation(
    () =>
      new NodeReadableStream<Uint8Array>({
        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<void> {
    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<void>((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<typeof import("../logging/console.js")>();
  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<void> {
      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<NodeJS.Signals, () => 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<unknown[]> {
    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<unknown[]> {
    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<NodeJS.Signals, () => void>,
    servePromise: Promise<void>,
  ) {
    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<void>((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<typeof import("../state/openclaw-state-db.js")>(
      "../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);
  });
});