fix(mobile): stop pinning sent messages below newer history

apps/mobile/src/state/acknowledged-thread-messages.test.ts

@@ -0,0 +1,60 @@
+import { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
+import { describe, expect, it, vi } from "vite-plus/test";
+
+import { unsettledAcknowledgedThreadMessages } from "./acknowledged-thread-messages";
+
+vi.mock("./threads", () => ({ environmentThreadDetails: {} }));
+
+const environmentId = EnvironmentId.make("environment-1");
+const threadId = ThreadId.make("thread-1");
+
+function acknowledged(messageId: string, acknowledgedSequence?: number) {
+  return {
+    environmentId,
+    threadId,
+    messageId: MessageId.make(messageId),
+    commandId: CommandId.make(`command-${messageId}`),
+    text: messageId,
+    attachments: [],
+    // Client time can be ahead of the server, or equal for different sends.
+    createdAt: "2099-01-01T00:00:00.000Z",
+    ...(acknowledgedSequence === undefined ? {} : { acknowledgedSequence }),
+  };
+}
+
+describe("acknowledged message reconciliation", () => {
+  it("settles an unloaded echo only after a bounded snapshot includes its receipt", () => {
+    const old = acknowledged("accepted-outside-window", 40);
+    const inFlight = acknowledged("accepted-after-snapshot", 51);
+    const unknown = acknowledged("no-ordering-evidence");
+    const otherThread = { ...old, threadId: ThreadId.make("thread-2") };
+    const otherEnvironment = { ...old, environmentId: EnvironmentId.make("environment-2") };
+    const echoed = acknowledged("loaded-echo", 52);
+    const messages = [old, inFlight, unknown, otherThread, otherEnvironment, echoed];
+    const echoes = new Set([echoed.messageId]);
+    const history = { environmentId, threadId, hasMoreHistory: true, snapshotSequence: 40 };
+
+    expect(unsettledAcknowledgedThreadMessages(messages, echoes, history)).toEqual([
+      inFlight,
+      unknown,
+      otherThread,
+      otherEnvironment,
+    ]);
+    expect(
+      unsettledAcknowledgedThreadMessages(messages, echoes, { ...history, snapshotSequence: 39 }),
+    ).toEqual([old, inFlight, unknown, otherThread, otherEnvironment]);
+  });
+
+  it("keeps missing echoes without evidence of a bounded, committed history window", () => {
+    const message = acknowledged("still-waiting", 40);
+    const history = { environmentId, threadId, hasMoreHistory: false, snapshotSequence: 50 };
+    expect(unsettledAcknowledgedThreadMessages([message], new Set(), history)).toEqual([message]);
+    expect(
+      unsettledAcknowledgedThreadMessages([message], new Set(), {
+        environmentId,
+        threadId,
+        hasMoreHistory: true,
+      }),
+    ).toEqual([message]);
+  });
+});

apps/mobile/src/state/acknowledged-thread-messages.ts

@@ -1,12 +1,44 @@
+import type { MessageId, ScopedThreadRef } from "@t3tools/contracts";
 import * as Option from "effect/Option";
 import { Atom } from "effect/reactivity";
 import { appAtomRegistry } from "./atom-registry";
 import type { QueuedThreadMessage } from "./thread-outbox-model";
 import { environmentThreadDetails } from "./threads";
 
+interface AcknowledgedThreadMessage extends QueuedThreadMessage {
+  readonly acknowledgedSequence?: number;
+}
+
+export function unsettledAcknowledgedThreadMessages(
+  messages: ReadonlyArray<AcknowledgedThreadMessage>,
+  echoedIds: ReadonlySet<MessageId>,
+  history?: ScopedThreadRef & {
+    readonly hasMoreHistory: boolean;
+    readonly snapshotSequence?: number;
+  },
+): ReadonlyArray<AcknowledgedThreadMessage> {
+  return messages.filter((message) => {
+    if (
+      history !== undefined &&
+      (message.environmentId !== history.environmentId || message.threadId !== history.threadId)
+    ) {
+      return true;
+    }
+    if (echoedIds.has(message.messageId)) return false;
+    // A bounded snapshot taken after the send committed can omit its echo.
+    // Client timestamps cannot establish this ordering, especially after offline sends.
+    return !(
+      history?.hasMoreHistory &&
+      history.snapshotSequence !== undefined &&
+      message.acknowledgedSequence !== undefined &&
+      message.acknowledgedSequence <= history.snapshotSequence
+    );
+  });
+}
+
 // Keep acknowledged rows until their subscribed projection arrives; command receipts can precede messages.
 // Render hides rows the projection has echoed, so dropping them here only frees memory.
-export const acknowledgedThreadMessagesAtom = Atom.make<ReadonlyArray<QueuedThreadMessage>>(
+export const acknowledgedThreadMessagesAtom = Atom.make<ReadonlyArray<AcknowledgedThreadMessage>>(
   [],
 ).pipe(Atom.keepAlive);
 
@@ -25,7 +57,9 @@ function isEchoedInMountedThread(message: QueuedThreadMessage): boolean {
 }
 
 function writeAcknowledgedThreadMessages(
-  update: (current: ReadonlyArray<QueuedThreadMessage>) => ReadonlyArray<QueuedThreadMessage>,
+  update: (
+    current: ReadonlyArray<AcknowledgedThreadMessage>,
+  ) => ReadonlyArray<AcknowledgedThreadMessage>,
 ) {
   const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom);
   const next = update(current).filter((entry) => !isEchoedInMountedThread(entry));
@@ -35,11 +69,16 @@ function writeAcknowledgedThreadMessages(
   appAtomRegistry.set(acknowledgedThreadMessagesAtom, next);
 }
 
-export function retainAcknowledgedThreadMessage(message: QueuedThreadMessage) {
+export function retainAcknowledgedThreadMessage(
+  message: QueuedThreadMessage,
+  acknowledgedSequence?: number,
+) {
+  const acknowledgedMessage =
+    acknowledgedSequence === undefined ? message : { ...message, acknowledgedSequence };
   writeAcknowledgedThreadMessages((current) =>
     current.some((entry) => entry.messageId === message.messageId)
       ? current
-      : [...current, message],
+      : [...current, acknowledgedMessage],
   );
 }
 

apps/mobile/src/state/thread-task-permissions.test.ts

@@ -1,6 +1,9 @@
 import {
   AuthOrchestrationOperateScope,
   DEFAULT_SERVER_SETTINGS,
+  CommandId,
+  EnvironmentId,
+  ThreadId,
   MessageId,
   ProjectId,
   ProviderDriverKind,
@@ -32,6 +35,7 @@ const state = vi.hoisted(() => ({
   },
   serverConfig: null as ServerConfig | null,
   projection: null as OrchestrationV2ThreadProjection | null,
+  history: { hasMoreHistory: true, snapshotSequence: 39 },
   pendingRequests: { approvals: [], userInputs: [] } as PendingThreadRequests,
   enqueue: vi.fn(async () => undefined),
   clearDraft: vi.fn(),
@@ -94,6 +98,7 @@ vi.mock("./use-thread-selection", () => ({
   useThreadSelection: () => ({
     selectedThread: state.thread,
     selectedThreadCreation: null,
+    selectedThreadDetailRef: { environmentId: "secondary", threadId: "thread" },
     selectedEnvironmentRuntime: {
       connectionState: state.connectionState,
       serverConfig: state.serverConfig,
@@ -113,6 +118,7 @@ vi.mock("./threads", async () => {
     environmentThreadDetails: {
       queueWorkflowAtom: () => Atom.make(null),
       checkpointSummariesAtom: () => Atom.make([]),
+      historyAtom: () => Atom.make(state.history),
       pendingRequestsAtom: () => Atom.make(state.pendingRequests),
     },
     threadEnvironment: {
@@ -174,6 +180,9 @@ vi.mock("../lib/commandMetadata", () => ({
 vi.mock("../lib/copyTextWithHaptic", () => ({ copyTextWithHaptic: vi.fn() }));
 vi.mock("./use-remote-environment-registry", () => ({ setPendingConnectionError: vi.fn() }));
 
+import { appAtomRegistry } from "./atom-registry";
+import { acknowledgedThreadMessagesAtom } from "./acknowledged-thread-messages";
+
 import { useThreadComposerState } from "./use-thread-composer-state";
 import { useSelectedThreadRequests } from "./use-selected-thread-requests";
 
@@ -182,6 +191,8 @@ beforeEach(() => {
   state.connectionState = "connected";
   state.draft = { text: "Keep my draft", attachments: [], modelSelection: undefined };
   state.projection = null;
+  state.history = { hasMoreHistory: true, snapshotSequence: 39 };
+  appAtomRegistry.set(acknowledgedThreadMessagesAtom, []);
   state.serverConfig = {
     providers: [{ instanceId: "codex", driver: "codex", enabled: true }],
     environment: { capabilities: {} },
@@ -428,3 +439,27 @@ describe("mobile handoff preview and submission", () => {
     expect(state.clearDraft).toHaveBeenCalledTimes(1);
   });
 });
+
+it("reopening bounded history removes an accepted row only after its receipt", () => {
+  const message = {
+    environmentId: EnvironmentId.make("secondary"),
+    threadId: ThreadId.make("thread"),
+    messageId: MessageId.make("accepted-outside-window"),
+    commandId: CommandId.make("accepted-command"),
+    text: "old accepted message",
+    attachments: [],
+    createdAt: "2099-01-01T00:00:00.000Z",
+    acknowledgedSequence: 40,
+  };
+  appAtomRegistry.set(acknowledgedThreadMessagesAtom, [message]);
+  state.projection = { ...v2Projection, messages: [] };
+  expect(useThreadComposerState().selectedThreadFeed.map((entry) => entry.id)).toEqual([
+    message.messageId,
+  ]);
+  state.history = { hasMoreHistory: true, snapshotSequence: 40 };
+  expect(useThreadComposerState().selectedThreadFeed).toEqual([]);
+  state.history = { hasMoreHistory: false, snapshotSequence: 40 };
+  expect(useThreadComposerState().selectedThreadFeed.map((entry) => entry.id)).toEqual([
+    message.messageId,
+  ]);
+});

apps/mobile/src/state/threadOutboxDelivery.permissions.test.ts

@@ -224,6 +224,7 @@ beforeEach(() => {
   ]) {
     command.mockReset().mockResolvedValue(AsyncResult.success(undefined));
   }
+  state.start.mockResolvedValue(AsyncResult.success({ sequence: 1 }));
   state.prepare.mockReset().mockImplementation(async ({ attachments }) => ({
     status: "ready",
     attachments: [],
@@ -427,7 +428,8 @@ describe("queued task operation access", () => {
       );
       await state.manager.enqueue(queued);
       const started = Promise.withResolvers<void>();
-      const delivery = Promise.withResolvers<AsyncResult.AsyncResult<void, Error>>();
+      const delivery =
+        Promise.withResolvers<AsyncResult.AsyncResult<{ readonly sequence: number }, Error>>();
       state.start.mockImplementationOnce(() => {
         started.resolve();
         return delivery.promise;
@@ -437,7 +439,7 @@ describe("queued task operation access", () => {
       setOperateGrant("secondary", false);
       delivery.resolve(
         accepted
-          ? AsyncResult.success(undefined)
+          ? AsyncResult.success({ sequence: 1 })
           : AsyncResult.failure(Cause.fail(new Error("This connection cannot control this task."))),
       );
       await drained;

apps/mobile/src/state/threadOutboxDelivery.test.ts

@@ -458,10 +458,14 @@ describe("thread outbox drain delivery cleanup", () => {
     await harness.manager.enqueue(message);
     const deliveryRevision = harness.manager.revisionOf(message.messageId);
 
-    await expect(completeQueuedMessageDelivery(message, deliveryRevision)).resolves.toBe("removed");
+    await expect(completeQueuedMessageDelivery(message, deliveryRevision, 40)).resolves.toBe(
+      "removed",
+    );
 
     expect(remainingMessages()).toEqual([]);
-    expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([message]);
+    expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([
+      { ...message, acknowledgedSequence: 40 },
+    ]);
   });
 
   it("keeps a delivered message when its editor opens during storage removal", async () => {

apps/mobile/src/state/threadOutboxDelivery.ts

@@ -254,6 +254,7 @@ function isQueuedMessagePayloadCurrent(
 export async function completeQueuedMessageDelivery(
   queuedMessage: QueuedThreadMessage,
   deliveryRevision: number,
+  acknowledgedSequence?: number,
 ): Promise<"removed" | "edited" | "failed"> {
   clearThreadComposerError(
     scopedThreadKey(queuedMessage.environmentId, queuedMessage.threadId),
@@ -270,7 +271,7 @@ export async function completeQueuedMessageDelivery(
     if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) {
       return "edited";
     }
-    retainAcknowledgedThreadMessage(queuedMessage);
+    retainAcknowledgedThreadMessage(queuedMessage, acknowledgedSequence);
     const removed = await removeThreadOutboxMessage(
       queuedMessage,
       deliveryRevision,
@@ -904,7 +905,13 @@ export function createThreadOutboxDelivery(): {
     }
     acknowledgedExistingThreadMessageIds.add(persistedMessage.messageId);
     const delivered =
-      (await completeQueuedMessageDelivery(persistedMessage, deliveryRevision)) === "removed";
+      (await completeQueuedMessageDelivery(
+        persistedMessage,
+        deliveryRevision,
+        AsyncResult.isSuccess(deliveryResult) && "sequence" in deliveryResult.value
+          ? deliveryResult.value.sequence
+          : undefined,
+      )) === "removed";
     if (delivered) {
       acknowledgedExistingThreadMessageIds.delete(persistedMessage.messageId);
     }

apps/mobile/src/state/use-thread-composer-state.ts

@@ -13,7 +13,7 @@ import {
   deriveThreadRuntime,
   threadRuntimeHasInterruptibleRun,
 } from "@t3tools/client-runtime/state/thread-execution";
-import { useRef, useState } from "react";
+import { useEffect, useRef, useState } from "react";
 
 import {
   CommandId,
@@ -30,6 +30,7 @@ import {
 import { clampFileAttachmentUploadBytes } from "@t3tools/client-runtime/state/attachments";
 import { nextPastedTextFileName, pastedTextDisposition } from "@t3tools/client-runtime/text-paste";
 import {
+  EMPTY_THREAD_HISTORY_META,
   parseCodexFeedbackCommand,
   submitCodexFeedback,
   type CodexFeedbackSubmission,
@@ -53,7 +54,10 @@ import {
 import type { DraftComposerImageAttachment } from "../lib/composerImages";
 import { scopedThreadKey } from "../lib/scopedEntities";
 import { buildThreadFeed, createThreadFeedCache } from "../lib/threadActivity";
-import { acknowledgedThreadMessagesAtom } from "./acknowledged-thread-messages";
+import {
+  acknowledgedThreadMessagesAtom,
+  unsettledAcknowledgedThreadMessages,
+} from "./acknowledged-thread-messages";
 import {
   appendPendingThreadMessages,
   selectPendingAcknowledgedMessages,
@@ -119,6 +123,7 @@ const EMPTY_QUEUE_WORKFLOW_ATOM = Atom.make<null>(null).pipe(
   Atom.withLabel("mobile-thread-queue-workflow:empty"),
 );
 const EMPTY_CHECKPOINT_SUMMARIES_ATOM = Atom.make<ReadonlyArray<ThreadCheckpointSummary>>([]);
+const EMPTY_THREAD_HISTORY_ATOM = Atom.make(EMPTY_THREAD_HISTORY_META);
 
 export function appendReviewCommentToDraft(input: {
   readonly environmentId: EnvironmentId;
@@ -250,6 +255,16 @@ export function useThreadComposerState() {
   const selectedThreadMessages = selectedThreadProjection?.projection.messages;
   const selectedThreadAttempts = selectedThreadProjection?.projection.attempts;
   const selectedThreadNodes = selectedThreadProjection?.projection.nodes;
+  const selectedThreadHistory = useAtomValue(
+    selectedThreadDetailRef === null
+      ? EMPTY_THREAD_HISTORY_ATOM
+      : environmentThreadDetails.historyAtom(selectedThreadDetailRef),
+  );
+  const echoedIds = new Set(selectedThreadMessages?.map((message) => message.id));
+  const acknowledgmentHistory =
+    selectedThreadDetailRef === null
+      ? undefined
+      : { ...selectedThreadDetailRef, ...selectedThreadHistory };
   const checkpointSummaries = useAtomValue(
     selectedThreadDetailRef === null
       ? EMPTY_CHECKPOINT_SUMMARIES_ATOM
@@ -274,16 +289,39 @@ export function useThreadComposerState() {
       checkpointSummaries,
       cache: feedCache,
     });
-    const pendingAcknowledgments = selectPendingAcknowledgedMessages(acknowledgedMessages, {
-      threadKey: selectedThreadKey,
-      queuedMessages: selectedThreadQueuedMessages,
-      projectedMessages: selectedThreadMessages,
-    });
+    const pendingAcknowledgments = selectPendingAcknowledgedMessages(
+      unsettledAcknowledgedThreadMessages(acknowledgedMessages, echoedIds, acknowledgmentHistory),
+      {
+        threadKey: selectedThreadKey,
+        queuedMessages: selectedThreadQueuedMessages,
+        projectedMessages: selectedThreadMessages,
+      },
+    );
     if (pendingAcknowledgments.length === 0) return feed;
     return appendPendingThreadMessages(feed, feed, pendingAcknowledgments).map((entry) =>
       entry.pendingMessage ? { ...entry, acknowledged: true } : entry,
     );
   })();
+  // oxlint-disable-next-line t3code/react-effect-budget -- Retire acknowledged rows when the subscribed history window advances.
+  useEffect(() => {
+    if (acknowledgedMessages.length === 0) return;
+    const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom);
+    const unsettled = unsettledAcknowledgedThreadMessages(
+      current,
+      new Set(selectedThreadMessages?.map((message) => message.id)),
+      selectedThreadDetailRef === null
+        ? undefined
+        : { ...selectedThreadDetailRef, ...selectedThreadHistory },
+    );
+    if (unsettled.length !== current.length) {
+      appAtomRegistry.set(acknowledgedThreadMessagesAtom, unsettled);
+    }
+  }, [
+    acknowledgedMessages,
+    selectedThreadMessages,
+    selectedThreadDetailRef,
+    selectedThreadHistory,
+  ]);
 
   const preferencesResult = useAtomValue(mobilePreferencesAtom);
   const followUpBehavior = AsyncResult.isSuccess(preferencesResult)