diff --git a/CHANGELOG.md b/CHANGELOG.md index c4d22d90ac..8829e87019 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ ### ✨ New Features +- **feat(combo):** new **`pipeline` (sequential) combo strategy** ([#6297](https://github.com/diegosouzapw/OmniRoute/issues/6297)) — the 18th routing strategy runs targets **in order**, threading each step's output into the next step's input, with an optional per-step `prompt` (system instruction); only the final step's response is returned. Distinct from `fusion` (parallel fan-out + judge). Implemented as a self-contained `open-sse/services/pipeline.ts` (sibling to `fusion.ts`), dispatched from `combo.ts`; the step list reuses `combo.models` order and reads an optional `prompt` off each target (backward-compatible — ignored by every other strategy). Intermediate steps run non-streaming with tools stripped (complete prose to thread forward); the final step keeps the client's `stream` flag + tools. A failing/empty/unparseable intermediate step fails the whole pipeline explicitly via a sanitized error (never silently swallowed). Kilo's dup flag vs #563 was a false positive (that's model→chain selection; this is a sequential chain). Regression guard: `tests/unit/combo-pipeline-strategy.test.ts` (5). (thanks @ofekbetzalel) - **feat(ci):** `check:test-masking` now flags **inline-reimplemented prod conditions** ([#6348](https://github.com/diegosouzapw/OmniRoute/issues/6348)) — a new **report-only** subcheck (v2, 6A.10 family) catches the wrong-shape contract test: a test that recomputes the condition under test inline instead of importing/exercising the real function (the #6216 class, where `=== 500` → `>= 500` stayed green because the test re-implemented the branch). For each added/modified test file it warns when the file textually duplicates a ≥3-token conditional from a production file touched in the same PR **and** does not import the symbol/module owning it, via a pure, fixture-tested `findReimplementedConditions()` with an allowlist mirroring `assertReductionAllowlist`. Report-only for now (does not fail the gate) — to be promoted to blocking after a triage cycle. Regression guard: `tests/unit/check-test-masking.test.ts` (45). - **feat(sse):** per-connection routing override (native vs CLIProxyAPI) ([#6339](https://github.com/diegosouzapw/OmniRoute/issues/6339)) — the previously-dead `isCliproxyapiDeepModeEnabled` helper is now wired into `resolveExecutorWithProxy`: a single connection can opt itself into the CLIProxyAPI passthrough executor via `providerSpecificData.cliproxyapiMode="claude-native"`, with precedence **connection override > provider `upstream_proxy_config` mode > default**. `resolveExecutorWithProxy` now receives the resolved connection's `providerSpecificData` (threaded from `chatCore.ts`), so one connection can deep-route while the provider's other connections stay native — no DB schema change (the toggle rides in `providerSpecificData`). Also resolves the same-provider mixing ask in #6340. Regression guard: `tests/unit/chatcore-executor-proxy.test.ts` (9). (thanks @RaviTharuma) - **feat(dashboard):** "Add session cookie" modal now shows a prominent **"Open ‹host› →"** link to the provider's own site ([#6268](https://github.com/diegosouzapw/OmniRoute/issues/6268)) — every `-web` cookie-session provider (chatgpt-web, claude-web, gemini-web, kimi-web, lmarena, qwen-web, m365-copilot-web, …) renders a one-click external link (opening the provider's login/home page in a new tab) so operators no longer tab away to retype the URL mid-setup. The host resolves from a pure, unit-tested `resolveWebProviderHost()` (prefers `WEB_COOKIE_PROVIDERS[id].website`, falls back to the registry `baseUrl` origin); non-web providers render exactly as before. Kilo's dup flag vs #6265 (modal-too-small-on-1080p) was a false positive — distinct concern. Regression guard: `tests/unit/resolve-web-provider-host.test.ts` (5). (thanks @chirag127) diff --git a/open-sse/services/combo.ts b/open-sse/services/combo.ts index 6b17ead328..3b7c274061 100644 --- a/open-sse/services/combo.ts +++ b/open-sse/services/combo.ts @@ -119,6 +119,7 @@ import { waitForCooldownAwareRetry, } from "../../src/sse/services/cooldownAwareRetry.ts"; import { handleFusionChat, type FusionTuning } from "./fusion.ts"; +import { handlePipelineChat, type PipelineStep } from "./pipeline.ts"; import { TRANSIENT_FOR_SEMAPHORE, MAX_FALLBACK_WAIT_MS, @@ -797,6 +798,37 @@ export async function handleComboChat({ }); } + // 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 + // machinery below — it runs targets in order, threading output → input. The step + // list is `combo.models` (in order); an optional per-step `prompt` is read off the + // target object (comboModelStepInputSchema.prompt). + if (strategy === "pipeline") { + const pipelineSteps = (combo.models || []) + .map((m): PipelineStep | null => { + if (typeof m === "string") return { model: m }; + if (m && typeof m === "object") { + const obj = m as Record; + if (typeof obj.model === "string") { + return { + model: obj.model, + prompt: typeof obj.prompt === "string" ? obj.prompt : undefined, + }; + } + } + return null; + }) + .filter((s): s is PipelineStep => Boolean(s)); + return handlePipelineChat({ + body, + steps: pipelineSteps, + handleSingleModel: handleSingleModelWithTimeout, + log, + comboName: combo.name, + }); + } + const nestingContext = nesting || { depth: 0, maxDepth: clampComboDepth(config.maxComboDepth), diff --git a/open-sse/services/pipeline.ts b/open-sse/services/pipeline.ts new file mode 100644 index 0000000000..3c1c86b06d --- /dev/null +++ b/open-sse/services/pipeline.ts @@ -0,0 +1,189 @@ +/** + * Pipeline combo strategy — sequential chain. + * + * A pipeline combo runs its targets IN ORDER: step N's output is fed into step + * N+1 as input, each step carries its own optional `prompt` (instruction), and + * only the FINAL step's response is returned to the client. This is the sequential + * counterpart to `fusion` (parallel fan-out + judge synthesis). + * + * ── Per-step config shape ───────────────────────────────────────────────────── + * The ordered step list IS `combo.models` — we reuse the existing target order + * rather than introducing a parallel `pipelineSteps` array that could drift out of + * sync with the models. Each step's optional instruction is read from a `prompt` + * field on the target object (`comboModelStepInputSchema.prompt`); a plain-string + * model entry is simply a step with no prompt. The field is optional and ignored by + * every other strategy, so this is fully backward-compatible. + * + * ── Prompt injection ────────────────────────────────────────────────────────── + * The engine passes prompts through the request's message array (OpenAI `messages`, + * Responses `input`, or Gemini `contents`). Each step's `prompt` is injected as a + * leading system instruction in whichever format the request uses: + * - step 1 keeps the client's original conversation and (if set) prepends its + * prompt as an extra system turn, so the first model sees the real user request; + * - steps 2..N are transforms — the conversation is replaced with the previous + * step's output as the user turn, plus this step's prompt as the system turn. + * + * Intermediate steps are forced non-streaming with tools stripped (we need the + * complete text to thread forward). The FINAL step keeps the client's original + * `stream` flag + tools, so streaming and downstream tool use still work. + * + * A step failure fails the whole pipeline EXPLICITLY (never silently swallowed): + * a non-OK intermediate response, an unparseable body, or an intermediate step that + * yields no text short-circuits with a sanitized error response. + */ +import { errorResponse } from "../utils/error.ts"; +import type { ComboLogger, HandleSingleModel } from "./combo/types.ts"; +// extractPanelText is a generic assistant-text extractor (OpenAI chat / Claude / +// Gemini / Responses) — reused here to read each step's output, not fusion-specific. +import { extractPanelText } from "./fusion.ts"; + +type Body = Record; + +export type PipelineStep = { model: string; prompt?: string | null }; + +/** + * Prepend a system instruction to the client's original conversation (format-aware), + * so step 1 sees the real user request plus its own step prompt. No-op when the + * step has no prompt. + */ +export function prependSystemInstruction(body: Body, prompt: string | null | undefined): Body { + const sys = typeof prompt === "string" && prompt.trim() ? prompt.trim() : null; + const next: Body = { ...body }; + if (!sys) return next; + if (Array.isArray(body.input)) { + next.input = [{ role: "system", content: sys }, ...(body.input as unknown[])]; + } else if (Array.isArray(body.contents)) { + // Gemini contents have no system role — a leading user turn is the closest analog. + next.contents = [{ role: "user", parts: [{ text: sys }] }, ...(body.contents as unknown[])]; + } else if (Array.isArray(body.messages)) { + next.messages = [{ role: "system", content: sys }, ...(body.messages as unknown[])]; + } else { + next.messages = [{ role: "system", content: sys }]; + } + return next; +} + +/** + * Replace the request's conversation with a fresh transform turn set (the previous + * step's output as the user turn + this step's prompt as the system turn), + * preserving whichever message-array shape the request format uses. Non-message + * fields are carried over; the caller overrides the model per step. + */ +export function buildTransformBody( + body: Body, + prompt: string | null | undefined, + input: string +): Body { + const next: Body = { ...body }; + const sys = typeof prompt === "string" && prompt.trim() ? prompt.trim() : null; + if (Array.isArray(body.input)) { + const turns: unknown[] = []; + if (sys) turns.push({ role: "system", content: sys }); + turns.push({ role: "user", content: input }); + next.input = turns; + delete next.messages; + delete next.contents; + } else if (Array.isArray(body.contents)) { + // Gemini contents have no system role — fold the instruction into the user turn. + const text = sys ? `${sys}\n\n${input}` : input; + next.contents = [{ role: "user", parts: [{ text }] }]; + delete next.messages; + delete next.input; + } else { + const turns: unknown[] = []; + if (sys) turns.push({ role: "system", content: sys }); + turns.push({ role: "user", content: input }); + next.messages = turns; + } + return next; +} + +/** Force non-streaming and strip tools so an intermediate step yields complete prose. */ +function stripStreaming(body: Body): Body { + const { tools: _tools, tool_choice: _tc, ...rest } = body; + void _tools; + void _tc; + return { ...rest, stream: false }; +} + +export type HandlePipelineChatOptions = { + body: Body; + steps: PipelineStep[]; + handleSingleModel: HandleSingleModel; + log: ComboLogger; + comboName?: string; +}; + +/** + * Handle a pipeline combo: run the steps in order, threading each step's output + * into the next step's input, and return only the final step's response. + */ +export async function handlePipelineChat({ + body, + steps, + handleSingleModel, + log, + comboName, +}: HandlePipelineChatOptions): Promise { + const chain = (Array.isArray(steps) ? steps : []).filter((s) => s && s.model); + if (chain.length === 0) { + return errorResponse(400, "Pipeline combo has no models"); + } + log.info( + "PIPELINE", + `Combo "${comboName ?? ""}" | steps=${chain.length} [${chain.map((s) => s.model).join(" -> ")}]` + ); + + // Single-step pipeline: nothing to chain — run it directly (streams to client). + if (chain.length === 1) { + return handleSingleModel(prependSystemInstruction(body, chain[0].prompt), chain[0].model); + } + + let prevOutput = ""; + for (let i = 0; i < chain.length; i++) { + const step = chain[i]; + const isFinal = i === chain.length - 1; + const isFirst = i === 0; + + let stepBody: Body = isFirst + ? prependSystemInstruction(body, step.prompt) + : buildTransformBody(body, step.prompt, prevOutput); + // Intermediate steps: complete prose only (no stream, no tools). The final step + // keeps the client's original stream flag + tools. + if (!isFinal) stepBody = stripStreaming(stepBody); + + const t0 = Date.now(); + const res = await handleSingleModel(stepBody, step.model); + + if (isFinal) { + log.info("PIPELINE", `Final step ${step.model} responded (${Date.now() - t0}ms)`); + return res; + } + + // An intermediate step must succeed with usable text — otherwise fail the whole + // pipeline (never silently swallow; the client gets a clear, sanitized error). + if (!res.ok) { + log.warn("PIPELINE", `Step ${i + 1} (${step.model}) failed`, { status: res.status }); + const status = res.status >= 400 && res.status <= 599 ? res.status : 502; + return errorResponse(status, `Pipeline step ${i + 1} (${step.model}) failed`); + } + try { + const json = await res.clone().json(); + prevOutput = extractPanelText(json); + } catch { + log.warn("PIPELINE", `Step ${i + 1} (${step.model}) returned an unparseable body`); + return errorResponse(502, `Pipeline step ${i + 1} (${step.model}) returned an unparseable body`); + } + if (!prevOutput.trim()) { + log.warn("PIPELINE", `Step ${i + 1} (${step.model}) returned empty output`); + return errorResponse(502, `Pipeline step ${i + 1} (${step.model}) returned empty output`); + } + log.info( + "PIPELINE", + `Step ${i + 1} ${step.model} ok (${prevOutput.length} chars, ${Date.now() - t0}ms)` + ); + } + + // Unreachable — the final step returns inside the loop. + return errorResponse(500, "Pipeline produced no final response"); +} diff --git a/src/shared/constants/routingStrategies.ts b/src/shared/constants/routingStrategies.ts index 736e4a7c07..a1287af553 100644 --- a/src/shared/constants/routingStrategies.ts +++ b/src/shared/constants/routingStrategies.ts @@ -16,6 +16,7 @@ export const ROUTING_STRATEGY_VALUES = [ "lkgp", "context-optimized", "fusion", + "pipeline", ] as const; export type RoutingStrategyValue = (typeof ROUTING_STRATEGY_VALUES)[number]; @@ -203,6 +204,13 @@ export const ROUTING_STRATEGIES: RoutingStrategyOption[] = [ settingsDescKey: "fusionDesc", icon: "hub", }, + { + value: "pipeline", + labelKey: "pipeline", + combosDescKey: "pipelineDesc", + settingsDescKey: "pipelineDesc", + icon: "linear_scale", + }, ]; export const SETTINGS_FALLBACK_STRATEGY_VALUES = ACCOUNT_FALLBACK_STRATEGY_VALUES; diff --git a/src/shared/validation/schemas/combo.ts b/src/shared/validation/schemas/combo.ts index 568fd7bfcf..96a76673e9 100644 --- a/src/shared/validation/schemas/combo.ts +++ b/src/shared/validation/schemas/combo.ts @@ -29,6 +29,11 @@ export const comboModelStepInputSchema = z.object({ model: z.string().trim().min(1).max(300), connectionId: z.string().trim().min(1).max(200).nullable().optional(), tags: z.array(z.string().trim().min(1).max(100)).max(20).optional(), + // Pipeline strategy (open-sse/services/pipeline.ts): an optional per-step + // instruction. Steps run in `models` order — each step's output feeds the next + // step's input, and this `prompt` is injected as that step's system instruction. + // Ignored by every other strategy, so it is fully backward-compatible. + prompt: z.string().trim().min(1).max(20000).optional(), ...comboStepMetaSchema, }); diff --git a/tests/unit/combo-pipeline-strategy.test.ts b/tests/unit/combo-pipeline-strategy.test.ts new file mode 100644 index 0000000000..594479f6c1 --- /dev/null +++ b/tests/unit/combo-pipeline-strategy.test.ts @@ -0,0 +1,205 @@ +/** + * Pipeline combo strategy — sequential chain (#6297). + * + * The 18th combo strategy: run targets IN ORDER, thread each step's output into the + * next step's input (each step has its own optional prompt), and return only the + * final step's response. Sequential counterpart to `fusion` (parallel + judge). + */ +import test from "node:test"; +import assert from "node:assert/strict"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-combo-pipeline-")); +process.env.DATA_DIR = TEST_DATA_DIR; +process.env.API_KEY_SECRET = process.env.API_KEY_SECRET || "combo-pipeline-test-secret"; + +const { handleComboChat } = await import("../../open-sse/services/combo.ts"); + +const noop = () => {}; +const log = { info: noop, warn: noop, debug: noop, error: noop }; + +type Body = Record; + +// Minimal OpenAI-chat Response-shaped object compatible with the engine's +// .ok + .clone().json() + .json() surface. +function okResponse(content: string): Response { + const body = JSON.stringify({ choices: [{ message: { role: "assistant", content } }] }); + return new Response(body, { status: 200, headers: { "Content-Type": "application/json" } }); +} + +function errResponse(status = 500): Response { + return new Response(JSON.stringify({ error: { message: "boom" } }), { + status, + headers: { "Content-Type": "application/json" }, + }); +} + +// steps: array of { model, prompt? }. +function pipelineCombo(steps: Array<{ model: string; prompt?: string }>) { + return { + name: "test-pipeline-combo", + strategy: "pipeline", + models: steps, + config: {}, + }; +} + +function lastUserContent(b: Body): string { + const msgs = b.messages as Array<{ role: string; content: string }>; + for (let i = msgs.length - 1; i >= 0; i--) { + if (msgs[i].role === "user") return msgs[i].content; + } + return ""; +} + +test("pipeline: 2 steps — step 1 gets the original input, step 2 gets step 1's output + its own prompt, only step 2's output is returned", async () => { + const seen: string[] = []; + const seenBodies: Body[] = []; + const handleSingleModel = async (b: Body, m: string) => { + seen.push(m); + seenBodies.push(b); + if (m === "p/a") return okResponse("OUT_A"); + return okResponse("FINAL_B"); + }; + + const res = await handleComboChat({ + body: { + messages: [{ role: "user", content: "hi" }], + stream: true, + tools: [{ name: "x" }], + }, + combo: pipelineCombo([{ model: "p/a" }, { model: "p/b", prompt: "SUMMARIZE" }]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + + // Exactly 2 calls, in order. + assert.deepEqual(seen, ["p/a", "p/b"]); + + // Step 1 sees the original user request. + assert.equal(lastUserContent(seenBodies[0]), "hi"); + // Step 1 is intermediate → non-streaming, tools stripped. + assert.equal(seenBodies[0].stream, false); + assert.equal(seenBodies[0].tools, undefined); + + // Step 2 receives step 1's output as its user input + its own prompt as a system turn. + assert.equal(lastUserContent(seenBodies[1]), "OUT_A"); + const step2Msgs = seenBodies[1].messages as Array<{ role: string; content: string }>; + assert.ok( + step2Msgs.some((m) => m.role === "system" && m.content === "SUMMARIZE"), + "step 2 should carry its own prompt as a system instruction" + ); + // Final step keeps the client's original stream flag + tools. + assert.equal(seenBodies[1].stream, true); + assert.ok(Array.isArray(seenBodies[1].tools)); + + // Only the final step's output is returned. + assert.equal(res.status, 200); + const json = (await res.json()) as { choices: Array<{ message: { content: string } }> }; + assert.equal(json.choices[0].message.content, "FINAL_B"); +}); + +test("pipeline: 3-step chain threads output → input correctly", async () => { + const seen: string[] = []; + const seenBodies: Body[] = []; + const outputs: Record = { "p/1": "o1", "p/2": "o2", "p/3": "o3" }; + const handleSingleModel = async (b: Body, m: string) => { + seen.push(m); + seenBodies.push(b); + return okResponse(outputs[m]); + }; + + const res = await handleComboChat({ + body: { messages: [{ role: "user", content: "start" }] }, + combo: pipelineCombo([{ model: "p/1" }, { model: "p/2" }, { model: "p/3" }]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + + assert.deepEqual(seen, ["p/1", "p/2", "p/3"]); + assert.equal(lastUserContent(seenBodies[0]), "start"); // original + assert.equal(lastUserContent(seenBodies[1]), "o1"); // step 1 output + assert.equal(lastUserContent(seenBodies[2]), "o2"); // step 2 output + + const json = (await res.json()) as { choices: Array<{ message: { content: string } }> }; + assert.equal(json.choices[0].message.content, "o3"); // final step output +}); + +test("pipeline: a failing middle step surfaces an error and short-circuits the chain", async () => { + const seen: string[] = []; + const handleSingleModel = async (_b: Body, m: string) => { + seen.push(m); + if (m === "p/a") return okResponse("OK"); + if (m === "p/bad") return errResponse(500); + return okResponse("SHOULD_NOT_RUN"); + }; + + const res = await handleComboChat({ + body: { messages: [{ role: "user", content: "go" }] }, + combo: pipelineCombo([{ model: "p/a" }, { model: "p/bad" }, { model: "p/c" }]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + + // Failure is surfaced (not a silent pass), and the downstream step never runs. + assert.equal(res.status, 500); + assert.ok(!seen.includes("p/c"), "the step after the failure must not execute"); + + // Error body must not leak a stack trace. + const body = (await res.json()) as { error?: { message?: string } }; + const msg = body.error?.message ?? ""; + assert.ok(!msg.includes("at /"), "error response must not leak a stack trace"); +}); + +test("pipeline: an intermediate step that returns empty output fails the pipeline", async () => { + const seen: string[] = []; + const handleSingleModel = async (_b: Body, m: string) => { + seen.push(m); + if (m === "p/empty") return okResponse(""); + return okResponse("FINAL"); + }; + + const res = await handleComboChat({ + body: { messages: [{ role: "user", content: "go" }] }, + combo: pipelineCombo([{ model: "p/empty" }, { model: "p/final" }]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + + assert.equal(res.status, 502); + assert.ok(!seen.includes("p/final"), "the final step must not run after an empty intermediate"); +}); + +test("pipeline: a single-step pipeline runs the one model directly and streams through", async () => { + const seen: string[] = []; + const seenBodies: Body[] = []; + const handleSingleModel = async (b: Body, m: string) => { + seen.push(m); + seenBodies.push(b); + return okResponse("solo"); + }; + + const res = await handleComboChat({ + body: { messages: [{ role: "user", content: "hi" }], stream: true }, + combo: pipelineCombo([{ model: "p/only" }]), + handleSingleModel, + log, + settings: {}, + allCombos: [], + }); + + assert.deepEqual(seen, ["p/only"]); + // Single step is the final step → keeps the client's stream flag. + assert.equal(seenBodies[0].stream, true); + assert.equal(res.status, 200); +});