From f5a37acbcb208022b7d4860b595f6a00b48d2ace Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Thu, 20 Aug 2026 17:03:46 +0000 Subject: [PATCH 1/2] Refuse queued messages for threads whose environment is gone (#1789) POST /threads/:id/queued-messages only checked archived/stopping/deleted and never applied the gone-environment rule the direct send path uses. A thread whose managed worktree was destroyed still answered 201, the message sat in the queue forever, and the auto-send sweep failed every 10 s. Apply goneThreadEnvironmentDetails in createQueuedMessageForThread, refuse threads that ran before but lost their environment row, and label destroyed worktrees "Destroyed" instead of "Provisioning" in formatEnvironmentDisplay. Fixes #1789 Co-Authored-By: Claude --- apps/server/src/routes/threads/actions.ts | 40 ++++ ...blic-thread-queue-gone-environment.test.ts | 197 ++++++++++++++++++ packages/core-ui/src/environment-display.ts | 26 ++- .../core-ui/test/environment-display.test.ts | 17 ++ 4 files changed, 271 insertions(+), 9 deletions(-) create mode 100644 apps/server/test/public/public-thread-queue-gone-environment.test.ts diff --git a/apps/server/src/routes/threads/actions.ts b/apps/server/src/routes/threads/actions.ts index a59673fa2e..e2c1d4f223 100644 --- a/apps/server/src/routes/threads/actions.ts +++ b/apps/server/src/routes/threads/actions.ts @@ -40,6 +40,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,12 +257,47 @@ function queuedMessagePayloadFromSendRequest( }; } +/** + * 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` and the auto-send sweep fails every cycle (#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 ensureQueuedMessageEnvironmentIsNotGone( + deps: AppDeps, + thread: Thread, +): void { + if (thread.environmentId === null) { + if (getLastProviderThreadId(deps, thread.id) !== null) { + throwThreadEnvironmentUnavailable( + threadEnvironmentUnavailableDetails("never_attached", null), + ); + } + return; + } + const environment = getEnvironment(deps.db, thread.environmentId); + if (!environment) { + return; + } + const goneDetails = goneThreadEnvironmentDetails(environment); + if (goneDetails) { + throwThreadEnvironmentUnavailable(goneDetails); + } +} + async function createQueuedMessageForThread( deps: AppDeps, args: CreateQueuedMessageForThreadArgs, ): Promise { const { payload, thread } = args; ensureThreadIsWritable(thread); + ensureQueuedMessageEnvironmentIsNotGone(deps, thread); await validatePromptAttachmentReferences({ dataDir: deps.config.dataDir, input: payload.input, 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..704c229367 --- /dev/null +++ b/apps/server/test/public/public-thread-queue-gone-environment.test.ts @@ -0,0 +1,197 @@ +import { getThread, listQueuedThreadMessages } from "@bb/db"; +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, + 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); + }); + }); +}); diff --git a/packages/core-ui/src/environment-display.ts b/packages/core-ui/src/environment-display.ts index bb6ca22065..bccad12f98 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, "Destroyed" + * once it is being torn down or gone, otherwise "Working locally", + * "Working remotely", or "Worktree". */ modeLabel: string; /** @@ -62,22 +63,29 @@ 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 isGoneDisplay = + environment.status === "destroying" || environment.status === "destroyed"; const isProvisioningDisplay = - environment.status === "provisioning" || - (environment.workspaceProvisionType === "managed-worktree" && - environment.path === null); + !isGoneDisplay && + (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 = isGoneDisplay + ? "Destroyed" + : isProvisioningDisplay ? "Provisioning" : mode === "worktree" ? "Worktree" : directModeLabel; - const generatedCompactModeLabel = - isProvisioningDisplay + const generatedCompactModeLabel = isGoneDisplay + ? "Destroyed" + : 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..c914c5fe19 100644 --- a/packages/core-ui/test/environment-display.test.ts +++ b/packages/core-ui/test/environment-display.test.ts @@ -143,6 +143,23 @@ describe("formatEnvironmentDisplay", () => { expect(result.mode).toBe("direct"); }); + it("reports 'Destroyed' for a destroyed managed worktree instead of 'Provisioning' (#1789)", () => { + for (const status of ["destroying", "destroyed"] as const) { + const result = formatEnvironmentDisplay({ + environment: makeEnvironment({ + managed: true, + isWorktree: true, + workspaceProvisionType: "managed-worktree", + path: null, + status, + }), + host: localHostContext, + }); + expect(result.modeLabel).toBe("Destroyed"); + expect(result.compactModeLabel).toBe("Destroyed"); + } + }); + it("reports 'Provisioning' for a prepared managed worktree before the workspace path exists", () => { const result = formatEnvironmentDisplay({ environment: makeEnvironment({ From c4c6df9cde59593c0574e9411521198744e5055a Mon Sep 17 00:00:00 2001 From: Sawyer Hood Date: Thu, 20 Aug 2026 17:56:53 +0000 Subject: [PATCH 2/2] Admit queued messages inside the insert transaction and skip gone environments in the sweep SlopCop review on #2043: - Pre-existing queued rows survived archive -> destroy -> unarchive and kept the 10 s sweep failing. listIdleThreadsWithQueuedMessages now joins environments and skips destroying/destroyed ones. - The lifecycle check ran before two awaits. Admission (writable + environment) now runs on fresh rows inside the same immediate transaction as the insert, and reads the provider thread id once. - formatEnvironmentDisplay says Destroying for destroying and Destroyed for destroyed. Co-Authored-By: Claude --- apps/server/src/routes/threads/actions.ts | 74 +++++++++------ ...blic-thread-queue-gone-environment.test.ts | 90 ++++++++++++++++++- packages/core-ui/src/environment-display.ts | 24 ++--- .../core-ui/test/environment-display.test.ts | 11 ++- packages/db/src/data/index.ts | 1 + .../db/src/data/queued-thread-messages.ts | 76 +++++++++------- 6 files changed, 204 insertions(+), 72 deletions(-) diff --git a/apps/server/src/routes/threads/actions.ts b/apps/server/src/routes/threads/actions.ts index e2c1d4f223..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, @@ -258,37 +260,43 @@ 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` and the auto-send sweep fails every cycle (#1789). Refuse - * with the same 409 the direct send path returns. + * 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 ensureQueuedMessageEnvironmentIsNotGone( - deps: AppDeps, +function admitQueuedMessage( + db: DbQueryConnection, thread: Thread, -): void { +): { providerThreadId: string | null } { + ensureThreadIsWritable(thread); + const providerThreadId = getLastProviderThreadId({ db }, thread.id); if (thread.environmentId === null) { - if (getLastProviderThreadId(deps, thread.id) !== null) { + if (providerThreadId !== null) { throwThreadEnvironmentUnavailable( threadEnvironmentUnavailableDetails("never_attached", null), ); } - return; - } - const environment = getEnvironment(deps.db, thread.environmentId); - if (!environment) { - return; + return { providerThreadId }; } - const goneDetails = goneThreadEnvironmentDetails(environment); + const environment = getEnvironment(db, thread.environmentId); + const goneDetails = environment + ? goneThreadEnvironmentDetails(environment) + : null; if (goneDetails) { throwThreadEnvironmentUnavailable(goneDetails); } + return { providerThreadId }; } async function createQueuedMessageForThread( @@ -297,7 +305,6 @@ async function createQueuedMessageForThread( ): Promise { const { payload, thread } = args; ensureThreadIsWritable(thread); - ensureQueuedMessageEnvironmentIsNotGone(deps, thread); await validatePromptAttachmentReferences({ dataDir: deps.config.dataDir, input: payload.input, @@ -315,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", @@ -334,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 index 704c229367..eec3e1ab18 100644 --- a/apps/server/test/public/public-thread-queue-gone-environment.test.ts +++ b/apps/server/test/public/public-thread-queue-gone-environment.test.ts @@ -1,4 +1,14 @@ -import { getThread, listQueuedThreadMessages } from "@bb/db"; +import { + archiveThread, + getEnvironment, + getThread, + listIdleThreadsWithQueuedMessages, + listQueuedThreadMessages, +} from "@bb/db"; +import { + applyEnvironmentLifecycleEvent, + requireEnvironmentLifecycleEventApplied, +} from "@bb/db/internal-environment-lifecycle"; import { encodeClientTurnRequestIdNumber, threadScope, @@ -11,6 +21,7 @@ import { seedEvent, seedHostSession, seedProjectWithSource, + seedQueuedMessage, seedThread, seedTurnStarted, } from "../helpers/seed.js"; @@ -194,4 +205,81 @@ describe("queued message into a thread whose environment is gone (#1789)", () => 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 bccad12f98..468015dd59 100644 --- a/packages/core-ui/src/environment-display.ts +++ b/packages/core-ui/src/environment-display.ts @@ -20,9 +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, "Destroyed" - * once it is being torn down or gone, 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; /** @@ -65,10 +65,14 @@ export function formatEnvironmentDisplay({ // 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 isGoneDisplay = - environment.status === "destroying" || environment.status === "destroyed"; + const goneLabel = + environment.status === "destroying" + ? "Destroying" + : environment.status === "destroyed" + ? "Destroyed" + : null; const isProvisioningDisplay = - !isGoneDisplay && + goneLabel === null && (environment.status === "provisioning" || (environment.workspaceProvisionType === "managed-worktree" && environment.path === null)); @@ -76,15 +80,15 @@ export function formatEnvironmentDisplay({ host.locality === "remote" ? "Working remotely" : "Working locally"; const directCompactModeLabel = host.locality === "remote" ? "Remote" : "Local"; - const generatedModeLabel = isGoneDisplay - ? "Destroyed" + const generatedModeLabel = goneLabel + ? goneLabel : isProvisioningDisplay ? "Provisioning" : mode === "worktree" ? "Worktree" : directModeLabel; - const generatedCompactModeLabel = isGoneDisplay - ? "Destroyed" + const generatedCompactModeLabel = goneLabel + ? goneLabel : isProvisioningDisplay ? "Provisioning" : mode === "worktree" diff --git a/packages/core-ui/test/environment-display.test.ts b/packages/core-ui/test/environment-display.test.ts index c914c5fe19..12fc0b0b91 100644 --- a/packages/core-ui/test/environment-display.test.ts +++ b/packages/core-ui/test/environment-display.test.ts @@ -143,8 +143,11 @@ describe("formatEnvironmentDisplay", () => { expect(result.mode).toBe("direct"); }); - it("reports 'Destroyed' for a destroyed managed worktree instead of 'Provisioning' (#1789)", () => { - for (const status of ["destroying", "destroyed"] as const) { + 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, @@ -155,8 +158,8 @@ describe("formatEnvironmentDisplay", () => { }), host: localHostContext, }); - expect(result.modeLabel).toBe("Destroyed"); - expect(result.compactModeLabel).toBe("Destroyed"); + expect(result.modeLabel).toBe(label); + expect(result.compactModeLabel).toBe(label); } }); 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), ),