import { duplexPair } from "node:stream";
import type { Duplex } from "node:stream";
import http2 from "node:http2";

import { beforeEach, describe, expect, it, vi } from "vitest";

import {
  buildHttp2BridgeForwardUrl,
  classifyStreamAgainstGoaway,
  createBridgeBodyReservation,
  createBridgeRouteBodyLedger,
  createHttp2BridgeServer,
  getBridgeBodyReservedBytesForTest,
  parseCanonicalBridgeRequestPath,
  resetBridgeBodyReservationsForTest,
  wrapDuplexChannelAsNodeDuplex,
  BridgeProcessCapacityError,
  DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES,
  DEFAULT_HTTP2_BRIDGE_PING_INTERVAL_MS,
  DEFAULT_HTTP2_BRIDGE_PING_STALL_MS,
  HTTP2_BRIDGE_ENABLE_PUSH,
  HTTP2_BRIDGE_HEADER_TABLE_SIZE,
  HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS,
  HTTP2_BRIDGE_MAX_DEFLATE_DYNAMIC_TABLE_SIZE,
  HTTP2_BRIDGE_MAX_HEADER_LIST_PAIRS,
  HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE,
  HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES,
  HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES,
  HTTP2_BRIDGE_MAX_SESSION_INVALID_FRAMES,
  HTTP2_BRIDGE_MAX_SESSION_MEMORY,
  HTTP2_BRIDGE_MAX_SESSION_REJECTED_STREAMS,
  HTTP2_BRIDGE_SERVER_OPTIONS,
  HTTP2_BRIDGE_STREAM_RESET_BURST,
  HTTP2_BRIDGE_STREAM_RESET_RATE,
  type Http2BridgeForwardRequest,
  type Http2BridgeForwardResult,
  type Http2BridgeGoawayRecord,
} from "./http2-bridge-server.js";
import {
  createSandboxHttp2BridgeGateway,
  DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES,
  HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST,
  type SandboxCallbackBridgeRouteRule,
} from "./sandbox-callback-bridge.js";
import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";

/**
 * Unit harness for the host HTTP/2 server and the sandbox HTTP/2 client
 * gateway. Every test connects the pair over one paired in-memory `Duplex`
 * (`node:stream`'s `duplexPair`) — no real TCP socket and no spawned sandbox
 * process. This proves the two halves speak one wire-compatible HTTP/2
 * session with no network in between.
 */

const BRIDGE_TOKEN = "test-bridge-token-fixed-length-32";

/** Wrap one side of a paired `Duplex` as a minimal fake `CommandManagedDuplexChannel`. */
function fakeChannelFromDuplex(duplex: Duplex): CommandManagedDuplexChannel {
  const dataListeners: Array<(chunk: Uint8Array) => void> = [];
  const exitListeners: Array<(exit: { exitCode: number | null; transportClosed?: boolean }) => void> = [];
  duplex.on("data", (chunk: Buffer) => {
    for (const listener of dataListeners) listener(chunk);
  });
  duplex.on("end", () => {
    for (const listener of exitListeners) listener({ exitCode: null, transportClosed: true });
  });
  return {
    write: (data) => {
      duplex.write(Buffer.from(data));
    },
    onData: (listener) => {
      dataListeners.push(listener);
    },
    onExit: (listener) => {
      exitListeners.push(listener);
    },
    stop: () => {
      duplex.destroy();
    },
    close: async () => {
      duplex.end();
    },
  };
}

interface TestPairOptions {
  forwardRequest?: (request: Http2BridgeForwardRequest) => Promise<Http2BridgeForwardResult>;
  bridgeToken?: string;
  pingIntervalMs?: number;
  pingStallMs?: number;
  requestBodyTimeoutMs?: number;
  requestBodyLifetimeCeilingMs?: number;
  closeGraceMs?: number;
  capacityDenialSettleDeadlineMs?: number;
  responseWriteSettleDeadlineMs?: number;
  maxBodyBytes?: number;
  routes?: readonly SandboxCallbackBridgeRouteRule[];
  onGoaway?: (record: Http2BridgeGoawayRecord) => void;
  onSessionError?: (error: Error) => void;
  onSession?: (session: http2.ServerHttp2Session) => void;
}

/** Bind the host server to one side of a fresh paired in-memory `Duplex`.
 * Returns the other side, unbound, for a caller to connect a client to. */
function bindTestServer(options: TestPairOptions = {}) {
  const bridgeToken = options.bridgeToken ?? BRIDGE_TOKEN;
  const [serverSide, clientSide] = duplexPair();
  const forwardRequest =
    options.forwardRequest ??
    (async (request: Http2BridgeForwardRequest) => ({
      status: 200,
      headers: { "content-type": "application/json" },
      body: Buffer.from(JSON.stringify({ echoedMethod: request.method, echoedPath: request.pathname }), "utf8"),
    }));
  const handle = createHttp2BridgeServer({
    bridgeToken,
    forwardRequest,
    pingIntervalMs: options.pingIntervalMs,
    pingStallMs: options.pingStallMs,
    requestBodyTimeoutMs: options.requestBodyTimeoutMs,
    requestBodyLifetimeCeilingMs: options.requestBodyLifetimeCeilingMs,
    closeGraceMs: options.closeGraceMs,
    capacityDenialSettleDeadlineMs: options.capacityDenialSettleDeadlineMs,
    responseWriteSettleDeadlineMs: options.responseWriteSettleDeadlineMs,
    maxBodyBytes: options.maxBodyBytes,
    routes: options.routes,
    onGoaway: options.onGoaway,
    onSessionError: options.onSessionError,
    onSession: options.onSession,
  });
  const channel = fakeChannelFromDuplex(serverSide);
  handle.bindChannel(channel);
  return { handle, bridgeToken, clientSide, serverSide };
}

/** Bind the server, then connect the sandbox HTTP/2 client gateway to the
 * other side of the same paired in-memory `Duplex`. */
function createTestPair(options: TestPairOptions = {}) {
  const { handle, bridgeToken, clientSide, serverSide } = bindTestServer(options);
  const gateway = createSandboxHttp2BridgeGateway({
    bridgeToken,
    createConnection: () => clientSide,
  });
  return { handle, gateway, bridgeToken, clientSide, serverSide };
}

/** Open a raw HTTP/2 client session directly against one side of the pair,
 * bypassing the sandbox gateway. Some tests need direct stream control (an
 * explicit RST_STREAM, an explicit GOAWAY) the gateway's `forwardRequest`
 * abstraction does not expose. `settings`, when given, rides the client's
 * own initial SETTINGS frame — for example `{ initialWindowSize: 0 }` to
 * deny the server any flow-control credit to send response bytes with, a
 * deterministic stall independent of the fake transport's own buffering. */
function connectRawClient(
  clientSide: Duplex,
  settings?: http2.Settings,
): http2.ClientHttp2Session {
  return http2.connect("http://bridge.internal", { createConnection: () => clientSide, settings });
}

/** Track whether `forwardRequest` ran, so a test can prove a denied or
 * destroyed stream never reached it. Call `markCalled()` from inside
 * `forwardRequest`, then assert on `.called`. */
function createForwarderCallTracker(): { called: boolean; markCalled: () => void } {
  const tracker = {
    called: false,
    markCalled(): void {
      tracker.called = true;
    },
  };
  return tracker;
}

/** Send one more request over `rawClient` and wait for it to complete, so a
 * test can prove the session survived a prior faulted, stalled, or denied
 * stream. */
async function expectSessionStillServesARequest(
  rawClient: http2.ClientHttp2Session,
  request: { method: string; path: string; token: string },
): Promise<{ status: number; body: string }> {
  return new Promise((resolve, reject) => {
    const req = rawClient.request({
      ":method": request.method,
      ":path": request.path,
      authorization: `Bearer ${request.token}`,
    });
    let status = 0;
    let body = "";
    req.setEncoding("utf8");
    req.on("response", (headers) => {
      status = Number(headers[":status"]) || 0;
    });
    req.on("data", (chunk) => (body += chunk));
    req.on("end", () => resolve({ status, body }));
    req.on("error", reject);
    req.end();
  });
}

describe("createHttp2BridgeServer + createSandboxHttp2BridgeGateway", () => {
  // Every test in this file shares the module-scope process reservation
  // total. A reservation one test leaves unreleased would otherwise lower
  // the ceiling every later test sees, so each test starts from zero.
  beforeEach(() => {
    resetBridgeBodyReservationsForTest();
  });

  it("test_one_session_over_a_fake_channel_forwards_a_request", async () => {
    const { handle, gateway } = createTestPair();
    try {
      const response = await gateway.forwardRequest({
        method: "GET",
        path: "/api/agents/me",
        query: "",
        headers: {},
        body: Buffer.alloc(0),
        receivedToken: BRIDGE_TOKEN,
      });
      expect(response.status).toBe(200);
      expect(JSON.parse(response.body.toString("utf8"))).toEqual({
        echoedMethod: "GET",
        echoedPath: "/api/agents/me",
      });
    } finally {
      await gateway.close();
      await handle.close();
    }
  });

  it("test_sixty_four_concurrent_streams_all_complete", async () => {
    const seenPaths = new Set<string>();
    const { handle, gateway } = createTestPair({
      forwardRequest: async (request) => {
        // Force real overlap: every forward call waits one macrotask before it
        // resolves, so 64 concurrent streams are genuinely in flight together.
        await new Promise((resolve) => setTimeout(resolve, 5));
        seenPaths.add(request.pathname);
        return {
          status: 200,
          headers: {},
          body: Buffer.from(JSON.stringify({ echoedPath: request.pathname }), "utf8"),
        };
      },
    });
    try {
      const requests = Array.from({ length: 64 }, (_, index) =>
        gateway.forwardRequest({
          method: "GET",
          path: `/api/issues/${index}`,
          query: "",
          headers: {},
          body: Buffer.alloc(0),
          receivedToken: BRIDGE_TOKEN,
        }),
      );
      const responses = await Promise.all(requests);
      expect(responses).toHaveLength(64);
      for (const [index, response] of responses.entries()) {
        expect(response.status).toBe(200);
        expect(JSON.parse(response.body.toString("utf8"))).toEqual({
          echoedPath: `/api/issues/${index}`,
        });
      }
      expect(seenPaths.size).toBe(64);
    } finally {
      await gateway.close();
      await handle.close();
    }
  });

  it("test_ping_detects_a_silent_stall_within_twenty_seconds", async () => {
    // The production defaults name the twenty-second bound this test name
    // promises. The mechanism test below exercises the same code path with
    // small overrides, so the suite stays fast.
    expect(DEFAULT_HTTP2_BRIDGE_PING_STALL_MS).toBe(20_000);
    expect(DEFAULT_HTTP2_BRIDGE_PING_INTERVAL_MS).toBeLessThan(DEFAULT_HTTP2_BRIDGE_PING_STALL_MS);

    const [serverSide] = duplexPair();
    // Nothing consumes the other side of the pair, so every PING frame the
    // server sends goes unacknowledged: a silent stall.
    const stallErrors: Error[] = [];
    const handle = createHttp2BridgeServer({
      bridgeToken: BRIDGE_TOKEN,
      forwardRequest: async () => ({ status: 200 }),
      pingIntervalMs: 15,
      pingStallMs: 40,
      onSessionError: (error) => stallErrors.push(error),
    });
    handle.bindChannel(fakeChannelFromDuplex(serverSide));

    await new Promise((resolve) => setTimeout(resolve, 300));
    expect(stallErrors).toHaveLength(1);
    expect(stallErrors[0]?.message).toMatch(/stall/i);
    await handle.close();
  });

  it("test_goaway_reports_the_last_processed_stream_identifier", async () => {
    // GOAWAY's Last-Stream-ID names the highest stream the SENDER processed
    // for the peer's own initiated streams. Every request stream in this
    // transport originates from the sandbox, so the meaningful, valid
    // direction is the host naming the last client stream it processed — not
    // the reverse (a client-sent GOAWAY only ever carries 0 here, since
    // server push is disabled). The sandbox gateway observes it and can
    // classify its own dispatched stream IDs with `classifyStreamAgainstGoaway`.
    let hostSession: http2.ServerHttp2Session | undefined;
    const goawayRecords: Array<{ lastStreamId: number; errorCode: number }> = [];
    const { handle, bridgeToken, clientSide } = bindTestServer({
      onSession: (session) => {
        hostSession = session;
      },
    });
    const gateway = createSandboxHttp2BridgeGateway({
      bridgeToken,
      createConnection: () => clientSide,
      onGoaway: (record) => goawayRecords.push(record),
    });
    try {
      const response = await gateway.forwardRequest({
        method: "GET",
        path: "/api/agents/me",
        query: "",
        headers: {},
        body: Buffer.alloc(0),
        receivedToken: bridgeToken,
      });
      expect(response.status).toBe(200);

      await new Promise<void>((resolve) => {
        hostSession?.goaway(http2.constants.NGHTTP2_NO_ERROR, 1);
        setTimeout(resolve, 50);
      });
      expect(goawayRecords).toHaveLength(1);
      expect(goawayRecords[0]?.lastStreamId).toBe(1);
      expect(classifyStreamAgainstGoaway(1, goawayRecords[0]!.lastStreamId)).toBe("accepted");
      expect(classifyStreamAgainstGoaway(3, goawayRecords[0]!.lastStreamId)).toBe("not_accepted");
    } finally {
      await gateway.close();
      await handle.close();
    }
  });

  it("test_rst_stream_fails_one_request_and_keeps_the_session", async () => {
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        if (request.pathname === "/api/agents/me") {
          // Hold this one open long enough for the test to RST it.
          await new Promise((resolve) => setTimeout(resolve, 200));
        }
        return { status: 200, headers: {}, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const abortedStream = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/me",
        authorization: `Bearer ${bridgeToken}`,
      });
      const abortedOutcome = new Promise<"aborted" | "closed">((resolve) => {
        abortedStream.on("error", () => resolve("aborted"));
        abortedStream.on("close", () => resolve("closed"));
      });
      abortedStream.end();
      // Give the request time to reach the server before the reset.
      await new Promise((resolve) => setTimeout(resolve, 20));
      abortedStream.close(http2.constants.NGHTTP2_CANCEL);
      await abortedOutcome;

      // The session survives: a second request completes normally.
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/companies/co1",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
      expect(JSON.parse(survivingResponse.body)).toEqual({ path: "/api/companies/co1" });
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_the_server_aborts_the_forward_when_the_client_closes_the_stream", async () => {
    let capturedRequest: Http2BridgeForwardRequest | undefined;
    let releaseForward: (() => void) | undefined;
    const forwardHeld = new Promise<void>((resolve) => {
      releaseForward = resolve;
    });
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        capturedRequest = request;
        // Hold the handler open, so the test controls exactly when the
        // client-side stream close happens relative to the forward.
        await forwardHeld;
        return { status: 200, body: Buffer.from("{}", "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const stream = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/me",
        authorization: `Bearer ${bridgeToken}`,
      });
      const streamClosed = new Promise<void>((resolve) => {
        stream.on("error", () => resolve());
        stream.on("close", () => resolve());
      });
      stream.end();
      // Give the request time to reach the server and enter forwardRequest.
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(capturedRequest).toBeDefined();
      expect(capturedRequest!.signal.aborted).toBe(false);
      stream.close(http2.constants.NGHTTP2_CANCEL);
      await streamClosed;
      // Give the server's own stream `close` listener a turn to run.
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(capturedRequest!.signal.aborted).toBe(true);
      releaseForward!();
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_one_aborted_stream_leaves_the_session_and_the_other_streams_open", async () => {
    const signalsByPath = new Map<string, AbortSignal>();
    let releaseSurvivor: (() => void) | undefined;
    const survivorHeld = new Promise<void>((resolve) => {
      releaseSurvivor = resolve;
    });
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        signalsByPath.set(request.pathname, request.signal);
        if (request.pathname === "/api/agents/aborted") {
          // The RST — not a manual release — is what must end this forward:
          // it settles only when the stream's own signal fires, so the test
          // proves the signal itself unblocks the handler.
          await new Promise<void>((_resolve, reject) => {
            request.signal.addEventListener("abort", () => reject(new Error("aborted")), {
              once: true,
            });
          }).catch(() => undefined);
        } else if (request.pathname === "/api/agents/survivor") {
          await survivorHeld;
        }
        return { status: 200, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const abortedStream = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/aborted",
        authorization: `Bearer ${bridgeToken}`,
      });
      const survivorStream = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/survivor",
        authorization: `Bearer ${bridgeToken}`,
      });
      const abortedStreamClosed = new Promise<void>((resolve) => {
        abortedStream.on("error", () => resolve());
        abortedStream.on("close", () => resolve());
      });
      // Capture the survivor stream's own response, so the test proves this
      // exact stream — the sibling of the aborted one, on the same session —
      // completes normally, not merely that the session admits a fresh one.
      const survivorResponse = new Promise<{ status: number; body: string }>((resolve, reject) => {
        let status = 0;
        let body = "";
        survivorStream.setEncoding("utf8");
        survivorStream.on("response", (h) => {
          status = Number(h[":status"]) || 0;
        });
        survivorStream.on("data", (chunk) => (body += chunk));
        survivorStream.on("end", () => resolve({ status, body }));
        survivorStream.on("error", reject);
      });
      abortedStream.end();
      survivorStream.end();
      // Give both requests time to reach the server and enter forwardRequest.
      await new Promise((resolve) => setTimeout(resolve, 20));
      abortedStream.close(http2.constants.NGHTTP2_CANCEL);
      await abortedStreamClosed;
      // Give the server's own stream `close` listener a turn to run.
      await new Promise((resolve) => setTimeout(resolve, 20));
      // The one aborted stream's signal fires. The other, still-open stream's
      // signal stays untouched: the abort is local to its own stream.
      expect(signalsByPath.get("/api/agents/aborted")?.aborted).toBe(true);
      expect(signalsByPath.get("/api/agents/survivor")?.aborted).toBe(false);
      releaseSurvivor!();

      const response = await survivorResponse;
      expect(response.status).toBe(200);
      expect(JSON.parse(response.body)).toEqual({ path: "/api/agents/survivor" });
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_live_forward_work_never_passes_the_stream_limit", async () => {
    // A forward that is never aborted holds open until the caller aborts it
    // — this stands in for a forward stuck at its own long timeout. Only the
    // per-stream abort binding this file adds can free such a forward before
    // that timeout, so this count proves the binding, not merely the HTTP/2
    // session's own stream-slot accounting (a stream's protocol slot frees on
    // RST regardless of whether its forward call ever settles).
    let liveForwards = 0;
    let maxLiveForwards = 0;
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        liveForwards += 1;
        maxLiveForwards = Math.max(maxLiveForwards, liveForwards);
        try {
          await new Promise<void>((resolve, reject) => {
            if (request.signal.aborted) {
              reject(new Error("aborted"));
              return;
            }
            request.signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true });
          });
          return { status: 200, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") };
        } finally {
          liveForwards -= 1;
        }
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const openAndCancelOneBatch = async (label: string) => {
        const streams = Array.from({ length: HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS }, (_, index) =>
          rawClient.request({
            ":method": "GET",
            ":path": `/api/issues/${label}-${index}`,
            authorization: `Bearer ${bridgeToken}`,
          }),
        );
        for (const stream of streams) stream.end();
        await new Promise((resolve) => setTimeout(resolve, 30));
        for (const stream of streams) stream.close(http2.constants.NGHTTP2_CANCEL);
        await new Promise((resolve) => setTimeout(resolve, 30));
      };

      // One full batch of maximum-concurrency streams, closed while their
      // forwards are still held open, then a second full batch opened right
      // after. Without the abort binding, the first batch's forwards would
      // still be alive when the second batch dispatches, doubling the live-
      // forward count past the stream limit.
      await openAndCancelOneBatch("first");
      await openAndCancelOneBatch("second");

      expect(maxLiveForwards).toBeLessThanOrEqual(HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS);
      expect(liveForwards).toBe(0);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_stalled_request_body_settles_instead_of_hanging_forever", async () => {
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 30,
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const startMs = Date.now();
      // A partial, never-ended body stalls the read. The timeout destroys the
      // stream, so the request settles (through an `error` or a `close`, not
      // a clean response) well inside the bound, instead of hanging forever.
      await new Promise<void>((resolve) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        req.on("error", () => resolve());
        req.on("close", () => resolve());
        req.write("partial-body");
      });
      expect(Date.now() - startMs).toBeLessThan(5_000);
      expect(forwarderTracker.called).toBe(false);

      // The session survives the timed-out stream: a second, complete request
      // still succeeds.
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "POST",
        path: "/api/issues/abc/comments",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_slow_but_progressing_request_body_completes_instead_of_timing_out", async () => {
    // The idle bound is well under the total time the whole body takes, so a
    // one-shot bound over the full read would trip. Each chunk resets the
    // bound, so the request still completes.
    const { handle, bridgeToken, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 80,
      forwardRequest: async (request) => ({
        status: 200,
        body: Buffer.from(JSON.stringify({ bodyLength: request.body.byteLength }), "utf8"),
      }),
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        let status = 0;
        let body = "";
        req.setEncoding("utf8");
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        // Five chunks, each inside the idle bound, summing past it.
        let sent = 0;
        const sendNext = () => {
          if (sent >= 5) {
            req.end();
            return;
          }
          sent += 1;
          req.write("chunk");
          setTimeout(sendNext, 40);
        };
        sendNext();
      });
      expect(response.status).toBe(200);
      expect(JSON.parse(response.body)).toEqual({ bodyLength: "chunk".length * 5 });
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_slow_trickle_that_never_ends_stops_at_the_one_shot_ceiling", async () => {
    // The idle bound is generous, so a chunk every 40ms never lets it
    // expire. The ceiling (150ms) arms once, when the read starts, and
    // never renews on a chunk: this proves a peer that keeps every chunk
    // gap under the idle bound, but never finishes the body, still loses
    // the stream once the read's total age passes the ceiling. This is the
    // failure mode an authenticated sandbox could otherwise use to hold one
    // of the concurrent-stream slots open indefinitely: keep sending just
    // enough to dodge the idle bound, and never finish the body.
    //
    // Every write below lands well before the ceiling, and the test makes
    // no further call on the stream after the last one: nothing races the
    // server's own reset of it once the ceiling passes, so this proves the
    // bound from the server's own, deterministic effect (the forward
    // handler never runs) instead of from a raw client-stream event whose
    // timing Node does not guarantee here.
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 5_000,
      requestBodyLifetimeCeilingMs: 150,
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const req = rawClient.request({
        ":method": "POST",
        ":path": "/api/issues/abc/comments",
        authorization: `Bearer ${bridgeToken}`,
      });
      req.on("error", () => undefined);
      // Four chunks, 40ms apart, well inside the idle bound. The total send
      // time, about 160ms, passes the 150ms ceiling. The body never ends:
      // this request never calls `.end()`.
      for (let chunkIndex = 0; chunkIndex < 4; chunkIndex += 1) {
        req.write("chunk");
        await new Promise((resolve) => setTimeout(resolve, 40));
      }
      // Wait past the ceiling with no further write, so the server has time
      // to destroy the stalled stream on its own.
      await new Promise((resolve) => setTimeout(resolve, 150));
      expect(forwarderTracker.called).toBe(false);

      // The session survives the stopped stream: a second, complete request
      // still succeeds.
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "POST",
        path: "/api/issues/abc/comments",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_request_body_within_the_lifetime_bound_still_completes", async () => {
    // A generous ceiling must not interfere with an ordinary request that
    // finishes well inside it.
    const { handle, bridgeToken, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 5_000,
      requestBodyLifetimeCeilingMs: 5_000,
      forwardRequest: async (request) => ({
        status: 200,
        body: Buffer.from(JSON.stringify({ bodyLength: request.body.byteLength }), "utf8"),
      }),
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        let status = 0;
        let body = "";
        req.setEncoding("utf8");
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        req.write("chunk");
        req.end();
      });
      expect(response.status).toBe(200);
      expect(JSON.parse(response.body)).toEqual({ bodyLength: "chunk".length });
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_slow_but_progressing_upload_completes_within_the_ceiling", async () => {
    // The idle bound and the ceiling are both generous, so a slow but
    // genuinely progressing upload completes normally, instead of losing
    // its stream mid-upload.
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 5_000,
      requestBodyLifetimeCeilingMs: 5_000,
      forwardRequest: async (request) => {
        forwarderTracker.markCalled();
        return { status: 200, body: Buffer.from(JSON.stringify({ bodyLength: request.body.byteLength }), "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        let status = 0;
        let body = "";
        req.setEncoding("utf8");
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        void (async () => {
          // Six chunks, 40ms apart (well inside the idle bound), each real
          // progress.
          for (let sent = 0; sent < 6; sent += 1) {
            req.write("chunk");
            await new Promise((r) => setTimeout(r, 40));
          }
          req.end();
        })();
      });
      expect(forwarderTracker.called).toBe(true);
      expect(response.status).toBe(200);
      expect(JSON.parse(response.body)).toEqual({ bodyLength: "chunk".length * 6 });
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_close_force_destroys_a_session_with_a_stalled_stream_instead_of_waiting_forever", async () => {
    const { handle, bridgeToken, clientSide } = bindTestServer({
      // A body timeout far longer than the close grace, so `close()` is the
      // only thing that bounds the wait in this test.
      requestBodyTimeoutMs: 60_000,
      closeGraceMs: 30,
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const req = rawClient.request({
        ":method": "POST",
        ":path": "/api/issues/abc/comments",
        authorization: `Bearer ${bridgeToken}`,
      });
      req.write("partial-body");
      // Give the request time to reach the server before close() runs.
      await new Promise((resolve) => setTimeout(resolve, 20));

      const closeStart = Date.now();
      await handle.close();
      const closeElapsedMs = Date.now() - closeStart;
      // `close()` force-destroyed the stalled session at the grace bound,
      // instead of waiting on `session.close()` forever.
      expect(closeElapsedMs).toBeLessThan(5_000);
    } finally {
      rawClient.close();
    }
  });

  it("test_the_server_sets_every_bounded_option", async () => {
    expect(HTTP2_BRIDGE_SERVER_OPTIONS).toEqual({
      settings: {
        enablePush: HTTP2_BRIDGE_ENABLE_PUSH,
        maxConcurrentStreams: HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS,
        maxHeaderListSize: HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE,
        headerTableSize: HTTP2_BRIDGE_HEADER_TABLE_SIZE,
      },
      maxSessionMemory: HTTP2_BRIDGE_MAX_SESSION_MEMORY,
      maxHeaderListPairs: HTTP2_BRIDGE_MAX_HEADER_LIST_PAIRS,
      maxDeflateDynamicTableSize: HTTP2_BRIDGE_MAX_DEFLATE_DYNAMIC_TABLE_SIZE,
      maxSessionInvalidFrames: HTTP2_BRIDGE_MAX_SESSION_INVALID_FRAMES,
      maxSessionRejectedStreams: HTTP2_BRIDGE_MAX_SESSION_REJECTED_STREAMS,
      streamResetRate: HTTP2_BRIDGE_STREAM_RESET_RATE,
      streamResetBurst: HTTP2_BRIDGE_STREAM_RESET_BURST,
    });
    expect(HTTP2_BRIDGE_ENABLE_PUSH).toBe(false);
    expect(HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS).toBe(4);
    expect(HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE).toBe(16384);
    expect(HTTP2_BRIDGE_HEADER_TABLE_SIZE).toBe(4096);
    expect(HTTP2_BRIDGE_MAX_SESSION_MEMORY).toBe(16);
    expect(HTTP2_BRIDGE_MAX_HEADER_LIST_PAIRS).toBe(128);
    expect(HTTP2_BRIDGE_MAX_DEFLATE_DYNAMIC_TABLE_SIZE).toBe(4096);
    expect(HTTP2_BRIDGE_MAX_SESSION_INVALID_FRAMES).toBe(100);
    expect(HTTP2_BRIDGE_MAX_SESSION_REJECTED_STREAMS).toBe(100);
    expect(HTTP2_BRIDGE_STREAM_RESET_RATE).toBe(10);
    expect(HTTP2_BRIDGE_STREAM_RESET_BURST).toBe(100);

    // The running server actually carries the bound, not only the constant:
    // the four `settings` values ride the server's outbound SETTINGS frame,
    // so the connected client's `remoteSettings` reflects them after the
    // handshake completes.
    const { handle, clientSide } = bindTestServer();
    const rawClient = connectRawClient(clientSide);
    try {
      const remoteSettings = await new Promise<http2.Settings>((resolve) => {
        rawClient.once("remoteSettings", resolve);
      });
      expect(remoteSettings.enablePush).toBe(HTTP2_BRIDGE_ENABLE_PUSH);
      expect(remoteSettings.maxConcurrentStreams).toBe(HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS);
      expect(remoteSettings.maxHeaderListSize).toBe(HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE);
      expect(remoteSettings.headerTableSize).toBe(HTTP2_BRIDGE_HEADER_TABLE_SIZE);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_the_host_body_budget_matches_the_stream_limit", () => {
    // The multiplier counts every retained `Buffer` copy of one live
    // forward's request and response body: four exact `Buffer` rows. The
    // forward path decodes no body to a string, so no row applies the
    // two-bytes-per-UTF-16-code-unit string overhead any more.
    // `test_live_forward_work_never_passes_the_stream_limit` proves the
    // count of live forwards never passes `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS`,
    // so this multiplier bounds live forwards, not merely open streams.
    // `HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES` uses this exact same formula to
    // enforce it as a real per-route cap, not merely a derived figure.
    const expectedRouteBudget =
      HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS * 4 * DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES;
    expect(expectedRouteBudget).toBe(168_820_736);
    expect(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES).toBe(expectedRouteBudget);
  });

  describe("createBridgeBodyReservation", () => {
    it("test_a_reservation_denies_a_chunk_that_passes_the_process_ceiling", () => {
      const owner = createBridgeBodyReservation();
      try {
        expect(owner.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES)).toBe(true);
        expect(owner.heldBytes).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES);
        expect(getBridgeBodyReservedBytesForTest()).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES);

        // One more byte passes the ceiling: denied, and it holds no bytes.
        expect(owner.reserve(1)).toBe(false);
        expect(owner.heldBytes).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES);
        expect(getBridgeBodyReservedBytesForTest()).toBe(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES);
      } finally {
        owner.release();
      }
    });

    it("test_a_reservation_denies_the_concatenated_copy_that_passes_the_process_ceiling", () => {
      const owner = createBridgeBodyReservation();
      try {
        // The chunk-array reservation alone passes comfortably.
        const chunkArrayBytes = Math.floor(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES * 0.6);
        expect(owner.reserve(chunkArrayBytes)).toBe(true);

        // The concatenated `Buffer.concat` copy is a second, separate live
        // copy of the same bytes: reserving it on top of the chunk array
        // this owner already holds passes the ceiling, even though the
        // chunk-array reservation alone did not.
        expect(owner.reserve(chunkArrayBytes)).toBe(false);
        expect(owner.heldBytes).toBe(chunkArrayBytes);
        expect(getBridgeBodyReservedBytesForTest()).toBe(chunkArrayBytes);
      } finally {
        owner.release();
      }
    });

    it("test_a_reservation_releases_every_held_byte_one_time_only", () => {
      const owner = createBridgeBodyReservation();
      expect(owner.reserve(1_000)).toBe(true);
      expect(getBridgeBodyReservedBytesForTest()).toBe(1_000);

      owner.release();
      expect(owner.heldBytes).toBe(0);
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);

      // A second release must not double-subtract: the total must not go
      // below zero.
      owner.release();
      expect(owner.heldBytes).toBe(0);
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);
    });
  });

  describe("route isolation", () => {
    it("test_a_route_ledger_denies_a_reservation_that_passes_its_own_ceiling_with_the_process_ceiling_still_open", () => {
      // The process-wide ceiling has ample room (1 GiB); only this one
      // route's own share is tight. A route-scoped owner must still deny
      // the second reservation, proving the route ceiling is a real,
      // independent check, not merely a reflection of the process total.
      const routeLedger = createBridgeRouteBodyLedger();
      const owner = createBridgeBodyReservation(routeLedger);
      try {
        expect(owner.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)).toBe(true);
        expect(routeLedger.reservedBytes).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES);

        expect(owner.reserve(1)).toBe(false);
        expect(owner.heldBytes).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES);
        expect(routeLedger.reservedBytes).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES);
        // The process-wide total only ever grew by what this owner actually
        // holds: the denied byte reserved against neither total.
        expect(getBridgeBodyReservedBytesForTest()).toBe(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES);
      } finally {
        owner.release();
      }
    });

    it("test_one_route_at_its_own_ceiling_never_blocks_a_sibling_routes_reservation", () => {
      // Two routes, two ledgers. Route A spends its own entire ceiling.
      // Route B's reservation, against its own separate ledger, must still
      // succeed: the process-wide total (1 GiB) has room for both routes'
      // ceilings many times over, so only route isolation — not the shared
      // total — could explain a denial here.
      const routeLedgerA = createBridgeRouteBodyLedger();
      const routeLedgerB = createBridgeRouteBodyLedger();
      const ownerA = createBridgeBodyReservation(routeLedgerA);
      const ownerB = createBridgeBodyReservation(routeLedgerB);
      try {
        expect(ownerA.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)).toBe(true);
        expect(ownerA.reserve(1)).toBe(false);

        expect(ownerB.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)).toBe(true);
      } finally {
        ownerA.release();
        ownerB.release();
      }
    });

    it("test_two_bridge_server_routes_isolate_their_own_reservations_end_to_end", async () => {
      // The end-to-end proof: while route A's own forward holds its own
      // entire per-route ceiling in flight, a second, independent route's
      // request must still succeed. Each `bindTestServer` call is its own
      // route (its own `createHttp2BridgeServer` call, so its own route
      // ledger). Route A's forward parks on `releaseRouteAHold` after it
      // reserves, so its reservation stays live — not merely reserved and
      // immediately released — for the whole time route B's request runs.
      let markRouteAReserved: (() => void) | undefined;
      const routeAReserved = new Promise<void>((resolve) => {
        markRouteAReserved = resolve;
      });
      let releaseRouteAHold: (() => void) | undefined;
      const routeAHold = new Promise<void>((resolve) => {
        releaseRouteAHold = resolve;
      });
      const routeA = bindTestServer({
        forwardRequest: async (request) => {
          // Reserve this route's own entire ceiling against its own ledger,
          // simulating route A at its own documented peak.
          if (!request.reservation.reserve(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES)) {
            throw new BridgeProcessCapacityError();
          }
          markRouteAReserved!();
          await routeAHold;
          return { status: 200 };
        },
      });
      const routeB = bindTestServer({
        forwardRequest: async () => ({ status: 200 }),
      });
      const rawClientA = connectRawClient(routeA.clientSide);
      const rawClientB = connectRawClient(routeB.clientSide);
      try {
        const pendingResponseA = expectSessionStillServesARequest(rawClientA, {
          method: "GET",
          path: "/api/agents/me",
          token: routeA.bridgeToken,
        });

        // Wait until route A's forward actually holds its own ceiling —
        // not merely until the request was sent — before checking route B.
        await routeAReserved;
        expect(getBridgeBodyReservedBytesForTest()).toBeGreaterThanOrEqual(HTTP2_BRIDGE_MAX_ROUTE_BODY_BYTES);

        // Route A now holds its own entire per-route ceiling, live. The
        // process-wide total still has headroom (1 GiB minus one 161 MiB
        // route), so route B's independent request succeeds only if its own
        // ledger is genuinely separate from route A's.
        const responseB = await expectSessionStillServesARequest(rawClientB, {
          method: "GET",
          path: "/api/agents/me",
          token: routeB.bridgeToken,
        });
        expect(responseB.status).toBe(200);

        releaseRouteAHold!();
        const responseA = await pendingResponseA;
        expect(responseA.status).toBe(200);
      } finally {
        rawClientA.close();
        rawClientB.close();
        await routeA.handle.close();
        await routeB.handle.close();
      }
    });
  });

  it("test_the_stream_handler_releases_its_reservation_on_completion_error_abort_timeout_and_close", async () => {
    // Every stream, no matter how it ends, must leave the process-wide
    // reservation total at zero: `handleStream`'s `finally` block releases
    // exactly one owner exactly one time on every exit path.

    async function runCompletionCase(): Promise<void> {
      const { handle, bridgeToken, clientSide } = bindTestServer({
        forwardRequest: async (request) => {
          request.reservation.reserve(1_000);
          return { status: 200, body: Buffer.from("{}", "utf8") };
        },
      });
      const rawClient = connectRawClient(clientSide);
      try {
        const response = await expectSessionStillServesARequest(rawClient, {
          method: "GET",
          path: "/api/agents/me",
          token: bridgeToken,
        });
        expect(response.status).toBe(200);
        expect(getBridgeBodyReservedBytesForTest()).toBe(0);
      } finally {
        rawClient.close();
        await handle.close();
      }
    }

    async function runErrorCase(): Promise<void> {
      const { handle, bridgeToken, clientSide } = bindTestServer({
        forwardRequest: async (request) => {
          request.reservation.reserve(1_000);
          throw new Error("forward handler fault");
        },
      });
      const rawClient = connectRawClient(clientSide);
      try {
        const response = await expectSessionStillServesARequest(rawClient, {
          method: "GET",
          path: "/api/agents/me",
          token: bridgeToken,
        });
        expect(response.status).toBe(502);
        expect(getBridgeBodyReservedBytesForTest()).toBe(0);
      } finally {
        rawClient.close();
        await handle.close();
      }
    }

    async function runAbortCase(): Promise<void> {
      const { handle, bridgeToken, clientSide } = bindTestServer({
        forwardRequest: async (request) => {
          request.reservation.reserve(1_000);
          await new Promise<void>((resolve) => {
            if (request.signal.aborted) {
              resolve();
              return;
            }
            request.signal.addEventListener("abort", () => resolve(), { once: true });
          });
          return { status: 200 };
        },
      });
      const rawClient = connectRawClient(clientSide);
      try {
        const stream = rawClient.request({
          ":method": "GET",
          ":path": "/api/agents/me",
          authorization: `Bearer ${bridgeToken}`,
        });
        const streamClosed = new Promise<void>((resolve) => {
          stream.on("error", () => resolve());
          stream.on("close", () => resolve());
        });
        stream.end();
        await new Promise((resolve) => setTimeout(resolve, 20));
        stream.close(http2.constants.NGHTTP2_CANCEL);
        await streamClosed;
        await new Promise((resolve) => setTimeout(resolve, 20));
        expect(getBridgeBodyReservedBytesForTest()).toBe(0);
      } finally {
        rawClient.close();
        await handle.close();
      }
    }

    async function runTimeoutCase(): Promise<void> {
      const { handle, bridgeToken, clientSide } = bindTestServer({
        requestBodyTimeoutMs: 30,
        forwardRequest: async () => ({ status: 200 }),
      });
      const rawClient = connectRawClient(clientSide);
      try {
        await new Promise<void>((resolve) => {
          const req = rawClient.request({
            ":method": "POST",
            ":path": "/api/issues/abc/comments",
            authorization: `Bearer ${bridgeToken}`,
          });
          req.on("error", () => resolve());
          req.on("close", () => resolve());
          req.write("partial-body");
        });
        await new Promise((resolve) => setTimeout(resolve, 20));
        expect(getBridgeBodyReservedBytesForTest()).toBe(0);
      } finally {
        rawClient.close();
        await handle.close();
      }
    }

    async function runCloseCase(): Promise<void> {
      const { handle, bridgeToken, clientSide } = bindTestServer({
        requestBodyTimeoutMs: 60_000,
        closeGraceMs: 30,
      });
      const rawClient = connectRawClient(clientSide);
      try {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        req.write("partial-body");
        await new Promise((resolve) => setTimeout(resolve, 20));
        await handle.close();
        await new Promise((resolve) => setTimeout(resolve, 20));
        expect(getBridgeBodyReservedBytesForTest()).toBe(0);
      } finally {
        rawClient.close();
      }
    }

    await runCompletionCase();
    await runErrorCase();
    await runAbortCase();
    await runTimeoutCase();
    await runCloseCase();
  }, 10_000);

  it("test_a_backpressured_response_holds_its_reservation_until_the_write_settles", async () => {
    // `stream.end(body)` only queues `body` for asynchronous transmission.
    // A client that never reads its response leaves those bytes in process
    // memory well after `end()` returns, so the reservation covering them
    // must stay held until the write actually settles, not merely until the
    // call to `end()` returns.
    const responseBytes = 4 * 1024 * 1024;
    const responseBody = Buffer.alloc(responseBytes, 0x61);
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        // Mirror the real forward path: the handler reserves the response
        // body it read from the sandbox target before returning it.
        if (!request.reservation.reserve(responseBytes)) {
          throw new BridgeProcessCapacityError();
        }
        return { status: 200, body: responseBody };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const req = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/me",
        authorization: `Bearer ${bridgeToken}`,
      });
      // Deliberately paused: no `resume()` and no `data` listener yet, so
      // the client never issues the flow-control credit the host needs to
      // finish writing a response this size.
      const headersReceived = new Promise<void>((resolve) => {
        req.once("response", () => resolve());
      });
      req.end();
      await headersReceived;
      // Give the host's `stream.end()` call, and its microtask queue, a
      // turn to run — the write is now queued but the client still is not
      // draining it.
      await new Promise((resolve) => setImmediate(resolve));
      expect(getBridgeBodyReservedBytesForTest()).toBeGreaterThan(0);

      const drained = new Promise<void>((resolve) => {
        req.on("data", () => {
          // Discard: this test only cares that the bytes left the host.
        });
        req.once("end", () => resolve());
      });
      req.resume();
      await drained;

      // Give the host's `finally` block a turn to run after the write
      // settles.
      await new Promise((resolve) => setImmediate(resolve));
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_peer_reset_holds_the_reservation_until_the_active_forward_settles", async () => {
    let capturedRequest: Http2BridgeForwardRequest | undefined;
    let releaseForward: (() => void) | undefined;
    const forwardHeld = new Promise<void>((resolve) => {
      releaseForward = resolve;
    });
    let forwardSettled = false;
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        capturedRequest = request;
        // Stand in for a live response-body copy `readBridgeForwardResponseBody`
        // (`execution-target.ts`) would reserve against this same owner.
        request.reservation.reserve(10_000);
        // The forward stays active past the peer's own reset: it settles
        // only when the test releases it below, not merely when the abort
        // signal fires — the same shape a real outbound `fetch` bound to
        // `request.signal` has, since an abort does not settle the fetch
        // promise synchronously.
        await new Promise<void>((resolve) => {
          if (request.signal.aborted) {
            resolve();
            return;
          }
          request.signal.addEventListener("abort", () => resolve(), { once: true });
        });
        await forwardHeld;
        forwardSettled = true;
        return { status: 200, body: Buffer.from("{}", "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const stream = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/me",
        authorization: `Bearer ${bridgeToken}`,
      });
      const streamClosed = new Promise<void>((resolve) => {
        stream.on("error", () => resolve());
        stream.on("close", () => resolve());
      });
      stream.end();
      // Give the request time to reach the server, enter `forwardRequest`,
      // and land its reservation.
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(capturedRequest).toBeDefined();
      expect(getBridgeBodyReservedBytesForTest()).toBe(10_000);

      // The peer resets the stream. The forward call is still pending.
      stream.close(http2.constants.NGHTTP2_CANCEL);
      await streamClosed;
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(capturedRequest!.signal.aborted).toBe(true);
      // The owner still holds its bytes: the active forward has not
      // settled yet, so `handleStream`'s `finally` has not released it.
      expect(forwardSettled).toBe(false);
      expect(getBridgeBodyReservedBytesForTest()).toBe(10_000);

      // Let the forward settle. The `finally` block releases the owner
      // exactly one time, and the process total returns to zero.
      releaseForward!();
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(forwardSettled).toBe(true);
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_stalled_body_releases_its_reservation_while_other_streams_hold_theirs", async () => {
    const releaseByPath = new Map<string, () => void>();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 60,
      forwardRequest: async (request) => {
        // Stand in for a live response-body buffer: reserve, then hold the
        // forward open until the test releases this exact path.
        request.reservation.reserve(20_000);
        await new Promise<void>((resolve) => {
          releaseByPath.set(request.pathname, resolve);
        });
        return { status: 200, body: Buffer.from("{}", "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const streamA = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/a",
        authorization: `Bearer ${bridgeToken}`,
      });
      const streamB = rawClient.request({
        ":method": "GET",
        ":path": "/api/agents/b",
        authorization: `Bearer ${bridgeToken}`,
      });
      // Drain each response as it arrives: `session.close()` later in this
      // test's teardown waits for every stream to fully end on both sides,
      // and an unread response can hold a stream open past that point.
      streamA.resume();
      streamB.resume();
      streamA.end();
      streamB.end();
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(getBridgeBodyReservedBytesForTest()).toBe(40_000);

      // The third stream's request body stalls past the idle bound. It
      // never reaches `forwardRequest`, so it never reserves anything of
      // its own; its owner still holds zero bytes at the moment it releases.
      const streamC = rawClient.request({
        ":method": "POST",
        ":path": "/api/issues/abc/comments",
        authorization: `Bearer ${bridgeToken}`,
      });
      const streamCClosed = new Promise<void>((resolve) => {
        streamC.on("error", () => resolve());
        streamC.on("close", () => resolve());
      });
      streamC.write("partial-body");
      await streamCClosed;
      await new Promise((resolve) => setTimeout(resolve, 20));

      // The stalled stream's release changed nothing: the other two streams
      // keep their reservations.
      expect(getBridgeBodyReservedBytesForTest()).toBe(40_000);

      releaseByPath.get("/api/agents/a")?.();
      releaseByPath.get("/api/agents/b")?.();
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_denied_stream_drains_its_body_and_holds_no_bytes", async () => {
    const { handle, clientSide } = bindTestServer({
      forwardRequest: async () => ({ status: 200 }),
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number }>((resolve, reject) => {
        // No `authorization` header: the token check denies this stream
        // before its body is ever read.
        const req = rawClient.request({ ":method": "POST", ":path": "/api/issues/abc/comments" });
        let status = 0;
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", () => undefined);
        req.on("end", () => resolve({ status }));
        req.on("error", reject);
        req.write(Buffer.alloc(50_000, "a"));
        req.end();
      });
      expect(response.status).toBe(401);
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_request_body_reservation_denial_answers_503_on_the_wire", async () => {
    // A denied request-body chunk must leave the stream alive long enough
    // to answer 503 for real, on the wire — not merely destroy the stream
    // and leave the caller with a bare reset. `respondJson` only queues the
    // 503 write; the handler must wait for that write to settle before it
    // destroys the stream, or the client can see a reset instead of the
    // full response body asserted below.
    const filler = createBridgeBodyReservation();
    expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - 100)).toBe(true);
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        let status = 0;
        let body = "";
        req.setEncoding("utf8");
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        // Ten times the 100 bytes of headroom the filler above left.
        req.end(Buffer.alloc(1_000, "a"));
      });
      expect(response.status).toBe(503);
      expect(forwarderTracker.called).toBe(false);
      // A raw reset delivers no body at all: parsing it here proves the
      // full JSON response actually reached the client, not merely the
      // status line.
      expect(JSON.parse(response.body)).toEqual({
        error: "The bridge host reached its reserved process body byte ceiling. Retry later.",
      });

      // The stream ended with a real response, not a raw reset: the session
      // stays healthy, so a second, complete request still succeeds.
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/agents/me",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
      filler.release();
    }
  });

  it("test_a_stalled_client_still_frees_the_capacity_denial_reservation_and_slot", async () => {
    // The capacity-denial (503) path must not wait forever for its queued
    // write to settle: a client that grants no flow-control credit for the
    // response would otherwise hold this stream's reservation and
    // concurrent-stream slot open indefinitely. A zero initial window
    // denies the server any credit to send the 503 body with, at the
    // protocol layer, regardless of the fake transport's own buffering —
    // the same deterministic stall a real stalled peer produces.
    const filler = createBridgeBodyReservation();
    expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - 100)).toBe(true);
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      capacityDenialSettleDeadlineMs: 30,
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide, { initialWindowSize: 0 });
    try {
      const startMs = Date.now();
      const closed = new Promise<void>((resolve) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        req.on("error", () => undefined);
        req.on("close", () => resolve());
        // Ten times the 100 bytes of headroom the filler above left: the
        // request body itself denies the reservation and enters the 503
        // path.
        req.end(Buffer.alloc(1_000, "a"));
      });
      await closed;
      // The deadline bounded the wait: the stream closed near the deadline
      // bound, not after some much longer natural settle that never comes.
      expect(Date.now() - startMs).toBeLessThan(5_000);
      expect(forwarderTracker.called).toBe(false);

      // The stream's own reservation released even though the client never
      // drained the 503 body — only the filler's own bytes remain held.
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(getBridgeBodyReservedBytesForTest()).toBe(filler.heldBytes);

      // The stream's slot freed too. Restore a normal flow-control window
      // first: only the denied stream above needs to stall.
      await new Promise<void>((resolve) => rawClient.settings({ initialWindowSize: 65_535 }, () => resolve()));
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/agents/me",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
      filler.release();
    }
  });

  it("test_a_stalled_normal_response_still_frees_its_reservation_and_slot", async () => {
    // The completed-response (200) write path must not wait forever for its
    // queued write to settle either: a client that grants no flow-control
    // credit for the response would otherwise hold this stream's
    // reservation and concurrent-stream slot open indefinitely, the same
    // failure mode the capacity-denial path already guards against. A zero
    // initial window denies the server any credit to send the response body
    // with, at the protocol layer, regardless of the fake transport's own
    // buffering.
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      responseWriteSettleDeadlineMs: 30,
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200, headers: {}, body: Buffer.alloc(1_000, "a") };
      },
    });
    const rawClient = connectRawClient(clientSide, { initialWindowSize: 0 });
    try {
      const startMs = Date.now();
      const closed = new Promise<void>((resolve) => {
        const req = rawClient.request({
          ":method": "GET",
          ":path": "/api/agents/me",
          authorization: `Bearer ${bridgeToken}`,
        });
        req.on("error", () => undefined);
        req.on("close", () => resolve());
        req.end();
      });
      await closed;
      expect(forwarderTracker.called).toBe(true);
      // The deadline bounded the wait: the stream closed near the deadline
      // bound, not after some much longer natural settle that never comes.
      expect(Date.now() - startMs).toBeLessThan(5_000);

      // This stream's own reservation released even though the client never
      // drained the response body.
      await new Promise((resolve) => setTimeout(resolve, 20));
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);

      // The stream's slot freed too. Restore a normal flow-control window
      // first: only the stalled stream above needs to stall.
      await new Promise<void>((resolve) => rawClient.settings({ initialWindowSize: 65_535 }, () => resolve()));
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/agents/me",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_normal_response_that_settles_in_time_is_left_alone", async () => {
    // A response that genuinely settles inside the deadline must not be
    // force-destroyed: the bounded wait exists only for a stream that never
    // settles on its own.
    const { handle, bridgeToken, clientSide } = bindTestServer({
      responseWriteSettleDeadlineMs: 5_000,
      forwardRequest: async () => ({
        status: 200,
        headers: { "content-type": "application/json" },
        body: Buffer.from(JSON.stringify({ ok: true }), "utf8"),
      }),
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/agents/me",
        token: bridgeToken,
      });
      expect(response.status).toBe(200);
      expect(JSON.parse(response.body)).toEqual({ ok: true });
      expect(getBridgeBodyReservedBytesForTest()).toBe(0);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_response_reservation_denial_answers_503_on_the_wire", async () => {
    // The response reader on the sandbox target side
    // (`readBridgeForwardResponseBody`) denies its own reservation and
    // throws `BridgeProcessCapacityError` from inside `forwardRequest`,
    // after the forward handler already ran — a different call site than
    // the request-body denial above, but the same 503-before-destroy
    // contract must hold: the stream must stay alive long enough to answer
    // 503 for real, on the wire, not merely destroy and leave the caller
    // with a bare reset.
    const forwarderTracker = createForwarderCallTracker();
    // Only the first call denies: the surviving-request check below reuses
    // the same session for a second, independent request, which must
    // succeed once this first stream's reservation released.
    let forwardCalls = 0;
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        forwardCalls += 1;
        if (forwardCalls === 1) {
          throw new BridgeProcessCapacityError();
        }
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        const req = rawClient.request({
          ":method": "GET",
          ":path": "/api/agents/me",
          authorization: `Bearer ${bridgeToken}`,
        });
        let status = 0;
        let body = "";
        req.setEncoding("utf8");
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        req.end();
      });
      expect(response.status).toBe(503);
      expect(forwarderTracker.called).toBe(true);
      // A raw reset delivers no body at all: parsing it here proves the
      // full JSON response actually reached the client, not merely the
      // status line.
      expect(JSON.parse(response.body)).toEqual({
        error: "The bridge host reached its reserved process body byte ceiling. Retry later.",
      });

      // The stream ended with a real response, not a raw reset: the session
      // stays healthy, so a second, complete request still succeeds.
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/agents/me",
        token: bridgeToken,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_denied_concatenated_request_body_never_allocates_the_copy", async () => {
    // The chunk-array copy and the concatenated copy are two separate live
    // buffers, so each must reserve on its own. Leave room for the first but
    // not the second: a correct reader checks the reservation before
    // `Buffer.concat` allocates the copy, so the denied concatenated copy
    // must never call `Buffer.concat` at all.
    const chunkBytes = 200_000;
    const filler = createBridgeBodyReservation();
    expect(filler.reserve(HTTP2_BRIDGE_MAX_PROCESS_BODY_BYTES - Math.floor(chunkBytes * 1.5))).toBe(true);
    const concatSpy = vi.spyOn(Buffer, "concat");
    const forwarderTracker = createForwarderCallTracker();
    const { handle, bridgeToken, clientSide } = bindTestServer({
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          authorization: `Bearer ${bridgeToken}`,
        });
        let status = 0;
        let body = "";
        req.setEncoding("utf8");
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        req.write(Buffer.alloc(chunkBytes, "a"), () => {
          // Give the host a turn to receive the one chunk and reserve its
          // bytes before this test ends the stream and triggers the
          // concatenated-copy reservation attempt.
          setTimeout(() => {
            // The chunk-array reservation alone must already hold, on top of
            // the filler above, exactly the one chunk's bytes: the
            // concatenated-copy reservation has not run yet, because the
            // stream has not ended.
            expect(getBridgeBodyReservedBytesForTest()).toBe(filler.heldBytes + chunkBytes);
            req.end();
          }, 50);
        });
      });
      expect(response.status).toBe(503);
      expect(forwarderTracker.called).toBe(false);
      expect(concatSpy).not.toHaveBeenCalled();
    } finally {
      concatSpy.mockRestore();
      rawClient.close();
      await handle.close();
      filler.release();
    }
  });

  describe("parseCanonicalBridgeRequestPath", () => {
    it("parses an origin-form path with a query exactly one time", () => {
      const result = parseCanonicalBridgeRequestPath({ ":path": "/api/issues/abc?foo=bar" });
      expect(result).toEqual({ ok: true, value: { pathname: "/api/issues/abc", query: "?foo=bar" } });
    });

    it("test_a_non_origin_form_path_is_rejected", () => {
      expect(parseCanonicalBridgeRequestPath({ ":path": "http://evil.example/api" })).toEqual({
        ok: false,
        reason: "non_origin_form",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "//evil.example/api" })).toEqual({
        ok: false,
        reason: "non_origin_form",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "*" })).toEqual({
        ok: false,
        reason: "non_origin_form",
      });
    });

    it("test_an_encoded_separator_or_a_dot_segment_is_rejected", () => {
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api%2fissues/abc" })).toEqual({
        ok: false,
        reason: "encoded_slash",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api\\issues/abc" })).toEqual({
        ok: false,
        reason: "backslash",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api/issues\0/abc" })).toEqual({
        ok: false,
        reason: "nul_byte",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api/%00/abc" })).toEqual({
        ok: false,
        reason: "nul_byte",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api/../secrets" })).toEqual({
        ok: false,
        reason: "dot_segment",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api/%2e%2e/secrets" })).toEqual({
        ok: false,
        reason: "dot_segment",
      });
      expect(parseCanonicalBridgeRequestPath({ ":path": "/api/./issues" })).toEqual({
        ok: false,
        reason: "dot_segment",
      });
    });

    it("rejects a duplicate pseudo-header", () => {
      expect(
        parseCanonicalBridgeRequestPath({ ":path": ["/api/agents/me", "/api/agents/other"] } as never),
      ).toEqual({ ok: false, reason: "duplicate_pseudo_header" });
    });

    it("rejects a missing path", () => {
      expect(parseCanonicalBridgeRequestPath({})).toEqual({ ok: false, reason: "missing_path" });
    });
  });

  it("buildHttp2BridgeForwardUrl resolves the parsed pathname and query against the base URL", () => {
    const url = buildHttp2BridgeForwardUrl("http://127.0.0.1:4000", {
      pathname: "/api/issues/abc",
      query: "?foo=bar",
    });
    expect(url.toString()).toBe("http://127.0.0.1:4000/api/issues/abc?foo=bar");
  });

  it("classifyStreamAgainstGoaway classifies by the last processed stream id", () => {
    expect(classifyStreamAgainstGoaway(1, 3)).toBe("accepted");
    expect(classifyStreamAgainstGoaway(3, 3)).toBe("accepted");
    expect(classifyStreamAgainstGoaway(5, 3)).toBe("not_accepted");
  });

  it("test_a_stream_without_the_bridge_token_never_reaches_the_forwarder", async () => {
    const forwarderTracker = createForwarderCallTracker();
    const { handle, clientSide } = bindTestServer({
      forwardRequest: async (request) => {
        forwarderTracker.markCalled();
        return { status: 200, body: Buffer.from(JSON.stringify({ path: request.pathname }), "utf8") };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const response = await new Promise<{ status: number; body: string }>((resolve, reject) => {
        // No `authorization` header at all: the stream never carries a token.
        const req = rawClient.request({ ":method": "GET", ":path": "/api/agents/me" });
        let status = 0;
        let body = "";
        req.on("response", (headers) => {
          status = Number(headers[":status"]) || 0;
        });
        req.setEncoding("utf8");
        req.on("data", (chunk) => (body += chunk));
        req.on("end", () => resolve({ status, body }));
        req.on("error", reject);
        req.end();
      });
      expect(response.status).toBe(401);
      expect(forwarderTracker.called).toBe(false);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("test_a_denied_stream_with_an_unfinished_body_still_frees_its_stream_slot", async () => {
    // The server answers a denied stream (an invalid token, here) at once,
    // before it ever reads the request body. A peer that leaves that body
    // unfinished must not hold the stream open past the same idle bound an
    // authenticated request gets — otherwise, up to
    // `HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS` denied streams could each retain
    // a slot forever and block every legitimate callback.
    const forwarderTracker = createForwarderCallTracker();
    const { handle, clientSide } = bindTestServer({
      requestBodyTimeoutMs: 30,
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    const rawClient = connectRawClient(clientSide);
    try {
      const startMs = Date.now();
      const closed = new Promise<void>((resolve) => {
        const req = rawClient.request({
          ":method": "POST",
          ":path": "/api/issues/abc/comments",
          // No `authorization` header: the token check denies this stream
          // before the body is ever read.
        });
        req.on("error", () => undefined);
        req.on("close", () => resolve());
        // Drain the response so its readable side reaches `end`: nothing in
        // this test reads the response body otherwise, and an unread
        // response can itself hold the client stream open past `close`.
        req.resume();
        // A partial body the request never ends: without its own bound, the
        // inbound half of this stream would stay open indefinitely.
        req.write("partial-body");
      });
      await closed;
      expect(Date.now() - startMs).toBeLessThan(5_000);
      expect(forwarderTracker.called).toBe(false);

      // The session survives the freed stream: a second, complete request
      // still succeeds, proving the session itself stayed open and healthy.
      const survivingResponse = await expectSessionStillServesARequest(rawClient, {
        method: "GET",
        path: "/api/agents/me",
        token: BRIDGE_TOKEN,
      });
      expect(survivingResponse.status).toBe(200);
    } finally {
      rawClient.close();
      await handle.close();
    }
  });

  it("serves only exact GET schema discovery through HTTP/2", async () => {
    const schema = { openapi: "3.1.0", paths: {} };
    const forwarded: string[] = [];
    const { gateway, handle } = createTestPair({
      forwardRequest: async (request) => {
        forwarded.push(`${request.method} ${request.pathname}`);
        return { status: 200, body: Buffer.from(JSON.stringify(schema)) };
      },
    });
    try {
      for (const [method, path, status] of [
        ["GET", "/api/openapi.json", 200],
        ["POST", "/api/openapi.json", 403],
        ["PATCH", "/api/openapi.json", 403],
        ["DELETE", "/api/openapi.json", 403],
        ["GET", "/api/openapi.json/extra", 403],
        ["GET", "/api/openapiXjson", 403],
        ["GET", "/api/secrets", 403],
      ] as const) {
        const response = await gateway.forwardRequest({
          method, path, query: "", headers: {}, body: Buffer.alloc(0), receivedToken: BRIDGE_TOKEN,
        });
        expect(response.status).toBe(status);
        if (status === 200) expect(JSON.parse(response.body!.toString())).toEqual(schema);
      }
      expect(forwarded).toEqual(["GET /api/openapi.json"]);
    } finally {
      await gateway.close();
      await handle.close();
    }
  });

  it("rejects a route the allowlist does not carry, before the forwarder runs", async () => {
    const forwarderTracker = createForwarderCallTracker();
    const { gateway, handle } = createTestPair({
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    try {
      const response = await gateway.forwardRequest({
        method: "DELETE",
        path: "/api/agents/me",
        query: "",
        headers: {},
        body: Buffer.alloc(0),
        receivedToken: BRIDGE_TOKEN,
      });
      expect(response.status).toBe(403);
      expect(forwarderTracker.called).toBe(false);
    } finally {
      await gateway.close();
      await handle.close();
    }
  });

  it("test_a_maximum_size_multipart_attachment_upload_fits_through_the_bridge", async () => {
    // A file at the exact attachment content ceiling (10 MiB — the same
    // figure `MAX_ATTACHMENT_BYTES` in `server/src/attachment-types.ts`
    // enforces) still crosses the bridge wrapped in a multipart body: the
    // boundary line and the part's own headers add bytes on top of that file
    // content. The bridge's own body limit needs headroom for that framing,
    // or a valid maximum-size attachment fails here before the server ever
    // sees it.
    const maxAttachmentBytes = 10 * 1024 * 1024;
    const boundary = "----PaperclipTestBoundary1234567890abcdef";
    const preamble = Buffer.from(
      `--${boundary}\r\n` +
        `Content-Disposition: form-data; name="file"; filename="attachment.bin"\r\n` +
        `Content-Type: application/octet-stream\r\n\r\n`,
      "utf8",
    );
    const fileContent = Buffer.alloc(maxAttachmentBytes, 0x61);
    const epilogue = Buffer.from(`\r\n--${boundary}--\r\n`, "utf8");
    const multipartBody = Buffer.concat([preamble, fileContent, epilogue]);

    // The wrapper really does add bytes on top of the file content alone —
    // otherwise this test would prove nothing about framing headroom.
    expect(multipartBody.byteLength).toBeGreaterThan(maxAttachmentBytes);
    expect(multipartBody.byteLength).toBeLessThanOrEqual(DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES);

    let receivedBodyBytes = 0;
    const { gateway, handle } = createTestPair({
      routes: HTTP2_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST,
      forwardRequest: async (request) => {
        receivedBodyBytes = request.body.byteLength;
        return { status: 201, body: Buffer.from("{}", "utf8") };
      },
    });
    try {
      const response = await gateway.forwardRequest({
        method: "POST",
        path: "/api/companies/co-1/issues/issue-1/attachments",
        query: "",
        headers: { "content-type": `multipart/form-data; boundary=${boundary}` },
        body: multipartBody,
        receivedToken: BRIDGE_TOKEN,
      });
      expect(response.status).toBe(201);
      expect(receivedBodyBytes).toBe(multipartBody.byteLength);
    } finally {
      await gateway.close();
      await handle.close();
    }
  }, 20_000);

  it("the sandbox gateway keeps its own token check before it opens a stream", async () => {
    const forwarderTracker = createForwarderCallTracker();
    const { gateway, handle } = createTestPair({
      forwardRequest: async () => {
        forwarderTracker.markCalled();
        return { status: 200 };
      },
    });
    try {
      await expect(
        gateway.forwardRequest({
          method: "GET",
          path: "/api/agents/me",
          query: "",
          headers: {},
          body: Buffer.alloc(0),
          receivedToken: "wrong-token",
        }),
      ).rejects.toThrow(/invalid bridge token/i);
      expect(forwarderTracker.called).toBe(false);
    } finally {
      await gateway.close();
      await handle.close();
    }
  });

  it("wrapDuplexChannelAsNodeDuplex relays writes and pushes through onData", async () => {
    const written: Buffer[] = [];
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    let exitListener: ((exit: { exitCode: number | null }) => void) | undefined;
    const channel: CommandManagedDuplexChannel = {
      write: (data) => {
        written.push(Buffer.from(data));
      },
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: (listener) => {
        exitListener = listener;
      },
      stop: () => undefined,
      close: async () => undefined,
    };
    const duplex = wrapDuplexChannelAsNodeDuplex(channel);
    const received: Buffer[] = [];
    duplex.on("data", (chunk: Buffer) => received.push(chunk));

    duplex.write(Buffer.from("outbound"));
    await new Promise((resolve) => setImmediate(resolve));
    expect(Buffer.concat(written).toString("utf8")).toBe("outbound");

    dataListener?.(Buffer.from("inbound"));
    await new Promise((resolve) => setImmediate(resolve));
    expect(Buffer.concat(received).toString("utf8")).toBe("inbound");

    const ended = new Promise<void>((resolve) => duplex.on("end", resolve));
    exitListener?.({ exitCode: 0 });
    await ended;
  });

  it("wrapDuplexChannelAsNodeDuplex queues chunks past a full readable side and drains them once the consumer reads again", async () => {
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    const channel: CommandManagedDuplexChannel = {
      write: () => undefined,
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: () => undefined,
      stop: () => undefined,
      close: async () => undefined,
    };
    const duplex = wrapDuplexChannelAsNodeDuplex(channel, { maxBufferedReadBytes: 1_000_000 });
    const received: Buffer[] = [];
    // The first chunk alone passes the readable side's default 64 KiB
    // high-water mark, so `push()` reports the readable side full. The two
    // chunks that follow queue in the wrapper instead of pushing past that
    // signal — this is the bounded flow-control path this fix adds.
    dataListener?.(Buffer.alloc(70_000, "a"));
    dataListener?.(Buffer.from("-second-"));
    dataListener?.(Buffer.from("-third-"));

    // Attaching a "data" listener now starts flowing mode, which drives the
    // wrapper's own `read()` until every queued chunk drains, proving no
    // queued chunk was dropped and the arrival order held.
    const drainedAll = new Promise<void>((resolve) => {
      duplex.on("data", (chunk: Buffer) => {
        received.push(chunk);
        if (Buffer.concat(received).includes("-third-")) resolve();
      });
    });
    await drainedAll;

    expect(Buffer.concat(received).toString("utf8").endsWith("-second--third-")).toBe(true);
  });

  it("wrapDuplexChannelAsNodeDuplex stops the channel and destroys the duplex once the bounded read backpressure buffer overflows", async () => {
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    let stopped = false;
    const channel: CommandManagedDuplexChannel = {
      write: () => undefined,
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: () => undefined,
      stop: () => {
        stopped = true;
      },
      close: async () => undefined,
    };
    // The cap must clear the default 65,536-byte readable high-water mark,
    // so the first chunk below still passes the direct-push check and
    // forces `push()` to report the readable side full; the overflow this
    // test proves comes from the queue that follows, not from that first
    // chunk on its own.
    const duplex = wrapDuplexChannelAsNodeDuplex(channel, { maxBufferedReadBytes: 80_000 });
    // No consumer ever attaches, so the readable side never drains: every
    // chunk past the first, which alone passes the default high-water mark,
    // fills the bounded queue instead of the unbounded internal buffer a
    // caller ignoring `push()`'s return would grow.
    const errored = new Promise<Error>((resolve) => duplex.on("error", resolve));

    dataListener?.(Buffer.alloc(70_000, "a")); // passes the high-water mark and the cap; still pushed directly.
    dataListener?.(Buffer.alloc(30_000, "b")); // queues: 30,000 of the 80,000-byte cap.
    dataListener?.(Buffer.alloc(30_000, "b")); // queues: 60,000 of the 80,000-byte cap.
    dataListener?.(Buffer.alloc(30_000, "b")); // 90,000 queued bytes passes the cap.

    const error = await errored;
    expect(error.message).toMatch(/backpressure/i);
    expect(stopped).toBe(true);
  });

  it("stops the channel when the read queue passes the lowered byte bound", async () => {
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    let stopped = false;
    const channel: CommandManagedDuplexChannel = {
      write: () => undefined,
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: () => undefined,
      stop: () => {
        stopped = true;
      },
      close: async () => undefined,
    };
    // No override: this test proves the default bound itself is the direct
    // 524,288-byte value, no longer derived from `HTTP2_BRIDGE_MAX_SESSION_MEMORY`.
    // No consumer ever attaches, so the readable side never drains.
    expect(DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES).toBe(524_288);
    const duplex = wrapDuplexChannelAsNodeDuplex(channel);
    const errored = new Promise<Error>((resolve) => duplex.on("error", resolve));

    dataListener?.(Buffer.alloc(70_000, "a")); // passes the default high-water mark; still pushed directly.
    dataListener?.(Buffer.alloc(500_000, "b")); // queues most of the 524,288-byte default cap.
    dataListener?.(Buffer.alloc(60_000, "b")); // passes the default cap.

    const error = await errored;
    expect(error.message).toMatch(/backpressure/i);
    expect(stopped).toBe(true);
  });

  it("wrapDuplexChannelAsNodeDuplex fails closed on one inbound chunk larger than the bounded read backpressure buffer, before the queue holds anything to compare it against", async () => {
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    let stopped = false;
    const channel: CommandManagedDuplexChannel = {
      write: () => undefined,
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: () => undefined,
      stop: () => {
        stopped = true;
      },
      close: async () => undefined,
    };
    // No consumer ever attaches, and the queue is empty when this chunk
    // arrives, so a check that only bounds the queue's cumulative size (and
    // not one chunk's own size) would let this chunk reach `push()` unbound.
    const duplex = wrapDuplexChannelAsNodeDuplex(channel, { maxBufferedReadBytes: 5_000 });
    const errored = new Promise<Error>((resolve) => duplex.on("error", resolve));

    dataListener?.(Buffer.alloc(10_000, "a")); // exceeds the 5,000-byte cap on its own, on the very first chunk.

    const error = await errored;
    expect(error.message).toMatch(/backpressure/i);
    expect(stopped).toBe(true);
  });

  it("wrapDuplexChannelAsNodeDuplex fails closed once the read backpressure queue stalls with no drain, even though it stays under the byte cap", async () => {
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    let stopped = false;
    const channel: CommandManagedDuplexChannel = {
      write: () => undefined,
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: () => undefined,
      stop: () => {
        stopped = true;
      },
      close: async () => undefined,
    };
    // A generous byte cap, so the byte-cap check above never fires: this
    // test proves the independent time bound catches a stuck consumer the
    // byte cap alone would miss.
    const duplex = wrapDuplexChannelAsNodeDuplex(channel, {
      maxBufferedReadBytes: 10_000_000,
      readBackpressureStallMs: 40,
    });
    const errored = new Promise<Error>((resolve) => duplex.on("error", resolve));

    dataListener?.(Buffer.alloc(70_000, "a")); // passes the high-water mark; pushed directly, no consumer drains it.
    dataListener?.(Buffer.from("queued-and-never-drained")); // queues; no "data" listener ever attaches to drain it.

    const error = await errored;
    expect(error.message).toMatch(/stall/i);
    expect(stopped).toBe(true);
  });

  it("wrapDuplexChannelAsNodeDuplex clears the stall bound once the queue fully drains, instead of firing later on an idle channel", async () => {
    let dataListener: ((chunk: Uint8Array) => void) | undefined;
    const channel: CommandManagedDuplexChannel = {
      write: () => undefined,
      onData: (listener) => {
        dataListener = listener;
      },
      onExit: () => undefined,
      stop: () => undefined,
      close: async () => undefined,
    };
    // A short stall bound, so this test proves the drain below clears the
    // timer instead of merely finishing before a long one would have fired.
    const duplex = wrapDuplexChannelAsNodeDuplex(channel, {
      maxBufferedReadBytes: 10_000_000,
      readBackpressureStallMs: 20,
    });
    const received: Buffer[] = [];
    const drainedAll = new Promise<void>((resolve) => {
      duplex.on("data", (chunk: Buffer) => {
        received.push(chunk);
        if (Buffer.concat(received).includes("-third-")) resolve();
      });
    });

    dataListener?.(Buffer.alloc(70_000, "a")); // passes the high-water mark; queues the chunks that follow.
    dataListener?.(Buffer.from("-second-"));
    dataListener?.(Buffer.from("-third-"));
    await drainedAll;

    // Wait past the stall bound with the channel now idle and the queue
    // empty. A timer the full drain above did not clear would fire here.
    await new Promise((resolve) => setTimeout(resolve, 60));
    expect(duplex.destroyed).toBe(false);
  });
});