一个 Agent 正在调用模型、执行脚本、生成文件。用户刷新页面，或者地铁里网络断了几秒：任务应该停止，还是继续？重新打开页面之后，应该重新执行，还是接着看原来的任务？

这看起来像一个流式接口问题，实际决定了 Agent 的执行生命周期由谁拥有。

我的项目把一轮工作交给后台 Turn 管理，HTTP/SSE 连接只订阅它的输出。因此，浏览器断连只结束订阅，任务可以继续执行；用户明确点击停止，才请求取消任务。

但这套设计有一个必须讲清楚的边界：它实现了单进程内的执行与连接解耦，并没有因此获得任意任务的跨进程恢复能力。 最近的事件可以补播，已经落库的消息可以重读，特定的用户输入等待点可以重建执行；这些能力各有条件。

这篇文章只讨论这一个技术点：从“连接承载任务”走向“连接观察任务”，实现上需要解决什么，又能承诺到哪一步。

本文项目代码基线为 4da2c8adc9a1319e39b1d9f058c5ec0d7add618e。外部资料核对日期为 2026 年 11 月 13 日。厂商部分描述公开接口契约；涉及设计动机时会明确标为分析，不推测内部基础设施。

1. 为什么这是 Harness 的问题

这里的 harness，可以理解为模型外面负责组织和约束执行的运行系统。模型输出下一步行动，harness 负责调用工具、保存结果、管理取消，并决定连接消失之后谁继续承担这轮工作的责任。

先看一种常见实现，下面是问题示意，并非项目当前代码：

1
2
3
4
5
6

func handleChat(w http.ResponseWriter, r *http.Request) {
    // 模型调用、工具执行都继承这次 HTTP 请求的生命周期。
    runAgent(r.Context(), func(event Event) {
        writeSSE(w, event)
    })
}

在 Go 服务端，请求的 Context 会在客户端连接关闭、HTTP/2 请求被取消或 ServeHTTP 返回时取消。如果模型请求、工具执行、数据库操作都沿用它，取消就可能一路传播下去。Go：Request.Context

对于一次普通查询，这通常合理；对于已经开始产生外部副作用的 Agent 工作流，需要重新选择语义。例如，脚本已经写好了文件，网络随后中断。如果前端把“没收到结果”理解成“没有执行”，再提交一次，可能造成重复工作。

即使改成 WebSocket，这个问题也仍然存在。传输协议决定如何交换数据，执行所有权决定连接断开后由谁继续工作。

因此，先区分四种保证：

“支持断线重连”没有说明后三项的答案。接口设计、测试和面试表达，都应该把这四种保证分开。

2. 当前项目：Turn 拥有执行，连接拥有订阅

项目里有三个不同寿命的对象：Session 保存多轮会话，Turn 表示其中一轮后台执行，SSE 订阅表示某个客户端正在观察它。同一个 Turn 可以经历多次订阅，也可以暂时没有订阅者。

核心入口是 TurnRegistry.Start。下面节选关键结构，省略了部分初始化与删除保护代码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

r.mu.Lock()
defer r.mu.Unlock()

if t, ok := r.turns[sessionID]; ok && t.Running() {
    return nil, ErrTurnRunning
}

ctx, cancel := context.WithCancel(context.Background())
t := &Turn{
    sessionID: sessionID,
    streamID:  uuid.NewString(),
    cancel:    cancel,
    wake:      make(chan struct{}),
}
r.turns[sessionID] = t
go t.pump(ctx, cancel, run)

这里包含两个设计决定。

第一，任务 Context 不从 HTTP 请求派生。 Start 创建独立 Context，把取消句柄交给 Turn 管理。API 层随后显式注入数据库依赖。HTTP 请求自己的 Context 只传给 Follow，用于终止本次订阅。

这也意味着，独立 Context 不能自动继承请求中的所有值。认证、租户权限需要在入口完成校验，后台需要的依赖与执行身份要明确传递。未来如果增加追踪信息，也应显式传入，而不是把整个请求生命周期重新绑回来。

第二，同一个 Session 同时只允许一个正在运行的 Turn。 检查和登记都发生在注册表锁内，避免两个并发请求同时通过检查。历史加载也在被接纳的后台工作里执行，减少两轮同时读写同一份历史的机会。

为什么当前选择拒绝，而不是自动排队？从实现取舍看，排队还需要定义用户后续消息究竟是“下一轮输入”“修改当前任务”还是“取消后的替代任务”，并处理排队期间的历史变化。先限制单会话并发，能把时序语义保持清楚。代价是暂不支持同一会话同时运行多个独立 Turn；同一 Turn 内仍可以有并发工具或子 Agent。

需要注意，这把锁只保护本进程。部署两个实例，每个实例各有一张 turns 表，并不会自然形成全局互斥。

3. 补播协议：游标为什么需要两个字段

任务开始之后，事件写入 Turn 的内存缓冲，再由订阅者读取。当前默认保留最近 3000 个事件。

首次发送消息走 POST；断线后重新订阅走 GET：

1
2
3
4
5

POST /api/sessions/{session_id}/messages

GET /api/sessions/{session_id}/stream
    ?from_seq=43
    &stream_id=本轮事件流的UUID

假设客户端已经处理了第 42 号事件，就从 43 开始继续。项目的 from_seq 是“下一条要读取的序号”，不是“最后读到的序号”。事件从 0 开始编号，SSE 帧大致如下：

1
2

id: 42
data: {"type":"delta","content":"继续生成的内容","stream_id":"本轮UUID"}

单独一个 seq 不够。同一 Session 的下一轮，以及输入等待点恢复后新建的 worker，都会产生新的事件流；如果还拿旧流的 200 号游标去读新流，可能跳过新输出。

因此，真正的游标是：

1

(stream_id, seq)

stream_id 标识一轮具体的事件流，seq 才在该流内部表示位置。服务端发现流身份变化时把读取位置重置为 0；输入等待点的恢复还会发出 stream_reset，让前端重建临时展示。

只重接订阅，不自动重发任务

前端 followStream 的自动重试只重新 GET 订阅，不重新发送创建、编辑或重新生成消息的 POST。

这是为了处理一个很现实的状态：客户端没有收到确认，但服务端可能已经接纳了任务。 此时重复创建，可能重复调用模型和工具。

当前还有一个保守分支：首次 POST 尚未确认 SSE 已建立就断开时，前端不会盲目接入 Session 上残留的旧 Turn，而是报告连接中断。这减少了误接上一轮的风险，但不能提供“提交一定成功或一定失败”的结论。

这个缺口需要任务提交幂等协议来进一步解决，例如给一次逻辑提交保存稳定的 request_id，由服务端返回同一个执行对象。当前普通消息创建接口还没有这样的幂等键，不能把“不自动重发”写成“已经实现 exactly-once”。

游标什么时候前移

网络分块不等于 SSE 事件边界。id: 和 data: 可能落在两次读取中，JSON 也可能只收到一半。

当前解析器保留跨分块的 pendingSeq；只有 JSON 成功解析、事件回调执行之后，才更新序号。因此，“读到了 id 行”不等于“已消费这个事件”。这里的消费确认属于前端本地逻辑，不是浏览器持久存储成功的事务确认。

4. 最值得推敲的并发细节：历史与实时之间不能有空窗

实现补播时，很容易写出这样的逻辑：

1
2
3

1. 读取历史，发现已经追平。
2. 解锁。
3. 注册通知，等待下一条事件。

如果生产者恰好在第 2 步和第 3 步之间写入事件并发出通知，订阅者可能错过通知。数据已经在缓冲里，订阅者却还在等待。

项目用同一把 t.mu 协调“检查事件”和“取得通知通道”。下面是 Follow 的核心流程，省略了旧游标钳位与心跳：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

for {
    t.mu.Lock()
    if i < t.dropped+len(t.events) {
        ev := t.events[i-t.dropped]
        t.mu.Unlock()
        onEvent(i, ev) // 网络写发生在锁外
        i++
        continue
    }
    finished := !t.finishedAt.IsZero()
    wake := t.wake // 与“已追平”的判断处于同一个临界区
    t.mu.Unlock()

    if finished {
        return
    }
    select {
    case <-ctx.Done():
        return
    case <-wake:
    }
}

生产者在同一把锁内追加事件，随后关闭旧通道，并创建新的通道：

1
2
3
4
5

// 持有 t.mu 时调用
func (t *Turn) wakeLocked() {
    close(t.wake)
    t.wake = make(chan struct{})
}

可以按生产者发生的时间来证明：

1. 生产者先获得锁。 新事件已写入，订阅者随后检查缓冲，就会看到未读数据。
2. 订阅者先获得锁。 它在确认已追平的同时取得旧 wake。生产者随后关闭这个通道；即使订阅者还没进入 select，读取已关闭的通道也会立即就绪。

这里的 channel 只表示“状态发生变化”，不承载事件内容，也不要求一个通知对应一个事件。唤醒后重新检查缓冲即可。关闭通道能唤醒所有正在等待的订阅者，每个订阅者维护自己的读取位置。

这和让多个订阅者从一个 chan Event 竞争消费不同：后者通常会把事件分给其中一个读者，无法直接实现每个页面都看到同一份输出。

另外，onEvent 在锁外执行。慢客户端的网络写不会长期占住生产者的锁；生产者只负责追加和唤醒。代价是慢客户端可能落后到保留窗口之外。

这个证明说明的是“不丢唤醒”，没有证明“不丢历史数据”。 两者分别由并发协议和保留策略决定。

5. 有界缓冲的代价：可以补播多久

当前缓冲使用切片追加，超过上限就移出最旧事件，并增加 dropped。源码注释中出现过“环形缓冲”的称呼，但实现并不是预分配数组上的循环下标；更准确的称呼是“有界滑动事件缓冲”。

其序号关系很简单：

1
2

events[i] 的真实序号 = dropped + i
下一条事件的序号    = dropped + len(events)

当客户端的 from_seq 小于 dropped 时，当前实现直接把它钳到最早可用位置。它没有额外发送 gap 事件，也没有在这一步自动返回完整快照。 因此，不能承诺任意长时间断线后都能还原完整的流式过程。

3000 是事件数，不是字节数，也不是固定时长。用一个仅供理解的算例：如果平均每秒产生 30 个事件，3000 条约覆盖 100 秒；如果变成每秒 300 个事件，就只覆盖约 10 秒。这不是项目实测吞吐，真实窗口取决于事件产生速度与负载。

内存成本也取决于载荷。一个很小的文本 delta 和一个很大的工具结果，都各算一个事件。即使条数相同，字节占用也可能差很多。

项目还定义了完成后 5 分钟的清理阈值。但清理只在下一次 Start 时顺带执行；同一 Session 的新 Turn 也会覆盖旧 Turn。因此不能把它描述成“每轮保证保留 5 分钟”或“到了 5 分钟一定释放”。

这套选择的收益是实现简单、无需额外事件存储，适合当前单实例运行。后续如果需要明确的断线服务保证，可以增加以下协议，这些属于改进方向，尚不是现有能力：

• 返回 earliest_seq，检测到窗口缺口时显式通知客户端重建视图。
• 以已保存消息重建快照，再接入实时流，并定义去重规则。
• 同时设置事件数量和字节预算；需要长时间补播时使用持久事件存储。

引入快照之后，也必须解决上一节的历史与订阅空窗，不能简单地“先 GET 历史，再连 SSE”。

6. 事件展示、持久化与取消，分别承诺什么

用户看到结果，不等于结果已经提交数据库

当前项目的模型 delta 即时发送，普通工具执行后的顺序也大致是：

1
2
3
4
5
6
7

工具执行完成
    ↓
emit(tool_call_result / tool_call_error)
    ↓
广播工件信息
    ↓
saveMessage(...)

也就是说，SSE 是实时展示流，不是提交之后才发布的持久化事件日志。进程如果在 emit 和 saveMessage 之间崩溃，用户可能已经看到结果，但数据库还没有对应记录。

前端在轮次收尾后会重新获取 Session，以已保存消息替换临时流式状态。这是一种展示收敛机制，能够校准已经落库的内容；它不能补回从未成功写入数据库的事实。

“持久化后再发布”可以缩小这一窗口，但也会把存储延迟带入展示链路。对逐 token 增量和对业务结果，不一定需要采用完全相同的可靠性级别。厂商方案中也能看到这种分层。

停止是取消请求，结束标记也不等于成功

浏览器断开，只取消 Follow。显式停止接口则先处理等待输入的停止事实，再调用 Abort 取消 Turn，并通知等待者。

Abort 是协作式取消：worker、模型调用和工具需要响应 Context。返回“停止请求已接受”，不代表所有子过程已经退出，更不代表已经发生的文件写入或外部 API 调用被回滚。

项目的 saveMessage 使用 context.WithoutCancel，避免收尾落库立刻继承任务取消。Go 对这个函数的定义是移除父 Context 的取消与截止时间约束，同时保留其值；它不会让代码跨进程死亡继续运行。Go：WithoutCancel

当前这条持久化路径没有再附加独立超时。后续可考虑在脱离执行取消之后，设置一个有限的收尾时间，并记录失败，这样才能兼顾“尽量保存”和“不会无限等待”。

pump 还会在错误或 panic 路径补齐终止事件，并设置完成时间。这里的 done 表示流已经收尾；失败路径也可以出现 done。前端判断成功时必须结合错误与结果，不能只看到结束标记就展示成功。

7. 服务重启以后：只恢复有明确定义的等待点

TurnRegistry、缓冲和取消句柄都在内存。服务进程死亡后，它们一起消失。后台 goroutine 解决的是与请求解耦，没有把程序计数器保存到数据库。

当前项目有一个有意设计的例外：request_input 等待用户回答。

问题与答案保存在 messages 中。回答先提交数据库，再通知 worker；等待者还有每两秒查询结果的兜底，因此通知丢失并不必然丢失答案。服务重启后，重新订阅或提交回答的路径可以识别这个等待点，重建 worker，并使用新的事件流身份继续。

它之所以能恢复，是因为暂停边界有业务定义：知道在等哪个问题、是否已有答案、接下来如何继续。同批次中其他普通工具会被标记为“未执行”，恢复时不重放它们的副作用。这里不展开输入协议，只强调：恢复点需要保存“接下来该做什么”的语义，只有一段聊天历史还不够。

如果进程在任意 bash、文件修改或外部工具调用中死亡，当前没有通用检查点来恢复原执行。历史组装时给缺失的工具结果补占位，可以修复消息协议配对，却不能推断工具究竟成功了没有。

这一边界也决定了之后如何升级：应该先识别允许重试的工作单元与副作用，再选择持久化机制。

8. 横向对比：主流方案把恢复责任放在哪里

下面比较的是公开产品或框架接口，而不是各家未公开的内部架构。特别是同一厂商的模型 API 与托管 Agent API，可能有不同的事件恢复契约。

OpenAI：Background Response 与托管 Agent Session 是两层能力

Responses 的 background mode 允许异步运行模型响应，客户端可以轮询或取消。创建时同时开启流式输出后，可记住 sequence_number，通过 starting_after 重新接流。这体现了“执行对象独立于监听连接”。它没有替应用提供任意自定义工具工作流的崩溃恢复；函数调用中的本地工具仍需要应用执行并回传结果。Background mode、Function calling

OpenAI 当前的 Agents API 则以 Session 和异步 Turn 管理工作，区分实时 events 与保存的 items。其流不重放错过的事件；官方恢复流程是先开新流并缓冲，再取 Session 与 saved items，按 item 身份重建并合并。对已终结的 item，丢弃过时更新。完整输出可以修正临时 delta，但无法还原所有中间事件。Run and continue sessions、Events and items

**设计分析：**前一种契约接近有游标的响应流；后一种把页面恢复建立在保存的工作结果上。它们支持的抽象层次不同，不能把 Responses 的补播参数直接套到 Agents API，也不能把一次模型响应的后台运行解释成整个应用已经获得 durable execution。

Anthropic：持久事件与逐步预览分层

Claude Managed Agents 的新 stream 默认只接建立后的事件。官方给出的重连顺序是：先订阅并缓冲，读取事件历史，再按事件 ID 去重后接上实时；显式中断使用 user.interrupt，入队也不意味着立即停止。Session event stream

它进一步区分完整的 agent.message 与 event_start、event_delta 预览。预览是尽力提供的，不作为持久历史补播；重连后应该用完成的消息校正临时内容。Preview responses with event deltas

**设计分析：**不必把打字动画的每一个增量都提升成永久事实。完整消息承担恢复职责，临时增量承担即时体验。相应代价是无法保证重现断线期间的逐 token 动画。客户端必须具备“预览被完整结果替换”的能力。

另一个容易混淆的入口是 Claude Agent SDK 的 resume：它从保存的 session 恢复会话上下文，默认使用本地会话存储。它本身不意味着恢复原来的 SSE 连接、任意指令位置或外部文件快照。把 SDK 包成 Web 服务后，执行托管与浏览器重连协议仍需要应用设计。Work with sessions

LangGraph：图检查点负责执行恢复，流接口负责观察

Agent Server 可以独立创建后台 run。流恢复需要看具体接口：官方 thread stream 支持 Last-Event-ID，但普通 runs.join_stream 的指南仍说明不会补齐加入之前的输出。不能把“加入运行”统一理解为“完整重放”。Background runs、Streaming API

LangGraph 的执行恢复依靠 checkpointer 保存图状态，而不是依靠 SSE。sync 持久化后再前进，async 与后续执行并行，exit 到退出时才保存；这些模式在开销与中途崩溃窗口之间作不同选择。跨重启恢复还需要持久存储后端，内存 saver 不具备这个条件。Persistence、Checkpointers

**设计分析：**明确的图步骤提供了可保存、可重试的边界，代价是业务需要按这些边界组织状态和执行。检查点可以恢复“运行到了哪里”，外部副作用依然需要另外处理。

放在同一张表里比较

表中的事实依据分别见上面各方案的官方链接；这是恢复语义比较，不是性能排名。

对当前项目最直接的启发有两点。其一，展示层可以有临时数据，但恢复必须有可信的事实来源。其二，读取历史和订阅实时数据之间必须有衔接协议：项目在进程内用同一把锁保证观察一致；OpenAI Agents API 与 Claude Managed Agents 则要求客户端先订阅，再读取保存的状态或历史，最后按对象身份合并或去重。这些方案解决了相似的空窗问题，但并不代表厂商内部使用了与项目相同的锁实现。

9. 面试中如何把设计讲到边界

介绍这一设计时，可以先用一段话建立主线：

项目把一轮 Agent 执行托管给独立的 Background Turn，HTTP/SSE 只订阅输出。Turn 维护有界事件缓冲，前端用流身份和序号重接；检查缓冲与取得唤醒通道在同一个临界区里，避免丢唤醒。断连只取消订阅，停止接口才请求取消执行。目前保证限定在单进程，通用跨进程恢复与外部副作用幂等还需要额外设计。

接下来的追问通常会落到下面几个故障场景。

场景一：工具已经成功，保存结果前进程崩溃

1

外部操作成功 ──→ 进程崩溃 ──→ 数据库没有结果 ──→ 恢复后准备重试

仅凭数据库没有结果，不能判定外部动作没发生。这个状态应该视为“结果未知”。LangGraph 的故障容错文档也要求在重试边界考虑幂等，不能仅依靠节点恢复保证副作用正确。Fault tolerance：Retry boundaries and idempotency

后续设计可以为逻辑动作保存稳定的 execution ID，把它作为外部工具认可的幂等键，并支持按键查询结果。没有幂等能力的系统，可能需要人工确认或补偿。每次重试重新生成一个 UUID，会失去防重作用。

事务 Outbox 可以协调“数据库提交”与“事件发布”，但它不能单独把外部系统的动作和本地数据库变成一个原子事务。回答到这里，才算解释清楚“消息可靠”与“业务动作不重复”的区别。

场景二：扩成多个实例，重连命中了另一台机器

当前实例内的 TurnRegistry 在另一台机器上不存在。粘性路由可以缓解重连路由问题，却无法处理原实例宕机；单独增加 Pub/Sub 可以转发实时通知，也不能自动补齐历史或转移执行所有权。

进一步的设计至少需要同时明确：执行记录保存在哪里、谁拥有当前执行、何时允许接管、旧执行者恢复后是否还能提交结果。

一种可讨论的方案是持久化 run 状态，使用租约与续租协调所有权，并为每次所有权变更增加 fencing token，让状态存储或受控写入端拒绝旧执行者的提交。它仍然依赖下游是否配合，不能凭一个 token 就约束所有外部 API。

这是当前方案的演进方向，不是已经完成的分布式能力。

场景三：慢客户端一直追不上生产者

本项目选择让生产者继续运行，让订阅者从共享缓冲读取。网络写不占生产者锁，但落后太多会越过窗口。

如果要求“不丢任何展示事件”，就需要支付更长保存、持久化或背压的成本；如果允许丢临时增量，则可以在缺口后读取完整消息。应该先定义产品需要恢复的是什么，再决定保存什么。

场景四：既然落了消息，为什么不能直接恢复所有任务

消息记录描述已经观察到或保存下来的内容，不必然包含尚未完成的调用、外部操作是否成功，以及下一步是否安全重试。

request_input 能恢复，是因为它有明确等待对象、已保存答案和不重放兄弟工具的协议。将来增加其他恢复点，也需要同样明确的边界。直接重新喂给模型整段历史，是一次基于历史继续推理，不能自动等同于恢复原来的执行。

10. 这次验证了什么

本次检查以实际代码为准，并运行了已有的针对性测试，没有修改 Agent 实现。

在 backend 目录执行：

1
2

go test ./pkg/agent ./api \
  -run '^(TestTurn|TestDelete|TestFailedDelete|TestInput)' -count=1

两个包均通过。前端使用 Node 24.13.0 执行 frontend/tests/stream.test.ts，37 个流处理测试全部通过。相关测试覆盖断连不取消 Turn、并发启动拒绝、补播与旧游标处理、取消和异常收尾，以及前端分块解析和只重接订阅等行为。

这些是使用内存或替身依赖的针对性测试，不是真实模型调用、线上网络故障、强制杀进程或多实例容灾实验。本文没有据此给出吞吐、延迟或 exactly-once 的实测结论。

便于继续阅读的源码入口如下，路径均相对项目根目录：

这个技术点体现的设计价值，在于明确了执行与观察的责任：连接可以短暂存在，任务拥有独立生命周期；临时输出可以追求及时，恢复依靠明确保存的事实；取消可以被请求，副作用是否完成仍需单独确认。能够准确说出这些保证的边界，才有基础继续演进更可靠的 Agent Harness。