fix(proxy): prevent shared mutation and replay memory retention

src/app/api/v1/resources/providers/handlers.ts

@@ -1,5 +1,4 @@
 import type { Context } from "hono";
-import type { ZodError } from "zod";
 import { z } from "zod";
 import type { ActionResult } from "@/actions/types";
 import { hasLegacyRedactedWritePlaceholders } from "@/lib/api/legacy-action-sanitizers";
@@ -773,11 +772,10 @@ function providerNotFound(c: Context): Response {
   });
 }
 
-type JsonBodySchema<T> = {
-  safeParse: (value: unknown) => { success: true; data: T } | { success: false; error: ZodError };
-};
-
-async function parseJson<T>(c: Context, schema: JsonBodySchema<T>): Promise<T | Response> {
+async function parseJson<S extends z.ZodType>(
+  c: Context,
+  schema: S
+): Promise<z.output<S> | Response> {
   const body = await parseHonoJsonBody(c, schema);
   if (!body.ok) return body.response;
   return body.data;

src/app/v1/_lib/proxy/billing-header-rectifier.ts

@@ -20,7 +20,7 @@ const BILLING_HEADER_PATTERN = /^\s*x-anthropic-billing-header\s*:/i;
 
 /**
  * Remove x-anthropic-billing-header text blocks from the request system prompt.
- * Mutates the message object in place (matches existing rectifier conventions).
+ * Writes changes back through the top-level message object without mutating shared nested arrays.
  */
 export function rectifyBillingHeader(
   message: Record<string, unknown>
@@ -62,11 +62,7 @@ export function rectifyBillingHeader(
     }
 
     if (extractedValues.length > 0) {
-      // Mutate in place: replace system array contents
-      system.length = 0;
-      for (const item of filtered) {
-        system.push(item);
-      }
+      message.system = filtered;
       return { applied: true, removedCount: extractedValues.length, extractedValues };
     }
 

src/app/v1/_lib/proxy/forwarder.ts

@@ -543,17 +543,23 @@ export function applyCacheTtlOverrideToMessage(
   // messages[].content[]
   const messages = message.messages;
   if (Array.isArray(messages)) {
-    for (const msg of messages) {
+    let nextMessages: unknown[] | null = null;
+    for (let index = 0; index < messages.length; index += 1) {
+      const msg = messages[index];
       if (!msg || typeof msg !== "object") continue;
       const msgObj = msg as Record<string, unknown>;
       const content = msgObj.content;
       if (!Array.isArray(content)) continue;
       const result = applyTtlToContentBlocks(content, ttl);
       if (result.applied) {
-        msgObj.content = result.blocks;
+        nextMessages ??= [...messages];
+        nextMessages[index] = { ...msgObj, content: result.blocks };
         applied = true;
       }
     }
+    if (nextMessages) {
+      message.messages = nextMessages;
+    }
   }
 
   return applied;
@@ -1166,7 +1172,11 @@ async function tryApplyReactiveRectifier(params: {
     }
 
     const requestDetailsBeforeRectify = buildRequestDetails(requestSession);
-    const rectified = descriptor.rectify(requestSession.request.message as Record<string, unknown>);
+    const mutableMessage = structuredClone(
+      requestSession.request.message as Record<string, unknown>
+    );
+    requestSession.request.message = mutableMessage;
+    const rectified = descriptor.rectify(mutableMessage);
 
     addSpecialSettingForPersistence(
       requestSession,
@@ -3325,13 +3335,7 @@ export class ProxyForwarder {
           const bodyString = JSON.stringify(messageToSend);
           requestBody = bodyString;
           session.forwardedRequestBody = bodyString;
-
-          try {
-            const parsed = JSON.parse(bodyString);
-            isStreaming = parsed.stream === true;
-          } catch {
-            isStreaming = false;
-          }
+          isStreaming = messageToSend.stream === true;
 
           if (process.env.NODE_ENV === "development") {
             logger.trace("ProxyForwarder: Forwarding request", {
@@ -7770,8 +7774,10 @@ export class ProxyForwarder {
 
     shadowState.request = {
       ...session.request,
-      message: structuredClone(session.request.message),
-      buffer: session.request.buffer ? session.request.buffer.slice(0) : undefined,
+      // attempt 改写采用顶层 copy-on-write；发送前的私有参数过滤会生成独立深拷贝。
+      message: { ...session.request.message },
+      // 原始请求字节只读；shadow 共享底层 buffer，任何改写都必须整体替换属性。
+      buffer: session.request.buffer,
       imageRequestMetadata: cloneOpenAIImageRequestMetadata(session.request.imageRequestMetadata),
     };
     shadow.requestUrl = new URL(session.requestUrl.toString());

src/app/v1/_lib/proxy/replay/replay-spool.ts

@@ -40,6 +40,7 @@ export class ReplaySpool {
   private readonly store = getReplayStore();
   private readonly decoder = new TextDecoder("utf-8");
   private readonly parts: string[] = [];
+  private readonly queuedBatches = new Set<string[]>();
   private pending: string[] = [];
   private pendingBytes = 0;
   private totalBytes = 0;
@@ -120,19 +121,20 @@ export class ReplaySpool {
     if (batch.length === 0) return;
     this.pending = [];
     this.pendingBytes = 0;
+    this.queuedBatches.add(batch);
     // 续接体自带 try/catch：链永不 rejected；每个 await 之后复查 disabled，
     // 防止与 disable/halt 竞态时在清理之后又写回 owning meta
     this.writeChain = this.writeChain.then(async () => {
       try {
-        if (this.disabled) return;
+        if (this.disabled || this.aborting) return;
         const expectedChunkCount = this.chunkCount + batch.length;
         const appended = await this.store.writeOwned(
           this.identity.replayId,
           this.ownerToken,
           this.buildMeta("owning", { chunkCount: expectedChunkCount }),
           batch
         );
-        if (this.disabled) return;
+        if (this.disabled || this.aborting) return;
         if (appended === null) {
           // Redis 不可用：本次 replay 放弃（热层写是原子的，不会留下半批数据）
           this.disable("redis_unavailable");
@@ -145,10 +147,13 @@ export class ReplaySpool {
         this.chunkCount = appended;
         this.metaWritten = true;
       } catch (error) {
+        if (this.aborting) return;
         logger.debug("[ReplaySpool] flush failed, disabling spool", {
           error: error instanceof Error ? error.message : String(error),
         });
         this.disable("flush_error");
+      } finally {
+        this.queuedBatches.delete(batch);
       }
     });
   }
@@ -172,15 +177,15 @@ export class ReplaySpool {
 
   private startOwnerHeartbeat(): void {
     this.ownerHeartbeatTimer = setInterval(() => {
-      if (this.disabled || this.released || this.ownerHeartbeatInFlight) return;
+      if (this.disabled || this.aborting || this.released || this.ownerHeartbeatInFlight) return;
       this.ownerHeartbeatInFlight = true;
       void this.store
         .renewOwnerLease(this.identity.replayId, this.ownerToken)
         .then((leaseHeld) => {
-          if (!leaseHeld && !this.released) this.halt("owner_lease_lost");
+          if (!leaseHeld && !this.aborting && !this.released) this.halt("owner_lease_lost");
         })
         .catch(() => {
-          if (!this.released) this.halt("owner_lease_lost");
+          if (!this.aborting && !this.released) this.halt("owner_lease_lost");
         })
         .finally(() => {
           this.ownerHeartbeatInFlight = false;
@@ -193,12 +198,13 @@ export class ReplaySpool {
   bootstrap(): void {
     this.writeChain = this.writeChain.then(async () => {
       try {
-        if (this.disabled || this.metaWritten) return;
+        if (this.disabled || this.aborting || this.metaWritten) return;
         const chunkCount = await this.store.writeOwned(
           this.identity.replayId,
           this.ownerToken,
           this.buildMeta("owning")
         );
+        if (this.disabled || this.aborting) return;
         if (chunkCount === null) {
           this.disable("redis_unavailable");
           return;
@@ -210,6 +216,7 @@ export class ReplaySpool {
         this.chunkCount = chunkCount;
         this.metaWritten = true;
       } catch (error) {
+        if (this.aborting) return;
         logger.debug("[ReplaySpool] bootstrap failed, disabling spool", {
           error: error instanceof Error ? error.message : String(error),
         });
@@ -234,11 +241,12 @@ export class ReplaySpool {
     const batch = this.pending;
     this.pending = [];
     this.pendingBytes = 0;
+    this.queuedBatches.add(batch);
 
     this.writeChain = this.writeChain.then(async () => {
       let pgPersisted = false;
       try {
-        if (this.disabled) return;
+        if (this.disabled || this.aborting) return;
         const expectedChunkCount = this.chunkCount + batch.length;
         const appended = await this.store.writeOwned(
           this.identity.replayId,
@@ -255,6 +263,7 @@ export class ReplaySpool {
         }
         this.chunkCount = appended;
         this.metaWritten = true;
+        const payload = this.takePayload();
         // 先写 PG（持久 payload），再翻 Redis meta 为 completed（热层可服务）
         const persistResult = await this.store.persistCompleted({
           replayId: this.identity.replayId,
@@ -266,7 +275,7 @@ export class ReplaySpool {
           model: this.identity.model,
           statusCode: this.statusCode,
           headers: this.headers,
-          payload: this.parts.join(""),
+          payload,
           byteSize: this.totalBytes,
           sourceMessageRequestId: messageRequestId,
         });
@@ -312,6 +321,8 @@ export class ReplaySpool {
           )
           .catch(() => false);
       } finally {
+        this.queuedBatches.delete(batch);
+        this.clearPayload();
         this.release();
       }
     });
@@ -320,12 +331,19 @@ export class ReplaySpool {
 
   /** 终态失败：meta 置 aborted + 删块；已 aborted 的条目绝不被重放命中。 */
   async abort(reason: string): Promise<void> {
+    if (this.abortPromise) {
+      await this.abortPromise;
+      return;
+    }
     if (this.terminal) return;
     this.terminal = true;
+    this.aborting = true;
     this.clearTimer();
     this.pending = [];
     this.pendingBytes = 0;
-    this.writeChain = this.writeChain.then(async () => {
+    this.clearPayload();
+    this.clearQueuedBatches();
+    this.abortPromise = this.writeChain.then(async () => {
       try {
         // 已失效（disable 已清理 / halt 已让渡所有权）：不得再写 meta 覆盖新 owner
         if (this.disabled) return;
@@ -340,7 +358,8 @@ export class ReplaySpool {
         this.release();
       }
     });
-    await this.writeChain;
+    this.writeChain = this.abortPromise;
+    await this.abortPromise;
   }
 
   /** 失效并删除条目（payload 超限 / Redis 不可用 / 冲刷异常等本 spool 自身的失败）。 */
@@ -360,31 +379,37 @@ export class ReplaySpool {
     this.pending = [];
     this.parts.length = 0;
     this.pendingBytes = 0;
+    this.clearQueuedBatches();
     // 清理顺着 writeChain 串行：与 in-flight append 竞态时绝不出现「删除后又写回」
     this.writeChain = this.writeChain.then(async () => {
-      if (deleteEntry) {
-        await this.store
-          .abortOwned(
-            this.identity.replayId,
-            this.ownerToken,
-            this.buildMeta("aborted", { abortReason: reason })
-          )
-          .catch(() => false);
-      } else {
-        // compare-delete 只删自己的 token：所有权已失时为安全 no-op
-        await this.store
-          .releaseOwner(this.identity.replayId, this.ownerToken)
-          .catch(() => undefined);
+      try {
+        if (deleteEntry) {
+          await this.store
+            .abortOwned(
+              this.identity.replayId,
+              this.ownerToken,
+              this.buildMeta("aborted", { abortReason: reason })
+            )
+            .catch(() => false);
+        } else {
+          // compare-delete 只删自己的 token：所有权已失时为安全 no-op
+          await this.store
+            .releaseOwner(this.identity.replayId, this.ownerToken)
+            .catch(() => undefined);
+        }
+      } finally {
+        this.release();
       }
     });
     logger.debug("[ReplaySpool] spool disabled", {
       replayId: this.identity.replayId.slice(0, 12),
       reason,
     });
-    this.release();
   }
 
   private released = false;
+  private aborting = false;
+  private abortPromise: Promise<void> | null = null;
 
   private release(): void {
     if (this.released) return;
@@ -409,6 +434,21 @@ export class ReplaySpool {
     this.clearFlushTimer();
     this.clearOwnerHeartbeat();
   }
+
+  private takePayload(): string {
+    const payload = this.parts.join("");
+    this.clearPayload();
+    return payload;
+  }
+
+  private clearPayload(): void {
+    this.parts.length = 0;
+  }
+
+  private clearQueuedBatches(): void {
+    for (const batch of this.queuedBatches) batch.length = 0;
+    this.queuedBatches.clear();
+  }
 }
 
 /** Forwarder 在 spool 创建前终止时，以 owner token 原子封死 Replay 条目。 */

src/lib/api/v1/_shared/request-body.ts

@@ -3,10 +3,6 @@ import { createProblemResponse, normalizeZodPath } from "./error-envelope";
 
 export type ParsedBodyResult<T> = { ok: true; data: T } | { ok: false; response: Response };
 
-type JsonBodySchema<T> = {
-  safeParse: (value: unknown) => { success: true; data: T } | { success: false; error: z.ZodError };
-};
-
 type ParseJsonBodyOptions = {
   validationErrorCode?: (error: z.ZodError) => string | undefined;
 };
@@ -20,10 +16,10 @@ type HonoJsonRequest = {
   };
 };
 
-export async function parseJsonBody<T>(
+export async function parseJsonBody<S extends z.ZodType>(
   request: Request,
-  schema: JsonBodySchema<T>
-): Promise<ParsedBodyResult<T>> {
+  schema: S
+): Promise<ParsedBodyResult<z.output<S>>> {
   const contentType = request.headers.get("content-type") ?? "";
   if (!contentType.toLowerCase().includes("application/json")) {
     return {
@@ -73,11 +69,11 @@ export async function parseJsonBody<T>(
   return { ok: true, data: parsed.data };
 }
 
-export async function parseHonoJsonBody<T>(
+export async function parseHonoJsonBody<S extends z.ZodType>(
   c: HonoJsonRequest,
-  schema: JsonBodySchema<T>,
+  schema: S,
   options?: ParseJsonBodyOptions
-): Promise<ParsedBodyResult<T>> {
+): Promise<ParsedBodyResult<z.output<S>>> {
   const contentType =
     c.req.header("content-type") ??
     c.req.header("Content-Type") ??

src/lib/api/v1/schemas/audit-logs.ts

@@ -23,7 +23,9 @@ export const AuditLogListQuerySchema = z.object({
   success: z
     .enum(["true", "false"])
     .optional()
-    .transform((val) => (val === undefined ? undefined : val === "true"))
+    .transform((val: "true" | "false" | undefined) =>
+      val === undefined ? undefined : val === "true"
+    )
     .describe("Optional success filter."),
   from: IsoDateTimeStringSchema.optional().describe("Optional inclusive start time."),
   to: IsoDateTimeStringSchema.optional().describe("Optional inclusive end time."),