| 260 | // through reach the client unmodified. The caller owns producing |
| 261 | // correctly-framed SSE; this method only relays bytes. |
| 262 | async _streamResponse(res, result) { |
| 263 | const headers = Object.assign({ |
| 264 | 'Content-Type': 'text/event-stream', |
| 265 | 'Cache-Control': 'no-cache', |
| 266 | 'Connection': 'keep-alive', |
| 267 | }, result.headers || {}); |
| 268 | res.writeHead(result.status || 200, headers); |
| 269 | |
| 270 | // Browsers, network blips, and Ctrl-C all close the SSE socket mid-stream. |
| 271 | // If we await `drain` without watching for `close`, the drain event never |
| 272 | // fires on a destroyed socket and this coroutine hangs forever — which |
| 273 | // also pins the upstream fetch body open (real Anthropic socket leak, not |
| 274 | // just coroutine leak). For Web ReadableStream upstreams we read via an |
| 275 | // explicit reader so cancellation can go through the same lock; for sync |
| 276 | // generators / Node Readables we fall back to for-await. |
| 277 | const stream = result.stream; |
| 278 | const reader = stream && typeof stream.getReader === 'function' ? stream.getReader() : null; |
| 279 | |
| 280 | let clientGone = false; |
| 281 | const onClose = () => { |
| 282 | clientGone = true; |
| 283 | if (reader) { |
| 284 | reader.cancel().catch(() => { /* upstream already settled */ }); |
| 285 | } else if (stream && typeof stream.destroy === 'function') { |
| 286 | try { stream.destroy(); } catch { /* ignore */ } |
| 287 | } |
| 288 | }; |
| 289 | res.once('close', onClose); |
| 290 | |
| 291 | const awaitBackpressure = () => new Promise((resolve) => { |
| 292 | let settled = false; |
| 293 | const onDrain = () => { |
| 294 | if (settled) return; |
| 295 | settled = true; |
| 296 | res.off('close', onCloseInner); |
| 297 | resolve(); |
| 298 | }; |
| 299 | const onCloseInner = () => { |
| 300 | if (settled) return; |
| 301 | settled = true; |
| 302 | res.off('drain', onDrain); |
| 303 | resolve(); |
| 304 | }; |
| 305 | res.once('drain', onDrain); |
| 306 | res.once('close', onCloseInner); |
| 307 | }); |
| 308 | |
| 309 | try { |
| 310 | if (reader) { |
| 311 | for (;;) { |
| 312 | const { value, done } = await reader.read(); |
| 313 | if (done || clientGone) break; |
| 314 | if (!res.write(value)) { |
| 315 | await awaitBackpressure(); |
| 316 | if (clientGone) break; |
| 317 | } |
| 318 | } |
| 319 | } else { |