feat(threads): add send-now action for queued follow-up messages

apps/api/src/router.ts

@@ -272,6 +272,7 @@ import {
   groupDefaultBotId,
   reactToThreadMessage,
   resolveThreadTarget,
+  sendQueuedThreadMessagesNow,
   sendThreadMessage,
   setThreadUnreadState,
   stopThreadRuns,
@@ -2062,6 +2063,18 @@ export function createRouter(deps: RouterDeps) {
         await stopThreadRuns(deps, context.actor, target);
         return { ok: true as const };
       }),
+      sendNow: authed.threads.sendNow.handler(async ({ context, input }) => {
+        if ((await modelSetup(deps, context.actor)).needsModel) {
+          throw new ORPCError("BAD_REQUEST", { message: "Connect a model to start a run." });
+        }
+        const target = await resolveThreadTarget(deps.prisma, context.actor, input);
+        if (target.kind === "bot") {
+          await assertTeachingSendAllowed(deps.prisma, context.actor.spaceId, target.botId);
+        }
+        return sendQueuedThreadMessagesNow(deps, context.actor, target, {
+          messageId: input.messageId,
+        });
+      }),
       clear: authed.threads.clear.handler(async ({ context, input }) => {
         const target = await resolveThreadTarget(deps.prisma, context.actor, input);
         const contextBotId = target.kind === "bot" ? target.botId : groupDefaultBotId(target);

apps/api/src/thread-target.test.ts

@@ -8,6 +8,7 @@ import {
   groupDefaultBotId,
   reactToThreadMessage,
   resolveGroupFollowUpBotId,
+  sendQueuedThreadMessagesNow,
   sendThreadMessage,
   stopThreadRuns,
   type ThreadTarget,
@@ -2118,4 +2119,115 @@ describe("stopThreadRuns", () => {
       expect.objectContaining({ runId: "run-b" }),
     );
   });
+
+  it("sendQueuedThreadMessagesNow cancels the active run, preserves pending steering, and enqueues an immediate continuation run", async () => {
+    const queuedCreatedAt = new Date("2026-01-01T10:00:05.000Z");
+    const transaction = {
+      $queryRaw: vi.fn(),
+      steeringMessage: {
+        findMany: vi.fn().mockResolvedValue([
+          {
+            id: "steer-1",
+            botId: "bot-a",
+            userId: "user-1",
+            messageId: "msg-queued-1",
+            claimedAt: null,
+            message: {
+              id: "msg-queued-1",
+              blocks: [{ kind: "text", text: "Send this right now" }],
+              seq: 2,
+              createdAt: queuedCreatedAt,
+            },
+          },
+        ]),
+        deleteMany: vi.fn().mockResolvedValue({ count: 0 }),
+        updateMany: vi.fn().mockResolvedValue({ count: 1 }),
+      },
+      run: {
+        updateManyAndReturn: vi
+          .fn()
+          .mockResolvedValue([{ id: "run-old", taskId: "task-old" }]),
+        create: vi.fn().mockResolvedValue({ id: "run-new", status: "queued" }),
+      },
+      task: {
+        updateMany: vi.fn().mockResolvedValue({ count: 1 }),
+        create: vi.fn().mockResolvedValue({ id: "task-new" }),
+      },
+      message: {
+        updateMany: vi.fn().mockResolvedValue({ count: 1 }),
+      },
+      computerExecutionLease: {
+        findMany: vi.fn().mockResolvedValue([]),
+      },
+      computer: {
+        findMany: vi.fn().mockResolvedValue([]),
+      },
+      event: {
+        findFirst: vi.fn().mockResolvedValue({ seq: 42 }),
+      },
+    };
+    const prisma = {
+      $transaction: vi.fn(async (callback: (client: typeof transaction) => unknown) =>
+        callback(transaction),
+      ),
+      computerExecutionLease: {
+        updateMany: vi.fn().mockResolvedValue({ count: 0 }),
+      },
+      computer: {
+        updateMany: vi.fn().mockResolvedValue({ count: 0 }),
+      },
+      event: {
+        deleteMany: vi.fn().mockResolvedValue({ count: 0 }),
+      },
+    } as unknown as PrismaClient;
+    const jobs = {
+      enqueue: vi.fn().mockResolvedValue(undefined),
+      cancel: vi.fn().mockResolvedValue(undefined),
+    };
+    const events = {
+      notify: vi.fn().mockResolvedValue(undefined),
+    };
+    const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor;
+    const target = {
+      kind: "bot",
+      botId: "bot-a",
+      threadId: "thread-1",
+      bot: {} as never,
+    } satisfies ThreadTarget;
+
+    const result = await sendQueuedThreadMessagesNow(
+      {
+        prisma,
+        sandbox: {} as SandboxProvider,
+        events: events as never,
+        jobs: jobs as never,
+      },
+      actor,
+      target,
+      { messageId: "msg-queued-1" },
+    );
+
+    expect(result).toEqual({ ok: true, runIds: ["run-new"] });
+    expect(transaction.run.updateManyAndReturn).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: expect.objectContaining({
+          threadId: "thread-1",
+          botId: { in: ["bot-a"] },
+        }),
+        data: expect.objectContaining({ status: "cancelled" }),
+      }),
+    );
+    expect(transaction.steeringMessage.updateMany).toHaveBeenCalledWith({
+      where: { id: { in: ["steer-1"] } },
+      data: { runId: "run-new", claimedAt: null },
+    });
+    expect(jobs.cancel).toHaveBeenCalledWith("run:run-old");
+    expect(jobs.enqueue).toHaveBeenCalledWith({
+      name: "run.continue",
+      payload: { runId: "run-new" },
+      replaceKey: "run:run-new",
+    });
+    expect(events.notify).toHaveBeenCalledWith("thread-1", 42);
+  });
 });
+

apps/api/src/thread-target.ts

@@ -1,5 +1,10 @@
 import { ORPCError } from "@orpc/server";
-import { type JobPublisher, runContinueJob, type SandboxProvider } from "@rakazo/adapter-kit";
+import {
+  type JobPublisher,
+  runContinueJob,
+  runJobKey,
+  type SandboxProvider,
+} from "@rakazo/adapter-kit";
 import { cancelComputerRunWork, screenLeaseIdForRun, toComputerRef } from "@rakazo/adapters";
 import {
   type Actor,
@@ -1151,63 +1156,50 @@ export async function reactToThreadMessage(
   });
 }
 
-export async function stopThreadRuns(
+async function snapshotCancelledRunTeardown(tx: Prisma.TransactionClient, runIds: string[]) {
+  // Snapshot teardown coordinates before commit. Once cancellation becomes
+  // visible, a worker can release its lease / execution columns immediately; a
+  // later lookup would then miss the sandbox work this request must stop.
+  // Team ownership lives on ComputerExecutionLease; Computer.executionRunId is
+  // only a legacy secondary path and is not written by current acquisition.
+  const leases = runIds.length
+    ? await tx.computerExecutionLease.findMany({
+        where: { runId: { in: runIds } },
+        select: { computerId: true, botId: true, runId: true, fence: true },
+      })
+    : [];
+  const leaseComputerIds = [...new Set(leases.map((lease) => lease.computerId))];
+  const computers = runIds.length
+    ? await tx.computer.findMany({
+        where:
+          leaseComputerIds.length > 0
+            ? {
+                OR: [{ id: { in: leaseComputerIds } }, { executionRunId: { in: runIds } }],
+              }
+            : { executionRunId: { in: runIds } },
+        select: {
+          id: true,
+          homeKey: true,
+          kind: true,
+          providerRef: true,
+          executionBotId: true,
+          executionRunId: true,
+        },
+      })
+    : [];
+  return { runIds, computers, leases };
+}
+
+async function teardownCancelledRuns(
   deps: {
     prisma: PrismaClient;
     sandbox: SandboxProvider;
   },
   actor: Actor,
-  target: ThreadTarget,
+  snapshot: Awaited<ReturnType<typeof snapshotCancelledRunTeardown>>,
 ) {
-  const { runIds, computers, leases } = await deps.prisma.$transaction(async (tx) => {
+  const { runIds, computers, leases } = snapshot;
-    await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR UPDATE`;
+  if (runIds.length === 0) return;
-    const cancelled = await tx.run.updateManyAndReturn({
-      where: {
-        threadId: target.threadId,
-        status: { in: [...ACTIVE_RUN_STATUSES] },
-      },
-      data: { status: "cancelled", completedAt: new Date() },
-      select: { id: true },
-    });
-    const ids = cancelled.map((run) => run.id);
-    await tx.steeringMessage.deleteMany({
-      where: {
-        botId: { in: target.kind === "bot" ? [target.botId] : target.memberBotIds },
-        message: { threadId: target.threadId },
-      },
-    });
-    // Snapshot teardown coordinates before commit. Once cancellation becomes
-    // visible, a worker can release its lease / execution columns immediately; a
-    // later lookup would then miss the sandbox work this request must stop.
-    // Team ownership lives on ComputerExecutionLease; Computer.executionRunId is
-    // only a legacy secondary path and is not written by current acquisition.
-    const leases = ids.length
-      ? await tx.computerExecutionLease.findMany({
-          where: { runId: { in: ids } },
-          select: { computerId: true, botId: true, runId: true, fence: true },
-        })
-      : [];
-    const leaseComputerIds = [...new Set(leases.map((lease) => lease.computerId))];
-    const computers = ids.length
-      ? await tx.computer.findMany({
-          where:
-            leaseComputerIds.length > 0
-              ? {
-                  OR: [{ id: { in: leaseComputerIds } }, { executionRunId: { in: ids } }],
-                }
-              : { executionRunId: { in: ids } },
-          select: {
-            id: true,
-            homeKey: true,
-            kind: true,
-            providerRef: true,
-            executionBotId: true,
-            executionRunId: true,
-          },
-        })
-      : [];
-    return { runIds: ids, computers, leases };
-  });
   // Keep the DB lease until after teardown so a replacement run cannot claim the
   // screen while we still need the cancelled run's screenLeaseId to release it.
   const computerById = new Map(computers.map((computer) => [computer.id, computer]));
@@ -1285,6 +1277,218 @@ export async function stopThreadRuns(
   });
 }
 
+export async function stopThreadRuns(
+  deps: {
+    prisma: PrismaClient;
+    sandbox: SandboxProvider;
+  },
+  actor: Actor,
+  target: ThreadTarget,
+) {
+  const snapshot = await deps.prisma.$transaction(async (tx) => {
+    await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR UPDATE`;
+    const cancelled = await tx.run.updateManyAndReturn({
+      where: {
+        threadId: target.threadId,
+        status: { in: [...ACTIVE_RUN_STATUSES] },
+      },
+      data: { status: "cancelled", completedAt: new Date() },
+      select: { id: true },
+    });
+    const ids = cancelled.map((run) => run.id);
+    await tx.steeringMessage.deleteMany({
+      where: {
+        botId: { in: target.kind === "bot" ? [target.botId] : target.memberBotIds },
+        message: { threadId: target.threadId },
+      },
+    });
+    return snapshotCancelledRunTeardown(tx, ids);
+  });
+  await teardownCancelledRuns(deps, actor, snapshot);
+}
+
+/**
+ * Promote queued follow-up messages immediately: cancel any currently active runs
+ * holding up the queue without discarding unclaimed steering messages, create a
+ * new continuation run for the pending steering messages, and enqueue it right away.
+ */
+export async function sendQueuedThreadMessagesNow(
+  deps: {
+    prisma: PrismaClient;
+    sandbox: SandboxProvider;
+    events: ThreadEvents;
+    jobs: JobPublisher;
+  },
+  actor: Actor,
+  target: ThreadTarget,
+  input?: { messageId?: string },
+) {
+  const botIds = target.kind === "bot" ? [target.botId] : target.memberBotIds;
+  const committed = await deps.prisma.$transaction(async (tx) => {
+    await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR UPDATE`;
+    const pendingSteering = await tx.steeringMessage.findMany({
+      where: {
+        botId: { in: botIds },
+        claimedAt: null,
+        message: { threadId: target.threadId },
+      },
+      include: {
+        message: { select: { id: true, blocks: true, seq: true, createdAt: true } },
+      },
+      orderBy: [{ message: { seq: "asc" } }, { id: "asc" }],
+    });
+
+    const pendingByBotId = new Map<string, typeof pendingSteering>();
+    for (const item of pendingSteering) {
+      const list = pendingByBotId.get(item.botId) ?? [];
+      list.push(item);
+      pendingByBotId.set(item.botId, list);
+    }
+
+    if (pendingByBotId.size === 0 && input?.messageId) {
+      const targetMsg = await tx.message.findFirst({
+        where: { id: input.messageId, threadId: target.threadId, role: "user" },
+        select: { id: true, blocks: true, seq: true, createdAt: true },
+      });
+      if (targetMsg) {
+        const activeRuns = await tx.run.findMany({
+          where: {
+            threadId: target.threadId,
+            botId: { in: botIds },
+            status: { in: [...ACTIVE_RUN_STATUSES] },
+            createdAt: { lt: targetMsg.createdAt },
+          },
+          select: { id: true, botId: true },
+        });
+        const fallbackBotId =
+          activeRuns[0]?.botId ??
+          (target.kind === "bot" ? target.botId : groupDefaultBotId(target));
+        if (fallbackBotId && activeRuns.length > 0) {
+          const steering = await tx.steeringMessage.create({
+            data: {
+              messageId: targetMsg.id,
+              botId: fallbackBotId,
+              userId: actor.userId,
+              runId: activeRuns[0]?.id ?? null,
+            },
+          });
+          pendingByBotId.set(fallbackBotId, [{ ...steering, message: targetMsg }]);
+        }
+      }
+    }
+
+    if (pendingByBotId.size === 0) {
+      return {
+        teardown: { runIds: [] as string[], computers: [], leases: [] },
+        runs: [] as Array<{ id: string; status: string }>,
+        eventSeq: null as number | null,
+      };
+    }
+
+    const affectedBotIds = [...pendingByBotId.keys()];
+    const now = new Date();
+    const cancelled = await tx.run.updateManyAndReturn({
+      where: {
+        threadId: target.threadId,
+        botId: { in: affectedBotIds },
+        status: { in: [...ACTIVE_RUN_STATUSES] },
+      },
+      data: { status: "cancelled", completedAt: now },
+      select: { id: true, taskId: true },
+    });
+    const cancelledRunIds = cancelled.map((run) => run.id);
+    if (cancelled.length > 0) {
+      await tx.task.updateMany({
+        where: { id: { in: cancelled.map((run) => run.taskId) } },
+        data: { status: "cancelled" },
+      });
+      await tx.steeringMessage.deleteMany({
+        where: {
+          runId: { in: cancelledRunIds },
+          claimedAt: { not: null },
+        },
+      });
+    }
+    const teardown = await snapshotCancelledRunTeardown(tx, cancelledRunIds);
+
+    const createdRuns: Array<{ id: string; status: string }> = [];
+    for (const [botId, items] of pendingByBotId) {
+      const last = items.at(-1)!;
+      const task = await tx.task.create({
+        data: {
+          spaceId: actor.spaceId,
+          botId,
+          threadId: target.threadId,
+          userId: items[0]!.userId,
+          prompt: "Respond to the user's steering context.",
+          status: "queued",
+        },
+      });
+      const runCreatedAt = new Date(
+        Math.max(Date.now(), last.message.createdAt.getTime() + 1),
+      );
+      const run = await tx.run.create({
+        data: {
+          spaceId: actor.spaceId,
+          botId,
+          threadId: target.threadId,
+          taskId: task.id,
+          userId: items[0]!.userId,
+          status: "queued",
+          trigger: "follow_up",
+          sourceMessageId: last.message.id,
+          createdAt: runCreatedAt,
+        },
+        select: { id: true, status: true },
+      });
+      await tx.steeringMessage.updateMany({
+        where: { id: { in: items.map((item) => item.id) } },
+        data: { runId: run.id, claimedAt: null },
+      });
+      await tx.message.updateMany({
+        where: { id: { in: items.map((item) => item.message.id) } },
+        data: { runId: run.id },
+      });
+      createdRuns.push(run);
+    }
+
+    if (target.kind === "group") {
+      await touchGroupUpdatedAt(tx, target.groupId);
+    }
+    const latestEvent = await tx.event.findFirst({
+      where: { threadId: target.threadId },
+      orderBy: { seq: "desc" },
+      select: { seq: true },
+    });
+    return {
+      teardown,
+      runs: createdRuns,
+      eventSeq: latestEvent?.seq ?? null,
+    };
+  });
+
+  if (committed.teardown.runIds.length > 0) {
+    await Promise.all([
+      teardownCancelledRuns(deps, actor, committed.teardown),
+      ...committed.teardown.runIds.map((runId) =>
+        deps.jobs.cancel(runJobKey(runId)).catch(() => undefined),
+      ),
+    ]);
+  }
+  if (committed.runs.length > 0) {
+    await enqueueRunsNeedingContinue(deps.jobs, committed.runs);
+    if (committed.eventSeq != null) {
+      await deps.events.notify(target.threadId, committed.eventSeq).catch((error) => {
+        getLogger().error("thread sendNow realtime notification", error);
+      });
+    }
+  }
+  return {
+    ok: true as const,
+    runIds: committed.runs.map((run) => run.id),
+  };
+}
+
 export async function setThreadUnreadState(
   prisma: PrismaClient,
   actor: Actor,

apps/mobile/app/thread.tsx

@@ -29,6 +29,7 @@ import {
   liveScreenRunId,
   mentionChipKey,
   projectMessageReactions,
+  queuedFollowUps,
   readComposerDraft,
   resolveComposerSendPlan,
   SLASH_ACTIONS,
@@ -435,6 +436,20 @@ function Thread() {
     });
   }, [inGroup, snap?.activeRuns, snap?.members, snap?.run]);
   const working = inGroup ? workingGroupBots.length > 0 : isWorkingStatus(currentBotStatus);
+  const activeRunCreatedAt = useMemo(() => {
+    const candidates = snap?.activeRuns ?? (snap?.run ? [snap.run] : []);
+    return candidates.find((run) => isWorkingStatus(run.status))?.createdAt;
+  }, [snap?.activeRuns, snap?.run]);
+  const queuedIds = useMemo(
+    () =>
+      new Set(
+        (working
+          ? queuedFollowUps(snap?.messages ?? [], activeRunCreatedAt, reactionView.reactions)
+          : []
+        ).map((message) => message.id),
+      ),
+    [activeRunCreatedAt, reactionView.reactions, snap?.messages, working],
+  );
 
   useEffect(() => {
     void rpc<AgentSkillCatalogEntry[]>("agentSkills/list")
@@ -1258,6 +1273,31 @@ function Thread() {
     }
   }
 
+  const [sendingNowId, setSendingNowId] = useState<string | null>(null);
+
+  async function sendQueuedNow(message: MobileMessage) {
+    const targetBotId = botId;
+    const targetGroupId = groupId;
+    if ((!targetBotId && !targetGroupId) || sendingNowId) return;
+    setSendingNowId(message.id);
+    setError(null);
+    try {
+      await rpc("threads/sendNow", {
+        ...(targetGroupId ? { groupId: targetGroupId } : { botId: targetBotId! }),
+        messageId: message.id,
+      });
+      if (isCurrentTarget(targetBotId, targetGroupId)) {
+        await refresh();
+      }
+    } catch (err) {
+      if (isCurrentTarget(targetBotId, targetGroupId)) {
+        setError(err instanceof Error ? err.message : t("Failed to send message"));
+      }
+    } finally {
+      setSendingNowId(null);
+    }
+  }
+
   const answerMessage = useCallback(
     async (message: MobileMessage, answer: string) => {
       const targetBotId = botId;
@@ -1610,6 +1650,38 @@ function Thread() {
               actionProps={actionProps}
             />
           </Pressable>
+          {queuedIds.has(message.id) ? (
+            <View
+              style={{
+                flexDirection: "row",
+                alignItems: "center",
+                justifyContent: "flex-end",
+                gap: 8,
+                marginTop: 4,
+              }}
+            >
+              <Text style={{ color: tokens.mutedForeground, fontSize: 12 }}>
+                {t("Queued")}
+              </Text>
+              <Pressable
+                accessibilityRole="button"
+                disabled={sendingNowId === message.id}
+                hitSlop={6}
+                onPress={() => void sendQueuedNow(message)}
+              >
+                <Text
+                  style={{
+                    color: tokens.foreground,
+                    fontSize: 12,
+                    fontWeight: "600",
+                    opacity: sendingNowId === message.id ? 0.5 : 1,
+                  }}
+                >
+                  {t("Send now")}
+                </Text>
+              </Pressable>
+            </View>
+          ) : null}
           {messageReactions ? (
             <View
               style={{

apps/mobile/lib/api.ts

@@ -784,7 +784,7 @@ export type MobileGroup = Pick<
   | "updatedAt"
   | "members"
 > &
-  Partial<Pick<Group, "spaceId">>;
+  Partial<Pick<Group, "spaceId" | "kind">>;
 
 export type MobileSpace = Space;
 export type MobileSpaceNavigation = SpaceNavigation;
@@ -797,8 +797,14 @@ export type MobileSnapshot = {
   cursor?: number;
   messages: MobileMessage[];
   olderCursor: number | null;
-  run: { id: string; botId?: string; status: string; error?: string | null } | null;
+  run: {
-  activeRuns?: Array<{ id: string; botId?: string; status: string }>;
+    id: string;
+    botId?: string;
+    status: string;
+    error?: string | null;
+    createdAt?: string;
+  } | null;
+  activeRuns?: Array<{ id: string; botId?: string; status: string; createdAt?: string }>;
   members?: MobileGroup["members"];
   computer?: {
     kind?: string;

apps/mobile/lib/locales/de.ts

@@ -497,6 +497,7 @@ export const DE_MESSAGES: Record<string, string> = {
   "Section name": "Abschnittsname",
   Send: "Senden",
   "Send answer": "Antwort senden",
+  "Send now": "Jetzt gleich senden",
   "Send reset link": "Link zum Zurücksetzen senden",
   "Sending…": "Wird gesendet…",
   Server: "Server",

apps/mobile/lib/locales/ru.ts

@@ -429,6 +429,7 @@ export const RU_MESSAGES: Record<string, string> = {
   Send: "Отправить",
   "Send reset link": "Отправить ссылку для сброса",
   "Send answer": "Отправить ответ",
+  "Send now": "Отправить сейчас",
   "Sending…": "Отправка…",
   Server: "Сервер",
   "Server URL": "URL-адрес сервера",

apps/mobile/lib/locales/zh.ts

@@ -410,6 +410,7 @@ export const ZH_MESSAGES: Record<string, string> = {
   Send: "发送",
   "Send reset link": "发送重置链接",
   "Send answer": "发送回答",
+  "Send now": "立即发送",
   "Sending…": "正在发送…",
   Server: "服务器",
   "Server URL": "服务器地址",

apps/web/src/lib/queued-follow-ups.test.ts

@@ -37,4 +37,10 @@ describe("queuedFollowUps", () => {
     const synthetic = userMessage("progress:1", "2026-01-01T10:00:05.000Z");
     expect(queuedFollowUps([synthetic], RUN_START, new Map())).toEqual([]);
   });
+
+  it("clears queued state once sendNow starts a newer run after the message", () => {
+    const followUp = userMessage("queued-1", "2026-01-01T10:00:05.000Z");
+    expect(queuedFollowUps([followUp], RUN_START, new Map())).toEqual([followUp]);
+    expect(queuedFollowUps([followUp], "2026-01-01T10:00:05.001Z", new Map())).toEqual([]);
+  });
 });

apps/web/src/lib/queued-follow-ups.ts

@@ -1,25 +1 @@
-import type { ThreadMessage } from "@rakazo/contracts";
+export { queuedFollowUps } from "@rakazo/core";
-
-/** The bot marks a steering message with this reaction once it has picked it up. */
-const SEEN_REACTION = "👀";
-
-/**
- * User messages sent after the active run started that the bot has not picked
- * up yet. The run's own trigger message predates the run, so it never counts.
- */
-export function queuedFollowUps(
-  messages: readonly ThreadMessage[],
-  runCreatedAt: string | undefined,
-  reactions: ReadonlyMap<string, ReadonlyMap<string, number>>,
-): ThreadMessage[] {
-  if (!runCreatedAt) return [];
-  const runStart = Date.parse(runCreatedAt);
-  if (Number.isNaN(runStart)) return [];
-  return messages.filter(
-    (message) =>
-      message.role === "user" &&
-      !message.id.includes(":") &&
-      Date.parse(message.createdAt) > runStart &&
-      !reactions.get(message.id)?.has(SEEN_REACTION),
-  );
-}

apps/web/src/locales/de/messages.po

@@ -4933,6 +4933,10 @@ msgstr "Antwort senden"
 msgid "Send it"
 msgstr "Senden"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "Jetzt gleich senden"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "Link zum Zurücksetzen senden"

apps/web/src/locales/en/messages.po

@@ -4933,6 +4933,10 @@ msgstr "Send answer"
 msgid "Send it"
 msgstr "Send it"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "Send now"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "Send reset link"

apps/web/src/locales/es/messages.po

@@ -4933,6 +4933,10 @@ msgstr "Enviar respuesta"
 msgid "Send it"
 msgstr "Enviarlo"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "Enviar ahora"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "Enviar enlace de restablecimiento"

apps/web/src/locales/hi/messages.po

@@ -4933,6 +4933,10 @@ msgstr "उत्तर भेजें"
 msgid "Send it"
 msgstr "इसे भेजें"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "अभी भेजें"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "रीसेट लिंक भेजें"

apps/web/src/locales/ko/messages.po

@@ -4933,6 +4933,10 @@ msgstr "답변 보내기"
 msgid "Send it"
 msgstr "보내기"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "지금 보내기"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "재설정 링크 보내기"

apps/web/src/locales/pt-BR/messages.po

@@ -4933,6 +4933,10 @@ msgstr "Enviar resposta"
 msgid "Send it"
 msgstr "Enviar"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "Enviar agora"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "Enviar link de redefinição"

apps/web/src/locales/ru/messages.po

@@ -4933,6 +4933,10 @@ msgstr "Отправить ответ"
 msgid "Send it"
 msgstr "Отправить"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "Отправить сейчас"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "Отправить ссылку для сброса"

apps/web/src/locales/tr/messages.po

@@ -4933,6 +4933,10 @@ msgstr "Yanıtı gönder"
 msgid "Send it"
 msgstr "Gönder"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "Şimdi gönder"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "Sıfırlama bağlantısı gönder"

apps/web/src/locales/zh-CN/messages.po

@@ -4933,6 +4933,10 @@ msgstr "发送回答"
 msgid "Send it"
 msgstr "发送"
 
+#: src/pages/shell/transcript.tsx
+msgid "Send now"
+msgstr "立即发送"
+
 #: src/pages/Auth.tsx
 msgid "Send reset link"
 msgstr "发送重置链接"

apps/web/src/pages/Shell.tsx

@@ -2662,6 +2662,27 @@ export function ShellPage() {
     await rpc.threads.followUp({ botId: id, text });
     await refreshThreadRef.current(id);
   }, []);
+  const sendQueuedMessageNow = useCallback(
+    async (message: ThreadMessage) => {
+      const groupTarget = activeGroupId.current;
+      const botTarget = activeBotId.current;
+      const messageId = message.id.startsWith("temp-") ? undefined : message.id;
+      setSendError(null);
+      try {
+        if (groupTarget) {
+          await rpc.threads.sendNow({ groupId: groupTarget, messageId });
+          await refreshGroupThreadRef.current(groupTarget);
+        } else if (botTarget) {
+          await rpc.threads.sendNow({ botId: botTarget, messageId });
+          await refreshThreadRef.current(botTarget);
+        }
+        void refreshBots().catch(() => undefined);
+      } catch (error) {
+        setSendError(error instanceof Error ? error.message : t`Failed to send message`);
+      }
+    },
+    [refreshBots, t],
+  );
   const stopRun = useCallback(async () => {
     inputQueueRef.current.requestStop();
     setSending(true);
@@ -4128,6 +4149,7 @@ export function ShellPage() {
                       speakingMessageId={speakingMessageId}
                       onSpeak={speakMessage}
                       onRestartTask={sendMessage}
+                      onSendQueuedNow={sendQueuedMessageNow}
                     />
                   </>
                 )}

apps/web/src/pages/shell/transcript.tsx

@@ -122,6 +122,7 @@ export const Transcript = memo(function Transcript({
   encounterMode = false,
   voiceEncounterActive = false,
   onRestartTask,
+  onSendQueuedNow,
 }: {
   scrollRef: RefObject<HTMLDivElement | null>;
   artifactTarget: ArtifactTarget;
@@ -155,9 +156,11 @@ export const Transcript = memo(function Transcript({
   encounterMode?: boolean;
   voiceEncounterActive?: boolean;
   onRestartTask?: (taskPrompt: string) => void | Promise<void>;
+  onSendQueuedNow?: (message: ThreadMessage) => void | Promise<void>;
 }) {
   const { t } = useLingui();
   const [restartingPrompt, setRestartingPrompt] = useState<string | null>(null);
+  const [sendingQueuedId, setSendingQueuedId] = useState<string | null>(null);
 
   const handleRestart = useCallback(
     async (prompt: string) => {
@@ -172,6 +175,19 @@ export const Transcript = memo(function Transcript({
     [onRestartTask, restartingPrompt],
   );
 
+  const handleSendQueuedNow = useCallback(
+    async (message: ThreadMessage) => {
+      if (!onSendQueuedNow || sendingQueuedId) return;
+      setSendingQueuedId(message.id);
+      try {
+        await onSendQueuedNow(message);
+      } finally {
+        setSendingQueuedId(null);
+      }
+    },
+    [onSendQueuedNow, sendingQueuedId],
+  );
+
   const [atEnd, setAtEnd] = useState(true);
   const following = useRef(true);
   const autoScrolling = useRef(false);
@@ -564,12 +580,23 @@ export const Transcript = memo(function Transcript({
                       onSpeak={() => onSpeak(message)}
                     />
                     {queuedIds.has(message.id) ? (
-                      <p
+                      <div
                         data-testid="message-queued"
-                        className="mt-1 text-end text-ui-label text-muted-foreground"
+                        className="mt-1 flex items-center justify-end gap-2 text-end text-ui-label text-muted-foreground"
                       >
-                        {t`Queued`}
+                        <span>{t`Queued`}</span>
-                      </p>
+                        {onSendQueuedNow ? (
+                          <button
+                            type="button"
+                            data-testid="message-send-now"
+                            disabled={sendingQueuedId === message.id}
+                            onClick={() => void handleSendQueuedNow(message)}
+                            className="font-medium text-foreground underline-offset-4 transition-colors hover:underline disabled:opacity-50"
+                          >
+                            {t`Send now`}
+                          </button>
+                        ) : null}
+                      </div>
                     ) : null}
                     {!peerReceipt && messageReactions ? (
                       <div

packages/contracts/src/rpc.ts

@@ -509,6 +509,14 @@ export const appContract = {
       )
       .output(z.object({ ok: z.literal(true) })),
     stop: oc.input(threadTarget).output(z.object({ ok: z.literal(true) })),
+    sendNow: oc
+      .input(threadTarget.safeExtend({ messageId: Id.optional() }))
+      .output(
+        z.object({
+          ok: z.literal(true),
+          runIds: z.array(Id),
+        }),
+      ),
     followUp: oc
       .input(threadTarget.safeExtend({ text: z.string().min(1) }))
       .output(z.object({ ok: z.literal(true) })),

packages/core/src/index.ts

@@ -45,6 +45,7 @@ export * from "./oo-rig.js";
 export * from "./optimistic-queue.js";
 export * from "./project-look.js";
 export * from "./project-threads.js";
+export * from "./queued-follow-ups.js";
 export * from "./quiet-hours.js";
 export * from "./response-bytes.js";
 export * from "./routine-events.js";

packages/core/src/queued-follow-ups.ts

@@ -0,0 +1,26 @@
+/** The bot marks a steering message with this reaction once it has picked it up. */
+const SEEN_REACTION = "👀";
+
+/**
+ * User messages sent after the active run started that the bot has not picked
+ * up yet. The run's own trigger message predates the run, so it never counts.
+ */
+export function queuedFollowUps<
+  T extends { id: string; role: string; createdAt?: string | undefined },
+>(
+  messages: readonly T[],
+  runCreatedAt: string | undefined,
+  reactions: ReadonlyMap<string, ReadonlyMap<string, number>>,
+): T[] {
+  if (!runCreatedAt) return [];
+  const runStart = Date.parse(runCreatedAt);
+  if (Number.isNaN(runStart)) return [];
+  return messages.filter(
+    (message) =>
+      message.role === "user" &&
+      !message.id.includes(":") &&
+      typeof message.createdAt === "string" &&
+      Date.parse(message.createdAt) > runStart &&
+      !reactions.get(message.id)?.has(SEEN_REACTION),
+  );
+}