跳到主要内容

数据截至 (上游 commit ee230f304a1a)

一次回合是怎么跑的:后端抽象与 approval 环路

30 秒导读: Letta Code 的对话历史不在 CLI 进程里,在服务端。所以"一个回合"不是一次函数调用,而是客户端和服务端来回好几趟:模型说"我要跑 Bash",服务端把回合停在 requires_approval;CLI 本地执行工具,把结果当成一条新消息发回去,服务端接着往下跑。这一章讲透这个环路,以及它带来的两个好处——中途断线能续,关掉重开能恢复。

本章只讲主线:一个 turn 在客户端如何流转。工具本身怎么实现看 本地工具层,放不放行的规则看 批准这件事。

1. 先建立直觉:"有状态"到底状态在哪​

1.1 两种编码 agent 的对照​

大多数终端 agent 是无状态客户端 + 无状态 API:进程内存里攒着 messages[] 数组,每一轮把整个数组重新发给模型 API。进程一死,数组就没了。

Letta Code 是有状态服务端:conversation 的消息历史、系统提示、未决的工具调用,全部由后端持有。CLI 每次只发增量(这一轮的新消息),不发全量历史。

维度无状态客户端(常见做法)Letta Code
谁存 messages[]CLI 进程内存服务端 conversation
每次请求发什么全量历史只发本轮新增的 message / approval
进程被 kill 后历史丢失,只能靠本地 session 文件重放服务端仍持有,重开即接上
未决的工具调用丢失服务端仍停在 requires_approval

1.2 一句话直觉​

把 conversation 当成一台停在断点上的虚拟机。 CLI 不是虚拟机本身,而是调试器:它 attach 上去、看寄存器(消息流)、执行一条外部指令(工具),再把结果写回去让虚拟机继续跑。调试器崩了,虚拟机还停在原来那个断点上。

requires_approval 就是那个断点。

1.3 用起来什么样​

你: 帮我把 src/ 下所有 console.log 删掉

  ● 我先看看有多少处
  ⏺ Grep(pattern: "console\.log", path: "src")     ← 自动放行,直接跑
    找到 14 处

  ⏺ Bash(rm -rf node_modules/.cache)               ← 停下来问你
    ┌ 允许执行这条命令吗?
    │ ❯ 允许    允许(本会话不再问)    拒绝
    └

那个"停下来问你"的瞬间,服务端的回合已经结束了(stop_reason = requires_approval)。你按 Ctrl-C 关掉 CLI、重新 letta,那条待批准的 Bash 还在,提示框会重新弹出来。这不是 CLI 把它存到了本地文件,而是它本来就在服务端。

2. 顶层全景:一个 turn 的骨架​

2.1 怎么读这张图​

从左上开始顺时针。粗箭头是一次 HTTP 往返;虚线框里的事全在客户端本地发生,不经过网络。带 ↺ 的那条回边就是本章标题里的 "approval 环路"——它把一个用户回合拆成 N 次服务端往返。

   用户敲回车
        │
        ▼
 ┌──────────────────┐   ① 组请求体(只发增量)
 │ processConversa- │──────────────────────────────┐
 │ tion (while true)│                              │
 └──────────────────┘                              ▼
        ▲                              ┌───────────────────────┐
        │                              │ sendMessageStream     │
        │ ↺ 把工具结果当              │  → Backend 接口        │
        │   新消息再发一次             └───────────┬───────────┘
        │                                          │ SSE 流
   ┌────┴─────────────────┐                        ▼
   │ ④ 本地执行工具       │            ┌───────────────────────┐
   │   executeApproval-   │            │ ② drainStream         │
   │   Batch              │            │   逐 chunk 拼装 + 落 UI│
   └────▲─────────────────┘            └───────────┬───────────┘
        │                                          │
        │  ┌───────────────────────────────────────┘
        │  ▼
   ┌────┴──────────────────────────────────┐
   │ ③ 看 stop_reason                      │
   │   end_turn ────────────► 回合结束      │
   │   requires_approval ───► 分三堆:      │
   │        自动放行 / 自动拒 / 问用户      │
   │   error / llm_api_error ► 重试或报错   │
   └───────────────────────────────────────┘

2.2 部件一句话职责​

部件干什么在哪
Backend 接口把"和服务端说话"这件事抽象成一组方法,好换实现src/backend/backend.ts:161
APIBackend直连 Letta 云端 API 的实现src/backend/backend.ts:305
LocalBackend本机自跑 provider 循环的实现src/backend/local/local-backend.ts:245
sendMessageStream组请求体、开 SSE 流、挂上下文元数据src/agent/message.ts:350
drainStream消费 SSE,累积 approvals 与 stop_reasonsrc/cli/helpers/stream.ts:90
onChunk(accumulator)把分片 chunk 拼成可渲染的行src/cli/helpers/accumulator.ts:1090
executeApprovalBatch批量执行审批决定,带资源锁src/agent/approval-execution.ts:372
useConversationLoop上面这些的编排者(整个 turn 的状态机)src/cli/app/use-conversation-loop.ts:284
getResumeDataFromBackend重开 CLI 时从服务端捞回未决审批 + 历史src/agent/check-approval.ts:513

3. 后端抽象:一个接口,两种世界​

3.1 它要解决的小问题​

同一份 CLI,既要能连云端跑 Claude/GPT,也要能在本机自己跑 provider 循环(离线、自带 key、本地模型)。如果 UI 层到处 if (isLocal),很快就会烂掉。

3.2 做法:接口 + 能力位 + 单例​

Backend 是一个约 25 个方法的接口(src/backend/backend.ts:161),覆盖 agent / conversation / message / run 四类操作。方法签名直接从 SDK 客户端的参数类型推导出来:

export type ConversationMessageCreateParams = Parameters<
  APIClient["conversations"]["messages"]["create"]
>;

src/backend/backend.ts:28-33 —— 这样接口不会和 SDK 漂移,SDK 改签名会直接编译报错。

但两种后端能力不一样,不是每个方法都有意义。于是有一组能力位 BackendCapabilities(src/backend/backend.ts:150),调用方先查再用:

能力位APIBackendLocalBackend含义
remoteMemfstruefalse记忆文件系统在服务端
serverSideToolManagementtruefalse工具注册由服务端管
serverSecretstruefalse密钥托管在服务端
agentFileImportExporttruefalse支持 .af 导入导出
promptRecompiletruetrue支持重编译系统提示
byokProviderRefreshtruefalse自带 key 的 provider 刷新
localModelCatalogfalsetrue模型目录来自本机
localMemfsfalsetrue记忆文件系统在本机

对照两处定义:src/backend/backend.ts:306-315(API)与 src/backend/local/local-backend.ts:246-255(local)——几乎是镜像取反。

用法举例:回合循环用 localModelCatalog 判断"本机一个模型都没配"时该不该合成一段本地提示回复(src/cli/app/use-conversation-loop.ts:380-387);重试路径用 localModelCatalog && !remoteMemfs 判断"本地已经把这轮输入落盘了,重试时别再发一遍"(src/cli/app/use-conversation-loop.ts:2594-2603)。

3.3 模式怎么定​

模式解析被单独拆到一个叶子模块,免得读一个布尔值就得把整个 backend 模块拖进来:

export function resolveBackendMode(): BackendMode {
  return (
    configuredBackendMode ?? (isLocalBackendEnvEnabled() ? "local" : "api")
  );
}

src/backend/backend-mode.ts:24 —— 运行时 override 优先,否则看环境变量 LETTA_LOCAL_BACKEND_EXPERIMENTAL。isExperimentalLocalBackendEnabled()(src/backend/backend-mode.ts:38)只是 mode === "local" 的语义化包装。

三个取实例的入口,各有分工:

函数行为用途
getBackend()懒初始化的进程单例绝大多数调用点
getBackendForMode(mode)新建一个实例,不动全局跨后端读数据(如从另一边捞 pinned agent)
configureBackendMode(mode)换全局实例 + 同步写回 env用户在会话中切换后端

依据:src/backend/backend.ts:557、:566、:570。注意 configureBackendMode 会同时改 process.env,因为 settings 命名空间那套逻辑刻意仍读 env 而不是这个可变全局(理由见 src/backend/backend-mode.ts:10-12 的注释:全局变量会跨测试文件泄漏)。

4. 发出去:一次请求带了什么​

4.1 只发增量​

请求体的构造在 buildRequestBodyFromPreparedMessages(src/agent/message.ts:314),导出的纯函数版本是 buildConversationMessagesCreateRequestBody(src/agent/message.ts:296)。核心字段:

字段值为什么
messages本轮新增的 message / approval历史在服务端,不重发
streaming / stream_tokenstrue逐 token 出字
include_pingstrue服务端约 20s 一个心跳,静默即代表连接死了
backgroundtrue(默认)服务端后台跑 run,断线可 resume
client_tools本地工具定义快照告诉模型客户端有哪些工具
client_skills本地技能清单见 自我扩展
include_compaction_messagestrue压缩事件也进流,UI 能显示
agent_id仅当 conversationId === "default"默认会话没有独立 conversation 对象

最后一条有个硬校验:用 "default" 却没给 agentId 直接抛错(src/agent/message.ts:321-326)。

4.2 一个刻意的默认值:不重试​

  // Disable SDK retries by default - state management happens outside the stream,
  // so retries would violate idempotency and create race conditions
  requestOptions: SendMessageStreamRequestOptions = { maxRetries: 0 },

src/agent/message.ts:354-358 —— SDK 层重试被关死。因为状态在服务端,SDK 盲目重发同一个请求会让服务端跑两个 run。重试必须发生在更高层,由懂上下文的代码带着新 OTID 重发(见 §7)。

唯一的例外是云端主动缩容:503 + errorCode: cloud_api_shutting_down + admitted: false 才在 sendMessageStreamWithBackend 里重试最多 3 次(src/agent/message.ts:53-68、:535-593)。判据里的 admitted: false 是关键——请求没被受理,重发才安全。

4.3 给流挂上"身份证"​

流对象本身不带上下文,于是用三个 WeakMap 把元数据挂上去:

export type StreamRequestContext = {
  conversationId: string;
  resolvedConversationId: string;
  agentId: string | null;
  requestStartedAtMs: number;
  otid?: string;
};

src/agent/message.ts:103,写入点在 :608-614。其中 otid 是断线续传的钥匙:服务端能凭它精确定位"这个客户端那条消息对应哪个 run",比时间戳猜测安全得多(§6.2)。

4.4 一个小优化:response state 复用​

审批续跑时可以带上 X-Letta-Response-State 头,让服务端复用上一次的 response 缓存。但有个刻意的限制:

  const canUsePreviousResponseState =
    isApprovalContinuation && opts.allowResponseStateReuse === true;

src/agent/message.ts:433-434 —— 只有全自动处理完、人没插手的审批续跑才允许复用。理由写在紧邻注释里:人在审批框前停留时,agent/conversation 的可见状态可能被改动(比如切了模型),缓存就不再有效。所以调用点也只在"全自动"分支传 true(src/cli/app/use-conversation-loop.ts:2120-2123)。

5. 读回来:分片如何变成一个回合的结论​

5.1 三层拼装​

一条 SSE 流里的 chunk 是碎的:一个工具调用的 arguments 可能分十几个 chunk 到齐。有三处各自做累积,职责不同:

  SSE chunk
      │
      ├──► StreamProcessor.processChunk   ── 累积「协议事实」
      │      pendingApprovals / stopReason / lastRunId / lastSeqId
      │
      ├──► onChunk(buffers, chunk)        ── 累积「可渲染的行」
      │      按 tool_call_id 建行,argsText 字符串拼接
      │
      └──► DrainStreamHook (可选)         ── 让上层改写行为
             setExecutionPhase / shouldOutput / stopReason

StreamProcessor 按 tool_call_id 建条目,名字取一次、参数字符串累加:

        if (toolCall.arguments) {
          existing.toolArgs += toolCall.arguments;
        }

src/cli/helpers/stream-processor.ts:176-178。同一套逻辑在 UI 侧也来一遍,行 id 直接用 tool_call_id(src/cli/helpers/accumulator.ts:1108、:1161-1170)。

这里有个容易看漏的区分:tool_call_message 与 approval_request_message 在 UI 累积器里走同一个 case(src/cli/helpers/accumulator.ts:1090-1091),但在 StreamProcessor 里只有后者进 pendingApprovals(src/cli/helpers/stream-processor.ts:136-139 的注释说得很直白:tool_call_message 是服务端自己执行的,不需要客户端插手)。

5.2 DrainStreamHook:给上层留的口子​

export type DrainStreamHook = (ctx: DrainStreamHookContext) =>
  DrainStreamHookResult | undefined | Promise<DrainStreamHookResult | undefined>;

src/cli/helpers/stream.ts:70-75。钩子能改三件事:这块要不要显示(shouldOutput)、要不要进累积器(shouldAccumulate)、要不要直接判定 stop(stopReason)。TUI 用它做的事很轻——只是根据 chunk 类型切换"思考中/用工具中/回答中"的动画阶段(src/cli/app/use-conversation-loop.ts:150-161)。

5.3 一致性兜底:end_turn 被强改成 requires_approval​

如果流里已经吐过 approval_request_message,最后却给了 end_turn,那是服务端的不一致。drainStream 直接改判并上报遥测:

  if (stopReason === "end_turn" && approvals.length > 0) {
    ...
    stopReason = "requires_approval";
  }

src/cli/helpers/stream.ts:418-432。回合循环里还有第二道同样的兜底(src/cli/app/use-conversation-loop.ts:1395-1411)。同一个不变量守两遍,是这份代码的一个稳定风格。

6. 核心机制:approval 环路​

6.1 三条消息,一次往返​

这是本章最重要的一节。三种消息类型构成一个完整的握手:

消息类型方向含义
approval_request_message服务端 → 客户端模型想调工具 X,参数 Y,回合就此暂停
tool_return_message(type: "tool")客户端 → 服务端我跑了,结果是 Z
approval_response_message(type: "approval")客户端 → 服务端我拒绝跑,理由是 R

关键点:后两者不是"HTTP 响应",是新消息。 客户端把它们塞进一个 type: "approval" 的 ApprovalCreate 里,当作下一轮请求的 messages 发出去。所以一次工具调用 = 一次完整的服务端往返。

用户消息
   │
   ▼  POST messages:[{type:"message", role:"user", ...}]
┌────────────┐
│ 服务端 run │──► reasoning / assistant / approval_request_message
└────────────┘    stop_reason: requires_approval
   │
   ▼  客户端分类 + 本地执行
   │
   ▼  POST messages:[{type:"approval", approvals:[
   │        {type:"tool",     tool_call_id, tool_return, status},   ← 执行了
   │        {type:"approval", tool_call_id, approve:false, reason}  ← 拒绝了
   │      ]}]
┌────────────┐
│ 服务端 run │──► 继续推理…… 可能又是 requires_approval(↺ 回到上面)
└────────────┘    或者 stop_reason: end_turn(结束)

代码上,这个 ↺ 就是 processConversation 调用自己:

                await processConversation(
                  [{ type: "approval", approvals: allResults, otid: randomUUID() }],
                  { allowReentry: true, allowResponseStateReuse: true },
                );

src/cli/app/use-conversation-loop.ts:2112-2124。allowReentry: true 用来绕过并发守卫(src/cli/app/use-conversation-loop.ts:554-556),并且不重置重试计数(:595-600)——因为这仍然是同一个用户回合,重试预算要共享。

6.2 分成三堆​

拿到 approvals 后,先按权限规则分流(规则本身见 批准这件事):

            const { needsUserInput, autoAllowed, autoDenied } =
              await classifyApprovals(approvalsToProcess, { ... });

src/cli/app/use-conversation-loop.ts:1821-1829

堆处理结果消息
autoAllowed直接跑,不打扰用户type: "tool" + 真实返回
autoDenied直接拒,附权限理由type: "approval", approve: false
needsUserInput弹审批框,turn 在此挂起用户点了才有

只有 needsUserInput 为空时才走全自动续跑;否则把三堆状态灌进 UI,setStreaming(false) 并返回(src/cli/app/use-conversation-loop.ts:2203-2216)。注意此时自动放行的结果已经算好了并暂存在 state 里,等用户点完再和用户的决定拼成一个批次发出去。

6.3 批量执行:能并行的并行,会打架的排队​

executeApprovalBatch(src/agent/approval-execution.ts:372)按三种策略分组:

  decisions
     ├─ 拒绝               ──────────────► 全并行(不用真执行)
     ├─ 并行安全(只读)     ──────────────► 全并行
     └─ 写工具 ─ 按 resourceKey 分组 ──► 组间并行,组内串行
                              │
                              ├── "/abs/path/a.ts"  → Edit, Edit  (串行)
                              ├── "/abs/path/b.ts"  → Write       (与上组并行)
                              └── "__global__"      → Bash, memory(串行,且和谁都不并)

资源键的算法很短,值得记住:

export function getResourceKey(toolName, toolArgs, workingDirectory) {
  if (GLOBAL_LOCK_TOOLS.has(toolName)) return "__global__";
  if (FILE_PATH_TOOLS.has(toolName)) { /* 归一化成绝对路径 */ }
  return "__global__";  // 未知工具 / 缺 file_path → 保守取全局锁
}

src/agent/approval-execution.ts:143-166。两处保守设计值得学:Shell 类工具永远拿全局锁(它能干任何事,:114-131),认不出来的工具也拿全局锁(:165)。宁可慢,不可乱。

并行安全名单 PARALLEL_SAFE_TOOLS(src/agent/approval-execution.ts:43-85)横跨三套工具族(Anthropic / Codex / Gemini)的同义词,并且注释里明确写了 Bash 被故意排除。

结果数组是预分配再回填的(src/agent/approval-execution.ts:399-401),所以无论并发怎么调度,返回顺序都和输入决定的顺序一致——服务端按顺序对齐 tool_call_id 时不会错位。

6.4 一段最小示意​

把上面整件事剥到只剩骨架,大概是这样:

// 示意,非源码:一个用户回合 = 若干次服务端往返
async function runTurn(conversationId, userMessage) {
  let outgoing = [{ type: "message", role: "user", content: userMessage }];

  while (true) {
    const stream = await backend.createConversationMessageStream(conversationId, {
      messages: outgoing,          // 只发增量,历史在服务端
      streaming: true,
      client_tools: localToolDefs,
    });

    const { stopReason, approvals } = await drainStream(stream);  // 消费 SSE
    if (stopReason !== "requires_approval") return stopReason;    // end_turn / error

    const { autoAllowed, autoDenied, needsUserInput } = classify(approvals);
    if (needsUserInput.length > 0) return "paused";               // 交给 UI,turn 挂起

    const results = await executeApprovalBatch([...autoAllowed, ...autoDenied]);
    outgoing = [{ type: "approval", approvals: results }];        // 结果 = 下一轮的输入
  }
}

重点看最后一行:工具结果不是"返回值",是下一轮请求的消息体。真实代码里这个 while 循环在 src/cli/app/use-conversation-loop.ts:737,只是掺了重试、队列、模式钉住等十几种旁路。

7. 为什么关掉重开还能恢复​

7.1 恢复靠的不是本地文件​

因为未决审批本来就在服务端,恢复动作只是去问一遍。入口是 getResumeDataFromBackend(src/agent/check-approval.ts:513),返回三样东西:未决审批、可回放的历史、conversation 对象。

真相源写在函数文档注释里:

The source of truth for pending approvals is conversation.in_context_message_ids. We anchor our message fetch to that, not arbitrary recent cursor messages.

src/agent/check-approval.ts:506-507。取 in_context_message_ids 的最后一个 id当锚点(:553),再从消息尾巴里挑出属于这个 id 的所有变体(一条消息在 API 里可能拆成 :assistant: / :reasoning: / :tool: 多个变体,靠 sourceMessageIdFromVariant 归并,:222-227)。

7.2 去重:哪些审批其实已经结了​

光看 approval_request_message 会把已经处理完的也算成未决。所以先扫一遍已完成的 tool_call_id 集合:

function completedToolCallIdsFromMessages(messages: Message[]): Set<string> {

src/agent/check-approval.ts:149-180 —— 两个来源都收:tool_return_message(含 tool_returns[] 数组形态)和 approval_response_message(含 approvals[])。然后过滤:

      if (completedToolCallIds.has(approval.toolCallId)) continue;

src/agent/check-approval.ts:202。再用一个 Map<toolCallId, ApprovalRequest> 去重同 id 的重复请求(:203-206)。

7.3 backfill:让重开后的屏幕不是空的​

恢复审批只需要一条消息,但用户重开 CLI 时希望看见上一段对话。于是同一次抓取顺带做 backfill,要拉的类型列成常量:

const RESUME_BACKFILL_MESSAGE_TYPES: MessageType[] = [
  "user_message", "assistant_message", "reasoning_message",
  "event_message", "summary_message",
  "approval_request_message", "tool_return_message", "approval_response_message",
];

src/agent/check-approval.ts:25-34

裁剪逻辑在 prepareMessageHistory(src/agent/check-approval.ts:313),思路是"按主消息数倒着数,而不是按原始条数":

常量值作用
BACKFILL_PAGE_LIMIT50向服务端要 50 条(拉得比渲染的多)
BACKFILL_PRIMARY_MESSAGE_LIMIT10倒数够 10 条"主消息"(user/assistant/reasoning/event/summary)就停
BACKFILL_MAX_RENDERABLE_MESSAGES40安全上限

src/agent/check-approval.ts:17-22。为什么拉 50 渲染 10?注释写了:工具密集的回合会把最后一条用户可见的 assistant 消息挤出窗口(:20-21)。最后还要把开头孤儿的 tool_return_message 削掉——裁剪可能砍在一个回合中间,留下没有对应调用的返回(:368-371)。

顺带一提,消息顺序不能信 API:sortChronological(src/agent/check-approval.ts:380)先按 date 排,同一毫秒的再按类型等级排(user=0 → reasoning=1 → call=2 → return=3 → assistant=4),保证重放出来的顺序符合直觉。

7.4 启动时怎么串起来​

letta 启动
   │
   ▼  src/index.ts:2512  getResumeDataFromBackend(agent, "default")
   │
   ▼  src/index.ts:2707  startupApprovals = resumeData.pendingApprovals
   │
   ▼  src/cli/app/use-approval-flow.ts:338  useEffect
   │     if (loadingState === "ready" && approvals.length > 0)
   │        → recoverRestoredPendingApprovals(approvals)
   ▼
 审批框重新弹出

--resume 指定会话、切换会话、fork 会话走的是同一个函数的不同调用点(src/cli/app/AppView.tsx:1217、src/cli/app/use-conversation-switching.ts:367、src/cli/app/AppCoordinator.tsx:4527)。一个恢复函数,所有入口共用。

8. 中断与重试:四层各管一段​

这四层容易混,先用表分清:

层触发谁处理动作
① 用户中断Ctrl-C / EschandleInterrupt客户端先急停,再异步通知服务端取消
② 流中途断SSE 连接掉,没收到 stop_reasondrainStreamWithResume用 run_id / OTID 从断点续读
③ 回合后错误stop_reason = error / llm_api_errorisRetriableError + 循环 continue换新 OTID 重发,起一个新 run
④ 发送前冲突POST 直接 409 / 5xxgetPreStreamErrorAction四选一:解审批 / 等忙 / 退避重试 / 抛出

8.1 ① 客户端先停,服务端后停​

handleInterrupt(src/cli/app/use-interrupt-handler.ts:123)在 EAGER_CANCEL 打开时(常量恒为 true,src/cli/app/constants.ts:14)采取"先斩后奏"策略:

1. buffers.abortGeneration++        ← 让还在跑的 drainStream 立刻发现自己过期
2. markIncompleteToolsAsCancelled   ← UI 上把未完成的工具标成"已中断"
3. abortController.abort()          ← 掐掉 HTTP
4. conversationGenerationRef++      ← 让在途的 processConversation 变"陈旧"并静默退出
5. setStreaming(false) / 刷新 UI    ← 用户立刻看到反应
6. 然后才 fire-and-forget:
     getBackend().cancelConversation(...)     ← 通知服务端

依据:src/cli/app/use-interrupt-handler.ts:238-286、:202-215。

第 6 步不能省,注释解释得很清楚:不发这个取消,服务端会一直停在 requires_approval,导致下一条用户消息直接 409 冲突(src/cli/app/use-interrupt-handler.ts:199-201)。这就是 §8.4 里 approval_pending 那条恢复路径要处理的场景。

"陈旧代"这个机制在 drainStream 里也有对应检查——每个 chunk 都比一次 abortGeneration(src/cli/helpers/stream.ts:175-180),变了就当 cancelled 退出。

8.2 ② 断线续读​

drainStreamWithResume(src/cli/helpers/stream.ts:513)先正常 drain 一次,失败了再判断能不能续:

  const canResume =
    result.stopReason === "error" &&
    (!result.sawStopReasonChunk || replayGenericError) &&
    !isApprovalPendingConflict &&
    (runIdToResume || runIdSource === "otid") &&
    abortSignal && !abortSignal.aborted;

src/cli/helpers/stream.ts:639-645

逐条读:收到过终态 stop_reason 就别续(那是服务端正常结束,不是掉线);审批冲突不算掉线(交给审批恢复路径);用户已取消就别续。

拿不到 run_id 时的两条路,优先级有讲究:

run_id 从哪来?
  ├─ 流里直接带了      → source = "stream_chunk"   (最好)
  ├─ 有 OTID          → source = "otid"           (次好:服务端按 OTID 精确定位 run)
  └─ 都没有           → 时间戳猜测 discoverFallbackRunIdWithTimeout (兜底)

src/cli/helpers/stream.ts:558-609。注释点明了为什么偏爱 OTID:多客户端场景下时间戳启发式不安全(:554-557)。

续读成功后还有一道细活。恢复流用的是全新的 StreamProcessor,不会重放断点之前的 approval chunk,所以要把原始的 approvals 合回来,分两种情况:

情况现象处理
审批 chunk 全在断点前恢复流一个 approval 都没有整体沿用原来的
参数被断点劈成两半两边都有同一个 tool_call_idorigArgs + resumeArgs 字符串拼接

src/cli/helpers/stream.ts:804-826。第二种情况的处理就是一行 toolArgs: (orig.toolArgs \|\| "") + (resumeApproval.toolArgs \|\| "")(:822)——JSON 参数是纯字符串累加的,所以断点续接才这么简单。这是 §5.1 那个设计决定的回报。

8.3 ③ 回合后重试​

判定入口很薄:

export async function isRetriableError(stopReason, lastRunId, fallbackDetail)

src/cli/app/retry.ts:6 —— 先按 stop_reason 快速排除 8 种绝不重试的(cancelled / requires_approval / max_steps / end_turn / tool_rule 等,:12-22),再去拉 run 的 metadata.error 做细分类,最后交给纯函数 shouldRetryPostStreamRunError(src/agent/turn-recovery-policy.ts:222)。

turn-recovery-policy.ts 整个文件是没有网络、没有 React、没有 IO 的纯策略,文件头注释说明了原因:TUI 和 headless 共用,保证同样的冲突输入产生同样的恢复动作(src/agent/turn-recovery-policy.ts:1-7)。它维护四张模式表:

表例子结论
RETRYABLE_PROVIDER_DETAIL_PATTERNS"overloaded"、"socket hang up"、"fetch failed"重试
NON_RETRYABLE_PROVIDER_DETAIL_PATTERNS"invalid api key"、"context_length_exceeded"不重试
NON_RETRYABLE_RUN_ERROR_TYPESllm_authentication、llm_insufficient_credits不重试
NON_RETRYABLE_429_REASONS"exceeded-quota"、"not-enough-credits"不重试(429 不等于可重试)

src/agent/turn-recovery-policy.ts:26-101。退避策略按类别分三种(getRetryDelayMs,:323):瞬时错误指数退避(Cloudflare 52x 基数 5s,其余 1s,有 Retry-After 就听它的)、会话忙 10s 指数、空响应 500ms 线性。

真正重试时有个后端相关的分叉:

              currentInput = retryFromPersistedLocalState
                ? []
                : refreshInputOtidsForNewRequest(currentInput);

src/cli/app/use-conversation-loop.ts:2601-2603 —— 本地后端在失败的 run 之前就把输入落盘了,重发会造成重复;云端则要换一批新 OTID 重发(refreshInputOtidsForNewRequest,src/agent/turn-recovery-policy.ts:482)。OTID 必须换,因为这是一个新 run,复用旧 OTID 会被当成同一条消息。

8.4 ④ 发送前的 409​

POST 还没出流就报错时,getPreStreamErrorAction(src/agent/turn-recovery-policy.ts:391)给出四选一:

动作触发条件后续
resolve_approval_pending详情含 "waiting for approval"去服务端捞真实的未决审批,重建输入
retry_conversation_busy含 "is currently being processed" 等,且预算未耗尽10s 指数退避后重发
retry_transient429 / 5xx / 可重试模式,且预算未耗尽退避重发
rethrow其余抛给外层 catch 报错

第一条是解决 §8.1 埋下的那个坑。修复手段是用一段自白式的理由把陈旧审批关掉:

export const STALE_APPROVAL_RECOVERY_DENIAL_REASON =
  "The agent harness automatically closed this stale pending tool call to recover
   from a client/server state desync: ... It was not denied by the user or a
   permissions policy. Re-issue the tool call if you still need it.";

src/agent/turn-recovery-policy.ts:464-465(为可读性折行)。这段话是写给模型看的——明确告诉它"这不是用户拒绝你,是握手错位,需要就重发"。然后 rebuildInputWithFreshDenials(:495)把旧 approval 载荷剥掉、把新的拒绝拼到最前面。用提示词修状态机,是这份代码里挺聪明的一手。

9. 两种后端:同一套环路,两个世界​

9.1 分工对照​

维度APIBackendLocalBackend
定义位置src/backend/backend.ts:305src/backend/local/local-backend.ts:245
每个方法在干嘛转调 SDK 客户端,几乎零逻辑自己跑完整回合
谁调 LLM云端本进程(PiStreamAdapter)
状态存哪云端本机磁盘(LocalStore)
上下文压缩云端做自己做(§9.4)
系统提示编译云端做自己做,带 memfs 修订号缓存
记忆文件系统远端本机 git 仓(见 记忆系统)

APIBackend 的方法体几乎都是三行(拿 client、转调、返回),典型如 src/backend/backend.ts:485-492。唯一有真逻辑的是 getConversationResumeTail(:442):默认会话走 listAgentMessages,具名会话并发拉 conversation + 消息页(:454-460)。还有一处边界修正值得注意——分页游标的语义翻译:

The Backend contract uses chronological cursors: before always means older and after always means newer. The API interprets them relative to sort order, so descending requests need their cursor keys swapped at this boundary.

src/backend/backend.ts:108-110,实现在 :111-115。在抽象边界上把外部的怪异语义修正掉,这样上层不必知道 API 的排序习惯。

9.2 LocalBackend 的继承链​

Backend (interface)
   └── HeadlessBackend  ← src/backend/dev/headless-backend.ts:2 只是重导出
          │              实体在 fake-headless-backend.ts:184
          │              管:store 读写、run 生命周期、chunk 落盘与重放
          └── LocalBackend  ← 覆写模型目录、系统提示编译、压缩、memfs

src/backend/dev/headless-backend.ts 只有两行,是一个刻意的重导出别名。真正的基类 HeadlessBackend 提供了"有状态后端"的通用骨架,LocalBackend 只补本地特有的部分。

9.3 本地一个回合怎么跑​

executeConversationTurn(src/backend/dev/fake-headless-backend.ts:441)的顺序:

1. 结算上一轮遗留的孤儿工具调用 settleInterruptedToolCalls
     ↑ 但如果本轮输入是 approval,跳过 —— 那个 pending 是故意留的,正等着这轮的结果
2. appendTurnInput            把本轮输入落盘
3. startRun                   建 run 记录
4. listConversationMessages / listLocalMessages   取历史
5. resolveSystemPromptForTurn 编系统提示(LocalBackend 覆写)
6. executor.execute(...)      交给 ProviderTurnExecutor 真的调模型
7. persistExecutorStream      边吐边落盘 + 边录制

第 1 步的例外判断是这段代码里最精细的一处:

    if (!isApprovalTurn) {
      this.store.settleInterruptedToolCalls(conversationId, { reason: TURN_DID_NOT_COMPLETE });
    }

src/backend/dev/fake-headless-backend.ts:455-462。理由在注释里:进程被 kill 后会留下没有结果的 tool_use 块,Anthropic 会拒绝含它们的后续请求,所以要补合成的错误结果;但审批回合的 pending 是有意为之,不能一起清掉。

9.4 本地的"有状态"是怎么做到的​

persistExecutorStream(src/backend/dev/fake-headless-backend.ts:653)是一个包装迭代器,每个 chunk 做三件事:落盘、录制、透传。

provider chunk
   ├─► store.appendStreamChunk(...)      ← 落盘:这就是本地版的"服务端状态"
   ├─► backend.recordRunChunk(runId,...) ← 录制:塞进 runChunksByRunId,带 seq_id
   └─► yield                             ← 透传给 CLI

src/backend/dev/fake-headless-backend.ts:683-693。录制的那份让 streamRunMessages(runId, { starting_after }) 能按 seq_id 过滤重放(:421-434)——§8.2 的断线续读在本地后端一样能用。

流结束若没见 stop_reason,按见过什么补一个:见过 approval 就补 requires_approval(:702-707),有错误信息就补 error(:696-701),都没有就当协议错误(:708-716)。

9.5 本地怎么把模型输出翻成 Letta 协议​

createProviderLettaStream(src/backend/dev/provider-turn-executor.ts:398)是本章的收束点——它把 pi-ai 的事件翻译成和云端一模一样的 chunk:

pi-ai 事件翻成
text_deltaassistant_message
thinking_deltareasoning_message
toolcall_endapproval_request_message
done(有工具调用 / toolUse)stop_reason: requires_approval
done(length)stop_reason: max_tokens_exceeded
done(其他)stop_reason: end_turn

src/backend/dev/provider-turn-executor.ts:429-493。所以 §6 那整套 approval 环路在本地后端一字不改地成立——CLI 根本不知道自己在跟谁说话。这就是 Backend 抽象真正的价值所在。

一个细节:多段 text 会被归并到同一个 OTID。otidForContentSegment(:379)先向前找同类型内容的连续起点(contiguousContentStartIndex,:364),同一段共用一个 OTID,UI 才不会把一段话拆成几行。

9.6 本地自己做上下文压缩​

云端后端不管这事;本地必须自己算、自己压。PiStreamAdapter.stream(src/backend/dev/pi-stream-adapter.ts:811)是一个带五种恢复手段的循环:

  ┌─► ① 预检压缩(每回合一次)
  │      估算 contextTokens > 阈值? → 压缩 → 重来
  │
  ├─► streamOnce(调 provider)
  │      成功 ──────────────────────────────► 返回
  │      失败 ↓
  │   ② provider 报 overflow  → 压缩(最多 3 次)→ 重来
  │   ③ 可重试传输错误 + 载荷超大 → 剔图片 / 压缩 → 重来
  │   ④ 重试若干次仍失败      → 自适应剔图片    → 重来
  └─  ⑤ 其余可重试错误       → 退避后重试(最多 3 次)

依据 src/backend/dev/pi-stream-adapter.ts:817-966。

第 ① 步存在的理由写在 contextCompactionThreshold 的长注释里(src/backend/dev/provider-turn-executor.ts:223-239):pi-ai 会把超量请求的输出配额压到 1 token 来让它"合法",于是一个快满的请求会以 length 正常结束,而不是抛 overflow ——那样后置的 overflow 恢复路径根本抓不到。所以必须在发请求之前就判断:

  const reserveTokens = Math.min(
    LOCAL_CONTEXT_COMPACTION_RESERVE_TOKENS,                    // 16384
    Math.max(1, Math.floor(contextWindow * 0.2)),               // 小窗口取 20%
  );
  return Math.max(0, contextWindow - reserveTokens);

src/backend/dev/provider-turn-executor.ts:251-258

压多少,由 LocalBackend 侧的两种模式决定(src/backend/local/local-backend.ts:743 起):

模式做法默认
sliding_window从头 evict 一批,摘要放前面,保留近期消息✔(30%)
all全部摘要,只留最后一条挂着工具调用的消息降级用

src/backend/local/compaction.ts:42-43(默认值)、:575(滑窗计划)、:643(全量计划)。滑窗从 30% 起试,不够就每次 +10% 直到保留部分低于目标 token 数(:601-622);切点必须落在 assistant 消息上(isValidSlidingWindowCutoff,:564),避免把一个工具调用和它的返回劈开。滑窗规划失败或压完还超窗,就降级到 all(src/backend/local/local-backend.ts:784-826)。

token 怎么估?不是数 JSON 长度,而是以最后一条带 usage 的 assistant 消息为锚,加上它之后所有消息的估算:

  return { tokens: usageTokens + trailingTokens, usageTokens, trailingTokens, lastUsageIndex };

src/backend/local/local-context-estimate.ts:188-193。文件头列了三条相对 pi-mono 的刻意偏离(:8-22),其中第 3 条是关键:压缩边界时间戳之前的 usage 反映的是旧的(更大的)上下文,必须当作过期忽略(:145-148),否则刚压缩完的会话会立刻被判定成又满了。

压缩过程会往流里插两个 chunk(event_message: compaction + summary_message,src/backend/dev/pi-stream-adapter.ts:554-569),于是 UI 上会显示"已压缩"——和云端的表现一致。

10. 巧妙之处(可以带走的)​

  • 协议里的"结果"就是"下一条消息"。 工具返回不走响应通道,而是构造成 type: "approval" 的新消息。代价是多一次往返,收益是回合可中断、可持久、可恢复,而且客户端和服务端不用维护长连接语义。(src/cli/app/use-conversation-loop.ts:2112)

  • 参数用字符串累加,而不是 JSON 增量合并。 正因为 toolArgs 只是 +=,断线续传时把两半直接拼起来就还原了完整参数(src/cli/helpers/stream.ts:822)。选一个可结合的累积表示,恢复逻辑就免费了。

  • 不可信的东西一律拿全局锁。 未知工具、缺 file_path、Shell 类工具,全部映射到 "__global__"(src/agent/approval-execution.ts:149-165)。默认串行、显式并行,而不是反过来。

  • 恢复策略做成零依赖的纯函数。 turn-recovery-policy.ts 全文没有网络和 UI,TUI 与 headless 共用,保证"同样的冲突 → 同样的动作"(src/agent/turn-recovery-policy.ts:1-7)。

  • 用提示词修状态机。 客户端/服务端审批错位时,自动发一条写给模型看的拒绝理由,说明"这不是人拒绝你,需要就重发"(src/agent/turn-recovery-policy.ts:464)。

  • 同一个不变量守两遍。 "有 approval 就不可能是 end_turn" 在 drainStream(src/cli/helpers/stream.ts:418)和回合循环(src/cli/app/use-conversation-loop.ts:1395)各兜一次。协议边界上的偏执是划算的。

  • 先斩后奏的中断。 Ctrl-C 先把客户端全停、UI 立刻响应(src/cli/app/use-interrupt-handler.ts:238-286),取消请求 fire-and-forget 发出去(:202-215)。用户感知的延迟是 0,一致性靠后续的审批恢复路径补齐。

11. 边界与局限​

  • 多一次往返就是多一次延迟。 每个需要审批的工具都要一整轮 HTTP。allowResponseStateReuse 只能省掉服务端的部分重算(src/agent/message.ts:433),省不掉往返本身。

  • 恢复的粒度是"未决审批 + 最近若干条",不是完整会话。 backfill 上限 40 条渲染消息(src/agent/check-approval.ts:18),更早的历史在服务端但不会重放到屏幕上。

  • run 发现有猜的成分。 拿不到 run_id 又没有 OTID 时,只能退回时间戳启发式(src/cli/helpers/stream.ts:574-592),代码注释自己承认多客户端场景下这不安全。

  • 本地上下文估算是估算。 无 usage 锚点时按 字符数 / 4 算,图片按固定 1200 token 记(src/backend/local/local-context-estimate.ts:25)。压缩阈值因此是保守而非精确的。

  • 本地滑窗压缩会放弃。 消息少于 4 条、或找不到合法的 assistant 切点,就抛 LocalSlidingWindowCompactionPlanningError 并降级到全量压缩(src/backend/local/compaction.ts:579-583、:624-634)。

  • useConversationLoop 2871 行。 主线清晰,但被队列、模式钉住、模型自动切换、trajectory 统计等旁路缠绕。读的时候建议只顺着 while (true) → stopReasonToHandle 的分支走,别被旁路带偏。

12. 横向对比​

同 shelf 的兄弟项目多数是无状态客户端:会话数组在进程内,审批停在一次函数调用里,进程死了就得靠本地 session 文件重放。Letta Code 把状态推到服务端,换来的差异是:

关切无状态客户端典型做法Letta Code
审批暂停一个 await 卡在函数里一次 stop_reason: requires_approval,回合真的结束了
进程重启读本地 session 文件重放向服务端要 in_context_message_ids
上下文压缩客户端做云端做;本地后端才自己做
断线重发整轮按 run_id/seq_id 续读

代价是协议复杂度和往返次数;收益是同一个 conversation 可以被多个入口(CLI / headless / 远程环境 / 消息渠道)接管——这一点在 多入口与常驻 里展开。

13. 代码地图​

主题文件符号
后端接口与能力位src/backend/backend.tsBackend、BackendCapabilities
云端实现src/backend/backend.tsAPIBackend、getConversationResumeTail
单例与模式切换src/backend/backend.tsgetBackend、getBackendForMode、configureBackendMode
模式解析src/backend/backend-mode.tsresolveBackendMode、isExperimentalLocalBackendEnabled
发送与请求体src/agent/message.tssendMessageStream、sendMessageStreamWithBackend、buildConversationMessagesCreateRequestBody
流元数据src/agent/message.tsStreamRequestContext、getStreamRequestContext
流消费src/cli/helpers/stream.tsdrainStream、DrainStreamHook、DrainResult
断线续读src/cli/helpers/stream.tsdrainStreamWithResume
协议累积src/cli/helpers/stream-processor.tsStreamProcessor、pendingApprovals
UI 分片拼装src/cli/helpers/accumulator.tsonChunk、markIncompleteToolsAsCancelled、setToolCallsRunning
回合编排src/cli/app/use-conversation-loop.tsuseConversationLoop、processConversation
审批批量执行src/agent/approval-execution.tsexecuteApprovalBatch、executeAutoAllowedTools、getResourceKey、ApprovalDecision
恢复与 backfillsrc/agent/check-approval.tsgetResumeDataFromBackend、extractApprovals、prepareMessageHistory、RESUME_BACKFILL_MESSAGE_TYPES
恢复策略(纯函数)src/agent/turn-recovery-policy.tsgetPreStreamErrorAction、getRetryDelayMs、rebuildInputWithFreshDenials、STALE_APPROVAL_RECOVERY_DENIAL_REASON
重试判定src/cli/app/retry.tsisRetriableError
中断src/cli/app/use-interrupt-handler.tshandleInterrupt
启动恢复接入src/cli/app/use-approval-flow.tsrecoverRestoredPendingApprovals
本地后端src/backend/local/local-backend.tsLocalBackend、compactLocalConversation、resolveSystemPromptForTurn
有状态骨架src/backend/dev/fake-headless-backend.tsHeadlessBackend、executeConversationTurn、persistExecutorStream
协议翻译src/backend/dev/provider-turn-executor.tsProviderTurnExecutor、shouldCompactForContextPressure、contextCompactionThreshold
provider 循环src/backend/dev/pi-stream-adapter.tsPiStreamAdapter、compactBeforeProviderCall、streamOnce
压缩计划src/backend/local/compaction.tsplanLocalSlidingWindowCompaction、planLocalAllCompaction、packageLocalSummaryMessage
上下文估算src/backend/local/local-context-estimate.tsestimateLocalContextTokens、contextTokensFromLocalUsage

相邻章节: 架构与原理总览 · 本地工具层 · 批准这件事 · 记忆系统 · 自我扩展 · 多入口与常驻