diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 1b1ff3738d..694cb7c9c8 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -5991,6 +5991,138 @@ describe("Codex app-server provider", () => { }); describe("Codex importable sessions", () => { + const CODEX_THREAD_PAGE_CAP = 100; + + // Codex answers thread/list with at most 100 rows per response whatever limit + // the caller asks for, and hands back a cursor for the remaining threads. + function cappedThreadListHandler(threads: Array>) { + return (input: unknown) => { + const params = (input ?? {}) as { + limit?: number; + cursor?: string; + cwd?: string; + sortKey?: string; + }; + const recencyKey = params.sortKey === "updated_at" ? "updatedAt" : "createdAt"; + const matching = threads + .filter((thread) => !params.cwd || thread.cwd === params.cwd) + .sort((a, b) => Number(b[recencyKey]) - Number(a[recencyKey])); + const offset = Number(params.cursor ?? 0); + const page = matching.slice( + offset, + offset + Math.min(params.limit ?? CODEX_THREAD_PAGE_CAP, CODEX_THREAD_PAGE_CAP), + ); + const nextOffset = offset + page.length; + return { + data: page, + nextCursor: nextOffset < matching.length ? String(nextOffset) : null, + }; + }; + } + + function threadsUsedInReverseCreationOrder(count: number) { + // The most recently used thread is also the oldest one created, so a scan + // that follows Codex's creation-time page order misses it. + return Array.from({ length: count }, (_, index) => ({ + id: `thread-${index}`, + cwd: "/workspace/project-a", + preview: `Session ${index}`, + createdAt: index, + updatedAt: count - index, + })); + } + + test("fills the requested window across the pages Codex caps at 100 rows", async () => { + const threads = threadsUsedInReverseCreationOrder(250); + const appServer = createFakeCodexAppServer({ + "thread/list": cappedThreadListHandler([ + ...threads, + { id: "other-project", cwd: "/workspace/project-b", createdAt: 9000, updatedAt: 9000 }, + ]), + }); + const provider = createProviderWithFakeAppServer(appServer); + + const sessions = await provider.listImportableSessions({ + limit: 240, + cwd: "/workspace/project-a", + }); + + expect(sessions.map((session) => session.providerHandleId)).toEqual( + threads.slice(0, 240).map((thread) => thread.id), + ); + appServer.assertNoErrors(); + }); + + test("keeps the scanned window on the most recently used threads", async () => { + const threads = threadsUsedInReverseCreationOrder(550); + const appServer = createFakeCodexAppServer({ + "thread/list": cappedThreadListHandler(threads), + }); + const provider = createProviderWithFakeAppServer(appServer); + + const sessions = await provider.listImportableSessions({ limit: 500, scanLimit: 700 }); + + expect(sessions.map((session) => session.providerHandleId)).toEqual( + threads.slice(0, 500).map((thread) => thread.id), + ); + appServer.assertNoErrors(); + }); + + test("keeps a thread once when it moves onto a later page mid-scan", async () => { + // A thread used while the scan is paging sorts to the front and comes back + // on the next page. + const appServer = createFakeCodexAppServer({ + "thread/list": (input) => { + const { cursor } = (input ?? {}) as { cursor?: string }; + return cursor + ? { + data: [ + { id: "thread-1", cwd: "/workspace/project-a", updatedAt: 9000 }, + { id: "thread-3", cwd: "/workspace/project-a", updatedAt: 1000 }, + ], + nextCursor: null, + } + : { + data: [ + { id: "thread-1", cwd: "/workspace/project-a", updatedAt: 3000 }, + { id: "thread-2", cwd: "/workspace/project-a", updatedAt: 2000 }, + ], + nextCursor: "page-2", + }; + }, + }); + const provider = createProviderWithFakeAppServer(appServer); + + const sessions = await provider.listImportableSessions({ limit: 500, scanLimit: 500 }); + + expect(sessions.map((session) => session.providerHandleId)).toEqual([ + "thread-1", + "thread-2", + "thread-3", + ]); + appServer.assertNoErrors(); + }); + + test("stops scanning when Codex pages without handing back a new thread", async () => { + const appServer = createFakeCodexAppServer({ + // A cursor cycle: the two pages after the first serve threads the scan + // already holds and point at each other, so it can never fill its window. + "thread/list": (input) => { + const { cursor } = (input ?? {}) as { cursor?: string }; + const thread = { id: "thread-1", cwd: "/workspace/project-a" }; + if (cursor === "page-a") return { data: [thread], nextCursor: "page-b" }; + if (cursor === "page-b") return { data: [thread], nextCursor: "page-a" }; + return { data: [thread], nextCursor: "page-a" }; + }, + }); + const provider = createProviderWithFakeAppServer(appServer); + + const sessions = await provider.listImportableSessions({ limit: 500, scanLimit: 500 }); + + expect(sessions.map((session) => session.providerHandleId)).toEqual(["thread-1"]); + appServer.assertNoErrors(); + }); + test("listImportableSessions uses thread list metadata without hydrating thread history", async () => { const allThreads = [ { @@ -6071,7 +6203,10 @@ describe("Codex importable sessions", () => { capabilities: { experimentalApi: true, mcpServerOpenaiFormElicitation: true }, }, }, - { method: "thread/list", params: { limit: 50, cwd: "/workspace/project-a" } }, + { + method: "thread/list", + params: { limit: 50, sortKey: "updated_at", cwd: "/workspace/project-a" }, + }, ]); }); }); diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 8bf9ede447..1c43612fa4 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -914,6 +914,60 @@ const CodexModelListResponseSchema = z.object({ .optional(), }); +/** + * Read the newest `window.limit` threads Codex knows about. + * + * Codex caps every `thread/list` response at 100 rows whatever limit it is + * given (codex-cli 0.153 and 0.155), so one request only ever sees the newest + * page and hands back a cursor for the rest. Follow that cursor until the + * caller's window is full or Codex runs out of threads. + */ +async function readCodexThreadWindow( + client: Pick, + logger: Logger, + window: { limit: number; cwd?: string }, +): Promise>> { + // A thread updated while the scan is paging moves under `updated_at` order + // and can come back on a later page, so identify each one and keep it once. + // A row Codex sends without an id stands for itself. + const threads = new Map>(); + let cursor: string | undefined; + while (threads.size < window.limit) { + const response = toObjectRecord( + await client.request("thread/list", { + limit: window.limit - threads.size, + // Rank the window by last use. Codex pages by creation time by default, + // which drops an old conversation that is still being worked in. + // Older Codex builds ignore the unknown key and keep that order. + sortKey: "updated_at", + ...(window.cwd ? { cwd: window.cwd } : {}), + ...(cursor ? { cursor } : {}), + }), + ); + const page = Array.isArray(response?.data) ? response.data.filter(isRecord) : []; + const sizeBeforePage = threads.size; + for (const thread of page) { + const identity = typeof thread.id === "string" ? thread.id : thread; + if (!threads.has(identity)) threads.set(identity, thread); + } + const nextCursor = typeof response?.nextCursor === "string" ? response.nextCursor : undefined; + if (!nextCursor) break; + if (threads.size === sizeBeforePage) { + // Every page that continues the scan brings a thread the scan has not + // seen, so the loop is bounded by the window. A page that brings none + // would let a Codex-side cursor bug (a stuck cursor, or a cycle) page for + // ever behind a caller that has already timed out. + logger.warn( + { cursor: nextCursor }, + "codex thread/list returned no new threads mid-scan, stopping the session scan", + ); + break; + } + cursor = nextCursor; + } + return [...threads.values()]; +} + function filterCodexThreadsByCwd( threads: Array>, cwd: string | undefined, @@ -7143,13 +7197,10 @@ export class CodexAppServerAgentClient implements AgentClient { // filtering since most threads will be from other cwds, then keep the // local realpath-aware filter for symlink-equivalent workspace paths. const listLimit = options?.cwd ? Math.max(scanLimit, 50) : scanLimit; - const response = toObjectRecord( - await client.request("thread/list", { - limit: listLimit, - ...(options?.cwd ? { cwd: options.cwd } : {}), - }), - ); - const allThreads = Array.isArray(response?.data) ? response.data.filter(isRecord) : []; + const allThreads = await readCodexThreadWindow(client, this.logger, { + limit: listLimit, + cwd: options?.cwd, + }); const threads = filterCodexThreadsByCwd(allThreads, options?.cwd); return threads.slice(0, limit).map((thread) => { const threadId = typeof thread.id === "string" ? thread.id : "";