diff --git a/apps/server/src/routes/threads/actions.ts b/apps/server/src/routes/threads/actions.ts index a59673fa2e..ab49377258 100644 --- a/apps/server/src/routes/threads/actions.ts +++ b/apps/server/src/routes/threads/actions.ts @@ -1,8 +1,9 @@ import { - createQueuedThreadMessage, + createQueuedThreadMessageInTransaction, deleteQueuedThreadMessage, getEnvironment, getQueuedThreadMessage, + getThread, listActiveVisiblePinnedThreadRootsWithPendingInteractionState, pinThread, reorderPinnedThread, @@ -15,6 +16,7 @@ import { type ReorderPinnedThreadResult, type ReorderQueuedThreadMessageResult, type SetQueuedThreadMessageGroupBoundaryResult, + type DbQueryConnection, } from "@bb/db"; import { publicApiRoutes, @@ -40,6 +42,11 @@ import { } from "../../services/environments/environment-cleanup-internal.js"; import { applyLoggedEnvironmentLifecycleEvent } from "../../services/environments/lifecycle-outcome.js"; import { requirePublicThread } from "../../services/lib/entity-lookup.js"; +import { + goneThreadEnvironmentDetails, + threadEnvironmentUnavailableDetails, + throwThreadEnvironmentUnavailable, +} from "../../services/lib/lifecycle-api-errors.js"; import { parseSafeRelativeRoutePath } from "../relative-route-path.js"; import { validatePromptAttachmentReferences } from "../../services/projects/attachments.js"; import { @@ -252,6 +259,46 @@ function queuedMessagePayloadFromSendRequest( }; } +/** + * Admits a queued message against the current thread and environment rows. + * Returns the provider thread id so the caller can decide on auto-send without + * a second event-history read. + * + * A queued message can only drain into the thread's environment. A gone + * environment (`destroying`/`destroyed`) is never reprovisioned, so accepting + * the message would park it in the queue forever while the thread keeps + * reporting `idle` (#1789). Refuse with the same 409 the direct send path + * returns. + * + * A thread with no environment row is accepted while it has never run: the + * queue is how messages wait for provisioning. Once the thread has a provider + * thread id, a missing environment means the row was pruned after destroy, and + * the direct send path already refuses with `never_attached`. + */ +function admitQueuedMessage( + db: DbQueryConnection, + thread: Thread, +): { providerThreadId: string | null } { + ensureThreadIsWritable(thread); + const providerThreadId = getLastProviderThreadId({ db }, thread.id); + if (thread.environmentId === null) { + if (providerThreadId !== null) { + throwThreadEnvironmentUnavailable( + threadEnvironmentUnavailableDetails("never_attached", null), + ); + } + return { providerThreadId }; + } + const environment = getEnvironment(db, thread.environmentId); + const goneDetails = environment + ? goneThreadEnvironmentDetails(environment) + : null; + if (goneDetails) { + throwThreadEnvironmentUnavailable(goneDetails); + } + return { providerThreadId }; +} + async function createQueuedMessageForThread( deps: AppDeps, args: CreateQueuedMessageForThreadArgs, @@ -275,15 +322,31 @@ async function createQueuedMessageForThread( senderThreadId: payload.senderThreadId, targetThread: thread, }); - const queuedMessage = createQueuedThreadMessage(deps.db, deps.hub, { - threadId: thread.id, - content: payload.input, - senderThreadId, - model: execution.model, - reasoningLevel: execution.reasoningLevel, - permissionMode: execution.permissionMode, - serviceTier: execution.serviceTier, - }); + // The awaits above can interleave with an archive or environment destroy, so + // admit against the rows as they are at insert time, in the same immediate + // transaction as the insert. + const { currentThread, providerThreadId, queuedMessage } = + deps.db.transaction( + (tx) => { + const currentThread = getThread(tx, thread.id); + if (!currentThread) { + throw new ApiError(404, "thread_not_found", "Thread not found"); + } + const { providerThreadId } = admitQueuedMessage(tx, currentThread); + const queuedMessage = createQueuedThreadMessageInTransaction(tx, { + threadId: thread.id, + content: payload.input, + senderThreadId, + model: execution.model, + reasoningLevel: execution.reasoningLevel, + permissionMode: execution.permissionMode, + serviceTier: execution.serviceTier, + }); + return { currentThread, providerThreadId, queuedMessage }; + }, + { behavior: "immediate" }, + ); + deps.hub.notifyThread(thread.id, ["queue-changed"]); if (senderThreadId === null && payload.input.length > 0) { deps.telemetry.capture({ name: "user_message_sent", @@ -294,10 +357,7 @@ async function createQueuedMessageForThread( }, }); } - if ( - thread.status === "idle" && - getLastProviderThreadId(deps, thread.id) !== null - ) { + if (currentThread.status === "idle" && providerThreadId !== null) { requestQueuedMessageAutoSendForThread(deps, { queuedMessageId: queuedMessage.id, threadId: thread.id, diff --git a/apps/server/test/public/public-thread-queue-gone-environment.test.ts b/apps/server/test/public/public-thread-queue-gone-environment.test.ts new file mode 100644 index 0000000000..eec3e1ab18 --- /dev/null +++ b/apps/server/test/public/public-thread-queue-gone-environment.test.ts @@ -0,0 +1,285 @@ +import { + archiveThread, + getEnvironment, + getThread, + listIdleThreadsWithQueuedMessages, + listQueuedThreadMessages, +} from "@bb/db"; +import { + applyEnvironmentLifecycleEvent, + requireEnvironmentLifecycleEventApplied, +} from "@bb/db/internal-environment-lifecycle"; +import { + encodeClientTurnRequestIdNumber, + threadScope, + type EnvironmentStatus, +} from "@bb/domain"; +import { describe, expect, it } from "vitest"; +import { readJson } from "../helpers/json.js"; +import { + seedEnvironment, + seedEvent, + seedHostSession, + seedProjectWithSource, + seedQueuedMessage, + seedThread, + seedTurnStarted, +} from "../helpers/seed.js"; +import { withTestHarness, type TestAppHarness } from "../helpers/test-app.js"; + +/** + * Regression test for get-bb/bb#1789. + * + * A direct send into a thread whose environment is destroying/destroyed is + * rejected with 409 `thread_environment_unavailable` (see + * public-thread-environment-decoupling.test.ts). The queued-message create + * route used to have no such guard: it answered 201 with a queued-message id, + * the thread kept reporting `idle`, and the message could never drain because + * the auto-send path threw the very 409 the create route should have returned. + */ +const earlierTurnEventData = { + direction: "outbound", + requestId: encodeClientTurnRequestIdNumber({ value: 1 }), + input: [{ type: "text", text: "Earlier work" }], + target: { kind: "new-turn" }, + execution: { + model: "gpt-5", + serviceTier: "default", + reasoningLevel: "medium", + permissionMode: "full", + source: "client/turn/requested", + }, + initiator: "user", + senderThreadId: null, + request: { method: "turn/start", params: {} }, + source: "tell", +} as const; + +async function postQueuedMessage( + harness: TestAppHarness, + threadId: string, + text: string, + options: { model?: string } = {}, +): Promise { + return harness.app.request(`/api/v1/threads/${threadId}/queued-messages`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ input: [{ type: "text", text }], ...options }), + }); +} + +describe("queued message into a thread whose environment is gone (#1789)", () => { + for (const status of [ + "destroying", + "destroyed", + ] as const satisfies readonly EnvironmentStatus[]) { + it(`rejects queue-create when the environment is ${status}`, async () => { + await withTestHarness(async (harness) => { + const { host } = seedHostSession(harness.deps, { + id: `host-queue-${status}`, + }); + const { project } = seedProjectWithSource(harness.deps, { + hostId: host.id, + }); + const environment = seedEnvironment(harness.deps, { + hostId: host.id, + managed: true, + projectId: project.id, + path: null, + status, + workspaceProvisionType: "managed-worktree", + }); + const thread = seedThread(harness.deps, { + projectId: project.id, + environmentId: environment.id, + status: "idle", + }); + + // A prior accepted turn: the thread has a stored execution model and + // a provider thread id, exactly like a real idle thread that ran once. + seedEvent(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + sequence: 1, + type: "client/turn/requested", + scope: threadScope(), + data: earlierTurnEventData, + }); + + // Control: the direct send path already refuses. + const sendResponse = await harness.app.request( + `/api/v1/threads/${thread.id}/send`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + mode: "auto", + input: [{ type: "text", text: "direct send" }], + }), + }, + ); + expect(sendResponse.status).toBe(409); + + // The queue-create path must refuse the same message. + const queueResponse = await postQueuedMessage( + harness, + thread.id, + "queued into a gone env", + ); + const body = await readJson(queueResponse); + expect(getThread(harness.db, thread.id)?.status).toBe("idle"); + expect( + queueResponse.status, + `queue-create answered ${queueResponse.status} ${JSON.stringify(body)}; queued rows: ${ + listQueuedThreadMessages(harness.db, thread.id).length + }`, + ).toBe(409); + expect(body).toMatchObject({ + code: "thread_environment_unavailable", + details: { reason: status, environmentStatus: status }, + }); + expect(listQueuedThreadMessages(harness.db, thread.id)).toHaveLength(0); + }); + }); + } + + it("rejects queue-create when a thread that ran before lost its environment row", async () => { + await withTestHarness(async (harness) => { + const { host } = seedHostSession(harness.deps, { id: "host-queue-pruned" }); + const { project } = seedProjectWithSource(harness.deps, { + hostId: host.id, + }); + // After the destroyed row is pruned, the FK sets environmentId to NULL. + const thread = seedThread(harness.deps, { + projectId: project.id, + environmentId: null, + status: "idle", + }); + seedEvent(harness.deps, { + threadId: thread.id, + environmentId: null, + sequence: 1, + type: "client/turn/requested", + scope: threadScope(), + data: earlierTurnEventData, + }); + seedTurnStarted(harness.deps, { + threadId: thread.id, + environmentId: null, + turnId: "turn-1", + }); + + const queueResponse = await postQueuedMessage( + harness, + thread.id, + "queued into a pruned env", + ); + expect(queueResponse.status).toBe(409); + expect(await readJson(queueResponse)).toMatchObject({ + code: "thread_environment_unavailable", + details: { reason: "never_attached", environmentStatus: null }, + }); + expect(listQueuedThreadMessages(harness.db, thread.id)).toHaveLength(0); + }); + }); + + it("still accepts queue-create for a thread that has not run and has no environment yet", async () => { + await withTestHarness(async (harness) => { + const { host } = seedHostSession(harness.deps, { id: "host-queue-new" }); + const { project } = seedProjectWithSource(harness.deps, { + hostId: host.id, + }); + const thread = seedThread(harness.deps, { + projectId: project.id, + environmentId: null, + status: "idle", + }); + + const queueResponse = await postQueuedMessage( + harness, + thread.id, + "queued before provisioning", + { model: "gpt-5" }, + ); + expect(queueResponse.status, await queueResponse.text()).toBe(201); + expect(listQueuedThreadMessages(harness.db, thread.id)).toHaveLength(1); + }); + }); + + it("drops a thread from the auto-send sweep once its environment is gone, so pre-existing queued rows stop failing every cycle", async () => { + await withTestHarness(async (harness) => { + const { host } = seedHostSession(harness.deps, { id: "host-queue-sweep" }); + const { project } = seedProjectWithSource(harness.deps, { + hostId: host.id, + }); + const environment = seedEnvironment(harness.deps, { + hostId: host.id, + managed: true, + projectId: project.id, + status: "ready", + workspaceProvisionType: "managed-worktree", + }); + const thread = seedThread(harness.deps, { + projectId: project.id, + environmentId: environment.id, + status: "idle", + }); + seedTurnStarted(harness.deps, { + threadId: thread.id, + environmentId: environment.id, + turnId: "turn-1", + }); + // The row was queued while the environment was still ready. + seedQueuedMessage(harness.deps, { + threadId: thread.id, + content: [{ type: "text", text: "queued before destroy", mentions: [] }], + }); + const sweepCandidates = () => + listIdleThreadsWithQueuedMessages(harness.db).map((row) => row.threadId); + expect(sweepCandidates()).toEqual([thread.id]); + + // Archive → grace window elapses → destroy → unarchive. The queued row + // survives all of it. + archiveThread(harness.db, harness.hub, thread.id); + requireEnvironmentLifecycleEventApplied( + applyEnvironmentLifecycleEvent(harness.db, harness.hub, { + environmentId: environment.id, + event: { type: "retire.requested" }, + }), + ); + requireEnvironmentLifecycleEventApplied( + applyEnvironmentLifecycleEvent(harness.db, harness.hub, { + environmentId: environment.id, + event: { type: "destroy.started", destroyAttemptId: "rpc_sweep" }, + }), + ); + expect(getEnvironment(harness.db, environment.id)?.status).toBe( + "destroying", + ); + expect(sweepCandidates()).toEqual([]); + requireEnvironmentLifecycleEventApplied( + applyEnvironmentLifecycleEvent(harness.db, harness.hub, { + environmentId: environment.id, + event: { type: "destroy.completed", destroyAttemptId: "rpc_sweep" }, + }), + ); + expect(getEnvironment(harness.db, environment.id)?.status).toBe( + "destroyed", + ); + const unarchiveResponse = await harness.app.request( + `/api/v1/threads/${thread.id}/unarchive`, + { method: "POST" }, + ); + expect(unarchiveResponse.status).toBe(200); + expect(getThread(harness.db, thread.id)).toMatchObject({ + archivedAt: null, + status: "idle", + }); + + // The row is still there for the user to see, but the sweep no longer + // picks the thread up, so there is no send to fail every 10 s. + expect(listQueuedThreadMessages(harness.db, thread.id)).toHaveLength(1); + expect(sweepCandidates()).toEqual([]); + }); + }); +}); diff --git a/packages/core-ui/src/environment-display.ts b/packages/core-ui/src/environment-display.ts index bb6ca22065..468015dd59 100644 --- a/packages/core-ui/src/environment-display.ts +++ b/packages/core-ui/src/environment-display.ts @@ -20,8 +20,9 @@ export interface EnvironmentDisplayHostContext { export interface EnvironmentDisplayInfo { /** * Human-readable environment label: a custom environment name when present, - * "Provisioning" while the environment is still being set up, otherwise - * "Working locally", "Working remotely", or "Worktree". + * "Provisioning" while the environment is still being set up, "Destroying" + * while it is torn down, "Destroyed" once it is gone, otherwise "Working + * locally", "Working remotely", or "Worktree". */ modeLabel: string; /** @@ -62,22 +63,33 @@ export function formatEnvironmentDisplay({ // Managed worktrees can also sit in a prepared metadata-inference stage with // no workspace path before the actual provision request is queued. Report the // setup lifecycle honestly instead of guessing "Working locally". + // A destroyed managed worktree also has no path. It is gone, not being set + // up, so it must not read as "Provisioning" (#1789). + const goneLabel = + environment.status === "destroying" + ? "Destroying" + : environment.status === "destroyed" + ? "Destroyed" + : null; const isProvisioningDisplay = - environment.status === "provisioning" || - (environment.workspaceProvisionType === "managed-worktree" && - environment.path === null); + goneLabel === null && + (environment.status === "provisioning" || + (environment.workspaceProvisionType === "managed-worktree" && + environment.path === null)); const directModeLabel = host.locality === "remote" ? "Working remotely" : "Working locally"; const directCompactModeLabel = host.locality === "remote" ? "Remote" : "Local"; - const generatedModeLabel = - isProvisioningDisplay + const generatedModeLabel = goneLabel + ? goneLabel + : isProvisioningDisplay ? "Provisioning" : mode === "worktree" ? "Worktree" : directModeLabel; - const generatedCompactModeLabel = - isProvisioningDisplay + const generatedCompactModeLabel = goneLabel + ? goneLabel + : isProvisioningDisplay ? "Provisioning" : mode === "worktree" ? "Worktree" diff --git a/packages/core-ui/test/environment-display.test.ts b/packages/core-ui/test/environment-display.test.ts index 1f3ea67dfd..12fc0b0b91 100644 --- a/packages/core-ui/test/environment-display.test.ts +++ b/packages/core-ui/test/environment-display.test.ts @@ -143,6 +143,26 @@ describe("formatEnvironmentDisplay", () => { expect(result.mode).toBe("direct"); }); + it("reports 'Destroying'/'Destroyed' for a gone managed worktree instead of 'Provisioning' (#1789)", () => { + for (const [status, label] of [ + ["destroying", "Destroying"], + ["destroyed", "Destroyed"], + ] as const) { + const result = formatEnvironmentDisplay({ + environment: makeEnvironment({ + managed: true, + isWorktree: true, + workspaceProvisionType: "managed-worktree", + path: null, + status, + }), + host: localHostContext, + }); + expect(result.modeLabel).toBe(label); + expect(result.compactModeLabel).toBe(label); + } + }); + it("reports 'Provisioning' for a prepared managed worktree before the workspace path exists", () => { const result = formatEnvironmentDisplay({ environment: makeEnvironment({ diff --git a/packages/db/src/data/index.ts b/packages/db/src/data/index.ts index b0253b7fd7..ded9bbfb5b 100644 --- a/packages/db/src/data/index.ts +++ b/packages/db/src/data/index.ts @@ -512,6 +512,7 @@ export { claimNextQueuedThreadMessage, claimNextQueuedThreadMessageGroup, createQueuedThreadMessage, + createQueuedThreadMessageInTransaction, deleteClaimedQueuedThreadMessageBatchInTransaction, deleteClaimedQueuedThreadMessage, deleteClaimedQueuedThreadMessageInTransaction, diff --git a/packages/db/src/data/queued-thread-messages.ts b/packages/db/src/data/queued-thread-messages.ts index e3420cbb32..d02f51df93 100644 --- a/packages/db/src/data/queued-thread-messages.ts +++ b/packages/db/src/data/queued-thread-messages.ts @@ -18,7 +18,7 @@ import type { DbTransaction, } from "../connection.js"; import type { DbNotifier } from "../notifier.js"; -import { queuedThreadMessages, threads } from "../schema.js"; +import { environments, queuedThreadMessages, threads } from "../schema.js"; import { createQueuedThreadMessageClaimToken, createQueuedThreadMessageId } from "../ids.js"; import { createOrderKeyAfter, @@ -471,40 +471,52 @@ function applyPreservedLeadGroupAfterReorder( : queuedMessages; } +/** + * Inserts a queued message inside a caller-owned transaction. Callers that + * must admit the message against the current thread and environment rows (so + * an archive or destroy that lands between their read and this insert cannot + * slip a row in) run their checks in the same transaction, then call this. + * The caller notifies `queue-changed` after the transaction commits. + */ +export function createQueuedThreadMessageInTransaction( + tx: DbTransaction, + input: CreateQueuedThreadMessageInput, +) { + const now = Date.now(); + const id = createQueuedThreadMessageId(); + const lastQueuedMessage = getLastQueuedThreadMessage(tx, input.threadId); + const sortKey = lastQueuedMessage + ? createOrderKeyAfter({ previousKey: lastQueuedMessage.sortKey }) + : createOrderKeyBetween({ previousKey: null, nextKey: null }); + return tx + .insert(queuedThreadMessages) + .values({ + id, + threadId: input.threadId, + content: JSON.stringify(input.content), + senderThreadId: input.senderThreadId ?? null, + model: input.model, + reasoningLevel: input.reasoningLevel, + permissionMode: input.permissionMode, + serviceTier: input.serviceTier, + groupWithNext: false, + claimedAt: null, + claimToken: null, + sortKey, + createdAt: now, + updatedAt: now, + }) + .returning() + .get(); +} + export function createQueuedThreadMessage( db: DbConnection, notifier: DbNotifier, input: CreateQueuedThreadMessageInput, ) { - const now = Date.now(); - const id = createQueuedThreadMessageId(); const row = db.transaction( - (tx) => { - const lastQueuedMessage = getLastQueuedThreadMessage(tx, input.threadId); - const sortKey = lastQueuedMessage - ? createOrderKeyAfter({ previousKey: lastQueuedMessage.sortKey }) - : createOrderKeyBetween({ previousKey: null, nextKey: null }); - return tx - .insert(queuedThreadMessages) - .values({ - id, - threadId: input.threadId, - content: JSON.stringify(input.content), - senderThreadId: input.senderThreadId ?? null, - model: input.model, - reasoningLevel: input.reasoningLevel, - permissionMode: input.permissionMode, - serviceTier: input.serviceTier, - groupWithNext: false, - claimedAt: null, - claimToken: null, - sortKey, - createdAt: now, - updatedAt: now, - }) - .returning() - .get(); - }, + (tx) => createQueuedThreadMessageInTransaction(tx, input), { behavior: "immediate" }, ); notifier.notifyThread(input.threadId, ["queue-changed"]); @@ -594,12 +606,16 @@ export function listIdleThreadsWithQueuedMessages( }) .from(queuedThreadMessages) .innerJoin(threads, eq(threads.id, queuedThreadMessages.threadId)) + // A gone environment (destroying/destroyed) is never reprovisioned, so its + // queued rows can never drain. Leave them out of the sweep instead of + // failing the same send every cycle (#1789). + .innerJoin(environments, eq(environments.id, threads.environmentId)) .where( and( eq(threads.status, "idle"), isNull(threads.archivedAt), isNull(threads.deletedAt), - isNotNull(threads.environmentId), + notInArray(environments.status, ["destroying", "destroyed"]), isNull(queuedThreadMessages.claimedAt), isNull(queuedThreadMessages.claimToken), ),