Retry failed queued messages and expose manual recovery

apps/app/src/components/promptbox/banner/QueuedMessagesList.test.tsx

@@ -169,6 +169,41 @@ afterEach(() => {
 });
 
 describe("QueuedMessagesList", () => {
+  it.each([
+    { kind: "host-offline", hostName: "M4" },
+    { kind: "provisioning" },
+    { kind: "interaction" },
+    { kind: "turn-starting" },
+    { kind: "stopping" },
+  ] as const)("offers Send now for a failed $kind row", (waitingOn) => {
+    const onSend = vi.fn();
+    const { getByRole, getByText } = render(
+      <QueuedMessagesList
+        queuedMessages={[
+          makeThreadQueuedMessage({
+            id: "failed-row",
+            waitingOn,
+            failureReason: "Provider unavailable",
+          }),
+        ]}
+        sendDisabled={false}
+        actionDisabled={false}
+        processingMessageId={null}
+        processingAction={null}
+        onSend={onSend}
+        onReorder={noop}
+        onSetGroupBoundary={noop}
+        onEdit={noop}
+        onDelete={noop}
+      />,
+    );
+    expect(getByText("Provider unavailable")).toBeTruthy();
+    const button = getByRole("button", { name: "Send queued message 1 now" });
+    expect(button.hasAttribute("disabled")).toBe(false);
+    fireEvent.click(button);
+    expect(onSend).toHaveBeenCalledWith("failed-row");
+  });
+
   it("labels non-user senders and refreshes their names from the thread cache", async () => {
     const queryClient = new QueryClient();
     const messages = [

apps/app/src/components/promptbox/banner/QueuedMessagesList.tsx

@@ -803,7 +803,7 @@ const QueuedMessageRow = memo(function QueuedMessageRow({
   const hasWaitLine = queuedMessageHasWaitLine(queuedMessage);
   const sendAllowed =
     sendAction === "steer-when-ready" ||
-    isQueuedMessageSendNowAllowed(queuedMessage.waitingOn);
+    isQueuedMessageSendNowAllowed(queuedMessage);
   const sendAriaLabel =
     sendAction === "steer-when-ready"
       ? `Steer queued message ${index + 1} when ready`

apps/app/src/lib/queued-message-wait.test.ts

@@ -201,27 +201,33 @@ describe("queuedMessageFallbackTitle", () => {
 });
 
 describe("isQueuedMessageSendNowAllowed", () => {
-  it("hides send-now only for the waits a re-attempt cannot clear", () => {
-    expect(isQueuedMessageSendNowAllowed({ kind: "time" })).toBe(true);
-    expect(
-      isQueuedMessageSendNowAllowed({
-        kind: "plugin",
-        pluginId: "limiter",
-        reason: "busy",
-      }),
-    ).toBe(true);
-    expect(isQueuedMessageSendNowAllowed({ kind: "thread-busy" })).toBe(true);
-    expect(isQueuedMessageSendNowAllowed({ kind: "stopping" })).toBe(false);
-    expect(isQueuedMessageSendNowAllowed({ kind: "turn-starting" })).toBe(
-      false,
-    );
-    expect(isQueuedMessageSendNowAllowed(null)).toBe(true);
-    expect(isQueuedMessageSendNowAllowed({ kind: "provisioning" })).toBe(false);
-    expect(isQueuedMessageSendNowAllowed({ kind: "interaction" })).toBe(false);
-    expect(
-      isQueuedMessageSendNowAllowed({ kind: "host-offline", hostName: "M4" }),
-    ).toBe(false);
-  });
+  it.each([
+    { waitingOn: null, allowed: true },
+    { waitingOn: { kind: "time" }, allowed: true },
+    {
+      waitingOn: { kind: "plugin", pluginId: "limiter", reason: "busy" },
+      allowed: true,
+    },
+    { waitingOn: { kind: "thread-busy" }, allowed: true },
+    { waitingOn: { kind: "stopping" }, allowed: false },
+    { waitingOn: { kind: "turn-starting" }, allowed: false },
+    { waitingOn: { kind: "provisioning" }, allowed: false },
+    { waitingOn: { kind: "interaction" }, allowed: false },
+    { waitingOn: { kind: "host-offline", hostName: "M4" }, allowed: false },
+  ] as const)(
+    "allows manual recovery for $waitingOn",
+    ({ waitingOn, allowed }) => {
+      expect(
+        isQueuedMessageSendNowAllowed({ waitingOn, failureReason: null }),
+      ).toBe(allowed);
+      expect(
+        isQueuedMessageSendNowAllowed({
+          waitingOn,
+          failureReason: "Provider unavailable",
+        }),
+      ).toBe(true);
+    },
+  );
 });
 
 describe("formatQueuedMessageCountdown", () => {

apps/app/src/lib/queued-message-wait.ts

@@ -22,9 +22,14 @@ export function formatQueuedMessageCountdown(
   return `in ${Math.floor(remainingMs / DAY_MS)}d`;
 }
 
-export function isQueuedMessageSendNowAllowed(
-  waitingOn: QueuedMessageWaitingOn | null,
-): boolean {
+export function isQueuedMessageSendNowAllowed({
+  waitingOn,
+  failureReason,
+}: {
+  waitingOn: QueuedMessageWaitingOn | null;
+  failureReason: string | null;
+}): boolean {
+  if (failureReason !== null) return true;
   if (waitingOn === null) return true;
   switch (waitingOn.kind) {
     case "provisioning":

apps/cli/src/__tests__/command-output/thread-organization.test.ts

@@ -105,6 +105,36 @@ describe("bb thread organization commands", () => {
     expect(output).toContain("System");
   });
 
+  it.each([false, true])(
+    "shows failures and recovery commands in queue list (scoped: %s)",
+    async (scoped) => {
+      const failureReason =
+        "The provider bridge is unavailable while the plugin is still building";
+      const list = vi.fn(async () => [
+        queuedMessage({
+          waitingOn: { kind: "host-offline", hostName: "Michael-M4" },
+          failureReason,
+        }),
+      ]);
+      stubServerApi({
+        [scoped
+          ? "v1.threads.:id.queued-messages.$get"
+          : "v1.queued-messages.$get"]: list,
+      });
+      await runCommand(
+        ["thread", "queue", "list", ...(scoped ? ["thread-1"] : [])],
+        register,
+      );
+      const output = vi
+        .mocked(console.log)
+        .mock.calls.map((args) => args.join(" "))
+        .join("\n");
+      expect(output).toContain(`Failed queued-1: ${failureReason}`);
+      expect(output).toContain("bb thread queue send thread-1 queued-1");
+      expect(output).not.toContain("waiting for Michael-M4 to reconnect");
+    },
+  );
+
   it("updates a queued message in place", async () => {
     const list = vi.fn(async () => [
       { id: "queued-1", updatedAt: 42 },

apps/cli/src/commands/thread/organization.ts

@@ -120,7 +120,12 @@ function printQueueTable(rows: ThreadQueuedMessagesResult): void {
         ? "System"
         : (row.senderThreadId ?? "Agent"),
     truncateCell(queuedMessagePreview(row.content), MAX_QUEUE_TEXT_WIDTH),
-    truncateCell(describeQueueWait(row), MAX_QUEUE_TEXT_WIDTH),
+    truncateCell(
+      row.failureReason === null
+        ? describeQueueWait(row)
+        : `Failed: ${row.failureReason}`,
+      MAX_QUEUE_TEXT_WIDTH,
+    ),
     formatQueueSendCountdown(row.sendAt, now),
   ]);
   printBorderlessTable(
@@ -130,6 +135,11 @@ function printQueueTable(rows: ThreadQueuedMessagesResult): void {
     },
     table,
   );
+  for (const row of rows) {
+    if (row.failureReason === null) continue;
+    console.log(`Failed ${row.id}: ${row.failureReason}`);
+    console.log(`Retry: bb thread queue send ${row.threadId} ${row.id}`);
+  }
 }
 
 function queuedMessagePreview(content: PromptInput[]): string {

apps/server/src/services/system/periodic-sweeps.ts

@@ -591,6 +591,13 @@ const PERIODIC_SWEEP_JOBS: PeriodicSweepJob[] = [
     run: (deps, now) =>
       runQueuedMessageDispatch(deps, { kind: "time-reached", now }),
   },
+  {
+    cadenceMs: 0,
+    category: "durable-intent-retry",
+    name: "failed-queue-message-retry",
+    run: (deps, now) =>
+      runQueuedMessageDispatch(deps, { kind: "failed-retry", now }),
+  },
   {
     cadenceMs: 0,
     category: "durable-intent-retry",

apps/server/src/services/threads/queue-drain-failure.ts

@@ -12,6 +12,10 @@ import { dispatchEnvironmentAndHost } from "./dispatch-hooks.js";
 
 type QueueDrainFailureDeps = Pick<AppDeps, "db" | "hub">;
 
+export const QUEUED_MESSAGE_RETRY_DELAYS_MS: readonly number[] = [
+  15_000, 60_000, 300_000,
+];
+
 /**
  * What a failed dispatch says to the person whose message did not go.
  *
@@ -39,7 +43,10 @@ export function describeDispatchFailure(error: unknown): string {
  * host-reconnect drain clears when the machine comes back. Any other failure
  * is recorded as the row's failure reason, leaving its existing wait alone —
  * the row is still waiting on whatever it was waiting on, and what went wrong
- * last time is a different fact from what it is waiting for.
+ * last time is a different fact from what it is waiting for — and spends one
+ * of the row's attempts, booking the next on
+ * {@link QUEUED_MESSAGE_RETRY_DELAYS_MS}. The row gives up only once that
+ * budget runs out.
  *
  * Only the drain calls this. An inline attempt has a caller still listening
  * and surfaces its error to them instead, which is why a queued row never
@@ -49,6 +56,7 @@ export function recordQueuedMessageDrainFailure(
   deps: QueueDrainFailureDeps,
   args: {
     error: unknown;
+    now: number;
     row: { id: string; threadId: string };
     thread: Thread;
   },
@@ -71,5 +79,7 @@ export function recordQueuedMessageDrainFailure(
     id: args.row.id,
     threadId: args.row.threadId,
     failureReason: describeDispatchFailure(args.error),
+    now: args.now,
+    retryDelaysMs: QUEUED_MESSAGE_RETRY_DELAYS_MS,
   });
 }

apps/server/src/services/threads/queued-message-dispatch.ts

@@ -10,6 +10,7 @@ import {
   listQueuedThreadMessagePluginWaitRefs,
   listQueuedThreadMessagesByWaitHolder,
   listQueuedThreadMessagesWaitingOnKind,
+  listRetryableFailedQueuedThreadMessages,
   listThreadIdsWithHostOfflineQueueWaits,
 } from "@bb/db";
 import {
@@ -47,6 +48,7 @@ export type QueuedMessageDispatchWake =
   | { kind: "plugin-recheck" }
   | { kind: "plugin-unregistered"; pluginId: string }
   | { kind: "idle-recovery"; now: number }
+  | { kind: "failed-retry"; now: number }
   | {
       kind: "orphaned-plugin-recovery";
       plugins: QueueWaitPluginDirectory;
@@ -116,6 +118,7 @@ function dispatchWakeContext(
     case "interaction-settled":
       return { threadId: wake.threadId, wake: wake.kind };
     case "idle-recovery":
+    case "failed-retry":
       return { now: wake.now, wake: wake.kind };
     case "orphaned-plugin-recovery":
       return { wake: wake.kind };
@@ -231,6 +234,7 @@ async function executePreparedQueuedMessageDispatch(
           await attemptAutomaticQueuedMessage(deps, row, {
             now: Date.now(),
             respectRequeuePacing: false,
+            retryingFailure: false,
           });
         }
       }
@@ -260,6 +264,9 @@ async function executePreparedQueuedMessageDispatch(
       releaseStaleQueuedMessageDispatchClaims(deps, wake.now);
       await runIdleThreadRecovery(deps);
       return;
+    case "failed-retry":
+      await runFailedRetryDispatch(deps, wake.now);
+      return;
     case "orphaned-plugin-recovery":
       await runOrphanedPluginWaitRecovery(deps, wake.plugins);
       return;
@@ -319,6 +326,7 @@ async function runTurnStartedDispatch(
     await attemptAutomaticQueuedMessage(deps, row, {
       now: Date.now(),
       respectRequeuePacing: false,
+      retryingFailure: false,
     });
   }
 }
@@ -343,6 +351,7 @@ async function runWorkspaceReadyDispatch(
     await attemptAutomaticQueuedMessage(deps, row, {
       now: Date.now(),
       respectRequeuePacing: false,
+      retryingFailure: false,
     });
   }
 }
@@ -364,7 +373,11 @@ async function runInteractionSettledDispatch(
 async function attemptAutomaticQueuedMessage(
   deps: QueueDispatchDeps,
   row: QueuedMessageDispatchRef,
-  args: { now: number; respectRequeuePacing: boolean },
+  args: {
+    now: number;
+    respectRequeuePacing: boolean;
+    retryingFailure: boolean;
+  },
 ): Promise<void> {
   if (args.respectRequeuePacing && isDispatchRequeuedRecently(row.threadId))
     return;
@@ -377,8 +390,10 @@ async function attemptAutomaticQueuedMessage(
         kind: "automatic",
         isGroupEligible: createAutomaticQueuedMessageGroupEligibility(deps, {
           now: args.now,
+          retryingFailure: args.retryingFailure,
           thread,
         }),
+        retryingFailure: args.retryingFailure,
       },
       mode: "auto",
       queuedMessageId: row.id,
@@ -396,7 +411,12 @@ async function attemptAutomaticQueuedMessage(
       );
       return;
     }
-    recordQueuedMessageDrainFailure(deps, { error, row, thread });
+    recordQueuedMessageDrainFailure(deps, {
+      error,
+      now: args.now,
+      row,
+      thread,
+    });
     deps.logger.warn(
       {
         queuedMessageId: row.id,
@@ -416,6 +436,7 @@ async function runPluginRecheckDispatch(
     await attemptAutomaticQueuedMessage(deps, row, {
       now,
       respectRequeuePacing: true,
+      retryingFailure: false,
     });
   }
 }
@@ -449,6 +470,40 @@ async function runDueScheduledDispatch(
     await attemptAutomaticQueuedMessage(deps, row, {
       now,
       respectRequeuePacing: true,
+      retryingFailure: false,
+    });
+  }
+}
+
+/**
+ * Re-attempts rows whose booked retry has come due.
+ *
+ * This is the only automatic path that may claim a row with a recorded
+ * failure, and it exists because a failure is not a verdict about the message:
+ * it is what the server was able to do at one instant, usually an instant
+ * during a restart. Every other wake asks "did the thing this row waits for
+ * happen?", which a row that failed can no longer be asked — its wait may have
+ * gone stale while it sat there, and the edge that would have cleared it has
+ * passed. So this one re-asks the whole question instead, and the row's
+ * remaining attempts are what stop it asking forever.
+ */
+async function runFailedRetryDispatch(
+  deps: QueueDispatchDeps,
+  now: number,
+): Promise<void> {
+  for (const row of listRetryableFailedQueuedThreadMessages(deps.db, now)) {
+    deps.logger.info(
+      {
+        failureCount: row.failureCount,
+        queuedMessageId: row.id,
+        threadId: row.threadId,
+      },
+      "Retrying a queued message whose dispatch failed",
+    );
+    await attemptAutomaticQueuedMessage(deps, row, {
+      now,
+      respectRequeuePacing: true,
+      retryingFailure: true,
     });
   }
 }