From 7db445d38fe4108e996caca79509e982fa5ab62e Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 29 Oct 2026 16:41:35 -0700 Subject: [PATCH 1/4] Continue threads after restarts behind an experiment With the continueThreadsAfterRestarts experiment on, a thread that a quit, crash, update or machine restart interrupted gets one message naming what it was doing, such as the command it was running, so the agent checks it and carries on. bb records a resume for each running thread when the server shuts down, and for each cut turn or awaited background work (not background shells) when a host daemon comes back as a new process. Resumes go out a few seconds apart per machine, child threads before their parents, ahead of messages already queued for the thread. A resume is dropped if the turn finished, the user stopped it, or the user sent something newer, and bb stops after three automatic resumes in a row. A turn a restart cut is not announced to retry policies, the parent of an interrupted child hears only when the child finishes, and bb thread wait keeps waiting on a failed thread whose resume is scheduled. Co-Authored-By: Claude Opus 5.5 --- .../timeline/GeneratedConversationMessage.tsx | 4 + apps/app/src/lib/system-config-atoms.ts | 1 + apps/app/src/views/SettingsView.tsx | 5 + .../command-output/thread-wait.test.ts | 53 +- .../internal/session-owner-side-effects.ts | 5 + .../threads/background-task-reconciliation.ts | 17 + .../src/services/threads/lifecycle-outcome.ts | 8 +- .../src/services/threads/queued-messages.ts | 29 +- .../src/services/threads/restart-resume.ts | 506 ++++++++++++++++++ .../src/services/threads/thread-lifecycle.ts | 17 +- apps/server/src/start-server.ts | 26 + apps/server/test/helpers/restart-resume.ts | 175 ++++++ apps/server/test/system/experiments.test.ts | 6 + .../test/threads/restart-resume.test.ts | 366 +++++++++++++ .../test/threads/turn-failed-retry.test.ts | 107 +++- docs/configuration.md | 14 + packages/db/src/data/events.ts | 85 +++ packages/db/src/data/index.ts | 8 + packages/db/src/data/pending-interactions.ts | 11 + .../db/src/data/queued-thread-messages.ts | 102 ++++ packages/domain/src/experiments.ts | 3 + packages/domain/src/system-message.ts | 6 + packages/sdk/src/areas/threads.ts | 16 +- .../src/templates/bb-guide-customization.md | 4 + .../templates/bb-guide-thread-inspection.md | 4 + .../system-message-restart-resume.md | 14 + 26 files changed, 1578 insertions(+), 14 deletions(-) create mode 100644 apps/server/src/services/threads/restart-resume.ts create mode 100644 apps/server/test/helpers/restart-resume.ts create mode 100644 apps/server/test/threads/restart-resume.test.ts create mode 100644 packages/templates/src/templates/system-message-restart-resume.md diff --git a/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx b/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx index 89656bbf003..e4eb310325e 100644 --- a/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx +++ b/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 "RotateCcw"; case "unlabeled": return "Info"; } diff --git a/apps/app/src/lib/system-config-atoms.ts b/apps/app/src/lib/system-config-atoms.ts index dd997936032..e2d756d2bc4 100644 --- a/apps/app/src/lib/system-config-atoms.ts +++ b/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, diff --git a/apps/app/src/views/SettingsView.tsx b/apps/app/src/views/SettingsView.tsx index 15cca3d047f..608fd0c7932 100644 --- a/apps/app/src/views/SettingsView.tsx +++ b/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: diff --git a/apps/cli/src/__tests__/command-output/thread-wait.test.ts b/apps/cli/src/__tests__/command-output/thread-wait.test.ts index 551a5364750..e100c70a4e0 100644 --- a/apps/cli/src/__tests__/command-output/thread-wait.test.ts +++ b/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 () => diff --git a/apps/server/src/internal/session-owner-side-effects.ts b/apps/server/src/internal/session-owner-side-effects.ts index 9b0bd4fed22..36aee5bd30a 100644 --- a/apps/server/src/internal/session-owner-side-effects.ts +++ b/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, diff --git a/apps/server/src/services/threads/background-task-reconciliation.ts b/apps/server/src/services/threads/background-task-reconciliation.ts index f26397cc662..b5b96fa11c6 100644 --- a/apps/server/src/services/threads/background-task-reconciliation.ts +++ b/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 { + const byThread = new Map(); + 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, diff --git a/apps/server/src/services/threads/lifecycle-outcome.ts b/apps/server/src/services/threads/lifecycle-outcome.ts index 4f5427e6c9c..63b1dfb5a5d 100644 --- a/apps/server/src/services/threads/lifecycle-outcome.ts +++ b/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; } diff --git a/apps/server/src/services/threads/queued-messages.ts b/apps/server/src/services/threads/queued-messages.ts index d33975087ea..d869ab026be 100644 --- a/apps/server/src/services/threads/queued-messages.ts +++ b/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, + 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. diff --git a/apps/server/src/services/threads/restart-resume.ts b/apps/server/src/services/threads/restart-resume.ts new file mode 100644 index 00000000000..84268e04f49 --- /dev/null +++ b/apps/server/src/services/threads/restart-resume.ts @@ -0,0 +1,506 @@ +import { z } from "zod"; +import { + createQueuedThreadMessage, + getActiveStoredTurnId, + getEnvironment, + getExperiments, + getLastStoredTurnRequestEvent, + getLatestThreadInterruption, + getLatestThreadSequence, + getPendingQueuedSystemNotice, + getRunningCommandItemData, + getThread, + listLatestTurnRequestSystemMessageKinds, + listQueuedThreadMessages, + listRunningThreads, + listStoredTurnCompletedRowsByTurnIds, + listThreadIdsWithActivePendingInteractions, + listUnclaimedQueuedSystemNotices, + moveQueuedThreadMessageToFront, + rescheduleQueuedThreadMessage, + setQueuedThreadMessageWaitingOn, + type DbConnection, + type DbQueryConnection, + type QueuedThreadMessageRow, +} from "@bb/db"; +import { + queuedMessageSystemNoticeSchema, + threadEventTurnStatusSchema, + type SystemThreadInterruptedReason, +} from "@bb/domain"; +import { renderTemplate } from "@bb/templates"; +import type { LoggedPendingInteractionWorkSessionDeps } from "../../types.js"; +import { listAwaitedBackgroundWorkByThread } from "./background-task-reconciliation.js"; +import { buildExecutionOptions } from "./thread-commands.js"; +import { parseStoredQueuedThreadMessageWaitingOn } from "./thread-queued-messages.js"; + +const MAX_CONSECUTIVE_RESTART_RESUMES = 3; +const RESUME_DELAY_MS = 3_000; +const RESUME_STAGGER_MS = 5_000; +const RESUME_PARKED_FOR_MS = 24 * 60 * 60 * 1000; +const RESTART_INTERRUPTION_REASONS: ReadonlySet = + new Set(["host-daemon-restarted"]); +const ORDERED_BEHIND_RESUME_WAITS = new Set([ + "host-offline", + "stopping", + "thread-busy", + "turn-starting", +]); + +const runningCommandItemDataSchema = z.object({ + item: z.object({ command: z.string().min(1) }), +}); +const turnCompletedDataSchema = z.object({ + status: threadEventTurnStatusSchema, +}); + +export type RestartWork = + | { kind: "turn"; threadId: string } + | { kind: "background"; jobs: readonly string[]; threadId: string }; + +interface RestartResumeSubject { + afterSequence: number; + turnId: string | null; +} + +interface PendingRestartResume { + row: QueuedThreadMessageRow; + subject: RestartResumeSubject; +} + +interface ResumeText { + check: string; + interrupted: string; + restarted: string; +} + +type RestartResumeDeps = LoggedPendingInteractionWorkSessionDeps; + +export function isRestartContinuationEnabled(db: DbConnection): boolean { + return getExperiments(db).continueThreadsAfterRestarts; +} + +function readSubject(row: QueuedThreadMessageRow): RestartResumeSubject | null { + if (row.systemNotice === null) return null; + const notice = queuedMessageSystemNoticeSchema.safeParse( + JSON.parse(row.systemNotice), + ); + if ( + !notice.success || + notice.data.kind !== "restart-resume" || + notice.data.subject?.kind !== "restart" + ) { + return null; + } + return { + afterSequence: notice.data.subject.afterSequence, + turnId: notice.data.subject.turnId, + }; +} + +function getPendingRestartResume( + db: DbQueryConnection, + threadId: string, +): PendingRestartResume | null { + const row = getPendingQueuedSystemNotice(db, { + kind: "restart-resume", + threadId, + }); + if (row === null) return null; + const subject = readSubject(row); + return subject === null ? null : { row, subject }; +} + +function threadHostId(db: DbConnection, threadId: string): string | null { + const environmentId = getThread(db, threadId)?.environmentId ?? null; + return environmentId === null + ? null + : (getEnvironment(db, environmentId)?.hostId ?? null); +} + +function runningCommand(db: DbQueryConnection, threadId: string): string | null { + const data = getRunningCommandItemData(db, { threadId }); + if (data === null) return null; + try { + const parsed = runningCommandItemDataSchema.safeParse(JSON.parse(data)); + return parsed.success ? parsed.data.item.command : null; + } catch { + return null; + } +} + +const RESTARTED = "bb restarted, or the machine running this thread did,"; + +function turnResumeText(args: { + awaitingUser: boolean; + command: string | null; +}): ResumeText { + const restarted = RESTARTED; + if (args.awaitingUser) { + return { + restarted, + interrupted: + args.command === null + ? "while you were waiting for the user" + : `while you were waiting for the user to approve \`${args.command}\``, + check: + "The restart cancelled that request; the user did not answer it. Ask again if you still need it, and do not decide for them.", + }; + } + if (args.command === null) { + return { + restarted, + interrupted: "while you were working", + check: + "The restart interrupted your turn; the user did not. Check whether your last step finished or is still running, then continue the task without repeating work that already happened.", + }; + } + return { + restarted, + interrupted: `while you were running \`${args.command}\``, + check: + "The restart interrupted it; the user did not reject it. It may have completed, partly completed, never run, or still be running, so check its effects before you continue, and rerun it only if that is safe.", + }; +} + +function backgroundResumeText(jobs: readonly string[]): ResumeText { + return { + restarted: RESTARTED, + interrupted: `while this background work was running: ${jobs + .map((job) => `\`${job}\``) + .join(", ")}`, + check: + "The restart may have stopped it. Check whether each is still running and how far it got, and restart one only when that is safe; otherwise explain what needs attention.", + }; +} + +function hasReachedResumeLimit(db: DbQueryConnection, threadId: string): boolean { + const kinds = listLatestTurnRequestSystemMessageKinds(db, { + limit: MAX_CONSECUTIVE_RESTART_RESUMES, + threadId, + }); + return ( + kinds.length === MAX_CONSECUTIVE_RESTART_RESUMES && + kinds.every((kind) => kind === "restart-resume") + ); +} + +export async function recordRestartResumes( + deps: RestartResumeDeps, + args: { now: number; work: readonly RestartWork[] }, +): Promise { + const awaitingUser = new Set( + listThreadIdsWithActivePendingInteractions(deps.db), + ); + const prepared = []; + const seen = new Set(); + for (const work of args.work) { + if (seen.has(work.threadId)) continue; + seen.add(work.threadId); + const thread = getThread(deps.db, work.threadId); + if ( + thread === null || + thread.archivedAt !== null || + thread.deletedAt !== null || + getPendingRestartResume(deps.db, thread.id) !== null + ) { + continue; + } + if (hasReachedResumeLimit(deps.db, thread.id)) { + deps.logger.warn( + { threadId: thread.id }, + "Not resuming a thread that restarts keep interrupting", + ); + continue; + } + const text = + work.kind === "turn" + ? turnResumeText({ + awaitingUser: awaitingUser.has(thread.id), + command: runningCommand(deps.db, thread.id), + }) + : backgroundResumeText(work.jobs); + try { + prepared.push({ + execution: await buildExecutionOptions( + deps, + {}, + { threadId: thread.id }, + ), + text, + threadId: thread.id, + turnId: + work.kind === "turn" ? getActiveStoredTurnId(deps.db, thread.id) : null, + }); + } catch (error) { + deps.logger.warn( + { error, threadId: thread.id }, + "Could not prepare a resume message for a restart", + ); + } + } + const recorded: string[] = []; + for (const entry of prepared) { + if (getPendingRestartResume(deps.db, entry.threadId) !== null) continue; + createQueuedThreadMessage(deps.db, deps.hub, { + threadId: entry.threadId, + content: [ + { + type: "text", + text: renderTemplate("systemMessageRestartResume", entry.text), + mentions: [], + }, + ], + senderThreadId: null, + origin: null, + originPluginId: null, + model: entry.execution.model, + reasoningLevel: entry.execution.reasoningLevel, + permissionMode: entry.execution.permissionMode, + serviceTier: entry.execution.serviceTier, + waitingOn: { kind: "time" }, + sendAt: args.now + RESUME_PARKED_FOR_MS, + payload: { kind: "inline" }, + systemNotice: { + kind: "restart-resume", + subject: { + kind: "restart", + afterSequence: getLatestThreadSequence(deps.db, { + threadId: entry.threadId, + }), + turnId: entry.turnId, + }, + }, + }); + recorded.push(entry.threadId); + } + return recorded; +} + +function resumingAncestorCount( + db: DbConnection, + threadId: string, + resuming: ReadonlyMap, +): number { + let count = 0; + const seen = new Set([threadId]); + let parentId = getThread(db, threadId)?.parentThreadId ?? null; + while (parentId !== null && !seen.has(parentId)) { + seen.add(parentId); + if (resuming.has(parentId)) count += 1; + parentId = getThread(db, parentId)?.parentThreadId ?? null; + } + return count; +} + +function childrenFirst( + db: DbConnection, + hostIdByThread: ReadonlyMap, +): [string, string | null][] { + return [...hostIdByThread] + .map((entry) => ({ + depth: resumingAncestorCount(db, entry[0], hostIdByThread), + entry, + })) + .sort((a, b) => b.depth - a.depth) + .map(({ entry }) => entry); +} + +export function scheduleRestartResumes( + deps: RestartResumeDeps, + args: { hostIdByThread: ReadonlyMap; startAt: number }, +): Map { + const sendAtByThread = new Map(); + const offsetByHost = new Map(); + for (const [threadId, hostId] of childrenFirst( + deps.db, + args.hostIdByThread, + )) { + const pending = getPendingRestartResume(deps.db, threadId); + if (pending === null) continue; + const hostKey = hostId ?? threadId; + const offset = offsetByHost.get(hostKey) ?? 0; + const sendAt = args.startAt + offset * RESUME_STAGGER_MS; + if ( + rescheduleQueuedThreadMessage(deps.db, deps.hub, { + id: pending.row.id, + sendAt, + }) === null + ) { + continue; + } + offsetByHost.set(hostKey, offset + 1); + moveQueuedThreadMessageToFront(deps.db, deps.hub, { + id: pending.row.id, + threadId, + }); + let behind = 0; + for (const row of listQueuedThreadMessages(deps.db, threadId)) { + if (row.id === pending.row.id || row.failureReason !== null) continue; + const waitingOn = parseStoredQueuedThreadMessageWaitingOn(row); + if (waitingOn === null || !ORDERED_BEHIND_RESUME_WAITS.has(waitingOn.kind)) { + continue; + } + behind += 1; + rescheduleQueuedThreadMessage(deps.db, deps.hub, { + id: row.id, + sendAt: sendAt + behind, + }); + } + sendAtByThread.set(threadId, sendAt); + } + return sendAtByThread; +} + +export async function resumeWorkCutByHostRestart( + deps: RestartResumeDeps, + args: { hostId: string; now: number }, +): Promise { + if (!isRestartContinuationEnabled(deps.db)) return; + const running = listRunningThreads(deps.db) + .filter((row) => row.hostId === args.hostId) + .map((row) => row.id); + const runningSet = new Set(running); + const background = [ + ...listAwaitedBackgroundWorkByThread(deps.db, args.hostId), + ].filter(([threadId]) => !runningSet.has(threadId)); + await recordRestartResumes(deps, { + now: args.now, + work: [ + ...running.map((threadId) => ({ kind: "turn" as const, threadId })), + ...background.map(([threadId, jobs]) => ({ + jobs, + kind: "background" as const, + threadId, + })), + ], + }); + scheduleRestartResumes(deps, { + hostIdByThread: new Map( + [...running, ...background.map(([threadId]) => threadId)].map( + (threadId) => [threadId, args.hostId], + ), + ), + startAt: args.now + RESUME_DELAY_MS, + }); +} + +export async function resumeTurnsCutByRestart( + deps: RestartResumeDeps, + args: { hostId: string; now: number; threadIds: readonly string[] }, +): Promise { + if (args.threadIds.length === 0 || !isRestartContinuationEnabled(deps.db)) { + return; + } + await recordRestartResumes(deps, { + now: args.now, + work: args.threadIds.map((threadId) => ({ kind: "turn", threadId })), + }); + scheduleRestartResumes(deps, { + hostIdByThread: new Map( + args.threadIds.map((threadId) => [threadId, args.hostId]), + ), + startAt: args.now + RESUME_DELAY_MS, + }); +} + +export function releaseRestartResumes( + deps: RestartResumeDeps, + args: { now: number }, +): string[] { + const running = new Map( + listRunningThreads(deps.db).map((row) => [row.id, row.hostId]), + ); + const ready = new Map(); + for (const row of listUnclaimedQueuedSystemNotices(deps.db, "restart-resume")) { + if (row.failureReason !== null) continue; + if (running.has(row.threadId)) { + setQueuedThreadMessageWaitingOn(deps.db, deps.hub, { + id: row.id, + threadId: row.threadId, + waitingOn: { kind: "thread-busy" }, + sendAt: null, + }); + continue; + } + ready.set(row.threadId, threadHostId(deps.db, row.threadId)); + } + return [ + ...scheduleRestartResumes(deps, { + hostIdByThread: ready, + startAt: args.now + RESUME_DELAY_MS, + }).keys(), + ]; +} + +export async function recordResumesForShutdown( + deps: RestartResumeDeps, + args: { now: number }, +): Promise { + if (!isRestartContinuationEnabled(deps.db)) return; + await recordRestartResumes(deps, { + now: args.now, + work: listRunningThreads(deps.db).map((row) => ({ + kind: "turn", + threadId: row.id, + })), + }); +} + +export function isStaleRestartResume( + db: DbConnection, + args: { subject: RestartResumeSubject; threadId: string }, +): boolean { + const latestTurnRequest = getLastStoredTurnRequestEvent(db, args.threadId); + if ( + latestTurnRequest !== null && + latestTurnRequest.sequence > args.subject.afterSequence + ) { + return true; + } + if (args.subject.turnId !== null) { + const [completed] = listStoredTurnCompletedRowsByTurnIds(db, { + threadId: args.threadId, + turnIds: [args.subject.turnId], + }); + if (completed !== undefined) { + const data = turnCompletedDataSchema.safeParse(JSON.parse(completed.data)); + if (data.success && data.data.status === "completed") return true; + } + } + const interruption = getLatestThreadInterruption(db, { + threadId: args.threadId, + }); + return ( + interruption !== null && + interruption.sequence > args.subject.afterSequence && + !RESTART_INTERRUPTION_REASONS.has(interruption.reason) + ); +} + +export function readRestartResumeSubject( + row: QueuedThreadMessageRow, +): RestartResumeSubject | null { + return readSubject(row); +} + +export function isTurnFailureHandledByRestartResume( + db: DbQueryConnection, + threadId: string, +): boolean { + const pending = getPendingRestartResume(db, threadId); + if (pending === null) return false; + const interruption = getLatestThreadInterruption(db, { threadId }); + return ( + interruption !== null && + RESTART_INTERRUPTION_REASONS.has(interruption.reason) && + interruption.sequence > pending.subject.afterSequence + ); +} + +export function hasPendingRestartResume( + db: DbQueryConnection, + threadId: string, +): boolean { + return getPendingRestartResume(db, threadId) !== null; +} diff --git a/apps/server/src/services/threads/thread-lifecycle.ts b/apps/server/src/services/threads/thread-lifecycle.ts index a54805aebd0..b793caf4c88 100644 --- a/apps/server/src/services/threads/thread-lifecycle.ts +++ b/apps/server/src/services/threads/thread-lifecycle.ts @@ -113,6 +113,10 @@ import { scheduleThreadProvisioningAdvance } from "./thread-provisioning.js"; import { isPreStartThreadStatus } from "./thread-status.js"; import { settleDanglingBackgroundTasksForStoppedThreadInTransaction } from "./background-task-reconciliation.js"; import { abortPluginToolCallsForThreads } from "../plugins/plugin-tool-calls.js"; +import { + hasPendingRestartResume, + resumeTurnsCutByRestart, +} from "./restart-resume.js"; type ThreadStartCommand = Awaited>; type ThreadStopCommand = ReturnType; @@ -1808,8 +1812,9 @@ function interruptActiveThreads( data: buildSystemErrorEventData({ code: "thread_command_failed", message: failureMessage, - detail: - threadCommandFailureDetailForInterruption(effectiveReason), + detail: hasPendingRestartResume(tx, thread.threadId) + ? "bb will send this thread a message to continue." + : threadCommandFailureDetailForInterruption(effectiveReason), }), }); } @@ -1882,7 +1887,8 @@ function interruptActiveThreads( if ( result.interruptedTurnId !== null && thread && - isParentNotifiableChildThread(thread, "interrupted") + isParentNotifiableChildThread(thread, "interrupted") && + !hasPendingRestartResume(deps.db, thread.id) ) { void queueChildThreadTurnNotificationBestEffort(deps, { childThread: thread, @@ -2158,6 +2164,11 @@ export async function reconcileDaemonReportedThreads( ) .all(); + await resumeTurnsCutByRestart(deps, { + hostId: args.hostId, + now: Date.now(), + threadIds: activeButMissing.map((thread) => thread.id), + }); interruptActiveThreads(deps, { threads: activeButMissing.map((thread) => ({ environmentId: thread.environmentId, diff --git a/apps/server/src/start-server.ts b/apps/server/src/start-server.ts index e01e7c9cbdb..d6a25d528a7 100644 --- a/apps/server/src/start-server.ts +++ b/apps/server/src/start-server.ts @@ -39,6 +39,10 @@ import { runPeriodicSweeps, runStartupRecoverySweep, } from "./services/system/periodic-sweeps.js"; +import { + recordResumesForShutdown, + releaseRestartResumes, +} from "./services/threads/restart-resume.js"; import { installProviderModelCatalogPrewarm } from "./services/providers/provider-model-catalog-prewarm.js"; import { createProviderRegistryService, @@ -420,6 +424,20 @@ export async function runServer(serverConfig: ServerConfig): Promise { await runStartupRecoverySweep(sweepDeps).catch((error) => { logger.error({ err: error }, "Startup recovery sweep failed"); }); + try { + const resumed = releaseRestartResumes(sweepDeps, { now: Date.now() }); + if (resumed.length > 0) { + logger.info( + { threadCount: resumed.length }, + "Resuming threads a restart interrupted", + ); + } + } catch (error) { + logger.error( + { err: error }, + "Could not resume the threads a restart interrupted", + ); + } } if (!isLoopbackHostname(serverConfig.BB_SERVER_BIND_HOST)) { @@ -485,6 +503,14 @@ export async function runServer(serverConfig: ServerConfig): Promise { return shutdownPromise; } shutdownPromise = (async () => { + await recordResumesForShutdown(sweepDeps, { now: Date.now() }).catch( + (error: unknown) => { + logger.warn( + { err: error }, + "Could not record the threads this shutdown interrupts", + ); + }, + ); hub.stopClientChangeNotifications(); serverMove.dispose(); appUpdate.dispose(); diff --git a/apps/server/test/helpers/restart-resume.ts b/apps/server/test/helpers/restart-resume.ts new file mode 100644 index 00000000000..b5d89988af9 --- /dev/null +++ b/apps/server/test/helpers/restart-resume.ts @@ -0,0 +1,175 @@ +import { setExperiments } from "@bb/db"; +import { + queuedMessageWaitingOnSchema, + threadScope, + turnScope, +} from "@bb/domain"; +import { + seedEnvironment, + seedProjectWithSource, + seedStoredEvent, + seedThread, + seedThreadRuntimeState, + seedTurnStarted, +} from "./seed.js"; +import type { TestAppHarness } from "./test-app.js"; + +export const ACTIVE_TURN_ID = "turn-1"; + +export function enableRestartContinuation(harness: TestAppHarness): void { + setExperiments(harness.db, { continueThreadsAfterRestarts: true }); +} + +export function seedThreadOnHost( + harness: TestAppHarness, + hostId: string, + status: "idle" | "active", +) { + const { project } = seedProjectWithSource(harness.deps, { hostId }); + const environment = seedEnvironment(harness.deps, { + hostId, + projectId: project.id, + path: `/tmp/restart-resume-${project.id}`, + }); + const thread = seedThread(harness.deps, { + projectId: project.id, + environmentId: environment.id, + status, + }); + seedThreadRuntimeState(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + providerThreadId: `provider-${thread.id}`, + }); + if (status === "active") { + seedTurnStarted(harness.deps, { + environmentId: environment.id, + providerThreadId: `provider-${thread.id}`, + sequence: 3, + threadId: thread.id, + turnId: ACTIVE_TURN_ID, + }); + } + return { environment, thread }; +} + +export function seedCommand( + harness: TestAppHarness, + args: { command: string; environmentId: string; threadId: string }, +) { + seedStoredEvent(harness.deps, { + threadId: args.threadId, + environmentId: args.environmentId, + scope: turnScope(ACTIVE_TURN_ID), + providerThreadId: `provider-${args.threadId}`, + itemId: "command-open", + itemKind: "commandExecution", + sequence: 10, + type: "item/started", + data: { + item: { + id: "command-open", + type: "commandExecution", + command: args.command, + cwd: "/tmp/project", + status: "inProgress", + approvalStatus: null, + }, + }, + }); +} + +export function seedBackgroundJob( + harness: TestAppHarness, + args: { + description: string; + environmentId: string; + taskType: "local_agent" | "local_bash"; + threadId: string; + }, +) { + const item = { + id: "task:job", + type: "backgroundTask", + taskType: args.taskType, + description: args.description, + status: "pending", + taskStatus: "running", + skipTranscript: false, + foreground: false, + }; + const base = { + threadId: args.threadId, + environmentId: args.environmentId, + providerThreadId: `provider-${args.threadId}`, + itemId: item.id, + itemKind: "backgroundTask" as const, + data: { providerThreadId: `provider-${args.threadId}`, item }, + }; + seedStoredEvent(harness.deps, { + ...base, + scope: turnScope("turn-job"), + sequence: 20, + type: "item/started", + }); + seedStoredEvent(harness.deps, { + ...base, + scope: threadScope(), + sequence: 21, + type: "item/backgroundTask/progress", + }); +} + +export function finishBackgroundJob( + harness: TestAppHarness, + args: { environmentId: string; threadId: string }, +) { + seedStoredEvent(harness.deps, { + threadId: args.threadId, + environmentId: args.environmentId, + providerThreadId: `provider-${args.threadId}`, + itemId: "task:job", + itemKind: "backgroundTask", + scope: threadScope(), + sequence: 22, + type: "item/backgroundTask/completed", + data: { + providerThreadId: `provider-${args.threadId}`, + item: { + id: "task:job", + type: "backgroundTask", + taskType: "local_agent", + description: "done", + status: "completed", + taskStatus: "completed", + skipTranscript: false, + foreground: false, + }, + }, + }); +} + +export function sendText( + harness: TestAppHarness, + threadId: string, + text: string, +) { + return harness.app.request(`/api/v1/threads/${threadId}/send`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + input: [{ type: "text", text, mentions: [] }], + mode: "steer-if-active", + model: "gpt-5", + permissionMode: "full", + reasoningLevel: "medium", + serviceTier: "default", + }), + }); +} + +export function waitingOnKind(row: { waitingOn: string | null }): string | null { + return row.waitingOn === null + ? null + : queuedMessageWaitingOnSchema.parse(JSON.parse(row.waitingOn)).kind; +} diff --git a/apps/server/test/system/experiments.test.ts b/apps/server/test/system/experiments.test.ts index 08feac71a1f..515754b04cb 100644 --- a/apps/server/test/system/experiments.test.ts +++ b/apps/server/test/system/experiments.test.ts @@ -41,6 +41,7 @@ describe("experiments settings", () => { applicationLinks: false, changelogPreview: false, compactExpiredCache: false, + continueThreadsAfterRestarts: false, serverMove: false, performanceDiagnostics: false, pluginUpdates: false, @@ -78,6 +79,7 @@ describe("experiments settings", () => { applicationLinks: true, changelogPreview: true, compactExpiredCache: true, + continueThreadsAfterRestarts: false, serverMove: true, performanceDiagnostics: true, pluginUpdates: true, @@ -95,6 +97,7 @@ describe("experiments settings", () => { applicationLinks: true, changelogPreview: true, compactExpiredCache: true, + continueThreadsAfterRestarts: false, serverMove: true, performanceDiagnostics: true, pluginUpdates: true, @@ -110,6 +113,7 @@ describe("experiments settings", () => { applicationLinks: true, changelogPreview: true, compactExpiredCache: true, + continueThreadsAfterRestarts: false, serverMove: true, performanceDiagnostics: true, pluginUpdates: true, @@ -129,6 +133,7 @@ describe("experiments settings", () => { applicationLinks: true, changelogPreview: true, compactExpiredCache: true, + continueThreadsAfterRestarts: false, serverMove: true, performanceDiagnostics: true, pluginUpdates: true, @@ -152,6 +157,7 @@ describe("experiments settings", () => { applicationLinks: false, changelogPreview: true, compactExpiredCache: false, + continueThreadsAfterRestarts: false, serverMove: true, performanceDiagnostics: false, pluginUpdates: false, diff --git a/apps/server/test/threads/restart-resume.test.ts b/apps/server/test/threads/restart-resume.test.ts new file mode 100644 index 00000000000..1acc5043e91 --- /dev/null +++ b/apps/server/test/threads/restart-resume.test.ts @@ -0,0 +1,366 @@ +import { + getLatestThreadSequence, + getThread, + listEvents, + listQueuedThreadMessages, + setExperiments, +} from "@bb/db"; +import { + encodeClientTurnRequestIdNumber, + threadScope, + turnScope, +} from "@bb/domain"; +import { afterEach, describe, expect, it } from "vitest"; +import { handleHostSessionOpened } from "../../src/internal/session-owner-side-effects.js"; +import { setPluginThreadEventEmitter } from "../../src/services/plugins/plugin-thread-events.js"; +import { applyLoggedThreadLifecycleEvent } from "../../src/services/threads/lifecycle-outcome.js"; +import { runQueuedMessageDispatch } from "../../src/services/threads/queued-message-dispatch.js"; +import { + recordResumesForShutdown, + releaseRestartResumes, + scheduleRestartResumes, +} from "../../src/services/threads/restart-resume.js"; +import { + ACTIVE_TURN_ID, + enableRestartContinuation, + seedBackgroundJob, + seedCommand, + seedThreadOnHost, + waitingOnKind, +} from "../helpers/restart-resume.js"; +import { + seedEvent, + seedHostSession, + seedSession, +} from "../helpers/seed.js"; +import { withTestHarness, type TestAppHarness } from "../helpers/test-app.js"; + +afterEach(() => { + setPluginThreadEventEmitter(undefined); +}); + +function recordTurnFailedAnnouncements(): string[] { + const announced: string[] = []; + setPluginThreadEventEmitter({ + emitThreadEvents: () => {}, + emitTerminalInput: () => {}, + emitHostDeleted: () => {}, + emitEnvironmentRemoved: () => {}, + emitThreadCreated: () => {}, + emitThreadActive: () => {}, + emitThreadIdle: () => {}, + emitThreadFailed: () => {}, + emitThreadArchived: () => {}, + emitThreadDeleted: () => {}, + emitInteractionPending: () => {}, + emitMessageQueued: () => {}, + emitMessageDispatched: () => {}, + emitMessageCancelled: () => {}, + emitThreadUnarchived: () => {}, + emitThreadParentChanged: () => {}, + emitTurnFailed: (threadId) => announced.push(threadId), + }); + return announced; +} + +function turnRequests(harness: TestAppHarness, threadId: string) { + return listEvents(harness.db, { threadId }).filter( + (event) => event.type === "client/turn/requested", + ); +} + +async function restartDaemon( + harness: TestAppHarness, + host: { session: Parameters[1]["openedSession"]; hostId: string }, +) { + const replacement = seedSession(harness.deps, host.hostId, { + instanceId: "instance-2", + }); + await handleHostSessionOpened(harness.deps, { + activeThreads: [], + hostId: host.hostId, + openedSession: replacement, + previousSession: host.session, + undeliveredEventThreadIds: [], + }); +} + +async function sweepPastResumes(harness: TestAppHarness): Promise { + await runQueuedMessageDispatch(harness.deps, { + kind: "time-reached", + now: Date.now() + 120_000, + }); +} + +function seedAutomaticResumeTurns( + harness: TestAppHarness, + args: { count: number; environmentId: string; threadId: string }, +) { + for (let index = 0; index < args.count; index += 1) { + const sequence = + getLatestThreadSequence(harness.db, { threadId: args.threadId }) + 1; + seedEvent(harness.deps, { + threadId: args.threadId, + environmentId: args.environmentId, + providerThreadId: `provider-${args.threadId}`, + sequence, + type: "client/turn/requested", + scope: threadScope(), + data: { + direction: "outbound", + requestId: encodeClientTurnRequestIdNumber({ value: sequence }), + input: [{ type: "text", text: "bb restarted" }], + target: { kind: "new-turn" }, + execution: { + model: "gpt-5", + serviceTier: "default", + reasoningLevel: "medium", + permissionMode: "full", + source: "client/turn/requested", + }, + initiator: "system", + senderThreadId: null, + systemMessageKind: "restart-resume", + request: { method: "turn/start", params: {} }, + source: "tell", + }, + }); + } +} + +describe("continuing threads after a restart", () => { + it("resumes the turns and awaited background work a daemon restart cut, without a retry", () => + withTestHarness(async (harness) => { + enableRestartContinuation(harness); + const announced = recordTurnFailedAnnouncements(); + const { host, session } = seedHostSession(harness.deps, { + id: "host-crashed", + name: "Crashed", + }); + const working = seedThreadOnHost(harness, host.id, "active"); + seedCommand(harness, { + command: "pnpm test", + environmentId: working.environment.id, + threadId: working.thread.id, + }); + const delegating = seedThreadOnHost(harness, host.id, "idle"); + seedBackgroundJob(harness, { + description: "Review the migration", + environmentId: delegating.environment.id, + taskType: "local_agent", + threadId: delegating.thread.id, + }); + const serving = seedThreadOnHost(harness, host.id, "idle"); + seedBackgroundJob(harness, { + description: "pnpm dev", + environmentId: serving.environment.id, + taskType: "local_bash", + threadId: serving.thread.id, + }); + + await restartDaemon(harness, { hostId: host.id, session }); + + expect(getThread(harness.db, working.thread.id)?.status).toBe("error"); + expect(announced).toEqual([]); + expect( + listEvents(harness.db, { threadId: working.thread.id }) + .filter((event) => event.type === "system/error") + .map((event) => JSON.parse(event.data).detail), + ).toEqual(["bb will send this thread a message to continue."]); + const [resume] = listQueuedThreadMessages(harness.db, working.thread.id); + expect(resume?.content).toContain( + "bb restarted, or the machine running this thread did, while you were running `pnpm test`. The restart interrupted it; the user did not reject it. It may have completed, partly completed, never run, or still be running", + ); + expect(resume && waitingOnKind(resume)).toBe("time"); + expect( + listQueuedThreadMessages(harness.db, delegating.thread.id)[0]?.content, + ).toContain( + "while this background work was running: `Review the migration`", + ); + expect( + listQueuedThreadMessages(harness.db, serving.thread.id), + ).toHaveLength(0); + + await sweepPastResumes(harness); + + expect(turnRequests(harness, working.thread.id)).toHaveLength(2); + expect( + listQueuedThreadMessages(harness.db, working.thread.id), + ).toHaveLength(0); + })); + + it("keeps a thread's resume parked across a quit, then drops it if the turn finished or the user stopped it", () => + withTestHarness(async (harness) => { + enableRestartContinuation(harness); + const surviving = seedHostSession(harness.deps, { + id: "host-surviving", + name: "Surviving", + }); + const restarted = seedHostSession(harness.deps, { + id: "host-restarted", + name: "Restarted", + }); + const finishes = seedThreadOnHost(harness, surviving.host.id, "active"); + const stopped = seedThreadOnHost(harness, surviving.host.id, "active"); + const cut = seedThreadOnHost(harness, restarted.host.id, "active"); + await recordResumesForShutdown(harness.deps, { now: Date.now() }); + for (const { thread } of [finishes, stopped, cut]) { + expect( + listQueuedThreadMessages(harness.db, thread.id).map(waitingOnKind), + ).toEqual(["time"]); + } + releaseRestartResumes(harness.deps, { now: Date.now() }); + + for (const { thread } of [finishes, stopped, cut]) { + expect( + listQueuedThreadMessages(harness.db, thread.id).map(waitingOnKind), + ).toEqual(["thread-busy"]); + } + + await restartDaemon(harness, { + hostId: restarted.host.id, + session: restarted.session, + }); + seedEvent(harness.deps, { + threadId: finishes.thread.id, + environmentId: finishes.environment.id, + providerThreadId: `provider-${finishes.thread.id}`, + sequence: + getLatestThreadSequence(harness.db, { + threadId: finishes.thread.id, + }) + 1, + type: "turn/completed", + scope: turnScope(ACTIVE_TURN_ID), + data: { + providerThreadId: `provider-${finishes.thread.id}`, + status: "completed", + }, + }); + applyLoggedThreadLifecycleEvent(harness.deps, { + event: { type: "run.succeeded" }, + threadId: finishes.thread.id, + }); + seedEvent(harness.deps, { + threadId: stopped.thread.id, + environmentId: stopped.environment.id, + sequence: + getLatestThreadSequence(harness.db, { threadId: stopped.thread.id }) + + 1, + type: "system/thread/interrupted", + scope: threadScope(), + data: { reason: "manual-stop" }, + }); + harness.db.$client + .prepare("UPDATE threads SET status = ? WHERE id = ?") + .run("idle", stopped.thread.id); + for (const { thread } of [finishes, stopped]) { + await runQueuedMessageDispatch(harness.deps, { + kind: "thread-ready", + threadId: thread.id, + }); + } + await sweepPastResumes(harness); + + expect(turnRequests(harness, finishes.thread.id)).toHaveLength(1); + expect(turnRequests(harness, stopped.thread.id)).toHaveLength(1); + expect(turnRequests(harness, cut.thread.id)).toHaveLength(2); + for (const { thread } of [finishes, stopped, cut]) { + expect(listQueuedThreadMessages(harness.db, thread.id)).toHaveLength(0); + } + })); + + it("resumes a child before the parent waiting on it, and tells the parent nothing else", () => + withTestHarness(async (harness) => { + enableRestartContinuation(harness); + const { host, session } = seedHostSession(harness.deps, { + id: "host-family", + name: "Family", + }); + const parent = seedThreadOnHost(harness, host.id, "active"); + seedCommand(harness, { + command: "bb thread wait thr_child", + environmentId: parent.environment.id, + threadId: parent.thread.id, + }); + const child = seedThreadOnHost(harness, host.id, "active"); + harness.db.$client + .prepare("UPDATE threads SET parent_thread_id = ? WHERE id = ?") + .run(parent.thread.id, child.thread.id); + + await restartDaemon(harness, { hostId: host.id, session }); + scheduleRestartResumes(harness.deps, { + hostIdByThread: new Map([ + [parent.thread.id, host.id], + [child.thread.id, host.id], + ]), + startAt: Date.now(), + }); + + const [parentResume] = listQueuedThreadMessages( + harness.db, + parent.thread.id, + ); + const [childResume] = listQueuedThreadMessages( + harness.db, + child.thread.id, + ); + expect(childResume?.sendAt ?? Infinity).toBeLessThan( + parentResume?.sendAt ?? 0, + ); + expect(parentResume?.content).toContain( + "while you were running `bb thread wait thr_child`", + ); + await new Promise((resolvePromise) => setTimeout(resolvePromise, 2_300)); + expect( + listQueuedThreadMessages(harness.db, parent.thread.id), + ).toHaveLength(1); + expect(turnRequests(harness, parent.thread.id)).toHaveLength(1); + })); + + it("stops resuming a thread after three automatic resumes in a row", () => + withTestHarness(async (harness) => { + enableRestartContinuation(harness); + const { host, session } = seedHostSession(harness.deps, { + id: "host-looping", + name: "Looping", + }); + const looping = seedThreadOnHost(harness, host.id, "active"); + seedAutomaticResumeTurns(harness, { + count: 3, + environmentId: looping.environment.id, + threadId: looping.thread.id, + }); + const recovering = seedThreadOnHost(harness, host.id, "active"); + seedAutomaticResumeTurns(harness, { + count: 2, + environmentId: recovering.environment.id, + threadId: recovering.thread.id, + }); + + await restartDaemon(harness, { hostId: host.id, session }); + + expect( + listQueuedThreadMessages(harness.db, looping.thread.id), + ).toHaveLength(0); + expect( + listQueuedThreadMessages(harness.db, recovering.thread.id), + ).toHaveLength(1); + })); + + it("leaves interrupted threads alone while the experiment is off", () => + withTestHarness(async (harness) => { + setExperiments(harness.db, { continueThreadsAfterRestarts: false }); + const { host, session } = seedHostSession(harness.deps, { + id: "host-off", + name: "Off", + }); + const working = seedThreadOnHost(harness, host.id, "active"); + + await restartDaemon(harness, { hostId: host.id, session }); + + expect(getThread(harness.db, working.thread.id)?.status).toBe("error"); + expect( + listQueuedThreadMessages(harness.db, working.thread.id), + ).toHaveLength(0); + })); +}); diff --git a/apps/server/test/threads/turn-failed-retry.test.ts b/apps/server/test/threads/turn-failed-retry.test.ts index 603c778f9de..d70f5ecb36f 100644 --- a/apps/server/test/threads/turn-failed-retry.test.ts +++ b/apps/server/test/threads/turn-failed-retry.test.ts @@ -10,6 +10,8 @@ import { getThread, listEvents, listQueuedThreadMessages, + setExperiments, + setQueuedThreadMessageFailureReason, setThreadExecutionOverride, } from "@bb/db"; import type { ThreadQueuedMessage } from "@bb/domain"; @@ -20,6 +22,7 @@ import { setPluginHookProvider, type PluginHookRegistration, } from "../../src/services/plugins/plugin-hook-registry.js"; +import { handleHostSessionOpened } from "../../src/internal/session-owner-side-effects.js"; import { setPluginThreadEventEmitter } from "../../src/services/plugins/plugin-thread-events.js"; import { applyLoggedThreadLifecycleEvent } from "../../src/services/threads/lifecycle-outcome.js"; import { runQueuedMessageDispatch } from "../../src/services/threads/queued-message-dispatch.js"; @@ -31,6 +34,7 @@ import { seedEvent, seedHostSession, seedProjectWithSource, + seedSession, seedThread, seedThreadRuntimeState, } from "../helpers/seed.js"; @@ -102,7 +106,7 @@ function recordTurnFailedAnnouncements(): string[] { * requested` events exist, and a second seeded one would hide a duplicate. */ function seedFailableThread(harness: TestAppHarness, hostId: string) { - const { host } = seedHostSession(harness.deps, { id: hostId }); + const { host, session } = seedHostSession(harness.deps, { id: hostId }); const { project } = seedProjectWithSource(harness.deps, { hostId: host.id, path: WORKSPACE_PATH, @@ -127,6 +131,7 @@ function seedFailableThread(harness: TestAppHarness, hostId: string) { environment, host, project, + session, thread, requestId: lastTurnRequest(harness, thread.id).requestId, }; @@ -319,6 +324,37 @@ async function sweepPastResume(harness: TestAppHarness): Promise { }); } +function sendUserText(harness: TestAppHarness, threadId: string, text: string) { + return harness.app.request(`/api/v1/threads/${threadId}/send`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + input: [{ type: "text", text, mentions: [] }], + mode: "steer-if-active", + model: "gpt-5", + permissionMode: "full", + reasoningLevel: "medium", + serviceTier: "default", + }), + }); +} + +async function restartDaemon( + harness: TestAppHarness, + failable: ReturnType, +): Promise { + setExperiments(harness.db, { continueThreadsAfterRestarts: true }); + await handleHostSessionOpened(harness.deps, { + activeThreads: [], + hostId: failable.host.id, + openedSession: seedSession(harness.deps, failable.host.id, { + instanceId: "instance-2", + }), + previousSession: failable.session, + undeliveredEventThreadIds: [], + }); +} + describe("the turn.failed announcement", () => { it("carries the failed turn's ids and the provider's failure facts", async () => { await withTestHarness(async (harness) => { @@ -434,6 +470,75 @@ describe("the turn.failed announcement", () => { expect(announced).toEqual([failing.id]); }); }); + + it("is not announced for a turn a daemon restart cut, and the resume then starts the next turn", async () => { + await withTestHarness(async (harness) => { + const announced = recordTurnFailedAnnouncements(); + const failable = seedFailableThread(harness, "host-restart"); + await restartDaemon(harness, failable); + expect(getThread(harness.db, failable.thread.id)?.status).toBe("error"); + expect(announced).toEqual([]); + + await sweepPastResume(harness); + + expect(turnRequests(harness, failable.thread.id)).toHaveLength(2); + expect( + listQueuedThreadMessages(harness.db, failable.thread.id), + ).toHaveLength(0); + failThread(harness, failable.thread.id); + expect(announced).toEqual([failable.thread.id]); + }); + }); + + it("is announced again once the thread's resume could not be delivered", async () => { + await withTestHarness(async (harness) => { + const announced = recordTurnFailedAnnouncements(); + const failable = seedFailableThread(harness, "host-restart"); + await restartDaemon(harness, failable); + const [resume] = listQueuedThreadMessages(harness.db, failable.thread.id); + if (resume === undefined) throw new Error("expected a resume"); + + setQueuedThreadMessageFailureReason(harness.db, harness.hub, { + id: resume.id, + threadId: failable.thread.id, + failureReason: "Provider unavailable", + now: Date.now(), + retryDelaysMs: [], + }); + expect( + (await sendUserText(harness, failable.thread.id, "Try again")).status, + ).toBe(200); + failThread(harness, failable.thread.id); + + expect(announced).toEqual([failable.thread.id]); + }); + }); + + it("drops the resume when the user started a newer turn after the restart", async () => { + await withTestHarness(async (harness) => { + const failable = seedFailableThread(harness, "host-restart"); + await restartDaemon(harness, failable); + + expect( + (await sendUserText(harness, failable.thread.id, "Do something else")) + .status, + ).toBe(200); + expect(turnRequests(harness, failable.thread.id)).toHaveLength(2); + applyLoggedThreadLifecycleEvent(harness.deps, { + event: { type: "run.succeeded" }, + threadId: failable.thread.id, + }); + await sweepPastResume(harness); + + expect(turnRequests(harness, failable.thread.id)).toHaveLength(2); + expect(lastTurnRequest(harness, failable.thread.id).input[0]?.text).toBe( + "Do something else", + ); + expect( + listQueuedThreadMessages(harness.db, failable.thread.id), + ).toHaveLength(0); + }); + }); }); describe("retrying a failed turn", () => { diff --git a/docs/configuration.md b/docs/configuration.md index da0751ebe17..2941e691ecd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -895,6 +895,20 @@ off. An experiment you set yourself keeps your value either way. next send passes `compactFirst` so the server runs `/compact` and queues the message behind it. Clicking the chip keeps the full history for that one send. +- `continueThreadsAfterRestarts` resumes threads a restart interrupted. bb + records one resume message per running thread when the server shuts down, + and per cut turn or awaited background work (not background shells) when a + machine's daemon comes back as a new process after a quit, crash, update, or + reboot. The message names what the thread was doing, such as the command it + was running, and asks the agent to check whether it finished or is still + running before continuing. Resumes go a few seconds apart per machine, child + threads before their parents, ahead of messages already queued for the + thread. bb drops a resume if the turn finished after all, the user stopped + it, or the user sent a newer message first, and stops after three automatic + resumes in a row. A turn a restart cut off is not announced to retry + policies, and the parent of an interrupted child is told only when the child + finishes. `bb thread wait` keeps waiting on a failed thread whose resume is + scheduled. - `serverMove` enables Move server here in Settings → Machines, `bb server move`, `bb server export`, and old server copy deletion; while off those routes return 403 `server_move_experiment_disabled` (status and cancel diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 85f75411bbd..3b1a25e5c2f 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -2195,6 +2195,55 @@ export function getLatestThreadInterruptedReason( return systemThreadInterruptedReasonSchema.parse(row.reason); } +export function getLatestThreadInterruption( + db: DbQueryConnection, + args: { threadId: string }, +): { reason: SystemThreadInterruptedReason; sequence: number } | null { + const row = db + .select({ + reason: sql`json_extract(${events.data}, '$.reason')`, + sequence: events.sequence, + }) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + eq(events.type, "system/thread/interrupted"), + ), + ) + .orderBy(desc(events.sequence)) + .limit(1) + .get(); + if (!row) return null; + return { + reason: systemThreadInterruptedReasonSchema.parse(row.reason), + sequence: row.sequence, + }; +} + +export function listLatestTurnRequestSystemMessageKinds( + db: DbQueryConnection, + args: { limit: number; threadId: string }, +): (string | null)[] { + return db + .select({ + kind: sql< + string | null + >`json_extract(${events.data}, '$.systemMessageKind')`, + }) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + eq(events.type, "client/turn/requested"), + ), + ) + .orderBy(desc(events.sequence)) + .limit(args.limit) + .all() + .map((row) => row.kind); +} + export function listStoredTurnStartedRowsByTurnIdsUpToSequence( db: DbConnection, args: ListStoredTurnStartedRowsByTurnIdsUpToSequenceArgs, @@ -4399,6 +4448,42 @@ export function pruneTokenUsageEventsInTransaction( }); } +export function getRunningCommandItemData( + db: DbQueryConnection, + args: { threadId: string }, +): string | null { + const startedType = "item/started" satisfies ThreadEventType; + const completedType = "item/completed" satisfies ThreadEventType; + const finished = alias(events, "finished_command"); + const row = db + .select({ data: events.data }) + .from(events) + .where( + and( + eq(events.threadId, args.threadId), + sql`${events.itemKind} = 'commandExecution'`, + eq(events.type, startedType), + isNotNull(events.itemId), + notExists( + db + .select({ one: sql`1` }) + .from(finished) + .where( + and( + eq(finished.threadId, events.threadId), + eq(finished.itemId, events.itemId), + eq(finished.type, completedType), + ), + ), + ), + ), + ) + .orderBy(desc(events.sequence)) + .limit(1) + .get(); + return row?.data ?? null; +} + export function listOpenBackgroundTaskItemRowsForHost( db: DbQueryConnection, args: ListOpenBackgroundTaskItemRowsForHostArgs, diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index c731ff0c21f..3c5edf6d4d0 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -365,6 +365,9 @@ export { listTodoSnapshotEventRowsForThread, listOpenTurnInputAcceptedRowsByThreadIds, listOpenBackgroundTaskItemRowsForHost, + getRunningCommandItemData, + getLatestThreadInterruption, + listLatestTurnRequestSystemMessageKinds, listOpenBackgroundTaskItemRowsForThread, listThreadEventSuffixBackgroundSettlementRows, listThreadIdsStoppedSinceLastTurnStart, @@ -419,6 +422,7 @@ export type { export { createPendingInteraction, getActivePendingInteractionForThread, + listThreadIdsWithActivePendingInteractions, hasTurnBoundActivePendingInteractionForThread, getPendingInteraction, getPendingInteractionByProviderRequest, @@ -467,6 +471,10 @@ export { listQueuedThreadMessagesForApi, listQueuedThreadMessagesByWaitHolder, listQueuedThreadMessagesWaitingOnKind, + listUnclaimedQueuedSystemNotices, + getPendingQueuedSystemNotice, + moveQueuedThreadMessageToFront, + rescheduleQueuedThreadMessage, listRetryableFailedQueuedThreadMessages, listThreadIdsWithHostOfflineQueueWaits, releaseQueuedMessageClaim, diff --git a/packages/db/src/data/pending-interactions.ts b/packages/db/src/data/pending-interactions.ts index ad6802fe085..e1b003ffb7a 100644 --- a/packages/db/src/data/pending-interactions.ts +++ b/packages/db/src/data/pending-interactions.ts @@ -399,3 +399,14 @@ export function interruptPendingInteractionsForThreadIds( threadIds: args.threadIds, }); } + +export function listThreadIdsWithActivePendingInteractions( + db: PendingInteractionReadConnection, +): string[] { + return db + .selectDistinct({ threadId: pendingInteractions.threadId }) + .from(pendingInteractions) + .where(inArray(pendingInteractions.status, ["pending", "resolving"])) + .all() + .map((row) => row.threadId); +} diff --git a/packages/db/src/data/queued-thread-messages.ts b/packages/db/src/data/queued-thread-messages.ts index 57314fe08ed..20efe8e86a3 100644 --- a/packages/db/src/data/queued-thread-messages.ts +++ b/packages/db/src/data/queued-thread-messages.ts @@ -36,6 +36,7 @@ import type { PromptInput, QueuedMessagePayload, QueuedMessageSystemNotice, + SystemMessageKind, StartedOnBehalfOf, ThreadCreateOrigin, QueuedMessageWaitHolder, @@ -2202,6 +2203,107 @@ export function listQueuedThreadMessagesWaitingOnKind( .all(); } +export function listUnclaimedQueuedSystemNotices( + db: DbQueryConnection, + kind: SystemMessageKind, +): QueuedThreadMessageRow[] { + return db + .select() + .from(queuedThreadMessages) + .where( + and( + sql`json_extract(${queuedThreadMessages.systemNotice}, '$.kind') = ${kind}`, + isNull(queuedThreadMessages.claimToken), + ), + ) + .orderBy(asc(queuedThreadMessages.createdAt), asc(queuedThreadMessages.id)) + .all(); +} + +export function getPendingQueuedSystemNotice( + db: DbQueryConnection, + args: { kind: SystemMessageKind; threadId: string }, +): QueuedThreadMessageRow | null { + return ( + db + .select() + .from(queuedThreadMessages) + .where( + and( + eq(queuedThreadMessages.threadId, args.threadId), + sql`json_extract(${queuedThreadMessages.systemNotice}, '$.kind') = ${args.kind}`, + isNull(queuedThreadMessages.claimToken), + isNull(queuedThreadMessages.failureReason), + ), + ) + .limit(1) + .get() ?? null + ); +} + +export function moveQueuedThreadMessageToFront( + db: DbConnection, + notifier: DbNotifier, + args: { id: string; threadId: string }, +): boolean { + const moved = db.transaction( + (tx) => { + const [first] = listQueuedThreadMessages(tx, args.threadId); + if (first === undefined || first.id === args.id) return false; + return ( + tx + .update(queuedThreadMessages) + .set({ + sortKey: createOrderKeyBetween({ + previousKey: null, + nextKey: first.sortKey, + }), + updatedAt: Date.now(), + }) + .where( + and( + eq(queuedThreadMessages.id, args.id), + eq(queuedThreadMessages.threadId, args.threadId), + isNull(queuedThreadMessages.claimToken), + ), + ) + .returning({ id: queuedThreadMessages.id }) + .get() !== undefined + ); + }, + { behavior: "immediate" }, + ); + if (moved) notifier.notifyThread(args.threadId, ["queue-changed"]); + return moved; +} + +export function rescheduleQueuedThreadMessage( + db: DbConnection, + notifier: DbNotifier, + args: { id: string; sendAt: number }, +): QueuedThreadMessageRow | null { + const updated = db.transaction( + (tx) => { + const row = getQueuedThreadMessage(tx, args.id); + if (!row || isQueuedThreadMessageClaimed(row)) return null; + return transitionQueuedMessage(tx, { + now: Date.now(), + row, + transition: { + event: "wait.recorded", + sendAt: args.sendAt, + waitingOn: { kind: "time" }, + }, + }); + }, + { behavior: "immediate" }, + ); + if (updated !== null) { + notifier.notifyThread(updated.threadId, ["queue-changed"]); + } + return updated; +} + /** * Whether a retry of this original turn request already exists on the queue. * diff --git a/packages/domain/src/experiments.ts b/packages/domain/src/experiments.ts index acea6082837..ea3ea4a669f 100644 --- a/packages/domain/src/experiments.ts +++ b/packages/domain/src/experiments.ts @@ -4,6 +4,7 @@ export const experimentKeys = [ "applicationLinks", "changelogPreview", "compactExpiredCache", + "continueThreadsAfterRestarts", "serverMove", "performanceDiagnostics", "pluginUpdates", @@ -31,6 +32,7 @@ export const defaultExperiments: Experiments = { applicationLinks: false, changelogPreview: false, compactExpiredCache: false, + continueThreadsAfterRestarts: false, serverMove: false, performanceDiagnostics: false, pluginUpdates: false, @@ -47,6 +49,7 @@ export const earlyAdopterExperiments: Experiments = { applicationLinks: true, changelogPreview: false, compactExpiredCache: true, + continueThreadsAfterRestarts: false, serverMove: false, performanceDiagnostics: false, pluginUpdates: true, diff --git a/packages/domain/src/system-message.ts b/packages/domain/src/system-message.ts index 631cba6a7a1..60685b23b38 100644 --- a/packages/domain/src/system-message.ts +++ b/packages/domain/src/system-message.ts @@ -23,6 +23,7 @@ const systemMessageKindValues = [ "child-outcome-batch", "tool-result-delivered", "environment-directory-updated", + "restart-resume", "unlabeled", ] as const; export const systemMessageKindSchema = z.enum(systemMessageKindValues); @@ -71,5 +72,10 @@ export const systemMessageSubjectSchema = z.discriminatedUnion("kind", [ kind: z.literal("directory"), path: z.string(), }), + z.object({ + kind: z.literal("restart"), + afterSequence: z.number().int(), + turnId: z.string().nullable(), + }), ]); export type SystemMessageSubject = z.infer; diff --git a/packages/sdk/src/areas/threads.ts b/packages/sdk/src/areas/threads.ts index 48b7d3da5ed..a3216d573d0 100644 --- a/packages/sdk/src/areas/threads.ts +++ b/packages/sdk/src/areas/threads.ts @@ -1120,6 +1120,14 @@ export function createThreadsArea(args: CreateSdkAreaArgs): ThreadsArea { ); }, }; + const hasScheduledQueuedMessage = async (input: { + signal?: AbortSignal | undefined; + threadId: string; + }): Promise => + (await queuedMessages.list(input)).some( + (row) => row.waitingOn?.kind === "time" && row.failureReason === null, + ); + const queuedMessages: ThreadQueuedMessagesArea = { async create(input) { const { threadId, ...json } = input; @@ -1722,7 +1730,13 @@ export function createThreadsArea(args: CreateSdkAreaArgs): ThreadsArea { threadId: input.threadId, }; } - if (isThreadWaitTargetUnreachable(thread.status, target)) { + if ( + isThreadWaitTargetUnreachable(thread.status, target) && + !(await hasScheduledQueuedMessage({ + signal: input.signal, + threadId: input.threadId, + })) + ) { throw new ThreadWaitUnreachableError({ currentStatus: thread.status, target, diff --git a/packages/templates/src/templates/bb-guide-customization.md b/packages/templates/src/templates/bb-guide-customization.md index 39d749e5d44..955811aee23 100644 --- a/packages/templates/src/templates/bb-guide-customization.md +++ b/packages/templates/src/templates/bb-guide-customization.md @@ -187,6 +187,10 @@ remote client's localStorage. release notes as a dismissible card on Settings → Updates. Experiments that gate a feature are documented with it: `serverMove` in `bb guide machines`, `performanceDiagnostics` under Server performance diagnostics below. +`continueThreadsAfterRestarts` sends each thread a quit, crash, update, or +machine restart interrupted one message naming what it was doing, so the agent +checks it and carries on; bb drops it if the turn finished, the user stopped +it, or the user sent something newer. `timelinePush` sends an open thread's timeline updates over the live connection as row deltas instead of reloading the latest window after each update. `compactExpiredCache` offers a Compact chip in the composer when an idle diff --git a/packages/templates/src/templates/bb-guide-thread-inspection.md b/packages/templates/src/templates/bb-guide-thread-inspection.md index 6cfb8041dd0..01d01d40c7d 100644 --- a/packages/templates/src/templates/bb-guide-thread-inspection.md +++ b/packages/templates/src/templates/bb-guide-thread-inspection.md @@ -109,6 +109,10 @@ Inspecting: --timeout Seconds, or a duration with a unit: 90s, 20m, 4h (default: 1200s / 20 min) --poll-interval Milliseconds, or a duration with a unit + Waiting for idle on a thread in `error` exits 4 at once, unless the thread + has a message scheduled to send on its own, such as a retry or a resume after + a restart; then it keeps waiting. + Opening threads and files in the app: In chat, reference a thread as @thread:thr_abc123, substituting its actual ID. diff --git a/packages/templates/src/templates/system-message-restart-resume.md b/packages/templates/src/templates/system-message-restart-resume.md new file mode 100644 index 00000000000..66e2a174d1f --- /dev/null +++ b/packages/templates/src/templates/system-message-restart-resume.md @@ -0,0 +1,14 @@ +--- +kind: prompt +title: Resume after a restart +summary: Starts the next turn for a thread that a bb restart, a bb update or a machine restart interrupted. +intent: Let the agent carry on from its timeline without the user having to say "continue", checking the one step the restart may have cut off. +editingNotes: The thread's timeline already holds every finished step, so the message only names what was interrupted. Keep the check-before-continuing line; a stopped command or job may be partly applied, and repeating it may duplicate its effects. Keep saying that the restart, not the user, stopped the work; agents otherwise read the stop as a rejection. +variables: + restarted: What restarted, already phrased, such as "bb restarted to install an update". + interrupted: What the restart interrupted, already phrased, such as "while you were running `pnpm test`". + check: What to check before continuing, already phrased. +--- +[bb system] + +{{restarted}} {{interrupted}}. {{check}} From 344d27137855087f20e9894062cd13df6c15a0d9 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Thu, 29 Oct 2026 16:47:22 -0700 Subject: [PATCH 2/4] Tell a resumed agent that a rejection in its history was the restart A graceful shutdown interrupts the provider, and Claude Code records the cut tool call as rejected by the user. A resumed agent trusted that over the resume message and stopped to ask, so the message now says a recorded interruption or rejection was the restart. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/services/threads/restart-resume.ts | 6 +++--- apps/server/test/threads/restart-resume.test.ts | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/apps/server/src/services/threads/restart-resume.ts b/apps/server/src/services/threads/restart-resume.ts index 84268e04f49..7e46b64fcd8 100644 --- a/apps/server/src/services/threads/restart-resume.ts +++ b/apps/server/src/services/threads/restart-resume.ts @@ -144,7 +144,7 @@ function turnResumeText(args: { ? "while you were waiting for the user" : `while you were waiting for the user to approve \`${args.command}\``, check: - "The restart cancelled that request; the user did not answer it. Ask again if you still need it, and do not decide for them.", + "The restart cancelled that request; the user did not answer it, even if your history shows it as interrupted or rejected. Ask again if you still need it, and do not decide for them.", }; } if (args.command === null) { @@ -152,14 +152,14 @@ function turnResumeText(args: { restarted, interrupted: "while you were working", check: - "The restart interrupted your turn; the user did not. Check whether your last step finished or is still running, then continue the task without repeating work that already happened.", + "The restart interrupted your turn; the user did not, even if your history shows it as interrupted or rejected by the user. Check whether your last step finished or is still running, then continue the task without repeating work that already happened.", }; } return { restarted, interrupted: `while you were running \`${args.command}\``, check: - "The restart interrupted it; the user did not reject it. It may have completed, partly completed, never run, or still be running, so check its effects before you continue, and rerun it only if that is safe.", + "The restart interrupted it; the user did not reject it, even if your history shows it as interrupted or rejected by the user. It may have completed, partly completed, never run, or still be running, so check its effects, then continue the task, rerunning it only if that is safe.", }; } diff --git a/apps/server/test/threads/restart-resume.test.ts b/apps/server/test/threads/restart-resume.test.ts index 1acc5043e91..6421728e62d 100644 --- a/apps/server/test/threads/restart-resume.test.ts +++ b/apps/server/test/threads/restart-resume.test.ts @@ -169,7 +169,7 @@ describe("continuing threads after a restart", () => { ).toEqual(["bb will send this thread a message to continue."]); const [resume] = listQueuedThreadMessages(harness.db, working.thread.id); expect(resume?.content).toContain( - "bb restarted, or the machine running this thread did, while you were running `pnpm test`. The restart interrupted it; the user did not reject it. It may have completed, partly completed, never run, or still be running", + "bb restarted, or the machine running this thread did, while you were running `pnpm test`. The restart interrupted it; the user did not reject it, even if your history shows it as interrupted or rejected by the user.", ); expect(resume && waitingOnKind(resume)).toBe("time"); expect( From 0df5d9ced663140dd314272a214cfccd8ff50ddb Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Fri, 30 Oct 2026 10:43:15 -0700 Subject: [PATCH 3/4] Simplify restart continuation prompts --- .../src/services/threads/restart-resume.ts | 81 ++----------------- .../test/threads/restart-resume.test.ts | 12 ++- docs/configuration.md | 8 +- packages/db/src/data/events.ts | 36 --------- packages/db/src/data/index.ts | 2 - packages/db/src/data/pending-interactions.ts | 11 --- .../src/templates/bb-guide-customization.md | 4 +- ...ystem-message-restart-resume-background.md | 11 +++ .../system-message-restart-resume.md | 13 +-- 9 files changed, 33 insertions(+), 145 deletions(-) create mode 100644 packages/templates/src/templates/system-message-restart-resume-background.md diff --git a/apps/server/src/services/threads/restart-resume.ts b/apps/server/src/services/threads/restart-resume.ts index 7e46b64fcd8..31359db1567 100644 --- a/apps/server/src/services/threads/restart-resume.ts +++ b/apps/server/src/services/threads/restart-resume.ts @@ -8,13 +8,11 @@ import { getLatestThreadInterruption, getLatestThreadSequence, getPendingQueuedSystemNotice, - getRunningCommandItemData, getThread, listLatestTurnRequestSystemMessageKinds, listQueuedThreadMessages, listRunningThreads, listStoredTurnCompletedRowsByTurnIds, - listThreadIdsWithActivePendingInteractions, listUnclaimedQueuedSystemNotices, moveQueuedThreadMessageToFront, rescheduleQueuedThreadMessage, @@ -47,9 +45,6 @@ const ORDERED_BEHIND_RESUME_WAITS = new Set([ "turn-starting", ]); -const runningCommandItemDataSchema = z.object({ - item: z.object({ command: z.string().min(1) }), -}); const turnCompletedDataSchema = z.object({ status: threadEventTurnStatusSchema, }); @@ -68,12 +63,6 @@ interface PendingRestartResume { subject: RestartResumeSubject; } -interface ResumeText { - check: string; - interrupted: string; - restarted: string; -} - type RestartResumeDeps = LoggedPendingInteractionWorkSessionDeps; export function isRestartContinuationEnabled(db: DbConnection): boolean { @@ -118,62 +107,6 @@ function threadHostId(db: DbConnection, threadId: string): string | null { : (getEnvironment(db, environmentId)?.hostId ?? null); } -function runningCommand(db: DbQueryConnection, threadId: string): string | null { - const data = getRunningCommandItemData(db, { threadId }); - if (data === null) return null; - try { - const parsed = runningCommandItemDataSchema.safeParse(JSON.parse(data)); - return parsed.success ? parsed.data.item.command : null; - } catch { - return null; - } -} - -const RESTARTED = "bb restarted, or the machine running this thread did,"; - -function turnResumeText(args: { - awaitingUser: boolean; - command: string | null; -}): ResumeText { - const restarted = RESTARTED; - if (args.awaitingUser) { - return { - restarted, - interrupted: - args.command === null - ? "while you were waiting for the user" - : `while you were waiting for the user to approve \`${args.command}\``, - check: - "The restart cancelled that request; the user did not answer it, even if your history shows it as interrupted or rejected. Ask again if you still need it, and do not decide for them.", - }; - } - if (args.command === null) { - return { - restarted, - interrupted: "while you were working", - check: - "The restart interrupted your turn; the user did not, even if your history shows it as interrupted or rejected by the user. Check whether your last step finished or is still running, then continue the task without repeating work that already happened.", - }; - } - return { - restarted, - interrupted: `while you were running \`${args.command}\``, - check: - "The restart interrupted it; the user did not reject it, even if your history shows it as interrupted or rejected by the user. It may have completed, partly completed, never run, or still be running, so check its effects, then continue the task, rerunning it only if that is safe.", - }; -} - -function backgroundResumeText(jobs: readonly string[]): ResumeText { - return { - restarted: RESTARTED, - interrupted: `while this background work was running: ${jobs - .map((job) => `\`${job}\``) - .join(", ")}`, - check: - "The restart may have stopped it. Check whether each is still running and how far it got, and restart one only when that is safe; otherwise explain what needs attention.", - }; -} - function hasReachedResumeLimit(db: DbQueryConnection, threadId: string): boolean { const kinds = listLatestTurnRequestSystemMessageKinds(db, { limit: MAX_CONSECUTIVE_RESTART_RESUMES, @@ -189,9 +122,6 @@ export async function recordRestartResumes( deps: RestartResumeDeps, args: { now: number; work: readonly RestartWork[] }, ): Promise { - const awaitingUser = new Set( - listThreadIdsWithActivePendingInteractions(deps.db), - ); const prepared = []; const seen = new Set(); for (const work of args.work) { @@ -215,11 +145,10 @@ export async function recordRestartResumes( } const text = work.kind === "turn" - ? turnResumeText({ - awaitingUser: awaitingUser.has(thread.id), - command: runningCommand(deps.db, thread.id), - }) - : backgroundResumeText(work.jobs); + ? renderTemplate("systemMessageRestartResume", {}) + : renderTemplate("systemMessageRestartResumeBackground", { + jobs: work.jobs.map((job) => `- ${job}`).join("\n"), + }); try { prepared.push({ execution: await buildExecutionOptions( @@ -247,7 +176,7 @@ export async function recordRestartResumes( content: [ { type: "text", - text: renderTemplate("systemMessageRestartResume", entry.text), + text: entry.text, mentions: [], }, ], diff --git a/apps/server/test/threads/restart-resume.test.ts b/apps/server/test/threads/restart-resume.test.ts index 6421728e62d..2e36d248537 100644 --- a/apps/server/test/threads/restart-resume.test.ts +++ b/apps/server/test/threads/restart-resume.test.ts @@ -169,13 +169,13 @@ describe("continuing threads after a restart", () => { ).toEqual(["bb will send this thread a message to continue."]); const [resume] = listQueuedThreadMessages(harness.db, working.thread.id); expect(resume?.content).toContain( - "bb restarted, or the machine running this thread did, while you were running `pnpm test`. The restart interrupted it; the user did not reject it, even if your history shows it as interrupted or rejected by the user.", + "Continue where you left off.", ); expect(resume && waitingOnKind(resume)).toBe("time"); expect( listQueuedThreadMessages(harness.db, delegating.thread.id)[0]?.content, ).toContain( - "while this background work was running: `Review the migration`", + "- Review the migration", ); expect( listQueuedThreadMessages(harness.db, serving.thread.id), @@ -183,7 +183,11 @@ describe("continuing threads after a restart", () => { await sweepPastResumes(harness); - expect(turnRequests(harness, working.thread.id)).toHaveLength(2); + const requests = turnRequests(harness, working.thread.id); + expect(requests).toHaveLength(2); + expect(JSON.parse(requests[1]!.data).input).toEqual([ + { type: "text", text: "Continue where you left off.", mentions: [] }, + ]); expect( listQueuedThreadMessages(harness.db, working.thread.id), ).toHaveLength(0); @@ -308,7 +312,7 @@ describe("continuing threads after a restart", () => { parentResume?.sendAt ?? 0, ); expect(parentResume?.content).toContain( - "while you were running `bb thread wait thr_child`", + "Continue where you left off.", ); await new Promise((resolvePromise) => setTimeout(resolvePromise, 2_300)); expect( diff --git a/docs/configuration.md b/docs/configuration.md index 2941e691ecd..2ee4f5168f4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -899,10 +899,10 @@ off. An experiment you set yourself keeps your value either way. records one resume message per running thread when the server shuts down, and per cut turn or awaited background work (not background shells) when a machine's daemon comes back as a new process after a quit, crash, update, or - reboot. The message names what the thread was doing, such as the command it - was running, and asks the agent to check whether it finished or is still - running before continuing. Resumes go a few seconds apart per machine, child - threads before their parents, ahead of messages already queued for the + reboot. The message says "Continue where you left off." Interrupted awaited + background work gets a note explaining which jobs will no longer report back + and asking the agent to resume or restart what it still needs. Resumes go a + few seconds apart per machine, child threads before their parents, ahead of messages already queued for the thread. bb drops a resume if the turn finished after all, the user stopped it, or the user sent a newer message first, and stops after three automatic resumes in a row. A turn a restart cut off is not announced to retry diff --git a/packages/db/src/data/events.ts b/packages/db/src/data/events.ts index 3b1a25e5c2f..3798805c9de 100644 --- a/packages/db/src/data/events.ts +++ b/packages/db/src/data/events.ts @@ -4448,42 +4448,6 @@ export function pruneTokenUsageEventsInTransaction( }); } -export function getRunningCommandItemData( - db: DbQueryConnection, - args: { threadId: string }, -): string | null { - const startedType = "item/started" satisfies ThreadEventType; - const completedType = "item/completed" satisfies ThreadEventType; - const finished = alias(events, "finished_command"); - const row = db - .select({ data: events.data }) - .from(events) - .where( - and( - eq(events.threadId, args.threadId), - sql`${events.itemKind} = 'commandExecution'`, - eq(events.type, startedType), - isNotNull(events.itemId), - notExists( - db - .select({ one: sql`1` }) - .from(finished) - .where( - and( - eq(finished.threadId, events.threadId), - eq(finished.itemId, events.itemId), - eq(finished.type, completedType), - ), - ), - ), - ), - ) - .orderBy(desc(events.sequence)) - .limit(1) - .get(); - return row?.data ?? null; -} - export function listOpenBackgroundTaskItemRowsForHost( db: DbQueryConnection, args: ListOpenBackgroundTaskItemRowsForHostArgs, diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index 3c5edf6d4d0..30525aba87f 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -365,7 +365,6 @@ export { listTodoSnapshotEventRowsForThread, listOpenTurnInputAcceptedRowsByThreadIds, listOpenBackgroundTaskItemRowsForHost, - getRunningCommandItemData, getLatestThreadInterruption, listLatestTurnRequestSystemMessageKinds, listOpenBackgroundTaskItemRowsForThread, @@ -422,7 +421,6 @@ export type { export { createPendingInteraction, getActivePendingInteractionForThread, - listThreadIdsWithActivePendingInteractions, hasTurnBoundActivePendingInteractionForThread, getPendingInteraction, getPendingInteractionByProviderRequest, diff --git a/packages/db/src/data/pending-interactions.ts b/packages/db/src/data/pending-interactions.ts index e1b003ffb7a..ad6802fe085 100644 --- a/packages/db/src/data/pending-interactions.ts +++ b/packages/db/src/data/pending-interactions.ts @@ -399,14 +399,3 @@ export function interruptPendingInteractionsForThreadIds( threadIds: args.threadIds, }); } - -export function listThreadIdsWithActivePendingInteractions( - db: PendingInteractionReadConnection, -): string[] { - return db - .selectDistinct({ threadId: pendingInteractions.threadId }) - .from(pendingInteractions) - .where(inArray(pendingInteractions.status, ["pending", "resolving"])) - .all() - .map((row) => row.threadId); -} diff --git a/packages/templates/src/templates/bb-guide-customization.md b/packages/templates/src/templates/bb-guide-customization.md index 955811aee23..7d107a5d3fd 100644 --- a/packages/templates/src/templates/bb-guide-customization.md +++ b/packages/templates/src/templates/bb-guide-customization.md @@ -188,8 +188,8 @@ release notes as a dismissible card on Settings → Updates. Experiments that gate a feature are documented with it: `serverMove` in `bb guide machines`, `performanceDiagnostics` under Server performance diagnostics below. `continueThreadsAfterRestarts` sends each thread a quit, crash, update, or -machine restart interrupted one message naming what it was doing, so the agent -checks it and carries on; bb drops it if the turn finished, the user stopped +machine restart interrupted a "Continue where you left off." message, or a +note about interrupted awaited background work; bb drops it if the turn finished, the user stopped it, or the user sent something newer. `timelinePush` sends an open thread's timeline updates over the live connection as row deltas instead of reloading the latest window after each diff --git a/packages/templates/src/templates/system-message-restart-resume-background.md b/packages/templates/src/templates/system-message-restart-resume-background.md new file mode 100644 index 00000000000..b9c6d3b1b26 --- /dev/null +++ b/packages/templates/src/templates/system-message-restart-resume-background.md @@ -0,0 +1,11 @@ +--- +kind: prompt +title: Resume background work after a restart +summary: Explains which awaited background work was interrupted by a restart. +intent: Let the agent recover background work that will no longer report back. +variables: + jobs: Interrupted background work, formatted as a list. +--- +bb restarted while you were waiting on this background work. It will no longer report back. Restart any work you still need, then continue from its results: + +{{jobs}} diff --git a/packages/templates/src/templates/system-message-restart-resume.md b/packages/templates/src/templates/system-message-restart-resume.md index 66e2a174d1f..6fbddb10992 100644 --- a/packages/templates/src/templates/system-message-restart-resume.md +++ b/packages/templates/src/templates/system-message-restart-resume.md @@ -1,14 +1,7 @@ --- kind: prompt title: Resume after a restart -summary: Starts the next turn for a thread that a bb restart, a bb update or a machine restart interrupted. -intent: Let the agent carry on from its timeline without the user having to say "continue", checking the one step the restart may have cut off. -editingNotes: The thread's timeline already holds every finished step, so the message only names what was interrupted. Keep the check-before-continuing line; a stopped command or job may be partly applied, and repeating it may duplicate its effects. Keep saying that the restart, not the user, stopped the work; agents otherwise read the stop as a rejection. -variables: - restarted: What restarted, already phrased, such as "bb restarted to install an update". - interrupted: What the restart interrupted, already phrased, such as "while you were running `pnpm test`". - check: What to check before continuing, already phrased. +summary: Continues a thread interrupted by a bb or machine restart. +intent: Let the agent carry on from its existing conversation. --- -[bb system] - -{{restarted}} {{interrupted}}. {{check}} +Continue where you left off. From b860c60bb34daccff3506e42308b2de65ed50d57 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Fri, 30 Oct 2026 11:24:23 -0700 Subject: [PATCH 4/4] Use the Replay icon for restart resumes Co-Authored-By: Claude Opus 5.5 --- .../components/thread/timeline/GeneratedConversationMessage.tsx | 2 +- packages/shared-ui/src/components/ui/icon-extended.tsx | 2 ++ packages/shared-ui/src/components/ui/icon-registry.ts | 1 + 3 files changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx b/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx index e4eb310325e..e06190a1364 100644 --- a/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx +++ b/apps/app/src/components/thread/timeline/GeneratedConversationMessage.tsx @@ -356,7 +356,7 @@ function systemMessageIconName(systemMessageKind: SystemMessageKind): IconName { case "environment-directory-updated": return "FolderOpen"; case "restart-resume": - return "RotateCcw"; + return "Replay"; case "unlabeled": return "Info"; } diff --git a/packages/shared-ui/src/components/ui/icon-extended.tsx b/packages/shared-ui/src/components/ui/icon-extended.tsx index 2ece1ebe329..46544160b51 100644 --- a/packages/shared-ui/src/components/ui/icon-extended.tsx +++ b/packages/shared-ui/src/components/ui/icon-extended.tsx @@ -91,6 +91,7 @@ import PlusSignIcon from "@hugeicons/core-free-icons/PlusSignIcon"; import PuzzleIcon from "@hugeicons/core-free-icons/PuzzleIcon"; import Refresh01Icon from "@hugeicons/core-free-icons/Refresh01Icon"; import RepeatIcon from "@hugeicons/core-free-icons/RepeatIcon"; +import ReplayIcon from "@hugeicons/core-free-icons/ReplayIcon"; import SecurityCheckIcon from "@hugeicons/core-free-icons/SecurityCheckIcon"; import SentIcon from "@hugeicons/core-free-icons/SentIcon"; import SidebarBottomIcon from "@hugeicons/core-free-icons/SidebarBottomIcon"; @@ -332,6 +333,7 @@ export const EXTENDED_ICON_MAP: ExtendedIconMap = { Plus: PlusSignIcon, Puzzle: PuzzleIcon, Repeat: RepeatIcon, + Replay: ReplayIcon, SecurityCheck: SecurityCheckIcon, RotateCcw: Refresh01Icon, Rows2: LayoutTwoRowIcon, diff --git a/packages/shared-ui/src/components/ui/icon-registry.ts b/packages/shared-ui/src/components/ui/icon-registry.ts index 32e2a6b8e55..52de4c48eb6 100644 --- a/packages/shared-ui/src/components/ui/icon-registry.ts +++ b/packages/shared-ui/src/components/ui/icon-registry.ts @@ -107,6 +107,7 @@ export const EXTENDED_ICON_NAMES = [ "Plus", "Puzzle", "Repeat", + "Replay", "RotateCcw", "Rows2", "SecurityCheck",