fix(rollback): fence stale clients and queued turns

Refs #200

Author: rynfar

apps/mobile/src/features/threads/ThreadDetailScreen.tsx

@@ -11,7 +11,8 @@ import {
   type CodexArtifactTemplate,
 } from "@t3tools/client-runtime/codex-artifact-templates";
 import type { EnvironmentThreadStatus } from "@t3tools/client-runtime/state/threads";
-import type { RollbackTarget } from "@t3tools/client-runtime/rollback";
+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";
@@ -305,32 +306,20 @@ function RollbackStatusSurface(props: {
   readonly onRecover: (action: "retry-verification" | "resume-compensation") => Promise<void>;
 }) {
   if (!props.status) return null;
-  const severe = props.status.state === "manual-recovery" || props.status.state === "failed";
-  const title =
-    props.status.state === "pending"
-      ? "Rollback pending"
-      : props.status.state === "recovering"
-        ? "Rollback recovering"
-        : props.status.state === "manual-recovery"
-          ? "Manual recovery required"
-          : props.status.state === "completed"
-            ? "Rollback completed"
-            : "Rollback failed safely";
-  const actions = props.status.allowedActions ?? [];
+  const presentation = getMobileRollbackStatusPresentation(props.status);
+  const { severe, title, actions } = presentation;
   return (
     <View
-      accessibilityRole={severe ? "alert" : "summary"}
-      accessibilityLiveRegion={severe ? "assertive" : "polite"}
+      accessibilityRole={presentation.accessibilityRole}
+      accessibilityLiveRegion={presentation.accessibilityLiveRegion}
       className={`mb-3 gap-2 rounded-2xl border px-3 py-2.5 ${
         severe ? "border-red-500/50 bg-red-500/10" : "border-adaptive-neutral-300-700 bg-screen"
       }`}
     >
       <Text className={`font-t3-semibold text-sm ${severe ? "text-red-500" : "text-foreground"}`}>
         {title}
       </Text>
-      <Text className="font-t3-regular text-xs text-foreground-muted">
-        {props.status.detail ?? "Pylon is verifying rollback state."}
-      </Text>
+      <Text className="font-t3-regular text-xs text-foreground-muted">{presentation.detail}</Text>
       {actions.length > 0 ? (
         <View className="flex-row flex-wrap gap-2">
           {actions.includes("retry-verification") ? (
@@ -945,6 +934,23 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread
                     pending={props.rollbackCommandPending}
                     onRecover={props.onRecoverRollback}
                   />
+                  {props.localOutboxCount > 0 &&
+                  props.rollbackTargets.size > 0 &&
+                  !isRollbackActive(props.rollbackStatus) ? (
+                    <View
+                      accessibilityRole="summary"
+                      className="mb-3 rounded-xl border border-adaptive-neutral-300-700 bg-screen px-3 py-2"
+                    >
+                      <Text className="text-sm font-semibold text-foreground">
+                        Rollback paused for queued messages
+                      </Text>
+                      <Text className="mt-1 text-xs leading-5 text-foreground-muted">
+                        Send or cancel the queued{" "}
+                        {props.localOutboxCount === 1 ? "message" : "messages"} before starting
+                        rollback.
+                      </Text>
+                    </View>
+                  ) : null}
                 </View>
 
                 {props.activePendingApproval ||

apps/mobile/src/features/threads/ThreadRouteScreen.tsx

@@ -24,6 +24,7 @@ import {
   isRollbackActive,
   type RollbackTarget,
 } from "@t3tools/client-runtime/rollback";
+import { resolveMobileRollbackStatus } from "./rollback-status-presentation";
 import {
   requestOlderThreadTurns,
   threadHasOlderTurns,
@@ -726,7 +727,10 @@ function ThreadRouteContent(
         : deriveRollbackTargets(selectedThreadDetail),
     [selectedThreadDetail],
   );
-  const rollbackStatus = selectedThreadDetail?.rollbackStatus ?? selectedThread?.rollbackStatus;
+  const rollbackStatus = resolveMobileRollbackStatus(
+    selectedThreadDetail?.rollbackStatus,
+    selectedThread?.rollbackStatus,
+  );
   const rollbackActive = isRollbackActive(rollbackStatus);
   const rollbackTargetIdle =
     selectedThreadDetail?.session !== null &&
@@ -738,6 +742,7 @@ function ThreadRouteContent(
     selectedThreadDetail.session.activeTurnRequestId === undefined &&
     selectedThreadDetail.latestTurn?.state !== "running" &&
     !rollbackActive &&
+    composer.selectedThreadQueueCount === 0 &&
     !composer.activeThreadBusy &&
     !rollbackCommandPending;
 

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"],
+    });
+  });
+});

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 ?? [],
+  };
+}

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";