fix(gateway): centralize streamed media descriptor ownership (#117000)

Co-authored-by: Peter Steinberger <steipete@macos.shared>

src/gateway/control-ui.ts

@@ -81,7 +81,11 @@ import {
   sendControlUiHtmlBody,
   serveControlUiAsset,
 } from "./control-ui-static.js";
-import { resolveByteResponse, writeByteHeaders } from "./http-byte-range.js";
+import {
+  createGatewayByteStream,
+  resolveByteResponse,
+  writeByteHeaders,
+} from "./http-byte-range.js";
 import { buildMissingScopeForbiddenBody, sendGatewayAuthFailure } from "./http-common.js";
 import {
   getBearerToken,
@@ -667,21 +671,13 @@ export async function handleControlUiAssistantMediaRequest(
     return true;
   }
 
-  let opened: Awaited<ReturnType<typeof openLocalFileSafely>> | null = null;
-  let localPath;
-  let handleClosed = false;
-  const closeOpenedHandle = async () => {
-    if (!opened || handleClosed) {
-      return;
-    }
-    handleClosed = true;
-    await opened.handle.close().catch(() => {});
-  };
+  let byteStream: ReturnType<typeof createGatewayByteStream> | undefined;
   try {
     const resolvedReference = await resolveMediaReferenceLocalPathInfo(source);
-    localPath = resolvedReference.path;
+    const localPath = resolvedReference.path;
     await assertLocalMediaAllowed(localPath, localRoots);
-    opened = await openLocalFileSafely({ filePath: localPath });
+    let opened = await openLocalFileSafely({ filePath: localPath });
+    byteStream = createGatewayByteStream(res, opened.handle, () => respondControlUiNotFound(res));
     const sniffLength = Math.min(opened.stat.size, 8192);
     const sniffBuffer = sniffLength > 0 ? Buffer.allocUnsafe(sniffLength) : undefined;
     const bytesRead =
@@ -709,16 +705,18 @@ export async function handleControlUiAssistantMediaRequest(
         kind: mediaKind,
       });
       if (playback.kind === "preparing") {
-        await closeOpenedHandle();
+        await byteStream.close();
         sendJson(res, 202, { status: "preparing" });
         return true;
       }
       if (playback.kind === "transcoded") {
         const transcoded = await openLocalFileSafely({ filePath: playback.path }).catch(() => null);
         if (transcoded) {
-          await closeOpenedHandle();
+          await byteStream.close();
           opened = transcoded;
-          handleClosed = false;
+          byteStream = createGatewayByteStream(res, opened.handle, () =>
+            respondControlUiNotFound(res),
+          );
           contentType = playback.contentType;
           filename = replacePlaybackFileExtension(filename, playback.extension);
         }
@@ -738,39 +736,10 @@ export async function handleControlUiAssistantMediaRequest(
       ifNoneMatchHeader: req.headers["if-none-match"],
     });
     writeByteHeaders(res, byteResponse);
-    if (
-      req.method === "HEAD" ||
-      byteResponse.kind === "not-modified" ||
-      byteResponse.kind === "unsatisfiable" ||
-      opened.stat.size === 0
-    ) {
-      await closeOpenedHandle();
-      res.end();
-      return true;
-    }
-    const stream = opened.handle.createReadStream({
-      start: byteResponse.kind === "partial" ? byteResponse.range.start : 0,
-      end: byteResponse.kind === "partial" ? byteResponse.range.end : opened.stat.size - 1,
-      autoClose: false,
-    });
-    const finishClose = () => {
-      void closeOpenedHandle();
-    };
-    stream.once("end", finishClose);
-    stream.once("close", finishClose);
-    stream.once("error", () => {
-      void closeOpenedHandle();
-      if (!res.headersSent) {
-        respondControlUiNotFound(res);
-      } else {
-        res.destroy();
-      }
-    });
-    res.once("close", finishClose);
-    stream.pipe(res);
+    await byteStream.pipe(byteResponse, req.method);
     return true;
   } catch {
-    await closeOpenedHandle();
+    await byteStream?.close();
     respondControlUiNotFound(res);
     return true;
   }

src/gateway/http-byte-range.test.ts

@@ -1,6 +1,13 @@
-import type { ServerResponse } from "node:http";
+import fs from "node:fs/promises";
+import http, { type ServerResponse } from "node:http";
+import os from "node:os";
+import path from "node:path";
 import { describe, expect, it, vi } from "vitest";
-import { resolveByteResponse, writeByteHeaders } from "./http-byte-range.js";
+import {
+  createGatewayByteStream,
+  resolveByteResponse,
+  writeByteHeaders,
+} from "./http-byte-range.js";
 
 const FILE = { size: 10, mtimeMs: 1_752_000_000_123.5 };
 
@@ -154,3 +161,91 @@ describe("byte ETag generation", () => {
     expect(etag).toMatch(/^"[A-Za-z0-9_-]+"$/);
   });
 });
+
+describe("Gateway byte response descriptor lifecycle", () => {
+  it("destroys the real file stream and closes its descriptor once when its HTTP client disconnects", async () => {
+    const directory = await fs.mkdtemp(path.join(os.tmpdir(), "gateway-byte-stream-"));
+    const filePath = path.join(directory, "media.bin");
+    const body = Buffer.alloc(8 * 1024 * 1024, 7);
+    await fs.writeFile(filePath, body);
+    const handle = await fs.open(filePath, "r");
+    const closeHandle = vi.spyOn(handle, "close");
+    const createReadStream = vi.spyOn(handle, "createReadStream");
+    let resolveResponseClose!: () => void;
+    const responseClosed = new Promise<void>((resolve) => {
+      resolveResponseClose = resolve;
+    });
+    const server = http.createServer((_request, response) => {
+      const owner = createGatewayByteStream(response, handle, () => {
+        response.statusCode = 404;
+        response.end("not found");
+      });
+      const byteResponse = resolveByteResponse({
+        file: { size: body.byteLength, mtimeMs: 1 },
+        method: "GET",
+      });
+      writeByteHeaders(response, byteResponse);
+      void owner.pipe(byteResponse, "GET");
+      response.once("close", resolveResponseClose);
+    });
+    await new Promise<void>((resolve) => {
+      server.listen(0, "127.0.0.1", resolve);
+    });
+    const address = server.address();
+    if (!address || typeof address === "string") {
+      throw new Error("expected test HTTP server to bind to a TCP port");
+    }
+
+    try {
+      await new Promise<void>((resolve, reject) => {
+        const request = http.get({ host: "127.0.0.1", port: address.port }, (response) => {
+          response.once("data", () => {
+            response.destroy();
+            resolve();
+          });
+        });
+        request.once("error", reject);
+      });
+      await responseClosed;
+      await vi.waitFor(() => {
+        expect(closeHandle).toHaveBeenCalledOnce();
+        expect(handle.fd).toBe(-1);
+      });
+
+      const streamedFile = createReadStream.mock.results[0]?.value;
+      expect(streamedFile?.destroyed).toBe(true);
+      expect(streamedFile?.readableEnded).toBe(false);
+    } finally {
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      });
+      await fs.rm(directory, { recursive: true, force: true });
+    }
+  });
+
+  it("closes a newly opened descriptor when its response ended before streaming began", async () => {
+    const directory = await fs.mkdtemp(path.join(os.tmpdir(), "gateway-byte-ended-"));
+    const filePath = path.join(directory, "media.bin");
+    await fs.writeFile(filePath, "media");
+    const handle = await fs.open(filePath, "r");
+    const closeHandle = vi.spyOn(handle, "close");
+    const response = new http.ServerResponse({ method: "GET" } as http.IncomingMessage);
+    response.end();
+    const owner = createGatewayByteStream(response, handle, () => {});
+
+    try {
+      await owner.pipe(
+        resolveByteResponse({ file: { size: 5, mtimeMs: 1 }, method: "GET" }),
+        "GET",
+      );
+      expect(closeHandle).toHaveBeenCalledOnce();
+      expect(handle.fd).toBe(-1);
+    } finally {
+      if (handle.fd >= 0) {
+        await handle.close();
+      }
+      await fs.rm(directory, { recursive: true, force: true });
+    }
+  });
+});

src/gateway/http-byte-range.ts

@@ -1,4 +1,5 @@
 import { createHash } from "node:crypto";
+import type { FileHandle } from "node:fs/promises";
 import type { ServerResponse } from "node:http";
 import { matchesHttpIfNoneMatch } from "./http-conditional.js";
 
@@ -146,3 +147,60 @@ export function writeByteHeaders(res: ServerResponse, plan: ByteResponsePlan): v
     res.setHeader("Content-Range", `bytes */${plan.size}`);
   }
 }
+
+export function createGatewayByteStream(
+  res: ServerResponse,
+  handle: Pick<FileHandle, "close" | "createReadStream">,
+  onReadError: () => void,
+) {
+  let stream: ReturnType<FileHandle["createReadStream"]> | undefined;
+  let closed = false;
+  const close = async () => {
+    if (closed) {
+      return;
+    }
+    closed = true;
+    if (stream) {
+      stream.destroy();
+      return;
+    }
+    await handle.close().catch(() => {});
+  };
+  const release = () => {
+    void close();
+  };
+  // The ReadStream owns the FileHandle after creation; destroying it closes the descriptor once.
+  res.once("close", release);
+
+  return {
+    close,
+    async pipe(plan: ByteResponsePlan, method: string | undefined) {
+      if (method === "HEAD" || !("contentLength" in plan) || plan.contentLength === 0) {
+        await close();
+        res.end();
+        return;
+      }
+      if (closed || res.destroyed || res.writableEnded) {
+        await close();
+        return;
+      }
+      stream = handle.createReadStream({
+        start: plan.kind === "partial" ? plan.range.start : 0,
+        end: plan.kind === "partial" ? plan.range.end : plan.contentLength - 1,
+        autoClose: true,
+      });
+      stream.once("end", release).once("close", release);
+      stream.once("error", () => {
+        release();
+        if (!res.destroyed && !res.writableEnded) {
+          if (res.headersSent) {
+            res.destroy();
+          } else {
+            onReadError();
+          }
+        }
+      });
+      stream.pipe(res);
+    },
+  };
+}

src/gateway/managed-image-attachments.test.ts

@@ -6,7 +6,7 @@ import type { AddressInfo } from "node:net";
 import path from "node:path";
 import { pathToFileURL } from "node:url";
 import { maxBytesForKind } from "@openclaw/media-core/constants";
-import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from "vitest";
 import {
   createNoisyPngBuffer as createNoisyPngFixtureBuffer,
   createSolidPngBuffer,
@@ -21,6 +21,7 @@ import {
   MANAGED_OUTGOING_ORIGINALS_SUBDIR,
   readManagedImageRecord,
 } from "./managed-image-record-store.js";
+import { makeMockHttpResponse } from "./test-http-response.js";
 
 type PlaybackTranscodeResolution = Awaited<
   ReturnType<(typeof import("../media/playback-transcode.js"))["resolvePlaybackTranscode"]>
@@ -535,6 +536,62 @@ describe("handleManagedOutgoingImageHttpRequest", () => {
     expect(result.body).toEqual(body);
   });
 
+  it("closes the opened managed-media descriptor when playback resolution rejects", async () => {
+    const { attachmentId, sessionKey, originalPath } = await createFixture(stateDir, {
+      filename: "voice.caf",
+      contentType: "audio/x-caf",
+      body: Buffer.from("caff-original"),
+    });
+    authorizeGatewayHttpRequestOrReplyMock.mockResolvedValue({ ok: true, authMethod: "token" });
+    resolveOpenAiCompatibleHttpOperatorScopesMock.mockReturnValue(["operator.read"]);
+    resolveOpenAiCompatibleHttpSenderIsOwnerMock.mockReturnValue(true);
+    loadSessionEntryMock.mockReturnValue({
+      storePath: path.join(stateDir, "gateway-sessions.json"),
+      entry: { sessionId: "sess-1", sessionFile: "session.jsonl" },
+    });
+    resolveSessionHistoryTranscriptPathMock.mockResolvedValue("session.jsonl");
+    readSessionMessagesMock.mockResolvedValue([
+      {
+        role: "assistant",
+        content: [
+          {
+            type: "audio",
+            url: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full`,
+          },
+        ],
+        __openclaw: { id: "msg-1" },
+      },
+    ]);
+    resolvePlaybackTranscodeMock.mockRejectedValueOnce(new Error("playback inspection failed"));
+    const originalOpen = fs.open;
+    let closeOpenedHandle: MockInstance<() => Promise<void>> | undefined;
+    const openSpy = vi.spyOn(fs, "open").mockImplementation(async (...args) => {
+      const handle = await originalOpen(...args);
+      if (String(args[0]) === originalPath) {
+        closeOpenedHandle = vi.spyOn(handle, "close");
+      }
+      return handle;
+    });
+    const { res } = makeMockHttpResponse();
+
+    try {
+      await expect(
+        handleManagedOutgoingImageHttpRequest(
+          {
+            url: `/api/chat/media/outgoing/${encodeURIComponent(sessionKey)}/${attachmentId}/full?playback=1`,
+            method: "GET",
+            headers: {},
+          } as http.IncomingMessage,
+          res,
+          { auth: { mode: "test" } as never, stateDir },
+        ),
+      ).rejects.toThrow("playback inspection failed");
+      expect(closeOpenedHandle).toHaveBeenCalledOnce();
+    } finally {
+      openSpy.mockRestore();
+    }
+  });
+
   it("passes native managed playback bytes through unchanged", async () => {
     const body = Buffer.from("ID3-native-audio");
     const { attachmentId, sessionKey } = await createFixture(stateDir, {

src/gateway/managed-image-attachments.ts

@@ -34,7 +34,11 @@ import { safeEqualSecret } from "../security/secret-equal.js";
 import { buildAssistantMediaContentDisposition } from "./assistant-media-content-disposition.js";
 import type { AuthRateLimiter } from "./auth-rate-limit.js";
 import type { ResolvedGatewayAuth } from "./auth.js";
-import { resolveByteResponse, writeByteHeaders } from "./http-byte-range.js";
+import {
+  createGatewayByteStream,
+  resolveByteResponse,
+  writeByteHeaders,
+} from "./http-byte-range.js";
 import { sendJson, sendMethodNotAllowed, sendMissingScopeForbidden } from "./http-common.js";
 import {
   authorizeGatewayHttpRequestOrReply,
@@ -1459,6 +1463,8 @@ export async function handleManagedOutgoingMediaHttpRequest(
     sendStatus(res, 404, "not found");
     return true;
   }
+  const respondNotFound = () => sendStatus(res, 404, "not found");
+  let byteStream = createGatewayByteStream(res, opened.handle, respondNotFound);
 
   let responseContentType = record.original.contentType || "application/octet-stream";
   let responseFilename = record.original.filename;
@@ -1472,17 +1478,21 @@ export async function handleManagedOutgoingMediaHttpRequest(
       sourceStat: opened.stat,
       mimeType: responseContentType,
       kind: mediaKind,
+    }).catch(async (error: unknown) => {
+      await byteStream.close();
+      throw error;
     });
     if (playback.kind === "preparing") {
-      await opened.handle.close().catch(() => {});
+      await byteStream.close();
       sendJson(res, 202, { status: "preparing" });
       return true;
     }
     if (playback.kind === "transcoded") {
       const transcoded = await openLocalFileSafely({ filePath: playback.path }).catch(() => null);
       if (transcoded) {
-        await opened.handle.close().catch(() => {});
+        await byteStream.close();
         opened = transcoded;
+        byteStream = createGatewayByteStream(res, opened.handle, respondNotFound);
         responseContentType = playback.contentType;
         responseFilename = replacePlaybackFileExtension(
           responseFilename ?? "generated-media",
@@ -1492,14 +1502,6 @@ export async function handleManagedOutgoingMediaHttpRequest(
     }
   }
 
-  let handleClosed = false;
-  const closeOpenedHandle = async () => {
-    if (handleClosed) {
-      return;
-    }
-    handleClosed = true;
-    await opened.handle.close().catch(() => {});
-  };
   res.setHeader("content-type", responseContentType);
   res.setHeader("x-content-type-options", "nosniff");
   res.setHeader("referrer-policy", "no-referrer");
@@ -1521,38 +1523,8 @@ export async function handleManagedOutgoingMediaHttpRequest(
     ifNoneMatchHeader: req.headers["if-none-match"],
   });
   writeByteHeaders(res, byteResponse);
-  if (
-    req.method === "HEAD" ||
-    byteResponse.kind === "not-modified" ||
-    byteResponse.kind === "unsatisfiable" ||
-    opened.stat.size === 0
-  ) {
-    await closeOpenedHandle();
-    res.end();
-    return true;
-  }
-
   // Stream from the verified descriptor so a path swap cannot bypass fs-safe after validation.
-  const stream = opened.handle.createReadStream({
-    start: byteResponse.kind === "partial" ? byteResponse.range.start : 0,
-    end: byteResponse.kind === "partial" ? byteResponse.range.end : opened.stat.size - 1,
-    autoClose: false,
-  });
-  const finishClose = () => {
-    void closeOpenedHandle();
-  };
-  stream.once("end", finishClose);
-  stream.once("close", finishClose);
-  stream.once("error", () => {
-    void closeOpenedHandle();
-    if (!res.headersSent) {
-      sendStatus(res, 404, "not found");
-    } else {
-      res.destroy();
-    }
-  });
-  res.once("close", finishClose);
-  stream.pipe(res);
+  await byteStream.pipe(byteResponse, req.method);
   return true;
 }
 /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */