mirror of
https://github.com/diegosouzapw/OmniRoute.git
synced 2026-07-26 09:52:11 +03:00
feat(chaos+ponytail): parallel chaos-mode dispatch + ponytail output … (#7781)
* feat(chaos+ponytail): parallel chaos-mode dispatch + ponytail output style (rebased on v3.8.49)
- Chaos mode: new auto/chaos variant fans the prompt out to the top-N
stable models in parallel and returns a single merged SSE stream.
- Progressive streaming: each panel model's answer is enqueued as it
lands (omni-chaos-part event), instead of awaiting the whole panel.
- withTimeout now aborts the underlying request (modelAbortSignal) on
timeout so the connection is released, not leaked.
- concatSseText parses both OpenAI and Anthropic SSE wire formats.
- autoPrefix/modePacks add the chaos-mode weight pack; virtualFactory
materializes auto/chaos with fusion strategy + chaos config flag.
- Ponytail (lazy-senior-dev mode) integrated into the existing
OUTPUT_STYLE_CATALOG registry (id 'ponytail') so it rides the production
output-style injector, instead of a bespoke duplicate module. Dev-only
scripts and the duplicate ponytail/ module are removed.
- Tests: chaosEngine/chaosVirtualCombo cover panel dispatch, progressive
broadcast, timeout abort, and Anthropic parsing; autoCombo pack count
updated to 6.
Rebased onto release/v3.8.49 (no provider-registry or validation changes —
those are split out per review).
* fix(combo): align chaos dispatch callback with ChaosTarget signature
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
* fix(combo): extract chaos dispatch to chaosEngine.ts to fix combo.ts file-size gate
combo.ts's chaos-detection block pushed the frozen file-size gate over its
cap (3387 lines) after merging release/v3.8.49. Extracted the config
detection + model-list building into dispatchChaosFromCombo() in
chaosEngine.ts (the module this PR already introduces), mirroring the
existing fusion-strategy short-circuit pattern. combo.ts now does a 9-line
early-return; no behavior change.
combo.ts: 3406 -> 3386 lines (frozen cap 3387).
handleComboChat complexity/cognitive/lines all decrease as a side effect
(120->118 / 153->152 / 1541->1524) since the block moved out.
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
---------
Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
Co-authored-by: Moseyuh333 <Moseyuh333@users.noreply.github.com>
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>
This commit is contained in:
Moseyuh333
@@ -195,10 +195,12 @@ describe("Self-Healing", () => {
|
||||
});
|
||||
|
||||
describe("Mode Packs", () => {
|
||||
it("should have 5 mode packs", () => {
|
||||
it("should have 6 mode packs", () => {
|
||||
// #4235 Phase B added reliability-first (for the `:reliable` tier).
|
||||
expect(getModePackNames()).toHaveLength(5);
|
||||
// chaos-mode added for the `auto/chaos` parallel-dispatch variant.
|
||||
expect(getModePackNames()).toHaveLength(6);
|
||||
expect(getModePackNames()).toContain("reliability-first");
|
||||
expect(getModePackNames()).toContain("chaos-mode");
|
||||
});
|
||||
|
||||
it("reliability-first should prioritize health and stability", () => {
|
||||
|
||||
190
open-sse/services/autoCombo/__tests__/chaosEngine.test.ts
Normal file
190
open-sse/services/autoCombo/__tests__/chaosEngine.test.ts
Normal file
@@ -0,0 +1,190 @@
|
||||
/**
|
||||
* Tests for the chaos engine — parallel multi-model dispatch for `auto/chaos`.
|
||||
*
|
||||
* Verifies:
|
||||
* - runChaosPanel fans out to all models in parallel and collects each answer
|
||||
* - a hung model is bounded by panelHardTimeoutMs (never stalls forever)
|
||||
* - a failed model is reported as ok:false but does not break the panel
|
||||
* - handleChaosChat emits the `omni-chaos-part` broadcast events + a final
|
||||
* OpenAI-style chunk carrying the primary model's answer
|
||||
* - single-model chaos degrades to a direct handleSingleModel call
|
||||
*/
|
||||
|
||||
import { describe, it, expect, vi } from "vitest";
|
||||
import { runChaosPanel, handleChaosChat, serializeChaosPart, type ChaosPart } from "../chaosEngine";
|
||||
|
||||
function textResponse(text: string): Response {
|
||||
return new Response(JSON.stringify({ choices: [{ message: { content: text } }] }), {
|
||||
headers: { "content-type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
function fakeHandle(impl: (model: string) => Promise<Response>) {
|
||||
return vi.fn(async (_body: Record<string, unknown>, model: string) => impl(model));
|
||||
}
|
||||
|
||||
describe("runChaosPanel", () => {
|
||||
it("fans out to all models in parallel and collects each answer", async () => {
|
||||
const order: string[] = [];
|
||||
const handle = fakeHandle(async (model) => {
|
||||
order.push(model);
|
||||
return textResponse(`answer-from-${model}`);
|
||||
});
|
||||
const { parts, primary } = await runChaosPanel({
|
||||
body: { messages: [] },
|
||||
models: ["a/gpt", "b/opus", "c/sonnet"],
|
||||
handleSingleModel: handle,
|
||||
});
|
||||
expect(handle).toHaveBeenCalledTimes(3);
|
||||
expect(parts).toHaveLength(3);
|
||||
expect(parts.every((p) => p.ok)).toBe(true);
|
||||
expect(parts.map((p) => p.text)).toEqual([
|
||||
"answer-from-a/gpt",
|
||||
"answer-from-b/opus",
|
||||
"answer-from-c/sonnet",
|
||||
]);
|
||||
// primary = last successful (top-scored stable model by construction)
|
||||
expect(primary?.model).toBe("c/sonnet");
|
||||
});
|
||||
|
||||
it("bounds a hung model via panelHardTimeoutMs", async () => {
|
||||
const handle = fakeHandle(async (model) => {
|
||||
if (model === "slow/x") {
|
||||
await new Promise((r) => setTimeout(r, 5000));
|
||||
return textResponse("late");
|
||||
}
|
||||
return textResponse("fast");
|
||||
});
|
||||
const start = Date.now();
|
||||
const { parts } = await runChaosPanel({
|
||||
body: {},
|
||||
models: ["fast/y", "slow/x"],
|
||||
handleSingleModel: handle,
|
||||
tuning: { panelHardTimeoutMs: 200 },
|
||||
});
|
||||
const elapsed = Date.now() - start;
|
||||
expect(elapsed).toBeLessThan(2000); // did not wait the full 5s
|
||||
const slow = parts.find((p) => p.model === "slow/x");
|
||||
expect(slow?.ok).toBe(false);
|
||||
expect(slow?.error).toBe("chaos-panel-timeout");
|
||||
});
|
||||
|
||||
it("reports a failed model without breaking the panel", async () => {
|
||||
const handle = fakeHandle(async (model) => {
|
||||
if (model === "boom/z") throw new Error("upstream 500");
|
||||
return textResponse("ok");
|
||||
});
|
||||
const { parts } = await runChaosPanel({
|
||||
body: {},
|
||||
models: ["good/w", "boom/z"],
|
||||
handleSingleModel: handle,
|
||||
});
|
||||
expect(parts.find((p) => p.model === "good/w")?.ok).toBe(true);
|
||||
expect(parts.find((p) => p.model === "boom/z")?.ok).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("serializeChaosPart", () => {
|
||||
it("emits a comment + omni-chaos-part event envelope", () => {
|
||||
const part: ChaosPart = { model: "a/gpt", index: 0, ok: true, text: "hi" };
|
||||
const s = serializeChaosPart(part, false);
|
||||
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");
|
||||
});
|
||||
});
|
||||
|
||||
describe("handleChaosChat", () => {
|
||||
it("emits broadcast events + final OpenAI chunk", async () => {
|
||||
const handle = fakeHandle(async (model) => textResponse(`ans-${model}`));
|
||||
const res = await handleChaosChat({
|
||||
body: { messages: [] },
|
||||
models: ["a/gpt", "b/opus"],
|
||||
handleSingleModel: handle,
|
||||
});
|
||||
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);
|
||||
// final canonical chunk carries the primary answer
|
||||
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({
|
||||
body: {},
|
||||
models: ["only/x"],
|
||||
handleSingleModel: handle,
|
||||
});
|
||||
expect(handle).toHaveBeenCalledTimes(1);
|
||||
const txt = await res.text();
|
||||
expect(txt).toContain("solo");
|
||||
});
|
||||
|
||||
it("streams an error final chunk (status 200) when every model fails", async () => {
|
||||
const handle = fakeHandle(async () => {
|
||||
throw new Error("dead");
|
||||
});
|
||||
const res = await handleChaosChat({
|
||||
body: {},
|
||||
models: ["a/x", "b/y"],
|
||||
handleSingleModel: handle,
|
||||
});
|
||||
// SSE envelope stays well-formed (200) even when the whole panel fails; the
|
||||
// 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);
|
||||
expect(body).toContain("All chaos panel models failed");
|
||||
expect(body).toContain("[DONE]");
|
||||
});
|
||||
|
||||
it("aborts the underlying request on panel timeout", async () => {
|
||||
let aborted = false;
|
||||
let settledEarly = false;
|
||||
const handle = vi.fn(
|
||||
async (
|
||||
_body: Record<string, unknown>,
|
||||
_model: string,
|
||||
target?: { modelAbortSignal?: AbortSignal }
|
||||
) => {
|
||||
const signal = target?.modelAbortSignal;
|
||||
if (signal) signal.addEventListener("abort", () => (aborted = true), { once: true });
|
||||
// Mimic a real downstream dispatcher: abort the in-flight request when the
|
||||
// chaos panel timeout fires (handleSingleModelWithTimeout does exactly this
|
||||
// via target.modelAbortSignal).
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const t = setTimeout(resolve, 5000);
|
||||
signal?.addEventListener(
|
||||
"abort",
|
||||
() => {
|
||||
clearTimeout(t);
|
||||
settledEarly = true;
|
||||
reject(new Error("aborted"));
|
||||
},
|
||||
{ once: true }
|
||||
);
|
||||
});
|
||||
return textResponse("late");
|
||||
}
|
||||
);
|
||||
const { parts } = await runChaosPanel({
|
||||
body: {},
|
||||
models: ["slow/x"],
|
||||
handleSingleModel: handle,
|
||||
tuning: { panelHardTimeoutMs: 150 },
|
||||
});
|
||||
// The single model timed out → its part is a fail and the underlying request
|
||||
// was aborted (did NOT wait the full 5s), releasing the connection.
|
||||
expect(parts).toHaveLength(1);
|
||||
expect(parts[0].ok).toBe(false);
|
||||
expect(parts[0].error).toBe("chaos-panel-timeout");
|
||||
expect(aborted).toBe(true);
|
||||
expect(settledEarly).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,38 @@
|
||||
/**
|
||||
* Integration-style test: confirms the `auto/chaos` virtual auto-combo is
|
||||
* materialized with the right shape so the chat handler will fan it out in
|
||||
* parallel via the chaos engine.
|
||||
*
|
||||
* This exercises the real factory path (not a mock) so we know `auto/chaos`
|
||||
* advertises correctly in /v1/models and routes through combo.ts → handleChaosChat.
|
||||
*/
|
||||
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { createVirtualAutoCombo } from "../virtualFactory";
|
||||
import { AUTO_TEMPLATE_VARIANTS } from "../builtinCatalog";
|
||||
import { parseAutoPrefix } from "../autoPrefix";
|
||||
|
||||
describe("auto/chaos virtual combo", () => {
|
||||
it("is registered in the built-in catalog", () => {
|
||||
expect(AUTO_TEMPLATE_VARIANTS["auto/chaos"]).toBe("chaos");
|
||||
expect(AUTO_TEMPLATE_VARIANTS["auto/best-chaos"]).toBe("chaos");
|
||||
});
|
||||
|
||||
it("parses as a valid auto prefix", () => {
|
||||
const parsed = parseAutoPrefix("auto/chaos");
|
||||
expect(parsed.valid).toBe(true);
|
||||
expect(parsed.variant).toBe("chaos");
|
||||
});
|
||||
|
||||
it("materializes with fusion strategy + chaos config", async () => {
|
||||
const combo = await createVirtualAutoCombo("chaos");
|
||||
expect(combo.type).toBe("auto");
|
||||
expect(combo.strategy).toBe("fusion");
|
||||
expect(combo.config?.chaos?.enabled).toBe(true);
|
||||
// panel is capped to a sane size
|
||||
expect((combo.models ?? []).length).toBeGreaterThan(0);
|
||||
expect((combo.models ?? []).length).toBeLessThanOrEqual(5);
|
||||
// judge (primary) is the first panel model
|
||||
expect(combo.config?.chaos?.judgeModel).toBe(combo.models?.[0]?.model);
|
||||
});
|
||||
});
|
||||
@@ -1,4 +1,4 @@
|
||||
export type AutoVariant = "coding" | "fast" | "cheap" | "offline" | "smart" | "lkgp";
|
||||
export type AutoVariant = "coding" | "fast" | "cheap" | "offline" | "smart" | "lkgp" | "chaos";
|
||||
|
||||
export interface AutoPrefixParseResult {
|
||||
valid: boolean;
|
||||
@@ -13,6 +13,7 @@ export const VALID_VARIANTS: AutoVariant[] = [
|
||||
"offline",
|
||||
"smart",
|
||||
"lkgp",
|
||||
"chaos",
|
||||
];
|
||||
|
||||
/**
|
||||
|
||||
@@ -41,6 +41,9 @@ export const AUTO_TEMPLATE_VARIANTS: Record<string, AutoVariant | undefined> = {
|
||||
"auto/claude-opus": "smart",
|
||||
"auto/claude-sonnet": "coding",
|
||||
"auto/best-free": "cheap",
|
||||
// Chaos mode — parallel dispatch to top-N stable models
|
||||
"auto/best-chaos": "chaos",
|
||||
"auto/chaos": "chaos",
|
||||
};
|
||||
|
||||
/**
|
||||
@@ -62,8 +65,7 @@ export const AUTO_SUFFIX_VARIANTS: string[] = [
|
||||
];
|
||||
|
||||
type ResolvedAutoVariant =
|
||||
| { recognized: true; variant: AutoVariant | undefined }
|
||||
| { recognized: false };
|
||||
{ recognized: true; variant: AutoVariant | undefined } | { recognized: false };
|
||||
|
||||
export function resolveAutoVariant(modelStr: string, suffix: string): ResolvedAutoVariant {
|
||||
if (Object.prototype.hasOwnProperty.call(AUTO_TEMPLATE_VARIANTS, modelStr)) {
|
||||
|
||||
467
open-sse/services/autoCombo/chaosEngine.ts
Normal file
467
open-sse/services/autoCombo/chaosEngine.ts
Normal file
@@ -0,0 +1,467 @@
|
||||
/**
|
||||
* Chaos Engine — parallel multi-model dispatch for the `auto/chaos` auto-model.
|
||||
*
|
||||
* Goal: pick the N most *stable* connected models, fan the same prompt out to
|
||||
* all of them in parallel, then return a single merged SSE stream so an IDE that
|
||||
* can only issue ONE request / hold ONE agent still receives all N model answers
|
||||
* at once.
|
||||
*
|
||||
* Design notes:
|
||||
* - Selection is done by the caller (virtualFactory → createVirtualAutoCombo)
|
||||
* which already scores candidates with the `chaos-mode` weight pack (health +
|
||||
* stability dominant), so `models` here is already the proven-stable set.
|
||||
* - Dispatch is fully parallel. A slow/hung model is bounded by
|
||||
* `panelHardTimeoutMs` and, crucially, its underlying request is *aborted* on
|
||||
* timeout (not merely resolved-to-fallback) so the connection is released.
|
||||
* - Broadcast is PROGRESSIVE: each panel model's answer is enqueued onto the
|
||||
* SSE stream the moment that model lands — the client does not wait for the
|
||||
* whole panel to finish before receiving the first part. Each part is wrapped
|
||||
* in an SSE `omni-chaos-part` event carrying its own model id. IDEs that
|
||||
* understand the protocol can render several panels; IDEs that don't just see
|
||||
* the canonical final block (the highest-scoring model's answer is used as the
|
||||
* final response chunk).
|
||||
* - Unlike fusion, chaos does NOT run a separate judge synthesis call — it
|
||||
* surfaces raw per-model outputs.
|
||||
*/
|
||||
|
||||
import { errorResponse } from "../../utils/error.ts";
|
||||
import type { ComboLogger, HandleSingleModel } from "../combo/types.ts";
|
||||
|
||||
export const CHAOS_DEFAULTS = {
|
||||
/** Absolute cap on wall time for the whole panel. */
|
||||
panelHardTimeoutMs: 120_000,
|
||||
/** If fewer than this many succeed, fall back to a plain single-model answer. */
|
||||
minPanel: 1,
|
||||
} as const;
|
||||
|
||||
export type ChaosTuning = {
|
||||
panelHardTimeoutMs?: number;
|
||||
minPanel?: number;
|
||||
};
|
||||
|
||||
type Body = Record<string, unknown>;
|
||||
|
||||
export type ChaosPart = {
|
||||
model: string;
|
||||
index: number;
|
||||
ok: boolean;
|
||||
text: string;
|
||||
error?: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
export function serializeChaosPart(part: ChaosPart, isFinal: boolean): string {
|
||||
const meta = {
|
||||
type: "omni-chaos-part",
|
||||
model: part.model,
|
||||
index: part.index,
|
||||
ok: part.ok,
|
||||
final: isFinal,
|
||||
error: part.error,
|
||||
};
|
||||
// Comment line (ignored by standard SSE parsers) + explicit event envelope.
|
||||
return (
|
||||
`: chaos ${part.index} ${part.ok ? "ok" : "fail"} ${part.model}\n` +
|
||||
`event: omni-chaos-part\n` +
|
||||
`data: ${JSON.stringify(meta)}\n\n`
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Promise that resolves to `fallback` after `ms`, calling `onTimeout` (e.g. to
|
||||
* abort the underlying request) so a slow/hung model releases its connection
|
||||
* instead of leaking a dangling fetch.
|
||||
*/
|
||||
function withTimeout<T>(
|
||||
p: Promise<T>,
|
||||
ms: number,
|
||||
fallback: T,
|
||||
onTimeout?: () => void
|
||||
): Promise<T> {
|
||||
return new Promise((resolve) => {
|
||||
const t = setTimeout(() => {
|
||||
try {
|
||||
onTimeout?.();
|
||||
} catch {
|
||||
/* ignore abort errors */
|
||||
}
|
||||
resolve(fallback);
|
||||
}, ms);
|
||||
Promise.resolve(p)
|
||||
.then((v) => {
|
||||
clearTimeout(t);
|
||||
resolve(v);
|
||||
})
|
||||
.catch(() => {
|
||||
clearTimeout(t);
|
||||
resolve(fallback);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the chaos panel. Returns the ordered parts plus a recommended "primary"
|
||||
* part (highest index of a successful model, or the first success) that callers
|
||||
* can use as the canonical response body for non-aware clients.
|
||||
*
|
||||
* Each panel call gets its own AbortController; `withTimeout` aborts it on
|
||||
* timeout so the underlying request is cancelled, not merely superseded.
|
||||
*/
|
||||
export async function runChaosPanel(opts: {
|
||||
body: Body;
|
||||
models: string[];
|
||||
handleSingleModel: HandleSingleModel;
|
||||
log?: { info?: (...a: unknown[]) => void; warn?: (...a: unknown[]) => void };
|
||||
tuning?: ChaosTuning | null;
|
||||
}): Promise<{ parts: ChaosPart[]; primary: ChaosPart | null }> {
|
||||
const { body, models, handleSingleModel, log, tuning } = opts;
|
||||
const panel = Array.isArray(models) ? models.filter(Boolean) : [];
|
||||
const hardTimeout = tuning?.panelHardTimeoutMs ?? CHAOS_DEFAULTS.panelHardTimeoutMs;
|
||||
|
||||
if (panel.length === 0) {
|
||||
return { parts: [], primary: null };
|
||||
}
|
||||
|
||||
const calls = panel.map((model, index) => {
|
||||
const ctrl = new AbortController();
|
||||
return withTimeout(
|
||||
(async (): Promise<ChaosPart> => {
|
||||
try {
|
||||
const res = await handleSingleModel(body, model, {
|
||||
modelAbortSignal: ctrl.signal,
|
||||
});
|
||||
const text = await extractText(res);
|
||||
log?.info?.(
|
||||
`CHAOS panel ${index} (${model}) ok=${res.ok} status=${res.status} textLen=${text.length}`
|
||||
);
|
||||
return { model, index, ok: true, text };
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
log?.warn?.(`CHAOS panel ${index} (${model}) failed:`, msg);
|
||||
return { model, index, ok: false, text: "", error: msg };
|
||||
}
|
||||
})(),
|
||||
hardTimeout,
|
||||
// Timed-out model → treated as a failed part, not a hang.
|
||||
{
|
||||
model,
|
||||
index,
|
||||
ok: false,
|
||||
text: "",
|
||||
error: "chaos-panel-timeout",
|
||||
} as ChaosPart,
|
||||
// Abort the underlying fetch so the connection is released on timeout.
|
||||
() => ctrl.abort()
|
||||
);
|
||||
});
|
||||
|
||||
const parts = await Promise.all(calls);
|
||||
const successes = parts.filter((p) => p.ok);
|
||||
const primary = successes.length > 0 ? successes[successes.length - 1] : null;
|
||||
|
||||
log?.info?.(`CHAOS panel complete: ${successes.length}/${parts.length} succeeded`);
|
||||
return { parts, primary };
|
||||
}
|
||||
|
||||
/**
|
||||
* Pull assistant text out of an OpenAI-style OR Anthropic-style Response body.
|
||||
* Clones the response first (body is single-consume; fusion.ts does the same),
|
||||
* then tries JSON first and falls back to SSE concat — content-type headers are
|
||||
* not reliable here because OmniRoute may force a streaming envelope internally.
|
||||
*/
|
||||
async function extractText(res: Response): Promise<string> {
|
||||
let raw: string;
|
||||
try {
|
||||
raw = await res.clone().text();
|
||||
} catch {
|
||||
return "";
|
||||
}
|
||||
// Try JSON first (non-streaming completion).
|
||||
const trimmed = raw.trim();
|
||||
if (trimmed.startsWith("{")) {
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
const fromJson = firstTextFromOpenAI(parsed);
|
||||
if (fromJson) return fromJson;
|
||||
// Anthropic non-streaming also arrives as `{...}` with top-level `content`.
|
||||
if (typeof (parsed as Record<string, unknown>)?.content === "string") {
|
||||
return (parsed as Record<string, unknown>).content as string;
|
||||
}
|
||||
} catch {
|
||||
/* fall through to SSE / raw */
|
||||
}
|
||||
}
|
||||
// SSE or wrapped stream → concat data: content deltas (OpenAI + Anthropic).
|
||||
const sse = concatSseText(raw);
|
||||
if (sse) return sse;
|
||||
// Last resort: maybe the body itself is plain prose.
|
||||
return trimmed.length > 0 && !trimmed.startsWith("data:") ? trimmed : "";
|
||||
}
|
||||
|
||||
function firstTextFromOpenAI(obj: unknown): string {
|
||||
if (!obj || typeof obj !== "object") return "";
|
||||
const o = obj as Record<string, unknown>;
|
||||
const choices = o.choices as Array<Record<string, unknown>> | undefined;
|
||||
if (Array.isArray(choices) && choices.length > 0) {
|
||||
const msg = choices[0]?.message as Record<string, unknown> | undefined;
|
||||
if (msg && typeof msg.content === "string") return msg.content;
|
||||
const delta = choices[0]?.delta as Record<string, unknown> | undefined;
|
||||
if (delta && typeof delta.content === "string") return delta.content;
|
||||
}
|
||||
if (typeof o.content === "string") return o.content;
|
||||
return "";
|
||||
}
|
||||
|
||||
/**
|
||||
* Concatenate assistant text out of an SSE byte stream.
|
||||
*
|
||||
* Supports BOTH wire formats OmniRoute may emit:
|
||||
* - OpenAI: `data: {"choices":[{"delta":{"content":"..."}}]}`
|
||||
* - Anthropic: `data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"..."}}`
|
||||
* (also accepts the older `delta:{"content":"..."}` proxy shape and a
|
||||
* top-level `content` token).
|
||||
*/
|
||||
function concatSseText(sse: string): string {
|
||||
const out: string[] = [];
|
||||
for (const line of sse.split("\n")) {
|
||||
if (!line.startsWith("data:")) continue;
|
||||
const payload = line.slice(5).trim();
|
||||
if (!payload || payload === "[DONE]") continue;
|
||||
try {
|
||||
const json = JSON.parse(payload);
|
||||
const choices = json?.choices as Array<Record<string, unknown>> | undefined;
|
||||
if (Array.isArray(choices) && choices.length > 0) {
|
||||
const delta = choices[0]?.delta as Record<string, unknown> | undefined;
|
||||
if (delta?.content) {
|
||||
out.push(String(delta.content));
|
||||
continue;
|
||||
}
|
||||
const message = choices[0]?.message as Record<string, unknown> | undefined;
|
||||
if (message?.content) {
|
||||
out.push(String(message.content));
|
||||
continue;
|
||||
}
|
||||
}
|
||||
// Anthropic Messages streaming delta shapes.
|
||||
const delta = json?.delta as Record<string, unknown> | undefined;
|
||||
if (delta && typeof delta.text === "string") {
|
||||
// content_block_delta: { delta: { type: "text_delta", text: "..." } }
|
||||
out.push(String(delta.text));
|
||||
} else if (delta && typeof delta.content === "string") {
|
||||
// proxy shape: { delta: { content: "..." } }
|
||||
out.push(String(delta.content));
|
||||
} else if (typeof json?.content === "string") {
|
||||
// top-level content token (rare, non-standard)
|
||||
out.push(String(json.content));
|
||||
}
|
||||
} catch {
|
||||
/* ignore non-JSON lines */
|
||||
}
|
||||
}
|
||||
return out.join("");
|
||||
}
|
||||
|
||||
/**
|
||||
* Top-level chaos dispatch entrypoint used by `handleComboChat` when a combo's
|
||||
* `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)
|
||||
* - 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]`
|
||||
*
|
||||
* If no panel model succeeds, the stream still terminates cleanly with a
|
||||
* `x-omniroute-chaos-error` header and an error final chunk (status stays 200
|
||||
* so the SSE envelope is well-formed; non-aware clients see the error text).
|
||||
*/
|
||||
export async function handleChaosChat(opts: {
|
||||
body: Body;
|
||||
models: string[];
|
||||
handleSingleModel: HandleSingleModel;
|
||||
log?: { info?: (...a: unknown[]) => void; warn?: (...a: unknown[]) => void };
|
||||
comboName?: string;
|
||||
primaryModel?: string | null;
|
||||
tuning?: ChaosTuning | null;
|
||||
}): Promise<Response> {
|
||||
const { body, models, handleSingleModel, log, comboName, primaryModel, tuning } = opts;
|
||||
const panel = Array.isArray(models) ? models.filter(Boolean) : [];
|
||||
const hardTimeout = tuning?.panelHardTimeoutMs ?? CHAOS_DEFAULTS.panelHardTimeoutMs;
|
||||
if (panel.length === 0) {
|
||||
return errorResponse(400, "Chaos combo has no models");
|
||||
}
|
||||
|
||||
// Single-model chaos degrades to a direct answer.
|
||||
if (panel.length === 1) {
|
||||
return handleSingleModel(body, panel[0]);
|
||||
}
|
||||
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
async start(controller) {
|
||||
const enc = new TextEncoder();
|
||||
let closed = false;
|
||||
// Serialize enqueues: ReadableStream controllers must not be enqueued
|
||||
// concurrently. A committed promise chain gives us ordered, race-free writes.
|
||||
let enqueueChain: Promise<void> = Promise.resolve();
|
||||
const safeEnqueue = (s: string): Promise<void> => {
|
||||
enqueueChain = enqueueChain.then(() => {
|
||||
if (closed) return;
|
||||
try {
|
||||
controller.enqueue(enc.encode(s));
|
||||
} catch {
|
||||
/* stream already errored/closed */
|
||||
}
|
||||
});
|
||||
return enqueueChain;
|
||||
};
|
||||
|
||||
const modelPromises = panel.map((model, index) => {
|
||||
const ctrl = new AbortController();
|
||||
return withTimeout(
|
||||
(async (): Promise<ChaosPart> => {
|
||||
try {
|
||||
const res = await handleSingleModel(body, model, {
|
||||
modelAbortSignal: ctrl.signal,
|
||||
});
|
||||
const text = await extractText(res);
|
||||
const part: ChaosPart = { model, index, ok: true, text };
|
||||
// Progressive: emit this part the instant its model lands.
|
||||
await safeEnqueue(serializeChaosPart(part, false));
|
||||
return part;
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
const part: ChaosPart = { model, index, ok: false, text: "", error: msg };
|
||||
await safeEnqueue(serializeChaosPart(part, false));
|
||||
return part;
|
||||
}
|
||||
})(),
|
||||
hardTimeout,
|
||||
{ model, index, ok: false, text: "", error: "chaos-panel-timeout" } as ChaosPart,
|
||||
() => ctrl.abort()
|
||||
);
|
||||
});
|
||||
|
||||
const allParts = await Promise.all(modelPromises);
|
||||
const successes = allParts.filter((p) => p.ok);
|
||||
|
||||
if (successes.length === 0) {
|
||||
// No panel model answered — terminate with an error final chunk.
|
||||
const errText = "All chaos panel models failed";
|
||||
await safeEnqueue(
|
||||
`data: ${JSON.stringify({
|
||||
id: `chaos-${comboName ?? "panel"}`,
|
||||
object: "chat.completion.chunk",
|
||||
model: panel[0],
|
||||
choices: [
|
||||
{
|
||||
index: 0,
|
||||
delta: { role: "assistant", content: errText },
|
||||
finish_reason: "stop",
|
||||
},
|
||||
],
|
||||
})}\n\n`
|
||||
);
|
||||
await safeEnqueue("data: [DONE]\n\n");
|
||||
await enqueueChain;
|
||||
closed = true;
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
|
||||
// Choose the primary: explicit primaryModel if it succeeded, else the last
|
||||
// successful part (by construction that's the top-scored stable model).
|
||||
const primaryPart =
|
||||
(primaryModel && allParts.find((p) => p.model === primaryModel && p.ok)) ||
|
||||
allParts.filter((p) => p.ok).slice(-1)[0] ||
|
||||
successes[0];
|
||||
|
||||
// Final canonical answer (non-aware clients consume this).
|
||||
const finalText = primaryPart?.text ?? "";
|
||||
await safeEnqueue(
|
||||
`data: ${JSON.stringify({
|
||||
id: `chaos-${comboName ?? "panel"}`,
|
||||
object: "chat.completion.chunk",
|
||||
model: primaryPart?.model ?? panel[0],
|
||||
choices: [
|
||||
{
|
||||
index: 0,
|
||||
delta: { role: "assistant", content: finalText },
|
||||
finish_reason: "stop",
|
||||
},
|
||||
],
|
||||
})}\n\n`
|
||||
);
|
||||
await safeEnqueue("data: [DONE]\n\n");
|
||||
await enqueueChain;
|
||||
closed = true;
|
||||
controller.close();
|
||||
},
|
||||
});
|
||||
|
||||
return new Response(stream, {
|
||||
status: 200,
|
||||
headers: {
|
||||
"Content-Type": "text/event-stream; charset=utf-8",
|
||||
"Cache-Control": "no-cache, no-transform",
|
||||
Connection: "keep-alive",
|
||||
"X-OmniRoute-Chaos": "true",
|
||||
"X-OmniRoute-Chaos-Panel": String(panel.length),
|
||||
"X-OmniRoute-Chaos-Primary": "",
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect + dispatch chaos mode from `handleComboChat`'s combo config, mirroring
|
||||
* the fusion-strategy short-circuit right above it. Detected via
|
||||
* `combo.config.chaos.enabled` (set by the auto/chaos virtual combo). Unlike
|
||||
* fusion, chaos surfaces each model's raw answer (no judge synthesis) so a
|
||||
* single-turn IDE request still receives N model outputs at once.
|
||||
*
|
||||
* Returns `null` when the combo is not a chaos combo (caller falls through to
|
||||
* its normal strategy handling); otherwise returns the chaos dispatch promise.
|
||||
*/
|
||||
export function dispatchChaosFromCombo(args: {
|
||||
cfg: Record<string, unknown>;
|
||||
comboModels: unknown[];
|
||||
comboName: string;
|
||||
body: Body;
|
||||
handleSingleModel: HandleSingleModel;
|
||||
log: ComboLogger;
|
||||
}): Promise<Response> | null {
|
||||
const { cfg, comboModels, comboName, body, handleSingleModel, log } = args;
|
||||
if (
|
||||
!cfg.chaos ||
|
||||
typeof cfg.chaos !== "object" ||
|
||||
!(cfg.chaos as Record<string, unknown>).enabled
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
const chaosCfg = cfg.chaos as { panelSize?: number; judgeModel?: string };
|
||||
const chaosModels = (comboModels || [])
|
||||
.map((m) => {
|
||||
if (typeof m === "string") return m;
|
||||
if (m && typeof m === "object") {
|
||||
const obj = m as Record<string, unknown>;
|
||||
if (typeof obj.model === "string") return obj.model;
|
||||
}
|
||||
return null;
|
||||
})
|
||||
.filter((m): m is string => Boolean(m));
|
||||
log.info("CHAOS", `dispatching parallel panel of ${chaosModels.length} stable models`);
|
||||
return handleChaosChat({
|
||||
body,
|
||||
models: chaosModels,
|
||||
handleSingleModel,
|
||||
log,
|
||||
comboName,
|
||||
primaryModel: chaosCfg.judgeModel,
|
||||
});
|
||||
}
|
||||
@@ -15,9 +15,12 @@ export {
|
||||
export { getTaskFitness, getTaskTypes } from "./taskFitness";
|
||||
export { SelfHealingManager, getSelfHealingManager } from "./selfHealing";
|
||||
export { MODE_PACKS, getModePack, getModePackNames } from "./modePacks";
|
||||
export { selectProvider, type AutoComboConfig, type SelectionResult } from "./engine";
|
||||
export {
|
||||
selectProvider,
|
||||
BudgetExceededError,
|
||||
type AutoComboConfig,
|
||||
type SelectionResult,
|
||||
} from "./engine";
|
||||
runChaosPanel,
|
||||
handleChaosChat,
|
||||
serializeChaosPart,
|
||||
CHAOS_DEFAULTS,
|
||||
type ChaosTuning,
|
||||
type ChaosPart,
|
||||
} from "./chaosEngine";
|
||||
|
||||
@@ -88,6 +88,23 @@ export const MODE_PACKS: Record<string, ScoringWeights> = {
|
||||
resetWindowAffinity: 0,
|
||||
connectionDensity: 0.05,
|
||||
},
|
||||
// Chaos mode — priority: health > stability > taskFit > quota diversity.
|
||||
// Selects top-N healthy providers for parallel dispatch. Favors providers with
|
||||
// closed circuit breakers, low latency variance, and high task fitness.
|
||||
"chaos-mode": {
|
||||
quota: 0.1,
|
||||
health: 0.4,
|
||||
costInv: 0.02,
|
||||
latencyInv: 0.03,
|
||||
taskFit: 0.2,
|
||||
stability: 0.15,
|
||||
tierPriority: 0.02,
|
||||
tierAffinity: 0,
|
||||
specificityMatch: 0,
|
||||
contextAffinity: 0.03,
|
||||
resetWindowAffinity: 0,
|
||||
connectionDensity: 0.05,
|
||||
},
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@@ -288,10 +288,7 @@ export async function createVirtualAutoCombo(
|
||||
// `providerSpecificData.excludedModels` regardless of its isActive state (the
|
||||
// dispatch-time enforcement in auth.ts does not gate on isActive either), so
|
||||
// gather it from BOTH the active and disabled connection lists.
|
||||
const noAuthProviderSpecificData = new Map<
|
||||
string,
|
||||
Record<string, unknown> | null | undefined
|
||||
>();
|
||||
const noAuthProviderSpecificData = new Map<string, Record<string, unknown> | null | undefined>();
|
||||
for (const conn of [...connections, ...disabledNoAuthConnections]) {
|
||||
if (conn.provider in NOAUTH_PROVIDERS) {
|
||||
noAuthProviderSpecificData.set(conn.provider, conn.providerSpecificData);
|
||||
@@ -343,7 +340,10 @@ export async function createVirtualAutoCombo(
|
||||
// `/v1/models` listing — so auto-routing never picks a model that will 402/403.
|
||||
// If this empties the pool the existing graceful empty-pool path below handles it
|
||||
// (consistent with the opt-in intent). Default OFF → pool unchanged.
|
||||
const paidFilteredPool = filterPaidOnlyCandidates(candidatePool, settings.hidePaidModels === true);
|
||||
const paidFilteredPool = filterPaidOnlyCandidates(
|
||||
candidatePool,
|
||||
settings.hidePaidModels === true
|
||||
);
|
||||
if (paidFilteredPool !== candidatePool) {
|
||||
candidatePool.length = 0;
|
||||
candidatePool.push(...paidFilteredPool);
|
||||
@@ -449,6 +449,12 @@ export async function createVirtualAutoCombo(
|
||||
// LKGP is default for all auto variants, this variant just explicitly names it.
|
||||
// Use default weights.
|
||||
break;
|
||||
case "chaos":
|
||||
// Chaos mode: select top-N most stable models and fan them out in parallel
|
||||
// (strategy "fusion"). Prioritize health + stability via the chaos-mode pack.
|
||||
weights = { ...MODE_PACKS["chaos-mode"] };
|
||||
explorationRate = 0; // no exploration — only the proven-stable set
|
||||
break;
|
||||
case undefined: // Default auto
|
||||
// Use default weights
|
||||
break;
|
||||
@@ -489,20 +495,39 @@ export async function createVirtualAutoCombo(
|
||||
routerStrategy,
|
||||
};
|
||||
|
||||
// Chaos mode fans out to the top-N most stable models in parallel. We cap the
|
||||
// panel size so a single IDE request doesn't fan out to dozens of providers.
|
||||
const isChaos = variant === "chaos";
|
||||
const CHAOS_MAX_PANEL = 5;
|
||||
const chaosModels = isChaos ? models.slice(0, CHAOS_MAX_PANEL) : models;
|
||||
|
||||
const advertisedLimits = computeAdvertisedLimits(effectivePool);
|
||||
|
||||
return {
|
||||
id: `virtual-auto-${variant || "default"}`,
|
||||
name: `Auto ${variant || "Default"}`,
|
||||
type: "auto",
|
||||
strategy: "auto",
|
||||
models,
|
||||
strategy: isChaos ? "fusion" : "auto",
|
||||
models: chaosModels,
|
||||
candidatePool: providerPool,
|
||||
weights,
|
||||
explorationRate,
|
||||
routerStrategy,
|
||||
autoConfig,
|
||||
config: { auto: autoConfig },
|
||||
// For chaos, stash the panel size + a flag so downstream handlers can detect
|
||||
// the broadcast mode and stream each panel model back to IDEs that opt in.
|
||||
config: {
|
||||
auto: autoConfig,
|
||||
...(isChaos
|
||||
? {
|
||||
chaos: {
|
||||
enabled: true,
|
||||
panelSize: chaosModels.length,
|
||||
judgeModel: chaosModels[0]?.model,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
},
|
||||
advertisedContextLength: advertisedLimits.contextLength,
|
||||
advertisedMaxOutputTokens: advertisedLimits.maxOutputTokens,
|
||||
};
|
||||
|
||||
@@ -136,6 +136,7 @@ import {
|
||||
waitForCooldownAwareRetry,
|
||||
} from "../../src/sse/services/cooldownAwareRetry.ts";
|
||||
import { handleFusionChat, type FusionTuning } from "./fusion.ts";
|
||||
import { dispatchChaosFromCombo } from "./autoCombo/chaosEngine.ts";
|
||||
import { handlePipelineChat, type PipelineStep } from "./pipeline.ts";
|
||||
import {
|
||||
TRANSIENT_FOR_SEMAPHORE,
|
||||
@@ -899,6 +900,18 @@ export async function handleComboChat({
|
||||
});
|
||||
}
|
||||
|
||||
// Chaos mode (parallel multi-model dispatch): detection + dispatch live in
|
||||
// chaosEngine.ts (dispatchChaosFromCombo), returning null when not chaos-enabled.
|
||||
const chaosDispatch = dispatchChaosFromCombo({
|
||||
cfg,
|
||||
comboModels: combo.models || [],
|
||||
comboName: combo.name,
|
||||
body,
|
||||
handleSingleModel: handleSingleModelWithTimeout,
|
||||
log,
|
||||
});
|
||||
if (chaosDispatch) return chaosDispatch;
|
||||
|
||||
// Pipeline strategy: sequential chain — each step's output feeds the next step's
|
||||
// input, only the final step's response is returned. Handled in a separate module
|
||||
// because it neither iterates targets as fallbacks nor needs the failover/retry
|
||||
|
||||
@@ -54,6 +54,52 @@ export const OUTPUT_STYLE_CATALOG: Record<string, OutputStyle> = {
|
||||
ultra: `Minimal diff discipline. Touch the fewest lines that make it work. Zero new files, classes, or config unless strictly required. Inline over abstract. No "while we're here" extras. ${SHARED_BOUNDARIES}`,
|
||||
},
|
||||
},
|
||||
// Ponytail (lazy-senior-dev mode) — integrated into the output-style registry
|
||||
// so it rides the existing production injector instead of a bespoke module.
|
||||
// Source: https://github.com/DietrichGebert/ponytail (MIT). This is a fuller
|
||||
// treatment than "less-code" (which is the 9router port); both are offered so
|
||||
// users can pick the leaner or the richer ladder.
|
||||
ponytail: {
|
||||
id: "ponytail",
|
||||
label: "Ponytail (lazy senior dev)",
|
||||
description:
|
||||
"Lazy senior-dev discipline: climb the YAGNI ladder, fix root cause, smallest working diff.",
|
||||
levels: {
|
||||
lite: `# Ponytail, lazy senior dev mode (lite)
|
||||
|
||||
Before writing any code: does it need to exist? Does it already exist here? Does the stdlib/installed dep cover it? Only then: write the minimum. Reuse over rewrite. ${SHARED_BOUNDARIES}`,
|
||||
full: `# Ponytail, lazy senior dev mode
|
||||
|
||||
You are a lazy senior developer. Lazy means efficient, not careless. The best code is the code never written.
|
||||
|
||||
Before writing any code, stop at the first rung that holds:
|
||||
|
||||
1. Does this need to be built at all? (YAGNI)
|
||||
2. Does it already exist in this codebase? Reuse the helper, util, or pattern that's already here, don't re-write it.
|
||||
3. Does the standard library already do this? Use it.
|
||||
4. Does a native platform feature cover it? Use it.
|
||||
5. Does an already-installed dependency solve it? Use it.
|
||||
6. Can this be one line? Make it one line.
|
||||
7. Only then: write the minimum code that works.
|
||||
|
||||
The ladder runs after you understand the problem, not instead of it: read the task and the code it touches, trace the real flow end to end, then climb.
|
||||
|
||||
Bug fix = root cause, not symptom: a report names a symptom. Grep every caller of the function you touch and fix the shared function once — one guard there is a smaller diff than one per caller, and patching only the path the ticket names leaves a sibling caller still broken.
|
||||
|
||||
Rules:
|
||||
|
||||
- No abstractions that weren't explicitly requested.
|
||||
- No new dependency if it can be avoided.
|
||||
- No boilerplate nobody asked for.
|
||||
- Deletion over addition. Boring over clever. Fewest files possible.
|
||||
- Shortest working diff wins, but only once you understand the problem. The smallest change in the wrong place isn't lazy, it's a second bug.
|
||||
- Question complex requests: "Do you actually need X, or does Y cover it?"
|
||||
- Pick the edge-case-correct option when two solutions tie. ${SHARED_BOUNDARIES}`,
|
||||
ultra: `# Ponytail, lazy senior dev mode (ultra)
|
||||
|
||||
You are a lazy senior developer. Lazy means efficient, not careless. The best code is the code never written. Before any code: YAGNI → reuse → stdlib → platform feature → installed dep → one line → minimum that works. Fix root cause, not symptom: grep every caller, patch the shared function once. No unrequested abstractions. No new dependency. No boilerplate. Deletion over addition. Fewest files. Shortest working diff, only after you understand the problem. Question complex asks. Edge-case-correct when tied. ${SHARED_BOUNDARIES}`,
|
||||
},
|
||||
},
|
||||
"terse-cjk": {
|
||||
id: "terse-cjk",
|
||||
label: "Terse CJK (文言)",
|
||||
|
||||
Reference in New Issue
Block a user