API SSE exception path closes without the OpenAI terminal marker

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));