fix(mobile): sequenceless delivered messages stop reappearing once paged out

apps/mobile/src/features/threads/pending-thread-feed.test.ts

@@ -132,4 +132,29 @@ describe("pending timeline messages", () => {
       "waiting",
     ]);
   });
+
+  it("stops showing a sequenceless acknowledged message that predates the loaded window", () => {
+    // Older servers report no commit sequence, so a paged-out message is only
+    // recognizable as delivered once it is older than everything still loaded.
+    const agedOut = pending("aged-out");
+    const fresh = { ...pending("just-sent"), createdAt: "2026-09-06T12:00:00.000Z" };
+    // A reported sequence the thread has not reached outranks the window guess.
+    const sequenced = { ...pending("sequenced"), sequence: 50 };
+    const windowed = {
+      threadKey: "env:thread",
+      queuedMessages: [],
+      sequence: 45,
+      projectedMessages: [],
+      hasMoreHistory: true,
+      oldestLoadedAt: "2026-09-06T11:00:00.000Z",
+    };
+    expect(selectPendingAcknowledgedMessages([agedOut, fresh, sequenced], windowed)).toEqual([
+      fresh,
+      sequenced,
+    ]);
+    // With the thread fully loaded an older absence still means the echo is outstanding.
+    expect(
+      selectPendingAcknowledgedMessages([agedOut], { ...windowed, hasMoreHistory: false }),
+    ).toEqual([agedOut]);
+  });
 });

apps/mobile/src/features/threads/pending-thread-feed.ts

@@ -37,6 +37,8 @@ export function selectPendingAcknowledgedMessages(
     readonly queuedMessages: ReadonlyArray<QueuedThreadMessage>;
     readonly sequence: number;
     readonly projectedMessages: ReadonlyArray<{ readonly id: string }> | undefined;
+    readonly hasMoreHistory?: boolean;
+    readonly oldestLoadedAt?: string | null;
   },
 ): ReadonlyArray<AcknowledgedThreadMessage> {
   return acknowledged.filter(
@@ -46,6 +48,8 @@ export function selectPendingAcknowledgedMessages(
       !threadHasCaughtUpTo(message, {
         sequence: thread.sequence,
         messages: thread.projectedMessages,
+        hasMoreHistory: thread.hasMoreHistory,
+        oldestLoadedAt: thread.oldestLoadedAt,
       }),
   );
 }

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

@@ -1,7 +1,7 @@
 import * as Option from "effect/Option";
 import { Atom } from "effect/reactivity";
 import { appAtomRegistry } from "./atom-registry";
-import { threadHasCaughtUpTo } from "@t3tools/client-runtime/state/threads";
+import { threadHasCaughtUpTo, threadOldestLoadedAt } from "@t3tools/client-runtime/state/threads";
 import type { AcknowledgedThreadMessage, QueuedThreadMessage } from "./thread-outbox-model";
 import { environmentThreadDetails } from "./threads";
 
@@ -22,11 +22,15 @@ function isEchoedInMountedThread(message: AcknowledgedThreadMessage): boolean {
   const loaded: Option.Option<Atom.Type<typeof stateAtom>> = node.valueOption();
   return Option.match(loaded, {
     onNone: () => false,
-    onSome: (thread) =>
-      threadHasCaughtUpTo(message, {
+    onSome: (thread) => {
+      const projection = Option.getOrUndefined(thread.data);
+      return threadHasCaughtUpTo(message, {
         sequence: thread.sequence,
-        messages: Option.getOrUndefined(thread.data)?.messages,
-      }),
+        messages: projection?.messages,
+        hasMoreHistory: thread.history.hasMoreHistory,
+        oldestLoadedAt: threadOldestLoadedAt(projection?.visibleTurnItems ?? []),
+      });
+    },
   });
 }
 

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

@@ -16,8 +16,10 @@ const harness = vi.hoisted(() => ({
   threadState: null as unknown as import("effect/reactivity").Atom.Writable<{
     readonly data: import("effect/Option").Option<{
       readonly messages: ReadonlyArray<{ readonly id: string }>;
+      readonly visibleTurnItems: ReadonlyArray<unknown>;
     }>;
     readonly sequence: number;
+    readonly history: { readonly hasMoreHistory: boolean };
   }>,
   manager: null as unknown as ReturnType<
     typeof import("./thread-outbox-manager").createThreadOutboxManager
@@ -97,8 +99,12 @@ vi.mock("./threads", async () => {
   const { Atom } = await import("effect/reactivity");
   const Option = await import("effect/Option");
   harness.threadState = Atom.make({
-    data: Option.none<{ readonly messages: ReadonlyArray<{ readonly id: string }> }>(),
+    data: Option.none<{
+      readonly messages: ReadonlyArray<{ readonly id: string }>;
+      readonly visibleTurnItems: ReadonlyArray<unknown>;
+    }>(),
     sequence: 0,
+    history: { hasMoreHistory: false },
   }).pipe(Atom.keepAlive);
   return {
     threadEnvironment: {},
@@ -210,7 +216,11 @@ function remainingMessages(): ReadonlyArray<QueuedThreadMessage> {
 
 beforeEach(() => {
   appAtomRegistry.set(acknowledgedThreadMessagesAtom, []);
-  appAtomRegistry.set(harness.threadState, { data: Option.none(), sequence: 0 });
+  appAtomRegistry.set(harness.threadState, {
+    data: Option.none(),
+    sequence: 0,
+    history: { hasMoreHistory: false },
+  });
   harness.canOperate = true;
   harness.draftFile.setDocument({ schemaVersion: 1, drafts: {} });
 });
@@ -351,8 +361,9 @@ describe("thread outbox drain delivery cleanup", () => {
 
     // The thread applied the message, but its bounded window does not hold it.
     appAtomRegistry.set(harness.threadState, {
-      data: Option.some({ messages: [{ id: "unrelated" }] }),
+      data: Option.some({ messages: [{ id: "unrelated" }], visibleTurnItems: [] }),
       sequence: 7,
+      history: { hasMoreHistory: false },
     });
     expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toHaveLength(1);
 

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

@@ -28,8 +28,10 @@ import {
 } from "@t3tools/contracts";
 
 import {
+  EMPTY_THREAD_HISTORY_META,
   parseCodexFeedbackCommand,
   submitCodexFeedback,
+  threadOldestLoadedAt,
   type CodexFeedbackSubmission,
 } from "@t3tools/client-runtime/state/threads";
 import { resolveThreadWorkingStartedAt } from "@t3tools/client-runtime/state/models";
@@ -111,6 +113,7 @@ const EMPTY_QUEUE_WORKFLOW_ATOM = Atom.make<null>(null).pipe(
 );
 const EMPTY_CHECKPOINT_SUMMARIES_ATOM = Atom.make<ReadonlyArray<ThreadCheckpointSummary>>([]);
 const EMPTY_THREAD_SEQUENCE_ATOM = Atom.make(0);
+const EMPTY_THREAD_HISTORY_ATOM = Atom.make(EMPTY_THREAD_HISTORY_META);
 
 export function appendReviewCommentToDraft(input: {
   readonly environmentId: EnvironmentId;
@@ -233,6 +236,11 @@ export function useThreadComposerState() {
       ? EMPTY_THREAD_SEQUENCE_ATOM
       : environmentThreadDetails.sequenceAtom(selectedThreadDetailRef),
   );
+  const selectedThreadHistory = useAtomValue(
+    selectedThreadDetailRef === null
+      ? EMPTY_THREAD_HISTORY_ATOM
+      : environmentThreadDetails.historyAtom(selectedThreadDetailRef),
+  );
   const checkpointSummaries = useAtomValue(
     selectedThreadDetailRef === null
       ? EMPTY_CHECKPOINT_SUMMARIES_ATOM
@@ -262,6 +270,8 @@ export function useThreadComposerState() {
       queuedMessages: selectedThreadQueuedMessages,
       sequence: selectedThreadSequence,
       projectedMessages: selectedThreadMessages,
+      hasMoreHistory: selectedThreadHistory.hasMoreHistory,
+      oldestLoadedAt: threadOldestLoadedAt(selectedThreadVisibleTurnItems),
     });
     if (pendingAcknowledgments.length === 0) return feed;
     return appendPendingThreadMessages(feed, feed, pendingAcknowledgments).map((entry) =>

packages/client-runtime/src/state/threadState.test.ts

@@ -0,0 +1,105 @@
+import type { OrchestrationV2ProjectedTurnItem } from "@t3tools/contracts";
+import { describe, expect, it } from "@effect/vitest";
+import * as DateTime from "effect/DateTime";
+
+import { v2ThreadId } from "./orchestrationV2TestFixtures.ts";
+import { threadHasCaughtUpTo, threadOldestLoadedAt } from "./threadState.ts";
+
+const WINDOW = {
+  hasMoreHistory: true,
+  oldestLoadedAt: "2026-09-06T11:00:00.000Z",
+} as const;
+
+describe("threadHasCaughtUpTo", () => {
+  it("prefers the commit sequence over searching paged-out history", () => {
+    expect(
+      threadHasCaughtUpTo({ messageId: "m1", sequence: 40 }, { sequence: 45, messages: [] }),
+    ).toBe(true);
+    expect(
+      threadHasCaughtUpTo(
+        { messageId: "m1", sequence: 50 },
+        { sequence: 45, messages: [{ id: "m1" }] },
+      ),
+    ).toBe(true);
+  });
+
+  it("does not let the window bound override an unreached sequence", () => {
+    expect(
+      threadHasCaughtUpTo(
+        { messageId: "m1", sequence: 50, createdAt: "2026-09-06T10:00:00.000Z" },
+        { sequence: 45, messages: [], ...WINDOW },
+      ),
+    ).toBe(false);
+  });
+
+  it("treats a sequenceless write older than the loaded window as caught up", () => {
+    expect(
+      threadHasCaughtUpTo(
+        { messageId: "m1", createdAt: "2026-09-06T10:00:00.000Z" },
+        { sequence: 45, messages: [], ...WINDOW },
+      ),
+    ).toBe(true);
+    expect(
+      threadHasCaughtUpTo(
+        { messageId: "m1", createdAt: "2026-09-06T12:00:00.000Z" },
+        { sequence: 45, messages: [], ...WINDOW },
+      ),
+    ).toBe(false);
+    expect(
+      threadHasCaughtUpTo(
+        { messageId: "m1", createdAt: "2026-09-06T10:00:00.000Z" },
+        {
+          sequence: 45,
+          messages: [],
+          hasMoreHistory: false,
+          oldestLoadedAt: WINDOW.oldestLoadedAt,
+        },
+      ),
+    ).toBe(false);
+  });
+});
+
+describe("threadOldestLoadedAt", () => {
+  const row = (startedAt: string | null, updatedAt: string): OrchestrationV2ProjectedTurnItem => {
+    const id = "item-0" as never;
+    return {
+      position: 0,
+      visibility: "local",
+      sourceThreadId: v2ThreadId,
+      sourceItemId: id,
+      item: {
+        id,
+        type: "command_execution",
+        threadId: v2ThreadId,
+        runId: null,
+        nodeId: null,
+        providerThreadId: null,
+        providerTurnId: null,
+        nativeItemRef: null,
+        parentItemId: null,
+        ordinal: 1,
+        status: "completed",
+        title: "Command",
+        input: "cmd",
+        output: "out",
+        exitCode: 0,
+        startedAt: startedAt === null ? null : DateTime.makeUnsafe(startedAt),
+        completedAt: null,
+        updatedAt: DateTime.makeUnsafe(updatedAt),
+      },
+    };
+  };
+
+  it("returns the first item's start, falling back to its update time", () => {
+    expect(threadOldestLoadedAt([])).toBeNull();
+    expect(
+      threadOldestLoadedAt([
+        row("2026-09-06T11:00:00.000Z", "2026-09-06T11:05:00.000Z"),
+        row("2026-09-06T12:00:00.000Z", "2026-09-06T12:00:00.000Z"),
+      ]),
+    ).toBe("2026-09-06T11:00:00.000Z");
+    expect(threadOldestLoadedAt([row(null, "2026-09-06T11:05:00.000Z")])).toBe(
+      "2026-09-06T11:05:00.000Z",
+    );
+  });
+});

packages/client-runtime/src/state/threadState.ts

@@ -1,4 +1,5 @@
 import type { OrchestrationV2ThreadProjection } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 import * as Option from "effect/Option";
 
 import { EMPTY_THREAD_HISTORY_META, type ThreadHistoryMeta } from "./threadHistoryMerge.ts";
@@ -28,16 +29,37 @@ export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = {
 
 /**
  * A thread holds a write once it has applied the write's commit sequence. A bounded
- * projection can page the message out later, so finding it is only a fallback for
- * servers that do not report the sequence.
+ * projection can page the message out later, so finding it is a fallback for servers
+ * that do not report the sequence, as is a write older than the loaded window.
  */
 export function threadHasCaughtUpTo(
-  write: { readonly messageId: string; readonly sequence?: number | undefined },
+  write: {
+    readonly messageId: string;
+    readonly sequence?: number | undefined;
+    readonly createdAt?: string | undefined;
+  },
   thread: {
     readonly sequence: number;
     readonly messages: ReadonlyArray<{ readonly id: string }> | undefined;
+    readonly hasMoreHistory?: boolean | undefined;
+    readonly oldestLoadedAt?: string | null | undefined;
   },
 ): boolean {
   if (write.sequence !== undefined && thread.sequence >= write.sequence) return true;
-  return thread.messages?.some((entry) => entry.id === write.messageId) ?? false;
+  if (thread.messages?.some((entry) => entry.id === write.messageId) === true) return true;
+  return (
+    write.sequence === undefined &&
+    write.createdAt !== undefined &&
+    thread.hasMoreHistory === true &&
+    thread.oldestLoadedAt != null &&
+    write.createdAt < thread.oldestLoadedAt
+  );
+}
+
+/** ISO time of the oldest loaded timeline item, the bound sequenceless writes are judged against. */
+export function threadOldestLoadedAt(
+  visibleTurnItems: OrchestrationV2ThreadProjection["visibleTurnItems"],
+): string | null {
+  const oldest = visibleTurnItems[0]?.item;
+  return oldest === undefined ? null : DateTime.formatIso(oldest.startedAt ?? oldest.updatedAt);
 }