T2 — Streaming event demultiplexer

{`
const handleEvent = Effect.fnUntraced(function* (value: StreamEvent) {
  switch (value.type) {

    case "reasoning-start":
      ctx.reasoningMap[value.id] = {
        id: PartID.ascending(),
        text: "",
        time: { start: Date.now() }
      }
      yield* session.updatePart(ctx.reasoningMap[value.id])
      return

    case "reasoning-delta":
      ctx.reasoningMap[value.id].text += value.text
      yield* session.updatePartDelta({ delta: value.text })
      return

    case "text-delta":
      ctx.currentText = (ctx.currentText ?? "") + value.text
      yield* session.updatePartDelta({ delta: value.text })
      return

    case "tool-input-start":
      const part = yield* session.updatePart({ status: "pending", input: {} })
      ctx.toolcalls[value.id] = {
        done: yield* Deferred.make<void>(),  // ← T3: Deferred sync point
        partID: part.id
      }
      return

    case "tool-call":
      yield* updateToolCall(value.toolCallId, (m) => ({
        ...m, status: "running", input: value.input
      }))
      // Doom loop check — T6
      yield* checkDoomLoop(value.toolName, value.input)
      return

    case "finish":
      ctx.finish = value.finishReason
      return
  }
})
`}

{`
// llm.stream() trả Effect.Stream từ AI SDK async iterable
const stream = llm.stream(streamInput)

yield* stream.pipe(
  Stream.tap((event) => handleEvent(event)),     // route mỗi event
  Stream.takeUntil(() => ctx.needsCompaction),   // stop sớm nếu overflow
  Stream.runDrain,                                // consume toàn bộ
)
`}

{`
const ctx = {
  currentText: undefined as string | undefined,
  reasoningMap: {} as Record<string, ReasoningPart>,
  toolcalls: {} as Record<string, ToolCall>,   // T3 Deferred entries
  finish: undefined as string | undefined,
  needsCompaction: false,
  assistantMessage: ...,
}
`}

{`
// WRONG: không streaming
const response = await llm.generate(messages)  // block đến khi done

// Hậu quả:
// - User thấy màn hình trắng trong 10-30 giây
// - Không thể cancel giữa chừng
// - Reasoning text bị mất nếu error xảy ra giữa stream
`}

{`
// Nhiều SDK chỉ expose:
case "tool_call":  // đã có đủ name + input
  execute(tc)

// opencode tách làm 2 phase:
case "tool-input-start":  // UI hiển thị "pending"
  ctx.toolcalls[id] = { done: Deferred.make(), status: "pending" }
case "tool-call":          // UI chuyển sang "running"
  updateToolCall(id, { status: "running" })
`}

{`
// WRONG: không có cancel mechanism
for await (const event of stream) {
  handle(event)
  // nếu user cancel, vòng lặp vẫn tiếp tục
  // cho đến khi stream kết thúc tự nhiên
}

// CORRECT: cooperative cancel
for await (const event of stream) {
  if (abortSignal.aborted) break  // check mỗi event
  handle(event)
}
`}

{`
type StreamEvent =
  | { type: "text-delta"; text: string }
  | { type: "tool-input-start"; id: string; toolName: string }
  | { type: "tool-call"; toolCallId: string; toolName: string; input: unknown }
  | { type: "reasoning-start"; id: string }
  | { type: "reasoning-delta"; id: string; text: string }
  | { type: "finish"; finishReason: string }

interface DemuxContext {
  currentText: string
  reasoningMap: Record<string, { text: string }>
  pendingTools: Map<string, { resolve: () => void; promise: Promise<void> }>
  finishReason: string | null
  needsStop: boolean
}

async function demuxStream(
  stream: AsyncIterable<StreamEvent>,
  ctx: DemuxContext,
  signal?: AbortSignal
) {
  for await (const event of stream) {
    // Cooperative cancel
    if (signal?.aborted || ctx.needsStop) break

    switch (event.type) {
      case "reasoning-start":
        ctx.reasoningMap[event.id] = { text: "" }
        break

      case "reasoning-delta":
        ctx.reasoningMap[event.id].text += event.text
        break

      case "text-delta":
        ctx.currentText += event.text
        break

      case "tool-input-start": {
        // Track deferred cho T3 pattern
        let resolve!: () => void
        const promise = new Promise<void>((r) => { resolve = r })
        ctx.pendingTools.set(event.id, { resolve, promise })
        break
      }

      case "tool-call":
        // Execute async, resolve deferred khi done
        const deferred = ctx.pendingTools.get(event.toolCallId)
        executeToolAsync(event).then(() => deferred?.resolve())
        break

      case "finish":
        ctx.finishReason = event.finishReason
        break
    }
  }
}

// Tách execute async — không block stream consumer
async function executeToolAsync(tc: Extract<StreamEvent, { type: "tool-call" }>) {
  // tool execution logic
}
`}