fix(server): stop an oversized provider diff from exhausting the backend heap

apps/server/src/provider/Layers/CodexAdapter.test.ts

@@ -1233,6 +1233,45 @@ lifecycleLayer("CodexAdapterLive lifecycle", (it) => {
     }),
   );
 
+  it.effect("carries a turn diff once instead of mirroring it into the raw payload", () =>
+    Effect.gen(function* () {
+      const { adapter, runtime } = yield* startLifecycleRuntime();
+      const firstEventFiber = yield* Stream.runHead(adapter.streamEvents).pipe(Effect.forkChild);
+
+      const diff = `diff --git a/file.ts b/file.ts
++${"x".repeat(4_096)}
+`;
+      yield* runtime.emit({
+        id: asEventId("evt-turn-diff"),
+        kind: "notification",
+        provider: ProviderDriverKind.make("codex"),
+        createdAt: "2026-01-01T00:00:00.000Z",
+        method: "turn/diff/updated",
+        threadId: asThreadId("thread-1"),
+        turnId: asTurnId("turn-1"),
+        payload: { threadId: "thread-1", turnId: "turn-1", diff },
+      });
+
+      const firstEvent = yield* Fiber.join(firstEventFiber);
+      NodeAssert.equal(firstEvent._tag, "Some");
+      if (firstEvent._tag !== "Some") {
+        return;
+      }
+      const mapped = firstEvent.value;
+      NodeAssert.equal(mapped.type, "turn.diff.updated");
+      if (mapped.type !== "turn.diff.updated") {
+        return;
+      }
+      NodeAssert.equal(mapped.payload.unifiedDiff, diff);
+
+      const rawPayload = mapped.raw?.payload as { readonly diff?: unknown } | undefined;
+      NodeAssert.equal(
+        rawPayload?.diff,
+        `[omitted by t3, ${diff.length} characters in payload.unifiedDiff]`,
+      );
+    }),
+  );
+
   it.effect("labels MCP lifecycle entries with server and tool names", () =>
     Effect.gen(function* () {
       const { adapter, runtime } = yield* startLifecycleRuntime();

apps/server/src/provider/Layers/CodexAdapter.ts

@@ -990,6 +990,23 @@ function runtimeEventBase(
   };
 }
 
+/**
+ * Replaces the diff in a mirrored `turn/diff/updated` payload with a short
+ * marker. `payload.unifiedDiff` on the canonical event already carries the same
+ * string, and a turn diff is large enough that serializing it twice has
+ * exhausted the backend heap while logging the event.
+ */
+function elideNativeTurnDiff(payload: unknown): unknown {
+  if (typeof payload !== "object" || payload === null) return payload;
+  const fields = payload as Record<string, unknown>;
+  const diff = fields.diff;
+  if (typeof diff !== "string") return payload;
+  return {
+    ...fields,
+    diff: `[omitted by t3, ${diff.length} characters in payload.unifiedDiff]`,
+  };
+}
+
 function mapItemLifecycle(
   event: ProviderEvent,
   canonicalThreadId: ThreadId,
@@ -1648,9 +1665,12 @@ function mapToRuntimeEvents(
     if (!payload) {
       return [];
     }
+    const base = runtimeEventBase(event, canonicalThreadId);
+    const raw = base.raw;
     return [
       {
-        ...runtimeEventBase(event, canonicalThreadId),
+        ...base,
+        ...(raw ? { raw: { ...raw, payload: elideNativeTurnDiff(raw.payload) } } : {}),
         type: "turn.diff.updated",
         payload: {
           unifiedDiff: payload.diff,

apps/server/src/provider/Layers/EventNdjsonLogger.test.ts

@@ -127,6 +127,46 @@ describe("EventNdjsonLogger", () => {
     }),
   );
 
+  it.effect("truncates oversized string values before serializing an event", () =>
+    Effect.gen(function* () {
+      const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
+      const basePath = NodePath.join(tempDir, "provider-canonical.ndjson");
+
+      try {
+        const logger = yield* makeEventNdjsonLogger(basePath, {
+          stream: "canonical",
+          maxStringLength: 64,
+        });
+        assert.notEqual(logger, undefined);
+        if (!logger) {
+          return;
+        }
+
+        // A real Codex turn diff reaches hundreds of MiB. Serializing it whole,
+        // twice over, is what exhausted the backend heap.
+        const diff = "d".repeat(200_000);
+        yield* logger.write(
+          {
+            type: "turn.diff.updated",
+            id: "evt-diff",
+            payload: { unifiedDiff: diff },
+            raw: { method: "turn/diff/updated", payload: { diff } },
+          },
+          ThreadId.make("thread-diff"),
+        );
+        yield* logger.close();
+
+        const line = NodeFS.readFileSync(ownedLogPath(basePath, "thread-diff"), "utf8").trim();
+        assert.equal(line.length < 1_000, true);
+        assert.notInclude(line, "d".repeat(65));
+        assert.include(line, "[truncated by t3, 200000 characters total]");
+        assert.include(line, '"id":"evt-diff"');
+      } finally {
+        NodeFS.rmSync(tempDir, { recursive: true, force: true });
+      }
+    }),
+  );
+
   it.effect(
     "falls back to a global segment when orchestration thread id is missing or invalid",
     () =>

apps/server/src/provider/Layers/EventNdjsonLogger.ts

@@ -32,6 +32,9 @@ const DEFAULT_MAX_AGE_MS = 14 * DAY_MS;
 const DEFAULT_RETENTION_CHECK_INTERVAL_MS = 5 * 60 * 1_000;
 const DEFAULT_MAX_BUFFERED_BYTES = MEBIBYTE;
 const DEFAULT_MAX_BUFFERED_RECORDS = 512;
+// Per string value, in UTF-16 code units rather than bytes: `String.length` is
+// O(1), while measuring real bytes would scan every string on the write path.
+const DEFAULT_MAX_STRING_LENGTH = 256 * 1024;
 const GLOBAL_THREAD_SEGMENT = "_global";
 const LOG_SCOPE = "provider-observability";
 const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
@@ -80,6 +83,7 @@ export interface EventNdjsonLogStoreOptions {
   readonly retentionCheckIntervalMs?: number;
   readonly maxBufferedBytes?: number;
   readonly maxBufferedRecords?: number;
+  readonly maxStringLength?: number;
   readonly attribution?: ResourceAttribution["Service"];
 }
 
@@ -126,6 +130,7 @@ interface ResolvedOptions {
   readonly retentionCheckIntervalMs: number;
   readonly maxBufferedBytes: number;
   readonly maxBufferedRecords: number;
+  readonly maxStringLength: number;
   readonly attribution: ResourceAttribution["Service"] | undefined;
 }
 
@@ -373,6 +378,7 @@ function resolveOptions(
       options.retentionCheckIntervalMs ?? DEFAULT_RETENTION_CHECK_INTERVAL_MS,
     maxBufferedBytes: options.maxBufferedBytes ?? DEFAULT_MAX_BUFFERED_BYTES,
     maxBufferedRecords: options.maxBufferedRecords ?? DEFAULT_MAX_BUFFERED_RECORDS,
+    maxStringLength: options.maxStringLength ?? DEFAULT_MAX_STRING_LENGTH,
     attribution: options.attribution,
   } satisfies ResolvedOptions;
 
@@ -385,6 +391,7 @@ function resolveOptions(
     ["retentionCheckIntervalMs", resolved.retentionCheckIntervalMs, 1],
     ["maxBufferedBytes", resolved.maxBufferedBytes, 1],
     ["maxBufferedRecords", resolved.maxBufferedRecords, 1],
+    ["maxStringLength", resolved.maxStringLength, 1],
   ] as const;
 
   for (const [option, value, minimum] of validations) {
@@ -494,6 +501,68 @@ function drainPending(input: {
   ];
 }
 
+function truncateString(value: string, maxLength: number): string {
+  // Never cut between a surrogate pair; a lone surrogate would survive into the log line.
+  const lastCode = value.charCodeAt(maxLength - 1);
+  const keep = lastCode >= 0xd800 && lastCode <= 0xdbff ? maxLength - 1 : maxLength;
+  return `${value.slice(0, keep)}[truncated by t3, ${value.length} characters total]`;
+}
+
+function isPlainContainer(value: object): boolean {
+  if (Array.isArray(value)) return true;
+  const prototype = Reflect.getPrototypeOf(value);
+  return prototype === Object.prototype || prototype === null;
+}
+
+/**
+ * Bounds the string values of one event before it is serialized. A provider can
+ * hand us a payload far larger than any log record should be (a Codex
+ * `turn/diff/updated` diff reaches hundreds of MiB on a big turn), and encoding
+ * it whole materializes an equally large line that can exhaust the heap.
+ *
+ * Values that already fit are returned by reference, so an ordinary event
+ * allocates nothing. Anything that is not a plain object or array is left alone,
+ * because `toJSON` carriers such as `Date` must reach the encoder intact. A cycle
+ * is returned untouched at the point it closes, so serialization still fails the
+ * way it did before and `serializeEvent` reports it.
+ */
+function clampStrings(value: unknown, maxLength: number, ancestors: Set<object>): unknown {
+  if (typeof value === "string") {
+    return value.length > maxLength ? truncateString(value, maxLength) : value;
+  }
+  if (typeof value !== "object" || value === null) return value;
+  if (!isPlainContainer(value) || ancestors.has(value)) return value;
+
+  ancestors.add(value);
+  try {
+    if (Array.isArray(value)) {
+      const entries = value as ReadonlyArray<unknown>;
+      let clampedEntries: Array<unknown> | undefined;
+      for (let index = 0; index < entries.length; index += 1) {
+        const entry = entries[index];
+        const clamped = clampStrings(entry, maxLength, ancestors);
+        if (clamped !== entry && clampedEntries === undefined) {
+          clampedEntries = entries.slice(0, index);
+        }
+        clampedEntries?.push(clamped);
+      }
+      return clampedEntries ?? value;
+    }
+
+    const fields = value as Record<string, unknown>;
+    let clampedFields: Record<string, unknown> | undefined;
+    for (const [key, entry] of Object.entries(fields)) {
+      const clamped = clampStrings(entry, maxLength, ancestors);
+      if (clamped === entry) continue;
+      clampedFields ??= { ...fields };
+      clampedFields[key] = clamped;
+    }
+    return clampedFields ?? value;
+  } finally {
+    ancestors.delete(value);
+  }
+}
+
 const serializeEvent = Effect.fnUntraced(function* (event: unknown) {
   return yield* encodeUnknownJsonString(event).pipe(
     Effect.catch((error) =>
@@ -613,7 +682,9 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
 
     const write = Effect.fnUntraced(function* (event: unknown, threadId: ThreadId | null) {
       if (!shouldPersist(stream, event)) return;
-      const payload = yield* serializeEvent(event);
+      const payload = yield* serializeEvent(
+        clampStrings(event, resolved.maxStringLength, new Set()),
+      );
       if (payload === undefined) return;
 
       const observedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso));