diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index 024fc14179fd2..5441e3340c210 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -956,7 +956,7 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps ) : showResumeAction ? ( void handleResumeQueue()} @@ -1159,7 +1159,7 @@ export const ThreadComposer = function ThreadComposer(props: ThreadComposerProps ) : showResumeAction && voicePresentation.showsSend ? ( void handleResumeQueue()} diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 737e1aff90ada..5b2510802eb15 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -20,6 +20,7 @@ import { AuthTerminalReadScope, DEFAULT_SERVER_SETTINGS, EnvironmentId, + MessageId, ThreadId, type ProjectScript, } from "@t3tools/contracts"; @@ -54,6 +55,11 @@ import { } from "../../state/use-remote-environment-registry"; import { useKnownTerminalSessions } from "../../state/use-terminal-session"; import { uuidv4 } from "../../lib/uuid"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, + type AtomCommandResult, +} from "@t3tools/client-runtime/state/runtime"; import { readEnvironmentScope, useEnvironmentScope } from "../../state/session"; import { useSelectedThreadDetailState } from "../../state/use-thread-detail"; import { useThreadSelection } from "../../state/use-thread-selection"; @@ -385,7 +391,22 @@ function ThreadRouteContent( const gitActions = useSelectedThreadGitActions(); const requests = useSelectedThreadRequests(); const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, "thread interrupt"); - const resumeThreadQueue = useAtomCommand(threadEnvironment.resumeThreadQueue, "resume queue"); + const resumeThreadQueue = useAtomCommand(threadEnvironment.resumeThreadQueue, { + label: "resume queue", + reportFailure: false, + }); + const continueThreadRun = useAtomCommand(threadEnvironment.startTurn, { + label: "continue thread", + reportFailure: false, + }); + const setThreadRuntimeMode = useAtomCommand(threadEnvironment.setRuntimeMode, { + label: "set thread runtime mode", + reportFailure: false, + }); + const setThreadInteractionMode = useAtomCommand(threadEnvironment.setInteractionMode, { + label: "set thread interaction mode", + reportFailure: false, + }); const loadEarlierHistory = useAtomCommand(threadEnvironment.loadEarlierHistory, { label: "load earlier thread history", reportFailure: false, @@ -689,10 +710,71 @@ function ThreadRouteContent( const handleResumeQueue = async () => { if (!selectedThread) return; - await resumeThreadQueue({ - environmentId: selectedThread.environmentId, - input: { threadId: selectedThread.id }, - }); + const failed = (result: AtomCommandResult) => { + if (result._tag !== "Failure") return false; + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + Dialog.alert( + "Could not resume thread", + error instanceof Error ? error.message : "Try again.", + ); + } + return true; + }; + // The server holds the queue behind a stopped or usage-limited run until that run is continued. + if (composer.resumableRunId !== null) { + const runtimeMode = composer.runtimeMode ?? selectedThread.runtimeMode; + const interactionMode = composer.interactionMode ?? selectedThread.interactionMode; + const createdAt = new Date().toISOString(); + // A turn on an existing thread runs with the thread's saved modes, not the turn's. + if ( + runtimeMode !== selectedThread.runtimeMode && + failed( + await setThreadRuntimeMode({ + environmentId: selectedThread.environmentId, + input: { threadId: selectedThread.id, runtimeMode, createdAt }, + }), + ) + ) { + return; + } + if ( + interactionMode !== selectedThread.interactionMode && + failed( + await setThreadInteractionMode({ + environmentId: selectedThread.environmentId, + input: { threadId: selectedThread.id, interactionMode, createdAt }, + }), + ) + ) { + return; + } + const continued = await continueThreadRun({ + environmentId: selectedThread.environmentId, + input: { + creationSource: "mobile", + threadId: selectedThread.id, + manualContinuationOfRunId: composer.resumableRunId, + ...(composer.modelSelection ? { modelSelection: composer.modelSelection } : {}), + message: { + messageId: MessageId.make(uuidv4()), + role: "user", + text: "Continue where you left off.", + attachments: [], + }, + runtimeMode, + interactionMode, + dispatchMode: "start", + }, + }); + if (failed(continued) || !composer.queueHeld) return; + } + failed( + await resumeThreadQueue({ + environmentId: selectedThread.environmentId, + input: { threadId: selectedThread.id }, + }), + ); }; const handleOpenTerminal = (nextTerminalId?: string | null) => { @@ -1077,7 +1159,9 @@ function ThreadRouteContent( onRemoveDraftImage={composer.onRemoveDraftImage} serverConfig={serverConfig} onStopThread={awaitingBootstrapTurn ? handleCancelWorktreeSetup : handleStopThread} - onResumeQueue={composer.queueHeld ? handleResumeQueue : null} + onResumeQueue={ + composer.queueHeld || composer.resumableRunId !== null ? handleResumeQueue : null + } onSendMessage={composer.onSendMessage} onReconnectEnvironment={handleReconnectEnvironment} canSwitchThreadProvider={composer.canSwitchThreadProvider} diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 4c14fb56254cf..cf7a1fa00067b 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -8,6 +8,7 @@ import type { ThreadCheckpointSummary } from "@t3tools/client-runtime/state/thre import { resolvePendingHandoff } from "@t3tools/client-runtime/handoff"; import { deriveProviderSubagentStatus, + deriveResumableRunId, deriveRunlessWorkStartedAt, deriveThreadActivityRun, deriveThreadRuntime, @@ -359,6 +360,9 @@ export function useThreadComposerState() { // A stop or a restart holds the queue, completion wakes included, until it // is resumed; the queue sheet lists only the user's own messages. const queueHeld = queueWorkflow?.isHeld === true; + const resumableRunId = selectedThreadProjection + ? deriveResumableRunId(selectedThreadProjection.projection, selectedThreadRuntime) + : null; const runlessWorkStartedAt = selectedThreadProjection ? deriveRunlessWorkStartedAt(selectedThreadProjection.projection) @@ -743,6 +747,7 @@ export function useThreadComposerState() { followUpBehavior, canSteerActiveTurn, queueHeld, + resumableRunId, queuedRunEdit, isSavingQueuedEdit, cancelQueuedRunEdit, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index cf48ddcb842dd..71ff92ade8daa 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -42,10 +42,6 @@ import { RightPanelSurfaceContent } from "./chat/RightPanelSurfaceContent"; import { usePendingThreadRequests } from "./chat/usePendingThreadRequests"; import { isScratchProject } from "@t3tools/client-runtime/state/projects"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; -import { - latestExecutedRun, - latestRootProviderFailure, -} from "@t3tools/shared/orchestrationV2ThreadError"; import type { UsageLimitSourceSnapshots } from "@t3tools/contracts"; import { feedbackBannerItem } from "./chat/ComposerFeedback"; import { useUsageLimitsPanel } from "./chat/useUsageLimitsPanel"; @@ -76,6 +72,7 @@ import { import { deriveProviderSubagentStatus, deriveReportedModelSelection, + deriveResumableRunId, formatModelSelectionEffort, deriveRunlessWorkStartedAt, deriveThreadActivityRun, @@ -814,16 +811,10 @@ export default function ChatView(props: ChatViewProps) { isServerThread && serverProjection?.runs.some((run) => run.status === "queued" && run.queueHeld === true) === true; - const resumableRunId = (() => { - if (!isServerThread || serverProjection === null) return null; - const run = latestExecutedRun(serverProjection.runs); - if (run?.status === "interrupted") return run.id; - return run?.status === "failed" && - serverRuntime?.lastErrorClass === "usage_limit" && - latestRootProviderFailure(run, serverProjection.turnItems)?.class === "usage_limit" - ? run.id - : null; - })(); + const resumableRunId = + !isServerThread || serverProjection === null + ? null + : deriveResumableRunId(serverProjection, serverRuntime); const parentSubagentThreadId = activeThread?.lineage.relationshipToParent === "subagent" ? activeThread.lineage.parentThreadId diff --git a/docs/user/composer.md b/docs/user/composer.md index 714572dfd5005..39cf75dd1cbe4 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -211,8 +211,8 @@ images. **Send now** sends a message right away, into the running turn if there also edit or remove it. If you stop the agent or the server restarts, queued messages keep their order and are held. -Press **Resume** in an empty composer on web and desktop, or **Resume queue** on mobile, to -continue. Sending a new message also resumes the queue, unless it steers a turn that is already +Press **Resume** in an empty composer to continue. **Resume** also continues a stopped +run or one that hit a usage limit, even with nothing queued. Sending a new message also resumes the queue, unless it steers a turn that is already running. You can edit or remove held messages without starting them. The pencil on a queued message opens it in the composer for editing. The original message diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index fb5e12b08372b..0c2730f236121 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -237,8 +237,9 @@ from its context menu, or **Copy thread** from its long-press menu on mobile. **Limited** means the provider stopped on a usage or rate limit. The conversation keeps the provider's explanation. Retry after the limit resets, or switch to another provider instance. -On web and desktop, press **Resume** in an empty composer to continue a limited -or interrupted turn manually. +Press **Resume** in an empty composer to continue a limited or interrupted turn +manually. Do this when the provider reports no reset time, because nothing +continues the turn on its own. Queued messages stay saved while the limit blocks the thread. They run after the continuation finishes. If the queue was held by a restart, resume it then. diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index 3f8ef2a7f9d31..6718244bed258 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -22,6 +22,7 @@ import { deriveReportedModelSelection, deriveLatestThreadRun, deriveProviderSubagentStatus, + deriveResumableRunId, deriveRunlessWorkStartedAt, deriveThreadActivityRun, deriveThreadRuntime, @@ -82,6 +83,7 @@ describe("thread execution presentation", () => { lastError: "Plan limit reached", lastErrorClass: "usage_limit", }); + expect(deriveResumableRunId(projection, deriveThreadRuntime(projection))).toBe(failed.id); expect( deriveThreadRuntime({ ...projection, @@ -152,6 +154,7 @@ describe("thread execution presentation", () => { lastError: "Plan limit reached", lastErrorClass: "usage_limit", }); + expect(deriveResumableRunId(projection, deriveThreadRuntime(projection))).toBe(failed.id); const cancelledQueued = { ...run("cancelled-queued", 3, "cancelled"), startedAt: null, @@ -163,12 +166,15 @@ describe("thread execution presentation", () => { expect( deriveThreadRuntime({ ...projection, runs: [failed, queued, cancelledQueued] }), ).toMatchObject({ status: "failed", lastErrorClass: "usage_limit" }); - expect( - deriveThreadRuntime({ - ...projection, - turnItems: [{ ...item, failure: { ...item.failure, class: "provider_error" as const } }], - }), - ).toMatchObject({ status: "queued", lastErrorClass: null }); + const providerError = { + ...projection, + turnItems: [{ ...item, failure: { ...item.failure, class: "provider_error" as const } }], + }; + expect(deriveThreadRuntime(providerError)).toMatchObject({ + status: "queued", + lastErrorClass: null, + }); + expect(deriveResumableRunId(providerError, deriveThreadRuntime(providerError))).toBeNull(); }); it("keeps live activity attached to an executing run when a newer run is queued", () => { @@ -229,6 +235,7 @@ describe("thread execution presentation", () => { status: "interrupted", activeRunId: null, }); + expect(deriveResumableRunId(projection, deriveThreadRuntime(projection))).toBe(interrupted.id); const resumed = { ...projection, runs: [interrupted, { ...held, queueHeld: false }] }; expect(deriveThreadRuntime(resumed)).toMatchObject({ status: "queued" }); diff --git a/packages/client-runtime/src/state/threadExecution.ts b/packages/client-runtime/src/state/threadExecution.ts index 52d07babfbb61..2fbf428ea27e0 100644 --- a/packages/client-runtime/src/state/threadExecution.ts +++ b/packages/client-runtime/src/state/threadExecution.ts @@ -1,4 +1,5 @@ import { + latestExecutedRun, latestRootProviderFailure, latestUnheldRun, threadErrorSummary, @@ -16,6 +17,7 @@ import { type OrchestrationV2ExecutionNode, type OrchestrationV2ThreadProjection, orchestrationV2RunWorkStartedAt, + type RunId, type ThreadId, } from "@t3tools/contracts"; import { @@ -209,6 +211,20 @@ export function formatProviderSubagentStatus( return live ? `${label} ${elapsed}` : `${label} in ${elapsed}`; } +/** The run a manual continuation resumes; the server holds the queue until it does. */ +export function deriveResumableRunId( + projection: Pick, + runtime: Pick | null, +): RunId | null { + const run = latestExecutedRun(projection.runs); + if (run?.status === "interrupted") return run.id; + return run?.status === "failed" && + runtime?.lastErrorClass === "usage_limit" && + latestRootProviderFailure(run, projection.turnItems)?.class === "usage_limit" + ? run.id + : null; +} + export function deriveThreadRuntime( projection: ThreadStructure & Pick, ): ThreadRuntimeSummary | null {