fix(server): import transcripts with oversized tool records

apps/server/package.json

@@ -31,6 +31,8 @@
     "effect": "catalog:",
     "msgpackr-extract": "3.0.4",
     "node-pty": "^1.1.0",
+    "stream-chain": "^4.2.5",
+    "stream-json": "3.6.0",
     "yaml": "catalog:",
     "yauzl": "^3.4.0"
   },

apps/server/src/project/AgentSessionJson.ts

@@ -0,0 +1,155 @@
+import * as SchemaAST from "effect/SchemaAST";
+import { isMany, none, type Many } from "stream-chain/defs.js";
+import { Assembler } from "stream-json/core/assembler.js";
+import { filter } from "stream-json/core/filters/filter.js";
+import * as StreamJson from "stream-json/core/parser.js";
+import type { ParserOptions, Token } from "stream-json/core/parser.js";
+
+type JsonPath = ReadonlyArray<string | number | null>;
+
+/** Select schema fields before assembling their values, without a second field list. */
+export function createTranscriptJsonSelector(schema: { readonly ast: SchemaAST.AST }) {
+  const ast = SchemaAST.toEncoded(schema.ast);
+  const includes = (node: SchemaAST.AST, path: JsonPath, index: number): boolean => {
+    if (index === path.length) return true;
+    switch (node._tag) {
+      case "Objects":
+        // Records have dynamic keys. Keep their values for the decoder to validate.
+        if (node.indexSignatures.length > 0) return true;
+        return node.propertySignatures.some(
+          (property) =>
+            String(property.name) === path[index] && includes(property.type, path, index + 1),
+        );
+      case "Arrays": {
+        const key = path[index];
+        if (typeof key !== "number") return true;
+        const element = node.elements[key];
+        if (element) return includes(element, path, index + 1);
+        return node.rest.length === 0 || node.rest.some((item) => includes(item, path, index + 1));
+      }
+      case "Union":
+        return node.types.some((type) => includes(type, path, index));
+      case "Suspend":
+        return includes(node.thunk(), path, index);
+      case "Unknown":
+      case "Any":
+      case "ObjectKeyword":
+      case "Declaration":
+        // Unstructured/custom schemas must reach the decoder intact. The
+        // shared budget still bounds their allocations.
+        return true;
+      default:
+        return false;
+    }
+  };
+  return (path: JsonPath) => includes(ast, path, 0);
+}
+
+export class TranscriptJsonLimitError extends Error {}
+
+/**
+ * Project a single JSONL record without materializing unselected string values.
+ * The caller supplies a shared allocation budget for the entire transcript.
+ * Budget exhaustion rejects the transcript, never a message within it.
+ */
+export function createTranscriptJsonReader(
+  reserve: (bytes: number) => void,
+  selectPath: (path: JsonPath) => boolean,
+) {
+  // The synchronous tokenizer is exported at runtime in 3.6.0, but omitted
+  // from its bundled types. Unlike parser(), it does not wrap tokens in an
+  // async generator; the file reader already supplies backpressure and UTF-8.
+  const { jsonParser } = StreamJson as typeof StreamJson & {
+    jsonParser: (
+      options: ParserOptions,
+    ) => (input: string | typeof none) => Many<Token> | typeof none;
+  };
+  const tokenize = jsonParser({ packValues: false });
+  const select = filter({ filter: selectPath, streamKeys: false }) as (
+    input: Token | typeof none,
+  ) => Token | Many<Token> | typeof none;
+  const assembler = new Assembler();
+  let key: string | null = null;
+  let value = "";
+  let depth = 0;
+  let complete = false;
+  let malformed = false;
+
+  const assemble = (token: Token) => {
+    reserve(
+      64 + ("value" in token && typeof token.value === "string" ? token.value.length * 2 : 0),
+    );
+    switch (token.name) {
+      case "startString":
+      case "startNumber":
+        value = "";
+        break;
+      case "stringChunk":
+      case "numberChunk":
+        value += token.value;
+        break;
+      case "endString":
+        assembler.consume({ name: "stringValue", value });
+        value = "";
+        break;
+      case "endNumber":
+        assembler.consume({ name: "numberValue", value });
+        value = "";
+        break;
+      default:
+        assembler.consume(token);
+    }
+  };
+  const selectToken = (token: Token | typeof none) => {
+    const selected = select(token);
+    if (selected === none) return;
+    if (isMany(selected)) {
+      for (const item of selected.values) assemble(item);
+    } else {
+      assemble(selected);
+    }
+  };
+  const consume = (input: string | typeof none) => {
+    if (malformed) return;
+    try {
+      const tokens = tokenize(input);
+      if (tokens === none) return;
+      for (const token of tokens.values) {
+        if (token.name === "startObject" || token.name === "startArray") {
+          if (++depth > 128)
+            throw new TranscriptJsonLimitError("Transcript JSON nesting exceeds 128 levels");
+        } else if (token.name === "endObject" || token.name === "endArray") {
+          if (--depth === 0) complete = true;
+        }
+        // Charge keys before assembling them, including unknown names. Reject
+        // the transcript on exhaustion instead of silently shortening a key.
+        if (token.name === "startKey") {
+          key = "";
+        } else if (token.name === "stringChunk" && key !== null) {
+          reserve(token.value.length * 2);
+          key += token.value;
+        } else if (token.name === "endKey") {
+          selectToken({ name: "keyValue", value: key ?? "" });
+          key = null;
+        } else {
+          selectToken(token);
+        }
+      }
+    } catch (cause) {
+      if (cause instanceof Error && cause.message.startsWith("Parser ")) {
+        malformed = true;
+      } else {
+        throw cause;
+      }
+    }
+  };
+  return {
+    write: (chunk: string) => consume(chunk),
+    finish: (): unknown => {
+      consume(none);
+      if (malformed || !complete) return undefined;
+      selectToken(none);
+      return assembler.done ? assembler.current : undefined;
+    },
+  };
+}

apps/server/src/project/AgentSessionScanner.test.ts

@@ -1522,7 +1522,7 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => {
       }),
     );
 
-    it.effect("shares a 64 MiB full-read budget across providers without hiding projects", () =>
+    it.effect("streams large transcripts across providers without hiding projects", () =>
       Effect.gen(function* () {
         const path = yield* Path.Path;
         const fileSystem = yield* FileSystem.FileSystem;
@@ -1619,9 +1619,9 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => {
           "Importable",
           "Importable",
           "Importable",
-          "Skipped",
+          "Importable",
         ]);
-        expect(fullReadBytes).toBe(64 * 1024 * 1024);
+        expect(fullReadBytes).toBe(80 * 1024 * 1024);
       }),
     );
 
@@ -1835,15 +1835,15 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => {
       }),
     );
 
-    it.effect("reports an eligible transcript over 16 MiB as skipped", () =>
+    it.effect("imports visible history from a transcript with an oversized tool record", () =>
       Effect.gen(function* () {
         const path = yield* Path.Path;
         const nowMs = Date.parse("2026-08-24T12:00:00.000Z");
         yield* TestClock.setTime(nowMs);
         const claudeHomePath = yield* makeTempDir("t3code-claude-home-");
         const codexHomePath = yield* makeTempDir("t3code-codex-home-");
         const workspace = yield* makeTempDir("t3code-workspace-");
-        const transcript = [
+        const transcript = `${[
           encodeTranscriptRecord({
             type: "session_meta",
             payload: { id: "large-session", cwd: workspace },
@@ -1852,9 +1852,10 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => {
             type: "event_msg",
             payload: { type: "user_message", message: "Import this large session" },
           }),
-        ]
-          .join("\n")
-          .padEnd(16 * 1024 * 1024 + 1, " ");
+        ].join("\n")}\n${encodeTranscriptRecord({ type: "tool_result", data: "" }).padEnd(
+          16 * 1024 * 1024 + 1,
+          " ",
+        )}`;
         yield* writeTranscript({
           filePath: path.join(codexHomePath, "sessions", "2026", "08", "24", "rollout-large.jsonl"),
           contents: transcript,
@@ -1867,7 +1868,15 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => {
           workspaceRoot: workspace,
         });
 
-        expect(outcomes).toEqual([{ _tag: "Skipped" }]);
+        expect(outcomes).toMatchObject([
+          {
+            _tag: "Importable",
+            thread: {
+              providerSessionId: "large-session",
+              messages: [{ role: "user", text: "Import this large session" }],
+            },
+          },
+        ]);
       }),
     );