一、前言：为什么"逐字输出"在鸿蒙上这么难

假设你要在鸿蒙应用里做流式输出。你查了 HarmonyOS 文档，发现 @ohos.net.http 有个 dataReceive 回调——好，那就监听它：

client.on('dataReceive', (data: ArrayBuffer) => {
  // data 是一块字节，然后呢？
})

然后你拿到智谱或 DeepSeek 的接口文档，发现流式返回长这样：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

data: [DONE]

看起来不难？但你真正上真机一跑，至少会撞上这几个坑：

1. 中文乱码：dataReceive 给你的 ArrayBuffer 边界是任意的，一个"你好"的 UTF-8 编码（6 字节）可能被切成 E4 BD A0 和 E5 A5 BD 两次回调。直接 TextDecoder 或手动按字节转字符串，第二个 chunk 开头就乱码了。

2. 帧粘在一起：一次 dataReceive 可能包含半条 data:、一整条、或者两条半粘在一起。你不能假设"一次回调 = 一条 SSE 事件"。

3. [DONE] 之外还有终止信号：有些 Provider 最后一条带 finish_reason，有些用 data: [DONE]，有些两者都有，有些只有 usage。你靠什么判断"流真的结束了"？

4. reasoning 和 text 混着来：智谱 glm-5.2 真机流式会先吐 94 个 reasoning_delta，再吐 2 个 text_delta。推理过程和正文是分开的，你要不要都显示？

5. 工具参数被切成碎片：流式 tool call 的 arguments 是按 delta 分片到达的（{"ci → ty":"北京 → "}），多个工具还会按 index 交错。你没等拼完就执行，一定解析失败。

6. 取消后还在蹦事件：用户点了"停止"，但 dataReceive 回调队列里还积压着几条，它们还会触发你的 onNext，UI 上又蹦出几个字。

这些问题互相耦合——UTF-8 解码依赖字节缓冲、SSE 解帧依赖 UTF-8 解码、事件组装依赖解帧、终止判断依赖事件流、取消依赖终止状态。你不能把它们拆成五个独立的小函数糊在一起，你需要一条有状态的管道。

ArkAgent 的解法是：在鸿蒙 HTTP 回调和业务观察者之间，插入四个专职组件——IncrementalUtf8Decoder（管残字节）、SseParser（管帧边界）、ToolCallAccumulator（管分片拼装）、StreamAssembler（管事件分发和终止）。本文就拆解这条管道。

---

二、流式输出的三层挑战

先建立全局认知。流式输出从最底层的字节到最上层的事件，要穿过三层：

┌─────────────────────────────────────────────────────────────┐
│  语义层（StreamAssembler）                                    │
│  把 JSON chunk 翻译成 8 种 StreamingEvent                     │
│  累积 text / reasoning / toolCalls，判断终止                   │
├─────────────────────────────────────────────────────────────┤
│  协议层（SseParser）                                          │
│  把字节流按 SSE 协议拆成 data: 帧                              │
│  处理 LF/CRLF、多行 data、注释、空行分帧、[DONE]                │
├─────────────────────────────────────────────────────────────┤
│  编码层（IncrementalUtf8Decoder）                             │
│  把任意边界的字节 chunk 解码成字符串                            │
│  跨 chunk 保留不完整的多字节残字节                              │
├─────────────────────────────────────────────────────────────┤
│  传输层（HttpTransport SPI → HarmonyHttpTransport）           │
│  鸿蒙 @ohos.net.http 的 dataReceive / dataEnd / 取消           │
└─────────────────────────────────────────────────────────────┘

每一层的职责、输入、输出、状态：

层	组件	输入	输出	持有状态
传输	HarmonyHttpTransport	HttpRequest	status / headers / byte chunk / complete	鸿蒙 HTTP client、terminated 标志
编码	IncrementalUtf8Decoder	Uint8Array	string	pending: number[]（残字节）
协议	SseParser	string	SseEvent[]	buffer、dataLines
语义	StreamAssembler	SseEvent	StreamingEvent（8 种）	text/reasoning/usage/tools/terminated

关键设计判断：这四层职责正交，状态隔离。编码层只管字节→字符，不关心 SSE；协议层只管帧边界，不关心 JSON 内容；语义层只管事件语义，不关心字节从哪来。这样每一层都可以用固定的离线 Fixture 独立测试，不依赖真机网络。

为什么强调"不依赖真机"？因为真机网络行为是不可控、不可复现的——chunk 大小、回调时序、缓冲行为都受设备和网络环境影响。如果流式逻辑只在真机上能测，那它的正确性就成了玄学。ArkAgent 的策略是：协议层和编码层用固定字节 Fixture 做单元测试，真机只验证"鸿蒙 HTTP 能不能把字节传上来"这一件事。

---

三、八种流式事件：统一的事件契约

在深入管道细节之前，先看管道的"出口"——8 种 StreamingEvent。这是 ArkAgent 流式能力的核心契约，定义在 domain/Streaming.ets：

export enum StreamingEventType {
  modelStart = 'model_start',
  reasoningDelta = 'reasoning_delta',
  textDelta = 'text_delta',
  toolCallDelta = 'tool_call_delta',
  usage = 'usage',
  retry = 'retry',
  modelComplete = 'model_complete',
  error = 'error'
}

每种事件的含义和触发时机：

事件	何时触发	携带数据	UI 典型处理
modelStart	流开始（拿到首个 chunk 前）	无	显示 loading
reasoningDelta	收到 reasoning_content 增量	text	可选：显示 thinking
textDelta	收到 content 增量	text	追加到回答区
toolCallDelta	收到 tool_calls 增量	toolCall（快照）	可选：显示参数拼装
usage	收到 token 用量	usage	可选：显示消耗
retry	触发了一次重试	metadata（attempt/错误/延迟）	可选：提示重试
modelComplete	流正常结束	response（完整 ModelResponse）	隐藏 loading
error	流异常结束	error	显示错误

3.1 为什么是 8 种，不是 3 种或 20 种

最朴素的设计只有三种：delta（增量）、complete（完成）、error（错误）。但这不够：

• reasoning 和 text 必须分开。智谱和 DeepSeek 都把推理过程（reasoning_content）和正文（content）分开返回。如果合成一种 delta，UI 没法区分"这是模型在想"还是"这是模型在说"。真机数据很能说明问题——一次 glm-5.2 调用产生了 94 个 reasoning_delta + 2 个 text_delta，如果混在一起，用户会看到一大段思考过程刷屏，最后才蹦出正文。

• tool_call 必须独立。工具调用的增量不是文本，而是结构化的 ToolCall 快照，UI 可能想显示"正在拼装 get_weather 的参数"。

• usage 必须独立。token 用量通常在最后一个 chunk，但它和文本增量是两回事。

• retry 必须独立。流式重试时，UI 需要告诉用户"第一次失败了，正在重试"，而不是默默吞掉错误。

反过来，为什么不做 20 种？因为再多就泄漏 Provider 细节了。比如智谱有 sensitive finish reason、DeepSeek 有 insufficient_system_resource——这些不应该成为独立事件类型，它们在语义层被归一化成 error 或 modelComplete，Provider 特有的东西留在 ArkAgentError.providerCode 里。8 种是"恰好覆盖所有业务语义"的最小集合。

3.2 modelComplete 必须携带完整响应

这条规则容易被忽视，但极其关键：

// ❌ 错误理解：onComplete 就代表完成了，不用带数据
observer.onComplete()  // 那数据呢？

// ✅ 正确：modelComplete 事件本身就携带完整 ModelResponse
observer.onNext(new StreamingEvent(
  StreamingEventType.modelComplete, undefined, undefined, undefined, response))

为什么不让消费者自己累积 delta？ 因为"累积"这件事本身就是 bug 温床——reasoning 要不要进正文？tool_call 的 arguments 拼到一半怎么办？finish_reason 是 tool_calls 还是 stop？这些判断逻辑如果散落在每个消费者里，每个消费者都会重复实现且各错各的。ArkAgent 把累积逻辑收敛在 StreamAssembler 内部，modelComplete 事件直接给出一个已经拼好的、可信的 ModelResponse——消费者可以完全忽略 delta 事件，只等 modelComplete，也能拿到正确结果。

delta 事件的价值是体验（逐字显示），不是数据（完整数据）。modelComplete 才是数据的唯一权威来源。这条契约记录在 ADR-0004，是整个流式设计的地基。

3.3 StreamObserver：观察者契约

export interface StreamObserver<T> {
  onNext(value: T): void
  onError(error: ArkAgentError): void
  onComplete(): void
}

export interface StreamSubscription {
  cancel(reason?: string): Promise<void>
}

export interface LLMClient {
  generate(request: ModelRequest, signal?: CancellationSignal): Promise<ModelResponse>
  stream(request: ModelRequest, observer: StreamObserver<StreamingEvent>,
    signal?: CancellationSignal): Promise<StreamSubscription>
}

三个回调，含义明确：

• onNext：收到一个 StreamingEvent（8 种之一），可能被调用很多次。

• onError：流异常终止，最多调用一次，调用后不再有 onNext/onComplete。

• onComplete：流正常结束的信号（传输层完成），最多调用一次。

stream 方法返回 StreamSubscription，消费者通过 subscription.cancel(reason) 取消。取消是异步的，因为底层要销毁 HTTP 连接。

⚠️ ArkTS 的约束：StreamObserver 必须用显式 class 实现，不能用对象字面量。ArkTS 严格模式禁止未类型化的对象字面量（arkts-no-untyped-obj-literals），写 { onNext: (e) => {...} } 会直接编译报错。第二篇博客里 SessionCoordinator 就是这么做的。

---

四、Transport SPI：把鸿蒙 HTTP 和协议解析解耦

现在进入管道最底层。先看传输层。

4.1 为什么不直接用 @ohos.net.http？

如果让 StreamAssembler 直接依赖 @ohos.net.http 的 dataReceive，会有两个问题：

1. 无法离线测试。StreamAssembler 的单元测试就得连真机或 mock 整个鸿蒙 HTTP 栈。

2. 被鸿蒙 HTTP 行为绑死。鸿蒙 HTTP 在某些场景下可能缓冲整个响应（详见踩坑二），如果协议解析和传输绑死，缓冲问题就无处可改。

ArkAgent 的解法是定义一个传输层 SPI（Service Provider Interface），把"字节怎么来"和"字节怎么解析"彻底分开：

export interface HttpStreamObserver {
  onHeaders(statusCode: number, headers: HttpHeaders): void
  onBytes(chunk: Uint8Array): void
  onError(error: ArkAgentError): void
  onComplete(): void
}

export interface HttpTransport {
  request(request: HttpRequest, signal?: CancellationSignal): Promise<HttpResponse>
  stream(request: HttpRequest, observer: HttpStreamObserver,
    signal?: CancellationSignal): Promise<HttpStreamSubscription>
}

注意 SPI 的设计哲学：只暴露 status、headers、byte chunk、complete/error/cancel，绝不暴露 SSE 语义。传输层只负责"把字节搬上来"，不知道也不需要知道这些字节是 SSE 还是别的什么。SSE 的解帧完全在上层的 StreamAssembler 里做。

4.2 HarmonyHttpTransport：鸿蒙 HTTP 的实现

HarmonyHttpTransport 是 HttpTransport SPI 的默认实现，把鸿蒙 @ohos.net.http 的回调映射成 HttpStreamObserver 的方法：

export class HarmonyHttpTransport implements HttpTransport {
  async stream(request: HttpRequest, observer: HttpStreamObserver,
    signal?: CancellationSignal): Promise<HttpStreamSubscription> {
    const client = http.createHttp()
    let terminated = false

    // 取消信号 → 立即终止 + 销毁连接
    const cancelSub = signal?.onCancel((reason?: string) => {
      terminateError(ArkAgentError.cancelled('HTTP stream cancelled', reason))
    })

    // 鸿蒙 HTTP 三个核心回调
    client.on('headersReceive', (header: Object) => {
      if (terminated) return               // ← 终止后不再派发
      observer.onHeaders(statusCode, this.toHeaders(header))
    })

    client.on('dataReceive', (data: ArrayBuffer) => {
      if (terminated) return               // ← 终止后不再派发
      observer.onBytes(new Uint8Array(data))
    })

    client.on('dataEnd', () => {
      cancelSub?.unsubscribe()
      terminateComplete()
    })

    // ...省略 requestInStream 调用和错误处理
  }
}

三个要点：

第一，terminated 标志是终止的唯一闸门。 每个回调进来先检查 terminated，是 true 就直接 return。这保证了"取消后不再派发 onNext"这条不变量——即使用户在取消时，鸿蒙 HTTP 回调队列里还积压着几条 dataReceive，它们都会被 terminated 挡掉。

第二，取消会销毁底层连接。 signal.onCancel 触发时，不只是设标志，还会 client.destroy()，真正释放网络资源。光设标志不销毁连接，取消后连接还在那儿烧 token。

第三，dataEnd 才是传输完成的标志。 不是 [DONE]，不是 finish_reason——那些是协议层的事。传输层只认 dataEnd 回调，它代表"鸿蒙 HTTP 认为这个响应的字节流结束了"。

4.3 流式请求的关键配置

流式请求和非流式请求在 HTTP 层有个重要差异——expectDataType：

private toOptions(request: HttpRequest, streaming: boolean): http.HttpRequestOptions {
  const options: http.HttpRequestOptions = {
    method: this.toMethod(request.method),
    header: headerObj,
    connectTimeout: request.connectTimeoutMs,
    readTimeout: request.readTimeoutMs,
    expectDataType: http.HttpDataType.STRING     // ← 非流式默认
  }
  if (streaming) {
    options.expectDataType = http.HttpDataType.ARRAY_BUFFER  // ← 流式必须用这个
  }
  return options
}

为什么流式要用 ARRAY_BUFFER？ 因为用 STRING 模式时，鸿蒙 HTTP 会尝试自己把整个响应体解码成字符串再返回——这会破坏流式语义（要么缓冲整体、要么在多字节字符边界出错）。ARRAY_BUFFER 模式下，dataReceive 给的是原始字节，UTF-8 解码完全由我们的 IncrementalUtf8Decoder 接管。这是"把字节解码权从平台手里拿回来"的关键一步。

---

五、SseParser：逐行解帧，不依赖平台类型

字节通过 IncrementalUtf8Decoder 变成字符串后，进入 SseParser。它的职责是：把一段可能粘在一起的 SSE 文本，按空行拆成一条条 SseEvent。

5.1 SSE 协议速览

Server-Sent Events 是一个极简的文本协议。服务端发的每一条事件由若干行组成，事件之间用空行分隔：

data: {"choices":[{"delta":{"content":"你"}}]}

data: {"choices":[{"delta":{"content":"好"}}]}

: 这是一个注释，可以忽略

event: ping
data: keepalive

data: [DONE]

规则：

• 每行格式是 字段: 值，冒号后可选一个空格。

• data: 是数据行，一个事件可以有多行 data:，最终用 \n 拼接。

• event:、id: 是可选的元数据行。

• 以 : 开头的行是注释，忽略。

• 空行是事件的分隔符——遇到空行，就把累积的 data: 行组装成一条事件发出。

• [DONE] 是 OpenAI 兼容协议的约定终止标记。

5.2 SseParser 的实现

SseParser（openai/SseParser.ets）是一个有状态的逐行解析器。核心方法 push(text) 追加文本，返回这次解析出的 0~N 条事件：

export class SseParser {
  private buffer: string = ''           // 尚未形成完整行的尾巴
  private dataLines: string[] = []      // 当前事件累积的 data 行
  private eventName?: string
  private eventId?: string
  private comments: string[] = []

  push(text: string): Result<SseEvent[], ArkAgentError> {
    this.buffer += text
    const events: SseEvent[] = []
    while (true) {
      // 找下一个换行（同时处理 LF 和 CRLF）
      const idxLf = this.buffer.indexOf('\n')
      const idxCr = this.buffer.indexOf('\r')
      if (idxLf < 0 && idxCr < 0) break   // 没有完整行了，留着下次
      // ...切出一行，交给 handleLine
      const line = this.buffer.substring(0, idx)
      this.buffer = this.buffer.substring(idx + sepLen)
      const maybe = this.handleLine(line)
      if (maybe !== undefined) events.push(maybe)
    }
    return Result.success<SseEvent[], ArkAgentError>(events)
  }
}

逐行处理的逻辑：

private handleLine(line: string): SseEvent | undefined {
  let value = line
  if (value.endsWith('\r')) value = value.substring(0, value.length - 1)

  // 空行 → 当前事件完成，发出
  if (value.length === 0) {
    if (this.dataLines.length === 0 && this.eventName === undefined &&
        this.eventId === undefined && this.comments.length === 0) {
      return undefined    // 连续空行，忽略
    }
    return this.emitEvent()
  }

  // 注释行（以 : 开头）
  if (value.startsWith(':')) {
    this.comments.push(value.substring(1).trimStart())
    return undefined
  }

  // 字段: 值
  const colon = value.indexOf(':')
  let field = value
  let fieldValue = ''
  if (colon >= 0) {
    field = value.substring(0, colon)
    fieldValue = value.substring(colon + 1)
    if (fieldValue.startsWith(' ')) fieldValue = fieldValue.substring(1)  // 去掉可选空格
  }

  if (field === 'data') {
    this.dataLines.push(fieldValue)
  } else if (field === 'event') {
    this.eventName = fieldValue
  } else if (field === 'id') {
    this.eventId = fieldValue
  }
  // retry 字段故意忽略（重试逻辑由 RetryPolicy 管）
  return undefined
}

private emitEvent(): SseEvent {
  const data = this.dataLines.join('\n')   // 多行 data 用 \n 拼接
  const event = new SseEvent(data, this.eventName, this.eventId, this.comments.slice())
  this.dataLines = []
  this.eventName = undefined
  this.eventId = undefined
  this.comments = []
  return event
}

几个关键设计点：

第一，buffer 跨 push 调用保留。 一次 dataReceive 可能给出半行（data: {"choices"），SseParser 把它留在 buffer 里，等下次 push 补全。这就是"帧粘包"的解法。

第二，LF 和 CRLF 同时兼容。 智谱和 DeepSeek 的 SSE 分隔符可能不同（有些服务用 \n，有些用 \r\n）。SseParser 同时检查 \n 和 \r，谁先出现按谁切，还专门处理了 \r\n 的情况。这避免了"在某个 Provider 上好好的，换一家就丢帧"。

第三，[DONE] 只是数据内容，不是协议事件。 SseEvent 有个 isDone getter 检查 data === '[DONE]'，但解析器本身不特殊处理它——它仍然是一条普通的 data: 事件，由上层 StreamAssembler 判断。这保持了协议层的纯粹性。

第四，finish() 处理流尾。 如果流结束时 buffer 还有内容，或者有未发出的 dataLines，finish() 把它们强制成一条事件发出。这应对"最后一条事件后面没有空行"的边缘情况。

5.3 SseEvent 的结构

export class SseEvent {
  readonly event?: string
  readonly data: string
  readonly id?: string
  readonly comments: string[]

  get isDone(): boolean {
    return this.data.trim() === '[DONE]'
  }
}

SseEvent 是一个纯值对象——只有数据，没有行为（除了 isDone）。它不依赖任何鸿蒙类型，不依赖 JSON 解析。这让 SseParser 可以用纯字符串 Fixture 完整测试，不需要 mock 任何平台能力。

---

六、⚠️ 踩坑一：UTF-8 跨 chunk 切断，中文变乱码

这是流式输出里最高频、最隐蔽的坑，也是 IncrementalUtf8Decoder 存在的唯一理由。

6.1 症状

中文场景下，流式输出偶尔出现乱码字符（\uFFFD 替换符，或者干脆丢字）。更诡异的是：不是每次都乱码，同样的代码，有时正常有时乱，且乱码位置随机。

6.2 根因

UTF-8 是变长编码：ASCII 字符 1 字节，中文常用字 3 字节，emoji 4 字节。而鸿蒙 dataReceive 给的 ArrayBuffer 边界是任意的——它按网络包到达的节奏切分，完全不管字符边界。一个中文字"你"的 UTF-8 编码是 E4 BD A0（3 字节），它极有可能被切成两次回调：

dataReceive #1: [..., E4 BD]      ← "你"的前两字节
dataReceive #2: [A0, ...]         ← "你"的最后一字节 + 后续内容

如果每个 chunk 独立解码，第一次回调的 E4 BD 是不完整序列——直接 TextDecoder 会输出一个替换符或丢掉，第二次回调的 A0 单独看是非法起始字节，又是一个替换符。结果：一个"你"字变成两个乱码。

6.3 修复：IncrementalUtf8Decoder 的残字节保留

IncrementalUtf8Decoder（openai/IncrementalUtf8Decoder.ets）的核心思想：遇到不完整的多字节序列，把残字节存起来，等下一个 chunk 到了再拼。

export class IncrementalUtf8Decoder {
  private pending: number[] = []    // ← 跨 chunk 保留的残字节

  push(chunk: Uint8Array): Result<string, ArkAgentError> {
    const bytes: number[] = this.pending.concat(Array.from(chunk))  // 拼上次的残字节
    this.pending = []
    let out = ''
    let i = 0
    while (i < bytes.length) {
      const b0 = bytes[i]
      if (b0 < 0x80) {                  // ASCII，1 字节
        out += String.fromCharCode(b0)
        i++
        continue
      }
      // 根据首字节判断这个字符需要几个字节
      let needed = 0
      if ((b0 & 0xE0) === 0xC0) needed = 2
      else if ((b0 & 0xF0) === 0xE0) needed = 3
      else if ((b0 & 0xF8) === 0xF0) needed = 4
      else {
        return Result.failure<string, ArkAgentError>(
          ArkAgentError.decode('utf8_invalid_byte', `Invalid UTF-8 lead byte at ${i}`))
      }
      // 关键：剩余字节不够凑成一个完整字符 → 存起来，等下次
      if (i + needed > bytes.length) {
        this.pending = bytes.slice(i)   // ← 残字节进 pending
        break
      }
      // 校验后续字节的高两位是 10，并解码
      if (needed === 2) {
        const b1 = bytes[i + 1]
        if ((b1 & 0xC0) !== 0x80) {
          return Result.failure(...'utf8_invalid_sequence'...)
        }
        out += String.fromCharCode(((b0 & 0x1F) << 6) | (b1 & 0x3F))
      } else if (needed === 3) {
        // 3 字节（中文主力）...
      } else {
        // 4 字节（emoji），需要代理对
      }
      i += needed
    }
    return Result.success<string, ArkAgentError>(out)
  }

  finish(): Result<string, ArkAgentError> {
    // 流结束时还有残字节 → 非法截断
    if (this.pending.length > 0) {
      return Result.failure<string, ArkAgentError>(
        ArkAgentError.decode('utf8_truncated', 'Incomplete UTF-8 sequence at end of stream'))
    }
    return Result.success<string, ArkAgentError>('')
  }
}

走一遍前面那个"你"字被切断的例子：

push([..., E4, BD])    ← pending=[]，bytes=[...,E4,BD]
                         扫到 E4：(E4 & 0xF0)==0xE0 → needed=3
                         i+3 > bytes.length → pending=[E4,BD]，break
                         返回 out="..."（不含"你"，但也不乱码）
                         pending=[E4,BD]

push([A0, ...])        ← pending=[E4,BD]，bytes=[E4,BD,A0,...]
                         扫到 E4：needed=3，i+3 <= length
                         b1=BD(b0&0x1F=04,b1&0x3F=3D) b2=A0(A0&0x3F=20)
                         码点 = (04<<12)|(3D<<6)|20 = 0x4F60 = "你"  ✅
                         返回 out="你..."

残字节在两次 push 之间被完美衔接，"你"字正确还原。

关键设计：pending 是 number[] 而不是 Uint8Array，因为 Uint8Array 在 ArkTS 里不方便动态拼接。用 number[] 配合 concat，简单直接。reset() 清空 pending，用于重试时复用解码器。

6.4 为什么不直接用鸿蒙的 TextDecoder

鸿蒙确实提供 util.TextDecoder，但它的问题是：默认行为是"尽力解码"——遇到不完整序列，它要么输出替换符、要么忽略，而不是把它留到下次。也就是说，它没有"跨 chunk 残字节保留"的语义。你可以在每个 chunk 上调 TextDecoder，但你挡不住它在 chunk 边界乱码。

自己实现一个增量解码器，代价是 77 行代码，换来的是完全可控的字节边界行为。这个代价非常划算。而且这个解码器不依赖任何鸿蒙类型，纯算法，可以喂任意字节序列做测试。

测试维度：这个解码器的 Fixture 必须覆盖——UTF-8 跨 chunk（在 2 字节、3 字节、4 字节字符的每个字节边界都切一次）、CRLF/LF、4 字节 emoji 代理对、非法字节序列、流尾截断。这些都在离线单元测试里固化，不依赖真机。

---

七、StreamAssembler：字节→事件的完整管道

前面三层（编码、协议、语义）的组件都有了，StreamAssembler（openai/StreamAssembler.ets）是把它们串起来的总装车间。它持有四个内部组件的实例，把字节流变成 StreamingEvent 流：

export class StreamAssembler {
  private readonly profile: ProviderProfile
  private readonly observer: StreamObserver<StreamingEvent>
  private readonly utf8 = new IncrementalUtf8Decoder()
  private readonly sse = new SseParser()
  private readonly tools = new ToolCallAccumulator()
  private text: string = ''
  private reasoning: string = ''
  private usage?: ModelUsage
  private rawFinish?: string
  private terminated: boolean = false
  private modelStartSent: boolean = false
  // ...
}

7.1 字节进，事件出

pushBytes 是管道的入口，按"解码 → 解帧 → 处理 JSON"三步走：

pushBytes(chunk: Uint8Array): Result<void, ArkAgentError> {
  if (this.terminated) {
    return Result.success<void, ArkAgentError>(undefined as void)  // 终止后不再处理
  }
  // 第一步：字节 → 字符串（残字节在内部处理）
  const decoded = this.utf8.push(chunk)
  if (!decoded.ok || decoded.value === undefined) {
    return Result.failure<void, ArkAgentError>(decoded.error as ArkAgentError)
  }
  if (decoded.value.length === 0) {
    return Result.success<void, ArkAgentError>(undefined as void)  // 全是残字节，没有可处理字符
  }
  // 第二步：字符串 → SSE 事件
  const events = this.sse.push(decoded.value)
  if (!events.ok || events.value === undefined) {
    return Result.failure<void, ArkAgentError>(events.error as ArkAgentError)
  }
  // 第三步：逐条 SSE 事件 → JSON → StreamingEvent
  for (let i = 0; i < events.value.length; i++) {
    const handled = this.handleSse(events.value[i])
    if (!handled.ok) return handled
  }
  return Result.success<void, ArkAgentError>(undefined as void)
}

注意一个细节：UTF-8 解码返回空字符串时直接 return，不算错误。因为一个 chunk 可能全是某个多字节字符的前半段（残字节），解出来是空——这是正常的，等下个 chunk。把它当错误处理，会导致 chunk 边界上的合法字符丢失。

7.2 处理一条 SSE 事件

private handleSse(event: SseEvent): Result<void, ArkAgentError> {
  if (event.isDone) {                       // [DONE] 标记
    this.doneSeen = true
    return Result.success<void, ArkAgentError>(undefined as void)
  }
  if (event.data.trim().length === 0) {     // 空数据（心跳等）
    return Result.success<void, ArkAgentError>(undefined as void)
  }
  // 解析 JSON
  const parsed = JsonCodec.parseObject(event.data)
  if (!parsed.ok || parsed.value === undefined) {
    return Result.failure<void, ArkAgentError>(
      ArkAgentError.decode('sse_malformed_json', 'Malformed SSE JSON event'))
  }
  return this.handleChunk(parsed.value)
}

这里有个 ArkTS 特有的约束值得展开：为什么用 JsonCodec.parseObject 而不是 JSON.parse？ 因为 ArkTS 严格模式禁止 any，而 JSON.parse 返回 any——你拿到的动态对象无法用 obj.xxx 访问属性（会触发 arkts-no-indexed-signatures）。ArkAgent 设计了一套类型安全的 JsonObject/JsonValue 体系（下一篇博客会专门讲），所有 JSON 都走严格编解码器。流式场景下，每条 SSE 事件的 JSON 都经过这套编解码器，错误返回 Result 而不是抛异常。

7.3 从 JSON chunk 到事件

handleChunk 是把 Provider 的 wire JSON 翻译成领域事件的翻译器：

private handleChunk(root: JsonObject): Result<void, ArkAgentError> {
  const id = root.get('id')?.asString()
  if (id !== undefined) this.responseId = id
  const model = root.get('model')?.asString()
  if (model !== undefined) this.model = model

  // usage 可能出现在任意 chunk（通常是最后一个）
  const usageObj = root.get('usage')?.asObject()
  if (usageObj !== undefined) {
    this.usage = OpenAIWireCodec.parseUsage(usageObj)
    if (this.usage !== undefined) {
      this.safeNext(new StreamingEvent(StreamingEventType.usage, undefined, undefined, this.usage))
    }
  }

  const choices = root.get('choices')?.asArray()
  if (choices === undefined || choices.length === 0) {
    return Result.success<void, ArkAgentError>(undefined as void)
    // ↑ 空 choices 可能出现在 usage-only 的最终 chunk，不是错误
  }

  for (let c = 0; c < choices.length; c++) {
    const choice = choices[c].asObject()
    const choiceIndex = choice.get('index')?.asNumber() ?? c
    const finish = choice.get('finish_reason')?.asString()
    if (finish !== undefined && finish.length > 0) {
      this.rawFinish = finish
    }
    const delta = choice.get('delta')?.asObject()
    if (delta !== undefined) {
      const handled = this.handleDelta(choiceIndex, delta)
      if (!handled.ok) return handled
    }
  }
  return Result.success<void, ArkAgentError>(undefined as void)
}

handleDelta 是真正的"事件分发器"，它决定一个 delta 产生哪些事件：

private handleDelta(choiceIndex: number, delta: JsonObject): Result<void, ArkAgentError> {
  // 推理过程增量（reasoning_content，智谱/DeepSeek 都用这个字段名）
  const reasoning = delta.get('reasoning_content')?.asString()
  if (reasoning !== undefined && reasoning.length > 0) {
    this.reasoning += reasoning
    this.safeNext(new StreamingEvent(StreamingEventType.reasoningDelta, reasoning))
  }
  // 正文增量
  const content = delta.get('content')?.asString()
  if (content !== undefined && content.length > 0) {
    this.text += content
    this.safeNext(new StreamingEvent(StreamingEventType.textDelta, content))
  }
  // 工具调用增量（按 index 累积）
  const toolCalls = delta.get('tool_calls')?.asArray()
  if (toolCalls !== undefined) {
    for (let i = 0; i < toolCalls.length; i++) {
      const tc = toolCalls[i].asObject()
      const toolIndex = tc.get('index')?.asNumber() ?? i
      const id = tc.get('id')?.asString()
      const fn = tc.get('function')?.asObject()
      const name = fn?.get('name')?.asString()
      const args = fn?.get('arguments')?.asString()
      const snapshot = this.tools.applyDelta(choiceIndex, toolIndex, id, name, args)
      this.safeNext(new StreamingEvent(StreamingEventType.toolCallDelta, undefined, snapshot))
      // ↑ 注意：这里不标记"工具调用完成"，要等流结束 build() 才算完成
    }
  }
  return Result.success<void, ArkAgentError>(undefined as void)
}

这里有两个关键的"为什么"：

为什么 reasoning 和 text 都累积，但只发增量？ 累积是为了最后在 modelComplete 里给出完整内容；发增量是为了让 UI 能逐字显示。两者都需要——增量服务体验，累积服务数据完整性。

为什么 tool_call 每次都发快照，但不标记完成？ 因为工具参数是分片到达的，UI 可能想显示"正在拼装 get_weather 的参数：当前已收到 {"ci"。每次 applyDelta 返回一个包含当前累积进度的 ToolCall 快照，UI 可以选择展示。但"完成"这个状态必须等流结束（build()）才能确定——中间任何一刻都可能是半成品。

7.4 safeNext：终止后的安全阀

注意所有事件派发都走 safeNext：

private safeNext(event: StreamingEvent): void {
  if (this.terminated) return    // ← 终止后一个事件都不发
  this.observer.onNext(event)
}

这个守卫是终止状态机的最后一道防线。即使上层逻辑失误（比如错误处理路径漏了设标志），safeNext 也能保证终止后不再有事件泄漏给消费者。

---

八、⚠️ 踩坑二：终止事件被"自己"吞掉

这是 StreamAssembler 和终止状态机交互时最容易踩的坑，ArkAgent 在阶段 3 真实遇到过并写进了报告。

8.1 症状

流式调用，onComplete 触发了，但消费者收不到 modelComplete 事件。表现是：UI 上 loading 一直在转（因为靠 modelComplete 隐藏 loading），但日志显示流已经结束。

8.2 根因

错误的终止状态机写法——先 mark terminated，再 emit 终止事件：

// ❌ 错误顺序
private buildComplete(): Result<ModelResponse, ArkAgentError> {
  // ...拼装 response ...
  this.terminated = true                          // ← 先设标志
  this.observer.onNext(new StreamingEvent(        // ← 再发事件 → 被 safeNext 吞掉！
    StreamingEventType.modelComplete, undefined, undefined, undefined, response))
  return Result.success<ModelResponse, ArkAgentError>(response)
}

因为 safeNext 检查 terminated，如果你先把 terminated 设成 true，紧接着的 modelComplete 事件就被 safeNext 挡掉了。消费者永远收不到这条最关键的事件。

8.3 修复：先 emit 再 mark terminated

正确顺序：先派发终止事件，再设标志。

// ✅ 正确顺序（StreamAssembler 的真实实现）
private buildComplete(): Result<ModelResponse, ArkAgentError> {
  const toolResult = this.tools.build()
  // ... 拼 finish_reason、contents、reasoningContents、message、metadata ...

  const response = new ModelResponse(message, finish, this.usage, this.rawFinish,
    this.responseId, metadata)

  // 第一步：先 emit modelComplete（此时 terminated 还是 false，safeNext 放行）
  this.safeNext(new StreamingEvent(StreamingEventType.modelComplete, undefined, undefined,
    undefined, response))

  // 第二步：再 mark terminated（之后任何事件都被挡掉）
  this.terminated = true
  return Result.success<ModelResponse, ArkAgentError>(response)
}

这条规则被沉淀成一条铁律：emit terminal → set terminated。终止事件（modelComplete/error）必须在 terminated 置 true 之前发出。置 true 之后，管道进入"只读不写"的密封状态。

8.4 为什么这条规则容易被破坏

因为这个 bug 是静默的——不会报错、不会崩，就是少了一条事件。如果你不在测试里断言"终止后恰好收到 N 条事件、且最后一条是 modelComplete"，这个 bug 会潜伏很久。ArkAgent 的流式测试有一条专门的断言：统计 afterTerminal（终止后收到的事件数），它必须严格为 0——真机数据里 zhipu cancel 和 deepseek cancel 两条都显示 afterTerminal=0，证明这条规则在真机上成立。

预防方式：流式测试必须断言"单终止"（modelComplete 和 error 二选一，且只出现一次）和"终止后零事件"（afterTerminal=0）。这两条断言是把终止状态机的正确性变成可验证事实的唯一手段。

---

九、⚠️ 踩坑三：HTTP 流可能被缓冲（R-004）

这是 ArkAgent 风险登记册里编号 R-004 的开放风险，也是 Transport 设计成 SPI 的根本原因。

9.1 症状

在某些设备或网络环境下，流式响应不是逐 chunk 回调，而是等整个响应体到齐后一次性回调。表现是：用户看不到逐字输出，等了好几秒后突然蹦出一整段文字——流式退化成了"伪流式"。

9.2 根因

这是鸿蒙 @ohos.net.http 的平台行为，不是代码 bug。HTTP 栈在某些场景下（比如经过某些代理、某些网络配置）会缓冲整个响应体，不保证逐 chunk 回调 dataReceive。ArkAgent 在阶段 0 的 streaming spike 里就诚实地记录了这一点：

"本阶段没有创建网络 spike 页面或执行真机测试，因此尚未确认 API 24 下实际 chunk 粒度、后台行为和取消后的回调顺序。"

阶段 2 真机验收时也观察到 HarmonyHttpTransport 的 status 与 headersReceive 时序在极端缓冲场景下需要持续观察。这个风险至今在风险登记册里保持开放状态，没有假装它已解决。

9.3 修复思路：SPI 逃生通道

ArkAgent 没法改鸿蒙 HTTP 栈的行为，但它做了一个关键的架构决策——把 Transport 设计成 SPI，而不是绑死 HarmonyHttpTransport。

如果你遇到缓冲问题：

方案 A：换传输实现
  实现 HttpTransport 接口 → 用原生扩展或替代 HTTP 库
  StreamAssembler / Runtime / domain 完全不用改
  因为它们只依赖 HttpTransport SPI，不依赖 HarmonyHttpTransport

方案 B：观察 + 记录
  在 CountingHttpTransport（装饰器）里统计字节到达时序
  把数据反馈给平台方或留作真机验证证据

CountingHttpTransport 是一个装饰器实现，包在 HarmonyHttpTransport 外面，统计每个 chunk 的字节大小和到达时间戳，用于真机观察缓冲行为。这不解决缓冲，但让缓冲可观测。

9.4 真机验证的策略

R-004 的处理方式体现了 ArkAgent 对"平台未知数"的工程态度——不掩盖、不假装、留逃生通道：

做法	目的
Transport 设计成 SPI	缓冲问题出现时能整体替换实现
真机 spike 持续观察	积累设备/网络环境的行为数据
风险登记册保持开放	不在文档里假装它已解决
CountingHttpTransport 装饰器	让缓冲行为可度量
协议层与传输层正交	缓冲只影响"字节何时到"，不影响"字节如何解析"

教训：面对平台行为的不确定性，正确的工程姿势不是"我相信它没问题"，而是"假设它可能出问题，并确保出问题时我有退路"。SPI 就是这个退路。

---

十、ToolCallAccumulator：工具参数的分片拼装

流式工具调用是流式输出里最绕的部分。单独拎出来讲。

10.1 为什么工具参数是分片的

非流式响应里，一个 tool call 是完整的：

{"tool_calls":[{"id":"call_1","function":{"name":"get_weather","arguments":"{\"city\":\"北京\"}"}}]}

但流式响应里，同一个 tool call 会被切成多个 delta，按 index 区分，arguments 是字符串片段：

data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"get_weather","arguments":""}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"ci"}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"ty\":\"北京\""}}]}}]}
data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"}"}}]}}]}

更麻烦的是，多个工具会交错到达——index 0 和 index 1 的 arguments 片段可能交替出现。你不能假设"先收完工具 A，再收工具 B"。

10.2 按 index 累积

ToolCallAccumulator（openai/ToolCallAccumulator.ets）用 choiceIndex:toolIndex 作为 key 累积每个工具的片段：

class PartialToolCall {
  id: string = ''
  name: string = ''
  argumentsJson: string = ''
  index: number
  choiceIndex: number
}

export class ToolCallAccumulator {
  private readonly parts: Map<string, PartialToolCall> = new Map<string, PartialToolCall>()

  applyDelta(choiceIndex: number, toolIndex: number, id?: string, name?: string,
    argumentsFragment?: string): ToolCall {
    const key = `${choiceIndex}:${toolIndex}`        // ← 用 index 组合做 key
    let part = this.parts.get(key)
    if (part === undefined) {
      part = new PartialToolCall(choiceIndex, toolIndex)
      this.parts.set(key, part)
    }
    // id 和 name 只在首个 delta 出现，后续 delta 只有 arguments 片段
    if (id !== undefined && id.length > 0) part.id = id
    if (name !== undefined && name.length > 0) part.name = name
    if (argumentsFragment !== undefined) part.argumentsJson += argumentsFragment  // ← 字符串拼接
    // 返回当前进度的快照（供 toolCallDelta 事件）
    return new ToolCall(part.id, part.name, part.argumentsJson, part.index)
  }

  build(): Result<ToolCall[], ArkAgentError> {
    const keys = Array.from(this.parts.keys()).sort()   // ← 按 index 排序
    const result: ToolCall[] = []
    for (let i = 0; i < keys.length; i++) {
      const part = this.parts.get(keys[i]) as PartialToolCall
      if (part.id.length === 0) {
        return Result.failure(...'tool_call_missing_id'...)
      }
      if (part.name.length === 0) {
        return Result.failure(...'tool_call_missing_name'...)
      }
      // 注意：这里不解析 argumentsJson，原样保留
      result.push(new ToolCall(part.id, part.name, part.argumentsJson, part.index))
    }
    return Result.success<ToolCall[], ArkAgentError>(result)
  }
}

三个要点：

第一，key 是 choiceIndex:toolIndex，不是 id。 因为首个 delta 可能还没带 id，你不能用 id 做 key。用 Provider 给的 index 组合，保证同一个工具的片段一定落到同一个 PartialToolCall。

第二，build() 时排序。 Map 的遍历顺序是插入顺序，而工具可能交错到达。按 key 字符串排序后，工具顺序和 Provider 声明的 index 一致。

第三，build() 只校验 id 和 name，不解析 arguments。 注释写得很明确：// Arguments may be empty object string; do not parse until complete. arguments 的 JSON 解析在更上层（Runtime 执行工具前）做，并配合 JSON Schema 校验。这里只负责把字符串片段拼完整。

10.3 真机数据：分片确实在发生

真机验收数据印证了分片机制：

zhipu tool:    toolCount=1 firstTool=get_weather argsLen=20 ms=2953
deepseek tool: toolCount=1 firstTool=get_weather argsLen=21 ms=951

两家 Provider 最终都正确累积出 1 个 get_weather 工具调用，但 argsLen（最终 arguments JSON 的长度）分别是 20 和 21——说明两家的参数分片粒度/格式略有差异，但都被 ToolCallAccumulator 正确聚合。如果分片拼装有 bug，argsLen 会对不上，或者 toolCount 会变成 0/2（拼错或拼重）。

10.4 "只在 modelComplete 后执行工具"的天然约束

这里有个优雅的架构决策：Runtime 完全不感知 wire delta，只等 modelComplete。

流式 delta 到达 → ToolCallAccumulator 累积（但不执行）
流结束 → build() 拼出完整 ToolCall[] → modelComplete 事件携带
Runtime 收到 modelComplete → 取出完整 ToolCall[] → 此时才执行工具

这意味着：工具执行天然发生在参数拼装完成之后。Runtime 不需要自己判断"参数拼完了没"，StreamAssembler 的设计已经保证了 modelComplete 里的 ToolCall 一定是完整的。这是一个把"数据完整性约束"内化到管道结构的例子——不用靠运行时检查，靠架构保证。

---

十一、Provider 差异：智谱 vs DeepSeek 的流式参数

智谱和 DeepSeek 都兼容 OpenAI Chat Completions，但流式相关有几个关键差异。这些差异集中在各自的 Provider Profile 里。

11.1 差异矩阵

差异点	智谱 GLM	DeepSeek
流式扩展参数	thinking、tool_stream、reasoning_effort	thinking、reasoning_effort
推理内容字段	reasoning_content（与 content 分开）	reasoning_content（与 content 分开）
特有 finish reason	sensitive（→ contentFilter）、network_error（可重试）、model_context_window_exceeded	insufficient_system_resource（可重试）
thinking 与 tool_choice	可共存	V4 thinking 默认开启时不支持 tool_choice
图片输入	支持（VLM 模型）	不支持

11.2 扩展参数怎么注入

Provider 特有的流式参数不进入 Core 的固定字段，而是通过 ModelConfig.extensions 传入，由各自的 Profile 消费。智谱的 Profile 消费三个扩展：

// ZhipuProviderProfile.applyRequestExtensions
applyRequestExtensions(body: JsonObject, request: ModelRequest): Result<void, ArkAgentError> {
  const extensions = request.modelConfig.extensions
  const thinking = ProviderExtensionGuard.readObject(extensions, 'thinking')
  if (thinking !== undefined) {
    body.set('thinking', JsonValue.object(thinking))       // 推理开关
  }
  const toolStream = ProviderExtensionGuard.readBoolean(extensions, 'tool_stream')
  if (toolStream !== undefined) {
    body.set('tool_stream', JsonValue.boolean(toolStream)) // 工具流式输出
  }
  const effort = ProviderExtensionGuard.readString(extensions, 'reasoning_effort')
  if (effort !== undefined) {
    body.set('reasoning_effort', JsonValue.string(effort)) // 推理强度
  }
  return Result.success<void, ArkAgentError>(undefined as void)
}

DeepSeek 的 Profile 只消费 thinking 和 reasoning_effort——没有 tool_stream：

// DeepSeekProviderProfile.applyRequestExtensions
applyRequestExtensions(body: JsonObject, request: ModelRequest): Result<void, ArkAgentError> {
  const extensions = request.modelConfig.extensions
  const thinking = ProviderExtensionGuard.readObject(extensions, 'thinking')
  if (thinking !== undefined) {
    body.set('thinking', JsonValue.object(thinking))
  }
  const effort = ProviderExtensionGuard.readString(extensions, 'reasoning_effort')
  if (effort !== undefined) {
    body.set('reasoning_effort', JsonValue.string(effort))
  }
  return Result.success<void, ArkAgentError>(undefined as void)
}

关键设计：未消费的扩展必须报配置错误，不能静默发送。如果用户给 DeepSeek 配了 tool_stream: true，但 DeepSeek Profile 不消费它，这个扩展会被 ProviderExtensionGuard 标记为未消费，调用前报 config error。为什么这么严格？因为静默发送一个 Provider 不认识的参数，行为不可预测——有的 Provider 会报错，有的会忽略，有的会产生奇怪副作用。与其让 bug 隐藏在"静默忽略"里，不如显式拒绝。

11.3 DeepSeek thinking 与 tool_choice 的冲突

这是真机踩出来的一个配置陷阱：DeepSeek V4 在 thinking 默认开启时，不支持 tool_choice。

如果你同时开了 thinking 和 tool_choice（比如强制 tool_choice: auto 或 named tool），DeepSeek 会报错或忽略 tool_choice。ArkAgent 的示例工程在 DeepSeek 的工具调用场景里，显式把 thinking 关掉：

DeepSeek tool 用例：
  thinking.type = disabled
  tool_choice = named:get_weather
  → toolCount=1, finish=tool_calls  ✅

这条经验写进了阶段报告，Demo 和 Profile 文档同步：DeepSeek 工具调用场景默认 thinking.disabled。这不是 SDK 的 bug，是 Provider 的能力边界，但如果不记录，每个新接入的人都会踩一遍。

11.4 finish reason 的归一化

两家都有 OpenAI 之外的特有 finish reason，各自在 Profile 里做归一化：

// 智谱
mapFinishReason(raw?: string): FinishReason {
  if (raw === 'sensitive' || raw === 'content_filter') return FinishReason.contentFilter
  if (raw === 'network_error') return FinishReason.error
  if (raw === 'model_context_window_exceeded') return FinishReason.length
  // ...
}

// DeepSeek
mapFinishReason(raw?: string): FinishReason {
  if (raw === 'insufficient_system_resource') return FinishReason.error
  // ...
}

而且有些 finish reason 直接映射成错误（而不是正常的 modelComplete）：

// 智谱：network_error 和 context_window_exceeded 直接当错误
mapProviderFinishAsError(raw: string): ArkAgentError | undefined {
  if (raw === 'network_error') {
    return ArkAgentError.provider('network_error', 'Provider network error',
      ErrorRetryability.conditional, undefined, raw)    // ← 可条件重试
  }
  if (raw === 'model_context_window_exceeded') {
    return ArkAgentError.provider('context_window_exceeded',
      'Model context window exceeded', ErrorRetryability.never, undefined, raw)  // ← 永不重试
  }
  return undefined
}

StreamAssembler.buildComplete() 在拼装 modelComplete 之前先调 mapProviderFinishAsError——如果某个 finish reason 代表错误，流以 error 事件终止，而不是 modelComplete。这样 network_error 能进入重试流程，而 context_window_exceeded（上下文超限，重试也没用）被标记为 never 重试。

---

十二、流式重试：错误边界与安全约束

流式场景的重试比非流式复杂得多，因为有个致命约束：一旦工具已经执行（有副作用），就不能重试了。

12.1 RetryPolicy 的核心判断

export class RetryPolicy {
  decide(error: ArkAgentError, attempt: number, headers?: HttpHeaders,
    toolCallCompleted: boolean = false): RetryDecision {
    // 第一道闸：工具已执行 → 永不重试
    if (toolCallCompleted) {
      return new RetryDecision(false)
    }
    // 第二道闸：超过最大次数
    if (attempt >= this.maxRetries) {
      return new RetryDecision(false)
    }
    // 第三道闸：用户取消 → 永不重试
    if (error.layer === 'cancelled') {
      return new RetryDecision(false)
    }
    // 第四道闸：不可重试的错误（401/403 等）
    if (error.retryability === ErrorRetryability.never) {
      return new RetryDecision(false)
    }
    // 可重试的错误 → 指数退避
    if (error.retryability === ErrorRetryability.safe ||
      error.retryability === ErrorRetryability.conditional) {
      return new RetryDecision(true, this.computeDelay(attempt, headers))
    }
    return new RetryDecision(false)
  }

  private computeDelay(attempt: number, headers?: HttpHeaders): number {
    // 优先尊重 Retry-After 响应头
    const retryAfter = headers?.get('retry-after')
    if (retryAfter !== undefined) {
      const seconds = Number(retryAfter)
      if (Number.isFinite(seconds) && seconds >= 0) {
        return Math.min(Math.floor(seconds * 1000), this.maxDelayMs)
      }
    }
    // 否则指数退避：initial * 2^attempt，封顶 maxDelay
    let delay = this.initialDelayMs * Math.pow(2, attempt)
    if (delay > this.maxDelayMs) delay = this.maxDelayMs
    return delay
  }
}

重试决策的优先级清晰：工具已执行 > 次数上限 > 用户取消 > 错误不可重试 > 可重试。

12.2 流式重试的特殊场景

OpenAICompatibleClient.stream 在几个地方调 RetryPolicy.decide，每个场景传的 toolCallCompleted 参数不同：

场景	toolCallCompleted	能否重试
HTTP 4xx/5xx（还没开始流）	assembler.hasEmittedCompletedToolCall	通常可重试
流中途字节错误（已收字节）	assembler.hasEmittedCompletedToolCall || state.sawBytes	已收字节则不重试
流结束解析失败	assembler.hasEmittedCompletedToolCall	未拼出工具才可重试

注意第二个场景的严格性：只要已经开始接收字节（sawBytes=true），即使还没拼出完整工具，也不重试。为什么？因为字节已经部分到达，说明 Provider 已经处理了请求、可能已经产生了副作用（比如计费、限流计数）。重试可能导致重复计费或触发限流。宁可失败，不可冒险。

12.3 重试是可观测的

每次重试都会发一个 retry 事件，让 UI 有机会提示用户：

private emitRetry(observer: StreamObserver<StreamingEvent>, attempt: number,
  error: ArkAgentError, delayMs: number): void {
  const metadata = new JsonObject()
    .set('attempt', JsonValue.number(attempt))
    .set('errorCode', JsonValue.string(error.code))
    .set('delayMs', JsonValue.number(delayMs))
  observer.onNext(new StreamingEvent(StreamingEventType.retry, undefined, undefined, undefined,
    undefined, undefined, metadata))
}

retry 事件携带 attempt 编号、错误码和延迟毫秒数。UI 可以显示"网络波动，第 2 次重试中（1.5 秒后）"，而不是让用户对着卡住的 loading 干瞪眼。

12.4 错误的统一出口

不管是重试耗尽、不可重试错误、还是用户取消，最终都走 emitError：

private emitError(observer: StreamObserver<StreamingEvent>, error: ArkAgentError): void {
  observer.onNext(new StreamingEvent(StreamingEventType.error, undefined, undefined, undefined,
    undefined, error))
  observer.onError(error)    // ← 同时触发 onError 回调
}

error 事件和 onError 回调是成对的——先发 error 事件（让关心事件流的消费者处理），再触发 onError（让关心"流结束"语义的消费者收尾）。两者携带同一个 ArkAgentError，消费者可以选择处理哪个。

---

十三、真机流式数据：两家 Provider 的真实表现

理论讲完了，看真机数据。以下是 ArkAgent 阶段 2 Provider 在线验收的真实流式事件统计（脱敏，不含密钥/prompt/响应内容）：

zhipu stream model=glm-5.2 status=ok
  events=model_complete=1,model_start=1,reasoning_delta=94,text_delta=2,usage=1
  hasDelta=true ms=2058

deepseek stream model=deepseek-v4-flash status=ok
  events=model_complete=1,model_start=1,reasoning_delta=21,text_delta=1,usage=1
  hasDelta=true ms=879

几个值得关注的点：

第一，事件计数完全符合契约。 两家都是 model_start=1（恰好一次）、model_complete=1（恰好一次）、usage=1。这验证了"单终止"和"modelComplete 恰好一次"的契约在真机上成立。

第二，reasoning_delta 数量差异巨大。 智谱 94 个、DeepSeek 21 个——智谱的推理过程分片比 DeepSeek 细约 4.5 倍。这说明两家 Provider 的流式分片粒度不同，但都被同一个 StreamAssembler 正确处理。如果你的流式管道假设"reasoning_delta 不会超过 N 个"，在智谱上就会踩坑。

第三，text_delta 都很少。 智谱 2 个、DeepSeek 1 个。说明两家都把大部分生成都放在了 reasoning（思考）阶段，正文反而很简短。这印证了"reasoning 和 text 必须分开事件"的必要性——如果合成一种 delta，94+2=96 个事件里你分不清哪些是想、哪些是说。

取消行为的真机验证：

zhipu cancel:    errorLayer=cancelled completed=false hasComplete=false afterTerminal=0 events=model_start=1 ms=1376
deepseek cancel: errorLayer=cancelled completed=false hasComplete=false afterTerminal=0 events=model_start=1 ms=282

afterTerminal=0 是最关键的指标——取消（终止）之后，消费者没有收到任何额外事件。这验证了 HarmonyHttpTransport 的 terminated 标志和 StreamAssembler 的 safeNext 守卫在真机上有效阻断了"取消后的事件泄漏"。

---

十四、最佳实践清单

14.1 架构分层

• ✅ 把流式管道分成编码、协议、语义三层，职责正交、状态隔离。

• ✅ 传输层设计成 SPI，不绑死 @ohos.net.http，给缓冲问题留逃生通道。

• ✅ 协议层（SseParser）和编码层（IncrementalUtf8Decoder）不依赖任何鸿蒙类型，可用纯 Fixture 测试。

• ✅ modelComplete 必须携带完整 ModelResponse，delta 事件只服务体验，不服务数据完整性。

• ✅ Runtime 只在 modelComplete 后执行工具，天然保证参数拼装完成。

14.2 事件契约

• ✅ 固定 8 种 StreamingEvent，不多不少——再多会泄漏 Provider 细节。

• ✅ Provider 特有 finish reason 在各自 Profile 里归一化，不新增事件类型。

• ✅ 终止事件（modelComplete/error）必须先 emit 再 mark terminated。

• ✅ 取消后终止后零事件（afterTerminal=0），用测试断言固化。

• ✅ error 事件和 onError 回调成对触发，携带同一个 ArkAgentError。

14.3 编码与协议

• ✅ 流式 HTTP 用 ARRAY_BUFFER 模式，把字节解码权拿回自己手里。

• ✅ UTF-8 解码必须跨 chunk 保留残字节，用 IncrementalUtf8Decoder。

• ✅ SSE 解析同时兼容 LF 和 CRLF，处理多行 data、注释、空行分帧。

• ✅ [DONE] 只是数据内容，不是协议事件，由上层判断。

• ✅ JSON 解析用类型安全的 JsonCodec，不用 JSON.parse（ArkTS 禁 any）。

14.4 工具与重试

• ✅ 工具参数按 choiceIndex:toolIndex 累积，build 时排序。

• ✅ 工具 arguments 在 modelComplete 后才解析，不在 delta 阶段解析半成品。

• ✅ 工具已执行（toolCallCompleted）后永不重试。

• ✅ 已开始接收字节（sawBytes）后不重试，避免重复计费/限流。

• ✅ 重试发 retry 事件，让 UI 可观测。

• ✅ 用户取消永不重试，取消错误自成一类。

14.5 Provider 差异

• ✅ Provider 特有参数走 ModelConfig.extensions，由 Profile 消费。

• ✅ 未消费的扩展必须报 config error，不静默发送。

• ✅ DeepSeek 工具调用场景默认 thinking.disabled（V4 thinking 与 tool_choice 冲突）。

• ✅ 模型名不进入 Core enum，作为配置字符串传入。

---

十五、常见错误对照表

错误做法	问题	正确做法
用 STRING 模式收流式响应	鸿蒙自动解码，破坏流式语义	用 ARRAY_BUFFER + 自己的 UTF-8 解码器
每个 chunk 独立 UTF-8 解码	多字节字符跨 chunk 切断 → 乱码	用 IncrementalUtf8Decoder 保留残字节
假设"一次 dataReceive = 一条事件"	帧粘包，事件截断	用有状态的 SseParser 跨 push 保留 buffer
只认 [DONE] 判断结束	有些 Provider 不发 [DONE]	以 dataEnd/finish_reason/modelComplete 综合判断
reasoning 和 text 合成一种 delta	UI 分不清"在想"还是"在说"	分成 reasoningDelta 和 textDelta
让消费者自己累积 delta	每个消费者重复实现且各错各的	StreamAssembler 累积，modelComplete 给完整响应
先 mark terminated 再 emit 终止事件	终止事件被 safeNext 吞掉	先 emit terminal 再 set terminated
工具参数在 delta 阶段就解析	参数是半成品，解析必失败	等 modelComplete 的 build() 后再解析
工具执行后还允许重试	副作用已发生，重试导致重复	toolCallCompleted=true 后永不重试
已收字节后还重试	可能重复计费/触发限流	sawBytes=true 后不重试
用 JSON.parse 解析 SSE 数据	ArkTS 禁 any，编译拒绝	用 JsonCodec.parseObject
把 tool_stream 配给 DeepSeek	DeepSeek 不支持，静默忽略行为未知	未消费扩展必须报 config error
DeepSeek 工具场景不关 thinking	thinking 与 tool_choice 冲突，工具失败	默认 thinking.type=disabled
取消后不销毁 HTTP 连接	连接继续烧 token	cancel 立即 client.destroy()
把缓冲问题当 bug 修	平台行为，不可在 SDK 层"修复"	Transport SPI + 真机观察 + 风险登记
用对象字面量实现 StreamObserver	ArkTS 禁未类型化字面量，编译报错	用显式 class 实现

---

十六、验证清单（真机 Review）

基础流式

• 智谱 glm-5.2 流式输出逐字显示，不乱码

• DeepSeek deepseek-v4-flash 流式输出逐字显示，不乱码

• 中文长文本流式无残字节乱码

• emoji 流式输出正确（4 字节 UTF-8 代理对）

• reasoning 和 text 分开显示

事件契约

• model_start 恰好 1 次

• model_complete 恰好 1 次（正常路径）

• 终止后 afterTerminal=0（无泄漏事件）

• modelComplete 携带完整 ModelResponse

• error 和 onError 成对触发

工具调用流式

• 流式 tool call 参数正确拼装（argsLen 与预期一致）

• 多个工具交错到达时按 index 正确聚合

• toolCount 与预期一致

• 工具在 modelComplete 后才执行

取消与重试

• 用户取消后立即停止接收事件（afterTerminal=0）

• 取消后 HTTP 连接销毁

• 429/5xx 触发指数退避重试

• 重试时收到 retry 事件

• 401/403 不重试

• 工具执行后不重试

Provider 差异

• 智谱 thinking 扩展生效

• 智谱 tool_stream 扩展生效

• DeepSeek thinking 扩展生效

• DeepSeek 工具场景 thinking 已 disabled

• 给 DeepSeek 配 tool_stream 报 config error

• 智谱 network_error finish reason 触发可重试错误

• DeepSeek insufficient_system_resource 触发可重试错误

离线 Fixture（不依赖真机）

• UTF-8 跨 chunk（2/3/4 字节字符每个边界切一次）解码正确

• CRLF 和 LF 混合的 SSE 正确切帧

• 多行 data 拼接正确

• 注释行和心跳不影响事件

• [DONE] 正确识别

• 畸形 JSON 返回 sse_malformed_json 错误

• 流尾截断返回 utf8_truncated 错误

---

十七、构建验证

# 构建 agent_core HAR
NODE_HOME=/Applications/DevEco-Studio.app/Contents/tools/node \
  DEVECO_SDK_HOME=/Applications/DevEco-Studio.app/Contents/sdk \
  /Applications/DevEco-Studio.app/Contents/tools/hvigor/bin/hvigorw assembleHar --no-daemon

# 运行流式相关单元测试（含 SSE / UTF-8 / Streaming）
NODE_HOME=/Applications/DevEco-Studio.app/Contents/tools/node \
  DEVECO_SDK_HOME=/Applications/DevEco-Studio.app/Contents/sdk \
  /Applications/DevEco-Studio.app/Contents/tools/hvigor/bin/hvigorw test --no-daemon

构建结果：

CompileArkTS passed
BUILD SUCCESSFUL

流式相关的离线测试覆盖：SseAndUrl.test.ets（SSE 解帧 + URL 工具）、Streaming.test.ets（流式事件组装）、FakeHttpTransport（注入固定字节序列验证 UTF-8 跨 chunk）。这些测试不依赖真机网络，用固定的字节 Fixture 复现 chunk 边界场景。

真机流式验收数据（脱敏）见 docs/reports/phases/phase-02-live-acceptance-summary.txt，覆盖智谱和 DeepSeek 的流式文本、流式工具、取消、鉴权失败四个场景。

---

十八、写在最后

流式输出是 AI Agent 体验的灵魂——用户对"智能"的感知，很大程度上来自"它在逐字跟我说话"。但在鸿蒙上做好流式，要穿越编码、协议、语义三层挑战，每一层都有自己的陷阱：UTF-8 跨 chunk 切断会让中文乱码、SSE 帧粘包会让事件错乱、终止状态机写反会让事件丢失、HTTP 缓冲会让流式退化成伪流式、工具参数分片会让执行失败、重试时机不对会让副作用翻倍。

ArkAgent 的流式管道把这些挑战收敛成四个正交的组件：

HarmonyHttpTransport    字节怎么来（SPI，可替换）
  ↓ Uint8Array
IncrementalUtf8Decoder  字节怎么解码（残字节保留）
  ↓ string
SseParser               帧怎么拆（LF/CRLF/多行/注释）
  ↓ SseEvent
StreamAssembler         事件怎么组装（8 种 + 终止状态机）
  ↓ StreamingEvent
你的 observer           业务怎么响应

记住这几条口诀：

流式三层要分清，编码协议加语义。 UTF-8 切跨块，残字节留到下一回。 SSE 帧会粘包，buffer 留尾按行切。 CRLF 和 LF 都要认，换行符不认就丢帧。 八种事件不多不少，reasoning 和 text 要分家。 modelComplete 带全量，delta 只管逐字蹦。 终止先 emit 后标记，顺序反了事件没。 取消之后零事件，afterTerminal 必须零。 工具参数按 index，拼完才许去执行。 工具执行不重试，副作用了别回头。 已收字节不重试，重复计费风险高。 Transport 做成 SPI，缓冲来了能换鞋。 Provider 差异进 Profile，未消费扩展要报错。

本文是 ArkAgent 鸿蒙教程系列的第四篇。第一篇讲了架构全景，第二篇讲了快速接入，这篇拆解了流式输出这条最核心、最多坑的管道。下一篇会讲 ArkTS 里另一个基础且致命的主题——JSON 类型安全：为什么 ArkTS 禁止 any 和动态 key，以及怎么设计一套类型安全的 JsonValue 体系。

如果你正在鸿蒙上做流式 AI 应用，可以把这条四层管道直接作为你的流式架构基线。不要从"dataReceive 里 JSON.parse"开始——那是乱码和丢帧的开始。