API SSE exception path closes without the OpenAI terminal marker

[MicroMilo]

Summary

When the Gateway throws, the route emits channel_submit_failed and closes without a finish chunk or data: [DONE]. The normal baseline emits both markers.

Expected behavior

An SSE error path should emit a terminal finish chunk and data: [DONE] so clients can deterministically complete a failed stream.

Actual behavior

When the Gateway throws, the route emits channel_submit_failed and closes without a finish chunk or data: [DONE]. The normal baseline emits both markers.

Impact

Clients that wait for the OpenAI terminal marker can leave failed streams pending or misclassify their final state.

Reproduction

Open an API streaming completion and make the Gateway throw after the request has been accepted. Compare the error stream with a normal completion. The error stream should emit a terminal finish chunk and data: [DONE]; the observed result is channel_submit_failed followed by connection close without either terminal marker.

Minimal reproduction script

From the repository root, save this as repro_api_stream_error_terminal.mts and run:

pnpm install --frozen-lockfile
pnpm exec tsx repro_api_stream_error_terminal.mts

import { ApiServerChannel } from "./src/adapters/channel/api-server/ApiServerChannel.js";
import type { Gateway, GatewayEvent, GatewaySubmitTurnInput } from "./src/gateway/index.js";
import { Readable } from "node:stream";

const calls: Array<{ sessionKey: string; channelKey: string; message: string }> = [];

const fakeGateway = {
  submitTurn(input: GatewaySubmitTurnInput): AsyncIterable<GatewayEvent> {
    calls.push({ sessionKey: input.sessionKey, channelKey: input.channelKey, message: input.message });
    return input.message === "trigger-failure" ? failingStream() : successfulStream();
  },
} as Pick<Gateway, "submitTurn"> as Gateway;

class CaptureResponse {
  headers = new Map<string, string>();
  statusCode = 0;
  chunks: string[] = [];
  ended = false;

  setHeader(name: string, value: string): void {
    this.headers.set(name.toLowerCase(), value);
  }

  flushHeaders(): void {}

  write(chunk: string): boolean {
    this.chunks.push(String(chunk));
    return true;
  }

  end(chunk?: string): void {
    if (chunk) this.chunks.push(String(chunk));
    this.ended = true;
  }
}

async function* failingStream(): AsyncIterable<GatewayEvent> {
  yield { type: "assistant_text_delta", text: "partial" };
  throw new Error("sanitized gateway iterator failure");
}

async function* successfulStream(): AsyncIterable<GatewayEvent> {
  yield { type: "assistant_text_delta", text: "complete" };
}

const channel = new ApiServerChannel({ modelName: "witness-model" });
const channelInternals = channel as unknown as { gateway?: Gateway; handleRequest(req: unknown, res: CaptureResponse): Promise<void> };
channelInternals.gateway = fakeGateway;

async function send(message: string): Promise<{
  status: number;
  contentType: string | null;
  sessionHeader: string | null;
  body: string;
  ended: boolean;
}> {
  const request = Readable.from([Buffer.from(JSON.stringify({
    model: "witness-model",
    stream: true,
    messages: [{ role: "user", content: message }],
  }))]) as Readable & { method?: string; url?: string; headers: Record<string, string> };
  request.method = "POST";
  request.url = "/v1/chat/completions";
  request.headers = {
    host: "fixture.invalid",
    "content-type": "application/json",
    "x-hermes-session-id": "witness-session",
  };
  const response = new CaptureResponse();
  await channelInternals.handleRequest(request, response);
  return {
    status: response.statusCode,
    contentType: response.headers.get("content-type") ?? null,
    sessionHeader: response.headers.get("x-hermes-session-id") ?? null,
    body: response.chunks.join(""),
    ended: response.ended,
  };
}

const failure = await send("trigger-failure");
const normal = await send("normal-completion");

function dataFrames(body: string): string[] {
  return body
    .split("\n")
    .filter((line) => line.startsWith("data: "))
    .map((line) => line.slice("data: ".length));
}

const failureFrames = dataFrames(failure.body);
const normalFrames = dataFrames(normal.body);
const failureError = failureFrames
  .map((frame) => {
    try {
      return JSON.parse(frame) as Record<string, unknown>;
    } catch {
      return null;
    }
  })
  .find((frame) => frame?.event === "channel_submit_failed");

const artifact = {
  fixture: "ApiServerChannel.handleRequest browserless boundary with fake Gateway async iterable",
  input: {
    method: "POST",
    path: "/v1/chat/completions",
    stream: true,
    sessionHeader: "witness-session",
    failureMessage: "trigger-failure",
    baselineMessage: "normal-completion",
  },
  expected: {
    failure: "A streaming error may carry a structured error event, but the OpenAI SSE response must still expose the terminal data: [DONE] marker so a consumer can finish parsing deterministically.",
    baseline: "A normally completed stream exposes a finish chunk followed by data: [DONE].",
  },
  actual: {
    failure: {
      status: failure.status,
      contentType: failure.contentType,
      sessionHeader: failure.sessionHeader,
      responseEnded: failure.ended,
      body: failure.body,
      frames: failureFrames,
      hasDoneMarker: failure.body.includes("data: [DONE]"),
      errorEvent: failureError,
    },
    baseline: {
      status: normal.status,
      contentType: normal.contentType,
      sessionHeader: normal.sessionHeader,
      responseEnded: normal.ended,
      body: normal.body,
      frames: normalFrames,
      hasDoneMarker: normal.body.includes("data: [DONE]"),
    },
    gatewayCalls: calls,
  },
  source: {
    catchPath: "src/adapters/channel/api-server/ApiServerChannel.ts:333-349",
    normalTerminalPath: "src/adapters/channel/api-server/ApiServerChannel.ts:505-514",
  },
};

console.log(JSON.stringify(artifact, null, 2));

Relevant source locations

• src/adapters/channel/api-server/ApiServerChannel.ts:317-349
• src/adapters/channel/api-server/ApiServerChannel.ts:505-514

Suggested direction

Make the external-input path establish one durable, identity-bound state/receipt before returning success; propagate explicit terminal outcomes to every channel and client; and add a regression test for the reproduced boundary.

This report is about functional behavior, not security. The reproduction uses deterministic in-memory or isolated fixtures and contains no credentials or private data.