diff --git a/apps/mobile/src/connection/platform.ts b/apps/mobile/src/connection/platform.ts index 407016ab484d9..8ed62f641a339 100644 --- a/apps/mobile/src/connection/platform.ts +++ b/apps/mobile/src/connection/platform.ts @@ -28,6 +28,7 @@ import { appAtomRegistry } from "../state/atom-registry"; import { clearThreadOutboxEnvironment } from "../state/thread-outbox-removal"; import { clearComposerDraftsEnvironment } from "../state/use-composer-drafts"; import { clearThreadComposerErrorsForEnvironment } from "../state/thread-composer-error"; +import { mobileThreadTriageOutboxStore } from "../state/thread-triage-outbox-storage"; import { mobileApplicationActiveWakeup } from "./app-state-wakeups"; import * as ConnectionStorage from "./storage"; @@ -264,6 +265,7 @@ const layerEnvironmentOwnedDataCleanup = Layer.succeed( Effect.promise(() => clearThreadOutboxEnvironment(environmentId)), Effect.promise(() => clearComposerDraftsEnvironment(environmentId)), Effect.sync(() => clearThreadComposerErrorsForEnvironment(environmentId)), + Effect.ignore(mobileThreadTriageOutboxStore.clearEnvironment(environmentId)), ], { concurrency: "unbounded", discard: true }, ), diff --git a/apps/mobile/src/connection/storage.ts b/apps/mobile/src/connection/storage.ts index 91a6ac71919da..b412fadede0d9 100644 --- a/apps/mobile/src/connection/storage.ts +++ b/apps/mobile/src/connection/storage.ts @@ -7,6 +7,7 @@ import { removeCatalogValue, replaceCatalogValue, Persistence, + ThreadTriageOutboxStore, } from "@t3tools/client-runtime/platform"; import { TokenStore } from "@t3tools/client-runtime/authorization"; import { @@ -21,6 +22,7 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import { mobileThreadTriageOutboxStore } from "../state/thread-triage-outbox-storage"; import * as CatalogStore from "./catalog-store"; function targetPersistenceError( @@ -164,6 +166,7 @@ export const layer = Layer.effectContext( Context.add(ProfileStore.ConnectionProfileStore, profileStore), Context.add(CredentialStore.ConnectionCredentialStore, credentialStore), Context.add(TokenStore.RemoteDpopAccessTokenStore, remoteTokenStore), + Context.add(ThreadTriageOutboxStore, mobileThreadTriageOutboxStore), ); }), ); diff --git a/apps/mobile/src/features/home/useThreadListActions.test.ts b/apps/mobile/src/features/home/useThreadListActions.test.ts index 0bea39e419e21..89685cc4eb82b 100644 --- a/apps/mobile/src/features/home/useThreadListActions.test.ts +++ b/apps/mobile/src/features/home/useThreadListActions.test.ts @@ -22,6 +22,8 @@ const state = vi.hoisted(() => ({ titleOutcome: null as ThreadTitleRegenerationOutcome | null, archivedRefreshes: [] as string[], scopes: new Map>(), + sessionNotLoaded: new Set(), + successValues: new Map(), shells: [] as EnvironmentThreadShell[], requests: [] as { action: string; @@ -64,6 +66,9 @@ vi.mock("../archive/useArchivedThreadSnapshots", () => ({ vi.mock("../../state/session", () => ({ readEnvironmentScope: (environmentId: string, scope: string) => state.scopes.get(environmentId)?.has(scope) === true, + readThreadTriageScope: (environmentId: string) => + state.scopes.get(environmentId)?.has(AuthOrchestrationOperateScope) === true || + state.sessionNotLoaded.has(environmentId), })); vi.mock("../../state/presentation", () => ({ connectedEnvironmentIdsAtom: "connected-environments", @@ -152,13 +157,19 @@ vi.mock("../../state/threads", () => ({ }) => { state.requests.push({ action, ...request }); await state.heldActions.get(action); - if (!state.scopes.get(request.environmentId)?.has(AuthOrchestrationOperateScope)) { + // Without a loaded session the command itself queues settle and snooze. + if ( + !state.scopes.get(request.environmentId)?.has(AuthOrchestrationOperateScope) && + !state.sessionNotLoaded.has(request.environmentId) + ) { return AsyncResult.failure(Cause.fail(new Error("Thread operation denied"))); } if (action === state.failedAction) return AsyncResult.failure(Cause.fail(new Error("Server rejected action"))); state.afterRequest?.(); - return AsyncResult.success(action === "regenerateTitle" ? state.titleOutcome : undefined); + return AsyncResult.success( + action === "regenerateTitle" ? state.titleOutcome : state.successValues.get(action), + ); }, ]), ), @@ -199,6 +210,8 @@ beforeEach(() => { state.archivedRefreshes = []; state.pendingOrder = null; state.dropBusy = false; + state.sessionNotLoaded = new Set(); + state.successValues = new Map(); state.holdingRows = false; state.scopes = new Map([ [primaryEnvironmentId, new Set([AuthOrchestrationOperateScope])], @@ -250,6 +263,19 @@ describe("thread list operation permissions", () => { expect(state.requests).toEqual([]); }); + it.each(mutationCases)( + "%s runs before the session loads only for settle, snooze and their undo", + async (handler, action) => { + state.scopes.get(primaryEnvironmentId)!.clear(); + state.sessionNotLoaded.add(primaryEnvironmentId); + + await useThreadListActions()[handler](makeThread(), "2099-01-01T00:00:00.000Z"); + + const triage = ["settle", "unsettle", "snooze", "unsnooze"].includes(action); + expect(state.requests.map((request) => request.action)).toEqual(triage ? [action] : []); + }, + ); + it("rechecks permission after a retained delete confirmation", async () => { useThreadListActions().confirmDeleteThread(makeThread()); const confirm = state.dialogs[0]?.onConfirm; @@ -474,6 +500,22 @@ describe("thread action undo", () => { expect(readThreadUndoNotice()).toBeNull(); }); + // Whether the queued settle reached the server is known only when Undo runs. + it.each([ + ["cancels it before it syncs", { sequence: 0, cancelledPendingSync: true }, ["snooze"]], + ["finds it delivered", { sequence: 3 }, ["pin", "snooze"]], + ])("restores a queued settle only as far as needed when Undo %s", async (_, undone, restored) => { + state.successValues.set("unsettle", undone); + const thread = makeThread({ + pinnedAt: "2026-09-02T00:00:00.000Z", + pinOrderKey: "m", + snoozedUntil: "2099-01-01T09:00:00.000Z", + }); + expect(await useThreadListActions().settleThread(thread)).toBe(true); + await readThreadUndoNotice()!.undo(); + expect(state.requests.map(({ action }) => action)).toEqual(["settle", "unsettle", ...restored]); + }); + // regression: #21401 restore archive and thread action undo it("wakes a snoozed thread through the undo action", async () => { const thread = makeThread(); diff --git a/apps/mobile/src/features/home/useThreadListActions.ts b/apps/mobile/src/features/home/useThreadListActions.ts index 3cd180db6273f..f346e35bb30cd 100644 --- a/apps/mobile/src/features/home/useThreadListActions.ts +++ b/apps/mobile/src/features/home/useThreadListActions.ts @@ -16,7 +16,7 @@ import { pinOrderKeyBetween } from "@t3tools/client-runtime/state/thread-sort"; import { appAtomRegistry } from "../../state/atom-registry"; import { environmentServerConfigsAtom } from "../../state/server"; import { connectedEnvironmentIdsAtom } from "../../state/presentation"; -import { readEnvironmentScope } from "../../state/session"; +import { readEnvironmentScope, readThreadTriageScope } from "../../state/session"; import { environmentThreadShells, steadyNavigationThreadShellsAtom, @@ -82,8 +82,18 @@ function actionFailureTitle(action: ThreadListAction): string { return "Could not delete thread"; } -function checkThreadOperationPermission(thread: EnvironmentThreadShell, title: string): boolean { - if (readEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope)) return true; +/** Settle, snooze and their undo may queue before the environment's session loads. */ +function checkThreadOperationPermission( + thread: EnvironmentThreadShell, + title: string, + triage = false, +): boolean { + if ( + triage + ? readThreadTriageScope(thread.environmentId) + : readEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope) + ) + return true; Dialog.alert(title, "This connection cannot change threads."); return false; } @@ -115,8 +125,10 @@ function useThreadActionExecutor( action: ThreadListAction, thread: EnvironmentThreadShell, onStarted?: () => void, + onSucceeded?: (value: unknown) => void, ) => { - if (!checkThreadOperationPermission(thread, actionFailureTitle(action))) return false; + const triage = action === "settle" || action === "unsettle"; + if (!checkThreadOperationPermission(thread, actionFailureTitle(action), triage)) return false; const key = scopedThreadKey(thread.environmentId, thread.id); // Drop only a repeat of the latest pending action; Settle → Un-settle → Settle must all send. if (latestPendingActions.current.get(key)?.action === action) { @@ -180,6 +192,7 @@ function useThreadActionExecutor( if (action === "archive" || action === "unarchive" || action === "delete") { refreshArchivedThreadsForEnvironment(thread.environmentId); } + onSucceeded?.(result.value); onCompleted?.(action, thread); return true; })(); @@ -257,21 +270,27 @@ export function useThreadListActions(): { action, claim, undo: async () => { + // A local cancellation keeps pin placement, but may have replaced a queued snooze. + let cancelledQueued = false; try { await ThreadUndo.executeThreadUndo( plan, async (step) => { switch (step.action) { case "unsettle": - return unsettleThread(thread); + return ( + (await executeAction("unsettle", thread, undefined, (value) => { + cancelledQueued = ThreadUndo.cancelledPendingSync(value); + })) === true + ); case "unsnooze": return unsnoozeThread(thread); case "pin": - return pinThread(thread, step.orderKey); + return cancelledQueued || pinThread(thread, step.orderKey); case "snooze": { const failureTitle = action === "Unpinned" ? "Could not undo unpin" : "Could not undo settle"; - if (!checkThreadOperationPermission(thread, failureTitle)) return false; + if (!checkThreadOperationPermission(thread, failureTitle, true)) return false; const result = await snoozeMutation({ environmentId: thread.environmentId, input: { threadId: thread.id, snoozedUntil: step.snoozedUntil }, @@ -314,7 +333,7 @@ export function useThreadListActions(): { return succeeded; }; const snoozeThread = async (thread: EnvironmentThreadShell, snoozedUntil: string) => { - if (!checkThreadOperationPermission(thread, "Could not snooze thread")) return false; + if (!checkThreadOperationPermission(thread, "Could not snooze thread", true)) return false; const key = scopedThreadKey(thread.environmentId, thread.id); if (snoozeInFlightThreadKeys.current.has(key)) { return false; @@ -370,7 +389,7 @@ export function useThreadListActions(): { return succeeded; }; const unsnoozeThread = async (thread: EnvironmentThreadShell) => { - if (!checkThreadOperationPermission(thread, "Could not wake thread")) return false; + if (!checkThreadOperationPermission(thread, "Could not wake thread", true)) return false; const key = scopedThreadKey(thread.environmentId, thread.id); if (snoozeInFlightThreadKeys.current.has(key)) { return false; diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index cb9e5eb715dc9..daf375fb42113 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -163,6 +163,7 @@ import { appAtomRegistry } from "../../state/atom-registry"; import { threadStreamHaptics } from "./threadStreamHaptics"; import { useAtomCommand } from "../../state/use-atom-command"; import { orchestrationEnvironment } from "../../state/orchestration"; +import { useThreadTriageScope } from "../../state/session"; import { useDelayedStatus } from "../../lib/useDelayedStatus"; import type { PendingUserInputDraftAnswer } from "@t3tools/client-runtime/state/pending-user-input"; import type { RunStopPresentation } from "@t3tools/client-runtime/state/thread-execution"; @@ -594,6 +595,7 @@ function ThreadDetailBody( }); const unsnoozeThread = useAtomCommand(threadEnvironment.unsnooze, { reportFailure: false }); const unsettleThread = useAtomCommand(threadEnvironment.unsettle, { reportFailure: false }); + const canTriageThread = useThreadTriageScope(props.environmentId); const updateMetadata = useAtomCommand(threadEnvironment.updateMetadata, { reportFailure: false }); const visitThread = useAtomCommand(threadEnvironment.visit, { reportFailure: false }); const anyRunStop = useAtomValue( @@ -690,7 +692,9 @@ function ThreadDetailBody( latestRunCompletedAt: props.selectedThread.latestRun?.completedAt ?? null, }); const runStatusAction = async (action: "stop" | "wake" | "unsettle" | "dismiss") => { - if (statusPending || !props.canOperateThread) return; + const permitted = + action === "wake" || action === "unsettle" ? canTriageThread : props.canOperateThread; + if (statusPending || !permitted) return; const request: BackgroundWorkStopRequest = { threadKey: selectedThreadKey, taskIds: @@ -882,7 +886,9 @@ function ThreadDetailBody( : "Dismissing..." : actionLabel, accessibilityLabel: actionLabel, - disabled: !props.canOperateThread || statusPending, + disabled: + !(lifecycleSnoozed || settled ? canTriageThread : props.canOperateThread) || + statusPending, onPress: () => void runStatusAction(lifecycleSnoozed ? "wake" : settled ? "unsettle" : "dismiss"), }, diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index b6bf0c98e349c..3c8a73de2b69f 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -1,3 +1,4 @@ +import { useAtomValue } from "@effect/atom-react"; import * as Dialog from "../../components/ui/Dialog"; import type { ThreadRowProviderInstance } from "./thread-provider-instance"; import { @@ -55,7 +56,7 @@ import { cn } from "../../lib/cn"; import { copyTextWithHaptic } from "../../lib/copyTextWithHaptic"; import { copyThreadReference } from "../../lib/copyThreadReference"; import { useUniwindTheme } from "../../lib/useUniwindTheme"; -import { useEnvironmentScope } from "../../state/session"; +import { useEnvironmentScope, useThreadTriageScope } from "../../state/session"; import type { PendingNewTask } from "../../state/use-pending-new-tasks"; import { pendingDraftBranchNeedsRefs, @@ -287,6 +288,8 @@ type ThreadListV2ShelfHeaderProps = { readonly pane?: "screen" | "sidebar"; }; +const TRIAGE_MENU_ACTION_IDS = new Set(["settle", "snooze", "unsettle", "unsnooze"]); + const SHELF_LABEL = { working: "Working", snoozed: "Snoozed", settled: "Settled" } as const; function ThreadListV2ShelfHeader( @@ -781,9 +784,12 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { ? { ...workingLabel, label: "Goal" } : workingLabel) ?? (isUnread ? DONE_STATUS_LABEL : undefined); const recede = shouldRecedeThreadRow({ status, selected }); + const pendingSync = useAtomValue(threadEnvironment.pendingSyncAtom(thread.environmentId)).has( + thread.id, + ); + const timeLabel = pendingSync ? "Waiting to sync" : props.timeLabel; const isRegeneratingTitle = thread.titleRegeneration != null; - const rowAccessibilityLabel = `${thread.title}${providerNameLabel ? `, ${providerNameLabel}` : ""}${props.hasQueuedMessages ? ", messages queued to send" : ""}${isRegeneratingTitle ? ", regenerating title" : ""}`; - const timeLabel = props.timeLabel; + const rowAccessibilityLabel = `${thread.title}${providerNameLabel ? `, ${providerNameLabel}` : ""}${pendingSync ? ", waiting to sync" : ""}${props.hasQueuedMessages ? ", messages queued to send" : ""}${isRegeneratingTitle ? ", regenerating title" : ""}`; const visitThread = useAtomCommand(threadEnvironment.visit, { reportFailure: false }); const handleAcknowledgeWoke = useCallback(() => { @@ -805,6 +811,9 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { const handleDelete = useCallback(() => onDeleteThread(thread), [onDeleteThread, thread]); const handleRename = useCallback(() => onRenameThread(thread), [onRenameThread, thread]); const canOperateThread = useEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope); + // Before the session loads, only settle, snooze and their undo are offered: they can queue. + const triageOnly = useThreadTriageScope(thread.environmentId) && !canOperateThread; + const triageAllows = (action: string) => TRIAGE_MENU_ACTION_IDS.has(action); const handleRegenerateTitle = useCallback( () => onRegenerateThreadTitle(thread), [onRegenerateThreadTitle, thread], @@ -1154,11 +1163,12 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { : null, [handleMenuAction, snoozePresetActions, swipeActions.secondary, thread.title], ); - const swipeAccessibilityHint = !canOperateThread - ? "Opens the thread" - : secondaryAction === null - ? `Opens the thread. Swipe left to ${primaryAction.label.toLowerCase()}.` - : `Opens the thread. Swipe left for ${primaryAction.label.toLowerCase()} and snooze actions.`; + const swipeAccessibilityHint = + !canOperateThread && !(triageOnly && triageAllows(swipeActions.primary)) + ? "Opens the thread" + : secondaryAction === null + ? `Opens the thread. Swipe left to ${primaryAction.label.toLowerCase()}.` + : `Opens the thread. Swipe left for ${primaryAction.label.toLowerCase()} and snooze actions.`; const titleTextInset = props.project ? projectIconSize + 8 : 0; const cardContent = ( @@ -1544,7 +1554,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { )} style={{ fontFamily: MONO_FONT }} > - {snoozedRow && props.snoozeWakeLabelText !== undefined + {!pendingSync && snoozedRow && props.snoozeWakeLabelText !== undefined ? props.snoozeWakeLabelText : timeLabel} @@ -1552,9 +1562,8 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { ); - if (!canOperateThread) + if (!canOperateThread && !(triageOnly && triageAllows(swipeActions.primary))) return {rowContent(() => {})}; - return ( {customSnoozeOpen && ( @@ -1579,7 +1588,15 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { threadTitle={thread.title} > {(close) => ( - + triageAllows(action.id ?? "")) + : menuActions + } + onPressAction={handleMenuAction} + shouldOpenOnLongPress + > {rowContent(close)} )} diff --git a/apps/mobile/src/state/session.test.ts b/apps/mobile/src/state/session.test.ts index b47054e258663..5579da2bcc9ac 100644 --- a/apps/mobile/src/state/session.test.ts +++ b/apps/mobile/src/state/session.test.ts @@ -5,6 +5,7 @@ import { type AuthSessionState, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Option from "effect/Option"; import { AsyncResult, Atom } from "effect/reactivity"; import { afterEach, beforeEach, expect, it, vi } from "vite-plus/test"; @@ -33,6 +34,7 @@ import { appAtomRegistry } from "./atom-registry"; import { environmentSession, readEnvironmentScope, + readThreadTriageScope, useEnvironmentScope, useEnvironmentsWithScope, } from "./session"; @@ -100,6 +102,22 @@ it("reacts to grant revocation, failure, and regrant without a connection list c expect(changes).not.toHaveLength(0); }); +it("lets settle and snooze queue until a loaded session denies them, also after a failed refresh", () => { + const failedAfter = (canOperate: boolean) => + AsyncResult.failure(Cause.fail(new Error("Session lookup failed")), { + previousSuccess: Option.some(AsyncResult.success(session(canOperate))), + }); + expect(readThreadTriageScope(secondary)).toBe(true); + appAtomRegistry.set(source(secondary), AsyncResult.failure(Cause.fail(new Error("offline")))); + expect(readThreadTriageScope(secondary)).toBe(true); + appAtomRegistry.set(source(secondary), AsyncResult.success(session(false))); + expect(readThreadTriageScope(secondary)).toBe(false); + appAtomRegistry.set(source(secondary), failedAfter(false)); + expect(readThreadTriageScope(secondary)).toBe(false); + appAtomRegistry.set(source(secondary), failedAfter(true)); + expect(readThreadTriageScope(secondary)).toBe(true); +}); + it("uses exact new-server permissions and keeps old-server grants usable", () => { appAtomRegistry.set(source(primary), AsyncResult.success({ ...session(true), permissions: [] })); expect(useEnvironmentScope(primary, AuthOrchestrationOperateScope)).toBe(false); diff --git a/apps/mobile/src/state/session.ts b/apps/mobile/src/state/session.ts index 5d6a441e9b9e4..48b9ace054dd8 100644 --- a/apps/mobile/src/state/session.ts +++ b/apps/mobile/src/state/session.ts @@ -1,6 +1,7 @@ import { useAtomValue } from "@effect/atom-react"; import { createEnvironmentSessionAtoms } from "@t3tools/client-runtime/state/session"; import { + AuthOrchestrationOperateScope, type AuthEnvironmentScope, type AuthSessionState, type EnvironmentId, @@ -66,6 +67,33 @@ export function readEnvironmentScope( return sessionHasScope(result, scope); } +/** + * Settle and snooze can queue for an environment whose session has not loaded yet, such as one + * that was offline at launch. The server checks the grant when the change is sent. + */ +function sessionAllowsThreadTriage( + result: AsyncResult.AsyncResult, +): boolean { + // A failed refresh still carries the last loaded session, whose answer stands. + const session = AsyncResult.getOrElse(result, () => null); + return session === null || sessionGrantsScope(session, AuthOrchestrationOperateScope); +} + +export function useThreadTriageScope(environmentId: EnvironmentId | null): boolean { + return useAtomValue( + environmentId === null + ? EMPTY_SESSION_STATE_ATOM + : environmentSession.sessionStateAtom(environmentId), + sessionAllowsThreadTriage, + ); +} + +export function readThreadTriageScope(environmentId: EnvironmentId): boolean { + return sessionAllowsThreadTriage( + appAtomRegistry.get(environmentSession.sessionStateAtom(environmentId)), + ); +} + const EMPTY_PREPARED_CONNECTION_ATOM = Atom.make(Option.none()).pipe( Atom.withLabel("mobile-prepared-connection:empty"), ); diff --git a/apps/mobile/src/state/thread-triage-outbox-storage.ts b/apps/mobile/src/state/thread-triage-outbox-storage.ts new file mode 100644 index 0000000000000..26b531518a9d2 --- /dev/null +++ b/apps/mobile/src/state/thread-triage-outbox-storage.ts @@ -0,0 +1,20 @@ +import { + makeThreadTriageOutboxStore, + ThreadTriageOutboxStorageError, +} from "@t3tools/client-runtime/platform"; +import * as Effect from "effect/Effect"; + +import { writeFileAtomically } from "../lib/atomic-file"; + +/** A file outside the client cache, so clearing cached data keeps unsynced settles and snoozes. */ +export const mobileThreadTriageOutboxStore = makeThreadTriageOutboxStore((next) => + Effect.tryPromise({ + try: async () => { + const { File, Paths } = await import("expo-file-system"); + const file = new File(Paths.document, "thread-triage-outbox.json"); + const value = next(file.exists ? await file.text() : null); + if (value !== null) await writeFileAtomically(file, value); + }, + catch: (cause) => new ThreadTriageOutboxStorageError({ cause }), + }), +); diff --git a/apps/mobile/src/state/threads.ts b/apps/mobile/src/state/threads.ts index bf2013aef5398..a0acea9187249 100644 --- a/apps/mobile/src/state/threads.ts +++ b/apps/mobile/src/state/threads.ts @@ -37,6 +37,7 @@ export const olderSettledThreads = createOlderSettledThreadAtoms( export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, threadFamilies.snapshotAtom, + environmentShell.statusAtom, ); export const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index ec9cde4684b9c..81557875f9e72 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -154,7 +154,7 @@ import { threadEnvironment } from "../state/threads"; import { useOlderSettledShelf, useOlderSettledShelfGrowth } from "../state/olderSettledShelf"; import { useThreadSearch } from "../state/queries"; import { useOrchestrationCommand } from "../state/use-orchestration-command"; -import { readEnvironmentScope, useEnvironmentScope } from "../state/session"; +import { readEnvironmentScope, readThreadTriageScope, useEnvironmentScope } from "../state/session"; import { appAtomRegistry } from "../rpc/atomRegistry"; import { canShareWithGuestAtom } from "./chat/guestShareAvailability"; import { openGuestShareDialog } from "./chat/guestShareDialogState"; @@ -171,6 +171,7 @@ import { ProjectRowActions } from "./ProjectComboboxRow"; import { buildDraftActionMenuItems, buildThreadActionMenuItems, + isThreadTriageAction, threadActionRequiresOperate, } from "./threadActionMenu.logic"; import { @@ -859,18 +860,23 @@ function ProjectScopeRowIcon(props: { isSelected: boolean; children: ReactNode } ); } +/** Settle, snooze and their undo can queue before the environment's session loads. */ function canOperateThreads( threads: ReadonlyArray>, + triage = false, ): boolean { return threads.every((thread) => - readEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope), + triage + ? readThreadTriageScope(thread.environmentId) + : readEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope), ); } function checkThreadOperations( threads: ReadonlyArray>, + triage = false, ): boolean { - if (canOperateThreads(threads)) return true; + if (canOperateThreads(threads, triage)) return true; toastManager.add({ type: "error", title: "Thread action unavailable", @@ -2337,26 +2343,26 @@ export default function Sidebar({ routeReveal }: { routeReveal: SidebarRouteReve { id: "settle", label: `Settle (${count})`, - disabled: settlingThreads.length === 0 || !canOperateThreads(settlingThreads), + disabled: settlingThreads.length === 0 || !canOperateThreads(settlingThreads, true), }, ...(canSnoozeSelection ? [ { id: "snooze", label: `Snooze (${count})`, - disabled: !canOperateThreads(selectedThreads), + disabled: !canOperateThreads(selectedThreads, true), children: [ ...snoozePresets.map((preset) => ({ id: `snooze:${preset.id}`, label: preset.label, detail: preset.whenLabel, - disabled: !canOperateThreads(selectedThreads), + disabled: !canOperateThreads(selectedThreads, true), })), { id: "snooze:custom", label: "Custom…", separatorBefore: snoozePresets.length > 0, - disabled: !canOperateThreads(selectedThreads), + disabled: !canOperateThreads(selectedThreads, true), }, ], }, @@ -2398,7 +2404,11 @@ export default function Sidebar({ routeReveal }: { routeReveal: SidebarRouteReve }) : selectedThreads; if (clicked.value === "settle" && actionTargets.length === 0) return; - if (clicked.value !== "mark-unread" && !checkThreadOperations(actionTargets)) return; + if ( + clicked.value !== "mark-unread" && + !checkThreadOperations(actionTargets, isThreadTriageAction(clicked.value)) + ) + return; if (clicked.value?.startsWith("snooze:")) { const preset = clicked.value === "snooze:custom" @@ -2629,6 +2639,7 @@ export default function Sidebar({ routeReveal }: { routeReveal: SidebarRouteReve threadRef.environmentId, AuthOrchestrationOperateScope, ), + canTriage: readThreadTriageScope(threadRef.environmentId), branch: thread.branch ?? null, // An archived project is not a filter choice. projectFilter: @@ -2662,7 +2673,11 @@ export default function Sidebar({ routeReveal }: { routeReveal: SidebarRouteReve ), ); if (clicked._tag === "Failure" || clicked.value === null) return; - if (threadActionRequiresOperate(clicked.value) && !checkThreadOperations([thread])) return; + if ( + threadActionRequiresOperate(clicked.value) && + !checkThreadOperations([thread], isThreadTriageAction(clicked.value)) + ) + return; if (clicked.value?.startsWith("snooze:")) { const preset = clicked.value === "snooze:custom" diff --git a/apps/web/src/components/chat/useThreadLifecycle.tsx b/apps/web/src/components/chat/useThreadLifecycle.tsx index 66b2a9f6b3fd3..99f0e38ba2d3f 100644 --- a/apps/web/src/components/chat/useThreadLifecycle.tsx +++ b/apps/web/src/components/chat/useThreadLifecycle.tsx @@ -53,7 +53,7 @@ import { titleRegenerationToast } from "../threadTitleRegenerationToast"; import { newMessageId } from "~/lib/utils"; import { buildThreadRouteParams } from "../../threadRoutes"; import { composerDraftHasUserContent, useComposerDraftStore } from "../../composerDraftStore"; -import { useEnvironmentScope } from "~/state/session"; +import { useEnvironmentScope, useThreadTriageScope } from "~/state/session"; import { threadEnvironment } from "../../state/threads"; import { vcsEnvironment } from "../../state/vcs"; import { useChildThreadInputs, readThreadShell, waitForThreadShell } from "../../state/entities"; @@ -62,7 +62,7 @@ import { environmentShell } from "../../state/shell"; import { ThreadStatusLine } from "./ThreadStatusLine"; import { formatRelativeTimeLabel } from "../../timestampFormat"; import { appAtomRegistry } from "../../rpc/atomRegistry"; -import { readEnvironmentScope } from "../../state/session"; +import { readEnvironmentScope, readThreadTriageScope } from "../../state/session"; import { useAtomCommand } from "../../state/use-atom-command"; import { useOrchestrationCommand } from "../../state/use-orchestration-command"; import { @@ -183,6 +183,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { const router = useRouter(); const { settleThread, pinThread, confirmAndUnpinThread } = useThreadActions(); const canWriteSourceControl = useEnvironmentScope(environmentId, AuthSourceControlWriteScope); + const canTriageThread = useThreadTriageScope(environmentId); const switchGitRef = useAtomCommand(vcsEnvironment.switchRef, { reportFailure: false }); const updateThreadMetadata = useOrchestrationCommand(threadEnvironment.updateMetadata, { reportFailure: false, @@ -233,17 +234,15 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { }); const activeThreadSettled = serverConfigLoaded && activeThreadShell?.settledOverride === "settled"; - const unsettleThreadMutation = useOrchestrationCommand(threadEnvironment.unsettle, { - reportFailure: false, - }); + const unsettleThreadMutation = useOrchestrationCommand( + threadEnvironment.unsettle, + { reportFailure: false }, + readThreadTriageScope, + ); const [unsettlingThreadKey, setUnsettlingThreadKey] = useState(null); const isUnsettling = unsettlingThreadKey !== null && unsettlingThreadKey === activeThreadKey; const handleUnsettleActiveThread = async () => { - if ( - !activeThreadRef || - !readEnvironmentScope(activeThreadRef.environmentId, AuthOrchestrationOperateScope) - ) - return; + if (!activeThreadRef || !readThreadTriageScope(activeThreadRef.environmentId)) return; const threadKey = scopedThreadKey(activeThreadRef); setUnsettlingThreadKey(threadKey); // Commands settle into a result, so this always runs. @@ -264,17 +263,15 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { setUnsettlingThreadKey((current) => (current === threadKey ? null : current)); }; const handleUnsettleActiveThreadStable = useStableCallback(handleUnsettleActiveThread); - const unsnoozeThreadMutation = useOrchestrationCommand(threadEnvironment.unsnooze, { - reportFailure: false, - }); + const unsnoozeThreadMutation = useOrchestrationCommand( + threadEnvironment.unsnooze, + { reportFailure: false }, + readThreadTriageScope, + ); const [unsnoozingThreadKey, setUnsnoozingThreadKey] = useState(null); const isUnsnoozing = unsnoozingThreadKey !== null && unsnoozingThreadKey === activeThreadKey; const handleUnsnoozeActiveThread = async () => { - if ( - !activeThreadRef || - !readEnvironmentScope(activeThreadRef.environmentId, AuthOrchestrationOperateScope) - ) - return; + if (!activeThreadRef || !readThreadTriageScope(activeThreadRef.environmentId)) return; const threadKey = scopedThreadKey(activeThreadRef); setUnsnoozingThreadKey(threadKey); // Commands settle into a result, so this always runs. @@ -626,7 +623,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { label={snoozeWake && snoozeWake !== "now" ? `Snoozed, ${snoozeWake} left` : "Snoozed"} clockMinute={nowMinute} actionLabel={isUnsnoozing ? "Waking..." : "Wake now"} - actionDisabled={!canOperateThread || isUnsnoozing} + actionDisabled={!canTriageThread || isUnsnoozing} onAction={() => void handleUnsnoozeActiveThreadStable()} /> ) @@ -636,7 +633,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { label={settledAt ? `Settled ${formatRelativeTimeLabel(settledAt)}` : "Settled"} clockMinute={nowMinute} actionLabel={isUnsettling ? "Un-settling..." : "Un-settle"} - actionDisabled={!canOperateThread || isUnsettling} + actionDisabled={!canTriageThread || isUnsettling} onAction={() => void handleUnsettleActiveThreadStable()} /> ) : activeThreadWokeVisible ? ( @@ -661,7 +658,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { }; const shortcuts = { "thread.settle": () => { - if (!readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return false; + if (!readThreadTriageScope(environmentId)) return false; if (!activeThreadShell || !activeThreadRef || !serverConfigLoaded) return; if (activeThreadSettled) { void handleUnsettleActiveThread(); diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 216f6635f7088..4aab44acaec7b 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -1,4 +1,4 @@ -import { useAtomMount } from "@effect/atom-react"; +import { useAtomMount, useAtomValue } from "@effect/atom-react"; import { asLaunchingThread } from "@t3tools/client-runtime/state/thread-inbox"; import { ThreadHoverCard, ThreadHoverCardPopup } from "../ThreadHoverCard"; import { ThreadTerminalStatusButton } from "../ThreadStatusIndicators"; @@ -74,7 +74,12 @@ import { useNowMinute } from "../../hooks/useNowMinute"; import { useNowSecond } from "../../hooks/useNowSecond"; import { vcsEnvironment, vcsBranchRevalidation } from "../../state/vcs"; import { useEnvironmentQuery } from "../../state/query"; -import { readEnvironmentScope, useEnvironmentScope } from "../../state/session"; +import { threadEnvironment } from "../../state/threads"; +import { + readEnvironmentScope, + useEnvironmentScope, + useThreadTriageScope, +} from "../../state/session"; import { formatRelativeTimeLabel, parseTimestampDate } from "../../timestampFormat"; import type { SidebarThreadSummary } from "../../types"; import type { EnvironmentProject } from "@t3tools/client-runtime/state/shell"; @@ -142,6 +147,15 @@ function compactSidebarTimeLabel(label: string): string { return label.endsWith(" ago") ? label.slice(0, -4) : label; } +function PendingSyncLabel() { + return ( + + }>Pending + Waiting to sync with its environment + + ); +} + // Subscribes to the minute tick itself, so only the label re-renders each // minute. The minute is passed as the clock so it is the formatter's input. function RelativeTimeLabel(props: { timestamp: string | null }) { @@ -669,6 +683,10 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP const threadRef = scopeThreadRef(thread.environmentId, thread.id); const threadKey = scopedThreadKey(threadRef); const canOperateThread = useEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope); + const canTriageThread = useThreadTriageScope(thread.environmentId); + const pendingSync = useAtomValue(threadEnvironment.pendingSyncAtom(thread.environmentId)).has( + thread.id, + ); const { leaseLiveStatus, rowRef: leaseRowRef } = useSidebarRowSubscriptionLease(props.isActive); const { revealRef } = props; const rowRef = (row: HTMLElement | null) => { @@ -957,7 +975,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP setSnoozeGateClock({ thread, now: new Date(snoozeGateOpensAt).toISOString() }); } const showSnoozeButton = - canOperateThread && props.snoozeSupported && canSnooze(thread, { now: snoozeGateNow }); + canTriageThread && props.snoozeSupported && canSnooze(thread, { now: snoozeGateNow }); // Closing here, not just masking, keeps the menu from reopening when the button returns. if (snoozeMenuOpenRaw && !showSnoozeButton) setSnoozeMenuOpen(false); const snoozeMenuOpen = snoozeMenuOpenRaw && showSnoozeButton; @@ -1028,7 +1046,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP const accessibility = resolveSidebarRowAccessibility({ title: thread.title, - statusLabel: topStatus?.label ?? null, + statusLabel: pendingSync ? "Pending" : (topStatus?.label ?? null), projectDisplayName: props.projectDisplayName, isActive: props.isActive, }); @@ -1302,7 +1320,11 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP !isWoke && "group-any-hover/sidebar-row:opacity-0", )} > - {variantAction === "unsnooze" && props.snoozeWakeLabelText !== null ? ( + {pendingSync ? ( + + + + ) : variantAction === "unsnooze" && props.snoozeWakeLabelText !== null ? ( {/* The clock tells a wake time from an age at a glance. */} @@ -1492,7 +1514,9 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP snoozeMenuOpen && "pointer-events-none absolute right-0 opacity-0", )} > - {topStatus ? ( + {pendingSync ? ( + + ) : topStatus ? ( isWokeStatus ? ( { }, ); + it.each([false, true])( + "keeps settle, snooze and their undo before the session loads (reversed: %s)", + (reversed) => { + const items = buildThreadActionMenuItems({ + ...baseState, + canOperate: false, + canTriage: true, + isPinned: reversed, + isSettled: reversed, + isSnoozed: reversed, + }); + expect(items.filter((item) => item.disabled).map((item) => item.id)).toEqual([ + reversed ? "unpin" : "pin", + "rename", + "regenerate-title", + "auto-settle", + "archive", + "delete", + ]); + expect( + items.find((item) => item.id === "snooze")?.children?.some((child) => child.disabled) ?? + false, + ).toBe(false); + }, + ); + it("disables archive and turning into a project while the thread is running", () => { const items = buildThreadActionMenuItems({ ...baseState, diff --git a/apps/web/src/components/threadActionMenu.logic.ts b/apps/web/src/components/threadActionMenu.logic.ts index 1e3eddd0b9389..76af8507dfde0 100644 --- a/apps/web/src/components/threadActionMenu.logic.ts +++ b/apps/web/src/components/threadActionMenu.logic.ts @@ -69,6 +69,8 @@ export function buildDraftActionMenuItems(options: { export interface ThreadActionMenuState { readonly canOperate: boolean; + /** Settle, snooze and their undo can queue before the environment's session loads. */ + readonly canTriage?: boolean; readonly canFind?: boolean; readonly branch: string | null; readonly projectFilter: { @@ -113,6 +115,16 @@ export function threadActionRequiresOperate(action: ThreadActionMenuId): boolean ].includes(action); } +export function isThreadTriageAction(action: string): boolean { + return ( + action === "settle" || + action === "unsettle" || + action === "snooze" || + action === "unsnooze" || + action.startsWith("snooze:") + ); +} + export function buildThreadActionMenuItems( state: ThreadActionMenuState, ): ReadonlyArray> { @@ -256,7 +268,8 @@ export function buildThreadActionMenuItems( return state.canOperate ? items : items.map((item) => - threadActionRequiresOperate(item.id) + threadActionRequiresOperate(item.id) && + !(state.canTriage === true && isThreadTriageAction(item.id)) ? { ...item, disabled: true, diff --git a/apps/web/src/connection/platform.ts b/apps/web/src/connection/platform.ts index 98335f5090c51..fc6b140837c0d 100644 --- a/apps/web/src/connection/platform.ts +++ b/apps/web/src/connection/platform.ts @@ -61,6 +61,7 @@ import { type DesktopSecondaryBootstrapsRead, } from "./desktopLocal"; import * as ConnectionStorage from "./storage"; +import { browserThreadTriageOutboxStore } from "./threadTriageOutboxStore"; import { clientPresentationMetadata } from "./clientMetadata"; let nextObservedRpcRequestId = 0; @@ -814,7 +815,11 @@ const layerEnvironmentOwnedDataCleanup = Layer.succeed( clear: Effect.fn("Persistence.EnvironmentOwnedDataCleanup.clear")(function (environmentId) { return Effect.sync(() => { clearComposerDraftsEnvironment(environmentId); - }); + }).pipe( + Effect.andThen( + Effect.ignore(browserThreadTriageOutboxStore.clearEnvironment(environmentId)), + ), + ); }), }), ); diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index b25abe442ba11..34b8accd7c750 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -14,6 +14,7 @@ import { setRoutesInCatalog, replaceCatalogValue, Persistence, + ThreadTriageOutboxStore, makeStoredShellSnapshotEncoder, makeStoredThreadSnapshotEncoder, makeThreadCacheWriteOrder, @@ -42,6 +43,7 @@ import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import { projectFaviconCache } from "../assets/projectFaviconCache"; +import { browserThreadTriageOutboxStore } from "./threadTriageOutboxStore"; const DATABASE_NAME = "t3code:connection-runtime"; const DATABASE_VERSION = 5; @@ -1113,6 +1115,7 @@ export const layer = Layer.effectContext( Context.add(CredentialStore.ConnectionCredentialStore, credentialStore), Context.add(TokenStore.RemoteDpopAccessTokenStore, remoteTokenStore), Context.add(Persistence.EnvironmentCacheStore, cacheStore), + Context.add(ThreadTriageOutboxStore, browserThreadTriageOutboxStore), Context.add(EnvironmentQueryStore, queryStore), ); }), diff --git a/apps/web/src/connection/threadTriageOutboxStore.ts b/apps/web/src/connection/threadTriageOutboxStore.ts new file mode 100644 index 0000000000000..2177ab3d441bf --- /dev/null +++ b/apps/web/src/connection/threadTriageOutboxStore.ts @@ -0,0 +1,18 @@ +import { + makeThreadTriageOutboxStore, + ThreadTriageOutboxStorageError, +} from "@t3tools/client-runtime/platform"; +import * as Effect from "effect/Effect"; + +const KEY = "t3code:thread-triage-outbox"; + +/** In localStorage, so tabs share it and clearing the environment cache keeps unsynced changes. */ +export const browserThreadTriageOutboxStore = makeThreadTriageOutboxStore((next) => + Effect.try({ + try: () => { + const value = next(window.localStorage.getItem(KEY)); + if (value !== null) window.localStorage.setItem(KEY, value); + }, + catch: (cause) => new ThreadTriageOutboxStorageError({ cause }), + }), +); diff --git a/apps/web/src/hooks/useThreadActionMenu.test.ts b/apps/web/src/hooks/useThreadActionMenu.test.ts index 3c8255b04c088..564764d1f6b04 100644 --- a/apps/web/src/hooks/useThreadActionMenu.test.ts +++ b/apps/web/src/hooks/useThreadActionMenu.test.ts @@ -45,6 +45,7 @@ vi.mock("@tanstack/react-router", () => ({ vi.mock("../state/session", () => ({ readEnvironmentScope: (environmentId: string, scope: string) => scope === AuthOrchestrationOperateScope && state.granted.has(environmentId), + readThreadTriageScope: (environmentId: string) => state.granted.has(environmentId), })); vi.mock("../state/entities", () => ({ readEnvironmentServerConfigLoaded: () => true, diff --git a/apps/web/src/hooks/useThreadActionMenu.ts b/apps/web/src/hooks/useThreadActionMenu.ts index f96baf908b3d9..ab2960262c225 100644 --- a/apps/web/src/hooks/useThreadActionMenu.ts +++ b/apps/web/src/hooks/useThreadActionMenu.ts @@ -18,6 +18,7 @@ import { useRouter } from "@tanstack/react-router"; import { pickSnoozePreset, resolveSnoozePresets } from "../components/Sidebar.snooze"; import { buildThreadActionMenuItems, + isThreadTriageAction, threadActionRequiresOperate, type ThreadActionMenuId, } from "../components/threadActionMenu.logic"; @@ -27,7 +28,7 @@ import { copyThreadReference } from "../lib/copyThreadReference"; import { threadChipCopyTarget } from "@t3tools/client-runtime/thread-reference"; import { threadEnvironment } from "../state/threads"; import { useOrchestrationCommand } from "../state/use-orchestration-command"; -import { readEnvironmentScope } from "../state/session"; +import { readEnvironmentScope, readThreadTriageScope } from "../state/session"; import { readEnvironmentServerConfigLoaded, readEnvironmentSupportsAutoSettleOptOut, @@ -130,6 +131,7 @@ export function useThreadActionMenu(input: { const snoozePresets = resolveSnoozePresets(now, timestampFormat, threadRef.environmentId); const items = buildThreadActionMenuItems({ canOperate: readEnvironmentScope(threadRef.environmentId, AuthOrchestrationOperateScope), + canTriage: readThreadTriageScope(threadRef.environmentId), canFind: onFind !== undefined, branch: thread.branch ?? null, projectFilter: null, @@ -150,7 +152,9 @@ export function useThreadActionMenu(input: { const action: ThreadActionMenuId = clicked.value; if ( threadActionRequiresOperate(action) && - !readEnvironmentScope(threadRef.environmentId, AuthOrchestrationOperateScope) + !(isThreadTriageAction(action) + ? readThreadTriageScope(threadRef.environmentId) + : readEnvironmentScope(threadRef.environmentId, AuthOrchestrationOperateScope)) ) { failureToast( "Thread action unavailable", diff --git a/apps/web/src/hooks/useThreadActions.permissions.test.ts b/apps/web/src/hooks/useThreadActions.permissions.test.ts index 649ec0c6bac1b..2e6ee88521481 100644 --- a/apps/web/src/hooks/useThreadActions.permissions.test.ts +++ b/apps/web/src/hooks/useThreadActions.permissions.test.ts @@ -27,6 +27,7 @@ const state = vi.hoisted(() => ({ serverConfig: { shellThreadFamilies: false }, familyLookupFails: false, scopes: new Map>(), + sessionNotLoaded: new Set(), threads: [] as { environmentId: EnvironmentId; id: ThreadId; @@ -65,6 +66,9 @@ vi.mock("../state/session", () => ({ environmentSession: { sessionStateAtom: "session" }, readEnvironmentScope: (environmentId: string, scope: string) => state.scopes.get(environmentId)?.has(scope) === true, + readThreadTriageScope: (environmentId: string) => + state.scopes.get(environmentId)?.has(AuthOrchestrationOperateScope) === true || + state.sessionNotLoaded.has(environmentId), })); vi.mock("../state/use-atom-command", () => ({ useAtomCommand: @@ -73,7 +77,11 @@ vi.mock("../state/use-atom-command", () => ({ state.requests.push({ action, ...request }); const scope = action === "removeWorktree" ? AuthSourceControlWriteScope : AuthOrchestrationOperateScope; - if (!state.scopes.get(request.environmentId)?.has(scope)) { + // Without a loaded session the command itself queues settle and snooze. + if ( + !state.scopes.get(request.environmentId)?.has(scope) && + !state.sessionNotLoaded.has(request.environmentId) + ) { return AsyncResult.failure(Cause.fail(new Error("Server denied the request"))); } state.afterRequest?.(action); @@ -274,6 +282,7 @@ beforeEach(() => { latestTurn: null, }, ]; + state.sessionNotLoaded = new Set(); state.requests = []; state.localEffects = []; state.confirm.mockReset().mockResolvedValue(true); @@ -314,6 +323,17 @@ describe("thread action permissions", () => { }, ); + it.each(operations)( + "$name runs before the session loads only for settle, snooze and their undo", + async ({ name, run }) => { + state.sessionNotLoaded.add(secondary); + const result = await run(useThreadActions(), target); + const triage = ["settle", "unsettle", "snooze", "unsnooze"].includes(name); + expect(result._tag).toBe(triage ? "Success" : "Failure"); + expect(state.requests.map(({ action }) => action)).toEqual(triage ? [name] : []); + }, + ); + it.each(["confirmAndDeleteThread", "confirmAndUnpinThread"] as const)( "%s blocks a forbidden confirmation", async (action) => { diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index cc9c27770886e..ef781c409f837 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -24,7 +24,7 @@ import { useRouter } from "@tanstack/react-router"; import { getFallbackThreadIdAfterDelete, pinOrderKeyBetween } from "../components/Sidebar.logic"; import { useComposerDraftStore } from "../composerDraftStore"; -import { environmentSession, readEnvironmentScope } from "../state/session"; +import { environmentSession, readEnvironmentScope, readThreadTriageScope } from "../state/session"; import { appAtomRegistry } from "../rpc/atomRegistry"; import { environmentServerConfigsAtom } from "../state/server"; import { isScratchProject } from "@t3tools/client-runtime/state/projects"; @@ -284,12 +284,16 @@ export function useThreadActions() { const deleteThreadMutation = useOrchestrationCommand(threadEnvironment.delete, { reportFailure: false, }); - const settleThreadMutation = useOrchestrationCommand(threadEnvironment.settle, { - reportFailure: false, - }); - const unsettleThreadMutation = useOrchestrationCommand(threadEnvironment.unsettle, { - reportFailure: false, - }); + const settleThreadMutation = useOrchestrationCommand( + threadEnvironment.settle, + { reportFailure: false }, + readThreadTriageScope, + ); + const unsettleThreadMutation = useOrchestrationCommand( + threadEnvironment.unsettle, + { reportFailure: false }, + readThreadTriageScope, + ); const pinThreadMutation = useOrchestrationCommand(threadEnvironment.pin, { reportFailure: false, }); @@ -305,12 +309,16 @@ export function useThreadActions() { const reorderActiveThreadMutation = useOrchestrationCommand(threadEnvironment.reorderActive, { reportFailure: false, }); - const snoozeThreadMutation = useOrchestrationCommand(threadEnvironment.snooze, { - reportFailure: false, - }); - const unsnoozeThreadMutation = useOrchestrationCommand(threadEnvironment.unsnooze, { - reportFailure: false, - }); + const snoozeThreadMutation = useOrchestrationCommand( + threadEnvironment.snooze, + { reportFailure: false }, + readThreadTriageScope, + ); + const unsnoozeThreadMutation = useOrchestrationCommand( + threadEnvironment.unsnooze, + { reportFailure: false }, + readThreadTriageScope, + ); const markThreadUnread = useMarkThreadUnread(); const createProjectFromThread = useAtomCommand(projectEnvironment.createFromThread, { reportFailure: false, @@ -950,16 +958,22 @@ export function useThreadActions() { return result; }; - const runUndoPlan = (target: ScopedThreadRef, plan: ThreadUndo.ThreadUndoPlan) => - ThreadUndo.executeThreadUndo( + const runUndoPlan = (target: ScopedThreadRef, plan: ThreadUndo.ThreadUndoPlan) => { + let cancelledQueued = false; + return ThreadUndo.executeThreadUndo( plan, async (step) => { switch (step.action) { - case "unsettle": - return unsettleThread(target); + case "unsettle": { + const result = await unsettleThread(target); + cancelledQueued = + result._tag === "Success" && ThreadUndo.cancelledPendingSync(result.value); + return result; + } case "unsnooze": return unsnoozeThread(target); case "pin": + if (cancelledQueued) return AsyncResult.success(undefined); return sendPin(target, step.orderKey === undefined ? {} : { orderKey: step.orderKey }); case "snooze": return snoozeThreadMutation({ @@ -970,6 +984,7 @@ export function useThreadActions() { }, (result) => result._tag === "Success", ); + }; const confirmAndDeleteThread = async (target: ScopedThreadRef, knownTitle?: string) => { const permissionFailure = threadOperationFailure(target); diff --git a/apps/web/src/hooks/useThreadActions.undo.test.ts b/apps/web/src/hooks/useThreadActions.undo.test.ts index 98f69e9de5a09..290d54e262c0d 100644 --- a/apps/web/src/hooks/useThreadActions.undo.test.ts +++ b/apps/web/src/hooks/useThreadActions.undo.test.ts @@ -115,6 +115,7 @@ beforeEach(() => { for (const command of Object.values(commands)) { command.mockReset().mockResolvedValue({ _tag: "Success", value: undefined }); } + commands.settle.mockResolvedValue({ _tag: "Success", value: { sequence: 1 } }); router.navigate.mockClear(); router.state.location.href = "/"; router.state.matches[0]!.params = {}; @@ -311,6 +312,35 @@ describe("settle and snooze Undo", () => { }); }); + // Whether the queued settle reached the server is known only when Undo runs. + it.each([ + ["cancels it before it syncs", { sequence: 0, cancelledPendingSync: true }, 0], + ["finds it delivered", { sequence: 3 }, 1], + ] as const)( + "restores a queued settle only as far as needed when Undo %s", + async (_, undone, restorations) => { + const snoozedUntil = "2030-01-01T09:00:00.000Z"; + threadShell.pinnedAt = "2026-01-01T00:00:00.000Z"; + threadShell.snoozedUntil = snoozedUntil; + commands.unsettle.mockResolvedValue({ _tag: "Success", value: undone }); + await useThreadActions().settleThread(target); + expect(useThreadUndoNotice.getState().notice).toMatchObject({ action: "Settled", count: 1 }); + await currentUndo()(); + expect(commands.unsettle).toHaveBeenCalledExactlyOnceWith({ + environmentId: target.environmentId, + input: { threadId: target.threadId, reason: "user" }, + }); + expect(commands.pin).toHaveBeenCalledTimes(restorations); + expect(commands.snooze).toHaveBeenCalledOnce(); + if (restorations > 0) { + expect(commands.snooze).toHaveBeenCalledWith({ + environmentId: target.environmentId, + input: { threadId: target.threadId, snoozedUntil }, + }); + } + }, + ); + it("stops settle restoration if restoring the pin fails", async () => { threadShell.pinnedAt = "2026-01-01T00:00:00.000Z"; threadShell.snoozedUntil = "2030-01-01T09:00:00.000Z"; diff --git a/apps/web/src/state/session.ts b/apps/web/src/state/session.ts index 2c2de0a107d36..7509f93da5e0d 100644 --- a/apps/web/src/state/session.ts +++ b/apps/web/src/state/session.ts @@ -1,6 +1,7 @@ import { useAtomValue } from "@effect/atom-react"; import { createEnvironmentSessionAtoms } from "@t3tools/client-runtime/state/session"; import { + AuthOrchestrationOperateScope, type AuthEnvironmentScope, type AuthSessionState, type EnvironmentId, @@ -73,6 +74,33 @@ export function readEnvironmentScope( ); } +/** + * Settle and snooze can queue for an environment whose session has not loaded yet, such as one + * that was offline at launch. The server checks the grant when the change is sent. + */ +function sessionAllowsThreadTriage( + result: AsyncResult.AsyncResult, +): boolean { + // A failed refresh still carries the last loaded session, whose answer stands. + const session = AsyncResult.getOrElse(result, () => null); + return session === null || sessionGrantsScope(session, AuthOrchestrationOperateScope); +} + +export function useThreadTriageScope(environmentId: EnvironmentId | null): boolean { + return useAtomValue( + environmentId === null + ? EMPTY_SESSION_STATE_ATOM + : environmentSession.sessionStateAtom(environmentId), + sessionAllowsThreadTriage, + ); +} + +export function readThreadTriageScope(environmentId: EnvironmentId): boolean { + return sessionAllowsThreadTriage( + appAtomRegistry.get(environmentSession.sessionStateAtom(environmentId)), + ); +} + const EMPTY_PREPARED_CONNECTION_ATOM = Atom.make(Option.none()).pipe( Atom.withLabel("web-prepared-connection:empty"), ); diff --git a/apps/web/src/state/threads.ts b/apps/web/src/state/threads.ts index 6d2ce2a07c90b..1cb2749055806 100644 --- a/apps/web/src/state/threads.ts +++ b/apps/web/src/state/threads.ts @@ -32,6 +32,7 @@ export const olderSettledThreads = createOlderSettledThreadAtoms( export const threadEnvironment = createThreadEnvironmentAtoms( connectionAtomRuntime, threadFamilies.snapshotAtom, + environmentShell.statusAtom, ); const environmentThreads = createEnvironmentThreadStateAtoms(connectionAtomRuntime); export const environmentThreadDetails = createEnvironmentThreadDetailAtoms( diff --git a/apps/web/src/state/use-orchestration-command.test.ts b/apps/web/src/state/use-orchestration-command.test.ts index c0ddca203f081..1087670ac58b9 100644 --- a/apps/web/src/state/use-orchestration-command.test.ts +++ b/apps/web/src/state/use-orchestration-command.test.ts @@ -54,6 +54,19 @@ it("checks the command's environment instead of borrowing the primary grant", as expect(state.run).toHaveBeenCalledWith(target(first)); }); +it("checks a caller's permission instead of the live grant on every call", async () => { + const remembered = new Set([second]); + const mutate = useOrchestrationCommand(command, { reportFailure: false }, (environmentId) => + remembered.has(environmentId), + ); + expect(await mutate(target(second))).toMatchObject({ _tag: "Success", value: "receipt" }); + remembered.clear(); + const denied = await mutate(target(second)); + assert(denied._tag === "Failure"); + expect(Cause.squash(denied.cause)).toBeInstanceOf(EnvironmentAuthorizationError); + expect(state.run).toHaveBeenCalledOnce(); +}); + it("does not dispatch a later mutation after access changes while awaiting a receipt", async () => { state.allowed.add(second); let finish: (() => void) | undefined; diff --git a/apps/web/src/state/use-orchestration-command.ts b/apps/web/src/state/use-orchestration-command.ts index 98f7efeecbfe6..7c37b83ae82f3 100644 --- a/apps/web/src/state/use-orchestration-command.ts +++ b/apps/web/src/state/use-orchestration-command.ts @@ -16,13 +16,23 @@ import { useAtomCommand } from "./use-atom-command"; const ORCHESTRATION_OPERATE_DENIED_MESSAGE = "This connection cannot change threads or projects."; +// Module scope: React Compiler cannot reorder an inline arrow used as a parameter default. +function hasLiveOperateGrant(environmentId: EnvironmentId): boolean { + return readEnvironmentScope(environmentId, AuthOrchestrationOperateScope); +} + +/** + * `permitted` replaces the operate check, so settle, snooze and their undo can queue before the + * environment's session loads. It runs on every call, like the default. + */ export function useOrchestrationCommand( command: AtomCommand, options?: string | AtomCommandOptions, + permitted: (environmentId: EnvironmentId) => boolean = hasLiveOperateGrant, ): (value: W) => Promise> { const run = useAtomCommand(command, options); return async (value: W): Promise> => { - if (!readEnvironmentScope(value.environmentId, AuthOrchestrationOperateScope)) { + if (!permitted(value.environmentId)) { return AsyncResult.failure( Cause.fail( new EnvironmentAuthorizationError({ diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index e5fe950e140a2..02bbe2a3399db 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -174,6 +174,12 @@ when the thread settles** enabled in Settings > Projects > Actions. You can also that action so T3 can stop its terminal and child processes. Commands started through an agent's own shell tools are managed by the provider. +You can settle or snooze a thread while its environment is offline. The thread shows +**Waiting to sync** on mobile or **Pending** on web and desktop, and the change is sent +when the environment reconnects. If the thread changed there in the meantime, that +environment's state wins and your change is dropped, as it is when this connection cannot +change threads there. A thread with a pending settle cannot be snoozed until it syncs. + To reclaim disk space from settled work, turn on **Run in the thread's worktree when the thread settles** for one of the project's actions, or set `"runOnSettle": true` on a `t3.json` script, for example `cargo clean`. It runs each time a thread in its own diff --git a/packages/client-runtime/src/platform/index.ts b/packages/client-runtime/src/platform/index.ts index fa7be91aecd3f..33619c378e3b7 100644 --- a/packages/client-runtime/src/platform/index.ts +++ b/packages/client-runtime/src/platform/index.ts @@ -7,3 +7,4 @@ export * from "./shellCacheEncoding.ts"; export * from "./threadCacheEncoding.ts"; export * as PlatformConnectionSource from "./source.ts"; export * from "./storageDocument.ts"; +export * from "./threadTriageOutboxStore.ts"; diff --git a/packages/client-runtime/src/platform/threadTriageOutboxStore.ts b/packages/client-runtime/src/platform/threadTriageOutboxStore.ts new file mode 100644 index 0000000000000..589bdc75433f5 --- /dev/null +++ b/packages/client-runtime/src/platform/threadTriageOutboxStore.ts @@ -0,0 +1,99 @@ +import { CommandId, EnvironmentId, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; + +export const PendingThreadTriage = Schema.Struct({ + environmentId: EnvironmentId, + threadId: ThreadId, + commandId: CommandId, + action: Schema.Union([ + Schema.Struct({ type: Schema.Literal("settle") }), + Schema.Struct({ type: Schema.Literal("snooze"), snoozedUntil: Schema.String }), + ]), + queuedAt: Schema.DateTimeUtcFromString, + /** The thread as the user saw it; if it shows otherwise on reconnect, the change is dropped. */ + seen: Schema.String, + /** Set before the first send and never cleared: the server may have it, so only it can undo. */ + attempted: Schema.optionalKey(Schema.Literal(true)), +}); +export type PendingThreadTriage = typeof PendingThreadTriage.Type; +export type PendingThreadTriages = ReadonlyArray; + +const Stored = Schema.fromJsonString(Schema.Struct({ entries: Schema.Array(PendingThreadTriage) })); +const decodeStored = Schema.decodeUnknownExit(Stored); +const encodeStored = Schema.encodeSync(Stored); + +export class ThreadTriageOutboxStorageError extends Schema.TaggedError()( + "ThreadTriageOutboxStorageError", + { cause: Schema.Defect() }, +) { + override get message(): string { + return "Could not save the change for later sync."; + } +} + +/** + * Settle and snooze taken while an environment is offline, kept apart from the environment + * cache so clearing the cache does not drop them. Removing the environment does. + */ +export interface ThreadTriageOutboxStoreShape { + /** Stores `change(stored)` in one step and returns it. Returning `stored` itself only reads. */ + readonly update: ( + change: (stored: PendingThreadTriages) => PendingThreadTriages, + ) => Effect.Effect; + readonly clearEnvironment: ( + environmentId: EnvironmentId, + ) => Effect.Effect; +} + +/** + * `modify` replaces the stored text with `next(stored)`, or keeps it when that is null. Text that + * no longer decodes reads as empty. + */ +export function makeThreadTriageOutboxStore( + modify: ( + next: (stored: string | null) => string | null, + ) => Effect.Effect, +): ThreadTriageOutboxStoreShape { + // Platforms whose `modify` reads and writes in separate steps rely on this to not interleave. + const lock = Semaphore.makeUnsafe(1); + const update: ThreadTriageOutboxStoreShape["update"] = (change) => + lock.withPermits(1)( + Effect.suspend(() => { + let result: PendingThreadTriages = []; + return modify((raw) => { + const decoded = raw === null ? undefined : decodeStored(raw); + const stored = + decoded !== undefined && Exit.isSuccess(decoded) ? decoded.value.entries : []; + result = change(stored); + return result === stored ? null : encodeStored({ entries: result }); + }).pipe(Effect.map(() => result)); + }), + ); + return { + update, + clearEnvironment: (environmentId) => + update((stored) => + stored.some((entry) => entry.environmentId === environmentId) + ? stored.filter((entry) => entry.environmentId !== environmentId) + : stored, + ).pipe(Effect.asVoid), + }; +} + +export class ThreadTriageOutboxStore extends Context.Reference( + "@t3tools/client-runtime/state/ThreadTriageOutboxStore", + { + defaultValue: () => { + let value: string | null = null; + return makeThreadTriageOutboxStore((next) => + Effect.sync(() => { + value = next(value) ?? value; + }), + ); + }, + }, +) {} diff --git a/packages/client-runtime/src/state/threadCommands.test.ts b/packages/client-runtime/src/state/threadCommands.test.ts index 031ef68a14502..b08a9fde70cd1 100644 --- a/packages/client-runtime/src/state/threadCommands.test.ts +++ b/packages/client-runtime/src/state/threadCommands.test.ts @@ -16,6 +16,7 @@ import * as DateTime from "effect/DateTime"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Cause from "effect/Cause"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; @@ -25,6 +26,7 @@ import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom, AtomRegistry } from "effect/reactivity"; +import { RpcClientError } from "effect/rpc"; import * as EnvironmentRegistry from "../connection/registry.ts"; import { @@ -34,7 +36,14 @@ import { } from "../connection/model.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import type { RpcSession } from "../rpc/session.ts"; +import { + makeThreadTriageOutboxStore, + ThreadTriageOutboxStorageError, + ThreadTriageOutboxStore, +} from "../platform/threadTriageOutboxStore.ts"; +import type { EnvironmentShellStatus } from "./shell.ts"; import { createThreadEnvironmentAtoms } from "./threadCommands.ts"; +import * as ThreadUndo from "./threadUndo.ts"; import * as ThreadHistoryController from "./threadHistoryController.ts"; const ENVIRONMENT_ID = EnvironmentId.make("remote"); @@ -92,15 +101,33 @@ const SNAPSHOT: OrchestrationV2ShellSnapshot = { ], }; +let randomCalls = 0; + const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (options?: { readonly loadEarlier?: () => Effect.Effect; readonly reconcilePendingWork?: (threadId: ThreadId) => Effect.Effect; + readonly connected?: boolean; + /** Shared between harnesses to stand in for a restart. */ + readonly storage?: { value: string | null; failWrites?: boolean; onModify?: () => void }; }) { const requests = yield* Queue.unbounded<{ command: OrchestrationV2Command; reply: Deferred.Deferred<{ sequence: number }, Error>; }>(); const reconciled = yield* Queue.unbounded(); + const liveSession = Option.some({ + client: { + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => + Effect.gen(function* () { + const reply = yield* Deferred.make<{ sequence: number }, Error>(); + yield* Queue.offer(requests, { command, reply }); + return yield* Deferred.await(reply); + }), + }, + } as unknown as RpcSession); + const session = yield* SubscriptionRef.make( + options?.connected === false ? Option.none() : liveSession, + ); const connectionFollowed = yield* Deferred.make(); const connection = yield* SubscriptionRef.make({ ...AVAILABLE_CONNECTION_STATE, @@ -110,23 +137,25 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (optio const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ target: new BearerConnectionTarget({ environmentId: ENVIRONMENT_ID, - label: "Test", + label: "Mac", connectionId: "test", }), state: connection, - session: yield* SubscriptionRef.make( - Option.some({ - client: { - [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command: OrchestrationV2Command) => - Effect.gen(function* () { - const reply = yield* Deferred.make<{ sequence: number }, Error>(); - yield* Queue.offer(requests, { command, reply }); - return yield* Deferred.await(reply); - }), - }, - } as unknown as RpcSession), - ), + session, } as EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const storage = options?.storage ?? { value: null }; + const store = makeThreadTriageOutboxStore((next) => + Effect.suspend(() => { + storage.onModify?.(); + const value = next(storage.value); + if (value === null) return Effect.void; + if (storage.failWrites) { + return Effect.fail(new ThreadTriageOutboxStorageError({ cause: "disk full" })); + } + storage.value = value; + return Effect.void; + }), + ); const runtime = Atom.runtime( Layer.mergeAll( Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, { @@ -141,8 +170,10 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (optio } as EnvironmentRegistry.EnvironmentRegistry["Service"]), Layer.succeed( SecureRandom.SecureRandom, - SecureRandom.make((size) => new Uint8Array(size)), + // Distinct bytes per call, so each queued change gets its own command id. + SecureRandom.make((size) => new Uint8Array(size).fill(++randomCalls % 256)), ), + Layer.succeed(ThreadTriageOutboxStore, store), Layer.succeed( ThreadHistoryController.ThreadHistoryController, ThreadHistoryController.ThreadHistoryController.of({ @@ -158,18 +189,36 @@ const makeHarness = Effect.fn("TestThreadCommands.makeHarness")(function* (optio const snapshotAtom = Atom.family((_environmentId: EnvironmentId) => Atom.make(SNAPSHOT), ); - const commands = createThreadEnvironmentAtoms(runtime, snapshotAtom); + const statusAtom = Atom.family((_environmentId: EnvironmentId) => + Atom.make(options?.connected === false ? "cached" : "live"), + ); + const commands = createThreadEnvironmentAtoms(runtime, snapshotAtom, statusAtom); const registry = AtomRegistry.make(); yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose())); const visibleAtom = commands.snapshotAtom(ENVIRONMENT_ID); registry.mount(visibleAtom); + const pendingSyncAtom = commands.pendingSyncAtom(ENVIRONMENT_ID); + registry.mount(pendingSyncAtom); + const reconnect = (status: EnvironmentShellStatus = "live") => + SubscriptionRef.set(session, liveSession).pipe( + Effect.andThen(Effect.sync(() => registry.set(statusAtom(ENVIRONMENT_ID), status))), + ); + const disconnect = SubscriptionRef.set(session, Option.none()).pipe( + Effect.andThen(Effect.sync(() => registry.set(statusAtom(ENVIRONMENT_ID), "cached"))), + ); return { registry, commands, snapshotAtom, + statusAtom, visibleAtom, + pendingSyncAtom, requests, reconciled, + storage, + store, + reconnect, + disconnect, connection, connectionFollowed, }; @@ -665,3 +714,358 @@ describe("remote thread lifecycle commands", () => { }), ); }); + +const waitFor = ( + registry: AtomRegistry.AtomRegistry, + atom: Atom.Atom, + done: (value: A) => boolean, +) => + Effect.callback((resume) => { + if (done(registry.get(atom))) return resume(Effect.void); + const unsubscribe = registry.subscribe(atom, (value) => { + if (!done(value)) return; + unsubscribe(); + resume(Effect.void); + }); + return Effect.sync(unsubscribe); + }); + +describe("settle and snooze while the environment is offline", () => { + const settle = { environmentId: ENVIRONMENT_ID, input: { threadId: THREAD_ID } }; + const snooze = { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, snoozedUntil: "2099-01-01T00:00:00.000Z" }, + }; + const undo = { + environmentId: ENVIRONMENT_ID, + input: { threadId: THREAD_ID, reason: "user" as const }, + }; + const stored = (storage: { value: string | null }) => + storage.value === null + ? [] + : (JSON.parse(storage.value) as { entries: Array> }).entries; + const changedThread = (change: Partial) => ({ + ...SNAPSHOT, + snapshotSequence: 5, + threads: [{ ...SNAPSHOT.threads[0]!, ...change }], + }); + /** The request may have reached the server before the socket closed. */ + const lostReply = () => + new RpcClientError.RpcClientError({ + reason: new RpcClientError.RpcClientDefect({ + message: "socket closed", + cause: new Error("socket closed"), + }), + }); + const queued = Effect.fn("queued")(function* ( + action: "settle" | "snooze" = "settle", + storage?: { value: string | null; onModify?: () => void }, + ) { + const h = yield* makeHarness({ connected: false, ...(storage ? { storage } : {}) }); + const result = yield* Effect.promise(() => h.commands[action].run(h.registry, snooze)); + expect(result._tag).toBe("Success"); + return h; + }); + type Harness = Effect.Success>; + const nothingPending = (h: Harness) => + waitFor(h.registry, h.pendingSyncAtom, (threads) => threads.size === 0); + /** Commands on one thread go out in order, so this one arrives first only if no other was sent. */ + const expectNothingSent = Effect.fn("expectNothingSent")(function* (h: Harness) { + const pin = h.commands.pin.run(h.registry, settle); + const request = yield* Queue.take(h.requests); + expect(request.command.type).toBe("thread.pin"); + yield* Deferred.succeed(request.reply, { sequence: 1 }); + yield* Effect.promise(() => pin); + }); + + it.effect("shows a settle at once and sends it when the environment is back", () => + Effect.gen(function* () { + const h = yield* queued(); + expect(h.registry.get(h.visibleAtom)?.threads[0]?.settledOverride).toBe("settled"); + expect(h.registry.get(h.pendingSyncAtom).has(THREAD_ID)).toBe(true); + const [entry] = stored(h.storage); + expect(entry).toMatchObject({ threadId: THREAD_ID, action: { type: "settle" } }); + + yield* h.reconnect(); + const request = yield* Queue.take(h.requests); + expect(request.command).toMatchObject({ + type: "thread.settle", + commandId: entry!.commandId, + threadId: THREAD_ID, + }); + // Recorded before the send, so a restart or Undo from here asks the server. + expect(stored(h.storage)[0]!.attempted).toBe(true); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + yield* nothingPending(h); + expect(stored(h.storage)).toEqual([]); + }), + ); + + it.effect("sends a queued snooze after a restart, under the command id it was stored with", () => + Effect.gen(function* () { + const storage = { value: null as string | null }; + yield* queued("snooze", storage); + const [entry] = stored(storage); + + const restarted = yield* makeHarness({ connected: false, storage }); + yield* waitFor( + restarted.registry, + restarted.pendingSyncAtom, + (threads) => threads.size === 1, + ); + expect(restarted.registry.get(restarted.visibleAtom)?.threads[0]?.snoozedUntil).toEqual( + FUTURE, + ); + yield* restarted.reconnect(); + const request = yield* Queue.take(restarted.requests); + expect(request.command).toMatchObject({ + type: "thread.snooze", + commandId: entry!.commandId, + snoozedUntil: "2099-01-01T00:00:00.000Z", + }); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + yield* nothingPending(restarted); + yield* expectNothingSent(restarted); + }), + ); + + it.effect.each([ + ["produced new output", { itemCount: 1 }], + ["was pinned on another device", { pinnedAt: NOW }], + ["is asking for approval", { pendingRuntimeRequest: APPROVAL }], + ] as const)("keeps the server's state for a thread that %s meanwhile", ([, change]) => + Effect.gen(function* () { + const h = yield* queued(); + // Judged against the synchronized shell: sending any earlier would leave it pending here. + yield* h.reconnect("synchronizing"); + const changed = changedThread(change); + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), changed); + h.registry.set(h.statusAtom(ENVIRONMENT_ID), "live"); + yield* nothingPending(h); + yield* expectNothingSent(h); + expect(stored(h.storage)).toEqual([]); + expect(h.registry.get(h.visibleAtom)).toBe(changed); + }), + ); + + it.effect.each([ + // A reconnect whose shell has not synchronized yet lands while the change is being marked. + { when: "the shell stops being synchronized while it is marked", connected: true }, + // The shell reports synchronized from a stale stream while the session is already gone. + { when: "its environment has no session", connected: false }, + ])("leaves a change unsent and cancellable when $when", ({ connected }) => + Effect.gen(function* () { + const unmarked = yield* Queue.unbounded(); + const storage: { value: string | null; onModify?: () => void } = { value: null }; + const h = yield* queued("settle", storage); + let calls = 0; + // Replay reads the queue, marks the change, and takes the mark off again. + storage.onModify = () => { + calls += 1; + if (calls === 2 && connected) { + h.registry.set(h.statusAtom(ENVIRONMENT_ID), "synchronizing"); + } + if (calls === 3) Queue.offerUnsafe(unmarked, undefined); + }; + if (connected) yield* h.reconnect(); + else h.registry.set(h.statusAtom(ENVIRONMENT_ID), "live"); + yield* Queue.take(unmarked); + const result = yield* Effect.promise(() => h.commands.unsettle.run(h.registry, undo)); + expect(result._tag === "Success" && ThreadUndo.cancelledPendingSync(result.value)).toBe(true); + expect(stored(storage)).toEqual([]); + yield* h.reconnect(); + yield* expectNothingSent(h); + }), + ); + + it.effect("keeps another tab's sent mark when this tab could not send", () => + Effect.gen(function* () { + const flipped = yield* Queue.unbounded(); + const storage: { value: string | null; onModify?: () => void } = { value: null }; + const h = yield* queued("settle", storage); + let calls = 0; + // Between this tab's read and its mark, another tab marks the change and sends it. Then + // this tab's shell stops being synchronized. + storage.onModify = () => { + if (++calls !== 2) return; + storage.value = storage.value!.replace('"seen"', '"attempted":true,"seen"'); + h.registry.set(h.statusAtom(ENVIRONMENT_ID), "synchronizing"); + Queue.offerUnsafe(flipped, undefined); + }; + yield* h.reconnect(); + yield* Queue.take(flipped); + const undone = h.commands.unsettle.run(h.registry, undo); + const request = yield* Queue.take(h.requests); + expect(request.command.type).toBe("thread.unsettle"); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + expect((yield* Effect.promise(() => undone))._tag).toBe("Success"); + }), + ); + + it.effect("sends a queued change again when the first send got no answer", () => + Effect.gen(function* () { + const h = yield* queued(); + yield* h.reconnect(); + const first = yield* Queue.take(h.requests); + yield* h.disconnect; + yield* Deferred.fail(first.reply, lostReply()); + expect(h.registry.get(h.pendingSyncAtom).has(THREAD_ID)).toBe(true); + yield* h.reconnect(); + const second = yield* Queue.take(h.requests); + expect(second.command.commandId).toBe(first.command.commandId); + yield* Deferred.succeed(second.reply, { sequence: 2 }); + yield* nothingPending(h); + }), + ); + + it.effect("drops a queued change the server refuses, and shows the server's state", () => + Effect.gen(function* () { + const h = yield* queued(); + yield* h.reconnect(); + const request = yield* Queue.take(h.requests); + yield* Deferred.fail(request.reply, new Error("This session cannot change threads.")); + yield* nothingPending(h); + expect(stored(h.storage)).toEqual([]); + expect(h.registry.get(h.visibleAtom)).toBe(SNAPSHOT); + }), + ); + + it.effect.each([ + ["settle", "unsettle"], + ["snooze", "unsnooze"], + ] as const)("undoing a queued %s cancels it without contacting the server", ([action, undoing]) => + Effect.gen(function* () { + const h = yield* queued(action); + const result = yield* Effect.promise(() => h.commands[undoing].run(h.registry, undo)); + expect(result._tag === "Success" && ThreadUndo.cancelledPendingSync(result.value)).toBe(true); + expect(h.registry.get(h.pendingSyncAtom).size).toBe(0); + expect(stored(h.storage)).toEqual([]); + expect(h.registry.get(h.visibleAtom)).toBe(SNAPSHOT); + yield* h.reconnect(); + yield* expectNothingSent(h); + }), + ); + + it.effect("after a lost reply, Undo asks the server and the change is not sent twice", () => + Effect.gen(function* () { + const h = yield* queued(); + yield* h.reconnect(); + const request = yield* Queue.take(h.requests); + yield* h.disconnect; + yield* Deferred.fail(request.reply, lostReply()); + + // The server may have settled it, so cancelling on the device would lose the Undo. + const undone = yield* Effect.promise(() => h.commands.unsettle.run(h.registry, undo)); + expect(undone._tag).toBe("Failure"); + expect(h.registry.get(h.pendingSyncAtom).has(THREAD_ID)).toBe(true); + + // It had: the returning shell shows the thread settled, which answers the queued change. + h.registry.set( + h.snapshotAtom(ENVIRONMENT_ID), + changedThread({ settledOverride: "settled", settledAt: NOW }), + ); + yield* h.reconnect(); + yield* nothingPending(h); + yield* expectNothingSent(h); + }), + ); + + it.effect("a later action that reaches the server replaces the queued one", () => + Effect.gen(function* () { + const h = yield* queued(); + yield* h.reconnect("synchronizing"); + const result = h.commands.snooze.run(h.registry, snooze); + const request = yield* Queue.take(h.requests); + expect(request.command.type).toBe("thread.snooze"); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + expect((yield* Effect.promise(() => result))._tag).toBe("Success"); + expect(stored(h.storage)).toEqual([]); + h.registry.set(h.statusAtom(ENVIRONMENT_ID), "live"); + yield* expectNothingSent(h); + }), + ); + + it.effect("drops a queued snooze that expired before the reconnect", () => + Effect.gen(function* () { + const storage = { value: null as string | null }; + yield* queued("snooze", storage); + storage.value = storage.value!.replace( + "2099-01-01T00:00:00.000Z", + "2000-01-01T00:00:00.000Z", + ); + const restarted = yield* makeHarness({ connected: false, storage }); + yield* waitFor( + restarted.registry, + restarted.pendingSyncAtom, + (threads) => threads.size === 1, + ); + yield* restarted.reconnect(); + yield* nothingPending(restarted); + yield* expectNothingSent(restarted); + expect(stored(storage)).toEqual([]); + }), + ); + + it.effect("sends what storage holds when another tab replaced the change", () => + Effect.gen(function* () { + const storage = { value: null as string | null }; + const first = yield* queued("settle", storage); + const second = yield* makeHarness({ connected: false, storage }); + yield* waitFor(second.registry, second.pendingSyncAtom, (threads) => threads.size === 1); + yield* Effect.promise(() => second.commands.unsettle.run(second.registry, undo)); + yield* Effect.promise(() => second.commands.snooze.run(second.registry, snooze)); + + yield* first.reconnect(); + const request = yield* Queue.take(first.requests); + expect(request.command).toMatchObject({ + type: "thread.snooze", + commandId: stored(storage)[0]!.commandId, + }); + yield* Deferred.succeed(request.reply, { sequence: 2 }); + yield* nothingPending(first); + }), + ); + + it.effect("does not queue a settle the cached thread cannot take", () => + Effect.gen(function* () { + const h = yield* makeHarness({ connected: false }); + h.registry.set(h.snapshotAtom(ENVIRONMENT_ID), changedThread({ status: "running" })); + const result = yield* Effect.promise(() => h.commands.settle.run(h.registry, settle)); + expect(result._tag).toBe("Failure"); + expect(stored(h.storage)).toEqual([]); + }), + ); + + // The other tab loaded before the settle was queued, so only storage knows about it. + it.effect( + "keeps a settle queued in another tab when a snooze of the same thread would replace it", + () => + Effect.gen(function* () { + const storage = { value: null as string | null }; + const other = yield* makeHarness({ connected: false, storage }); + // Store calls run in order, so the other tab's first read is done after this one. + yield* other.store.update((stored) => stored); + yield* queued("settle", storage); + const before = storage.value; + const result = yield* Effect.promise(() => + other.commands.snooze.run(other.registry, snooze), + ); + expect(result._tag).toBe("Failure"); + expect(storage.value).toBe(before); + }), + ); + + it.effect("keeps the not-connected failure when the change cannot be saved", () => + Effect.gen(function* () { + const h = yield* makeHarness({ + connected: false, + storage: { value: null, failWrites: true }, + }); + const result = yield* Effect.promise(() => h.commands.settle.run(h.registry, settle)); + expect(String(result._tag === "Failure" ? Cause.squash(result.cause) : "")).toContain( + "Mac is not connected.", + ); + expect(h.registry.get(h.pendingSyncAtom).size).toBe(0); + }), + ); +}); diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 865f6273b20ad..82012d21099d1 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -7,9 +7,12 @@ import { WS_METHODS, type EnvironmentId, type OrchestrationV2ShellSnapshot, + type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import { createOptimisticThreadLifecycle } from "./threadLifecycle.ts"; +import type { EnvironmentShellStatus } from "./shell.ts"; +import { createThreadTriageOutbox } from "./threadTriageOutbox.ts"; import * as DateTime from "effect/DateTime"; import { @@ -74,9 +77,53 @@ export type LoadEarlierThreadHistoryInput = { readonly workAnchorId?: string; }; +function applySettle( + thread: OrchestrationV2ThreadShell, + now: DateTime.Utc, + accepted: boolean, +): OrchestrationV2ThreadShell { + return !accepted && + (thread.pendingRuntimeRequest !== null || + ["preparing", "queued", "starting", "running", "waiting"].includes(thread.status)) + ? thread + : { + ...thread, + pendingRuntimeRequest: null, + settledOverride: "settled", + settledAt: thread.settledOverride === "settled" ? (thread.settledAt ?? now) : now, + unsettledAt: null, + activeOrderKey: null, + pinnedAt: null, + pinOrderKey: null, + snoozedAt: null, + snoozedUntil: null, + }; +} + +function applySnooze( + thread: OrchestrationV2ThreadShell, + snoozedUntil: string, + now: DateTime.Utc, + accepted: boolean, +): OrchestrationV2ThreadShell { + return (!accepted && + ((thread.pendingRuntimeRequest !== null && + thread.pendingRuntimeRequest.kind !== "user_input") || + ["preparing", "queued", "starting"].includes(thread.status))) || + !(Date.parse(snoozedUntil) > DateTime.toEpochMillis(now)) + ? thread + : { + ...thread, + snoozedUntil: DateTime.makeUnsafe(snoozedUntil), + snoozedAt: now, + }; +} + export function createThreadEnvironmentAtoms( runtime: Atom.AtomRuntime, snapshotAtom: (environmentId: EnvironmentId) => Atom.Atom, + /** Offline settle and snooze are sent once this reports a synchronized shell. */ + shellStatusAtom: (environmentId: EnvironmentId) => Atom.Atom, ) { const scheduler = createAtomCommandScheduler(); const concurrency = { @@ -434,7 +481,29 @@ export function createThreadEnvironmentAtoms( pin: rowAction(sent.pin), unpin: rowAction(sent.unpin), }; - const optimistic = createOptimisticThreadLifecycle(snapshotAtom); + const triage = createThreadTriageOutbox({ + runtime, + sourceSnapshotAtom: snapshotAtom, + shellStatusAtom, + apply: (thread, action, queuedAt) => + action.type === "settle" + ? applySettle(thread, queuedAt, false) + : applySnooze(thread, action.snoozedUntil, queuedAt, false), + send: (registry, { environmentId, threadId, commandId, action }) => + action.type === "settle" + ? settle.run(registry, { environmentId, input: { threadId, commandId } }) + : snooze.run(registry, { + environmentId, + input: { threadId, commandId, snoozedUntil: action.snoozedUntil }, + }), + }); + const optimistic = createOptimisticThreadLifecycle(triage.snapshotAtom); + const settle = optimistic.wrap(commands.settle, (thread, _input, now, accepted) => + applySettle(thread, now, accepted), + ); + const snooze = optimistic.wrap(commands.snooze, (thread, input, now, accepted) => + applySnooze(thread, input.snoozedUntil, now, accepted), + ); const disconnectedAtom = Atom.family((environmentId: EnvironmentId) => Atom.make((get) => { const connection = AsyncResult.getOrElse( @@ -471,48 +540,31 @@ export function createThreadEnvironmentAtoms( rowActionsAtom, snapshotAtom: optimistic.snapshotAtom, pendingThreadIdsAtom: optimistic.pendingThreadIdsAtom, - settle: optimistic.wrap(commands.settle, (thread, _input, now, accepted) => - !accepted && - (thread.pendingRuntimeRequest !== null || - ["preparing", "queued", "starting", "running", "waiting"].includes(thread.status)) - ? thread - : { - ...thread, - pendingRuntimeRequest: null, - settledOverride: "settled", - settledAt: thread.settledOverride === "settled" ? (thread.settledAt ?? now) : now, - unsettledAt: null, - activeOrderKey: null, - pinnedAt: null, - pinOrderKey: null, - snoozedAt: null, - snoozedUntil: null, - }, + pendingSyncAtom: triage.pendingSyncAtom, + settle: triage.queueWhenOffline(settle, () => ({ type: "settle" as const })), + unsettle: triage.cancelQueued( + optimistic.wrap(commands.unsettle, (thread, input, now) => ({ + ...thread, + settledOverride: input.reason === "user" ? "active" : null, + settledAt: null, + unsettledAt: thread.settledOverride === "active" ? (thread.unsettledAt ?? null) : now, + })), + "settle", + { sequence: 0, cancelledPendingSync: true as const }, ), - unsettle: optimistic.wrap(commands.unsettle, (thread, input, now) => ({ - ...thread, - settledOverride: input.reason === "user" ? "active" : null, - settledAt: null, - unsettledAt: thread.settledOverride === "active" ? (thread.unsettledAt ?? null) : now, + snooze: triage.queueWhenOffline(snooze, (input) => ({ + type: "snooze" as const, + snoozedUntil: input.snoozedUntil, })), - snooze: optimistic.wrap(commands.snooze, (thread, input, now, accepted) => - (!accepted && - ((thread.pendingRuntimeRequest !== null && - thread.pendingRuntimeRequest.kind !== "user_input") || - ["preparing", "queued", "starting"].includes(thread.status))) || - !(Date.parse(input.snoozedUntil) > DateTime.toEpochMillis(now)) - ? thread - : { - ...thread, - snoozedUntil: DateTime.makeUnsafe(input.snoozedUntil), - snoozedAt: now, - }, + unsnooze: triage.cancelQueued( + optimistic.wrap(commands.unsnooze, (thread) => ({ + ...thread, + snoozedUntil: null, + snoozedAt: null, + })), + "snooze", + { sequence: 0, cancelledPendingSync: true as const }, ), - unsnooze: optimistic.wrap(commands.unsnooze, (thread) => ({ - ...thread, - snoozedUntil: null, - snoozedAt: null, - })), setAutoSettle: optimistic.wrap(commands.setAutoSettle, (thread, input, now) => ({ ...thread, autoSettleDisabledAt: input.enabled ? null : (thread.autoSettleDisabledAt ?? now), diff --git a/packages/client-runtime/src/state/threadTriageOutbox.ts b/packages/client-runtime/src/state/threadTriageOutbox.ts new file mode 100644 index 0000000000000..7d035f94f560b --- /dev/null +++ b/packages/client-runtime/src/state/threadTriageOutbox.ts @@ -0,0 +1,372 @@ +import { + CommandId, + type EnvironmentId, + type OrchestrationV2ShellSnapshot, + type OrchestrationV2ThreadShell, + type ThreadId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Schema from "effect/Schema"; +import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity"; + +import { + type PendingThreadTriage, + type PendingThreadTriages, + ThreadTriageOutboxStore, + type ThreadTriageOutboxStorageError, +} from "../platform/threadTriageOutboxStore.ts"; +import { EnvironmentRpcUnavailableError, isRpcClientError } from "../rpc/client.ts"; +import type { EnvironmentShellStatus } from "./shell.ts"; +import { type AtomCommand, type AtomCommandResult, createRuntimeCommand } from "./runtime.ts"; + +type Target = { readonly environmentId: EnvironmentId; readonly input: I }; +type Change = (stored: PendingThreadTriages, newCommandId: CommandId) => PendingThreadTriages; + +const isUnavailable = Schema.is(EnvironmentRpcUnavailableError); + +/** Failures that say nothing about whether the server would accept the change. */ +function isTransient(cause: Cause.Cause) { + return ( + Cause.hasInterruptsOnly(cause) || + cause.reasons.some( + (reason) => + !Cause.isFailReason(reason) || + isUnavailable(reason.error) || + isRpcClientError(reason.error), + ) + ); +} + +/** + * What the user acted on: the thread's work, and every triage decision with its time, so a + * settle or snooze made and undone elsewhere still counts. A thread that shows otherwise on + * reconnect keeps its server state. + */ +function seenState(thread: OrchestrationV2ThreadShell) { + return JSON.stringify([ + thread.latestRunId, + thread.status, + thread.itemCount, + thread.pendingRuntimeRequest?.id ?? null, + thread.settledOverride, + ...[ + thread.settledAt, + thread.unsettledAt, + thread.snoozedUntil, + thread.snoozedAt, + thread.pinnedAt, + ].map((time) => (time == null ? null : DateTime.toEpochMillis(time))), + ]); +} + +const without = (entries: PendingThreadTriages, commandId: CommandId) => + entries.some((entry) => entry.commandId === commandId) + ? entries.filter((entry) => entry.commandId !== commandId) + : entries; + +/** The sent mark is left out: it is only ever read from storage. */ +const sameEntries = (a: PendingThreadTriages, b: PendingThreadTriages) => + a.length === b.length && a.every((entry, index) => entry.commandId === b[index]!.commandId); + +/** Raised before a request is sent, when the environment has no session. */ +const isOffline = (cause: Cause.Cause) => + cause.reasons.some((reason) => Cause.isFailReason(reason) && isUnavailable(reason.error)); + +/** + * Settle and snooze for an environment that is not connected: the change is stored on the + * device, shown at once, and sent when the environment is back. Storage is the truth, so tabs + * sharing it agree: memory mirrors what the last read or write returned, and a change is + * confirmed there right before it is sent. + */ +export function createThreadTriageOutbox(input: { + readonly runtime: Atom.AtomRuntime; + readonly sourceSnapshotAtom: ( + environmentId: EnvironmentId, + ) => Atom.Atom; + /** Queued changes are sent once this reports a synchronized shell. */ + readonly shellStatusAtom: (environmentId: EnvironmentId) => Atom.Atom; + /** Returns the thread itself when the action cannot apply to it. */ + readonly apply: ( + thread: OrchestrationV2ThreadShell, + action: PendingThreadTriage["action"], + queuedAt: DateTime.Utc, + ) => OrchestrationV2ThreadShell; + /** Sends the action under its command id, which the server answers only once. */ + readonly send: ( + registry: AtomRegistry.AtomRegistry, + entry: PendingThreadTriage, + ) => Promise>; +}) { + const update = createRuntimeCommand(input.runtime, { + label: "environment-data:thread-triage-outbox", + execute: Effect.fn("ThreadTriageOutbox.update")(function* (change: Change) { + const store = yield* ThreadTriageOutboxStore; + const random = yield* SecureRandom.SecureRandom; + const newCommandId = CommandId.make(yield* random.uuidV4()); + return yield* store + .update((stored) => change(stored, newCommandId)) + .pipe( + Effect.tapError(() => + Effect.logWarning("Could not read or save offline thread changes.", { + reason: "storage", + }), + ), + ); + }), + }); + + const entriesAtom = Atom.make(null).pipe(Atom.keepAlive); + /** Ends with the registry, so a replay in flight stops touching it. */ + const lifetimeAtom = Atom.make((get) => { + const lifetime = { ended: false }; + get.addFinalizer(() => { + lifetime.ended = true; + }); + return lifetime; + }).pipe(Atom.keepAlive); + + const write = async ( + registry: AtomRegistry.AtomRegistry, + change: Change = (stored) => stored, + ) => { + const lifetime = registry.get(lifetimeAtom); + const result = await update.run(registry, change); + if (AsyncResult.isSuccess(result) && !lifetime.ended) { + const current = registry.get(entriesAtom); + // An unchanged queue must not wake the replay that just read it. + if (current === null || !sameEntries(current, result.value)) { + registry.set(entriesAtom, result.value); + } + } + return result; + }; + const forget = (registry: AtomRegistry.AtomRegistry, entry: PendingThreadTriage) => + write(registry, (stored) => without(stored, entry.commandId)); + + let loading: Promise | undefined; + const load = (registry: AtomRegistry.AtomRegistry) => + (loading ??= write(registry).then((result) => { + // A failed read is tried again the next time the queue is looked at. + if (!AsyncResult.isSuccess(result)) loading = undefined; + })); + const environmentEntriesAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => { + const entries = get(entriesAtom); + if (entries === null) void load(get.registry); + return (entries ?? []).filter((entry) => entry.environmentId === environmentId); + }), + ); + const pendingFor = ( + registry: AtomRegistry.AtomRegistry, + environmentId: EnvironmentId, + threadId: ThreadId, + ) => + registry + .get(entriesAtom) + ?.find((entry) => entry.environmentId === environmentId && entry.threadId === threadId); + const sourceThread = ( + registry: AtomRegistry.AtomRegistry, + environmentId: EnvironmentId, + threadId: ThreadId, + ) => + registry + .get(input.sourceSnapshotAtom(environmentId)) + ?.threads.find((thread) => thread.id === threadId); + + const replaying = new Set(); + const replayAgain = new Set(); + const isLive = (registry: AtomRegistry.AtomRegistry, environmentId: EnvironmentId) => + registry.get(input.shellStatusAtom(environmentId)) === "live"; + const replay = async (registry: AtomRegistry.AtomRegistry, environmentId: EnvironmentId) => { + if (replaying.has(environmentId)) { + replayAgain.add(environmentId); + return; + } + const lifetime = registry.get(lifetimeAtom); + replaying.add(environmentId); + try { + await replayEntries(registry, environmentId, lifetime); + } finally { + replaying.delete(environmentId); + } + if (!lifetime.ended && replayAgain.delete(environmentId) && isLive(registry, environmentId)) { + await replay(registry, environmentId); + } + }; + const replayEntries = async ( + registry: AtomRegistry.AtomRegistry, + environmentId: EnvironmentId, + lifetime: { readonly ended: boolean }, + ) => { + // Read again: another tab may have queued a change this one has not seen. + const stored = await write(registry); + if (!AsyncResult.isSuccess(stored)) return; + for (const entry of stored.value) { + if (entry.environmentId !== environmentId) continue; + // A change is only marked as sent while it can be sent, so Undo can still cancel the rest. + if (lifetime.ended || !isLive(registry, environmentId)) return; + // Marked before sending, so an Undo after a lost reply or a restart asks the server. + let markedHere = false; + const marked = await write(registry, (current) => { + markedHere = current.some( + (item) => item.commandId === entry.commandId && item.attempted !== true, + ); + return markedHere + ? current.map((item) => + item.commandId === entry.commandId ? { ...item, attempted: true as const } : item, + ) + : current; + }); + if (lifetime.ended || !AsyncResult.isSuccess(marked)) return; + // Storage no longer has it: cancelled, replaced, sent by another tab, or its environment + // was removed. + if (!marked.value.some((item) => item.commandId === entry.commandId)) continue; + // A change that never left the device loses the mark this pass made, so Undo can cancel + // it. A mark from an earlier pass or another tab stays: that send may have arrived. + const unmark = async () => { + if (!markedHere) return; + await write(registry, (current) => + current.map(({ attempted, ...item }) => + item.commandId === entry.commandId || attempted !== true + ? item + : { ...item, attempted }, + ), + ); + }; + // Checked after the last wait, so the shell judged is the one the send goes through. + if (!isLive(registry, environmentId)) return unmark(); + const thread = sourceThread(registry, environmentId, entry.threadId); + // The live shell is the server's state. It wins over a change made against an older one, + // which also covers a change that was sent before but whose reply was lost. + if ( + thread === undefined || + seenState(thread) !== entry.seen || + input.apply(thread, entry.action, DateTime.nowUnsafe()) === thread + ) { + await forget(registry, entry); + continue; + } + const result = await input.send(registry, entry); + if (lifetime.ended) return; + if (!AsyncResult.isSuccess(result) && isTransient(result.cause)) { + return isOffline(result.cause) ? unmark() : undefined; + } + await forget(registry, entry); + } + }; + const replayAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => { + // Also runs when this tab's queue changes while the shell is synchronized. + get(environmentEntriesAtom(environmentId)); + if (get(input.shellStatusAtom(environmentId)) === "live") { + void replay(get.registry, environmentId); + } + }), + ); + + const snapshotAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => { + get.mount(replayAtom(environmentId)); + const snapshot = get(input.sourceSnapshotAtom(environmentId)); + const entries = get(environmentEntriesAtom(environmentId)); + if (snapshot === null || entries.length === 0) return snapshot; + const byThread = new Map(entries.map((entry) => [entry.threadId, entry])); + return { + ...snapshot, + threads: snapshot.threads.map((thread) => { + const entry = byThread.get(thread.id); + return entry === undefined ? thread : input.apply(thread, entry.action, entry.queuedAt); + }), + }; + }), + ); + + const pendingSyncAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make( + (get): ReadonlySet => + new Set(get(environmentEntriesAtom(environmentId)).map((entry) => entry.threadId)), + ), + ); + + /** Queues the action on the device when its environment is not connected. */ + function queueWhenOffline( + command: AtomCommand, { readonly sequence: number }, CE>, + toAction: (input: I) => PendingThreadTriage["action"], + ): typeof command { + return { + label: command.label, + run: async (registry, target) => { + const { environmentId } = target; + const { threadId } = target.input; + // Read before sending: the change is judged against what the user saw when acting. + const snapshot = registry.get(input.sourceSnapshotAtom(environmentId)); + const thread = snapshot?.threads.find((candidate) => candidate.id === threadId); + const result = await command.run(registry, target); + if (AsyncResult.isSuccess(result)) { + // The server now has a newer action than the queued one. + const previous = pendingFor(registry, environmentId, threadId); + if (previous !== undefined) await forget(registry, previous); + return result; + } + if (!isOffline(result.cause) || snapshot == null || thread === undefined) return result; + const action = toAction(target.input); + const queuedAt = DateTime.nowUnsafe(); + if (input.apply(thread, action, queuedAt) === thread) return result; + let refused = false; + const queued = await write(registry, (stored, commandId) => { + const same = (entry: PendingThreadTriage) => + entry.environmentId === environmentId && entry.threadId === threadId; + // One change per thread. A snooze cannot replace a settle, which online it keeps. + refused = action.type === "snooze" && stored.find(same)?.action.type === "settle"; + return refused + ? stored + : [ + ...stored.filter((entry) => !same(entry)), + { environmentId, threadId, commandId, action, queuedAt, seen: seenState(thread) }, + ]; + }); + return AsyncResult.isSuccess(queued) && !refused + ? AsyncResult.success({ sequence: snapshot.snapshotSequence }) + : result; + }, + }; + } + + /** + * Undo of a change that was never sent cancels it on the device. Once it was sent, even if the + * reply was lost, only the server can undo it. + */ + function cancelQueued( + command: AtomCommand, A, CE>, + cancels: PendingThreadTriage["action"]["type"], + cancelledValue: B, + ): AtomCommand, A | B, CE | ThreadTriageOutboxStorageError | E> { + return { + label: command.label, + run: async (registry, target) => { + const pending = pendingFor(registry, target.environmentId, target.input.threadId); + if (pending === undefined || pending.action.type !== cancels) { + return command.run(registry, target); + } + // Storage decides, because a replay, a restart or another tab may have sent it. + let unsent = false; + const stored = await write(registry, (current) => { + unsent = current.some( + (entry) => entry.commandId === pending.commandId && entry.attempted !== true, + ); + return unsent ? without(current, pending.commandId) : current; + }); + if (!AsyncResult.isSuccess(stored)) return AsyncResult.failure(stored.cause); + if (unsent) return AsyncResult.success(cancelledValue); + const result = await command.run(registry, target); + if (AsyncResult.isSuccess(result)) await forget(registry, pending); + return result; + }, + }; + } + + return { snapshotAtom, pendingSyncAtom, queueWhenOffline, cancelQueued }; +} diff --git a/packages/client-runtime/src/state/threadUndo.ts b/packages/client-runtime/src/state/threadUndo.ts index c9a8ac402bee5..7552afa9e81c9 100644 --- a/packages/client-runtime/src/state/threadUndo.ts +++ b/packages/client-runtime/src/state/threadUndo.ts @@ -65,6 +65,19 @@ export function planThreadUndo( return [{ action: "unsettle" }, ...(thread?.pinnedAt != null ? [pin] : []), ...snooze]; } +/** + * True when an Undo step cancelled a settle or snooze still waiting to sync. The server never + * saw the change, so the rest of the plan (re-pin, re-snooze) has nothing to restore. + */ +export function cancelledPendingSync(value: unknown): boolean { + return ( + typeof value === "object" && + value !== null && + "cancelledPendingSync" in value && + value.cancelledPendingSync === true + ); +} + export async function executeThreadUndo( plan: ThreadUndoPlan, execute: (step: ThreadUndoStep) => Promise,