Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 32 additions & 7 deletions open-sse/services/autoCombo/__tests__/chaosEngine.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -85,18 +85,29 @@ describe("runChaosPanel", () => {
});

describe("serializeChaosPart", () => {
it("emits a comment + omni-chaos-part event envelope", () => {
it("emits a comment + omni-chaos-part event envelope when custom event is requested", () => {
const part: ChaosPart = { model: "a/gpt", index: 0, ok: true, text: "hi" };
const s = serializeChaosPart(part, false);
const s = serializeChaosPart(part, false, true);
expect(s).toContain("event: omni-chaos-part");
expect(s).toContain('"type":"omni-chaos-part"');
expect(s).toContain('"model":"a/gpt"');
expect(s).toContain(": chaos 0 ok a/gpt");
});

it("emits ONLY the SSE comment (no event/data) by default for OpenAI-compatible clients", () => {
const part: ChaosPart = { model: "a/gpt", index: 0, ok: true, text: "hi" };
const s = serializeChaosPart(part, false);
// comment line kept (ignored by every SSE parser by spec)
expect(s).toContain(": chaos 0 ok a/gpt");
// NO custom event/data — those break openai-node / @ai-sdk validators
expect(s).not.toContain("event: omni-chaos-part");
expect(s).not.toContain('"type":"omni-chaos-part"');
expect(s).not.toMatch(/^data:/m);
});
});

describe("handleChaosChat", () => {
it("emits broadcast events + final OpenAI chunk", async () => {
it("emits ONLY SSE comments (no custom event) by default + final OpenAI chunk", async () => {
const handle = fakeHandle(async (model) => textResponse(`ans-${model}`));
const res = await handleChaosChat({
body: { messages: [] },
Expand All @@ -106,13 +117,27 @@ describe("handleChaosChat", () => {
expect(res.headers.get("X-OmniRoute-Chaos")).toBe("true");
expect(res.headers.get("X-OmniRoute-Chaos-Panel")).toBe("2");
const body = await res.text();
// each model gets a broadcast event
expect(body.match(/event: omni-chaos-part/g)?.length).toBe(2);
// NO custom event by default — OpenAI-compatible parsers choke on it
expect(body.match(/event: omni-chaos-part/g)?.length ?? 0).toBe(0);
expect(body.match(/^: chaos /gm)?.length ?? 0).toBe(2);
// final canonical chunk carries the primary answer
expect(body).toContain("ans-b/opus");
expect(body).toContain("[DONE]");
});

it("emits omni-chaos-part events when stream_options.include_chaos_parts is set", async () => {
const handle = fakeHandle(async (model) => textResponse(`ans-${model}`));
const res = await handleChaosChat({
body: { messages: [], stream_options: { include_chaos_parts: true } },
models: ["a/gpt", "b/opus"],
handleSingleModel: handle,
});
const body = await res.text();
expect(body.match(/event: omni-chaos-part/g)?.length).toBe(2);
expect(body).toContain("ans-b/opus");
expect(body).toContain("[DONE]");
});

it("degrades to a direct call when only one model", async () => {
const handle = fakeHandle(async () => textResponse("solo"));
const res = await handleChaosChat({
Expand All @@ -138,8 +163,8 @@ describe("handleChaosChat", () => {
// client learns via the error final chunk rather than a bare 503.
expect(res.status).toBe(200);
const body = await res.text();
// each model gets a broadcast fail event
expect(body.match(/event: omni-chaos-part/g)?.length).toBe(2);
// NO custom events by default (comments only), error conveyed via final chunk
expect(body.match(/event: omni-chaos-part/g)?.length ?? 0).toBe(0);
expect(body).toContain("All chaos panel models failed");
expect(body).toContain("[DONE]");
});
Expand Down
50 changes: 36 additions & 14 deletions open-sse/services/autoCombo/chaosEngine.ts
Original file line number Diff line number Diff line change
Expand Up @@ -53,16 +53,28 @@ export type ChaosPart = {
};

/**
* Build the SSE comment/event wrapper for one chaos panel part.
* We emit a custom event name `omni-chaos-part` so a protocol-aware IDE can
* split it out; non-aware clients reading OpenAI-style SSE will simply ignore
* the unknown event and use the final `data:` chunk below.
* Build the SSE wrapper for one chaos panel part.
*
* By DEFAULT only an SSE comment (`: chaos ...`) is emitted — comments are
* ignored by every SSE parser per spec, so OpenAI-compatible clients
* (openai-node, @ai-sdk/openai-compatible, …) never see a non-`choices`
* `data:` payload and their schema validation cannot fail with an
* `invalid_union` error.
*
* When `emitCustomEvent` is true (opt-in via
* `stream_options.include_chaos_parts`), the custom event name
* `omni-chaos-part` + metadata `data:` block is also emitted so a
* protocol-aware IDE can split panels out.
*
* The part's text is NOT included in the metadata event — it arrives in the
* final `data:` chunk for the primary model. This keeps each broadcast event
* small (metadata-only) so SSE buffering stays predictable.
*/
export function serializeChaosPart(part: ChaosPart, isFinal: boolean): string {
export function serializeChaosPart(
part: ChaosPart,
isFinal: boolean,
emitCustomEvent = false
): string {
const meta = {
type: "omni-chaos-part",
model: part.model,
Expand All @@ -71,11 +83,11 @@ export function serializeChaosPart(part: ChaosPart, isFinal: boolean): string {
final: isFinal,
...(part.error ? { error: part.error } : {}),
};
return (
`: chaos ${part.index} ${part.ok ? "ok" : "fail"} ${part.model}\n` +
`event: omni-chaos-part\n` +
`data: ${JSON.stringify(meta)}\n\n`
);
const comment = `: chaos ${part.index} ${part.ok ? "ok" : "fail"} ${part.model}\n`;
if (!emitCustomEvent) {
return comment + "\n";
}
return comment + `event: omni-chaos-part\n` + `data: ${JSON.stringify(meta)}\n\n`;
}

/**
Expand Down Expand Up @@ -330,9 +342,11 @@ function concatSseText(sse: string): string {
* `config.chaos.enabled` flag is set (the `auto/chaos` virtual combo).
*
* Returns a single Response whose body is an SSE stream:
* - one `omni-chaos-part` event per panel model, enqueued PROGRESSIVELY as
* each model lands (so the client starts receiving answers immediately,
* without waiting for the whole panel to finish)
* - one SSE comment (`: chaos N ...`) per panel model, enqueued
* PROGRESSIVELY as each model lands (comments are ignored by every SSE
* parser, so OpenAI-compatible clients see only the final chunk)
* - when `stream_options.include_chaos_parts: true` is set, the per-panel
* `omni-chaos-part` custom event is emitted instead of the bare comment
* - a final `data:` OpenAI-style chunk carrying the primary model's answer
* (so non-aware clients / IDEs still get a usable completion)
* - a terminating `data: [DONE]`
Expand All @@ -354,6 +368,14 @@ export async function handleChaosChat(opts: {
const panel = Array.isArray(models) ? models.filter(Boolean) : [];
const hardTimeout = tuning?.panelHardTimeoutMs ?? CHAOS_DEFAULTS.panelHardTimeoutMs;
const minPanel = tuning?.minPanel ?? CHAOS_DEFAULTS.minPanel;
// Opt-in gate: only protocol-aware clients request the custom event. OpenAI
// SDK validators choke on any `data:` payload without `choices`/`error`, so
// the default MUST be comment-only output.
const streamOptions = (body as Record<string, unknown> | null | undefined)?.stream_options;
const emitCustomEvent =
typeof streamOptions === "object" &&
streamOptions !== null &&
(streamOptions as Record<string, unknown>).include_chaos_parts === true;
if (panel.length === 0) {
return errorResponse(400, "Chaos combo has no models");
}
Expand Down Expand Up @@ -396,7 +418,7 @@ export async function handleChaosChat(opts: {
hardTimeout,
log,
onResult: async (part) => {
await safeEnqueue(serializeChaosPart(part, false));
await safeEnqueue(serializeChaosPart(part, false, emitCustomEvent));
},
});
});
Expand Down
Loading