const abortReason = new Error("live anthropic stream abort");

let requestBody = "";

let requestBodyPromise: Promise<string> | undefined;

let connectionClosed = false;

let resolveConnectionClosed: (() => void) | undefined;

const connectionClosedPromise = new Promise<void>((resolve) => {

resolveConnectionClosed = resolve;

let resolveResponseStarted: (() => void) | undefined;

const responseStartedPromise = new Promise<void>((resolve) => {

resolveResponseStarted = resolve;

});

const server = http.createServer((request, response) => {

const markConnectionClosed = () => {

connectionClosed = true;

resolveConnectionClosed?.();

};

request.on("aborted", markConnectionClosed);

request.on("close", markConnectionClosed);

response.on("close", () => {

markConnectionClosed();

});

requestBodyPromise = readRequestBody(request).then((body) => {

requestBody = body;

response.writeHead(200, {

response.write(

'data: {"type":"message_start","message":{"id":"msg_live","usage":{"input_tokens":1,"output_tokens":0}}}\n\n',

);

resolveResponseStarted?.();

return body;

});

});

const port = await waitForServerListening(server);

try {

setTimeout(() => controller.abort(abortReason), 50);

const model: AnthropicMessagesModel = {

id: "claude-sonnet-4-6",

name: "Claude Sonnet 4.6",

),

);

const responseStarted = await Promise.race([

responseStartedPromise.then(() => true),

delay(1_000, false),

]);

expect(responseStarted).toBe(true);

controller.abort(abortReason);

const timedOut = Symbol("timed out");

const result = await Promise.race([stream.result(), delay(1_000, timedOut)]);

if (result === timedOut) {

throw new Error("Anthropic live SSE stream did not abort within 1000ms");

}

const observedConnectionClose = await Promise.race([

connectionClosedPromise.then(() => true),

delay(2_000, false),

]);

expect(result.stopReason).toBe("aborted");

expect(result.errorMessage).toBe("live anthropic stream abort");

expect(observedConnectionClose || connectionClosed).toBe(true);

const capturedRequestBody = requestBodyPromise

? await Promise.race([requestBodyPromise, delay(500, requestBody)])

: requestBody;

});

}

} finally {

if (!controller.signal.aborted) {

controller.abort(abortReason);

}

await closeServer(server);

}

}, 10_000);