From 5d3ffda808f943cf402026be77d31b9a7f348b2f Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Sat, 25 Jul 2026 10:18:41 +0200 Subject: [PATCH] perf: import mobile pagination and bounded replay (upstream #3510) Source: pingdotgg/t3code#3510 Source SHA: 034f4936d7a1435887bb62ac3f2db61f08928cbf 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 #4018 web/server candidate and preserved Tim lifecycle handling plus our mobile composer changes. Excluded: #3510 server/web pagination duplicated by #4018, the later shared-hook refactor, formatting-only commits, and contract comments. The shared refactor can be revisited independently after production validation. --- .github/upstream-candidates.json | 6 + .../features/threads/ThreadDetailScreen.tsx | 6 + .../src/features/threads/ThreadFeed.tsx | 21 +++- .../features/threads/ThreadRouteScreen.tsx | 3 + .../src/state/use-thread-composer-state.ts | 118 +++++++++++++++++- .../Services/OrchestrationEngine.ts | 5 +- apps/server/src/server.test.ts | 78 ++++++++++++ apps/server/src/ws.ts | 73 ++++++++++- 8 files changed, 298 insertions(+), 12 deletions(-) diff --git a/.github/upstream-candidates.json b/.github/upstream-candidates.json index b68afc54b75..bced8e5cc34 100644 --- a/.github/upstream-candidates.json +++ b/.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" } ] } diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 8984e3d2ee7..160d11c3529 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/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} /> ) : ( diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 37a8639fdbd..2e2e5b3e7ba 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/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; + /** 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 : + usesNativeAutomaticInsets && !loadingOlder ? null : ( + + {loadingOlder ? : null} + + ) } contentContainerStyle={{ paddingTop: 12, diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 7f6ec925380..023f2a1fa38 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/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} diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 90831f8437a..a9e2b724017 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/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 = []; + 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 + >([]); + 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(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, diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index f8bcfd76ac0..b224887e465 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/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: ( diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 4db56fb341e..68a8cb3d683 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5955,6 +5955,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(); + 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; + + 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"); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index d47ef6f3549..241340af35c 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -119,6 +119,23 @@ import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; import { deriveLocalBranchNameFromRemoteRef } from "@t3tools/shared/git"; + +const ORCHESTRATION_SUBSCRIPTION_REPLAY_LIMIT = 1_000; + +const subscriptionReplayLimit = ( + afterSequence: number, + snapshotSequence: number, +): number | null => { + const eventCount = snapshotSequence - afterSequence; + if ( + !Number.isSafeInteger(eventCount) || + eventCount < 0 || + eventCount > ORCHESTRATION_SUBSCRIPTION_REPLAY_LIMIT + ) { + return null; + } + return eventCount; +}; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); @@ -1446,14 +1463,60 @@ const makeWsRpcLayer = ( // catch-up followed by the buffered/ongoing live events. Overlapping // events are deduped by sequence on the client. // - // Read the full range after the cursor (not the store's default - // page-bounded limit): the range is normally tiny (a fresh HTTP - // snapshot sequence) and the per-thread filter runs after reading, - // so a global cap could otherwise omit this thread's events. + // A recent cursor replays the exact global range through the + // per-thread filter. A stale cursor receives a current thread + // snapshot instead, keeping catch-up bounded even when the global + // event log is very large. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; + const { snapshotSequence } = yield* projectionSnapshotQuery + .getSnapshotSequence() + .pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration snapshot sequence", + cause, + }), + ), + ); + const replayLimit = subscriptionReplayLimit(afterSequence, snapshotSequence); + + if (replayLimit === null) { + const snapshot = yield* projectionSnapshotQuery + .getThreadDetailSnapshot(input.threadId) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to load thread ${input.threadId}`, + cause, + }), + ), + ); + + if (Option.isNone(snapshot)) { + return yield* new OrchestrationGetSnapshotError({ + message: `Thread ${input.threadId} was not found`, + cause: input.threadId, + }); + } + + const replacementSnapshot = + input.requestCompletionMarker === true + ? Stream.make( + { kind: "snapshot" as const, snapshot: snapshot.value }, + { kind: "synchronized" as const }, + ) + : Stream.make({ + kind: "snapshot" as const, + snapshot: snapshot.value, + }); + return Stream.concat(replacementSnapshot, bufferedLiveStream); + } + const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) + .readEvents(afterSequence, replayLimit) .pipe( Stream.filter(isThisThreadDetailEvent), Stream.map((event) => ({ kind: "event" as const, event })),