SSE

# chat_backend.py:681
def stream_generator() -> Generator[str, None, None]:
    state_container = ChatStateContainer()
    try:
        for obj in handle_stream_message_objects(...):
            yield get_json_line(obj.model_dump())   # 每个对象 → 一行 JSON
    except Exception as e:
        yield json.dumps({"error": str(e)})
    finally:
        logger.debug("Stream generator finished")

return StreamingResponse(stream_generator(), media_type="text/event-stream")

def get_json_line(
    json_dict: dict[str, Any], encoder: type[json.JSONEncoder] = OnyxJSONEncoder
) -> str:
    return json.dumps(json_dict, cls=encoder) + "\n"

class Packet(BaseModel):
    placement: Placement                      # 定位信息（哪个 turn / 哪个模型）
    obj: Annotated[PacketObj, Field(discriminator="type")]   # 具体内容，按 type 区分

stream_generator (chat_backend.py:682)
   └─ handle_stream_message_objects (process_message.py:1752)
        └─ _stream_chat_turn (process_message.py:1528)
             └─ _run_models (process_message.py:1062)
                  └─ _read_stream (process_message.py:1496)  ← 实际对外 yield

for packet in llm.stream(
    prompt=llm_msg_history,
    tools=tool_definitions,
    tool_choice=tool_choice,
    max_tokens=max_tokens,
    reasoning_effort=reasoning_effort,
    ...
):
    delta = packet.choice.delta

    if not answer_start:
        yield Packet(obj=AgentResponseStart(final_documents=..., ...))
        answer_start = True

    accumulated_answer += content_chunk                    # 后端也累加一份
    state_container.set_answer_tokens(accumulated_answer)  # 用于最终持久化
    yield Packet(
        placement=_current_placement(),
        obj=AgentResponseDelta(content=content_chunk),     # 每块一个增量包
    )

worker0 ──Emitter──▶ merged_queue ──┐
worker1 ──Emitter──▶ merged_queue ──┼─▶ writer线程 ──▶ tee队列 ──▶ reader → yield → HTTP
worker2 ──Emitter──▶ merged_queue ──┘      │
                                          └─▶ StreamBufferWriter（持久化缓冲）

const response = await fetch(`/api/chat/send-chat-message`, {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body,
  signal,                       // ← 取消信号
});

if (!response.ok) { /* 抛错 */ }

yield* withoutHeartbeats(handleSSEStream<PacketType>(response, signal));

async function* withoutHeartbeats(stream) {
  for await (const packet of stream) {
    if ("obj" in packet && packet.obj.type === "chat_heartbeat") continue;
    yield packet;
  }
}

export class CurrentMessageFIFO {
  private stack: PacketType[] = [];
  isComplete: boolean = false;
  error: string | null = null;

  push(packetBunch) { this.stack.push(packetBunch); }
  nextPacket() { return this.stack.shift(); }   // FIFO：先入先出
  isEmpty() { return this.stack.length === 0; }
}

export async function updateCurrentMessageFIFO(stack, params) {
  try {
    for await (const packet of sendMessage(params)) {
      if (params.signal?.aborted) throw new Error("AbortError");
      stack.push(packet);
    }
  } catch (error) { /* 记入 stack.error */ }
  finally { stack.isComplete = true; }
}

await delay(50);
while (...) {
  if (stack.isEmpty()) {
    if (pendingFlush) await flushViaRAF();  // 攒到下一帧再刷
  }
  ...
}

prev.packetCount === next.packetCount &&
prev.chatState.agent === next.chatState.agent &&
... // 其余字段

// useChatSessionController.ts:319 的注释点明了设计意图：
// "AgentMessage's memo compares packetCount, not the packets array."
node.packetCount = accumulated.length;

// 处理状态放在 ref 里：增量、同步、不触发双渲染
const stateRef = useRef<ProcessorState>(createInitialState(nodeId));
// 只有真正的 UI 状态才用 useState
const [renderComplete, setRenderComplete] = useState(false);

// 增量处理：只处理新到的包（避免重复处理已消费的）
if (rawPackets.length > stateRef.current.nextPacketIndex) {
  stateRef.current = processPackets(stateRef.current, rawPackets);
}

while True:
    try:
        item = tee.get(timeout=_CANCEL_POLL_INTERVAL_S)
    except queue.Empty:   # 队列空 = 没有真实包到来
        now = time.monotonic()
        if now - last_packet_yield >= CHAT_HEARTBEAT_INTERVAL_S:
            yield heartbeat_packet()          # 发个心跳保活
            last_packet_yield = now
        continue
    ...

def heartbeat_packet() -> Packet:
    """Keepalive for silent stretches; carries no run state."""
    return Packet(placement=Placement(turn_index=0), obj=ChatHeartbeat())

# process_message.py:1650
for pre_run_packet in pre_run_packets:
    stream_buffer.append_line(get_json_line(pre_run_packet.model_dump()))

def flush(self) -> None:
    payload = zlib.compress("".join(self._pending).encode("utf-8"))
    self._cache.set(
        _chunk_key(self._chat_session_id, self._run_id, self._meta.chunk_count),
        payload,
        ex=CHAT_STREAM_BUFFER_TTL_S,
    )

def resume_chat_stream(
    session_id: UUID,
    cursor: int = Query(0, ge=0),     # 客户端上次读到哪
    user: User = Depends(...),
) -> StreamingResponse:
    ...
    run_id = get_processing_run_id(session_id, cache)
    if run_id is None or not has_stream_buffer(cache, session_id, run_id):
        raise OnyxError(NOT_FOUND, "No resumable run for this chat session")

    def stream_buffered_run() -> Generator[str, None, None]:
        chunk_cursor = cursor
        while True:
            read = read_stream_chunks(cache, session_id, run_id, chunk_cursor, ...)
            if read is None or read.gap: return      # 缓冲丢了 → 终止，客户端重新拉全量
            if read.blocks:
                yield "".join(read.blocks)           # 回放缓冲内容
                chunk_cursor = read.next_cursor
                continue
            if read.done: return                     # 正常结束
            if not is_chat_session_processing(session_id, cache):
                # writer 死了但没标 done → 再补一次 drain
                ...
            # 缓冲读完了 → 实时 tail 新的包
            time.sleep(CHAT_RESUME_POLL_INTERVAL_S)
            if now - last_emit >= CHAT_HEARTBEAT_INTERVAL_S:
                yield get_json_line(heartbeat_packet().model_dump())  # 轮询中也发心跳
    ...

export async function* resumeStream(chatSessionId, cursor, signal?) {
  const response = await fetch(
    `/api/chat/chat-session/${chatSessionId}/resume-stream?cursor=${cursor}`,
    { signal }
  );
  ...
  yield* handleSSEStream<PacketType>(response, signal);
}

const abortController = new AbortController();
try {
  for await (const rawPacket of resumeStream(sessionId, 0, abortController.signal)) {
    if (!stillCurrent()) return;              // 用户已切走会话 → 停
    if (packet.obj.type === "chat_heartbeat") continue;
    accumulated.push(packet);
    // 每 100ms 或 trailing 120ms 刷一次到消息树
  }
} finally {
  abortController.abort();
  // 流结束后拉一次落库的完整 session 做"结算"
  const settled = await (await fetch(`/api/chat/get-chat-session/${sessionId}`)).json();
  updateSessionAndMessageTree(sessionId, processRawChatHistory(settled.messages, ...));
}