fix(combo): retry transient errors in pipeline strategy (#7794)

* chore: auto-sync from VM - 2026-07-10T16:29:57Z

* fix(combo): retry transient errors in pipeline strategy

Pipeline combo strategy (sequential chain) was hard-failing on ANY
intermediate step error, including transient ones like 429 rate-limit
and 503 service-unavailable. The combo config already exposes
maxRetries and retryDelayMs, but handlePipelineChat() ignored them
entirely — a single 429 from the first provider would kill the whole
pipeline without trying the remaining steps.

Now intermediate steps that fail with a transient HTTP status
(429, 502, 503, 504) are retried up to maxRetries times with
retryDelayMs delay, mirroring the retry behaviour already used by
priority/weighted strategies. Non-transient errors (400, 401, 403,
404) still fail immediately.

Changes:
- open-sse/services/pipeline.ts: add maxRetries/retryDelayMs params,
  retry loop for transient statuses
- open-sse/services/combo.ts: wire combo.config.maxRetries and
  combo.config.retryDelayMs to handlePipelineChat()
- tests/unit/combo-pipeline.test.ts: 7 tests covering retry success,
  retry exhaustion, non-transient skip, final-step passthrough,
  backward compat (maxRetries=0)

* chore: revert unrelated package-lock.json scope creep from PR #7794

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

* docs(changelog): add fragment for #7794 pipeline transient retry

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

---------

Co-authored-by: Andrian B. <andrewbalanesq@gmail.com>
Co-authored-by: Diego Rodrigues de Sa e Souza <diegosouza.pw@gmail.com>
Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

changelog.d/fixes/7794-pipeline-transient-retry.md

@@ -0,0 +1 @@
+- fix(combo): retry intermediate pipeline-strategy steps on transient upstream errors (429/502/503/504), gated by `combo.config.maxRetries`/`retryDelayMs`, without retrying terminal auth/request errors (#7794 — thanks @AndrianBalanescu)

open-sse/services/combo.ts

@@ -927,6 +927,8 @@ export async function handleComboChat({
       handleSingleModel: handleSingleModelWithTimeout,
       log,
       comboName: combo.name,
+      maxRetries: config.maxRetries ?? 0,
+      retryDelayMs: resolveDelayMs(config.retryDelayMs, 1000),
     });
   }
 

open-sse/services/pipeline.ts

@@ -30,6 +30,14 @@
  * 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.
+ *
+ * ── Transient retry ──────────────────────────────────────────────────────────
+ * Intermediate steps that fail with a transient HTTP status (429, 502, 503, 504)
+ * are retried up to `maxRetries` times with `retryDelayMs` delay between attempts.
+ * This mirrors the retry behaviour already used by the priority/weighted strategies
+ * and respects the same `combo.config.maxRetries` / `combo.config.retryDelayMs`
+ * fields. Non-transient errors (400, 401, 403, 404, …) fail immediately — retrying
+ * a bad-request or auth error wastes quota and will never succeed.
  */
 import { errorResponse } from "../utils/error.ts";
 import type { ComboLogger, HandleSingleModel } from "./combo/types.ts";
@@ -112,8 +120,19 @@ export type HandlePipelineChatOptions = {
   handleSingleModel: HandleSingleModel;
   log: ComboLogger;
   comboName?: string;
+  /** Max retry attempts on transient errors (429/502/503/504). Default: 0 (no retry). */
+  maxRetries?: number;
+  /** Delay between retries in milliseconds. Default: 1000. */
+  retryDelayMs?: number;
 };
 
+/** HTTP statuses that are worth retrying (transient / capacity / rate-limit). */
+const TRANSIENT_STATUS = new Set([429, 502, 503, 504]);
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
 /**
  * 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.
@@ -124,6 +143,8 @@ export async function handlePipelineChat({
   handleSingleModel,
   log,
   comboName,
+  maxRetries = 0,
+  retryDelayMs = 1000,
 }: HandlePipelineChatOptions): Promise<Response> {
   const chain = (Array.isArray(steps) ? steps : []).filter((s) => s && s.model);
   if (chain.length === 0) {
@@ -153,13 +174,25 @@ export async function handlePipelineChat({
     if (!isFinal) stepBody = stripStreaming(stepBody);
 
     const t0 = Date.now();
-    const res = await handleSingleModel(stepBody, step.model);
+    let res = await handleSingleModel(stepBody, step.model);
 
     if (isFinal) {
       log.info("PIPELINE", `Final step ${step.model} responded (${Date.now() - t0}ms)`);
       return res;
     }
 
+    // Transient retry: if the intermediate step failed with a retryable status
+    // (429/502/503/504), retry the same step up to maxRetries times before
+    // giving up. Non-transient errors (400/401/403/404) fail immediately.
+    for (let attempt = 0; attempt < maxRetries && !res.ok && TRANSIENT_STATUS.has(res.status); attempt++) {
+      log.warn(
+        "PIPELINE",
+        `Step ${i + 1} (${step.model}) transient ${res.status}, retrying ${attempt + 1}/${maxRetries} in ${retryDelayMs}ms`
+      );
+      await sleep(retryDelayMs);
+      res = await handleSingleModel(stepBody, step.model);
+    }
+
     // 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) {

tests/unit/combo-pipeline.test.ts

@@ -0,0 +1,198 @@
+/**
+ * Combo Pipeline Strategy Tests
+ *
+ * Tests for open-sse/services/pipeline.ts — the sequential chain combo strategy.
+ * Focus: transient retry behaviour (429/502/503/504) added to prevent hard-fail
+ * on rate-limited or temporarily unavailable upstream providers.
+ */
+
+import { describe, it, mock } from "node:test";
+import assert from "node:assert/strict";
+
+import { handlePipelineChat, type PipelineStep } from "../../open-sse/services/pipeline.ts";
+
+// ---------------------------------------------------------------------------
+// Types & helpers
+// ---------------------------------------------------------------------------
+
+type Body = Record<string, unknown>;
+
+interface MockResponse {
+  ok: boolean;
+  status: number;
+  json: () => Promise<unknown>;
+  clone: () => MockResponse;
+}
+
+/** Build a successful OpenAI-shaped response with given text. */
+function okResponse(text: string): MockResponse {
+  const body = {
+    choices: [{ message: { role: "assistant", content: text } }],
+  };
+  return {
+    ok: true,
+    status: 200,
+    json: async () => body,
+    // handlePipelineChat calls res.clone().json() on intermediate steps
+    clone: () => okResponse(text),
+  };
+}
+
+/** Build a failed response with given status. */
+function failResponse(status: number): MockResponse {
+  const body = { error: { message: `HTTP ${status}` } };
+  return {
+    ok: false,
+    status,
+    json: async () => body,
+    clone: () => failResponse(status),
+  };
+}
+
+type HandlerFn = (body: Body, model: string) => Promise<MockResponse>;
+
+/** Build a mock handleSingleModel that returns responses in sequence. */
+function makeHandler(responses: MockResponse[], opts?: { loopLast?: boolean }): HandlerFn {
+  let call = 0;
+  return async (_body: Body, _model: string): Promise<MockResponse> => {
+    const idx = call++;
+    if (idx < responses.length) return responses[idx];
+    if (opts?.loopLast && responses.length > 0) return responses[responses.length - 1];
+    return okResponse("fallback");
+  };
+}
+
+// Minimal stub type — the real type is more complex but we only need (body, model) => Response
+// (kept for reference, not used as value)
+
+const noopLog = {
+  info: () => {},
+  warn: () => {},
+  error: () => {},
+  debug: () => {},
+};
+
+const STEPS: PipelineStep[] = [
+  { model: "provider-a/model-a" },
+  { model: "provider-b/model-b" },
+];
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+describe("handlePipelineChat — transient retry", () => {
+  it("succeeds when all steps return 200", async () => {
+    const handler = makeHandler([okResponse("step 1 output"), okResponse("step 2 output")]);
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      maxRetries: 2,
+    });
+    assert.equal(res.ok, true);
+  });
+
+  it("retries on 429 then succeeds", async () => {
+    // First call to step 1 → 429, second call to step 1 → 200, step 2 → 200
+    const handler = makeHandler([
+      failResponse(429),
+      okResponse("recovered"),
+      okResponse("final"),
+    ]);
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      maxRetries: 2,
+      retryDelayMs: 1, // fast for tests
+    });
+    assert.equal(res.ok, true);
+  });
+
+  it("retries on 503 then succeeds", async () => {
+    const handler = makeHandler([
+      failResponse(503),
+      failResponse(503),
+      okResponse("recovered after 2"),
+      okResponse("final"),
+    ]);
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      maxRetries: 2,
+      retryDelayMs: 1,
+    });
+    assert.equal(res.ok, true);
+  });
+
+  it("fails after exhausting retries on persistent 429", async () => {
+    // Step 1 always returns 429, even after maxRetries=2 (3 total attempts)
+    const handler = makeHandler([failResponse(429)], { loopLast: true });
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      maxRetries: 2,
+      retryDelayMs: 1,
+    });
+    assert.equal(res.ok, false);
+    assert.equal(res.status, 429);
+  });
+
+  it("fails immediately on 400 (non-transient, no retry)", async () => {
+    let callCount = 0;
+    const handler = async (): Promise<MockResponse> => {
+      callCount++;
+      return failResponse(400);
+    };
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      maxRetries: 5, // should NOT retry on 400
+      retryDelayMs: 1,
+    });
+    assert.equal(res.ok, false);
+    assert.equal(res.status, 400);
+    // Only 1 call — no retry on non-transient error
+    assert.equal(callCount, 1);
+  });
+
+  it("does NOT retry the final step", async () => {
+    // Step 1 → 200, final step → 502 (should be returned as-is, no retry)
+    const handler = makeHandler([okResponse("step 1"), failResponse(502)], { loopLast: true });
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      maxRetries: 3,
+      retryDelayMs: 1,
+    });
+    // Final step result is returned directly regardless of status
+    assert.equal(res.status, 502);
+    assert.equal(res.ok, false);
+  });
+});
+
+describe("handlePipelineChat — backward compat (no retry)", () => {
+  it("fails on transient error when maxRetries=0 (default)", async () => {
+    const handler = makeHandler([failResponse(429)], { loopLast: true });
+    const res = await handlePipelineChat({
+      body: { messages: [{ role: "user", content: "hi" }] },
+      steps: STEPS,
+      handleSingleModel: handler as unknown as never,
+      log: noopLog as never,
+      // maxRetries defaults to 0 — no retry
+    });
+    assert.equal(res.ok, false);
+    assert.equal(res.status, 429);
+  });
+});