fix(server): bound Claude Desktop apply bodies

src/server/management/agent-settings-routes.ts

@@ -60,7 +60,7 @@ import { applySystemEnvToggle } from "../system-env";
 
 import { isPlainRecord, parseDebugLogQuery, tokPerSecondResult, unavailableCostReason, costResult, requestLogDto, stripRegistryOnlyStaticHeaders, fetchAllModels, fetchGrokCandidateModels, buildClaudeDesktopState } from "./shared";
 import type { MetricUnavailableReason, TokPerSecondResult, CostEstimateReason, CostResult, MetricSource } from "./shared";
-import { readManagementJsonBody, rethrowManagementBodyTooLarge } from "./body";
+import { readManagementJsonBody, readOptionalManagementJsonBody, rethrowManagementBodyTooLarge } from "./body";
 
 const GROK_APPLY_JOIN_MS = 120_000;
 export const GROK_APPLY_TERMINAL_MS = 10 * 60_000;
@@ -759,32 +759,22 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise
   }
   if (url.pathname === "/api/claude-desktop/apply" && req.method === "POST") {
     try {
-      const { setIntegrationEnabled, claudeDesktopIntegrationEnabled } = await import("../../codex/desired-state");
-      const desired = setIntegrationEnabled("claude-desktop", true);
-      if (!desired.ok) return jsonResponse({ error: desired.message }, desired.retryable ? 409 : 500);
-      // Disk now says ON; the reused server snapshot must agree, or the native
-      // GET reports OFF and a later whole-snapshot save undoes this transition.
-      mirrorDesiredEnabledOntoSnapshot(config, "claude-desktop", true);
-      // Disk now says ON; the reused server snapshot must agree, or the native
-      // GET reports OFF and a later whole-snapshot save undoes this transition.
       // #859: the CLI delegates here so the registry is built in the serving
       // process. Accept an optional mode; default stays static for back-compat.
       let mode: "static" | "hybrid" | "discovery" = "static";
-      const rawBody = await req.text();
       let parsed: unknown;
-      if (rawBody.trim()) {
-        try {
-          parsed = JSON.parse(rawBody);
-        } catch {
-          return jsonResponse({ error: "invalid JSON body" }, 400);
-        }
-        const requested = (parsed as { mode?: unknown } | null)?.mode;
-        if (requested !== undefined) {
-          if (requested === "static" || requested === "hybrid" || requested === "discovery") {
-            mode = requested;
-          } else {
-            return jsonResponse({ error: "mode must be static, hybrid, or discovery" }, 400);
-          }
+      try {
+        parsed = await readOptionalManagementJsonBody(req);
+      } catch (error) {
+        rethrowManagementBodyTooLarge(error);
+        return jsonResponse({ error: "invalid JSON body" }, 400);
+      }
+      const requested = (parsed as { mode?: unknown } | null)?.mode;
+      if (requested !== undefined) {
+        if (requested === "static" || requested === "hybrid" || requested === "discovery") {
+          mode = requested;
+        } else {
+          return jsonResponse({ error: "mode must be static, hybrid, or discovery" }, 400);
         }
       }
       // #859: a delegated CLI apply carries the profile it just saved — the
@@ -800,6 +790,12 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise
           return jsonResponse({ error: error instanceof Error ? error.message : String(error) }, 400);
         }
       }
+      const { setIntegrationEnabled, claudeDesktopIntegrationEnabled } = await import("../../codex/desired-state");
+      const desired = setIntegrationEnabled("claude-desktop", true);
+      if (!desired.ok) return jsonResponse({ error: desired.message }, desired.retryable ? 409 : 500);
+      // Disk now says ON; the reused server snapshot must agree, or the native
+      // GET reports OFF and a later whole-snapshot save undoes this transition.
+      mirrorDesiredEnabledOntoSnapshot(config, "claude-desktop", true);
       const state = await buildClaudeDesktopState(config, profileOverride);
       // `setIntegrationEnabled` above wrote desired ON to DISK; it does not touch
       // this long-lived server snapshot. Saving the snapshot wholesale would carry
@@ -866,6 +862,7 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise
       }
       return jsonResponse({ ok: true, saved: true, applied: true, path: result.path, fingerprint: result.fingerprint });
     } catch (error) {
+      rethrowManagementBodyTooLarge(error);
       return jsonResponse({ error: error instanceof Error ? error.message : String(error) }, 400);
     }
   }

src/server/management/body.ts

@@ -11,6 +11,12 @@ export function readManagementJsonBody<T = unknown>(req: Request): Promise<T> {
   return readBoundedJsonRequestBody(req, MANAGEMENT_JSON_BODY_MAX_BYTES) as Promise<T>;
 }
 
+export function readOptionalManagementJsonBody<T = unknown>(req: Request): Promise<T> {
+  return readBoundedJsonRequestBody(req, MANAGEMENT_JSON_BODY_MAX_BYTES, undefined, {
+    emptyBodyFallback: {},
+  }) as Promise<T>;
+}
+
 export function managementBodyTooLargeResponse(
   error: unknown,
   req: Request,

src/server/request-decompress.ts

@@ -45,6 +45,101 @@ function declaredBodyLength(req: Request): number | null {
   return Number.isFinite(length) && length >= 0 ? length : null;
 }
 
+function cancelStreamWithoutWaiting(stream: ReadableStream<Uint8Array> | null, reason: unknown): void {
+  if (!stream || stream.locked) return;
+  try {
+    void stream.cancel(reason).catch(() => undefined);
+  } catch {
+    // A non-standard stream may throw synchronously from cancel().
+  }
+}
+
+function cancelReaderWithoutWaiting(reader: ReadableStreamDefaultReader<Uint8Array>, reason: unknown): void {
+  // Request.clone() tees can leave cancel() pending until the other branch
+  // drains. Cancellation must never extend this reader's own admission bound.
+  try {
+    void reader.cancel(reason).catch(() => undefined);
+  } catch {
+    // A non-standard reader may throw synchronously from cancel().
+  }
+}
+
+async function readRequestBodyBytesCapped(
+  body: ReadableStream<Uint8Array> | null,
+  maxBytes: number,
+  signal?: AbortSignal,
+): Promise<Uint8Array> {
+  if (signal?.aborted) {
+    cancelStreamWithoutWaiting(body, signal.reason);
+    throw signal.reason;
+  }
+  if (!body) return new Uint8Array(0);
+
+  const reader = body.getReader();
+  // Keep one geometric buffer instead of one object per transport chunk. A
+  // hostile peer can fragment a bounded payload into arbitrarily many chunks.
+  let retained = new Uint8Array(Math.min(maxBytes, 64 * 1024));
+  let retainedBytes = 0;
+  let aborted = false;
+  let abortReason: unknown;
+  let cancellationStarted = false;
+  const cancel = (reason: unknown): void => {
+    if (cancellationStarted) return;
+    cancellationStarted = true;
+    cancelReaderWithoutWaiting(reader, reason);
+  };
+  const onAbort = (): void => {
+    aborted = true;
+    abortReason = signal?.reason;
+    cancel(abortReason);
+  };
+  signal?.addEventListener("abort", onAbort, { once: true });
+  // Close the narrow race between the preflight check and listener install.
+  if (signal?.aborted) onAbort();
+
+  try {
+    while (true) {
+      if (aborted) throw abortReason;
+      const { value, done } = await reader.read();
+      // cancel() can resolve a pending read as EOF. Preserve the caller's
+      // original abort reason instead of misclassifying that as a clean body.
+      if (aborted) throw abortReason;
+      if (done) {
+        return retainedBytes === retained.byteLength
+          ? retained
+          : retained.slice(0, retainedBytes);
+      }
+      if (!value || value.byteLength === 0) continue;
+
+      if (value.byteLength > maxBytes - retainedBytes) {
+        const error = new DecompressedBodyTooLargeError(retainedBytes + value.byteLength, maxBytes);
+        cancel(error);
+        throw error;
+      }
+
+      const required = retainedBytes + value.byteLength;
+      if (required > retained.byteLength) {
+        const grown = new Uint8Array(Math.min(maxBytes, Math.max(retained.byteLength * 2, required)));
+        grown.set(retained.subarray(0, retainedBytes));
+        retained = grown;
+      }
+      retained.set(value, retainedBytes);
+      retainedBytes = required;
+    }
+  } catch (error) {
+    const failure = aborted ? abortReason : error;
+    cancel(failure);
+    throw failure;
+  } finally {
+    signal?.removeEventListener("abort", onAbort);
+    try {
+      reader.releaseLock();
+    } catch {
+      // A pending cancellation can retain the lock briefly; never await it.
+    }
+  }
+}
+
 function inflateDeflateBody(compressed: Uint8Array<ArrayBuffer>, opts: { maxOutputLength: number }): Uint8Array {
   // HTTP "deflate" appears both zlib-wrapped and raw in the wild (Bun.deflateSync emits raw,
   // which the previous Bun.inflateSync accepted). Try zlib-wrapped first, fall back to raw —
@@ -90,24 +185,27 @@ export async function readBoundedJsonRequestBody(
   req: Request,
   maxBytes: number,
   budget?: TranslatorBudget,
+  options?: { emptyBodyFallback?: unknown },
 ): Promise<unknown> {
   const encoding = req.headers.get("content-encoding");
   const declaredLength = declaredBodyLength(req);
-  // Reject an honest oversized declaration before req.arrayBuffer() can allocate it.
-  // Missing, malformed, or dishonest declarations remain covered by decodeRequestBody's
-  // post-read cap below.
+  // Reject an honest oversized declaration before reading. Missing, malformed,
+  // and dishonest declarations remain bounded by the streaming reader below.
   if (declaredLength !== null && declaredLength > maxBytes) {
-    throw new DecompressedBodyTooLargeError(declaredLength, maxBytes);
+    const error = new DecompressedBodyTooLargeError(declaredLength, maxBytes);
+    cancelStreamWithoutWaiting(req.body, error);
+    throw error;
   }
   const releaseReservation = budget && declaredLength !== null && declaredLength > 0
     ? budget.observeAcceptedRequestCopy(declaredLength)
     : undefined;
   let raw: Uint8Array;
   try {
-    raw = new Uint8Array(await req.arrayBuffer());
+    raw = await readRequestBodyBytesCapped(req.body, maxBytes, req.signal);
   } finally {
     releaseReservation?.();
   }
+  assertBodySizeWithinLimit(raw, maxBytes);
   const releaseRaw = budget?.observeAcceptedRequestCopy(raw.byteLength);
   let releaseDecoded: (() => void) | undefined;
   let releaseText: (() => void) | undefined;
@@ -116,6 +214,9 @@ export async function readBoundedJsonRequestBody(
     releaseDecoded = decoded === raw ? undefined : budget?.observeAcceptedRequestCopy(decoded.byteLength);
     const text = new TextDecoder().decode(decoded);
     releaseText = budget?.observeAcceptedRequestCopy(new TextEncoder().encode(text).byteLength);
+    if (options && "emptyBodyFallback" in options && text.trim() === "") {
+      return options.emptyBodyFallback;
+    }
     const parsed = JSON.parse(text);
     budget?.observeAcceptedRequestCopy(new TextEncoder().encode(JSON.stringify(parsed)).byteLength);
     return parsed;

tests/claude-management-api.test.ts

@@ -8,6 +8,7 @@ import { startServer } from "../src/server";
 import * as systemEnv from "../src/server/system-env";
 import type { OcxConfig } from "../src/types";
 import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home";
+import { MANAGEMENT_JSON_BODY_MAX_BYTES } from "../src/server/management/body";
 
 // Full-suite Windows load: startServer + multi-PUT management flows often exceed bun's
 // default 5s per-test budget (same flake class as 810fa115 / kiro-oauth).
@@ -690,12 +691,33 @@ test("Claude Desktop apply honors the profile in the request body over daemon-st
 test("Claude Desktop apply validates the mode body", async () => {
   const server = startServer(0);
   try {
+    const beforeMalformed = structuredClone(loadConfig());
+    const malformed = await fetch(new URL("/api/claude-desktop/apply", server.url), {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: "{",
+    });
+    expect(malformed.status).toBe(400);
+    expect(await malformed.json()).toEqual({ error: "invalid JSON body" });
+    expect(loadConfig()).toEqual(beforeMalformed);
+
+    const beforeBadMode = structuredClone(loadConfig());
     const bad = await fetch(new URL("/api/claude-desktop/apply", server.url), {
       method: "POST",
       headers: { "Content-Type": "application/json" },
       body: JSON.stringify({ mode: "nonsense" }),
     });
     expect(bad.status).toBe(400);
+    expect(loadConfig()).toEqual(beforeBadMode);
+
+    const beforeBadProfile = structuredClone(loadConfig());
+    const badProfile = await fetch(new URL("/api/claude-desktop/apply", server.url), {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ profile: { version: 2 } }),
+    });
+    expect(badProfile.status).toBe(400);
+    expect(loadConfig()).toEqual(beforeBadProfile);
 
     const hybrid = await fetch(new URL("/api/claude-desktop/apply", server.url), {
       method: "POST",
@@ -711,6 +733,24 @@ test("Claude Desktop apply validates the mode body", async () => {
   }
 });
 
+test("Claude Desktop apply rejects an oversized decompressed body without mutating config", async () => {
+  const server = startServer(0);
+  try {
+    const before = structuredClone(loadConfig());
+    const oversized = JSON.stringify({ pad: "x".repeat(MANAGEMENT_JSON_BODY_MAX_BYTES) });
+    const response = await fetch(new URL("/api/claude-desktop/apply", server.url), {
+      method: "POST",
+      headers: { "Content-Type": "application/json", "Content-Encoding": "gzip" },
+      body: Bun.gzipSync(new TextEncoder().encode(oversized)),
+    });
+    expect(response.status).toBe(413);
+    expect(await response.json()).toEqual({ error: "request body too large" });
+    expect(loadConfig()).toEqual(before);
+  } finally {
+    await server.stop(true);
+  }
+});
+
 test("Claude Desktop PUT rejects invalid JSON profile without mutating saved config", async () => {
   const server = startServer(0);
   try {

tests/native-claude-desktop-toggle.test.ts

@@ -4,6 +4,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { handleManagementAPI } from "../src/server/management-api";
 import { setIntegrationEnabled } from "../src/codex/desired-state";
+import { MANAGEMENT_JSON_BODY_MAX_BYTES } from "../src/server/management/body";
 import type { ManagementApiDeps } from "../src/server/management/context";
 import type { OcxConfig } from "../src/types";
 
@@ -42,6 +43,35 @@ async function toggle(enabled: boolean, deps: ManagementApiDeps = {}) {
   return { status: response!.status, body: await response!.json() as Record<string, unknown> };
 }
 
+function oversizedTrackedBody(): {
+  body: ReadableStream<Uint8Array>;
+  stats: { pulls: number; cancelled: number; sentinelPulled: boolean };
+} {
+  const sentinel = Uint8Array.of(0x7f);
+  const chunks = [
+    new Uint8Array(MANAGEMENT_JSON_BODY_MAX_BYTES / 2),
+    new Uint8Array(MANAGEMENT_JSON_BODY_MAX_BYTES / 2 + 1),
+    sentinel,
+  ];
+  const stats = { pulls: 0, cancelled: 0, sentinelPulled: false };
+  const body = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      stats.pulls += 1;
+      const chunk = chunks.shift();
+      if (!chunk) {
+        controller.close();
+        return;
+      }
+      if (chunk === sentinel) stats.sentinelPulled = true;
+      controller.enqueue(chunk);
+    },
+    cancel() {
+      stats.cancelled += 1;
+    },
+  }, { highWaterMark: 0 });
+  return { body, stats };
+}
+
 beforeEach(() => {
   root = mkdtempSync(join(tmpdir(), "ocx-desktop-toggle-"));
   library = join(root, "desktop-library");
@@ -214,6 +244,40 @@ test("POST /apply enables from a stale OFF server snapshot instead of cancelling
   expect(persistedIntent()).toBeUndefined();
 });
 
+for (const [label, declaration] of [
+  ["missing Content-Length", undefined],
+  ["a lying low Content-Length", "1"],
+] as const) {
+  test(`POST /apply stops an oversized stream with ${label} before any mutation`, async () => {
+    const inputConfig = config();
+    const beforeInputConfig = structuredClone(inputConfig);
+    const beforePersistedConfig = readFileSync(join(root, "config.json"), "utf8");
+    const { body, stats } = oversizedTrackedBody();
+    let writes = 0;
+    const headers: Record<string, string> = { "Content-Type": "application/json" };
+    if (declaration !== undefined) headers["Content-Length"] = declaration;
+
+    const response = await dispatch("/api/claude-desktop/apply", {
+      method: "POST",
+      headers,
+      body,
+    }, {
+      writeDesktop3pConfig: () => {
+        writes += 1;
+        return { written: true, path: join(library, "unexpected.json"), fingerprint: "unexpected" };
+      },
+    }, inputConfig);
+
+    expect(response!.status).toBe(413);
+    expect(await response!.json()).toEqual({ error: "request body too large" });
+    expect(stats).toEqual({ pulls: 2, cancelled: 1, sentinelPulled: false });
+    expect(writes).toBe(0);
+    expect(inputConfig).toEqual(beforeInputConfig);
+    expect(readFileSync(join(root, "config.json"), "utf8")).toBe(beforePersistedConfig);
+    expect(existsSync(library)).toBe(false);
+  });
+}
+
 test("POST /apply leaves the reused server snapshot agreeing with disk", async () => {
   // Disk-only repair is not enough: the server reuses ONE config object per
   // request, so a stale snapshot makes the native GET report the opposite of