From bad96a987bffda68365fd979ed46b3ec2b6cc0ea Mon Sep 17 00:00:00 2001 From: Kelvin Jayanoris Date: Fri, 31 Jul 2026 11:07:01 +0300 Subject: [PATCH 1/6] feat: add shared App Server live-peer mode --- README.md | 37 +++ package-lock.json | 33 +++ package.json | 2 + src/AcpExtensions.ts | 1 + src/CodexAcpClient.ts | 11 +- src/CodexAcpServer.ts | 117 +++++---- src/CodexEventHandler.ts | 40 ++- src/CodexJsonRpcConnection.ts | 196 +++++++++++++- src/CodexMeta.ts | 45 ++++ src/UserInputMapper.ts | 30 +++ .../CodexACPAgent/initialize.test.ts | 9 + src/__tests__/CodexACPAgent/live-peer.test.ts | 248 ++++++++++++++++++ .../CodexACPAgent/mcp-config-merge.test.ts | 2 +- .../CodexACPAgent/process-exit-error.test.ts | 11 +- .../CodexACPAgent/steer-events.test.ts | 12 + src/__tests__/CodexJsonRpcConnection.test.ts | 115 ++++++++ src/__tests__/acp-test-utils.ts | 4 +- src/index.ts | 14 +- src/login.ts | 2 +- 19 files changed, 856 insertions(+), 73 deletions(-) create mode 100644 src/CodexMeta.ts create mode 100644 src/UserInputMapper.ts create mode 100644 src/__tests__/CodexACPAgent/live-peer.test.ts create mode 100644 src/__tests__/CodexJsonRpcConnection.test.ts diff --git a/README.md b/README.md index 7cf19dae..24df94ba 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,41 @@ Use [OpenAI Codex](https://github.com/openai/codex) from [Agent Client Protocol] `codex-acp` is a stdio ACP agent server. It starts the Codex App Server, translates ACP requests into Codex operations, and maps Codex events back into the client. +## Trunkline live-peer mode + +This downstream branch can connect to an App Server that is already shared with +the stock Codex TUI: + +```bash +CODEX_APP_SERVER_URL=unix:// codex-acp +``` + +`unix://PATH`, `ws://`, and `wss://` endpoints are also supported. + +Live-peer behavior is opt-in through ACP initialization metadata: + +```json +{ + "_meta": { + "codex": { + "livePeer": { + "version": 1 + } + } + } +} +``` + +The initialize response advertises the supported live-peer version and +capabilities. When enabled, loaded sessions continue streaming events and +interaction requests even when the ACP client does not own the active prompt. +Live user-message updates carry `_meta.codex.clientUserMessageId` when Codex +provides one. + +ACP prompt and `_session/steering` requests may set the same metadata field. +Codex persists it on the user item, allowing clients to suppress their own echo +without comparing message text. + ## Features - ChatGPT, API key, and client-provided custom gateway authentication. @@ -56,6 +91,8 @@ The adapter advertises ACP auth methods during initialization. Clients can authe - `INITIAL_AGENT_MODE` - initial mode id: `read-only`, `agent`, or `agent-full-access`. - `NO_BROWSER` - hide browser-based ChatGPT auth when set. - `APP_SERVER_LOGS` - directory for adapter logs. +- `CODEX_APP_SERVER_URL` - connect to an existing App Server over `unix://`, + `unix://PATH`, `ws://`, or `wss://` instead of spawning a private server. ## Development diff --git a/package-lock.json b/package-lock.json index 3847268a..5d817639 100644 --- a/package-lock.json +++ b/package-lock.json @@ -14,6 +14,7 @@ "diff": "^9.0.0", "open": "^11.0.0", "vscode-jsonrpc": "^9.0.1", + "ws": "^8.21.1", "zod": "^4.0.0" }, "bin": { @@ -21,6 +22,7 @@ }, "devDependencies": { "@types/node": "^26.1.0", + "@types/ws": "^8.18.1", "esbuild": "^0.28.1", "mcp-hello-world": "^1.1.2", "tsx": "^4.23.1", @@ -1395,6 +1397,16 @@ "undici-types": "~8.3.0" } }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@typescript/typescript-aix-ppc64": { "version": "7.0.2", "resolved": "https://registry.npmjs.org/@typescript/typescript-aix-ppc64/-/typescript-aix-ppc64-7.0.2.tgz", @@ -4242,6 +4254,27 @@ "dev": true, "license": "ISC" }, + "node_modules/ws": { + "version": "8.21.1", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.1.tgz", + "integrity": "sha512-+0NTnW77fFN/DjQi6k/Sq/Yvk4Sgajw7urW8V+asjXnRgDs9gyGkdb7EzgfhA4goXsRIZKE28fzIXBHEzhuiWw==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/wsl-utils": { "version": "0.3.1", "resolved": "https://registry.npmjs.org/wsl-utils/-/wsl-utils-0.3.1.tgz", diff --git a/package.json b/package.json index fcdaf98f..2d6a8d8c 100644 --- a/package.json +++ b/package.json @@ -56,6 +56,7 @@ "type": "module", "devDependencies": { "@types/node": "^26.1.0", + "@types/ws": "^8.18.1", "esbuild": "^0.28.1", "mcp-hello-world": "^1.1.2", "tsx": "^4.23.1", @@ -68,6 +69,7 @@ "diff": "^9.0.0", "open": "^11.0.0", "vscode-jsonrpc": "^9.0.1", + "ws": "^8.21.1", "zod": "^4.0.0" } } diff --git a/src/AcpExtensions.ts b/src/AcpExtensions.ts index 5d8e4602..8751b6f8 100644 --- a/src/AcpExtensions.ts +++ b/src/AcpExtensions.ts @@ -87,6 +87,7 @@ export async function legacySetSessionModel( export type SessionSteerRequest = { sessionId: SessionId; prompt: ContentBlock[]; + _meta?: Record; } export type SessionSteeringResponse = { diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 0c7af207..c2d23869 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -18,6 +18,7 @@ import type { } from "./app-server"; import type {JsonValue} from "./app-server/serde_json/JsonValue"; import {ModelId} from "./ModelId"; +import {getClientUserMessageId} from "./CodexMeta"; import {AgentMode} from "./AgentMode"; import path from "node:path"; import {logger} from "./Logger"; @@ -700,8 +701,10 @@ export class CodexAcpClient { if (shouldCancel?.()) { return null; } + const clientUserMessageId = getClientUserMessageId(request._meta); return await this.codexClient.runTurn({ threadId: request.sessionId, + ...(clientUserMessageId ? {clientUserMessageId} : {}), input: input, approvalPolicy: agentMode.approvalPolicy, sandboxPolicy: addAdditionalDirectoriesToSandboxPolicy(agentMode.sandboxPolicy, additionalDirectories), @@ -845,10 +848,16 @@ export class CodexAcpClient { }); } - async steerTurn(params: { threadId: string, turnId: string, prompt: acp.ContentBlock[] }): Promise { + async steerTurn(params: { + threadId: string, + turnId: string, + prompt: acp.ContentBlock[], + clientUserMessageId?: string | null, + }): Promise { return await this.codexClient.turnSteer({ threadId: params.threadId, expectedTurnId: params.turnId, + ...(params.clientUserMessageId ? {clientUserMessageId: params.clientUserMessageId} : {}), input: buildPromptItems(params.prompt), }); } diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index 928f2b16..c6cdf5f5 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -90,6 +90,13 @@ import { type ThreadGoalSnapshot, toThreadGoalSnapshot, } from "./ThreadGoalSnapshot"; +import { + createClientUserMessageIdMeta, + getClientUserMessageId, + LIVE_PEER_CAPABILITIES, + requestsLivePeer, +} from "./CodexMeta"; +import {userInputToContentBlocks} from "./UserInputMapper"; export interface SessionState { sessionId: string, @@ -162,6 +169,7 @@ export class CodexAcpServer { private clientCapabilities: acp.ClientCapabilities | null; private terminalOutputMode: TerminalOutputMode; private booleanConfigOptionsSupported: boolean; + private livePeerEnabled: boolean; private readonly sessions: Map; private readonly pendingMcpStartupSessions: Map; @@ -196,6 +204,7 @@ export class CodexAcpServer { this.clientCapabilities = null; this.terminalOutputMode = "terminal_output_delta"; this.booleanConfigOptionsSupported = false; + this.livePeerEnabled = false; this.availableCommands = new CodexCommands( connection, codexAcpClient, @@ -210,6 +219,7 @@ export class CodexAcpServer { logger.log("Initialize request received"); this.clientInfo = _params.clientInfo ?? null; this.clientCapabilities = _params.clientCapabilities ?? null; + this.livePeerEnabled = requestsLivePeer(_params._meta); this.terminalOutputMode = resolveTerminalOutputMode(_params.clientCapabilities); this.booleanConfigOptionsSupported = clientSupportsBooleanConfigOptions(_params.clientCapabilities); await this.runWithProcessCheck(() => this.codexAcpClient.initialize(_params)); @@ -248,6 +258,9 @@ export class CodexAcpServer { steering: { supported: true, }, + codex: { + livePeer: LIVE_PEER_CAPABILITIES, + }, }, }; } @@ -469,6 +482,9 @@ export class CodexAcpServer { sessionTitleSource: "sessionId" in request ? "unknown" : "unset", }; this.sessions.set(sessionId, sessionState); + if (this.livePeerEnabled) { + await this.installSessionHandlers(sessionState); + } resumeSubscribed = false; if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { @@ -988,6 +1004,7 @@ export class CodexAcpServer { threadId: params.sessionId, turnId, prompt: params.prompt, + clientUserMessageId: getClientUserMessageId(params._meta), })); return true; } catch (err) { @@ -1097,12 +1114,16 @@ export class CodexAcpServer { private parseSessionSteerParams(params: Record): SessionSteerRequest { const sessionId = params["sessionId"]; const prompt = params["prompt"]; + const meta = params["_meta"]; if (typeof sessionId !== "string" || !Array.isArray(prompt)) { throw RequestError.invalidParams(); } return { sessionId: sessionId, prompt: prompt as acp.ContentBlock[], + ...(typeof meta === "object" && meta !== null && !Array.isArray(meta) + ? {_meta: meta as Record} + : {}), }; } @@ -1311,6 +1332,9 @@ export class CodexAcpServer { sessionTitleSource: "unset", }; this.sessions.set(sessionId, sessionState); + if (this.livePeerEnabled) { + await this.installSessionHandlers(sessionState); + } subscribed = false; if (requestedMcpServers.length > 0 && mcpServerStartupVersion !== null) { @@ -1477,9 +1501,15 @@ export class CodexAcpServer { const updates: UpdateSessionEvent[] = []; const messageId = item.id; for (const input of item.content) { - const blocks = this.userInputToContentBlocks(input); + const blocks = userInputToContentBlocks(input); for (const block of blocks) { - updates.push(createUserMessageChunk(block, messageId)); + updates.push(createUserMessageChunk( + block, + messageId, + this.livePeerEnabled + ? createClientUserMessageIdMeta(item.clientId) + : undefined, + )); } } return updates; @@ -1530,34 +1560,6 @@ export class CodexAcpServer { ); } - private userInputToContentBlocks(input: UserInput): acp.ContentBlock[] { - switch (input.type) { - case "text": - return input.text.length > 0 ? [{ type: "text", text: input.text }] : []; - case "image": - return [{ type: "text", text: this.formatUriAsLink("image", input.url) }]; - case "localImage": { - const uri = input.path.startsWith("file://") ? input.path : `file://${input.path}`; - return [{ type: "text", text: this.formatUriAsLink(null, uri) }]; - } - case "skill": - return [{ type: "text", text: `skill:${input.name} (${input.path})` }]; - } - return []; - } - - private formatUriAsLink(name: string | null, uri: string): string { - if (name && name.length > 0) { - return `[@${name}](${uri})`; - } - if (uri.startsWith("file://")) { - const path = uri.replace("file://", ""); - const fileName = path.split("/").pop() ?? path; - return `[@${fileName}](${uri})`; - } - return uri; - } - getSessionState(sessionId: string): SessionState { const sessionState = this.sessions.get(sessionId); if (!sessionState) { @@ -1876,21 +1878,7 @@ export class CodexAcpServer { const disposePromptRequestCancellation = this.observePromptRequestCancellation(signal, sessionState, activePrompt); try { - const eventHandler = new CodexEventHandler(this.connection, sessionState); - const approvalHandler = new CodexApprovalHandler(this.connection, sessionState, activePrompt.signal); - const elicitationHandler = new CodexElicitationHandler( - this.connection, - sessionState, - this.clientCapabilities, - activePrompt.signal, - ); - await this.codexAcpClient.subscribeToSessionEvents(params.sessionId, - async (event) => { - await elicitationHandler.handleNotification(event); - return eventHandler.handleNotification(event); - }, - approvalHandler, - elicitationHandler); + const eventHandler = await this.installSessionHandlers(sessionState, activePrompt.signal); if (activePrompt.signal.aborted) { return this.cancelledPromptResponse(sessionState); @@ -2053,10 +2041,49 @@ export class CodexAcpServer { this.pendingTurnStarts.delete(params.sessionId); registeredPendingTurnStart.resolve(null); } + if ( + this.livePeerEnabled + && this.sessions.get(params.sessionId) === sessionState + && !this.sessionIsClosing(params.sessionId) + ) { + await this.installSessionHandlers(sessionState); + } activePrompt.complete(); } } + private async installSessionHandlers( + sessionState: SessionState, + cancellationSignal?: AbortSignal, + ): Promise { + const eventHandler = new CodexEventHandler( + this.connection, + sessionState, + {emitUserMessages: this.livePeerEnabled}, + ); + const approvalHandler = new CodexApprovalHandler( + this.connection, + sessionState, + cancellationSignal, + ); + const elicitationHandler = new CodexElicitationHandler( + this.connection, + sessionState, + this.clientCapabilities, + cancellationSignal, + ); + await this.codexAcpClient.subscribeToSessionEvents( + sessionState.sessionId, + async (event) => { + await elicitationHandler.handleNotification(event); + return eventHandler.handleNotification(event); + }, + approvalHandler, + elicitationHandler, + ); + return eventHandler; + } + private cancelledPromptResponse(sessionState: SessionState): acp.PromptResponse { return { stopReason: "cancelled", diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 89e91eb5..1aee836e 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -63,8 +63,11 @@ import { createCodexMessagePhaseMeta, createAgentTextMessageChunk, createAgentTextThoughtChunk, + createUserMessageChunk, } from "./ContentChunks"; import {sameThreadGoalSnapshot, toThreadGoalSnapshot} from "./ThreadGoalSnapshot"; +import {createClientUserMessageIdMeta} from "./CodexMeta"; +import {userInputToContentBlocks} from "./UserInputMapper"; export { stripShellPrefix }; @@ -83,10 +86,16 @@ export class CodexEventHandler { private readonly terminalCommandOutputIds = new Set(); private readonly agentMessagePhases = new Map(); private readonly activeSubAgentActivities = new Set(); + private readonly emitUserMessages: boolean; - constructor(connection: AcpClientConnection, sessionState: SessionState) { + constructor( + connection: AcpClientConnection, + sessionState: SessionState, + options: {emitUserMessages?: boolean} = {}, + ) { this.connection = connection; this.sessionState = sessionState; + this.emitUserMessages = options.emitUserMessages ?? false; } getFailure(): RequestError | null { @@ -95,13 +104,18 @@ export class CodexEventHandler { async handleNotification(notification: ServerNotification) { const session = new ACPSessionConnection(this.connection, this.sessionState.sessionId); - const updateEvent = await this.createUpdateEvent(notification); - if (updateEvent) { + const updateEvents = await this.createUpdateEvent(notification); + for (const updateEvent of Array.isArray(updateEvents) ? updateEvents : [updateEvents]) { + if (!updateEvent) { + continue; + } await session.update(updateEvent); } } - private async createUpdateEvent(notification: ServerNotification): Promise { + private async createUpdateEvent( + notification: ServerNotification, + ): Promise { /* TODO split UpdateSessionEvent to improve completion createUpdateEvent({ @@ -314,7 +328,9 @@ export class CodexEventHandler { return createAgentTextThoughtChunk(text, messageId); } - private async createItemEvent(event: ItemStartedNotification): Promise { + private async createItemEvent( + event: ItemStartedNotification, + ): Promise { switch (event.item.type) { case "fileChange": return await createFileChangeUpdate(event.item); @@ -344,13 +360,25 @@ export class CodexEventHandler { case "agentMessage": this.rememberAgentMessagePhase(event.item); return null; + case "userMessage": { + if (!this.emitUserMessages) { + return null; + } + const clientUserMessageMeta = createClientUserMessageIdMeta(event.item.clientId); + return event.item.content + .flatMap(userInputToContentBlocks) + .map(content => createUserMessageChunk( + content, + event.item.id, + clientUserMessageMeta, + )); + } case "contextCompaction": return createContextCompactionStartUpdate(event.item); case "subAgentActivity": this.activeSubAgentActivities.add(event.item.id); return createSubAgentActivityUpdate(event.item, "in_progress", "tool_call"); case "sleep": - case "userMessage": case "hookPrompt": case "reasoning": case "enteredReviewMode": diff --git a/src/CodexJsonRpcConnection.ts b/src/CodexJsonRpcConnection.ts index 0f79117a..85d047f4 100644 --- a/src/CodexJsonRpcConnection.ts +++ b/src/CodexJsonRpcConnection.ts @@ -1,19 +1,36 @@ import * as rpc from "vscode-jsonrpc/node"; -import type {MessageConnection} from "vscode-jsonrpc/node"; +import type {DataCallback, Disposable, Message, MessageConnection} from "vscode-jsonrpc/node"; import type {ChildProcessWithoutNullStreams} from "node:child_process"; import {spawn} from "node:child_process"; import {createRequire} from "node:module"; +import net from "node:net"; +import {homedir} from "node:os"; +import path from "node:path"; +import {PassThrough, type Readable, Writable} from "node:stream"; +import WebSocket, {type RawData} from "ws"; import {createJSONRPCReader, createJSONRPCWriter} from "./StdUtils"; import {logger} from "./Logger"; export interface CodexConnection { - readonly connection: MessageConnection - readonly process: ChildProcessWithoutNullStreams; + readonly connection: MessageConnection; + readonly runtime: CodexRuntime; +} + +export interface CodexRuntime { + readonly stderr: Readable; + readonly exitCode: number | null; + close(): void; + terminate(): boolean; + onExit(listener: (exitCode: number | null) => void): () => void; } export function startCodexConnection(codexPath?: string, env?: NodeJS.ProcessEnv): CodexConnection { const spawnEnv = env ?? process.env; + const appServerUrl = spawnEnv["CODEX_APP_SERVER_URL"]; + if (appServerUrl) { + return connectCodexAppServer(appServerUrl, spawnEnv); + } let codex: ChildProcessWithoutNullStreams; if (codexPath) { @@ -35,11 +52,178 @@ export function startCodexConnection(codexPath?: string, env?: NodeJS.ProcessEnv connection.listen(); // Terminate all current activities on process termination - codex.on("exit", _ => { - connection.dispose(); + const runtime = new ChildProcessRuntime(codex); + runtime.onExit(() => connection.dispose()); + + return {connection, runtime}; +} + +function connectCodexAppServer(url: string, env: NodeJS.ProcessEnv): CodexConnection { + const socket = createWebSocket(url, env); + const runtime = new WebSocketRuntime(socket); + const connection = rpc.createMessageConnection( + new WebSocketMessageReader(socket), + new WebSocketMessageWriter(socket), + ); + + connection.listen(); + runtime.onExit(() => connection.dispose()); + + return {connection, runtime}; +} + +class ChildProcessRuntime implements CodexRuntime { + constructor(private readonly process: ChildProcessWithoutNullStreams) {} + + get stderr(): Readable { + return this.process.stderr; + } + + get exitCode(): number | null { + return this.process.exitCode; + } + + close(): void { + this.process.stdin.end(); + } + + terminate(): boolean { + return this.process.kill(); + } + + onExit(listener: (exitCode: number | null) => void): () => void { + this.process.on("close", listener); + return () => this.process.off("close", listener); + } +} + +class WebSocketRuntime implements CodexRuntime { + readonly stderr = new PassThrough(); + exitCode: number | null = null; + + constructor(private readonly socket: WebSocket) { + socket.once("error", (error) => { + this.stderr.write(`${String(error)}\n`); + }); + socket.once("close", (code) => { + this.exitCode = code === 1000 ? 0 : 1; + }); + } + + close(): void { + this.socket.close(); + } + + terminate(): boolean { + if (this.exitCode !== null || this.socket.readyState === WebSocket.CLOSED) { + return false; + } + this.socket.terminate(); + return true; + } + + onExit(listener: (exitCode: number | null) => void): () => void { + const onClose = () => listener(this.exitCode); + this.socket.on("close", onClose); + return () => this.socket.off("close", onClose); + } +} + +function createWebSocket(url: string, env: NodeJS.ProcessEnv): WebSocket { + const options: WebSocket.ClientOptions = {perMessageDeflate: false}; + if (!url.startsWith("unix://")) { + return new WebSocket(url, options); + } + + const configuredPath = url.slice("unix://".length); + const codexHome = env["CODEX_HOME"] || path.join(env["HOME"] || homedir(), ".codex"); + const socketPath = configuredPath + || path.join(codexHome, "app-server-control", "app-server-control.sock"); + return new WebSocket("ws://localhost/", { + ...options, + createConnection: () => net.createConnection(socketPath), }); +} + +class WebSocketMessageReader extends rpc.AbstractMessageReader { + constructor(private readonly socket: WebSocket) { + super(); + } + + listen(callback: DataCallback): Disposable { + const onMessage = (data: RawData) => { + try { + callback(JSON.parse(webSocketDataToString(data)) as Message); + } catch (error) { + this.fireError(error); + } + }; + const onError = (error: Error) => this.fireError(error); + const onClose = () => this.fireClose(); + this.socket.on("message", onMessage); + this.socket.on("error", onError); + this.socket.on("close", onClose); + return rpc.Disposable.create(() => { + this.socket.off("message", onMessage); + this.socket.off("error", onError); + this.socket.off("close", onClose); + }); + } +} - return {connection: connection, process: codex}; +class WebSocketMessageWriter extends rpc.AbstractMessageWriter { + constructor(private readonly socket: WebSocket) { + super(); + } + + async write(message: Message): Promise { + await waitUntilOpen(this.socket); + if (this.socket.readyState !== WebSocket.OPEN) { + throw new Error("Codex App Server WebSocket is not open"); + } + await new Promise((resolve, reject) => { + this.socket.send( + JSON.stringify(message), + (error) => error ? reject(error) : resolve(), + ); + }); + } + + end(): void { + this.socket.close(); + } +} + +async function waitUntilOpen(socket: WebSocket): Promise { + if (socket.readyState !== WebSocket.CONNECTING) { + return; + } + await new Promise((resolve, reject) => { + const cleanup = () => { + socket.off("open", onOpen); + socket.off("error", onError); + }; + const onOpen = () => { + cleanup(); + resolve(); + }; + const onError = (error: Error) => { + cleanup(); + reject(error); + }; + socket.on("open", onOpen); + socket.on("error", onError); + }); +} + +function webSocketDataToString(data: RawData): string { + if (Buffer.isBuffer(data)) { + return data.toString("utf8"); + } + if (Array.isArray(data)) { + return Buffer.concat(data).toString("utf8"); + } + return Buffer.from(data).toString("utf8"); } function attachLogs(proc: ChildProcessWithoutNullStreams) { diff --git a/src/CodexMeta.ts b/src/CodexMeta.ts new file mode 100644 index 00000000..ccd218ec --- /dev/null +++ b/src/CodexMeta.ts @@ -0,0 +1,45 @@ +type CodexMeta = { + codex?: { + livePeer?: { + version?: unknown; + }; + clientUserMessageId?: unknown; + }; +}; + +export const LIVE_PEER_CAPABILITIES = { + version: 1, + ambientEvents: true, + interactions: true, + userMessages: true, + clientUserMessageIds: true, +} as const; + +export function requestsLivePeer(meta: unknown): boolean { + if (!isRecord(meta)) { + return false; + } + return (meta as CodexMeta).codex?.livePeer?.version === LIVE_PEER_CAPABILITIES.version; +} + +export function getClientUserMessageId(meta: unknown): string | null { + if (!isRecord(meta)) { + return null; + } + const clientUserMessageId = (meta as CodexMeta).codex?.clientUserMessageId; + return typeof clientUserMessageId === "string" && clientUserMessageId.length > 0 + ? clientUserMessageId + : null; +} + +export function createClientUserMessageIdMeta( + clientUserMessageId: string | null, +): Record | undefined { + return clientUserMessageId + ? {codex: {clientUserMessageId}} + : undefined; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/UserInputMapper.ts b/src/UserInputMapper.ts new file mode 100644 index 00000000..aba427d1 --- /dev/null +++ b/src/UserInputMapper.ts @@ -0,0 +1,30 @@ +import type {ContentBlock} from "@agentclientprotocol/sdk"; +import type {UserInput} from "./app-server/v2"; + +export function userInputToContentBlocks(input: UserInput): ContentBlock[] { + switch (input.type) { + case "text": + return input.text.length > 0 ? [{type: "text", text: input.text}] : []; + case "image": + return [{type: "text", text: formatUriAsLink("image", input.url)}]; + case "localImage": { + const uri = input.path.startsWith("file://") ? input.path : `file://${input.path}`; + return [{type: "text", text: formatUriAsLink(null, uri)}]; + } + case "skill": + return [{type: "text", text: `skill:${input.name} (${input.path})`}]; + } + return []; +} + +function formatUriAsLink(name: string | null, uri: string): string { + if (name) { + return `[@${name}](${uri})`; + } + if (uri.startsWith("file://")) { + const imagePath = uri.slice("file://".length); + const fileName = imagePath.split("/").pop() ?? imagePath; + return `[@${fileName}](${uri})`; + } + return uri; +} diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 9d6dc2b8..a5f94699 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -65,6 +65,15 @@ describe('CodexACPAgent - initialize', () => { steering: { supported: true, }, + codex: { + livePeer: { + version: 1, + ambientEvents: true, + interactions: true, + userMessages: true, + clientUserMessageIds: true, + }, + }, }, }); }); diff --git a/src/__tests__/CodexACPAgent/live-peer.test.ts b/src/__tests__/CodexACPAgent/live-peer.test.ts new file mode 100644 index 00000000..49110fb6 --- /dev/null +++ b/src/__tests__/CodexACPAgent/live-peer.test.ts @@ -0,0 +1,248 @@ +import * as acp from "@agentclientprotocol/sdk"; +import {beforeEach, describe, expect, it, vi} from "vitest"; +import type {ServerNotification} from "../../app-server"; +import type {CommandExecutionRequestApprovalParams, UserInput} from "../../app-server/v2"; +import { + createCodexMockTestFixture, + createTestModel, + type CodexMockTestFixture, +} from "../acp-test-utils"; + +describe("Codex ACP live peer", () => { + let fixture: CodexMockTestFixture; + + beforeEach(() => { + fixture = createCodexMockTestFixture(); + configureNewSession(fixture); + }); + + it("advertises live-peer support without enabling it for existing clients", async () => { + const response = await fixture.getCodexAcpAgent().initialize({ + protocolVersion: acp.PROTOCOL_VERSION, + }); + expect(response._meta).toMatchObject({ + codex: { + livePeer: { + version: 1, + ambientEvents: true, + interactions: true, + userMessages: true, + clientUserMessageIds: true, + }, + }, + }); + + const session = await fixture.getCodexAcpAgent().newSession({ + cwd: "/workspace", + mcpServers: [], + }); + fixture.clearAcpConnectionDump(); + fixture.sendServerNotification(userMessageStarted(session.sessionId)); + + await new Promise(resolve => setImmediate(resolve)); + expect(fixture.getAcpConnectionDump([])).not.toContain("terminal says hello"); + }); + + it("streams ambient terminal user messages with their origin ID", async () => { + await initializeLivePeer(fixture); + const session = await fixture.getCodexAcpAgent().newSession({ + cwd: "/workspace", + mcpServers: [], + }); + fixture.clearAcpConnectionDump(); + + fixture.sendServerNotification(userMessageStarted( + session.sessionId, + "telegram:chat-1:update-42", + )); + + await vi.waitFor(() => { + expect(fixture.getAcpConnectionDump([])).toContain("terminal says hello"); + }); + expect(fixture.getAcpConnectionDump([])).toContain( + '"clientUserMessageId": "telegram:chat-1:update-42"', + ); + }); + + it("uses the history mapper for ambient user-message content", async () => { + await initializeLivePeer(fixture); + const session = await fixture.getCodexAcpAgent().newSession({ + cwd: "/workspace", + mcpServers: [], + }); + fixture.clearAcpConnectionDump(); + + fixture.sendServerNotification(userMessageStarted( + session.sessionId, + null, + [ + {type: "text", text: "inspect this", text_elements: []}, + {type: "localImage", path: "/workspace/screenshot.png"}, + ], + )); + + await vi.waitFor(() => { + const updates = fixture.getAcpConnectionDump([]); + expect(updates).toContain("inspect this"); + expect(updates).toContain("[@screenshot.png](file:///workspace/screenshot.png)"); + }); + }); + + it("relays an ambient approval after the session is created", async () => { + await initializeLivePeer(fixture); + const session = await fixture.getCodexAcpAgent().newSession({ + cwd: "/workspace", + mcpServers: [], + }); + fixture.setPermissionResponse({ + outcome: {outcome: "selected", optionId: "allow_once"}, + }); + + const params: CommandExecutionRequestApprovalParams = { + threadId: session.sessionId, + turnId: "turn-1", + itemId: "command-1", + reason: "Run the requested test", + command: "npm test", + cwd: "/workspace", + startedAtMs: 0, + environmentId: null, + proposedExecpolicyAmendment: null, + }; + + await expect(fixture.sendServerRequest( + "item/commandExecution/requestApproval", + params, + )).resolves.toEqual({decision: "accept"}); + expect(fixture.getAcpConnectionDump([])).toContain("requestPermission"); + }); + + it("passes a client user-message ID into a normal turn", async () => { + const agent = fixture.getCodexAcpAgent(); + const session = await agent.newSession({ + cwd: "/workspace", + mcpServers: [], + }); + const turnStart = mockCompletedTurn(fixture, session.sessionId); + + await agent.prompt({ + sessionId: session.sessionId, + prompt: [{type: "text", text: "hello"}], + _meta: { + codex: { + clientUserMessageId: "telegram:chat-1:update-99", + }, + }, + }); + + expect(turnStart).toHaveBeenCalledWith(expect.objectContaining({ + clientUserMessageId: "telegram:chat-1:update-99", + })); + }); + + it("restores ambient handlers after an ACP-owned prompt", async () => { + await initializeLivePeer(fixture); + const agent = fixture.getCodexAcpAgent(); + const session = await agent.newSession({ + cwd: "/workspace", + mcpServers: [], + }); + mockCompletedTurn(fixture, session.sessionId); + + await agent.prompt({ + sessionId: session.sessionId, + prompt: [{type: "text", text: "remote prompt"}], + }); + fixture.clearAcpConnectionDump(); + fixture.sendServerNotification(userMessageStarted(session.sessionId)); + + await vi.waitFor(() => { + expect(fixture.getAcpConnectionDump([])).toContain("terminal says hello"); + }); + }); +}); + +async function initializeLivePeer(fixture: CodexMockTestFixture): Promise { + await fixture.getCodexAcpAgent().initialize({ + protocolVersion: acp.PROTOCOL_VERSION, + _meta: { + codex: { + livePeer: {version: 1}, + }, + }, + }); +} + +function configureNewSession(fixture: CodexMockTestFixture): void { + const client = fixture.getCodexAcpClient(); + const appServer = fixture.getCodexAppServerClient(); + client.authRequired = vi.fn().mockResolvedValue(false); + client.getAccount = vi.fn().mockResolvedValue({ + account: null, + requiresOpenaiAuth: false, + }); + vi.spyOn(appServer, "listSkills").mockResolvedValue({data: []}); + vi.spyOn(appServer, "listModels").mockResolvedValue({ + data: [createTestModel()], + nextCursor: null, + }); + vi.spyOn(appServer, "threadStart").mockResolvedValue({ + thread: {id: "session-1"}, + model: "model-id", + modelProvider: "openai", + cwd: "/workspace", + approvalPolicy: "on-request", + sandbox: {type: "workspaceWrite", writableRoots: []}, + reasoningEffort: "medium", + } as never); +} + +function mockCompletedTurn( + fixture: CodexMockTestFixture, + threadId: string, +): ReturnType { + const appServer = fixture.getCodexAppServerClient(); + const turn = { + id: "turn-1", + items: [], + itemsView: "notLoaded" as const, + status: "completed" as const, + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }; + const turnStart = vi.spyOn(appServer, "turnStart").mockResolvedValue({ + turn: {...turn, status: "inProgress"}, + }); + vi.spyOn(appServer, "awaitTurnCompleted").mockResolvedValue({ + threadId, + turn, + }); + return turnStart; +} + +function userMessageStarted( + threadId: string, + clientId: string | null = null, + content: UserInput[] = [{ + type: "text", + text: "terminal says hello", + text_elements: [], + }], +): ServerNotification { + return { + method: "item/started", + params: { + threadId, + turnId: "turn-1", + startedAtMs: 0, + item: { + type: "userMessage", + id: "message-1", + clientId, + content, + }, + }, + }; +} diff --git a/src/__tests__/CodexACPAgent/mcp-config-merge.test.ts b/src/__tests__/CodexACPAgent/mcp-config-merge.test.ts index 94a92d71..65564a3d 100644 --- a/src/__tests__/CodexACPAgent/mcp-config-merge.test.ts +++ b/src/__tests__/CodexACPAgent/mcp-config-merge.test.ts @@ -40,7 +40,7 @@ url = "https://example.com/mcp" fixture = createBaseTestFixture({ connection: codexConnection.connection, - getExitCode: () => codexConnection.process.exitCode, + getExitCode: () => codexConnection.runtime.exitCode, }); }); diff --git a/src/__tests__/CodexACPAgent/process-exit-error.test.ts b/src/__tests__/CodexACPAgent/process-exit-error.test.ts index 99f27d52..06a9800d 100644 --- a/src/__tests__/CodexACPAgent/process-exit-error.test.ts +++ b/src/__tests__/CodexACPAgent/process-exit-error.test.ts @@ -2,7 +2,6 @@ import { describe, it, expect } from 'vitest'; import * as fs from 'node:fs'; import * as os from 'node:os'; import * as path from 'node:path'; -import { once } from 'node:events'; import * as acp from '@agentclientprotocol/sdk'; import { startCodexConnection } from '../../CodexJsonRpcConnection'; import { CodexAppServerClient } from '../../CodexAppServerClient'; @@ -18,18 +17,22 @@ describe('CodexACPAgent - process exit error', () => { const connection = startCodexConnection(fakeCodex); let stderr = ''; - connection.process.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString(); }); + connection.runtime.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString(); }); const codexClient = new CodexAcpClient(new CodexAppServerClient(connection.connection)); const agent = new CodexAcpServer( createMockConnections().mockAcpConnection, codexClient, undefined, - () => connection.process.exitCode, + () => connection.runtime.exitCode, () => stderr, ); - await once(connection.process, 'close'); // process exited and stderr flushed + if (connection.runtime.exitCode === null) { + await new Promise(resolve => { + connection.runtime.onExit(() => resolve()); + }); + } await expect(agent.initialize({ protocolVersion: acp.PROTOCOL_VERSION })) .rejects.toThrow("Codex process has exited with code 1:\ncodex: failed to launch"); diff --git a/src/__tests__/CodexACPAgent/steer-events.test.ts b/src/__tests__/CodexACPAgent/steer-events.test.ts index c13a719b..72d700f9 100644 --- a/src/__tests__/CodexACPAgent/steer-events.test.ts +++ b/src/__tests__/CodexACPAgent/steer-events.test.ts @@ -65,11 +65,17 @@ describe('_session/steering', () => { await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { sessionId: "session-id", prompt: [{type: "text", text: "also keep backward compatibility"}], + _meta: { + codex: { + clientUserMessageId: "telegram:chat-1:update-100", + }, + }, })).resolves.toEqual({outcome: "injected"}); expect(turnSteerSpy).toHaveBeenCalledWith({ threadId: "session-id", expectedTurnId: "turn-id", + clientUserMessageId: "telegram:chat-1:update-100", input: [{type: "text", text: "also keep backward compatibility", text_elements: []}], }); @@ -93,10 +99,16 @@ describe('_session/steering', () => { await expect(mockFixture.getCodexAcpAgent().extMethod(SESSION_STEERING_METHOD, { sessionId: "session-id", prompt: [{type: "text", text: "too late for the previous turn"}], + _meta: { + codex: { + clientUserMessageId: "telegram:chat-1:update-101", + }, + }, })).resolves.toEqual({outcome: "startedNewTurn"}); expect(turnStartSpy).toHaveBeenCalledWith(expect.objectContaining({ threadId: "session-id", + clientUserMessageId: "telegram:chat-1:update-101", input: [{type: "text", text: "too late for the previous turn", text_elements: []}], })); diff --git a/src/__tests__/CodexJsonRpcConnection.test.ts b/src/__tests__/CodexJsonRpcConnection.test.ts new file mode 100644 index 00000000..0820dcee --- /dev/null +++ b/src/__tests__/CodexJsonRpcConnection.test.ts @@ -0,0 +1,115 @@ +import {once} from "node:events"; +import {createServer, type Server} from "node:http"; +import {mkdir, mkdtemp, rm} from "node:fs/promises"; +import {tmpdir} from "node:os"; +import path from "node:path"; +import {afterEach, describe, expect, it} from "vitest"; +import {WebSocketServer} from "ws"; +import {startCodexConnection} from "../CodexJsonRpcConnection"; + +describe("Codex JSON-RPC connection", () => { + const cleanups: Array<() => Promise> = []; + + afterEach(async () => { + await Promise.all(cleanups.splice(0).map(cleanup => cleanup())); + }); + + it("connects to an existing WebSocket App Server", async () => { + const webSocketServer = new WebSocketServer({port: 0}); + await once(webSocketServer, "listening"); + cleanups.push(() => closeWebSocketServer(webSocketServer)); + + const address = webSocketServer.address(); + if (typeof address === "string" || address === null) { + throw new Error("WebSocket server did not expose a TCP port"); + } + answerProbeRequests(webSocketServer); + + const codex = startCodexConnection("/path/that/must/not/run", { + CODEX_APP_SERVER_URL: `ws://127.0.0.1:${address.port}`, + }); + cleanups.push(() => closeCodexConnection(codex)); + + await expect(codex.connection.sendRequest("probe", {value: 42})) + .resolves.toEqual({value: 42}); + }); + + it("resolves unix:// through CODEX_HOME and connects to the default socket", async () => { + const directory = await mkdtemp(path.join(tmpdir(), "codex-acp-ws-")); + const socketDirectory = path.join(directory, "app-server-control"); + const socketPath = path.join(socketDirectory, "app-server-control.sock"); + await mkdir(socketDirectory); + const httpServer = createServer(); + const webSocketServer = new WebSocketServer({server: httpServer}); + await new Promise((resolve, reject) => { + httpServer.once("error", reject); + httpServer.listen(socketPath, resolve); + }); + cleanups.push(async () => { + await closeWebSocketServer(webSocketServer); + await closeHttpServer(httpServer); + await rm(directory, {recursive: true, force: true}); + }); + answerProbeRequests(webSocketServer); + + const codex = startCodexConnection(undefined, { + CODEX_APP_SERVER_URL: "unix://", + CODEX_HOME: directory, + }); + cleanups.push(() => closeCodexConnection(codex)); + + await expect(codex.connection.sendRequest("probe", {transport: "unix"})) + .resolves.toEqual({transport: "unix"}); + }); +}); + +function answerProbeRequests(server: WebSocketServer): void { + server.on("connection", socket => { + socket.on("message", raw => { + const request = JSON.parse(raw.toString()) as { + id: string | number; + method: string; + params: unknown; + }; + socket.send(JSON.stringify({ + jsonrpc: "2.0", + id: request.id, + result: request.params, + })); + }); + }); +} + +async function closeCodexConnection( + codex: ReturnType, +): Promise { + codex.connection.dispose(); + if (codex.runtime.exitCode !== null) { + return; + } + const closed = new Promise(resolve => { + const dispose = codex.runtime.onExit(() => { + dispose(); + resolve(); + }); + }); + codex.runtime.terminate(); + await closed; +} + +async function closeWebSocketServer(server: WebSocketServer): Promise { + for (const client of server.clients) { + client.terminate(); + } + if (server.address() === null) { + return; + } + await new Promise(resolve => server.close(() => resolve())); +} + +async function closeHttpServer(server: Server): Promise { + if (!server.listening) { + return; + } + await new Promise(resolve => server.close(() => resolve())); +} diff --git a/src/__tests__/acp-test-utils.ts b/src/__tests__/acp-test-utils.ts index de4ad962..ee448af9 100644 --- a/src/__tests__/acp-test-utils.ts +++ b/src/__tests__/acp-test-utils.ts @@ -180,13 +180,13 @@ export function createTestFixture(): TestFixture { ...process.env, CODEX_HOME: codexHome, }); - codexConnection.process.on("exit", () => { + codexConnection.runtime.onExit(() => { removeDirectoryWithRetry(codexHome); }); return createBaseTestFixture({ connection: codexConnection.connection, - getExitCode: () => codexConnection.process.exitCode + getExitCode: () => codexConnection.runtime.exitCode }); } diff --git a/src/index.ts b/src/index.ts index 014801ff..73df92af 100644 --- a/src/index.ts +++ b/src/index.ts @@ -85,17 +85,17 @@ function startAcpServer() { const maxStderrTailChars = 2 * 1024; let stderr = ""; - codexConnection.process.stderr.addListener("data", (data: Buffer) => { + codexConnection.runtime.stderr.addListener("data", (data: Buffer) => { stderr = (stderr + data.toString()).slice(-maxStderrTailChars); }); process.stdin.on("close", () => { - codexConnection.process.stdin.end(); - // Kill the codex process if it doesn't exit naturally + codexConnection.runtime.close(); + // Terminate the App Server connection if it does not close naturally. setTimeout(() => { - if (!codexConnection.process.killed) { - logger.log("Codex still running 2s after stdin closed; terminating process"); - codexConnection.process.kill(); + if (codexConnection.runtime.exitCode === null) { + logger.log("Codex connection still open 2s after stdin closed; terminating"); + codexConnection.runtime.terminate(); } }, 2000); }); @@ -105,7 +105,7 @@ function startAcpServer() { function createAgent(connection: acp.AgentContext): CodexAcpServer { const appServerClient = new CodexAppServerClient(codexConnection.connection); const codexClient = new CodexAcpClient(appServerClient, config, modelProvider); - return new CodexAcpServer(connection, codexClient, defaultAuthRequest, () => codexConnection.process.exitCode, () => stderr); + return new CodexAcpServer(connection, codexClient, defaultAuthRequest, () => codexConnection.runtime.exitCode, () => stderr); } let codexAcpServer: CodexAcpServer | null = null; diff --git a/src/login.ts b/src/login.ts index a46bdf89..bdfdd65d 100644 --- a/src/login.ts +++ b/src/login.ts @@ -112,7 +112,7 @@ async function login(options: LoginOptions): Promise { } } finally { codexConnection.connection.dispose(); - codexConnection.process.kill(); + codexConnection.runtime.close(); } } From ceab62f86f26aa9c0c013abb51a743c2c975b046 Mon Sep 17 00:00:00 2001 From: Kelvin Jayanoris Date: Fri, 31 Jul 2026 12:23:59 +0300 Subject: [PATCH 2/6] feat: expose live-peer turn lifecycle --- src/CodexEventHandler.ts | 15 +++- src/CodexMeta.ts | 1 + .../CodexACPAgent/initialize.test.ts | 1 + src/__tests__/CodexACPAgent/live-peer.test.ts | 69 +++++++++++++++++++ 4 files changed, 84 insertions(+), 2 deletions(-) diff --git a/src/CodexEventHandler.ts b/src/CodexEventHandler.ts index 1aee836e..96ce0553 100644 --- a/src/CodexEventHandler.ts +++ b/src/CodexEventHandler.ts @@ -138,10 +138,21 @@ export class CodexEventHandler { return await this.createErrorEvent(notification.params); case "turn/started": this.sessionState.currentTurnId = notification.params.turn.id; - return null; + return this.createCodexSessionInfoUpdate({ + turn: { + id: notification.params.turn.id, + status: notification.params.turn.status, + }, + }); case "turn/completed": this.sessionState.currentTurnId = null; - return null; + return this.createCodexSessionInfoUpdate({ + turn: { + id: notification.params.turn.id, + status: notification.params.turn.status, + error: notification.params.turn.error, + }, + }); case "thread/tokenUsage/updated": return this.createUsageUpdate(notification.params); case "thread/name/updated": diff --git a/src/CodexMeta.ts b/src/CodexMeta.ts index ccd218ec..0b973e00 100644 --- a/src/CodexMeta.ts +++ b/src/CodexMeta.ts @@ -13,6 +13,7 @@ export const LIVE_PEER_CAPABILITIES = { interactions: true, userMessages: true, clientUserMessageIds: true, + turnLifecycle: true, } as const; export function requestsLivePeer(meta: unknown): boolean { diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index a5f94699..02bf2a8d 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -72,6 +72,7 @@ describe('CodexACPAgent - initialize', () => { interactions: true, userMessages: true, clientUserMessageIds: true, + turnLifecycle: true, }, }, }, diff --git a/src/__tests__/CodexACPAgent/live-peer.test.ts b/src/__tests__/CodexACPAgent/live-peer.test.ts index 49110fb6..44f58787 100644 --- a/src/__tests__/CodexACPAgent/live-peer.test.ts +++ b/src/__tests__/CodexACPAgent/live-peer.test.ts @@ -28,6 +28,7 @@ describe("Codex ACP live peer", () => { interactions: true, userMessages: true, clientUserMessageIds: true, + turnLifecycle: true, }, }, }); @@ -64,6 +65,74 @@ describe("Codex ACP live peer", () => { ); }); + it("streams ambient turn lifecycle without guessing from message silence", async () => { + await initializeLivePeer(fixture); + const session = await fixture.getCodexAcpAgent().newSession({ + cwd: "/workspace", + mcpServers: [], + }); + fixture.clearAcpConnectionDump(); + + fixture.sendServerNotification({ + method: "turn/started", + params: { + threadId: session.sessionId, + turn: { + id: "turn-ambient", + items: [], + itemsView: "notLoaded", + status: "inProgress", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }, + }); + fixture.sendServerNotification({ + method: "turn/completed", + params: { + threadId: session.sessionId, + turn: { + id: "turn-ambient", + items: [], + itemsView: "notLoaded", + status: "completed", + error: null, + startedAt: null, + completedAt: null, + durationMs: null, + }, + }, + }); + + await vi.waitFor(() => { + const updates = fixture.getAcpConnectionEvents([]) + .map(event => event.args[0]?.update); + expect(updates).toContainEqual(expect.objectContaining({ + _meta: { + codex: { + turn: { + id: "turn-ambient", + status: "inProgress", + }, + }, + }, + })); + expect(updates).toContainEqual(expect.objectContaining({ + _meta: { + codex: { + turn: { + id: "turn-ambient", + status: "completed", + error: null, + }, + }, + }, + })); + }); + }); + it("uses the history mapper for ambient user-message content", async () => { await initializeLivePeer(fixture); const session = await fixture.getCodexAcpAgent().newSession({ From 8f257bb6ab51507aaf1705ae96050feac986db6c Mon Sep 17 00:00:00 2001 From: Kelvin Jayanoris Date: Fri, 31 Jul 2026 14:31:45 +0300 Subject: [PATCH 3/6] feat: frame live-peer history replay --- src/CodexAcpClient.ts | 7 ++ src/CodexAcpServer.ts | 87 ++++++++++++++- src/CodexMeta.ts | 1 + .../CodexACPAgent/initialize.test.ts | 1 + src/__tests__/CodexACPAgent/live-peer.test.ts | 102 ++++++++++++++++++ 5 files changed, 194 insertions(+), 4 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index c2d23869..4f243bd0 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -344,6 +344,7 @@ export class CodexAcpClient { sessionId: request.sessionId, currentModelId: currentModelId, models: codexModels, + activeTurnId: activeTurnId(response.thread), collaborationMode: this.getCollaborationMode(response.thread.id), modelProvider: response.modelProvider, currentServiceTier: response.serviceTier as ServiceTier ?? null, @@ -372,6 +373,7 @@ export class CodexAcpClient { sessionId: request.sessionId, currentModelId: currentModelId, models: codexModels, + activeTurnId: activeTurnId(historyResponse.thread), collaborationMode: this.getCollaborationMode(response.thread.id), modelProvider: response.modelProvider, currentServiceTier: response.serviceTier as ServiceTier ?? null, @@ -910,12 +912,17 @@ export type SessionMetadata = { modelProvider?: string | null, currentServiceTier?: ServiceTier | null, additionalDirectories: string[], + activeTurnId?: string | null, } export type SessionMetadataWithThread = SessionMetadata & { thread: Thread, } +function activeTurnId(thread: Thread): string | null { + return thread.turns?.find((turn) => turn.status === "inProgress")?.id ?? null; +} + function buildPromptItems(prompt: acp.ContentBlock[]): UserInput[] { return prompt.map((block): UserInput | null => { switch (block.type) { diff --git a/src/CodexAcpServer.ts b/src/CodexAcpServer.ts index c6cdf5f5..a6c11643 100644 --- a/src/CodexAcpServer.ts +++ b/src/CodexAcpServer.ts @@ -14,6 +14,7 @@ import type { ReasoningEffortOption, Thread, ThreadItem, + Turn, UserInput } from "./app-server/v2"; import type {RateLimitsMap} from "./RateLimitsMap"; @@ -320,7 +321,7 @@ export class CodexAcpServer { } } - async getOrCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState]> { + async getOrCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState, string | null]> { try { return await this.tryCreateSession(request); } catch (e) { @@ -405,7 +406,7 @@ export class CodexAcpServer { return generation; } - async tryCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState]> { + async tryCreateSession(request: acp.NewSessionRequest | acp.ResumeSessionRequest): Promise<[SessionId, LegacySessionModelState, SessionModeState, string | null]> { const requestedSessionGeneration = "sessionId" in request ? this.beginSessionOpen(request.sessionId) : null; @@ -502,7 +503,12 @@ export class CodexAcpServer { const sessionModelState: LegacySessionModelState = this.createModelState(models, currentModelId); const sessionModeState: SessionModeState = sessionState.agentMode.toSessionModeState(); - return [sessionId, sessionModelState, sessionModeState]; + return [ + sessionId, + sessionModelState, + sessionModeState, + sessionMetadata.activeTurnId ?? null, + ]; } private async getAuthStateForProvider(authProvider: string | null): Promise { @@ -544,6 +550,7 @@ export class CodexAcpServer { modelState, modeState, thread, + activeTurnId, } = await this.getOrCreateSessionWithHistory(params); await this.streamThreadHistory(sessionId, thread); @@ -557,12 +564,21 @@ export class CodexAcpServer { models: modelState, modes: modeState, ...this.createSessionConfigOptionsResponse(this.getSessionState(sessionId)), + ...(this.livePeerEnabled ? { + _meta: { + codex: { + livePeer: { + activeTurnId, + }, + }, + }, + } : {}), }; } async resumeSession(params: acp.ResumeSessionRequest): Promise { logger.log("Resuming session...", {sessionId: params.sessionId}); - const [sessionId, modelState, modeState] = await this.getOrCreateSession(params); + const [sessionId, modelState, modeState, activeTurnId] = await this.getOrCreateSession(params); logger.log("Session resumed", { sessionId: sessionId, @@ -573,6 +589,15 @@ export class CodexAcpServer { models: modelState, modes: modeState, ...this.createSessionConfigOptionsResponse(this.getSessionState(sessionId)), + ...(this.livePeerEnabled ? { + _meta: { + codex: { + livePeer: { + activeTurnId, + }, + }, + }, + } : {}), }; } @@ -1263,6 +1288,7 @@ export class CodexAcpServer { modelState: LegacySessionModelState; modeState: SessionModeState; thread: Thread; + activeTurnId: string | null; }> { const requestedSessionGeneration = this.beginSessionOpen(request.sessionId); await this.checkAuthorization(); @@ -1355,6 +1381,7 @@ export class CodexAcpServer { modelState: sessionModelState, modeState: sessionModeState, thread: thread, + activeTurnId: sessionMetadata.activeTurnId ?? null, }; } @@ -1362,6 +1389,22 @@ export class CodexAcpServer { const session = new ACPSessionConnection(this.connection, sessionId); const sessionState = this.getSessionState(sessionId); await this.publishThreadHistoryTitle(session, sessionState, thread); + if (this.livePeerEnabled) { + await session.update(historySnapshotUpdate("started")); + for (const turn of thread.turns) { + await session.update(historyTurnUpdate(turn, "started")); + for (const item of turn.items) { + for (const update of await this.createHistoryUpdates(item, sessionState)) { + await session.update(update); + } + } + if (turn.status !== "inProgress") { + await session.update(historyTurnUpdate(turn, "finished")); + } + } + await session.update(historySnapshotUpdate("finished")); + return; + } const responseItemFallbackUpdates = await createResponseItemHistoryFallbackUpdates( thread, sessionState.terminalOutputMode, @@ -2158,6 +2201,42 @@ export class CodexAcpServer { } } +function historyTurnUpdate( + turn: Turn, + phase: "started" | "finished", +): UpdateSessionEvent { + return { + sessionUpdate: "session_info_update", + _meta: { + codex: { + turn: { + id: turn.id, + status: phase === "started" ? "inProgress" : turn.status, + error: turn.error?.message ?? null, + replayed: true, + startedAt: turn.startedAt, + completedAt: turn.completedAt, + }, + }, + }, + }; +} + +function historySnapshotUpdate( + phase: "started" | "finished", +): UpdateSessionEvent { + return { + sessionUpdate: "session_info_update", + _meta: { + codex: { + historyReplay: { + phase, + }, + }, + }, + }; +} + function mergeHistoryUpdates( responseItemFallbackUpdates: UpdateSessionEvent[], threadUpdates: UpdateSessionEvent[], diff --git a/src/CodexMeta.ts b/src/CodexMeta.ts index 0b973e00..32e6e987 100644 --- a/src/CodexMeta.ts +++ b/src/CodexMeta.ts @@ -14,6 +14,7 @@ export const LIVE_PEER_CAPABILITIES = { userMessages: true, clientUserMessageIds: true, turnLifecycle: true, + historyReplay: true, } as const; export function requestsLivePeer(meta: unknown): boolean { diff --git a/src/__tests__/CodexACPAgent/initialize.test.ts b/src/__tests__/CodexACPAgent/initialize.test.ts index 02bf2a8d..c99e4fcd 100644 --- a/src/__tests__/CodexACPAgent/initialize.test.ts +++ b/src/__tests__/CodexACPAgent/initialize.test.ts @@ -73,6 +73,7 @@ describe('CodexACPAgent - initialize', () => { userMessages: true, clientUserMessageIds: true, turnLifecycle: true, + historyReplay: true, }, }, }, diff --git a/src/__tests__/CodexACPAgent/live-peer.test.ts b/src/__tests__/CodexACPAgent/live-peer.test.ts index 44f58787..a4cb86ae 100644 --- a/src/__tests__/CodexACPAgent/live-peer.test.ts +++ b/src/__tests__/CodexACPAgent/live-peer.test.ts @@ -29,6 +29,7 @@ describe("Codex ACP live peer", () => { userMessages: true, clientUserMessageIds: true, turnLifecycle: true, + historyReplay: true, }, }, }); @@ -157,6 +158,107 @@ describe("Codex ACP live peer", () => { }); }); + it("frames replayed turns and reports the active turn when loading", async () => { + await initializeLivePeer(fixture); + const appServer = fixture.getCodexAppServerClient(); + const completedTurn = { + id: "turn-completed", + items: [ + { + type: "userMessage", + id: "user-completed", + clientId: null, + content: [{type: "text", text: "old prompt", text_elements: []}], + }, + { + type: "agentMessage", + id: "agent-completed", + text: "old answer", + phase: null, + memoryCitation: null, + }, + ], + itemsView: "full", + status: "completed", + error: null, + startedAt: 10, + completedAt: 20, + durationMs: 10_000, + }; + const activeTurn = { + id: "turn-active", + items: [], + itemsView: "full", + status: "inProgress", + error: null, + startedAt: 30, + completedAt: null, + durationMs: null, + }; + const thread = { + id: "session-history", + sessionId: "session-history", + parentThreadId: null, + threadSource: null, + forkedFromId: null, + preview: "old prompt", + ephemeral: false, + modelProvider: "openai", + createdAt: 1, + updatedAt: 30, + recencyAt: null, + status: {type: "active", activeFlags: []}, + path: null, + cwd: "/workspace", + cliVersion: "test", + source: "cli", + agentNickname: null, + agentRole: null, + gitInfo: null, + name: null, + turns: [completedTurn, activeTurn], + }; + vi.spyOn(appServer, "threadResume").mockResolvedValue({ + thread, + model: "model-id", + modelProvider: "openai", + cwd: "/workspace", + approvalPolicy: "on-request", + sandbox: {type: "workspaceWrite", writableRoots: []}, + reasoningEffort: "medium", + } as never); + vi.spyOn(appServer, "threadRead").mockResolvedValue({thread} as never); + + const response = await fixture.getCodexAcpAgent().loadSession({ + sessionId: thread.id, + cwd: "/workspace", + mcpServers: [], + }); + + expect(response._meta).toMatchObject({ + codex: {livePeer: {activeTurnId: "turn-active"}}, + }); + const updates = fixture.getAcpConnectionEvents([]) + .map(event => event.args[0]?.update); + expect(updates).toContainEqual(expect.objectContaining({ + _meta: {codex: {historyReplay: {phase: "started"}}}, + })); + expect(updates).toContainEqual(expect.objectContaining({ + _meta: expect.objectContaining({ + codex: { + turn: expect.objectContaining({ + id: "turn-completed", + status: "completed", + replayed: true, + }), + }, + }), + })); + expect(updates).toContainEqual(expect.objectContaining({ + _meta: {codex: {historyReplay: {phase: "finished"}}}, + })); + }); + it("relays an ambient approval after the session is created", async () => { await initializeLivePeer(fixture); const session = await fixture.getCodexAcpAgent().newSession({ From 4b0102d8ce7adfc4f89e47cd3350611ab5ead4bd Mon Sep 17 00:00:00 2001 From: Kelvin Jayanoris Date: Fri, 31 Jul 2026 15:51:41 +0300 Subject: [PATCH 4/6] fix: bound session list pages --- src/CodexAcpClient.ts | 3 +++ src/__tests__/CodexACPAgent/list-sessions.test.ts | 1 + 2 files changed, 4 insertions(+) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 4f243bd0..5c81d69b 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -61,6 +61,8 @@ const SUPPORTED_GATEWAY_PROTOCOLS: Record = { openai: "responses", }; +const SESSION_LIST_PAGE_SIZE = 20; + /** * API for accessing the Codex App Server using ACP requests. * Converts ACP requests into corresponding app-server operations. @@ -809,6 +811,7 @@ export class CodexAcpClient { const modelProviders = preferredProvider ? [preferredProvider] : []; const listResponse = await this.codexClient.threadList({ cursor: request.cursor ?? null, + limit: SESSION_LIST_PAGE_SIZE, modelProviders: modelProviders, sourceKinds: sourceKinds, }); diff --git a/src/__tests__/CodexACPAgent/list-sessions.test.ts b/src/__tests__/CodexACPAgent/list-sessions.test.ts index 845ee0fd..f4a68977 100644 --- a/src/__tests__/CodexACPAgent/list-sessions.test.ts +++ b/src/__tests__/CodexACPAgent/list-sessions.test.ts @@ -76,6 +76,7 @@ describe("CodexACPAgent - list sessions", () => { const response = await codexAcpAgent.listSessions(params); expect(codexAppServerClient.threadList).toHaveBeenCalledWith(expect.objectContaining({ + limit: 20, sourceKinds: [ "cli", "vscode", From a22a59e89f5e8f53814feed4f3d8aaf5cb418e1d Mon Sep 17 00:00:00 2001 From: Kelvin Jayanoris Date: Fri, 31 Jul 2026 16:47:39 +0300 Subject: [PATCH 5/6] fix: match Codex interactive resume sources --- src/CodexAcpClient.ts | 3 --- src/__tests__/CodexACPAgent/list-sessions.test.ts | 3 --- 2 files changed, 6 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 5c81d69b..655d4265 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -793,9 +793,6 @@ export class CodexAcpClient { const sourceKinds: ThreadSourceKind[] = [ "cli", "vscode", - "exec", - "appServer", - "unknown", ]; const requestedCwd = request.cwd?.trim() ?? null; const filterByCwd = (thread: Thread): boolean => { diff --git a/src/__tests__/CodexACPAgent/list-sessions.test.ts b/src/__tests__/CodexACPAgent/list-sessions.test.ts index f4a68977..f7c8d83e 100644 --- a/src/__tests__/CodexACPAgent/list-sessions.test.ts +++ b/src/__tests__/CodexACPAgent/list-sessions.test.ts @@ -80,9 +80,6 @@ describe("CodexACPAgent - list sessions", () => { sourceKinds: [ "cli", "vscode", - "exec", - "appServer", - "unknown", ], })); await expect(JSON.stringify(response, null, 2)).toMatchFileSnapshot( From a4774ba6c0de793f0298ee526a7dcfdbde7378d2 Mon Sep 17 00:00:00 2001 From: Kelvin Jayanoris Date: Fri, 31 Jul 2026 19:17:53 +0300 Subject: [PATCH 6/6] feat: report live session presence --- src/CodexAcpClient.ts | 11 +++++ .../data/list-sessions-thread-name.json | 9 +++- .../CodexACPAgent/data/list-sessions.json | 9 +++- .../CodexACPAgent/list-sessions.test.ts | 48 +++++++++++++++++++ 4 files changed, 75 insertions(+), 2 deletions(-) diff --git a/src/CodexAcpClient.ts b/src/CodexAcpClient.ts index 655d4265..6a7d7d92 100644 --- a/src/CodexAcpClient.ts +++ b/src/CodexAcpClient.ts @@ -818,6 +818,17 @@ export class CodexAcpClient { cwd: thread.cwd, title: (thread.name ?? thread.preview) || null, updatedAt: new Date(thread.updatedAt * 1000).toISOString(), + _meta: { + codex: { + livePeer: { + presence: thread.status.type === "active" + ? "working" + : thread.status.type === "idle" + ? "live" + : "saved", + }, + }, + }, }); if (listResponse.data.length === 0) { diff --git a/src/__tests__/CodexACPAgent/data/list-sessions-thread-name.json b/src/__tests__/CodexACPAgent/data/list-sessions-thread-name.json index 406c9701..3aa434eb 100644 --- a/src/__tests__/CodexACPAgent/data/list-sessions-thread-name.json +++ b/src/__tests__/CodexACPAgent/data/list-sessions-thread-name.json @@ -4,7 +4,14 @@ "sessionId": "sess-1", "cwd": "/repo/project", "title": "Saved title", - "updatedAt": "1970-01-01T00:03:20.000Z" + "updatedAt": "1970-01-01T00:03:20.000Z", + "_meta": { + "codex": { + "livePeer": { + "presence": "live" + } + } + } } ], "nextCursor": null diff --git a/src/__tests__/CodexACPAgent/data/list-sessions.json b/src/__tests__/CodexACPAgent/data/list-sessions.json index 2791400d..54407eed 100644 --- a/src/__tests__/CodexACPAgent/data/list-sessions.json +++ b/src/__tests__/CodexACPAgent/data/list-sessions.json @@ -4,7 +4,14 @@ "sessionId": "sess-1", "cwd": "/repo/project", "title": "First session", - "updatedAt": "1970-01-01T00:03:20.000Z" + "updatedAt": "1970-01-01T00:03:20.000Z", + "_meta": { + "codex": { + "livePeer": { + "presence": "live" + } + } + } } ], "nextCursor": "next-cursor" diff --git a/src/__tests__/CodexACPAgent/list-sessions.test.ts b/src/__tests__/CodexACPAgent/list-sessions.test.ts index f7c8d83e..57b971d3 100644 --- a/src/__tests__/CodexACPAgent/list-sessions.test.ts +++ b/src/__tests__/CodexACPAgent/list-sessions.test.ts @@ -134,6 +134,54 @@ describe("CodexACPAgent - list sessions", () => { ); }); + it("reports whether a session is working, live, or only saved", async () => { + const fixture = createCodexMockTestFixture(); + const codexAcpAgent = fixture.getCodexAcpAgent(); + const codexAcpClient = fixture.getCodexAcpClient(); + const codexAppServerClient = fixture.getCodexAppServerClient(); + + codexAcpClient.authRequired = vi.fn().mockResolvedValue(false); + const base: Thread = { + id: "working", + sessionId: "working", + parentThreadId: null, + threadSource: null, + forkedFromId: null, + preview: "Working session", + ephemeral: false, + modelProvider: "openai", + createdAt: 100, + updatedAt: 200, + recencyAt: null, + status: {type: "active", activeFlags: []}, + path: null, + cwd: "/repo/project", + cliVersion: "0.0.0", + source: "cli", + agentNickname: null, + agentRole: null, + gitInfo: null, + name: null, + turns: [], + }; + codexAppServerClient.threadList = vi.fn().mockResolvedValue({ + data: [ + base, + {...base, id: "live", sessionId: "live", status: {type: "idle"}}, + {...base, id: "saved", sessionId: "saved", status: {type: "notLoaded"}}, + ], + nextCursor: null, + }); + + const response = await codexAcpAgent.listSessions({cwd: null, cursor: null}); + + expect(response.sessions.map((session) => session._meta)).toEqual([ + {codex: {livePeer: {presence: "working"}}}, + {codex: {livePeer: {presence: "live"}}}, + {codex: {livePeer: {presence: "saved"}}}, + ]); + }); + it("includes tracked additional directories for active sessions", async () => { const fixture = createCodexMockTestFixture(); const codexAcpAgent = fixture.getCodexAcpAgent();