fix(hub): retry model errors before ending the turn

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");
+});

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,61 @@ 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<InferenceError["category"]> = 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;
+  if (
+    !TRANSIENT_ERRORS.has(error.category) ||
+    delayMs === undefined ||
+    (error.category === "timeout" && attempt > TIMEOUT_RETRIES) ||
+    elapsedMs > RETRY_BUDGET_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 =>
+  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 +370,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,16 +384,16 @@ 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({
       contextWindow: "number.integer > 0",
       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 +407,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<ReactorDirector["decide"]>[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 +522,6 @@ export const compactingDirector = (packageName: string) =>
           return actions;
         },
       };
+      return withRetryPolicy(director, !config.replyOnFailure);
     },
   });

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: [],