luulam
e438a03f96
- new open-sse/utils/streamErrorPeek.js: bounded peek of the first bytes of a 200 stream; configured pattern match → 502 so account/combo fallback can run before any byte reaches the client (streaming included). Re-emits RAW bytes so split multi-byte UTF-8 sequences survive the peek (never re-encode decoded text — TextDecoder flush corrupts a lone leading byte to U+FFFD). - chatCore: run the peek after executor.execute when the provider has streamErrorPatterns configured; chat.js passes the settings through. - commandcode executor: same raw-bytes fix in peekForUpstreamError + regression test that fails against the old flush-based re-encode.
100 lines
3.0 KiB
JavaScript
100 lines
3.0 KiB
JavaScript
import { describe, it, expect } from "vitest";
|
|
import { maybeRejectEarlyStreamError } from "../../open-sse/utils/streamErrorPeek.js";
|
|
|
|
const encoder = new TextEncoder();
|
|
const sseResp = (lines) =>
|
|
new Response(
|
|
new ReadableStream({
|
|
start(c) {
|
|
for (const l of lines) c.enqueue(encoder.encode(l));
|
|
c.close();
|
|
},
|
|
}),
|
|
{ status: 200, headers: { "content-type": "text/event-stream" } },
|
|
);
|
|
|
|
describe("maybeRejectEarlyStreamError", () => {
|
|
it("returns 502 when a pattern matches early stream text", async () => {
|
|
const res = await maybeRejectEarlyStreamError(
|
|
sseResp([
|
|
'{"type":"start"}\n{"type":"error","error":{"type":"server_error","message":"Network connection lost."}}\n',
|
|
]),
|
|
["server_error"],
|
|
);
|
|
expect(res.status).toBe(502);
|
|
const body = await res.json();
|
|
expect(body.error.message).toContain("server_error");
|
|
});
|
|
|
|
it("passes the stream through unchanged when nothing matches", async () => {
|
|
const res = await maybeRejectEarlyStreamError(
|
|
sseResp([
|
|
'data: {"choices":[{"delta":{"content":"hello"},"finish_reason":null}]}\n\ndata: [DONE]\n\n',
|
|
]),
|
|
["server_error"],
|
|
);
|
|
expect(res.status).toBe(200);
|
|
const text = await res.text();
|
|
expect(text).toContain("hello");
|
|
expect(text).toContain("[DONE]");
|
|
});
|
|
|
|
it("commits on timeout without hanging", async () => {
|
|
const stalled = new Response(new ReadableStream({ start() {} }), {
|
|
status: 200,
|
|
});
|
|
const res = await maybeRejectEarlyStreamError(stalled, ["x"], {
|
|
timeoutMs: 50,
|
|
});
|
|
expect(res.status).toBe(200);
|
|
await res.body.cancel();
|
|
});
|
|
|
|
it("commits on abort without hanging", async () => {
|
|
const ctrl = new AbortController();
|
|
const stalled = new Response(new ReadableStream({ start() {} }), {
|
|
status: 200,
|
|
});
|
|
setTimeout(() => ctrl.abort(new Error("gone")), 10);
|
|
const res = await maybeRejectEarlyStreamError(stalled, ["x"], {
|
|
signal: ctrl.signal,
|
|
timeoutMs: 2000,
|
|
});
|
|
expect(res.status).toBe(200);
|
|
await res.body.cancel();
|
|
});
|
|
|
|
it("commits (passthrough) when patterns are empty", async () => {
|
|
const res = await maybeRejectEarlyStreamError(
|
|
sseResp(["data: hi\n\n"]),
|
|
[],
|
|
);
|
|
expect(res.status).toBe(200);
|
|
expect(await res.text()).toBe("data: hi\n\n");
|
|
});
|
|
|
|
it("multi-byte UTF-8 split across peek boundary round-trips losslessly", async () => {
|
|
// "café" split mid-é: the peek consumes bytes up to and including 0xC3
|
|
// (the first half of é); the re-emitted stream must contain the RAW bytes
|
|
// (never a re-encoded decoded string — TextDecoder flush would replace the
|
|
// lone 0xC3 with U+FFFD and corrupt the output).
|
|
const bytes = [
|
|
0x64, 0x61, 0x74, 0x61, 0x3a, 0x20, 0x22, 0x63, 0x61, 0x66, 0xc3,
|
|
];
|
|
const rest = new Uint8Array([0xa9, 0x22, 0x0a, 0x0a]);
|
|
const body = new ReadableStream({
|
|
start(c) {
|
|
c.enqueue(new Uint8Array(bytes));
|
|
c.enqueue(rest);
|
|
c.close();
|
|
},
|
|
});
|
|
const res = await maybeRejectEarlyStreamError(
|
|
new Response(body, { status: 200 }),
|
|
["nomatch"],
|
|
{ maxBytes: 11 },
|
|
);
|
|
expect(await res.text()).toBe('data: "café"\n\n');
|
|
});
|
|
});
|