feat: advance the fallback chain after inference failures (#6)

README.md

@@ -22,6 +22,8 @@ v2-native.
   a prompt is a state-changing action.
 - **Cooldown.** A model whose switch fails is skipped for `cooldownMs`, so a
   broken target is not retried on every failure.
+- **Forward progress.** Each failure advances along your chain instead of
+  re-selecting the entry that just failed, so later providers are reachable.
 
 ## Requirements
 
@@ -88,40 +90,67 @@ plugin entry in your OpenCode config:
 }
 ```
 
-| Option           | Type       | Default | Description                                     |
-| ---------------- | ---------- | ------- | ----------------------------------------------- |
-| `fallbackModels` | `string[]` | `[]`    | Ordered `provider/model` chain, tried in order. |
-| `maxAttempts`    | `number`   | `2`     | Failovers per session before giving up.         |
-| `cooldownMs`     | `number`   | `60000` | Skip a model this long after a failed switch.   |
-| `autoRetry`      | `boolean`  | `false` | Replay the last user message after switching.   |
-| `notify`         | `boolean`  | `true`  | Log each failover to stderr.                    |
-| `logFile`        | `string`   | `""`    | Also append log lines to this file.             |
+| Option           | Type       | Default | Description                                       |
+| ---------------- | ---------- | ------- | ------------------------------------------------- |
+| `fallbackModels` | `string[]` | `[]`    | Ordered `provider/model` chain, tried in order.   |
+| `maxAttempts`    | `number`   | `2`     | Failovers per failure sequence before giving up.  |
+| `cooldownMs`     | `number`   | `60000` | Skip a model this long after a failed switch.     |
+| `autoRetry`      | `boolean`  | `false` | Replay the last user message after switching.     |
+| `notify`         | `boolean`  | `true`  | Log each failover to stderr.                      |
+| `logFile`        | `string`   | `""`    | Also append log lines to this file.               |
 
 A `provider/model` string is split on the first `/`. Entries without a `/` are
 dropped from the chain.
 
 ## How it works
 
-The plugin subscribes to the v2 event stream and reacts to one event type:
+The plugin subscribes to the v2 event stream:
 
 ```js
 ctx.event.subscribe({ signal })
-// session.execution.failed
-//   -> { data: { sessionID, error: { type, message } } }
+// session.execution.failed        -> { data: { sessionID, error: { type, message } } }
+// session.execution.succeeded     -> { data: { sessionID } }
+// session.execution.interrupted   -> { data: { sessionID } }
+// session.model.selected          -> { data: { sessionID, model: { providerID, id } } }
+// session.step.started            -> { data: { sessionID, model: { providerID, id } } }
+// session.deleted                 -> { data: { sessionID } }
 ```
 
-On a retryable failure it:
+On a retryable `session.execution.failed` it:
 
 1. reads the models that actually resolve with `ctx.model.list()`;
-2. picks the first chain entry that resolves and is not cooling down;
+2. picks the first chain entry that resolves, is not cooling down, is not the
+   model the session is already on, and has not been tried in this failure
+   sequence;
 3. moves the session with `ctx.session.switchModel({ sessionID, model })`.
 
 If `autoRetry` is on, it then reads the transcript with
 `ctx.session.context({ sessionID })` and replays the last user message with
 `ctx.session.prompt({ sessionID, text })`.
 
-The `id`/`setup` plugin shape, the `plugins` config key, and the four methods
-above are the whole v2 surface it uses.
+### Progression and reset
+
+`maxAttempts` counts failovers within one failure sequence, not per session
+lifetime, and each completed switch is remembered so the next failure moves to
+the entry after it. That is what lets a session reach a later provider: without
+it every failure re-selected the first available entry.
+
+- A failed `switchModel` is a provider problem, not a failed inference: the
+  model is cooled down for `cooldownMs`, but no attempt is used and the chain
+  position does not move.
+- A failed *inference* does not cool the model down. It advances this session's
+  position instead, so a sibling session can still land on that model: the
+  failure is the model's problem for this request, and a process-wide cooldown
+  would take a working model away from everyone.
+- `session.execution.succeeded` and `session.execution.interrupted` end the
+  sequence, so the next failure starts at the top of the chain again.
+- `session.model.selected` and `session.step.started` record the model the
+  session is on, so a failure never selects that model again in the same
+  sequence.
+- `session.deleted` drops the session's state.
+
+The `id`/`setup` plugin shape, the `plugins` config key, and the methods above
+are the whole v2 surface it uses.
 
 ## Observing it
 

index.js

@@ -30,7 +30,9 @@ export default {
     const logFile =
       typeof opt.logFile === "string" ? opt.logFile : process.env.MODEL_FALLBACK_LOG || "";
 
-    const attempts = new Map(); // sessionID -> failovers used
+    const attempts = new Map(); // sessionID -> failovers used in this failure sequence
+    const attempted = new Map(); // sessionID -> Set of refs already tried in this sequence
+    const current = new Map(); // sessionID -> "provider/model" selected for the session
     const busy = new Set(); // sessionID -> failover in flight
     const cooldown = new Map(); // "provider/model" -> epoch ms it is usable again
 
@@ -86,8 +88,22 @@ export default {
 
       const set = await availableRefs();
       const now = Date.now();
-      log(`availability=${set.size} models; looking for next in chain`);
-      const next = chain.find((ref) => set.has(ref) && (cooldown.get(ref) || 0) <= now);
+      const tried = attempted.get(sessionID) || new Set();
+      const selected = current.get(sessionID);
+      // The model that just failed must not be picked again even if this
+      // sequence never switched to it (for example the first failure of a
+      // session that started on chain[0]).
+      if (selected) tried.add(selected);
+      log(
+        `availability=${set.size} models; already tried=${tried.size}; looking for next in chain`,
+      );
+      // Skip what the session is already on and what this failure sequence
+      // has already moved to, otherwise the first available entry is picked
+      // forever and later providers are never reached.
+      const next = chain.find(
+        (ref) =>
+          set.has(ref) && (cooldown.get(ref) || 0) <= now && !tried.has(ref) && ref !== selected,
+      );
       if (!next) {
         log("no available model left in chain");
         return;
@@ -96,22 +112,31 @@ export default {
       const ref = splitRef(next);
       if (!ref) return;
 
-      attempts.set(sessionID, used + 1);
       try {
         await ctx.session.switchModel({
           sessionID,
           model: { providerID: ref.providerID, id: ref.modelID },
         });
-        log(
-          `session ${sessionID} moved to ${next} (attempt ${used + 1}/${maxAttempts}) after ` +
-            `${error?.type || "error"}`,
-        );
       } catch (e) {
+        // A failed switch is a provider problem, not a failed inference: cool
+        // the model down and leave the attempt budget and progression alone,
+        // so the next failure can pick the entry after it.
         cooldown.set(next, Date.now() + cooldownMs);
         log(`switchModel to ${next} failed: ${e?.message}`);
         return;
       }
 
+      tried.add(next);
+      attempted.set(sessionID, tried);
+      attempts.set(sessionID, used + 1);
+      // Record the new model immediately: the session.model.selected event
+      // that follows may arrive after the next failure.
+      current.set(sessionID, next);
+      log(
+        `session ${sessionID} moved to ${next} (attempt ${used + 1}/${maxAttempts}) after ` +
+          `${error?.type || "error"}`,
+      );
+
       if (!autoRetry) return;
       try {
         const transcript = await ctx.session.context({ sessionID });
@@ -126,26 +151,50 @@ export default {
     void (async () => {
       try {
         for await (const event of ctx.event.subscribe({ signal: controller.signal })) {
-          if (event?.type !== "session.execution.failed") continue;
-          const data = event.data || {};
-          const sessionID = data.sessionID;
-          const error = data.error || {};
+          const type = event?.type;
+          const sid = event?.data?.sessionID;
+
+          // A completed or user-interrupted run ends the failure sequence, so
+          // the next failure starts at the top of the chain again instead of
+          // inheriting another sequence's progress.
+          if (type === "session.execution.succeeded" || type === "session.execution.interrupted") {
+            if (!sid) continue;
+            attempted.delete(sid);
+            attempts.delete(sid);
+            continue;
+          }
+          if (type === "session.model.selected" || type === "session.step.started") {
+            const model = event.data?.model || {};
+            const id = model.id || model.modelID;
+            if (sid && model.providerID && id) current.set(sid, `${model.providerID}/${id}`);
+            continue;
+          }
+          if (type === "session.deleted") {
+            if (!sid) continue;
+            attempted.delete(sid);
+            attempts.delete(sid);
+            current.delete(sid);
+            continue;
+          }
+          if (type !== "session.execution.failed") continue;
+
+          const error = event.data?.error || {};
           log(
-            `failed event: session=${sessionID} error=${error.type} msg=${String(error.message).slice(0, 120)}`,
+            `failed event: session=${sid} error=${error.type} msg=${String(error.message).slice(0, 120)}`,
           );
-          if (!sessionID || busy.has(sessionID)) continue;
+          if (!sid || busy.has(sid)) continue;
 
           const haystack = `${error.type || ""} ${error.message || ""}`;
           if (!RETRYABLE.test(haystack)) {
             log(`not retryable, ignoring: ${haystack.slice(0, 120)}`);
             continue;
           }
 
-          busy.add(sessionID);
+          busy.add(sid);
           try {
-            await failover(sessionID, error);
+            await failover(sid, error);
           } finally {
-            busy.delete(sessionID);
+            busy.delete(sid);
           }
         }
       } catch {}

test/failover.test.mjs

@@ -77,6 +77,14 @@ function failedEvent(sessionID, error) {
   return { type: "session.execution.failed", data: { sessionID, error } };
 }
 
+function succeededEvent(sessionID) {
+  return { type: "session.execution.succeeded", data: { sessionID } };
+}
+
+function modelSelectedEvent(sessionID, model) {
+  return { type: "session.model.selected", data: { sessionID, model } };
+}
+
 async function start(t, options, cfg = {}) {
   const stream = createEventStream();
   const calls = { switchModel: [], prompt: [], context: 0, list: 0 };
@@ -224,3 +232,134 @@ test("replays the last user message when autoRetry is on", async (t) => {
 
   assert.deepEqual(calls.prompt[0], { sessionID: "s1", text: "second prompt" });
 });
+
+test("advances to the next chain entry after each inference failure", async (t) => {
+  const logFile = mkLog();
+  const { calls, push } = await start(
+    t,
+    { fallbackModels: ["a/one", "a/two", "b/three"], maxAttempts: 5, notify: false, logFile },
+    { models: ["a/one", "a/two", "b/three"] },
+  );
+
+  push(failedEvent("s1", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 1);
+  push(failedEvent("s1", { type: "provider.overloaded", message: "503" }));
+  await waitFor(() => calls.switchModel.length === 2);
+  push(failedEvent("s1", { type: "provider.timeout", message: "timed out" }));
+  await waitFor(() => calls.switchModel.length === 3);
+
+  assert.deepEqual(
+    calls.switchModel.map((c) => `${c.model.providerID}/${c.model.id}`),
+    ["a/one", "a/two", "b/three"],
+  );
+});
+
+test("starts a fresh sequence after a successful execution", async (t) => {
+  const logFile = mkLog();
+  const { calls, push } = await start(
+    t,
+    { fallbackModels: ["a/one", "a/two"], maxAttempts: 1, notify: false, logFile },
+    { models: ["a/one", "a/two"] },
+  );
+
+  push(modelSelectedEvent("s1", { providerID: "a", id: "one" }));
+  await delay(20);
+
+  push(failedEvent("s1", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 1);
+  assert.equal(calls.switchModel[0].model.id, "two");
+
+  push(succeededEvent("s1"));
+  await delay(30);
+
+  // New sequence, fresh budget: the chain restarts, so the entry it is not
+  // currently on (`a/one`) is eligible again.
+  push(failedEvent("s1", { type: "provider.timeout", message: "timed out" }));
+  await waitFor(() => calls.switchModel.length === 2);
+
+  assert.equal(calls.switchModel[1].model.id, "one");
+  assert.ok(!readLog(logFile).includes("max attempts reached"));
+});
+
+test("keeps attempt state per session", async (t) => {
+  const logFile = mkLog();
+  const { calls, push } = await start(
+    t,
+    { fallbackModels: ["a/one"], maxAttempts: 1, notify: false, logFile },
+    { models: ["a/one"] },
+  );
+
+  push(failedEvent("s1", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 1);
+  push(failedEvent("s2", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 2);
+
+  assert.deepEqual(
+    calls.switchModel.map((c) => c.sessionID),
+    ["s1", "s2"],
+  );
+});
+
+test("skips the model already selected for the session", async (t) => {
+  const logFile = mkLog();
+  const { calls, push } = await start(
+    t,
+    { fallbackModels: ["a/one", "a/two"], notify: false, logFile },
+    { models: ["a/one", "a/two"] },
+  );
+
+  push(modelSelectedEvent("s1", { providerID: "a", id: "one" }));
+  await delay(20);
+
+  push(failedEvent("s1", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 1);
+
+  assert.equal(calls.switchModel[0].model.id, "two");
+});
+
+test("a failed switch cools the model down without using an attempt", async (t) => {
+  const logFile = mkLog();
+  const { calls, push } = await start(
+    t,
+    { fallbackModels: ["a/one", "a/two"], maxAttempts: 2, notify: false, logFile },
+    {
+      models: ["a/one", "a/two"],
+      onSwitch: (args) => {
+        if (args.model.id === "one") throw new Error("switch rejected");
+      },
+    },
+  );
+
+  push(failedEvent("s1", { type: "provider.timeout", message: "timed out" }));
+  await waitFor(() => readLog(logFile).includes("switchModel to a/one failed"));
+
+  push(failedEvent("s1", { type: "provider.timeout", message: "timed out" }));
+  await waitFor(() => calls.switchModel.length === 2);
+
+  assert.equal(calls.switchModel[1].model.id, "two");
+  assert.ok(readLog(logFile).includes("(attempt 1/2)"));
+});
+
+test("never returns to the model that just failed", async (t) => {
+  const logFile = mkLog();
+  const { calls, push } = await start(
+    t,
+    { fallbackModels: ["a/one", "a/two", "b/three"], maxAttempts: 5, notify: false, logFile },
+    { models: ["a/one", "a/two", "b/three"] },
+  );
+
+  push(modelSelectedEvent("s1", { providerID: "a", id: "one" }));
+  await delay(20);
+  push(failedEvent("s1", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 1);
+
+  push(modelSelectedEvent("s1", { providerID: "a", id: "two" }));
+  await delay(20);
+  push(failedEvent("s1", { type: "provider.rateLimit", message: "429" }));
+  await waitFor(() => calls.switchModel.length === 2);
+
+  assert.deepEqual(
+    calls.switchModel.map((c) => c.model.id),
+    ["two", "three"],
+  );
+});