Queue messages sent to a thread that awaits user interaction

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

@@ -124,7 +124,7 @@ 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,12 +27,55 @@ describe("bb thread tell command output", () => {
     ).toEqual({
       threadId: "thread-json-tell",
       ok: true,
+      delivery: "sent",
+      mode: "steer",
+    });
+  });
+
+  it("bb thread tell reports a queued delivery when the thread awaits user interaction", async () => {
+    const post = vi.fn(async () => ({
+      ok: true,
+      delivery: "queued",
+      queuedReason: "awaiting_user_interaction",
+    }));
+    stubServerApi({ "v1.threads.:id.send.$post": post });
+
+    await runCommand(
+      ["thread", "tell", "thread-blocked-tell", "hello"],
+      register,
+    );
+
+    expect(vi.mocked(console.log).mock.calls[0]?.[0]).toBe(
+      "Thread thread-blocked-tell is awaiting user interaction; message queued and delivers when the thread is next idle",
+    );
+  });
+
+  it("bb thread tell --json includes the queued delivery outcome", async () => {
+    const post = vi.fn(async () => ({
+      ok: true,
+      delivery: "queued",
+      queuedReason: "awaiting_user_interaction",
+    }));
+    stubServerApi({ "v1.threads.:id.send.$post": post });
+
+    await runCommand(
+      ["thread", "tell", "thread-blocked-json", "hello", "--json"],
+      register,
+    );
+
+    expect(
+      JSON.parse(String(vi.mocked(console.log).mock.calls[0]?.[0])),
+    ).toEqual({
+      threadId: "thread-blocked-json",
+      ok: true,
+      delivery: "queued",
+      queuedReason: "awaiting_user_interaction",
       mode: "steer",
     });
   });
 
   it("bb thread tell --mode queue preserves non-urgent queued delivery", 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(
@@ -50,7 +93,7 @@ describe("bb thread tell command output", () => {
   });
 
   it("bb thread tell --mode auto preserves explicit legacy auto delivery", 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(
@@ -68,7 +111,7 @@ describe("bb thread tell command output", () => {
   });
 
   it("bb thread tell forwards execution options", 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(
@@ -103,7 +146,7 @@ describe("bb thread tell command output", () => {
   });
 
   it("bb thread tell forwards automatic review mode", 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(
@@ -129,7 +172,7 @@ describe("bb thread tell command output", () => {
   });
 
   it("bb thread tell forwards host-readable paths without reading them on the CLI machine", 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(
@@ -161,7 +204,7 @@ describe("bb thread tell command output", () => {
 
   it("bb thread tell includes sender thread metadata when run inside another thread", async () => {
     vi.stubEnv("BB_THREAD_ID", "thread-sender");
-    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(
@@ -181,7 +224,7 @@ describe("bb thread tell command output", () => {
 
   it("bb thread tell omits sender metadata when targeting the current thread", async () => {
     vi.stubEnv("BB_THREAD_ID", "thread-self");
-    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(["thread", "tell", "thread-self", "self note"], register);

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,
@@ -105,10 +106,9 @@ interface PostThreadMessageArgs {
   images?: readonly string[];
 }
 
-interface PostThreadMessageResult {
-  ok: true;
+type PostThreadMessageResult = ThreadSendResult & {
   mode: ThreadTellDeliveryMode;
-}
+};
 
 interface ThreadUpdateBody {
   title?: string;
@@ -456,11 +456,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));
         },
       ),
     );
@@ -566,7 +562,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,
@@ -586,11 +582,30 @@ 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 === "queued") {
+    switch (response.queuedReason) {
+      case "awaiting_user_interaction":
+        return `Thread ${threadId} is awaiting user interaction; message queued and delivers when the thread is next idle`;
+      case "manual_compaction":
+        return `Thread ${threadId} is compacting; message queued and delivers when the thread is next idle`;
+      case "requested":
+        return `Thread ${threadId} message queued`;
+    }
+  }
+  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/routes/threads/actions.ts

@@ -23,6 +23,7 @@ import {
   type ThreadListResponse,
   type PublicApiSchema,
   type SendMessageRequest,
+  type SendMessageResponse,
 } from "@bb/server-contract";
 import type { Hono } from "hono";
 import {
@@ -301,7 +302,8 @@ async function createQueuedMessageForThread(
   }
   if (
     thread.status === "idle" &&
-    getLastProviderThreadId(deps, thread.id) !== null
+    getLastProviderThreadId(deps, thread.id) !== null &&
+    !deps.pendingInteractions.hasPendingThreadInteraction(thread.id)
   ) {
     requestQueuedMessageAutoSendForThread(deps, {
       queuedMessageId: queuedMessage.id,
@@ -311,6 +313,44 @@ async function createQueuedMessageForThread(
   return toThreadQueuedMessage(queuedMessage);
 }
 
+type SendQueuedReason = Extract<
+  SendMessageResponse,
+  { delivery: "queued" }
+>["queuedReason"];
+
+/**
+ * Decides whether a `send` request must wait in the thread queue instead of
+ * dispatching now. A thread that awaits user interaction (for example an
+ * `AskUserQuestion`) cannot take a new prompt, but a queued message is not
+ * lost: it auto-sends when the thread is next idle, and it stays visible in
+ * the queue meanwhile (#1650). `start` never queues; `sendThreadMessage`
+ * rejects it with the usual conflict errors.
+ */
+function resolveSendQueuedReason(
+  deps: AppDeps,
+  args: { payload: SendMessageRequest; thread: Thread },
+): SendQueuedReason | null {
+  const { payload, thread } = args;
+  if (payload.mode === "start") {
+    return null;
+  }
+  // Checked before the status guard: a plugin input request can block an idle
+  // thread too, and that message must wait in the queue as well.
+  if (deps.pendingInteractions.hasPendingThreadInteraction(thread.id)) {
+    return "awaiting_user_interaction";
+  }
+  if (thread.status !== "active") {
+    return null;
+  }
+  if (payload.mode === "queue-if-active") {
+    return "requested";
+  }
+  if (isManualCompactionActive(deps, thread)) {
+    return "manual_compaction";
+  }
+  return null;
+}
+
 export function registerThreadActionRoutes(app: Hono, deps: AppDeps): void {
   const { get, post, patch, del } = typedRoutes<PublicApiSchema>(app, {
     onValidationError: (msg) => new ApiError(400, "invalid_request", msg),
@@ -319,17 +359,13 @@ 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);
+    const queuedReason = resolveSendQueuedReason(deps, { payload, thread });
+    if (queuedReason !== null) {
       await createQueuedMessageForThread(deps, {
         payload: queuedMessagePayloadFromSendRequest(payload),
         thread,
       });
-      return context.json({ ok: true });
+      return context.json({ ok: true, delivery: "queued", queuedReason });
     }
     const environment = await requireThreadCommandEnvironment(deps, {
       thread,
@@ -340,7 +376,7 @@ export function registerThreadActionRoutes(app: Hono, deps: AppDeps): void {
       thread,
       trigger: "user",
     });
-    return context.json({ ok: true });
+    return context.json({ ok: true, delivery: "sent" });
   });
 
   get(routes.rateLimitRecovery, async (context) => {

apps/server/src/server.ts

@@ -74,6 +74,8 @@ import {
 } from "./services/plugin-catalog/plugin-catalog-service.js";
 import { callHostRetryableOnlineRpc } from "./services/hosts/online-rpc.js";
 import { browserRequestProblem } from "./browser-request-guard.js";
+import { requestDeferredParentSystemMessageFlush } from "./services/threads/parent-system-messages.js";
+import { requestQueuedMessageAutoSendForIdleThread } from "./services/threads/queued-messages.js";
 
 /**
  * `/api/v1/plugins/<id>/http/...` — the plugin-owned wire, whose auth mode is
@@ -419,6 +421,13 @@ export function createApp(
     watchBuiltinPluginSources:
       process.env.BB_MANAGED_DEV_BUILTIN_PLUGIN_HOT_RELOAD === "1",
   });
+  // Work held back while a thread awaited user interaction resumes once that
+  // interaction settles (#1650): deferred parent system messages flush, and an
+  // idle thread (a plugin input request can block one) drains its queue.
+  deps.pendingInteractions.setThreadInteractionSettledListener((threadId) => {
+    requestDeferredParentSystemMessageFlush(deps, threadId);
+    requestQueuedMessageAutoSendForIdleThread(deps, threadId);
+  });
   // Bridge the thread lifecycle seams to this service's plugins (§4.5).
   setPluginThreadEventEmitter(pluginService.events);
   // Bridge runtime-config assembly to plugin skills + context (§4.4).

apps/server/src/services/interactions/pending-interactions.ts

@@ -222,6 +222,8 @@ function buildInteractiveResolveCommand(
 
 type PendingInteractionLifecycleArgs = CreateLifecycleDeps;
 
+export type ThreadInteractionSettledListener = (threadId: string) => void;
+
 function buildInteractionChangeMetadata({
   db,
   hasPendingInteraction,
@@ -261,6 +263,8 @@ export class PendingInteractionLifecycle {
   private readonly deps: CreateLifecycleDeps;
   private readonly pluginWaiters = new Map<string, PluginInteractionWaiter>();
   private started = false;
+  private interactionSettledListener: ThreadInteractionSettledListener | null =
+    null;
 
   constructor(args: PendingInteractionLifecycleArgs) {
     this.deps = {
@@ -292,6 +296,18 @@ export class PendingInteractionLifecycle {
     );
   }
 
+  /**
+   * Registers the one listener that runs after an interaction reaches a
+   * terminal state. Callers use it to release work held back while the thread
+   * was blocked; they must re-check `hasPendingThreadInteraction`, because a
+   * thread can settle one interaction and still hold another.
+   */
+  setThreadInteractionSettledListener(
+    listener: ThreadInteractionSettledListener,
+  ): void {
+    this.interactionSettledListener = listener;
+  }
+
   listThreadInteractions(threadId: string): PendingInteraction[] {
     return this.parseListRows(
       listPendingInteractionsByThread(this.deps.db, { threadId }),
@@ -894,6 +910,7 @@ export class PendingInteractionLifecycle {
       hasPendingInteraction: false,
       threadId: interaction.threadId,
     });
+    this.notifyInteractionSettled(interaction.threadId);
   }
 
   private settleInteractionTerminalStateInTransaction(
@@ -906,6 +923,21 @@ export class PendingInteractionLifecycle {
       hasPendingInteraction: false,
       threadId: interaction.threadId,
     });
+    this.notifyInteractionSettled(interaction.threadId);
+  }
+
+  private notifyInteractionSettled(threadId: string): void {
+    if (!this.interactionSettledListener) {
+      return;
+    }
+    try {
+      this.interactionSettledListener(threadId);
+    } catch (error) {
+      this.deps.logger.warn(
+        { err: error, threadId },
+        "Pending interaction settled listener failed",
+      );
+    }
   }
 
   private cancelPluginInteractionFromCallback(args: {