diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index 4eb03a5cc036..5bc4fb2678e0 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -318,6 +318,263 @@ 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.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", + }, + ]); + + 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", + titleState: { + source: "manual", + version: CommandId.make("manual-title"), + needsRefinement: false, + }, + }), + }), + }), + ).toEqual({ importedCount: 1, skippedCount: 0 }); + expect(commands).toEqual([]); + }), + ); + + 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 5ebb41a1bb54..e0176a6c6b2b 100644 --- a/apps/server/src/project/AgentSessionImporter.ts +++ b/apps/server/src/project/AgentSessionImporter.ts @@ -70,6 +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 === "" || + 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, @@ -136,6 +156,32 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu let importedCount = 0; let skippedCount = 0; + const repairImportedCodexTitle = Effect.fn("repairImportedCodexTitle")(function* ( + threadId: ThreadId, + 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 ( + replacementTitle === null || + !hasLegacyCodexContextTitle(existingTitle) || + existingTitle === replacementTitle + ) { + return; + } + yield* engine.dispatch({ + type: "thread.title.generate.complete", + commandId: CommandId.make(yield* crypto.randomUUIDv4), + threadId, + title: replacementTitle, + expectedTitle: existingTitle, + expectedVersion: existingThread.value.titleState?.version ?? null, + needsRefinement: false, + }); + }); + yield* Stream.runForEach(threads, (outcome) => Effect.gen(function* () { if (outcome._tag === "Skipped") { @@ -147,6 +193,16 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu `import:${outcome.source.providerInstanceId}:${outcome.source.providerSessionId}`, ); if (outcome._tag === "AlreadyImported") { + if (outcome.source.provider === "codex") { + 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 +255,16 @@ export const importRecentAgentThreads = Effect.fn("importRecentAgentThreads")(fu importedHistoryPresent && Option.isSome(existingBinding) ) { + if (thread.source === "codex") { + 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 b1cba460f959..c9a2d5ece58a 100644 --- a/apps/server/src/project/AgentSessionScanner.test.ts +++ b/apps/server/src/project/AgentSessionScanner.test.ts @@ -2606,10 +2606,175 @@ 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", + encodeTranscriptRecord({ + 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"); + }), + ); + + 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); + }), + ); }); }); 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: "" }), + 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", "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])( "handles the exact record limit and an interior blank overflow=%s", (overflow) => { @@ -3099,13 +3264,103 @@ 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.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."; @@ -3126,7 +3381,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..3f7a43e0ec2d 100644 --- a/apps/server/src/project/AgentSessionScanner.ts +++ b/apps/server/src/project/AgentSessionScanner.ts @@ -94,6 +94,9 @@ 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 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), @@ -110,6 +113,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 +149,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 +160,7 @@ interface AgentSessionTranscriptMetadata { readonly providerInstanceId: ProviderInstanceId; readonly fallbackSessionId: string; readonly lastActiveAtMs: number; + readonly canonicalTitle?: string; } export interface AgentSessionThreadMessage { @@ -174,7 +186,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 +224,7 @@ interface RawCandidate { readonly transcripts: ReadonlyArray<{ readonly filePath: string; readonly mtimeMs: number | null; + readonly canonicalTitle: string | null; }>; } @@ -216,6 +233,32 @@ interface TranscriptCandidate { readonly mtimeMs: number; readonly providerInstanceId: ProviderInstanceId; readonly size: number; + 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(); + 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 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 new Map(Array.from(titles).toReversed()); } interface MetadataReadBudget { @@ -283,6 +326,67 @@ function codexTurnId(metadata: unknown): string | null { return decoded.value.turn_id; } +/** 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; + } + 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; +} + /** Keep visible user and assistant text while ignoring tools, reasoning, and malformed records. */ export function parseAgentSessionTranscript( input: AgentSessionTranscriptMetadata & { @@ -303,13 +407,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 +477,9 @@ function parseAgentSessionRecords( if (firstUserMessage === undefined && message.role === "user") { firstUserMessage = message; } + if (firstDerivedTitle === null && message.role === "user") { + firstDerivedTitle = deriveImportedThreadTitle(message.text, input.source); + } messages.push(message); if (messages.length > MAX_IMPORTED_MESSAGES) messages.shift(); }; @@ -492,13 +600,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, @@ -664,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, @@ -966,6 +1106,7 @@ export const make = Effect.gen(function* () { mtimeMs: stats.value.mtime.value.getTime(), providerInstanceId, size: Number(stats.value.size), + canonicalTitle: null, }); } } @@ -976,6 +1117,13 @@ 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 indexContents = yield* readFileStringBounded( + indexPath, + MAX_CODEX_SESSION_INDEX_BYTES, + ).pipe(Effect.orElseSucceed(() => null)); + const indexedTitles = + indexContents === null ? new Map() : parseCodexSessionIndex(indexContents); const transcripts: Array = []; let operationsRemaining = operationBudget; @@ -1029,6 +1177,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 +1204,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 +1565,7 @@ export const make = Effect.gen(function* () { return Option.some({ _tag: "AlreadyImported", source: completedSource, + canonicalTitle: transcript.canonicalTitle, }); } if ( @@ -1454,6 +1613,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, );