長考の途中で切れる接続は、上流ではなく中継が作っています

// origin.mjs — 上流の模擬。最初の1フレームのあと長考に入り、ping だけを流す
import http from 'node:http';
 
const SILENCE = Number(process.env.SILENCE ?? 6000); // 長考の長さ
const PING = Number(process.env.PING_MS ?? 0);       // 0 なら ping を出さない
 
http.createServer((req, res) => {
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  });
  res.write('event: message_start\ndata: {"type":"message_start"}\n\n');
 
  // 長考区間。本文は出ず、コメント行だけが流れる
  const iv = PING ? setInterval(() => res.write(': ping\n\n'), PING) : null;
 
  setTimeout(() => {
    if (iv) clearInterval(iv);
    res.write('event: content_block_delta\ndata: {"text":"done"}\n\n');
    res.write('event: message_stop\ndata: {}\n\n');
    res.end();
  }, SILENCE);
}).listen(4001);

// client.mjs — バイトが1つでも届けばタイマーを戻す
const IDLE = Number(process.env.IDLE ?? 4000);
const t0 = Date.now();
const ctrl = new AbortController();
let timer = setTimeout(() => ctrl.abort(), IDLE);
let bytes = 0;
let wakeups = 0;
 
try {
  const r = await fetch('http://127.0.0.1:4002/v1/messages', { signal: ctrl.signal });
  const reader = r.body.getReader();
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    bytes += value.length;
    wakeups++;
    clearTimeout(timer);                                  // ← 内容は一切見ていない
    timer = setTimeout(() => ctrl.abort(), IDLE);
  }
  clearTimeout(timer);
  console.log(`OK    elapsed=${Date.now() - t0}ms bytes=${bytes} wakeups=${wakeups}`);
} catch (e) {
  clearTimeout(timer);
  console.log(`ABORT elapsed=${Date.now() - t0}ms bytes=${bytes} wakeups=${wakeups}`);
}

// proxy.mjs — 中継。MODE を差し替えて挙動を比較する
import http from 'node:http';
 
const MODE = process.env.MODE ?? 'stream';
const FLUSH = Number(process.env.FLUSH ?? 5000);
 
http.createServer(async (req, res) => {
  const up = await fetch('http://127.0.0.1:4001/v1/messages');
  res.writeHead(200, { 'Content-Type': 'text/event-stream' });
 
  // よくある一行。ここで全部溜まる
  if (MODE === 'buffer') {
    res.end(await up.text());
    return;
  }
 
  const reader = up.body.getReader();
  const dec = new TextDecoder();
  let pending = '';
  const iv = MODE === 'aggregate'
    ? setInterval(() => { if (pending) { res.write(pending); pending = ''; } }, FLUSH)
    : null;
 
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    let chunk = dec.decode(value, { stream: true });
 
    if (MODE === 'strip')       chunk = chunk.replace(/^:[^\n]*\n\n/gm, ''); // フレームごと除去
    if (MODE === 'strip-loose') chunk = chunk.split('\n').filter((l) => !l.startsWith(':')).join('\n');
 
    if (!chunk.length) continue;
    if (MODE === 'aggregate') pending += chunk;
    else res.write(chunk);
  }
 
  if (iv) clearInterval(iv);
  if (pending) res.write(pending);
  res.end();
}).listen(4002);

// Before — 動くけれど、長考の日に落ちる
export default {
  async fetch(request, env) {
    const upstream = await fetch(env.UPSTREAM, request);
    const body = await upstream.text();          // ← ここで全部待つ
    return new Response(body, {
      status: upstream.status,
      headers: upstream.headers,
    });
  },
};

// After — 本体には触れず、そのまま素通しする
export default {
  async fetch(request, env) {
    const upstream = await fetch(env.UPSTREAM, request);
 
    // 本文を消費しない。ReadableStream をそのまま渡す
    const headers = new Headers(upstream.headers);
    headers.set('Cache-Control', 'no-cache, no-transform');
    headers.set('X-Accel-Buffering', 'no');       // 前段のバッファリングを抑止する
    headers.delete('Content-Length');             // ストリームに長さはない
 
    return new Response(upstream.body, {
      status: upstream.status,
      headers,
    });
  },
};