From d119eb4fafc15dbe3eae5b14f7281d983f5e7c19 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 1 Aug 2026 23:21:57 +0800 Subject: [PATCH] feat: share Codex app-server sessions with Desktop --- README.en.md | 13 ++ README.md | 6 + npm-shrinkwrap.json | 3 +- package.json | 3 +- src/bridge/background-turns.ts | 13 +- src/bridge/bridge.ts | 40 +++++ src/bridge/formatters.ts | 8 + src/bridge/route-queue.ts | 12 ++ src/codex/app-server-codex-adapter.ts | 3 + src/codex/app-server/rpc-client.ts | 147 +++++++++++++++++-- src/codex/app-server/turn-controller.ts | 38 +++++ src/codex/types.ts | 1 + src/state/file-state-store.ts | 2 +- src/state/memory-state-store.ts | 4 + src/vendor.d.ts | 31 ++++ tests/integration/bridge-persistence.test.ts | 77 ++++++++++ tests/unit/app-server-core-modules.test.ts | 103 +++++++++++++ 17 files changed, 485 insertions(+), 19 deletions(-) diff --git a/README.en.md b/README.en.md index e7dc90e..d39960c 100644 --- a/README.en.md +++ b/README.en.md @@ -105,6 +105,19 @@ Chat-Codex enables pairing protection for real Weixin and Feishu chats by defaul After pairing succeeds, that chat is stored as a trusted route and remains usable after restarts. Pairing is scoped per chat route: different Weixin contacts and different Feishu private `chat_id` values each need to be paired once. +### Shared Codex App Server + +By default, Chat-Codex starts its own stdio Codex app-server. Advanced setups can connect to an existing app-server over a Unix socket: + +| Variable | Default | Description | +| --- | --- | --- | +| `CHAT_CODEX_APP_SERVER_ENDPOINT` | unset | Shared endpoint, currently `unix:///absolute/path.sock`. | +| `CHAT_CODEX_APP_SERVER_DAEMON` | unset | Set to `1` to use the shared Unix socket mode. | +| `CHAT_CODEX_APP_SERVER_SOCKET` | `~/.codex/app-server-control/app-server-control.sock` | Socket path used in daemon mode. | +| `CHAT_CODEX_MIRROR_DESKTOP_PROMPTS` | unset | Set to `1` to mirror Desktop user prompts to the chat route that owns the Codex session. | + +The shared connection reconnects with backoff and restores trusted routes, session ownership, and active sessions after restart. Desktop prompt mirroring is opt-in and route-scoped; messages originating from Chat-Codex are not mirrored back. + ## Tech Stack | Area | Technology | diff --git a/README.md b/README.md index 1d9fdf5..c18327e 100644 --- a/README.md +++ b/README.md @@ -123,9 +123,15 @@ Chat-Codex 对真实微信/飞书聊天默认启用配对保护。第一次从 | `CHAT_CODEX_BIN` | 未设置 | 覆盖 Codex CLI 可执行文件路径；主要用于 Windows Codex CLI 路径排障。 | | `CHAT_CODEX_STATE_DIR` | 未设置 | 覆盖状态根目录；相对路径按启动 `chat-codex` 时的工作目录解析。 | | `CHAT_CODEX_UPLOAD_DIR` | 未设置 | 覆盖上传目录；相对路径按启动 `chat-codex` 时的工作目录解析。 | +| `CHAT_CODEX_APP_SERVER_ENDPOINT` | 未设置 | 连接已有 Codex app-server；目前支持 `unix:///absolute/path.sock`。未设置时仍由 Chat-Codex 启动独立的 stdio app-server。 | +| `CHAT_CODEX_APP_SERVER_DAEMON` | 未设置 | 设为 `1` 时连接共享 Unix socket；可配合 `CHAT_CODEX_APP_SERVER_SOCKET` 指定路径。 | +| `CHAT_CODEX_APP_SERVER_SOCKET` | `~/.codex/app-server-control/app-server-control.sock` | 共享 app-server 的 Unix socket 路径，仅在 daemon 模式下使用。 | +| `CHAT_CODEX_MIRROR_DESKTOP_PROMPTS` | 未设置 | 设为 `1` 时，将同一 Codex session 中来自 Desktop 的用户消息镜像到其所属聊天 route。 | 旧版本曾默认写入启动目录下的 `state/` 和 `.chat-codex-uploads/`。升级后如果需要读取旧数据，可以把旧 `state/` 移到 `~/.chat-codex/state/`，或临时设置 `CHAT_CODEX_STATE_DIR=/old/start/dir/state`。 +共享 app-server 属于高级配置。它会在断线后自动退避重连，并在重启后恢复可信 route、session owner 和活动 session。Desktop 消息镜像默认关闭，且只投递给该 session 已绑定的 route；Chat-Codex 自己发送的消息不会被再次镜像。 + ## 技术栈 | 模块 | 技术 | diff --git a/npm-shrinkwrap.json b/npm-shrinkwrap.json index 27213e2..6a81707 100644 --- a/npm-shrinkwrap.json +++ b/npm-shrinkwrap.json @@ -16,7 +16,8 @@ "@larksuiteoapi/node-sdk": "^1.71.1", "ink": "^7.0.3", "qrcode-terminal": "^0.12.0", - "react": "^19.2.6" + "react": "^19.2.6", + "ws": "^8.20.1" }, "bin": { "chat-codex": "dist/src/cli.js" diff --git a/package.json b/package.json index ade46dc..2e4f1fb 100644 --- a/package.json +++ b/package.json @@ -71,7 +71,8 @@ "@larksuiteoapi/node-sdk": "^1.71.1", "ink": "^7.0.3", "qrcode-terminal": "^0.12.0", - "react": "^19.2.6" + "react": "^19.2.6", + "ws": "^8.20.1" }, "overrides": { "axios": "^1.16.1", diff --git a/src/bridge/background-turns.ts b/src/bridge/background-turns.ts index 9b2fe4a..6cb95fd 100644 --- a/src/bridge/background-turns.ts +++ b/src/bridge/background-turns.ts @@ -7,7 +7,7 @@ import type { ChannelMessage, ChannelTarget } from "../protocol/channel.js"; import type { ChannelDeliveryPolicy } from "../protocol/delivery-policy.js"; import type { MemoryStateStore } from "../state/memory-state-store.js"; import type { BackgroundTurnState } from "./bridge-types.js"; -import { composeFinalAnswer } from "./formatters.js"; +import { composeFinalAnswer, desktopPromptMirrorText } from "./formatters.js"; import type { BridgeDelivery } from "./delivery.js"; import { contextCompactionNotice } from "./context-compaction.js"; import { BridgeProgressDelivery } from "./progress-delivery.js"; @@ -115,6 +115,17 @@ export class BridgeBackgroundTurns { startedAt: event.startedAt ?? new Date().toISOString(), }); this.startTypingKeepalive(state); + } else if (event.type === "user.input") { + try { + await this.delivery.sendText(state.target, desktopPromptMirrorText(event.text)); + } catch (error) { + this.logger.warn("desktop prompt mirror delivery failed", { + routeKey: state.routeKey, + sessionId: event.sessionId, + turnId: event.turnId, + error: error instanceof Error ? error.message : String(error), + }); + } } else if (event.type === "context.compaction") { if (this.shouldDeliverContextCompaction(state.routeKey, event.sessionId)) { await this.delivery.sendText(state.target, contextCompactionNotice(event)); diff --git a/src/bridge/bridge.ts b/src/bridge/bridge.ts index dffe739..3585357 100644 --- a/src/bridge/bridge.ts +++ b/src/bridge/bridge.ts @@ -390,6 +390,7 @@ export class Bridge { } async start(): Promise { + await this.restorePersistedRoutes(); this.stopBackgroundEvents = this.codex.onBackgroundEvent?.((event) => this.backgroundTurns.handle(event)); this.channels.onMessage((message) => this.handleMessage(message)); this.channels.onApprovalAction((action) => this.handleChannelApprovalAction(action)); @@ -397,6 +398,45 @@ export class Bridge { this.logger.info("bridge started", { channels: this.channels.ids().join(",") }); } + private async restorePersistedRoutes(): Promise { + const resumed = new Set(); + for (const route of this.state.listRoutes()) { + if (!route.routeKey || !route.channelId || !route.conversationId) continue; + if (!this.state.isRouteTrusted(route.routeKey)) continue; + const senderId = route.identity?.lastSenderId ?? route.conversationId; + const message: ChannelMessage = { + id: `restored-route:${route.routeKey}`, + channelId: route.channelId, + accountId: route.accountId, + routeKey: route.routeKey, + conversation: { + kind: route.conversationKind, + id: route.conversationId, + displayName: route.displayName, + }, + sender: { id: senderId }, + timestamp: route.lastSeenAt ?? new Date().toISOString(), + text: "", + }; + this.routeMessages.set(route.routeKey, message); + this.routeTargets.set(route.routeKey, replyTargetFromMessage(message)); + const sessionId = route.activeSessionId; + const owner = sessionId ? this.state.getSessionOwner(sessionId) : undefined; + if (!sessionId || owner?.ownerRouteKey !== route.routeKey || resumed.has(sessionId)) continue; + try { + await this.codex.resumeSession(sessionId); + this.applyStoredSessionRunPolicy(sessionId); + resumed.add(sessionId); + } catch (error) { + this.logger.warn("persisted Codex session resume failed", { + sessionId, + routeKey: route.routeKey, + error: error instanceof Error ? error.message : String(error), + }); + } + } + } + async stop(): Promise { this.routeSteering.clearAll(); this.pendingMedia.clearAll(); diff --git a/src/bridge/formatters.ts b/src/bridge/formatters.ts index b3d1ed1..5a221dc 100644 --- a/src/bridge/formatters.ts +++ b/src/bridge/formatters.ts @@ -28,6 +28,14 @@ export function truncateForChannel(text: string, maxLength = 600): string { return `${normalized.slice(0, maxLength)}...`; } +export function desktopPromptMirrorText(text: string): string { + const normalized = text.trim(); + const visible = normalized.length > 3000 + ? `${normalized.slice(0, 3000)}\n…（电脑端提示词过长，手机镜像已截断）` + : normalized; + return `【电脑端提示词】\n${visible}`; +} + export function isSteerableStatus(status: CodexSessionStatus["type"]): boolean { return status === "running" || status === "waiting_approval" || status === "waiting_input"; } diff --git a/src/bridge/route-queue.ts b/src/bridge/route-queue.ts index 72e2f17..b39b57b 100644 --- a/src/bridge/route-queue.ts +++ b/src/bridge/route-queue.ts @@ -23,6 +23,7 @@ import type { BridgeSessionFlow } from "./session-flow.js"; import { StaleCodexSessionBindingError, type StaleCodexSessionBindingInfo } from "./session-flow.js"; import { composeFinalAnswer, + desktopPromptMirrorText, truncateForChannel, withSendFileInstruction, } from "./formatters.js"; @@ -286,6 +287,17 @@ export class BridgeRouteQueue { task: truncateForChannel(promptText || codexInputPlainText(prompt), 120), startedAt: currentTurnStartedAt, }); + } else if (event.type === "user.input") { + try { + await this.delivery.sendText(target, desktopPromptMirrorText(event.text)); + } catch (error) { + this.logger.warn("desktop prompt mirror delivery failed", { + routeKey: message.routeKey, + sessionId: event.sessionId, + turnId: event.turnId, + error: error instanceof Error ? error.message : String(error), + }); + } } else if (event.type === "context.compaction") { await this.delivery.sendText(target, contextCompactionNotice(event)); } else if (event.type === "assistant.progress") { diff --git a/src/codex/app-server-codex-adapter.ts b/src/codex/app-server-codex-adapter.ts index 81f4cfd..46925dc 100644 --- a/src/codex/app-server-codex-adapter.ts +++ b/src/codex/app-server-codex-adapter.ts @@ -70,6 +70,7 @@ export interface AppServerCodexAdapterOptions { requestTimeoutMs?: number; interruptTimeoutMs?: number; compactTimeoutMs?: number; + appServerEndpoint?: string; } interface CompactWaiter { @@ -117,6 +118,7 @@ export class AppServerCodexAdapter implements CodexAdapter { onServerRequest: (request) => this.handleServerRequest(request), onNotification: (notification) => this.handleNotification(notification), onFatalError: (error) => this.handleFatalAppServerError(error), + appServerEndpoint: options.appServerEndpoint, }); } @@ -147,6 +149,7 @@ export class AppServerCodexAdapter implements CodexAdapter { approvalsReviewer: approvalsReviewerForRunPolicy(this.defaultRunPolicy), sandbox: sandboxModeForRunPolicy(this.defaultRunPolicy), serviceName: "codex-chat-bridge", + threadSource: "user", sessionStartSource: "startup", }); const thread = objectValue(response.thread); diff --git a/src/codex/app-server/rpc-client.ts b/src/codex/app-server/rpc-client.ts index 8525b32..88b496a 100644 --- a/src/codex/app-server/rpc-client.ts +++ b/src/codex/app-server/rpc-client.ts @@ -1,6 +1,10 @@ import type { ChildProcess } from "node:child_process"; +import net from "node:net"; +import os from "node:os"; +import path from "node:path"; import { createInterface } from "node:readline"; import type { Interface as ReadlineInterface } from "node:readline"; +import WebSocket from "ws"; import { resolveCodexCommand, spawnCodex, type CodexCommandResolution } from "../codex-process.js"; import type { JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, PendingResponse } from "./types.js"; @@ -10,6 +14,7 @@ export interface AppServerRpcClientOptions { onServerRequest: (request: JsonRpcRequest) => Promise | void; onNotification: (notification: JsonRpcNotification) => void; onFatalError: (error: Error) => void; + appServerEndpoint?: string; } export class AppServerRpcClient { @@ -18,14 +23,18 @@ export class AppServerRpcClient { private readonly onServerRequest: (request: JsonRpcRequest) => Promise | void; private readonly onNotification: (notification: JsonRpcNotification) => void; private readonly onFatalError: (error: Error) => void; + private readonly appServerEndpoint?: string; private readonly pendingResponses = new Map(); private requestSequence = 0; private child?: ChildProcess; + private webSocket?: WebSocket; private stdoutLines?: ReadlineInterface; private stderr = ""; private initialized?: Promise; private stopping = false; private processGeneration = 0; + private reconnectAttempt = 0; + private reconnectTimer?: ReturnType; constructor(options: AppServerRpcClientOptions) { this.codexCommand = typeof options.codexBin === "string" ? resolveCodexCommand({ codexBin: options.codexBin }) : options.codexBin; @@ -33,16 +42,28 @@ export class AppServerRpcClient { this.onServerRequest = options.onServerRequest; this.onNotification = options.onNotification; this.onFatalError = options.onFatalError; + this.appServerEndpoint = options.appServerEndpoint ?? sharedAppServerEndpointFromEnvironment(); } start(): Promise { - this.initialized ??= this.startProcessAndInitialize(); + this.initialized ??= this.startProcessAndInitialize() + .then(() => { + this.reconnectAttempt = 0; + }) + .catch((error) => { + this.initialized = undefined; + this.webSocket = undefined; + this.scheduleSharedReconnect(); + throw error; + }); return this.initialized; } stop(): void { this.stopping = true; this.processGeneration += 1; + if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + this.reconnectTimer = undefined; for (const pending of this.pendingResponses.values()) { pending.reject(new Error("codex app-server stopped")); } @@ -51,7 +72,15 @@ export class AppServerRpcClient { if (this.child && !this.child.killed) { this.child.kill("SIGTERM"); } + if (this.webSocket) { + try { + this.webSocket.terminate(); + } catch { + // Best effort while shutting down. + } + } this.child = undefined; + this.webSocket = undefined; this.initialized = undefined; } @@ -100,6 +129,10 @@ export class AppServerRpcClient { } writeMessage(message: unknown): void { + if (this.webSocket?.readyState === WebSocket.OPEN) { + this.webSocket.send(JSON.stringify(message)); + return; + } if (!this.child?.stdin || this.child.stdin.destroyed) { throw new Error("codex app-server stdin is closed"); } @@ -110,20 +143,24 @@ export class AppServerRpcClient { this.stopping = false; const generation = ++this.processGeneration; this.stderr = ""; - this.child = spawnCodex(this.codexCommand, ["app-server", "--listen", "stdio://"], { - stdio: ["pipe", "pipe", "pipe"], - }); - this.child.stderr?.setEncoding("utf8"); - this.child.stderr?.on("data", (chunk: string) => { - this.stderr += chunk; - }); - this.child.on("error", (error) => this.handleProcessEnd(error, generation)); - this.child.on("close", (code) => { - this.handleProcessEnd(new Error(this.stderr.trim() || `codex app-server exited with code ${code}`), generation); - }); - if (!this.child.stdout || !this.child.stdin) throw new Error("failed to start codex app-server stdio"); - this.stdoutLines = createInterface({ input: this.child.stdout }); - void this.readLoop(); + if (this.appServerEndpoint) { + await this.connectToSharedAppServer(this.appServerEndpoint, generation); + } else { + this.child = spawnCodex(this.codexCommand, ["app-server", "--listen", "stdio://"], { + stdio: ["pipe", "pipe", "pipe"], + }); + this.child.stderr?.setEncoding("utf8"); + this.child.stderr?.on("data", (chunk: string) => { + this.stderr += chunk; + }); + this.child.on("error", (error) => this.handleProcessEnd(error, generation)); + this.child.on("close", (code) => { + this.handleProcessEnd(new Error(this.stderr.trim() || `codex app-server exited with code ${code}`), generation); + }); + if (!this.child.stdout || !this.child.stdin) throw new Error("failed to start codex app-server stdio"); + this.stdoutLines = createInterface({ input: this.child.stdout }); + void this.readLoop(); + } await this.request("initialize", { clientInfo: { name: "codex-chat-bridge", @@ -142,6 +179,50 @@ export class AppServerRpcClient { this.writeMessage({ method: "initialized" }); } + private async connectToSharedAppServer(endpoint: string, generation: number): Promise { + const socketPath = unixSocketPath(endpoint); + await new Promise((resolve, reject) => { + let opened = false; + let ended = false; + const webSocket = new WebSocket("ws://localhost/rpc", { + perMessageDeflate: false, + createConnection: () => net.createConnection({ path: socketPath }), + }); + this.webSocket = webSocket; + const endOnce = (error: Error) => { + if (ended) return; + ended = true; + if (!opened) { + reject(error); + return; + } + this.handleProcessEnd(error, generation); + }; + webSocket.once("open", () => { + opened = true; + resolve(); + }); + webSocket.on("message", (data) => { + try { + const message = JSON.parse(data.toString()) as JsonRpcResponse | JsonRpcRequest | JsonRpcNotification; + void this.handleMessage(message); + } catch (error) { + endOnce(error instanceof Error ? error : new Error(String(error))); + try { + webSocket.terminate(); + } catch { + // Best effort after malformed transport data. + } + } + }); + webSocket.on("error", (error) => endOnce(error)); + webSocket.on("close", (code, reason) => { + const detail = reason.toString().trim(); + endOnce(new Error(`shared codex app-server closed (${code})${detail ? `: ${detail}` : ""}`)); + }); + }); + } + private handleProcessEnd(error: Error, generation: number): void { if (generation !== this.processGeneration) return; if (this.stopping) { @@ -154,15 +235,29 @@ export class AppServerRpcClient { this.pendingResponses.clear(); this.initialized = undefined; this.child = undefined; + this.webSocket = undefined; this.onFatalError(error); + this.scheduleSharedReconnect(); } private async ensureChildOpen(): Promise { + if (this.webSocket?.readyState === WebSocket.OPEN) return; if (!this.child?.stdin || this.child.killed) { throw new Error("codex app-server is not running"); } } + private scheduleSharedReconnect(): void { + if (!this.appServerEndpoint || this.stopping || this.reconnectTimer) return; + const delayMs = Math.min(1000 * (2 ** this.reconnectAttempt), 30_000); + this.reconnectAttempt += 1; + this.reconnectTimer = setTimeout(() => { + this.reconnectTimer = undefined; + void this.start().catch(() => undefined); + }, delayMs); + this.reconnectTimer.unref?.(); + } + private async readLoop(): Promise { if (!this.stdoutLines) return; try { @@ -201,3 +296,25 @@ export class AppServerRpcClient { } } } + +function sharedAppServerEndpointFromEnvironment(): string | undefined { + const explicit = process.env.CHAT_CODEX_APP_SERVER_ENDPOINT?.trim(); + if (explicit) return explicit; + if (process.env.CHAT_CODEX_APP_SERVER_DAEMON !== "1") return undefined; + const configuredSocket = process.env.CHAT_CODEX_APP_SERVER_SOCKET?.trim(); + const codexHome = process.env.CODEX_HOME?.trim() || path.join(os.homedir(), ".codex"); + const socketPath = configuredSocket || path.join(codexHome, "app-server-control", "app-server-control.sock"); + return `unix://${socketPath}`; +} + +function unixSocketPath(endpoint: string): string { + const parsed = new URL(endpoint); + if (parsed.protocol !== "unix:" || parsed.host) { + throw new Error(`unsupported codex app-server endpoint: ${endpoint}`); + } + const socketPath = decodeURIComponent(parsed.pathname); + if (!path.isAbsolute(socketPath)) { + throw new Error(`codex app-server Unix socket must be absolute: ${endpoint}`); + } + return socketPath; +} diff --git a/src/codex/app-server/turn-controller.ts b/src/codex/app-server/turn-controller.ts index 0912151..cfd8ec7 100644 --- a/src/codex/app-server/turn-controller.ts +++ b/src/codex/app-server/turn-controller.ts @@ -31,6 +31,7 @@ export class AppServerTurnController { private readonly earlyTurnEvents = new Map(); private readonly closedTurnIds = new Set(); private readonly backgroundHandlers = new Set(); + private readonly mirroredDesktopUserItems = new Set(); constructor(options: AppServerTurnControllerOptions) { this.sessions = options.sessions; @@ -282,6 +283,28 @@ export class AppServerTurnController { const itemType = stringValue(item.type); const itemId = stringValue(item.id); if (itemId) this.flushProgressDraft(turn, sessionId, turnId, itemId); + if (itemType === "userMessage") { + if (process.env.CHAT_CODEX_MIRROR_DESKTOP_PROMPTS !== "1") return; + const clientId = stringValue(item.clientId); + const text = textFromDesktopUserMessage(item); + if (!isDesktopClientId(clientId) || !text) return; + const mirrorKey = `${sessionId}:${itemId ?? clientId}`; + if (this.mirroredDesktopUserItems.has(mirrorKey)) return; + this.mirroredDesktopUserItems.add(mirrorKey); + if (this.mirroredDesktopUserItems.size > 512) { + const oldest = this.mirroredDesktopUserItems.values().next().value; + if (oldest) this.mirroredDesktopUserItems.delete(oldest); + } + turn.queue.push({ + type: "user.input", + sessionId, + turnId, + clientId, + text, + ...(itemId ? { itemId } : {}), + }); + return; + } if (itemType === "contextCompaction" || itemType === "context_compaction") { this.pushTurnEvent(turnId, { type: "context.compaction", sessionId, turnId, phase: "completed" }); return; @@ -500,6 +523,21 @@ export class AppServerTurnController { } } +function textFromDesktopUserMessage(item: Record): string { + return arrayValue(item.content) + .map((entry) => { + const part = objectValue(entry); + return stringValue(part.type) === "text" ? stringValue(part.text) : undefined; + }) + .filter((entry): entry is string => Boolean(entry)) + .join("\n") + .trim(); +} + +function isDesktopClientId(clientId: string | undefined): clientId is string { + return Boolean(clientId && /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(clientId)); +} + function startedAtFromNotification(params: Record): string | undefined { const turn = objectValue(params.turn); return isoFromSeconds(numberValue(turn.startedAt)) diff --git a/src/codex/types.ts b/src/codex/types.ts index 33ce34b..fe1ee68 100644 --- a/src/codex/types.ts +++ b/src/codex/types.ts @@ -183,6 +183,7 @@ export type CodexEvent = | { type: "context.compaction"; sessionId: string; turnId: string; phase: "started" | "completed" } | { type: "assistant.progress"; sessionId: string; turnId: string; text: string; kind?: CodexProgressKind } | { type: "assistant.commentary"; sessionId: string; turnId: string; text: string; itemId?: string } + | { type: "user.input"; sessionId: string; turnId: string; text: string; itemId?: string; clientId: string } | { type: "tool.progress"; sessionId: string; turnId: string; progress: CodexToolProgress } | { type: "codex.notification"; sessionId: string; turnId: string; notification: CodexNotification } | { type: "assistant.plan"; sessionId: string; turnId: string; text: string } diff --git a/src/state/file-state-store.ts b/src/state/file-state-store.ts index 36e6e14..53093cc 100644 --- a/src/state/file-state-store.ts +++ b/src/state/file-state-store.ts @@ -339,7 +339,7 @@ export class FileStateStore extends MemoryStateStore { .sort((left, right) => left.routeKey.localeCompare(right.routeKey)); } - listRoutes(): RouteRecord[] { + override listRoutes(): RouteRecord[] { return [...this.routes.values()].sort((left, right) => left.routeKey.localeCompare(right.routeKey)); } diff --git a/src/state/memory-state-store.ts b/src/state/memory-state-store.ts index 2e44f8b..63aa1bb 100644 --- a/src/state/memory-state-store.ts +++ b/src/state/memory-state-store.ts @@ -98,6 +98,10 @@ export class MemoryStateStore { return undefined; } + listRoutes(): RouteRecord[] { + return []; + } + isRouteTrusted(routeKey: string): boolean { return this.trustedRoutes.has(routeKey); } diff --git a/src/vendor.d.ts b/src/vendor.d.ts index 15ba038..c5aa99b 100644 --- a/src/vendor.d.ts +++ b/src/vendor.d.ts @@ -8,3 +8,34 @@ declare module "qrcode-terminal" { }; export default qrcodeTerminal; } + +declare module "ws" { + import type { Duplex } from "node:stream"; + + export type RawData = Buffer | ArrayBuffer | Buffer[]; + + export interface ClientOptions { + perMessageDeflate?: boolean; + createConnection?: () => Duplex; + } + + export default class WebSocket { + static readonly OPEN: number; + readonly readyState: number; + + constructor(address: string, options?: ClientOptions); + + send(data: string): void; + terminate(): void; + once(event: "open", listener: () => void): this; + on(event: "message", listener: (data: RawData) => void): this; + on(event: "error", listener: (error: Error) => void): this; + on(event: "close", listener: (code: number, reason: Buffer) => void): this; + } + + export class WebSocketServer { + constructor(options: { server: import("node:http").Server }); + on(event: "connection", listener: (socket: WebSocket) => void): this; + close(callback: (error?: Error) => void): void; + } +} diff --git a/tests/integration/bridge-persistence.test.ts b/tests/integration/bridge-persistence.test.ts index ea5cfd7..6643535 100644 --- a/tests/integration/bridge-persistence.test.ts +++ b/tests/integration/bridge-persistence.test.ts @@ -6,11 +6,31 @@ import path from "node:path"; import { Bridge } from "../../src/bridge/bridge.js"; import { MockChannelAdapter } from "../../src/channels/mock/mock-channel-adapter.js"; import { MockCodexAdapter } from "../../src/codex/mock-codex-adapter.js"; +import type { CodexBackgroundEventHandler, CodexEvent } from "../../src/codex/types.js"; import { SilentLogger } from "../../src/logging/logger.js"; import type { TranscriptSink } from "../../src/logging/transcript.js"; import type { ChannelMessage } from "../../src/protocol/channel.js"; import { FileStateStore } from "../../src/state/file-state-store.js"; +class BackgroundMockCodexAdapter extends MockCodexAdapter { + readonly resumedSessions: string[] = []; + private readonly backgroundHandlers = new Set(); + + override async resumeSession(sessionId: string) { + this.resumedSessions.push(sessionId); + return await super.resumeSession(sessionId); + } + + onBackgroundEvent(handler: CodexBackgroundEventHandler): () => void { + this.backgroundHandlers.add(handler); + return () => this.backgroundHandlers.delete(handler); + } + + async emitBackground(event: CodexEvent): Promise { + for (const handler of this.backgroundHandlers) await handler(event); + } +} + test("Bridge restores route session binding from FileStateStore after restart", async () => { const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "codex-bridge-persist-")); const codex = new MockCodexAdapter(); @@ -51,6 +71,63 @@ test("Bridge restores route session binding from FileStateStore after restart", assert.equal(fs.existsSync(path.join(rootDir, "session-owners.json")), true); }); +test("Bridge restores trusted route targets before delivering Desktop prompt mirrors", async () => { + const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "codex-bridge-desktop-mirror-")); + const routeKey = "weixin-main:wx-account:direct:wx-user"; + const codex = new BackgroundMockCodexAdapter(); + const session = await codex.startSession({ routeKey, cwd: "/tmp/work", title: "desktop mirror" }); + const state = new FileStateStore({ rootDir }); + const message: ChannelMessage = { + id: "wx-message-1", + routeKey, + channelId: "weixin-main", + accountId: "wx-account", + sender: { id: "wx-user" }, + conversation: { id: "wx-user", kind: "direct", displayName: "微信私聊" }, + text: "你好", + timestamp: "2026-08-01T00:00:00.000Z", + }; + state.recordRouteMessage(message); + state.trustRoute({ + routeKey, + channelId: "weixin-main", + accountId: "wx-account", + conversationKind: "direct", + conversationId: "wx-user", + trustedAt: "2026-08-01T00:00:00.000Z", + trustedBySenderId: "wx-user", + trustMethod: "manual", + createdAt: "2026-08-01T00:00:00.000Z", + updatedAt: "2026-08-01T00:00:00.000Z", + }); + state.bindSession(routeKey, session); + + const channel = new MockChannelAdapter({ id: "weixin-main", accountId: "wx-account" }); + const bridge = new Bridge({ + channel, + codex, + state: new FileStateStore({ rootDir }), + logger: new SilentLogger(), + cwd: "/tmp/work", + }); + await bridge.start(); + try { + await codex.emitBackground({ + type: "user.input", + sessionId: session.id, + turnId: "desktop-turn-1", + itemId: "desktop-item-1", + clientId: "7cb06f4c-7d2d-4f1a-a9f5-ef9cf2754898", + text: "电脑端复测", + }); + assert.deepEqual(codex.resumedSessions, [session.id]); + assert.equal(channel.sentMessages.at(-1)?.target.routeKey, routeKey); + assert.equal(channel.sentMessages.at(-1)?.text, "【电脑端提示词】\n电脑端复测"); + } finally { + await bridge.stop(); + } +}); + test("Bridge applies persisted session run policy when restoring a binding", async () => { const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "codex-bridge-policy-")); const routeKey = "feishu-main:default:direct:oc_user"; diff --git a/tests/unit/app-server-core-modules.test.ts b/tests/unit/app-server-core-modules.test.ts index 0b24bb2..d81871f 100644 --- a/tests/unit/app-server-core-modules.test.ts +++ b/tests/unit/app-server-core-modules.test.ts @@ -1,8 +1,10 @@ import test from "node:test"; import assert from "node:assert/strict"; import { chmod, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { createServer } from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { WebSocketServer } from "ws"; import type { CodexEvent } from "../../src/codex/types.js"; import { AppServerRpcClient } from "../../src/codex/app-server/rpc-client.js"; import { AppServerSessionStore } from "../../src/codex/app-server/session-store.js"; @@ -89,6 +91,69 @@ test("app-server turn controller maps notifications to queued events and status assert.equal(controller.hasActiveTurns(), false); }); +test("app-server turn controller mirrors Desktop user messages once and ignores bridge-origin inputs", async () => { + const previous = process.env.CHAT_CODEX_MIRROR_DESKTOP_PROMPTS; + process.env.CHAT_CODEX_MIRROR_DESKTOP_PROMPTS = "1"; + try { + const sessions = new Map(); + sessions.set("session-1", { + session: { id: "session-1", cwd: "/repo", createdAt: "now" }, + status: { type: "idle" }, + updatedAt: "now", + }); + const controller = new AppServerTurnController({ sessions, threadToSession: new Map([["thread-1", "session-1"]]) }); + const queue = new AsyncEventQueue(); + const iterator = queue[Symbol.asyncIterator](); + controller.registerTurn("session-1", "turn-1", queue); + const desktopItem = { + type: "userMessage", + id: "desktop-item-1", + clientId: "7cb06f4c-7d2d-4f1a-a9f5-ef9cf2754898", + content: [{ type: "text", text: "电脑端提示词" }], + }; + + controller.handleNotification({ + method: "item/completed", + params: { threadId: "thread-1", turnId: "turn-1", item: desktopItem }, + }); + assert.deepEqual(await iterator.next(), { + value: { + type: "user.input", + sessionId: "session-1", + turnId: "turn-1", + itemId: "desktop-item-1", + clientId: "7cb06f4c-7d2d-4f1a-a9f5-ef9cf2754898", + text: "电脑端提示词", + }, + done: false, + }); + + controller.handleNotification({ + method: "item/completed", + params: { threadId: "thread-1", turnId: "turn-1", item: desktopItem }, + }); + controller.handleNotification({ + method: "item/completed", + params: { + threadId: "thread-1", + turnId: "turn-1", + item: { ...desktopItem, id: "weixin-item-1", clientId: null }, + }, + }); + controller.handleNotification({ + method: "turn/completed", + params: { threadId: "thread-1", turnId: "turn-1", turn: { status: "completed" } }, + }); + assert.deepEqual(await iterator.next(), { + value: { type: "turn.completed", sessionId: "session-1", turnId: "turn-1" }, + done: false, + }); + } finally { + if (previous === undefined) delete process.env.CHAT_CODEX_MIRROR_DESKTOP_PROMPTS; + else process.env.CHAT_CODEX_MIRROR_DESKTOP_PROMPTS = previous; + } +}); + test("app-server turn controller emits notifications even after the turn is closed", async () => { const sessions = new Map(); sessions.set("session-1", { @@ -218,3 +283,41 @@ for await (const line of rl) { await rm(dir, { recursive: true, force: true }); } }); + +test("app-server rpc client connects to a shared app-server over a Unix socket", async () => { + const dir = await mkdtemp(join(tmpdir(), "chat-codex-unix-rpc-")); + const socketPath = join(dir, "app-server.sock"); + const server = createServer(); + const webSockets = new WebSocketServer({ server }); + webSockets.on("connection", (socket) => { + socket.on("message", (data) => { + const message = JSON.parse(data.toString()); + if (message.method === "initialize") { + socket.send(JSON.stringify({ id: message.id, result: { ok: true } })); + } else if (message.method === "model/list") { + socket.send(JSON.stringify({ id: message.id, result: { data: [{ id: "shared" }], nextCursor: null } })); + } + }); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(socketPath, resolve); + }); + const client = new AppServerRpcClient({ + codexBin: process.execPath, + requestTimeoutMs: 1000, + appServerEndpoint: `unix://${socketPath}`, + onServerRequest: () => undefined, + onNotification: () => undefined, + onFatalError: () => undefined, + }); + try { + await client.start(); + assert.deepEqual(await client.request("model/list"), { data: [{ id: "shared" }], nextCursor: null }); + } finally { + client.stop(); + await new Promise((resolve) => webSockets.close(() => resolve())); + await new Promise((resolve) => server.close(() => resolve())); + await rm(dir, { recursive: true, force: true }); + } +});