SSE の done は「出力完了」ではなく「副作用の確定」 — ストリーミング API の完了をどう定義するか

リクエスト・レスポンス型の API では、「完了」はレスポンスが返ることと同義です。ステータスコードで成否が表現でき、レスポンスが返った時点で副作用は確定しています。ストリーミングにすると、この 2 つが分解されます。成否はストリームの中身になり、副作用の確定は別途どこかで宣言しなければならなくなる。

本記事の主題はこの分解に対する答えです。SSE の done イベントを「モデルの出力が終わった」という合図ではなく、このストリームが引き起こす副作用がすべて確定したという合図として設計しました。言い換えると、done を API 間の read-after-write 契約の表明として使っています。

題材は前回と同じ、B2B の業務 SaaS のバックエンドで動く AI エージェント API です。Azure Functions v4 (Node.js 22 / TypeScript) の上で 14 のエンドポイントが動き、そのうち 3 つが SSE でチャットの応答を流します。モデルは Azure OpenAI の gpt-5.4-mini、API バージョンは 2025-04-01-preview です。

SSE プロトコルそのものの解説と、Azure Functions v4 プログラミングモデルの入門は書きません。どちらも一次情報が十分に読みやすい形で揃っています。

前提: まず経路を通す

done の設計に入る前に、前提を短くまとめておきます。SSE は「作る側」のコードは短く済みますが、そのストリームがクライアントへ届くまでの経路に詰まりどころがあります。実際、本番で「数十秒待たされたあとに全チャンクが一度に届く」という症状が出て、直した箇所は 3 つありました。厄介なのは、どれか 1 つでも欠けていると症状がまったく同じという点です。

1 つ目と 2 つ目はセットで踏みました。設定は専用ファイル 1 つに置いています。

// src/index.ts（ファイル全体）
import { app } from "@azure/functions";

// HTTP レスポンスストリーミング機能を有効化
// SSE (text/event-stream) で ReadableStream を逐次クライアントへ送出するために必須
// 参照: https://aka.ms/AzFuncNodeHttpStreams
app.setup({ enableHttpStream: true });

公式のマイグレーションガイドによると、v4 ではコード構造を自由に決められ、ランタイムが読み込むファイルは package.json の main フィールド（glob パターン可）で指定します。裏を返せば、main に含まれないファイルは存在しないのと同じで、app.setup() を書いても何のエラーも出ないまま無効になります。v4 で設定が効かないときは、コードより先に main とビルド出力の実際のパスを突き合わせるのが早いと思います。

3 つ目はレスポンスヘッダでの宣言です。no-transform は RFC 9111 で「キャッシュを実装しているかどうかに関わらず、中間者はコンテンツを変換してはならない (MUST NOT)」と定義されたディレクティブ、X-Accel-Buffering は nginx が解釈する慣習的なヘッダで、レスポンス側からバッファリングを無効化できます。この 2 つは効いているかどうかがローカルでは分からない種類の対処です。中間プロキシは開発環境には存在しないので、ローカルでストリーミングできることは何の保証にもなりません。

経路まわりであと 3 点だけ。

• タイムアウトは下流（発生源に近い側）から上流（クライアントに近い側）に向かって余裕を持たせる。 どこか 1 層でも下流より短いと、正常に流れているストリームがその層で切られます。nginx の proxy_read_timeout の既定値は 60 秒で、しかも「連続する 2 回の読み取りの間隔」への制限なので、最初のトークンが出るまで待たされる LLM では何も設定しないとプロキシが先に諦めます
• Azure Functions の HTTP トリガーには 230 秒の上限がある。 functionTimeout の設定とは別に文書化されており、Azure Load Balancer の既定アイドルタイムアウト由来なので伸ばせません。このシステムのチャットが 180 秒なのは、この壁より内側に収めるためです
• 受信側は EventSource ではなく fetch + ReadableStream。 EventSource のコンストラクタが受け取るのは URL と withCredentials だけで、リクエストボディを渡す口がないからです。チャットの入力はボディで送るので、SSE の受信側は自作になります

流す中身: エラーは HTTP ステータスで返せない

経路が通ったら、次は「何を流すか」です。ここに done の設計を要請してくる制約が 2 つあります。

1 つ目は改行の扱いです。SSE のフレームは改行で区切られ、空行が来た時点でイベントが 1 つ確定します。つまり \n はプロトコルの構文要素で、これは改行を平気で吐く LLM の出力と相性が最悪です。デルタをそのまま data: に載せるとフレームが割れるので、JSON 文字列リテラルとして書き出します。

// src/shared/responses.ts  sendMessageStream()
if (event.type === "response.output_text.delta") {
  fullText.push(event.delta ?? "");
  yield `data: ${JSON.stringify(event.delta ?? "")}\n\n`;
}

JSON.stringify("a\nb") は改行を含まない文字列リテラルになり、クライアントは JSON.parse で元のデルタに戻します。

2 つ目が本質的な制約です。ヘッダを送出したあとに問題が起きても、HTTP ステータスはもう変えられません。ストリームの途中で上流が落ちても、返した 200 はそのままです。したがってエラーはストリームの中身として表現するしかありません。このシステムのイベントは 5 種類です。

meta が先頭にあるのは、クライアントが thread_id を持たずにリクエストしてくる場合があり、その場合はサーバ側が新規に発行するからです。ヘッダで返せない値を、ストリームの先頭で返しているという形です。同じように、リクエスト・レスポンス型ならステータスコードとレスポンスボディが担っていたものが、すべてストリームの中のイベントへ移っていきます。では「完了」はどのイベントが、何を保証する形で担うのか。それが次章です。

done は「出力完了」ではなく「副作用の確定」

ここが本記事の核です。

このシステムのフロントは、done を受け取った直後に別の API を叩ける状態になります。たとえば問い合わせチャットでは、会話を終えたユーザーが「改善要望を送信」を押すと、別のエンドポイントが会話ログを Table Storage から読んで内容を構造化します。

ここに順序の問題があります。会話ログを書いているのはストリームを流している側です。もし done を先に送ってしまうと、直前の発話がまだ書かれていないログに対して後段の処理が走ります。ユーザーから見ると「今しゃべった内容が反映されていない」という形で出ます。しかも毎回ではなく、書き込みが遅れたときだけ起きます。

なので、done の前で書き込みを待ちます。

// src/shared/responses.ts  sendMessageStream() の末尾
const outputText = stripCitations(fullText.join(""));

if (responseId) {
  // done 直後に inquiry/complete 等が会話ログを読むため、書き込み完了を待ってから done を送出する
  //（失敗しても done は返す = Table障害でもAIレスポンスは返す方針）
  await Promise.allSettled([
    upsertSession(params.threadId, responseId),
    logMessage(params.threadId, "user", params.userMessage, params.endpoint),
    logMessage(params.threadId, "assistant", outputText, params.endpoint),
  ]);
}

const doneData = JSON.stringify({
  status: "success",
  thread_id: params.threadId,
  response_id: responseId ?? "",
});
yield `event: done\ndata: ${doneData}\n\n`;

Promise.all ではなく Promise.allSettled なのは意図的です。Promise.all だと 1 つでも失敗した時点で reject し、done が送られないまま例外側へ抜けます。Table Storage が落ちているときに AI の応答まで届かなくなるのは、このシステムでは割に合いません。会話ログの欠落は後段の品質低下として現れますが、応答が返らないのは即座の障害です。だから「待つが、失敗しても進む」を選んでいます。

言い換えると、done に持たせている意味は次のようになります。

• 「モデルの出力が終わった」ではない — それは最後のデルタを送った時点で分かる
• 「このストリームが引き起こす副作用が、成否はともかく確定した」 — だから次の API を叩いてよい

これは要するに、SSE の完了イベントを API 間の read-after-write 契約の表明として使っているということです。ストリーミング API とその結果を読む API が別々にある構成では、この境界をどこかで宣言しないと、クライアント側が「少し待ってから叩く」という類の回避策を持つことになります。完了イベントはその宣言を置く場所として使えます。

そう決めたぶん、正直に書いておくべきことが 2 つあります。

1 つは、done が成功の合図ではないことです。ストリームの開始そのものに失敗した経路でも、event: error の直後に event: done（status: "failed"）を送っています。

// src/shared/responses.ts  sendMessageStream()
} catch (e) {
  // signal による中断は呼び出し側（タイムアウト経路）で error/done を送出済みなので静かに終了
  if (signal?.aborted) return;
  console.error(`ストリーム開始失敗: thread_id=${params.threadId}, error=${e}`);
  const errorData = JSON.stringify({
    status: "failed",
    thread_id: params.threadId,
    error: "ストリーム開始に失敗しました。",
  });
  yield `event: error\ndata: ${errorData}\n\n`;
  yield `event: done\ndata: ${JSON.stringify({ status: "failed", thread_id: params.threadId, response_id: "" })}\n\n`;
  return;
}

done は終端の契約です。クライアントは done を待てば必ず終わり、成否は status で判定する。「成功したときだけ届くイベント」にすると、クライアントは失敗を検出するためにタイムアウトを持つことになります。それは経路上にもう 1 つタイムアウトを増やすのと同じで、前章の「下流から上流へ」の管理対象を自分で増やす行為です。

もう 1 つは、このバリアが経路ごとに置かれていることです。ツールを実行した場合はモデルへの再問い合わせが入るので処理が分岐し、done の直前で永続化を待つ同じ形のブロックが 2 箇所に存在します。共通化されていないので、3 つ目の経路を足す人は同じことを忘れずに書く必要があります。「バリアは経路に属する」という当たり前の帰結ですが、コードの見た目としては重複です。

終わらせ方: 中断を指示した側が、通知に責任を持つ

done を終端の契約にすると、今度は契約を果たせない経路、つまりタイムアウトの終わらせ方を設計することになります。

最初の実装は素朴でした。setTimeout で event: error を流して controller.close() するだけ。これで約 3 ヶ月動きましたが、タイムアウトが実際に発火する状況になって 3 つのバグが表面化しました。いずれも「タイムアウトが発火したあとも周りの世界が動き続ける」ことが原因です。

1. 閉じたストリームへの書き込み。 close() しても for await で回している生成器は止まらず、次のチャンクの enqueue() が TypeError になる。finally の close() で二重 close も起きる
2. 上流の Azure OpenAI が走り続ける。 クライアントへの経路を閉じても上流は止まらないので、モデルは最後まで生成を続け、その分のトークンは課金される
3. エラーの二重送出。 タイムアウト処理が event: error を送ったあと、中断で生成器側が例外を投げ、catch がもう一度 event: error を送る

修正後がこうなります。対処は 3 つとも「状態を 1 つ持つ」だけです。

// src/functions/chatFacilitator.ts
const sseStream = new ReadableStream({
  async start(controller) {
    const encoder = new TextEncoder();
    const abortCtrl = new AbortController();
    let closed = false;
    const safeEnqueue = (chunk) => {
      if (closed) return;
      controller.enqueue(encoder.encode(chunk));
    };
    const safeClose = () => {
      if (closed) return;
      closed = true;
      controller.close();
    };
    // SSEストリームタイムアウト（3分）: 上流 Azure OpenAI も abort、controller は idempotent close
    const timeoutId = setTimeout(() => {
      abortCtrl.abort();
      const errorData = JSON.stringify({ status: "failed", error: "ストリームがタイムアウトしました。" });
      safeEnqueue(`event: error\ndata: ${errorData}\n\n`);
      safeClose();
    }, RUN_TIMEOUT_SEC * 1000);
    try {
      // ...（中略：meta 送出、ナレッジ検索、generator の生成）
      for await (const chunk of generator) {
        safeEnqueue(chunk);
      }
    } catch (e) {
      if (!abortCtrl.signal.aborted) {
        context.error(`SSEストリームエラー: ${e}`);
        const errorData = JSON.stringify({ status: "failed", error: "ストリーム処理中にエラーが発生しました。" });
        safeEnqueue(`event: error\ndata: ${errorData}\n\n`);
      }
    } finally {
      clearTimeout(timeoutId);
      safeClose();
    }
  },
});

closed フラグで enqueue と close を冪等にし、AbortController の signal を上流へ渡し、その signal.aborted を catch のガードに使う。そして signal を受け取った側は、こう応じます。

// src/shared/responses.ts  sendMessageStream()
} catch (e) {
  // signal 経由の中断は呼び出し側で error/done を送出済み
  if (signal?.aborted) return;
  throw e;
}

自分が中断させられた場合は、何も言わずに return する。 エラーイベントは中断を指示した側が既に送っているからです。ここで丁寧に throw すると、呼び出し側の catch がまた event: error を送ることになります。

この分担は、AbortSignal を経路に通したときに毎回考えることになる部類の話だと思います。中断は「エラー」ではなく「こちらが決めたこと」なので、中断を指示した側がユーザーへの通知に責任を持ち、指示された側は黙って止まる。逆にすると通知が重複するか、あるいは誰も通知しないかのどちらかになります。

3 ヶ月表面化しなかった理由も一言で書けます。この 3 つはいずれもタイムアウトが実際に発火しないと起きず、通常の応答は数十秒で終わるので開発中はまず踏みません。タイムアウト処理を書くというのは、「めったに通らない経路を新設する」ことでもある、ということです。

まとめ — ストリーミング API の設計は、完了の契約を決めること

リクエスト・レスポンス型の API では、完了を設計する必要はありませんでした。レスポンスが返れば成否が分かり、副作用も確定している — この 3 つを HTTP が 1 つに束ねて保証していたからです。ストリーミングはこの束を解きます。解けた保証をどう束ね直すかが、ストリーミング API の設計だ、というのが本記事の結論です。

このシステムの答えは 1 行で書けます。done とは「このストリームが引き起こす副作用はすべて確定した。だから次の API を叩いてよい」という契約である。 モデルの出力の終わりは最後のデルタで分かるので、完了イベントにその意味を持たせるのは保証の重複です。完了イベントには、それにしか宣言できないこと — read-after-write の成立 — を宣言させます。

この一文を決めると、本記事で書いた実装は次のように従属的に決まります。

逆に、done を「モデルの出力完了」と定義しても API は動きます。ただしそのとき「いつ次を叩いてよいか」という契約はどこにも書かれていないので、クライアントは「少し待ってから叩く」という推測でそれを補うことになります。ストリーミング API を作るとき最初に決めるべきは、イベントの種類でもタイムアウトの値でもなく、完了イベントが何を保証するかの一文です。

次回は、このシステムの Structured Outputs について書きます。minimum も multipleOf も minItems も strict モードでは強制されないのに、スキーマに全部書き続けている理由と、バリデーションの失敗を「モデルが読んで直せる文章」として返す設計の話です。

検証環境は Node.js 22 / Azure Functions v4 / Azure OpenAI api-version: 2025-04-01-preview / gpt-5.4-mini、2026 年 8 月時点のものです。enableHttpStream を含む Node.js v4 プログラミングモデルの仕様は変わりえるので、最新の情報は公式ドキュメントで確認してください。

参考リンク

• Node.js developer reference for Azure Functions (Microsoft Learn) — HTTP ストリームがオプトイン機能である旨、app.setup({ enableHttpStream: true }) を main フィールドに含まれるファイルへ置く必要がある旨
• Migrate to v4 of the Node.js model for Azure Functions (Microsoft Learn) — main フィールドに単一ファイルまたは glob パターンを指定してファイル構造を定義できる旨（dist/src/{index.js,functions/*.js} の例を含む）
• Function app timeout duration (Microsoft Learn) — プラン別の functionTimeout と、HTTP トリガーには設定に関わらず 230 秒の上限がある（Azure Load Balancer の既定アイドルタイムアウト由来）という注記
• Server-sent events (WHATWG HTML Standard) — イベントストリームの解析規則。空行でイベントが dispatch されること、data フィールドは値の後に U+000A LINE FEED を追加してバッファに積むこと
• EventSource (MDN) — コンストラクタが受け取るのは URL と withCredentials のみで、リクエストボディを渡す手段がないこと
• ngx_http_proxy_module (nginx) — X-Accel-Buffering レスポンスヘッダに yes / no を渡すことでバッファリングを制御できる旨、および proxy_read_timeout が「連続する 2 回の読み取りの間隔」に対する制限で既定 60 秒であること
• RFC 9111 — HTTP Caching, §5.2.2.6 no-transform — no-transform レスポンスディレクティブが、キャッシュを実装しているかに関わらず中間者にコンテンツの変換を禁じる (MUST NOT) 旨