reverse-engineering-anything/tests/boundary/browser/cdpTargetSessionOwner.test.ts

import { createServer } from "node:http";
import type { Duplex } from "node:stream";

import { WebSocket, WebSocketServer } from "ws";
import { expect, it, onTestFinished, vi } from "vitest";

import { CdpTargetSessionOwner } from "../../../src/browser/CdpTargetSessionOwner.js";
import { CdpConnection } from "../../../src/browser/CdpConnection.js";
import { CdpBrowserProvider } from "../../../src/browser/CdpBrowserProvider.js";
import { inspectWebPageInputSchema } from "../../../src/domain/browserObservation.js";
import { AnalysisError } from "../../../src/domain/analysisErrorBase.js";
import {
  AnalysisCancelledError,
  AnalysisCapabilityUnavailableError,
} from "../../../src/domain/analysisErrorCore.js";
import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js";
import { startFakeCdpBrowser } from "../../fixtures/fakeCdpBrowser.js";
import { awaitAcknowledgement } from "../../fixtures/lifecycleAcknowledgement.js";
import { waitFor } from "../../support/testWait.js";

it.each(["connect", "acquireBrowserSocket"] as const)(
  "refuses %s after clean shutdown without inventing retained cleanup",
  async (method) => {
    const owner = new CdpTargetSessionOwner();
    await owner.close();
    const endpoint = "ws://127.0.0.1:9/devtools/browser/closed";
    const failure = await (
      method === "connect"
        ? owner.connect(endpoint, "inspect_web_page", {}, "not-acquired")
        : owner.acquireBrowserSocket(
            endpoint,
            "inspect_web_page",
            {},
            {
              commandTimeoutMs: 100,
              handshakeTimeoutMs: 100,
            },
          )
    ).catch((cause: unknown) => cause);
    expect(failure).toBeInstanceOf(AnalysisCapabilityUnavailableError);
    expect(failure).toMatchObject({
      providerId: "rea-cdp-browser",
      operation: "inspect_web_page",
      cleanupIncomplete: false,
      cleanupResources: [],
    });
    await owner.close();
  },
);

it("joins a late owned handshake and waits for its active operation", async () => {
  const http = createServer();
  const webSockets = new WebSocketServer({ noServer: true });
  let port = 0;
  let releaseUpgrade: (() => void) | undefined;
  let upgradeReceived: (() => void) | undefined;
  const received = new Promise<void>((resolve) => {
    upgradeReceived = resolve;
  });
  const allow = new Promise<void>((resolve) => {
    releaseUpgrade = resolve;
  });
  http.on("upgrade", (request, socket, head) => {
    upgradeReceived?.();
    void allow.then(() =>
      webSockets.handleUpgrade(request, socket, head, (webSocket) =>
        webSockets.emit("connection", webSocket, request),
      ),
    );
  });
  await new Promise<void>((resolve) => http.listen(0, "127.0.0.1", resolve));
  const address = http.address();
  if (address === null || typeof address === "string")
    throw new Error("Fixture did not bind a TCP port");
  port = address.port;
  const owner = new CdpTargetSessionOwner();
  let connecting: Promise<CdpConnection> | undefined;
  onTestFinished(async () => {
    releaseUpgrade?.();
    if (connecting !== undefined) {
      const pendingConnection = await connecting;
      await owner.finish(pendingConnection);
    }
    await owner.close();
    await new Promise<void>((resolve, reject) =>
      webSockets.close((cause) => (cause ? reject(cause) : resolve())),
    );
    await new Promise<void>((resolve, reject) =>
      http.close((cause) => (cause ? reject(cause) : resolve())),
    );
  });

  connecting = owner.connect(
    `ws://127.0.0.1:${String(port)}/devtools/browser/late`,
    "inspect_web_page",
    {},
    "late-handshake",
  );
  await received;
  const closing = owner.close();
  let closed = false;
  void closing.then(() => {
    closed = true;
  });
  await new Promise<void>((resolve) => setImmediate(resolve));
  expect(closed).toBe(false);
  releaseUpgrade?.();
  const connection = await connecting;
  await new Promise<void>((resolve) => setImmediate(resolve));
  expect(closed).toBe(false);
  await owner.finish(connection);
  await closing;
  await waitFor(() => expect(webSockets.clients.size).toBe(0));
});

it("retries only the failed transport and does not repeat successful child cleanup", async () => {
  const browser = await startFakeCdpBrowser();
  const owner = new CdpTargetSessionOwner();
  let active: CdpConnection | undefined;
  onTestFinished(async () => {
    if (active !== undefined) await owner.finish(active);
    await owner.close();
    await browser.close();
  });
  const failed = await owner.connect(
    browser.browserWebSocketUrl,
    "inspect_web_page",
    {},
    "failed-session",
  );
  active = await owner.connect(
    browser.browserWebSocketUrl,
    "inspect_web_page",
    {},
    "active-peer",
  );
  const close = vi
    .spyOn(WebSocket.prototype, "close")
    .mockImplementationOnce(() => {
      throw new Error("controlled close failure");
    });
  const terminate = vi
    .spyOn(WebSocket.prototype, "terminate")
    .mockImplementation(() => undefined);
  try {
    const failure = await owner.finish(failed).then(
      () => undefined,
      (cause: unknown) => cause,
    );
    expect(failure).toBeInstanceOf(ProviderCleanupError);
    expect(failure).toMatchObject({
      cleanupIncomplete: true,
      cleanupResources: ["failed-session"],
    });
  } finally {
    close.mockRestore();
    terminate.mockRestore();
  }

  await owner.retryPending();
  await waitFor(() => expect(browser.activeConnectionCount()).toBe(1));
  const cleanup = vi.fn(async () => undefined);
  await owner.finish(active, cleanup);
  await owner.retryPending();
  await owner.close();
  expect(cleanup).toHaveBeenCalledTimes(1);
  await waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
  expect(browser.commands.map(({ method }) => method)).not.toContain(
    "Target.closeTarget",
  );
  expect(browser.commands.map(({ method }) => method)).not.toContain(
    "Browser.close",
  );
});

it("accepts asynchronous terminate success only after the close event", async () => {
  const browser = await startFakeCdpBrowser();
  const owner = new CdpTargetSessionOwner();
  onTestFinished(async () => {
    await owner.close();
    await browser.close();
  });
  const connection = await owner.connect(
    browser.browserWebSocketUrl,
    "inspect_web_page",
    {},
    "forced-close",
  );
  const originalTerminate = WebSocket.prototype.terminate;
  const close = vi
    .spyOn(WebSocket.prototype, "close")
    .mockImplementation(() => undefined);
  const terminate = vi
    .spyOn(WebSocket.prototype, "terminate")
    .mockImplementation(function (this: WebSocket) {
      return originalTerminate.call(this);
    });
  try {
    await owner.finish(connection);
  } finally {
    close.mockRestore();
    terminate.mockRestore();
  }
  await owner.close();
  await waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
});

it("clears returned inspection and shutdown cleanup status only after socket release", async () => {
  const browser = await startFakeCdpBrowser();
  const provider = new CdpBrowserProvider();
  const originalClose = CdpConnection.prototype.close;
  let denied = true;
  const closing = vi
    .spyOn(CdpConnection.prototype, "close")
    .mockImplementation(function (this: CdpConnection) {
      return denied
        ? Promise.reject(new Error("controlled transport close failure"))
        : originalClose.call(this);
    });
  onTestFinished(async () => {
    closing.mockRestore();
    await provider.close();
    await browser.close();
  });
  const result = await provider.inspectPage(
    inspectWebPageInputSchema.parse({
      cdp_endpoint: browser.endpoint,
      allowed_origins: [browser.allowedOrigin],
      target_id: "allowed-page",
      observation_ms: 0,
    }),
  );
  if (result.ok) throw new Error("Expected retained socket cleanup failure");
  expect(result.error.cleanupIncomplete).toBe(true);
  expect(result.error.partialObservation).toBeDefined();
  const observation = result.error.partialObservation;
  const retry = await provider.close().catch((cause: unknown) => cause);
  expect(retry).toBeInstanceOf(AnalysisError);
  if (!(retry instanceof AnalysisError))
    throw new Error("Missing retry failure");
  expect(retry.cleanupIncomplete).toBe(true);
  expect(browser.activeConnectionCount()).toBe(1);
  denied = false;
  await provider.close();
  await waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
  expect(result.error.cleanupIncomplete).toBe(false);
  expect(retry.cleanupIncomplete).toBe(false);
  expect(result.error.partialObservation).toBe(observation);
  expect(browser.commands.map(({ method }) => method)).not.toContain(
    "Browser.close",
  );
  expect(browser.commands.map(({ method }) => method)).not.toContain(
    "Target.closeTarget",
  );
});

it("keeps retry aggregates incomplete until every reported socket is released", async () => {
  const browser = await startFakeCdpBrowser();
  const owner = new CdpTargetSessionOwner();
  const first = await owner.connect(
    browser.browserWebSocketUrl,
    "inspect_web_page",
    {},
    "first-socket",
  );
  const second = await owner.connect(
    browser.browserWebSocketUrl,
    "inspect_web_page",
    {},
    "second-socket",
  );
  const active = await owner.connect(
    browser.browserWebSocketUrl,
    "inspect_web_page",
    {},
    "active-peer",
  );
  const originalClose = CdpConnection.prototype.close;
  const denied = new Set([first, second]);
  const closing = vi
    .spyOn(CdpConnection.prototype, "close")
    .mockImplementation(function (this: CdpConnection) {
      return denied.has(this)
        ? Promise.reject(new Error("controlled mixed close failure"))
        : originalClose.call(this);
    });
  onTestFinished(async () => {
    closing.mockRestore();
    await owner.finish(active);
    await owner.close();
    await browser.close();
  });
  const errors = await Promise.all(
    [first, second].map((connection) =>
      owner.finish(connection).catch((cause: unknown) => cause),
    ),
  );
  const aggregate = await owner.retryPending().catch((cause: unknown) => cause);
  if (
    !(aggregate instanceof AnalysisError) ||
    !errors.every((error) => error instanceof AnalysisError)
  )
    throw new Error("Missing typed cleanup errors");
  expect(aggregate.cleanupResources).toEqual(["first-socket", "second-socket"]);
  denied.delete(first);
  const mixed = await owner.retryPending().catch((cause: unknown) => cause);
  if (!(mixed instanceof AnalysisError))
    throw new Error("Missing mixed retry error");
  expect(errors.map((error) => error.cleanupIncomplete)).toEqual([false, true]);
  expect(aggregate.cleanupIncomplete).toBe(true);
  expect(mixed.cleanupResources).toEqual(["second-socket"]);
  await waitFor(() => expect(browser.activeConnectionCount()).toBe(2));
  denied.clear();
  await owner.retryPending();
  expect(errors.map((error) => error.cleanupIncomplete)).toEqual([
    false,
    false,
  ]);
  expect(aggregate.cleanupIncomplete).toBe(false);
  expect(mixed.cleanupIncomplete).toBe(false);
  await waitFor(() => expect(browser.activeConnectionCount()).toBe(1));
  await owner.finish(active);
});

it("surfaces cold-start acquireBrowserSocket cleanup failure to live waiters without unhandled rejections", async () => {
  const http = createServer();
  http.on("upgrade", (_request, socket) => {
    socket.destroy();
  });
  await new Promise<void>((resolve) => http.listen(0, "127.0.0.1", resolve));
  const address = http.address();
  if (address === null || typeof address === "string")
    throw new Error("Fixture did not bind a TCP port");
  const url = `ws://127.0.0.1:${String(address.port)}/devtools/browser/rejected`;
  const owner = new CdpTargetSessionOwner();
  const originalClose = CdpConnection.prototype.close;
  let denyClose = true;
  const closeSpy = vi
    .spyOn(CdpConnection.prototype, "close")
    .mockImplementation(function (this: CdpConnection) {
      return denyClose
        ? Promise.reject(new Error("controlled cold-connect close failure"))
        : originalClose.call(this);
    });
  const unhandled: unknown[] = [];
  const onUnhandled = (reason: unknown): void => {
    unhandled.push(reason);
  };
  process.on("unhandledRejection", onUnhandled);
  let first:
    | ReturnType<CdpTargetSessionOwner["acquireBrowserSocket"]>
    | undefined;
  let second:
    | ReturnType<CdpTargetSessionOwner["acquireBrowserSocket"]>
    | undefined;
  try {
    const limits = { commandTimeoutMs: 1_000, handshakeTimeoutMs: 1_000 };
    first = owner.acquireBrowserSocket(url, "list_browser_targets", {}, limits);
    second = owner.acquireBrowserSocket(url, "inspect_web_page", {}, limits);
    const [firstError, secondError] = await Promise.all([
      first.then(
        () => undefined,
        (cause: unknown) => cause,
      ),
      second.then(
        () => undefined,
        (cause: unknown) => cause,
      ),
    ]);
    await new Promise<void>((resolve) => setImmediate(resolve));
    await new Promise<void>((resolve) => setTimeout(resolve, 20));
    expect(unhandled).toEqual([]);
    if (
      !(firstError instanceof AnalysisError) ||
      !(secondError instanceof AnalysisError)
    )
      throw new Error("Expected typed cleanup-incomplete handshake errors");
    expect(firstError.cleanupIncomplete).toBe(true);
    expect(secondError.cleanupIncomplete).toBe(true);
    expect(firstError.cleanupResources).toEqual(["browser-endpoint"]);
    expect(secondError.cleanupResources).toEqual(["browser-endpoint"]);
    expect(secondError.cleanupPrimary).toMatchObject({
      _tag: "BrowserObservationError",
      operation: "inspect_web_page",
    });

    const blockedRetry = await owner
      .retryPending()
      .catch((cause: unknown) => cause);
    expect(blockedRetry).toBeInstanceOf(AnalysisError);

    denyClose = false;
    await owner.retryPending();
    expect(firstError.cleanupIncomplete).toBe(false);
    expect(secondError.cleanupIncomplete).toBe(false);
  } finally {
    process.off("unhandledRejection", onUnhandled);
    denyClose = false;
    await Promise.allSettled([first, second]);
    closeSpy.mockRestore();
    await owner.close();
    await new Promise<void>((resolve, reject) =>
      http.close((cause) => (cause ? reject(cause) : resolve())),
    );
  }
});

it("surfaces cold-start acquireBrowserSocket cleanup failure when the last waiter aborts without unhandled rejections", async () => {
  const http = createServer();
  const upgradeSockets = new Set<Duplex>();
  let upgradeReceived: (() => void) | undefined;
  const received = new Promise<void>((resolve) => {
    upgradeReceived = resolve;
  });
  http.on("upgrade", (_request, socket) => {
    upgradeSockets.add(socket);
    socket.on("close", () => upgradeSockets.delete(socket));
    socket.on("error", () => undefined);
    upgradeReceived?.();
  });
  await new Promise<void>((resolve) => http.listen(0, "127.0.0.1", resolve));
  const address = http.address();
  if (address === null || typeof address === "string")
    throw new Error("Fixture did not bind a TCP port");
  const url = `ws://127.0.0.1:${String(address.port)}/devtools/browser/stalled`;
  const owner = new CdpTargetSessionOwner();
  const originalClose = CdpConnection.prototype.close;
  let denyClose = true;
  const closeSpy = vi
    .spyOn(CdpConnection.prototype, "close")
    .mockImplementation(function (this: CdpConnection) {
      return denyClose
        ? Promise.reject(new Error("controlled abort close failure"))
        : originalClose.call(this);
    });
  const unhandled: unknown[] = [];
  const onUnhandled = (reason: unknown): void => {
    unhandled.push(reason);
  };
  process.on("unhandledRejection", onUnhandled);
  const firstController = new AbortController();
  const secondController = new AbortController();
  let first:
    | ReturnType<CdpTargetSessionOwner["acquireBrowserSocket"]>
    | undefined;
  let second:
    | ReturnType<CdpTargetSessionOwner["acquireBrowserSocket"]>
    | undefined;
  try {
    const limits = { commandTimeoutMs: 5_000, handshakeTimeoutMs: 5_000 };
    first = owner.acquireBrowserSocket(
      url,
      "list_browser_targets",
      { signal: firstController.signal },
      limits,
    );
    second = owner.acquireBrowserSocket(
      url,
      "inspect_web_page",
      { signal: secondController.signal },
      limits,
    );
    await awaitAcknowledgement(
      received,
      Promise.all([first, second]),
      "cold browser-socket upgrade",
    );

    firstController.abort(new Error("first caller cancelled"));
    const firstError = await first.then(
      () => undefined,
      (cause: unknown) => cause,
    );
    expect(firstError).toBeInstanceOf(AnalysisCancelledError);
    if (!(firstError instanceof AnalysisCancelledError))
      throw new Error("Expected first waiter cancellation");
    expect(firstError.cleanupIncomplete).toBe(false);

    secondController.abort(new Error("last caller cancelled"));
    const secondError = await second.then(
      () => undefined,
      (cause: unknown) => cause,
    );
    await new Promise<void>((resolve) => setImmediate(resolve));
    await new Promise<void>((resolve) => setTimeout(resolve, 20));
    expect(unhandled).toEqual([]);
    if (!(secondError instanceof AnalysisError))
      throw new Error("Expected last waiter cleanup-incomplete failure");
    expect(secondError.cleanupIncomplete).toBe(true);
    expect(secondError.cleanupResources).toEqual(["browser-endpoint"]);
    expect(secondError.cleanupPrimary).toMatchObject({
      _tag: "AnalysisCancelledError",
      operation: "inspect_web_page",
    });

    denyClose = false;
    await owner.retryPending();
    expect(secondError.cleanupIncomplete).toBe(false);
  } finally {
    process.off("unhandledRejection", onUnhandled);
    denyClose = false;
    firstController.abort(new Error("test teardown"));
    secondController.abort(new Error("test teardown"));
    await Promise.allSettled([first, second]);
    closeSpy.mockRestore();
    await owner.close();
    for (const socket of upgradeSockets) socket.destroy();
    await new Promise<void>((resolve, reject) =>
      http.close((cause) => (cause ? reject(cause) : resolve())),
    );
  }
});