From 2b6a3c68c30a5c785ee833ce42cfdae5dfc2a0cd Mon Sep 17 00:00:00 2001 From: Deepak Chauhan Date: Tue, 15 Sep 2026 21:45:09 +0530 Subject: [PATCH 1/2] fix(server): preserve Codex titles on import Use Codex session index titles when available, fall back past injected context, and repair legacy placeholder titles during a repeated import. Closes #10513 --- .../src/project/AgentSessionImporter.test.ts | 80 +++++++++++++ .../src/project/AgentSessionImporter.ts | 41 +++++++ .../src/project/AgentSessionScanner.test.ts | 106 +++++++++++++++++- .../server/src/project/AgentSessionScanner.ts | 91 ++++++++++++++- 4 files changed, 310 insertions(+), 8 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 4eb03a5cc036..7de1df5446d3 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -318,6 +318,86 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { }), ); + it.effect("repairs legacy context titles on already imported Codex threads", () => + Effect.gen(function* () { + const thread = makeThread("codex"); + const source = makeThreadOutcome(thread).source; + const threadId = ThreadId.make( + `import:${source.providerInstanceId}:${source.providerSessionId}`, + ); + const commands: Array = []; + const scanner = AgentSessionScanner.AgentSessionScanner.of({ + scan: Effect.die("unused"), + recentThreads: () => + Stream.succeed({ + _tag: "AlreadyImported", + source, + canonicalTitle: "Prototype MetaApi trade replication", + }), + }); + const engine = OrchestrationEngine.OrchestrationEngineService.of({ + dispatch: (command) => Effect.sync(() => ({ sequence: commands.push(command) })), + readEvents: () => Stream.empty, + readThreadEvents: () => Stream.empty, + getThreadReplayStats: () => Effect.die("unused"), + streamDomainEvents: Stream.empty, + subscribeDomainEvents: Effect.succeed(Stream.empty), + latestSequence: Effect.succeed(0), + }); + const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ + upsert: () => Effect.die("must not replace an existing binding"), + getProvider: () => Effect.die("unused"), + recordImportedTranscript: () => Effect.die("must not rewrite completed history"), + getBinding: () => Effect.die("must not read a completed binding"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.die("unused"), + }); + + const result = yield* runImport({ + scanner, + engine, + directory, + snapshots: makeSnapshotsLayer({ + project: makeProject(), + getThread: (id) => + id === threadId + ? Option.some({ + ...makeProjectedThread({ source: "codex", imported: true }), + title: "", + }) + : Option.none(), + }), + }); + + expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); + expect(commands).toMatchObject([ + { + type: "thread.meta.update", + threadId, + title: "Prototype MetaApi trade replication", + }, + ]); + + commands.length = 0; + expect( + yield* runImport({ + scanner, + engine, + directory, + snapshots: makeSnapshotsLayer({ + project: makeProject(), + getThread: () => + Option.some({ + ...makeProjectedThread({ source: "codex", imported: true }), + title: "My custom thread title", + }), + }), + }), + ).toEqual({ importedCount: 1, skippedCount: 0 }); + expect(commands).toEqual([]); + }), + ); + it.effect("counts scanner skips without writing a thread or binding", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index 5ebb41a1bb54..e33a2b8976ba 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -70,6 +70,14 @@ function hasImportedHistory(thread: OrchestrationThread): boolean { return thread.messages.some((message) => isImportedAgentSessionMessageId(message.id)); } +function hasLegacyCodexContextTitle(title: string): boolean { + return ( + title === "" || + title === "# AGENTS.md instructions" || + title === "" + ); +} + function hasImportBlockingActivity( thread: OrchestrationThread, importedHistoryPresent: boolean, @@ -136,6 +144,26 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu let importedCount = 0; let skippedCount = 0; + const repairImportedCodexTitle = Effect.fn("repairImportedCodexTitle")(function* ( + threadId: ThreadId, + canonicalTitle: string, + ) { + const existingThread = yield* snapshots.getThreadDetailById(threadId); + if ( + Option.isNone(existingThread) || + !hasLegacyCodexContextTitle(existingThread.value.title) || + existingThread.value.title === canonicalTitle + ) { + return; + } + yield* engine.dispatch({ + type: "thread.meta.update", + commandId: CommandId.make(yield* crypto.randomUUIDv4), + threadId, + title: canonicalTitle, + }); + }); + yield* Stream.runForEach(threads, (outcome) => Effect.gen(function* () { if (outcome._tag === "Skipped") { @@ -147,6 +175,16 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu `import:${outcome.source.providerInstanceId}:${outcome.source.providerSessionId}`, ); if (outcome._tag === "AlreadyImported") { + if (outcome.source.provider === "codex" && outcome.canonicalTitle !== null) { + yield* repairImportedCodexTitle(threadId, outcome.canonicalTitle).pipe( + Effect.catch((cause) => + Effect.logWarning("Could not repair an imported Codex thread title", { + threadId, + cause, + }), + ), + ); + } importedThreadIds.add(threadId); importedCount += 1; } else if (importedThreadIds.has(threadId)) { @@ -199,6 +237,9 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu importedHistoryPresent && Option.isSome(existingBinding) ) { + if (thread.source === "codex") { + yield* repairImportedCodexTitle(threadId, thread.title); + } yield* directory.recordImportedTranscript({ threadId, source: outcome.source }); return true; } diff --git a/apps/server/src/project/AgentSessionScanner.test.ts b/apps/server/src/project/AgentSessionScanner.test.ts index b1cba460f959..373fceeaed71 100644 --- a/apps/server/src/project/AgentSessionScanner.test.ts +++ b/apps/server/src/project/AgentSessionScanner.test.ts @@ -2606,10 +2606,76 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => { ).toEqual(["recent-session"]); }), ); + + it.effect("uses the canonical Codex title from the session index", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const fileSystem = yield* FileSystem.FileSystem; + const nowMs = Date.parse("2026-08-24T12:00:00.000Z"); + const sessionId = "01a0a0b4-3958-7382-82b2-b22b8bb830ba"; + yield* TestClock.setTime(nowMs); + const claudeHomePath = yield* makeTempDir("t3code-index-title-claude-"); + const codexHomePath = yield* makeTempDir("t3code-index-title-codex-"); + const workspaceRoot = yield* makeTempDir("t3code-index-title-workspace-"); + + yield* writeTranscript({ + filePath: path.join( + codexHomePath, + "sessions", + "2026", + "08", + "24", + `rollout-2026-08-24T12-00-00-${sessionId}.jsonl`, + ), + contents: [ + encodeTranscriptRecord({ + type: "session_meta", + payload: { id: sessionId, cwd: workspaceRoot }, + }), + encodeTranscriptRecord({ + type: "response_item", + payload: { + type: "message", + role: "user", + content: [{ type: "input_text", text: "" }], + }, + }), + ].join("\n"), + mtimeMs: nowMs, + }); + yield* fileSystem.writeFileString( + path.join(codexHomePath, "session_index.jsonl"), + [ + "not json", + JSON.stringify({ id: sessionId, thread_name: "Prototype MetaApi trade replication" }), + ].join("\n"), + ); + + const threads = yield* runRecentThreads({ + claudeHomePath, + codexHomePath, + workspaceRoot, + }); + + expect(threads[0]?.title).toBe("Prototype MetaApi trade replication"); + }), + ); }); }); describe("parseAgentSessionTranscript", () => { + it("parses valid, named Codex session index entries", () => { + expect( + AgentSessionScanner.parseCodexSessionIndex( + [ + "not json", + JSON.stringify({ id: "session-1", thread_name: " Fix imported titles " }), + JSON.stringify({ id: "session-2", thread_name: "" }), + ].join("\n"), + ), + ).toEqual(new Map([["session-1", "Fix imported titles"]])); + }); + it.each([false, true])( "handles the exact record limit and an interior blank overflow=%s", (overflow) => { @@ -3099,13 +3165,49 @@ describe("parseAgentSessionTranscript", () => { lastActiveAtMs: Date.parse("2026-08-25T08:00:00.000Z"), }); - expect(thread?.title).toBe(""); + expect(thread?.title).toBe("Initialize Git and add a README."); expect(thread?.messages.map((message) => message.text)).toEqual([ context, "Initialize Git and add a README.", ]); }); + it("skips recommended plugin context when deriving a Codex title", () => { + const plugins = + "\n- GitHub (github@openai-curated-remote)\n"; + const thread = AgentSessionScanner.parseAgentSessionTranscript({ + contents: [ + encodeTranscriptRecord({ type: "session_meta", payload: { id: "codex-session" } }), + encodeTranscriptRecord({ + type: "response_item", + payload: { + type: "message", + role: "user", + content: [{ type: "input_text", text: plugins }], + }, + }), + encodeTranscriptRecord({ + type: "response_item", + payload: { + type: "message", + role: "user", + content: [{ type: "input_text", text: "Build MetaApi trade replication" }], + }, + }), + ].join("\n"), + source: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + fallbackSessionId: "fallback", + lastActiveAtMs: Date.parse("2026-08-25T08:00:00.000Z"), + }); + + expect(thread?.title).toBe("Build MetaApi trade replication"); + expect(thread?.messages.map((message) => message.text)).toEqual([ + plugins, + "Build MetaApi trade replication", + ]); + }); + it("preserves a canonical Codex event that starts with context markup", () => { const prompt = "\n/tmp/project\n\n\nCreate a useful project."; @@ -3126,7 +3228,7 @@ describe("parseAgentSessionTranscript", () => { lastActiveAtMs: Date.parse("2026-08-25T08:00:00.000Z"), }); - expect(thread?.title).toBe(""); + expect(thread?.title).toBe("Create a useful project."); expect(thread?.messages.map((message) => message.text)).toEqual([prompt]); }); diff --git a/apps/server/src/project/AgentSessionScanner.ts b/apps/server/src/project/AgentSessionScanner.ts index 975192e70033..1395331c90e7 100644 --- a/apps/server/src/project/AgentSessionScanner.ts +++ b/apps/server/src/project/AgentSessionScanner.ts @@ -94,6 +94,7 @@ 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; +const MAX_CODEX_SESSION_INDEX_BYTES = 16 * 1024 * 1024; const TranscriptContentBlock = Schema.Struct({ type: Schema.optional(Schema.String), @@ -110,6 +111,11 @@ const CodexTurnMetadata = Schema.Struct({ turn_id: Schema.optional(Schema.Union([Schema.String, Schema.Null])), }); +const CodexSessionIndexEntry = Schema.Struct({ + id: Schema.String, + thread_name: Schema.String, +}); + const TranscriptRecord = Schema.Struct({ type: Schema.optional(Schema.String), timestamp: Schema.optional(Schema.String), @@ -141,6 +147,9 @@ const decodeTranscriptRecord = Schema.decodeUnknownOption(Schema.fromJsonString( const decodeTranscriptValue = Schema.decodeUnknownOption(TranscriptRecord); const selectTranscriptPath = createTranscriptJsonSelector(TranscriptRecord); const decodeCodexTurnMetadata = Schema.decodeUnknownOption(CodexTurnMetadata); +const decodeCodexSessionIndexEntry = Schema.decodeUnknownOption( + Schema.fromJsonString(CodexSessionIndexEntry), +); type DecodedTranscriptRecord = typeof TranscriptRecord.Type; @@ -149,6 +158,7 @@ interface AgentSessionTranscriptMetadata { readonly providerInstanceId: ProviderInstanceId; readonly fallbackSessionId: string; readonly lastActiveAtMs: number; + readonly canonicalTitle?: string; } export interface AgentSessionThreadMessage { @@ -174,7 +184,11 @@ export type AgentSessionRecentThread = readonly thread: AgentSessionThread; readonly source: AgentSessionImportSource; } - | { readonly _tag: "AlreadyImported"; readonly source: AgentSessionImportSource } + | { + readonly _tag: "AlreadyImported"; + readonly source: AgentSessionImportSource; + readonly canonicalTitle: string | null; + } | { readonly _tag: "Duplicate"; readonly source: AgentSessionImportSource } | { readonly _tag: "Skipped" }; @@ -208,6 +222,7 @@ interface RawCandidate { readonly transcripts: ReadonlyArray<{ readonly filePath: string; readonly mtimeMs: number | null; + readonly canonicalTitle: string | null; }>; } @@ -216,6 +231,19 @@ interface TranscriptCandidate { readonly mtimeMs: number; readonly providerInstanceId: ProviderInstanceId; readonly size: number; + readonly canonicalTitle: string | null; +} + +export function parseCodexSessionIndex(contents: string): ReadonlyMap { + const titles = new Map(); + for (const line of contents.split("\n")) { + const decoded = decodeCodexSessionIndexEntry(line); + if (Option.isNone(decoded)) continue; + const id = decoded.value.id.trim(); + const title = decoded.value.thread_name.trim(); + if (id.length > 0 && title.length > 0) titles.set(id, title); + } + return titles; } interface MetadataReadBudget { @@ -283,6 +311,29 @@ function codexTurnId(metadata: unknown): string | null { return decoded.value.turn_id; } +function deriveImportedThreadTitle(text: string): string | null { + const injectedContextPatterns = [ + /^[\s\S]*?<\/recommended_plugins>\s*/, + /^[\s\S]*?<\/environment_context>\s*/, + /^[\s\S]*?<\/user_instructions>\s*/, + /^# AGENTS\.md instructions[^\n]*(?:\n+[\s\S]*?<\/INSTRUCTIONS>)?\s*/, + ]; + let visibleText = text.trimStart(); + let removedContext = true; + while (removedContext) { + removedContext = false; + for (const pattern of injectedContextPatterns) { + const withoutContext = visibleText.replace(pattern, ""); + if (withoutContext === visibleText) continue; + visibleText = withoutContext.trimStart(); + removedContext = true; + break; + } + } + const firstLine = visibleText.split("\n")[0]?.slice(0, 100).trim(); + return firstLine && firstLine.length > 0 ? firstLine : null; +} + /** Keep visible user and assistant text while ignoring tools, reasoning, and malformed records. */ export function parseAgentSessionTranscript( input: AgentSessionTranscriptMetadata & { @@ -303,13 +354,14 @@ function parseAgentSessionRecords( // Claude filenames are session IDs. Codex rollout filenames include extra // timestamp text, so only transcript metadata can provide a resumable ID. let providerSessionId = input.source === "codex" ? "" : input.fallbackSessionId; - let title: string | null = null; + let title = input.canonicalTitle?.trim() || null; let model: string | null = null; let hasCodexSessionId = false; const messages: Array = []; let firstUserMessage: | (AgentSessionThreadMessage & { readonly codexResponseUser: boolean }) | undefined; + let firstDerivedTitle: string | null = null; // 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. @@ -372,6 +424,9 @@ function parseAgentSessionRecords( if (firstUserMessage === undefined && message.role === "user") { firstUserMessage = message; } + if (firstDerivedTitle === null && message.role === "user") { + firstDerivedTitle = deriveImportedThreadTitle(message.text); + } messages.push(message); if (messages.length > MAX_IMPORTED_MESSAGES) messages.shift(); }; @@ -492,13 +547,11 @@ function parseAgentSessionRecords( const retainedMessages = firstUserMessageRetained ? visibleMessages : [visibleFirstUserMessage, ...visibleMessages.slice(-(MAX_IMPORTED_MESSAGES - 1))]; - const derivedTitle = visibleFirstUserMessage.text.trim().split("\n")[0]?.slice(0, 100).trim(); - return { source: input.source, providerInstanceId: input.providerInstanceId, providerSessionId, - title: title ?? (derivedTitle && derivedTitle.length > 0 ? derivedTitle : "Imported thread"), + title: title ?? firstDerivedTitle ?? "Imported thread", model, createdAt: retainedMessages[0]?.createdAt ?? fallbackTimestamp, updatedAt: fallbackTimestamp, @@ -966,6 +1019,7 @@ export const make = Effect.gen(function* () { mtimeMs: stats.value.mtime.value.getTime(), providerInstanceId, size: Number(stats.value.size), + canonicalTitle: null, }); } } @@ -976,6 +1030,17 @@ export const make = Effect.gen(function* () { const discoverCodexTranscripts = Effect.fn("AgentSessionScanner.discoverCodexTranscripts")( function* (homePath: string, providerInstanceId: ProviderInstanceId, operationBudget: number) { const sessionsDir = path.join(homePath, "sessions"); + const indexPath = path.join(homePath, "session_index.jsonl"); + const indexStats = yield* statOption(indexPath); + const indexedTitles = + Option.isSome(indexStats) && + indexStats.value.type === "File" && + Number(indexStats.value.size) <= MAX_CODEX_SESSION_INDEX_BYTES + ? yield* fileSystem.readFileString(indexPath).pipe( + Effect.map(parseCodexSessionIndex), + Effect.orElseSucceed(() => new Map()), + ) + : new Map(); const transcripts: Array = []; let operationsRemaining = operationBudget; @@ -1029,6 +1094,12 @@ export const make = Effect.gen(function* () { mtimeMs: stats.value.mtime.value.getTime(), providerInstanceId, size: Number(stats.value.size), + canonicalTitle: + indexedTitles.get( + entry.match( + /([0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12})\.jsonl$/i, + )?.[1] ?? "", + ) ?? null, }); } } @@ -1050,7 +1121,11 @@ export const make = Effect.gen(function* () { cwd: string; providerInstanceId: ProviderInstanceId; lastActiveAtMs: number; - transcripts: Array<{ filePath: string; mtimeMs: number }>; + transcripts: Array<{ + filePath: string; + mtimeMs: number; + canonicalTitle: string | null; + }>; } >(); @@ -1407,6 +1482,7 @@ export const make = Effect.gen(function* () { return Option.some({ _tag: "AlreadyImported", source: completedSource, + canonicalTitle: transcript.canonicalTitle, }); } if ( @@ -1454,6 +1530,9 @@ export const make = Effect.gen(function* () { providerInstanceId: candidate.providerInstanceId, fallbackSessionId: path.basename(transcript.filePath, ".jsonl"), lastActiveAtMs: transcript.mtimeMs, + ...(transcript.canonicalTitle === null + ? {} + : { canonicalTitle: transcript.canonicalTitle }), }, snapshot.records, ); From dcfd375ca7db1a8e807d6d1d6379b48aa8eca885 Mon Sep 17 00:00:00 2001 From: Deepak Chauhan Date: Wed, 16 Sep 2026 01:39:42 +0530 Subject: [PATCH 2/2] fix(server): harden Codex title repair Bound session-index reads, preserve concurrent manual renames, recover fallback titles from imported history, and isolate repair failures. --- .../src/project/AgentSessionImporter.test.ts | 179 +++++++++++++++++- .../src/project/AgentSessionImporter.ts | 45 ++++- .../src/project/AgentSessionScanner.test.ts | 159 +++++++++++++++- .../server/src/project/AgentSessionScanner.ts | 143 +++++++++++--- 4 files changed, 482 insertions(+), 44 deletions(-) diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 7de1df5446d3..5bc4fb2678e0 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -372,9 +372,36 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); expect(commands).toMatchObject([ { - type: "thread.meta.update", + type: "thread.title.generate.complete", threadId, title: "Prototype MetaApi trade replication", + expectedTitle: "", + expectedVersion: null, + needsRefinement: false, + }, + ]); + + commands.length = 0; + expect( + yield* runImport({ + scanner, + engine, + directory, + snapshots: makeSnapshotsLayer({ + project: makeProject(), + getThread: () => + Option.some({ + ...makeProjectedThread({ source: "codex", imported: true }), + title: "# AGENTS.md instructions for /tmp/project", + }), + }), + }), + ).toEqual({ importedCount: 1, skippedCount: 0 }); + expect(commands).toMatchObject([ + { + type: "thread.title.generate.complete", + expectedTitle: "# AGENTS.md instructions for /tmp/project", + title: "Prototype MetaApi trade replication", }, ]); @@ -390,6 +417,11 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { Option.some({ ...makeProjectedThread({ source: "codex", imported: true }), title: "My custom thread title", + titleState: { + source: "manual", + version: CommandId.make("manual-title"), + needsRefinement: false, + }, }), }), }), @@ -398,6 +430,151 @@ it.layer(NodeServices.layer)("AgentSessionImporter", (it) => { }), ); + it.effect( + "repairs a legacy title from imported history when the Codex index has no title", + () => + Effect.gen(function* () { + const thread = makeThread("codex"); + const source = makeThreadOutcome(thread).source; + const threadId = ThreadId.make( + `import:${source.providerInstanceId}:${source.providerSessionId}`, + ); + const commands: Array = []; + const scanner = AgentSessionScanner.AgentSessionScanner.of({ + scan: Effect.die("unused"), + recentThreads: () => + Stream.succeed({ _tag: "AlreadyImported", source, canonicalTitle: null }), + }); + const engine = OrchestrationEngine.OrchestrationEngineService.of({ + dispatch: (command) => Effect.sync(() => ({ sequence: commands.push(command) })), + readEvents: () => Stream.empty, + readThreadEvents: () => Stream.empty, + getThreadReplayStats: () => Effect.die("unused"), + streamDomainEvents: Stream.empty, + subscribeDomainEvents: Effect.succeed(Stream.empty), + latestSequence: Effect.succeed(0), + }); + const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ + upsert: () => Effect.die("unused"), + getProvider: () => Effect.die("unused"), + recordImportedTranscript: () => Effect.die("must not rewrite completed history"), + getBinding: () => Effect.die("unused"), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.die("unused"), + }); + const projected = makeProjectedThread({ source: "codex", imported: true }); + + const result = yield* runImport({ + scanner, + engine, + directory, + snapshots: makeSnapshotsLayer({ + project: makeProject(), + getThread: () => + Option.some({ + ...projected, + title: "", + messages: [ + { + ...projected.messages[0]!, + text: "\nInternal setup instructions", + }, + { + ...projected.messages[0]!, + id: MessageId.make("user-followup"), + text: "Do not use this follow-up", + }, + { + ...projected.messages[0]!, + id: MessageId.make(`${threadId}:000001`), + text: "Recovered from imported history", + }, + ], + }), + }), + }); + + expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); + expect(commands).toMatchObject([ + { + type: "thread.title.generate.complete", + threadId, + expectedTitle: "", + title: "Recovered from imported history", + needsRefinement: false, + }, + ]); + }), + ); + + it.effect("records an existing import when Codex title repair fails", () => + Effect.gen(function* () { + const thread = makeThread("codex"); + const outcome = makeThreadOutcome(thread); + const threadId = ThreadId.make( + `import:${thread.providerInstanceId}:${thread.providerSessionId}`, + ); + let recorded = 0; + const scanner = AgentSessionScanner.AgentSessionScanner.of({ + scan: Effect.die("unused"), + recentThreads: () => Stream.succeed(outcome), + }); + const engine = OrchestrationEngine.OrchestrationEngineService.of({ + dispatch: (command) => + command.type === "thread.title.generate.complete" + ? Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: "Temporary title repair failure.", + }), + ) + : Effect.die("must not dispatch another command"), + readEvents: () => Stream.empty, + readThreadEvents: () => Stream.empty, + getThreadReplayStats: () => Effect.die("unused"), + streamDomainEvents: Stream.empty, + subscribeDomainEvents: Effect.succeed(Stream.empty), + latestSequence: Effect.succeed(0), + }); + const directory = ProviderSessionDirectory.ProviderSessionDirectory.of({ + upsert: () => Effect.die("must not replace the existing binding"), + getProvider: () => Effect.die("unused"), + recordImportedTranscript: () => Effect.sync(() => void (recorded += 1)), + getBinding: () => + Effect.succeed( + Option.some({ + threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex"), + status: "stopped" as const, + resumeCursor: { threadId: thread.providerSessionId }, + runtimeMode: "full-access" as const, + runtimePayload: { cwd: WORKSPACE_ROOT }, + }), + ), + listThreadIds: () => Effect.die("unused"), + listBindings: () => Effect.die("unused"), + }); + + const result = yield* runImport({ + scanner, + engine, + directory, + snapshots: makeSnapshotsLayer({ + project: makeProject(), + getThread: () => + Option.some({ + ...makeProjectedThread({ source: "codex", imported: true }), + title: "", + }), + }), + }); + + expect(result).toEqual({ importedCount: 1, skippedCount: 0 }); + expect(recorded).toBe(1); + }), + ); + it.effect("counts scanner skips without writing a thread or binding", () => Effect.gen(function* () { const scanner = AgentSessionScanner.AgentSessionScanner.of({ diff --git a/apps/server/src/project/AgentSessionImporter.ts b/apps/server/src/project/AgentSessionImporter.ts index e33a2b8976ba..e0176a6c6b2b 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -70,14 +70,26 @@ function hasImportedHistory(thread: OrchestrationThread): boolean { return thread.messages.some((message) => isImportedAgentSessionMessageId(message.id)); } +/** Identify titles produced by the old first-line Codex import fallback. */ function hasLegacyCodexContextTitle(title: string): boolean { return ( title === "" || - title === "# AGENTS.md instructions" || - title === "" + title === "" || + title === "" || + /^# AGENTS\.md instructions(?:\s|$)/.test(title) ); } +/** Recover a fallback title from the imported user messages already in the projection. */ +function deriveImportedCodexTitle(thread: OrchestrationThread): string | null { + for (const message of thread.messages) { + if (message.role !== "user" || !isImportedAgentSessionMessageId(message.id)) continue; + const title = AgentSessionScanner.deriveImportedThreadTitle(message.text, "codex"); + if (title !== null) return title; + } + return null; +} + function hasImportBlockingActivity( thread: OrchestrationThread, importedHistoryPresent: boolean, @@ -146,21 +158,27 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu const repairImportedCodexTitle = Effect.fn("repairImportedCodexTitle")(function* ( threadId: ThreadId, - canonicalTitle: string, + canonicalTitle: string | null, ) { const existingThread = yield* snapshots.getThreadDetailById(threadId); + if (Option.isNone(existingThread)) return; + const existingTitle = existingThread.value.title; + const replacementTitle = canonicalTitle ?? deriveImportedCodexTitle(existingThread.value); if ( - Option.isNone(existingThread) || - !hasLegacyCodexContextTitle(existingThread.value.title) || - existingThread.value.title === canonicalTitle + replacementTitle === null || + !hasLegacyCodexContextTitle(existingTitle) || + existingTitle === replacementTitle ) { return; } yield* engine.dispatch({ - type: "thread.meta.update", + type: "thread.title.generate.complete", commandId: CommandId.make(yield* crypto.randomUUIDv4), threadId, - title: canonicalTitle, + title: replacementTitle, + expectedTitle: existingTitle, + expectedVersion: existingThread.value.titleState?.version ?? null, + needsRefinement: false, }); }); @@ -175,7 +193,7 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu `import:${outcome.source.providerInstanceId}:${outcome.source.providerSessionId}`, ); if (outcome._tag === "AlreadyImported") { - if (outcome.source.provider === "codex" && outcome.canonicalTitle !== null) { + if (outcome.source.provider === "codex") { yield* repairImportedCodexTitle(threadId, outcome.canonicalTitle).pipe( Effect.catch((cause) => Effect.logWarning("Could not repair an imported Codex thread title", { @@ -238,7 +256,14 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu Option.isSome(existingBinding) ) { if (thread.source === "codex") { - yield* repairImportedCodexTitle(threadId, thread.title); + yield* repairImportedCodexTitle(threadId, thread.title).pipe( + Effect.catch((cause) => + Effect.logWarning("Could not repair an imported Codex thread title", { + threadId, + cause, + }), + ), + ); } yield* directory.recordImportedTranscript({ threadId, source: outcome.source }); return true; diff --git a/apps/server/src/project/AgentSessionScanner.test.ts b/apps/server/src/project/AgentSessionScanner.test.ts index 373fceeaed71..c9a2d5ece58a 100644 --- a/apps/server/src/project/AgentSessionScanner.test.ts +++ b/apps/server/src/project/AgentSessionScanner.test.ts @@ -2647,7 +2647,10 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => { path.join(codexHomePath, "session_index.jsonl"), [ "not json", - JSON.stringify({ id: sessionId, thread_name: "Prototype MetaApi trade replication" }), + encodeTranscriptRecord({ + id: sessionId, + thread_name: "Prototype MetaApi trade replication", + }), ].join("\n"), ); @@ -2660,6 +2663,83 @@ it.layer(NodeServices.layer)("AgentSessionScanner", (it) => { expect(threads[0]?.title).toBe("Prototype MetaApi trade replication"); }), ); + + it.effect("bounds index growth after the opened-handle stat", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const fileSystem = yield* FileSystem.FileSystem; + const nowMs = Date.parse("2026-08-24T12:00:00.000Z"); + const sessionId = "01a0a0b4-3958-7382-82b2-b22b8bb830ba"; + yield* TestClock.setTime(nowMs); + const claudeHomePath = yield* makeTempDir("t3code-index-race-claude-"); + const codexHomePath = yield* makeTempDir("t3code-index-race-codex-"); + const workspaceRoot = yield* makeTempDir("t3code-index-race-workspace-"); + const transcriptPath = path.join( + codexHomePath, + "sessions", + "2026", + "08", + "24", + `rollout-2026-08-24T12-00-00-${sessionId}.jsonl`, + ); + const indexPath = path.join(codexHomePath, "session_index.jsonl"); + + yield* writeTranscript({ + filePath: transcriptPath, + contents: [ + encodeTranscriptRecord({ + type: "session_meta", + payload: { id: sessionId, cwd: workspaceRoot }, + }), + encodeTranscriptRecord({ + type: "event_msg", + payload: { type: "user_message", message: "Use the transcript title" }, + }), + ].join("\n"), + mtimeMs: nowMs, + }); + yield* fileSystem.writeFileString( + indexPath, + `${encodeTranscriptRecord({ id: sessionId, thread_name: "Unsafe index title" })}\n${"x".repeat(16 * 1024 * 1024)}`, + ); + let indexReads = 0; + let indexBytesRequested = 0; + const simulatedFileSystem = FileSystem.FileSystem.of({ + ...fileSystem, + stat: (target) => fileSystem.stat(target === indexPath ? transcriptPath : target), + open: (target, options) => + fileSystem.open(target, options).pipe( + Effect.map((file) => + target === indexPath + ? new Proxy(file, { + get(inner, key) { + if (key === "stat") return fileSystem.stat(transcriptPath); + if (key === "readAlloc") { + return (size: FileSystem.SizeInput) => { + indexReads += 1; + indexBytesRequested += Number(size); + return inner.readAlloc(size); + }; + } + return Reflect.get(inner, key, inner); + }, + }) + : file, + ), + ), + }); + + const threads = yield* runRecentThreads({ + claudeHomePath, + codexHomePath, + workspaceRoot, + }).pipe(Effect.provideService(FileSystem.FileSystem, simulatedFileSystem)); + + expect(threads[0]?.title).toBe("Use the transcript title"); + expect(indexReads).toBeGreaterThan(0); + expect(indexBytesRequested).toBeLessThanOrEqual(16 * 1024 * 1024 + 1); + }), + ); }); }); @@ -2671,9 +2751,28 @@ describe("parseAgentSessionTranscript", () => { "not json", JSON.stringify({ id: "session-1", thread_name: " Fix imported titles " }), JSON.stringify({ id: "session-2", thread_name: "" }), - ].join("\n"), + JSON.stringify({ id: "session-3", thread_name: ` ${"x".repeat(150)} ` }), + JSON.stringify({ id: "session-1", thread_name: "Latest imported title" }), + ].join("\r\n") + "\r\n", ), - ).toEqual(new Map([["session-1", "Fix imported titles"]])); + ).toEqual( + new Map([ + ["session-1", "Latest imported title"], + ["session-3", "x".repeat(100)], + ]), + ); + }); + + it("bounds the session index map while retaining its newest entries", () => { + const titles = AgentSessionScanner.parseCodexSessionIndex( + Array.from({ length: 5_001 }, (_, index) => + encodeTranscriptRecord({ id: `session-${index}`, thread_name: `Title ${index}` }), + ).join("\n"), + ); + + expect(titles.size).toBe(5_000); + expect(titles.has("session-0")).toBe(false); + expect(titles.get("session-5000")).toBe("Title 5000"); }); it.each([false, true])( @@ -3208,6 +3307,60 @@ describe("parseAgentSessionTranscript", () => { ]); }); + it.each([ + "", + "\n- GitHub", + "\n/tmp/project", + "\nInternal setup instructions", + "# AGENTS.md instructions for /tmp/project\n\n\nInternal rules", + ])("ignores a bare or unclosed Codex context message: %s", (context) => { + const thread = AgentSessionScanner.parseAgentSessionTranscript({ + contents: [ + encodeTranscriptRecord({ type: "session_meta", payload: { id: "codex-session" } }), + encodeTranscriptRecord({ + type: "response_item", + payload: { + type: "message", + role: "user", + content: [{ type: "input_text", text: context }], + }, + }), + encodeTranscriptRecord({ + type: "event_msg", + payload: { type: "user_message", message: "Build the actual project" }, + }), + ].join("\n"), + source: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + fallbackSessionId: "fallback", + lastActiveAtMs: Date.parse("2026-08-25T08:00:00.000Z"), + }); + + expect(thread?.title).toBe("Build the actual project"); + }); + + it("strips stacked Codex context in one pass", () => { + const text = [ + "\n- GitHub\n", + "\n/tmp/project\n", + "# AGENTS.md instructions for /tmp/project\n\nRules\n", + "Build the actual project", + ].join("\n\n"); + + expect(AgentSessionScanner.deriveImportedThreadTitle(text, "codex")).toBe( + "Build the actual project", + ); + }); + + it("does not strip context-like text from Claude titles", () => { + expect( + AgentSessionScanner.deriveImportedThreadTitle( + "\nThis is a literal Claude request", + "claudeAgent", + ), + ).toBe(""); + }); + it("preserves a canonical Codex event that starts with context markup", () => { const prompt = "\n/tmp/project\n\n\nCreate a useful project."; diff --git a/apps/server/src/project/AgentSessionScanner.ts b/apps/server/src/project/AgentSessionScanner.ts index 1395331c90e7..3f7a43e0ec2d 100644 --- a/apps/server/src/project/AgentSessionScanner.ts +++ b/apps/server/src/project/AgentSessionScanner.ts @@ -95,6 +95,8 @@ const MAX_IMPORT_BYTES = 4 * 1024 * 1024 * 1024; const MAX_IMPORT_TRANSCRIPTS = 100; const MAX_IMPORT_RECORDS = 100_000; const MAX_CODEX_SESSION_INDEX_BYTES = 16 * 1024 * 1024; +const MAX_CODEX_SESSION_INDEX_ENTRIES = MAX_TRANSCRIPTS_PER_SOURCE; +const MAX_IMPORTED_THREAD_TITLE_CHARS = 100; const TranscriptContentBlock = Schema.Struct({ type: Schema.optional(Schema.String), @@ -234,16 +236,29 @@ interface TranscriptCandidate { readonly canonicalTitle: string | null; } +/** Parse the valid named entries from Codex's best-effort session index. */ export function parseCodexSessionIndex(contents: string): ReadonlyMap { const titles = new Map(); - for (const line of contents.split("\n")) { + let lineEnd = contents.length; + while (lineEnd > 0 && /[\r\n]/.test(contents[lineEnd - 1] ?? "")) lineEnd -= 1; + let entriesRead = 0; + while (lineEnd > 0 && entriesRead < MAX_CODEX_SESSION_INDEX_ENTRIES) { + const lineStart = contents.lastIndexOf("\n", lineEnd - 1) + 1; + const line = contents.slice(lineStart, lineEnd); + lineEnd = lineStart === 0 ? 0 : lineStart - 1; + entriesRead += 1; const decoded = decodeCodexSessionIndexEntry(line); if (Option.isNone(decoded)) continue; const id = decoded.value.id.trim(); - const title = decoded.value.thread_name.trim(); - if (id.length > 0 && title.length > 0) titles.set(id, title); + const normalizedTitle = decoded.value.thread_name.trim(); + const titleLineEnd = normalizedTitle.indexOf("\n"); + const title = normalizedTitle + .slice(0, titleLineEnd === -1 ? normalizedTitle.length : titleLineEnd) + .slice(0, MAX_IMPORTED_THREAD_TITLE_CHARS) + .trim(); + if (id.length > 0 && title.length > 0 && !titles.has(id)) titles.set(id, title); } - return titles; + return new Map(Array.from(titles).toReversed()); } interface MetadataReadBudget { @@ -311,26 +326,64 @@ function codexTurnId(metadata: unknown): string | null { return decoded.value.turn_id; } -function deriveImportedThreadTitle(text: string): string | null { - const injectedContextPatterns = [ - /^[\s\S]*?<\/recommended_plugins>\s*/, - /^[\s\S]*?<\/environment_context>\s*/, - /^[\s\S]*?<\/user_instructions>\s*/, - /^# AGENTS\.md instructions[^\n]*(?:\n+[\s\S]*?<\/INSTRUCTIONS>)?\s*/, - ]; - let visibleText = text.trimStart(); - let removedContext = true; - while (removedContext) { - removedContext = false; - for (const pattern of injectedContextPatterns) { - const withoutContext = visibleText.replace(pattern, ""); - if (withoutContext === visibleText) continue; - visibleText = withoutContext.trimStart(); +/** Find the first code-unit offset after context envelopes injected into a Codex message. */ +function skipLeadingCodexContext(text: string): number { + const contextTags = ["recommended_plugins", "environment_context", "user_instructions"]; + let offset = 0; + const skipWhitespace = () => { + while (offset < text.length && /\s/.test(text[offset] ?? "")) offset += 1; + }; + + skipWhitespace(); + while (offset < text.length) { + let removedContext = false; + for (const tag of contextTags) { + const openingTag = `<${tag}>`; + if (!text.startsWith(openingTag, offset)) continue; + const closingTag = ``; + const closingOffset = text.indexOf(closingTag, offset + openingTag.length); + if (closingOffset === -1) return text.length; + offset = closingOffset + closingTag.length; + skipWhitespace(); removedContext = true; break; } + if (removedContext) continue; + + const agentsHeading = "# AGENTS.md instructions"; + const agentsHeadingEnd = offset + agentsHeading.length; + if ( + text.startsWith(agentsHeading, offset) && + (agentsHeadingEnd === text.length || /\s/.test(text[agentsHeadingEnd] ?? "")) + ) { + const headingEnd = text.indexOf("\n", offset); + if (headingEnd === -1) return text.length; + offset = headingEnd + 1; + skipWhitespace(); + const openingTag = ""; + if (text.startsWith(openingTag, offset)) { + const closingTag = ""; + const closingOffset = text.indexOf(closingTag, offset + openingTag.length); + if (closingOffset === -1) return text.length; + offset = closingOffset + closingTag.length; + skipWhitespace(); + } + continue; + } + break; } - const firstLine = visibleText.split("\n")[0]?.slice(0, 100).trim(); + return offset; +} + +/** Derive a visible title without mistaking Codex-injected context for the user request. */ +export function deriveImportedThreadTitle(text: string, source: AgentSessionSource): string | null { + const visibleText = + source === "codex" ? text.slice(skipLeadingCodexContext(text)) : text.trimStart(); + const lineEnd = visibleText.indexOf("\n"); + const firstLine = visibleText + .slice(0, lineEnd === -1 ? visibleText.length : lineEnd) + .slice(0, MAX_IMPORTED_THREAD_TITLE_CHARS) + .trim(); return firstLine && firstLine.length > 0 ? firstLine : null; } @@ -425,7 +478,7 @@ function parseAgentSessionRecords( firstUserMessage = message; } if (firstDerivedTitle === null && message.role === "user") { - firstDerivedTitle = deriveImportedThreadTitle(message.text); + firstDerivedTitle = deriveImportedThreadTitle(message.text, input.source); } messages.push(message); if (messages.length > MAX_IMPORTED_MESSAGES) messages.shift(); @@ -717,6 +770,40 @@ export const make = Effect.gen(function* () { const statOption = (target: string) => fileSystem.stat(target).pipe(Effect.map(Option.some), Effect.orElseSucceed(Option.none)); + /** Read with a maxBytes payload cap and a one-byte probe for concurrent growth. */ + const readFileStringBounded = Effect.fn("AgentSessionScanner.readFileStringBounded")(function* ( + target: string, + maxBytes: number, + ) { + return yield* Effect.scoped( + fileSystem.open(target, { flag: "r" }).pipe( + Effect.flatMap((file) => + Effect.gen(function* () { + const info = yield* file.stat; + if (info.type !== "File" || info.size > BigInt(maxBytes)) return null; + + const decoder = new TextDecoder(); + const chunks: Array = []; + let bytesRead = 0; + while (bytesRead <= maxBytes) { + const next = yield* file.readAlloc( + Math.min(TRANSCRIPT_PREFIX_BYTES, maxBytes + 1 - bytesRead), + ); + if (Option.isNone(next) || next.value.byteLength === 0) { + chunks.push(decoder.decode()); + return chunks.join(""); + } + bytesRead += next.value.byteLength; + if (bytesRead > maxBytes) return null; + chunks.push(decoder.decode(next.value, { stream: true })); + } + return null; + }), + ), + ), + ); + }); + /** Match directory aliases without assuming the host volume is case-insensitive. */ const directoryIdentity = Effect.fn("AgentSessionScanner.directoryIdentity")(function* ( target: string, @@ -1031,16 +1118,12 @@ export const make = Effect.gen(function* () { function* (homePath: string, providerInstanceId: ProviderInstanceId, operationBudget: number) { const sessionsDir = path.join(homePath, "sessions"); const indexPath = path.join(homePath, "session_index.jsonl"); - const indexStats = yield* statOption(indexPath); + const indexContents = yield* readFileStringBounded( + indexPath, + MAX_CODEX_SESSION_INDEX_BYTES, + ).pipe(Effect.orElseSucceed(() => null)); const indexedTitles = - Option.isSome(indexStats) && - indexStats.value.type === "File" && - Number(indexStats.value.size) <= MAX_CODEX_SESSION_INDEX_BYTES - ? yield* fileSystem.readFileString(indexPath).pipe( - Effect.map(parseCodexSessionIndex), - Effect.orElseSucceed(() => new Map()), - ) - : new Map(); + indexContents === null ? new Map() : parseCodexSessionIndex(indexContents); const transcripts: Array = []; let operationsRemaining = operationBudget;