diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index f3a8ab3..12de279 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -273,6 +273,7 @@ import { reactToThreadMessage, removeQueuedMessage, resolveThreadTarget, + resumeThreadRuns, sendQueuedThreadMessagesNow, sendThreadMessage, setThreadUnreadState, @@ -2064,6 +2065,16 @@ export function createRouter(deps: RouterDeps) { await stopThreadRuns(deps, context.actor, target); return { ok: true as const }; }), + resume: authed.threads.resume.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 resumeThreadRuns(deps, context.actor, target); + }), 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." }); diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index 326848a..1ee5174 100644 --- a/apps/api/src/thread-target.test.ts +++ b/apps/api/src/thread-target.test.ts @@ -2,16 +2,17 @@ import type { SandboxProvider } from "@rakazo/adapter-kit"; import type { Actor } from "@rakazo/contracts"; import type { PrismaClient } from "@rakazo/db"; import { describe, expect, it, vi } from "vitest"; +import type { ThreadTarget } from "./thread-target.js"; import { cancelSupersededQueuedRuns, followUpGroupThreadMessage, groupDefaultBotId, reactToThreadMessage, resolveGroupFollowUpBotId, + resumeThreadRuns, sendQueuedThreadMessagesNow, sendThreadMessage, stopThreadRuns, - type ThreadTarget, threadHead, threadSnapshot, } from "./thread-target.js"; @@ -2228,3 +2229,83 @@ describe("stopThreadRuns", () => { expect(events.notify).toHaveBeenCalledWith("thread-1", 42); }); }); + +describe("resumeThreadRuns", () => { + it("re-queues the latest cancelled or failed run without creating a duplicate message", async () => { + const transaction = { + $queryRaw: vi.fn().mockResolvedValue([{ id: "thread-1" }]), + run: { + findMany: vi + .fn() + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([ + { + id: "run-cancelled", + taskId: "task-cancelled", + botId: "bot-a", + status: "cancelled", + createdAt: new Date("2026-04-20T10:00:00.000Z"), + completedAt: new Date("2026-04-20T10:02:00.000Z"), + }, + ]), + update: vi.fn().mockResolvedValue({}), + }, + task: { + updateMany: vi.fn().mockResolvedValue({ count: 1 }), + }, + event: { + findFirst: vi.fn().mockResolvedValue({ seq: 19 }), + }, + }; + const prisma = { + $transaction: vi.fn(async (fn: (tx: typeof transaction) => Promise) => + fn(transaction), + ), + } as unknown as PrismaClient; + const jobs = { + enqueue: 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 resumeThreadRuns( + { + prisma, + events: events as never, + jobs: jobs as never, + }, + actor, + target, + ); + + expect(result).toEqual({ ok: true, runIds: ["run-cancelled"] }); + expect(transaction.task.updateMany).toHaveBeenCalledWith({ + where: { id: "task-cancelled" }, + data: { status: "queued" }, + }); + expect(transaction.run.update).toHaveBeenCalledWith({ + where: { id: "run-cancelled" }, + data: { + status: "queued", + error: null, + completedAt: null, + leaseOwner: null, + leaseExpiresAt: null, + }, + }); + expect(jobs.enqueue).toHaveBeenCalledWith({ + name: "run.continue", + payload: { runId: "run-cancelled" }, + replaceKey: "run:run-cancelled", + }); + expect(events.notify).toHaveBeenCalledWith("thread-1", 19); + }); +}); diff --git a/apps/api/src/thread-target.ts b/apps/api/src/thread-target.ts index 3d6fe4e..45ed32b 100644 --- a/apps/api/src/thread-target.ts +++ b/apps/api/src/thread-target.ts @@ -1307,6 +1307,106 @@ export async function stopThreadRuns( await teardownCancelledRuns(deps, actor, snapshot); } +/** + * Resume the latest stopped (`cancelled`) or interrupted (`failed`) run in the thread + * without creating a duplicate user message row. + */ +export async function resumeThreadRuns( + deps: { + prisma: PrismaClient; + events: ThreadEvents; + jobs: JobPublisher; + }, + actor: Actor, + target: ThreadTarget, +) { + 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 activeRuns = await tx.run.findMany({ + where: { + threadId: target.threadId, + botId: { in: botIds }, + status: { in: [...ACTIVE_RUN_STATUSES] }, + }, + select: { id: true, status: true }, + }); + if (activeRuns.length > 0) { + return { + runs: activeRuns, + eventSeq: null as number | null, + }; + } + + const recentTerminals = await tx.run.findMany({ + where: { + threadId: target.threadId, + botId: { in: botIds }, + trigger: { not: "bot_message" }, + status: { in: ["failed", "completed", "cancelled"] }, + }, + orderBy: [{ updatedAt: "desc" }, { id: "desc" }], + take: 20, + select: { + id: true, + botId: true, + taskId: true, + status: true, + createdAt: true, + completedAt: true, + }, + }); + const latestTerminal = pickLatestTerminalRun(recentTerminals); + if ( + !latestTerminal || + (latestTerminal.status !== "cancelled" && latestTerminal.status !== "failed") + ) { + return { + runs: [] as Array<{ id: string; status: string }>, + eventSeq: null as number | null, + }; + } + + await tx.task.updateMany({ + where: { id: latestTerminal.taskId }, + data: { status: "queued" }, + }); + await tx.run.update({ + where: { id: latestTerminal.id }, + data: { + status: "queued", + error: null, + completedAt: null, + leaseOwner: null, + leaseExpiresAt: null, + }, + }); + 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 { + runs: [{ id: latestTerminal.id, status: "queued" }], + eventSeq: latestEvent?.seq ?? null, + }; + }); + + if (committed.eventSeq !== null) { + await deps.events.notify(target.threadId, committed.eventSeq).catch((error) => { + getLogger().error("thread resume realtime notification", error); + }); + } + await enqueueRunsNeedingContinue(deps.jobs, committed.runs); + return { + ok: true as const, + runIds: committed.runs.map((run) => run.id), + }; +} + /** * Promote queued follow-up messages immediately: cancel any currently active runs * holding up the queue without discarding unclaimed steering messages, create a diff --git a/apps/api/src/voice.ts b/apps/api/src/voice.ts index b5e9a8c..55351a0 100644 --- a/apps/api/src/voice.ts +++ b/apps/api/src/voice.ts @@ -147,7 +147,11 @@ export async function persistVoiceCredential( let voiceId = input.voiceId?.trim() ?? ""; if (!voiceId) { const voices = await provider.listVoices(input.plaintext, voiceContext(actor, input.signal)); - voiceId = voices[0]?.id ?? ""; + const germanVoice = + input.provider === "kokoro" + ? voices.find((v) => v.id === "df_eva" || /^d[fm]_/i.test(v.id)) + : undefined; + voiceId = germanVoice?.id ?? voices[0]?.id ?? ""; } const stored = await deps.secrets.put(input.plaintext, voiceContext(actor, input.signal)); const cred = await withSerializableRetry(() => diff --git a/apps/mobile/app/thread.tsx b/apps/mobile/app/thread.tsx index 5d3533e..71ef8ac 100644 --- a/apps/mobile/app/thread.tsx +++ b/apps/mobile/app/thread.tsx @@ -336,6 +336,7 @@ function Thread() { const [replyTarget, setReplyTarget] = useState(null); const [attachmentNotice, setAttachmentNotice] = useState(null); const [sending, setSending] = useState(false); + const [stopped, setStopped] = useState(false); const [error, setError] = useState(null); const [loadingOlder, setLoadingOlder] = useState(false); const [markdownPreview, setMarkdownPreview] = useState( @@ -1134,6 +1135,7 @@ function Thread() { setReplyTarget(null); setAttachmentNotice(null); setError(null); + setStopped(false); }, [threadKey]); function updateDraft(value: string) { @@ -1228,6 +1230,7 @@ function Thread() { if (initialBotTarget) cancelFocusPrompt(initialBotTarget); }; setSending(true); + setStopped(false); setError(null); try { if (plan.shouldRunRoutines) { @@ -1337,12 +1340,16 @@ function Thread() { const targetGroupId = groupId; if ((!targetBotId && !targetGroupId) || sending) return; setSending(true); + setStopped(false); setError(null); try { await rpc( "threads/stop", targetGroupId ? { groupId: targetGroupId } : { botId: targetBotId! }, ); + if (isCurrentTarget(targetBotId, targetGroupId)) { + setStopped(true); + } } catch (err) { if (isCurrentTarget(targetBotId, targetGroupId)) { setError(err instanceof Error ? err.message : t("Failed to stop work")); @@ -1362,6 +1369,30 @@ function Thread() { } } + async function resume() { + const targetBotId = botId; + const targetGroupId = groupId; + if ((!targetBotId && !targetGroupId) || sending) return; + setSending(true); + setStopped(false); + setError(null); + try { + await rpc( + "threads/resume", + targetGroupId ? { groupId: targetGroupId } : { botId: targetBotId! }, + ); + if (isCurrentTarget(targetBotId, targetGroupId)) { + await refresh(); + } + } catch (err) { + if (isCurrentTarget(targetBotId, targetGroupId)) { + setError(err instanceof Error ? err.message : t("Could not continue")); + } + } finally { + setSending(false); + } + } + const [sendingNowId, setSendingNowId] = useState(null); async function sendQueuedNow(message: MobileMessage) { @@ -1925,7 +1956,70 @@ function Thread() { > {error ? {error} : null} {runError ? ( - {runError} + + {runError} + {!working ? ( + void resume()} + style={{ + borderRadius: 999, + borderWidth: 1, + borderColor: tokens.border, + backgroundColor: tokens.card, + paddingHorizontal: 12, + paddingVertical: 6, + opacity: sending ? 0.5 : 1, + }} + > + + {t("Resume")} + + + ) : null} + + ) : stopped && !working ? ( + + {t("Stopped")} + void resume()} + style={{ + borderRadius: 999, + borderWidth: 1, + borderColor: tokens.border, + backgroundColor: tokens.card, + paddingHorizontal: 12, + paddingVertical: 6, + opacity: sending ? 0.5 : 1, + }} + > + + {t("Resume")} + + + ) : null} {showPinnedPage ? ( diff --git a/apps/web/src/pages/Shell.tsx b/apps/web/src/pages/Shell.tsx index c142636..db4cb16 100644 --- a/apps/web/src/pages/Shell.tsx +++ b/apps/web/src/pages/Shell.tsx @@ -413,6 +413,7 @@ export function ShellPage() { const inputQueueRef = useRef(new OptimisticInputQueue()); const [optimisticMessages, setOptimisticMessages] = useState([]); const [sendError, setSendError] = useState(null); + const [stoppedThreadKey, setStoppedThreadKey] = useState(null); const [attachmentNotice, setAttachmentNotice] = useState(null); // Safety watchdog: clear a stuck send only when no run is active (local models can take longer). @@ -1245,15 +1246,29 @@ export function ShellPage() { "df_kerstin", ] as const; if (!botId) return "df_eva"; - let hash = 0; + const idx = botsRef.current.findIndex((b) => b.id === botId); + if (idx >= 0) { + return germanVoices[idx % germanVoices.length] ?? "df_eva"; + } + let hash = 2166136261; for (let i = 0; i < botId.length; i += 1) { - hash = (hash * 31 + botId.charCodeAt(i)) >>> 0; + hash ^= botId.charCodeAt(i); + hash = Math.imul(hash, 16777619) >>> 0; } return germanVoices[hash % germanVoices.length] ?? "df_eva"; }, [], ); + useEffect(() => { + const unsubSpeech = speaker.subscribe((state) => { + setSpeakingMessageId(state.status === "idle" ? null : (state.messageId ?? null)); + }); + return () => { + unsubSpeech(); + }; + }, []); + useEffect(() => { void rpc.voice .status() @@ -1277,27 +1292,18 @@ export function ShellPage() { setVoiceStatus(status); }) .catch(() => undefined); - const unsubSpeech = speaker.subscribe((state) => { - setSpeakingMessageId(state.status === "idle" ? null : (state.messageId ?? null)); - }); - return () => { - unsubSpeech(); - }; - }, []); + }, [currentSpaceId]); + const migratedBotIdsRef = useRef(new Set()); useEffect(() => { if (!bots.length) return; - const migratedKey = "rakazo:voice-de-default-v2"; - try { - if (localStorage.getItem(migratedKey) === "1") return; - localStorage.setItem(migratedKey, "1"); - } catch { - return; - } const staleBots = bots.filter( - (b) => b.autoSpeak === false || !b.voiceId || /^a[fm]_/i.test(b.voiceId), + (b) => + !migratedBotIdsRef.current.has(b.id) && + (b.autoSpeak === false || !b.voiceId || /^a[fm]_/i.test(b.voiceId)), ); if (!staleBots.length) return; + for (const b of staleBots) migratedBotIdsRef.current.add(b.id); void Promise.all( staleBots.map((b) => rpc.bots @@ -1325,7 +1331,13 @@ export function ShellPage() { if (!threadKey || !currentSnapshot) return; const lastBot = [...currentSnapshot.messages] .reverse() - .find((message) => message.role === "bot"); + .find( + (message) => + message.role === "bot" && + !message.id.startsWith("progress:") && + !message.id.startsWith("subagent:") && + Boolean(spokenSummaryFromBlocks(message.blocks)), + ); if (autoSpokenBotId.current !== threadKey) { autoSpokenBotId.current = threadKey; autoSpoken.current = lastBot?.id ?? null; @@ -1344,12 +1356,6 @@ export function ShellPage() { autoSpoken.current = lastBot?.id ?? null; return; } - if ( - currentSnapshot.run && - ["running", "queued", "leased"].includes(currentSnapshot.run.status) - ) { - return; - } if (!lastBot || lastBot.id === autoSpoken.current) return; const text = spokenSummaryFromBlocks(lastBot.blocks); if (!text) return; @@ -2592,6 +2598,7 @@ export function ShellPage() { hasAttachments: attachments.length > 0, }); if (plan.isNoOp) return; + setStoppedThreadKey(null); const reroutedToGroup = Boolean( plan.rerouteGroupId && plan.rerouteGroupId !== initialGroupTarget, ); @@ -2881,6 +2888,7 @@ export function ShellPage() { } // Stop has no terminal event; clear run UI before refresh races with in-flight gets. if (activeGroupId.current === groupTarget) { + setStoppedThreadKey(`group:${groupTarget}`); updateSnapshot((prev) => prev && prev.groupId === groupTarget ? clearActiveThreadRuns(prev) : prev, ); @@ -2902,6 +2910,7 @@ export function ShellPage() { // superseded in-flight refresh (older cursor) cannot leave Stop enabled / Take control // blocked while the API is already idle. if (activeBotId.current === botTarget) { + setStoppedThreadKey(`bot:${botTarget}`); updateSnapshot((prev) => !prev || (prev.botId !== botTarget && prev.botId) ? prev : clearActiveThreadRuns(prev), ); @@ -2915,6 +2924,45 @@ export function ShellPage() { setSending(false); } }, [sending, t]); + const resumeRun = useCallback( + async (fallbackPrompt?: string) => { + const botTarget = activeBotId.current; + const groupTarget = activeGroupId.current; + if (!botTarget && !groupTarget) return; + setSending(true); + setSendError(null); + setStoppedThreadKey(null); + const failedRunId = snapshotRef.current?.run?.id ?? null; + if (failedRunId) { + rememberSeenRunErrorId(failedRunId); + setDismissedRunErrorIds((current) => new Set(current).add(failedRunId)); + } + try { + const res = await rpc.threads.resume( + groupTarget ? { groupId: groupTarget } : { botId: botTarget! }, + ); + if (res.runIds.length === 0 && fallbackPrompt?.trim()) { + await sendMessage(fallbackPrompt); + return; + } + if (groupTarget && activeGroupId.current === groupTarget) { + await refreshGroupThreadRef.current(groupTarget).catch(() => undefined); + } else if (botTarget && activeBotId.current === botTarget) { + await refreshThreadRef.current(botTarget).catch(() => undefined); + } + } catch (error) { + const stillHere = groupTarget + ? activeGroupId.current === groupTarget + : activeBotId.current === botTarget; + if (stillHere) { + setSendError(error instanceof Error ? error.message : t`Could not continue`); + } + } finally { + setSending(false); + } + }, + [sendMessage, t], + ); const stopTeaching = useCallback(async () => { const id = activeBotId.current; if (!id || teachBusy) return; @@ -4333,7 +4381,7 @@ export function ShellPage() { voiceReady={Boolean(voiceStatus?.ready)} speakingMessageId={speakingMessageId} onSpeak={speakMessage} - onRestartTask={sendMessage} + onRestartTask={resumeRun} onSendQueuedNow={sendQueuedMessageNow} onRemoveQueuedMessage={removeQueuedMessage} /> @@ -4366,6 +4414,13 @@ export function ShellPage() { onRemoveAttachment={removeAttachment} onSend={sendMessage} onStop={stopRun} + stopped={ + !composerRunning && + stoppedThreadKey === + (inGroup ? `group:${groupId}` : `bot:${active?.id}`) + } + onResume={resumeRun} + onDismissStopped={() => setStoppedThreadKey(null)} onVoice={ active || inGroup ? () => { diff --git a/apps/web/src/pages/shell/bot-panel.tsx b/apps/web/src/pages/shell/bot-panel.tsx index 786b1f7..f5e7e64 100644 --- a/apps/web/src/pages/shell/bot-panel.tsx +++ b/apps/web/src/pages/shell/bot-panel.tsx @@ -20,6 +20,7 @@ import { ModelTierSchema, normalizeBotTags, } from "@rakazo/contracts"; +import { GERMAN_VOICE_PRESETS, germanVoiceForBot, voicesForLocale } from "@rakazo/core"; import { Button, Input, @@ -37,6 +38,7 @@ import { connectedModelOptions } from "../../lib/connected-models"; import { lazyWithRecovery } from "../../lib/lazy-recovery"; import { modelTierLabel } from "../../lib/model-tier-label"; import { rpc } from "../../lib/rpc"; +import { readUiLocale } from "../../lib/ui-locale"; import { BotCapabilitiesFields } from "./bot-capabilities"; const ScratchpadSection = lazyWithRecovery("ScratchpadSection", () => @@ -280,8 +282,8 @@ export function BotSettings({ const [computerMode, setComputerMode] = useState(bot.computerMode); const [memoryScope, setMemoryScope] = useState(bot.memoryScope); const [autoSpeak, setAutoSpeak] = useState(bot.autoSpeak); - const [voiceId, setVoiceId] = useState(bot.voiceId ?? ""); - const [voices, setVoices] = useState([]); + const [voiceId, setVoiceId] = useState(bot.voiceId || germanVoiceForBot(bot.id, bot.voiceId)); + const [voices, setVoices] = useState(GERMAN_VOICE_PRESETS); const [modelKey, setModelKey] = useState( bot.modelTier ? tierOptionKey(bot.modelTier) @@ -314,8 +316,15 @@ export function BotSettings({ useEffect(() => { void rpc.voice .voices({}) - .then(setVoices) - .catch(() => setVoices([])); + .then((listed) => { + const uiLocale = readUiLocale(); + const isGerman = uiLocale.toLowerCase().startsWith("de"); + const hasGerman = listed.some((v) => /^d[fm]_/i.test(v.id)); + const combined = isGerman && !hasGerman ? [...GERMAN_VOICE_PRESETS, ...listed] : listed; + const filtered = voicesForLocale(combined, uiLocale); + setVoices(filtered.length > 0 ? filtered : GERMAN_VOICE_PRESETS); + }) + .catch(() => setVoices(GERMAN_VOICE_PRESETS)); void rpc.bots .builtinTools({ botId: bot.id }) .then(setToolNames) @@ -675,7 +684,7 @@ export function BotSettings({ {t`Account default`} {voices.map((voice) => ( - {voice.label} + {voice.description ? `${voice.label} (${voice.description})` : voice.label} ))} diff --git a/apps/web/src/pages/shell/composer.tsx b/apps/web/src/pages/shell/composer.tsx index f6e1c7b..20f4c1e 100644 --- a/apps/web/src/pages/shell/composer.tsx +++ b/apps/web/src/pages/shell/composer.tsx @@ -31,6 +31,7 @@ import { Play, Plus, Puzzle, + RotateCcw, Settings, Square, X, @@ -83,6 +84,9 @@ export const Composer = memo(function Composer({ onConnectModel, onAutopilot, activePlan, + stopped = false, + onResume, + onDismissStopped, }: { draftKey?: string; running: boolean; @@ -116,6 +120,9 @@ export const Composer = memo(function Composer({ /** Hand the order to the autopilot; the composer menu offers it only when set. */ onAutopilot?: (goal: string, until?: string) => Promise; activePlan?: ComposerActivePlan | null; + stopped?: boolean; + onResume?: () => void | Promise; + onDismissStopped?: () => void; }) { const { t } = useLingui(); const initialDraft = useMemo(() => readComposerDraft(draftKey), [draftKey]); @@ -484,6 +491,18 @@ export const Composer = memo(function Composer({ className="mb-3 flex items-center gap-2 rounded-[14px] border border-destructive/40 bg-destructive/10 px-4 py-2 text-ui-body-sm text-destructive" > {sendError ?? runError} + {runError && !sendError && onResume ? ( + + ) : null} + ) : stopped && !running && onResume ? ( +
+ + Stopped + + + {onDismissStopped ? ( + + ) : null} +
) : null} {replyTarget ? (
) : null} - {activePlan ? : null} + {activePlan ? : null}
void | Promise; +}) { const { t } = useLingui(); const drawerId = useId(); const { block, steps, running, startedAt } = activePlan; @@ -1431,119 +1439,132 @@ export function ComposerTaskPill({ activePlan }: { activePlan: ComposerActivePla data-expanded={expanded ? "true" : "false"} className="mb-2.5 rounded-[22px] border border-border bg-card/95 px-3.5 pt-2 pb-2.5 shadow-md backdrop-blur-md transition-all duration-200" > - + + {!running && !isCompleted && onResume ? ( + + ) : null} +
{done ? (