From df2a05b6d80aba7ddaaac3ef5c547d25046c152b Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 27 Aug 2026 06:37:26 +0400 Subject: [PATCH 1/6] feat: support session/fork --- src/CodexAcpClient.ts | 23 ++++++++ src/CodexAcpServer.ts | 52 +++++++++++++++---- .../CodexACPAgent/CodexAcpClient.test.ts | 40 ++++++++++++++ .../CodexACPAgent/initialize.test.ts | 1 + .../CodexACPAgent/session-fork.test.ts | 37 +++++++++++++ src/index.ts | 1 + 6 files changed, 143 insertions(+), 11 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/session-fork.test.ts diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 893dc85d..fe6ecf47 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -487,6 +487,29 @@ export class CodexAcpClient { } } + async forkSession(request: acp.ForkSessionRequest): Promise { + const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); + await this.refreshSkills(request.cwd, additionalDirectories); + + const response = await this.codexClient.threadFork({ + config: await this.createSessionConfig(request.cwd, additionalDirectories, request.mcpServers ?? []), + cwd: request.cwd, + modelProvider: await this.getResumeModelProvider(), + threadId: request.sessionId, + }); + const codexModels = await this.fetchAvailableModels(); + const currentModelId = this.createModelId(codexModels, response.model, response.reasoningEffort).toString(); + return { + sessionId: response.thread.id, + currentModelId, + models: codexModels, + collaborationMode: this.getCollaborationMode(response.thread.id), + modelProvider: response.modelProvider, + currentServiceTier: response.serviceTier as ServiceTier ?? null, + additionalDirectories, + }; + } + async loadSession(request: acp.LoadSessionRequest, onSubscribed?: () => void): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index f94978e4..2b278307 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -315,6 +315,7 @@ export class CodexAcpServer { list: { }, close: { }, delete: { }, + fork: { }, additionalDirectories: {}, subagents: {}, }; @@ -553,9 +554,13 @@ export class CodexAcpServer { return generation; } - async tryCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState]> { - const requestedSessionGeneration = "sessionId" in request - ? this.beginSessionOpen(request.sessionId) + async tryCreateSession( + request: acp.NewSessionRequest | acp.ResumeSessionRequest | acp.ForkSessionRequest, + operation: "new" | "resume" | "fork" = "sessionId" in request ? "resume" : "new", + ): Promise<[SessionId, LegacySessionModelState, SessionModeState]> { + const existingSessionRequest = request as acp.ResumeSessionRequest | acp.ForkSessionRequest; + const requestedSessionGeneration = operation === "resume" + ? this.beginSessionOpen(existingSessionRequest.sessionId) : null; await this.checkAuthorization(); const requestedMcpServers = request.mcpServers ?? []; @@ -565,23 +570,28 @@ export class CodexAcpServer { let sessionMetadata: SessionMetadata; let resumeSubscribed = false; - if ("sessionId" in request) { - logger.log(`Resume existing session: ${request.sessionId}...`); + if (operation === "resume") { + const resumeRequest = request as acp.ResumeSessionRequest; + logger.log(`Resume existing session: ${resumeRequest.sessionId}...`); try { sessionMetadata = await this.runWithProcessCheck(() => - this.codexAcpClient.resumeSession(request, () => { + this.codexAcpClient.resumeSession(resumeRequest, () => { resumeSubscribed = true; }) ); } catch (err) { if (resumeSubscribed && requestedSessionGeneration !== null) { - await this.cleanupStaleSessionOpen(request.sessionId, requestedSessionGeneration); + await this.cleanupStaleSessionOpen(resumeRequest.sessionId, requestedSessionGeneration); } throw err; } + } else if (operation === "fork") { + const forkRequest = request as acp.ForkSessionRequest; + logger.log(`Fork existing session: ${forkRequest.sessionId}...`); + sessionMetadata = await this.runWithProcessCheck(() => this.codexAcpClient.forkSession(forkRequest)); } else { logger.log(`Create new session...`); - sessionMetadata = await this.runWithProcessCheck(() => this.codexAcpClient.newSession(request)); + sessionMetadata = await this.runWithProcessCheck(() => this.codexAcpClient.newSession(request as acp.NewSessionRequest)); } const {sessionId, currentModelId, models} = sessionMetadata; @@ -600,7 +610,7 @@ export class CodexAcpServer { resumeSubscribed = false; await this.closeStaleSessionOpen(sessionId, sessionGeneration); } - const sessionMcpServers = this.resolveSessionMcpServers(requestedMcpServers, "sessionId" in request); + const sessionMcpServers = this.resolveSessionMcpServers(requestedMcpServers, operation === "resume"); const currentModel = this.findCurrentModel(models, currentModelId); const currentModelSupportsFast = modelSupportsFast(currentModel); const sessionState: SessionState = { @@ -628,7 +638,7 @@ export class CodexAcpServer { terminalOutputMode: this.terminalOutputMode, goalRevision: 0, sessionTitle: null, - sessionTitleSource: "sessionId" in request ? "unknown" : "unset", + sessionTitleSource: operation === "resume" ? "unknown" : "unset", subagents: new CodexSubagentEventRouter( sessionId, clientSupportsSubagents(this.clientCapabilities), @@ -647,7 +657,7 @@ export class CodexAcpServer { } this.publishAvailableCommandsAsync(sessionState, sessionGeneration); - if ("sessionId" in request) { + if (operation === "resume") { this.publishCurrentGoalAsync(sessionState, sessionGeneration); } const sessionModelState: LegacySessionModelState = this.createModelState(models, currentModelId); @@ -733,6 +743,26 @@ export class CodexAcpServer { }; } + async forkSession(params: acp.ForkSessionRequest): Promise { + if (this.providerUpdate !== null) { + await this.providerUpdate; + } + logger.log("Forking session...", {sessionId: params.sessionId}); + try { + const [sessionId, , modeState] = await this.tryCreateSession(params, "fork"); + logger.log("Session forked", {sourceSessionId: params.sessionId, sessionId}); + return { + sessionId, + modes: modeState, + ...this.createSessionConfigOptionsResponse(this.getSessionState(sessionId)), + }; + } catch (e) { + const error = e instanceof Error ? e : new Error(String(e)); + await this.handleError(error); + throw e; + } + } + async listSessions(params: acp.ListSessionsRequest): Promise { logger.log("Listing sessions...", {cwd: params.cwd, cursor: params.cursor}); await this.checkAuthorization(); diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index ba2f8cf5..5a5cb12c 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -563,6 +563,46 @@ describe('ACP server test', { timeout: 40_000 }, () => { }); }); + it('forks an ACP session through thread/fork with the requested workspace', async () => { + const mockFixture = createCodexMockTestFixture(); + const codexAcpClient = mockFixture.getCodexAcpClient(); + const codexAppServerClient = mockFixture.getCodexAppServerClient(); + + vi.spyOn(codexAppServerClient, "skillsExtraRootsSet").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({data: []}); + const threadForkSpy = vi.spyOn(codexAppServerClient, "threadFork").mockResolvedValue({ + thread: {id: "fork-id"} as any, + model: "gpt-5", + modelProvider: "openai", + reasoningEffort: "medium", + serviceTier: null, + } as any); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [createTestModel({id: "gpt-5"})], + nextCursor: null, + }); + + const forked = await codexAcpClient.forkSession({ + sessionId: "source-id", + cwd: "/workspace", + additionalDirectories: ["/workspace/extra"], + mcpServers: [], + }); + + expect(forked.sessionId).toBe("fork-id"); + expect(forked.additionalDirectories).toEqual(["/workspace/extra"]); + expect(threadForkSpy).toHaveBeenCalledWith(expect.objectContaining({ + threadId: "source-id", + cwd: "/workspace", + config: expect.objectContaining({ + projects: { + "/workspace": {trust_level: "trusted"}, + "/workspace/extra": {trust_level: "trusted"}, + }, + }), + })); + }); + it('restores collaboration mode for resumed and loaded sessions', async () => { const mockFixture = createCodexMockTestFixture(); const codexAcpAgent = mockFixture.getCodexAcpAgent(); diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 65350488..9f8fe458 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -52,6 +52,7 @@ describe('CodexACPAgent - initialize', () => { list: {}, close: {}, delete: {}, + fork: {}, additionalDirectories: {}, subagents: {}, }, diff --git a/src/__tests__/CodexACPAgent/session-fork.test.ts b/src/__tests__/CodexACPAgent/session-fork.test.ts new file mode 100644 index 00000000..aee6546b --- /dev/null +++ b/src/__tests__/CodexACPAgent/session-fork.test.ts @@ -0,0 +1,37 @@ +import {describe, expect, it, vi} from "vitest"; +import {createCodexMockTestFixture, createTestModel} from "../acp-test-utils"; + +describe("ACP session fork", () => { + it("creates and installs a forked session", async () => { + const fixture = createCodexMockTestFixture(); + const agent = fixture.getCodexAcpAgent(); + const client = fixture.getCodexAcpClient(); + const model = createTestModel({id: "gpt-5"}); + + vi.spyOn(client, "authRequired").mockResolvedValue(false); + vi.spyOn(client, "getAccount").mockResolvedValue({account: null, requiresOpenaiAuth: false}); + const forkSpy = vi.spyOn(client, "forkSession").mockResolvedValue({ + sessionId: "fork-id", + currentModelId: "gpt-5[medium]", + models: [model], + collaborationMode: "default", + modelProvider: "openai", + currentServiceTier: null, + additionalDirectories: [], + }); + + const response = await agent.forkSession({ + sessionId: "source-id", + cwd: "/workspace", + mcpServers: [], + }); + + expect(response.sessionId).toBe("fork-id"); + expect(agent.getSessionState("fork-id").cwd).toBe("/workspace"); + expect(forkSpy).toHaveBeenCalledWith({ + sessionId: "source-id", + cwd: "/workspace", + mcpServers: [], + }); + }); +}); diff --git a/src/index.ts b/src/index.ts index 15ad2256..68df2ccd 100644 --- a/src/index.ts +++ b/src/index.ts @@ -143,6 +143,7 @@ function startAcpServer() { .onRequest(acp.methods.agent.initialize, (ctx) => getAgent().initialize(ctx.params)) .onRequest(acp.methods.agent.session.new, (ctx) => getAgent().newSession(ctx.params)) .onRequest(acp.methods.agent.session.load, (ctx) => getAgent().loadSession(ctx.params)) + .onRequest(acp.methods.agent.session.fork, (ctx) => getAgent().forkSession(ctx.params)) .onRequest(acp.methods.agent.session.list, (ctx) => getAgent().listSessions(ctx.params)) .onRequest(acp.methods.agent.session.delete, (ctx) => getAgent().deleteSession(ctx.params)) .onRequest(acp.methods.agent.session.resume, (ctx) => getAgent().resumeSession(ctx.params)) From 2b192fe30478b511640e65e438972133387705c3 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 27 Aug 2026 09:27:12 +0400 Subject: [PATCH 2/6] feat: support message-specific session forks --- src/CodexAcpClient.ts | 34 +++++++++++++++ .../CodexACPAgent/CodexAcpClient.test.ts | 42 +++++++++++++++++++ 2 files changed, 76 insertions(+) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index fe6ecf47..94aef2ec 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -490,10 +490,26 @@ export class CodexAcpClient { async forkSession(request: acp.ForkSessionRequest): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); + const messageId = readAirForkMessageId(request._meta); + let lastTurnId: string | undefined; + if (messageId) { + const history = await this.codexClient.threadRead({ + threadId: request.sessionId, + includeTurns: true, + }); + lastTurnId = history.thread.turns.find(turn => turn.items.some(item => item.id === messageId))?.id; + if (!lastTurnId) { + throw RequestError.invalidParams( + {messageId}, + `Fork point message ${messageId} was not found in session ${request.sessionId}`, + ); + } + } const response = await this.codexClient.threadFork({ config: await this.createSessionConfig(request.cwd, additionalDirectories, request.mcpServers ?? []), cwd: request.cwd, + ...(lastTurnId !== undefined && {lastTurnId}), modelProvider: await this.getResumeModelProvider(), threadId: request.sessionId, }); @@ -1353,6 +1369,24 @@ function readMetaAdditionalRoots(meta?: Record | null): string[ .filter(value => value.length > 0)); } +function readAirForkMessageId(meta?: Record | null): string | undefined { + const jetbrains = meta?.["jetbrains"]; + if (!isUnknownRecord(jetbrains)) return undefined; + const air = jetbrains["air"]; + if (!isUnknownRecord(air)) return undefined; + const fork = air["fork"]; + if (!isUnknownRecord(fork) || fork["version"] !== 1) return undefined; + const messageId = fork["messageId"]; + if (typeof messageId !== "string" || messageId.trim().length === 0) { + throw RequestError.invalidParams(undefined, "AIR fork messageId must be a non-empty string"); + } + return messageId.trim(); +} + +function isUnknownRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + function readAdditionalDirectories(cwd: string, additionalDirectories?: string[], meta?: Record | null): string[] { const rawDirectories = additionalDirectories ?? readMetaAdditionalRoots(meta); if (!rawDirectories) { diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 5a5cb12c..3895b51d 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -603,6 +603,48 @@ describe('ACP server test', { timeout: 40_000 }, () => { })); }); + it('maps an AIR fork message id to the containing Codex turn', async () => { + const mockFixture = createCodexMockTestFixture(); + const codexAcpClient = mockFixture.getCodexAcpClient(); + const codexAppServerClient = mockFixture.getCodexAppServerClient(); + + vi.spyOn(codexAppServerClient, "skillsExtraRootsSet").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({data: []}); + vi.spyOn(codexAppServerClient, "threadRead").mockResolvedValue({ + thread: { + id: "source-id", + turns: [ + {id: "turn-1", items: [{id: "item-1"}]}, + {id: "turn-2", items: [{id: "agent-message-2"}]}, + ], + }, + } as any); + const threadForkSpy = vi.spyOn(codexAppServerClient, "threadFork").mockResolvedValue({ + thread: {id: "fork-id"}, + model: "gpt-5", + modelProvider: "openai", + reasoningEffort: "medium", + serviceTier: null, + } as any); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [createTestModel({id: "gpt-5"})], + nextCursor: null, + }); + + await codexAcpClient.forkSession({ + sessionId: "source-id", + cwd: "/workspace", + _meta: { + jetbrains: {air: {fork: {version: 1, messageId: "agent-message-2"}}}, + }, + }); + + expect(threadForkSpy).toHaveBeenCalledWith(expect.objectContaining({ + threadId: "source-id", + lastTurnId: "turn-2", + })); + }); + it('restores collaboration mode for resumed and loaded sessions', async () => { const mockFixture = createCodexMockTestFixture(); const codexAcpAgent = mockFixture.getCodexAcpAgent(); From b0f87e83bf701eaa7e540adb866c63e83cafb8e2 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 27 Aug 2026 09:45:14 +0400 Subject: [PATCH 3/6] fix: accept AIR segment IDs for session forks Resolve the exact item id first. Fall back to the ACP source id for compatibility with older AIR builds. --- src/CodexAcpClient.ts | 12 +++++++++++- src/__tests__/CodexACPAgent/CodexAcpClient.test.ts | 2 +- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 94aef2ec..5d48e55e 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -497,7 +497,10 @@ export class CodexAcpClient { threadId: request.sessionId, includeTurns: true, }); - lastTurnId = history.thread.turns.find(turn => turn.items.some(item => item.id === messageId))?.id; + const candidateIds = airForkMessageIdCandidates(messageId); + lastTurnId = candidateIds + .map(candidateId => history.thread.turns.find(turn => turn.items.some(item => item.id === candidateId))?.id) + .find(turnId => turnId !== undefined); if (!lastTurnId) { throw RequestError.invalidParams( {messageId}, @@ -1383,6 +1386,13 @@ function readAirForkMessageId(meta?: Record | null): string | u return messageId.trim(); } +function airForkMessageIdCandidates(messageId: string): string[] { + // Older AIR builds sent their visible segment id. Prefer the exact id before its ACP source id. + const visibleSegmentSuffix = /:segment:\d+$/; + const protocolMessageId = messageId.replace(visibleSegmentSuffix, ""); + return protocolMessageId === messageId ? [messageId] : [messageId, protocolMessageId]; +} + function isUnknownRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 3895b51d..3637cfac 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -635,7 +635,7 @@ describe('ACP server test', { timeout: 40_000 }, () => { sessionId: "source-id", cwd: "/workspace", _meta: { - jetbrains: {air: {fork: {version: 1, messageId: "agent-message-2"}}}, + jetbrains: {air: {fork: {version: 1, messageId: "agent-message-2:segment:0"}}}, }, }); From 75b1809de792d61493633b43d635deac9c7e9fac Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 27 Aug 2026 11:49:12 +0400 Subject: [PATCH 4/6] fix: restore exact AIR session forks Codex can rebuild item ids when it restores a session. Match the persisted message fingerprint and occurrence before the thread fork. --- src/CodexAcpClient.ts | 45 +++++++++++++++--- .../CodexACPAgent/CodexAcpClient.test.ts | 47 +++++++++++++++++++ 2 files changed, 85 insertions(+), 7 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 5d48e55e..58f5adce 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -27,6 +27,7 @@ import type {JsonValue} from "./app-server/serde_json/JsonValue"; import {ModelId} from "./ModelId"; import {AgentMode} from "./AgentMode"; import path from "node:path"; +import {createHash} from "node:crypto"; import {logger} from "./Logger"; import {sanitizeMcpServerName} from "./McpServerName"; import type { @@ -490,21 +491,28 @@ export class CodexAcpClient { async forkSession(request: acp.ForkSessionRequest): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); - const messageId = readAirForkMessageId(request._meta); + const forkPoint = readAirForkPoint(request._meta); let lastTurnId: string | undefined; - if (messageId) { + if (forkPoint) { const history = await this.codexClient.threadRead({ threadId: request.sessionId, includeTurns: true, }); - const candidateIds = airForkMessageIdCandidates(messageId); + const candidateIds = airForkMessageIdCandidates(forkPoint.messageId); lastTurnId = candidateIds .map(candidateId => history.thread.turns.find(turn => turn.items.some(item => item.id === candidateId))?.id) .find(turnId => turnId !== undefined); + if (!lastTurnId && forkPoint.messageFingerprint) { + const matchingTurns = history.thread.turns.flatMap(turn => turn.items + .filter(item => item.type === "agentMessage" + && fingerprintAgentMessage(item.text) === forkPoint.messageFingerprint) + .map(() => turn.id)); + lastTurnId = matchingTurns[forkPoint.messageOccurrence - 1]; + } if (!lastTurnId) { throw RequestError.invalidParams( - {messageId}, - `Fork point message ${messageId} was not found in session ${request.sessionId}`, + {messageId: forkPoint.messageId}, + `Fork point message ${forkPoint.messageId} was not found in session ${request.sessionId}`, ); } } @@ -1372,7 +1380,13 @@ function readMetaAdditionalRoots(meta?: Record | null): string[ .filter(value => value.length > 0)); } -function readAirForkMessageId(meta?: Record | null): string | undefined { +type AirForkPoint = { + messageId: string; + messageFingerprint?: string; + messageOccurrence: number; +}; + +function readAirForkPoint(meta?: Record | null): AirForkPoint | undefined { const jetbrains = meta?.["jetbrains"]; if (!isUnknownRecord(jetbrains)) return undefined; const air = jetbrains["air"]; @@ -1383,7 +1397,24 @@ function readAirForkMessageId(meta?: Record | null): string | u if (typeof messageId !== "string" || messageId.trim().length === 0) { throw RequestError.invalidParams(undefined, "AIR fork messageId must be a non-empty string"); } - return messageId.trim(); + const messageFingerprint = fork["messageFingerprint"]; + if (messageFingerprint !== undefined + && (typeof messageFingerprint !== "string" || !/^sha256:[0-9a-f]{64}$/.test(messageFingerprint))) { + throw RequestError.invalidParams(undefined, "AIR fork messageFingerprint must be a SHA-256 fingerprint"); + } + const messageOccurrence = fork["messageOccurrence"] ?? 1; + if (!Number.isSafeInteger(messageOccurrence) || (messageOccurrence as number) < 1) { + throw RequestError.invalidParams(undefined, "AIR fork messageOccurrence must be a positive integer"); + } + return { + messageId: messageId.trim(), + ...(typeof messageFingerprint === "string" && {messageFingerprint}), + messageOccurrence: messageOccurrence as number, + }; +} + +function fingerprintAgentMessage(text: string): string { + return `sha256:${createHash("sha256").update(text, "utf8").digest("hex")}`; } function airForkMessageIdCandidates(messageId: string): string[] { diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 3637cfac..3d254ab7 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -645,6 +645,53 @@ describe('ACP server test', { timeout: 40_000 }, () => { })); }); + it('maps a persisted AIR message fingerprint when Codex item ids changed', async () => { + const mockFixture = createCodexMockTestFixture(); + const codexAcpClient = mockFixture.getCodexAcpClient(); + const codexAppServerClient = mockFixture.getCodexAppServerClient(); + + vi.spyOn(codexAppServerClient, "skillsExtraRootsSet").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({data: []}); + vi.spyOn(codexAppServerClient, "threadRead").mockResolvedValue({ + thread: { + id: "source-id", + turns: [ + {id: "turn-1", items: [{type: "agentMessage", id: "new-item-1", text: "Same answer"}]}, + {id: "turn-2", items: [{type: "agentMessage", id: "new-item-2", text: "Same answer"}]}, + ], + }, + } as any); + const threadForkSpy = vi.spyOn(codexAppServerClient, "threadFork").mockResolvedValue({ + thread: {id: "fork-id"}, + model: "gpt-5", + modelProvider: "openai", + reasoningEffort: "medium", + serviceTier: null, + } as any); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [createTestModel({id: "gpt-5"})], + nextCursor: null, + }); + + await codexAcpClient.forkSession({ + sessionId: "source-id", + cwd: "/workspace", + _meta: { + jetbrains: {air: {fork: { + version: 1, + messageId: "old-item-2", + messageFingerprint: "sha256:41153d2b46c2869f4021958d44dac18888247fd999507c28970be299a8de4a0f", + messageOccurrence: 2, + }}}, + }, + }); + + expect(threadForkSpy).toHaveBeenCalledWith(expect.objectContaining({ + threadId: "source-id", + lastTurnId: "turn-2", + })); + }); + it('restores collaboration mode for resumed and loaded sessions', async () => { const mockFixture = createCodexMockTestFixture(); const codexAcpAgent = mockFixture.getCodexAcpAgent(); From b7c84ac565c9e50f9f16970da9d42d8da76668a8 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 27 Aug 2026 12:27:29 +0400 Subject: [PATCH 5/6] fix: release fork sessions before ACP load Do not send updates for a fork before the client loads it. Release the Codex app-server writer after thread/fork. This lets AIR load the fork through ACP without a terminal fallback. --- src/CodexAcpClient.ts | 1 + src/CodexAcpServer.ts | 7 +++++-- src/__tests__/CodexACPAgent/CodexAcpClient.test.ts | 4 ++++ src/__tests__/CodexACPAgent/session-fork.test.ts | 2 ++ 4 files changed, 12 insertions(+), 2 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 58f5adce..2e9ae4fb 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -524,6 +524,7 @@ export class CodexAcpClient { modelProvider: await this.getResumeModelProvider(), threadId: request.sessionId, }); + await this.codexClient.threadUnsubscribe({threadId: response.thread.id}); const codexModels = await this.fetchAvailableModels(); const currentModelId = this.createModelId(codexModels, response.model, response.reasoningEffort).toString(); return { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 2b278307..2c7937f3 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -648,7 +648,8 @@ export class CodexAcpServer { this.sessions.set(sessionId, sessionState); resumeSubscribed = false; - if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { + const canPublishSessionUpdates = operation !== "fork"; + if (canPublishSessionUpdates && requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { this.pendingMcpStartupSessions.set(sessionId, { requestedServers: new Set(getRequestedMcpServerNames(requestedMcpServers)), afterVersion: mcpServerStartupVersion, @@ -656,7 +657,9 @@ export class CodexAcpServer { this.publishMcpStartupStatusAsync(sessionId); } - this.publishAvailableCommandsAsync(sessionState, sessionGeneration); + if (canPublishSessionUpdates) { + this.publishAvailableCommandsAsync(sessionState, sessionGeneration); + } if (operation === "resume") { this.publishCurrentGoalAsync(sessionState, sessionGeneration); } diff --git a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts index 3d254ab7..ced1752a 100644 --- a/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts +++ b/src/__tests__/CodexACPAgent/CodexAcpClient.test.ts @@ -577,6 +577,9 @@ describe('ACP server test', { timeout: 40_000 }, () => { reasoningEffort: "medium", serviceTier: null, } as any); + const threadUnsubscribeSpy = vi.spyOn(codexAppServerClient, "threadUnsubscribe").mockResolvedValue({ + status: "unsubscribed", + }); vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ data: [createTestModel({id: "gpt-5"})], nextCursor: null, @@ -601,6 +604,7 @@ describe('ACP server test', { timeout: 40_000 }, () => { }, }), })); + expect(threadUnsubscribeSpy).toHaveBeenCalledWith({threadId: "fork-id"}); }); it('maps an AIR fork message id to the containing Codex turn', async () => { diff --git a/src/__tests__/CodexACPAgent/session-fork.test.ts b/src/__tests__/CodexACPAgent/session-fork.test.ts index aee6546b..edc4007e 100644 --- a/src/__tests__/CodexACPAgent/session-fork.test.ts +++ b/src/__tests__/CodexACPAgent/session-fork.test.ts @@ -10,6 +10,7 @@ describe("ACP session fork", () => { vi.spyOn(client, "authRequired").mockResolvedValue(false); vi.spyOn(client, "getAccount").mockResolvedValue({account: null, requiresOpenaiAuth: false}); + vi.spyOn(client, "listSkills").mockResolvedValue({data: []}); const forkSpy = vi.spyOn(client, "forkSession").mockResolvedValue({ sessionId: "fork-id", currentModelId: "gpt-5[medium]", @@ -28,6 +29,7 @@ describe("ACP session fork", () => { expect(response.sessionId).toBe("fork-id"); expect(agent.getSessionState("fork-id").cwd).toBe("/workspace"); + expect(fixture.getAcpConnectionEvents([])).toEqual([]); expect(forkSpy).toHaveBeenCalledWith({ sessionId: "source-id", cwd: "/workspace", From 155c1589235356a9798db9a6cbce1e5e0bb7f7b6 Mon Sep 17 00:00:00 2001 From: Nikita Ashikhmin Date: Thu, 27 Aug 2026 19:49:57 +0400 Subject: [PATCH 6/6] refactor: extract session fork logic --- src/CodexAcpClient.ts | 121 ++++--------------------------------- src/SessionFork.ts | 134 +++++++++++++++++++++++++++++++++++++++++ src/SessionMetadata.ts | 17 ++++++ 3 files changed, 164 insertions(+), 108 deletions(-) create mode 100644 src/SessionFork.ts create mode 100644 src/SessionMetadata.ts diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 2e9ae4fb..68b74e00 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -27,7 +27,6 @@ import type {JsonValue} from "./app-server/serde_json/JsonValue"; import {ModelId} from "./ModelId"; import {AgentMode} from "./AgentMode"; import path from "node:path"; -import {createHash} from "node:crypto"; import {logger} from "./Logger"; import {sanitizeMcpServerName} from "./McpServerName"; import type { @@ -65,6 +64,9 @@ import { createUnavailableAgentFileChangeReport, } from "./AgentFileChangeReport"; import {CodexSubagentSubscriptions} from "./subagents/CodexSubagentSubscriptions"; +import {forkSession as runForkSession} from "./SessionFork"; +import type {SessionMetadata, SessionMetadataWithThread} from "./SessionMetadata"; +export type {SessionMetadata, SessionMetadataWithThread} from "./SessionMetadata"; /** * Well-known provider id for the client-configurable custom LLM gateway. @@ -490,52 +492,17 @@ export class CodexAcpClient { async forkSession(request: acp.ForkSessionRequest): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); - await this.refreshSkills(request.cwd, additionalDirectories); - const forkPoint = readAirForkPoint(request._meta); - let lastTurnId: string | undefined; - if (forkPoint) { - const history = await this.codexClient.threadRead({ - threadId: request.sessionId, - includeTurns: true, - }); - const candidateIds = airForkMessageIdCandidates(forkPoint.messageId); - lastTurnId = candidateIds - .map(candidateId => history.thread.turns.find(turn => turn.items.some(item => item.id === candidateId))?.id) - .find(turnId => turnId !== undefined); - if (!lastTurnId && forkPoint.messageFingerprint) { - const matchingTurns = history.thread.turns.flatMap(turn => turn.items - .filter(item => item.type === "agentMessage" - && fingerprintAgentMessage(item.text) === forkPoint.messageFingerprint) - .map(() => turn.id)); - lastTurnId = matchingTurns[forkPoint.messageOccurrence - 1]; - } - if (!lastTurnId) { - throw RequestError.invalidParams( - {messageId: forkPoint.messageId}, - `Fork point message ${forkPoint.messageId} was not found in session ${request.sessionId}`, - ); - } - } - - const response = await this.codexClient.threadFork({ - config: await this.createSessionConfig(request.cwd, additionalDirectories, request.mcpServers ?? []), - cwd: request.cwd, - ...(lastTurnId !== undefined && {lastTurnId}), - modelProvider: await this.getResumeModelProvider(), - threadId: request.sessionId, + return await runForkSession(request, additionalDirectories, { + codexClient: this.codexClient, + refreshSkills: (cwd, directories) => this.refreshSkills(cwd, directories), + createSessionConfig: (cwd, directories, mcpServers) => + this.createSessionConfig(cwd, directories, mcpServers), + getResumeModelProvider: () => this.getResumeModelProvider(), + fetchAvailableModels: () => this.fetchAvailableModels(), + createCurrentModelId: (models, model, reasoningEffort) => + this.createModelId(models, model, reasoningEffort).toString(), + getCollaborationMode: sessionId => this.getCollaborationMode(sessionId), }); - await this.codexClient.threadUnsubscribe({threadId: response.thread.id}); - const codexModels = await this.fetchAvailableModels(); - const currentModelId = this.createModelId(codexModels, response.model, response.reasoningEffort).toString(); - return { - sessionId: response.thread.id, - currentModelId, - models: codexModels, - collaborationMode: this.getCollaborationMode(response.thread.id), - modelProvider: response.modelProvider, - currentServiceTier: response.serviceTier as ServiceTier ?? null, - additionalDirectories, - }; } async loadSession(request: acp.LoadSessionRequest, onSubscribed?: () => void): Promise { @@ -1274,20 +1241,6 @@ class AgentFileChangeReportBudget { export type JsonObject = { [key in string]?: JsonValue } -export type SessionMetadata = { - sessionId: string, - currentModelId: string, - models: Model[], - collaborationMode: ModeKind, - modelProvider?: string | null, - currentServiceTier?: ServiceTier | null, - additionalDirectories: string[], -} - -export type SessionMetadataWithThread = SessionMetadata & { - thread: Thread, -} - function buildPromptItems(prompt: acp.ContentBlock[]): UserInput[] { return prompt.map((block): UserInput | null => { switch (block.type) { @@ -1381,54 +1334,6 @@ function readMetaAdditionalRoots(meta?: Record | null): string[ .filter(value => value.length > 0)); } -type AirForkPoint = { - messageId: string; - messageFingerprint?: string; - messageOccurrence: number; -}; - -function readAirForkPoint(meta?: Record | null): AirForkPoint | undefined { - const jetbrains = meta?.["jetbrains"]; - if (!isUnknownRecord(jetbrains)) return undefined; - const air = jetbrains["air"]; - if (!isUnknownRecord(air)) return undefined; - const fork = air["fork"]; - if (!isUnknownRecord(fork) || fork["version"] !== 1) return undefined; - const messageId = fork["messageId"]; - if (typeof messageId !== "string" || messageId.trim().length === 0) { - throw RequestError.invalidParams(undefined, "AIR fork messageId must be a non-empty string"); - } - const messageFingerprint = fork["messageFingerprint"]; - if (messageFingerprint !== undefined - && (typeof messageFingerprint !== "string" || !/^sha256:[0-9a-f]{64}$/.test(messageFingerprint))) { - throw RequestError.invalidParams(undefined, "AIR fork messageFingerprint must be a SHA-256 fingerprint"); - } - const messageOccurrence = fork["messageOccurrence"] ?? 1; - if (!Number.isSafeInteger(messageOccurrence) || (messageOccurrence as number) < 1) { - throw RequestError.invalidParams(undefined, "AIR fork messageOccurrence must be a positive integer"); - } - return { - messageId: messageId.trim(), - ...(typeof messageFingerprint === "string" && {messageFingerprint}), - messageOccurrence: messageOccurrence as number, - }; -} - -function fingerprintAgentMessage(text: string): string { - return `sha256:${createHash("sha256").update(text, "utf8").digest("hex")}`; -} - -function airForkMessageIdCandidates(messageId: string): string[] { - // Older AIR builds sent their visible segment id. Prefer the exact id before its ACP source id. - const visibleSegmentSuffix = /:segment:\d+$/; - const protocolMessageId = messageId.replace(visibleSegmentSuffix, ""); - return protocolMessageId === messageId ? [messageId] : [messageId, protocolMessageId]; -} - -function isUnknownRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null && !Array.isArray(value); -} - function readAdditionalDirectories(cwd: string, additionalDirectories?: string[], meta?: Record | null): string[] { const rawDirectories = additionalDirectories ?? readMetaAdditionalRoots(meta); if (!rawDirectories) { diff --git a/src/SessionFork.ts b/src/SessionFork.ts new file mode 100644 index 00000000..b52b7cae --- /dev/null +++ b/src/SessionFork.ts @@ -0,0 +1,134 @@ +import {createHash} from "node:crypto"; +import type * as acp from "@agentclientprotocol/sdk"; +import {RequestError} from "@agentclientprotocol/sdk"; +import type {CodexAppServerClient} from "./CodexAppServerClient"; +import type {ModeKind} from "./app-server/ModeKind"; +import type {ServiceTier} from "./app-server/ServiceTier"; +import type {Model, ThreadForkParams} from "./app-server/v2"; +import type {SessionMetadata} from "./SessionMetadata"; + +export type SessionForkDependencies = { + codexClient: CodexAppServerClient; + refreshSkills(cwd: string, additionalDirectories: string[]): Promise; + createSessionConfig( + cwd: string, + additionalDirectories: string[], + mcpServers: acp.McpServer[], + ): Promise>; + getResumeModelProvider(): Promise; + fetchAvailableModels(): Promise; + createCurrentModelId(models: Model[], model: string, reasoningEffort: string | null): string; + getCollaborationMode(sessionId: string): ModeKind; +}; + +export async function forkSession( + request: acp.ForkSessionRequest, + additionalDirectories: string[], + dependencies: SessionForkDependencies, +): Promise { + await dependencies.refreshSkills(request.cwd, additionalDirectories); + const lastTurnId = await resolveForkTurnId(request, dependencies.codexClient); + const response = await dependencies.codexClient.threadFork({ + config: await dependencies.createSessionConfig( + request.cwd, + additionalDirectories, + request.mcpServers ?? [], + ), + cwd: request.cwd, + ...(lastTurnId !== undefined && {lastTurnId}), + modelProvider: await dependencies.getResumeModelProvider(), + threadId: request.sessionId, + }); + await dependencies.codexClient.threadUnsubscribe({threadId: response.thread.id}); + + const models = await dependencies.fetchAvailableModels(); + return { + sessionId: response.thread.id, + currentModelId: dependencies.createCurrentModelId(models, response.model, response.reasoningEffort), + models, + collaborationMode: dependencies.getCollaborationMode(response.thread.id), + modelProvider: response.modelProvider, + currentServiceTier: response.serviceTier as ServiceTier ?? null, + additionalDirectories, + }; +} + +async function resolveForkTurnId( + request: acp.ForkSessionRequest, + codexClient: CodexAppServerClient, +): Promise { + const forkPoint = readAirForkPoint(request._meta); + if (!forkPoint) return undefined; + + const history = await codexClient.threadRead({ + threadId: request.sessionId, + includeTurns: true, + }); + const candidateIds = airForkMessageIdCandidates(forkPoint.messageId); + const itemTurnId = candidateIds + .map(candidateId => history.thread.turns.find(turn => turn.items.some(item => item.id === candidateId))?.id) + .find(turnId => turnId !== undefined); + if (itemTurnId) return itemTurnId; + + if (forkPoint.messageFingerprint) { + const matchingTurns = history.thread.turns.flatMap(turn => turn.items + .filter(item => item.type === "agentMessage" + && fingerprintAgentMessage(item.text) === forkPoint.messageFingerprint) + .map(() => turn.id)); + const fingerprintTurnId = matchingTurns[forkPoint.messageOccurrence - 1]; + if (fingerprintTurnId) return fingerprintTurnId; + } + + throw RequestError.invalidParams( + {messageId: forkPoint.messageId}, + `Fork point message ${forkPoint.messageId} was not found in session ${request.sessionId}`, + ); +} + +type AirForkPoint = { + messageId: string; + messageFingerprint?: string; + messageOccurrence: number; +}; + +function readAirForkPoint(meta?: Record | null): AirForkPoint | undefined { + const jetbrains = meta?.["jetbrains"]; + if (!isUnknownRecord(jetbrains)) return undefined; + const air = jetbrains["air"]; + if (!isUnknownRecord(air)) return undefined; + const fork = air["fork"]; + if (!isUnknownRecord(fork) || fork["version"] !== 1) return undefined; + const messageId = fork["messageId"]; + if (typeof messageId !== "string" || messageId.trim().length === 0) { + throw RequestError.invalidParams(undefined, "AIR fork messageId must be a non-empty string"); + } + const messageFingerprint = fork["messageFingerprint"]; + if (messageFingerprint !== undefined + && (typeof messageFingerprint !== "string" || !/^sha256:[0-9a-f]{64}$/.test(messageFingerprint))) { + throw RequestError.invalidParams(undefined, "AIR fork messageFingerprint must be a SHA-256 fingerprint"); + } + const messageOccurrence = fork["messageOccurrence"] ?? 1; + if (!Number.isSafeInteger(messageOccurrence) || (messageOccurrence as number) < 1) { + throw RequestError.invalidParams(undefined, "AIR fork messageOccurrence must be a positive integer"); + } + return { + messageId: messageId.trim(), + ...(typeof messageFingerprint === "string" && {messageFingerprint}), + messageOccurrence: messageOccurrence as number, + }; +} + +function fingerprintAgentMessage(text: string): string { + return `sha256:${createHash("sha256").update(text, "utf8").digest("hex")}`; +} + +function airForkMessageIdCandidates(messageId: string): string[] { + // Older AIR builds sent their visible segment id. Prefer the exact id before its ACP source id. + const visibleSegmentSuffix = /:segment:\d+$/; + const protocolMessageId = messageId.replace(visibleSegmentSuffix, ""); + return protocolMessageId === messageId ? [messageId] : [messageId, protocolMessageId]; +} + +function isUnknownRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/SessionMetadata.ts b/src/SessionMetadata.ts new file mode 100644 index 00000000..50562058 --- /dev/null +++ b/src/SessionMetadata.ts @@ -0,0 +1,17 @@ +import type {ModeKind} from "./app-server/ModeKind"; +import type {ServiceTier} from "./app-server/ServiceTier"; +import type {Model, Thread} from "./app-server/v2"; + +export type SessionMetadata = { + sessionId: string, + currentModelId: string, + models: Model[], + collaborationMode: ModeKind, + modelProvider?: string | null, + currentServiceTier?: ServiceTier | null, + additionalDirectories: string[], +} + +export type SessionMetadataWithThread = SessionMetadata & { + thread: Thread, +}