diff --git a/AGENTS.md b/AGENTS.md index 56f93557..dc075fdb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,7 +34,11 @@ - 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. +- Session fork uses app-server `thread/fork` and installs the returned child as an independent ACP + session. The temporary `_meta.lody.forkAtMessage` extension carries a standard ACP `messageId`; + resolve the containing Codex turn with `thread/read` inside this adapter before setting + `thread/fork.lastTurnId`. Never expose Codex turn IDs as ACP message IDs or 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/src/AcpExtensions.ts b/src/AcpExtensions.ts index 1f8f15a3..5ba7dfd6 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -12,6 +12,21 @@ export const ACP_EXT_SESSION_USAGE_UPDATE_METHOD = "_acp_ext:session_usage_updat export const ACP_EXT_SESSION_RATE_LIMITS_METHOD = "_acp_ext:session_rate_limits"; export const ACP_EXT_CODEX_PROPOSED_PLAN_METHOD = "_acp_ext:codex_proposed_plan"; export const CODEX_STEER_APPLIED_METHOD = "_codex/steerApplied"; +export const LODY_FORK_MESSAGE_BEFORE_ACTIVE_TURN_METHOD = + "_lody/session/fork-message-before-active-turn"; + +export function getLodyForkMessageId(meta: unknown): string | null { + if (typeof meta !== "object" || meta === null) return null; + const lody = (meta as Record)["lody"]; + if (typeof lody !== "object" || lody === null) return null; + const forkAtMessage = (lody as Record)["forkAtMessage"]; + if (typeof forkAtMessage !== "object" || forkAtMessage === null) return null; + const version = (forkAtMessage as Record)["version"]; + const messageId = (forkAtMessage as Record)["messageId"]; + return version === 1 && typeof messageId === "string" && messageId.length > 0 + ? messageId + : null; +} export type CodexSteerCapability = { version: 1; diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index a780eecd..80df60bd 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -23,6 +23,7 @@ import {AgentMode} from "./AgentMode"; import path from "node:path"; import {logger} from "./Logger"; import {sanitizeMcpServerName} from "./McpServerName"; +import {getLodyForkMessageId} from "./AcpExtensions"; import type { AccountLoginCompletedNotification, AccountUpdatedNotification, @@ -355,11 +356,16 @@ export class CodexAcpClient { ): Promise { const additionalDirectories = readAdditionalDirectories(request.cwd, request.additionalDirectories, request._meta); await this.refreshSkills(request.cwd, additionalDirectories); + const forkMessageId = getLodyForkMessageId(request._meta); + const forkTurnId = forkMessageId + ? await this.resolveForkTurnIdForMessage(request.sessionId, forkMessageId) + : null; const response = await this.codexClient.threadFork({ config: await this.createSessionConfig(request.cwd, additionalDirectories, request.mcpServers ?? []), cwd: request.cwd, excludeTurns: true, + ...(forkTurnId ? {lastTurnId: forkTurnId} : {}), modelProvider: await this.getResumeModelProvider(), threadId: request.sessionId, }); @@ -376,6 +382,44 @@ export class CodexAcpClient { }; } + async resolveForkTurnIdForMessage(threadId: string, messageId: string): Promise { + const response = await this.codexClient.threadRead({ + threadId, + includeTurns: true, + }); + const turn = response.thread.turns.find((candidate) => + candidate.items.some((item) => item.type === "agentMessage" && item.id === messageId) + ); + if (!turn) { + throw RequestError.invalidRequest("ACP message is not a forkable Codex turn boundary"); + } + return turn.id; + } + + async findMessageBeforeTurn(threadId: string, activeTurnId: string): Promise { + const response = await this.codexClient.threadRead({ + threadId, + includeTurns: true, + }); + const turns = response.thread.turns; + const activeIndex = turns.findIndex((turn) => turn.id === activeTurnId); + if (activeIndex < 0) { + throw RequestError.invalidRequest("Active turn changed before its preceding message was captured"); + } + for (let index = activeIndex - 1; index >= 0; index--) { + const turn = turns[index]; + if (turn && turn.status !== "inProgress") { + const agentMessage = [...turn.items] + .reverse() + .find((item) => item.type === "agentMessage"); + if (agentMessage) { + return agentMessage.id; + } + } + } + throw RequestError.invalidRequest("No completed assistant message exists before the active turn"); + } + 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 7486ff55..fc18f307 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -47,6 +47,7 @@ import { getCodexSteerId, isExtMethodRequest, LEGACY_SET_SESSION_MODEL_METHOD, + LODY_FORK_MESSAGE_BEFORE_ACTIVE_TURN_METHOD, } from "./AcpExtensions"; import { createCollabAgentToolCallUpdate, @@ -266,6 +267,9 @@ export class CodexAcpServer { codex: { steer: CODEX_STEER_CAPABILITY, }, + lody: { + forkAtMessage: {version: 1, beforeActiveTurn: true}, + }, }, }, authMethods: getCodexAuthMethods(_params.clientCapabilities), @@ -665,6 +669,22 @@ export class CodexAcpServer { }; } + async resolveMessageBeforeActiveTurn(params: {sessionId: string}): Promise<{messageId: string}> { + const session = this.sessions.get(params.sessionId); + const activeTurnId = session?.currentTurnId; + if (!activeTurnId) { + throw RequestError.invalidRequest("Session has no active turn"); + } + const messageId = await this.runWithProcessCheck(() => + this.codexAcpClient.findMessageBeforeTurn(params.sessionId, activeTurnId) + ); + logger.log("Resolved ACP message before active turn", { + sessionId: params.sessionId, + method: LODY_FORK_MESSAGE_BEFORE_ACTIVE_TURN_METHOD, + }); + return {messageId}; + } + async listSessions(params: acp.ListSessionsRequest): Promise { logger.log("Listing sessions...", {cwd: params.cwd, cursor: params.cursor}); await this.checkAuthorization(); @@ -1997,7 +2017,7 @@ export class CodexAcpServer { quota: { token_count: sessionState.lastTokenUsage, model_usage: modelUsage - } + }, }; } diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 75785673..1c15c11c 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -69,6 +69,12 @@ describe('CodexACPAgent - initialize', () => { configPolicy: "active", }, }, + lody: { + forkAtMessage: { + version: 1, + beforeActiveTurn: true, + }, + }, }, }, authMethods: getCodexAuthMethods(), diff --git a/src/__tests__/CodexACPAgent/session-fork.test.ts b/src/__tests__/CodexACPAgent/session-fork.test.ts index 8841bf63..a424fd92 100644 --- a/src/__tests__/CodexACPAgent/session-fork.test.ts +++ b/src/__tests__/CodexACPAgent/session-fork.test.ts @@ -21,7 +21,14 @@ describe("ACP session fork", () => { 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 threadReadSpy = vi.spyOn(codexAppServerClient, "threadRead").mockResolvedValue({ + thread: { + turns: [{ + id: "completed-turn-id", + items: [{type: "agentMessage", id: "assistant-message-id"}], + }], + }, + } as never); const threadForkSpy = vi.spyOn(codexAppServerClient, "threadFork").mockResolvedValue({ thread: {id: "child-session-id"}, model: model.id, @@ -40,6 +47,14 @@ describe("ACP session fork", () => { cwd: "/workspace", additionalDirectories: ["/workspace/extra"], mcpServers: [mcpServer], + _meta: { + lody: { + forkAtMessage: { + version: 1, + messageId: "assistant-message-id", + }, + }, + }, }, subscribed); expect(result).toEqual({ @@ -53,6 +68,7 @@ describe("ACP session fork", () => { expect(subscribed).toHaveBeenCalledWith("child-session-id"); expect(threadForkSpy).toHaveBeenCalledWith({ threadId: "source-session-id", + lastTurnId: "completed-turn-id", cwd: "/workspace", excludeTurns: true, modelProvider: "openai", @@ -73,7 +89,47 @@ describe("ACP session fork", () => { }, }, }); - expect(threadReadSpy).not.toHaveBeenCalled(); + expect(threadReadSpy).toHaveBeenCalledWith({ + threadId: "source-session-id", + includeTurns: true, + }); + }); + + it("resolves the last terminal Codex turn before the active turn", async () => { + const fixture = createCodexMockTestFixture(); + const codexAcpClient = fixture.getCodexAcpClient(); + const codexAppServerClient = fixture.getCodexAppServerClient(); + vi.spyOn(codexAppServerClient, "threadRead").mockResolvedValue({ + thread: { + turns: [ + { + id: "completed-turn", + status: "completed", + items: [{type: "agentMessage", id: "assistant-message-id"}], + }, + {id: "active-turn", status: "inProgress", items: []}, + ], + }, + } as never); + + await expect( + codexAcpClient.findMessageBeforeTurn("source-session-id", "active-turn"), + ).resolves.toBe("assistant-message-id"); + }); + + it("rejects when the active Codex turn changed during capture", async () => { + const fixture = createCodexMockTestFixture(); + const codexAcpClient = fixture.getCodexAcpClient(); + const codexAppServerClient = fixture.getCodexAppServerClient(); + vi.spyOn(codexAppServerClient, "threadRead").mockResolvedValue({ + thread: { + turns: [{id: "completed-turn", status: "completed"}], + }, + } as never); + + await expect( + codexAcpClient.findMessageBeforeTurn("source-session-id", "stale-active-turn"), + ).rejects.toThrow("Invalid request"); }); it("installs the fork as an independent promptable ACP session", async () => { diff --git a/src/index.ts b/src/index.ts index 5d33486b..cc5bc28c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -12,7 +12,10 @@ import packageJson from "../package.json"; import {logger} from "./Logger"; import {runLoginCommand} from "./login"; import {runCodexCli} from "./CodexCli"; -import {LEGACY_SET_SESSION_MODEL_METHOD} from "./AcpExtensions"; +import { + LEGACY_SET_SESSION_MODEL_METHOD, + LODY_FORK_MESSAGE_BEFORE_ACTIVE_TURN_METHOD, +} from "./AcpExtensions"; const emptyExtensionParamsParser = z.preprocess( (params) => params ?? {}, @@ -24,6 +27,10 @@ const legacySetSessionModelParamsParser = z.object({ modelId: z.string(), }).passthrough(); +const activeTurnForkMessageParamsParser = z.object({ + sessionId: z.string().min(1), +}); + if (process.argv.includes("--version")) { console.log(`${packageJson.name} ${packageJson.version}`); process.exit(0); @@ -133,5 +140,8 @@ function startAcpServer() { .onRequest("authentication/status", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/status", ctx.params)) .onRequest("authentication/logout", emptyExtensionParamsParser, (ctx) => getAgent().extMethod("authentication/logout", ctx.params)) .onRequest(LEGACY_SET_SESSION_MODEL_METHOD, legacySetSessionModelParamsParser, (ctx) => getAgent().extMethod(LEGACY_SET_SESSION_MODEL_METHOD, ctx.params)) + .onRequest(LODY_FORK_MESSAGE_BEFORE_ACTIVE_TURN_METHOD, activeTurnForkMessageParamsParser, (ctx) => + getAgent().resolveMessageBeforeActiveTurn(ctx.params) + ) .connect(acpJsonStream); }