Refuse queued messages for threads whose environment is gone

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,