Function tailRun

examples/sandbox-cloudflare/src/routes/api.run.ts:178–266

* Open the run's WebSocket tail (the coordinator's `fetch` returns a `101` with a * `webSocket`) and yield each chat `StreamChunk` as it arrives. Resolves when the * coordinator sends its terminal `status` frame (or the socket closes / the client * disconnects).

(
  coordinator: Coordinator,
  runId: string,
  threadId: string,
  signal: AbortSignal,
)

 * disconnects).
 */
async function* tailRun(
  coordinator: Coordinator,
  runId: string,
  threadId: string,
  signal: AbortSignal,
): AsyncGenerator<StreamChunk> {
  // The host is irrelevant — the DO routes on the pathname; this is an in-process
  // DO `fetch`, not a public request.
  const streamUrl = `https://do/runs/${runId}/stream?threadId=${encodeURIComponent(threadId)}&lastSeq=-1`
  const res = await coordinator.fetch(streamUrl, {
    headers: { Upgrade: 'websocket' },
  })
  const socket = res.webSocket
  if (!socket) {
    throw new Error(
      `agent stream did not upgrade to a WebSocket (status ${res.status})`,
    )
  }
  socket.accept()

  const queue: Array<StreamChunk> = []
  // Mutated from the socket/abort callbacks below. Held on an object (rather than
  // bare `let`s) so the generator loop's checks aren't flagged as constant.
  const state: { finished: boolean; failure: Error | null } = {
    finished: false,
    failure: null,
  }
  let wake: (() => void) | null = null
  const signalReady = () => {
    wake?.()
    wake = null
  }

  socket.addEventListener('message', (event: MessageEvent) => {
    let parsed: unknown
    try {
      parsed = JSON.parse(typeof event.data === 'string' ? event.data : '')
    } catch {
      return
    }
    if (parsed === null || typeof parsed !== 'object') return
    if ('type' in parsed && parsed.type === 'status') {
      state.finished = true
    } else if ('chunk' in parsed) {
      queue.push(parsed.chunk as StreamChunk)
    }
    signalReady()
  })
  socket.addEventListener('close', () => {
    state.finished = true
    signalReady()
  })
  socket.addEventListener('error', () => {
    state.failure = new Error('agent stream socket error')
    state.finished = true
    signalReady()
  })
  const onAbort = () => {