Week 2 · D11About 6 hours

The Run State Machine, Streaming Output Back, Ordering by runId, SSE Waiters, Merging Interruptions Within 30 Seconds

Build a run state machine for every conversation turn, stream the worker's output back in order by runId to the waiting SSE connection, and merge interruptions that arrive within 30 seconds.

Today's goals 0/3

能画出一个 run 从创建到完成的完整状态机（含失败、打断状态）
能实现按 runId 把 worker 的流式输出保序回传给等待中的 SSE 客户端
能实现 30 秒内的用户打断合并成同一次输入，而不是开两个 run

今日目标

能画出一个 run 从创建到完成的完整状态机（含失败、打断状态）
能实现按 runId 把 worker 的流式输出保序回传给等待中的 SSE 客户端
能实现 30 秒内的用户打断合并成同一次输入，而不是开两个 run

D10 把执行侧的顺序保住了，但用户那边还是「发出去就没声了」。今天把输出接回去，这条链路才算通。读完回来把上面三条勾掉。

小白版讲解

先把今天和 D7 的分界画清楚，因为整章都立在这条线上。D7 是同一个进程既生成又推送：模型吐一个字，同一个函数栈里立刻 res.write 一帧，生成和推送之间没有任何缝隙，所以那时候「顺序」根本不是问题——它天然就是对的。今天生成在 Worker、推送在 Gateway，中间隔着一条消息总线：字在一个进程里产生，连接挂在另一个进程上，中间那条总线不承诺你什么时候拿到、以什么顺序拿到。SSE 的服务端写法 D7 已经讲透了（四个响应头、: ping 心跳、连接断开处理），今天一行都不重讲，全部精力用在这条缝隙带来的四个新问题上：状态、顺序、多个订阅者、重连。

一次执行得有个身份：run 状态机

电台做一期节目：策划立项、进棚录制、开始播出、播完归档。任何时刻问「这期节目现在到哪一步了」，都有一个明确答案，而且这些步骤是单向的——播完了不可能退回去变成「正在录制」。

单进程时代你不需要这个东西。D7 那个 POST /chat 里，「执行到哪一步了」就是那个函数栈本身：函数在跑就是在跑，函数返回了就是结束了，函数抛异常了就是失败了。这份状态存在于进程内存里，没有名字，也不需要名字。

拆开之后它必须有名字，因为至少有三方要同时回答同一个问题：Gateway 要知道「这次执行还活着吗」才能决定挂不挂 SSE；Worker 要知道「这条消息是不是已经有人在做了」才能决定要不要执行；用户重新打开页面时，前端要知道「上一次问的问题还在生成吗」。三方不在同一个进程里，只能靠一张表对齐。这就是 D8 定的 runs 表，和它的六个状态：

TextText

       ┌──────────────────────────── 正常路径 ────────────────────────────┐
       │                                                                 │
  ┌─────────┐   投递被消费   ┌─────────┐  第一个片段  ┌───────────┐        ▼
  │ pending │ ────────────→ │ running │ ──────────→ │ streaming │ ───→ ┌──────┐
  └─────────┘               └─────────┘             └───────────┘      │ done │
       │                         │                        │            └──────┘
       │                         ▼                        ▼
       │                    ┌────────┐              ┌────────┐
       └──────────────────→ │ failed │              │ failed │   重试耗尽
                            └────────┘              └────────┘
       │                         │                        │
       └──────────────────→ ┌───────────┐ ←──────────────-─┘
                            │ cancelled │   被打断合并或用户取消
                            └───────────┘

四个正常状态一路向前，两个异常出口随时可以走：failed 是重试耗尽，cancelled 是被打断合并或用户主动取消。running 和 streaming 要分开，是因为它们对用户的含义完全不同：running 是「已经有 Worker 领走了，但还没有一个字」，streaming 是「第一个字已经出来了」。这条分界线就是首字延迟（time-to-first-token）的观测点，也是前端决定「继续转圈」还是「开始打字机效果」的依据。

状态机的价值不在于把这几个词列出来，而在于把不合法的转换写进代码里挡住：

state.js

// 转换表，不是一串 if：它能被读、能被测，也能直接贴进设计文档
const TRANSITIONS = {
  pending: ['running', 'cancelled', 'failed'],
  running: ['streaming', 'done', 'failed', 'cancelled'],
  streaming: ['done', 'failed', 'cancelled'],
  done: [], // 终态没有出边，这是整张表最值钱的一行
  failed: [],
  cancelled: [],
}
 
function transition(from, to) {
  if (!TRANSITIONS[from].includes(to)) {
    throw new Error(`非法状态转换：${from} → ${to}`)
  }
  return to
}

state.py

from enum import StrEnum
 
 
class RunStatus(StrEnum):
    PENDING = "pending"
    RUNNING = "running"
    STREAMING = "streaming"
    DONE = "done"
    FAILED = "failed"
    CANCELLED = "cancelled"
 
 
# frozenset 表达「这是一张不会被改的表」，比 list 更贴合语义
TRANSITIONS: dict[RunStatus, frozenset[RunStatus]] = {
    RunStatus.PENDING: frozenset({RunStatus.RUNNING, RunStatus.CANCELLED, RunStatus.FAILED}),
    RunStatus.RUNNING: frozenset({RunStatus.STREAMING, RunStatus.DONE, RunStatus.FAILED, RunStatus.CANCELLED}),
    RunStatus.STREAMING: frozenset({RunStatus.DONE, RunStatus.FAILED, RunStatus.CANCELLED}),
    RunStatus.DONE: frozenset(),
    RunStatus.FAILED: frozenset(),
    RunStatus.CANCELLED: frozenset(),
}
 
 
def transition(current: RunStatus, target: RunStatus) -> RunStatus:
    if target not in TRANSITIONS[current]:
        raise ValueError(f"非法状态转换：{current} → {target}")
    return target

RunStatus.java

// Java 里状态机的地道写法是把出边挂在枚举自己身上，用 EnumSet 存——
// 位图实现，查一次是常数时间，而且忘了给新状态配出边编译期就能发现
enum RunStatus {
    PENDING, RUNNING, STREAMING, DONE, FAILED, CANCELLED;
 
    Set<RunStatus> allowedNext() {
        return switch (this) {
            case PENDING -> EnumSet.of(RUNNING, CANCELLED, FAILED);
            case RUNNING -> EnumSet.of(STREAMING, DONE, FAILED, CANCELLED);
            case STREAMING -> EnumSet.of(DONE, FAILED, CANCELLED);
            case DONE, FAILED, CANCELLED -> EnumSet.noneOf(RunStatus.class);
        };
    }
 
    RunStatus to(RunStatus target) {
        if (!allowedNext().contains(target)) {
            throw new IllegalStateException("非法状态转换：" + this + " → " + target);
        }
        return target;
    }
}

RunStatus.swift

// Swift 用穷尽 switch：以后往枚举里加一个状态，这个 switch 编译不过，
// 编译器会逼你回来补出边——四门语言里它对「漏写一个状态」的防护最强
// 出边表与 D8 那份完全一致（含 pending → failed）。区别只在方法：
// D8 给的是 canTransition(to:) -> Bool，今天升级成 transition(to:) throws——
// 状态机要真的挡住非法迁移，返回 Bool 太容易被调用方忽略掉
enum RunStatus: String, Codable {
    case pending, running, streaming, done, failed, cancelled
 
    var allowedNext: Set<RunStatus> {
        switch self {
        case .pending: [.running, .cancelled, .failed]
        case .running: [.streaming, .done, .failed, .cancelled]
        case .streaming: [.done, .failed, .cancelled]
        case .done, .failed, .cancelled: []
        }
    }
 
    func transition(to target: RunStatus) throws -> RunStatus {
        guard allowedNext.contains(target) else {
            throw RunError.illegalTransition(from: self, to: target)
        }
        return target
    }
}

那句 throw 换来的是一类事故被挡在写入那一刻。分布式系统里「一个已经 done 的 run 又收到一个片段」是常态而不是意外——总线至少投递一次（D9），网络抖一下就会重放一条旧消息，慢半拍的 Worker 也可能在收尾之后又醒过来写一笔。没有这张表，那一笔会安静地写进库里，用户看到回复末尾莫名多出半句话，你查半天日志找不出是谁写的；有这张表，它当场抛错、当场进日志。

状态机真正的产出不是那几个词，是一条「凡是写状态都必须过这个函数」的纪律。 绕过它写一次 UPDATE runs SET status = 'done'，这张表就退化成注释了。

那么第二个问题来了：Worker 在自己的进程里一个字一个字地生成，这些字要怎么按顺序回到 Gateway 上那条还挂着的连接？

保序回传：seq 从 0 开始，一号都不许跳

电台节目是按时间顺序播出的，第 12 分钟的内容不会跑到第 3 分钟前面去。但今天的「播出」不是一个人从头做到尾——录制在录音棚（Worker），播出在导播台（Gateway），中间隔着一条线路。线路不保证你收到的顺序就是录制的顺序。

做法是固定的，一句话：Worker 每产出一个片段，就带着 runId + seq 写进 koda:out:{runId} 这条流；Gateway 侧的 SSE 处理器按 seq 顺序推。 三个细节都不能改：

一是每个 run 一条自己的输出流，而不是所有 run 挤在一条流上让 Gateway 自己筛。流名里带 runId，Gateway 只订阅它关心的那一条，读到的每一条都是自己要的；否则一万个并发对话意味着每条连接都要扫描一万倍的数据量。

二是seq 从 0 开始、连续、不跳号。这不是洁癖，是断线重连能续上的前提：客户端说「我最后收到 5 号」，Gateway 要能确定「那就从 6 号开始」。如果 seq 是时间戳、是随机 id、或者中间会跳号，这句推理立刻不成立。

三是SSE 的 id: 字段就写 seq。这样客户端不需要额外记账，浏览器原生的 EventSource 甚至会自动把它放进重连请求的 Last-Event-ID 头里。一帧报文长这样：

TextText

id: 41
event: delta
data: {"text":"运"}
 
: ping
 
id: 42
event: delta
data: {"text":"单"}

心跳还是 D7 那条 15 秒的注释行，原样照抄，不必改动。

Gateway 侧的核心是一个小小的保序器。它只做一件事：只按号交付，缺号就先攒着。

orderer.js

class SeqOrderer {
  constructor(from, emit) {
    this.next = from // 下一个该交付的号
    this.buffer = new Map()
    this.emit = emit
  }
 
  offer(seq, text) {
    if (seq < this.next) return // 已经推过：重连回放与流上的片段必然重叠
    this.buffer.set(seq, text)
    // 缺号就停在这里等，等到齐了一次性交付出去
    while (this.buffer.has(this.next)) {
      this.emit(this.next, this.buffer.get(this.next))
      this.buffer.delete(this.next)
      this.next += 1
    }
  }
}

orderer.py

from collections.abc import Callable
 
 
class SeqOrderer:
    def __init__(self, start: int, emit: Callable[[int, str], None]) -> None:
        self._next = start
        self._buffer: dict[int, str] = {}
        self._emit = emit
 
    def offer(self, seq: int, text: str) -> None:
        if seq < self._next:
            return  # 已经推过：重连回放与流上的片段必然重叠
        self._buffer[seq] = text
        # 用 walrus 一步「取出并判断」，缺号时循环自然停下
        while (ready := self._buffer.pop(self._next, None)) is not None:
            self._emit(self._next, ready)
            self._next += 1

SeqOrderer.java

// TreeMap 按键有序，缺号时 firstKey 一眼就能看出「等的是哪一号」，
// 顺手还能拿到「攒了多少号没交付」这个可以直接上报的指标
final class SeqOrderer {
    private final TreeMap<Integer, String> buffer = new TreeMap<>();
    private final BiConsumer<Integer, String> emit;
    private int next;
 
    SeqOrderer(int from, BiConsumer<Integer, String> emit) {
        this.next = from;
        this.emit = emit;
    }
 
    void offer(int seq, String text) {
        if (seq < next) return; // 已经推过
        buffer.put(seq, text);
        for (String ready = buffer.remove(next); ready != null; ready = buffer.remove(next)) {
            emit.accept(next, ready);
            next++;
        }
    }
}

SeqOrderer.swift

// 用 struct + mutating：调用方持有唯一一份状态，不可能被两个地方同时改坏
struct SeqOrderer {
    private var next: Int
    private var buffer: [Int: String] = [:]
    private let emit: (Int, String) -> Void
 
    init(from: Int, emit: @escaping (Int, String) -> Void) {
        self.next = from
        self.emit = emit
    }
 
    mutating func offer(seq: Int, text: String) {
        guard seq >= next else { return } // 已经推过
        buffer[seq] = text
        while let ready = buffer.removeValue(forKey: next) {
            emit(next, ready)
            next += 1
        }
    }
}

这段代码只有十几行，工程代价却在别处：攒着的片段要不要设上限？ 如果 5 号迟迟不到，6 号到 500 号全在内存里排队，一万条连接同时这样就是一次内存事故。生产做法是给缓冲设一个上限（比如 64 个片段）和一个等待上限（比如 2 秒），超了就认为 5 号丢了，从库里补读一次；补不到就发一个 error 事件让客户端重连。能等，但不能无限等，这是所有保序缓冲的通用纪律。

还有一件事必须说清：Worker 写片段是双写——既进输出流，也进 messages 表（D8 定的 unique(run_id, seq) 正好在这里派上用场）。因为它们服务两种人：流是给「现在正挂着的连接」用的，库是给「等一下才回来的人」用的。 只有流，重连时片段早被消费掉了；只有库，就得轮询查库，首字延迟从几十毫秒涨到几百毫秒。代价是写放大：一次回答 500 个片段就是 500 行，生产里会攒批落库（每 20 到 40 个片段或每 200 毫秒一次），实验里为了看清保序才一个片段一行。

一个频道，很多听众：多个 SSE 等待者

同一档节目可以有很多人同时收听，电台不会因为第二个人打开收音机就重新录一遍。

对应到 Agent 上：同一个 run 完全可能被多处订阅。用户在手机上问完，顺手在电脑上打开了同一个会话；网页开了两个标签页；断线重连的一瞬间，旧连接还没被服务端察觉、新连接已经建上来了——这几秒里同一个 run 上真的挂着两条 SSE。

关键判断只有一条：run 是执行的单位，SSE 连接是观看的单位，两者不是一对一。 想清楚这句话，很多看似棘手的问题就消失了：不需要「谁先连上谁独占」的锁，也不需要让第二个连接排队等第一个断开。每条连接各自维护自己的读取位置和自己的保序器，各读各的，互不影响。这也顺带解释了为什么输出流的内容要用 XREAD 这种「旁观者读法」而不是消费组——消费组是分摊，一条消息只给一个消费者；这里要的是广播，每个订阅者都得看到全部。 用消费组去做扇出，结果就是两条连接各拿到半段话。

扇出有两种做法。一是每条连接各自去读流：最简单，代价是同一批数据在 Redis 侧被读了 N 次。二是一个 Gateway 进程只读一次，再在进程内广播给本地订阅者：省掉重复读取，但要自己维护订阅者表、要处理「最后一个订阅者走了怎么停」，跨实例依然要各读一次。判据是每个 run 的平均订阅者数——多数产品里它接近 1，那就选第一种，别为不存在的规模先写一层订阅者管理。

主持人还没播完，新点播来了：30 秒打断合并

直播中主持人正在念一封听众来信，这时候同一个听众又发来一条「等等，我刚说错了，是第三首歌不是第二首」。正常主持人的做法是把这条并进本期节目，而不是另开一期节目同时播——两期节目同时占着一个频率，听众听到的只会是一团噪音。

用户在 Agent 面前的行为一模一样：发出去三秒钟，看到回复刚出来两行，突然想起漏说了一个条件，于是又发一条。如果你老老实实为第二条消息新建一个 run，会同时出现两个后果：两个回答同时往同一条会话里写（前端看到两段交错的文字），以及第一个回答基于不完整的信息，注定是错的。

判据要写死，三个条件全中才合并：同一个 sessionId、上一个 run 处于 running 或 streaming、距它创建不到 30 秒。命中就把新消息追加进同一个 run 的输入，并把它标记为需要重跑，不新建 run；超过 30 秒、或者上一个 run 已经 done，就正常新建。

merge.js

const MERGE_WINDOW_MS = 30_000
 
function shouldMerge(prev, now) {
  if (!prev) return false
  // 只合并「正在干活」的：pending 只存在几毫秒，done 的说完了合并没有意义
  if (prev.status !== 'running' && prev.status !== 'streaming') return false
  return now - prev.createdAt < MERGE_WINDOW_MS
}

merge.py

from datetime import UTC, datetime, timedelta
 
MERGE_WINDOW = timedelta(seconds=30)
ACTIVE = {RunStatus.RUNNING, RunStatus.STREAMING}
 
 
def should_merge(prev: Run | None, now: datetime | None = None) -> bool:
    if prev is None or prev.status not in ACTIVE:
        return False
    # 用 timedelta 而不是裸秒数：单位写在类型里，就不会有人误传毫秒
    return (now or datetime.now(UTC)) - prev.created_at < MERGE_WINDOW

MergePolicy.java

// Instant + Duration 是 JDK 的时间语义，别用 long 存毫秒——
// 那样调用方永远搞不清传进来的是秒、毫秒还是纳秒
record MergePolicy(Duration window) {
    static final MergePolicy DEFAULT = new MergePolicy(Duration.ofSeconds(30));
 
    boolean shouldMerge(Run prev, Instant now) {
        if (prev == null) return false;
        var active = switch (prev.status()) {
            case RUNNING, STREAMING -> true;
            default -> false;
        };
        return active && Duration.between(prev.createdAt(), now).compareTo(window) < 0;
    }
}

MergePolicy.swift

struct MergePolicy {
    static let window: TimeInterval = 30
 
    // guard let 把「没有上一个 run」这条最常见的分支挑出去，主逻辑只剩一行
    static func shouldMerge(previous: Run?, now: Date = .now) throws -> Bool {
        guard let previous, previous.status == .running || previous.status == .streaming
        else { return false }
        // createdAt 是 Fluent 的 @Timestamp，类型是 Date?——用 D8 收口好的 requireCreatedAt()
        return try now.timeIntervalSince(previous.requireCreatedAt()) < window
    }
}

两个实现细节值得单独说。第一，「需要重跑」这个标记不要写进 runs 表。它只在本次执行期间有意义，属于控制信号而不是持久状态；写进业务表，进程崩在半路就会留下一个脏标记，重启后那个 run 会无限重跑。放在一个带过期时间的 Redis 键上，进程没了它自己就消失。第二，Worker 要在片段之间的间隙检查这个标记——不是每毫秒轮询，而是每吐出一个片段顺手看一眼。这样最坏的响应延迟就是一个片段的生成时间，几十毫秒，用户感觉不到。重跑时 seq 必须接着往上加、不能重置，否则重连的人按 Last-Event-ID 续号会续到一段已经作废的历史上。

那 30 秒是怎么来的？它不是算出来的，是一个可调的产品判断：短了合并不到（用户改口通常在 5 到 15 秒之间），长了会把「一个新问题」误合成「上一个问题的补充」。要点是这个数必须在一个地方定义、被三处引用（判据、标记的过期时间、前端的输入框提示），不要在代码里散落三个 30000。

顺带算一笔账，好让你知道打断合并不是一次成本优化。一次回答按输入 2000 token、输出 500 token 估：输入 2000 除以一百万再乘 0.15 美元约等于 0.0003 美元，输出 500 除以一百万再乘 0.60 美元也约等于 0.0003 美元，合计约 0.0006 美元。不合并的话两个 run 各跑完一遍，约 0.0012 美元；合并的话第一遍在三分之一处被掐掉（约 0.0004 美元）加第二遍完整的 0.0006 美元，约 0.0010 美元。省下来的那 17% 摊到一天一万次改口也就两美元。所以合并的理由从来不是钱，是不让两个回答同时对着用户说话。 面试时把这笔账主动算出来，比说「为了节省成本」有力得多。

中途调进来，从录音第 N 秒续听：幂等回放

电台的听众可以中途调进来。如果只有直播，他就只能从当下听；但节目同时在录音，他就能从第 N 秒接着听。今天的重连靠的正是那份录音——写进 messages 表的每一行。

流程只有三步，但每一步都有一个精确的坑。第一步，算出该从哪一号开始。 客户端带回来的 Last-Event-ID 是它最后收到的那一号，不是它想要的下一号，所以要加一。少加一，重连后的第一帧是重复的；多加一，用户丢掉一个字。这一行是整段重连逻辑里唯一的算术，也是最常写错的地方。第二步，先从库里回放。 库里一定是全的，先把缺的补齐。第三步，再接上输出流。 流上还在飘的片段和刚回放的必然重叠，靠保序器那句「小于当前指针的一律丢弃」去重——这就是幂等回放的全部秘密，一次比较而已。

resume.js

function resumeFrom(lastEventId) {
  if (!lastEventId) return 0 // 首次连接
  const n = Number(lastEventId)
  return Number.isInteger(n) && n >= 0 ? n + 1 : 0 // 带回来的是「最后收到」，所以加一
}
 
// 先回放、再接流，两边重叠部分由 orderer 自己丢弃
const from = resumeFrom(req.headers['last-event-id'])
const orderer = new SeqOrderer(from, (seq, text) => sendEvent(res, 'delta', { text }, String(seq)))
for (const row of await store.listDeltas(runId, from)) orderer.offer(row.seq, row.content)
await subscribe(outStream(runId), (msg) => orderer.offer(Number(msg.seq), msg.text))

resume.py

def resume_from(last_event_id: str | None) -> int:
    if not last_event_id or not last_event_id.isdigit():
        return 0
    return int(last_event_id) + 1  # 带回来的是「最后收到」的那一号
 
 
start = resume_from(request.headers.get("last-event-id"))
orderer = SeqOrderer(start, lambda seq, text: send_event(response, "delta", text, seq))
# 先把库里错过的补齐，再接上还在流动的那部分
for row in await store.list_deltas(run_id, start):
    orderer.offer(row.seq, row.content)
async for msg in bus.subscribe(out_stream(run_id)):
    orderer.offer(int(msg["seq"]), msg["text"])

Resume.java

// Optional 链式处理「头可能不存在、也可能不是数字」，避免嵌套判空
static int resumeFrom(String lastEventId) {
    return Optional.ofNullable(lastEventId)
        // 限长到 9 位：Last-Event-ID 是客户端可控的头，"\\d+" 放过一串 30 位数字，
        // 后面的 parseInt 会抛 NumberFormatException，一条 curl 就能把这个接口打成 500
        .filter(s -> s.matches("\\d{1,9}"))
        .map(s -> Integer.parseInt(s) + 1) // 客户端给的是「最后收到」的号
        .orElse(0);
}
 
int from = resumeFrom(request.getHeader("Last-Event-ID"));
var orderer = new SeqOrderer(from, (seq, text) -> sink.next(delta(seq, text)));
store.listDeltas(runId, from).forEach(row -> orderer.offer(row.seq(), row.content()));
bus.subscribe(outStream(runId)).subscribe(msg -> orderer.offer(msg.seq(), msg.text()));

Resume.swift

// 三个前提（头存在、是数字、非负）用一个 guard 一次性挡掉，主逻辑只剩「加一」这一行。
// Java 版那边是 Optional 链式写法，Swift 这边 guard let 更地道——同一件事两种表达
func resumeFrom(_ lastEventID: String?) -> Int {
    guard let raw = lastEventID, let n = Int(raw), n >= 0 else { return 0 }
    return n + 1 // 客户端带回来的是它最后收到的那一号
}
 
let from = resumeFrom(request.headers.first(name: "Last-Event-ID"))
var orderer = SeqOrderer(from: from) { seq, text in writer.send(delta: text, id: seq) }
for row in try await store.listDeltas(runID: runID, from: from) {
    orderer.offer(seq: row.seq, text: row.content)
}
for try await msg in bus.subscribe(outStream(runID)) {
    orderer.offer(seq: msg.seq, text: msg.text)
}

最后回到全局：D8 把状态搬进数据库，D9 把执行搬进总线，D10 保住同用户的处理顺序，今天把输出的顺序与体验接回来。分布式化的每一步都在拆掉一个「顺手就有」的保证，然后你得显式地重建它一遍——顺序、状态、身份、恰好一次，单进程里免费，跨进程之后每一个都要花代码去买。某生产级 IM Agent 平台也是把这几层踩实之后，功能才有地方长。

源码导读

动手实验

🧪 D11 实验：完整聊天往返 + 打断合并

Code location: labs/agent-30days/day-11-run-state-machine

验收标准：

MOCK=1 SELFTEST=1 pnpm start 的六项自检全是 ✅。starter/ 原样跑是第 3、6 项 ✅，第 1、2、4、5 项 ❌，四个练习点各对应一项。
第 2 项的 seq 前 12 号是 0,1,2,3,4,5,6,7,8,9,10,11，拼回来的回复是通顺的一句话；starter/ 会看到 0,1,2,3,5,4,… 和一句被打乱的话。
第 4 项显示「断开前收到 6 个（最后一号 5），重连后第一号 6，合起来无重复、无缺号」——不丢也不重，两件事同时成立。
第 5 项显示第二条消息 merged=true、仍是同一个 run、输出里出现「换货」、seq 仍连续，日志里能看到 Worker 打印「带合并后的输入重跑，seq 从 N 继续」。
pnpm typecheck 通过、没有 any；docker compose up -d 之后带上 REDIS_URL 与 DATABASE_URL 再跑一次同一条命令，六项结果与内存版完全一致。

这个实验第一次把三段接成闭环，src/ 里 Gateway、Worker、基础设施都有——但你只需要动四个函数，全在 src/shared/ 下的两个文件里：状态转换表、保序器、Last-Event-ID 换算、打断合并判据。基础设施走端口加适配器：MOCK=1 下 Redis 与 Postgres 都是内存实现（不是打桩，是把语义写出来），设了 REDIS_URL 与 DATABASE_URL 就换成 ioredis 与 pg，业务代码一行不改。先原样跑一次自检，那句被打乱的模拟回复会让你一眼看懂「不保序」长什么样。

原样跑 MOCK=1 SELFTEST=1 pnpm start，记住第 1、2、4、5 项 ❌ 的样子，特别是第 2 项那句被打乱的回复。
补齐状态转换表，让第 1 项变 ✅：非法转换（done 到 streaming、pending 到 streaming）必须被拒。
补齐保序器的三件事——丢弃旧号、缺号入缓冲、连号批量交付，第 2 项的 seq 应当变成从 0 开始连续。
补齐 Last-Event-ID 换算，第 4 项要同时满足「重连后第一号 = 最后收到的号加一」和「合起来无重复无缺号」。
补齐打断合并判据，第 5 项变 ✅ 之后确认第 6 项仍然 ✅——超过 30 秒必须新建，两者是同一个判据的两侧。

面试题

今天 4 道题在下方题库区，覆盖 run 状态机的设计、多订阅者的有序性、打断合并的产品与工程取舍、断线重连的不丢不重。展开后先看"分析过程"再看要点——第 4 题的追问（保留期与回放上限）是这一章最容易被追到底的地方，别跳过。

检查清单与明日预告

能画出一个 run 从创建到完成的完整状态机（含失败、打断状态）
能实现按 runId 把 worker 的流式输出保序回传给等待中的 SSE 客户端
能实现 30 秒内的用户打断合并成同一次输入，而不是开两个 run
能说清为什么输出流用广播读法而输入流用消费组，以及用错会出什么现象
能解释「seq 从 0 开始、连续、不跳号」为什么是断线重连的前提
实验的 5 条验收标准全部通过（六项自检全 ✅）
4 道面试题不看要点也能答出至少 3 道

明天（D12）我们给这个 Agent 装长期记忆。为什么是现在？因为链路到今天才算通——用户发一句话能拿到有序、可重连、会合并的回答。但这个 Agent 只记得当前会话：上周他说过「我住的地方没有电梯，大件包裹请放物业」，今天再来问，它一无所知。D6 解决的是「这一轮塞不下」，明天解决的是「上个月说过的事想不起来」，不是同一个问题，也不是同一套机制。

Interview questions

How would you design the state machine for one agent run, and which failure states must it cover?怎么设计一次 Agent 执行（run）的状态机？需要覆盖哪些异常状态？
Common in ChinaCommon overseasBasic#state-machine#distributed-systems
How to reason about it · think before answering
The discriminator is not listing states, it is explaining why a single-process service does not need them at all. Without that, you have only memorized a diagram.
Start from motivation: in one process the call stack *is* the state. Once you split gateway and worker, three parties must answer the same question independently — the gateway decides whether to keep an SSE connection open, the worker decides whether someone already claimed the message, and a reopened browser tab asks whether the previous question is still generating. Different processes, so the answer has to live in a table.
Then the states: pending to running to streaming to done on the happy path, with failed (retries exhausted) and cancelled (superseded by a merge, or user-cancelled) as exits available from anywhere. Volunteer why running and streaming are separate: running means claimed but no token yet, streaming means the first token is out. That boundary is your time-to-first-token probe and the frontend's cue to switch from spinner to typewriter.
Land on the real purpose: the machine exists to reject writes. Terminal states having no outgoing edges is the most valuable row in the table. Under at-least-once delivery, a done run receiving one more chunk is routine, and without the table that chunk lands silently — the user sees half a sentence appended and the logs show nothing wrong.
Add the discipline that separates shipped from read-about: every status write goes through one transition function. One raw UPDATE that bypasses it and the state machine is just a comment.
Expect the follow-up on storage and concurrency: the database row is the single source of truth, and transitions are conditional updates that include the expected current status in the WHERE clause. Zero rows affected means someone moved first — re-read and decide, never blindly overwrite.
分析过程 · 先想清楚再作答
这题的区分度不在「能不能列出几个状态」，而在你有没有说出「为什么单进程时代不需要它」。答不出这一点，说明你只是抄过一张状态图。
先给动机：单进程里「执行到哪一步了」就是那个函数栈，状态存在于进程内存里，不需要名字。拆成 Gateway 与 Worker 之后，至少三方要同时回答同一个问题——接入层要判断还挂不挂 SSE，执行层要判断这条消息是否已被人领走，前端重开页面要判断上次的问题还在不在生成。三方不同进程，只能靠一张表对齐。
再给状态：pending 到 running 到 streaming 到 done 是正常路径，failed（重试耗尽）与 cancelled（被打断合并或用户取消）是两个随时可以走的异常出口。主动说明为什么 running 和 streaming 要分开：前者是「有人领走了但还没有一个字」，后者是「第一个字已出来」，这条线就是首字延迟的观测点，也是前端决定转圈还是打字机的依据。
结论要落到「状态机是用来挡写入的」：终态没有出边这一条最值钱。至少一次投递下「已经 done 的 run 又收到一个片段」是常态，没有转换表，那一笔会安静地写进库，用户看到回复末尾多出半句话，而日志里查不出是谁写的。
补一条纪律，这是有没有落地过的分水岭：所有写状态的地方都必须过同一个转换函数。绕过它直接执行一条更新语句，状态机就退化成注释了。
可以预期的追问：状态存哪、并发怎么办？答数据库那一行是唯一真相，转换用带条件的更新（更新时把当前状态写进 where 子句），失败说明有人抢先改过，这时候重读再决定，而不是覆盖。
Key points
In one process the call stack is the state; after splitting gateway and worker, three parties need the same answer, so it has to be a table
Happy path pending, running, streaming, done; exits are failed (retries exhausted) and cancelled (merged or user-cancelled)
Separating running from streaming gives you a time-to-first-token probe and tells the UI when to switch from spinner to typewriter
Terminal states with no outgoing edges reject the late chunks that at-least-once delivery guarantees you will get
Every status write goes through one transition function, implemented as a conditional update on the expected current status
答题要点
单进程里状态就是函数栈；拆成 Gateway 与 Worker 后有三方要独立回答「这次执行到哪了」，必须落成一张表
正常路径 pending 到 running 到 streaming 到 done；异常出口 failed（重试耗尽）与 cancelled（打断合并或用户取消）
running 与 streaming 分开，是为了观测首字延迟，也让前端知道该转圈还是该开始打字机效果
终态没有出边是核心：至少一次投递下的迟到片段会被当场挡住，而不是安静写进库
纪律：所有状态写入都过同一个转换函数，并用带当前状态条件的更新来处理并发
Several clients subscribe to the same run's streaming output at once. How do you guarantee each of them receives the full content in order?多个客户端同时订阅同一次执行的流式输出，怎么保证每个客户端都收到完整且有序的内容？
Common in ChinaCommon overseasIntermediate#sse#ordering#fan-out
How to reason about it · think before answering
Two words carry the question: complete and ordered. Most candidates answer only ordering and drop completeness — which is the half that is easy to get structurally wrong, because it depends on which read primitive you pick.
Set the frame first: a run is the unit of execution, a connection is the unit of viewing, and they are not one-to-one. Phone plus laptop, two browser tabs, or the overlap window during a reconnect all put multiple streams on one run. Once that is clear, 'first connection wins the lock' schemes fall away on their own.
Name the trap in 'complete': broadcast reads and consumer groups are different semantics. A consumer group divides work — each message goes to exactly one consumer — while here every subscriber must see everything. Using a consumer group for fan-out gives you two connections each holding half the answer, and that is the classic wrong answer here.
Then ordering: every chunk carries a sequence number starting at 0, contiguous, never skipping, and is written to the stream. The reader keeps a 'next to deliver' cursor, discards anything below it, buffers anything above it, and flushes contiguous runs. Put the same number in the SSE id field so the client keeps no separate bookkeeping.
Volunteer the cost: the reorder buffer needs bounds. If chunk 5 is late, 6 onward pile up in memory, and ten thousand connections doing that is an outage. Cap the buffer size and the wait, then backfill the gap from the database, and if that fails emit an error event and let the client reconnect. Wait, but never wait forever.
Expect the fan-out follow-up: either every connection reads the stream itself (simple, at the cost of reading the same data N times) or one read per process broadcast to local subscribers (fewer reads, but you now own a subscriber registry, teardown when the last one leaves, and still one read per instance). Decide by average subscribers per run — usually close to one, so take the simple path.
分析过程 · 先想清楚再作答
题眼有两个词：完整、有序。很多人只答有序，漏掉完整——而「完整」那一半恰好是最容易设计错的，因为它取决于你用了哪种读法。
先给一句能定调的判断：一次执行是执行的单位，一条连接是观看的单位，两者不是一对一。手机和电脑同开、两个标签页、重连瞬间新旧连接并存，都会让同一次执行上挂着多条流。想清楚这句话，「谁先连谁独占」这种锁的方案就自然被排除了。
接着点出「完整」的真正机关：广播读法与消费组是两种语义。消费组是分摊，一条消息只给一个消费者；这里要的是广播，每个订阅者都要看到全部。用消费组做扇出，结果就是两条连接各拿到半段话——这是这道题最常见的错误答案。
再答「有序」：每个片段带一个从 0 开始、连续、不跳号的序号，写进流；接收侧维护「下一个该交付的号」，小于它的丢弃，大于它的先入缓冲，连号了再批量推出去。序号同时写进 SSE 的 id 字段，客户端不用另记一套账。
然后是必须主动说的工程代价：缓冲要有上限。如果 5 号迟迟不到，6 号往后全在内存里排队，一万条连接同时这样就是一次内存事故。做法是给缓冲设条数上限和等待上限，超时就从库里补读，补不到就发 error 让客户端重连——能等，但不能无限等。
可以预期的追问：扇出实现怎么选？两种——每条连接各自去读一遍流（简单，代价是同一批数据被读 N 次），或进程内只读一次再广播给本地订阅者（省读取，但要维护订阅者表、要处理最后一个订阅者离开，跨实例仍要各读一次）。判据是每次执行的平均订阅者数，多数产品接近 1，那就选前者，别为不存在的规模提前写一层。
Key points
A run is the unit of execution and a connection is the unit of viewing; they are not one-to-one, so no first-wins lock is needed
Read the output stream as a broadcast, not through a consumer group — a group divides messages and leaves each connection with half the answer
Tag every chunk with a contiguous sequence starting at 0; the reader discards older, buffers newer, and flushes contiguous ranges
Mirror that sequence into the SSE id field so clients need no extra bookkeeping and can resume from it
Bound the reorder buffer by size and time, backfill gaps from the database, and fall back to an error event plus reconnect
答题要点
一次执行是执行单位、一条连接是观看单位，两者不是一对一，不需要「谁先连谁独占」的锁
输出流必须用广播读法而不是消费组：消费组是分摊，会让两条连接各拿到半段话
每个片段带从 0 开始、连续、不跳号的序号，接收侧按序交付：小于当前号丢弃、大于当前号入缓冲、连号批量推
序号同时写进 SSE 的 id 字段，客户端不必自己记账，也是重连续号的依据
缓冲必须有条数与时间上限，超时从库里补读，补不到就发 error 让客户端重连
A user sends another message while the agent is still answering the previous one. How should the system handle it?用户在 Agent 还没回复完的时候又发来一条消息，应该怎么处理？
Common in ChinaCommon overseasIntermediate#interrupt-merge#state-machine#cost
How to reason about it · think before answering
It reads like a product question but tests whether you have thought through two concurrent runs. 'Queue it' or 'cancel the previous one' are not wrong, just incomplete — they want the criteria and the costs.
Start with what happens if you ignore it: two runs write into the same conversation, so the UI shows two interleaved answers, and the first run was computed from incomplete input, so its answer is already wrong. Those two consequences point straight at merging rather than concurrency.
Then give the actual test — all three must hold: same session, the previous run is running or streaming, and it was created less than 30 seconds ago. On a hit, append the new message to that run's input and flag it for a rerun instead of creating a new run; outside the window, or if the previous run finished, create a new one. Excluding pending is deliberate: that window lasts milliseconds, and excluding it keeps the rule free of races with the worker reading the input.
Two implementation details show hands-on experience. First, the rerun flag does not belong in the business table — it is meaningful only during this execution, and persisting it means a crash mid-flight leaves a dirty flag that makes the run loop forever after restart; an expiring key is the right home. Second, on rerun the sequence must keep counting up rather than resetting, or a reconnecting client resuming from its last id lands in a history that has been invalidated.
Volunteer the arithmetic to kill the 'saves money' answer: at roughly 2000 input and 500 output tokens, one answer costs about $0.0006. Not merging means two full runs, about $0.0012; merging means a first pass cut off a third of the way in (about $0.0004) plus a full second pass ($0.0006), about $0.0010 — a 17% saving, which is two dollars a day even at ten thousand corrections. Merging is a user-experience decision, not a cost optimization.
Expect: where does 30 seconds come from? It is a product judgment, not a derivation — corrections usually arrive 5 to 15 seconds in, too short misses them and too long merges genuinely new questions into old ones. What matters is defining it once and referencing it from both the rule and the UI hint rather than scattering the constant.
分析过程 · 先想清楚再作答
这题看起来是产品题，其实考的是你有没有想过「并发两次执行」的后果。答「排队处理」或「直接取消上一条」都不算错，但都不完整——面试官想听的是判据和代价。
先说清不处理会怎样：两次执行同时往同一个会话里写输出，前端看到两段交错的文字；而且第一次执行是基于不完整的信息跑的，它的答案注定要被推翻。这两条后果一说，方案的方向就定了——要合并，不要并发。
然后给可执行的判据，三个条件全中才合并：同一个会话、上一次执行正处于 running 或 streaming、距它创建不到 30 秒。命中就把新消息追加进同一次执行的输入并标记为需要重跑，不新建；超窗或上一次已完成就正常新建。把 pending 排除掉是有意的——那段窗口只有几毫秒，排除后判据不必考虑「执行侧正好在这一刻读输入」的竞态。
两个实现细节最能体现动手过：一是「需要重跑」这个标记不要写进业务表，它只在本次执行期间有意义，写进表里进程崩在半路就留下脏标记、重启后无限重跑，放一个带过期时间的键上更合适；二是重跑时序号必须接着往上加、不能重置，否则重连的客户端按上次收到的号续，会续到一段已经作废的历史上。
主动算一笔账，把「为了省钱」这个错误理由挡回去：按输入 2000、输出 500 个 token 估，单次约 0.0006 美元；不合并是两次跑完约 0.0012 美元，合并是第一遍被掐在三分之一处约 0.0004 美元加第二遍 0.0006 美元约 0.0010 美元，只省 17%，一天一万次改口也就两美元。所以合并的理由是体验，不是成本。
可以预期的追问：30 秒怎么定的？答它是产品判断不是推导结果——用户改口通常在 5 到 15 秒之间，窗口太短合并不到、太长会把新问题误并成补充；关键是这个数只在一处定义、被判据与前端提示共同引用，不要在代码里散落三份。
Key points
Without merging you get two interleaved answers in one conversation, and the first was computed from incomplete input
Merge only when all three hold: same session, previous run running or streaming, created under 30 seconds ago; otherwise create a new run
On a merge, append to the same run's input and flag a rerun, keeping that flag in an expiring key rather than the business table
Sequence numbers keep counting on rerun and are never reset, or reconnects resume into an invalidated history
The cost saving is small (about 17%); the real reason is to avoid two answers talking over each other
答题要点
不合并的两个后果：两段输出交错写进同一个会话，且第一次执行基于不完整信息注定被推翻
判据三条全中才合并：同一会话、上一次执行处于 running 或 streaming、距创建不到 30 秒；否则正常新建
命中就把新消息追加进同一次执行的输入并标记需要重跑，标记放带过期时间的键上而不是业务表
重跑时序号继续往上加、绝不重置，否则断线重连会续到作废的历史上
合并省的钱有限（约 17%），真正的理由是不让两个回答同时对着用户说话
After a streaming client reconnects, how do you deliver every missed chunk exactly once — no gaps, no duplicates?流式接口的客户端断线重连后，怎么做到既不丢片段也不重复？
Common in ChinaCommon overseasDeep dive#sse#idempotency#reconnect
How to reason about it · think before answering
Answer 'no gaps' and 'no duplicates' separately. Plenty of candidates cover only the first — they backfill from storage but never say how the overlap is deduplicated.
The chain is short: the client knows the last id it received, it sends that id back on reconnect, the server resumes from the next one — and all of that requires contiguous, monotonic numbering. Whether resumption is possible at all was decided when you chose the sequence scheme; timestamps or random ids break the chain at step one.
Then the three steps and their individual traps. Convert: the client reports the last id it *received*, so add one — off by minus one repeats a frame, off by plus one drops a character, and this is the only arithmetic in the whole flow and the most commonly wrong line. Replay: read the missing range from durable storage, which is always complete. Attach: resume the live stream, whose overlap with the replay is guaranteed, and drop anything below the cursor. That single comparison is all there is to idempotent replay.
Explain why the dual write is mandatory: chunks go both to the stream and to the table. Stream only, and the early chunks are gone by reconnect time; table only, and you are polling the database, pushing time-to-first-token from tens to hundreds of milliseconds. The cost is write amplification — hundreds of rows per answer — so production batches the writes, every few dozen chunks or every couple hundred milliseconds.
Get the protocol detail right: the browser's native event source replays the last id in a request header for you, but model endpoints generally need POST while that API only issues GET, so real frontends hand-roll the parser and must resend the id themselves. Mentioning this proves you have actually wired up the client side.
Expect: how long do you keep replayable data? Give two bounds — a retention window (per-chunk rows only for runs from the last few hours, then collapsed into one complete message) and a replay cap (beyond N chunks, send the full text once instead of re-enacting it character by character). Without both, that table becomes the largest in the database while 99% of its rows are never read again after ten seconds.
分析过程 · 先想清楚再作答
「不丢」和「不重复」要分开答。只答一半的人很多：说了从库里补发（不丢），却没说重叠部分怎么去重（不重复）。
推导链很短：客户端知道自己最后收到的编号 → 它重连时把这个编号带回来 → 服务端从下一号开始给 → 前提是编号连续不跳号。所以能不能重连，取决于当初有没有把序号设计成从 0 开始、连续、单调。序号一旦是时间戳或随机 id，这条链第一步就断了。
然后给三步实现和各自的坑：第一步换算，带回来的是「最后收到」的那一号，要加一，少加一重复一帧、多加一丢一个字，这是整段逻辑里唯一的算术也最常写错；第二步先从持久化里回放缺的部分，因为库里一定是全的；第三步再接上还在流动的那条流，两边必然重叠，靠「小于当前指针的一律丢弃」去重——幂等回放的全部秘密就是这一次比较。
这里要点出为什么必须双写：片段既进流也进库。只有流，重连时早期片段已被消费掉；只有库，就得轮询查库，首字延迟从几十毫秒涨到几百毫秒。代价是写放大，一次回答几百个片段就是几百行，生产里按批落库（每几十个片段或每两百毫秒一次）。
对齐一下协议细节：浏览器原生的事件源会自动把上次的编号放进重连请求头带回来；但大模型接口通常要用 POST，原生事件源只能发 GET，所以真实前端多是手写解析，重连时要自己把编号带上——这个细节能证明你真接过前端。
可以预期的追问：回放要保留多久？必须给两个边界——保留期（逐片段的行只对最近若干小时的执行保留，之后归档成一整条完整回复并删掉碎行）和回放上限（一次重连最多回放多少片段，超了就一次性发完整文本而不是逐字重演）。不定这两条，那张表会变成全库最大且 99% 的行写完十秒后再没人读。
Key points
Resumption requires a contiguous, monotonic sequence starting at 0; timestamps or random ids make it impossible
The client reports its last received id, so the server resumes from that id plus one — the single most error-prone line
Replay the gap from durable storage first, then attach the live stream, discarding anything below the cursor to dedupe the overlap
Dual-write every chunk: the stream serves currently attached connections, the table serves clients that come back later; batch the writes in production
Set a retention window and a replay cap — archive old runs into one complete message and send full text instead of re-enacting long replays
答题要点
重连的前提是序号从 0 开始、连续、单调；序号是时间戳或随机 id 就无法续传
客户端带回来的是「最后收到」的那一号，服务端要加一再开始，这是唯一的算术也最容易错
先从库里回放缺的片段（库一定是全的），再接上还在流动的流，重叠部分靠「小于当前指针一律丢弃」去重
片段必须双写：流服务当前挂着的连接，库服务等一下才回来的人；代价是写放大，生产里按批落库
必须定保留期与回放上限：过期的执行归档成一整条完整回复，超长回放直接一次性发完整文本

Comments

No comments yet — be the first.