From 50d36e6db098c990305551da0a3fc678dfa0b2d4 Mon Sep 17 00:00:00 2001 From: CoderLambert Date: Fri, 11 Sep 2026 19:43:56 +0800 Subject: [PATCH 1/3] docs: add TanStack streaming state notes --- docs/notes/03-tanstack-streaming-state.md | 1505 +++++++++++++++++++++ 1 file changed, 1505 insertions(+) create mode 100644 docs/notes/03-tanstack-streaming-state.md diff --git a/docs/notes/03-tanstack-streaming-state.md b/docs/notes/03-tanstack-streaming-state.md new file mode 100644 index 0000000..905f4a2 --- /dev/null +++ b/docs/notes/03-tanstack-streaming-state.md @@ -0,0 +1,1505 @@ +# TanStack Start 流式 RPC 与 React 状态建模 + +> 本文基于 `codex-tanstack-start` 当前 `main`（`6f4584d2`）的真实实现，目标不是解释“怎么把一个聊天页面跑起来”，而是沉淀一套可迁移到其他 Agent / LLM 前端的流式交互设计方法。 + +## 1. 先建立正确的问题模型 + +传统 Web 请求通常是： + +```text +用户提交 + ↓ +HTTP request + ↓ +服务端完成全部工作 + ↓ +一次性 response + ↓ +setState(finalResult) +``` + +Agent / LLM 场景不同。一次 turn 在真正结束前会持续产生状态： + +```text +thread started +turn running +assistant item started +assistant delta +assistant delta +command/tool activity +assistant completed +turn completed +``` + +因此问题不再只是“请求一个结果”，而是： + +> 如何把服务端持续产生的事件安全地穿过 RPC 边界，转换成稳定的应用事件，再用确定性的客户端状态机持续折叠成 UI？ + +当前项目的主链路是： + +```text +React UI + ↓ +useChatController() + ↓ +TanStack Start createServerFn + ↓ +async generator + ↓ +streamNormalizedChatEvents() + ↓ +ChatEvent + ↓ +for await...of + ↓ +toChatStateEvent() + ↓ +chatReducer() + ↓ +React render +``` + +核心不是某一个 API，而是 **Boundary → Event Contract → State Machine** 三层分离。 + +--- + +## 2. `createServerFn` 的 Browser / Server 边界 + +当前入口： + +```ts +export const streamChat = createServerFn({ method: 'POST' }) + .validator(validateChatRequest) + .handler(async function* ({ data }) { + yield* streamNormalizedChatEvents(data, streamCodexTurn) + }) +``` + +文件： + +```text +src/server-functions/chat.ts +``` + +### 2.1 一个容易误解的点 + +这个模块 **允许被浏览器代码 import**，但 handler 不是因此就进入浏览器 bundle。 + +浏览器写： + +```ts +import { streamChat } from '../../server-functions/chat' +``` + +调用： + +```ts +const events = await streamChat({ data }) +``` + +语义上看像普通函数；构建后实际是： + +```text +browser function call + ↓ +TanStack Start RPC bridge + ↓ +server-side handler +``` + +这就是 Server Function 的价值： + +- 浏览器侧获得类型安全调用接口； +- handler 在服务端执行； +- 不需要手写传统 `/api/chat` controller； +- 参数 validator 放在 RPC 入口； +- async iterable 可以继续作为流返回。 + +### 2.2 为什么 Codex Runtime 必须继续放在 `*.server.ts` + +`chat.ts` 虽然 client-importable，但它依赖： + +```ts +import { streamCodexTurn } from './chat.runtime.server' +``` + +真正 runtime 再进入： + +```text +src/server/codex/** +``` + +原因不是目录美观，而是安全与运行环境隔离： + +```text +Browser +不能拥有： +- child_process +- 本地 Codex 认证 +- filesystem path +- app-server stdio +- command/tool 原始数据 +``` + +如果把 Runtime 直接写进一个可能进入 client graph 的普通模块，会产生两类风险： + +1. **构建风险**：Node-only API 被浏览器构建引用； +2. **安全风险**：认证、路径、协议对象意外越过边界。 + +### 2.3 设计原则 + +```text +client-importable API ≠ client-executable implementation +``` + +正确结构： + +```text +chat.ts // 浏览器可以 import 的 RPC declaration +chat.runtime.server.ts // server-only integration seam +server/codex/** // Node runtime / app-server +``` + +这是 full-stack framework 中非常重要的模块边界意识。 + +--- + +## 3. async generator 如何形成真正的 RPC Streaming + +服务端： + +```ts +.handler(async function* ({ data }) { + yield* streamNormalizedChatEvents(data, streamCodexTurn) +}) +``` + +bridge： + +```ts +export async function* streamNormalizedChatEvents( + request, + createCodexTurnStream, +) { + const source = await createCodexTurnStream(request) + const normalizer = createCodexEventNormalizer() + + for await (const codexEvent of source) { + for (const chatEvent of normalizer.normalize(codexEvent)) { + yield chatEvent + } + } +} +``` + +客户端： + +```ts +const events = await streamChat({ data }) + +for await (const event of events) { + dispatch({ + type: 'event.received', + event: toChatStateEvent(event, Date.now()), + }) +} +``` + +### 3.1 关键点 + +真正让 UI 流起来的不是 React，也不是 `setState`，而是整条链路每一层都没有把事件收集成数组： + +```text +Codex event + ↓ immediately +normalize + ↓ immediately +yield ChatEvent + ↓ immediately +RPC transport + ↓ immediately +for await + ↓ immediately +dispatch + ↓ +render +``` + +如果任何一层写成： + +```ts +const events = [] +for await (...) events.push(...) +return events +``` + +流式就会在这一层退化成批量返回。 + +### 3.2 流式系统的 Backpressure 心智模型 + +async iterator 天然是消费者驱动： + +```text +producer → yield → consumer next() → producer continues +``` + +它比“无限 callback 推送”更容易控制生命周期，也更适合服务端逐事件转换。 + +--- + +## 4. 为什么需要两套 Event Contract + +项目当前有两层事件： + +### Transport Contract：`ChatEvent` + +位置： + +```text +src/features/chat/chat.types.ts +``` + +例如： + +```ts +type ChatEvent = + | { type: 'thread.started'; threadId: string } + | { type: 'assistant.started'; id: string } + | { type: 'assistant.delta'; id: string; delta: string } + | { type: 'assistant.completed'; id: string; text: string } + | { type: 'turn.completed'; usage: ChatUsage } + | { type: 'error'; message: string } +``` + +它回答： + +> 服务端允许通过网络告诉浏览器什么？ + +### State Contract：`ChatStateEvent` + +位置： + +```text +src/features/chat/chat.reducer.ts +``` + +例如： + +```ts +type ChatStateEvent = + | { type: 'assistant.started'; id: string; createdAt: number } + | { type: 'assistant.delta'; id: string; delta: string } + | { type: 'assistant.completed'; id: string; text: string } +``` + +它回答： + +> reducer 需要什么信息才能确定性地更新本地状态？ + +### 4.1 两者为什么不能偷懒合并 + +因为网络事件和 UI 状态的关注点不同。 + +例如 `assistant.started`： + +```text +Transport: +{id} + +State: +{id, createdAt} +``` + +`createdAt` 是客户端持久化消息需要的字段，但没必要让 Codex Runtime 负责。 + +再例如： + +```text +turn.completed transport event +``` + +携带 token usage；但当前 reducer 只关心： + +```text +status: running → idle +``` + +所以 adapter 会把它收敛为： + +```ts +{ type: 'turn.completed' } +``` + +这是一种 **Anti-Corruption Layer（防腐层）** 思路： + +```text +Runtime Protocol + ↓ normalize +Transport Contract + ↓ adapt +State Contract + ↓ reduce +UI State +``` + +每层只暴露下一层真正需要知道的内容。 + +--- + +## 5. Adapter 层不是“多余的一层” + +当前 adapter： + +```ts +export function toChatStateEvent( + event: ChatEvent, + receivedAt: number, +): ChatStateEvent +``` + +它至少承担三个职责： + +### 5.1 注入客户端语义 + +```ts +case 'assistant.started': + return { + type: 'assistant.started', + id: event.id, + createdAt: receivedAt, + } +``` + +时间属于客户端接收时刻，不应该污染 server transport contract。 + +### 5.2 丢弃 state 不需要的数据 + +例如当前 reducer 并不存储 token usage： + +```ts +case 'turn.completed': + return { type: 'turn.completed' } +``` + +### 5.3 阻止 runtime shape 渗透进 React + +如果未来底层从 Codex 换成 Claude / Pi / Qwen，理想情况是： + +```text +新 runtime adapter + ↓ +仍然输出 ChatEvent + ↓ +React reducer 不改 +``` + +这就是应用协议稳定性的价值。 + +--- + +## 6. `useReducer`：把流式 UI 当状态机，而不是字符串拼接器 + +当前主状态： + +```ts +interface ChatState { + threadId: string | null + messages: ChatMessage[] + activities: AgentActivity[] + status: 'idle' | 'running' | 'error' + error: string | null +} +``` + +一个 assistant response 的生命周期： + +```mermaid +stateDiagram-v2 + [*] --> Waiting + Waiting --> Started: assistant.started + Started --> Streaming: assistant.delta + Streaming --> Streaming: assistant.delta + Streaming --> Completed: assistant.completed + Started --> Completed: assistant.completed + Completed --> [*] +``` + +整个 turn： + +```mermaid +stateDiagram-v2 + Idle --> Running: turn.started + Running --> Running: assistant/activity events + Running --> Idle: turn.completed + Running --> Error: error + Error --> Running: next turn + Idle --> Idle: new-chat +``` + +### 6.1 `assistant.started` + +创建空 assistant message： + +```ts +{ + id, + role: 'assistant', + content: '', + createdAt, +} +``` + +为什么不是等第一段 delta 再创建？ + +因为 `started` 是领域事件： + +```text +“这条 assistant message 已经存在，只是内容暂时为空。” +``` + +这样 UI 可以独立展示 loading cursor、message shell 或 metadata。 + +### 6.2 `assistant.delta` + +核心逻辑： + +```ts +messages[index] = { + ...message, + content: message.content + event.delta, +} +``` + +这里的 `delta` 是 **增量**，不是 snapshot。 + +所以： + +```text +"React" ++ " 是" ++ "一个 UI 库" +``` + +最终才得到完整文本。 + +### 6.3 `assistant.completed` + +completed 不只是“状态结束”，还承担最终校准： + +```ts +content: event.text +``` + +即： + +```text +delta stream 用于即时体验 +completed snapshot 用于最终正确性 +``` + +如果中间某个 delta 因协议、网络或 adapter 问题缺失，最终 snapshot 仍可以纠正消息。 + +这是非常值得复用的模式： + +```text +optimistic incremental state + + +authoritative final snapshot +``` + +--- + +## 7. Reducer 必须是纯函数：当前代码中的真实反例 + +React reducer 的核心契约： + +```text +same state + same action + ↓ +same next state +``` + +也就是： + +```ts +next = reducer(state, action) +``` + +不能偷偷依赖外界时间、随机数、网络、localStorage。 + +当前代码有两个值得修正的反例： + +```ts +createdAt: Date.now() +``` + +分别出现在： + +- `assistant.delta` 找不到目标 message 的 fallback； +- `assistant.completed` 找不到旧 message 的 fallback。 + +这意味着同一个 event 重放两次，结果可能不同。 + +### 正确做法 + +时间应该在 reducer 外生成： + +```ts +toChatStateEvent(event, receivedAt) +``` + +然后 event 明确携带： + +```ts +{ + type: 'assistant.delta', + id, + delta, + receivedAt, +} +``` + +Reducer 只消费输入。 + +### 为什么这不仅是“代码洁癖” + +纯 reducer 带来： + +- 可重复测试； +- event replay； +- time-travel debugging； +- 并发 React 下更稳； +- 更容易迁移到 Zustand reducer / Redux / XState。 + +--- + +## 8. Controller：副作用编排层 + +`useChatController()` 才是副作用应该存在的地方。 + +它负责： + +```text +时间 +localStorage +RPC +async iterator +generation guard +dispatch +``` + +而 reducer 负责： + +```text +state + event → next state +``` + +这是经典分层： + +```text +Controller = orchestration / side effects +Reducer = deterministic state transition +UI = render(state) +``` + +### 8.1 为什么需要 `stateRef` + +`sendMessage` 被： + +```ts +useCallback(..., []) +``` + +固定住，因此闭包中的 `state` 会变旧。 + +代码使用： + +```ts +const stateRef = useRef(state) + +useEffect(() => { + stateRef.current = state +}, [state]) +``` + +然后： + +```ts +stateRef.current.threadId +stateRef.current.status +``` + +这样长期稳定的 callback 可以读取最新状态。 + +需要注意：这是一种显式 escape hatch。若 callback 依赖越来越复杂，应重新评估是否值得固定为空依赖。 + +--- + +## 9. generation guard：解决 stale stream，但不是 Abort + +当前机制： + +```ts +const generationRef = useRef(0) +``` + +开始请求时： + +```ts +const generation = generationRef.current +``` + +消费事件时： + +```ts +if (generation !== generationRef.current) return +``` + +点击 New Chat： + +```ts +generationRef.current += 1 +``` + +### 9.1 它解决什么 + +假设旧会话 A 仍在返回 delta： + +```text +A stream + ↓ +New Chat + ↓ generation 0 → 1 +A next delta + ↓ +0 !== 1 + ↓ +ignore +``` + +因此旧流不会污染新会话 B。 + +这是 **stale-result invalidation**。 + +### 9.2 它没有解决什么 + +底层 Agent 可能仍然： + +```text +继续推理 +继续产生 token +继续执行工具 +继续占用 app-server process +``` + +所以： + +```text +generation guard ≠ cancellation +``` + +真正取消需要把中断一路传播到 Runtime，例如： + +```text +UI Stop/New Chat + ↓ +AbortController / interrupt API + ↓ +TanStack server request + ↓ +Codex Runtime + ↓ +turn/interrupt +``` + +两者职责不同： + +| 机制 | 解决的问题 | +|---|---| +| generation guard | 旧结果不能修改当前 UI | +| Abort / interrupt | 停止底层工作 | + +健壮系统通常两者都需要。 + +--- + +## 10. Persistence：为什么只保存 `threadId + messages` + +当前： + +```ts +interface PersistedConversation { + threadId: string | null + messages: ChatMessage[] +} +``` + +没有保存： + +```text +status +activities +error +``` + +这是合理的边界。 + +这些字段可以分为： + +### Durable state + +```text +threadId +messages +``` + +刷新后仍然有意义。 + +### Ephemeral state + +```text +running/error/activity +``` + +它们描述的是“当前页面这一刻发生什么”，刷新后不能安全地假装还能继续。 + +### 10.1 Versioned storage + +当前 key： + +```ts +codex-tanstack-demo:conversation:v1 +``` + +并且 envelope 有： + +```ts +{ version, conversation } +``` + +这是正确做法，因为本地持久化格式也是一种 schema。 + +未来 message shape 改变时，可以： + +```text +v1 → migrate → v2 +``` + +或者明确丢弃旧数据。 + +### 10.2 为什么不能直接 `JSON.parse()` 后信任 + +代码会逐字段验证： + +```text +threadId +messages array +message.id +message.role +message.content +message.createdAt +``` + +localStorage 是不可信输入： + +- 老版本残留； +- 用户 DevTools 手工修改； +- 浏览器插件修改； +- 半写入 / 非预期数据。 + +因此 parse 与 validation 必须在边界完成。 + +--- + +## 11. New Chat 与 Resume 的真正语义 + +### Resume + +发送新消息时 controller 带上： + +```ts +threadId: stateRef.current.threadId ?? undefined +``` + +服务端 Runtime 决定： + +```text +有 threadId → resume +无 threadId → start +``` + +所以浏览器并不保存完整 Agent 上下文，只保存一个 opaque thread identifier。 + +### New Chat + +当前 New Chat： + +```ts +dispatch({ type: 'new-chat' }) +clearConversation(storage) +``` + +它表示： + +```text +清空浏览器当前 conversation pointer +``` + +它并不表示： + +```text +删除 Codex 后端持久化 thread +``` + +这是很重要的语义区别： + +```text +new local conversation ≠ delete remote/runtime history +``` + +未来如果提供“历史会话管理”，应该把： + +```text +New Chat +Delete Thread +Archive Thread +Resume Thread +``` + +拆成不同动作。 + +--- + +## 12. UI 流式渲染：每个 delta 都会触发什么 + +当前每个 delta： + +```text +for await event + ↓ +dispatch + ↓ +reducer 创建新 messages array + ↓ +React render + ↓ +ChatMessage 更新 +``` + +同时 `ChatPage` 有： + +```ts +useEffect(() => { + scrollAnchorRef.current?.scrollIntoView({ block: 'end' }) +}, [messages, activities, status]) +``` + +所以 messages 每变化一次，就会尝试滚到底部。 + +### 12.1 优点 + +简单、确定、V0 足够。 + +### 12.2 高速 token stream 下的潜在问题 + +如果 delta 非常密集： + +```text +每 10ms 一个 delta +``` + +则可能产生大量： + +```text +render +DOM update +scrollIntoView +``` + +成熟方案可考虑： + +- transport 层适度 batching； +- `requestAnimationFrame` 合并 UI append； +- 自动滚动只在用户本来位于底部附近时执行； +- 用户手工向上滚动时暂停 auto-follow。 + +不要一开始就优化，但必须理解成本来自哪里。 + +--- + +## 13. 当前其实还没有 Markdown Streaming + +`ChatMessage` 当前是： + +```tsx +
{message.content}
+``` + +也就是纯文本渲染。 + +这点非常重要： + +```text +“文本是流式的” +并不等于 +“Markdown renderer 能安全流式增量解析” +``` + +如果以后接 Markdown，需要处理未闭合语法： + +```md +```ts +const a = +``` + +在中间 delta 阶段可能只有： + +```md +```ts +const +``` + +因此 Markdown streaming 需要考虑： + +- parser 对 incomplete syntax 的容忍； +- code block 重解析成本； +- DOM 抖动； +- link / HTML 安全； +- syntax highlighting 是否每个 token 重跑。 + +实践上更合理： + +```text +streaming phase：容忍增量 Markdown +completed phase：最终完整重渲染/校准 +``` + +而不是自己做“打字机动画”。 + +--- + +## 14. Error 状态与恢复 + +Reducer： + +```ts +case 'error': + return { + ...state, + status: 'error', + error: event.message, + } +``` + +下一次发送时： + +```ts +dispatch({ type: 'user.message.added', ... }) +dispatch({ type: 'turn.started' }) +``` + +`turn.started` 会： + +```text +status = running +error = null +activities = [] +``` + +这意味着错误不是永久锁死状态，而是可恢复状态。 + +### 一个工程上需要继续思考的问题 + +如果 turn 失败时已经收到部分 assistant delta： + +```text +assistant: "已经生成一半..." +error +``` + +应该： + +- 保留 partial message？ +- 标记 incomplete？ +- 删除？ +- 提供 retry？ + +当前模型只记录全局 `error`，还没有 message-level failure metadata。 + +未来成熟聊天 UI 可以把 message 扩展为： + +```ts +status: 'streaming' | 'completed' | 'failed' +``` + +--- + +## 15. 事件驱动 UI vs “完成后 setState” + +### 完成后更新 + +```ts +const result = await askAgent(prompt) +setMessages([...messages, result]) +``` + +优点：简单。 + +缺点： + +- 没有流式体验； +- 无法展示 activity； +- 中断困难； +- 工具调用期间 UI 没状态； +- 很难表达 partial failure。 + +### 事件驱动 + +```ts +for await (const event of stream) { + dispatch(event) +} +``` + +优点： + +- streaming 是自然结果； +- 支持 tool/activity； +- 状态迁移可测试； +- 容易增加 interrupt / approval； +- Runtime 和 UI 解耦。 + +代价： + +- 必须认真设计 event schema； +- 必须处理乱序、重复、缺失、stale stream； +- 测试从“结果断言”升级成“事件序列断言”。 + +Agent UI 更接近事件系统，而不是传统 CRUD 表单。 + +--- + +## 16. SSE / WebSocket / Fetch Stream / TanStack Server Function 怎么选 + +| 方案 | 优点 | 缺点 | 更适合 | +|---|---|---|---| +| TanStack `createServerFn` + async iterable | 类型整合好、项目内调用自然、少样板代码 | 与框架耦合 | 单体 TanStack Start 应用 | +| SSE | 简单、浏览器原生、服务端单向推送语义清晰 | 客户端→服务端控制通常需另一路请求 | token/event 单向流 | +| `fetch()` + ReadableStream | 标准 Web API、控制力强 | framing / parse / type contract 自己维护 | 自定义流协议 | +| WebSocket | 双向、长连接、适合 interrupt/approval/live control | 生命周期、重连、状态同步更复杂 | 高频双向 Agent 会话 | + +### 当前项目为什么选择 Server Function 合理 + +当前目标是： + +```text +单人、本地、TanStack Start、一个 chat UI +``` + +因此不需要为了“未来可能双向”提前引入 WebSocket。 + +设计原则： + +> 用满足当前语义的最简单 transport，同时通过 `ChatEvent` contract 保留未来替换 transport 的能力。 + +--- + +## 17. 关键源码地图 + +```text +src/server-functions/chat.ts + createServerFn RPC 边界 + +src/server-functions/chat.runtime.server.ts + server-only runtime seam + +src/server-functions/chat-stream.ts + runtime event → ChatEvent streaming bridge + +src/server-functions/codex-event-normalizer.ts + Codex event → application transport event + +src/features/chat/chat.types.ts + ChatRequest / ChatEvent transport contract + +src/features/chat/chat-event.adapter.ts + transport event → state event + +src/features/chat/chat.reducer.ts + deterministic state machine + +src/features/chat/use-chat-controller.ts + RPC / storage / async stream / generation orchestration + +src/features/chat/chat.storage.ts + local durable-state schema + +src/features/chat/components/chat-page.tsx + controlled UI + auto scroll + +src/features/chat/components/chat-message.tsx + message rendering（当前纯文本） +``` + +推荐调试顺序也按这个链路反向定位： + +```text +UI 不更新 + ↓ +Reducer 是否收到 event？ + ↓ +Adapter 是否转换？ + ↓ +for await 是否收到 ChatEvent？ + ↓ +Server function 是否 yield？ + ↓ +Normalizer 是否产生事件？ + ↓ +Runtime 是否收到 app-server notification？ +``` + +不要一看到“前端没流式”就直接改 React。 + +--- + +## 18. 常见 Bug 模式 + +### Bug 1：服务端有 stream，UI 仍一次性出现 + +检查是否某层做了 buffer： + +```ts +const all = [] +for await (...) all.push(...) +return all +``` + +### Bug 2：delta 重复 + +常见原因：把 snapshot 当 delta： + +```text +old = "Hello" +new = "Hello world" + +错误：append new +→ "HelloHello world" +``` + +### Bug 3：切换会话后旧回复串进来 + +缺少 generation / request identity guard。 + +### Bug 4：Reducer 测试偶发不一致 + +Reducer 内使用： + +```text +Date.now() +Math.random() +localStorage +``` + +### Bug 5：刷新后出现 running 状态假象 + +错误持久化 ephemeral state。 + +### Bug 6：SSR 阶段访问 `window.localStorage` + +当前 `getBrowserStorage()` 用： + +```ts +if (typeof window === 'undefined') return null +``` + +就是为了防这个问题。 + +### Bug 7：用户往上翻历史时被强制拉到底部 + +无条件 `scrollIntoView()` 的典型 UX 副作用。 + +--- + +## 19. 测试应该按事件层分层 + +### 19.1 Transport normalizer + +验证： + +```text +runtime event sequence +→ expected ChatEvent sequence +``` + +### 19.2 Adapter + +验证： + +```text +ChatEvent + receivedAt +→ deterministic ChatStateEvent +``` + +重点：时间是否只在 adapter 注入。 + +### 19.3 Reducer + +应该覆盖： + +```text +assistant.started +assistant.delta × N +assistant.completed +重复 started +未知 id delta +不同 message id +turn.completed +error → next turn recovery +new-chat +conversation restored +``` + +并增加纯函数测试： + +```ts +expect(reducer(state, event)).toEqual(reducer(state, event)) +``` + +前提是先移除 reducer 内部时钟。 + +### 19.4 Storage + +覆盖： + +```text +valid v1 +invalid JSON +wrong version +wrong message shape +storage exception +clear +SSR no-window +``` + +### 19.5 Controller + +当前 controller 测试非常轻，应进一步覆盖： + +```text +stream event dispatch order +new-chat invalidates stale stream +running 时拒绝重复 send +resume 使用最新 threadId +error path +``` + +### 19.6 UI + +建议使用 React Testing Library 验证行为，而不是 DOM implementation detail： + +```text +running → input disabled +message delta → visible text grows +error → alert visible +new chat → history cleared +auto-scroll policy +``` + +--- + +## 20. 一条完整时序图 + +```mermaid +sequenceDiagram + participant U as User + participant C as useChatController + participant S as TanStack ServerFn + participant B as Streaming Bridge + participant R as Codex Runtime + participant A as Adapter + participant D as Reducer + participant UI as React UI + + U->>C: sendMessage("解释项目") + C->>D: user.message.added + C->>D: turn.started + D-->>UI: render running + + C->>S: streamChat({message, threadId}) + S->>B: streamNormalizedChatEvents() + B->>R: streamTurn() + + R-->>B: agent message started + B-->>S: assistant.started + S-->>C: assistant.started + C->>A: toChatStateEvent + A->>D: assistant.started + D-->>UI: empty assistant shell + + loop each delta + R-->>B: agentMessage/delta + B-->>S: assistant.delta + S-->>C: assistant.delta + C->>A: adapt + A->>D: assistant.delta + D-->>UI: append + rerender + end + + R-->>B: item completed snapshot + B-->>S: assistant.completed + S-->>C: assistant.completed + C->>D: final calibration + + R-->>B: turn completed + B-->>S: turn.completed + S-->>C: turn.completed + C->>D: status = idle + D-->>UI: enable input +``` + +--- + +## 21. 可复用设计模式 + +### Pattern A：Application-owned Event Contract + +不要让第三方 SDK 类型进入 UI。 + +```text +Provider Event +→ Adapter +→ App Event +``` + +### Pattern B：Streaming Delta + Final Snapshot + +```text +delta = responsiveness +snapshot = correctness +``` + +### Pattern C：Controller / Reducer / View 分离 + +```text +Controller: side effects +Reducer: deterministic transitions +View: render only +``` + +### Pattern D：Durable / Ephemeral State 分离 + +只持久化刷新后仍然有真实语义的状态。 + +### Pattern E：Stale Result Guard + Real Cancellation + +两者不是替代关系： + +```text +guard 保 UI +interrupt 保资源 +``` + +### Pattern F：Boundary Validation + +所有跨边界数据都应该验证： + +```text +RPC input +localStorage +provider protocol +``` + +--- + +## 22. 当前实现值得继续改进的地方 + +按优先级： + +1. **移除 reducer 内 `Date.now()`**，把时间全部移到 adapter/controller； +2. **增加真正的 abort / `turn/interrupt`**，generation guard 继续保留； +3. **完善 controller integration tests**； +4. **自动滚动改为“仅用户位于底部附近时 follow”**； +5. **引入 Markdown 前先设计 incomplete Markdown streaming 策略**； +6. **消息增加 streaming/completed/failed 状态**，更准确表达 partial failure； +7. 若 delta 频率过高，再引入 rAF batching，而不是预优化。 + +--- + +## 23. Review Checklist + +以后审核类似 Agent Chat UI，可以逐项检查： + +- [ ] 浏览器是否只依赖 application-owned contract？ +- [ ] Node/runtime-only 依赖是否明确隔离？ +- [ ] 流是否在中间某层被 buffer？ +- [ ] delta 与 snapshot 是否语义明确？ +- [ ] completed 是否可做最终校准？ +- [ ] reducer 是否完全纯函数？ +- [ ] controller 是否集中管理副作用？ +- [ ] stale stream 是否有 request/generation identity？ +- [ ] stale guard 之外是否有真实 cancellation？ +- [ ] durable 与 ephemeral state 是否分开？ +- [ ] persisted schema 是否 versioned + validated？ +- [ ] SSR 是否安全访问 browser API？ +- [ ] 自动滚动是否尊重用户手动阅读历史？ +- [ ] 流式 Markdown 是否能处理不完整语法？ +- [ ] 测试是否覆盖事件序列而不只是最终结果？ + +--- + +## 24. 复习题 + +### 基础 + +1. 为什么 `createServerFn` 模块可以被浏览器 import，而 Codex runtime 仍必须放在 server-only 模块？ +2. async generator 为什么适合表达 LLM / Agent 输出？ +3. `ChatEvent` 和 `ChatStateEvent` 分别属于哪一层？ +4. 为什么 `assistant.completed` 仍应携带完整文本，而不是只发一个 `done: true`？ +5. 为什么只持久化 `threadId + messages`？ + +### 进阶 + +6. generation guard 和 AbortController 的问题域有什么区别？ +7. 如果 provider 提供的是累计 snapshot，而 UI contract 要的是 delta，应在哪一层转换？为什么？ +8. reducer 内调用 `Date.now()` 会破坏哪些工程能力？ +9. 如果每 5ms 收到一个 delta，React UI 可能出现哪些性能问题？你会先优化哪里？ +10. 为什么未闭合 Markdown 会让 streaming renderer 比纯文本复杂得多？ + +### 架构思考 + +11. 如果以后底层从 Codex 改成 Pi Agent，哪些层应该保持不变？ +12. 如果产品需要实时 approval、interrupt、steer，多路双向消息增加后，继续使用 Server Function streaming 还是切 WebSocket？判断依据是什么？ +13. 如果刷新页面时一个 turn 尚未结束，应用应该恢复成 `running` 吗？为什么？ +14. 如果收到 `assistant.delta` 但从未收到 `assistant.started`，系统应丢弃、补建还是报错？不同策略分别有什么 trade-off？ +15. 如何设计测试证明“流式 UI 真的是逐事件更新”，而不是最后一次性渲染？ + +--- + +## 25. 最终心智模型 + +不要把 Agent Chat 理解成： + +```text +prompt → response +``` + +更准确的是： + +```text +Provider Runtime + ↓ produces domain events +Server Boundary + ↓ normalizes +Application Transport Events + ↓ streams +Client Adapter + ↓ enriches / narrows +State Events + ↓ folds +Deterministic State Machine + ↓ renders +UI +``` + +一旦建立这个模型，流式文本、工具活动、中断、审批、多 Agent、持久化都只是“新增事件和状态迁移”，而不是不断给一个巨型 `sendMessage()` 打补丁。 -- 2.54.0 From 8632e0702e3cab9eda806b3ab3341381864e5bb4 Mon Sep 17 00:00:00 2001 From: CoderLambert Date: Fri, 11 Sep 2026 20:16:51 +0800 Subject: [PATCH 2/3] docs: add system architecture notes --- docs/notes/01-system-architecture.md | 931 +++++++++++++++++++++++++++ 1 file changed, 931 insertions(+) create mode 100644 docs/notes/01-system-architecture.md diff --git a/docs/notes/01-system-architecture.md b/docs/notes/01-system-architecture.md new file mode 100644 index 0000000..e25c6bf --- /dev/null +++ b/docs/notes/01-system-architecture.md @@ -0,0 +1,931 @@ +# Codex + TanStack Start：系统架构学习笔记 + +> 目标：这不是项目进度记录，而是一篇可以脱离当前上下文独立复习的架构笔记。它回答三个问题：**系统边界在哪里、数据如何流动、为什么要这样分层。** + +## 1. 问题背景：我们真正要解决什么 + +这个项目不是在“做一个聊天框”，而是在验证一套本地 Agent Web 架构：浏览器负责交互，TanStack Start 负责 Web/RPC 边界，Codex app-server 负责 Agent runtime，本机已有的 Codex/ChatGPT 登录态负责认证。 + +核心约束有四个： + +1. 浏览器不能拿到 Codex 凭据、`~/.codex`、原始协议对象或本地敏感信息。 +2. Agent 回复必须是真正的流式文本，而不是请求完成后一次性返回。 +3. Web 层不能直接绑定 Codex 协议，否则以后换 Claude、Pi、Qwen 或其他 runtime 会牵动整个 UI。 +4. V0 必须保持只读：可以分析仓库，但不允许文件写入、网络访问或交互式审批绕过。 + +因此，架构目标不是“最少代码”，而是建立几个清晰的边界： + +```text +Browser/UI boundary +Transport/RPC boundary +Application event boundary +Agent runtime boundary +Local machine / workspace boundary +``` + +真正重要的是：每一层只理解自己需要理解的协议。 + +--- + +## 2. 总体架构 + +```mermaid +flowchart TD + U[User] --> UI[React Chat UI] + UI --> C[useChatController] + C --> SF[TanStack Start createServerFn] + SF --> BR[Streaming Bridge] + BR --> N[Codex Event Normalizer] + N --> RT[CodexRuntime interface] + RT --> AR[CodexAppServerRuntime] + AR --> AS[codex app-server] + AS --> FS[Workspace / Git Repository] + AS --> AUTH[Local Codex / ChatGPT Auth] + + N -->|ChatEvent| BR + BR -->|Async stream| SF + SF --> C + C --> AD[ChatEvent adapter] + AD --> R[chatReducer] + R --> UI +``` + +从浏览器看，它只知道： + +```text +ChatRequest -> stream +``` + +它不知道： + +```text +JSON-RPC +codex app-server +thread/start +turn/start +item/agentMessage/delta +~/.codex +子进程 +stdio +``` + +这就是架构中的第一原则：**把基础设施协议封装成应用协议。** + +--- + +## 3. 三层系统边界 + +### 3.1 Browser：只负责产品状态 + +浏览器职责： + +- 收集用户输入； +- 调用 `streamChat()`； +- 消费 `ChatEvent`； +- 维护消息、运行状态、活动状态； +- 保存最小会话信息； +- 渲染 UI。 + +浏览器不应该： + +- 调用 `codex app-server`； +- 读取 Codex 登录文件； +- 接触 MCP 参数/result； +- 接触 command stdout/stderr； +- 接触 reasoning 原文； +- 决定 sandbox 权限。 + +**前端状态不是 Agent runtime 状态。** 前端保存的是“产品需要展示的投影”。 + +### 3.2 TanStack Start Server：安全网关 + 协议翻译层 + +Server 层做三类工作： + +1. 输入验证； +2. runtime 调用； +3. 将 runtime 事件变成浏览器安全的应用事件。 + +`src/server-functions/chat.ts` 很关键，因为它是浏览器可导入模块，但真正的 Codex runtime 被隔离在 `*.server.ts` 后面。 + +```text +client import + | + v +createServerFn() + | + | server execution only + v +chat.runtime.server.ts + | + v +Codex runtime +``` + +这解决了一个典型全栈框架问题：**同一个 TypeScript 工程不等于所有模块都可以进入 browser bundle。** + +### 3.3 Codex Runtime：负责 Agent 生命周期 + +Runtime 层负责： + +- spawn `codex app-server --stdio`； +- initialize； +- start/resume thread； +- start turn； +- 接收 item notification； +- 接收 assistant delta； +- 收集 usage； +- 清理进程； +- 把协议对象转换为内部 `CodexThreadEvent`。 + +这里最重要的架构点不是 Codex，而是 `CodexRuntime` interface。 + +```ts +interface CodexRuntime { + streamTurn(input: StreamCodexTurnInput): AsyncGenerator +} +``` + +UI 不依赖 `CodexAppServerRuntime`，server function 也不应该依赖 JSON-RPC 细节。 + +--- + +## 4. thread / turn / item：必须建立的心智模型 + +这是理解 Codex Agent runtime 的基础。 + +### Thread + +`thread` 是长期会话上下文。 + +它类似： + +```text +Conversation / Agent Session +``` + +第一条消息： + +```text +thread/start +``` + +后续继续聊天： + +```text +thread/resume(threadId) +``` + +浏览器 localStorage 保存 `threadId` 的原因，就是要把产品侧会话重新连接到 Codex 侧的长期上下文。 + +### Turn + +`turn` 是 thread 中一次用户输入对应的一轮 Agent 工作。 + +```text +Thread +├── Turn 1 +├── Turn 2 +└── Turn 3 +``` + +一次 turn 可能包含： + +- reasoning； +- command execution； +- MCP tool call； +- web search； +- assistant message； +- token usage； +- error。 + +因此： + +> 一个 HTTP/RPC 请求不等于一个 assistant message，而更接近一个完整 turn。 + +### Item + +`item` 是 turn 内部的工作单元。 + +```text +Turn +├── reasoning item +├── command item +├── tool item +└── agentMessage item +``` + +Assistant 文本本身也是一个 item，它有生命周期： + +```text +item/started + ↓ +item/agentMessage/delta × N + ↓ +item/completed +``` + +这也是为什么 UI 应使用： + +```text +assistant.started +assistant.delta +assistant.completed +``` + +而不是只使用一个： + +```text +assistant.message +``` + +--- + +## 5. 一条消息的完整生命周期 + +```mermaid +sequenceDiagram + participant User + participant React + participant Start as TanStack Start + participant Runtime + participant Codex as codex app-server + + User->>React: 输入 prompt + React->>Start: streamChat({message, threadId}) + Start->>Runtime: streamTurn() + Runtime->>Codex: initialize + Codex-->>Runtime: initialize result + Runtime->>Codex: initialized + + alt 新会话 + Runtime->>Codex: thread/start + else 已有会话 + Runtime->>Codex: thread/resume(threadId) + end + + Codex-->>Runtime: thread id + Runtime-->>React: thread.started + + Runtime->>Codex: turn/start + Codex-->>Runtime: item/started(agentMessage) + Runtime-->>React: assistant.started + + loop 模型生成文本 + Codex-->>Runtime: item/agentMessage/delta + Runtime-->>React: assistant.delta + React->>React: reducer append delta + end + + Codex-->>Runtime: item/completed(agentMessage) + Runtime-->>React: assistant.completed(full snapshot) + Codex-->>Runtime: thread/tokenUsage/updated + Codex-->>Runtime: turn/completed + Runtime-->>React: turn.completed +``` + +这里有两个特别值得记住的设计： + +### Delta 用于体验 + +`assistant.delta` 提供实时输出。 + +### Completed snapshot 用于校准 + +最终 `item/completed` 里的完整文本不是多余的。 + +它可以校准： + +- delta 丢失； +- delta 重复； +- snapshot 修订； +- 中间状态异常。 + +设计原则： + +```text +delta = 实时体验 +completed snapshot = 最终事实 +``` + +--- + +## 6. 为什么需要 application-owned `ChatEvent` + +这是整个架构最值得迁移到其他项目的设计之一。 + +如果 UI 直接消费 Codex： + +```ts +if (event.method === 'item/agentMessage/delta') { ... } +``` + +那么 UI 已经被 Codex 协议绑死。 + +现在使用： + +```ts +ChatEvent = + | assistant.started + | assistant.delta + | assistant.completed + | activity.started + | activity.updated + | activity.completed + | turn.completed + | error +``` + +于是关系变成： + +```text +Codex protocol + ↓ adapter +Application protocol + ↓ +React UI +``` + +未来换 runtime： + +```text +Claude events ─┐ +Pi events ├─> ChatEvent ─> UI +Qwen events ─┘ +``` + +UI 不变。 + +### `ChatEvent` 不是简单 DTO + +它承担三个职责： + +1. **解耦**：隔离供应商协议； +2. **安全**：只允许白名单数据进入浏览器； +3. **产品语义**：把 Agent 底层事件转换成 UI 真正关心的生命周期。 + +这比“直接透传原始 event，然后前端自己判断”健壮得多。 + +--- + +## 7. 为什么还要有 `ChatStateEvent` + +项目里实际上有两个事件层： + +```text +ChatEvent -> transport contract +ChatStateEvent -> reducer contract +``` + +它们不是重复设计。 + +例如 server 只需要告诉浏览器： + +```text +assistant.started(id) +``` + +而 reducer 创建消息时需要： + +```text +createdAt +``` + +于是 adapter 可以在客户端补产品状态需要的信息： + +```text +ChatEvent + ↓ toChatStateEvent(receivedAt) +ChatStateEvent + ↓ +Reducer +``` + +这遵循一个很重要的原则： + +> Transport model、domain model、view state model 不应因为字段看起来相似就强行合并。 + +--- + +## 8. Server-only 边界为什么重要 + +全栈 TypeScript 很容易制造一种错觉： + +> “既然都是 TS 文件，直接 import 不就行了吗？” + +问题是 browser bundle 一旦导入 runtime 模块，就可能把： + +- Node API； +- 本地路径； +- 子进程逻辑； +- 协议类型； +- 甚至认证相关实现 + +带进浏览器构建图。 + +当前项目的边界是： + +```text +chat.ts browser-importable RPC declaration +chat.runtime.server.ts server-only seam +src/server/codex/** server-only runtime +``` + +`createServerFn` 的价值之一，就是让客户端引用一个“函数形状”，实际执行发生在服务器。 + +--- + +## 9. 持久化边界：谁保存什么 + +当前有两套状态所有者。 + +### Browser 持久化 + +只保存： + +```text +threadId +messages[] +``` + +这解决的是产品体验：刷新页面后仍然能看到聊天记录并继续原线程。 + +### Codex 持久化 + +Codex 自己维护 thread/session 数据，例如本地 session。 + +这解决的是 Agent 上下文。 + +两者不能混为一谈： + +```text +Browser transcript ≠ Codex thread state +``` + +浏览器消息只是 UI 投影，不应该被当成 Agent 的唯一真实上下文。 + +### New Chat 的语义 + +当前 New Chat 做的是： + +```text +清浏览器 active conversation +``` + +不是： + +```text +删除 Codex 历史 session +``` + +这是正确的职责分离。 + +--- + +## 10. V0 read-only 安全模型 + +当前策略： + +```text +model: luna +effort: high +sandbox: read-only +sandboxPolicy.networkAccess: false +approvalPolicy: never +``` + +安全思路不是依赖一个开关，而是多层防线。 + +### 第一层：权限限制 + +```text +read-only sandbox +network disabled +``` + +### 第二层：无交互审批 + +```text +approvalPolicy: never +``` + +避免 runtime 临时要求更高权限后由 Web UI 放行。 + +### 第三层：数据最小化 + +浏览器不接收： + +```text +reasoning raw text +command stdout/stderr +MCP arguments/results +auth data +raw runtime event +``` + +### 第四层：workspace root + +工作目录需要 canonicalize，并限制在允许 root 下。 + +安全原则： + +> Agent 的权限控制和 UI 的数据脱敏是两个不同问题，二者都必须做。 + +只读 sandbox 防止 Agent 修改系统；事件白名单防止敏感信息泄漏到 Browser。 + +--- + +## 11. 为什么不直接把 Codex 协议给前端 + +看起来直接转发 JSON-RPC 最省代码： + +```text +Codex -> WebSocket -> Browser +``` + +但会带来几个长期问题。 + +| 方案 | 优点 | 代价 | +|---|---|---| +| 原始协议直传 | 开发快、信息完整 | 前端强耦合 Codex、敏感字段难控制、协议升级影响 UI | +| Server normalize | 安全边界清楚、UI 稳定、可换 runtime | server adapter 代码更多 | +| Browser 自己 adapter | server 简单 | 安全与兼容逻辑散落前端，不推荐 | + +对于 Agent 产品，推荐： + +```text +Raw runtime protocol + ↓ +Server-side normalization + ↓ +Stable app protocol +``` + +因为 runtime event 通常比 UI 所需的数据丰富得多。 + +--- + +## 12. 当前关键源码映射 + +| 关注点 | 文件 | 职责 | +|---|---|---| +| Web RPC | `src/server-functions/chat.ts` | 输入校验、streaming server function、错误收敛 | +| Runtime seam | `src/server-functions/chat.runtime.server.ts` | Web 层进入 Codex 层的唯一桥梁 | +| Streaming bridge | `src/server-functions/chat-stream.ts` | 消费 runtime event 并输出 ChatEvent | +| Event normalization | `src/server-functions/codex-event-normalizer.ts` | Codex event -> application event | +| Runtime contract | `src/server/codex/codex-runtime.ts` | 定义可替换 runtime interface | +| App-server client | `src/server/codex/codex-app-server.server.ts` | 子进程、JSON-RPC、thread/turn/item 生命周期 | +| Workspace policy | `src/server/codex/workspace.server.ts` | 工作目录约束 | +| App transport types | `src/features/chat/chat.types.ts` | Browser-safe ChatEvent contract | +| State adapter | `src/features/chat/chat-event.adapter.ts` | transport event -> reducer event | +| Reducer | `src/features/chat/chat.reducer.ts` | 确定性状态变化 | +| Controller | `src/features/chat/use-chat-controller.ts` | 调 RPC、消费 stream、generation guard、持久化 | +| Storage | `src/features/chat/chat.storage.ts` | localStorage versioned persistence | +| UI | `src/features/chat/components/**` | 纯展示与交互 | + +复习源码时推荐按这个顺序： + +```text +chat.ts +-> chat-stream.ts +-> codex-event-normalizer.ts +-> codex-runtime.ts +-> codex-app-server.server.ts +-> chat-event.adapter.ts +-> chat.reducer.ts +-> use-chat-controller.ts +``` + +这样是在顺着数据流读，而不是按目录读。 + +--- + +## 13. 当前架构里的几个关键 trade-off + +### process-per-turn vs long-lived app-server + +当前 `streamTurn()` 每次创建一个 app-server connection。 + +优点： + +- 生命周期简单； +- 故障隔离强； +- turn 结束即可清理； +- V0 易调试。 + +缺点： + +- 重复 initialize； +- 多 turn 成本更高； +- interrupt/steer/多并发管理不自然； +- 不适合未来复杂 Agent desktop runtime。 + +成熟版本通常更适合： + +```text +App lifecycle + ↓ +long-lived AppServerClient + ├── Thread A / Turn 1 + ├── Thread A / Turn 2 + └── Thread B / Turn 1 +``` + +### async generator RPC vs SSE/WebSocket + +当前使用 TanStack Start async generator。 + +| 技术 | 适合场景 | +|---|---| +| async generator RPC | 请求-流式响应、类型整合好、当前 Demo 简洁 | +| SSE | 单向服务器推送，协议简单，浏览器原生支持 | +| WebSocket | 双向长期会话、interrupt/steer/approval/实时协作 | +| raw fetch stream | 控制力高，但协议、解析、类型都需要自己维护 | + +当前需求主要是： + +```text +user request -> server stream response +``` + +因此 async generator 很合理。 + +当未来加入： + +```text +steer +approval +interrupt +multi-agent events +``` + +WebSocket 或独立 runtime transport 的价值会提高。 + +--- + +## 14. 典型失败场景 + +### 14.1 Codex 模型不可用 + +表现： + +```text +turn 很快失败 +没有 agentMessage delta +``` + +调试顺序： + +1. 先直接测试本机 Codex； +2. 验证账号模型权限； +3. 记录 app-server event type，而不是直接猜 UI； +4. 检查 `turn/completed` status/error。 + +### 14.2 浏览器不是流式输出 + +分层排查： + +```text +Codex 是否产生 delta？ + ↓ yes +Normalizer 是否产生 assistant.delta？ + ↓ yes +TanStack stream 是否逐事件到达？ + ↓ yes +Reducer 是否 append？ + ↓ yes +UI 是否被 memo/render 阻断？ +``` + +不要一开始就在 React 层加“打字机动画”。伪流式会掩盖真正的数据链路问题。 + +### 14.3 New Chat 后旧回复出现 + +这是 stale stream 问题。 + +当前客户端用 generation guard： + +```text +stream generation != current generation +=> ignore +``` + +它解决 UI 污染，但不等于 runtime 已取消。 + +### 14.4 turn 事件串线 + +未来存在并发 turn 时，不能只看： + +```text +message.method === 'turn/completed' +``` + +应该同时关联： + +```text +threadId + turnId +``` + +否则别的 turn completed 可能误结束当前流。 + +### 14.5 子进程退出异常 + +需要区分： + +```text +JSON-RPC error +protocol parse error +app-server process exit +stderr diagnostics +user abort +turn failure +``` + +如果全部变成一个 `Error('failed')`，系统后续会很难观测。 + +--- + +## 15. 调试方法：按边界观察，而不是全链路乱打日志 + +推荐在开发期临时记录结构化信息： + +```text +method +threadId +turnId +item.type +item.id +delta.length +status +``` + +不要记录： + +```text +完整 reasoning +command output +MCP payload +auth token +敏感文件内容 +``` + +### 一条标准调试链 + +```text +1. 本机 codex CLI 是否正常 +2. app-server initialize 是否成功 +3. thread/start 或 resume 是否成功 +4. turn/start 是否返回 +5. item/started 是否出现 +6. agentMessage delta 是否持续出现 +7. item/completed 是否包含最终 snapshot +8. turn/completed 是否正确 +9. ChatEvent 是否正确映射 +10. reducer 是否按 id 更新 +``` + +这是比“浏览器没显示，先看 React”更专业的定位方式。 + +--- + +## 16. 架构演进路线 + +### Stage 1：当前 V0 + +```text +read-only +single workspace +single active browser conversation +process-per-turn +streaming assistant +``` + +### Stage 2：可靠 runtime client + +增加： + +- `turnId` correlation； +- `turn/interrupt`； +- 强制进程退出兜底； +- fake app-server integration tests； +- 结构化错误类型。 + +### Stage 3：长期 app-server + +```text +one runtime process +multiple threads +multiple turns +``` + +增加： + +- turn registry； +- notification routing； +- concurrency control； +- reconnect/recovery。 + +### Stage 4：可写 Agent + +必须新增： + +- approval UI； +- permission model； +- diff preview； +- write sandbox； +- destructive action confirmation。 + +不能简单把： + +```text +read-only -> workspace-write +``` + +当成一个配置切换。 + +### Stage 5：多 Runtime + +保持： + +```text +CodexAdapter ─┐ +ClaudeAdapter ├─> ChatEvent +PiAdapter ─┘ +``` + +这时今天设计的 application-owned contract 才真正体现价值。 + +--- + +## 17. 可迁移的架构原则 + +这套项目最值得记住的不是某个 Codex API，而是下面这些模式： + +1. **Agent runtime 必须放在可信 server boundary 后面。** +2. **供应商协议和产品协议要分开。** +3. **流式 UI 应基于真实 delta，而不是字符串动画。** +4. **delta 是实时状态，completed snapshot 是最终事实。** +5. **Thread、Turn、Item 必须分层理解。** +6. **Transport state 与 UI state 不应该强行共用同一模型。** +7. **权限隔离和数据脱敏是两条不同安全防线。** +8. **持久化必须明确数据所有者：Browser transcript 和 Agent thread 不是同一份状态。** +9. **并发系统必须使用稳定 correlation id，而不是靠事件顺序猜归属。** +10. **调试应该沿系统边界逐层验证。** + +--- + +## 18. 复习检查表 + +如果可以不看代码回答下面问题，说明架构已经基本掌握： + +- [ ] 为什么 `codex app-server` 必须运行在 server-side？ +- [ ] `thread`、`turn`、`item` 分别表示什么？ +- [ ] 为什么一个 turn 不等于一个 assistant message？ +- [ ] `item/agentMessage/delta` 在系统里经过了哪些层？ +- [ ] 为什么 `ChatEvent` 不直接复用 Codex 协议类型？ +- [ ] `ChatEvent` 和 `ChatStateEvent` 为什么要分开？ +- [ ] `assistant.completed` 已经有最终文本，为什么还需要 delta？ +- [ ] 有 delta 以后为什么仍需要 completed snapshot？ +- [ ] localStorage 和 Codex session 分别保存什么？ +- [ ] New Chat 为什么不应该等同于删除 Codex session？ +- [ ] read-only sandbox 与浏览器数据脱敏分别解决什么问题？ +- [ ] generation guard 能解决什么，不能解决什么？ +- [ ] 为什么并发 turn 必须按 `turnId` correlation？ +- [ ] process-per-turn 的优缺点是什么？ +- [ ] 什么时候应该考虑 SSE 或 WebSocket？ +- [ ] 如果未来替换成 Claude/Pi，哪些层应该变化、哪些层应该保持不变？ + +--- + +## 19. 思考题 + +1. 如果用户同时打开两个浏览器 Tab，两个 Tab resume 同一个 thread，会有哪些竞态？应该在哪一层解决？ +2. 如果 `assistant.delta` 已经追加了 200 字，但最终 `assistant.completed.text` 只有 180 字，Reducer 应该怎么处理？为什么？ +3. 如果未来允许 Agent 写文件，仅增加 `sandbox: workspace-write` 为什么不够？至少还需要哪些产品能力？ +4. 如果 app-server 变成长驻进程，一个 JSON-RPC reader 如何同时服务多个 turn？你需要哪些 registry/correlation 数据结构？ +5. 如果换成一个只提供 SSE 的 Agent provider，现有 `ChatEvent` 层还能否保留？哪些 adapter 需要变化？ +6. 为什么“浏览器永远不接触 raw runtime event”不仅是解耦设计，也是安全设计？ +7. 哪些错误应该展示给用户，哪些错误只应该进入 server log？如何给它们建立稳定 error code？ + +--- + +## 20. 最终心智模型 + +可以把整个系统压缩成一句话： + +> **浏览器维护产品状态，TanStack Start 建立可信 RPC 边界，Application Event Contract 隔离产品与供应商协议，CodexRuntime 管理 Agent 生命周期，Codex app-server 负责真正的 thread/turn/item 执行。** + +再进一步抽象： + +```text +External Agent Runtime + ↓ +Trusted Adapter + ↓ +Stable Application Events + ↓ +Deterministic Client State + ↓ +UI +``` + +这才是这个 Demo 最有价值的架构成果。 \ No newline at end of file -- 2.54.0 From b9b4eb9d4a156b6b4c307ef11b131444cf158928 Mon Sep 17 00:00:00 2001 From: CoderLambert Date: Fri, 11 Sep 2026 20:17:26 +0800 Subject: [PATCH 3/3] docs: add Codex app-server streaming notes --- docs/notes/02-codex-app-server-streaming.md | 1905 +++++++++++++++++++ 1 file changed, 1905 insertions(+) create mode 100644 docs/notes/02-codex-app-server-streaming.md diff --git a/docs/notes/02-codex-app-server-streaming.md b/docs/notes/02-codex-app-server-streaming.md new file mode 100644 index 0000000..822193f --- /dev/null +++ b/docs/notes/02-codex-app-server-streaming.md @@ -0,0 +1,1905 @@ +# Codex App Server 真流式：从协议、运行时到前端事件模型 + +> 适用项目：`CoderLambert/codex-tanstack-start` +> +> 审查基线：`main@6f4584d2f88c93db8b5059ef6e08a1efa3578a19` +> +> 本文不是项目进度记录，而是一篇围绕 **Codex App Server + Agent 流式运行时** 的可迁移技术笔记。阅读目标是：即使离开本仓库，也能独立设计一个可靠的本地 Agent Runtime Adapter。 + +--- + +## 1. 先建立正确问题：我们到底要“流式”什么？ + +在 Agent 应用里，“流式”至少有三种含义： + +1. **请求仍在进行**：客户端知道任务没结束。 +2. **Agent 生命周期事件在流动**：例如 reasoning、command started、tool completed、file change。 +3. **Assistant 正文按增量持续输出**：用户看到回答逐步增长，而不是最后一次性出现。 + +本项目最初使用 `@openai/codex-sdk` 的 `runStreamed()`。它能够返回 `thread.started`、`item.started`、`item.updated`、`item.completed`、`turn.completed` 等结构化事件，因此属于第 2 类流式；但在实际 `luna` 验证中，没有获得可用的 `agent_message.item.updated` 文本增量，而且当前账号对该模型配置还直接返回了 item error。 + +真正满足聊天 UI 需求的是第 3 类：**正文 delta streaming**。 + +Codex App Server 官方协议提供： + +```text +item/agentMessage/delta +``` + +因此项目从： + +```text +@openai/codex-sdk + -> runStreamed() + -> ThreadEvent +``` + +迁移为： + +```text +codex app-server + -> JSON messages over stdio + -> item/agentMessage/delta + -> application-owned ChatEvent +``` + +这不是为了“换一个 API”，而是为了换到更底层、信息更完整的 Agent 协议层。 + +### 核心原则 + +> **事件流（event streaming）不等于文本增量流（text delta streaming）。** + +判断一个 Agent SDK 是否能实现 ChatGPT 式输出，不能只看 API 名字有没有 `stream`，必须确认协议中是否存在可消费的正文 delta。 + +--- + +## 2. 为什么 App Server 更适合作为 Agent Runtime + +Codex App Server 是一个本地长运行/可长运行的 Agent 协议服务。宿主程序启动 `codex app-server`，通过 stdio 与它交换结构化消息。 + +本项目当前的边界： + +```text +Browser + │ + │ application ChatEvent + ▼ +TanStack Start Server Function + │ + ▼ +CodexRuntime interface + │ + ▼ +CodexAppServerRuntime + │ + │ protocol messages + ▼ +codex app-server + │ + ▼ +local Codex / ChatGPT authentication + agent loop +``` + +与直接在前端消费 App Server 协议相比，这个额外的 Runtime/Normalizer 层非常重要： + +- 浏览器不知道 Codex 原始协议。 +- 浏览器拿不到本地认证材料。 +- UI 不依赖某个模型供应商的事件结构。 +- command stdout/stderr、reasoning、MCP arguments/results 可以在 server boundary 被过滤。 +- 将来可以增加 Claude、Pi、Qwen Runtime，而不用重写 Chat UI。 + +这属于典型的 **Anti-Corruption Layer / Adapter Boundary**：把第三方协议转换成自己的稳定领域协议。 + +--- + +## 3. JSON-RPC 思维模型：Request、Response、Notification、Server Request + +App Server 使用结构化消息通过 stdio 交换。工程上可以把它理解成 JSON-RPC 风格的双向消息协议。 + +### 3.1 Request + +客户端发出带 `id` 的请求： + +```json +{ + "id": 3, + "method": "turn/start", + "params": { + "threadId": "thr_123", + "input": [{ "type": "text", "text": "解释这段代码" }] + } +} +``` + +### 3.2 Response + +服务端用相同 `id` 返回结果： + +```json +{ + "id": 3, + "result": { + "turn": { + "id": "turn_456", + "status": "inProgress", + "items": [] + } + } +} +``` + +因此客户端必须维护： + +```text +request id -> pending request +``` + +不能简单假设下一条 stdout 消息就是当前请求的响应。 + +### 3.3 Notification + +没有请求 `id`，表示服务端主动广播状态变化： + +```json +{ + "method": "item/agentMessage/delta", + "params": { + "threadId": "thr_123", + "turnId": "turn_456", + "itemId": "msg_1", + "delta": "React " + } +} +``` + +一个请求等待 response 期间，notification 完全可能先到。 + +本项目因此在 `AppServerConnection.request()` 中缓存 notification： + +```ts +const notifications: AppServerMessage[] = [] + +while (true) { + const message = await this.nextMessage() + + if (message.id === id) { + return { response: message, notifications } + } + + if (message.method) { + notifications.push(message) + } +} +``` + +这个设计比“发请求 -> 读下一行”正确得多。 + +### 3.4 Server Request + +App Server 还可能主动向宿主发送带 `id + method` 的请求，例如 approval / elicitation / attestation 类交互。 + +因此双向协议客户端至少要识别四类消息： + +| 类型 | `id` | `method` | 宿主行为 | +|---|---:|---|---| +| Request | 有 | 有 | 发往 server | +| Response | 有 | 无 | 匹配 pending request | +| Notification | 无 | 有 | 分发为事件 | +| Server Request | 有 | 有 | 必须响应/拒绝 | + +当前项目为了 V0 read-only + no approval，统一拒绝 server request： + +```ts +error: { + code: -32000, + message: 'Interactive server requests are disabled.' +} +``` + +这在当前安全模型下合理，但未来启用写操作、审批或 MCP elicitation 时必须升级为真正的 request handler。 + +--- + +## 4. 连接初始化：为什么 initialize / initialized 是协议握手 + +官方要求每个连接先执行： + +```text +initialize request + ↓ +initialize response + ↓ +initialized notification +``` + +本项目： + +```ts +await connection.request('initialize', { + clientInfo: { + name: 'codex-tanstack-demo', + title: 'Codex TanStack Start Demo', + version: '0.1.0', + }, + capabilities: { + experimentalApi: true, + requestAttestation: false, + }, +}) + +connection.notify('initialized') +``` + +这不是可选的礼貌流程，而是连接状态机的一部分。未初始化连接上的后续请求会被拒绝。 + +### 能力协商的意义 + +`capabilities` 不是普通配置，而是“这个客户端会不会理解某些协议行为”的声明。 + +例如： + +- `experimentalApi` +- notification opt-out +- attestation +- MCP form elicitation + +因此长期实现不要在协议客户端里把 capability 当作随手拼的 JSON；它应该属于 Connection Configuration。 + +官方参考： + +- https://developers.openai.com/codex/app-server/ + +--- + +## 5. Thread / Turn / Item：理解 Codex 的三层领域模型 + +这是整套 App Server 最重要的心智模型。 + +```text +Thread +├── Turn 1 +│ ├── userMessage +│ ├── reasoning +│ ├── commandExecution +│ ├── fileChange +│ └── agentMessage +│ +├── Turn 2 +│ ├── userMessage +│ └── agentMessage +│ +└── Turn 3 ... +``` + +### Thread + +一个持久对话上下文。 + +- 新会话：`thread/start` +- 继续会话：`thread/resume` +- 后续还可以 list/read/fork/archive + +### Turn + +一次“用户请求 + Agent 完成这次工作的全过程”。 + +调用： + +```text +turn/start +``` + +开始一个 turn。 + +Turn 不是一条 Assistant 消息。它可能包含： + +- 多个 reasoning item +- 多个 command execution +- MCP tool calls +- file changes +- 一个或多个 agent message 阶段 + +### Item + +Turn 中的工作单元。 + +常见： + +```text +userMessage +agentMessage +reasoning +commandExecution +fileChange +mcpToolCall +webSearch +``` + +### 为什么不能把 Thread 等同于前端 Message[]？ + +因为： + +```text +前端 transcript +``` + +是展示模型； + +而： + +```text +Codex thread +``` + +是 Agent 执行上下文模型。 + +二者可能相关，但不应该互相替代。 + +本项目目前只把： + +```text +threadId + visible messages +``` + +持久化到浏览器；真正 thread history 仍由 Codex 管理。这种边界非常适合 V0。 + +--- + +## 6. thread/start 与 thread/resume + +当前 Runtime： + +```ts +const threadRequest = threadId ? 'thread/resume' : 'thread/start' + +const threadResult = await connection.request(threadRequest, { + ...(threadId ? { threadId } : {}), + cwd: workingDirectory, + model: LUNA_HIGH_OPTIONS.model, + approvalPolicy: LUNA_HIGH_OPTIONS.approvalPolicy, + sandbox: LUNA_HIGH_OPTIONS.sandbox, +}) +``` + +逻辑可以抽象成： + +```text +没有 threadId + -> thread/start + -> 得到 activeThreadId + +已有 threadId + -> thread/resume(threadId) + -> 得到 activeThreadId +``` + +为什么仍然从 response 再解析 `activeThreadId`，而不是盲信输入？ + +因为 response 才是 Runtime 的权威确认值。 + +```ts +const activeThreadId = threadIdFromResponse(threadResult.response) +``` + +这是很值得迁移的工程原则： + +> **状态切换完成后，以服务端确认的 canonical identity 为准。** + +--- + +## 7. turn/start：一次 Agent 执行真正从这里开始 + +本项目： + +```ts +const turnResult = await connection.request('turn/start', { + threadId: activeThreadId, + input: [{ type: 'text', text: prompt }], + cwd: workingDirectory, + model: LUNA_HIGH_OPTIONS.model, + effort: LUNA_HIGH_OPTIONS.effort, + approvalPolicy: LUNA_HIGH_OPTIONS.approvalPolicy, + sandboxPolicy: LUNA_HIGH_OPTIONS.sandboxPolicy, +}) +``` + +这组字段实际分成三类： + +### 内容 + +```text +threadId +input +``` + +### 模型执行参数 + +```text +model +effort +cwd +``` + +### 权限 / 安全参数 + +```text +approvalPolicy +sandboxPolicy +``` + +把这三类概念区分开，对后续做“用户可选模型”和“管理员固定安全策略”非常重要。 + +UI 可以控制： + +```text +model / effort +``` + +但不代表 UI 应该能控制： + +```text +sandbox / network / approvals +``` + +安全策略应该留在可信 server boundary。 + +--- + +## 8. 真正的文本流：item/agentMessage/delta + +实际验证序列： + +```text +thread/started +→ turn/started +→ item/started(agentMessage) +→ item/agentMessage/delta +→ item/agentMessage/delta +→ ... +→ item/completed(agentMessage) +→ thread/tokenUsage/updated +→ turn/completed +``` + +这才是“用户看到正文逐渐长出来”的数据来源。 + +### 生命周期映射 + +Codex 协议： + +```text +item/started(agentMessage) +item/agentMessage/delta +item/completed(agentMessage) +``` + +应用协议： + +```text +assistant.started +assistant.delta +assistant.completed +``` + +映射的目的不是改名字，而是**切断 UI 与 Codex 协议耦合**。 + +```text +Codex App Server + │ + │ provider-specific + ▼ +CodexThreadEvent + │ + │ normalization boundary + ▼ +ChatEvent + │ + │ application-specific + ▼ +React reducer +``` + +### 为什么 completed 仍然重要？ + +因为 delta 是增量过程，`item/completed` 才是最终 authoritative state。 + +官方文档同样强调：最终 `item/completed` 应作为最终状态依据。 + +因此推荐模型： + +```text +started -> 建立 UI entity +delta -> 乐观追加 +completed -> 用最终 snapshot 校准 +``` + +这是一种 **stream + final reconciliation** 模式。 + +它不仅适用于 LLM：文件上传、语音转写、协作编辑、长任务进度都可复用。 + +--- + +## 9. Snapshot 与 Delta：为什么两种模型都要理解 + +有的上游提供 delta： + +```text +"React " +"是一个 " +"UI 库" +``` + +有的上游提供累计 snapshot： + +```text +"React" +"React 是一个" +"React 是一个 UI 库" +``` + +项目 normalizer 保留了对 snapshot update 的兼容： + +```ts +if (item.text.startsWith(previous)) { + const delta = item.text.slice(previous.length) + return [{ type: 'assistant.delta', id: item.id, delta }] +} +``` + +如果新 snapshot 不是旧 snapshot 的前缀： + +```ts +return [{ + type: 'assistant.completed', + id: item.id, + text: item.text, +}] +``` + +这是正确的防重复思想： + +> 非 append-only snapshot 不能安全地转换成 append-only delta。 + +例如： + +```text +old = "React isa" +new = "React is a" +``` + +简单： + +```ts +new.slice(old.length) +``` + +会得到错误结果。 + +### 更可靠的通用策略 + +```text +如果上游明确给 delta：直接追加 +如果上游给 snapshot： + - prefix -> 算 suffix delta + - non-prefix -> full replacement +最终 completed -> authoritative replace +``` + +--- + +## 10. threadId / turnId correlation：流式系统不能只看事件类型 + +这是当前 `main` 最重要的可靠性缺口之一。 + +官方 notification 通常带： + +```text +threadId +turnId +itemId +``` + +而当前实现结束循环的条件是： + +```ts +if (message.method === 'turn/completed') return +``` + +问题是它没有验证这个 `turn/completed` 是否属于当前 turn。 + +在“每个请求 spawn 一个 app-server，且单 turn 独占连接”的 V0 中，碰撞概率较低；但一旦演进到： + +- long-lived connection +- 多 thread +- 并发 turn +- steer +- interrupt +- detached review + +仅靠 method 已经不够。 + +### 推荐实现 + +从 `turn/start` response 保存： + +```text +activeThreadId +activeTurnId +``` + +每个 notification 进入业务层前判断： + +```ts +belongsToTurn(message, activeThreadId, activeTurnId) +``` + +至少： + +```text +threadId == activeThreadId +turnId == activeTurnId +``` + +然后： + +```text +turn/completed(other turn) -> 忽略/路由给其他订阅者 +turn/completed(active turn) -> 结束当前 iterator +``` + +### 可迁移结论 + +> **在 multiplexed event stream 中，事件类型解决“是什么”，correlation id 解决“是谁的”。** + +两者缺一不可。 + +--- + +## 11. tokenUsage：为什么 usage 是流中的独立状态 + +当前 runtime 维护： + +```ts +type UsageState = { value: CodexUsage } +``` + +收到： + +```text +thread/tokenUsage/updated +``` + +后更新： + +```ts +usage.value = last +``` + +最终在 `turn.completed` 转换成： + +```ts +{ + type: 'turn.completed', + usage: usage.value, +} +``` + +这里值得注意：官方把它定义为 thread usage update；当前代码把 `tokenUsage.last` 暂存并附着到应用层 turn completion。 + +这是一种 Adapter 层聚合： + +```text +provider notifications + ↓ +small local state + ↓ +application event +``` + +它说明 normalizer 不一定必须是纯粹 1:1 映射；只要状态范围明确，它也可以承担协议重组。 + +长期应该明确字段语义： + +```text +last +cumulative +thread total +turn usage +``` + +不要仅凭变量名 `usage` 假定含义。 + +--- + +## 12. Read-only Sandbox：安全不是一个开关 + +项目固定： + +```ts +const LUNA_HIGH_OPTIONS = { + model: 'luna', + effort: 'high', + approvalPolicy: 'never', + sandbox: 'read-only', + sandboxPolicy: { + type: 'readOnly', + networkAccess: false, + }, +} +``` + +这里其实有三条独立防线： + +### sandbox = read-only + +限制 filesystem mutation 能力。 + +### networkAccess = false + +限制 sandbox 中的网络访问。 + +### approvalPolicy = never + +表示当前宿主不走交互审批升级权限的路径。 + +三者不能互相替代。 + +例如： + +```text +read-only != no network +no approval != sandboxed +no network != no local data read +``` + +### 安全模型的正确表达 + +不要写： + +> “开启 read-only，所以安全。” + +应该写： + +> “当前 V0 通过只读 sandbox、关闭 sandbox network access、不提供 interactive approval、限制浏览器可见事件四层策略，把能力收敛到 repository inspection / explanation 类工作流。” + +这才是 threat model 语言。 + +--- + +## 13. ChatGPT 本地认证复用：认证为什么必须留在 Server + +本项目不要求把 OpenAI API Key 发给浏览器。 + +App Server 运行在本机 server process 上，复用宿主已有的 Codex / ChatGPT 登录状态。 + +边界： + +```text +Browser + X 不能访问 ~/.codex 等认证状态 + +TanStack server + ↓ spawn +codex app-server + ↓ +local Codex auth +``` + +应用只需要知道： + +```text +runtime 能不能成功工作 +``` + +不应该知道： + +```text +具体 token +session +cookie +credential path +``` + +### 通用原则 + +> **把 credential consumption 放在离 credential 最近的可信进程；上层只消费能力，不消费凭证。** + +这和数据库连接、SSH agent、OS keychain、云凭据代理是同一个设计模式。 + +--- + +## 14. 敏感信息隔离：不要把“结构化事件”误认为“安全事件” + +App Server 原始事件可能包含： + +- reasoning text +- command stdout/stderr +- command cwd +- file path / diff +- MCP arguments/results +- upstream error detail +- local runtime path + +因此 server normalizer 是安全边界。 + +当前项目明确不把这些内容发给浏览器： + +```text +raw reasoning -> 不转发 +command output -> 不转发 +MCP arguments -> 不转发 +MCP result -> 不转发 +``` + +例如 MCP item 映射： + +```ts +return { + id, + type: 'mcp_tool_call', + server, + tool, + arguments: undefined, + result: undefined, + error: undefined, + status: ..., +} +``` + +测试还专门检查： + +```ts +expect(JSON.stringify(events)).not.toContain('SECRET') +``` + +### 当前仍需加强的地方 + +`sanitizeErrorMessage()` 主要处理 token 风格秘密： + +```text +Bearer ... +sk-... +sess-... +session-... +token-... +``` + +但绝对文件路径等 diagnostics 仍可能泄漏。 + +更稳健的生产策略是： + +```text +Server log: 原始详细错误 +Browser: code + 泛化 message + traceId +``` + +而不是试图用越来越复杂的 regex 清洗所有错误字符串。 + +--- + +## 15. 进程生命周期：当前 process-per-turn 方案 + +现在每次 `streamTurn()`： + +```text +spawn codex app-server +→ initialize +→ thread/start|resume +→ turn/start +→ consume stream +→ turn/completed +→ close process +``` + +### 优点 + +- 隔离简单。 +- 一个 turn 出故障不污染后续连接。 +- 无需实现复杂的订阅路由。 +- Demo 容易理解和验证。 + +### 缺点 + +- 每个 turn 都有进程启动成本。 +- 每次都要 initialize。 +- 无法自然承载多 thread multiplexing。 +- `turn/steer` / `turn/interrupt` / approval 等跨请求操作变复杂。 +- 不适合未来多 Agent / 多窗口 desktop host。 + +### process-per-turn vs long-lived + +| 维度 | Process per turn | Long-lived app-server | +|---|---|---| +| 实现复杂度 | 低 | 高 | +| 故障隔离 | 强 | 需要客户端治理 | +| 启动开销 | 每次都有 | 一次 | +| 多 thread | 不自然 | 自然 | +| interrupt/steer | 较难 | 自然 | +| approval routing | 简单拒绝 | 可完整实现 | +| Desktop Agent | 不理想 | 推荐 | +| V0 Demo | 推荐 | 可能过度设计 | + +### 推荐演进 + +保持现有： + +```ts +interface CodexRuntime +``` + +不要让 UI 感知进程模型。 + +未来只需把实现从： + +```text +CodexAppServerRuntime + -> connection per turn +``` + +换成： + +```text +CodexAppServerRuntime + -> shared connection manager + -> request router + -> turn subscriptions +``` + +上层不变。 + +--- + +## 16. Abort 与 turn/interrupt：这两个动作不是一回事 + +当前 runtime： + +```ts +const abortHandler = () => connection.close().catch(() => undefined) +input.signal?.addEventListener('abort', abortHandler, { once: true }) +``` + +也就是说 AbortSignal 当前语义是： + +```text +abort +→ close app-server connection/process +``` + +这属于 **transport/process cancellation**。 + +官方协议还提供： + +```json +{ + "method": "turn/interrupt", + "id": 31, + "params": { + "threadId": "thr_123", + "turnId": "turn_456" + } +} +``` + +成功后当前 turn 最终会： + +```text +turn/completed status=interrupted +``` + +这属于 **domain cancellation**。 + +### 二者区别 + +```text +Abort HTTP / stream / process += 我不再等待这个 transport + +turn/interrupt += 请 Agent runtime 正式中止这个 turn +``` + +关闭浏览器 stream 并不自动等于 Agent 已停止。 + +### 推荐 Stop 流程 + +```text +User clicks Stop + ↓ +Client aborts UI transport (optional, for responsiveness) + ↓ +Server sends turn/interrupt(threadId, turnId) + ↓ +App Server acknowledges + ↓ +turn/completed(status=interrupted) + ↓ +State converges to interrupted +``` + +如果是 process-per-turn，可把 kill process 作为最后兜底，但不应把 kill 当作唯一领域协议。 + +--- + +## 17. Child Process shutdown：为什么需要 timeout + SIGKILL 兜底 + +当前： + +```ts +await new Promise((resolve) => { + const finish = () => resolve() + this.child.once('exit', finish) + this.child.kill() +}) +``` + +这里隐含假设： + +```text +SIGTERM -> child 一定退出 +``` + +生产环境不能依赖这个假设。 + +更健壮的模式： + +```text +close called + ↓ +mark closing (idempotent) + ↓ +close stdin/readline + ↓ +SIGTERM + ↓ +wait N ms + ├─ exited -> done + └─ alive -> SIGKILL +``` + +还应考虑： + +- spawn error +- stdin EPIPE +- stdout EOF +- malformed JSON +- stderr 无限增长 +- child exits before pending requests settle + +这部分属于 Agent Host 的基础设施能力，而不是 Codex 特有知识。 + +--- + +## 18. 关键源码映射 + +### `src/server/codex/codex-app-server.server.ts` + +职责： + +```text +spawn app-server +stdio framing +request / response matching +notification buffering +server request rejection +protocol -> CodexThreadEvent +thread start/resume +turn start +usage accumulation +process cleanup +``` + +这是当前最核心的 provider adapter。 + +### `src/server/codex/codex-runtime.ts` + +定义稳定 Runtime interface： + +```ts +interface CodexRuntime { + streamTurn( + input: StreamCodexTurnInput + ): AsyncGenerator +} +``` + +价值：把“上层需要什么”与“Codex 怎么实现”拆开。 + +### `src/server-functions/codex-event.types.ts` + +是 server-side structural mirror。 + +注意：它不是浏览器 contract，也不是官方完整协议 schema。 + +### `src/server-functions/codex-event-normalizer.ts` + +职责： + +```text +CodexThreadEvent + -> ChatEvent +``` + +这里同时承担： + +- provider decoupling +- data minimization +- security filtering +- assistant snapshot state +- usage normalization + +### `src/server-functions/chat-stream.ts` + +每个 turn 创建一个 stateful normalizer： + +```ts +const normalizer = createCodexEventNormalizer() +``` + +这是关键，因为 snapshot map 必须跨同一 stream 的多个事件存活。 + +--- + +## 19. 完整时序图 + +```mermaid +sequenceDiagram + actor U as User + participant UI as React UI + participant TS as TanStack Start + participant RT as CodexAppServerRuntime + participant AS as codex app-server + + U->>UI: Send prompt + UI->>TS: streamChat() + TS->>RT: streamTurn(prompt, threadId?) + RT->>AS: spawn app-server --stdio + RT->>AS: initialize(id=1) + AS-->>RT: response(id=1) + RT->>AS: initialized + + alt New conversation + RT->>AS: thread/start + else Existing conversation + RT->>AS: thread/resume(threadId) + end + + AS-->>RT: thread response + RT-->>TS: thread.started + + RT->>AS: turn/start(threadId, input) + AS-->>RT: turn/start response(turnId) + AS-->>RT: turn/started + AS-->>RT: item/started(agentMessage) + RT-->>TS: assistant.started + TS-->>UI: create empty assistant message + + loop Generated text + AS-->>RT: item/agentMessage/delta + RT-->>TS: assistant.delta + TS-->>UI: append delta + end + + AS-->>RT: item/completed(agentMessage) + RT-->>TS: assistant.completed(final text) + TS-->>UI: calibrate final text + + AS-->>RT: thread/tokenUsage/updated + AS-->>RT: turn/completed + RT-->>TS: turn.completed(usage) + TS-->>UI: status=idle + RT->>AS: close process +``` + +--- + +## 20. 当前代码中值得保留的设计 + +### 20.1 Runtime interface + +正确。它避免把 App Server 进程模型扩散到 UI。 + +### 20.2 application-owned ChatEvent + +正确。UI 不应该消费 raw Codex protocol。 + +### 20.3 delta + completed 校准 + +正确。过程响应和最终权威状态分开。 + +### 20.4 normalizer 内过滤敏感 payload + +正确。安全边界必须放在 server。 + +### 20.5 每个 stream 一个 normalizer instance + +正确。否则 snapshot 状态每次调用都会丢失。 + +--- + +## 21. 当前代码中需要继续修的地方 + +### P1：缺少 threadId / turnId correlation + +不要只靠： + +```ts +message.method === 'turn/completed' +``` + +结束 stream。 + +### P1：Stop 还不是协议级 interrupt + +AbortSignal 关闭 process != `turn/interrupt`。 + +### P1/P2：process close 缺少强制退出 timeout + +避免 child 不退出导致请求永久等待。 + +### P2：error sanitization 应改为结构化错误 + +不要依赖 regex 完成全部数据防泄漏。 + +### P2：AppServerConnection 测试深度不足 + +现有测试主要验证 notification mapper 和 MCP secret 不泄漏，还没有锁死： + +- response id routing +- interleaved notification buffering +- wrong turn notification filtering +- server request +- premature child exit +- malformed JSON +- abort +- shutdown escalation + +--- + +## 22. 如何抓真实事件验证协议 + +调试时不要一开始打印完整 payload；Agent 事件可能包含敏感数据。 + +推荐临时只记录 metadata： + +```ts +console.error('[codex-event]', { + method: message.method, + id: message.id, + threadId: getThreadId(message), + turnId: getTurnId(message), + itemType: getItemType(message), + itemId: getItemId(message), + textLength: getTextLength(message), +}) +``` + +验证目标： + +```text +1. initialize handshake 是否正确 +2. thread/start|resume response 是否返回 thread id +3. turn/start response 是否返回 turn id +4. agentMessage 是否先 started +5. delta 是否连续出现 +6. completed text 是否等于最终可见文本 +7. turn/completed 的 id/status 是否匹配本 turn +8. usage update 在哪个时点出现 +``` + +### 不推荐 + +```ts +console.log(JSON.stringify(message)) +``` + +长期保留在生产 server 中。 + +因为可能记录： + +- command output +- MCP result +- paths +- reasoning +- credentials-like strings + +--- + +## 23. 常见故障与排查路径 + +### 23.1 `codex` 找不到 + +检查： + +```bash +which codex +codex --version +``` + +项目支持： + +```text +CODEX_APP_SERVER_PATH +``` + +用来指定 executable。 + +### 23.2 app-server 启动后立刻退出 + +看 server stderr，不要把 stderr 原样发浏览器。 + +重点检查： + +```text +CLI version +auth state +unsupported flags +working directory +model availability +``` + +### 23.3 有生命周期事件，但没有文本流 + +确认是否真正收到： + +```text +item/agentMessage/delta +``` + +不要只看： + +```text +item/started +item/completed +``` + +### 23.4 出现 delta，但 UI 最后重复文本 + +检查是否把： + +```text +delta +``` + +误当成： + +```text +snapshot +``` + +或者同时将 snapshot update 与真实 delta 双重追加。 + +### 23.5 New Chat 后老任务还在跑 + +这通常是： + +```text +generation guard 只保护 UI +``` + +但没有： + +```text +turn/interrupt +``` + +### 23.6 stream 永远不结束 + +需要区分： + +- 模型仍在工作 +- tool call 卡住 +- response stream disconnected +- `turn/completed` 没收到 +- 收到了别的 turn 的 completed +- child process 半死不活 + +不要只加一个无限大的 HTTP timeout。 + +--- + +## 24. 测试策略：不要只测试 Mapper + +### Layer 1：纯函数测试 + +测试： + +```text +App Server item -> CodexThreadEvent +CodexThreadEvent -> ChatEvent +snapshot -> delta +error sanitization +usage mapping +``` + +当前已有一部分。 + +### Layer 2：Fake App Server transport + +建议实现 fake child/transport，脚本化输出： + +```text +client initialize +server response +notification A +notification B +request response +wrong-turn completed +right-turn completed +EOF +``` + +关键断言： + +```text +request ID 正确关联 +等待 response 期间 notification 不丢 +wrong turn 不结束 stream +right turn 才结束 +server request 被正确处理 +``` + +### Layer 3：Process integration test + +启动一个 fake executable，通过 stdin/stdout 模拟 app-server。 + +覆盖： + +```text +spawn failure +malformed JSON +stderr +exit before response +SIGTERM ignored -> SIGKILL +``` + +### Layer 4：Real smoke test + +在具备本地 Codex auth 的环境运行一次： + +```text +thread/start +turn/start +agentMessage delta >= 2 +item/completed +turn/completed +``` + +真实 smoke test 的意义是验证： + +```text +我们的协议假设 +≈ 当前安装版本真实行为 +``` + +但它不能替代 deterministic unit/integration test。 + +--- + +## 25. Fake transport 场景示例 + +最值得补的一组测试： + +```text +turn/start response: turn-A + +notification: + delta(turn-B) -> 不应进入 A + delta(turn-A,"Hi") -> 应进入 A + completed(turn-B) -> 不应结束 A + delta(turn-A,"!") -> 应进入 A + completed(turn-A) -> A 结束 +``` + +预期应用事件： + +```text +assistant.delta("Hi") +assistant.delta("!") +turn.completed +``` + +这个测试一旦存在，未来做 long-lived App Server 时会非常有价值。 + +--- + +## 26. Approval / Server Request：未来开启写能力后会发生什么 + +V0 统一拒绝 interactive server request 是合理的，因为： + +```text +sandbox=read-only +approvalPolicy=never +``` + +目标就是不给升级权限路径。 + +未来如果开启 workspace write，必须设计： + +```text +App Server server request + ↓ +Runtime maps to ApprovalRequest + ↓ +Backend assigns request identity + ↓ +Browser renders explicit approval UI + ↓ +User allow/deny + ↓ +Backend validates request still active + ↓ +response to App Server +``` + +不能把 raw server request 直接发前端然后信任前端原样回传。 + +安全相关决策必须 server-side 再验证。 + +--- + +## 27. MCP 支持：事件显示与协议执行应分层 + +当前浏览器只看到类似： + +```text +Tool: github / search +``` + +而不会看到： + +```text +arguments +result +``` + +这是一种很好的默认数据最小化策略。 + +将来若要做 MCP Inspector，可以新增一个显式权限等级： + +```text +summary-only +safe-details +full-debug(local development only) +``` + +而不是直接删除当前安全边界。 + +--- + +## 28. turn/steer：为什么 long-lived runtime 后会更有价值 + +`turn/steer` 用于向**当前仍在执行的 turn**追加输入。 + +它和新建下一轮： + +```text +turn/start +``` + +语义不同。 + +典型 UX： + +```text +Codex 正在做复杂任务 +User: “先别改测试文件” +``` + +如果支持 steer： + +```text +当前 turn 继续，但获得新约束 +``` + +如果不支持，只能： + +```text +Stop -> 新 turn +``` + +这也是为什么真正的 Agent UI 最终通常需要保存 active turn identity。 + +--- + +## 29. 多 Thread / 多 Agent：从单 iterator 演进到事件路由器 + +当前模型： + +```text +1 streamTurn + -> 1 process + -> 1 active turn +``` + +未来： + +```text +1 AppServerConnection + ├─ Thread A / Turn A1 + ├─ Thread B / Turn B1 + └─ Thread C / Turn C2 +``` + +这时需要： + +```text +Connection Reader + ↓ +Protocol Decoder + ↓ +Correlation Router + ├─ thread A subscribers + ├─ thread B subscribers + ├─ pending RPC promises + └─ server request handlers +``` + +也就是说，`AsyncGenerator` 仍然可以作为上层 API，但底层不能再由每个 generator 独占 stdout iterator。 + +这是从 Demo Runtime 到 Desktop Agent Host 的关键架构跃迁。 + +--- + +## 30. 与 SSE / WebSocket 的关系 + +Codex App Server stdio 只解决： + +```text +TanStack server <-> Codex +``` + +它没有规定： + +```text +Browser <-> TanStack server +``` + +浏览器侧仍可以选择： + +| Transport | 适合场景 | +|---|---| +| TanStack async server function | 当前 Demo，类型集成好 | +| SSE | server -> browser 单向事件，非常适合 Agent stream | +| Fetch ReadableStream | 自定义协议、控制直接 | +| WebSocket | steer、approval、双向 realtime、多任务 | + +因此不要把“Codex App Server 用 stdio”误解成“整个应用都必须用 stdio”。 + +正确分层： + +```text +Browser Transport +!= +Agent Runtime Transport +``` + +--- + +## 31. 可迁移到其他 Agent Runtime 的抽象 + +从本项目提炼出来，最有价值的不是 `codex app-server` 命令本身，而是下面这个通用模型： + +```text +Provider Runtime + ↓ +Provider Events + ↓ +Runtime Adapter + ↓ +Application Events + ↓ +State Machine + ↓ +UI +``` + +统一应用事件可以是： + +```ts +type AgentEvent = + | { type: 'message.started'; id: string } + | { type: 'message.delta'; id: string; delta: string } + | { type: 'message.completed'; id: string; text: string } + | { type: 'activity.started'; ... } + | { type: 'activity.updated'; ... } + | { type: 'activity.completed'; ... } + | { type: 'turn.completed'; ... } + | { type: 'error'; ... } +``` + +然后： + +```text +Codex -> adapter A +Claude -> adapter B +Pi -> adapter C +``` + +UI 不需要知道 provider。 + +--- + +## 32. 设计反例 + +### 反例 1：前端直接消费 raw App Server event + +问题： + +```text +provider lock-in +敏感信息泄漏 +协议升级影响 UI +测试困难 +``` + +### 反例 2：收到 delta 后每次创建新 message + +问题： + +```text +"React" +" 是" +" UI" +``` + +变成 3 条聊天消息。 + +应该按 `itemId` 聚合同一个 entity。 + +### 反例 3：completed 到来时继续 append final text + +会得到： + +```text +React is UIReact is UI +``` + +completed 应做 reconciliation/replacement。 + +### 反例 4：New Chat 只清 React state + +旧 runtime 继续耗 token / 执行工具。 + +UI cancellation 和 runtime interruption 必须分开考虑。 + +### 反例 5：所有错误原样透传 + +本地 Agent 的错误字符串往往比普通 SaaS API 更敏感，因为它能包含 filesystem/runtime 信息。 + +--- + +## 33. 推荐的下一阶段 Runtime 结构 + +```text +CodexRuntime +│ +├── startTurn(...): TurnHandle +│ +└── TurnHandle + ├── threadId + ├── turnId + ├── events: AsyncIterable + ├── interrupt() + └── dispose() +``` + +相比现在单一： + +```ts +streamTurn(): AsyncGenerator +``` + +`TurnHandle` 更适合后续： + +- Stop +- Steer +- active turn status +- multi-tab +- background turn +- telemetry + +但对当前 V0，不必急着改；先补 correlation 和 interrupt 即可。 + +--- + +## 34. 最小可靠性 Checklist + +实现任意本地 Agent Runtime Adapter 时，至少回答这些问题： + +- [ ] 是否区分 request / response / notification / server request？ +- [ ] pending request 是否按 request id 关联？ +- [ ] 等 response 时到达的 notification 会不会丢？ +- [ ] 事件是否按 threadId / turnId 关联？ +- [ ] message delta 是否按 itemId 聚合？ +- [ ] completed 是否作为最终权威 snapshot？ +- [ ] UI abort 是否真的停止 runtime？ +- [ ] 是否有领域级 interrupt？ +- [ ] child process 卡死怎么退出？ +- [ ] stderr 如何处理？ +- [ ] raw reasoning/command/MCP data 会不会泄漏到 browser？ +- [ ] error message 是否可能泄漏路径和凭据？ +- [ ] sandbox、network、approval 是否分别建模？ +- [ ] 真实 runtime 版本变化如何 smoke test？ + +--- + +## 35. 复习题 + +### 基础 + +1. `runStreamed()` 为什么不必然意味着 Assistant 正文是 token/delta 流？ +2. Thread、Turn、Item 各自解决什么问题？ +3. `item/agentMessage/delta` 与 `item/completed(agentMessage)` 的职责有什么区别？ +4. 为什么 UI 不应该直接依赖 Codex App Server event shape？ +5. `initialize` 与 `initialized` 分别是什么类型的消息？ + +### 进阶 + +6. 为什么等待某个 RPC response 时必须缓存穿插到达的 notification？ +7. 为什么 `turn/completed` 必须用 `turnId` correlation，而不能只判断 method？ +8. 为什么浏览器取消读取 stream 不等于 Agent turn 已中断？ +9. snapshot 转 delta 时，为什么 non-prefix update 不能直接做字符串 slice？ +10. `sandbox: read-only`、`networkAccess: false`、`approvalPolicy: never` 为什么是三种不同控制？ + +### 工程设计 + +11. 如果把 app-server 改成长连接，你会怎样设计 pending request map 与 event router？ +12. 怎样测试“wrong turn completed 不会结束 current turn”？ +13. 当 App Server 发来 approval request 时，为什么不能直接把原始 request 交给浏览器决定？ +14. process-per-turn 为什么适合 Demo，却不适合作为桌面 Agent 的最终架构？ +15. 如果未来增加 Claude/Pi Runtime，哪些层应复用，哪些层应该 provider-specific？ + +--- + +## 36. 一页总结 + +```text +真正的 Agent 流式 UI +不是“API 返回一个 stream”这么简单。 + +它需要： + +1. Runtime 有真实增量事件 + Codex: item/agentMessage/delta + +2. 协议客户端正确处理 multiplexing + request id + notification + server request + +3. 用 identity 做事件关联 + threadId + turnId + itemId + +4. 用应用协议隔离 provider + raw Codex event -> ChatEvent + +5. 流过程与最终状态分开 + delta append + completed reconcile + +6. 明确安全边界 + auth / reasoning / command output / MCP payload 不进浏览器 + +7. 明确取消语义 + transport abort != turn/interrupt + +8. 管好本地进程 + spawn / stderr / EOF / timeout / SIGTERM / SIGKILL + +9. 测试协议生命周期，而不只是 mapper + +10. 当系统走向多 Thread / 多 Agent 时， + 从“一个 generator 独占一个进程”演进为 + “长连接 + correlation router + TurnHandle”。 +``` + +--- + +## 参考资料 + +- Codex App Server 官方文档：https://developers.openai.com/codex/app-server/ +- 本项目 Runtime：`src/server/codex/codex-app-server.server.ts` +- Runtime interface：`src/server/codex/codex-runtime.ts` +- Runtime event mirror：`src/server-functions/codex-event.types.ts` +- Event normalizer：`src/server-functions/codex-event-normalizer.ts` +- Streaming bridge：`src/server-functions/chat-stream.ts` +- 当前 App Server notification 测试：`src/server/codex/codex-app-server.test.ts` + +> 文档中的“当前实现”以 `main@6f4584d2` 为准。协议能力以阅读本文时的官方 App Server 文档为权威来源；App Server 是活跃演进协议，升级 Codex CLI 后应重新进行真实事件 smoke verification。 -- 2.54.0