From 75a5f9df2697d4c842d433277eed740a87467637 Mon Sep 17 00:00:00 2001 From: Leeeon233 Date: Tue, 28 Jul 2026 10:07:44 +0800 Subject: [PATCH 1/2] feat: support ACP session fork Map ACP session/fork directly to Codex thread/fork, advertise the capability, and install forked threads through the shared session lifecycle. Model: GPT-5 --- AGENTS.md | 1 + README.md | 1 + src/CodexAcpClient.ts | 39 ++++- src/CodexAcpServer.ts | 149 +++++++++++++---- src/CodexAppServerClient.ts | 6 + .../CodexACPAgent/e2e/acp-e2e.test.ts | 27 ++++ .../CodexACPAgent/initialize.test.ts | 1 + .../CodexACPAgent/session-fork.test.ts | 153 ++++++++++++++++++ src/index.ts | 1 + 9 files changed, 345 insertions(+), 33 deletions(-) create mode 100644 src/__tests__/CodexACPAgent/session-fork.test.ts diff --git a/AGENTS.md b/AGENTS.md index f1cc5062..56f93557 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,6 +34,7 @@ - Codex app-server usage: see https://github.com/openai/codex/blob/main/codex-rs/app-server/README.md when touching protocol/transport details, adding or consuming JSON-RPC methods, handling approvals/turn events, or updating generated schema/clients. - App-server events: prefer `thread/*`, `turn/*`, and `item/*` event surfaces; avoid the deprecated `codex/event/*` API (planned removal). Keep implementations aligned with generated types in `src/app-server` (including `v2` exports). - Steer uses app-server `turn/steer` on the tracked active turn. Correlate `clientUserMessageId` and acknowledge only the matching `item/completed(userMessage)`; never emulate steer with a second `turn/start`. +- Session fork uses app-server `thread/fork` and installs the returned child as an independent ACP session. Never emulate fork by replaying source history. - Codex reasoning summaries can echo trailing empty HTML comments from model instructions. Keep that provider-specific cleanup in `src/ReasoningText.ts` across live deltas and history replay; do not filter assistant text, raw reasoning, or HTML globally in the client renderer. diff --git a/README.md b/README.md index 1e7418b2..c8a0b977 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol] - Text prompts, embedded context, images, resource links, and additional workspace directories. - Shell command, file change, permission request, MCP tool call, terminal output, reasoning, plan, web search, image generation, image view, token usage, and review events. - Client-provided MCP servers over command-based stdio config and HTTP transport. +- Native ACP session forking through Codex App Server `thread/fork`. - Acknowledged steering of an active Codex turn through app-server `turn/steer`. - Slash commands: `/status`, `/mcp`, `/skills`, `/review`, `/review-branch`, `/review-commit`, `/compact`, and `/logout`, as well as configured skills. diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 7ae09a92..37dc3ca4 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -323,7 +323,10 @@ export class CodexAcpClient { return this.codexClient.accountRateLimitsRead(); } - async resumeSession(request: acp.ResumeSessionRequest, onSubscribed?: () => void): Promise { + async resumeSession( + request: acp.ResumeSessionRequest, + onSubscribed?: (sessionId?: string) => void, + ): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); @@ -333,7 +336,7 @@ export class CodexAcpClient { modelProvider: await this.getResumeModelProvider(), threadId: request.sessionId, }); - onSubscribed?.(); + onSubscribed?.(request.sessionId); const codexModels = await this.fetchAvailableModels(); const currentModelId = this.createModelId(codexModels, response.model, response.reasoningEffort).toString(); return { @@ -346,6 +349,32 @@ export class CodexAcpClient { } } + async forkSession( + request: acp.ForkSessionRequest, + onSubscribed?: (sessionId?: string) => void, + ): 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, + }); + onSubscribed?.(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, + 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); @@ -374,7 +403,10 @@ export class CodexAcpClient { }; } - async newSession(request: acp.NewSessionRequest): Promise { + async newSession( + request: acp.NewSessionRequest, + onSubscribed?: (sessionId?: string) => void, + ): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); @@ -383,6 +415,7 @@ export class CodexAcpClient { modelProvider: this.getModelProvider(), cwd: request.cwd, }); + onSubscribed?.(response.thread.id); const codexModels = await this.fetchAvailableModels(); if (codexModels.length === 0) { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 973f0098..7486ff55 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -125,6 +125,18 @@ interface ActiveAuthState { authConfigured: boolean; } +type SessionOpenOperation = + | { kind: "new"; request: acp.NewSessionRequest } + | { kind: "resume"; request: acp.ResumeSessionRequest } + | { kind: "fork"; request: acp.ForkSessionRequest }; + +type SessionOpenResult = [ + SessionId, + LegacySessionModelState, + SessionModeState, + acp.AvailableCommand[], +]; + interface PendingMcpStartupSession { requestedServers: Set; afterVersion: number; @@ -239,6 +251,7 @@ export class CodexAcpServer { }, sessionCapabilities: { resume: { }, + fork: { }, list: { }, close: { }, delete: { }, @@ -293,7 +306,7 @@ export class CodexAcpServer { } } - async getOrCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState, acp.AvailableCommand[]]> { + async getOrCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise { try { return await this.tryCreateSession(request); } catch (e) { @@ -303,6 +316,16 @@ export class CodexAcpServer { } } + private async getOrForkSession(request: acp.ForkSessionRequest): Promise { + try { + return await this.tryOpenSession({kind: "fork", request}); + } catch (e) { + const error = e instanceof Error ? e : new Error(String(e)); + await this.handleError(error); + throw e; + } + } + async handleError(e: Error){ if (e.message.includes("log out") || e.message.includes("cloud requirements")) { await this.runWithProcessCheck(() => this.codexAcpClient.logout()); @@ -374,10 +397,35 @@ export class CodexAcpServer { return generation; } - async tryCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState, acp.AvailableCommand[]]> { - const requestedSessionGeneration = "sessionId" in request - ? this.beginSessionOpen(request.sessionId) + async tryCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise { + return await this.tryOpenSession("sessionId" in request + ? {kind: "resume", request} + : {kind: "new", request}); + } + + private async tryOpenSession(operation: SessionOpenOperation): Promise { + const {request} = operation; + let openedSession = operation.kind === "resume" + ? { + sessionId: operation.request.sessionId, + generation: this.beginSessionOpen(operation.request.sessionId), + } : null; + let subscribed = false; + const onSubscribed = (reportedSessionId?: string): void => { + const sessionId = reportedSessionId + ?? (operation.kind === "resume" ? operation.request.sessionId : null); + if (!sessionId) { + throw RequestError.internalError("Codex subscribed without reporting a session id"); + } + subscribed = true; + if (!openedSession) { + openedSession = { + sessionId, + generation: this.beginSessionOpen(sessionId), + }; + } + }; await this.checkAuthorization(); const requestedMcpServers = request.mcpServers ?? []; const mcpServerStartupVersion = requestedMcpServers.length > 0 @@ -385,43 +433,68 @@ export class CodexAcpServer { : null; let sessionMetadata: SessionMetadata; - let resumeSubscribed = false; - if ("sessionId" in request) { - logger.log(`Resume existing session: ${request.sessionId}...`); - try { - sessionMetadata = await this.runWithProcessCheck(() => - this.codexAcpClient.resumeSession(request, () => { - resumeSubscribed = true; - }) - ); - } catch (err) { - if (resumeSubscribed && requestedSessionGeneration !== null) { - await this.cleanupStaleSessionOpen(request.sessionId, requestedSessionGeneration); - } - throw err; + try { + switch (operation.kind) { + case "new": + logger.log("Create new session..."); + sessionMetadata = await this.runWithProcessCheck(() => + this.codexAcpClient.newSession(operation.request, onSubscribed) + ); + break; + case "resume": + logger.log(`Resume existing session: ${operation.request.sessionId}...`); + sessionMetadata = await this.runWithProcessCheck(() => + this.codexAcpClient.resumeSession(operation.request, onSubscribed) + ); + break; + case "fork": + logger.log(`Fork existing session: ${operation.request.sessionId}...`); + sessionMetadata = await this.runWithProcessCheck(() => + this.codexAcpClient.forkSession(operation.request, onSubscribed) + ); + break; } - } else { - logger.log(`Create new session...`); - sessionMetadata = await this.runWithProcessCheck(() => this.codexAcpClient.newSession(request)); + } catch (err) { + if (subscribed && openedSession) { + await this.cleanupStaleSessionOpen(openedSession.sessionId, openedSession.generation); + } + throw err; } const {sessionId, currentModelId, models} = sessionMetadata; + if (!openedSession) { + openedSession = { + sessionId, + generation: this.beginSessionOpen(sessionId), + }; + } else if (openedSession.sessionId !== sessionId) { + if (subscribed) { + await this.cleanupStaleSessionOpen(openedSession.sessionId, openedSession.generation); + } + throw RequestError.internalError( + {expectedSessionId: openedSession?.sessionId, actualSessionId: sessionId}, + "Codex opened a different session than it reported", + ); + } + subscribed = true; const authProvider = sessionMetadata.modelProvider ?? this.codexAcpClient.getModelProvider(); let authState: ActiveAuthState; try { authState = await this.getAuthStateForProvider(authProvider); } catch (err) { - if (resumeSubscribed && requestedSessionGeneration !== null) { - await this.cleanupStaleSessionOpen(sessionId, requestedSessionGeneration); + if (subscribed) { + await this.cleanupStaleSessionOpen(sessionId, openedSession.generation); } throw err; } - const sessionGeneration = requestedSessionGeneration ?? this.beginSessionOpen(sessionId); - if (!this.sessionOpenCanInstall(sessionId, sessionGeneration)) { - resumeSubscribed = false; - await this.closeStaleSessionOpen(sessionId, sessionGeneration); + if (!this.sessionOpenCanInstall(sessionId, openedSession.generation)) { + subscribed = false; + await this.closeStaleSessionOpen(sessionId, openedSession.generation); } - const sessionMcpServers = this.resolveSessionMcpServers(requestedMcpServers, "sessionId" in request); + const sessionMcpServers = this.resolveSessionMcpServers( + requestedMcpServers, + operation.kind !== "new", + ); const currentModel = this.findCurrentModel(models, currentModelId); const currentModelSupportsFast = modelSupportsFast(currentModel); const sessionState: SessionState = { @@ -448,11 +521,11 @@ export class CodexAcpServer { sessionMcpServers: sessionMcpServers, terminalOutputMode: this.terminalOutputMode, sessionTitle: null, - sessionTitleSource: "sessionId" in request ? "unknown" : "unset", + sessionTitleSource: operation.kind === "new" ? "unset" : "unknown", }; this.sessions.set(sessionId, sessionState); this.publishRateLimitsAsync(sessionState); - resumeSubscribed = false; + subscribed = false; if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { this.pendingMcpStartupSessions.set(sessionId, { @@ -576,6 +649,22 @@ export class CodexAcpServer { }; } + async unstable_forkSession(params: acp.ForkSessionRequest): Promise { + logger.log("Forking session...", {sessionId: params.sessionId}); + const [sessionId, , modeState, availableCommands] = await this.getOrForkSession(params); + this.publishAvailableCommandsAsync(sessionId, availableCommands); + + logger.log("Session forked", { + sourceSessionId: params.sessionId, + sessionId, + }); + return { + sessionId, + modes: modeState, + ...this.createSessionConfigOptionsResponse(this.getSessionState(sessionId)), + }; + } + async listSessions(params: acp.ListSessionsRequest): Promise { logger.log("Listing sessions...", {cwd: params.cwd, cursor: params.cursor}); await this.checkAuthorization(); diff --git a/src/CodexAppServerClient.ts b/src/CodexAppServerClient.ts index 38a40c13..33fe4996 100644 --- a/src/CodexAppServerClient.ts +++ b/src/CodexAppServerClient.ts @@ -40,6 +40,8 @@ import type { ThreadGoalClearResponse, ThreadGoalSetParams, ThreadGoalSetResponse, + ThreadForkParams, + ThreadForkResponse, ThreadLoadedListParams, ThreadLoadedListResponse, ThreadListParams, @@ -520,6 +522,10 @@ export class CodexAppServerClient { return await this.sendRequest({ method: "thread/resume", params: params }); } + async threadFork(params: ThreadForkParams): Promise { + return await this.sendRequest({ method: "thread/fork", params: params }); + } + async threadList(params: ThreadListParams): Promise { return await this.sendRequest({ method: "thread/list", params: params }); } diff --git a/src/__tests__/CodexACPAgent/e2e/acp-e2e.test.ts b/src/__tests__/CodexACPAgent/e2e/acp-e2e.test.ts index 9d5284c5..4e7b72c4 100644 --- a/src/__tests__/CodexACPAgent/e2e/acp-e2e.test.ts +++ b/src/__tests__/CodexACPAgent/e2e/acp-e2e.test.ts @@ -30,6 +30,33 @@ describeE2E("E2E tests", () => { }); }); + it("forks a session and continues from the copied Codex history", async () => { + fixture = await createAuthenticatedFixture(); + const source = await fixture.createSession(); + await fixture.expectPromptText( + source.sessionId, + "Remember the exact marker fork-history-ok and reply with exactly remembered.", + (text) => { + expect(text.toLowerCase()).toContain("remembered"); + }, + ); + + const forked = await fixture.connection.unstable_forkSession({ + sessionId: source.sessionId, + cwd: fixture.workspaceDir, + mcpServers: [], + }); + + expect(forked.sessionId).not.toBe(source.sessionId); + await fixture.expectPromptText( + forked.sessionId, + "Reply with only the exact marker I asked you to remember.", + (text) => { + expect(text.toLowerCase()).toContain("fork-history-ok"); + }, + ); + }); + it("returns model response when authenticated via gateway", async () => { const apiKey = requireLiveApiKey(); fixture = await createGatewayFixture("https://api.openai.com/v1", {Authorization: `Bearer ${apiKey}`}); diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 76a53176..75785673 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -49,6 +49,7 @@ describe('CodexACPAgent - initialize', () => { }, sessionCapabilities: { resume: {}, + fork: {}, list: {}, close: {}, delete: {}, diff --git a/src/__tests__/CodexACPAgent/session-fork.test.ts b/src/__tests__/CodexACPAgent/session-fork.test.ts new file mode 100644 index 00000000..3de35ced --- /dev/null +++ b/src/__tests__/CodexACPAgent/session-fork.test.ts @@ -0,0 +1,153 @@ +import {describe, expect, it, vi} from "vitest"; +import type {McpServerStdio} from "@agentclientprotocol/sdk"; +import { + createCodexMockTestFixture, + createTestModel, +} from "../acp-test-utils"; + +describe("ACP session fork", () => { + it("maps session/fork to thread/fork with lifecycle configuration", async () => { + const fixture = createCodexMockTestFixture(); + const codexAcpClient = fixture.getCodexAcpClient(); + const codexAppServerClient = fixture.getCodexAppServerClient(); + const model = createTestModel(); + const mcpServer: McpServerStdio = { + name: "fork-mcp", + command: "node", + args: ["server.js"], + env: [{name: "TOKEN", value: "test-token"}], + }; + + vi.spyOn(codexAppServerClient, "skillsExtraRootsSet").mockResolvedValue(undefined); + vi.spyOn(codexAppServerClient, "listSkills").mockResolvedValue({data: []}); + vi.spyOn(codexAppServerClient, "configRead").mockResolvedValue({config: {}} as never); + const threadReadSpy = vi.spyOn(codexAppServerClient, "threadRead"); + const threadForkSpy = vi.spyOn(codexAppServerClient, "threadFork").mockResolvedValue({ + thread: {id: "child-session-id"}, + model: model.id, + modelProvider: "openai", + serviceTier: null, + reasoningEffort: "medium", + } as never); + vi.spyOn(codexAppServerClient, "listModels").mockResolvedValue({ + data: [model], + nextCursor: null, + }); + const subscribed = vi.fn(); + + const result = await codexAcpClient.forkSession({ + sessionId: "source-session-id", + cwd: "/workspace", + additionalDirectories: ["/workspace/extra"], + mcpServers: [mcpServer], + }, subscribed); + + expect(result).toEqual({ + sessionId: "child-session-id", + currentModelId: "model-id[medium]", + models: [model], + modelProvider: "openai", + currentServiceTier: null, + additionalDirectories: ["/workspace/extra"], + }); + expect(subscribed).toHaveBeenCalledWith("child-session-id"); + expect(threadForkSpy).toHaveBeenCalledWith({ + threadId: "source-session-id", + cwd: "/workspace", + modelProvider: "openai", + config: { + projects: { + "/workspace": {trust_level: "trusted"}, + "/workspace/extra": {trust_level: "trusted"}, + }, + sandbox_workspace_write: { + writable_roots: ["/workspace/extra"], + }, + mcp_servers: { + "fork-mcp": { + command: "node", + args: ["server.js"], + env: {TOKEN: "test-token"}, + }, + }, + }, + }); + expect(threadReadSpy).not.toHaveBeenCalled(); + }); + + it("installs the fork as an independent promptable ACP session", async () => { + const fixture = createCodexMockTestFixture(); + const codexAcpAgent = fixture.getCodexAcpAgent(); + const codexAcpClient = fixture.getCodexAcpClient(); + const model = createTestModel(); + + vi.spyOn(codexAcpClient, "authRequired").mockResolvedValue(false); + vi.spyOn(codexAcpClient, "listSkills").mockResolvedValue({data: []}); + const forkSessionSpy = vi.spyOn(codexAcpClient, "forkSession").mockImplementation( + async (_request, onSubscribed) => { + onSubscribed?.("child-session-id"); + return { + sessionId: "child-session-id", + currentModelId: "model-id[medium]", + models: [model], + modelProvider: "custom-provider", + currentServiceTier: null, + additionalDirectories: ["/workspace/extra"], + }; + }, + ); + + const response = await codexAcpAgent.unstable_forkSession({ + sessionId: "source-session-id", + cwd: "/workspace", + additionalDirectories: ["/workspace/extra"], + mcpServers: [], + }); + + expect(forkSessionSpy).toHaveBeenCalledWith( + { + sessionId: "source-session-id", + cwd: "/workspace", + additionalDirectories: ["/workspace/extra"], + mcpServers: [], + }, + expect.any(Function), + ); + expect(response).toEqual(expect.objectContaining({ + sessionId: "child-session-id", + modes: expect.objectContaining({currentModeId: "agent"}), + configOptions: expect.any(Array), + })); + expect(codexAcpAgent.getSessionState("child-session-id")).toEqual(expect.objectContaining({ + sessionId: "child-session-id", + cwd: "/workspace", + additionalDirectories: ["/workspace/extra"], + currentTurnId: null, + })); + }); + + it("unsubscribes a child when fork setup fails after Codex creates it", async () => { + const fixture = createCodexMockTestFixture(); + const codexAcpAgent = fixture.getCodexAcpAgent(); + const codexAcpClient = fixture.getCodexAcpClient(); + + vi.spyOn(codexAcpClient, "authRequired").mockResolvedValue(false); + vi.spyOn(codexAcpClient, "forkSession").mockImplementation( + async (_request, onSubscribed) => { + onSubscribed?.("child-session-id"); + throw new Error("model catalog unavailable"); + }, + ); + const closeSessionSpy = vi.spyOn(codexAcpClient, "closeSession").mockResolvedValue(); + + await expect(codexAcpAgent.unstable_forkSession({ + sessionId: "source-session-id", + cwd: "/workspace", + mcpServers: [], + })).rejects.toThrow("model catalog unavailable"); + + expect(closeSessionSpy).toHaveBeenCalledWith("child-session-id"); + expect(() => codexAcpAgent.getSessionState("child-session-id")) + .toThrow("Session child-session-id not found"); + }); +}); diff --git a/src/index.ts b/src/index.ts index ca0b3d37..5d33486b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -116,6 +116,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().unstable_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 270051899cef17a7adc4e0f02c5ce7d6d0289195 Mon Sep 17 00:00:00 2001 From: Leeeon233 Date: Tue, 28 Jul 2026 10:29:17 +0800 Subject: [PATCH 2/2] chore: bump version to 1.3.0 Prepare the native ACP session fork release. Model: GPT-5 --- package-lock.json | 4 ++-- package.json | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index 0fc966ac..f10ac689 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "acp-extension-codex", - "version": "1.2.1", + "version": "1.3.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "acp-extension-codex", - "version": "1.2.1", + "version": "1.3.0", "license": "Apache-2.0", "dependencies": { "@agentclientprotocol/sdk": "^1.2.1", diff --git a/package.json b/package.json index b4ef2f9b..37bef3ea 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "publishConfig": { "access": "public" }, - "version": "1.2.1", + "version": "1.3.0", "description": "An ACP-compatible coding agent powered by Codex", "main": "dist/index.js", "bin": {