Add shared App Server live-peer mode

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
 

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"
   }
 }

src/AcpExtensions.ts

@@ -87,6 +87,7 @@ export async function legacySetSessionModel(
 export type SessionSteerRequest = {
     sessionId: SessionId;
     prompt: ContentBlock[];
+    _meta?: Record<string, unknown>;
 }
 
 export type SessionSteeringResponse = {

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";
@@ -60,6 +61,8 @@ const SUPPORTED_GATEWAY_PROTOCOLS: Record<acp.LlmProtocol, WireApi> = {
     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.
@@ -343,6 +346,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,
@@ -371,6 +375,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,
@@ -700,8 +705,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),
@@ -786,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 => {
@@ -804,6 +808,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,
         });
@@ -813,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) {
@@ -845,10 +861,16 @@ export class CodexAcpClient {
         });
     }
 
-    async steerTurn(params: { threadId: string, turnId: string, prompt: acp.ContentBlock[] }): Promise<TurnSteerResponse> {
+    async steerTurn(params: {
+        threadId: string,
+        turnId: string,
+        prompt: acp.ContentBlock[],
+        clientUserMessageId?: string | null,
+    }): Promise<TurnSteerResponse> {
         return await this.codexClient.turnSteer({
             threadId: params.threadId,
             expectedTurnId: params.turnId,
+            ...(params.clientUserMessageId ? {clientUserMessageId: params.clientUserMessageId} : {}),
             input: buildPromptItems(params.prompt),
         });
     }
@@ -901,12 +923,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) {