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
- Draw a run's complete state machine from creation to completion (including failure and interruption states)
- Stream a worker's output back in order by runId to a waiting SSE client
- Merge a user interruption within 30 seconds into the same input, rather than opening a second run
D10 preserved order on the execution side, and from the user's side it is still "sent it and heard nothing." Today we connect the output back, and only then is this chain complete.
Plain-Language Walkthrough
Draw the line between today and D7 first, because the whole chapter stands on it. On D7 one process both generated and pushed: the model emitted a character and the same call stack immediately wrote a frame, with no gap between generating and pushing, so order was simply not a problem — it was correct by construction. Today generation is in the worker and pushing is in the gateway, with a message bus between them: characters are produced in one process while the connection hangs off another, and the bus promises nothing about when or in what order you receive them. D7 covered the server side of SSE thoroughly (four headers, the : ping heartbeat, disconnect handling) and none of it is repeated; all the effort goes into the four new problems that gap creates: state, order, multiple subscribers, and reconnection.
An execution needs an identity: the run state machine
A radio station makes an episode: commissioned, recorded in studio, broadcast, archived. Ask at any moment which step the episode is at and there is a definite answer, and the steps are one-way — broadcast cannot revert to being recorded.
In the single-process era you did not need this. Inside D7's POST /chat, "how far the execution has got" was the call stack itself: running meant running, returned meant finished, thrown meant failed. That state lived in process memory, unnamed, and needed no name.
After the split it must have a name, because at least three parties have to answer the same question at once: the gateway needs to know whether this execution is still alive before deciding whether to hold an SSE connection; the worker needs to know whether somebody is already doing this message before deciding to execute; and when the user reopens the page the frontend needs to know whether the previous question is still generating. Those three are not in one process, and a table is the only thing that can align them. That is D8's runs table and its six statuses:
+---------------------------- normal path -------------------------+
| |
+---------+ delivery taken +---------+ first fragment +-----------+ v
| pending | ---------------) | running | -------------) | streaming |-) +------+
+---------+ +---------+ +-----------+ | done |
| | | +------+
| v v
| +--------+ +--------+
+------------------) | failed | | failed | retries exhausted
+--------+ +--------+
| | |
+------------------) +-----------+ (--------------+
| cancelled | merged away or cancelled by the user
+-----------+Four normal statuses run forward and two abnormal exits are available at any time: failed is retries exhausted, cancelled is merged away or cancelled by the user. running and streaming are separate because they mean entirely different things to the user: running is "a worker has taken it, and not one character yet," and streaming is "the first character is out." That boundary is the observation point for time-to-first-token, and the basis on which the frontend decides between keeping the spinner and starting the typewriter.
A state machine's value is not listing those words, it is putting the illegal transitions into code where they are blocked:
// A transition table rather than a chain of ifs: it can be read, tested, and pasted
// straight into a design document
const TRANSITIONS = {
pending: ['running', 'cancelled', 'failed'],
running: ['streaming', 'done', 'failed', 'cancelled'],
streaming: ['done', 'failed', 'cancelled'],
done: [], // a terminal state has no outgoing edges, the most valuable line in the table
failed: [],
cancelled: [],
}
function transition(from, to) {
if (!TRANSITIONS[from].includes(to)) {
throw new Error(`illegal state transition: ${from} -> ${to}`)
}
return to
}from enum import StrEnum
class RunStatus(StrEnum):
PENDING = "pending"
RUNNING = "running"
STREAMING = "streaming"
DONE = "done"
FAILED = "failed"
CANCELLED = "cancelled"
# frozenset says "this table will not be modified", which fits the meaning better than a 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"illegal state transition: {current} -> {target}")
return target// The idiomatic Java state machine hangs the outgoing edges off the enum itself, stored in
// an EnumSet - a bitmap implementation, constant-time lookups, and forgetting to give a
// new status its edges is caught at compile time
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("illegal state transition: " + this + " -> " + target);
}
return target;
}
}// Swift uses an exhaustive switch: add a status to the enum later and this switch stops
// compiling, so the compiler forces you back to fill in its edges - the strongest
// protection of the four languages against forgetting a status.
// The edge table matches D8's exactly (including pending to failed). Only the method
// differs: D8 gave canTransition(to:) -> Bool, and today it becomes transition(to:) throws -
// a state machine that really blocks illegal transitions cannot return a Bool a caller
// can quietly ignore
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
}
}What that throw buys is a class of incident blocked at the moment of writing. In a distributed system "a run that is already done receives another fragment" is routine rather than exceptional — the bus delivers at least once (D9), one network wobble replays an old message, and a lagging worker may wake after wrapping up and write once more. Without the table that write lands quietly, the user sees half a stray sentence appended to their reply, and you search the logs for ages without finding who wrote it. With the table it throws on the spot and lands in the log.
A state machine's real product is not those words, it is the discipline that every status write must go through this function. Bypass it once with an UPDATE runs SET status = 'done' and the table degrades into a comment.
Which raises the second problem: the worker generates characters one at a time inside its own process, so how do those characters get back, in order, to the connection still hanging off the gateway?
Ordered delivery: seq starts at 0 and skips nothing
A radio episode is broadcast in time order, and minute 12 does not jump ahead of minute 3. But today's broadcast is not one person doing it end to end — recording is in the studio (the worker) and transmission is in the gallery (the gateway), with a line in between. The line does not promise that the order you receive is the order that was recorded.
The approach is fixed, in one sentence: each fragment the worker produces is written with runId + seq into the stream koda:out:{runId}, and the gateway's SSE handler pushes in seq order. Three details are non-negotiable.
One, one output stream per run rather than every run crowding one stream for the gateway to filter. With runId in the stream name, the gateway subscribes only to the one it cares about and everything it reads is its own; otherwise ten thousand concurrent conversations mean every connection scans ten thousand times the data.
Two, seq starts at 0, is contiguous, and skips nothing. That is not fussiness, it is the prerequisite for resumption: the client says it last received number 5 and the gateway must be able to conclude "then start from 6." If seq were a timestamp, a random id, or something that skipped, that inference collapses.
Three, SSE's id: field is the seq. Then the client keeps no separate ledger, and the browser's native EventSource even puts it into the Last-Event-ID header on reconnect. A frame looks like this:
id: 41
event: delta
data: {"text":"wa"}
: ping
id: 42
event: delta
data: {"text":"ybill"}The heartbeat is still D7's 15-second comment line, copied unchanged.
The core of the gateway side is a small orderer. It does one thing: deliver strictly by number, and hold when a number is missing.
class SeqOrderer {
constructor(from, emit) {
this.next = from // the next number due for delivery
this.buffer = new Map()
this.emit = emit
}
offer(seq, text) {
if (seq < this.next) return // already pushed: replay and live fragments necessarily overlap
this.buffer.set(seq, text)
// A missing number stops us here; once it arrives, deliver the run in one go
while (this.buffer.has(this.next)) {
this.emit(this.next, this.buffer.get(this.next))
this.buffer.delete(this.next)
this.next += 1
}
}
}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 # already pushed: replay and live fragments necessarily overlap
self._buffer[seq] = text
# A walrus does pop-and-test in one step, and the loop stops naturally on a gap
while (ready := self._buffer.pop(self._next, None)) is not None:
self._emit(self._next, ready)
self._next += 1// A TreeMap is key-ordered, so on a gap firstKey shows at a glance which number we are
// waiting for, and it also hands you "how many numbers are buffered undelivered" as a
// metric you can report directly
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; // already pushed
buffer.put(seq, text);
for (String ready = buffer.remove(next); ready != null; ready = buffer.remove(next)) {
emit.accept(next, ready);
next++;
}
}
}// A struct plus mutating: the caller holds the single copy of this state, so two places
// cannot corrupt it concurrently
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 } // already pushed
buffer[seq] = text
while let ready = buffer.removeValue(forKey: next) {
emit(next, ready)
next += 1
}
}
}That is a dozen lines, and the engineering cost sits elsewhere: should the buffer have a limit? If number 5 never arrives, 6 through 500 queue in memory, and ten thousand connections doing that at once is a memory incident. Production practice gives the buffer a cap (say 64 fragments) and a wait cap (say 2 seconds), after which you assume 5 is lost and re-read it from the database; failing that, emit an error event and let the client reconnect. You may wait, but not indefinitely — the universal discipline for any ordering buffer.
One more thing to state clearly: the worker dual-writes each fragment, into the output stream and into the messages table (where D8's unique(run_id, seq) earns its keep). Because they serve two audiences: the stream serves the connection hanging there now, and the database serves whoever comes back later. With only the stream, the fragments are long consumed by the time somebody reconnects; with only the database, you poll it and time-to-first-token grows from tens to hundreds of milliseconds. The cost is write amplification: a 500-fragment answer is 500 rows, so production batches the writes (every 20 to 40 fragments or every 200 milliseconds), and the lab writes one row per fragment only so ordering is visible.
One channel, many listeners: multiple SSE waiters
One program can have many simultaneous listeners, and the station does not re-record because a second person switched on a radio.
Mapped onto an agent: one run may well be subscribed from several places. The user asks on their phone and then opens the same conversation on a laptop; a web app has two tabs; and in the instant of a reconnect the old connection has not been noticed as gone while the new one is already up — for those few seconds two SSE connections really are attached to one run.
There is only one key judgment: a run is the unit of execution and an SSE connection is the unit of watching, and they are not one to one. Grasp that and a lot of apparently thorny problems disappear: you need no lock making the first connection exclusive, and no queue making a second connection wait for the first to drop. Each connection maintains its own read position and its own orderer, reading independently. It also explains why the output stream is read with XREAD, the bystander's read, rather than a consumer group — a consumer group shares, giving one message to one consumer, and what we want is broadcast, where every subscriber sees everything. Use a consumer group for fan-out and two connections each get half a sentence.
Fan-out has two implementations. One, each connection reads the stream itself: simplest, at the cost of the same data being read N times on the Redis side. Two, one gateway process reads once and broadcasts to local subscribers: no duplicate reads, but you maintain a subscriber table and handle "the last subscriber left, so stop," and across instances each still reads once. The test is the average subscriber count per run — in most products it is close to 1, so choose the first and do not write a subscriber-management layer for scale that does not exist.
The host has not finished, and a new request arrives: 30-second interrupt merging
Live on air, the host is reading a listener's letter when the same listener sends "wait, I got it wrong, it is the third song not the second." A sane host folds that into this episode rather than starting a second episode to broadcast at the same time — two episodes on one frequency is just noise.
Users in front of an agent behave identically: three seconds after sending, with two lines of reply on screen, they remember a missing condition and send again. Dutifully creating a second run for that message produces two consequences: two answers writing into the same conversation at once (the frontend shows two interleaved streams of text), and the first answer resting on incomplete information, so it is bound to be wrong.
Fix the criteria and require all three: the same sessionId, the previous run in running or streaming, and less than 30 seconds since it was created. A hit appends the new message to the same run's input and marks it for a rerun without creating a run; past 30 seconds, or with the previous run already done, create one normally.
const MERGE_WINDOW_MS = 30_000
function shouldMerge(prev, now) {
if (!prev) return false
// Merge only what is actually working: pending lasts milliseconds, and merging into
// something already done is meaningless
if (prev.status !== 'running' && prev.status !== 'streaming') return false
return now - prev.createdAt < MERGE_WINDOW_MS
}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
# A timedelta rather than a bare number of seconds: the unit lives in the type, so
# nobody passes milliseconds by mistake
return (now or datetime.now(UTC)) - prev.created_at < MERGE_WINDOW// Instant plus Duration is the JDK's time vocabulary; do not store milliseconds in a long,
// because then callers never know whether the argument is seconds, millis, or nanos
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;
}
}struct MergePolicy {
static let window: TimeInterval = 30
// guard let peels off the most common branch, no previous run, leaving one line of logic
static func shouldMerge(previous: Run?, now: Date = .now) throws -> Bool {
guard let previous, previous.status == .running || previous.status == .streaming
else { return false }
// createdAt is Fluent's @Timestamp, typed Date? - use the requireCreatedAt()
// funnel established on D8
return try now.timeIntervalSince(previous.requireCreatedAt()) < window
}
}Two implementation details deserve their own note. First, the rerun mark does not go into the runs table. It is meaningful only during this execution, so it is a control signal rather than persistent state; put it in a business table and a crash midway leaves a dirty mark that makes that run rerun forever after a restart. Keep it in a Redis key with an expiry and it disappears with the process. Second, the worker checks that mark in the gaps between fragments — not by polling every millisecond, but by glancing at it after each fragment. The worst-case response delay is then one fragment's generation time, tens of milliseconds, which the user cannot feel. On a rerun the seq must continue upwards and never reset, or somebody reconnecting by Last-Event-ID resumes into a history that has been voided.
Where did 30 seconds come from? It is not computed, it is an adjustable product judgment: too short and you miss the merge (people correct themselves between 5 and 15 seconds), too long and you fold a genuinely new question into the previous one as an addendum. The point is that this number must be defined in one place and referenced from three (the criteria, the mark's expiry, and the frontend's input hint), not scattered as three literals.
Run the arithmetic in passing, so you know interrupt merging is not a cost optimization. Price one answer at 2,000 input tokens and 500 output: 2,000 over a million times 0.15 dollars is about 0.0003 dollars, and 500 over a million times 0.60 dollars is also about 0.0003, so roughly 0.0006 in total. Without merging, two runs each complete for about 0.0012; with merging, the first is cut off a third of the way through (about 0.0004) plus a complete second run at 0.0006, about 0.0010. The 17% saved, spread across ten thousand corrections a day, is two dollars. So the reason to merge is never money, it is not letting two answers talk to the user at once. Volunteering that arithmetic in an interview is far stronger than saying "to save cost."
Tuning in late, listening from second N: idempotent replay
A listener can tune in late. With only a live broadcast they can only hear from now; with the episode also being recorded they can pick up from second N. Today's reconnection rests on that recording — every row written into the messages table.
The flow is three steps and each has one precise trap. Step one, work out which number to start from. The Last-Event-ID the client returns is the number it last received, not the one it wants next, so add one. Add nothing and the first frame after reconnecting is a duplicate; add two and the user loses a character. That line is the only arithmetic in the whole reconnection path and the most commonly botched. Step two, replay from the database first. The database is necessarily complete, so fill the gap first. Step three, then attach to the output stream. Fragments still flowing on the stream necessarily overlap what you just replayed, and the orderer's "discard anything below the pointer" deduplicates them — which is the entire secret of idempotent replay, one comparison.
function resumeFrom(lastEventId) {
if (!lastEventId) return 0 // first connection
const n = Number(lastEventId)
return Number.isInteger(n) && n >= 0 ? n + 1 : 0 // it returns "last received", so add one
}
// Replay first, then attach; the orderer discards the overlap itself
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))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 # it returns the number it last received
start = resume_from(request.headers.get("last-event-id"))
orderer = SeqOrderer(start, lambda seq, text: send_event(response, "delta", text, seq))
# Fill in what was missed from the database, then attach to the part still flowing
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"])// An Optional chain handles "the header may be absent and may not be a number" without
// nested null checks
static int resumeFrom(String lastEventId) {
return Optional.ofNullable(lastEventId)
// Bounded to 9 digits: Last-Event-ID is a client-controlled header, and "\\d+"
// lets a 30-digit string through, so the parseInt below throws
// NumberFormatException and one curl turns this endpoint into a 500
.filter(s -> s.matches("\\d{1,9}"))
.map(s -> Integer.parseInt(s) + 1) // the client gives the number it last received
.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()));// Three preconditions (present, numeric, non-negative) fall to one guard, leaving "add one"
// as the only logic. The Java version chains Optional; guard let is more idiomatic here -
// the same thing said two ways
func resumeFrom(_ lastEventID: String?) -> Int {
guard let raw = lastEventID, let n = Int(raw), n >= 0 else { return 0 }
return n + 1 // the client returns the number it last received
}
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)
}Finally, zoom out: D8 moved state into the database, D9 moved execution onto the bus, D10 preserved per-user processing order, and today reconnects the order and the experience of the output. Every step of distribution removes a guarantee you got for free, and then you have to rebuild it explicitly — order, state, identity, exactly-once are all free in one process and each costs code across processes. A production-grade IM Agent platform also had to make these layers solid before features had anywhere to grow.
Source Reading
Hands-On Lab
This lab closes the loop across all three parts for the first time, and src/ holds the gateway, the worker, and the infrastructure — but you only touch four functions, all in two files under src/shared/: the transition table, the orderer, the Last-Event-ID arithmetic, and the merge criteria. Infrastructure goes through ports and adapters: under MOCK=1 both Redis and Postgres are in-memory implementations (semantics written out, not stubbed), and setting REDIS_URL and DATABASE_URL swaps in ioredis and pg with no change to business code. Run the self-check as-is first; that scrambled mock reply makes "unordered" instantly legible.
- Run
MOCK=1 SELFTEST=1 pnpm startas-is and remember what the failures on checks 1, 2, 4 and 5 look like, especially check 2's scrambled reply. - Complete the transition table so check 1 passes: illegal transitions (
donetostreaming,pendingtostreaming) must be refused. - Complete the orderer's three jobs — discard old numbers, buffer on a gap, deliver contiguous runs in a batch — and check 2's seq should become contiguous from 0.
- Complete the
Last-Event-IDarithmetic; check 4 must satisfy both "the first number after reconnecting equals the last received plus one" and "no duplicates and no gaps overall." - Complete the merge criteria, and once check 5 passes confirm check 6 still does — past 30 seconds a new run is mandatory, and the two are opposite sides of one criterion.
Interview Questions
Today's four questions are in the bank below, covering the design of the run state machine, ordering with multiple subscribers, the product and engineering trade-offs of interrupt merging, and losing nothing and duplicating nothing on reconnect. Expand a question and read the analysis before the key points — the follow-up on question 4, about the retention period and the replay cap, is where this chapter gets pressed hardest, so do not skip it.
Checklist and Tomorrow
- Draw a run's complete state machine from creation to completion (including failure and interruption states)
- Stream a worker's output back in order by runId to a waiting SSE client
- Merge a user interruption within 30 seconds into the same input, rather than opening a second run
- Say why the output stream uses a broadcast read while the input stream uses a consumer group, and what the symptom of getting it backwards is
- Explain why "seq starts at 0, contiguous, no gaps" is the prerequisite for resumption
- All 5 acceptance criteria of the lab pass (all six self-checks green)
- Answer at least 3 of the 4 interview questions without looking at the key points
Tomorrow (D12) we fit this agent with long-term memory. Why now? Because the chain only became complete today — a user sends a sentence and gets an ordered, resumable, mergeable answer. But this agent remembers only the current session: last week they said their building has no lift and large parcels should go to the management office, and asking again today it knows nothing. D6 solved "this round will not fit"; tomorrow solves "cannot recall what was said last month," which is neither the same problem nor the same mechanism.
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-systemsHow 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-outHow 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#costHow 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#reconnectHow 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 就无法续传
- 客户端带回来的是「最后收到」的那一号，服务端要加一再开始，这是唯一的算术也最容易错
- 先从库里回放缺的片段（库一定是全的），再接上还在流动的流，重叠部分靠「小于当前指针一律丢弃」去重
- 片段必须双写：流服务当前挂着的连接，库服务等一下才回来的人；代价是写放大，生产里按批落库
- 必须定保留期与回放上限：过期的执行归档成一整条完整回复，超长回放直接一次性发完整文本