feat(combo): sequential 'pipeline' combo strategy (#6297) (#6396)

sequential 'pipeline' combo strategy (#6297) (net +1/-0, tests OK). Integrated into release/v3.8.46.

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)

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<string, unknown>;
+          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),

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<string, unknown>;
+
+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<Response> {
+  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");
+}

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;

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,
 });
 

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<string, unknown>;
+
+// 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<string, string> = { "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);
+});