fix(lab): preserve CL-03 live transport failure classes

src/lib/lab-live-pinned-sender.ts

@@ -19,27 +19,38 @@ export function createLabAuthorizedPinnedSender(
       headers,
       maxBytes: limits.maxOutputBytes,
       connectTimeoutMs: limits.connectTimeoutMs,
-      idleTimeoutMs: Math.min(limits.firstByteTimeoutMs, limits.inactivityTimeoutMs),
+      firstByteTimeoutMs: limits.firstByteTimeoutMs,
+      inactivityTimeoutMs: limits.inactivityTimeoutMs,
       rejectUnauthorized: true,
       context: "Lab provider response",
     };
     let response: Response;
+    let body: string;
     try {
       response = request.method === "POST"
         ? await pinnedHttpPost(url, pinned, request.body ?? "", signal, options)
         : await pinnedHttpGet(url, pinned, signal, options);
+      body = await response.text();
     } catch (error) {
-      if (error instanceof PinnedHttpError && error.code === "connect_timeout") {
-        throw new TransportError("connect_timeout", "pinned provider connection timed out");
+      if (error instanceof PinnedHttpError) {
+        switch (error.code) {
+          case "connect_timeout":
+            throw new TransportError("connect_timeout", "pinned provider connection timed out");
+          case "first_byte_timeout":
+            throw new TransportError("first_byte_timeout", "pinned provider first byte timed out");
+          case "inactivity_timeout":
+            throw new TransportError("inactivity_timeout", "pinned provider response stalled");
+          case "output_byte_limit":
+            throw new TransportError("output_byte_limit", "pinned provider response exceeded byte budget");
+        }
       }
       throw error;
     }
-    const body = await response.text();
     const responseHeaders: Record<string, string> = {};
     for (const headerName of LAB_RESPONSE_HEADER_ALLOWLIST) {
       const value = response.headers.get(headerName);
       if (value !== null) responseHeaders[headerName] = value;
     }
     return { status: response.status, headers: responseHeaders, body };
   };
-}
+}

src/lib/pinned-http.ts

@@ -3,7 +3,11 @@ import https from "node:https";
 
 export type PinnedAddress = { address: string; family: number };
 
-export type PinnedHttpErrorCode = "connect_timeout";
+export type PinnedHttpErrorCode =
+  | "connect_timeout"
+  | "first_byte_timeout"
+  | "inactivity_timeout"
+  | "output_byte_limit";
 
 export class PinnedHttpError extends Error {
   override readonly name = "PinnedHttpError";
@@ -15,6 +19,11 @@ export interface PinnedHttpRequestOptions {
   maxBytes?: number;
   /** Optional deadline for establishing the TCP connection and, for HTTPS, completing TLS. */
   connectTimeoutMs?: number;
+  /** Optional deadline from connection establishment until response headers arrive. */
+  firstByteTimeoutMs?: number;
+  /** Optional maximum idle interval between response-body chunks. */
+  inactivityTimeoutMs?: number;
+  /** @deprecated Use firstByteTimeoutMs and inactivityTimeoutMs. */
   idleTimeoutMs?: number;
   rejectUnauthorized?: boolean;
   context?: string;
@@ -37,7 +46,11 @@ function pinnedHttpRequest(
   }
   const context = options?.context ?? "request";
   const connectTimeoutMs = options?.connectTimeoutMs;
-  const idleTimeoutMs = options?.idleTimeoutMs ?? 60_000;
+  const legacyIdleTimeoutMs = options?.idleTimeoutMs ?? 60_000;
+  const usesLegacyIdleTimeout = options?.firstByteTimeoutMs === undefined
+    && options?.inactivityTimeoutMs === undefined;
+  const firstByteTimeoutMs = options?.firstByteTimeoutMs ?? legacyIdleTimeoutMs;
+  const inactivityTimeoutMs = options?.inactivityTimeoutMs ?? legacyIdleTimeoutMs;
   const maxBytes = options?.maxBytes;
   const headers = new Headers(options?.headers);
   headers.set("host", parsed.host);
@@ -56,17 +69,30 @@ function pinnedHttpRequest(
     let settled = false;
     let req: ClientRequest | undefined;
     let connectTimer: ReturnType<typeof setTimeout> | undefined;
+    let firstByteTimer: ReturnType<typeof setTimeout> | undefined;
     const clearConnectTimer = () => {
       if (connectTimer !== undefined) clearTimeout(connectTimer);
       connectTimer = undefined;
     };
+    const clearFirstByteTimer = () => {
+      if (firstByteTimer !== undefined) clearTimeout(firstByteTimer);
+      firstByteTimer = undefined;
+    };
     const fail = (error: unknown) => {
       clearConnectTimer();
+      clearFirstByteTimer();
       try { req?.destroy(); } catch { /* ignore */ }
       if (settled) return;
       settled = true;
       reject(error instanceof Error ? error : new Error(String(error)));
     };
+    const startFirstByteTimer = () => {
+      clearFirstByteTimer();
+      firstByteTimer = setTimeout(
+        () => fail(new PinnedHttpError("first_byte_timeout", `${context} first byte timed out`)),
+        firstByteTimeoutMs,
+      );
+    };
     const requestOptions: RequestOptions & { servername?: string } = {
       protocol: parsed.protocol,
       hostname: parsed.hostname,
@@ -101,6 +127,7 @@ function pinnedHttpRequest(
 
     const onResponse = (response: IncomingMessage) => {
       clearConnectTimer();
+      clearFirstByteTimer();
       const status = response.statusCode ?? 0;
       const responseHeaders = new Headers();
       for (const [key, value] of Object.entries(response.headers)) {
@@ -124,28 +151,35 @@ function pinnedHttpRequest(
       let received = 0;
       const stream = new ReadableStream<Uint8Array>({
         start(controller) {
-          response.setTimeout(idleTimeoutMs, () => {
-            const error = new Error(`${context} stalled`);
-            fail(error);
+          let bodySettled = false;
+          const failBody = (error: Error) => {
+            if (bodySettled) return;
+            bodySettled = true;
             try { controller.error(error); } catch { /* closed */ }
+            try { response.destroy(); } catch { /* ignore */ }
+            try { req?.destroy(); } catch { /* ignore */ }
+          };
+
+          response.setTimeout(inactivityTimeoutMs, () => {
+            failBody(new PinnedHttpError("inactivity_timeout", `${context} stalled`));
           });
           response.on("data", (chunk: Buffer | string) => {
+            if (bodySettled) return;
             const buffer = typeof chunk === "string" ? Buffer.from(chunk) : chunk;
             received += buffer.byteLength;
             if (maxBytes !== undefined && received > maxBytes) {
-              const error = new Error(`${context} exceeds ${maxBytes} byte cap`);
-              fail(error);
-              try { controller.error(error); } catch { /* closed */ }
+              failBody(new PinnedHttpError("output_byte_limit", `${context} exceeds ${maxBytes} byte cap`));
               return;
             }
             try { controller.enqueue(buffer); } catch { /* closed */ }
           });
           response.on("end", () => {
+            if (bodySettled) return;
+            bodySettled = true;
             try { controller.close(); } catch { /* closed */ }
           });
           response.on("error", (error: Error) => {
-            fail(error);
-            try { controller.error(error); } catch { /* closed */ }
+            failBody(error);
           });
         },
         cancel() {
@@ -163,20 +197,40 @@ function pinnedHttpRequest(
     const onAbort = () => fail(signal?.reason instanceof Error ? signal.reason : new Error("aborted"));
     signal?.addEventListener("abort", onAbort, { once: true });
     req.on("socket", (socket) => {
-      if (!socket.connecting || connectTimeoutMs === undefined) return;
       const connectedEvent = parsed.protocol === "https:" ? "secureConnect" : "connect";
-      connectTimer = setTimeout(() => fail(new PinnedHttpError("connect_timeout", `${context} connect timed out`)), connectTimeoutMs);
-      socket.once(connectedEvent, clearConnectTimer);
-      socket.once("error", clearConnectTimer);
-      socket.once("close", clearConnectTimer);
+      if (!socket.connecting) {
+        if (!usesLegacyIdleTimeout) startFirstByteTimer();
+        return;
+      }
+      if (connectTimeoutMs !== undefined) {
+        connectTimer = setTimeout(
+          () => fail(new PinnedHttpError("connect_timeout", `${context} connect timed out`)),
+          connectTimeoutMs,
+        );
+      }
+      socket.once(connectedEvent, () => {
+        clearConnectTimer();
+        if (!usesLegacyIdleTimeout) startFirstByteTimer();
+      });
+      socket.once("error", () => {
+        clearConnectTimer();
+        clearFirstByteTimer();
+      });
+      socket.once("close", () => {
+        clearConnectTimer();
+        clearFirstByteTimer();
+      });
     });
-    req.setTimeout(idleTimeoutMs, () => fail(new Error(`${context} timed out`)));
+    if (usesLegacyIdleTimeout) {
+      req.setTimeout(legacyIdleTimeoutMs, () => fail(new Error(`${context} timed out`)));
+    }
     req.on("error", error => {
       signal?.removeEventListener("abort", onAbort);
       fail(error);
     });
     req.on("close", () => {
       clearConnectTimer();
+      clearFirstByteTimer();
       signal?.removeEventListener("abort", onAbort);
     });
     req.end(body);

tests/lab-live-pinned-timeouts.test.ts

@@ -0,0 +1,117 @@
+import { createServer, type IncomingMessage, type Server, type ServerResponse } from "node:http";
+import { afterEach, describe, expect, test } from "bun:test";
+import { createLabAuthorizedPinnedSender } from "../src/lib/lab-live-pinned-sender";
+import type { LabCredentialLeaseV1, LabDestinationV1, LiveRunConfig } from "../src/lab/live/types";
+
+const SERVERS: Server[] = [];
+
+afterEach(async () => {
+  for (const server of SERVERS.splice(0)) {
+    server.closeAllConnections?.();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+  }
+});
+
+async function listen(handler: (req: IncomingMessage, res: ServerResponse) => void): Promise<number> {
+  return await new Promise<number>((resolve, reject) => {
+    const server = createServer(handler);
+    SERVERS.push(server);
+    const onError = (error: Error) => reject(error);
+    server.once("error", onError);
+    server.listen(0, "127.0.0.1", () => {
+      server.off("error", onError);
+      const address = server.address();
+      if (!address || typeof address === "string") {
+        reject(new Error("loopback test server did not expose a TCP port"));
+        return;
+      }
+      resolve(address.port);
+    });
+  });
+}
+
+const BASE_LIMITS: LiveRunConfig = {
+  totalTimeoutMs: 1_000,
+  connectTimeoutMs: 250,
+  firstByteTimeoutMs: 30,
+  inactivityTimeoutMs: 30,
+  maxRequests: 2,
+  maxInputBytes: 1024,
+  maxOutputBytes: 1024,
+  maxOutputTokens: 1024,
+  maxToolCalls: 8,
+  maxMemoryBytes: 64 * 1024 * 1024,
+  maxChildProcesses: 0,
+  maxArtifacts: 4,
+  perArtifactBytes: 64 * 1024,
+  aggregateArtifactBytes: 256 * 1024,
+};
+
+function destination(port: number): LabDestinationV1 {
+  return {
+    scheme: "http",
+    host: "lab-timeout.invalid",
+    port,
+    basePath: "",
+    sniHost: "lab-timeout.invalid",
+    addresses: [{ address: "127.0.0.1", family: 4 }],
+    privateNetwork: true,
+    fingerprint: "a".repeat(64),
+  };
+}
+
+async function send(port: number, limitOverrides: Partial<LiveRunConfig> = {}) {
+  const sender = createLabAuthorizedPinnedSender(() => ({}));
+  return await sender(
+    {} as LabCredentialLeaseV1,
+    destination(port),
+    { address: "127.0.0.1", family: 4 },
+    { method: "POST", path: "/", body: "{}" },
+    new AbortController().signal,
+    { ...BASE_LIMITS, ...limitOverrides },
+  );
+}
+
+describe("CL-03 pinned live transport failure classification", () => {
+  test("preserves first-byte timeout as a transport timeout", async () => {
+    const port = await listen((_req, res) => {
+      setTimeout(() => {
+        if (res.destroyed) return;
+        res.writeHead(200, { "content-type": "application/json" });
+        res.end("{}");
+      }, 150);
+    });
+
+    await expect(send(port, { firstByteTimeoutMs: 30, inactivityTimeoutMs: 250 })).rejects.toMatchObject({
+      name: "TransportError",
+      code: "first_byte_timeout",
+    });
+  });
+
+  test("preserves response inactivity as inactivity_timeout", async () => {
+    const port = await listen((_req, res) => {
+      res.writeHead(200, { "content-type": "application/json" });
+      res.write("{\"ok\":");
+      setTimeout(() => {
+        if (!res.destroyed) res.end("true}");
+      }, 150);
+    });
+
+    await expect(send(port, { firstByteTimeoutMs: 250, inactivityTimeoutMs: 30 })).rejects.toMatchObject({
+      name: "TransportError",
+      code: "inactivity_timeout",
+    });
+  });
+
+  test("preserves the output byte ceiling as output_byte_limit", async () => {
+    const port = await listen((_req, res) => {
+      res.writeHead(200, { "content-type": "text/plain" });
+      res.end("x".repeat(128));
+    });
+
+    await expect(send(port, { maxOutputBytes: 16 })).rejects.toMatchObject({
+      name: "TransportError",
+      code: "output_byte_limit",
+    });
+  });
+});