diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 870ab31f00003..412a5010cd212 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -58,7 +58,11 @@ import { type ApprovalResponder, type ThreadUserInputQuestion, } from "@t3tools/client-runtime/state/thread-requests"; +import { useDeadlinePassed } from "../../lib/clock"; import { + BACKGROUND_WORK_STOP_RETRY_AFTER_MS, + type BackgroundWorkStopRequest, + isBackgroundWorkStopPending, presentPendingBackgroundWork, presentProviderGoal, } from "@t3tools/client-runtime/state/thread-execution"; @@ -353,15 +357,27 @@ const RECONNECT_ERROR_DELAY_MS = 5_000; // Anchors, the submitted message, follow state and pending scrolls belong to // one thread, so the body remounts when the thread changes. export function ThreadDetailScreen(props: ThreadDetailScreenProps) { + const [statusRequests, setStatusRequests] = useState< + Record + >({}); return ( ); } -function ThreadDetailBody(props: ThreadDetailScreenProps) { +function ThreadDetailBody( + props: ThreadDetailScreenProps & { + statusRequests: Record; + setStatusRequests: React.Dispatch< + React.SetStateAction> + >; + }, +) { const usesNativeWorkspaceColumns = useNativeWorkspaceColumnsSupported(); const { layout } = useAdaptiveWorkspaceLayout(); const deviceType = useDeviceType(); @@ -590,8 +606,13 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { ); } }; - const [statusRequest, setStatusRequest] = useState<{ threadKey: string } | null>(null); - const statusPending = statusRequest?.threadKey === selectedThreadKey; + const statusRequest = props.statusRequests[selectedThreadKey] ?? null; + const statusPending = isBackgroundWorkStopPending( + statusRequest, + selectedThreadKey, + props.selectedThread.pendingBackgroundTasks, + useDeadlinePassed(statusRequest?.retryAt ?? null), + ); const [lifecycleNow, setLifecycleNow] = useState(() => new Date().toISOString()); useFocusEffect( useCallback(() => { @@ -623,8 +644,15 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { }); const runStatusAction = async (action: "stop" | "wake" | "unsettle" | "dismiss") => { if (statusPending || !props.canOperateThread) return; - const request = { threadKey: selectedThreadKey }; - setStatusRequest(request); + const request: BackgroundWorkStopRequest = { + threadKey: selectedThreadKey, + taskIds: + action === "stop" + ? props.selectedThread.pendingBackgroundTasks.map((task) => task.taskId) + : [], + retryAt: null, + }; + props.setStatusRequests((current) => ({ ...current, [request.threadKey]: request })); const target = { environmentId: props.environmentId, input: { threadId: props.selectedThread.id }, @@ -658,8 +686,17 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { ...target, input: { ...target.input, visitedAt: wokeAt ?? lifecycleNow }, }); - // An accepted interrupt can leave tasks running. Release the action for retry. - setStatusRequest((current) => (current === request ? null : current)); + props.setStatusRequests((current) => + current[request.threadKey] !== request + ? current + : { + ...current, + [request.threadKey]: + action === "stop" && result._tag !== "Failure" + ? { ...request, retryAt: Date.now() + BACKGROUND_WORK_STOP_RETRY_AFTER_MS } + : undefined, + }, + ); if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { const error = squashAtomCommandFailure(result); Dialog.alert( diff --git a/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx b/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx index a0b10dd4c643f..6c02319d25a63 100644 --- a/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx +++ b/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx @@ -1,3 +1,4 @@ +import * as Cause from "effect/Cause"; import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; import { EnvironmentId, ThreadId } from "@t3tools/contracts"; import { expect, it, vi } from "vite-plus/test"; @@ -20,7 +21,7 @@ vi.mock(import("../../state/use-atom-command"), async (importOriginal) => ({ vi.mock(import("../../state/use-orchestration-command"), async (importOriginal) => ({ ...(await importOriginal()), - useOrchestrationCommand: (() => async () => ({ _tag: "Success", value: undefined })) as never, + useOrchestrationCommand: (() => (input: unknown) => state.switchRef(input)) as never, })); vi.mock(import("../../hooks/useThreadActions"), async (importOriginal) => ({ @@ -77,7 +78,7 @@ function RestoreBranchHarness(props: { serverConfig: null, runtimeMode: "full-access" as never, isWorking: false, - pendingBackgroundTasks: [], + pendingBackgroundTasks: [{ taskId: "background-command", kind: "command" }] as never, turnItems: [], activePendingApproval: null as never, activePendingUserInput: null as never, @@ -95,13 +96,16 @@ function RestoreBranchHarness(props: { persistThreadSettingsForNextTurn: async () => ({ _tag: "Success", value: undefined }) as never, }); return ( - + <> + {lifecycle.workNow?.actions} + + ); } @@ -181,3 +185,44 @@ it.each(["a", "b", "navigation"] as const)( expect(focusComposer).toHaveBeenCalledTimes(scenario === "navigation" ? 0 : 1); }, ); + +it.each(["Success", "Failure"] as const)( + "keeps background stop requests per thread after %s", + async (outcome) => { + setup(); + const completions: Array<(result: unknown) => void> = []; + state.switchRef = vi.fn(() => new Promise((resolve) => completions.push(resolve))); + const threadA = threadInput("thread:a", "feature-a"); + const threadB = threadInput("thread:b", "feature-b"); + const currentRouteThreadKeyRef = { current: threadA.routeThreadKey }; + const harness = (thread: ThreadInput) => ( + undefined} + focusComposer={() => undefined} + /> + ); + await using screen = await render(harness(threadA)); + await page.getByRole("button", { name: "Stop", exact: true }).click(); + await expect.element(page.getByRole("button", { name: "Stopping..." })).toBeDisabled(); + await screen.rerender(harness(threadB)); + await page.getByRole("button", { name: "Stop", exact: true }).click(); + await screen.rerender(harness(threadA)); + await expect.element(page.getByRole("button", { name: "Stopping..." })).toBeDisabled(); + completions[0]!( + outcome === "Success" + ? { _tag: "Success", value: undefined } + : { _tag: "Failure", cause: Cause.fail(new Error("stop failed")) }, + ); + if (outcome === "Success") { + await expect.element(page.getByRole("button", { name: "Stopping..." })).toBeDisabled(); + } else { + await expect.element(page.getByRole("button", { name: "Stop", exact: true })).toBeEnabled(); + } + await screen.rerender(harness(threadB)); + await expect.element(page.getByRole("button", { name: "Stopping..." })).toBeDisabled(); + expect(state.switchRef).toHaveBeenCalledTimes(2); + completions[1]!({ _tag: "Success", value: undefined }); + }, +); diff --git a/apps/web/src/components/chat/useThreadLifecycle.tsx b/apps/web/src/components/chat/useThreadLifecycle.tsx index 915f39a12a24e..6d71e3f59f05d 100644 --- a/apps/web/src/components/chat/useThreadLifecycle.tsx +++ b/apps/web/src/components/chat/useThreadLifecycle.tsx @@ -12,7 +12,7 @@ import { makeThreadErrorSetter, } from "../ChatView.logic"; import { useStableCallback } from "../../hooks/useStableCallback"; -import { useNowAfter } from "../../hooks/useDeadline"; +import { useDeadlinePassed, useNowAfter } from "../../hooks/useDeadline"; import type { usePendingThreadRequests } from "./usePendingThreadRequests"; import { AuthOrchestrationOperateScope, @@ -28,6 +28,11 @@ import { snoozeWakeLabel, threadWokeAt, } from "@t3tools/client-runtime/state/thread-settled"; +import { + BACKGROUND_WORK_STOP_RETRY_AFTER_MS, + type BackgroundWorkStopRequest, + isBackgroundWorkStopPending, +} from "@t3tools/client-runtime/state/thread-execution"; import { useAcknowledgeThreadWoke, useThreadActions } from "../../hooks/useThreadActions"; import { settleAndAdvance } from "../../settleAndAdvance"; import type { ChatShortcutCommands } from "./useChatShortcuts"; @@ -400,21 +405,44 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { // The interrupt path also accepts a completed run while its provider still // has background work. const activeBackgroundTasks = !isWorking && activeThread ? pendingBackgroundTasks : []; - const [stoppingBackgroundWorkKey, setStoppingBackgroundWorkKey] = useState(null); - const isStoppingBackgroundWork = - stoppingBackgroundWorkKey === `${environmentId}:${activeThreadId}`; + const [backgroundWorkStops, setBackgroundWorkStops] = useState< + Record + >({}); + const backgroundWorkStop = backgroundWorkStops[`${environmentId}:${activeThreadId}`] ?? null; + const isStoppingBackgroundWork = isBackgroundWorkStopPending( + backgroundWorkStop, + `${environmentId}:${activeThreadId}`, + pendingBackgroundTasks, + useDeadlinePassed(backgroundWorkStop?.retryAt ?? null), + ); const handleStopBackgroundWork = async () => { - if (!activeThread || !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) + if ( + isStoppingBackgroundWork || + !activeThread || + !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) + ) return; - const requestKey = `${environmentId}:${activeThread.id}`; - setStoppingBackgroundWorkKey(requestKey); + const request: BackgroundWorkStopRequest = { + threadKey: `${environmentId}:${activeThread.id}`, + taskIds: pendingBackgroundTasks.map((task) => task.taskId), + retryAt: null, + }; + setBackgroundWorkStops((current) => ({ ...current, [request.threadKey]: request })); const result = await interruptThreadTurn({ environmentId, input: { threadId: activeThread.id }, }); - // Acceptance does not confirm termination. Allow retry while the provider - // finishes stopping the tasks or reports a failure. - setStoppingBackgroundWorkKey((current) => (current === requestKey ? null : current)); + setBackgroundWorkStops((current) => + current[request.threadKey] !== request + ? current + : { + ...current, + [request.threadKey]: + result._tag === "Failure" + ? undefined + : { ...request, retryAt: Date.now() + BACKGROUND_WORK_STOP_RETRY_AFTER_MS }, + }, + ); if (result._tag === "Failure") { if (!isAtomCommandInterrupted(result)) { const error = squashAtomCommandFailure(result); diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index 9daa166e0b63d..ed60963ce7ef2 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -26,6 +26,7 @@ import { deriveRunlessWorkStartedAt, deriveThreadActivityRun, deriveThreadRuntime, + isBackgroundWorkStopPending, presentPendingBackgroundWork, presentProviderGoal, presentRunStop, @@ -472,6 +473,35 @@ describe("provider-reported model selection", () => { }); }); +describe("isBackgroundWorkStopPending", () => { + const tasks = [{ taskId: "dev-server" }]; + const sent = { threadKey: "env:thread", taskIds: ["dev-server"], retryAt: null }; + const accepted = { ...sent, retryAt: 10_000 }; + + it("stays pending after acceptance while a stopped task is still listed", () => { + expect(isBackgroundWorkStopPending(sent, "env:thread", tasks, false)).toBe(true); + expect(isBackgroundWorkStopPending(accepted, "env:thread", tasks, false)).toBe(true); + }); + + it("ends once the stopped tasks leave, even if new work starts", () => { + expect(isBackgroundWorkStopPending(accepted, "env:thread", [], false)).toBe(false); + expect(isBackgroundWorkStopPending(accepted, "env:thread", [{ taskId: "next" }], false)).toBe( + false, + ); + }); + + it("allows retry at the deadline even if a stopped task remains", () => { + const request = { ...accepted, retryAt: 10_000 }; + expect(isBackgroundWorkStopPending(request, "env:thread", tasks, false)).toBe(true); + expect(isBackgroundWorkStopPending(request, "env:thread", tasks, true)).toBe(false); + }); + + it("only applies to the thread that sent the Stop", () => { + expect(isBackgroundWorkStopPending(null, "env:thread", tasks, false)).toBe(false); + expect(isBackgroundWorkStopPending(sent, "env:other", tasks, false)).toBe(false); + }); +}); + describe("presentProviderGoal", () => { it("rolls token counts that round to 1000k over to 1m", () => { const usage = (tokensUsed: number) => diff --git a/packages/client-runtime/src/state/threadExecution.ts b/packages/client-runtime/src/state/threadExecution.ts index 3420c25de6cf2..361f44f717edb 100644 --- a/packages/client-runtime/src/state/threadExecution.ts +++ b/packages/client-runtime/src/state/threadExecution.ts @@ -378,6 +378,26 @@ function joinWithAnd(parts: ReadonlyArray): string { return `${parts.slice(0, -1).join(", ")} and ${parts.at(-1)}`; } +// The Claude adapter forces a stuck query closed after 10 seconds. +export const BACKGROUND_WORK_STOP_RETRY_AFTER_MS = 10_000; + +export interface BackgroundWorkStopRequest { + readonly threadKey: string; + readonly taskIds: ReadonlyArray; + readonly retryAt: number | null; +} + +/** Acceptance does not end the tasks, so an accepted Stop stays pending while they are listed. */ +export function isBackgroundWorkStopPending( + request: BackgroundWorkStopRequest | null, + threadKey: string, + tasks: ReadonlyArray>, + retryExpired: boolean, +): boolean { + if (request === null || request.threadKey !== threadKey || retryExpired) return false; + return request.retryAt === null || tasks.some((task) => request.taskIds.includes(task.taskId)); +} + export function presentPendingBackgroundWork( tasks: ReadonlyArray, ): PendingBackgroundWorkPresentation | null {