Search saved messages and protect CLI draft saves

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

@@ -27,6 +27,42 @@ describe("bb thread spawn command output", () => {
     return vi.spyOn(process.stderr, "write").mockImplementation(() => true);
   }
 
+  it("saves the first message with the shipped Drafts submission", async () => {
+    const post = vi.fn(async ({ json }: { json: unknown }) => {
+      createThreadRequestSchema.parse(json);
+      return fixtures.makeThread({
+        id: "thread-draft",
+        projectId: "proj-1",
+        providerId: "codex",
+        status: "pending",
+      });
+    });
+    stubServerApi({ "v1.threads.$post": post });
+
+    await runCommand(
+      [
+        "thread",
+        "spawn",
+        "--project",
+        "proj-1",
+        "--prompt",
+        "Save this",
+        "--draft",
+      ],
+      register,
+    );
+
+    expect(post).toHaveBeenCalledWith({
+      json: expect.objectContaining({
+        pluginSubmission: { pluginId: "drafts", data: { kind: "draft" } },
+        input: [{ type: "text", text: "Save this", mentions: [] }],
+      }),
+    });
+    expect(collectLogLines(vi.mocked(console.log))[0]).toBe(
+      "Draft saved: thread-draft",
+    );
+  });
+
   it("rejects explicitly empty lifecycle ownership instead of creating an independent thread", async () => {
     const post = vi.fn(async ({ json }: { json: unknown }) => {
       createThreadRequestSchema.parse(json);

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

@@ -18,6 +18,60 @@ describe("bb thread tell command output", () => {
   const register: CommandRegistrar = (program) =>
     registerThreadCommands(program, () => "http://server");
 
+  it("saves a follow-up draft without steering the active turn", async () => {
+    const post = vi.fn(async () => ({
+      ok: true,
+      delivery: "queued",
+      queuedMessage: {
+        id: "qm-draft",
+        waitingOn: { kind: "plugin", pluginId: "drafts", reason: "Draft" },
+        sendAt: null,
+      },
+    }));
+    stubServerApi({ "v1.threads.:id.send.$post": post });
+
+    await runCommand(
+      ["thread", "tell", "thread-draft", "Save this", "--draft"],
+      register,
+    );
+
+    expect(post).toHaveBeenCalledWith({
+      param: { id: "thread-draft" },
+      json: expect.objectContaining({
+        mode: "queue-if-active",
+        pluginSubmission: { pluginId: "drafts", data: { kind: "draft" } },
+      }),
+    });
+    expect(vi.mocked(console.log).mock.calls[0]?.[0]).toBe(
+      "Thread thread-draft draft saved; send it with bb thread queue send",
+    );
+  });
+
+  it("rejects conflicting draft and steer requests before sending", async () => {
+    const post = vi.fn();
+    stubServerApi({ "v1.threads.:id.send.$post": post });
+
+    await expect(
+      runCommand(
+        [
+          "thread",
+          "tell",
+          "thread-draft",
+          "Save this",
+          "--draft",
+          "--mode",
+          "steer",
+        ],
+        register,
+      ),
+    ).rejects.toThrow("process.exit:1");
+
+    expect(post).not.toHaveBeenCalled();
+    expect(vi.mocked(console.error).mock.calls[0]?.[0]).toBe(
+      "Error: --draft cannot be combined with --mode steer or auto.",
+    );
+  });
+
   it("bb thread tell --json prints the raw response plus thread id", async () => {
     const post = vi.fn(async () => ({ ok: true, delivery: "sent" }));
     stubServerApi({ "v1.threads.:id.send.$post": post });

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

@@ -69,6 +69,7 @@ interface ThreadDeleteCommandOptions {
 }
 
 interface ThreadTellCommandOptions {
+  draft?: boolean;
   json?: boolean;
   messageFile?: string;
   model?: string;
@@ -106,6 +107,7 @@ interface ThreadEditMessageCommandOptions {
 type ThreadTellDeliveryMode = "auto" | "queue" | "steer";
 
 interface PostThreadMessageArgs {
+  draft?: boolean;
   getUrl: () => string;
   threadId: string;
   message: string;
@@ -440,6 +442,7 @@ export function registerActionsCommands(
     .command("tell <id> [message]")
     .aliases(["message", "send"])
     .description("Send a follow-up message to a thread")
+    .option("--draft", "Save the message as a draft until you send it manually")
     .option(
       "--message-file <path>",
       `Read the message from a file instead of [message]; ${TEXT_FILE_HELP_SUFFIX}`,
@@ -483,11 +486,18 @@ export function registerActionsCommands(
             inline: inlineMessage,
             inlineLabel: "<message>",
           });
+          const mode = resolveThreadMessageMode(opts.mode);
+          if (opts.draft && opts.mode !== undefined && mode !== "queue") {
+            throw new Error(
+              "--draft cannot be combined with --mode steer or auto.",
+            );
+          }
           const response = await postThreadMessage({
             getUrl,
             threadId: id,
             message,
-            mode: resolveThreadMessageMode(opts.mode),
+            mode: opts.draft ? "queue" : mode,
+            draft: opts.draft,
             model: opts.model,
             permissionMode: parsePermissionMode(opts.permissionMode),
             reasoningLevel: parseReasoningLevel(opts.reasoningLevel),
@@ -618,6 +628,9 @@ async function postThreadMessage(
     ...(args.serviceTier ? { serviceTier: args.serviceTier } : {}),
     ...(args.senderThreadId ? { senderThreadId: args.senderThreadId } : {}),
     ...(args.sendAt === undefined ? {} : { sendAt: args.sendAt }),
+    ...(args.draft
+      ? { pluginSubmission: { pluginId: "drafts", data: { kind: "draft" } } }
+      : {}),
   });
   return { ...response, mode: args.mode };
 }
@@ -627,6 +640,12 @@ function describeThreadTellOutcome(
   response: PostThreadMessageResult,
 ): string {
   if (response.delivery === "queued") {
+    if (
+      response.queuedMessage.waitingOn?.kind === "plugin" &&
+      response.queuedMessage.waitingOn.pluginId === "drafts"
+    ) {
+      return `Thread ${threadId} draft saved; send it with bb thread queue send`;
+    }
     // The server says WHY it is waiting, so the CLI does not have to guess
     // from the flags it happened to send. `bb thread queue list` shows the
     // same reason for the row afterwards.

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

@@ -73,6 +73,7 @@ interface ThreadSpawnCommandOptions {
   sourceSeqEnd?: string;
   visibility?: string;
   sendAt?: string;
+  draft?: boolean;
 }
 
 export function looksLikePath(value: string): boolean {
@@ -392,6 +393,10 @@ export function registerSpawnCommand(
       "JSON value for an --environment-provider that declares inputs (`bb environment providers --json` shows the schema)",
     )
     .option("--send-at <when>", SEND_AT_HELP)
+    .option(
+      "--draft",
+      "Save the first message as a draft until you send it manually",
+    )
     .option("--origin-kind <kind>", "Thread origin: fork")
     .option("--source-thread <id>", "Source thread for a fork")
     .option(
@@ -596,14 +601,24 @@ export function registerSpawnCommand(
             ...(opts.sourceThread ? { sourceThreadId: opts.sourceThread } : {}),
             ...(sourceSeqEnd !== undefined ? { sourceSeqEnd } : {}),
             ...(sendAt !== undefined ? { sendAt } : {}),
+            ...(opts.draft
+              ? {
+                  pluginSubmission: {
+                    pluginId: "drafts",
+                    data: { kind: "draft" },
+                  },
+                }
+              : {}),
           });
         } catch (err: unknown) {
           throw prependErrorContext("Failed to create thread", err);
         }
 
         if (outputJson(opts, thread)) return;
-        console.log(`Thread spawned: ${thread.id}`);
-        if (sendAt !== undefined) {
+        console.log(
+          `${opts.draft ? "Draft saved" : "Thread spawned"}: ${thread.id}`,
+        );
+        if (sendAt !== undefined && !opts.draft) {
           console.log(
             `First message scheduled for ${new Date(sendAt).toLocaleString()}; the thread stays pending until then.`,
           );

apps/server/src/services/plugins/plugin-hook-registry.ts

@@ -52,6 +52,7 @@ export async function invokeBridgedProvider<T>(
  * seams — and because it is the seam a test substitutes fake handlers through.
  */
 export interface PluginHookProvider {
+  isPluginRunning?(pluginId: string): boolean;
   /** Registered handlers for a hook, in plugin install order. */
   listHooks<K extends PluginHookName>(hook: K): PluginHookRegistration<K>[];
   /**
@@ -77,7 +78,9 @@ export interface PluginHookProvider {
  */
 let provider: PluginHookProvider | undefined;
 
-export function setPluginHookProvider(next: PluginHookProvider | undefined): void {
+export function setPluginHookProvider(
+  next: PluginHookProvider | undefined,
+): void {
   provider = next;
 }
 

apps/server/src/services/plugins/plugin-service.ts

@@ -1266,6 +1266,15 @@ export function createPluginService(deps: PluginServiceDeps): PluginService {
     },
 
     hooks: {
+      isPluginRunning(pluginId) {
+        const row = getInstalledPlugin(deps.db, pluginId);
+        return (
+          row !== undefined &&
+          row.enabled &&
+          loaded.has(pluginId) &&
+          getStatus(row).status === "running"
+        );
+      },
       listHooks: listPluginHooks,
       invokeHook: invokeIsolated,
       decisionTimeoutMs: DEFAULT_PLUGIN_HOOK_TIMEOUT_MS,

apps/server/src/services/threads/dispatch-attempt.ts

@@ -43,6 +43,8 @@ import {
   dispatchExecutionSources,
   dispatchWaitReasonForPass,
   hasMessageDispatchHooks,
+  isDraftSubmission,
+  requireDraftSubmissionAvailable,
   noteDispatchRequeued,
   runMessageDispatchHookPass,
   type DispatchAttemptKind,
@@ -293,6 +295,7 @@ async function runDispatchAttempt(
   reattempted: boolean,
 ): Promise<DispatchAttemptOutcome> {
   const { payload, thread } = args;
+  requireDraftSubmissionAvailable(args.pluginSubmission);
   // A stopping thread is writable HERE and nowhere upstream: the checkpoint
   // below turns it into a core wait, which is a truthful "not yet" the row can
   // recover from, rather than the 409 that used to make a stop a dead end for
@@ -505,7 +508,10 @@ async function runDispatchAttempt(
     }
   };
 
-  if (!sendNow && hasMessageDispatchHooks()) {
+  if (
+    !sendNow &&
+    (isDraftSubmission(args.pluginSubmission) || hasMessageDispatchHooks())
+  ) {
     const outcome = await runMessageDispatchHookPass(deps, {
       thread,
       threadResponse: toThreadResponseFromThread(deps, { thread }),

apps/server/src/services/threads/dispatch-hooks.ts

@@ -166,6 +166,38 @@ export function hasMessageDispatchHooks(): boolean {
   );
 }
 
+export function isDraftSubmission(
+  submission: MessageDispatchHookContext["experimental_submission"] | undefined,
+): boolean {
+  return (
+    submission?.pluginId === "drafts" &&
+    submission.data !== null &&
+    typeof submission.data === "object" &&
+    !Array.isArray(submission.data) &&
+    submission.data.kind === "draft"
+  );
+}
+
+export function requireDraftSubmissionAvailable(
+  submission: MessageDispatchHookContext["experimental_submission"] | undefined,
+): void {
+  if (!isDraftSubmission(submission)) return;
+  const provider = pluginHookProvider();
+  if (
+    provider?.isPluginRunning?.("drafts") !== true ||
+    !provider
+      .listHooks("message.dispatch")
+      .some((hook) => hook.pluginId === "drafts")
+  ) {
+    throw new ApiError(
+      409,
+      "drafts_unavailable",
+      "Drafts must be installed, enabled, and running to save a draft.",
+      { details: { pluginId: "drafts" } },
+    );
+  }
+}
+
 /**
  * Server-wide evaluation lock.
  *
@@ -393,6 +425,7 @@ export async function runMessageDispatchHookPass(
   deps: DispatchHookDeps,
   request: MessageDispatchHookPassRequest,
 ): Promise<MessageDispatchHookPassOutcome> {
+  requireDraftSubmissionAvailable(request.pluginSubmission);
   const provider = pluginHookProvider();
   if (provider === undefined) {
     return { kind: "proceed" };
@@ -403,6 +436,7 @@ export async function runMessageDispatchHookPass(
   }
 
   return withEvaluationLock(async () => {
+    requireDraftSubmissionAvailable(request.pluginSubmission);
     const context = buildHookContext(deps, request);
     const waits: MessageDispatchWaitDecision[] = [];
 
@@ -447,12 +481,32 @@ export async function runMessageDispatchHookPass(
       }
     }
 
-    const waiter = waits[0];
+    const draftWait = waits.find((wait) => wait.pluginId === "drafts");
+    if (
+      isDraftSubmission(request.pluginSubmission) &&
+      (draftWait === undefined || draftWait.sendAt !== null)
+    ) {
+      throw messageDispatchHookFailure(
+        "drafts",
+        "did not hold the draft for manual dispatch",
+      );
+    }
+    const firstQueuedWait = request.queuedMessages[0]?.waitingOn;
+    const preserveDraftHold =
+      isDraftSubmission(request.pluginSubmission) ||
+      (firstQueuedWait?.kind === "plugin" &&
+        firstQueuedWait.pluginId === "drafts");
+    const waiter =
+      preserveDraftHold && draftWait !== undefined ? draftWait : waits[0];
     if (waiter === undefined) {
       await request.continueAfterHooks?.();
       return { kind: "proceed" };
     }
-    return { kind: "wait", waiter, additionalWaiters: waits.slice(1) };
+    return {
+      kind: "wait",
+      waiter,
+      additionalWaiters: waits.filter((wait) => wait !== waiter),
+    };
   });
 }
 

apps/server/src/services/threads/thread-create.ts

@@ -18,6 +18,7 @@ import type {
   LoggedPendingInteractionWorkSessionDeps,
 } from "../../types.js";
 import { ApiError } from "../../errors.js";
+import { requireDraftSubmissionAvailable } from "./dispatch-hooks.js";
 import { ensureHostSessionReadyForWork } from "../hosts/host-lifecycle.js";
 import { buildExecutionOptions } from "./thread-commands.js";
 import {
@@ -525,6 +526,7 @@ export async function createThreadFromRequest(
     forkSourceEnvironmentId?: string;
   } = {},
 ) {
+  requireDraftSubmissionAvailable(rawRequestInput.pluginSubmission);
   const project = requirePublicProjectForThreadCreate(
     deps,
     rawRequestInput.projectId,