From c2fd4c279ef7b7701dbee52a9938d856910be17a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 31 Oct 2026 09:58:06 -0700 Subject: [PATCH 1/2] fix(hub): retry model errors before ending the turn Every inference from the compacting director retries an overloaded, slow or rate-limited model on a 2/8/30 s schedule, bounded to 5 minutes and one retry for a stalled stream. Once it gives up, a designer or coder run fails so the coordinator gets the ended notice; the coordinator still replies with the error. replyOnRunaways becomes replyOnFailure. Closes CL-11871 --- apps/hub/src/workflows/compaction.test.ts | 122 ++++++++++++++++++ apps/hub/src/workflows/compaction.ts | 76 +++++++++-- apps/hub/src/workflows/entries/coordinator.ts | 2 +- 3 files changed, 191 insertions(+), 9 deletions(-) create mode 100644 apps/hub/src/workflows/compaction.test.ts diff --git a/apps/hub/src/workflows/compaction.test.ts b/apps/hub/src/workflows/compaction.test.ts new file mode 100644 index 00000000..0d347651 --- /dev/null +++ b/apps/hub/src/workflows/compaction.test.ts @@ -0,0 +1,122 @@ +import { expect, test } from "bun:test"; +import { createCapabilities } from "@intx/inference"; +import type { + InferenceError, + ReactorAction, + ReactorDirector, + ReactorInboundEvent, + ReactorState, + RetryPolicy, +} from "@intx/types/runtime"; +import { REPETITION_ABORT } from "../inference/repetition.ts"; +import { compactingDirector } from "./compaction.ts"; + +const director = (replyOnFailure: boolean) => + compactingDirector("@corbits-builder/test").factory( + { contextWindow: 200_000, compactAt: 0.8, replyOnFailure }, + {} as never, + { systemPrompt: "", toolDefinitions: [], compactorNames: [] }, + ) as ReactorDirector; + +const state: ReactorState = { + turns: [], + activeForks: [], + pendingOperations: [], + activeGates: [], + tokenUsage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, + lastCycleUsage: null, + lastCycleSource: null, + sessionId: "session", +}; + +const decide = async (subject: ReactorDirector, event: ReactorInboundEvent) => { + const actions = await subject.decide(event, state, createCapabilities()); + return (Array.isArray(actions) ? actions : [actions]) as ReactorAction[]; +}; + +const failed = (error: InferenceError): ReactorInboundEvent => ({ + type: "inference.error", + error, + partial: { text: "" }, +}); + +const retryPolicy = (actions: ReactorAction[]): RetryPolicy => { + const infer = actions.find((action) => action.type === "infer"); + const policy = + infer?.type === "infer" ? infer.options?.retryPolicy : undefined; + if (!policy) throw new Error("infer_without_retry_policy"); + return policy; +}; + +const overloaded: InferenceError = { + category: "retryable", + message: "overloaded", + statusCode: 529, +}; + +test("every inference retries a busy model on a bounded backoff", async () => { + const subject = director(false); + const policies = [ + retryPolicy( + await decide(subject, { + type: "reactor.gate.cleared", + gateId: "gate", + reason: "resolved", + }), + ), + retryPolicy( + await decide( + subject, + failed({ + category: "protocol_mismatch", + message: `${REPETITION_ABORT} repeated`, + }), + ), + ), + ]; + const timeout: InferenceError = { category: "timeout", message: "stall" }; + const quota = (retryAfterMs: number): InferenceError => ({ + category: "quota_exhausted", + message: "rate limited", + retryAfterMs, + }); + for (const policy of policies) { + const decision = (error: InferenceError, attempt: number, elapsedMs = 0) => + policy({ error, attempt, elapsedMs }); + expect( + await Promise.all([ + decision(overloaded, 1), + decision(overloaded, 2), + decision(overloaded, 3), + decision(overloaded, 4), + decision(overloaded, 2, 6 * 60_000), + decision(timeout, 1), + decision(timeout, 2), + decision(quota(45_000), 1), + decision(quota(90_000), 1), + decision({ category: "fatal", message: "bad request" }, 1), + ]), + ).toEqual([ + { kind: "retry", delayMs: 2_000 }, + { kind: "retry", delayMs: 8_000 }, + { kind: "retry", delayMs: 30_000 }, + { kind: "abort" }, + { kind: "abort" }, + { kind: "retry", delayMs: 2_000 }, + { kind: "abort" }, + { kind: "retry", delayMs: 45_000 }, + { kind: "abort" }, + { kind: "abort" }, + ]); + } +}); + +test("a model error that outlasts the retries fails a specialist and is the coordinator's reply", async () => { + await expect(decide(director(false), failed(overloaded))).rejects.toThrow( + "model_unavailable", + ); + const reply = (await decide(director(true), failed(overloaded))).find( + (action) => action.type === "reply", + ); + expect(reply?.type === "reply" ? reply.content : "").toContain("overloaded"); +}); diff --git a/apps/hub/src/workflows/compaction.ts b/apps/hub/src/workflows/compaction.ts index a8f9eafe..dafbbdad 100644 --- a/apps/hub/src/workflows/compaction.ts +++ b/apps/hub/src/workflows/compaction.ts @@ -5,7 +5,10 @@ import type { Compactor, ContentBlock, ConversationTurn, + InferenceError, + ReactorAction, ReactorDirector, + RetryPolicy, } from "@intx/types/runtime"; import { REPETITION_ABORT } from "../inference/repetition.ts"; import { CHECK_DESIGN_TOOL } from "../builds/design-check.ts"; @@ -235,6 +238,60 @@ const MAX_EMPTY_TURNS = 3; // After this many held replies the reply goes out as it is, so a model that cannot pass is not held forever. const MAX_UNCHECKED_REPLIES = 3; +const TRANSIENT_ERRORS: ReadonlySet = new Set([ + "retryable", + "timeout", + "quota_exhausted", +]); +const RETRY_DELAYS_MS = [2_000, 8_000, 30_000]; +const TIMEOUT_RETRIES = 1; +// Each timed-out attempt already waited minutes, and every model source the reactor fails over to +// starts its own retries, so one call stops retrying after this long. +const RETRY_BUDGET_MS = 5 * 60_000; +const MAX_RETRY_AFTER_MS = 60_000; +/** Retries an overloaded, slow or rate-limited model on a growing schedule; a rate limit's retry-after wins when longer. */ +const backoffRetryPolicy: RetryPolicy = ({ error, attempt, elapsedMs }) => { + const delayMs = RETRY_DELAYS_MS[attempt - 1]; + const retryAfterMs = error.retryAfterMs ?? 0; + if ( + !TRANSIENT_ERRORS.has(error.category) || + delayMs === undefined || + (error.category === "timeout" && attempt > TIMEOUT_RETRIES) || + elapsedMs > RETRY_BUDGET_MS || + retryAfterMs > MAX_RETRY_AFTER_MS + ) + return { kind: "abort" }; + return { kind: "retry", delayMs: Math.max(delayMs, retryAfterMs) }; +}; +const withBackoff = (action: ReactorAction): ReactorAction => + action.type === "infer" + ? { + ...action, + options: { ...action.options, retryPolicy: backoffRetryPolicy }, + } + : action; + +/** Every inference retries on backoffRetryPolicy; once it gives up, with failRun the run fails. */ +const withRetryPolicy = ( + director: ReactorDirector, + failRun: boolean, +): ReactorDirector => ({ + decide: async (event, state, caps) => { + if ( + failRun && + event.type === "inference.error" && + TRANSIENT_ERRORS.has(event.error.category) + ) + throw new Error( + `model_unavailable: ${event.error.category}: ${event.error.message}`, + ); + const actions = await director.decide(event, state, caps); + return Array.isArray(actions) + ? actions.map(withBackoff) + : withBackoff(actions); + }, +}); + // Shell edits the done gate counts as changes: redirection into a file (not 2>&1 or >/dev/null), // tee, sed -i, mv, cp, rm and touch. const SHELL_WRITE = @@ -312,9 +369,11 @@ const unfiled = (turns: readonly ConversationTurn[]) => * too long. A turn cut off with no text and no tool call is continued instead of parked, and a * turn that only repeats one sentence is dropped and sampled again, as is a stream the source adapter * cut for repeating one passage. The third such turn in one task fails the run, or with - * replyOnRunaways is the reply. With checkBeforeDone, a final reply waits for a passing check-app - * after the last file change; with fileBeforeDone, it waits for a filed document. The - * model-specific numbers come from the agent definition's director config. + * replyOnFailure is the reply. With checkBeforeDone, a final reply waits for a passing check-app + * after the last file change; with fileBeforeDone, it waits for a filed document. Every inference + * retries an overloaded, slow or rate-limited model a few times; once that gives up, the run + * fails, or with replyOnFailure the error is the reply. The model-specific numbers come from the + * agent definition's director config. */ export const compactingDirector = (packageName: string) => defineDirector<{ @@ -324,8 +383,8 @@ export const compactingDirector = (packageName: string) => checkBeforeDone?: boolean; // Whether a final reply waits for a write_artifact that filed the task's document. fileBeforeDone?: boolean; - // Whether the third runaway in one task is the reply rather than a failed run. - replyOnRunaways?: boolean; + // Whether a runaway or a model failure is the reply rather than a failed run. + replyOnFailure?: boolean; }>({ id: `${packageName}/compacting`, configSchema: type({ @@ -333,7 +392,7 @@ export const compactingDirector = (packageName: string) => compactAt: "0 < number < 1", "checkBeforeDone?": "boolean", "fileBeforeDone?": "boolean", - "replyOnRunaways?": "boolean", + "replyOnFailure?": "boolean", }), factory: (config, env, agent) => { const inner = defaultDirectorFactory({}, env, agent) as ReactorDirector; @@ -347,14 +406,14 @@ export const compactingDirector = (packageName: string) => // emits reactor.error and ends the run failed, which reaches the coordinator as a specialist // ended notice instead of a reply that reads as the specialist's result. const runaways = (caps: Parameters[2]) => { - if (config.replyOnRunaways) + if (config.replyOnFailure) return [ caps.checkpoint("degenerate-turns"), caps.reply(DEGENERATE_TURNS), ]; throw new Error(DEGENERATE_TURNS); }; - return { + const director: ReactorDirector = { decide: async (event, state, caps) => { if (event.type === "message.received") degenerateTurns = 0; // The source adapter cut a stream that repeated one passage; nothing of the turn was kept. @@ -462,5 +521,6 @@ export const compactingDirector = (packageName: string) => return actions; }, }; + return withRetryPolicy(director, !config.replyOnFailure); }, }); diff --git a/apps/hub/src/workflows/entries/coordinator.ts b/apps/hub/src/workflows/entries/coordinator.ts index 4f4919f7..d0077150 100644 --- a/apps/hub/src/workflows/entries/coordinator.ts +++ b/apps/hub/src/workflows/entries/coordinator.ts @@ -19,7 +19,7 @@ const agent = createCoordinatorAgent({ contextWindow, compactAt: 0.6, // The coordinator's run is the project's; nothing starts a failed one again. - replyOnRunaways: true, + replyOnFailure: true, }), id: "coordinator", capabilities: [], From 0eed1f8a455f9061725e4a2c92634364b89dbf4a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 31 Oct 2026 10:06:26 -0700 Subject: [PATCH 2/2] fix(hub): use a retry-after only when the error carries one --- apps/hub/src/workflows/compaction.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/apps/hub/src/workflows/compaction.ts b/apps/hub/src/workflows/compaction.ts index dafbbdad..b0cbbb77 100644 --- a/apps/hub/src/workflows/compaction.ts +++ b/apps/hub/src/workflows/compaction.ts @@ -252,15 +252,16 @@ const MAX_RETRY_AFTER_MS = 60_000; /** Retries an overloaded, slow or rate-limited model on a growing schedule; a rate limit's retry-after wins when longer. */ const backoffRetryPolicy: RetryPolicy = ({ error, attempt, elapsedMs }) => { const delayMs = RETRY_DELAYS_MS[attempt - 1]; - const retryAfterMs = error.retryAfterMs ?? 0; + const { retryAfterMs } = error; if ( !TRANSIENT_ERRORS.has(error.category) || delayMs === undefined || (error.category === "timeout" && attempt > TIMEOUT_RETRIES) || elapsedMs > RETRY_BUDGET_MS || - retryAfterMs > MAX_RETRY_AFTER_MS + (retryAfterMs !== undefined && retryAfterMs > MAX_RETRY_AFTER_MS) ) return { kind: "abort" }; + if (retryAfterMs === undefined) return { kind: "retry", delayMs }; return { kind: "retry", delayMs: Math.max(delayMs, retryAfterMs) }; }; const withBackoff = (action: ReactorAction): ReactorAction =>