| ... | ... | @@ -35,6 +35,7 @@ import * as Layer from "effect/Layer"; |
| 35 | 35 | import * as Option from "effect/Option"; |
| 36 | 36 | import * as Path from "effect/Path"; |
| 37 | 37 | import * as Schema from "effect/Schema"; |
| 38 | import * as Semaphore from "effect/Semaphore"; |
| 38 | 39 | import * as Stream from "effect/Stream"; |
| 39 | 40 | |
| 40 | 41 | import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess"; |
| ... | ... | @@ -45,6 +46,11 @@ import * as ProjectionSnapshotQuery from "../orchestration/Services/ProjectionSn |
| 45 | 46 | import { resolveCodexHomeLayout } from "../provider/Drivers/CodexHomeLayout.ts"; |
| 46 | 47 | import { expandHomePath } from "../pathExpansion.ts"; |
| 47 | 48 | import * as ServerSettings from "../serverSettings.ts"; |
| 49 | import { |
| 50 | createTranscriptJsonReader, |
| 51 | createTranscriptJsonSelector, |
| 52 | TranscriptJsonLimitError, |
| 53 | } from "./AgentSessionJson.ts"; |
| 48 | 54 | |
| 49 | 55 | /** Chunk size for full transcript reads. */ |
| 50 | 56 | const TRANSCRIPT_PREFIX_BYTES = 32 * 1024; |
| ... | ... | @@ -71,9 +77,15 @@ const MAX_METADATA_OPERATIONS_PER_SOURCE = MAX_TRANSCRIPTS_PER_SOURCE * 4; |
| 71 | 77 | const MAX_METADATA_RECORDS_PER_SOURCE = 100_000; |
| 72 | 78 | const MAX_METADATA_RECORDS_PER_TRANSCRIPT = 1_000; |
| 73 | 79 | const RECENT_THREAD_WINDOW_MS = 30 * 24 * 60 * 60 * 1000; |
| 74 | | const MAX_IMPORTED_TRANSCRIPT_BYTES = 16 * 1024 * 1024; |
| 80 | /** |
| 81 | * Large tool results (especially screenshots) can make an otherwise ordinary |
| 82 | * Codex transcript several GiB. Streaming field selection avoids allocating |
| 83 | * those payloads. Raw I/O and selected history have separate budgets. |
| 84 | */ |
| 85 | const MAX_IMPORTED_TRANSCRIPT_BYTES = 4 * 1024 * 1024 * 1024; |
| 75 | 86 | const MAX_IMPORTED_MESSAGES = 200; |
| 76 | | const MAX_IMPORT_BYTES = 64 * 1024 * 1024; |
| 87 | const MAX_IMPORT_HISTORY_BYTES = 32 * 1024 * 1024; |
| 88 | const MAX_IMPORT_BYTES = 4 * 1024 * 1024 * 1024; |
| 77 | 89 | const MAX_IMPORT_TRANSCRIPTS = 100; |
| 78 | 90 | const MAX_IMPORT_RECORDS = 100_000; |
| 79 | 91 | |
| ... | ... | @@ -95,6 +107,7 @@ const CodexTurnMetadata = Schema.Struct({ |
| 95 | 107 | const TranscriptRecord = Schema.Struct({ |
| 96 | 108 | type: Schema.optional(Schema.String), |
| 97 | 109 | timestamp: Schema.optional(Schema.String), |
| 110 | cwd: Schema.optional(Schema.String), |
| 98 | 111 | sessionId: Schema.optional(Schema.String), |
| 99 | 112 | aiTitle: Schema.optional(Schema.String), |
| 100 | 113 | isSidechain: Schema.optional(Schema.Boolean), |
| ... | ... | @@ -109,6 +122,7 @@ const TranscriptRecord = Schema.Struct({ |
| 109 | 122 | role: Schema.optional(Schema.String), |
| 110 | 123 | message: Schema.optional(Schema.String), |
| 111 | 124 | model: Schema.optional(Schema.String), |
| 125 | cwd: Schema.optional(Schema.String), |
| 112 | 126 | content: Schema.optional(Schema.Array(TranscriptContentBlock)), |
| 113 | 127 | internal_chat_message_metadata_passthrough: Schema.optional(Schema.Unknown), |
| 114 | 128 | }), |
| ... | ... | @@ -118,8 +132,19 @@ const TranscriptRecord = Schema.Struct({ |
| 118 | 132 | const decodeClaudeSettings = Schema.decodeUnknownOption(ClaudeSettings); |
| 119 | 133 | const decodeCodexSettings = Schema.decodeUnknownOption(CodexSettings); |
| 120 | 134 | const decodeTranscriptRecord = Schema.decodeUnknownOption(Schema.fromJsonString(TranscriptRecord)); |
| 135 | const decodeTranscriptValue = Schema.decodeUnknownOption(TranscriptRecord); |
| 136 | const selectTranscriptPath = createTranscriptJsonSelector(TranscriptRecord); |
| 121 | 137 | const decodeCodexTurnMetadata = Schema.decodeUnknownOption(CodexTurnMetadata); |
| 122 | 138 | |
| 139 | type DecodedTranscriptRecord = typeof TranscriptRecord.Type; |
| 140 | |
| 141 | interface AgentSessionTranscriptMetadata { |
| 142 | readonly source: AgentSessionSource; |
| 143 | readonly providerInstanceId: ProviderInstanceId; |
| 144 | readonly fallbackSessionId: string; |
| 145 | readonly lastActiveAtMs: number; |
| 146 | } |
| 147 | |
| 123 | 148 | export interface AgentSessionThreadMessage { |
| 124 | 149 | readonly role: "user" | "assistant"; |
| 125 | 150 | readonly text: string; |
| ... | ... | @@ -254,16 +279,20 @@ function codexTurnId(metadata: unknown): string | null { |
| 254 | 279 | |
| 255 | 280 | /** Keep visible user and assistant text while ignoring tools, reasoning, and malformed records. */ |
| 256 | 281 | export function parseAgentSessionTranscript( |
| 257 | | input: { |
| 282 | input: AgentSessionTranscriptMetadata & { |
| 258 | 283 | readonly contents: string; |
| 259 | | readonly source: AgentSessionSource; |
| 260 | | readonly providerInstanceId: ProviderInstanceId; |
| 261 | | readonly fallbackSessionId: string; |
| 262 | | readonly lastActiveAtMs: number; |
| 263 | 284 | }, |
| 264 | 285 | lines = splitTranscriptRecords(input.contents, MAX_IMPORT_RECORDS + 1), |
| 265 | 286 | ): AgentSessionThread | null { |
| 266 | 287 | if (lines.length > MAX_IMPORT_RECORDS) return null; |
| 288 | const records = lines.flatMap((line) => Option.toArray(decodeTranscriptRecord(line))); |
| 289 | return parseAgentSessionRecords(input, records); |
| 290 | } |
| 291 | |
| 292 | function parseAgentSessionRecords( |
| 293 | input: AgentSessionTranscriptMetadata, |
| 294 | records: ReadonlyArray<DecodedTranscriptRecord>, |
| 295 | ): AgentSessionThread | null { |
| 267 | 296 | const fallbackTimestamp = DateTime.formatIso(DateTime.makeUnsafe(input.lastActiveAtMs)); |
| 268 | 297 | // Claude filenames are session IDs. Codex rollout filenames include extra |
| 269 | 298 | // timestamp text, so only transcript metadata can provide a resumable ID. |
| ... | ... | @@ -275,13 +304,6 @@ export function parseAgentSessionTranscript( |
| 275 | 304 | let firstUserMessage: |
| 276 | 305 | | (AgentSessionThreadMessage & { readonly codexResponseUser: boolean }) |
| 277 | 306 | | undefined; |
| 278 | | function* decodedRecords() { |
| 279 | | for (const line of lines) { |
| 280 | | const decoded = decodeTranscriptRecord(line); |
| 281 | | if (Option.isSome(decoded)) yield decoded.value; |
| 282 | | } |
| 283 | | } |
| 284 | | |
| 285 | 307 | // A Codex response item can include generated setup text beside the real |
| 286 | 308 | // prompt. Suppress response-user records only when the shared turn ID and a |
| 287 | 309 | // verbatim event copy prove which prompt the user submitted. |
| ... | ... | @@ -308,7 +330,7 @@ export function parseAgentSessionTranscript( |
| 308 | 330 | }; |
| 309 | 331 | if (input.source === "codex") { |
| 310 | 332 | let recordIndex = -1; |
| 311 | | for (const record of decodedRecords()) { |
| 333 | for (const record of records) { |
| 312 | 334 | recordIndex += 1; |
| 313 | 335 | if ( |
| 314 | 336 | record.type === "response_item" && |
| ... | ... | @@ -365,7 +387,7 @@ export function parseAgentSessionTranscript( |
| 365 | 387 | }; |
| 366 | 388 | |
| 367 | 389 | let recordIndex = -1; |
| 368 | | for (const record of decodedRecords()) { |
| 390 | for (const record of records) { |
| 369 | 391 | recordIndex += 1; |
| 370 | 392 | if (input.source === "claudeAgent") { |
| 371 | 393 | if ( |
| ... | ... | @@ -478,6 +500,35 @@ export function parseAgentSessionTranscript( |
| 478 | 500 | }; |
| 479 | 501 | } |
| 480 | 502 | |
| 503 | function extractDecodedCwd(record: DecodedTranscriptRecord): string | null { |
| 504 | const cwd = record.cwd?.trim() || record.payload?.cwd?.trim(); |
| 505 | return cwd && cwd.length > 0 ? cwd : null; |
| 506 | } |
| 507 | |
| 508 | function shouldRetainDecodedRecord( |
| 509 | source: AgentSessionSource, |
| 510 | record: DecodedTranscriptRecord, |
| 511 | ): boolean { |
| 512 | if (extractDecodedCwd(record) !== null) return true; |
| 513 | if (source === "claudeAgent") { |
| 514 | return ( |
| 515 | record.type === "user" || |
| 516 | record.type === "assistant" || |
| 517 | record.sessionId !== undefined || |
| 518 | record.aiTitle !== undefined || |
| 519 | record.message?.model !== undefined |
| 520 | ); |
| 521 | } |
| 522 | return ( |
| 523 | record.type === "session_meta" || |
| 524 | record.type === "turn_context" || |
| 525 | (record.type === "event_msg" && record.payload?.type === "user_message") || |
| 526 | (record.type === "response_item" && |
| 527 | record.payload?.type === "message" && |
| 528 | (record.payload.role === "user" || record.payload.role === "assistant")) |
| 529 | ); |
| 530 | } |
| 531 | |
| 481 | 532 | /** |
| 482 | 533 | * T3 Code runs its own agent sessions inside disposable worktrees. Their |
| 483 | 534 | * transcripts look exactly like user sessions, but re-importing the app's own |
| ... | ... | @@ -560,6 +611,9 @@ function sameTranscriptIdentity( |
| 560 | 611 | |
| 561 | 612 | export const make = Effect.gen(function* () { |
| 562 | 613 | const fileSystem = yield* FileSystem.FileSystem; |
| 614 | // Different project imports can arrive concurrently from multiple clients. |
| 615 | // Only one transcript may hold its selected-history budget at a time. |
| 616 | const importReadLock = yield* Semaphore.make(1); |
| 563 | 617 | const path = yield* Path.Path; |
| 564 | 618 | const serverConfig = yield* ServerConfig.ServerConfig; |
| 565 | 619 | const serverSettings = yield* ServerSettings.ServerSettingsService; |
| ... | ... | @@ -692,10 +746,16 @@ export const make = Effect.gen(function* () { |
| 692 | 746 | ).pipe(Effect.orElseSucceed(() => null)); |
| 693 | 747 | }); |
| 694 | 748 | |
| 695 | | /** Check the open file before and after reading, without reading past its reserved byte budget. */ |
| 749 | /** |
| 750 | * Project history fields while reading, before allocating whole JSON records. |
| 751 | * Check the file identity on both sides of the read. A selected-history budget |
| 752 | * failure rejects the entire transcript before any imported messages persist. |
| 753 | */ |
| 696 | 754 | const readTranscript = Effect.fn("AgentSessionScanner.readTranscript")(function* ( |
| 697 | 755 | filePath: string, |
| 698 | 756 | expected: ReturnType<typeof transcriptIdentity>, |
| 757 | recordLimit: number, |
| 758 | source: AgentSessionSource, |
| 699 | 759 | ) { |
| 700 | 760 | if (expected.size > MAX_IMPORTED_TRANSCRIPT_BYTES) return null; |
| 701 | 761 | |
| ... | ... | @@ -706,9 +766,38 @@ export const make = Effect.gen(function* () { |
| 706 | 766 | if (!sameTranscriptIdentity(expected, transcriptIdentity(filePath, yield* file.stat))) { |
| 707 | 767 | return null; |
| 708 | 768 | } |
| 709 | | const decoder = new TextDecoder(); |
| 710 | | let contents = ""; |
| 769 | const records: Array<DecodedTranscriptRecord> = []; |
| 770 | let historyBytes = 0; |
| 771 | let recordBytes = 0; |
| 772 | let recordCount = 0; |
| 711 | 773 | let bytesRead = 0; |
| 774 | const reserve = (bytes: number) => { |
| 775 | recordBytes += bytes; |
| 776 | if (historyBytes + recordBytes > MAX_IMPORT_HISTORY_BYTES) { |
| 777 | throw new TranscriptJsonLimitError( |
| 778 | "Transcript selected history exceeds the 32 MiB memory budget", |
| 779 | ); |
| 780 | } |
| 781 | }; |
| 782 | let reader = createTranscriptJsonReader(reserve, selectTranscriptPath); |
| 783 | let decoder = new TextDecoder(); |
| 784 | let recordStarted = false; |
| 785 | |
| 786 | const finishRecord = () => { |
| 787 | reader.write(decoder.decode()); |
| 788 | recordCount += 1; |
| 789 | if (recordCount > recordLimit) return false; |
| 790 | const decoded = decodeTranscriptValue(reader.finish()); |
| 791 | if (Option.isSome(decoded) && shouldRetainDecodedRecord(source, decoded.value)) { |
| 792 | records.push(decoded.value); |
| 793 | historyBytes += recordBytes; |
| 794 | } |
| 795 | recordBytes = 0; |
| 796 | reader = createTranscriptJsonReader(reserve, selectTranscriptPath); |
| 797 | decoder = new TextDecoder(); |
| 798 | recordStarted = false; |
| 799 | return true; |
| 800 | }; |
| 712 | 801 | |
| 713 | 802 | while (bytesRead < expected.size) { |
| 714 | 803 | const next = yield* file.readAlloc( |
| ... | ... | @@ -719,16 +808,36 @@ export const make = Effect.gen(function* () { |
| 719 | 808 | } |
| 720 | 809 | |
| 721 | 810 | bytesRead += next.value.byteLength; |
| 722 | | contents += decoder.decode(next.value, { stream: true }); |
| 811 | const withinBudget = yield* Effect.try(() => { |
| 812 | let start = 0; |
| 813 | while (start < next.value.byteLength) { |
| 814 | const newline = next.value.indexOf(10, start); |
| 815 | const end = newline === -1 ? next.value.byteLength : newline; |
| 816 | recordStarted = true; |
| 817 | reader.write(decoder.decode(next.value.subarray(start, end), { stream: true })); |
| 818 | if (newline === -1) break; |
| 819 | if (!finishRecord()) return false; |
| 820 | start = newline + 1; |
| 821 | } |
| 822 | return true; |
| 823 | }); |
| 824 | if (!withinBudget) return null; |
| 723 | 825 | } |
| 724 | 826 | |
| 827 | if (recordStarted && !(yield* Effect.try(finishRecord))) return null; |
| 725 | 828 | return sameTranscriptIdentity(expected, transcriptIdentity(filePath, yield* file.stat)) |
| 726 | | ? contents + decoder.decode() |
| 829 | ? { records, recordCount } |
| 727 | 830 | : null; |
| 728 | 831 | }), |
| 729 | 832 | ), |
| 730 | 833 | ), |
| 731 | | ).pipe(Effect.orElseSucceed(() => null)); |
| 834 | ).pipe( |
| 835 | Effect.catch((cause) => |
| 836 | Effect.logWarning("Could not read imported transcript", { filePath, cause }).pipe( |
| 837 | Effect.as(null), |
| 838 | ), |
| 839 | ), |
| 840 | ); |
| 732 | 841 | }); |
| 733 | 842 | |
| 734 | 843 | /** |
| ... | ... | @@ -1238,20 +1347,21 @@ export const make = Effect.gen(function* () { |
| 1238 | 1347 | // Reserve the whole file even if its read or parse fails. |
| 1239 | 1348 | transcriptsRemaining -= 1; |
| 1240 | 1349 | bytesRemaining -= identity.size; |
| 1241 | | const contents = yield* readTranscript(transcript.filePath, identity); |
| 1242 | | if (contents === null) { |
| 1243 | | return Option.some<AgentSessionRecentThread>({ _tag: "Skipped" }); |
| 1244 | | } |
| 1245 | | const lines = splitTranscriptRecords(contents, recordsRemaining + 1); |
| 1246 | | if (lines.length > recordsRemaining) { |
| 1350 | const snapshot = yield* readTranscript( |
| 1351 | transcript.filePath, |
| 1352 | identity, |
| 1353 | recordsRemaining, |
| 1354 | candidate.source, |
| 1355 | ); |
| 1356 | if (snapshot === null) { |
| 1247 | 1357 | return Option.some<AgentSessionRecentThread>({ _tag: "Skipped" }); |
| 1248 | 1358 | } |
| 1249 | | recordsRemaining -= lines.length; |
| 1359 | recordsRemaining -= snapshot.recordCount; |
| 1250 | 1360 | |
| 1251 | 1361 | // A stable replacement file can belong to a different project than the cached candidate. |
| 1252 | 1362 | let snapshotCwd: string | null = null; |
| 1253 | | for (const line of lines) { |
| 1254 | | snapshotCwd = extractCwd(line); |
| 1363 | for (const record of snapshot.records) { |
| 1364 | snapshotCwd = extractDecodedCwd(record); |
| 1255 | 1365 | if (snapshotCwd !== null) break; |
| 1256 | 1366 | } |
| 1257 | 1367 | if (snapshotCwd === null) { |
| ... | ... | @@ -1265,15 +1375,14 @@ export const make = Effect.gen(function* () { |
| 1265 | 1375 | return Option.some<AgentSessionRecentThread>({ _tag: "Skipped" }); |
| 1266 | 1376 | } |
| 1267 | 1377 | |
| 1268 | | const parsedThread = parseAgentSessionTranscript( |
| 1378 | const parsedThread = parseAgentSessionRecords( |
| 1269 | 1379 | { |
| 1270 | | contents, |
| 1271 | 1380 | source: candidate.source, |
| 1272 | 1381 | providerInstanceId: candidate.providerInstanceId, |
| 1273 | 1382 | fallbackSessionId: path.basename(transcript.filePath, ".jsonl"), |
| 1274 | 1383 | lastActiveAtMs: transcript.mtimeMs, |
| 1275 | 1384 | }, |
| 1276 | | lines, |
| 1385 | snapshot.records, |
| 1277 | 1386 | ); |
| 1278 | 1387 | if (parsedThread === null) { |
| 1279 | 1388 | return Option.some<AgentSessionRecentThread>({ _tag: "Skipped" }); |
| ... | ... | @@ -1295,7 +1404,7 @@ export const make = Effect.gen(function* () { |
| 1295 | 1404 | thread: parsedThread, |
| 1296 | 1405 | source, |
| 1297 | 1406 | }); |
| 1298 | | }), |
| 1407 | }).pipe(importReadLock.withPermits(1)), |
| 1299 | 1408 | ), |
| 1300 | 1409 | Stream.map(Option.toArray), |
| 1301 | 1410 | Stream.flattenIterable, |