diff --git a/apps/server/package.json b/apps/server/package.json index 3d03fcd45dba..f602f37a38f9 100644 --- a/apps/server/package.json +++ b/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" }, diff --git a/apps/server/src/project/AgentSessionJson.ts b/apps/server/src/project/AgentSessionJson.ts new file mode 100644 index 000000000000..d31c47833d70 --- /dev/null +++ b/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; + +/** 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 | typeof none; + }; + const tokenize = jsonParser({ packValues: false }); + const select = filter({ filter: selectPath, streamKeys: false }) as ( + input: Token | typeof none, + ) => Token | Many | 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; + }, + }; +} diff --git a/apps/server/src/project/AgentSessionScanner.test.ts b/apps/server/src/project/AgentSessionScanner.test.ts index aee64cf4b5d6..0ca550160507 100644 --- a/apps/server/src/project/AgentSessionScanner.test.ts +++ b/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,7 +1835,7 @@ 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"); @@ -1843,7 +1843,7 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => { 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" }], + }, + }, + ]); }), ); diff --git a/apps/server/src/project/AgentSessionScanner.ts b/apps/server/src/project/AgentSessionScanner.ts index bc6d093d42a1..06d57441b54c 100644 --- a/apps/server/src/project/AgentSessionScanner.ts +++ b/apps/server/src/project/AgentSessionScanner.ts @@ -35,6 +35,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; @@ -45,6 +46,11 @@ import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSn import { resolveCodexHomeLayout } from "../provider/Drivers/CodexHomeLayout.ts"; import { expandHomePath } from "../pathExpansion.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { + createTranscriptJsonReader, + createTranscriptJsonSelector, + TranscriptJsonLimitError, +} from "./AgentSessionJson.ts"; /** Chunk size for full transcript reads. */ const TRANSCRIPT_PREFIX_BYTES = 32 * 1024; @@ -71,9 +77,15 @@ const MAX_METADATA_OPERATIONS_PER_SOURCE = MAX_TRANSCRIPTS_PER_SOURCE * 4; const MAX_METADATA_RECORDS_PER_SOURCE = 100_000; const MAX_METADATA_RECORDS_PER_TRANSCRIPT = 1_000; const RECENT_THREAD_WINDOW_MS = 30 * 24 * 60 * 60 * 1000; -const MAX_IMPORTED_TRANSCRIPT_BYTES = 16 * 1024 * 1024; +/** + * Large tool results (especially screenshots) can make an otherwise ordinary + * Codex transcript several GiB. Streaming field selection avoids allocating + * those payloads. Raw I/O and selected history have separate budgets. + */ +const MAX_IMPORTED_TRANSCRIPT_BYTES = 4 * 1024 * 1024 * 1024; const MAX_IMPORTED_MESSAGES = 200; -const MAX_IMPORT_BYTES = 64 * 1024 * 1024; +const MAX_IMPORT_HISTORY_BYTES = 32 * 1024 * 1024; +const MAX_IMPORT_BYTES = 4 * 1024 * 1024 * 1024; const MAX_IMPORT_TRANSCRIPTS = 100; const MAX_IMPORT_RECORDS = 100_000; @@ -95,6 +107,7 @@ const CodexTurnMetadata = Schema.Struct({ const TranscriptRecord = Schema.Struct({ type: Schema.optional(Schema.String), timestamp: Schema.optional(Schema.String), + cwd: Schema.optional(Schema.String), sessionId: Schema.optional(Schema.String), aiTitle: Schema.optional(Schema.String), isSidechain: Schema.optional(Schema.Boolean), @@ -109,6 +122,7 @@ const TranscriptRecord = Schema.Struct({ role: Schema.optional(Schema.String), message: Schema.optional(Schema.String), model: Schema.optional(Schema.String), + cwd: Schema.optional(Schema.String), content: Schema.optional(Schema.Array(TranscriptContentBlock)), internal_chat_message_metadata_passthrough: Schema.optional(Schema.Unknown), }), @@ -118,8 +132,19 @@ const TranscriptRecord = Schema.Struct({ const decodeClaudeSettings = Schema.decodeUnknownOption(ClaudeSettings); const decodeCodexSettings = Schema.decodeUnknownOption(CodexSettings); const decodeTranscriptRecord = Schema.decodeUnknownOption(Schema.fromJsonString(TranscriptRecord)); +const decodeTranscriptValue = Schema.decodeUnknownOption(TranscriptRecord); +const selectTranscriptPath = createTranscriptJsonSelector(TranscriptRecord); const decodeCodexTurnMetadata = Schema.decodeUnknownOption(CodexTurnMetadata); +type DecodedTranscriptRecord = typeof TranscriptRecord.Type; + +interface AgentSessionTranscriptMetadata { + readonly source: AgentSessionSource; + readonly providerInstanceId: ProviderInstanceId; + readonly fallbackSessionId: string; + readonly lastActiveAtMs: number; +} + export interface AgentSessionThreadMessage { readonly role: "user" | "assistant"; readonly text: string; @@ -254,16 +279,20 @@ function codexTurnId(metadata: unknown): string | null { /** Keep visible user and assistant text while ignoring tools, reasoning, and malformed records. */ export function parseAgentSessionTranscript( - input: { + input: AgentSessionTranscriptMetadata & { readonly contents: string; - readonly source: AgentSessionSource; - readonly providerInstanceId: ProviderInstanceId; - readonly fallbackSessionId: string; - readonly lastActiveAtMs: number; }, lines = splitTranscriptRecords(input.contents, MAX_IMPORT_RECORDS + 1), ): AgentSessionThread | null { if (lines.length > MAX_IMPORT_RECORDS) return null; + const records = lines.flatMap((line) => Option.toArray(decodeTranscriptRecord(line))); + return parseAgentSessionRecords(input, records); +} + +function parseAgentSessionRecords( + input: AgentSessionTranscriptMetadata, + records: ReadonlyArray, +): AgentSessionThread | null { const fallbackTimestamp = DateTime.formatIso(DateTime.makeUnsafe(input.lastActiveAtMs)); // Claude filenames are session IDs. Codex rollout filenames include extra // timestamp text, so only transcript metadata can provide a resumable ID. @@ -275,13 +304,6 @@ export function parseAgentSessionTranscript( let firstUserMessage: | (AgentSessionThreadMessage & { readonly codexResponseUser: boolean }) | undefined; - function* decodedRecords() { - for (const line of lines) { - const decoded = decodeTranscriptRecord(line); - if (Option.isSome(decoded)) yield decoded.value; - } - } - // A Codex response item can include generated setup text beside the real // prompt. Suppress response-user records only when the shared turn ID and a // verbatim event copy prove which prompt the user submitted. @@ -308,7 +330,7 @@ export function parseAgentSessionTranscript( }; if (input.source === "codex") { let recordIndex = -1; - for (const record of decodedRecords()) { + for (const record of records) { recordIndex += 1; if ( record.type === "response_item" && @@ -365,7 +387,7 @@ export function parseAgentSessionTranscript( }; let recordIndex = -1; - for (const record of decodedRecords()) { + for (const record of records) { recordIndex += 1; if (input.source === "claudeAgent") { if ( @@ -478,6 +500,35 @@ export function parseAgentSessionTranscript( }; } +function extractDecodedCwd(record: DecodedTranscriptRecord): string | null { + const cwd = record.cwd?.trim() || record.payload?.cwd?.trim(); + return cwd && cwd.length > 0 ? cwd : null; +} + +function shouldRetainDecodedRecord( + source: AgentSessionSource, + record: DecodedTranscriptRecord, +): boolean { + if (extractDecodedCwd(record) !== null) return true; + if (source === "claudeAgent") { + return ( + record.type === "user" || + record.type === "assistant" || + record.sessionId !== undefined || + record.aiTitle !== undefined || + record.message?.model !== undefined + ); + } + return ( + record.type === "session_meta" || + record.type === "turn_context" || + (record.type === "event_msg" && record.payload?.type === "user_message") || + (record.type === "response_item" && + record.payload?.type === "message" && + (record.payload.role === "user" || record.payload.role === "assistant")) + ); +} + /** * T3 Code runs its own agent sessions inside disposable worktrees. Their * transcripts look exactly like user sessions, but re-importing the app's own @@ -560,6 +611,9 @@ function sameTranscriptIdentity( export const make = Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; + // Different project imports can arrive concurrently from multiple clients. + // Only one transcript may hold its selected-history budget at a time. + const importReadLock = yield* Semaphore.make(1); const path = yield* Path.Path; const serverConfig = yield* ServerConfig.ServerConfig; const serverSettings = yield* ServerSettings.ServerSettingsService; @@ -692,10 +746,16 @@ export const make = Effect.gen(function* () { ).pipe(Effect.orElseSucceed(() => null)); }); - /** Check the open file before and after reading, without reading past its reserved byte budget. */ + /** + * Project history fields while reading, before allocating whole JSON records. + * Check the file identity on both sides of the read. A selected-history budget + * failure rejects the entire transcript before any imported messages persist. + */ const readTranscript = Effect.fn("AgentSessionScanner.readTranscript")(function* ( filePath: string, expected: ReturnType, + recordLimit: number, + source: AgentSessionSource, ) { if (expected.size > MAX_IMPORTED_TRANSCRIPT_BYTES) return null; @@ -706,9 +766,38 @@ export const make = Effect.gen(function* () { if (!sameTranscriptIdentity(expected, transcriptIdentity(filePath, yield* file.stat))) { return null; } - const decoder = new TextDecoder(); - let contents = ""; + const records: Array = []; + let historyBytes = 0; + let recordBytes = 0; + let recordCount = 0; let bytesRead = 0; + const reserve = (bytes: number) => { + recordBytes += bytes; + if (historyBytes + recordBytes > MAX_IMPORT_HISTORY_BYTES) { + throw new TranscriptJsonLimitError( + "Transcript selected history exceeds the 32 MiB memory budget", + ); + } + }; + let reader = createTranscriptJsonReader(reserve, selectTranscriptPath); + let decoder = new TextDecoder(); + let recordStarted = false; + + const finishRecord = () => { + reader.write(decoder.decode()); + recordCount += 1; + if (recordCount > recordLimit) return false; + const decoded = decodeTranscriptValue(reader.finish()); + if (Option.isSome(decoded) && shouldRetainDecodedRecord(source, decoded.value)) { + records.push(decoded.value); + historyBytes += recordBytes; + } + recordBytes = 0; + reader = createTranscriptJsonReader(reserve, selectTranscriptPath); + decoder = new TextDecoder(); + recordStarted = false; + return true; + }; while (bytesRead < expected.size) { const next = yield* file.readAlloc( @@ -719,16 +808,36 @@ export const make = Effect.gen(function* () { } bytesRead += next.value.byteLength; - contents += decoder.decode(next.value, { stream: true }); + const withinBudget = yield* Effect.try(() => { + let start = 0; + while (start < next.value.byteLength) { + const newline = next.value.indexOf(10, start); + const end = newline === -1 ? next.value.byteLength : newline; + recordStarted = true; + reader.write(decoder.decode(next.value.subarray(start, end), { stream: true })); + if (newline === -1) break; + if (!finishRecord()) return false; + start = newline + 1; + } + return true; + }); + if (!withinBudget) return null; } + if (recordStarted && !(yield* Effect.try(finishRecord))) return null; return sameTranscriptIdentity(expected, transcriptIdentity(filePath, yield* file.stat)) - ? contents + decoder.decode() + ? { records, recordCount } : null; }), ), ), - ).pipe(Effect.orElseSucceed(() => null)); + ).pipe( + Effect.catch((cause) => + Effect.logWarning("Could not read imported transcript", { filePath, cause }).pipe( + Effect.as(null), + ), + ), + ); }); /** @@ -1238,20 +1347,21 @@ export const make = Effect.gen(function* () { // Reserve the whole file even if its read or parse fails. transcriptsRemaining -= 1; bytesRemaining -= identity.size; - const contents = yield* readTranscript(transcript.filePath, identity); - if (contents === null) { - return Option.some({ _tag: "Skipped" }); - } - const lines = splitTranscriptRecords(contents, recordsRemaining + 1); - if (lines.length > recordsRemaining) { + const snapshot = yield* readTranscript( + transcript.filePath, + identity, + recordsRemaining, + candidate.source, + ); + if (snapshot === null) { return Option.some({ _tag: "Skipped" }); } - recordsRemaining -= lines.length; + recordsRemaining -= snapshot.recordCount; // A stable replacement file can belong to a different project than the cached candidate. let snapshotCwd: string | null = null; - for (const line of lines) { - snapshotCwd = extractCwd(line); + for (const record of snapshot.records) { + snapshotCwd = extractDecodedCwd(record); if (snapshotCwd !== null) break; } if (snapshotCwd === null) { @@ -1265,15 +1375,14 @@ export const make = Effect.gen(function* () { return Option.some({ _tag: "Skipped" }); } - const parsedThread = parseAgentSessionTranscript( + const parsedThread = parseAgentSessionRecords( { - contents, source: candidate.source, providerInstanceId: candidate.providerInstanceId, fallbackSessionId: path.basename(transcript.filePath, ".jsonl"), lastActiveAtMs: transcript.mtimeMs, }, - lines, + snapshot.records, ); if (parsedThread === null) { return Option.some({ _tag: "Skipped" }); @@ -1295,7 +1404,7 @@ export const make = Effect.gen(function* () { thread: parsedThread, source, }); - }), + }).pipe(importReadLock.withPermits(1)), ), Stream.map(Option.toArray), Stream.flattenIterable, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1b1598d0dcea..567d72e6da35 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -502,6 +502,12 @@ importers: node-pty: specifier: ^1.1.0 version: 1.1.0 + stream-chain: + specifier: ^4.2.5 + version: 4.2.5 + stream-json: + specifier: 3.6.0 + version: 3.6.0 yaml: specifier: ^2.9.0 version: 2.9.0 @@ -9848,6 +9854,13 @@ packages: resolution: {integrity: sha512-uyQK/mx5QjHun80FLJTfaWE7JtwfRMKBLkMne6udYOmvH0CawotVa7TfgYHzAnpphn4+TweIx1QKMnRIbipmUg==} engines: {node: '>= 0.10.0'} + stream-chain@4.2.5: + resolution: {integrity: sha512-Wtyq3bNE3ggLR0v2vftqvuhltym3WbZAkZpfIrkr5F/6vpeUmWmwTgXa16zD87gpahwJ/Qulq3zVfUlgIc0J2A==} + engines: {node: '>=22'} + + stream-json@3.6.0: + resolution: {integrity: sha512-NiJdqxKyau579z/E8vfqcjWfSDWxW/AT99javFXdPXF147Z5za85LRXSHEmSX9TKOakB7gaIccfD0fOIctb7KQ==} + strict-event-emitter@0.5.1: resolution: {integrity: sha512-vMgjE/GGEPEFnhFub6pa4FmJBRBVOLpIII2hvCZ8Kzb7K0hlHo7mQv6xYrBvCL2LtAIBwFUK8wvuJgTVSQ5MFQ==} @@ -20965,6 +20978,12 @@ snapshots: stream-buffers@2.2.0: {} + stream-chain@4.2.5: {} + + stream-json@3.6.0: + dependencies: + stream-chain: 4.2.5 + strict-event-emitter@0.5.1: optional: true