Continue threads after restarts behind an experiment

apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx

@@ -263,6 +263,8 @@ function systemMessageTitleSegments(
       return subject !== null && subject.kind === "directory"
         ? [verbSegment("Continued in"), subjectSegment(subject.path, null)]
         : SYSTEM_MESSAGE_FALLBACK_SEGMENTS;
+    case "restart-resume":
+      return [verbSegment("Resumed after a restart")];
     case "unlabeled":
       return SYSTEM_MESSAGE_FALLBACK_SEGMENTS;
   }
@@ -353,6 +355,8 @@ function systemMessageIconName(systemMessageKind: SystemMessageKind): IconName {
       return "Toolbox";
     case "environment-directory-updated":
       return "FolderOpen";
+    case "restart-resume":
+      return "Replay";
     case "unlabeled":
       return "Info";
   }

apps/app/src/lib/system-config-atoms.ts

@@ -33,6 +33,7 @@ const unavailableSystemConfig: SystemConfigResponse = {
     applicationLinks: false,
     changelogPreview: false,
     compactExpiredCache: false,
+    continueThreadsAfterRestarts: false,
     serverMove: false,
     performanceDiagnostics: false,
     pluginUpdates: false,

apps/app/src/views/SettingsView.tsx

@@ -1134,6 +1134,11 @@ const EXPERIMENT_DEFINITIONS: Record<
     description:
       "When an idle thread holds at least 100k tokens and its prompt cache has expired, the composer offers to compact before your next send. Works with supported providers (e.g. Claude Code, Pi).",
   },
+  continueThreadsAfterRestarts: {
+    label: "Continue threads after restarts",
+    description:
+      "When quitting bb, a crash, an update or restarting a machine interrupts a thread, send it a message so it picks its work back up. bb drops the message if the turn finished, you stopped it or you sent something newer.",
+  },
   performanceDiagnostics: {
     label: "Server performance diagnostics",
     description:

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

@@ -1,5 +1,6 @@
 import { describe, expect, it, vi } from "vitest";
 import * as domain from "@bb/domain";
+import { makeThreadQueuedMessage } from "@bb/test-helpers/domain-fixtures";
 import {
   setupCommandOutputTestEnvironment,
   collectLogLines,
@@ -99,7 +100,15 @@ describe("bb thread wait command output", () => {
         updatedAt: 2,
       }),
     );
-    stubServerApi({ "v1.threads.:id.$get": get });
+    stubServerApi({
+      "v1.threads.:id.$get": get,
+      "v1.threads.:id.queued-messages.$get": vi.fn(async () => [
+        makeThreadQueuedMessage({
+          threadId: "thread-wait-error",
+          waitingOn: { kind: "thread-busy" },
+        }),
+      ]),
+    });
 
     await expect(
       runCommand(
@@ -114,6 +123,48 @@ describe("bb thread wait command output", () => {
     expect(get).toHaveBeenCalledTimes(1);
   });
 
+  it("bb thread wait --status idle keeps waiting on a failed thread that will resume on its own", async () => {
+    const statuses = ["error", "error", "active", "idle"] as const;
+    const get = vi.fn(async () =>
+      fixtures.makeThread({
+        id: "thread-wait-resume",
+        projectId: "proj-1",
+        providerId: "codex",
+        status: statuses[Math.min(get.mock.calls.length - 1, 3)] ?? "idle",
+        createdAt: 1,
+        updatedAt: 2,
+      }),
+    );
+    stubServerApi({
+      "v1.threads.:id.$get": get,
+      "v1.threads.:id.queued-messages.$get": vi.fn(async () => [
+        makeThreadQueuedMessage({
+          initiator: "system",
+          threadId: "thread-wait-resume",
+          sendAt: 1,
+          waitingOn: { kind: "time" },
+        }),
+      ]),
+    });
+
+    await runCommand(
+      [
+        "thread",
+        "wait",
+        "thread-wait-resume",
+        "--status",
+        "idle",
+        "--poll-interval",
+        "1",
+      ],
+      register,
+    );
+
+    expect(collectLogLines(vi.mocked(console.log))).toContain(
+      "Thread thread-wait-resume reached status idle.",
+    );
+  });
+
   it("bb thread wait --event reports server errors instead of schema errors", async () => {
     const waitGet = vi.fn(
       async () =>

apps/server/src/internal/session-owner-side-effects.ts

@@ -21,6 +21,7 @@ import {
 } from "../services/threads/thread-lifecycle.js";
 import { buildThreadStatusChangeMetadataByThreadId } from "../services/threads/thread-runtime-display.js";
 import { settleDanglingBackgroundTasks } from "../services/threads/background-task-reconciliation.js";
+import { resumeWorkCutByHostRestart } from "../services/threads/restart-resume.js";
 import { interruptEnvironmentProvisioningForHost } from "../services/environments/environment-engine.js";
 
 const DAEMON_RESTARTED_PENDING_INTERACTION_REASON =
@@ -95,6 +96,10 @@ export async function handleHostSessionOpened(
     }
 
     if (!sameDaemonInstance) {
+      await resumeWorkCutByHostRestart(deps, {
+        hostId: args.hostId,
+        now: Date.now(),
+      });
       interruptEnvironmentProvisioningForHost(deps, {
         hostId: args.hostId,
         reason: DAEMON_RESTARTED_ENVIRONMENT_PROVISIONING_REASON,

apps/server/src/services/threads/background-task-reconciliation.ts

@@ -3,12 +3,14 @@ import {
   getLatestThreadSequence,
   listOpenBackgroundTaskItemRowsForHost,
   listOpenBackgroundTaskItemRowsForThread,
+  type DbConnection,
   type DbNotifier,
   type DbTransaction,
   type OpenBackgroundTaskItemRow,
 } from "@bb/db";
 import {
   backgroundTaskItemStatus,
+  isBackgroundCommandTaskType,
   isSettledBackgroundTaskStatus,
   threadEventBackgroundTaskItemSchema,
   threadScope,
@@ -45,6 +47,21 @@ function parseStoredBackgroundTaskItem(
   }
 }
 
+export function listAwaitedBackgroundWorkByThread(
+  db: DbConnection,
+  hostId: string,
+): Map<string, string[]> {
+  const byThread = new Map<string, string[]>();
+  for (const row of listOpenBackgroundTaskItemRowsForHost(db, { hostId })) {
+    const item = parseStoredBackgroundTaskItem(row);
+    if (!item || isBackgroundCommandTaskType(item.taskType)) continue;
+    const descriptions = byThread.get(row.threadId) ?? [];
+    descriptions.push(item.description);
+    byThread.set(row.threadId, descriptions);
+  }
+  return byThread;
+}
+
 export function settleDanglingBackgroundTasks(
   deps: SettleDanglingBackgroundTasksDeps,
   args: SettleDanglingBackgroundTasksArgs,

apps/server/src/services/threads/lifecycle-outcome.ts

@@ -5,6 +5,7 @@ import {
   type ApplyThreadLifecycleEventArgs,
   type ApplyThreadLifecycleEventOutcome,
   type DbConnection,
+  type DbQueryConnection,
   type DbTransaction,
 } from "@bb/db";
 import { recordFinishedTurn } from "../system/turn-telemetry.js";
@@ -15,6 +16,7 @@ import {
   emitPluginTurnFailed,
 } from "../plugins/plugin-thread-events.js";
 import type { ProviderRegistryService } from "../providers/provider-registry.js";
+import { isTurnFailureHandledByRestartResume } from "./restart-resume.js";
 import { buildThreadStatusChangeMetadata } from "./thread-runtime-display.js";
 
 /**
@@ -28,10 +30,12 @@ import { buildThreadStatusChangeMetadata } from "./thread-runtime-display.js";
  * handled.
  */
 function announceTurnFailed(
+  db: DbQueryConnection,
   args: ApplyThreadLifecycleEventArgs,
   outcome: ApplyThreadLifecycleEventOutcome,
 ): void {
   if (!outcome.applied || args.event.type !== "run.failed") return;
+  if (isTurnFailureHandledByRestartResume(db, args.threadId)) return;
   emitPluginTurnFailed(args.threadId);
 }
 
@@ -84,7 +88,7 @@ export function applyLoggedThreadLifecycleEvent(
   }
   logUnappliedThreadLifecycleEvent(deps.logger, args, outcome);
   emitPluginThreadLifecycleOutcome(outcome);
-  announceTurnFailed(args, outcome);
+  announceTurnFailed(deps.db, args, outcome);
   return outcome;
 }
 
@@ -99,6 +103,6 @@ export function applyLoggedThreadLifecycleEventInTransaction(
   }
   logUnappliedThreadLifecycleEvent(deps.logger, args, outcome);
   emitPluginThreadLifecycleOutcome(outcome);
-  announceTurnFailed(args, outcome);
+  announceTurnFailed(deps.db, args, outcome);
   return outcome;
 }

apps/server/src/services/threads/queued-messages.ts

@@ -105,6 +105,10 @@ import {
 } from "../lib/lifecycle-api-errors.js";
 import { resolvePromptAttachmentReferences } from "../projects/attachments.js";
 import { requestQueuedMessageDispatch } from "./queued-message-dispatch.js";
+import {
+  isStaleRestartResume,
+  readRestartResumeSubject,
+} from "./restart-resume.js";
 import { assertThreadHostAcceptsWork } from "./thread-host-admission.js";
 import {
   ThreadContextClearInProgressError,
@@ -649,6 +653,17 @@ async function sendClaimedQueuedMessageForIdleProviderThread(
  * queue is where a blocked dispatch waits. Null when the row is an ordinary
  * message, which is every row but these.
  */
+function isStaleNotice(
+  deps: Pick<AppDeps, "db">,
+  threadId: string,
+  row: ClaimedQueuedMessage,
+): boolean {
+  const subject = readRestartResumeSubject(row);
+  return (
+    subject !== null && isStaleRestartResume(deps.db, { subject, threadId })
+  );
+}
+
 async function sendClaimedSystemNotice(
   deps: LoggedPendingInteractionWorkSessionDeps,
   args: SendClaimedQueuedMessageForThreadArgs,
@@ -661,12 +676,14 @@ async function sendClaimedSystemNotice(
     JSON.parse(lead.systemNotice),
   );
   const queuedMessage = toThreadQueuedMessage(lead);
-  const delivered = await deliverParentSystemMessage(deps, {
-    input: queuedMessage.content,
-    parentThread: args.thread,
-    systemMessageKind: notice.kind,
-    systemMessageSubject: notice.subject,
-  });
+  const delivered =
+    isStaleNotice(deps, args.thread.id, lead) ||
+    (await deliverParentSystemMessage(deps, {
+      input: queuedMessage.content,
+      parentThread: args.thread,
+      systemMessageKind: notice.kind,
+      systemMessageSubject: notice.subject,
+    }));
   if (!delivered) {
     // The thread changed under the drain. Leave the row claimed-and-released
     // by the caller's error path rather than consuming a notice nobody got.