diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 80a3e6a..2225cca 100644 --- a/apps/api/src/router.ts +++ b/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); diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index 2027d31..c7d6b7c 100644 --- a/apps/api/src/thread-target.test.ts +++ b/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); + }); }); + diff --git a/apps/api/src/thread-target.ts b/apps/api/src/thread-target.ts index 0c26e27..326fa1d 100644 --- a/apps/api/src/thread-target.ts +++ b/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>, ) { - const { runIds, computers, leases } = 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 }, - }, - }); - // 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 }; - }); + const { runIds, computers, leases } = snapshot; + if (runIds.length === 0) return; // 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(); + 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, diff --git a/apps/mobile/app/thread.tsx b/apps/mobile/app/thread.tsx index 82feb24..511c88d 100644 --- a/apps/mobile/app/thread.tsx +++ b/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("agentSkills/list") @@ -1258,6 +1273,31 @@ function Thread() { } } + const [sendingNowId, setSendingNowId] = useState(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} /> + {queuedIds.has(message.id) ? ( + + + {t("Queued")} + + void sendQueuedNow(message)} + > + + {t("Send now")} + + + + ) : null} {messageReactions ? ( & - Partial>; + Partial>; 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; - activeRuns?: Array<{ id: string; botId?: string; status: string }>; + run: { + 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; diff --git a/apps/mobile/lib/locales/de.ts b/apps/mobile/lib/locales/de.ts index d105756..d10ae63 100644 --- a/apps/mobile/lib/locales/de.ts +++ b/apps/mobile/lib/locales/de.ts @@ -497,6 +497,7 @@ export const DE_MESSAGES: Record = { "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", diff --git a/apps/mobile/lib/locales/ru.ts b/apps/mobile/lib/locales/ru.ts index 9394f53..301bb7c 100644 --- a/apps/mobile/lib/locales/ru.ts +++ b/apps/mobile/lib/locales/ru.ts @@ -429,6 +429,7 @@ export const RU_MESSAGES: Record = { Send: "Отправить", "Send reset link": "Отправить ссылку для сброса", "Send answer": "Отправить ответ", + "Send now": "Отправить сейчас", "Sending…": "Отправка…", Server: "Сервер", "Server URL": "URL-адрес сервера", diff --git a/apps/mobile/lib/locales/zh.ts b/apps/mobile/lib/locales/zh.ts index a2af302..f1a8c23 100644 --- a/apps/mobile/lib/locales/zh.ts +++ b/apps/mobile/lib/locales/zh.ts @@ -410,6 +410,7 @@ export const ZH_MESSAGES: Record = { Send: "发送", "Send reset link": "发送重置链接", "Send answer": "发送回答", + "Send now": "立即发送", "Sending…": "正在发送…", Server: "服务器", "Server URL": "服务器地址", diff --git a/apps/web/src/lib/queued-follow-ups.test.ts b/apps/web/src/lib/queued-follow-ups.test.ts index 3da1ab0..7707c3d 100644 --- a/apps/web/src/lib/queued-follow-ups.test.ts +++ b/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([]); + }); }); diff --git a/apps/web/src/lib/queued-follow-ups.ts b/apps/web/src/lib/queued-follow-ups.ts index 8350266..7c0b977 100644 --- a/apps/web/src/lib/queued-follow-ups.ts +++ b/apps/web/src/lib/queued-follow-ups.ts @@ -1,25 +1 @@ -import type { ThreadMessage } from "@rakazo/contracts"; - -/** 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>, -): 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), - ); -} +export { queuedFollowUps } from "@rakazo/core"; diff --git a/apps/web/src/locales/de/messages.po b/apps/web/src/locales/de/messages.po index 7e1dc27..a471c4e 100644 --- a/apps/web/src/locales/de/messages.po +++ b/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" diff --git a/apps/web/src/locales/en/messages.po b/apps/web/src/locales/en/messages.po index baa3621..76cb20a 100644 --- a/apps/web/src/locales/en/messages.po +++ b/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" diff --git a/apps/web/src/locales/es/messages.po b/apps/web/src/locales/es/messages.po index 0b21eea..e9b23e8 100644 --- a/apps/web/src/locales/es/messages.po +++ b/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" diff --git a/apps/web/src/locales/hi/messages.po b/apps/web/src/locales/hi/messages.po index 60f2602..23b904a 100644 --- a/apps/web/src/locales/hi/messages.po +++ b/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 "रीसेट लिंक भेजें" diff --git a/apps/web/src/locales/ko/messages.po b/apps/web/src/locales/ko/messages.po index fa10821..815c754 100644 --- a/apps/web/src/locales/ko/messages.po +++ b/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 "재설정 링크 보내기" diff --git a/apps/web/src/locales/pt-BR/messages.po b/apps/web/src/locales/pt-BR/messages.po index d219d33..8664099 100644 --- a/apps/web/src/locales/pt-BR/messages.po +++ b/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" diff --git a/apps/web/src/locales/ru/messages.po b/apps/web/src/locales/ru/messages.po index 26781cb..a9fa9ef 100644 --- a/apps/web/src/locales/ru/messages.po +++ b/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 "Отправить ссылку для сброса" diff --git a/apps/web/src/locales/tr/messages.po b/apps/web/src/locales/tr/messages.po index c0908d8..11e7b77 100644 --- a/apps/web/src/locales/tr/messages.po +++ b/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" diff --git a/apps/web/src/locales/zh-CN/messages.po b/apps/web/src/locales/zh-CN/messages.po index 1c7e182..2e69d3c 100644 --- a/apps/web/src/locales/zh-CN/messages.po +++ b/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 "发送重置链接" diff --git a/apps/web/src/pages/Shell.tsx b/apps/web/src/pages/Shell.tsx index fc4f4a3..049fcda 100644 --- a/apps/web/src/pages/Shell.tsx +++ b/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} /> )} diff --git a/apps/web/src/pages/shell/transcript.tsx b/apps/web/src/pages/shell/transcript.tsx index e595fba..abb4944 100644 --- a/apps/web/src/pages/shell/transcript.tsx +++ b/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; artifactTarget: ArtifactTarget; @@ -155,9 +156,11 @@ export const Transcript = memo(function Transcript({ encounterMode?: boolean; voiceEncounterActive?: boolean; onRestartTask?: (taskPrompt: string) => void | Promise; + onSendQueuedNow?: (message: ThreadMessage) => void | Promise; }) { const { t } = useLingui(); const [restartingPrompt, setRestartingPrompt] = useState(null); + const [sendingQueuedId, setSendingQueuedId] = useState(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) ? ( -

- {t`Queued`} -

+ {t`Queued`} + {onSendQueuedNow ? ( + + ) : null} + ) : null} {!peerReceipt && messageReactions ? (
( + messages: readonly T[], + runCreatedAt: string | undefined, + reactions: ReadonlyMap>, +): 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), + ); +}