前言：封闭告一段落，开始系统复盘：一个生产 Agent 到底怎么跑起来。踩坑、经验和落地，最主要的是用户认可产品。尽量用几篇文章讲清，这篇文章是系列第5篇：多步、变慢、要确认时，前端不能靠猜。把 Run、中断、恢复和流式事件定成前后端都能认的协议。
用户说：“重启这个异常实例。” 或中间补一句：“继续查它。”
这篇只讲：一次请求怎么流式推给前端，以及人确认之后怎么从断点接着跑。
单轮聊天接口不难。业务一接进来，通常马上会碰到这三件事：
- 用户说「继续查它」，系统不知道「它」是谁；
- Agent 跑很久，前端长时间没反馈；
- 删除、发布、转账必须让人确认，确认后还要从中断处继续。
这三件事分别对应：多轮状态、SSE 流式通信、HITL（Human in the Loop，人在回路）。继续用案例 OpsPilot（企业运维助手，虚构案例），技术栈仍是 TypeScript；前端侧可按 Vue 3 + fetch 读流来接。
一次运行要有自己的状态
不要只存一串聊天文本。至少分成三张表（或三份记录）来存：
- 会话（Conversation）：左边那一列聊天列表里的一项，比如「排障 - payment-api」。一个人可以有很多个会话。
- 消息（Message）：会话里的每一句：用户说的、助手回的、工具结果，各算一条。
- 运行（Run）：用户点一次发送，后台为这句话开的一次任务（跑模型、调工具、等人确认）。同一会话里可以有很多次 Run。
type Conversation = { // 一次聊天会话（聊天列表里的一项）
id: string // 会话 id
userId: string // 属于哪个用户
createdAt: string // 创建时间
}
type Message = { // 会话里的一条消息
id: string // 消息 id
conversationId: string // 属于哪个会话
role: 'user' | 'assistant' | 'tool' // 谁说的
content: string // 正文（展示 / 写进数据库）
createdAt: string // 创建时间
}
type Run = { // 用户点一次发送对应的一次后台运行
id: string // Run id
conversationId: string // 落在哪个会话
userMessageId: string // 由哪条用户消息触发
status: // 这次运行进行到哪
| 'running' // 执行中
| 'interrupted' // 等人确认 / 补参
| 'completed' // 正常结束
| 'failed' // 失败
| 'cancelled' // 取消
checkpoint?: unknown // 中断后接着跑所需的最小状态
createdAt: string // 创建时间
updatedAt: string // 状态最后变更时间
}
一次用户请求对应一个 Run。用户确认后恢复的是这次运行，不用重新猜整段对话发生了什么。
三张表管「存哪」；下面三层管「每次喂模型什么」。
先能听懂「它」，别整段塞历史。 用户常说「继续查它」「刚才那个」。别把整段聊天记录都扔进模型；拼上下文时用三层就够：
- 已确认的事实 → 结构化 state（挂在会话上：Conversation 的字段或关联表）；只在工具成功或用户明确确认后写入，查找失败不要把猜的写进来。
- 更早的对话 → 压成一段摘要，控制长度。
- 最近几句原文 → 谁说了什么原样保留，别先压成摘要，不然模型不知道「它」是谁。
对应到一次模型调用：第 1、2 层放进 system；第 3 层作为最近消息列表接在后面（不是把 messages 全表倒进去）：
// 1) 已确认事实（挂在会话上）
const state = { service: 'payment-api', env: 'prod' as const }
// 2) 更早对话的摘要
const earlierSummary = '用户在排查生产环境报错；服务名与环境已确认。'
// 3) 最近几句原文（含当前这句「继续查它…」）
const recentMessages = [
{ role: 'user', content: '看下 payment-api 最近的错误' },
{ role: 'assistant', content: '生产环境过去 15 分钟有 12 次 5xx。' },
{ role: 'user', content: '继续查它最近的错误日志' },
]
// 拼装：1+2 → system；3 → 消息列表
const modelInput = [
{
role: 'system',
content:
`已知事实：${JSON.stringify(state)}\n` +
`更早摘要：${earlierSummary}\n` +
`用最近原文理解「它 / 刚才那个」是谁；不要靠关键词硬猜。`,
},
...recentMessages,
]
先看一条完整链路（SSE + HITL）
用 OpsPilot 举例：用户说「重启这个异常实例」。整件事分两段——先查清、等人点确认；点完再真重启。
第一段：发消息、看进度、弹出确认
- 前端把这句话发给会话服务，同时打开一条 SSE，用来收进度（开始了、正在查、查完了……）。
- 后台先跑只读步骤（比如定位是哪个实例），这些进度通过 SSE 推到页面。
- 模型提出要调「重启」工具。这时还不执行重启，而是发一条 interrupt 事件：页面弹出确认（风险说明、要重启谁）。
- 这次 Run 记成「等人」；这一轮推流可以结束。页面上是确认，不是红字报错。
第二段：用户点确认，从断点接着跑
- 用户点「确认」。前端调用恢复接口（带上这次中断的 id、同意/拒绝、以及防双击用的同一把请求键），不是再发一条全新聊天。
- 服务端先确认：这个中断还有效、当前用户仍有权限、实例现在还是那个状态；通过后才真正重启。
- 执行过程再次用 SSE 推进度（工具开始/结束、最终回答），直到结束。
sequenceDiagram
participant U as 用户
participant V as 前端
participant S as 会话服务
participant R as Runtime
U->>V: 重启这个异常实例
V->>S: 发消息并订 SSE
S->>R: 开始本次 Run
S-->>V: 进度：开始 / 正在查询…
R-->>S: 需要确认（先别重启）
S-->>V: 弹出确认
U->>V: 点确认
V->>S: 调用恢复接口
S->>R: 再验权限后执行重启
S-->>V: 进度：重启中 / 最终回答
就记两句：
- 等人确认用 interrupt，这是正常流程，不是失败。
- 点确认后走恢复接口接着这次 Run 跑，不要当成新开一轮对话。
SSE 流式推送
SSE（Server-Sent Events）基于普通 HTTP：前端发一次请求，服务端持续推文本事件。「用户发一句，服务端持续回进度」时通常比 WebSocket 简单；要高频双向或二进制时再评估 WebSocket。
有的编排框架每次推过来的不是「刚发生的那一步」，而是「到目前为止的全部状态」（例如步骤列表又完整发一遍）。这时前端或网关要自己对比上次看过的内容，只把新多出来的那几步转成 tool_started 这类事件；否则页面上同一步会显示两遍。
下面先定业务事件，再套 SSE 外壳；长连接上的注释行防超时、断开后重新连接，放在本节末。
事件协议
- 流必须是 UTF-8；MIME 为 text/event-stream（可写成 text/event-stream; charset=utf-8）。
- 建议每条业务事件里带一个本 Run 内越来越大的序号，比如 sequence: 1, 2, 3…。前端拿它判断「这条是不是已经渲染过」；以后若要「断开后把没收到的进度再拉回来」，也靠这个号。注意：字段先留着，不代表第一版就要做这件事；另外，这是你们自己 JSON 里的字段，和 SSE 协议自带的 id: 不是一回事。
import type { ServerResponse } from 'node:http'
type StreamEvent =
| {
type: 'run_started'
runId: string
sequence: number // 本 Run 内 1,2,3…；防重复渲染，以后也可按号把没收到的进度拉回来
}
| {
type: 'text_delta'
runId: string
sequence: number
delta: string // 增量字，不是全量覆盖
}
| {
type: 'tool_started'
runId: string
sequence: number
callId: string // 与 tool_finished 配对
tool: string // 工具名，步骤条展示
}
| {
type: 'tool_finished'
runId: string
sequence: number
callId: string
tool: string
status: 'success' | 'error'
summary?: string // 给前端的短摘要，不是原始大结果
}
| {
type: 'interrupt'
runId: string
sequence: number
interruptId: string // resume 时原样带回
payload: unknown // 确认：风险说明、参数摘要等（已脱敏）
}
| {
type: 'done'
runId: string
sequence: number
messageId: string // 已写入数据库的助手消息 id
}
| {
type: 'error'
runId: string
sequence: number
code: string
message: string // 可展示给用户；不要堆栈
}
// 片段：在已创建 Run、准备推流的路由里调用（res 是 Node/Express 响应，不是 fetch Response）
function openSse(res: ServerResponse) {
res.setHeader('Content-Type', 'text/event-stream; charset=utf-8')
res.setHeader('Cache-Control', 'no-cache, no-transform')
res.setHeader('X-Accel-Buffering', 'no') // 仅 nginx；不用可省略
res.flushHeaders()
}
function sendEvent(res: ServerResponse, event: StreamEvent) {
// 一条 SSE 消息：event 行 + data 行 + 空行结束
res.write(`event: ${event.type}\n`)
res.write(`data: ${JSON.stringify(event)}\n\n`)
}
// 需要等人时：先把 interrupt 和 Run=interrupted 写进数据库，再推事件（示意）
function emitInterrupt(
res: ServerResponse,
runId: string,
sequence: number,
interrupt: { id: string; payload: unknown }
) {
sendEvent(res, {
type: 'interrupt',
runId,
sequence,
interruptId: interrupt.id,
payload: interrupt.payload
})
// 随后可以 res.end() 结束本轮流；用户确认走 POST resume，不必占用同一条长连接
}
事件类型应带业务含义。不要让前端解析模型输出文本去猜「现在是不是在调工具」。
说明：若改用浏览器原生 EventSource，带了 event: 字段的消息不会进 onmessage，必须 addEventListener(事件名, ...)。本文用 fetch 自解析时，是靠 JSON 里的 type 分发，和 EventSource 的命名事件机制不是同一条路径。
前端怎么读流、怎么接到 interrupt
原生 EventSource 只能 GET、不能带请求体和自定义头（跨域最多 withCredentials 带 Cookie）。聊天要 POST，所以用 fetch 读流、自己切事件。
为什么必须用 buffer（先攒起来再切，不能来一段就当完整事件）：
- 网络一次读到的不一定是完整一条——可能只半截，也可能一次来好几条。
- 事件之间靠「空一行」分开。
- 空行怎么写因环境而异：有的是 \n，有的是 \r，有的是 \r\n（Windows 常见）。
- 解析时先统一收成 \n，再按连续两个换行 \n\n 切开。
下面骨架只负责拆包并回调；UI 自行接 onEvent。
// 可运行骨架：POST 聊天并解析 SSE
async function streamChat(
body: unknown,
onEvent: (event: StreamEvent) => void
) {
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body)
})
if (!response.ok || !response.body) {
throw new Error(`请求失败：${response.status}`)
}
const reader = response.body.getReader()
const decoder = new TextDecoder() // SSE 规定 UTF-8
let buffer = ''
while (true) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n').replace(/\r/g, '\n')
let boundary: number
while ((boundary = buffer.indexOf('\n\n')) >= 0) {
const block = buffer.slice(0, boundary)
buffer = buffer.slice(boundary + 2)
// 「: ...」是注释行，没有 data:，得到空串就跳过
const data = block
.split('\n')
.filter(line => line.startsWith('data:'))
.map(line => {
const value = line.slice(5)
return value.startsWith(' ') ? value.slice(1) : value
})
.join('\n')
if (data) onEvent(JSON.parse(data) as StreamEvent)
}
}
}
// 接到事件后怎么处理（示意，不是完整 UI）
function handleStreamEvent(event: StreamEvent) {
switch (event.type) {
case 'text_delta':
appendAssistantText(event.delta) // 追加字，不要整段替换
break
case 'tool_started':
showStep(event.callId, event.tool, 'running')
break
case 'tool_finished':
finishStep(event.callId, event.status, event.summary)
break
case 'interrupt':
// 正常等待人：弹确认；保存 interruptId 供 resume
openConfirmCard(event.interruptId, event.payload)
break
case 'done':
markRunDone(event.messageId)
break
case 'error':
showError(event.code, event.message)
break
default:
break
}
}
// 用户点确认：另发 resume，不是再 POST 一条「新聊天」
async function approveInterrupt(runId: string, interruptId: string) {
await fetch(`/api/runs/${runId}/resume`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
interruptId,
decision: 'approve',
idempotencyKey: `resume_${interruptId}_1` // 同一确认动作固定一键，防双击
})
})
// resume 若仍用 SSE 推进度：再发一次请求把 SSE 接上，或按你们约定复用同一条连接
}
连接别被空闲断开；断了也别指望浏览器替你补进度
推流时若半天没有业务事件，nginx / 网关常会按「空闲太久」把连接断开。做法很简单：隔一段时间发一行以 : 开头的注释（不是业务事件，前端解析时直接跳过）：
: keep-alive
间隔自己对着网关超时测（常见从 15～30 秒试起；nginx 的 proxy_read_timeout 默认往往是 60 秒）。
本文用的是 POST + fetch 读流，要心里有数：
- 连接断了，浏览器不会自动再连，也不会帮你把没收到的进度拉回来。
- 第一版优先保证：Run 状态在数据库里、断开或刷新后还能查到最终结果和确认卡；页面要继续看进度，就再发一次请求把 SSE 接上。
- 以后若要做「断开之后，把没收到的进度再向服务端拉回来」，靠 JSON 里的 sequence 告诉服务端「我已经收到几号了，从下一号继续给我」——前提是这些事件当时已经存进数据库；没存就拉不回来。
补充一句：浏览器原生的 EventSource 往往会自动重连，并带上 Last-Event-ID。那也只是告诉服务端「我上次收到哪条」；浏览器照样不会自动把中间没收到的进度补全，逻辑仍在服务端。聊天要 POST，本文不以 EventSource 为主路径。
HITL：暂停等人，不要当成报错
缺参数、要审批等时，这次 Run 应记成 interrupted（暂停等人），不要记成 failed。前端展示「等你确认 / 请补参数」，不要甩红字失败。
暂停时，至少留下两样东西：
- 一条中断记录（给前端弹确认卡、恢复时对得上号）；
- 一份断点（确认后从哪一步接着跑）。
中断记录与断点
type Interrupt = { // 中断记录（等人确认 / 补参）
id: string // 中断 id，恢复时必须原样带回
runId: string // 所属 Run
type: 'confirm_action' | 'select_option' | 'provide_input' // 确认 / 选择 / 补参
status: 'pending' | 'resolved' | 'expired' | 'cancelled' // 只有 pending 才能恢复
payload: unknown // 确认卡上要展示的内容（已脱敏）
expiresAt: string // 过了这个时间不能再点「同意」
}
type Checkpoint = { // 断点：继续跑所需的最小状态
runId: string
nextStep: string // 恢复后从哪一步继续
safeToolArgs: Record<string, unknown> // 已脱敏、恢复前还要再鉴权的参数
conversationVersion: number // 防止恢复期间会话被别人改乱
createdAt: string
}
payload 里不要塞数据库连接串、访问令牌、完整工具参数或内部堆栈。断点也只留继续跑真正需要的字段。
恢复接口
用户点「确认 / 拒绝 / 补参」时，打恢复接口，不是再发一条新聊天：
// POST /api/runs/:runId/resume 请求体
{
interruptId: 'int_456', // 仍为 pending 的中断 id
decision: 'approve', // approve | reject | 补参等
idempotencyKey: 'resume_int_456_1' // 同一确认动作固定一把键，防双击
}
服务端不能直接信中断里那份旧参数。下面骨架里的 authorize / reloadResource / continueFrom 换成你们自己的实现即可，顺序不要乱：
type ResumeInput = {
runId: string
interruptId: string
decision: 'approve' | 'reject' | { fill: Record<string, unknown> }
idempotencyKey: string
}
// 1) 找到仍 pending 的中断 2) 用当前登录态鉴权 3) 防重复提交
// 4) 拒绝则只改状态 5) 重读资源 6) 再验权限 7) 从断点续跑
async function resumeRun(input: ResumeInput, auth: AuthContext) {
const interrupt = await interruptRepo.findPending(input.interruptId)
if (!interrupt || interrupt.runId !== input.runId) {
throw new Error('INTERRUPT_NOT_FOUND')
}
await authorize(auth, interrupt) // 身份只信这次请求的登录态
await assertNotExpired(interrupt) // 看 expiresAt；过期直接拒绝
await idempotency.acquire(input.idempotencyKey) // 同一把键第二次进来：返回上次结果，别再执行
if (input.decision === 'reject') {
await markRejected(interrupt) // 只改状态
return { status: 'cancelled' as const } // 不要再调会改数据的工具
}
const checkpoint = await checkpointRepo.load(input.runId)
const latestResource = await reloadResource(checkpoint) // 等待期间实例可能已变
await revalidate(latestResource, auth)
return continueFrom(checkpoint, input.decision) // 按 nextStep 续跑
}
从弹出确认到用户点确认，资源和权限都可能变。所以，必须重读资源、重新鉴权。用户点拒绝时，只更新中断和 Run 状态，不要再去执行写操作。
记得要防双击、防重试打两次：
比如说：用户连点「确认」、网络抖一下浏览器自动重试、两个标签页各点一次，恢复接口都可能被打两遍。
所以我们的目标是：同一次确认，后面的写操作只真正执行一次。（在 OpsPilot 里就是「重启实例」只发生一次，不要重启两遍，这种动作很危险。）
常见两道防线，都写在恢复接口的后端代码里，由程序去访问数据库——不是让你打开数据库客户端手敲：
- 请求键 idempotencyKey：前端对「这一次点确认」生成一把固定的键（同一确认动作、连点、自动重试都带同一把）。服务端先查：这把键是不是已经处理过？处理过就直接返回上次结果，后面的重启逻辑不再跑。
- 改中断状态时加条件：真正去执行写操作之前，用代码执行一条带条件的更新——「只有这条中断现在还是 pending，才改成 resolved」。两个请求几乎同时进来时，数据库里只会有一个更新成功（改动行数 = 1）；另一个改动行数 = 0，说明已经有人抢先处理了，这个请求就不要再调重启工具。
// 写在 resume 接口里，通过数据库客户端执行（示意）
const result = await db.query(
`UPDATE interrupts
SET status = 'resolved', resolved_at = NOW()
WHERE id = $1 AND status = 'pending'`,
[interruptId]
)
if (result.rowCount === 0) {
// 已经处理过、过期或取消了 → 直接返回，禁止再执行工具
return alreadyHandledResponse()
}
// rowCount === 1：这是第一个抢到的请求，才允许继续重读资源、执行重启
记住一点就行：两个请求一起来时，靠上面这条带条件的更新，让数据库选出第一个成功的；只有它才能继续执行危险操作，另一个直接返回。
SSE 断了，后台要不要停
和「用户点取消」不是一回事。推流断开只表示前端暂时听不到进度；Run 要不要跟着停，看任务性质：
- 查询类：客户端断开后可以取消，省模型和工具费用。
- 发布、审批、重启实例等：通常应和 HTTP 连接拆开——后台 Run 继续跑，结果写进数据库；用户刷新后再把状态读回来。
// 默认只停推送；是否取消 Run 看策略
// Node：别听 req 的 close（POST 读完请求体就可能触发）；听 res 的 close
// 正常 res.end() 也会触发 close：用 !writableFinished 区分「客户端提前断开」
res.on('close', () => {
subscriber.detach(runId) // 停推送；不等于用户点了取消
if (!res.writableFinished && runPolicy.cancelWhenNoSubscriber) {
// 仅客户端提前断开；写操作 / 审批类通常不因断连取消
runController.cancel(runId)
}
})
和 LangGraph 怎么对照（点到为止）
不少团队用 LangGraph 跑 Agent 循环。它和本文解决的是同一类问题，分工不同：
- 本文定的是对外协议：会话、Run、SSE 业务事件、恢复接口怎么跟前端约定。
- LangGraph 管的是图怎么跑、中途状态怎么存。函数名、参数以你安装的版本文档为准，这里只记概念，避免版本一变例子就废。
一句话：可以用 LangGraph 跑循环、存断点，但 SSE 事件名、恢复接口、鉴权和防双击仍要按本文自己定——接上框架不等于这些都齐了。
怎么测、常见误区与落地顺序
先测这些（上线前尽量跑一遍）：
- SSE：一个事件被拆成好几块、一块里塞了多条、中文跨块切开、网关空闲超时、断网、刷新页面、重连后同一步是不是又画了一遍。
- HITL：双击「确认」、确认过期、恢复前权限被撤、实例在等待期间被别人改过、取消之后又点恢复。
- 会话上下文（顺带）
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
Preface: Close the door and start the system reset: how does a producer Agent actually get up? Penetrating, experience and landing, most notably user-approved products. As far as possible, this article is the fifth in the series: multisteps, slowness, and the front end cannot be guessed. Make Run, Interrupted, Restored, and Fluid Events an unrecognizable agreement.
The user says, "Reactivate this anomaly. Or add the following sentence: “Continue to look at it.”
This article only says: how a request will flow to the front end and how to run from the breakpoint after it has been confirmed.
One-wheel chat interface is easy. When business comes in, it usually comes up with three things:
- Users say "continue to check it" and the system does not know who "it" is;
- Agent runs long, no feedback at the front end;
- Delete, publish, transfer must be confirmed and then continue from the interruption.
These three things correspond: multi-wheel state, SSE flow communication, HITL (Human in the Loop, people on the way back). The case of OpsPilot (Enterprise Transport Assistant, Fiction Case) continues, and the technology warehouse remains TypeScript; the front end can be accessed on the Vue 3 + Fetch stream.
One run must have its own status.
Do not save only one chat text. At least three forms (or three records) are saved:
-Conversation: One of the rows of chats on the left, such as "Clause-payment-api". One can have a lot of meetings.
- Message: Each sentence of the session: the user says, the assistant returns, the tool results is one.
- Run (Run): a user-point sent at a time and a backstage mission (run model, modulation tool, and other confirmation). There can be many times in the same session, Run.
typeConversion = {/ / One chat (one in the chat list)
id: string // session id
user(s)
CreatedAt: string / create
}
type Message = a message from a session
id: string/ / message id
Organisation
/ Who said that?
Contact: string / / body (presentation / writing into database)
CreatedAt: string / create
}
type Run = {/ / / user point sent the corresponding backstage
id: string // Run id
ConversionId: string / / which session to drop
userMessageId: string / / Trigger from which user message
status: / / Where is this run going?
|runing '// Executing
|interrupted' / / / et al.
|completed '// Normal End
|failed '// Failed
|cancelled '/ / Canceled
Checkpoint?: unknown/ / Minimum state of running after interruption
CreatedAt: string / create
updatedAt: string / / state change time
}
One user request corresponds to one Run. The user confirmed that it was the operation that was restored without re-enumerating what had happened to the entire conversation.
Three standpipes "where to store" and three floors "what to feed the model each time."
You can understand "it" first, not the whole history. The user used to say, "Keep looking at it," "just now." Do not throw the whole chat record into the model; three layers of context is enough:
- Confirmed facts Structure state (on the session: Conversation field or associated table); write only after the tool has been successfully or the user has clearly confirmed, and do not write guesses when the search fails.
- Early conversation Z pressured into a summary, controlled length.
- The last few words say who said what to keep as it is, don't press summary, or the model doesn't know who it is.
Corresponds to a model call: 1st & 2nd floor in system; 3rd level is attached as the list of recent messages (not the complete list of messages):
/ 1) established facts ( hanging on session)
const state = { service: 'payment-api', env: 'prod' as const }
2) Early Dialogue summary
Const earlier Summary = 'User is referencing the production environment; service name and environment have been confirmed. ♪ I'm sorry ♪
/ 3) Recent words (with the current phrase "continue to look into it...")
const recentMessages = [
{rule: 'user', content: 'look at recent mistakes',
{rule: 'assistant', content: 'production environment has been 12 times 5x in 15 minutes. {\cHFFFFFF}{\cH00FFFF}
{rule: 'user', center: 'Continue to check its latest error log'},
]
// Collapse: 1+2 → system; 3 → message list
const modelInput = [
{
role: 'system',
content:
`Known Fact: $ {JSON.stream(state)}\
`Better summary: $ {earlierSummary}\
`In the most recent language, understand who it is / what it just was; don't use keywords. `...
},
...recentMessages,
]
See first full link (SSE + HITL)
For example, on OpsPilot: The user says, "Relaunch this anomaly." The whole thing is in two parts — first to be clear, to be confirmed; then to be counted and really restarted.
Paragraph 1: Sending messages, checking progress, eject confirmation
- The front end sends this sentence to the session service, while opening a SSE to collect progress (starting, checking, checking...).
- Run read-only steps in the backstage (e.g., which case is the location) and push these progress through SSE to the page.
- The model proposes a "restart" tool. At this point, the restart is not executed, but an interrupt event: popup confirmation (risk statement, who to restart).
- This time Run remembers "someone"; this wheel can end. page.
Second segment: User point confirmed, run from breakpoint
- User point "confirmation". The front-end calls for the restoration of the interface (with this interrupted id, consent/denied, and the same request key used to prevent double-clicking), rather than sending a whole new chat.
- The service confirmed that the break was still valid, that the current user still had permission, that the example was still in that state; and that it was actually restarted after adoption.
- The implementation process is again carried out using SSE propulsion (start/end of tool, final answer) until the end.
sequenceDiagram
Party U as user
Party V as Frontend
Party S as Session Service
participant R as Runtime
U->V: Restart this anomaly
V->S: Send messages and order SSE
Let's go, Run.
S->V: Progress: Start / Querying...
R-- > S: Require confirmation (not to restart)
S->V: Eject confirmation
U->V: Point confirmation
V->S: Call back interface
S->R: Re-test permission and execute restart
S->V: Progress: Restarting / Final answer
Just two words:
- People confirm using interrupt, which is a normal process, not a failure.
- Dot confirmation and go back to the interface and run this Run, not as a new round of conversation.
SSE Streaming
SSE(Server-Sent Events) is based on a regular HTTP: a front-end request, which continues to push text events at the service end. "User send a message that service-ends keep returning progress" is usually simpler than WebSocket; WebSocket is then evaluated when high frequency two-way or binary is required.
In some cases, the frame is not "the one that just happened" but "the whole state to date" (e.g. a full list of steps). This is when the front end or the gateway itself compares what was last seen, turning the new steps into a tool starded event; otherwise the same step on the page will appear twice.
Business events are determined below, followed by SSE casings; the comment line on long connections is protected against timeout, which is disconnected and reconnected at the end of this section.
Event protocol
- The flow must be UTF-8; MIME shall be text/event-stream (text/event-stream; charset=utf-8).
- It is recommended that each business event be accompanied by an increasing serial number within the Run, for example: 1, 2, 3... The front end uses it to determine whether the "this one has already been rendered" and, in the future, to "take back the unreceipted progress after it has been disconnected." Note: field keeps it, which doesn't mean that the first edition will do it; moreover, it's field from your own JSON, and SSE protocol id: It's not the same thing.
import type { ServerResponse } from 'node:http'
type StreamEvent =
| {
type: 'run_started'
runId: string
Sequence: number / / This Run 1, 2, 3...; repeat replicating, and later pull back unreceived progress by number
}
| {
type: 'text_delta'
runId: string
sequence: number
Delta: string // Incremental, not fully overwrite
}
| {
type: 'tool_started'
runId: string
sequence: number
Call Id: string/ / pair with tool finished
tool: string / toolname, stepbar presentation
}
| {
type: 'tool_finished'
runId: string
sequence: number
callId: string
tool: string
status: 'success' | 'error'
Short summary for front end, not the original big result.
}
| {
type: 'interrupt'
runId: string
sequence: number
InterruptId: string / / reme Back
Payload: unknown / Confirm: risk statement, parameter summary, etc. (sensitized)
}
| {
type: 'done'
runId: string
sequence: number
messageId:string / / Assistant message written to database id
}
| {
type: 'error'
runId: string
sequence: number
code: string
message: string/ can be displayed to users; no stacking
}
/ / Snippet: Called in created Run to prepare the stream (res is Node/Express response, not fitch Response)
function openSse(res: ServerResponse) {
res.setHeader('Content-Type', 'text/event-stream; charset=utf-8')
res.setHeader('Cache-Control', 'no-cache, no-transform')
rs. setHeader ('X-Accel-Buffering', 'no') / nginx only; not omitted
res.flushHeaders()
}
function sendEvent(res: ServerResponse, event: StreamEvent) {
/ / An SSE message: event line + data line + empty line end
res.write(`event: ${event.type}\n`)
res.write(`data: ${JSON.stringify(event)}\n\n`)
}
/ / When people are needed: enter interrupt and Run=interrupted in the database before pushing events (indicated)
function emitInterrupt(
res: ServerResponse,
runId: string,
sequence: number,
interrupt: { id: string; payload: unknown }
) {
sendEvent(res, {
type: 'interrupt',
runId,
sequence,
interruptId: interrupt.id,
payload: interrupt.payload
})
/ / can then end this rotation; users confirm that POST resume does not need the same long connection
}
The type of event should have business implications. Do not let the front-end solver output text guess "Is it a tool now?"
Note: If you change to the browser native EventSource, with information from event: field will not enter onmessage, you must add EventLister. This is a fech self-decomposition that was distributed by type in JSON, and the EventSource naming event mechanism is not the same path.
How do you read the stream at the front end? How do you get it?
EventSource can only get GET, not bring the requesting body and custom head (with Credentials with Cookie). Chats are POST, so use fech to read and cut events.
Why do you have to use a buffer?
- The web does not necessarily read a whole article at a time — it may be half-shot or several at a time.
- The event was separated by "empty line".
- How empty lines are written varies depending on the environment: some are \r, others are \\r\n (Windows common).
- Consistency harvest, then two consecutive lines cut.
The skeleton below is only for unpacking and turning back; UI is onevent.
/ Runable Bones: POST Chat and Parsing SSE
async function streamChat(
body: unknown,
onEvent: (event: StreamEvent) => void
) {
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body)
})
if (!response.ok || !response.body) {
Throw new Error
}
const reader = response.body.getReader()
Const decoder = new TextDecader() // SSE
let buffer = ''
while (true) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true }).replace(/\r\n/g, '\n').replace(/\r/g, '\n')
let boundary: number
while ((boundary = buffer.indexOf('\n\n')) >= 0) {
const block = buffer.slice(0, boundary)
buffer = buffer.slice(boundary + 2)
/ / ":" is a comment line, without data: Skip if you get an empty string
const data = block
.split('\n')
.filter(line => line.startsWith('data:'))
.map(line => {
const value = line.slice(5)
return value.startsWith(' ') ? value.slice(1) : value
})
.join('\n')
if (data) onEvent(JSON.parse(data) as StreamEvent)
}
}
}
/ Upon receipt of an incident (indicated, not complete UI)
function handleStreamEvent(event: StreamEvent) {
switch (event.type) {
case 'text_delta':
AppendAsistantText (event.delta) // Add Words, do not replace the whole paragraph
break
case 'tool_started':
showStep(event.callId, event.tool, 'running')
break
case 'tool_finished':
finishStep(event.callId, event.status, event.summary)
break
case 'interrupt':
/ / Normal waiter: bullet confirmation; save interruptId for resume
openConfirmCard(event.interruptId, event.payload)
break
case 'done':
markRunDone(event.messageId)
break
case 'error':
showError(event.code, event.message)
break
default:
break
}
}
/ / User Point Confirmation: Send another resume, not another POST "New Chat"
async function approveInterrupt(runId: string, interruptId: string) {
await fetch(`/api/runs/${runId}/resume`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
interruptId,
decision: 'approve',
: `resume ${interruptId} 1 // Hold the same confirmation action by one key, prevent double-click
})
})
/ Resume if SSE propulsion is still used: send another request to connect SSE, or repeat the same connection as you agreed
}
Connections are not disconnected; do not expect the browser to complement them degrees
nginx / gateways often cut off the connection by `too long to be idle' if there are no business events for half a day. The approach is simple: a line will be sent at a later time: a note at the beginning (not a business event, skipping directly when the front end is analyzed):
: keep-alive
Interval between timeout measurements (often from 15 to 30 seconds; nginx's program read timeout default is often 60 seconds).
This article is based on the POST + Fetch Reader.
- The connection's broken, the browser won't automatically reconnect, and it won't help you pull back the unreceived progress.
- First edition of the priority guarantee: the final results and confirmation cards can also be found in the Run status in the database, disconnected or refreshed; to continue to see progress on the page, a further request is sent to connect SSE.
- In the future, if you want to do "to pull back unreceipt progress to the service after it has been cut off", you can rely on JSON's security to tell the service that "I have received the number, continue to give it to me from the next number" — provided that these events were stored in the database; they could not be pulled back without it.
Additional sentence: EventSource, the original browser, tends to reconnect automatically with Last-Event-ID. It's just to tell the service which one I received the last time; the browser will not automatically complete the progress that was not received in the middle, and the logic is still at the service end. Chat to POST, this paper does not use EventSource as the main path.
HITL: Time out for people, don't take it as a mistake.
When parameters are missing, approval is to be granted, this time Run should be recorded as interrupted (suspended, etc.) and not failed. The frontend displays "Wait until you confirm"/please fill in the parameters, do not lose the red word.
At least two things are left on the pause:
- An interruption record (to the front-end confirmation card, to be marked when restored);
- A break point (from which step after confirmation).
Interrupt records and breakpoints
type Interrupt = {/ / / Interrupt record (waiting confirmation / supplementary / )
id: string / / interrupted id, must be as it is when restored Back
RunId: string / / belong to Run
type: 'conform action' |self option' |provide input'// confirm/ select / patch
status: 'pending'
Payload: unknown // confirmation card to display (dissensitized)
ExpiresAt: string / / Can't "approval" after this time
}
type Checkpoint = {/ / Breakpoint: Minimum state required to continue running
runId: string
NextStep: start / / Which step to continue after recovery
SafeTool Args: Record<string, unknown>// Parameters that are dissensitized and subject to re-examination before recovery
ConversionVersion: number / / prevent conversational distortion during recovery
createdAt: string
}
Payload does not plug in database links, access tokens, complete tool parameters or internal stacks. The breakpoint also leaves only the field that is really needed to keep running.
Restore interface
Recover the interface when the user point "confirms" / rejects / patches" instead of sending another chat:
/ POST / api/runs/:runId/resume request Body
{
InterruptId: 'int 456', / / still interrupting id
Deciment: 'approve', / approve | subject | supplementation, etc.
/ The same confirmation action fixes a key with double clicks
}
The service cannot directly connect to the old parameter in the break. Here's the restorize / reloadResource /continueFrom for your own realization.
type ResumeInput = {
runId: string
interruptId: string
decision: 'approve' | 'reject' | { fill: Record<string, unknown> }
idempotencyKey: string
}
/ 1) Disruption still pending 2) Use current login rights 3)
/ 4) Refusal is only changed to status 5) Reread Resource 6) Retest permission 7)
async function resumeRun(input: ResumeInput, auth: AuthContext) {
const interrupt = await interruptRepo.findPending(input.interruptId)
if (!interrupt || interrupt.runId !== input.runId) {
throw new Error('INTERRUPT_NOT_FOUND')
}
/ Identity only letters the login of this request State
Wait assertNotExpired (interrupt) //see extirresat; expired direct refusal
/ The same key comes in for the second time: return the last result and do not execute again
if (input.decision === 'reject') {
Wait markReject(interrupt)/ / just change
/ / Do not change the data again
}
const checkpoint = await checkpointRepo.load(input.runId)
Could change the example of waiting period
await revalidate(latestResource, auth)
Return continueFrom (checkpoint, input.description)// press nextStep
}
From ejection confirmation to user point confirmation, resources and permission can change. It is therefore necessary to revisit resources and re-examine. When the user point rejects, only the interrupted and Run state is updated and no more writing is performed.
Remember to double-click and double-click against retry:
For example, the user-point confirmation, the web shaker automatic retry, and the two tabs can be hit twice.
So our goal is to make sure the same thing is true for the later writing. (In OpsPilot, the "restart instance" only happens once, not twice, and it's dangerous.
The common two lines of defence are written in the back-end code for restoring the interface, and access to the database is provided by the program - not to allow you to open the database client:
- Request key idea key: the front-end generates a fixed key for this subpoint confirmation (the same confirmation action, nodes, automatic retry). The server checked: Has this key been processed? Once processed, you return directly to the last result, and the logic of the subsequent restart does not run.
- When the state of interruption is changed, the condition is added: before the actual writing operation is performed, an update with the condition is performed using the code - "only if the interruption is now pending, is changed to resolved". When both requests come in almost simultaneously, there will be only one successful update in the database (change line = 1); another change line = 0, indicating that someone has taken the lead, and this request does not reset the tool.
/ / Write in resume interface, through database client (indicated)
const result = await db.query(
`UPDATE interrupts
SET status = 'resolved', resolved_at = NOW()
WHERE id = $1 AND status = 'pending'`,
[interruptId]
)
if (result.rowCount === 0) {
// Processed, expired or cancelled
return alreadyHandledResponse()
}
/ /rowCount = = 1: This is the first request to be taken to allow the rereading of resources and the execution of restarts
Just remember: when the two requests come together, the first successful one will be selected by updating the above condition; only it will be able to continue the hazardous operation and the other will return directly.
SSE broke. Should we stop backstage?
It's not the same thing as "user point cancellation." Disconnection means only that the front end cannot hear progress; Run should stop and see the nature of the task:
- Query classes: Clients can cancel after disconnection, save for model and tool costs.
- publish, approvals, examples of restarts, etc.: usually to be unconnected with HTTP - Backstage Run continues to run and results are written into the database; users refresh and read back their status.
/ / Default only suspend; whether to cancel Run for strategy
/ Node: don't listen to close of req (possibly trigger if the POST reads the request); close of res
/ / Normal res.end() also triggers close: use!writableFinished to distinguish "customer breakup early"
res.on('close', () => {
Subscriber.detach (runId) // stopped; does not equal the cancellation of the user count
if (!res.writableFinished && runPolicy.cancelWhenNoSubscriber) {
/ / client-only early disconnect; writing operations / approval classes are not usually cancelled for disconnection
runController.cancel(runId)
}
})
How do you compare with LangGraph?
A lot of teams run Agent loops with LangGraph. It addresses the same category of issues, with a different division of labour:
- This is a foreign agreement: a session, a Run, a SSE business event, how to get back to the front end.
- LangGraph's running, midway. Function name, parameters are based on the version of the document you have installed, and here only the concept is taken into account to avoid the loss of the version as soon as it is changed.
One sentence: The LangGraph run cycle, breakpoint can be used, but the SSE event name, restoration of interface, assurance and protection against double-click still have to be determined by the text itself — not all of this is equal.
How do you measure it, the common error zone and the landing sequence?
Check these first.
SSE: Is it true that an incident was broken down into several pieces, several pieces were inserted in one piece, Chinese cross-sections were cut, the gateway was free timeout, the net was broken, the page was refreshed and the same step was drawn after reconnection.
HITL: Double-click "confirmation", confirm expiry, withdraw permission before reinstatement, change the example during the waiting period, and resume it after cancellation.
- Session context
Auto-fetched and translated by AI Atlas. All rights belong to the original author.