Hold messages for threads that await user interaction

apps/app/src/hooks/mutations/thread-runtime-mutations.test.tsx

@@ -124,7 +124,10 @@ beforeEach(() => {
     operationId: "edit-op-1",
     requestSequence: 42,
   });
-  vi.mocked(sdk.threads.send).mockResolvedValue({ ok: true });
+  vi.mocked(sdk.threads.send).mockResolvedValue({
+    ok: true,
+    delivery: "sent",
+  });
   vi.mocked(sdk.threads.queuedMessages.create).mockResolvedValue(
     makeQueuedMessage(),
   );

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

@@ -14,7 +14,7 @@ describe("bb thread tell command output", () => {
     registerThreadCommands(program, () => "http://server");
 
   it("bb thread tell --json prints the raw response plus thread id", async () => {
-    const post = vi.fn(async () => ({ ok: true }));
+    const post = vi.fn(async () => ({ ok: true, delivery: "sent" }));
     stubServerApi({ "v1.threads.:id.send.$post": post });
 
     await runCommand(
@@ -27,10 +27,33 @@ describe("bb thread tell command output", () => {
     ).toEqual({
       threadId: "thread-json-tell",
       ok: true,
+      delivery: "sent",
       mode: "steer",
     });
   });
 
+  it("bb thread tell says when the target is awaiting user interaction and the message is held", async () => {
+    const post = vi.fn(async () => ({ ok: true, delivery: "deferred" }));
+    stubServerApi({ "v1.threads.:id.send.$post": post });
+
+    await runCommand(["thread", "tell", "thread-blocked", "hello"], register);
+
+    expect(vi.mocked(console.log).mock.calls[0]?.[0]).toBe(
+      "Thread thread-blocked is awaiting user interaction; message held and delivers once the interaction settles",
+    );
+  });
+
+  it("bb thread tell keeps the steered wording for servers that only report ok", async () => {
+    const post = vi.fn(async () => ({ ok: true }));
+    stubServerApi({ "v1.threads.:id.send.$post": post });
+
+    await runCommand(["thread", "tell", "thread-legacy", "hello"], register);
+
+    expect(vi.mocked(console.log).mock.calls[0]?.[0]).toBe(
+      "Thread thread-legacy steered",
+    );
+  });
+
   it("bb thread tell --mode queue preserves non-urgent queued delivery", async () => {
     const post = vi.fn(async () => ({ ok: true }));
     stubServerApi({ "v1.threads.:id.send.$post": post });

apps/cli/src/commands/thread/actions.ts

@@ -9,6 +9,7 @@ import {
 } from "@bb/domain";
 import { action } from "../../action.js";
 import { createCliBbSdk } from "../../client.js";
+import type { ThreadSendResult } from "@bb/sdk";
 import {
   confirmDestructiveAction,
   outputJson,
@@ -101,10 +102,9 @@ interface PostThreadMessageArgs {
   images?: readonly string[];
 }
 
-interface PostThreadMessageResult {
-  ok: true;
+type PostThreadMessageResult = ThreadSendResult & {
   mode: ThreadTellDeliveryMode;
-}
+};
 
 interface ThreadUpdateBody {
   title?: string;
@@ -452,11 +452,7 @@ export function registerActionsCommands(
             images: opts.image,
           });
           if (outputJson(opts, { threadId: id, ...response })) return;
-          console.log(
-            response.mode === "steer"
-              ? `Thread ${id} steered`
-              : `Thread ${id} updated`,
-          );
+          console.log(describeThreadTellOutcome(id, response));
         },
       ),
     );
@@ -526,7 +522,7 @@ async function postThreadMessage(
   args: PostThreadMessageArgs,
 ): Promise<PostThreadMessageResult> {
   const sdk = createCliBbSdk(args.getUrl());
-  await sdk.threads.send({
+  const response = await sdk.threads.send({
     threadId: args.threadId,
     input: buildPromptInputs({
       message: args.message,
@@ -546,11 +542,27 @@ async function postThreadMessage(
     ...(args.senderThreadId ? { senderThreadId: args.senderThreadId } : {}),
   });
   return {
-    ok: true,
+    ...response,
     mode: args.mode,
   };
 }
 
+function describeThreadTellOutcome(
+  threadId: string,
+  response: PostThreadMessageResult,
+): string {
+  if (response.delivery === "deferred") {
+    return `Thread ${threadId} is awaiting user interaction; message held and delivers once the interaction settles`;
+  }
+  if (response.delivery === "queued") {
+    return `Thread ${threadId} message queued`;
+  }
+  // `sent`, or an older server that reports only `ok`.
+  return response.mode === "steer"
+    ? `Thread ${threadId} steered`
+    : `Thread ${threadId} updated`;
+}
+
 function resolveSenderThreadId(targetThreadId: string): string | undefined {
   const senderThreadId = resolveContextThreadId();
   if (!senderThreadId || senderThreadId === targetThreadId) {

apps/server/src/lifecycle-dedupers.ts

@@ -22,6 +22,7 @@ export interface ProviderModelListMemoValue {
 }
 
 export interface LifecycleDedupers {
+  deferredThreadMessageFlush: AsyncDeduper<string, void>;
   environmentCleanupAdvance: AsyncDeduper<string, void>;
   /**
    * Memo for host model probes: every execution-options read (each thread
@@ -34,6 +35,7 @@ export interface LifecycleDedupers {
 
 export function createLifecycleDedupers(): LifecycleDedupers {
   return {
+    deferredThreadMessageFlush: createAsyncDeduper<string, void>(),
     environmentCleanupAdvance: createAsyncDeduper<string, void>(),
     providerModelList: createAsyncTtlMemo<string, ProviderModelListMemoValue>({
       ttlMs: PROVIDER_MODEL_LIST_MEMO_TTL_MS,

apps/server/src/routes/threads/actions.ts

@@ -1,9 +1,7 @@
 import {
-  createQueuedThreadMessageInTransaction,
   deleteQueuedThreadMessage,
   getEnvironment,
   getQueuedThreadMessage,
-  getThread,
   listActiveVisiblePinnedThreadRootsWithPendingInteractionState,
   pinThread,
   reorderPinnedThread,
@@ -16,15 +14,12 @@ import {
   type ReorderPinnedThreadResult,
   type ReorderQueuedThreadMessageResult,
   type SetQueuedThreadMessageGroupBoundaryResult,
-  type DbQueryConnection,
 } from "@bb/db";
 import {
   publicApiRoutes,
   typedRoutes,
-  type CreateQueuedMessageRequest,
   type ThreadListResponse,
   type PublicApiSchema,
-  type SendMessageRequest,
 } from "@bb/server-contract";
 import type { Hono } from "hono";
 import {
@@ -42,33 +37,25 @@ 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 {
-  requestQueuedMessageAutoSendForThread,
+  createQueuedMessageForThread,
   sendQueuedMessage,
 } from "../../services/threads/queued-messages.js";
 import {
   ensureThreadIsNotAwaitingUserInteraction,
   ensureThreadIsWritable,
-  resolveMessageSenderThreadId,
   sendThreadMessage,
 } from "../../services/threads/thread-send.js";
+import { acceptThreadSendRequest } from "../../services/threads/thread-send-request.js";
 import { editThreadMessage } from "../../services/threads/thread-edit-message.js";
 import {
   buildExecutionOptions,
   dispatchThreadUnarchiveCommand,
   prepareTurnSubmitCommandPayload,
 } from "../../services/threads/thread-commands.js";
-import {
-  getLastProviderThreadId,
-  isManualCompactionActive,
-} from "../../services/threads/thread-events.js";
+import { getLastProviderThreadId } from "../../services/threads/thread-events.js";
 import { stopThreadForCurrentState } from "../../services/threads/thread-lifecycle.js";
 import {
   getThreadPromptBannerActivity,
@@ -232,142 +219,6 @@ function assertPinnedThreadOrderResult(
   }
 }
 
-interface CreateQueuedMessageForThreadArgs {
-  payload: CreateQueuedMessageRequest;
-  thread: Thread;
-}
-
-function queuedMessagePayloadFromSendRequest(
-  payload: SendMessageRequest,
-): CreateQueuedMessageRequest {
-  return {
-    input: payload.input,
-    ...(payload.model !== undefined ? { model: payload.model } : {}),
-    ...(payload.serviceTier !== undefined
-      ? { serviceTier: payload.serviceTier }
-      : {}),
-    ...(payload.reasoningLevel !== undefined
-      ? { reasoningLevel: payload.reasoningLevel }
-      : {}),
-    ...(payload.permissionMode !== undefined
-      ? { permissionMode: payload.permissionMode }
-      : {}),
-    ...(payload.executionInputSources !== undefined
-      ? { executionInputSources: payload.executionInputSources }
-      : {}),
-    ...(payload.senderThreadId !== undefined
-      ? { senderThreadId: payload.senderThreadId }
-      : {}),
-  };
-}
-
-/**
- * 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,
-): Promise<ThreadQueuedMessage> {
-  const { payload, thread } = args;
-  ensureThreadIsWritable(thread);
-  await validatePromptAttachmentReferences({
-    dataDir: deps.config.dataDir,
-    input: payload.input,
-    projectId: thread.projectId,
-  });
-  const execution = await buildExecutionOptions(
-    deps,
-    payload,
-    {
-      threadId: thread.id,
-    },
-    "client/turn/requested",
-  );
-  const senderThreadId = resolveMessageSenderThreadId(deps, {
-    senderThreadId: payload.senderThreadId,
-    targetThread: thread,
-  });
-  // 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",
-      properties: {
-        is_child_thread: thread.parentThreadId !== null,
-        message_source: "queued_message",
-        provider: thread.providerId,
-      },
-    });
-  }
-  if (currentThread.status === "idle" && providerThreadId !== null) {
-    requestQueuedMessageAutoSendForThread(deps, {
-      queuedMessageId: queuedMessage.id,
-      threadId: thread.id,
-    });
-  }
-  return toThreadQueuedMessage(queuedMessage);
-}
-
 export function registerThreadActionRoutes(app: Hono, deps: AppDeps): void {
   const { post, patch, del } = typedRoutes<PublicApiSchema>(app, {
     onValidationError: (msg) => new ApiError(400, "invalid_request", msg),
@@ -376,28 +227,9 @@ export function registerThreadActionRoutes(app: Hono, deps: AppDeps): void {
 
   post(routes.send, async (context, payload) => {
     const thread = requirePublicThread(deps.db, context.req.param("id"));
-    const shouldQueue =
-      thread.status === "active" &&
-      (payload.mode === "queue-if-active" ||
-        (payload.mode !== "start" && isManualCompactionActive(deps, thread)));
-    if (shouldQueue) {
-      ensureThreadIsNotAwaitingUserInteraction(deps, thread.id);
-      await createQueuedMessageForThread(deps, {
-        payload: queuedMessagePayloadFromSendRequest(payload),
-        thread,
-      });
-      return context.json({ ok: true });
-    }
-    const environment = await requireThreadCommandEnvironment(deps, {
-      thread,
-    });
-    await sendThreadMessage(deps, {
-      environment,
-      payload,
-      thread,
-      trigger: "user",
-    });
-    return context.json({ ok: true });
+    return context.json(
+      await acceptThreadSendRequest(deps, { payload, thread }),
+    );
   });
 
   post(routes.editMessage, async (context, payload) => {