diff --git a/README.md b/README.md index 55917ba..82598e1 100644 --- a/README.md +++ b/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 diff --git a/index.js b/index.js index 8abee13..e4a03d5 100644 --- a/index.js +++ b/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,14 +151,38 @@ 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)) { @@ -141,11 +190,11 @@ export default { continue; } - busy.add(sessionID); + busy.add(sid); try { - await failover(sessionID, error); + await failover(sid, error); } finally { - busy.delete(sessionID); + busy.delete(sid); } } } catch {} diff --git a/test/failover.test.mjs b/test/failover.test.mjs index b1457c4..8dbf5fe 100644 --- a/test/failover.test.mjs +++ b/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"], + ); +});