perf: import mobile pagination and bounded replay (upstream pingdotgg#3510) (#35)

Source: pingdotgg#3510
Source SHA: 034f493

Imported: native mobile lazy loading for older thread activity, a 1,000-event subscription catch-up ceiling with snapshot fallback, and synchronized stale snapshot watermarks.
Adapted: applied above the refreshed pingdotgg#4018 web/server candidate and preserved Tim lifecycle handling plus our mobile composer changes.
Excluded: pingdotgg#3510 server/web pagination duplicated by pingdotgg#4018, the later shared-hook refactor, formatting-only commits, and contract comments. The shared refactor can be revisited independently after production validation.

Author: patroza

.github/upstream-candidates.json

@@ -6,6 +6,12 @@
       "sourceSha": "de8fd65934768173819b93adcd6b92af3e8c7fc3",
       "status": "active",
       "purpose": "Bound server thread history and lazily page older web activity"
+    },
+    {
+      "upstreamPr": 3510,
+      "sourceSha": "034f4936d7a1435887bb62ac3f2db61f08928cbf",
+      "status": "active",
+      "purpose": "Page mobile history and bound stale subscription catch-up"
     }
   ]
 }

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

@@ -62,6 +62,9 @@ export interface ThreadDetailScreenProps {
   /** Message sync status for the selected thread (drives the composer status pill). */
   readonly threadSyncStatus?: EnvironmentThreadStatus;
   readonly activeThreadBusy: boolean;
+  readonly hasMoreOlderActivities: boolean;
+  readonly loadingOlderActivities: boolean;
+  readonly onLoadOlderActivities: () => void;
   readonly environmentId: EnvironmentId;
   readonly projectWorkspaceRoot: string | null;
   readonly threadCwd: string | null;
@@ -372,6 +375,9 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread
             usesAutomaticContentInsets={props.usesAutomaticContentInsets}
             onHeaderMaterialVisibilityChange={props.onHeaderMaterialVisibilityChange}
             skills={selectedProviderSkills}
+            hasMoreOlder={props.hasMoreOlderActivities}
+            loadingOlder={props.loadingOlderActivities}
+            onLoadOlder={props.onLoadOlderActivities}
           />
         </View>
       ) : (

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

@@ -141,6 +141,10 @@ export interface ThreadFeedProps {
   readonly usesAutomaticContentInsets?: boolean;
   readonly onHeaderMaterialVisibilityChange?: (visible: boolean) => void;
   readonly skills?: ReadonlyArray<SelectableMarkdownSkill>;
+  /** Older history beyond the live activity window can be lazy-loaded on scroll-up. */
+  readonly hasMoreOlder?: boolean;
+  readonly loadingOlder?: boolean;
+  readonly onLoadOlder?: () => void;
 }
 
 function MessageAttachmentImage(props: {
@@ -1498,6 +1502,15 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) {
       ? props.latestTurn.turnId
       : null;
 
+  // Reaching the top (oldest) lazy-loads older history. The hook keys an
+  // in-flight guard by thread, so repeated fires during scroll coalesce.
+  const { hasMoreOlder, loadingOlder, onLoadOlder } = props;
+  const onStartReachedOlderHistory = useCallback(() => {
+    if (hasMoreOlder && !loadingOlder) {
+      onLoadOlder?.();
+    }
+  }, [hasMoreOlder, loadingOlder, onLoadOlder]);
+
   useEffect(() => {
     const previous = previousLatestTurnRef.current;
     previousLatestTurnRef.current = props.latestTurn;
@@ -1790,9 +1803,15 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) {
             estimatedItemSize={180}
             initialScrollAtEnd
             onScroll={handleScroll}
+            onStartReached={onStartReachedOlderHistory}
+            onStartReachedThreshold={0.5}
             scrollEventThrottle={16}
             ListHeaderComponent={
-              usesNativeAutomaticInsets ? null : <View style={{ height: topContentInset }} />
+              usesNativeAutomaticInsets && !loadingOlder ? null : (
+                <View style={{ height: usesNativeAutomaticInsets ? undefined : topContentInset }}>
+                  {loadingOlder ? <ActivityIndicator style={{ marginTop: 8 }} /> : null}
+                </View>
+              )
             }
             contentContainerStyle={{
               paddingTop: 12,

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

@@ -785,6 +785,9 @@ function ThreadRouteContent(
           connectionStateLabel={routeConnectionState}
           threadSyncStatus={selectedThreadDetailState.status}
           activeThreadBusy={composer.activeThreadBusy}
+          hasMoreOlderActivities={composer.hasMoreOlderActivities}
+          loadingOlderActivities={composer.loadingOlderActivities}
+          onLoadOlderActivities={composer.onLoadOlderActivities}
           environmentId={selectedThread.environmentId}
           projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null}
           threadCwd={selectedThreadCwd}

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

@@ -1,11 +1,12 @@
 import { useAtomValue } from "@effect/atom-react";
-import { useCallback, useEffect, useMemo } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 
 import {
   CommandId,
   MessageId,
   type EnvironmentId,
   type ModelSelection,
+  type OrchestrationThreadActivity,
   type ProviderInteractionMode,
   type RuntimeMode,
   type ThreadId,
@@ -36,11 +37,15 @@ import {
   useComposerDraft,
 } from "./use-composer-drafts";
 import { setPendingConnectionError } from "../state/use-remote-environment-registry";
+import { orchestrationEnvironment } from "../state/orchestration";
 import { useSelectedThreadDetail } from "../state/use-thread-detail";
 import { useThreadSelection } from "../state/use-thread-selection";
+import { useAtomCommand } from "./use-atom-command";
 import { enqueueThreadOutboxMessage } from "./thread-outbox";
 import { useThreadOutboxMessages } from "./use-thread-outbox";
 
+const EMPTY_ACTIVITIES: ReadonlyArray<OrchestrationThreadActivity> = [];
+
 export function appendReviewCommentToDraft(input: {
   readonly environmentId: EnvironmentId;
   readonly threadId: ThreadId;
@@ -89,9 +94,113 @@ export function useThreadComposerState() {
     () => (selectedThreadKey ? (queuedMessagesByThreadKey[selectedThreadKey] ?? []) : []),
     [queuedMessagesByThreadKey, selectedThreadKey],
   );
+
+  // ── Older-history lazy-load (mirrors web ChatView) ──────────────────────────
+  // The detail snapshot windows activities to the most recent page (the server
+  // sets `hasMoreActivities`); older pages are fetched on demand and prepended.
+  const [olderActivities, setOlderActivities] = useState<
+    ReadonlyArray<OrchestrationThreadActivity>
+  >([]);
+  const [olderLoaded, setOlderLoaded] = useState(false);
+  const [olderHasMore, setOlderHasMore] = useState(false);
+  const [loadingOlderActivities, setLoadingOlderActivities] = useState(false);
+  const loadThreadActivities = useAtomCommand(orchestrationEnvironment.loadThreadActivities, {
+    reportFailure: false,
+  });
+
+  const activityRequestKey = selectedThreadShell
+    ? `${selectedThreadShell.environmentId}\u0000${selectedThreadShell.id}`
+    : null;
+  const activityRequestKeyRef = useRef(activityRequestKey);
+  activityRequestKeyRef.current = activityRequestKey;
+  useEffect(() => {
+    setOlderActivities([]);
+    setOlderLoaded(false);
+    setOlderHasMore(false);
+    setLoadingOlderActivities(false);
+  }, [activityRequestKey]);
+
+  const liveActivities = selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES;
+  const mergedActivities = useMemo(
+    () => (olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities),
+    [olderActivities, liveActivities],
+  );
+  // Before any page is loaded, the server tells us whether older history exists.
+  const hasMoreOlderActivities = olderLoaded
+    ? olderHasMore
+    : (selectedThreadDetail?.hasMoreActivities ?? false);
+
+  // Synchronous in-flight guard keyed by thread: the list fires onLoadOlder
+  // repeatedly while pinned at the top, but loading *state* only updates next
+  // render, so without this a fast scroll dispatches duplicate same-cursor calls.
+  const inFlightOlderKeyRef = useRef<string | null>(null);
+  const onLoadOlderActivities = useCallback(() => {
+    if (!selectedThreadShell || !hasMoreOlderActivities) {
+      return;
+    }
+    const oldestActivity = mergedActivities[0];
+    if (!oldestActivity || !activityRequestKey) {
+      return;
+    }
+    if (inFlightOlderKeyRef.current === activityRequestKey) {
+      return;
+    }
+    const cursorInput =
+      oldestActivity.sequence !== undefined
+        ? { beforeSequence: oldestActivity.sequence }
+        : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id };
+    const requestKey = activityRequestKey;
+    inFlightOlderKeyRef.current = requestKey;
+    setLoadingOlderActivities(true);
+    void loadThreadActivities({
+      environmentId: selectedThreadShell.environmentId,
+      input: { threadId: selectedThreadShell.id, ...cursorInput },
+    })
+      .then((result) => {
+        if (activityRequestKeyRef.current !== requestKey) {
+          return;
+        }
+        if (result._tag !== "Success") {
+          return;
+        }
+        const page = result.value;
+        setOlderActivities((prev) => {
+          // Dedup against both already-loaded older pages and the live window,
+          // since mobile merges everything into one array (duplicate ids would
+          // produce duplicate React keys in the feed).
+          const seen = new Set(prev.map((activity) => activity.id));
+          for (const activity of liveActivities) {
+            seen.add(activity.id);
+          }
+          const fresh = page.activities.filter((activity) => !seen.has(activity.id));
+          return [...fresh, ...prev];
+        });
+        setOlderLoaded(true);
+        setOlderHasMore(page.hasMore);
+      })
+      .finally(() => {
+        if (inFlightOlderKeyRef.current === requestKey) {
+          inFlightOlderKeyRef.current = null;
+        }
+        if (activityRequestKeyRef.current === requestKey) {
+          setLoadingOlderActivities(false);
+        }
+      });
+  }, [
+    selectedThreadShell,
+    hasMoreOlderActivities,
+    mergedActivities,
+    activityRequestKey,
+    liveActivities,
+    loadThreadActivities,
+  ]);
+
   const selectedThreadFeed = useMemo(
-    () => (selectedThreadDetail ? buildThreadFeed(selectedThreadDetail) : []),
-    [selectedThreadDetail],
+    () =>
+      selectedThreadDetail
+        ? buildThreadFeed({ ...selectedThreadDetail, activities: mergedActivities })
+        : [],
+    [selectedThreadDetail, mergedActivities],
   );
 
   const selectedDraft = selectedThreadKey ? composerDrafts[selectedThreadKey] : null;
@@ -299,6 +408,9 @@ export function useThreadComposerState() {
     runtimeMode,
     interactionMode,
     activeThreadBusy,
+    hasMoreOlderActivities,
+    loadingOlderActivities,
+    onLoadOlderActivities,
     onChangeDraftMessage,
     onPickDraftImages,
     onPasteIntoDraft,

apps/server/src/orchestration/Services/OrchestrationEngine.ts

@@ -27,9 +27,8 @@ export interface OrchestrationEngineShape {
    *
    * @param fromSequenceExclusive - Sequence cursor (exclusive).
    * @param limit - Maximum number of events to read. Defaults to the event
-   *   store's page-bounded default; pass a higher value when the caller must
-   *   read every event after the cursor (e.g. per-thread catch-up that filters
-   *   a small subset out of a potentially larger global range).
+   *   store's page-bounded default. Callers must keep this bounded; use a
+   *   projection snapshot instead of replaying an arbitrarily stale cursor.
    * @returns Stream containing ordered events.
    */
   readonly readEvents: (

apps/server/src/server.test.ts

@@ -5986,6 +5986,84 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
     }).pipe(Effect.provide(NodeHttpServer.layerTest)),
   );
 
+  it.effect("subscribeThread replaces a stale cursor with a fresh snapshot", () =>
+    Effect.gen(function* () {
+      const snapshotSequence = 5_000;
+      const thread = makeDefaultOrchestrationReadModel().threads[0]!;
+      const liveEvents = yield* PubSub.unbounded<OrchestrationEvent>();
+      let replayCalls = 0;
+      const messageEvent = {
+        sequence: snapshotSequence + 1,
+        eventId: EventId.make("event-stale-cursor-message"),
+        aggregateKind: "thread",
+        aggregateId: defaultThreadId,
+        occurredAt: "2026-01-01T00:00:01.000Z",
+        commandId: null,
+        causationEventId: null,
+        correlationId: null,
+        metadata: {},
+        type: "thread.message-sent",
+        payload: {
+          threadId: defaultThreadId,
+          messageId: MessageId.make("message-stale-cursor"),
+          role: "user",
+          text: "Published while loading the replacement snapshot",
+          turnId: null,
+          streaming: false,
+          createdAt: "2026-01-01T00:00:01.000Z",
+          updatedAt: "2026-01-01T00:00:01.000Z",
+        },
+      } satisfies Extract<OrchestrationEvent, { type: "thread.message-sent" }>;
+
+      yield* buildAppUnderTest({
+        layers: {
+          projectionSnapshotQuery: {
+            getSnapshotSequence: () => Effect.succeed({ snapshotSequence }),
+            getThreadDetailSnapshot: () =>
+              Effect.gen(function* () {
+                yield* Effect.sleep("25 millis");
+                yield* PubSub.publish(liveEvents, messageEvent);
+                return Option.some({
+                  snapshotSequence,
+                  thread,
+                });
+              }),
+          },
+          orchestrationEngine: {
+            streamDomainEvents: Stream.fromPubSub(liveEvents),
+            readEvents: () => {
+              replayCalls += 1;
+              return Stream.empty;
+            },
+          },
+        },
+      });
+
+      const wsUrl = yield* getWsServerUrl("/ws");
+      const result = yield* Effect.scoped(
+        withWsRpcClient(wsUrl, (client) =>
+          client[ORCHESTRATION_WS_METHODS.subscribeThread]({
+            threadId: defaultThreadId,
+            afterSequence: 1,
+            requestCompletionMarker: true,
+          }).pipe(Stream.take(3), Stream.runCollect),
+        ),
+      ).pipe(Effect.timeout("2 seconds"));
+
+      assert.equal(replayCalls, 0);
+      assert.equal(result[0]?.kind, "snapshot");
+      if (result[0]?.kind === "snapshot") {
+        assert.equal(result[0].snapshot.snapshotSequence, snapshotSequence);
+        assert.equal(result[0].snapshot.thread.id, defaultThreadId);
+      }
+      assert.deepEqual(result[1], { kind: "synchronized" });
+      assert.equal(result[2]?.kind, "event");
+      if (result[2]?.kind === "event") {
+        assert.equal(result[2].event.sequence, snapshotSequence + 1);
+      }
+    }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive),
+  );
+
   it.effect("subscribeShell coalesces a per-thread burst without stalling other threads", () =>
     Effect.gen(function* () {
       const busyThreadId = ThreadId.make("thread-busy");