diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index 192023c92..25e43ae68 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -606,6 +606,7 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer }); const canSend = hasContent && + !props.sessionInputBlocked && composerAuthority.providerAdmissionAvailable && props.projectCwd !== null && attachmentBlockReason === null && @@ -624,6 +625,9 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer activeSessionProviderStatus?.requiresNewThreadForModelChange === true); const getModelChangeDisabledReason = useCallback( (option: ModelOption) => { + if (props.sessionInputBlocked) { + return "Provider changes are blocked while this thread has a pending safety operation"; + } const boundInstanceId = props.selectedThread.session?.providerInstanceId; if (boundInstanceId) { const transition = resolveProviderContinuationTransition({ @@ -651,13 +655,21 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer ? "Start a new thread to use this model" : undefined; }, - [currentModelSelection, modelChangesLocked, props.selectedThread.session, props.serverConfig], + [ + currentModelSelection, + modelChangesLocked, + props.selectedThread.session, + props.serverConfig, + props.sessionInputBlocked, + ], ); - const quickQuestionAvailable = canOpenQuickQuestion({ - connectionState: props.connectionState, - session: props.selectedThread.session, - provider: activeSessionProviderStatus, - }); + const quickQuestionAvailable = + !props.sessionInputBlocked && + canOpenQuickQuestion({ + connectionState: props.connectionState, + session: props.selectedThread.session, + provider: activeSessionProviderStatus, + }); const quickQuestionScopeKey = quickQuestionSessionScopeKey({ environmentId: props.environmentId, threadId: props.selectedThread.id, @@ -824,12 +836,14 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer props.localOutboxCount === 0 && supportsSessionInputQueueFollowUp(activeSessionProviderStatus); const canClearSessionInputQueue = + !props.sessionInputBlocked && props.connectionState === "connected" && props.selectedThread.session?.status === "running" && props.selectedThread.session.activeTurnId != null && sessionQueueCount > 0 && supportsSessionInputQueueClear(activeSessionProviderStatus); const canRemoveOnlySessionInputQueueItem = + !props.sessionInputBlocked && props.connectionState === "connected" && props.selectedThread.session?.status === "running" && props.selectedThread.session.activeTurnId != null && @@ -845,6 +859,7 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer sessionInputQueueMutation?.scopeKey === sessionInputQueueScopeKey; const canSetSessionInputQueueModes = showSessionInputQueueModes && + !props.sessionInputBlocked && props.connectionState === "connected" && composerAuthority.providerAdmissionAvailable && (props.selectedThread.session?.status === "ready" || @@ -1316,6 +1331,7 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer sessionResources: props.sessionResources, showInteractionModeToggle, hasThread: true, + enabled: !props.sessionInputBlocked, onChangeDraftMessage: props.onChangeDraftMessage, onUpdateInteractionMode: props.onUpdateInteractionMode, }); @@ -1543,12 +1559,19 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer environmentId: props.environmentId, providerGroups: threadProviderGroups, selectedModel: currentModelSelection, - onSelectModel: (option) => props.onUpdateModelSelection(option.selection), + onSelectModel: (option) => { + if (!props.sessionInputBlocked) props.onUpdateModelSelection(option.selection); + }, optionDescriptors: providerOptionDescriptors, - onUpdateOptionSelections: (options) => - props.onUpdateModelSelection({ ...currentModelSelection, options }), + onUpdateOptionSelections: (options) => { + if (!props.sessionInputBlocked) { + props.onUpdateModelSelection({ ...currentModelSelection, options }); + } + }, runtimeMode: currentRuntimeMode, - onUpdateRuntimeMode: props.onUpdateRuntimeMode, + onUpdateRuntimeMode: (mode) => { + if (!props.sessionInputBlocked) props.onUpdateRuntimeMode(mode); + }, getModelDisabledReason: getModelChangeDisabledReason, }), [ @@ -1558,6 +1581,7 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer getModelChangeDisabledReason, props.onUpdateModelSelection, props.onUpdateRuntimeMode, + props.sessionInputBlocked, providerOptionDescriptors, settingsOwnerId, threadProviderGroups, @@ -1866,6 +1890,12 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer } label={currentModelOption?.label ?? currentModelSelection.model} maxWidth={152} + disabled={props.sessionInputBlocked} + accessibilityHint={ + props.sessionInputBlocked + ? "Provider changes are blocked while this thread has a pending safety operation" + : undefined + } onPress={openSettings} /> {sessionHarnessRefinementActions.length > 0 ? ( diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index cef0b2d0a..a58347f0e 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -11,6 +11,8 @@ import { type CodexArtifactTemplate, } from "@t3tools/client-runtime/codex-artifact-templates"; import type { EnvironmentThreadStatus } from "@t3tools/client-runtime/state/threads"; +import { isRollbackActive, type RollbackTarget } from "@t3tools/client-runtime/rollback"; +import { getMobileRollbackStatusPresentation } from "./rollback-status-presentation"; import { useKeyboardChatComposerInset, useKeyboardScrollToEnd } from "@legendapp/list/keyboard"; import type { LegendListRef } from "@legendapp/list/react-native"; import { HeaderHeightContext } from "@react-navigation/elements"; @@ -20,6 +22,7 @@ import type { MessageId, ModelSelection, OrchestrationThreadShell, + OrchestrationRollbackStatus, ProviderApprovalDecision, ProviderAskSessionSideQuestionResult, ProviderInteractionMode, @@ -47,6 +50,8 @@ import { AppState, Keyboard, Platform, + Pressable, + Text, useWindowDimensions, View, type GestureResponderEvent, @@ -144,6 +149,14 @@ export interface ThreadDetailScreenProps { readonly threadSyncStatus?: EnvironmentThreadStatus; /** Non-null when older turns exist beyond the loaded window. */ readonly loadEarlier?: { readonly loading: boolean; readonly onLoadEarlier: () => void } | null; + readonly rollbackStatus: OrchestrationRollbackStatus | null | undefined; + readonly rollbackTargets: ReadonlyMap; + readonly rollbackTargetIdle: boolean; + readonly rollbackCommandPending: boolean; + readonly onRevertMessage: (target: RollbackTarget) => void; + readonly onRecoverRollback: ( + action: "retry-verification" | "resume-compensation", + ) => Promise; readonly activeThreadBusy: boolean; readonly environmentId: EnvironmentId; readonly projectWorkspaceRoot: string | null; @@ -287,6 +300,56 @@ const USER_INPUT_TOGGLE_TIMING = { easing: Easing.out(Easing.cubic), }; +function RollbackStatusSurface(props: { + readonly status: OrchestrationRollbackStatus | null | undefined; + readonly pending: boolean; + readonly onRecover: (action: "retry-verification" | "resume-compensation") => Promise; +}) { + if (!props.status) return null; + const presentation = getMobileRollbackStatusPresentation(props.status); + const { severe, title, actions } = presentation; + return ( + + + {title} + + {presentation.detail} + {actions.length > 0 ? ( + + {actions.includes("retry-verification") ? ( + void props.onRecover("retry-verification")} + className="min-h-11 justify-center rounded-xl border border-adaptive-neutral-300-700 px-3 disabled:opacity-50" + > + Retry verification + + ) : null} + {actions.includes("resume-compensation") ? ( + void props.onRecover("resume-compensation")} + className="min-h-11 justify-center rounded-xl border border-adaptive-neutral-300-700 px-3 disabled:opacity-50" + > + Resume compensation + + ) : null} + + ) : null} + + ); +} + export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: ThreadDetailScreenProps) { const insets = useSafeAreaInsets(); const isKeyboardVisible = useKeyboardState((state) => state.isVisible); @@ -814,6 +877,10 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread onEndFollowEnabledChange={setEndFollowEnabled} skills={selectedProviderSkills} onUseArtifactTemplate={handleUseArtifactTemplate} + rollbackTargets={props.rollbackTargets} + rollbackTargetIdle={props.rollbackTargetIdle} + rollbackCommandPending={props.rollbackCommandPending} + onRevertMessage={props.onRevertMessage} loadEarlier={props.loadEarlier ?? null} /> @@ -861,6 +928,31 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread ) : null} + + + {props.localOutboxCount > 0 && + props.rollbackTargets.size > 0 && + !isRollbackActive(props.rollbackStatus) ? ( + + + Rollback paused for queued messages + + + Send or cancel the queued{" "} + {props.localOutboxCount === 1 ? "message" : "messages"} before starting + rollback. + + + ) : null} + + {props.activePendingApproval || props.activePendingUserInput || props.activePendingInteraction ? ( @@ -946,7 +1038,10 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread sessionInputBlocked={ props.activePendingApproval !== null || props.activePendingUserInput !== null || - props.activePendingInteraction !== null + props.activePendingInteraction !== null || + props.rollbackStatus?.state === "pending" || + props.rollbackStatus?.state === "recovering" || + props.rollbackStatus?.state === "manual-recovery" } environmentId={props.environmentId} projectCwd={props.projectWorkspaceRoot} diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index dc85cc376..5a3831086 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -23,6 +23,7 @@ import { classifyMarkdownImageSource, markdownImageSourceFragment, } from "@t3tools/client-runtime/markdown-images"; +import type { RollbackTarget } from "@t3tools/client-runtime/rollback"; import { resolveViewedImageAsset } from "@t3tools/client-runtime/work-log/presentation"; import { renderCodexFileCitationsAsMarkdown, @@ -230,6 +231,10 @@ export interface ThreadFeedProps { readonly onEndFollowEnabledChange?: (enabled: boolean) => void; readonly skills?: ReadonlyArray; readonly onUseArtifactTemplate?: (template: CodexArtifactTemplate) => void; + readonly rollbackTargets: ReadonlyMap; + readonly rollbackTargetIdle: boolean; + readonly rollbackCommandPending: boolean; + readonly onRevertMessage: (target: RollbackTarget) => void; /** Non-null when older turns exist beyond the loaded window. */ readonly loadEarlier?: { readonly loading: boolean; @@ -1449,7 +1454,16 @@ function useMarkdownStyles( function renderFeedEntry( info: { item: ThreadFeedEntry; index: number }, - props: Pick & { + props: Pick< + ThreadFeedProps, + | "environmentId" + | "onUseArtifactTemplate" + | "skills" + | "rollbackTargets" + | "rollbackTargetIdle" + | "rollbackCommandPending" + | "onRevertMessage" + > & { readonly copiedRowId: string | null; readonly expandedWorkRows: Record; readonly terminalAssistantMessageIds: ReadonlySet; @@ -1591,6 +1605,26 @@ function renderFeedEntry( {timestampLabel} + {props.rollbackTargetIdle && props.rollbackTargets.has(message.id) ? ( + { + const target = props.rollbackTargets.get(message.id); + if (target) props.onRevertMessage(target); + }} + className="size-11 items-center justify-center rounded-full disabled:opacity-50" + > + + + ) : null} {message.text.trim().length > 0 ? ( @@ -2665,6 +2703,10 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { props.environmentId, props.onUseArtifactTemplate, props.skills, + props.rollbackTargets, + props.rollbackTargetIdle, + props.rollbackCommandPending, + props.onRevertMessage, renderMarkdownImage, ], ); diff --git a/apps/mobile/src/features/threads/ThreadGitControls.tsx b/apps/mobile/src/features/threads/ThreadGitControls.tsx index 31b65f493..155046990 100644 --- a/apps/mobile/src/features/threads/ThreadGitControls.tsx +++ b/apps/mobile/src/features/threads/ThreadGitControls.tsx @@ -84,6 +84,7 @@ export type ThreadGitMenuProps = { readonly currentBranch: string | null; readonly gitStatus: VcsStatusResult | null; readonly gitOperationLabel: string | null; + readonly mutationBlocked?: boolean; readonly onOpenFilesInspector?: () => void; readonly onOpenGitInspector?: () => void; readonly onPull: () => Promise; @@ -113,7 +114,8 @@ function useThreadGitControlModel(props: ThreadGitMenuProps) { const { gitStatus, gitOperationLabel, onPull, onRunAction } = props; const currentBranchLabel = gitStatus?.refName ?? props.currentBranch ?? "Detached HEAD"; - const busy = gitOperationLabel !== null; + const mutationBlocked = props.mutationBlocked === true; + const busy = gitOperationLabel !== null || mutationBlocked; const isRepo = gitStatus?.isRepo ?? true; const hasPrimaryRemote = gitStatus?.hasPrimaryRemote ?? false; const isDefaultRef = gitStatus?.isDefaultRef ?? false; @@ -216,13 +218,15 @@ function useThreadGitControlModel(props: ThreadGitMenuProps) { }, [environmentId, props.onOpenFilesInspector, navigation, threadId]); const openReview = useCallback(() => { + if (mutationBlocked) return; navigation.navigate("ThreadReview", { environmentId: EnvironmentId.make(String(environmentId)), threadId: ThreadId.make(String(threadId)), }); - }, [environmentId, navigation, threadId]); + }, [environmentId, mutationBlocked, navigation, threadId]); const openGitInspector = useCallback(() => { + if (mutationBlocked) return; if (props.onOpenGitInspector) { props.onOpenGitInspector(); return; @@ -231,10 +235,11 @@ function useThreadGitControlModel(props: ThreadGitMenuProps) { environmentId: String(environmentId), threadId: String(threadId), }); - }, [environmentId, props.onOpenGitInspector, navigation, threadId]); + }, [environmentId, mutationBlocked, props.onOpenGitInspector, navigation, threadId]); return { currentBranchLabel, + mutationBlocked, isRepo, openFiles, openGitInspector, @@ -346,7 +351,7 @@ function useThreadGitHeaderActionItems(props: ThreadGitControlsProps): ThreadGit }, { description: "Turn diffs and worktree changes", - disabled: !model.isRepo, + disabled: !model.isRepo || model.mutationBlocked, icon: { name: "text.bubble", type: "sfSymbol" }, label: "Review changes", onPress: model.openReview, @@ -354,6 +359,7 @@ function useThreadGitHeaderActionItems(props: ThreadGitControlsProps): ThreadGit }, { description: "Commit, files, branches", + disabled: model.mutationBlocked, icon: { name: "ellipsis", type: "sfSymbol" }, label: "More", onPress: model.openGitInspector, @@ -373,6 +379,7 @@ function useThreadGitHeaderActionItems(props: ThreadGitControlsProps): ThreadGit model.openFiles, model.openGitInspector, model.openReview, + model.mutationBlocked, model.quickAction.disabled, model.quickAction.label, model.quickActionHint, @@ -524,7 +531,7 @@ export function ThreadGitMenu(props: ThreadGitMenuProps) { @@ -532,6 +539,7 @@ export function ThreadGitMenu(props: ThreadGitMenuProps) { diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 70de63b38..f1e7359c7 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -14,7 +14,17 @@ import { type ProviderAskSessionSideQuestionResult, type ProviderSessionSideQuestionRequestId, } from "@t3tools/contracts"; -import { isAtomCommandInterrupted } from "@t3tools/client-runtime/state/runtime"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; +import { + buildRollbackConfirmation, + deriveRollbackTargets, + isRollbackActive, + type RollbackTarget, +} from "@t3tools/client-runtime/rollback"; +import { resolveMobileRollbackStatus } from "./rollback-status-presentation"; import { requestOlderThreadTurns, threadHasOlderTurns, @@ -221,6 +231,15 @@ function ThreadRouteContent( const gitActions = useSelectedThreadGitActions(); const requests = useSelectedThreadRequests(); const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, "thread interrupt"); + const revertThreadCheckpoint = useAtomCommand(threadEnvironment.revertCheckpoint, { + label: "thread rollback", + reportFailure: false, + }); + const recoverThreadRollback = useAtomCommand(threadEnvironment.recoverRollback, { + label: "rollback recovery", + reportFailure: false, + }); + const [rollbackCommandPending, setRollbackCommandPending] = useState(false); const reloadThreadSessionResources = useAtomCommand( threadEnvironment.reloadSessionResources, "session resource reload", @@ -701,6 +720,86 @@ function ThreadRouteContent( terminalMenuSessions, ], ); + const rollbackTargets = useMemo( + () => + selectedThreadDetail === null + ? new Map() + : deriveRollbackTargets(selectedThreadDetail), + [selectedThreadDetail], + ); + const rollbackStatus = resolveMobileRollbackStatus( + selectedThreadDetail?.rollbackStatus, + selectedThread?.rollbackStatus, + ); + const rollbackActive = isRollbackActive(rollbackStatus); + const rollbackTargetIdle = + selectedThreadDetail?.session !== null && + selectedThreadDetail?.session !== undefined && + (selectedThreadDetail.session.status === "idle" || + selectedThreadDetail.session.status === "ready") && + selectedThreadDetail.session.activeTurnId === null && + selectedThreadDetail.session.pendingTurnRequestId === undefined && + selectedThreadDetail.session.activeTurnRequestId === undefined && + selectedThreadDetail.latestTurn?.state !== "running" && + !rollbackActive && + composer.selectedThreadQueueCount === 0 && + !composer.activeThreadBusy && + !rollbackCommandPending; + + const onRevertMessage = useCallback( + (target: RollbackTarget) => { + if (!selectedThread || !rollbackTargetIdle) return; + Alert.alert("Confirm exact rollback", buildRollbackConfirmation(target.label), [ + { text: "Cancel", style: "cancel" }, + { + text: "Revert", + style: "destructive", + onPress: () => { + setRollbackCommandPending(true); + void revertThreadCheckpoint({ + environmentId: selectedThread.environmentId, + input: { + threadId: selectedThread.id, + turnCount: target.targetTurnCount, + expectedSourceRevision: target.expectedSourceRevision, + }, + }).then((result) => { + setRollbackCommandPending(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const failure = squashAtomCommandFailure(result); + Alert.alert( + "Rollback unavailable", + failure instanceof Error ? failure.message : "Pylon rejected this rollback.", + ); + } + }); + }, + }, + ]); + }, + [revertThreadCheckpoint, rollbackTargetIdle, selectedThread], + ); + + const onRecoverRollback = useCallback( + async (action: "retry-verification" | "resume-compensation") => { + if (!selectedThread || rollbackCommandPending) return; + setRollbackCommandPending(true); + const result = await recoverThreadRollback({ + environmentId: selectedThread.environmentId, + input: { threadId: selectedThread.id, action }, + }); + setRollbackCommandPending(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const failure = squashAtomCommandFailure(result); + Alert.alert( + "Recovery could not resume", + failure instanceof Error ? failure.message : "Pylon rejected this recovery action.", + ); + } + }, + [recoverThreadRollback, rollbackCommandPending, selectedThread], + ); + const threadGitControlProps = { environmentId: environmentIdRaw ?? "", threadId: threadId ?? "", @@ -716,7 +815,10 @@ function ThreadRouteContent( onOpenGitInspector: fileInspector.supported ? handleOpenGitInspector : undefined, currentBranch: selectedThread?.branch ?? null, gitStatus: gitStatus.data, - gitOperationLabel: gitState.gitOperationLabel, + gitOperationLabel: rollbackActive + ? "Rollback verification in progress" + : gitState.gitOperationLabel, + mutationBlocked: rollbackActive, canOpenTerminal: Boolean(selectedThreadProject?.workspaceRoot), canOpenFiles: Boolean(selectedThreadProject?.workspaceRoot), projectScripts: selectedThreadProject?.scripts ?? [], @@ -796,11 +898,13 @@ function ThreadRouteContent( onPress: () => handleOpenTerminal(null), }); } - actions.push({ - accessibilityLabel: "Open git controls", - icon: "point.topleft.down.curvedto.point.bottomright.up", - onPress: handleOpenGitInspector, - }); + if (!rollbackActive) { + actions.push({ + accessibilityLabel: "Open git controls", + icon: "point.topleft.down.curvedto.point.bottomright.up", + onPress: handleOpenGitInspector, + }); + } if (fileInspector.supported && selectedThreadCwd !== null) { actions.push({ accessibilityLabel: "Toggle inspector", @@ -816,6 +920,7 @@ function ThreadRouteContent( handleOpenGitInspector, handleToggleInspector, props.onReturnToThread, + rollbackActive, selectedThreadCwd, selectedThreadProject?.workspaceRoot, ]); @@ -892,6 +997,12 @@ function ThreadRouteContent( connectionStateLabel={routeConnectionState} threadSyncStatus={selectedThreadDetailState.status} loadEarlier={loadEarlierTurns} + rollbackStatus={rollbackStatus} + rollbackTargets={rollbackTargets} + rollbackTargetIdle={rollbackTargetIdle} + rollbackCommandPending={rollbackCommandPending} + onRevertMessage={onRevertMessage} + onRecoverRollback={onRecoverRollback} activeThreadBusy={composer.activeThreadBusy} environmentId={selectedThread.environmentId} projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null} diff --git a/apps/mobile/src/features/threads/rollback-status-presentation.test.ts b/apps/mobile/src/features/threads/rollback-status-presentation.test.ts new file mode 100644 index 000000000..a3ab428d1 --- /dev/null +++ b/apps/mobile/src/features/threads/rollback-status-presentation.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from "vite-plus/test"; +import type { OrchestrationRollbackStatus } from "@t3tools/contracts"; + +import { + getMobileRollbackStatusPresentation, + resolveMobileRollbackStatus, +} from "./rollback-status-presentation"; + +const updatedAt = "2026-08-31T12:00:00.000Z"; + +describe("mobile rollback status presentation", () => { + it("keeps the durable detail status ahead of a stale shell status", () => { + const detail: OrchestrationRollbackStatus = { + state: "manual-recovery", + updatedAt, + detail: "Restore the provider transcript, then retry verification.", + allowedActions: ["retry-verification", "resume-compensation"], + }; + const shell: OrchestrationRollbackStatus = { state: "pending", updatedAt }; + expect(resolveMobileRollbackStatus(detail, shell)).toBe(detail); + expect(resolveMobileRollbackStatus(undefined, shell)).toBe(shell); + }); + + it("announces progress politely and manual recovery assertively with exact actions", () => { + expect(getMobileRollbackStatusPresentation({ state: "recovering", updatedAt })).toMatchObject({ + title: "Rollback recovering", + severe: false, + accessibilityRole: "summary", + accessibilityLiveRegion: "polite", + actions: [], + }); + + expect( + getMobileRollbackStatusPresentation({ + state: "manual-recovery", + updatedAt, + detail: "Manual repair is required.", + allowedActions: ["retry-verification", "resume-compensation"], + }), + ).toEqual({ + title: "Manual recovery required", + detail: "Manual repair is required.", + severe: true, + accessibilityRole: "alert", + accessibilityLiveRegion: "assertive", + actions: ["retry-verification", "resume-compensation"], + }); + }); +}); diff --git a/apps/mobile/src/features/threads/rollback-status-presentation.ts b/apps/mobile/src/features/threads/rollback-status-presentation.ts new file mode 100644 index 000000000..271e5e656 --- /dev/null +++ b/apps/mobile/src/features/threads/rollback-status-presentation.ts @@ -0,0 +1,41 @@ +import type { OrchestrationRollbackStatus } from "@t3tools/contracts"; + +export interface MobileRollbackStatusPresentation { + readonly title: string; + readonly detail: string; + readonly severe: boolean; + readonly accessibilityRole: "alert" | "summary"; + readonly accessibilityLiveRegion: "assertive" | "polite"; + readonly actions: ReadonlyArray<"retry-verification" | "resume-compensation">; +} + +export function resolveMobileRollbackStatus( + detailStatus: OrchestrationRollbackStatus | null | undefined, + shellStatus: OrchestrationRollbackStatus | null | undefined, +): OrchestrationRollbackStatus | null | undefined { + return detailStatus ?? shellStatus; +} + +export function getMobileRollbackStatusPresentation( + status: OrchestrationRollbackStatus, +): MobileRollbackStatusPresentation { + const severe = status.state === "manual-recovery" || status.state === "failed"; + const title = + status.state === "pending" + ? "Rollback pending" + : status.state === "recovering" + ? "Rollback recovering" + : status.state === "manual-recovery" + ? "Manual recovery required" + : status.state === "completed" + ? "Rollback completed" + : "Rollback failed safely"; + return { + title, + detail: status.detail ?? "Pylon is verifying rollback state.", + severe, + accessibilityRole: severe ? "alert" : "summary", + accessibilityLiveRegion: severe ? "assertive" : "polite", + actions: status.allowedActions ?? [], + }; +} diff --git a/apps/mobile/src/state/thread-outbox-model.ts b/apps/mobile/src/state/thread-outbox-model.ts index 999dae1fa..ad4892e8d 100644 --- a/apps/mobile/src/state/thread-outbox-model.ts +++ b/apps/mobile/src/state/thread-outbox-model.ts @@ -12,6 +12,7 @@ import { IsoDateTime, MessageId, ModelSelection, + NonNegativeInt, ProjectId, ProviderInteractionMode, RuntimeMode, @@ -29,7 +30,7 @@ import { DraftComposerAttachmentSchema } from "../lib/composer-image-schema"; import type { DraftComposerAttachment } from "../lib/composerImages"; import { scopedThreadKey } from "../lib/scopedEntities"; -const THREAD_OUTBOX_SCHEMA_VERSION = 6; +const THREAD_OUTBOX_SCHEMA_VERSION = 7; const THREAD_OUTBOX_MAX_RETRY_DELAY_MS = 16_000; const QueuedThreadCreationSchema = Schema.Struct({ @@ -51,15 +52,18 @@ const ThreadOutboxDeliveryHoldSchema = Schema.Struct({ "provider-binding-unresolved", "project-workspace-unavailable", "thread-missing", + "source-epoch-stale", "admission-rejected", ]), reason: Schema.String, boundInstanceId: Schema.optional(Schema.String), queuedInstanceId: Schema.optional(Schema.String), + queuedSourceEpoch: Schema.optional(NonNegativeInt), + currentSourceEpoch: Schema.optional(NonNegativeInt), }); export const QueuedThreadMessageSchema = Schema.Struct({ - schemaVersion: Schema.Literals([1, 2, 3, 4, 5, THREAD_OUTBOX_SCHEMA_VERSION]), + schemaVersion: Schema.Literals([1, 2, 3, 4, 5, 6, THREAD_OUTBOX_SCHEMA_VERSION]), environmentId: EnvironmentId, threadId: ThreadId, messageId: MessageId, @@ -69,6 +73,7 @@ export const QueuedThreadMessageSchema = Schema.Struct({ modelSelection: Schema.optional(ModelSelection), runtimeMode: Schema.optional(RuntimeMode), interactionMode: Schema.optional(ProviderInteractionMode), + sourceEpoch: Schema.optional(NonNegativeInt), deliveryHold: Schema.optional(ThreadOutboxDeliveryHoldSchema), // Present when the queued item creates a brand-new thread (pending task) // instead of appending a turn to an existing one. @@ -99,10 +104,13 @@ export interface ThreadOutboxDeliveryHold { | "provider-binding-unresolved" | "project-workspace-unavailable" | "thread-missing" + | "source-epoch-stale" | "admission-rejected"; readonly reason: string; readonly boundInstanceId?: string; readonly queuedInstanceId?: string; + readonly queuedSourceEpoch?: number; + readonly currentSourceEpoch?: number; } export interface QueuedThreadMessage { @@ -115,6 +123,7 @@ export interface QueuedThreadMessage { readonly modelSelection?: ModelSelectionType; readonly runtimeMode?: RuntimeModeType; readonly interactionMode?: ProviderInteractionModeType; + readonly sourceEpoch?: number; readonly deliveryHold?: ThreadOutboxDeliveryHold; readonly creation?: QueuedThreadCreation; readonly destination?: QueuedThreadCreation; @@ -125,6 +134,7 @@ export interface ThreadSettingsSnapshot { readonly modelSelection: ModelSelectionType; readonly runtimeMode: RuntimeModeType; readonly interactionMode: ProviderInteractionModeType; + readonly sourceEpoch?: number; readonly session?: { readonly providerInstanceId?: ModelSelectionType["instanceId"] | undefined; } | null; @@ -317,6 +327,7 @@ export function retryQueuedThreadMessage( readonly modelSelection?: ModelSelectionType; readonly runtimeMode?: RuntimeModeType; readonly interactionMode?: ProviderInteractionModeType; + readonly sourceEpoch?: number; }, ): QueuedThreadMessage { const { deliveryHold: _hold, ...retry } = message; @@ -327,6 +338,7 @@ export function retryQueuedThreadMessage( ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), ...(input.runtimeMode === undefined ? {} : { runtimeMode: input.runtimeMode }), ...(input.interactionMode === undefined ? {} : { interactionMode: input.interactionMode }), + ...(input.sourceEpoch === undefined ? {} : { sourceEpoch: input.sourceEpoch }), }; } @@ -464,6 +476,22 @@ export function resolveConfirmedThreadOutboxPlan(input: { }): ConfirmedThreadOutboxPlan { if (input.message.deliveryHold !== undefined) return { action: "wait" }; const creation = input.message.creation; + if (creation === undefined && input.thread != null) { + const queuedSourceEpoch = input.message.sourceEpoch ?? 0; + const currentSourceEpoch = input.thread.sourceEpoch ?? 0; + if (queuedSourceEpoch !== currentSourceEpoch) { + return { + action: "hold", + hold: { + kind: "source-epoch-stale", + reason: + "This message was composed before the thread was rolled back. Review it and explicitly reconfirm before sending.", + queuedSourceEpoch, + currentSourceEpoch, + }, + }; + } + } if (creation === undefined && input.thread == null) { if (input.shellStatus !== "live") return { action: "wait" }; return { @@ -601,6 +629,36 @@ export function shouldRetryThreadOutboxDelivery(error: unknown): boolean { return isTransportConnectionErrorMessage(errorMessage(error)); } +export function sourceEpochMismatchHold(error: unknown): ThreadOutboxDeliveryHold | null { + if ( + typeof error !== "object" || + error === null || + !("_tag" in error) || + error._tag !== "OrchestrationDispatchCommandError" || + !("reason" in error) || + error.reason !== "source-epoch-mismatch" + ) { + return null; + } + return { + kind: "source-epoch-stale", + reason: + "This message was composed before the thread was rolled back. Review it and explicitly reconfirm before sending.", + ...(typeof (error as unknown as { expectedSourceEpoch?: unknown }).expectedSourceEpoch === + "number" + ? { + queuedSourceEpoch: (error as unknown as { expectedSourceEpoch: number }) + .expectedSourceEpoch, + } + : {}), + ...(typeof (error as unknown as { actualSourceEpoch?: unknown }).actualSourceEpoch === "number" + ? { + currentSourceEpoch: (error as unknown as { actualSourceEpoch: number }).actualSourceEpoch, + } + : {}), + }; +} + export type ThreadOutboxCommandStage = "settings-sync" | "start-turn"; export type ThreadOutboxFailureAction = "retry" | "hold"; diff --git a/apps/mobile/src/state/thread-outbox.test.ts b/apps/mobile/src/state/thread-outbox.test.ts index 8daeb0677..d546fff49 100644 --- a/apps/mobile/src/state/thread-outbox.test.ts +++ b/apps/mobile/src/state/thread-outbox.test.ts @@ -28,6 +28,7 @@ import { retryQueuedThreadMessage, resolveQueuedThreadSettings, shouldRetryThreadOutboxDelivery, + sourceEpochMismatchHold, threadOutboxRetryDelayMs, type QueuedThreadMessage, } from "./thread-outbox-model"; @@ -1463,6 +1464,90 @@ describe("thread outbox", () => { ); }); + it("holds an offline turn across another client's rollback until explicit reconfirmation", () => { + const message = { + ...queuedMessage({ + messageId: "message-before-cross-device-rollback", + createdAt: "2026-06-08T10:00:01.000Z", + }), + sourceEpoch: 4, + }; + const thread = { + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + sourceEpoch: 5, + session: { status: "ready" as const }, + }; + const plan = (candidate: QueuedThreadMessage) => + resolveConfirmedThreadOutboxPlan({ + message: candidate, + thread, + shellStatus: "live", + environmentConnected: true, + providers: [provider({ instanceId: "codex" })], + project: null, + }); + + expect(plan(message)).toMatchObject({ + action: "hold", + hold: { + kind: "source-epoch-stale", + queuedSourceEpoch: 4, + currentSourceEpoch: 5, + }, + }); + + const ordinaryRetry = retryQueuedThreadMessage( + { + ...message, + deliveryHold: { + kind: "source-epoch-stale", + reason: "Review and reconfirm.", + queuedSourceEpoch: 4, + currentSourceEpoch: 5, + }, + }, + { + commandId: CommandId.make("retry-without-reconfirmation"), + createdAt: "2026-06-08T10:01:00.000Z", + }, + ); + expect(ordinaryRetry.sourceEpoch).toBe(4); + expect(plan(ordinaryRetry)).toMatchObject({ + action: "hold", + hold: { kind: "source-epoch-stale" }, + }); + + const reconfirmed = retryQueuedThreadMessage(ordinaryRetry, { + commandId: CommandId.make("retry-after-reconfirmation"), + createdAt: "2026-06-08T10:02:00.000Z", + sourceEpoch: 5, + }); + expect(reconfirmed.text).toBe(message.text); + expect(reconfirmed.attachments).toEqual(message.attachments); + expect(reconfirmed.sourceEpoch).toBe(5); + expect(plan(reconfirmed).action).toBe("send-existing"); + }); + + it("maps an atomic server epoch rejection to the durable review hold", () => { + expect( + sourceEpochMismatchHold({ + _tag: "OrchestrationDispatchCommandError", + reason: "source-epoch-mismatch", + expectedSourceEpoch: 7, + actualSourceEpoch: 8, + }), + ).toMatchObject({ + kind: "source-epoch-stale", + queuedSourceEpoch: 7, + currentSourceEpoch: 8, + }); + }); + it("quiesces after a cross-device delete is converted to a durable hold", () => { const destination = { projectId: ProjectId.make("project-cross-device-delete"), diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 318390d4f..f7bf0e980 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -700,6 +700,7 @@ export function useThreadComposerState() { modelSelection, runtimeMode, interactionMode, + sourceEpoch: selectedThreadShell.sourceEpoch ?? 0, destination: { projectId: selectedThreadShell.projectId, ...(selectedThreadProject?.title === undefined @@ -1330,6 +1331,7 @@ export function useThreadComposerState() { readonly modelSelection?: ModelSelection; readonly runtimeMode?: RuntimeMode; readonly interactionMode?: ProviderInteractionMode; + readonly sourceEpoch?: number; }) => { const metadata = makeQueuedMessageMetadata(); return retryQueuedThreadMessage(queuedMessage, { @@ -1396,7 +1398,31 @@ export function useThreadComposerState() { void recover({ draftKey: selectedThreadKey, startNewThread: false }); }, }); - if (selectedCompatibleSelection !== null) { + if (hold.kind === "source-epoch-stale") { + actions.push({ + text: "Review and reconfirm", + onPress: () => { + void updateThreadOutboxMessageIfCurrent( + queuedMessage, + freshRetry({ sourceEpoch: selectedThreadShell?.sourceEpoch ?? 0 }), + ) + .then((updated) => { + if (!updated) { + Alert.alert( + "Pending send changed", + "A newer queued copy was kept. Open Manage again to review it.", + ); + } + }) + .catch((error: unknown) => { + Alert.alert( + "Could not reconfirm pending send", + error instanceof Error ? error.message : "The original hold was kept.", + ); + }); + }, + }); + } else if (selectedCompatibleSelection !== null) { const selectedProviderName = selectedThreadServerConfig?.providers.find( (provider) => provider.instanceId === selectedCompatibleSelection.instanceId, diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index a6503b021..af7e40966 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -37,6 +37,7 @@ import { resolveThreadOutboxFailureAction, shouldRetryThreadOutboxDelivery, threadOutboxDeliveryHoldsEqual, + sourceEpochMismatchHold, threadOutboxRetryDelayMs, type QueuedThreadCreation, type QueuedThreadMessage, @@ -592,7 +593,11 @@ export function useThreadOutboxDrain(): void { const reportFailure = ( commandResult: AtomCommandResult, stage: ThreadOutboxCommandStage, - ): { readonly action: "retry" | "hold"; readonly message: string } | null => { + ): { + readonly action: "retry" | "hold"; + readonly message: string; + readonly hold: QueuedThreadMessage["deliveryHold"]; + } | null => { if (!AsyncResult.isFailure(commandResult)) { return null; } @@ -610,8 +615,10 @@ export function useThreadOutboxDrain(): void { cause: commandResult.cause, action, }); + const epochHold = stage === "start-turn" ? sourceEpochMismatchHold(error) : null; return { action, + hold: epochHold ?? undefined, message: error instanceof Error ? error.message @@ -624,12 +631,13 @@ export function useThreadOutboxDrain(): void { message: QueuedThreadMessage, expectedRevision: number, reason: string, + hold?: QueuedThreadMessage["deliveryHold"], ): Promise<"held" | "retry" | "complete"> => { try { const updated = await updateThreadOutboxMessage( { ...message, - deliveryHold: { + deliveryHold: hold ?? { kind: "admission-rejected", reason, ...(message.modelSelection === undefined @@ -709,6 +717,7 @@ export function useThreadOutboxDrain(): void { modelSelection: settings.modelSelection, runtimeMode: settings.runtimeMode, interactionMode: settings.interactionMode, + sourceEpoch: queuedMessage.sourceEpoch, createdAt: queuedMessage.createdAt, }, }); @@ -717,7 +726,12 @@ export function useThreadOutboxDrain(): void { return "retry"; } if (failure?.action === "hold") { - return persistRejectedAdmissionHold(persistedMessage, deliveryRevision, failure.message); + return persistRejectedAdmissionHold( + persistedMessage, + deliveryRevision, + failure.message, + failure.hold, + ); } acknowledgedExistingThreadMessageIdsRef.current.add(persistedMessage.messageId); @@ -808,7 +822,12 @@ export function useThreadOutboxDrain(): void { return "retry"; } if (failure?.action === "hold") { - return persistRejectedAdmissionHold(persistedMessage, deliveryRevision, failure.message); + return persistRejectedAdmissionHold( + persistedMessage, + deliveryRevision, + failure.message, + failure.hold, + ); } const outcome = await completeQueuedMessageDelivery(persistedMessage, deliveryRevision); diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 5e7f59acd..4770a7a56 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -50,6 +50,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.providerAbortSessionCompaction]: AuthOrchestrationOperateScope, [WS_METHODS.providerSetSessionAutoCompaction]: AuthOrchestrationOperateScope, [WS_METHODS.providerRefineSessionHarness]: AuthOrchestrationOperateScope, + [WS_METHODS.rollbackRecover]: AuthOrchestrationOperateScope, [WS_METHODS.serverUpdateProvider]: AuthOrchestrationOperateScope, [WS_METHODS.serverGetPrimeManagedMaintenance]: AuthOrchestrationReadScope, [WS_METHODS.serverRunPrimeManagedMaintenance]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index b649189b4..5210d6040 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -43,6 +43,15 @@ import { RollbackSagaRunner } from "../../rollback/RollbackSagaRunner.ts"; import { RollbackWorkspace } from "../../rollback/RollbackWorkspace.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); +const rollbackUnavailable = { + state: "unavailable" as const, + reason: + "Exact rollback requires an idle Pylon-managed native Prime session with a matching immutable checkpoint anchor.", +}; +const rollbackAvailable = { + state: "available" as const, + reason: "Pylon verified an exact native provider anchor for this immutable checkpoint.", +}; type ReactorInput = | { @@ -258,18 +267,18 @@ export const make = Effect.gen(function* () { Option.isNone(rollbackRepository) || Option.isNone(rollbackWorkspace) ) - return; + return rollbackUnavailable; const capabilities = yield* providerService .getCapabilities(session.value.providerInstanceId) .pipe(Effect.option); if (Option.isNone(capabilities) || capabilities.value.conversationRollback !== "absolute") - return; + return rollbackUnavailable; if ( providerService.hasAbsoluteConversationRollback === undefined || providerService.captureConversationAnchor === undefined || !(yield* providerService.hasAbsoluteConversationRollback(input.threadId)) ) - return; + return rollbackUnavailable; const checkpoint = yield* rollbackWorkspace.value.resolveCheckpoint({ cwd: input.cwd, checkpointRef: input.checkpointRef, @@ -298,6 +307,7 @@ export const make = Effect.gen(function* () { anchorDigest: anchor.digest, capturedAt: input.capturedAt, }); + return rollbackAvailable; }, ); @@ -340,14 +350,21 @@ export const make = Effect.gen(function* () { cwd: input.cwd, checkpointRef: targetCheckpointRef, }); - yield* capturePrivateCheckpointAnchor({ + const rollbackAvailability = yield* capturePrivateCheckpointAnchor({ threadId: input.threadId, cwd: input.cwd, checkpointTurnCount: input.turnCount, turnId: input.turnId, checkpointRef: targetCheckpointRef, capturedAt: input.createdAt, - }); + }).pipe( + Effect.catch(() => + Effect.logWarning("exact rollback anchor capture unavailable", { + threadId: input.threadId, + turnId: input.turnId, + }).pipe(Effect.as(rollbackUnavailable)), + ), + ); // Refresh the workspace entry index so the @-mention file picker // reflects files created or deleted during this turn. @@ -405,6 +422,7 @@ export const make = Effect.gen(function* () { files, ...(assistantMessageId === undefined ? {} : { assistantMessageId }), checkpointTurnCount: input.turnCount, + rollbackAvailability, createdAt: input.createdAt, }); yield* receiptBus.publish({ diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 78adee1f5..2fdc20419 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -900,6 +900,11 @@ describe("OrchestrationEngine", () => { status: "ready", files: [], assistantMessageId: null, + rollbackAvailability: { + state: "unavailable", + reason: + "Exact rollback requires an idle Pylon-managed native Prime session with a matching immutable checkpoint anchor.", + }, completedAt: createdAt, }, ]); diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts index feb957bce..0d65d6742 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts @@ -251,6 +251,44 @@ const makeOrchestrationEngine = Effect.gen(function* () { }); } + if ( + envelope.command.type === "thread.checkpoint.revert" && + Option.isSome(rollbackRepository) + ) { + const active = yield* rollbackRepository.value + .getActiveByThread(envelope.command.threadId) + .pipe( + Effect.mapError( + () => + new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: "Rollback admission state could not be read.", + }), + ), + ); + if (Option.isSome(active)) { + if ( + active.value.state.targetRevision !== envelope.command.turnCount || + active.value.state.sourceRevision !== envelope.command.expectedSourceRevision + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: envelope.command.type, + detail: "Another rollback target already owns this thread.", + }); + } + yield* commandReceiptRepository.upsert({ + commandId: envelope.command.commandId, + aggregateKind: "thread", + aggregateId: envelope.command.threadId, + acceptedAt: yield* nowIso, + resultSequence: commandReadModel.snapshotSequence, + status: "accepted", + error: null, + }); + return { sequence: commandReadModel.snapshotSequence }; + } + } + yield* assertRollbackFenceAllows(envelope.command); const eventBase = yield* decideOrchestrationCommand({ @@ -288,6 +326,11 @@ const makeOrchestrationEngine = Effect.gen(function* () { ), threadId: preparedRollback.value.threadId, status: "pending", + targetTurnCount: preparedRollback.value.targetRevision, + sourceRevision: preparedRollback.value.sourceRevision, + detail: + "Rewriting the provider conversation, Pylon history, and workspace to the selected message.", + allowedActions: [], createdAt: preparedRollback.value.createdAt, }, readModel: commandReadModel, @@ -414,7 +457,11 @@ const makeOrchestrationEngine = Effect.gen(function* () { sequence: committedCommand.lastSequence, eventCount: committedCommand.committedEvents.length, }; - }).pipe(Effect.withSpan(`orchestration.command.${envelope.command.type}`)), + }).pipe(Effect.withSpan(`orchestration.command.${envelope.command.type}`), (processCommand) => + envelope.command.type === "thread.checkpoint.revert" && Option.isSome(rollbackRepository) + ? rollbackRepository.value.withMutationFence(processCommand) + : processCommand, + ), ).pipe( Effect.flatMap((exit) => Effect.gen(function* () { diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 12b5f8b63..aa88be424 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -646,6 +646,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti latestTurnId: null, rollbackStatus: null, rollbackUpdatedAt: null, + sourceEpoch: 0, createdAt: event.payload.createdAt, updatedAt: event.payload.updatedAt, archivedAt: null, @@ -1012,6 +1013,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti latestTurnId, rollbackStatus: null, rollbackUpdatedAt: event.occurredAt, + sourceEpoch: existingRow.value.sourceEpoch + 1, updatedAt: event.occurredAt, }); yield* refreshThreadShellSummary(event.payload.threadId); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index b6a03b487..760625517 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -329,6 +329,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { interactionMode: "default", runtimeMode: "full-access", rollbackStatus: null, + sourceEpoch: 0, branch: null, worktreePath: null, linkedPullRequest: { @@ -403,6 +404,11 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { status: "ready", files: [{ path: "README.md", kind: "modified", additions: 2, deletions: 1 }], assistantMessageId: asMessageId("message-1"), + rollbackAvailability: { + state: "unavailable", + reason: + "Exact rollback requires an idle Pylon-managed native Prime session with a matching immutable checkpoint anchor.", + }, completedAt: "2026-02-24T00:00:08.000Z", }, ], @@ -460,6 +466,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { interactionMode: "default", runtimeMode: "full-access", rollbackStatus: null, + sourceEpoch: 0, branch: null, worktreePath: null, linkedPullRequest: { @@ -955,7 +962,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); - it.effect("reads single-thread checkpoint context without hydrating unrelated threads", () => + it.effect("reads exact checkpoint availability and durable recovery state", () => Effect.gen(function* () { const snapshotQuery = yield* ProjectionSnapshotQuery; const sql = yield* SqlClient.SqlClient; @@ -963,6 +970,9 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { yield* sql`DELETE FROM projection_projects`; yield* sql`DELETE FROM projection_threads`; yield* sql`DELETE FROM projection_turns`; + yield* sql`DELETE FROM projection_thread_sessions WHERE thread_id = 'thread-context'`; + yield* sql`DELETE FROM rollback_checkpoint_anchors WHERE thread_id = 'thread-context'`; + yield* sql`DELETE FROM rollback_sagas WHERE thread_id = 'thread-context'`; yield* sql` INSERT INTO projection_projects ( @@ -1020,6 +1030,13 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ) `; + yield* sql` + UPDATE projection_threads + SET rollback_status = 'manual-recovery', + rollback_updated_at = '2026-03-02T00:00:06.000Z' + WHERE thread_id = 'thread-context' + `; + yield* sql` INSERT INTO projection_turns ( thread_id, @@ -1072,6 +1089,82 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ) `; + yield* sql` + INSERT INTO projection_thread_sessions ( + thread_id, + status, + provider_name, + provider_instance_id, + session_incarnation_id, + runtime_mode, + restored, + harness_refinement_status, + updated_at + ) VALUES ( + 'thread-context', + 'idle', + 'prime', + 'prime', + 'incarnation-current', + 'full-access', + 0, + 'available', + '2026-03-02T00:00:06.000Z' + ) + `; + yield* sql` + INSERT INTO rollback_checkpoint_anchors ( + thread_id, + checkpoint_turn_count, + turn_id, + source_revision, + provider_instance_id, + session_incarnation_id, + checkpoint_ref, + checkpoint_oid, + anchor_json, + anchor_digest, + captured_at + ) VALUES ( + 'thread-context', + 1, + 'turn-1', + 1, + 'prime', + 'incarnation-current', + 'checkpoint-a', + 'checkpoint-oid-a', + '{"kind":"prime-native-leaf","leafId":"PRIVATE_LEAF_CANARY"}', + 'private-anchor-digest', + '2026-03-02T00:00:04.000Z' + ) + `; + yield* sql` + INSERT INTO rollback_sagas ( + operation_id, + request_event_id, + thread_id, + project_id, + workspace_key, + phase, + terminal, + private_state_json, + created_at, + updated_at + ) VALUES ( + 'operation-context', + 'request-context', + 'thread-context', + 'project-context', + 'PRIVATE_WORKSPACE_CANARY', + 'manual-recovery', + 0, + '{"targetRevision":1,"sourceRevision":2,"lastErrorCode":"workspace-compensation-unproved","compensation":"manual","projectionCommitSequence":null,"privateLeaf":"PRIVATE_SAGA_CANARY"}', + '2026-03-02T00:00:05.000Z', + '2026-03-02T00:00:06.000Z' + ) + `; + const context = yield* snapshotQuery.getThreadCheckpointContext( ThreadId.make("thread-context"), ); @@ -1090,6 +1183,11 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { status: "ready", files: [], assistantMessageId: null, + rollbackAvailability: { + state: "available", + reason: + "Pylon verified an exact native provider anchor for this immutable checkpoint.", + }, completedAt: "2026-03-02T00:00:04.000Z", }, { @@ -1099,11 +1197,91 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { status: "ready", files: [], assistantMessageId: null, + rollbackAvailability: { + state: "unavailable", + reason: + "Exact rollback requires an idle Pylon-managed native Prime session with a matching immutable checkpoint anchor.", + }, completedAt: "2026-03-02T00:00:05.000Z", }, ], }); } + + const detail = yield* snapshotQuery.getThreadDetailById(ThreadId.make("thread-context")); + assert.equal(detail._tag, "Some"); + if (detail._tag === "Some") { + assert.deepEqual(detail.value.rollbackStatus, { + state: "manual-recovery", + targetTurnCount: 1, + sourceRevision: 2, + detail: + "The thread remains fenced because automatic rollback recovery could not be proved (workspace-compensation-unproved).", + allowedActions: ["resume-compensation"], + updatedAt: "2026-03-02T00:00:06.000Z", + }); + const publicStrings: string[] = []; + const collectPublicStrings = (value: unknown): void => { + if (typeof value === "string") { + publicStrings.push(value); + } else if (Array.isArray(value)) { + value.forEach(collectPublicStrings); + } else if (value !== null && typeof value === "object") { + Object.values(value).forEach(collectPublicStrings); + } + }; + collectPublicStrings(detail.value); + assert.notInclude(publicStrings, "PRIVATE_LEAF_CANARY"); + assert.notInclude(publicStrings, "PRIVATE_SAGA_CANARY"); + assert.notInclude(publicStrings, "private-anchor-digest"); + assert.notInclude(publicStrings, "PRIVATE_WORKSPACE_CANARY"); + } + + yield* sql` + UPDATE projection_threads + SET rollback_status = 'completed', + rollback_updated_at = '2026-03-02T00:00:06.000Z' + WHERE thread_id = 'thread-context' + `; + yield* sql` + UPDATE rollback_sagas + SET phase = 'complete', + terminal = 0, + updated_at = '2026-03-02T00:00:06.000Z' + WHERE operation_id = 'operation-context' + `; + const stillFencedDetail = yield* snapshotQuery.getThreadDetailById( + ThreadId.make("thread-context"), + ); + assert.equal(stillFencedDetail._tag, "Some"); + if (stillFencedDetail._tag === "Some") { + assert.equal(stillFencedDetail.value.rollbackStatus?.state, "recovering"); + } + + yield* sql` + UPDATE rollback_sagas + SET terminal = 1, + updated_at = '2026-03-02T00:00:07.000Z' + WHERE operation_id = 'operation-context' + `; + const terminalDetail = yield* snapshotQuery.getThreadDetailById( + ThreadId.make("thread-context"), + ); + assert.equal(terminalDetail._tag, "Some"); + if (terminalDetail._tag === "Some") { + assert.deepEqual(terminalDetail.value.rollbackStatus, { + state: "completed", + targetTurnCount: 1, + sourceRevision: 2, + detail: "Rollback completed and all rewritten state was verified.", + allowedActions: [], + updatedAt: "2026-03-02T00:00:07.000Z", + }); + } + + yield* sql`DELETE FROM projection_thread_sessions WHERE thread_id = 'thread-context'`; + yield* sql`DELETE FROM rollback_checkpoint_anchors WHERE thread_id = 'thread-context'`; + yield* sql`DELETE FROM rollback_sagas WHERE thread_id = 'thread-context'`; }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 2fcae92dd..3ae93ac88 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -83,6 +83,15 @@ const THREAD_DETAIL_ACTIVITY_LIMIT = 500; // Snapshot payloads are decoded and projected in small sequential batches so // one client read does not retain the raw payloads for the full activity window. const THREAD_DETAIL_ACTIVITY_PAYLOAD_BATCH_SIZE = 25; +const EXACT_ROLLBACK_AVAILABLE = { + state: "available" as const, + reason: "Pylon verified an exact native provider anchor for this immutable checkpoint.", +}; +const EXACT_ROLLBACK_UNAVAILABLE = { + state: "unavailable" as const, + reason: + "Exact rollback requires an idle Pylon-managed native Prime session with a matching immutable checkpoint anchor.", +}; const ProjectionProjectDbRowSchema = ProjectionProject.mapFields( Struct.assign({ defaultModelSelection: Schema.NullOr(Schema.fromJsonString(ModelSelection)), @@ -128,6 +137,7 @@ const ProjectionThreadActivityIdRowSchema = Schema.Struct({ const ProjectionCheckpointDbRowSchema = ProjectionCheckpoint.mapFields( Struct.assign({ files: Schema.fromJsonString(Schema.Array(OrchestrationCheckpointFile)), + rollbackAvailable: Schema.Number, }), ); const ProjectionLatestTurnDbRowSchema = Schema.Struct({ @@ -213,6 +223,16 @@ const ProjectionThreadCheckpointContextThreadRowSchema = Schema.Struct({ workspaceRoot: Schema.String, worktreePath: Schema.NullOr(Schema.String), }); +const RollbackPublicDbRowSchema = Schema.Struct({ + targetRevision: NonNegativeInt, + sourceRevision: NonNegativeInt, + phase: Schema.String, + terminal: Schema.Number, + lastErrorCode: Schema.NullOr(Schema.String), + compensation: Schema.String, + projectionCommitSequence: Schema.NullOr(NonNegativeInt), + updatedAt: Schema.String, +}); const FullThreadDiffContextLookupInput = Schema.Struct({ threadId: ThreadId, checkpointTurnCount: NonNegativeInt, @@ -378,6 +398,76 @@ function mapSessionRow( }; } +function rollbackDetail(state: string, lastErrorCode: string | null): string { + if (state === "pending") { + return "Rewriting the provider conversation, Pylon history, and workspace to the selected message."; + } + if (state === "recovering") { + return "Verifying the provider conversation, Pylon history, and workspace before releasing the thread."; + } + if (state === "completed") { + return "Rollback completed and all rewritten state was verified."; + } + if (state === "failed") { + return "Rollback did not complete. Pylon restored and verified the original provider conversation and workspace; no thread content was removed."; + } + return `The thread remains fenced because automatic rollback recovery could not be proved (${lastErrorCode ?? "verification unavailable"}).`; +} + +function mapRollbackStatus( + status: string | null, + updatedAt: string | null, + saga: Schema.Schema.Type | null, +) { + const sagaStatus = + saga === null + ? null + : saga.phase === "complete" && saga.terminal === 1 + ? "completed" + : saga.phase === "compensated" && saga.terminal === 1 + ? "failed" + : saga.phase === "manual-recovery" + ? "manual-recovery" + : saga.phase === "projection-committed" || + saga.phase === "cleanup-started" || + saga.phase === "complete" || + saga.phase === "compensated" || + saga.phase.startsWith("compensation-") + ? "recovering" + : "pending"; + const shouldPreferSaga = + sagaStatus !== null && + (status === null || + status === "completed" || + status === "failed" || + sagaStatus === "completed" || + sagaStatus === "failed" || + sagaStatus === "manual-recovery"); + const effectiveStatus = shouldPreferSaga ? sagaStatus : status; + const effectiveUpdatedAt = shouldPreferSaga ? saga?.updatedAt : updatedAt; + if (effectiveStatus === null || effectiveUpdatedAt == null) return null; + const actions = + effectiveStatus !== "manual-recovery" || saga === null + ? [] + : saga.projectionCommitSequence !== null + ? (["retry-verification"] as const) + : saga.compensation === "manual" + ? (["resume-compensation"] as const) + : []; + return { + state: effectiveStatus as "pending" | "recovering" | "manual-recovery" | "completed" | "failed", + updatedAt: effectiveUpdatedAt, + ...(saga === null + ? {} + : { + targetTurnCount: saga.targetRevision, + sourceRevision: saga.sourceRevision, + }), + detail: rollbackDetail(effectiveStatus, saga?.lastErrorCode ?? null), + allowedActions: [...actions], + }; +} + function mapProjectShellRow( row: Schema.Schema.Type, repositoryIdentity: OrchestrationProject["repositoryIdentity"], @@ -509,6 +599,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_turn_id AS "latestTurnId", rollback_status AS "rollbackStatus", rollback_updated_at AS "rollbackUpdatedAt", + source_epoch AS "sourceEpoch", created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt", @@ -550,6 +641,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_turn_id AS "latestTurnId", rollback_status AS "rollbackStatus", rollback_updated_at AS "rollbackUpdatedAt", + source_epoch AS "sourceEpoch", created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt", @@ -593,6 +685,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_turn_id AS "latestTurnId", rollback_status AS "rollbackStatus", rollback_updated_at AS "rollbackUpdatedAt", + source_epoch AS "sourceEpoch", created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt", @@ -840,7 +933,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { checkpoint_status AS "status", checkpoint_files_json AS "files", assistant_message_id AS "assistantMessageId", - completed_at AS "completedAt" + completed_at AS "completedAt", + EXISTS ( + SELECT 1 + FROM rollback_checkpoint_anchors AS rollback_anchor + INNER JOIN projection_thread_sessions AS rollback_session + ON rollback_session.thread_id = projection_turns.thread_id + WHERE rollback_anchor.thread_id = projection_turns.thread_id + AND rollback_anchor.checkpoint_turn_count = projection_turns.checkpoint_turn_count + AND rollback_anchor.provider_instance_id = rollback_session.provider_instance_id + AND rollback_anchor.session_incarnation_id = rollback_session.session_incarnation_id + ) AS "rollbackAvailable" FROM projection_turns WHERE checkpoint_turn_count IS NOT NULL ORDER BY thread_id ASC, checkpoint_turn_count ASC @@ -1115,6 +1218,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { latest_turn_id AS "latestTurnId", rollback_status AS "rollbackStatus", rollback_updated_at AS "rollbackUpdatedAt", + source_epoch AS "sourceEpoch", created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt", @@ -1141,6 +1245,27 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const getLatestRollbackPublicRowByThread = SqlSchema.findOneOption({ + Request: ThreadIdLookupInput, + Result: RollbackPublicDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + json_extract(private_state_json, '$.targetRevision') AS "targetRevision", + json_extract(private_state_json, '$.sourceRevision') AS "sourceRevision", + phase, + terminal, + json_extract(private_state_json, '$.lastErrorCode') AS "lastErrorCode", + json_extract(private_state_json, '$.compensation') AS "compensation", + json_extract(private_state_json, '$.projectionCommitSequence') AS "projectionCommitSequence", + updated_at AS "updatedAt" + FROM rollback_sagas + WHERE thread_id = ${threadId} + ORDER BY created_at DESC, operation_id DESC + LIMIT 1 + `, + }); + const listThreadMessageRowsByThread = SqlSchema.findAll({ Request: ThreadIdLookupInput, Result: ProjectionThreadMessageDbRowSchema, @@ -1379,7 +1504,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { checkpoint_status AS "status", checkpoint_files_json AS "files", assistant_message_id AS "assistantMessageId", - completed_at AS "completedAt" + completed_at AS "completedAt", + EXISTS ( + SELECT 1 + FROM rollback_checkpoint_anchors AS rollback_anchor + INNER JOIN projection_thread_sessions AS rollback_session + ON rollback_session.thread_id = projection_turns.thread_id + WHERE rollback_anchor.thread_id = projection_turns.thread_id + AND rollback_anchor.checkpoint_turn_count = projection_turns.checkpoint_turn_count + AND rollback_anchor.provider_instance_id = rollback_session.provider_instance_id + AND rollback_anchor.session_incarnation_id = rollback_session.session_incarnation_id + ) AS "rollbackAvailable" FROM projection_turns WHERE thread_id = ${threadId} AND checkpoint_turn_count IS NOT NULL @@ -1963,6 +2098,10 @@ pending_approval_requests AS ( files: row.files, assistantMessageId: row.assistantMessageId, completedAt: row.completedAt, + rollbackAvailability: + row.rollbackAvailable === 1 + ? EXACT_ROLLBACK_AVAILABLE + : EXACT_ROLLBACK_UNAVAILABLE, }); checkpointsByThread.set(row.threadId, threadCheckpoints); } @@ -2093,6 +2232,7 @@ pending_approval_requests AS ( row.rollbackStatus == null || row.rollbackUpdatedAt == null ? null : { state: row.rollbackStatus, updatedAt: row.rollbackUpdatedAt }, + sourceEpoch: row.sourceEpoch, createdAt: row.createdAt, updatedAt: row.updatedAt, archivedAt: row.archivedAt, @@ -2309,6 +2449,7 @@ pending_approval_requests AS ( row.rollbackStatus == null || row.rollbackUpdatedAt == null ? null : { state: row.rollbackStatus, updatedAt: row.rollbackUpdatedAt }, + sourceEpoch: row.sourceEpoch, createdAt: row.createdAt, updatedAt: row.updatedAt, archivedAt: row.archivedAt, @@ -2454,6 +2595,7 @@ pending_approval_requests AS ( row.rollbackStatus == null || row.rollbackUpdatedAt == null ? null : { state: row.rollbackStatus, updatedAt: row.rollbackUpdatedAt }, + sourceEpoch: row.sourceEpoch, createdAt: row.createdAt, updatedAt: row.updatedAt, archivedAt: row.archivedAt, @@ -2608,6 +2750,7 @@ pending_approval_requests AS ( row.rollbackStatus == null || row.rollbackUpdatedAt == null ? null : { state: row.rollbackStatus, updatedAt: row.rollbackUpdatedAt }, + sourceEpoch: row.sourceEpoch, createdAt: row.createdAt, updatedAt: row.updatedAt, archivedAt: row.archivedAt, @@ -2823,6 +2966,8 @@ pending_approval_requests AS ( files: row.files, assistantMessageId: row.assistantMessageId, completedAt: row.completedAt, + rollbackAvailability: + row.rollbackAvailable === 1 ? EXACT_ROLLBACK_AVAILABLE : EXACT_ROLLBACK_UNAVAILABLE, }), ), }); @@ -2910,6 +3055,7 @@ pending_approval_requests AS ( state: threadRow.value.rollbackStatus, updatedAt: threadRow.value.rollbackUpdatedAt, }, + sourceEpoch: threadRow.value.sourceEpoch, createdAt: threadRow.value.createdAt, updatedAt: threadRow.value.updatedAt, archivedAt: threadRow.value.archivedAt, @@ -3077,6 +3223,7 @@ pending_approval_requests AS ( checkpointRows, latestTurnRow, sessionRow, + latestRollbackRow, ] = yield* Effect.all([ getActiveThreadRowById({ threadId }).pipe( Effect.mapError( @@ -3130,6 +3277,14 @@ pending_approval_requests AS ( ), ), ), + getLatestRollbackPublicRowByThread({ threadId }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadDetailById:getRollback:query", + "ProjectionSnapshotQuery.getThreadDetailById:getRollback:decodeRow", + ), + ), + ), ]); if (Option.isNone(threadRow)) { @@ -3149,13 +3304,12 @@ pending_approval_requests AS ( ? {} : { linkedPullRequest: threadRow.value.linkedPullRequest }), latestTurn: Option.isSome(latestTurnRow) ? mapLatestTurn(latestTurnRow.value) : null, - rollbackStatus: - threadRow.value.rollbackStatus == null || threadRow.value.rollbackUpdatedAt == null - ? null - : { - state: threadRow.value.rollbackStatus, - updatedAt: threadRow.value.rollbackUpdatedAt, - }, + rollbackStatus: mapRollbackStatus( + threadRow.value.rollbackStatus ?? null, + threadRow.value.rollbackUpdatedAt ?? null, + Option.getOrNull(latestRollbackRow), + ), + sourceEpoch: threadRow.value.sourceEpoch, createdAt: threadRow.value.createdAt, updatedAt: threadRow.value.updatedAt, archivedAt: threadRow.value.archivedAt, @@ -3194,6 +3348,8 @@ pending_approval_requests AS ( files: row.files, assistantMessageId: row.assistantMessageId, completedAt: row.completedAt, + rollbackAvailability: + row.rollbackAvailable === 1 ? EXACT_ROLLBACK_AVAILABLE : EXACT_ROLLBACK_UNAVAILABLE, })), session: Option.isSome(sessionRow) ? mapSessionRow(sessionRow.value) : null, }; diff --git a/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts b/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts index 9beafed81..3566e81dc 100644 --- a/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts +++ b/apps/server/src/orchestration/Layers/RollbackAdmissionAtomic.test.ts @@ -152,6 +152,16 @@ layer("durable rollback admission", (it) => { createdAt: now, }); + const joined = yield* orchestration.dispatch({ + type: "thread.checkpoint.revert", + commandId: CommandId.make("command-atomic-revert-same-target"), + threadId, + turnCount: 1, + expectedSourceRevision: 2, + createdAt: now, + }); + assert.equal(joined.sequence, 5); + const fencedCommands = [ orchestration.dispatch({ type: "thread.turn.start", diff --git a/apps/server/src/orchestration/decider.sessionLifecycle.test.ts b/apps/server/src/orchestration/decider.sessionLifecycle.test.ts index f8c0ffd6d..60141d003 100644 --- a/apps/server/src/orchestration/decider.sessionLifecycle.test.ts +++ b/apps/server/src/orchestration/decider.sessionLifecycle.test.ts @@ -40,7 +40,7 @@ const makeSession = (overrides: Partial = {}): Orchestrati ...overrides, }); -const makeReadModel = (session: OrchestrationSession): OrchestrationReadModel => ({ +const makeReadModel = (session: OrchestrationSession, sourceEpoch = 0): OrchestrationReadModel => ({ snapshotSequence: 0, projects: [], threads: [ @@ -54,6 +54,7 @@ const makeReadModel = (session: OrchestrationSession): OrchestrationReadModel => branch: null, worktreePath: null, latestTurn: null, + sourceEpoch, createdAt: NOW, updatedAt: NOW, archivedAt: null, @@ -473,4 +474,108 @@ it.layer(NodeServices.layer)("session lifecycle CAS decider", (it) => { }); }), ); + + it.effect("rejects a turn composed against an older rollback source epoch", () => + Effect.gen(function* () { + const session = makeSession({ + status: "ready", + pendingTurnRequestId: undefined, + pendingTurnMessageId: undefined, + pendingTurnRequestedAt: undefined, + pendingTurnDeadlineAt: undefined, + pendingTurnSessionId: undefined, + }); + const command = { + type: "thread.turn.start" as const, + commandId: CommandId.make("command-stale-source-epoch"), + threadId: THREAD_ID, + message: { + messageId: MessageId.make("message-stale-source-epoch"), + role: "user" as const, + text: "Keep this unsent", + attachments: [], + }, + modelSelection: { instanceId: INSTANCE_ID, model: "gpt-5.4" }, + runtimeMode: "full-access" as const, + interactionMode: "default" as const, + sourceEpoch: 1, + createdAt: NOW, + }; + + const rejected = yield* decideOrchestrationCommand({ + command, + readModel: makeReadModel(session, 2), + }).pipe(Effect.result); + expect(rejected._tag).toBe("Failure"); + if (rejected._tag === "Failure") { + expect(rejected.failure.message).toContain( + "Thread source epoch mismatch: expected 1; actual 2.", + ); + } + + const accepted = yield* decideOrchestrationCommand({ + command: { + ...command, + commandId: CommandId.make("command-current-source-epoch"), + sourceEpoch: 2, + }, + readModel: makeReadModel(session, 2), + }); + const events = Array.isArray(accepted) ? accepted : [accepted]; + expect(events.map((event) => event.type)).toContain("thread.message-sent"); + expect( + events.find((event) => event.type === "thread.turn-start-requested")?.payload.sourceEpoch, + ).toBe(2); + }), + ); + + it.effect("rejects a forged turn start while rollback recovery is active", () => + Effect.gen(function* () { + const session = makeSession({ + status: "ready", + pendingTurnRequestId: undefined, + pendingTurnMessageId: undefined, + pendingTurnRequestedAt: undefined, + pendingTurnDeadlineAt: undefined, + pendingTurnSessionId: undefined, + }); + const readModel = makeReadModel(session, 2); + const rollingBack = { + ...readModel, + threads: readModel.threads.map((thread) => ({ + ...thread, + rollbackStatus: { + state: "recovering" as const, + updatedAt: NOW, + }, + })), + }; + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.turn.start", + commandId: CommandId.make("command-during-rollback"), + threadId: THREAD_ID, + message: { + messageId: MessageId.make("message-during-rollback"), + role: "user", + text: "Do not race rollback", + attachments: [], + }, + modelSelection: { instanceId: INSTANCE_ID, model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + sourceEpoch: 2, + createdAt: NOW, + }, + readModel: rollingBack, + }).pipe(Effect.result); + + expect(result._tag).toBe("Failure"); + if (result._tag === "Failure") { + expect(result.failure.message).toContain( + "cannot start a turn while rollback recovery is active", + ); + } + }), + ); }); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 244c99214..455784475 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -985,6 +985,24 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command, threadId: command.threadId, }); + if ( + targetThread.rollbackStatus?.state === "pending" || + targetThread.rollbackStatus?.state === "recovering" || + targetThread.rollbackStatus?.state === "manual-recovery" + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${command.threadId}' cannot start a turn while rollback recovery is active.`, + }); + } + const observedSourceEpoch = command.sourceEpoch ?? 0; + const actualSourceEpoch = targetThread.sourceEpoch ?? 0; + if (observedSourceEpoch !== actualSourceEpoch) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread source epoch mismatch: expected ${observedSourceEpoch}; actual ${actualSourceEpoch}.`, + }); + } const effectiveModelSelection = command.modelSelection ?? targetThread.modelSelection; const providerSettingsChanged = !Equal.equals(effectiveModelSelection, targetThread.modelSelection) || @@ -1080,6 +1098,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" runtimeMode: command.runtimeMode, interactionMode: command.interactionMode, ...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}), + sourceEpoch: actualSourceEpoch, admissionRequestedAt, admissionDeadlineAt, createdAt: command.createdAt, @@ -1879,6 +1898,9 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" files: command.files, assistantMessageId: command.assistantMessageId ?? null, completedAt: command.completedAt, + ...(command.rollbackAvailability === undefined + ? {} + : { rollbackAvailability: command.rollbackAvailability }), }, }; } @@ -1934,6 +1956,16 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" payload: { threadId: command.threadId, status: command.status, + ...(command.targetTurnCount === undefined + ? {} + : { targetTurnCount: command.targetTurnCount }), + ...(command.sourceRevision === undefined + ? {} + : { sourceRevision: command.sourceRevision }), + ...(command.detail === undefined ? {} : { detail: command.detail }), + ...(command.allowedActions === undefined + ? {} + : { allowedActions: command.allowedActions }), updatedAt: command.createdAt, }, }; diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index c877e26d5..8c4f44abc 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -87,6 +87,7 @@ describe("orchestration projector", () => { worktreePath: null, continuedFromThreadId: null, latestTurn: null, + sourceEpoch: 0, createdAt: now, updatedAt: now, archivedAt: null, @@ -703,6 +704,7 @@ describe("orchestration projector", () => { ).toEqual([{ id: "activity-1", turnId: "turn-1" }]); expect(thread?.checkpoints.map((checkpoint) => checkpoint.checkpointTurnCount)).toEqual([1]); expect(thread?.latestTurn?.turnId).toBe("turn-1"); + expect(thread?.sourceEpoch).toBe(1); }); it("does not fallback-retain messages tied to removed turn IDs", async () => { diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index a8609282f..9b469f109 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -299,6 +299,7 @@ export function projectEvent( worktreePath: payload.worktreePath, continuedFromThreadId: payload.continuedFromThreadId ?? null, latestTurn: null, + sourceEpoch: 0, createdAt: payload.createdAt, updatedAt: payload.updatedAt, archivedAt: null, @@ -683,6 +684,9 @@ export function projectEvent( files: payload.files, assistantMessageId: payload.assistantMessageId, completedAt: payload.completedAt, + ...(payload.rollbackAvailability === undefined + ? {} + : { rollbackAvailability: payload.rollbackAvailability }), }, event.type, "checkpoint", @@ -742,7 +746,20 @@ export function projectEvent( rollbackStatus: event.payload.status === null ? null - : { state: event.payload.status, updatedAt: event.payload.updatedAt }, + : { + state: event.payload.status, + updatedAt: event.payload.updatedAt, + ...(event.payload.targetTurnCount === undefined + ? {} + : { targetTurnCount: event.payload.targetTurnCount }), + ...(event.payload.sourceRevision === undefined + ? {} + : { sourceRevision: event.payload.sourceRevision }), + ...(event.payload.detail === undefined ? {} : { detail: event.payload.detail }), + ...(event.payload.allowedActions === undefined + ? {} + : { allowedActions: event.payload.allowedActions }), + }, updatedAt: event.payload.updatedAt, }), }); @@ -792,7 +809,17 @@ export function projectEvent( proposedPlans, activities, latestTurn, - rollbackStatus: null, + sourceEpoch: (thread.sourceEpoch ?? 0) + 1, + rollbackStatus: + thread.rollbackStatus === null || thread.rollbackStatus === undefined + ? null + : { + ...thread.rollbackStatus, + state: "recovering", + detail: + "Rollback committed. Pylon is verifying cleanup before the thread is released.", + updatedAt: event.occurredAt, + }, updatedAt: event.occurredAt, }), }; diff --git a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts index 70a034932..217bee604 100644 --- a/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts +++ b/apps/server/src/persistence/Layers/ProjectionRepositories.test.ts @@ -89,6 +89,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { branch: null, worktreePath: null, latestTurnId: null, + sourceEpoch: 0, createdAt: "2026-03-24T00:00:00.000Z", updatedAt: "2026-03-24T00:00:00.000Z", archivedAt: null, @@ -153,6 +154,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { branch: null, worktreePath: null, latestTurnId: null, + sourceEpoch: 0, createdAt: "2026-03-24T00:00:00.000Z", updatedAt: "2026-03-25T00:00:00.000Z", archivedAt: null, @@ -230,6 +232,7 @@ projectionRepositoriesLayer("Projection repositories", (it) => { worktreePath: null, linkedPullRequest, latestTurnId: null, + sourceEpoch: 0, createdAt: "2026-03-24T00:00:00.000Z", updatedAt: "2026-03-24T00:00:00.000Z", archivedAt: null, diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index dddd97f4c..136a8232a 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -44,6 +44,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_turn_id, rollback_status, rollback_updated_at, + source_epoch, created_at, updated_at, archived_at, @@ -76,6 +77,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.latestTurnId}, ${row.rollbackStatus ?? null}, ${row.rollbackUpdatedAt ?? null}, + ${row.sourceEpoch}, ${row.createdAt}, ${row.updatedAt}, ${row.archivedAt}, @@ -108,6 +110,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_turn_id = excluded.latest_turn_id, rollback_status = excluded.rollback_status, rollback_updated_at = excluded.rollback_updated_at, + source_epoch = excluded.source_epoch, created_at = excluded.created_at, updated_at = excluded.updated_at, archived_at = excluded.archived_at, @@ -147,6 +150,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_turn_id AS "latestTurnId", rollback_status AS "rollbackStatus", rollback_updated_at AS "rollbackUpdatedAt", + source_epoch AS "sourceEpoch", created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt", @@ -188,6 +192,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { latest_turn_id AS "latestTurnId", rollback_status AS "rollbackStatus", rollback_updated_at AS "rollbackUpdatedAt", + source_epoch AS "sourceEpoch", created_at AS "createdAt", updated_at AS "updatedAt", archived_at AS "archivedAt", diff --git a/apps/server/src/persistence/Layers/RollbackSagas.test.ts b/apps/server/src/persistence/Layers/RollbackSagas.test.ts index af247053f..5e88db3c0 100644 --- a/apps/server/src/persistence/Layers/RollbackSagas.test.ts +++ b/apps/server/src/persistence/Layers/RollbackSagas.test.ts @@ -89,6 +89,41 @@ layer("RollbackSagaRepository", (it) => { }), ); + it.effect("blocks only provider instances owned by nonterminal rollback sagas", () => + Effect.gen(function* () { + const repository = yield* RollbackSagaRepository; + const fencedThread = ThreadId.make("thread-provider-fence"); + yield* repository.admit( + makeState("operation-provider-fence", fencedThread, "workspace-provider-fence"), + ); + let ranBusyMutation = false; + + const blocked = yield* repository + .withProviderMutationFence( + Effect.succeed([providerInstanceId]), + Effect.sync(() => { + ranBusyMutation = true; + }), + ) + .pipe(Effect.result); + assert.equal(blocked._tag, "Failure"); + if (blocked._tag === "Failure") { + assert.equal(blocked.failure._tag, "ServerProviderMutationBusyError"); + assert.equal(blocked.failure.reason, "rollback-active"); + assert.deepEqual(blocked.failure.providerInstanceIds, [providerInstanceId]); + assert.includeMembers([...blocked.failure.threadIds], [fencedThread]); + } + assert.isFalse(ranBusyMutation); + + const unrelated = ProviderInstanceId.make("unrelated-provider"); + const allowed = yield* repository.withProviderMutationFence( + Effect.succeed([unrelated]), + Effect.succeed("updated"), + ); + assert.equal(allowed, "updated"); + }), + ); + it.effect( "uses owner and version CAS, clears stale startup owners, and releases the lease last", () => diff --git a/apps/server/src/persistence/Layers/RollbackSagas.ts b/apps/server/src/persistence/Layers/RollbackSagas.ts index 614bc5d64..cdbb31c59 100644 --- a/apps/server/src/persistence/Layers/RollbackSagas.ts +++ b/apps/server/src/persistence/Layers/RollbackSagas.ts @@ -2,10 +2,17 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Semaphore from "effect/Semaphore"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import type { SqlError } from "effect/unstable/sql/SqlError"; -import { NonNegativeInt, ProjectId, ThreadId, TurnId } from "@t3tools/contracts"; +import { + NonNegativeInt, + ProjectId, + ServerProviderMutationBusyError, + ThreadId, + TurnId, +} from "@t3tools/contracts"; import { PersistenceDecodeError, toPersistenceDecodeError, @@ -60,7 +67,8 @@ const decodeAnchor = Schema.decodeUnknownEffect(RollbackCheckpointAnchor); const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - let nonterminalFenceCache: ReadonlyArray | null = null; + const mutationFence = yield* Semaphore.make(1); + let nonterminalFenceCache: ReadonlyArray | null = null; const mapSagaRow = Effect.fn("RollbackSagaRepository.mapSagaRow")(function* ( row: typeof SagaDbRow.Type, @@ -185,6 +193,42 @@ const make = Effect.gen(function* () { const listNonterminalForFence: RollbackSagaRepositoryShape["listNonterminalForFence"] = () => nonterminalFenceCache === null ? listNonterminal() : Effect.succeed(nonterminalFenceCache); + const withMutationFence: RollbackSagaRepositoryShape["withMutationFence"] = (effect) => + mutationFence.withPermits(1)(effect); + + const withProviderMutationFence: RollbackSagaRepositoryShape["withProviderMutationFence"] = ( + providerInstanceIds, + effect, + ) => + mutationFence.withPermits(1)( + Effect.gen(function* () { + const requested = new Set(yield* providerInstanceIds); + if (requested.size === 0) return yield* effect; + const active = yield* listNonterminalForFence().pipe( + Effect.mapError( + (cause) => + new ServerProviderMutationBusyError({ + reason: "rollback-state-unavailable", + providerInstanceIds: [...requested], + threadIds: [], + cause, + }), + ), + ); + const conflicts = active.filter((record) => requested.has(record.state.providerInstanceId)); + if (conflicts.length > 0) { + return yield* new ServerProviderMutationBusyError({ + reason: "rollback-active", + providerInstanceIds: [ + ...new Set(conflicts.map((record) => record.state.providerInstanceId)), + ], + threadIds: [...new Set(conflicts.map((record) => record.threadId))], + }); + } + return yield* effect; + }), + ); + const clearOwnersForStartup: RollbackSagaRepositoryShape["clearOwnersForStartup"] = () => sql`UPDATE rollback_sagas SET owner_id = NULL WHERE terminal = 0`.pipe( Effect.asVoid, @@ -250,7 +294,7 @@ const make = Effect.gen(function* () { phase, terminal, owner_id AS "ownerId", version, private_state_json AS "privateStateJson", created_at AS "createdAt", updated_at AS "updatedAt" `); - if (rows.length === 0) return Option.none(); + if (rows.length === 0) return Option.none(); yield* sql`DELETE FROM rollback_workspace_leases WHERE operation_id = ${input.operationId}`; return Option.some(rows[0]!); }), @@ -391,6 +435,8 @@ const make = Effect.gen(function* () { ); return RollbackSagaRepository.of({ + withMutationFence, + withProviderMutationFence, admit, get, getByRequestEvent, diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index cc285c327..e7d35dd17 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -63,6 +63,7 @@ import Migration0048 from "./Migrations/048_ProjectionThreadSessionPendingTurnRe import Migration0049 from "./Migrations/049_ProjectionThreadSessionPendingStop.ts"; import Migration0050 from "./Migrations/050_PrimeAgentRecoveryLedger.ts"; import Migration0051 from "./Migrations/051_DurableRollbackSagas.ts"; +import Migration0052 from "./Migrations/052_ProjectionThreadsSourceEpoch.ts"; /** * Migration loader with all migrations defined inline. * @@ -148,6 +149,7 @@ export const migrationEntries = [ [49, "ProjectionThreadSessionPendingStop", Migration0049], [50, "PrimeAgentRecoveryLedger", Migration0050], [51, "DurableRollbackSagas", Migration0051], + [52, "ProjectionThreadsSourceEpoch", Migration0052], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/052_ProjectionThreadsSourceEpoch.ts b/apps/server/src/persistence/Migrations/052_ProjectionThreadsSourceEpoch.ts new file mode 100644 index 000000000..bac5ddfe3 --- /dev/null +++ b/apps/server/src/persistence/Migrations/052_ProjectionThreadsSourceEpoch.ts @@ -0,0 +1,13 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Effect from "effect/Effect"; + +/** Monotonic server-owned generation for turns composed before/after rollback. */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>`PRAGMA table_info(projection_threads)`; + if (!columns.some((column) => column.name === "source_epoch")) { + yield* sql.unsafe( + "ALTER TABLE projection_threads ADD COLUMN source_epoch INTEGER NOT NULL DEFAULT 0", + ); + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index d5ab5a0f0..1ce2d4a87 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -39,6 +39,7 @@ export const ProjectionThread = Schema.Struct({ latestTurnId: Schema.NullOr(TurnId), rollbackStatus: Schema.optional(Schema.NullOr(OrchestrationRollbackPublicState)), rollbackUpdatedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + sourceEpoch: NonNegativeInt, createdAt: IsoDateTime, updatedAt: IsoDateTime, archivedAt: Schema.NullOr(IsoDateTime), diff --git a/apps/server/src/persistence/Services/RollbackSagas.ts b/apps/server/src/persistence/Services/RollbackSagas.ts index 36ad5ef45..9d7bce142 100644 --- a/apps/server/src/persistence/Services/RollbackSagas.ts +++ b/apps/server/src/persistence/Services/RollbackSagas.ts @@ -9,6 +9,7 @@ import { ProjectId, ProviderInstanceId, RuntimeSessionId, + ServerProviderMutationBusyError, ThreadId, TurnId, } from "@t3tools/contracts"; @@ -105,6 +106,17 @@ export type RollbackCheckpointAnchor = typeof RollbackCheckpointAnchor.Type; export type RollbackSagaRepositoryError = PersistenceSqlError | PersistenceDecodeError; export interface RollbackSagaRepositoryShape { + /** + * Process-wide serialization point shared by rollback admission and provider + * mutations. Holding it closes the query/admit race between independent RPC + * clients without exposing private saga state. + */ + readonly withMutationFence: (effect: Effect.Effect) => Effect.Effect; + /** Run a provider mutation only when none of its instances owns a saga. */ + readonly withProviderMutationFence: ( + providerInstanceIds: Effect.Effect, IdError, IdContext>, + effect: Effect.Effect, + ) => Effect.Effect; readonly admit: (state: RollbackSagaState) => Effect.Effect; readonly get: ( operationId: string, diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index f997e6aa4..eed24a4c1 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -17,9 +17,11 @@ import type { } from "@t3tools/contracts"; import { ApprovalRequestId, + CheckpointRef, CommandId, EnvironmentId, EventId, + ProjectId, ProviderDriverKind, ProviderInstanceId, ProviderSessionStartInput, @@ -65,6 +67,8 @@ import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; +import { RollbackSagaRepositoryLive } from "../../persistence/Layers/RollbackSagas.ts"; +import { RollbackSagaRepository } from "../../persistence/Services/RollbackSagas.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import { makeSqlitePersistenceLive, @@ -536,6 +540,9 @@ function makeProviderServiceLayer() { const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); + const rollbackRepositoryLayer = RollbackSagaRepositoryLive.pipe( + Layer.provide(SqlitePersistenceMemory), + ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); const layer = it.layer( @@ -545,6 +552,7 @@ function makeProviderServiceLayer() { }).pipe( Layer.provide(providerAdapterLayer), Layer.provide(directoryLayer), + Layer.provide(rollbackRepositoryLayer), Layer.provide(defaultServerSettingsLayer), Layer.provide(serverConfigTestLayer), Layer.provideMerge(AnalyticsService.layerTest), @@ -556,7 +564,7 @@ function makeProviderServiceLayer() { ), ), directoryLayer, - + rollbackRepositoryLayer, runtimeRepositoryLayer, NodeServices.layer, ), @@ -1285,6 +1293,68 @@ routing.layer("ProviderServiceLive routing", (it) => { }), ); + it.effect("fences direct provider mutations while exact rollback is active", () => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + const repository = yield* RollbackSagaRepository; + const threadId = asThreadId("thread-provider-rollback-fence"); + const projectId = ProjectId.make("project-provider-rollback-fence"); + const now = "2026-01-01T00:00:00.000Z"; + yield* repository.admit({ + operationId: "operation-provider-rollback-fence", + requestEventId: "request-provider-rollback-fence", + threadId, + projectId, + workspaceKey: "workspace-provider-rollback-fence", + workspaceCwd: "/tmp/project-provider-rollback-fence", + sourceRevision: 2, + targetRevision: 1, + sourceTurnId: null, + targetTurnId: null, + sourceCheckpointRef: CheckpointRef.make("checkpoint-source"), + sourceCheckpointOid: "2".repeat(40), + targetCheckpointRef: CheckpointRef.make("checkpoint-target"), + targetCheckpointOid: "1".repeat(40), + targetCheckpointDigest: "target-workspace", + providerInstanceId: codexInstanceId, + sessionIncarnationId: RuntimeSessionId.make("session-provider-rollback-fence"), + phase: "source-anchor-capture-started", + attempt: 0, + lastErrorCode: null, + compensation: "none", + cleanup: "pending", + sourceAnchor: null, + sourceAnchorDigest: null, + desiredAnchor: { leafId: "private-target" }, + desiredAnchorDigest: "target-provider", + preimage: null, + workspaceReceiptDigest: null, + providerReceiptDigest: null, + projectionCommitSequence: null, + createdAt: now, + updatedAt: now, + }); + routing.codex.messageSessionAgent.mockClear(); + routing.codex.compactSession.mockClear(); + + const messageFailure = yield* Effect.flip( + provider.messageSessionAgent({ + threadId, + agentId: RuntimeTaskId.make("agent-provider-rollback-fence"), + message: "Do not cross the rollback fence", + }), + ); + const compactionFailure = yield* Effect.flip(provider.compactSession({ threadId })); + + assert.instanceOf(messageFailure, ProviderValidationError); + assert.include(messageFailure.issue, "fenced by an active rollback operation"); + assert.instanceOf(compactionFailure, ProviderValidationError); + assert.include(compactionFailure.issue, "fenced by an active rollback operation"); + assert.equal(routing.codex.messageSessionAgent.mock.calls.length, 0); + assert.equal(routing.codex.compactSession.mock.calls.length, 0); + }), + ); + it.effect("routes side questions once without recovering inactive sessions", () => Effect.gen(function* () { const provider = yield* ProviderService.ProviderService; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 7268e9d0a..ab8f47e0a 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -1153,6 +1153,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderSessionStartInput, payload: rawInput, }); + yield* assertNotRollbackFenced(threadId, "ProviderService.startSession"); const resolvedInstanceId = yield* requireBindingInstanceId( "ProviderService.startSession", @@ -1899,6 +1900,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderReloadSessionResourcesInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.reloadSessionResources"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.reloadSessionResources", @@ -1931,6 +1933,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderAskSessionSideQuestionInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.askSessionSideQuestion"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.askSessionSideQuestion", @@ -1963,6 +1966,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderCancelSessionSideQuestionInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.cancelSessionSideQuestion"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.cancelSessionSideQuestion", @@ -1995,6 +1999,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderCancelSessionAgentInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.cancelSessionAgent"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.cancelSessionAgent", @@ -2042,6 +2047,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderMessageSessionAgentInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.messageSessionAgent"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.messageSessionAgent", @@ -2146,6 +2152,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( issue: `Agent depth must be an integer from 0 to ${PROVIDER_SESSION_AGENT_DEPTH_MAX_SETTABLE}.`, }); } + yield* assertNotRollbackFenced(input.threadId, "ProviderService.setSessionAgentDepth"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.setSessionAgentDepth", @@ -2392,6 +2399,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderCompactSessionInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.compactSession"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.compactSession", @@ -2425,6 +2433,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderAbortSessionCompactionInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.abortSessionCompaction"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.abortSessionCompaction", @@ -2457,6 +2466,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderSetSessionAutoCompactionInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.setSessionAutoCompaction"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.setSessionAutoCompaction", @@ -2490,6 +2500,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( schema: ProviderRefineSessionHarnessInput, payload: rawInput, }); + yield* assertNotRollbackFenced(input.threadId, "ProviderService.refineSessionHarness"); const routed = yield* resolveRoutableSession({ threadId: input.threadId, operation: "ProviderService.refineSessionHarness", diff --git a/apps/server/src/rollback/RollbackSagaRunner.test.ts b/apps/server/src/rollback/RollbackSagaRunner.test.ts index 8f9434377..bdfe656b8 100644 --- a/apps/server/src/rollback/RollbackSagaRunner.test.ts +++ b/apps/server/src/rollback/RollbackSagaRunner.test.ts @@ -123,6 +123,8 @@ const makeEnvironment = ( }; const repository: RollbackSagaRepositoryShape = { + withMutationFence: (effect) => effect, + withProviderMutationFence: (_providerInstanceIds, effect) => effect, admit: () => Effect.void, get: (id) => Effect.succeed(id === operationId ? Option.some(record) : Option.none()), getByRequestEvent: (eventId) => @@ -403,6 +405,13 @@ it.effect("commits last, clears private state, and never publishes private canar (command) => command.type === "thread.revert.complete", ); assert.isAtLeast(completeIndex, 0); + const terminalStatus = snapshot.commands.findLast( + (command) => command.type === "thread.rollback.status.set", + ); + assert.equal(terminalStatus?.status, "completed"); + assert.equal(terminalStatus?.targetTurnCount, 1); + assert.equal(terminalStatus?.sourceRevision, 2); + assert.deepEqual(terminalStatus?.allowedActions, []); assert.isTrue( snapshot.runtimeReceipts.some( (receipt) => receipt.type === "rollback.saga.phase" && receipt.phase === "complete", @@ -494,6 +503,12 @@ it.effect("compensates workspace and provider when the provider stays at source" assert.equal(snapshot.providerDigest, "provider-source"); assert.equal(snapshot.projectionCommits, 0); assert.isFalse(snapshot.lease); + assert.isFalse(snapshot.commands.some((command) => command.type === "thread.revert.complete")); + const status = snapshot.commands.findLast( + (command) => command.type === "thread.rollback.status.set", + ); + assert.equal(status?.status, "failed"); + assert.include(status?.detail ?? "", "no thread content was removed"); }), ); @@ -514,7 +529,9 @@ it.effect( assert.isTrue( snapshot.commands.some( (command) => - command.type === "thread.rollback.status.set" && command.status === "manual-recovery", + command.type === "thread.rollback.status.set" && + command.status === "manual-recovery" && + command.allowedActions?.includes("resume-compensation"), ), ); @@ -527,6 +544,98 @@ it.effect( }), ); +it.effect("resumes server-authorized compensation and reports a safe durable failure", () => + Effect.gen(function* () { + const operationId = "operation-manual-resume-compensation"; + const environment = makeEnvironment(operationId, "wrong-target"); + const runner = yield* environment.makeRunner(); + yield* runner.run(operationId, false); + assert.equal(environment.snapshot().record.state.phase, "manual-recovery"); + + yield* runner.recover({ threadId, action: "resume-compensation" }); + const snapshot = environment.snapshot(); + assert.equal(snapshot.record.state.phase, "compensated"); + assert.isTrue(snapshot.record.terminal); + assert.isFalse(snapshot.lease); + assert.equal(snapshot.workspaceDigest, "workspace-source"); + assert.equal(snapshot.providerDigest, "provider-source"); + assert.isFalse(snapshot.commands.some((command) => command.type === "thread.revert.complete")); + const status = snapshot.commands.findLast( + (command) => command.type === "thread.rollback.status.set", + ); + assert.equal(status?.status, "failed"); + assert.deepEqual(status?.allowedActions, []); + }), +); + +it.effect("retries post-commit verification without committing projection twice", () => + Effect.gen(function* () { + const operationId = "operation-manual-retry-verification"; + const environment = makeEnvironment(operationId); + const interrupted = yield* environment.makeRunner("persisted:projection-committed"); + yield* runInterrupted(interrupted, operationId); + environment.setProviderDigest("provider-source"); + environment.setProviderMode("stayed-source"); + yield* environment.repository.clearOwnersForStartup(); + const recovering = yield* environment.makeRunner(); + yield* recovering.run(operationId, true); + const manual = environment.snapshot(); + assert.equal(manual.record.state.phase, "manual-recovery"); + const manualStatus = manual.commands.findLast( + (command) => command.type === "thread.rollback.status.set", + ); + assert.deepEqual(manualStatus?.allowedActions, ["retry-verification"]); + + environment.setProviderMode("success"); + yield* recovering.recover({ threadId, action: "retry-verification" }); + const complete = environment.snapshot(); + assert.equal(complete.record.state.phase, "complete"); + assert.isTrue(complete.record.terminal); + assert.equal(complete.projectionCommits, 1); + assert.equal(complete.providerDigest, "provider-target"); + assert.equal( + complete.commands.findLast((command) => command.type === "thread.rollback.status.set") + ?.status, + "completed", + ); + }), +); + +it.effect("rejects recovery actions that the durable phase does not authorize", () => + Effect.gen(function* () { + const environment = makeEnvironment("operation-action-not-allowed"); + const runner = yield* environment.makeRunner(); + const error = yield* runner.recover({ threadId, action: "retry-verification" }).pipe( + Effect.match({ + onFailure: (failure) => failure, + onSuccess: () => null, + }), + ); + assert.equal(error?.reason, "action-not-allowed"); + assert.equal(environment.snapshot().record.state.phase, "source-anchor-capture-started"); + }), +); + +it.effect("lets only one client claim a permitted recovery action", () => + Effect.gen(function* () { + const operationId = "operation-recovery-busy"; + const environment = makeEnvironment(operationId, "wrong-target"); + const runner = yield* environment.makeRunner(); + yield* runner.run(operationId, false); + const claimed = yield* environment.repository.claim(operationId, "other-client"); + assert.equal(claimed._tag, "Some"); + + const error = yield* runner.recover({ threadId, action: "resume-compensation" }).pipe( + Effect.match({ + onFailure: (failure) => failure, + onSuccess: () => null, + }), + ); + assert.equal(error?.reason, "operation-busy"); + assert.equal(environment.snapshot().record.state.phase, "manual-recovery"); + }), +); + it.effect("releases only the worker owner and retries post-commit cleanup idempotently", () => Effect.gen(function* () { const environment = makeEnvironment("operation-cleanup-retry", "success", 1); diff --git a/apps/server/src/rollback/RollbackSagaRunner.ts b/apps/server/src/rollback/RollbackSagaRunner.ts index 481f94f87..f17f1f8f0 100644 --- a/apps/server/src/rollback/RollbackSagaRunner.ts +++ b/apps/server/src/rollback/RollbackSagaRunner.ts @@ -1,4 +1,10 @@ -import { CommandId, type CheckpointRef } from "@t3tools/contracts"; +import { + CommandId, + OrchestrationRollbackRecoveryError, + type CheckpointRef, + type OrchestrationRollbackRecoveryAction, + type ThreadId, +} from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; @@ -26,6 +32,10 @@ export const RollbackFaultInjector = Context.Reference( export interface RollbackSagaRunnerShape { readonly run: (operationId: string, recovering: boolean) => Effect.Effect; + readonly recover: (input: { + readonly threadId: ThreadId; + readonly action: OrchestrationRollbackRecoveryAction; + }) => Effect.Effect; } export class RollbackSagaRunner extends Context.Service< RollbackSagaRunner, @@ -61,14 +71,40 @@ export const make = Effect.gen(function* () { const after = (label: string, operationId: string) => fault(label, operationId); const statusCommand = Effect.fn("RollbackSagaRunner.statusCommand")(function* ( state: RollbackSagaState, - status: "pending" | "recovering" | "manual-recovery" | null, + status: "pending" | "recovering" | "manual-recovery" | "completed" | "failed" | null, ) { const createdAt = yield* nowIso; + const allowedActions = + status !== "manual-recovery" + ? [] + : state.projectionCommitSequence !== null + ? (["retry-verification"] as const) + : state.compensation === "manual" + ? (["resume-compensation"] as const) + : []; + const detail = + status === "pending" + ? "Rewriting the provider conversation, Pylon history, and workspace to the selected message." + : status === "recovering" + ? "Verifying the provider conversation, Pylon history, and workspace before releasing the thread." + : status === "manual-recovery" + ? `The thread remains fenced because automatic rollback recovery could not be proved (${state.lastErrorCode ?? "verification unavailable"}).` + : status === "completed" + ? "Rollback completed and all rewritten state was verified." + : status === "failed" + ? "Rollback did not complete. Pylon restored and verified the original provider conversation and workspace; no thread content was removed." + : undefined; yield* engine.dispatch({ type: "thread.rollback.status.set", - commandId: CommandId.make(`server:rollback-status:${state.operationId}:${status ?? "clear"}`), + commandId: CommandId.make( + `server:rollback-status:${state.operationId}:${status ?? "clear"}:${state.phase}:${state.updatedAt}`, + ), threadId: state.threadId, status, + targetTurnCount: state.targetRevision, + sourceRevision: state.sourceRevision, + ...(detail === undefined ? {} : { detail }), + allowedActions: [...allowedActions], createdAt, }); }); @@ -123,6 +159,7 @@ export const make = Effect.gen(function* () { yield* publishPhase(released.value.state); yield* after(`persisted:${released.value.state.phase}`, released.value.operationId); } + return released; }); const compensate = Effect.fn("RollbackSagaRunner.compensate")(function* ( @@ -226,8 +263,13 @@ export const make = Effect.gen(function* () { preimage: null, updatedAt: yield* nowIso, }; - yield* statusCommand(terminalState, null).pipe(Effect.ignore); - yield* releaseTerminal(record, terminalState); + const persisted = yield* update(record, terminalState); + if (Option.isNone(persisted)) return; + const statusPublished = yield* statusCommand(persisted.value.state, "failed").pipe( + Effect.result, + ); + if (statusPublished._tag === "Failure") return; + yield* releaseTerminal(persisted.value, persisted.value.state); }); const step = Effect.fn("RollbackSagaRunner.step")(function* (initial: RollbackSagaRecord) { @@ -536,7 +578,13 @@ export const make = Effect.gen(function* () { preimage: null, updatedAt: yield* nowIso, }; - yield* releaseTerminal(record, terminalState); + const persisted = yield* update(record, terminalState); + if (Option.isNone(persisted)) return; + const statusPublished = yield* statusCommand(persisted.value.state, "completed").pipe( + Effect.result, + ); + if (statusPublished._tag === "Failure") return; + yield* releaseTerminal(persisted.value, persisted.value.state); return; } case "compensation-workspace-started": @@ -544,9 +592,21 @@ export const make = Effect.gen(function* () { case "compensation-provider-started": return yield* compensate(record, state.lastErrorCode ?? "reconcile-compensation"); case "manual-recovery": - case "compensated": - case "complete": return; + case "compensated": { + const statusPublished = yield* statusCommand(state, "failed").pipe(Effect.result); + if (statusPublished._tag === "Success") { + yield* releaseTerminal(record, state); + } + return; + } + case "complete": { + const statusPublished = yield* statusCommand(state, "completed").pipe(Effect.result); + if (statusPublished._tag === "Success") { + yield* releaseTerminal(record, state); + } + return; + } } } }); @@ -579,7 +639,94 @@ export const make = Effect.gen(function* () { }, ); - return RollbackSagaRunner.of({ run }); + const recoveryError = ( + reason: "not-found" | "action-not-allowed" | "operation-busy", + message: string, + ) => new OrchestrationRollbackRecoveryError({ reason, message }); + + const recover: RollbackSagaRunnerShape["recover"] = Effect.fn("RollbackSagaRunner.recover")( + function* (input) { + const active = yield* repository + .getActiveByThread(input.threadId) + .pipe( + Effect.mapError(() => + recoveryError("not-found", "Pylon could not read the fenced rollback operation."), + ), + ); + if (Option.isNone(active)) { + return yield* recoveryError( + "not-found", + "No fenced rollback operation exists for this thread.", + ); + } + const state = active.value.state; + const allowed = + state.phase === "manual-recovery" && + ((input.action === "retry-verification" && state.projectionCommitSequence !== null) || + (input.action === "resume-compensation" && + state.projectionCommitSequence === null && + state.compensation === "manual")); + if (!allowed) { + return yield* recoveryError( + "action-not-allowed", + "That recovery action is not safe for the rollback's current durable phase.", + ); + } + + const claimed = yield* repository + .claim(active.value.operationId, ownerId) + .pipe( + Effect.mapError(() => + recoveryError("operation-busy", "Another client is already recovering this rollback."), + ), + ); + if (Option.isNone(claimed)) { + return yield* recoveryError( + "operation-busy", + "Another client is already recovering this rollback.", + ); + } + const claimedState = claimed.value.state; + const nextState: RollbackSagaState = { + ...claimedState, + phase: + input.action === "retry-verification" + ? "projection-committed" + : "compensation-workspace-started", + attempt: 0, + compensation: input.action === "retry-verification" ? "none" : "required", + lastErrorCode: null, + updatedAt: yield* nowIso, + }; + const updated = yield* repository + .updateOwned({ + operationId: claimed.value.operationId, + ownerId, + expectedVersion: claimed.value.version, + state: nextState, + }) + .pipe( + Effect.mapError(() => + recoveryError( + "operation-busy", + "The rollback recovery phase changed on another client.", + ), + ), + ); + if (Option.isNone(updated)) { + yield* repository.releaseOwnerOwned(claimed.value.operationId, ownerId).pipe(Effect.ignore); + return yield* recoveryError( + "operation-busy", + "The rollback recovery phase changed on another client.", + ); + } + yield* publishPhase(updated.value.state); + yield* repository.releaseOwnerOwned(updated.value.operationId, ownerId).pipe(Effect.ignore); + yield* run(updated.value.operationId, true); + }, + ); + + return RollbackSagaRunner.of({ run, recover }); }); export const layer = Layer.effect(RollbackSagaRunner, make); diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 0d6c60151..3ab040ae9 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -29,6 +29,7 @@ import { ProjectId, ProviderDriverKind, ProviderInstanceId, + ServerProviderMutationBusyError, ResolvedKeybindingRule, ThreadId, TurnId, @@ -118,10 +119,15 @@ import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; -import { OrchestrationListenerCallbackError } from "./orchestration/Errors.ts"; +import { + OrchestrationCommandInvariantError, + OrchestrationListenerCallbackError, +} from "./orchestration/Errors.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { ThreadDeletionReactor } from "./orchestration/Services/ThreadDeletionReactor.ts"; +import { RollbackSagaRunner } from "./rollback/RollbackSagaRunner.ts"; import { SqlitePersistenceMemory } from "./persistence/Layers/Sqlite.ts"; +import { RollbackSagaRepository } from "./persistence/Services/RollbackSagas.ts"; import { PersistenceSqlError } from "./persistence/Errors.ts"; import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts"; import { @@ -442,6 +448,7 @@ const buildAppUnderTest = (options?: { environmentTheme?: Partial; providerRegistry?: Partial; providerService?: Partial; + rollbackSagaRepository?: Partial; serverSettings?: Partial; externalLauncher?: Partial; vcsDriver?: Partial; @@ -857,6 +864,30 @@ const buildAppUnderTest = (options?: { drainThrough: () => Effect.void, ...options?.layers?.threadDeletionReactor, }), + Layer.mock(RollbackSagaRunner)({ + run: () => Effect.void, + recover: () => Effect.void, + }), + Layer.mock(RollbackSagaRepository)({ + withMutationFence: (effect) => effect, + withProviderMutationFence: (_providerInstanceIds, effect) => effect, + admit: () => Effect.void, + get: () => Effect.succeed(Option.none()), + getByRequestEvent: () => Effect.succeed(Option.none()), + getActiveByThread: () => Effect.succeed(Option.none()), + listNonterminal: () => Effect.succeed([]), + listNonterminalForFence: () => Effect.succeed([]), + clearOwnersForStartup: () => Effect.void, + claim: () => Effect.succeed(Option.none()), + updateOwned: () => Effect.succeed(Option.none()), + releaseOwnerOwned: () => Effect.void, + releaseLeaseOwned: () => Effect.succeed(Option.none()), + findLeaseByWorkspace: () => Effect.succeed(Option.none()), + putCheckpointAnchor: () => Effect.void, + getCheckpointAnchor: () => Effect.succeed(Option.none()), + deleteCheckpointAnchorsAfter: () => Effect.void, + ...options?.layers?.rollbackSagaRepository, + }), ), ), Layer.provide( @@ -5221,6 +5252,63 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("returns a typed busy error for provider maintenance and provider settings races", () => + Effect.gen(function* () { + const busyInstanceId = ProviderInstanceId.make("primeAgent-work"); + const busyThreadId = ThreadId.make("thread-active-rollback"); + yield* buildAppUnderTest({ + layers: { + rollbackSagaRepository: { + withProviderMutationFence: (providerInstanceIds, _effect) => + providerInstanceIds.pipe( + Effect.flatMap( + (ids) => + new ServerProviderMutationBusyError({ + reason: "rollback-active", + providerInstanceIds: ids, + threadIds: [busyThreadId], + }), + ), + ), + }, + }, + }); + const wsUrl = yield* getWsServerUrl("/ws"); + + const maintenance = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[WS_METHODS.serverUpdateProvider]({ + provider: ProviderDriverKind.make("primeAgent"), + instanceId: busyInstanceId, + }).pipe(Effect.result), + ), + ); + assertTrue(maintenance._tag === "Failure"); + assert.strictEqual(maintenance.failure._tag, "ServerProviderMutationBusyError"); + if (maintenance.failure._tag === "ServerProviderMutationBusyError") { + assert.deepEqual(maintenance.failure.providerInstanceIds, [busyInstanceId]); + assert.deepEqual(maintenance.failure.threadIds, [busyThreadId]); + } + + const settings = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[WS_METHODS.serverUpdateSettings]({ + patch: { + providerInstances: { + [busyInstanceId]: { driver: ProviderDriverKind.make("primeAgent") }, + }, + }, + }).pipe(Effect.result), + ), + ); + assertTrue(settings._tag === "Failure"); + assert.strictEqual(settings.failure._tag, "ServerProviderMutationBusyError"); + if (settings.failure._tag === "ServerProviderMutationBusyError") { + assert.includeMembers([...settings.failure.providerInstanceIds], [busyInstanceId]); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("refreshes providers for each subscribeServerConfig connection", () => Effect.gen(function* () { const refreshCalls = yield* Ref.make(0); @@ -6769,6 +6857,52 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("returns structured source epoch mismatch fields across websocket RPC", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + dispatch: () => + Effect.fail( + new OrchestrationCommandInvariantError({ + commandType: "thread.turn.start", + detail: "Thread source epoch mismatch: expected 3; actual 4.", + }), + ), + }, + }, + }); + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make("command-stale-epoch-rpc"), + threadId: defaultThreadId, + message: { + messageId: MessageId.make("message-stale-epoch-rpc"), + role: "user", + text: "Preserve this queued content", + attachments: [], + }, + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + sourceEpoch: 3, + createdAt: "2026-01-01T00:00:00.000Z", + }).pipe(Effect.result), + ), + ); + assertTrue(result._tag === "Failure"); + assert.strictEqual(result.failure._tag, "OrchestrationDispatchCommandError"); + if (result.failure._tag === "OrchestrationDispatchCommandError") { + assert.strictEqual(result.failure.reason, "source-epoch-mismatch"); + assert.strictEqual(result.failure.expectedSourceEpoch, 3); + assert.strictEqual(result.failure.actualSourceEpoch, 4); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("routes websocket rpc orchestration shell snapshot errors", () => Effect.gen(function* () { const projectionError = new PersistenceSqlError({ @@ -7244,6 +7378,163 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("filters rollback-status replay for legacy clients without blocking later events", () => + Effect.gen(function* () { + const occurredAt = "2026-01-01T00:00:00.000Z"; + const rollbackEvent = { + sequence: 2, + eventId: EventId.make("event-replay-rollback-status"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.rollback-status-updated", + payload: { + threadId: defaultThreadId, + status: "recovering", + updatedAt: occurredAt, + }, + } satisfies Extract; + const messageEvent = { + sequence: 3, + eventId: EventId.make("event-after-rollback-status"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: { + threadId: defaultThreadId, + messageId: MessageId.make("message-after-rollback-status"), + role: "assistant", + text: "Compatible later event", + turnId: null, + streaming: false, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(3), + readEvents: () => Stream.make(rollbackEvent, messageEvent), + }, + }, + }); + const wsUrl = yield* getWsServerUrl("/ws"); + + const legacyItems = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 1, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + assert.deepEqual( + Array.from(legacyItems).map((item) => + item.kind === "event" ? [item.event.type, item.event.sequence] : [item.kind], + ), + [["thread.message-sent", 3], ["synchronized"]], + ); + + const currentItems = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 1, + requestCompletionMarker: true, + rollbackStatusEvents: true, + }).pipe(Stream.take(3), Stream.runCollect), + ), + ); + assert.deepEqual( + Array.from(currentItems).map((item) => + item.kind === "event" ? [item.event.type, item.event.sequence] : [item.kind], + ), + [["thread.rollback-status-updated", 2], ["thread.message-sent", 3], ["synchronized"]], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect( + "filters rollback-status live events for legacy clients while later sequences flow", + () => + Effect.gen(function* () { + const occurredAt = "2026-01-01T00:00:00.000Z"; + const rollbackEvent = { + sequence: 4, + eventId: EventId.make("event-live-rollback-status"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.rollback-status-updated", + payload: { threadId: defaultThreadId, status: "recovering", updatedAt: occurredAt }, + } satisfies Extract; + const laterEvent = { + sequence: 5, + eventId: EventId.make("event-live-after-rollback-status"), + aggregateKind: "thread", + aggregateId: defaultThreadId, + occurredAt, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: { + threadId: defaultThreadId, + messageId: MessageId.make("message-live-after-rollback-status"), + role: "assistant", + text: "Later live event", + turnId: null, + streaming: false, + createdAt: occurredAt, + updatedAt: occurredAt, + }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(3), + readEvents: () => Stream.empty, + streamDomainEvents: Stream.make(rollbackEvent, laterEvent), + }, + }, + }); + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 3, + requestCompletionMarker: true, + }).pipe(Stream.take(2), Stream.runCollect), + ), + ); + assert.deepEqual( + Array.from(items).map((item) => + item.kind === "event" ? [item.event.type, item.event.sequence] : [item.kind], + ), + [["synchronized"], ["thread.message-sent", 5]], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("subscribeThread bounds catch-up replay to the captured head", () => Effect.gen(function* () { let replayLimit: number | undefined; diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index bf0ba4afe..6a142e492 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -292,6 +292,10 @@ const PersistenceLayerLive = PrimeAgentRecoveryLedger.layer.pipe( Layer.provideMerge(SqlitePersistenceLayerLive), ); +const PersistenceAndRollbackLayerLive = RollbackSagaRepositoryLive.pipe( + Layer.provideMerge(PersistenceLayerLive), +); + const VcsDriverRegistryLayerLive = VcsDriverRegistry.layer.pipe( Layer.provide(VcsProjectConfig.layer), ); @@ -410,7 +414,8 @@ const RuntimeCoreDependenciesLive = ReactorLayerLive.pipe( Layer.provideMerge(VcsLayerLive), Layer.provideMerge(ProviderRuntimeLayerLive), Layer.provideMerge(Layer.mergeAll(TerminalLayerLive, PreviewLayerLive)), - Layer.provideMerge(PersistenceLayerLive), + // Provider/settings RPC mutations and rollback admission must share this fence. + Layer.provideMerge(PersistenceAndRollbackLayerLive), // Both read a user-owned file out of the state directory and stream changes // to clients; neither depends on the other. Layer.provideMerge(Layer.mergeAll(Keybindings.layer, EnvironmentTheme.layer)), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index b905da4a0..15adccf58 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3,6 +3,7 @@ import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as Equal from "effect/Equal"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -17,6 +18,7 @@ import { AuthSessionId, ClientSurface, CommandId, + defaultInstanceIdForDriver, type DiscoveredLocalServerList, EventId, type EditorId, @@ -35,6 +37,9 @@ import { OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, type ProjectId, + type ProviderInstanceId, + type ServerSettings as ContractServerSettings, + type ServerSettingsPatch, type ProjectEntriesFailure, type ProjectFileFailure, type ProjectFileOperation, @@ -44,6 +49,7 @@ import { ProjectSearchEntriesError, ProjectWriteFileError, ProviderUploadFeedbackError, + ProviderDriverKind, RelayClientInstallFailedError, type RelayClientInstallProgressEvent, type ServerSelfUpdateError, @@ -87,6 +93,7 @@ import { import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { ThreadDeletionReactor } from "./orchestration/Services/ThreadDeletionReactor.ts"; +import { RollbackSagaRepository } from "./persistence/Services/RollbackSagas.ts"; import { observeRpcEffect as instrumentRpcEffect, observeRpcStream as instrumentRpcStream, @@ -128,6 +135,7 @@ import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as GitWorkflowService from "./git/GitWorkflowService.ts"; +import { RollbackSagaRunner } from "./rollback/RollbackSagaRunner.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; @@ -312,6 +320,55 @@ function projectSetupScriptCompatibilityDetail( } } +function sourceEpochMismatchFromError( + error: unknown, +): { readonly expectedSourceEpoch: number; readonly actualSourceEpoch: number } | null { + if (typeof error !== "object" || error === null) return null; + const record = error as { + readonly detail?: unknown; + readonly message?: unknown; + readonly cause?: unknown; + }; + for (const candidate of [record.detail, record.message]) { + if (typeof candidate !== "string") continue; + const match = /Thread source epoch mismatch: expected (\d+); actual (\d+)\./.exec(candidate); + if (match?.[1] !== undefined && match[2] !== undefined) { + return { + expectedSourceEpoch: Number(match[1]), + actualSourceEpoch: Number(match[2]), + }; + } + } + return record.cause === undefined ? null : sourceEpochMismatchFromError(record.cause); +} + +export function providerSettingsMutationInstanceIds( + current: ContractServerSettings, + patch: ServerSettingsPatch, +): ReadonlyArray { + const touched = new Set(); + if (patch.providerInstances !== undefined) { + const ids = new Set([ + ...Object.keys(current.providerInstances), + ...Object.keys(patch.providerInstances), + ]); + for (const rawId of ids) { + const instanceId = rawId as ProviderInstanceId; + if ( + !Equal.equals(current.providerInstances[instanceId], patch.providerInstances[instanceId]) + ) { + touched.add(instanceId); + } + } + } + if (patch.providers !== undefined) { + for (const driver of Object.keys(patch.providers)) { + touched.add(defaultInstanceIdForDriver(ProviderDriverKind.make(driver))); + } + } + return [...touched]; +} + export function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< OrchestrationEvent, { @@ -453,6 +510,8 @@ const makeWsRpcLayer = ( const portDiscovery = yield* PortScanner.PortDiscovery; const providerRegistry = yield* ProviderRegistry.ProviderRegistry; const providerService = yield* ProviderService.ProviderService; + const rollbackSagaRunner = yield* RollbackSagaRunner; + const rollbackSagaRepository = yield* RollbackSagaRepository; const sideQuestionOwnership = makeSessionSideQuestionOwnership(); const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const primeManagedMaintenance = yield* PrimeManagedMaintenance.PrimeManagedMaintenance; @@ -557,13 +616,30 @@ const makeWsRpcLayer = ( authorizeEffect(requiredScopeForRpcMethod(method), effect), traceAttributes, ); - const toDispatchCommandError = (cause: unknown, fallbackMessage: string) => - isOrchestrationDispatchCommandError(cause) - ? cause - : new OrchestrationDispatchCommandError({ - message: cause instanceof Error ? cause.message : fallbackMessage, - cause, - }); + const toDispatchCommandError = (cause: unknown, fallbackMessage: string) => { + const mismatch = sourceEpochMismatchFromError(cause); + if (isOrchestrationDispatchCommandError(cause) && mismatch === null) return cause; + return new OrchestrationDispatchCommandError({ + message: + mismatch === null + ? cause instanceof Error + ? cause.message + : fallbackMessage + : "This thread changed after the turn was composed. Review and reconfirm the message before sending it.", + cause, + ...(mismatch === null + ? {} + : { + reason: "source-epoch-mismatch" as const, + expectedSourceEpoch: mismatch.expectedSourceEpoch, + actualSourceEpoch: mismatch.actualSourceEpoch, + }), + ...(isOrchestrationDispatchCommandError(cause) && + cause.bootstrapThreadDisposition !== undefined + ? { bootstrapThreadDisposition: cause.bootstrapThreadDisposition } + : {}), + }); + }; const randomUUID = crypto.randomUUIDv4.pipe( Effect.mapError((cause) => toDispatchCommandError(cause, "Failed to generate orchestration command identifier."), @@ -1159,6 +1235,7 @@ const makeWsRpcLayer = ( }), threadResumeCompletionMarker: true, threadSnapshotPagination: true, + rollbackStatusStreaming: true, }; }); @@ -1445,7 +1522,9 @@ const makeWsRpcLayer = ( const isThisThreadDetailEvent = (event: OrchestrationEvent) => event.aggregateKind === "thread" && event.aggregateId === input.threadId && - isThreadDetailEvent(event); + isThreadDetailEvent(event) && + (input.rollbackStatusEvents === true || + event.type !== "thread.rollback-status-updated"); const liveStream = orchestrationEngine.streamDomainEvents.pipe( Stream.filter(isThisThreadDetailEvent), @@ -1569,6 +1648,10 @@ const makeWsRpcLayer = ( }), { "rpc.aggregate": "orchestration" }, ), + [WS_METHODS.rollbackRecover]: (input) => + observeRpcEffect(WS_METHODS.rollbackRecover, rollbackSagaRunner.recover(input), { + "rpc.aggregate": "rollback", + }), [WS_METHODS.serverProbe]: (_input) => observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), { "rpc.aggregate": "server", @@ -1844,7 +1927,10 @@ const makeWsRpcLayer = ( [WS_METHODS.serverUpdateProvider]: (input) => observeRpcEffect( WS_METHODS.serverUpdateProvider, - providerMaintenanceRunner.updateProvider(input), + rollbackSagaRepository.withProviderMutationFence( + Effect.succeed([input.instanceId ?? defaultInstanceIdForDriver(input.provider)]), + providerMaintenanceRunner.updateProvider(input), + ), { "rpc.aggregate": "server", }, @@ -1937,9 +2023,14 @@ const makeWsRpcLayer = ( [WS_METHODS.serverUpdateSettings]: ({ patch }) => observeRpcEffect( WS_METHODS.serverUpdateSettings, - serverSettings - .updateSettings(patch) - .pipe(Effect.map(ServerSettings.redactServerSettingsForClient)), + rollbackSagaRepository.withProviderMutationFence( + serverSettings.getSettings.pipe( + Effect.map((current) => providerSettingsMutationInstanceIds(current, patch)), + ), + serverSettings + .updateSettings(patch) + .pipe(Effect.map(ServerSettings.redactServerSettingsForClient)), + ), { "rpc.aggregate": "server", }, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 1ef4e9905..c549dd7e1 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -3,7 +3,6 @@ import { type ChatFileAttachment, DEFAULT_MODEL, defaultInstanceIdForDriver, - supportsServerProviderConversationRollback, type EnvironmentId, type MessageId, type ModelSelection, @@ -38,6 +37,12 @@ import { import { deriveReportedTurnCosts } from "@t3tools/client-runtime/state/turn-costs"; import { canAskSessionSideQuestion } from "@t3tools/client-runtime/state/session-side-question"; import { wasBootstrapThreadDeleted } from "@t3tools/client-runtime/errors"; +import { + buildRollbackConfirmation, + deriveRollbackTargets, + isRollbackActive, + type RollbackTarget, +} from "@t3tools/client-runtime/rollback"; import { type CodexArtifactTemplate } from "@t3tools/client-runtime/codex-artifact-templates"; import { changeRequestAutoSettles, @@ -191,6 +196,7 @@ import { PullRequestDetailGhost } from "./pullRequest/PullRequestGhosts"; import { PullRequestsUnavailableState } from "./pullRequest/PullRequestsUnavailableState"; import { RightPanelTabs, type PullRequestTabStatus } from "./RightPanelTabs"; import { AgentsPanel } from "./AgentsPanel"; +import { RollbackStatusBanner } from "./RollbackStatusBanner"; import { canMessageSessionAgent, deriveAgentPanelModel, @@ -1454,6 +1460,9 @@ function ChatViewContent(props: ChatViewProps) { const revertThreadCheckpoint = useAtomCommand(threadEnvironment.revertCheckpoint, { reportFailure: false, }); + const recoverThreadRollback = useAtomCommand(threadEnvironment.recoverRollback, { + reportFailure: false, + }); const openPreview = useAtomCommand(previewEnvironment.open, { reportFailure: false }); const closePreview = useAtomCommand(previewEnvironment.close, "preview close"); const { environments } = useEnvironments(); @@ -1604,6 +1613,7 @@ function ChatViewContent(props: ChatViewProps) { >({}); const [isConnecting, _setIsConnecting] = useState(false); const [isRevertingCheckpoint, setIsRevertingCheckpoint] = useState(false); + const [rollbackRecoveryPending, setRollbackRecoveryPending] = useState(false); const [maximizedRightPanelThreadKey, setMaximizedRightPanelThreadKey] = useState( null, ); @@ -3103,8 +3113,7 @@ function ChatViewContent(props: ChatViewProps) { attachDraftHeroComposerAnchorRef, captureDraftHeroComposerRect, ] = useDraftHeroLayoutTransition(isDraftHeroState); - const { turnDiffSummaries, inferredCheckpointTurnCountByTurnId } = - useTurnDiffSummaries(activeThread); + const { turnDiffSummaries } = useTurnDiffSummaries(activeThread); const turnDiffSummaryByAssistantMessageId = useMemo(() => { const byMessageId = new Map(); for (const summary of turnDiffSummaries) { @@ -3113,38 +3122,34 @@ function ChatViewContent(props: ChatViewProps) { } return byMessageId; }, [turnDiffSummaries]); - const revertTurnCountByUserMessageId = useMemo(() => { - const byUserMessageId = new Map(); - for (let index = 0; index < timelineEntries.length; index += 1) { - const entry = timelineEntries[index]; - if (!entry || entry.kind !== "message" || entry.message.role !== "user") { - continue; - } - - for (let nextIndex = index + 1; nextIndex < timelineEntries.length; nextIndex += 1) { - const nextEntry = timelineEntries[nextIndex]; - if (!nextEntry || nextEntry.kind !== "message") { - continue; - } - if (nextEntry.message.role === "user") { - break; - } - const summary = turnDiffSummaryByAssistantMessageId.get(nextEntry.message.id); - if (!summary) { - continue; - } - const turnCount = - summary.checkpointTurnCount ?? inferredCheckpointTurnCountByTurnId[summary.turnId]; - if (typeof turnCount !== "number") { - break; - } - byUserMessageId.set(entry.message.id, Math.max(0, turnCount - 1)); - break; - } - } - - return byUserMessageId; - }, [inferredCheckpointTurnCountByTurnId, timelineEntries, turnDiffSummaryByAssistantMessageId]); + const rollbackTargetsByUserMessageId = useMemo( + () => + activeThread ? deriveRollbackTargets(activeThread) : new Map(), + [activeThread], + ); + const revertTurnCountByUserMessageId = useMemo( + () => + new Map( + [...rollbackTargetsByUserMessageId].map(([messageId, target]) => [ + messageId, + target.targetTurnCount, + ]), + ), + [rollbackTargetsByUserMessageId], + ); + const rollbackActive = isRollbackActive(activeThread?.rollbackStatus); + const rollbackTargetIdle = + activeThread?.session !== null && + activeThread?.session !== undefined && + (activeThread.session.status === "idle" || activeThread.session.status === "ready") && + activeThread.session.activeTurnId === null && + activeThread.session.pendingTurnRequestId === undefined && + activeThread.session.activeTurnRequestId === undefined && + activeThread.latestTurn?.state !== "running" && + !rollbackActive && + phase !== "running" && + !isSendBusy && + !isConnecting; const gitCwd = activeProject ? projectScriptCwd({ @@ -3867,7 +3872,7 @@ function ChatViewContent(props: ChatViewProps) { const handleRuntimeModeChange = useCallback( (mode: RuntimeMode) => { - if (mode === runtimeMode) return; + if (rollbackActive || mode === runtimeMode) return; setComposerDraftRuntimeMode(composerDraftTarget, mode); if (isLocalDraftThread) { setDraftThreadContext(composerDraftTarget, { runtimeMode: mode }); @@ -3876,6 +3881,7 @@ function ChatViewContent(props: ChatViewProps) { }, [ isLocalDraftThread, + rollbackActive, runtimeMode, scheduleComposerFocus, composerDraftTarget, @@ -3886,7 +3892,7 @@ function ChatViewContent(props: ChatViewProps) { const handleInteractionModeChange = useCallback( (mode: ProviderInteractionMode) => { - if (mode === interactionMode) return; + if (rollbackActive || mode === interactionMode) return; setComposerDraftInteractionMode(composerDraftTarget, mode); if (isLocalDraftThread) { setDraftThreadContext(composerDraftTarget, { interactionMode: mode }); @@ -3896,6 +3902,7 @@ function ChatViewContent(props: ChatViewProps) { [ interactionMode, isLocalDraftThread, + rollbackActive, scheduleComposerFocus, composerDraftTarget, setComposerDraftInteractionMode, @@ -5855,9 +5862,9 @@ function ChatViewContent(props: ChatViewProps) { ]); const onRevertToTurnCount = useCallback( - async (turnCount: number) => { + async (target: RollbackTarget) => { const localApi = readLocalApi(); - if (!localApi || !activeThread || isRevertingCheckpoint) return; + if (!localApi || !activeThread || isRevertingCheckpoint || !rollbackTargetIdle) return; if (activeEnvironmentUnavailable && activeEnvironmentUnavailableLabel) { setThreadError( @@ -5866,18 +5873,13 @@ function ChatViewContent(props: ChatViewProps) { ); return; } - if (phase === "running" || isSendBusy || isConnecting) { + if (isSendBusy || isConnecting) { setThreadError(activeThread.id, "Interrupt the current turn before reverting checkpoints."); return; } - const confirmed = await localApi.dialogs.confirm( - [ - `Revert this thread to checkpoint ${turnCount}?`, - "This will discard newer messages and turn diffs in this thread.", - "This action cannot be undone.", - ].join("\n"), - { variant: "destructive" }, - ); + const confirmed = await localApi.dialogs.confirm(buildRollbackConfirmation(target.label), { + variant: "destructive", + }); if (!confirmed) { return; } @@ -5888,7 +5890,8 @@ function ChatViewContent(props: ChatViewProps) { environmentId, input: { threadId: activeThread.id, - turnCount, + turnCount: target.targetTurnCount, + expectedSourceRevision: target.expectedSourceRevision, }, }); if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { @@ -5910,10 +5913,32 @@ function ChatViewContent(props: ChatViewProps) { isSendBusy, phase, revertThreadCheckpoint, + rollbackTargetIdle, setThreadError, ], ); + const onRecoverRollback = useCallback( + async (action: "retry-verification" | "resume-compensation") => { + if (!activeThread || rollbackRecoveryPending) return; + setRollbackRecoveryPending(true); + setThreadError(activeThread.id, null); + const result = await recoverThreadRollback({ + environmentId, + input: { threadId: activeThread.id, action }, + }); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError( + activeThread.id, + error instanceof Error ? error.message : "Rollback recovery could not resume.", + ); + } + setRollbackRecoveryPending(false); + }, + [activeThread, environmentId, recoverThreadRollback, rollbackRecoveryPending, setThreadError], + ); + const onSend = async ( e?: { preventDefault: () => void }, submissionIntent: ComposerSubmissionIntent = "foreground", @@ -5924,6 +5949,10 @@ function ChatViewContent(props: ChatViewProps) { delivery: "immediate" | "follow-up" = "immediate", ) => { e?.preventDefault(); + if (rollbackActive && activeThread) { + setThreadError(activeThread.id, "This thread is fenced while Pylon verifies rollback state."); + return; + } const notifyDirectAnnotationAttached = () => { if (!directAnnotation) return; toastManager.add( @@ -6558,6 +6587,7 @@ function ChatViewContent(props: ChatViewProps) { titleSeed: title, runtimeMode, interactionMode, + sourceEpoch: activeThread.sourceEpoch ?? 0, ...(bootstrap ? { bootstrap } : {}), createdAt: messageCreatedAt, }, @@ -7011,7 +7041,7 @@ function ChatViewContent(props: ChatViewProps) { ); const onClearSessionInputQueue = useCallback(async () => { - if (!activeThreadId) return; + if (!activeThreadId || rollbackActive) return; const result = await clearThreadSessionInputQueue({ environmentId, input: { threadId: activeThreadId }, @@ -7027,11 +7057,11 @@ function ChatViewContent(props: ChatViewProps) { type: "success", title: "Pending session inputs cleared", }); - }, [activeThreadId, clearThreadSessionInputQueue, environmentId, setThreadError]); + }, [activeThreadId, clearThreadSessionInputQueue, environmentId, rollbackActive, setThreadError]); const onRemoveOnlySessionInputQueueItem = useCallback( async (queue: "steering" | "follow-up") => { - if (!activeThreadId) return; + if (!activeThreadId || rollbackActive) return; const result = await removeOnlyThreadSessionInputQueueItem({ environmentId, input: { threadId: activeThreadId, queue }, @@ -7048,12 +7078,18 @@ function ChatViewContent(props: ChatViewProps) { title: `Pending ${queue === "steering" ? "steering" : "follow-up"} input removed`, }); }, - [activeThreadId, environmentId, removeOnlyThreadSessionInputQueueItem, setThreadError], + [ + activeThreadId, + environmentId, + removeOnlyThreadSessionInputQueueItem, + rollbackActive, + setThreadError, + ], ); const onSetSessionInputQueueMode = useCallback( async (queue: "steering" | "follow-up", mode: "all-at-once" | "one-at-a-time") => { - if (!activeThreadId || !sessionProviderAdmissionAvailable) return; + if (!activeThreadId || rollbackActive || !sessionProviderAdmissionAvailable) return; const result = await setThreadSessionInputQueueMode({ environmentId, input: { threadId: activeThreadId, queue, mode }, @@ -7073,6 +7109,7 @@ function ChatViewContent(props: ChatViewProps) { [ activeThreadId, environmentId, + rollbackActive, sessionProviderAdmissionAvailable, setThreadError, setThreadSessionInputQueueMode, @@ -7396,6 +7433,7 @@ function ChatViewContent(props: ChatViewProps) { titleSeed: activeThread.title, runtimeMode, interactionMode: nextInteractionMode, + sourceEpoch: activeThread.sourceEpoch ?? 0, ...(nextInteractionMode === "default" && activeProposedPlan ? { sourceProposedPlan: { @@ -7532,6 +7570,7 @@ function ChatViewContent(props: ChatViewProps) { titleSeed: nextThreadTitle, runtimeMode, interactionMode: "default", + sourceEpoch: 0, sourceProposedPlan: { threadId: activeThread.id, planId: activeProposedPlan.id, @@ -7810,6 +7849,7 @@ function ChatViewContent(props: ChatViewProps) { titleSeed: nextThreadTitle, runtimeMode, interactionMode, + sourceEpoch: 0, createdAt, }, }); @@ -7893,6 +7933,9 @@ function ChatViewContent(props: ChatViewProps) { if (!activeThread) { return null; } + if (rollbackActive) { + return "Provider changes are blocked while Pylon verifies rollback state."; + } const reason = getStartedThreadModelChangeBlockReason({ providers: providerStatuses, hasStartedSession: activeThread.session !== null, @@ -7902,7 +7945,7 @@ function ChatViewContent(props: ChatViewProps) { }); return reason ? `${reason.description} Start a new thread to use this model.` : null; }, - [activeThread, providerStatuses], + [activeThread, providerStatuses, rollbackActive], ); const onProviderModelSelect = useCallback( @@ -8051,16 +8094,14 @@ function ChatViewContent(props: ChatViewProps) { ); // Both the Map and the revert handler are read from refs at call-time so // the callback reference is fully stable and never busts context identity. - const revertTurnCountRef = useRef(revertTurnCountByUserMessageId); - revertTurnCountRef.current = revertTurnCountByUserMessageId; + const rollbackTargetsRef = useRef(rollbackTargetsByUserMessageId); + rollbackTargetsRef.current = rollbackTargetsByUserMessageId; const onRevertToTurnCountRef = useRef(onRevertToTurnCount); onRevertToTurnCountRef.current = onRevertToTurnCount; const onRevertUserMessage = useCallback((messageId: MessageId) => { - const targetTurnCount = revertTurnCountRef.current.get(messageId); - if (typeof targetTurnCount !== "number") { - return; - } - void onRevertToTurnCountRef.current(targetTurnCount); + const target = rollbackTargetsRef.current.get(messageId); + if (!target) return; + void onRevertToTurnCountRef.current(target); }, []); // Empty state: no active thread @@ -8331,6 +8372,11 @@ function ChatViewContent(props: ChatViewProps) { status={visibleProviderStatus} onDismiss={() => setDismissedProviderStatusBannerKey(providerStatusBannerKey)} /> + {/* Messages Wrapper */}
@@ -8355,9 +8401,9 @@ function ChatViewContent(props: ChatViewProps) { onOpenTurnDiff={onOpenTurnDiff} revertTurnCountByUserMessageId={revertTurnCountByUserMessageId} onRevertUserMessage={onRevertUserMessage} - supportsConversationRollback={supportsServerProviderConversationRollback( - activeProviderStatus, - )} + supportsConversationRollback={ + rollbackTargetIdle && rollbackTargetsByUserMessageId.size > 0 + } onUseArtifactTemplate={useArtifactTemplate} isRevertingCheckpoint={isRevertingCheckpoint} onImageExpand={onExpandTimelineImage} @@ -8489,13 +8535,15 @@ function ChatViewContent(props: ChatViewProps) { isConnecting={isConnecting} isSendBusy={isSendBusy} sendDisabledReason={ - feedbackUploading - ? "Sending feedback" - : threadDetailLoading - ? "Messages loading" - : activeSessionInteraction - ? "Resolve the session request to continue" - : providerAdmissionDisabledReason + rollbackActive + ? "Rollback verification in progress" + : feedbackUploading + ? "Sending feedback" + : threadDetailLoading + ? "Messages loading" + : activeSessionInteraction + ? "Resolve the session request to continue" + : providerAdmissionDisabledReason } isPreparingWorktree={isPreparingWorktree} bannerItems={composerBannerItems} @@ -8603,7 +8651,7 @@ function ChatViewContent(props: ChatViewProps) { { + it("announces a manual recovery fence and exposes only authorized actions", () => { + const markup = renderToStaticMarkup( + {}} + />, + ); + + expect(markup).toContain('role="alert"'); + expect(markup).toContain('aria-live="assertive"'); + expect(markup).toContain('aria-atomic="true"'); + expect(markup).toContain("Manual recovery required"); + expect(markup).toContain('aria-label="Resume rollback compensation"'); + expect(markup).not.toContain("Retry rollback verification"); + }); + + it("uses a polite durable status for pending and completed operations", () => { + const pending = renderToStaticMarkup( + {}} + />, + ); + const completed = renderToStaticMarkup( + {}} + />, + ); + + expect(pending).toContain('role="status"'); + expect(pending).toContain('aria-live="polite"'); + expect(pending).toContain("Rollback pending"); + expect(completed).toContain("Rollback completed"); + }); + + it("disables recovery while a client request is in flight", () => { + const markup = renderToStaticMarkup( + {}} + />, + ); + + expect(markup).toContain('aria-label="Retry rollback verification"'); + expect(markup).toContain("disabled"); + }); +}); diff --git a/apps/web/src/components/RollbackStatusBanner.tsx b/apps/web/src/components/RollbackStatusBanner.tsx new file mode 100644 index 000000000..7c2dc3b43 --- /dev/null +++ b/apps/web/src/components/RollbackStatusBanner.tsx @@ -0,0 +1,66 @@ +import type { OrchestrationRollbackStatus } from "@t3tools/contracts"; +import { Button } from "./ui/button"; + +export function RollbackStatusBanner(props: { + readonly status: OrchestrationRollbackStatus | null | undefined; + readonly recoveryPending: boolean; + readonly onRecover: (action: "retry-verification" | "resume-compensation") => void; +}) { + const status = props.status; + if (!status) return null; + const severe = status.state === "manual-recovery" || status.state === "failed"; + const label = + status.state === "pending" + ? "Rollback pending" + : status.state === "recovering" + ? "Rollback recovering" + : status.state === "manual-recovery" + ? "Manual recovery required" + : status.state === "completed" + ? "Rollback completed" + : "Rollback failed safely"; + const actions = status.allowedActions ?? []; + return ( +
+
+

{label}

+

+ {status.detail ?? "Pylon is verifying rollback state."} +

+
+ {actions.includes("retry-verification") ? ( + + ) : null} + {actions.includes("resume-compensation") ? ( + + ) : null} +
+ ); +} diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index 78cc6ba10..bdfffdd27 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -524,6 +524,7 @@ const ComposerFooterModeControls = memo(function ComposerFooterModeControls(prop interactionMode: ProviderInteractionMode; runtimeMode: RuntimeMode; supportedRuntimeModes: ReadonlyArray; + disabled: boolean; onToggleInteractionMode: () => void; onRuntimeModeChange: (mode: RuntimeMode) => void; }) { @@ -548,6 +549,7 @@ const ComposerFooterModeControls = memo(function ComposerFooterModeControls(prop : "text-secondary-label hover:text-foreground", )} type="button" + disabled={props.disabled} onClick={props.onToggleInteractionMode} aria-label={interactionModeTooltip} /> @@ -574,6 +576,7 @@ const ComposerFooterModeControls = memo(function ComposerFooterModeControls(prop