authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-06 22:33:49-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-06 19:33:49-07:00
log6b73b3defe1dfb365de3b7bbb97ca56a26b50a43
tree2fb9f75f4313e6765e76e0f69cd5bf246aeb9b99
parent7aad7911f66c2fecba1cfd6601ea783a3fe2bf31
signature Signed by PGP key B5690EEEBB952194

feat: paginate thread loading with user-anchored turn windows (#5493)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

24 files changed, 2093 insertions(+), 30 deletions(-)

apps/mobile/src/connection/environment-cache-store.ts+4-1
......@@ -17,7 +17,10 @@ import * as Schema from "effect/Schema";
1717import * as MobileDatabase from "../persistence/mobile-database";
1818
1919const SHELL_SNAPSHOT_CACHE_SCHEMA_VERSION = 1;
20const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 2;
20// v3 adds windowed (paginated) snapshots carrying `page` metadata; the bump
21// makes pre-pagination clients discard the record instead of decoding a
22// partial thread as complete (rollback safety).
23const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 3;
2124const SERVER_CONFIG_CACHE_SCHEMA_VERSION = 1;
2225const VCS_REFS_CACHE_SCHEMA_VERSION = 1;
2326
apps/mobile/src/features/threads/ThreadDetailScreen.tsx+3
......@@ -61,6 +61,8 @@ export interface ThreadDetailScreenProps {
6161 readonly connectionStateLabel: EnvironmentConnectionPhase;
6262 /** Message sync status for the selected thread (drives the composer status pill). */
6363 readonly threadSyncStatus?: EnvironmentThreadStatus;
64 /** Non-null when older turns exist beyond the loaded window. */
65 readonly loadEarlier?: { readonly loading: boolean; readonly onLoadEarlier: () => void } | null;
6466 readonly activeThreadBusy: boolean;
6567 readonly environmentId: EnvironmentId;
6668 readonly projectWorkspaceRoot: string | null;
......@@ -371,6 +373,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread
371373 usesAutomaticContentInsets={props.usesAutomaticContentInsets}
372374 onHeaderMaterialVisibilityChange={props.onHeaderMaterialVisibilityChange}
373375 skills={selectedProviderSkills}
376 loadEarlier={props.loadEarlier ?? null}
374377 />
375378 </View>
376379 ) : (
apps/mobile/src/features/threads/ThreadFeed.tsx+19-1
......@@ -164,6 +164,11 @@ export interface ThreadFeedProps {
164164 readonly usesAutomaticContentInsets?: boolean;
165165 readonly onHeaderMaterialVisibilityChange?: (visible: boolean) => void;
166166 readonly skills?: ReadonlyArray<SelectableMarkdownSkill>;
167 /** Non-null when older turns exist beyond the loaded window. */
168 readonly loadEarlier?: {
169 readonly loading: boolean;
170 readonly onLoadEarlier: () => void;
171 } | null;
167172}
168173
169174function MessageAttachmentImage(props: {
......@@ -1893,7 +1898,20 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) {
18931898 onScroll={handleScroll}
18941899 scrollEventThrottle={16}
18951900 ListHeaderComponent={
1896 usesNativeAutomaticInsets ? null : <View style={{ height: topContentInset }} />
1901 <>
1902 {usesNativeAutomaticInsets ? null : <View style={{ height: topContentInset }} />}
1903 {props.loadEarlier != null ? (
1904 <Pressable
1905 onPress={props.loadEarlier.onLoadEarlier}
1906 disabled={props.loadEarlier.loading}
1907 className="items-center py-2"
1908 >
1909 <Text className="text-xs text-foreground-secondary">
1910 {props.loadEarlier.loading ? "Loading earlier turns…" : "Load earlier turns"}
1911 </Text>
1912 </Pressable>
1913 ) : null}
1914 </>
18971915 }
18981916 contentContainerStyle={{
18991917 paddingTop: 12,
apps/mobile/src/features/threads/ThreadRouteScreen.tsx+19
......@@ -8,6 +8,10 @@ import {
88import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react";
99import * as Option from "effect/Option";
1010import { EnvironmentId, ThreadId, type ProjectScript } from "@t3tools/contracts";
11import {
12 requestOlderThreadTurns,
13 threadHasOlderTurns,
14} from "@t3tools/client-runtime/state/threads";
1115import { projectScriptCwd, projectScriptRuntimeEnv } from "@t3tools/shared/projectScripts";
1216import { Platform, ScrollView, View } from "react-native";
1317import { useSafeAreaInsets } from "react-native-safe-area-context";
......@@ -190,6 +194,20 @@ function ThreadRouteContent(
190194 useThreadSelection();
191195 const selectedThreadDetailState = props.selectedThreadDetailState;
192196 const selectedThreadDetail = Option.getOrNull(selectedThreadDetailState.data);
197 // "Load earlier turns" header state for windowed (paginated) thread loads.
198 const loadEarlierTurns = useMemo(() => {
199 if (selectedThread === null || !threadHasOlderTurns(selectedThreadDetailState)) {
200 return null;
201 }
202 return {
203 loading:
204 selectedThreadDetailState.page._tag === "Some" &&
205 selectedThreadDetailState.page.value.loadingOlder,
206 onLoadEarlier: () => {
207 requestOlderThreadTurns(selectedThread.environmentId, selectedThread.id);
208 },
209 };
210 }, [selectedThread, selectedThreadDetailState]);
193211 const { selectedThreadCwd } = useSelectedThreadWorktree();
194212 const composer = useThreadComposerState();
195213 const gitState = useSelectedThreadGitState();
......@@ -766,6 +784,7 @@ function ThreadRouteContent(
766784 draftAttachments={composer.draftAttachments}
767785 connectionStateLabel={routeConnectionState}
768786 threadSyncStatus={selectedThreadDetailState.status}
787 loadEarlier={loadEarlierTurns}
769788 activeThreadBusy={composer.activeThreadBusy}
770789 environmentId={selectedThread.environmentId}
771790 projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null}
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+405
......@@ -19,6 +19,7 @@ import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
1919import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts";
2020import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts";
2121import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
22import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts";
2223
2324const asProjectId = (value: string): ProjectId => ProjectId.make(value);
2425const asTurnId = (value: string): TurnId => TurnId.make(value);
......@@ -1917,3 +1918,407 @@ it.effect(
19171918 }).pipe(Effect.provide(layer));
19181919 },
19191920);
1921
1922projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) => {
1923 // A thread shaped like real fan-out usage: user turns interleaved with
1924 // subagent turns (no user pending message), plus a turnless straggler user
1925 // message and a turnless activity anchored between turns.
1926 //
1927 // row turn pending msg anchor (requested_at)
1928 // 1 turn-1 user-msg-1 T00
1929 // 2 turn-2 (subagent) T01
1930 // 3 turn-3 (subagent) T02
1931 // 4 turn-4 user-msg-4 T03
1932 // 5 turn-5 user-msg-5 T04
1933 //
1934 // Straggler user message at T03.5 (turn_id NULL, not any pending_message_id)
1935 // and a turnless activity at T03.6 — both belong to the page containing T03+.
1936 const seedFanOutThread = Effect.fnUntraced(function* () {
1937 const sql = yield* SqlClient.SqlClient;
1938
1939 // Tests in this block share one in-memory database; reset before seeding.
1940 yield* sql`DELETE FROM projection_projects`;
1941 yield* sql`DELETE FROM projection_threads`;
1942 yield* sql`DELETE FROM projection_turns`;
1943 yield* sql`DELETE FROM projection_thread_messages`;
1944 yield* sql`DELETE FROM projection_thread_activities`;
1945 yield* sql`DELETE FROM projection_state`;
1946
1947 yield* sql`
1948 INSERT INTO projection_projects (
1949 project_id, title, workspace_root, scripts_json, created_at, updated_at, deleted_at
1950 )
1951 VALUES ('project-w', 'Windowed', '/tmp/project-w', '[]',
1952 '2026-03-01T00:00:00.000Z', '2026-03-01T00:00:00.000Z', NULL)
1953 `;
1954 yield* sql`
1955 INSERT INTO projection_threads (
1956 thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode,
1957 latest_turn_id, pending_approval_count, pending_user_input_count,
1958 has_actionable_proposed_plan, created_at, updated_at, deleted_at
1959 )
1960 VALUES ('thread-w', 'project-w', 'Windowed thread',
1961 '{"provider":"codex","model":"gpt-5-codex"}', 'full-access', 'default',
1962 'turn-5', 0, 0, 0, '2026-03-01T00:00:00.000Z', '2026-03-01T00:00:10.000Z', NULL)
1963 `;
1964
1965 const turns: ReadonlyArray<{
1966 turn: string;
1967 pendingMessage: string | null;
1968 at: string;
1969 }> = [
1970 { turn: "turn-1", pendingMessage: "user-msg-1", at: "2026-03-01T00:00:00.000Z" },
1971 { turn: "turn-2", pendingMessage: null, at: "2026-03-01T00:01:00.000Z" },
1972 { turn: "turn-3", pendingMessage: null, at: "2026-03-01T00:02:00.000Z" },
1973 { turn: "turn-4", pendingMessage: "user-msg-4", at: "2026-03-01T00:03:00.000Z" },
1974 { turn: "turn-5", pendingMessage: "user-msg-5", at: "2026-03-01T00:04:00.000Z" },
1975 ];
1976 for (const { turn, pendingMessage, at } of turns) {
1977 yield* sql`
1978 INSERT INTO projection_turns (
1979 thread_id, turn_id, pending_message_id, state, requested_at, started_at, completed_at,
1980 checkpoint_files_json
1981 )
1982 VALUES ('thread-w', ${turn}, ${pendingMessage}, 'completed', ${at}, ${at}, ${at}, '[]')
1983 `;
1984 if (pendingMessage !== null) {
1985 yield* sql`
1986 INSERT INTO projection_thread_messages (
1987 message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at
1988 )
1989 VALUES (${pendingMessage}, 'thread-w', NULL, 'user', ${"prompt for " + turn}, 0, ${at}, ${at})
1990 `;
1991 }
1992 yield* sql`
1993 INSERT INTO projection_thread_messages (
1994 message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at
1995 )
1996 VALUES (${turn + "-reply"}, 'thread-w', ${turn}, 'assistant', ${"reply from " + turn}, 0, ${at}, ${at})
1997 `;
1998 yield* sql`
1999 INSERT INTO projection_thread_activities (
2000 activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at
2001 )
2002 VALUES (${turn + "-activity"}, 'thread-w', ${turn}, 'tool', 'tool.completed',
2003 'ran tool', '{"ok":true}', ${at})
2004 `;
2005 }
2006
2007 // Straggler user message sent while turn-4 ran: turn_id NULL and not any
2008 // turn's pending_message_id.
2009 yield* sql`
2010 INSERT INTO projection_thread_messages (
2011 message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at
2012 )
2013 VALUES ('user-msg-straggler', 'thread-w', NULL, 'user', 'while you are at it',
2014 0, '2026-03-01T00:03:30.000Z', '2026-03-01T00:03:30.000Z')
2015 `;
2016 // Turnless activity in the same time range.
2017 yield* sql`
2018 INSERT INTO projection_thread_activities (
2019 activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at
2020 )
2021 VALUES ('turnless-activity', 'thread-w', NULL, 'info', 'context-window.updated',
2022 'usage', '{"usedTokens":1}', '2026-03-01T00:03:36.000Z')
2023 `;
2024
2025 for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) {
2026 yield* sql`
2027 INSERT INTO projection_state (projector, last_applied_sequence, updated_at)
2028 VALUES (${projector}, 42, '2026-03-01T00:00:10.000Z')
2029 `;
2030 }
2031 });
2032
2033 const threadW = ThreadId.make("thread-w");
2034 const messageIds = (snapshot: { thread: { messages: ReadonlyArray<{ id: string }> } }) =>
2035 snapshot.thread.messages.map((message) => message.id).toSorted();
2036 const activityIds = (snapshot: { thread: { activities: ReadonlyArray<{ id: string }> } }) =>
2037 snapshot.thread.activities.map((activity) => activity.id).toSorted();
2038
2039 it.effect("returns the full thread with no page metadata when no window is requested", () =>
2040 Effect.gen(function* () {
2041 yield* seedFanOutThread();
2042 const snapshotQuery = yield* ProjectionSnapshotQuery;
2043
2044 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW);
2045 assert.equal(snapshot._tag, "Some");
2046 if (snapshot._tag === "Some") {
2047 assert.equal(snapshot.value.page, undefined);
2048 assert.equal(snapshot.value.thread.messages.length, 9);
2049 assert.equal(snapshot.value.thread.activities.length, 6);
2050 assert.equal(snapshot.value.snapshotSequence, 42);
2051 }
2052 }),
2053 );
2054
2055 it.effect("windows to the last N user-anchored turns with subagent turns riding along", () =>
2056 Effect.gen(function* () {
2057 yield* seedFanOutThread();
2058 const snapshotQuery = yield* ProjectionSnapshotQuery;
2059
2060 // turnLimit 2 walks back: turn-5 (user), turn-4 (user) -> window is
2061 // rows 4..5. Subagent turns 2-3 are older than the 2nd user turn and
2062 // stay out; the straggler message and turnless activity (T03.5/T03.6,
2063 // after turn-4's anchor) ride along.
2064 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 });
2065 assert.equal(snapshot._tag, "Some");
2066 if (snapshot._tag === "Some") {
2067 assert.deepEqual(messageIds(snapshot.value), [
2068 "turn-4-reply",
2069 "turn-5-reply",
2070 "user-msg-4",
2071 "user-msg-5",
2072 "user-msg-straggler",
2073 ]);
2074 assert.deepEqual(activityIds(snapshot.value), [
2075 "turn-4-activity",
2076 "turn-5-activity",
2077 "turnless-activity",
2078 ]);
2079 assert.equal(snapshot.value.page?.hasMore, true);
2080 assert.notEqual(snapshot.value.page?.beforeCursor, null);
2081 assert.equal(snapshot.value.page?.snapshotSequence, 42);
2082 }
2083 }),
2084 );
2085
2086 it.effect("subagent turns between user turns ride along inside the window", () =>
2087 Effect.gen(function* () {
2088 yield* seedFanOutThread();
2089 const snapshotQuery = yield* ProjectionSnapshotQuery;
2090
2091 // turnLimit 3 reaches user turn-1, dragging subagent turns 2-3 along:
2092 // the full thread, so no further pages.
2093 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 3 });
2094 assert.equal(snapshot._tag, "Some");
2095 if (snapshot._tag === "Some") {
2096 assert.equal(snapshot.value.thread.messages.length, 9);
2097 assert.equal(snapshot.value.thread.activities.length, 6);
2098 assert.equal(snapshot.value.page?.hasMore, false);
2099 assert.equal(snapshot.value.page?.beforeCursor, null);
2100 }
2101 }),
2102 );
2103
2104 it.effect("cursors survive a projection rewrite that reassigns turn row ids", () =>
2105 Effect.gen(function* () {
2106 // The revert projector (and any projection rebuild) deletes and
2107 // re-upserts projection_turns, assigning fresh autoincrement row ids.
2108 // The keyset cursor is derived from event content, so a page cursor
2109 // minted before the rewrite must keep working after it.
2110 yield* seedFanOutThread();
2111 const sql = yield* SqlClient.SqlClient;
2112 const snapshotQuery = yield* ProjectionSnapshotQuery;
2113
2114 const firstPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 });
2115 assert.equal(firstPage._tag, "Some");
2116 if (firstPage._tag !== "Some") return;
2117 const cursor = firstPage.value.page?.beforeCursor;
2118 assert.notEqual(cursor, null);
2119 if (cursor === null || cursor === undefined) return;
2120
2121 // Simulate the rewrite: delete and re-insert every turn row with the
2122 // same content, which reassigns all row ids.
2123 const turnRows = yield* sql`
2124 SELECT thread_id, turn_id, pending_message_id, state, requested_at, started_at,
2125 completed_at, checkpoint_files_json
2126 FROM projection_turns WHERE thread_id = 'thread-w' ORDER BY row_id
2127 `;
2128 yield* sql`DELETE FROM projection_turns WHERE thread_id = 'thread-w'`;
2129 for (const row of turnRows) {
2130 yield* sql`
2131 INSERT INTO projection_turns (
2132 thread_id, turn_id, pending_message_id, state, requested_at, started_at,
2133 completed_at, checkpoint_files_json
2134 )
2135 VALUES (${row.thread_id as string}, ${row.turn_id as string},
2136 ${row.pending_message_id as string | null}, ${row.state as string},
2137 ${row.requested_at as string}, ${row.started_at as string},
2138 ${row.completed_at as string}, ${row.checkpoint_files_json as string})
2139 `;
2140 }
2141
2142 const olderPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2143 turnLimit: 1,
2144 beforeCursor: cursor,
2145 });
2146 assert.equal(olderPage._tag, "Some");
2147 if (olderPage._tag === "Some") {
2148 // Identical older slice to what the pre-rewrite cursor would return.
2149 assert.deepEqual(messageIds(olderPage.value), [
2150 "turn-1-reply",
2151 "turn-2-reply",
2152 "turn-3-reply",
2153 "user-msg-1",
2154 ]);
2155 assert.equal(olderPage.value.page?.hasMore, false);
2156 }
2157 }),
2158 );
2159
2160 it.effect("beforeCursor returns the disjoint adjacent older slice", () =>
2161 Effect.gen(function* () {
2162 yield* seedFanOutThread();
2163 const snapshotQuery = yield* ProjectionSnapshotQuery;
2164
2165 const firstPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 });
2166 assert.equal(firstPage._tag, "Some");
2167 if (firstPage._tag !== "Some") return;
2168 const cursor = firstPage.value.page?.beforeCursor;
2169 assert.notEqual(cursor, null);
2170 assert.notEqual(cursor, undefined);
2171 if (cursor === null || cursor === undefined) return;
2172
2173 // Older page: user turn-1 plus subagent turns 2-3 riding along. Disjoint
2174 // from the first page: no turn-4/5 rows, no straggler.
2175 const olderPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2176 turnLimit: 1,
2177 beforeCursor: cursor,
2178 });
2179 assert.equal(olderPage._tag, "Some");
2180 if (olderPage._tag === "Some") {
2181 assert.deepEqual(messageIds(olderPage.value), [
2182 "turn-1-reply",
2183 "turn-2-reply",
2184 "turn-3-reply",
2185 "user-msg-1",
2186 ]);
2187 assert.deepEqual(activityIds(olderPage.value), [
2188 "turn-1-activity",
2189 "turn-2-activity",
2190 "turn-3-activity",
2191 ]);
2192 assert.equal(olderPage.value.page?.hasMore, false);
2193 assert.equal(olderPage.value.page?.beforeCursor, null);
2194 }
2195 }),
2196 );
2197
2198 it.effect("a cursor for a different thread degrades to the first page", () =>
2199 Effect.gen(function* () {
2200 yield* seedFanOutThread();
2201 const snapshotQuery = yield* ProjectionSnapshotQuery;
2202
2203 const firstPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 });
2204 assert.equal(firstPage._tag, "Some");
2205 if (firstPage._tag !== "Some") return;
2206
2207 const foreign = encodeThreadDetailPageCursor({
2208 threadId: ThreadId.make("thread-other"),
2209 beforeAnchorAt: "2026-03-01T00:01:00.000Z",
2210 beforeTurnId: "turn-2",
2211 });
2212 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2213 turnLimit: 2,
2214 beforeCursor: foreign,
2215 });
2216 assert.equal(snapshot._tag, "Some");
2217 if (snapshot._tag === "Some") {
2218 assert.deepEqual(messageIds(snapshot.value), messageIds(firstPage.value));
2219 }
2220 }),
2221 );
2222
2223 it.effect("a malformed cursor degrades to the first page instead of failing", () =>
2224 Effect.gen(function* () {
2225 yield* seedFanOutThread();
2226 const snapshotQuery = yield* ProjectionSnapshotQuery;
2227
2228 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2229 turnLimit: 2,
2230 beforeCursor: "not-a-cursor",
2231 });
2232 assert.equal(snapshot._tag, "Some");
2233 if (snapshot._tag === "Some") {
2234 assert.equal(snapshot.value.page?.hasMore, true);
2235 assert.equal(snapshot.value.thread.messages.length, 5);
2236 }
2237 }),
2238 );
2239
2240 it.effect("windows never split below the raw-turn ceiling boundary contiguously", () =>
2241 Effect.gen(function* () {
2242 yield* seedFanOutThread();
2243 const snapshotQuery = yield* ProjectionSnapshotQuery;
2244
2245 // Page repeatedly with turnLimit 1 and assert the union of all pages is
2246 // exactly the full thread with no duplicates (disjointness + coverage).
2247 const seenMessages: string[] = [];
2248 const seenActivities: string[] = [];
2249 let cursor: string | undefined;
2250 for (let page = 0; page < 10; page += 1) {
2251 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, {
2252 turnLimit: 1,
2253 ...(cursor !== undefined ? { beforeCursor: cursor } : {}),
2254 });
2255 assert.equal(snapshot._tag, "Some");
2256 if (snapshot._tag !== "Some") return;
2257 seenMessages.push(...snapshot.value.thread.messages.map((message) => message.id));
2258 seenActivities.push(...snapshot.value.thread.activities.map((activity) => activity.id));
2259 const next = snapshot.value.page?.beforeCursor;
2260 if (next === null || next === undefined) break;
2261 cursor = next;
2262 }
2263 assert.equal(new Set(seenMessages).size, seenMessages.length);
2264 assert.equal(new Set(seenActivities).size, seenActivities.length);
2265 assert.equal(seenMessages.length, 9);
2266 assert.equal(seenActivities.length, 6);
2267 }),
2268 );
2269
2270 it.effect("a thread with no turns returns its content unwindowed on the first page", () =>
2271 Effect.gen(function* () {
2272 const sql = yield* SqlClient.SqlClient;
2273 const snapshotQuery = yield* ProjectionSnapshotQuery;
2274
2275 yield* sql`DELETE FROM projection_projects`;
2276 yield* sql`DELETE FROM projection_threads`;
2277 yield* sql`DELETE FROM projection_turns`;
2278 yield* sql`DELETE FROM projection_thread_messages`;
2279 yield* sql`DELETE FROM projection_thread_activities`;
2280 yield* sql`DELETE FROM projection_state`;
2281
2282 yield* sql`
2283 INSERT INTO projection_projects (
2284 project_id, title, workspace_root, scripts_json, created_at, updated_at, deleted_at
2285 )
2286 VALUES ('project-e', 'Empty', '/tmp/project-e', '[]',
2287 '2026-03-02T00:00:00.000Z', '2026-03-02T00:00:00.000Z', NULL)
2288 `;
2289 yield* sql`
2290 INSERT INTO projection_threads (
2291 thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode,
2292 pending_approval_count, pending_user_input_count, has_actionable_proposed_plan,
2293 created_at, updated_at, deleted_at
2294 )
2295 VALUES ('thread-e', 'project-e', 'Turnless thread',
2296 '{"provider":"codex","model":"gpt-5-codex"}', 'full-access', 'default',
2297 0, 0, 0, '2026-03-02T00:00:00.000Z', '2026-03-02T00:00:00.000Z', NULL)
2298 `;
2299 yield* sql`
2300 INSERT INTO projection_thread_messages (
2301 message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at
2302 )
2303 VALUES ('pre-turn-msg', 'thread-e', NULL, 'user', 'first prompt', 0,
2304 '2026-03-02T00:00:01.000Z', '2026-03-02T00:00:01.000Z')
2305 `;
2306 for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) {
2307 yield* sql`
2308 INSERT INTO projection_state (projector, last_applied_sequence, updated_at)
2309 VALUES (${projector}, 7, '2026-03-02T00:00:01.000Z')
2310 `;
2311 }
2312
2313 const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(ThreadId.make("thread-e"), {
2314 turnLimit: 5,
2315 });
2316 assert.equal(snapshot._tag, "Some");
2317 if (snapshot._tag === "Some") {
2318 assert.deepEqual(messageIds(snapshot.value), ["pre-turn-msg"]);
2319 assert.equal(snapshot.value.page?.hasMore, false);
2320 assert.equal(snapshot.value.page?.beforeCursor, null);
2321 }
2322 }),
2323 );
2324});
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+363-6
......@@ -51,6 +51,10 @@ import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionTh
5151import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts";
5252import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts";
5353import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts";
54import {
55 decodeThreadDetailPageCursor,
56 encodeThreadDetailPageCursor,
57} from "../threadDetailCursor.ts";
5458import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts";
5559import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts";
5660import {
......@@ -130,6 +134,36 @@ const ProjectIdLookupInput = Schema.Struct({
130134const ThreadIdLookupInput = Schema.Struct({
131135 threadId: ThreadId,
132136});
137// Windowed reads order turns by the stable keyset (anchor, turn key), where
138// anchor is requested_at and turn key is
139// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the
140// revert projector's row-id rewrite and full projection rebuilds.
141const ThreadTurnWindowLookupInput = Schema.Struct({
142 threadId: ThreadId,
143 // Exclusive keyset upper bound. Sentinels "~"/"" mean unbounded ("~" sorts
144 // after every ISO timestamp).
145 beforeAnchorAt: Schema.String,
146 beforeTurnKey: Schema.String,
147 userTurnLimit: Schema.Number,
148 maxRawTurns: Schema.Number,
149});
150const ProjectionTurnWindowRowSchema = Schema.Struct({
151 // The turn's timeline anchor, used to bound rows that have no turn linkage
152 // (user messages and turnless activities) to the same page window.
153 anchorAt: Schema.String,
154 turnKey: Schema.String,
155});
156const ThreadTurnRangeLookupInput = Schema.Struct({
157 threadId: ThreadId,
158 // Turn-linked rows are bounded by the keyset range [min, before) over
159 // (anchor, turn key); turnless rows by the matching [minAnchorAt,
160 // beforeAnchorAt) time range. Unbounded ends use sentinels: "" for the
161 // lower bound, "~" (sorts after ISO dates) for the upper bound.
162 minAnchorAt: Schema.String,
163 minTurnKey: Schema.String,
164 beforeAnchorAt: Schema.String,
165 beforeTurnKey: Schema.String,
166});
133167const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema;
134168const ProjectionThreadIdLookupRowSchema = Schema.Struct({
135169 threadId: ThreadId,
......@@ -1043,6 +1077,197 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
10431077 `,
10441078 });
10451079
1080 // Resolves a page of recent turns for a windowed thread detail read. Walks
1081 // back from the exclusive (beforeAnchorAt, beforeTurnKey) keyset boundary
1082 // (sentinels "~"/"" mean unbounded, i.e. the first page) until it has seen
1083 // `userTurnLimit` user-anchored turns — turns whose pending message is a
1084 // user message; subagent/fan-out turns between them ride along — or hits the
1085 // `maxRawTurns` ceiling that bounds pathological fan-out. The `candidates`
1086 // CTE applies the keyset bound and LIMIT before the window functions run;
1087 // its ORDER BY uses raw columns so the migration-037
1088 // (thread_id, requested_at, turn_id) index serves both range and order with
1089 // no temp B-tree — the scan is genuinely bounded by the LIMIT. (Raw
1090 // turn_id DESC places NULLs exactly where COALESCE-to-'' would, below every
1091 // real id.) The caller derives the continuation cursor from the oldest
1092 // returned row.
1093 // Highest thread-DETAIL event sequence for this thread that the projection
1094 // has applied (bounded by the global snapshot sequence read in the same
1095 // transaction). This is the thread-scoped watermark a windowed page carries
1096 // so clients can defer merging until their live subscription has caught up;
1097 // the global sequence is not waitable per-thread. The event_type filter
1098 // must match ws.ts's isThreadDetailEvent exactly: the subscription only
1099 // delivers these types, so a watermark counting any other event could
1100 // never be reached by the client and would park the page forever. Served
1101 // by the event store's (aggregate_kind, stream_id, sequence) index.
1102 const getThreadEventWatermarkRow = SqlSchema.findOneOption({
1103 Request: Schema.Struct({ threadId: ThreadId, maxSequence: Schema.Number }),
1104 Result: Schema.Struct({ threadSequence: Schema.NullOr(Schema.Number) }),
1105 execute: ({ threadId, maxSequence }) =>
1106 sql`
1107 SELECT MAX(sequence) AS "threadSequence"
1108 FROM orchestration_events
1109 WHERE aggregate_kind = 'thread'
1110 AND stream_id = ${threadId}
1111 AND sequence <= ${maxSequence}
1112 AND event_type IN (
1113 'thread.message-sent',
1114 'thread.proposed-plan-upserted',
1115 'thread.activity-appended',
1116 'thread.turn-diff-completed',
1117 'thread.reverted',
1118 'thread.session-set'
1119 )
1120 `,
1121 });
1122
1123 const listTurnWindowRows = SqlSchema.findAll({
1124 Request: ThreadTurnWindowLookupInput,
1125 Result: ProjectionTurnWindowRowSchema,
1126 execute: ({ threadId, beforeAnchorAt, beforeTurnKey, userTurnLimit, maxRawTurns }) =>
1127 sql`
1128 WITH candidates AS (
1129 SELECT
1130 turns.requested_at AS anchor_at,
1131 COALESCE(turns.turn_id, '') AS turn_key,
1132 turns.pending_message_id
1133 FROM projection_turns AS turns
1134 WHERE turns.thread_id = ${threadId}
1135 AND (
1136 turns.requested_at < ${beforeAnchorAt}
1137 OR (
1138 turns.requested_at = ${beforeAnchorAt}
1139 AND COALESCE(turns.turn_id, '') < ${beforeTurnKey}
1140 )
1141 )
1142 ORDER BY turns.requested_at DESC, turns.turn_id DESC
1143 LIMIT ${maxRawTurns}
1144 ),
1145 walked AS (
1146 SELECT
1147 candidates.anchor_at,
1148 candidates.turn_key,
1149 CASE WHEN messages.role = 'user' THEN 1 ELSE 0 END AS is_user_turn,
1150 SUM(CASE WHEN messages.role = 'user' THEN 1 ELSE 0 END) OVER (
1151 ORDER BY candidates.anchor_at DESC, candidates.turn_key DESC
1152 ) AS user_turns_seen
1153 FROM candidates
1154 LEFT JOIN projection_thread_messages AS messages
1155 ON messages.message_id = candidates.pending_message_id
1156 )
1157 SELECT
1158 anchor_at AS "anchorAt",
1159 turn_key AS "turnKey"
1160 FROM walked
1161 WHERE user_turns_seen < ${userTurnLimit}
1162 OR (user_turns_seen = ${userTurnLimit} AND is_user_turn = 1)
1163 ORDER BY anchor_at ASC, turn_key ASC
1164 `,
1165 });
1166
1167 // Windowed variants of the two heavy collections. Turn-linked rows are
1168 // bounded by the page's (anchor, turn key) keyset range over
1169 // projection_turns; rows with no turn linkage (user messages always, and
1170 // turnless activities like pre-turn context-window updates) are bounded by
1171 // the matching turn-anchor time range so they land on the same page as the
1172 // turns around them. Proposed plans and checkpoints stay unwindowed: they
1173 // are metadata-scale.
1174 const listThreadMessageRowsByThreadWindow = SqlSchema.findAll({
1175 Request: ThreadTurnRangeLookupInput,
1176 Result: ProjectionThreadMessageDbRowSchema,
1177 execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
1178 sql`
1179 SELECT
1180 message_id AS "messageId",
1181 thread_id AS "threadId",
1182 turn_id AS "turnId",
1183 role,
1184 text,
1185 attachments_json AS "attachments",
1186 is_streaming AS "isStreaming",
1187 created_at AS "createdAt",
1188 updated_at AS "updatedAt"
1189 FROM projection_thread_messages
1190 WHERE thread_id = ${threadId}
1191 AND (
1192 turn_id IN (
1193 SELECT turn_id FROM projection_turns
1194 WHERE thread_id = ${threadId}
1195 AND turn_id IS NOT NULL
1196 AND (
1197 requested_at > ${minAnchorAt}
1198 OR (
1199 requested_at = ${minAnchorAt}
1200 AND turn_id >= ${minTurnKey}
1201 )
1202 )
1203 AND (
1204 requested_at < ${beforeAnchorAt}
1205 OR (
1206 requested_at = ${beforeAnchorAt}
1207 AND turn_id < ${beforeTurnKey}
1208 )
1209 )
1210 )
1211 OR (
1212 turn_id IS NULL
1213 AND created_at >= ${minAnchorAt}
1214 AND created_at < ${beforeAnchorAt}
1215 )
1216 )
1217 ORDER BY created_at ASC, message_id ASC
1218 `,
1219 });
1220
1221 const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({
1222 Request: ThreadTurnRangeLookupInput,
1223 Result: ProjectionThreadActivityDbRowSchema,
1224 execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) =>
1225 sql`
1226 SELECT
1227 activity_id AS "activityId",
1228 thread_id AS "threadId",
1229 turn_id AS "turnId",
1230 tone,
1231 kind,
1232 summary,
1233 payload_json AS "payload",
1234 sequence,
1235 created_at AS "createdAt"
1236 FROM projection_thread_activities
1237 WHERE thread_id = ${threadId}
1238 AND (
1239 turn_id IN (
1240 SELECT turn_id FROM projection_turns
1241 WHERE thread_id = ${threadId}
1242 AND turn_id IS NOT NULL
1243 AND (
1244 requested_at > ${minAnchorAt}
1245 OR (
1246 requested_at = ${minAnchorAt}
1247 AND turn_id >= ${minTurnKey}
1248 )
1249 )
1250 AND (
1251 requested_at < ${beforeAnchorAt}
1252 OR (
1253 requested_at = ${beforeAnchorAt}
1254 AND turn_id < ${beforeTurnKey}
1255 )
1256 )
1257 )
1258 OR (
1259 turn_id IS NULL
1260 AND created_at >= ${minAnchorAt}
1261 AND created_at < ${beforeAnchorAt}
1262 )
1263 )
1264 ORDER BY
1265 sequence ASC,
1266 created_at ASC,
1267 activity_id ASC
1268 `,
1269 });
1270
10461271 const getFullThreadDiffContextRow = SqlSchema.findOneOption({
10471272 Request: FullThreadDiffContextLookupInput,
10481273 Result: ProjectionFullThreadDiffContextRowSchema,
......@@ -2104,7 +2329,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
21042329 } satisfies OrchestrationThreadShell);
21052330 });
21062331
2107 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (threadId) =>
2332 // Contiguous turn range bounding a windowed detail read; undefined loads the
2333 // full thread. Resolved from a window request inside the snapshot
2334 // transaction (see getThreadDetailSnapshot).
2335 interface ThreadDetailBounds {
2336 readonly minAnchorAt: string;
2337 readonly minTurnKey: string;
2338 readonly beforeAnchorAt: string;
2339 readonly beforeTurnKey: string;
2340 }
2341
2342 const getThreadDetailByIdBounded = (threadId: ThreadId, bounds: ThreadDetailBounds | undefined) =>
21082343 Effect.gen(function* () {
21092344 const [
21102345 threadRow,
......@@ -2123,7 +2358,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
21232358 ),
21242359 ),
21252360 ),
2126 listThreadMessageRowsByThread({ threadId }).pipe(
2361 (bounds === undefined
2362 ? listThreadMessageRowsByThread({ threadId })
2363 : listThreadMessageRowsByThreadWindow({ threadId, ...bounds })
2364 ).pipe(
21272365 Effect.mapError(
21282366 toPersistenceSqlOrDecodeError(
21292367 "ProjectionSnapshotQuery.getThreadDetailById:listMessages:query",
......@@ -2139,7 +2377,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
21392377 ),
21402378 ),
21412379 ),
2142 listThreadActivityRowsByThread({ threadId }).pipe(
2380 (bounds === undefined
2381 ? listThreadActivityRowsByThread({ threadId })
2382 : listThreadActivityRowsByThreadWindow({ threadId, ...bounds })
2383 ).pipe(
21432384 Effect.mapError(
21442385 toPersistenceSqlOrDecodeError(
21452386 "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query",
......@@ -2249,23 +2490,139 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
22492490 );
22502491 });
22512492
2493 const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (threadId) =>
2494 getThreadDetailByIdBounded(threadId, undefined);
2495
2496 // Bounds pathological fan-out: one user turn that spawned hundreds of
2497 // subagent turns still pages in bounded chunks, at the cost of splitting the
2498 // fan-out group across pages (the cursor continues the same group). Also
2499 // structurally bounds the window scan via the candidates CTE's LIMIT.
2500 const THREAD_DETAIL_MAX_RAW_TURNS_PER_PAGE = 150;
2501 // Sentinels for unbounded keyset ends; "~" sorts after any ISO timestamp.
2502 const ANCHOR_UNBOUNDED = "~";
2503
22522504 const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = (
22532505 threadId,
2506 window,
22542507 ) =>
22552508 // Read the thread detail and the snapshot sequence within a single
22562509 // transaction so the sequence is consistent with the returned state; a
22572510 // projector update landing between two separate reads could otherwise return
22582511 // a sequence ahead of the thread detail, causing the client to resume from
2259 // too far and drop events.
2512 // too far and drop events. Window resolution runs inside the same
2513 // transaction so the page boundary is consistent with the returned rows.
22602514 sql
22612515 .withTransaction(
22622516 Effect.gen(function* () {
2263 const thread = yield* getThreadDetailById(threadId);
2517 if (window?.turnLimit === undefined) {
2518 const thread = yield* getThreadDetailById(threadId);
2519 if (Option.isNone(thread)) {
2520 return Option.none<OrchestrationThreadDetailSnapshot>();
2521 }
2522 const { snapshotSequence } = yield* getSnapshotSequence();
2523 return Option.some({ snapshotSequence, thread: thread.value });
2524 }
2525
2526 // A malformed or foreign-thread cursor falls back to the first page
2527 // rather than failing: the client's stale cursor after a revert or
2528 // reconnect should degrade to "reload recent history", not error.
2529 const decodedCursor =
2530 window.beforeCursor === undefined
2531 ? null
2532 : decodeThreadDetailPageCursor(window.beforeCursor);
2533 const cursor = decodedCursor?.threadId === threadId ? decodedCursor : null;
2534
2535 const windowRows = yield* listTurnWindowRows({
2536 threadId,
2537 beforeAnchorAt: cursor?.beforeAnchorAt ?? ANCHOR_UNBOUNDED,
2538 beforeTurnKey: cursor?.beforeTurnId ?? "",
2539 userTurnLimit: window.turnLimit,
2540 maxRawTurns: THREAD_DETAIL_MAX_RAW_TURNS_PER_PAGE,
2541 }).pipe(
2542 Effect.mapError(
2543 toPersistenceSqlOrDecodeError(
2544 "ProjectionSnapshotQuery.getThreadDetailSnapshot:listTurnWindow:query",
2545 "ProjectionSnapshotQuery.getThreadDetailSnapshot:listTurnWindow:decodeRows",
2546 ),
2547 ),
2548 );
2549
2550 const oldest = windowRows[0];
2551 // An empty window (no turns before the cursor, or a thread with no
2552 // turns at all) still returns thread metadata with empty collections
2553 // for turn-linked rows; turnless rows are bounded to the same empty
2554 // range. The first page of a turnless thread stays unwindowed so
2555 // pre-turn content (e.g. a just-created thread) is not hidden.
2556 const bounds: ThreadDetailBounds | undefined =
2557 oldest === undefined && cursor === null
2558 ? undefined
2559 : {
2560 minAnchorAt: oldest?.anchorAt ?? "",
2561 minTurnKey: oldest?.turnKey ?? "",
2562 beforeAnchorAt: cursor?.beforeAnchorAt ?? ANCHOR_UNBOUNDED,
2563 beforeTurnKey: cursor?.beforeTurnId ?? "",
2564 };
2565 // Empty window behind a cursor: nothing older remains.
2566 const emptyBounds =
2567 oldest === undefined && cursor !== null
2568 ? { minAnchorAt: "", minTurnKey: "", beforeAnchorAt: "", beforeTurnKey: "" }
2569 : undefined;
2570
2571 const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds);
22642572 if (Option.isNone(thread)) {
22652573 return Option.none<OrchestrationThreadDetailSnapshot>();
22662574 }
2575
2576 const hasMore =
2577 oldest !== undefined &&
2578 (yield* listTurnWindowRows({
2579 threadId,
2580 beforeAnchorAt: oldest.anchorAt,
2581 beforeTurnKey: oldest.turnKey,
2582 userTurnLimit: 1,
2583 maxRawTurns: 1,
2584 }).pipe(
2585 Effect.mapError(
2586 toPersistenceSqlOrDecodeError(
2587 "ProjectionSnapshotQuery.getThreadDetailSnapshot:probeOlder:query",
2588 "ProjectionSnapshotQuery.getThreadDetailSnapshot:probeOlder:decodeRows",
2589 ),
2590 ),
2591 )).length > 0;
2592
22672593 const { snapshotSequence } = yield* getSnapshotSequence();
2268 return Option.some({ snapshotSequence, thread: thread.value });
2594 const watermarkRow = yield* getThreadEventWatermarkRow({
2595 threadId,
2596 maxSequence: snapshotSequence,
2597 }).pipe(
2598 Effect.mapError(
2599 toPersistenceSqlOrDecodeError(
2600 "ProjectionSnapshotQuery.getThreadDetailSnapshot:threadWatermark:query",
2601 "ProjectionSnapshotQuery.getThreadDetailSnapshot:threadWatermark:decodeRow",
2602 ),
2603 ),
2604 );
2605 const threadSequence = Option.match(watermarkRow, {
2606 onNone: () => 0,
2607 onSome: (row) => row.threadSequence ?? 0,
2608 });
2609 return Option.some({
2610 snapshotSequence,
2611 thread: thread.value,
2612 page: {
2613 beforeCursor:
2614 hasMore && oldest !== undefined
2615 ? encodeThreadDetailPageCursor({
2616 threadId,
2617 beforeAnchorAt: oldest.anchorAt,
2618 beforeTurnId: oldest.turnKey,
2619 })
2620 : null,
2621 hasMore,
2622 snapshotSequence,
2623 threadSequence,
2624 },
2625 });
22692626 }),
22702627 )
22712628 .pipe(
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+8
......@@ -17,6 +17,7 @@ import type {
1717 OrchestrationShellSnapshot,
1818 OrchestrationThread,
1919 OrchestrationThreadDetailSnapshot,
20 OrchestrationThreadDetailWindow,
2021 OrchestrationThreadShell,
2122 ProjectId,
2223 ThreadId,
......@@ -174,9 +175,16 @@ export interface ProjectionSnapshotQueryShape {
174175 * sequence in one consistent transaction, so the returned `snapshotSequence`
175176 * exactly matches the state reflected in `thread` (no interleaving projector
176177 * update between the two reads).
178 *
179 * When `window` is provided, the thread's messages, activities, proposed
180 * plans, and checkpoints are bounded to a page of recent turns and the
181 * response carries `page` metadata (see `OrchestrationThreadDetailWindow`).
182 * Without a window the full thread is returned with no `page` field —
183 * pagination is strictly opt-in.
177184 */
178185 readonly getThreadDetailSnapshot: (
179186 threadId: ThreadId,
187 window?: OrchestrationThreadDetailWindow,
180188 ) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>, ProjectionRepositoryError>;
181189}
182190
apps/server/src/orchestration/http.ts+11-1
......@@ -66,7 +66,17 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group(
6666 yield* annotateEnvironmentRequest(args.endpoint.name);
6767 yield* requireEnvironmentScope(AuthOrchestrationReadScope);
6868 const snapshot = yield* projectionSnapshotQuery
69 .getThreadDetailSnapshot(args.params.threadId)
69 .getThreadDetailSnapshot(
70 args.params.threadId,
71 args.payload.turnLimit === undefined
72 ? undefined
73 : {
74 turnLimit: args.payload.turnLimit,
75 ...(args.payload.beforeCursor !== undefined
76 ? { beforeCursor: args.payload.beforeCursor }
77 : {}),
78 },
79 )
7080 .pipe(
7181 Effect.catch((cause) =>
7282 failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
apps/server/src/orchestration/threadDetailCursor.test.ts created+44
......@@ -0,0 +1,44 @@
1import { ThreadId } from "@t3tools/contracts";
2import { describe, expect, it } from "@effect/vitest";
3
4import {
5 decodeThreadDetailPageCursor,
6 encodeThreadDetailPageCursor,
7} from "./threadDetailCursor.ts";
8
9describe("threadDetailCursor", () => {
10 it("round-trips a cursor", () => {
11 const cursor = {
12 threadId: ThreadId.make("thread-1"),
13 beforeAnchorAt: "2026-08-01T00:00:00.000Z",
14 beforeTurnId: "turn-9",
15 };
16 expect(decodeThreadDetailPageCursor(encodeThreadDetailPageCursor(cursor))).toEqual(cursor);
17 });
18
19 it("round-trips empty boundary values", () => {
20 // The anchor is COALESCE(requested_at, started_at, '') and the turn key
21 // is COALESCE(turn_id, ''), so a server-minted cursor can legitimately
22 // carry empty strings; rejecting them would degrade a valid cursor to a
23 // first-page request that repeats recent history (review finding).
24 const cursor = {
25 threadId: ThreadId.make("thread-1"),
26 beforeAnchorAt: "",
27 beforeTurnId: "",
28 };
29 expect(decodeThreadDetailPageCursor(encodeThreadDetailPageCursor(cursor))).toEqual(cursor);
30 });
31
32 it("rejects malformed input", () => {
33 expect(decodeThreadDetailPageCursor("not-base64-json")).toBeNull();
34 expect(decodeThreadDetailPageCursor(Buffer.from("[]").toString("base64url"))).toBeNull();
35 expect(
36 decodeThreadDetailPageCursor(Buffer.from(JSON.stringify({ t: "" })).toString("base64url")),
37 ).toBeNull();
38 expect(
39 decodeThreadDetailPageCursor(
40 Buffer.from(JSON.stringify({ t: "thread-1", a: 5, i: "x" })).toString("base64url"),
41 ),
42 ).toBeNull();
43 });
44});
apps/server/src/orchestration/threadDetailCursor.ts created+62
......@@ -0,0 +1,62 @@
1import type { ThreadId } from "@t3tools/contracts";
2
3/**
4 * Opaque, exclusive cursor for windowed thread detail reads. Encodes the thread
5 * id and the keyset boundary of an already-delivered page: the boundary turn's
6 * anchor timestamp (`COALESCE(requested_at, started_at, '')`) and turn id.
7 * Passing it back requests the adjacent disjoint slice of strictly older turns
8 * under `(anchor, turn_id)` ordering.
9 *
10 * The boundary is deliberately NOT a `projection_turns.row_id`: row ids are
11 * rewritten by the revert projector (delete + re-upsert) and by projection
12 * rebuilds, which would silently invalidate every persisted cursor with no
13 * event emitted. The (anchor, turnId) pair is derived from event content, so
14 * cursors survive both and no client-side refresh machinery is needed. The
15 * anchor doubles as the time bound for rows with no turn linkage (straggler
16 * user messages, turnless activities). The thread id is embedded so a cursor
17 * can never be replayed against a different thread. Clients must treat the
18 * string as opaque.
19 */
20export interface ThreadDetailPageCursor {
21 readonly threadId: ThreadId;
22 readonly beforeAnchorAt: string;
23 /** Boundary turn id; "" for the rare turn row with a null turn_id. */
24 readonly beforeTurnId: string;
25}
26
27export function encodeThreadDetailPageCursor(cursor: ThreadDetailPageCursor): string {
28 return Buffer.from(
29 JSON.stringify({ t: cursor.threadId, a: cursor.beforeAnchorAt, i: cursor.beforeTurnId }),
30 ).toString("base64url");
31}
32
33/**
34 * Returns null for anything that is not a well-formed cursor. Callers degrade
35 * a malformed or foreign-thread cursor to a first-page request.
36 */
37export function decodeThreadDetailPageCursor(encoded: string): ThreadDetailPageCursor | null {
38 let parsed: unknown;
39 try {
40 parsed = JSON.parse(Buffer.from(encoded, "base64url").toString("utf8"));
41 } catch {
42 return null;
43 }
44 if (parsed === null || typeof parsed !== "object") {
45 return null;
46 }
47 const record = parsed as Record<string, unknown>;
48 if (typeof record.t !== "string" || record.t.length === 0) {
49 return null;
50 }
51 // Empty strings are valid boundary values, not malformed input: the anchor
52 // is COALESCE(requested_at, started_at, ''), so a boundary turn with no
53 // timestamps encodes a: "" (and sorts before every real anchor, correctly
54 // ending the walk); the turn key is "" for a null turn_id.
55 if (typeof record.a !== "string") {
56 return null;
57 }
58 if (typeof record.i !== "string") {
59 return null;
60 }
61 return { threadId: record.t as ThreadId, beforeAnchorAt: record.a, beforeTurnId: record.i };
62}
apps/server/src/persistence/Migrations.ts+2
......@@ -49,6 +49,7 @@ import Migration0033 from "./Migrations/033_ProjectionThreadsSettled.ts";
4949import Migration0034 from "./Migrations/034_ProjectionThreadsSnoozed.ts";
5050import Migration0035 from "./Migrations/035_ProjectionThreadTitleRegeneration.ts";
5151import Migration0036 from "./Migrations/036_ProjectionThreadsPinned.ts";
52import Migration0037 from "./Migrations/037_ProjectionTurnsKeysetIndex.ts";
5253
5354/**
5455 * Migration loader with all migrations defined inline.
......@@ -97,6 +98,7 @@ export const migrationEntries = [
9798 [34, "ProjectionThreadsSnoozed", Migration0034],
9899 [35, "ProjectionThreadTitleRegeneration", Migration0035],
99100 [36, "ProjectionThreadsPinned", Migration0036],
101 [37, "ProjectionTurnsKeysetIndex", Migration0037],
100102] as const;
101103
102104export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);
apps/server/src/persistence/Migrations/037_ProjectionTurnsKeysetIndex.ts created+17
......@@ -0,0 +1,17 @@
1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/unstable/sql/SqlClient";
3
4/**
5 * Composite index for windowed thread detail reads. Pagination orders turns by
6 * the stable keyset (requested_at, turn_id); the pre-existing
7 * (thread_id, requested_at) index cannot serve the tiebreak order, forcing a
8 * temp B-tree over all of a thread's turns before the page LIMIT applies.
9 * With this index the candidates scan is genuinely bounded by the page size.
10 */
11export default Effect.gen(function* () {
12 const sql = yield* SqlClient.SqlClient;
13 yield* sql`
14 CREATE INDEX IF NOT EXISTS idx_projection_turns_thread_keyset
15 ON projection_turns(thread_id, requested_at, turn_id)
16 `;
17});
apps/server/src/ws.ts+9-1
......@@ -1019,6 +1019,7 @@ const makeWsRpcLayer = (
10191019 settings,
10201020 shellResumeCompletionMarker: true,
10211021 threadResumeCompletionMarker: true,
1022 threadSnapshotPagination: true,
10221023 };
10231024 });
10241025
......@@ -1351,7 +1352,14 @@ const makeWsRpcLayer = (
13511352 }
13521353
13531354 const snapshot = yield* projectionSnapshotQuery
1354 .getThreadDetailSnapshot(input.threadId)
1355 .getThreadDetailSnapshot(
1356 input.threadId,
1357 // Windowing the fallback snapshot is opt-in per subscription:
1358 // clients that don't send turnLimit (including all
1359 // pre-pagination clients) get the full thread, since they
1360 // have no way to load older pages.
1361 input.turnLimit === undefined ? undefined : { turnLimit: input.turnLimit },
1362 )
13551363 .pipe(
13561364 Effect.mapError(
13571365 (cause) =>
apps/web/src/components/ChatView.tsx+23-1
......@@ -223,7 +223,11 @@ import {
223223 serverEnvironment,
224224} from "../state/server";
225225import { terminalEnvironment } from "../state/terminal";
226import { threadEnvironment } from "../state/threads";
226import { threadEnvironment, useEnvironmentThread } from "../state/threads";
227import {
228 requestOlderThreadTurns,
229 threadHasOlderTurns,
230} from "@t3tools/client-runtime/state/threads";
227231import { vcsEnvironment } from "../state/vcs";
228232import { useEnvironments, usePrimaryEnvironment } from "../state/environments";
229233import {
......@@ -1239,6 +1243,23 @@ function ChatViewContent(props: ChatViewProps) {
12391243 [routeServerThreadShell, threadDetailLoading],
12401244 );
12411245 const activeServerThread = serverThread ?? loadingServerThread;
1246 // Pagination window state for the routed server thread: drives the
1247 // "load earlier turns" header when the loaded window has older history.
1248 const routeThreadState = useEnvironmentThread(
1249 routeKind === "server" ? routeThreadRef.environmentId : null,
1250 routeKind === "server" ? routeThreadRef.threadId : null,
1251 );
1252 const loadEarlierTurns = useMemo(() => {
1253 if (routeKind !== "server" || !threadHasOlderTurns(routeThreadState)) {
1254 return null;
1255 }
1256 return {
1257 loading: routeThreadState.page._tag === "Some" && routeThreadState.page.value.loadingOlder,
1258 onLoadEarlier: () => {
1259 requestOlderThreadTurns(routeThreadRef.environmentId, routeThreadRef.threadId);
1260 },
1261 };
1262 }, [routeKind, routeThreadRef, routeThreadState]);
12421263 const markThreadVisited = useUiStateStore((store) => store.markThreadVisited);
12431264 const settings = useEnvironmentSettings(environmentId);
12441265 // New-thread defaults live in the primary environment's settings.json (the
......@@ -6029,6 +6050,7 @@ function ChatViewContent(props: ChatViewProps) {
60296050 onManualNavigation={cancelTimelineLiveFollowForUserNavigation}
60306051 hideEmptyPlaceholder={isDraftHeroState || threadDetailLoading}
60316052 topFadeEnabled={!hasTimelineTopBanner}
6053 loadEarlier={loadEarlierTurns}
60326054 />
60336055
60346056 {/* scroll to end pill — shown when user has scrolled away from the live edge */}
apps/web/src/components/chat/MessagesTimeline.tsx+43-1
......@@ -158,6 +158,33 @@ const TimelineRowCtx = createContext<TimelineRowSharedState>(null!);
158158const TimelineRowActivityCtx = createContext<TimelineRowActivityState>(null!);
159159const TIMELINE_LIST_HEADER = <div className="h-3 sm:h-4" />;
160160const TIMELINE_LIST_FADE_HEADER = <div className="h-10 sm:h-12" />;
161
162// Header row shown when older turns exist beyond the loaded window. Plain
163// button, no spinner animation; the label change is the loading indicator.
164function TimelineLoadEarlierHeader({
165 loading,
166 onLoadEarlier,
167 fade,
168}: {
169 loading: boolean;
170 onLoadEarlier: () => void;
171 fade: boolean;
172}) {
173 return (
174 <div className={fade ? "pt-10 sm:pt-12" : "pt-3 sm:pt-4"}>
175 <div className="mx-auto w-full max-w-3xl pb-2">
176 <button
177 type="button"
178 onClick={onLoadEarlier}
179 disabled={loading}
180 className="w-full py-1.5 text-xs text-muted-foreground/60 hover:text-foreground disabled:cursor-default"
181 >
182 {loading ? "Loading earlier turns…" : "Load earlier turns"}
183 </button>
184 </div>
185 </div>
186 );
187}
161188const TIMELINE_LIST_FOOTER = <div className="h-3 sm:h-4" />;
162189const EMPTY_TIMELINE_SKILLS: ReadonlyArray<Pick<ServerProviderSkill, "name" | "displayName">> = [];
163190
......@@ -196,6 +223,8 @@ interface MessagesTimelineProps {
196223 onManualNavigation: () => void;
197224 hideEmptyPlaceholder?: boolean;
198225 topFadeEnabled?: boolean;
226 /** Non-null when older turns exist beyond the loaded window. */
227 loadEarlier?: { readonly loading: boolean; readonly onLoadEarlier: () => void } | null;
199228}
200229
201230// ---------------------------------------------------------------------------
......@@ -233,6 +262,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({
233262 onManualNavigation,
234263 hideEmptyPlaceholder = false,
235264 topFadeEnabled = false,
265 loadEarlier = null,
236266}: MessagesTimelineProps) {
237267 const [expandedTurnIds, setExpandedTurnIds] = useState<ReadonlySet<TurnId>>(new Set());
238268 const [expandedWorkGroupIds, setExpandedWorkGroupIds] = useState<ReadonlySet<string>>(new Set());
......@@ -533,7 +563,19 @@ export const MessagesTimeline = memo(function MessagesTimeline({
533563 "scrollbar-gutter-both h-full min-h-0 overflow-x-hidden overscroll-y-contain px-3 [overflow-anchor:none] sm:px-5",
534564 topFadeEnabled && "chat-timeline-scroll-fade",
535565 )}
536 ListHeaderComponent={topFadeEnabled ? TIMELINE_LIST_FADE_HEADER : TIMELINE_LIST_HEADER}
566 ListHeaderComponent={
567 loadEarlier !== null ? (
568 <TimelineLoadEarlierHeader
569 loading={loadEarlier.loading}
570 onLoadEarlier={loadEarlier.onLoadEarlier}
571 fade={topFadeEnabled}
572 />
573 ) : topFadeEnabled ? (
574 TIMELINE_LIST_FADE_HEADER
575 ) : (
576 TIMELINE_LIST_HEADER
577 )
578 }
537579 ListFooterComponent={TIMELINE_LIST_FOOTER}
538580 />
539581 <TimelineMinimap
apps/web/src/connection/storage.ts+6-3
......@@ -51,9 +51,12 @@ const StoredShellSnapshot = Schema.Struct({
5151const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot);
5252// v2 stores the snapshot sequence alongside the thread so a warm cache can
5353// resume via `afterSequence` instead of re-downloading the full thread body.
54// Older v1 entries (no sequence) fail to decode and are treated as a cold cache.
54// v3 adds windowed (paginated) snapshots carrying `page` metadata. The bump
55// exists for rollback safety: a pre-pagination client would decode a windowed
56// v2 record, silently drop the unknown `page` field, and treat the partial
57// thread as complete forever. Older entries fail to decode → cold cache.
5558const StoredThreadSnapshot = Schema.Struct({
56 schemaVersion: Schema.Literal(2),
59 schemaVersion: Schema.Literal(3),
5760 environmentId: EnvironmentId,
5861 threadId: ThreadId,
5962 snapshot: OrchestrationThreadDetailSnapshot,
......@@ -561,7 +564,7 @@ export const connectionStorageLayer = Layer.effectContext(
561564 saveThread: (environmentId, snapshot) =>
562565 Effect.gen(function* () {
563566 const encoded = yield* encodeStoredThreadSnapshot({
564 schemaVersion: 2,
567 schemaVersion: 3,
565568 environmentId,
566569 threadId: snapshot.thread.id,
567570 snapshot,
packages/client-runtime/src/state/entities.test.ts+2
......@@ -333,6 +333,7 @@ describe("environment entity projections", () => {
333333 data: Option.some(detail),
334334 status: "live",
335335 error: Option.none(),
336 page: Option.none(),
336337 }),
337338 );
338339
......@@ -361,6 +362,7 @@ describe("environment entity projections", () => {
361362 }),
362363 status: "live",
363364 error: Option.none(),
365 page: Option.none(),
364366 }),
365367 );
366368
packages/client-runtime/src/state/threadSnapshotHttp.ts+25-2
......@@ -26,6 +26,16 @@ const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000;
2626 * WebSocket subscription's first frame. The response is gzip-compressible by
2727 * the transport and keeps the (potentially multi-KB) snapshot off the socket.
2828 */
29/**
30 * Optional turn window for a snapshot fetch. Only send a window to servers
31 * that advertise `threadSnapshotPagination`; older servers reject unknown
32 * query parameters.
33 */
34export interface ThreadSnapshotWindow {
35 readonly turnLimit: number;
36 readonly beforeCursor?: string;
37}
38
2939export const fetchEnvironmentThreadSnapshot = Effect.fn(
3040 "clientRuntime.state.fetchEnvironmentThreadSnapshot",
3141)(function* (input: {
......@@ -33,6 +43,7 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(
3343 readonly threadId: ThreadId;
3444 readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;
3545 readonly timeoutMs?: number;
46 readonly window?: ThreadSnapshotWindow;
3647}) {
3748 const requestUrl = environmentEndpointUrl(
3849 input.prepared.httpBaseUrl,
......@@ -52,6 +63,12 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn(
5263 input.prepared.httpAuthorization,
5364 client.orchestration.threadSnapshot({
5465 params: { threadId: input.threadId },
66 payload: {
67 ...(input.window !== undefined ? { turnLimit: input.window.turnLimit } : {}),
68 ...(input.window?.beforeCursor !== undefined
69 ? { beforeCursor: input.window.beforeCursor }
70 : {}),
71 },
5572 headers,
5673 }),
5774 ),
......@@ -72,6 +89,7 @@ export class ThreadSnapshotLoader extends Context.Service<
7289 readonly load: (
7390 prepared: PreparedConnection,
7491 threadId: ThreadId,
92 window?: ThreadSnapshotWindow,
7593 ) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>>;
7694 }
7795>()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {}
......@@ -89,8 +107,13 @@ export const threadSnapshotLoaderLayer: Layer.Layer<
89107 // connections work without one).
90108 const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner);
91109 return ThreadSnapshotLoader.of({
92 load: (prepared: PreparedConnection, threadId: ThreadId) =>
93 fetchEnvironmentThreadSnapshot({ prepared, threadId, signer }).pipe(
110 load: (prepared: PreparedConnection, threadId: ThreadId, window?: ThreadSnapshotWindow) =>
111 fetchEnvironmentThreadSnapshot({
112 prepared,
113 threadId,
114 signer,
115 ...(window !== undefined ? { window } : {}),
116 }).pipe(
94117 Effect.map(Option.some<OrchestrationThreadDetailSnapshot>),
95118 Effect.provideService(HttpClient.HttpClient, httpClient),
96119 // A genuinely missing thread (404) is expected — the socket
packages/client-runtime/src/state/threadState.ts+24
......@@ -3,14 +3,38 @@ import * as Option from "effect/Option";
33
44export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted";
55
6/**
7 * Pagination state for a windowed thread. Present only when the loaded thread
8 * is a window (the server returned `page` metadata); absent means the thread is
9 * fully loaded — either the server predates pagination or the window reached
10 * the top.
11 */
12export interface EnvironmentThreadPageState {
13 /** Opaque exclusive cursor for the next older slice; null when fully loaded. */
14 readonly beforeCursor: string | null;
15 readonly hasMore: boolean;
16 /** True while an older page fetch is in flight. */
17 readonly loadingOlder: boolean;
18}
19
620export interface EnvironmentThreadState {
721 readonly data: Option.Option<OrchestrationThread>;
822 readonly status: EnvironmentThreadStatus;
923 readonly error: Option.Option<string>;
24 readonly page: Option.Option<EnvironmentThreadPageState>;
1025}
1126
1227export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = {
1328 data: Option.none(),
1429 status: "empty",
1530 error: Option.none(),
31 page: Option.none(),
1632};
33
34/** Whether the thread has older turns that can be loaded with more pages. */
35export function threadHasOlderTurns(state: EnvironmentThreadState): boolean {
36 return Option.match(state.page, {
37 onNone: () => false,
38 onSome: (page) => page.hasMore,
39 });
40}
packages/client-runtime/src/state/threads-pagination.test.ts created+543
......@@ -0,0 +1,543 @@
1import {
2 EnvironmentId,
3 EventId,
4 ORCHESTRATION_WS_METHODS,
5 ProjectId,
6 ProviderInstanceId,
7 ThreadId,
8 TurnId,
9 type OrchestrationMessage,
10 type OrchestrationThread,
11 type OrchestrationThreadDetailSnapshot,
12 type OrchestrationThreadStreamItem,
13} from "@t3tools/contracts";
14import { describe, expect, it } from "@effect/vitest";
15import * as Deferred from "effect/Deferred";
16import * as Effect from "effect/Effect";
17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Ref from "effect/Ref";
20import * as Stream from "effect/Stream";
21import * as SubscriptionRef from "effect/SubscriptionRef";
22
23import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
24import {
25 AVAILABLE_CONNECTION_STATE,
26 PrimaryConnectionTarget,
27 type PreparedConnection,
28 type SupervisorConnectionState,
29} from "../connection/model.ts";
30import * as EnvironmentSupervisor from "../connection/supervisor.ts";
31import * as Persistence from "../platform/persistence.ts";
32import * as RpcSession from "../rpc/session.ts";
33import type { ThreadSnapshotWindow } from "./threadSnapshotHttp.ts";
34import {
35 INITIAL_THREAD_USER_TURN_LIMIT,
36 makeEnvironmentThreadState,
37 requestOlderThreadTurns,
38 ThreadSnapshotLoader,
39 type EnvironmentThreadState,
40} from "./threads.ts";
41
42const TARGET = new PrimaryConnectionTarget({
43 environmentId: EnvironmentId.make("environment-1"),
44 label: "Test environment",
45 httpBaseUrl: "https://environment.example.test",
46 wsBaseUrl: "wss://environment.example.test",
47});
48const THREAD_ID = ThreadId.make("thread-1");
49const PREPARED: PreparedConnection = {
50 environmentId: TARGET.environmentId,
51 label: TARGET.label,
52 httpBaseUrl: TARGET.httpBaseUrl,
53 socketUrl: TARGET.wsBaseUrl,
54 httpAuthorization: null,
55 target: TARGET,
56};
57
58function message(id: string, turnId: string, createdAt: string): OrchestrationMessage {
59 return {
60 id: id as OrchestrationMessage["id"],
61 role: "assistant",
62 text: `text of ${id}`,
63 turnId: TurnId.make(turnId),
64 streaming: false,
65 createdAt,
66 updatedAt: createdAt,
67 };
68}
69
70const OLDER_MESSAGE = message("message-old", "turn-1", "2026-04-01T00:00:00.000Z");
71const RECENT_MESSAGE = message("message-recent", "turn-2", "2026-04-01T01:00:00.000Z");
72
73// Reverts retain turns via checkpoints with checkpointTurnCount <= the revert's
74// turnCount, so both fixture turns carry one: reverting to turnCount 1 keeps
75// turn-1 (the older page's turn) and discards turn-2 (the loaded window's).
76function checkpoint(turnId: string, turnCount: number): OrchestrationThread["checkpoints"][number] {
77 return {
78 turnId: TurnId.make(turnId),
79 checkpointTurnCount: turnCount,
80 checkpointRef:
81 `checkpoint-${turnCount}` as OrchestrationThread["checkpoints"][number]["checkpointRef"],
82 status: "ready",
83 files: [],
84 assistantMessageId: null,
85 completedAt: "2026-04-01T01:00:00.000Z",
86 };
87}
88
89const BASE_THREAD: OrchestrationThread = {
90 id: THREAD_ID,
91 projectId: ProjectId.make("project-1"),
92 title: "Windowed thread",
93 modelSelection: {
94 instanceId: ProviderInstanceId.make("codex"),
95 model: "gpt-5.4",
96 },
97 runtimeMode: "full-access",
98 interactionMode: "default",
99 branch: "main",
100 worktreePath: null,
101 latestTurn: null,
102 createdAt: "2026-04-01T00:00:00.000Z",
103 updatedAt: "2026-04-01T00:00:00.000Z",
104 archivedAt: null,
105 settledOverride: null,
106 settledAt: null,
107 deletedAt: null,
108 messages: [RECENT_MESSAGE],
109 proposedPlans: [],
110 activities: [],
111 checkpoints: [checkpoint("turn-2", 2)],
112 session: null,
113};
114
115const WINDOWED_SNAPSHOT: OrchestrationThreadDetailSnapshot = {
116 snapshotSequence: 10,
117 thread: BASE_THREAD,
118 page: { beforeCursor: "cursor-1", hasMore: true, snapshotSequence: 10 },
119};
120
121const OLDER_PAGE: OrchestrationThreadDetailSnapshot = {
122 snapshotSequence: 10,
123 thread: {
124 ...BASE_THREAD,
125 messages: [OLDER_MESSAGE],
126 checkpoints: [checkpoint("turn-1", 1)],
127 },
128 page: { beforeCursor: null, hasMore: false, snapshotSequence: 10 },
129};
130
131type LoaderResponse = Option.Option<OrchestrationThreadDetailSnapshot>;
132
133const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (options?: {
134 readonly paginationCapability?: boolean;
135 readonly initialResponse?: LoaderResponse;
136 /** Cached snapshot returned by the cache store (simulates a warm cache). */
137 readonly cached?: OrchestrationThreadDetailSnapshot;
138}) {
139 const inputs = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
140 const observed = yield* Queue.unbounded<EnvironmentThreadState>();
141 const loaderWindows = yield* Ref.make<ReadonlyArray<ThreadSnapshotWindow | undefined>>([]);
142 const lastSubscribeInput = yield* Ref.make<Record<string, unknown> | undefined>(undefined);
143 const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThreadDetailSnapshot>>([]);
144 // Older-page responses resolve through deferreds so tests can interleave
145 // live events with an in-flight page fetch.
146 const pendingPageResponses = yield* Queue.unbounded<Deferred.Deferred<LoaderResponse>>();
147 const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(
148 AVAILABLE_CONNECTION_STATE,
149 );
150 const client = {
151 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: Record<string, unknown>) =>
152 Stream.unwrap(Ref.set(lastSubscribeInput, input).pipe(Effect.as(Stream.fromQueue(inputs)))),
153 } as unknown as WsRpcProtocolClient;
154 const session: RpcSession.RpcSession = {
155 client,
156 initialConfig: Effect.succeed({
157 threadSnapshotPagination: options?.paginationCapability !== false,
158 } as never),
159 ready: Effect.void,
160 probe: Effect.void,
161 closed: Effect.never,
162 };
163 const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
164 Option.some(session),
165 );
166 const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
167 Option.some(PREPARED),
168 );
169 const snapshotLoader = ThreadSnapshotLoader.of({
170 load: (_prepared, _threadId, window) =>
171 Ref.update(loaderWindows, (current) => [...current, window]).pipe(
172 Effect.andThen(
173 window?.beforeCursor === undefined
174 ? Effect.succeed(
175 options?.initialResponse ?? Option.none<OrchestrationThreadDetailSnapshot>(),
176 )
177 : Deferred.make<LoaderResponse>().pipe(
178 Effect.tap((deferred) => Queue.offer(pendingPageResponses, deferred)),
179 Effect.flatMap(Deferred.await),
180 ),
181 ),
182 ),
183 });
184 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
185 target: TARGET,
186 state: supervisorState,
187 session: supervisorSession,
188 prepared,
189 connect: Effect.void,
190 disconnect: Effect.void,
191 retryNow: Effect.void,
192 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
193 const cache = Persistence.EnvironmentCacheStore.of({
194 loadShell: () => Effect.succeed(Option.none()),
195 saveShell: () => Effect.void,
196 loadThread: () =>
197 Effect.succeed(options?.cached !== undefined ? Option.some(options.cached) : Option.none()),
198 saveThread: (_environmentId, thread) =>
199 Ref.update(savedThreads, (current) => [...current, thread]),
200 removeThread: () => Effect.void,
201 loadServerConfig: () => Effect.succeed(Option.none()),
202 saveServerConfig: () => Effect.void,
203 loadVcsRefs: () => Effect.succeed(Option.none()),
204 saveVcsRefs: () => Effect.void,
205 removeVcsRefs: () => Effect.void,
206 clearVcsRefs: () => Effect.void,
207 clear: () => Effect.void,
208 });
209 const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe(
210 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
211 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
212 Effect.provideService(ThreadSnapshotLoader, snapshotLoader),
213 );
214 yield* SubscriptionRef.changes(threadState).pipe(
215 Stream.runForEach((state) => Queue.offer(observed, state)),
216 Effect.forkScoped,
217 );
218
219 const awaitState = (predicate: (state: EnvironmentThreadState) => boolean) =>
220 Queue.take(observed).pipe(Effect.repeat({ until: predicate }));
221 const resolveNextPage = (response: LoaderResponse) =>
222 Queue.take(pendingPageResponses).pipe(
223 Effect.flatMap((deferred) => Deferred.succeed(deferred, response)),
224 );
225
226 return {
227 inputs,
228 observed,
229 awaitState,
230 resolveNextPage,
231 loaderWindows,
232 lastSubscribeInput,
233 savedThreads,
234 threadState,
235 };
236});
237
238const hasMessage = (state: EnvironmentThreadState, id: string): boolean =>
239 Option.match(state.data, {
240 onNone: () => false,
241 onSome: (thread) => thread.messages.some((entry) => entry.id === id),
242 });
243
244const titleEvent = (title: string, sequence: number): OrchestrationThreadStreamItem => ({
245 kind: "event",
246 event: {
247 eventId: EventId.make(`event-title-${sequence}`),
248 sequence,
249 occurredAt: "2026-04-01T01:30:00.000Z",
250 commandId: null,
251 causationEventId: null,
252 correlationId: null,
253 metadata: {},
254 aggregateKind: "thread",
255 aggregateId: THREAD_ID,
256 type: "thread.meta-updated",
257 payload: {
258 threadId: THREAD_ID,
259 title,
260 updatedAt: "2026-04-01T01:30:00.000Z",
261 },
262 },
263});
264
265// Reverting to turnCount 1 retains only turns whose checkpoint count is <= 1:
266// turn-1 survives, turn-2 (the loaded window's newest turn) is discarded.
267const revertEvent = (sequence: number): OrchestrationThreadStreamItem => ({
268 kind: "event",
269 event: {
270 eventId: EventId.make(`event-revert-${sequence}`),
271 sequence,
272 occurredAt: "2026-04-01T02:00:00.000Z",
273 commandId: null,
274 causationEventId: null,
275 correlationId: null,
276 metadata: {},
277 aggregateKind: "thread",
278 aggregateId: THREAD_ID,
279 type: "thread.reverted",
280 payload: {
281 threadId: THREAD_ID,
282 turnCount: 1,
283 },
284 },
285});
286
287describe("thread pagination state", () => {
288 it.effect("windows the initial load when the server advertises pagination", () =>
289 Effect.gen(function* () {
290 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
291 const state = yield* harness.awaitState((value) => Option.isSome(value.page));
292 expect(Option.getOrThrow(state.page)).toEqual({
293 beforeCursor: "cursor-1",
294 hasMore: true,
295 loadingOlder: false,
296 });
297 const windows = yield* Ref.get(harness.loaderWindows);
298 expect(windows[0]?.turnLimit).toBe(INITIAL_THREAD_USER_TURN_LIMIT);
299 const subscribeInput = yield* Ref.get(harness.lastSubscribeInput);
300 expect(subscribeInput?.turnLimit).toBe(INITIAL_THREAD_USER_TURN_LIMIT);
301 }),
302 );
303
304 it.effect("does not send a window to servers without the capability", () =>
305 Effect.gen(function* () {
306 const harness = yield* makeHarness({
307 paginationCapability: false,
308 initialResponse: Option.some({ snapshotSequence: 10, thread: BASE_THREAD }),
309 });
310 const state = yield* harness.awaitState((value) => Option.isSome(value.data));
311 expect(Option.isNone(state.page)).toBe(true);
312 const windows = yield* Ref.get(harness.loaderWindows);
313 expect(windows[0]).toBeUndefined();
314 const subscribeInput = yield* Ref.get(harness.lastSubscribeInput);
315 expect(subscribeInput?.turnLimit).toBeUndefined();
316 }),
317 );
318
319 it.effect("merges an older page below the loaded window and clears the cursor", () =>
320 Effect.gen(function* () {
321 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
322 yield* harness.awaitState((value) => Option.isSome(value.page));
323
324 expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true);
325 yield* harness.awaitState((value) =>
326 Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }),
327 );
328 yield* harness.resolveNextPage(Option.some(OLDER_PAGE));
329
330 const state = yield* harness.awaitState((value) => hasMessage(value, "message-old"));
331 const thread = Option.getOrThrow(state.data);
332 // Older rows land before the loaded window's rows.
333 expect(thread.messages.map((entry) => entry.id)).toEqual(["message-old", "message-recent"]);
334 expect(Option.getOrThrow(state.page)).toEqual({
335 beforeCursor: null,
336 hasMore: false,
337 loadingOlder: false,
338 });
339 }),
340 );
341
342 it.effect("discards an in-flight older page when a revert rewrites history", () =>
343 Effect.gen(function* () {
344 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
345 yield* harness.awaitState((value) => Option.isSome(value.page));
346
347 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
348 yield* harness.awaitState((value) =>
349 Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }),
350 );
351 // Revert lands while the page fetch is in flight and removes turn-2.
352 yield* Queue.offer(harness.inputs, revertEvent(11));
353 yield* harness.awaitState((value) => !hasMessage(value, "message-recent"));
354 yield* harness.resolveNextPage(Option.some(OLDER_PAGE));
355
356 const state = yield* harness.awaitState((value) =>
357 Option.match(value.page, { onNone: () => false, onSome: (page) => !page.loadingOlder }),
358 );
359 // The stale page was dropped: no resurrected rows, cursor unchanged.
360 expect(hasMessage(state, "message-old")).toBe(false);
361 expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1");
362 }),
363 );
364
365 it.effect("discards an in-flight older page when a fresh snapshot replaces the thread", () =>
366 Effect.gen(function* () {
367 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
368 yield* harness.awaitState((value) => Option.isSome(value.page));
369
370 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
371 yield* harness.awaitState((value) =>
372 Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }),
373 );
374 yield* Queue.offer(harness.inputs, {
375 kind: "snapshot",
376 snapshot: {
377 snapshotSequence: 20,
378 thread: { ...BASE_THREAD, title: "Replaced thread" },
379 page: { beforeCursor: "cursor-2", hasMore: true, snapshotSequence: 20 },
380 },
381 });
382 yield* harness.awaitState((value) =>
383 Option.match(value.data, {
384 onNone: () => false,
385 onSome: (thread) => thread.title === "Replaced thread",
386 }),
387 );
388 yield* harness.resolveNextPage(Option.some(OLDER_PAGE));
389
390 const state = yield* harness.awaitState((value) =>
391 Option.match(value.page, { onNone: () => false, onSome: (page) => !page.loadingOlder }),
392 );
393 expect(hasMessage(state, "message-old")).toBe(false);
394 // The replacement snapshot's cursor wins over the discarded page's.
395 expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-2");
396 }),
397 );
398
399 it.effect("discards an older page read from a projection behind the loaded state", () =>
400 Effect.gen(function* () {
401 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
402 yield* harness.awaitState((value) => Option.isSome(value.page));
403
404 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
405 yield* harness.awaitState((value) =>
406 Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }),
407 );
408 yield* harness.resolveNextPage(Option.some({ ...OLDER_PAGE, snapshotSequence: 5 }));
409
410 const state = yield* harness.awaitState((value) =>
411 Option.match(value.page, { onNone: () => false, onSome: (page) => !page.loadingOlder }),
412 );
413 expect(hasMessage(state, "message-old")).toBe(false);
414 expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1");
415 }),
416 );
417
418 it.effect("a merged history page never advances the live-event dedupe sequence", () =>
419 Effect.gen(function* () {
420 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
421 yield* harness.awaitState((value) => Option.isSome(value.page));
422
423 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
424 yield* harness.awaitState((value) =>
425 Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }),
426 );
427 // The page was captured at a newer projection sequence (12) than the
428 // loaded state (10); merging it must not swallow events 11-12.
429 yield* harness.resolveNextPage(
430 Option.some({
431 ...OLDER_PAGE,
432 snapshotSequence: 12,
433 page: { beforeCursor: null, hasMore: false, snapshotSequence: 12 },
434 }),
435 );
436 yield* harness.awaitState((value) => hasMessage(value, "message-old"));
437
438 // Event at sequence 11 must still apply after the merge: the revert
439 // discards turn-2, so the loaded window's row disappears while the
440 // merged older turn-1 row survives. If the merge had advanced the
441 // dedupe sequence to the page's 12, this event would be swallowed.
442 yield* Queue.offer(harness.inputs, revertEvent(11));
443 const state = yield* harness.awaitState(
444 (value) => !hasMessage(value, "message-recent") && hasMessage(value, "message-old"),
445 );
446 expect(hasMessage(state, "message-old")).toBe(true);
447 }),
448 );
449
450 it.effect("parks a page read ahead of the live state until events catch up", () =>
451 Effect.gen(function* () {
452 // A page whose thread watermark is ahead of the loaded state may
453 // contain streaming content the subscription has not delivered yet
454 // (e.g. an out-of-window subagent turn mid-stream); merging it
455 // immediately and then replaying those deltas would duplicate text.
456 // The page parks until the live state reaches the watermark.
457 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
458 yield* harness.awaitState((value) => Option.isSome(value.page));
459
460 requestOlderThreadTurns(TARGET.environmentId, THREAD_ID);
461 yield* harness.awaitState((value) =>
462 Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }),
463 );
464 // Page watermark 11 > loaded sequence 10: must park, not merge.
465 yield* harness.resolveNextPage(
466 Option.some({
467 ...OLDER_PAGE,
468 snapshotSequence: 11,
469 page: { beforeCursor: null, hasMore: false, snapshotSequence: 11, threadSequence: 11 },
470 }),
471 );
472
473 // A live event at sequence 11 arrives; only then does the page merge.
474 yield* Queue.offer(harness.inputs, titleEvent("Advanced past watermark", 11));
475 const state = yield* harness.awaitState((value) => hasMessage(value, "message-old"));
476 expect(hasMessage(state, "message-recent")).toBe(true);
477 expect(Option.getOrThrow(state.page).loadingOlder).toBe(false);
478 }),
479 );
480
481 it.effect("a revert keeps the page cursor and triggers no refresh fetch", () =>
482 Effect.gen(function* () {
483 // Cursors are an (anchor, turnId) keyset derived from event content, so
484 // they survive the revert projector's row rewrite: the machine keeps
485 // the stored cursor and performs no snapshot re-fetch. The revert
486 // reducer's turn filtering alone handles loaded history.
487 const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) });
488 yield* harness.awaitState((value) => Option.isSome(value.page));
489
490 yield* Queue.offer(harness.inputs, revertEvent(11));
491 const state = yield* harness.awaitState((value) => !hasMessage(value, "message-recent"));
492
493 expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1");
494 const windows = yield* Ref.get(harness.loaderWindows);
495 // Only the initial load hit the loader — no post-revert refresh fetch.
496 expect(windows.length).toBe(1);
497 }),
498 );
499
500 it.effect("drops a windowed cache when the server lacks the pagination capability", () =>
501 Effect.gen(function* () {
502 // Resuming a windowed cache via afterSequence against a pre-pagination
503 // server would render only the window forever with no way to load the
504 // rest: the machine must discard the cache and take a full snapshot.
505 const fullSnapshot: OrchestrationThreadDetailSnapshot = {
506 snapshotSequence: 20,
507 thread: { ...BASE_THREAD, title: "Full reload" },
508 };
509 const harness = yield* makeHarness({
510 paginationCapability: false,
511 cached: WINDOWED_SNAPSHOT,
512 initialResponse: Option.some(fullSnapshot),
513 });
514
515 const state = yield* harness.awaitState((value) =>
516 Option.match(value.data, {
517 onNone: () => false,
518 onSome: (thread) => thread.title === "Full reload",
519 }),
520 );
521 expect(Option.isNone(state.page)).toBe(true);
522 // The subscription resumed from the fresh full snapshot, not the
523 // discarded windowed cache's watermark, and sent no window fields.
524 const subscribeInput = yield* Ref.get(harness.lastSubscribeInput);
525 expect(subscribeInput?.turnLimit).toBeUndefined();
526 expect(subscribeInput?.afterSequence).toBe(20);
527 }),
528 );
529
530 it.effect("keeps a windowed cache when the server supports pagination", () =>
531 Effect.gen(function* () {
532 const harness = yield* makeHarness({ cached: WINDOWED_SNAPSHOT });
533 const state = yield* harness.awaitState((value) => Option.isSome(value.page));
534 expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1");
535 // Wait for the subscription (recorded when the WS method is invoked)
536 // before asserting its input.
537 const subscribeInput = yield* Ref.get(harness.lastSubscribeInput).pipe(
538 Effect.repeat({ until: (input) => input !== undefined }),
539 );
540 expect(subscribeInput?.afterSequence).toBe(10);
541 }),
542 );
543});
packages/client-runtime/src/state/threads.ts+393-13
......@@ -2,15 +2,18 @@ import {
22 ORCHESTRATION_WS_METHODS,
33 type EnvironmentId as EnvironmentIdType,
44 type OrchestrationThread,
5 type OrchestrationThreadDetailPage,
56 type OrchestrationThreadDetailSnapshot,
67 type OrchestrationThreadStreamItem,
78 type ThreadId as ThreadIdType,
89} from "@t3tools/contracts";
910import * as Cause from "effect/Cause";
11import * as Context from "effect/Context";
1012import * as Effect from "effect/Effect";
1113import * as Option from "effect/Option";
1214import * as Queue from "effect/Queue";
1315import * as Ref from "effect/Ref";
16import * as Semaphore from "effect/Semaphore";
1417import * as Stream from "effect/Stream";
1518import * as SubscriptionRef from "effect/SubscriptionRef";
1619import { Atom } from "effect/unstable/reactivity";
......@@ -21,13 +24,14 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts";
2124import * as ConnectionWakeups from "../connection/wakeups.ts";
2225import { EnvironmentCacheStore } from "../platform/persistence.ts";
2326import { subscribeDynamic } from "../rpc/client.ts";
24import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts";
27import { ThreadSnapshotLoader, type ThreadSnapshotWindow } from "./threadSnapshotHttp.ts";
2528import { parseThreadKey, threadKey } from "./entities.ts";
2629import { applyThreadDetailEvent } from "./threadReducer.ts";
2730import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
2831import { followStreamInEnvironment } from "./runtime.ts";
2932import {
3033 EMPTY_ENVIRONMENT_THREAD_STATE,
34 type EnvironmentThreadPageState,
3135 type EnvironmentThreadState,
3236 type EnvironmentThreadStatus,
3337} from "./threadState.ts";
......@@ -36,6 +40,85 @@ function statusWithoutLiveData(data: Option.Option<OrchestrationThread>): Enviro
3640 return Option.isSome(data) ? "cached" : "empty";
3741}
3842
43/**
44 * Turn window sizes for paginated thread loads: the initial page covers the
45 * last 10 user-anchored turns (subagent/fan-out turns ride along), each
46 * "load earlier" tap fetches 20 more. Sized so first paint on the heaviest
47 * observed threads stays around 100K gzipped while median threads load fully.
48 */
49export const INITIAL_THREAD_USER_TURN_LIMIT = 10;
50export const OLDER_THREAD_PAGE_USER_TURN_LIMIT = 20;
51
52function pageStateFromSnapshot(
53 page: OrchestrationThreadDetailPage | undefined,
54): Option.Option<EnvironmentThreadPageState> {
55 return page === undefined
56 ? Option.none()
57 : Option.some({
58 beforeCursor: page.beforeCursor,
59 hasMore: page.hasMore,
60 loadingOlder: false,
61 });
62}
63
64interface ThreadOlderTurnRequestRegistry {
65 /**
66 * Registers the live state machine for a thread. Returns the deregistration
67 * cleanup; registration lives exactly as long as the machine's scope, and a
68 * successor machine for the same thread simply replaces the entry.
69 */
70 readonly register: (key: string, handler: () => void) => () => void;
71 readonly request: (key: string) => boolean;
72}
73
74function makeThreadOlderTurnRequestRegistry(): ThreadOlderTurnRequestRegistry {
75 const handlers = new Map<string, () => void>();
76 return {
77 register: (key, handler) => {
78 handlers.set(key, handler);
79 return () => {
80 if (handlers.get(key) === handler) {
81 handlers.delete(key);
82 }
83 };
84 },
85 request: (key) => {
86 const handler = handlers.get(key);
87 if (handler === undefined) {
88 return false;
89 }
90 handler();
91 return true;
92 },
93 };
94}
95
96const defaultOlderTurnRequestRegistry = makeThreadOlderTurnRequestRegistry();
97
98/**
99 * Channel from UI actions to the live per-thread state machines. The machines
100 * resolve it from the Effect environment (overridable in tests); the default
101 * instance is shared with the sync `requestOlderThreadTurns` entry point so
102 * the apps get working wiring without providing anything.
103 */
104export class ThreadOlderTurnRequests extends Context.Reference<ThreadOlderTurnRequestRegistry>(
105 "@t3tools/client-runtime/state/threads/ThreadOlderTurnRequests",
106 { defaultValue: () => defaultOlderTurnRequestRegistry },
107) {}
108
109/**
110 * Asks the live state machine for `threadId` to fetch the next older page.
111 * Returns false when no machine is live or no fetch was started (no cursor,
112 * already loading); callers render from `EnvironmentThreadState.page` and can
113 * treat false as "nothing to do".
114 */
115export function requestOlderThreadTurns(
116 environmentId: EnvironmentIdType,
117 threadId: ThreadIdType,
118): boolean {
119 return defaultOlderTurnRequestRegistry.request(threadKey({ environmentId, threadId }));
120}
121
39122function formatThreadError(cause: Cause.Cause<unknown>): string {
40123 const error = Cause.squash(cause);
41124 return error instanceof Error && error.message.trim().length > 0
......@@ -73,6 +156,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
73156 data: cachedThread,
74157 status: statusWithoutLiveData(cachedThread),
75158 error: Option.none(),
159 // A cached windowed snapshot restores its page cursor so "load earlier"
160 // works while rendering from cache; a cached full snapshot has no page.
161 page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)),
76162 });
77163 // Seed the resume cursor from the cached snapshot so a warm cache can catch up
78164 // via `afterSequence` instead of re-downloading the full thread body.
......@@ -80,6 +166,25 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
80166 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }),
81167 );
82168 const awaitingCompletion = yield* Ref.make(false);
169 // Bumped whenever loaded history may have been rewritten out from under an
170 // in-flight older-page fetch (snapshot replacement, revert, deletion). A
171 // page response captured under an older epoch is discarded, not merged.
172 const historyEpoch = yield* Ref.make(0);
173 // Serializes stream-item application against older-page staleness checks +
174 // merges. Without it, a revert or snapshot processed between loadOlderTurns'
175 // epoch check and its merge could still slip resurrected history in.
176 const applyLock = yield* Semaphore.make(1);
177 // Whether the connected server accepts windowed reads; set per subscription
178 // from the session config. Gates loadOlderTurns so a reconnect to a
179 // pre-pagination server never sends unsupported window parameters.
180 const paginationSupported = yield* Ref.make(false);
181 // An older page whose thread watermark is ahead of the live state, parked
182 // until the subscription catches up (see mergeOlderPage's caller). At most
183 // one can exist because loadOlderTurns no-ops while loadingOlder is true.
184 const pendingOlderPage = yield* Ref.make<{
185 readonly snapshot: OrchestrationThreadDetailSnapshot;
186 readonly epoch: number;
187 } | null>(null);
83188 const persistence = yield* Queue.sliding<OrchestrationThreadDetailSnapshot>(1);
84189
85190 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
......@@ -124,6 +229,12 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
124229 );
125230 const setDisconnected = Effect.gen(function* () {
126231 yield* Ref.set(awaitingCompletion, false);
232 // The capability belongs to the session that advertised it. During a
233 // reconnect, a new prepared connection can exist before the new session's
234 // config arrives; leaving the old value would let loadOlderTurns send
235 // window parameters to a server that may not accept them (review
236 // finding). makeSubscribeInput re-sets it from the next session's config.
237 yield* Ref.set(paginationSupported, false);
127238 yield* SubscriptionRef.update(state, (current) => ({
128239 ...current,
129240 status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data),
......@@ -143,28 +254,51 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
143254
144255 const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* (
145256 thread: OrchestrationThread,
257 // "keep" preserves the current page state (live events touch only loaded
258 // recent turns); a snapshot or merged page passes its own page state.
259 page: Option.Option<EnvironmentThreadPageState> | "keep",
146260 ) {
147261 const waiting = yield* Ref.get(awaitingCompletion);
148 yield* SubscriptionRef.set(state, {
262 yield* SubscriptionRef.update(state, (current) => ({
149263 data: Option.some(thread),
150 status: waiting ? "synchronizing" : "live",
264 status: waiting ? ("synchronizing" as const) : ("live" as const),
151265 error: Option.none(),
152 });
266 page: page === "keep" ? current.page : page,
267 }));
153268 // Active threads can update many times per second and retain large tool
154269 // payloads. The server remains the source of truth while a turn is active;
155270 // persist once it settles so cache encoding stays off the streaming path.
156271 if (shouldPersistThread(thread)) {
157272 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
158 yield* Queue.offer(persistence, { snapshotSequence, thread });
273 const currentPage = yield* SubscriptionRef.get(state).pipe(Effect.map((value) => value.page));
274 yield* Queue.offer(persistence, {
275 snapshotSequence,
276 thread,
277 // Persist the window boundary with the window's content so a cache
278 // restore can keep paging from where the loaded history ends.
279 ...Option.match(currentPage, {
280 onNone: () => ({}),
281 onSome: (value) =>
282 ({
283 page: {
284 beforeCursor: value.beforeCursor,
285 hasMore: value.hasMore,
286 snapshotSequence,
287 },
288 }) as const,
289 }),
290 });
159291 }
160292 });
161293
162294 const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
163295 yield* Ref.set(awaitingCompletion, false);
296 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);
164297 yield* SubscriptionRef.set(state, {
165298 data: Option.none(),
166299 status: "deleted",
167300 error: Option.none(),
301 page: Option.none(),
168302 });
169303 yield* cache.removeThread(environmentId, threadId).pipe(
170304 Effect.catch((error) =>
......@@ -179,7 +313,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
179313 );
180314 });
181315
182 const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (
316 // Body of applyItem, running under applyLock.
317 const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* (
183318 item: OrchestrationThreadStreamItem,
184319 ) {
185320 if (item.kind === "synchronized") {
......@@ -193,8 +328,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
193328 }
194329
195330 if (item.kind === "snapshot") {
331 // A fresh snapshot replaces all loaded history, including older
332 // pages: a turn reverted while disconnected would otherwise survive
333 // in the preserved history with no event left to remove it. The
334 // epoch bump discards any older-page fetch racing this snapshot.
335 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);
196336 yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence);
197 yield* setThread(item.snapshot.thread);
337 yield* setThread(item.snapshot.thread, pageStateFromSnapshot(item.snapshot.page));
198338 return;
199339 }
200340
......@@ -211,12 +351,184 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
211351 }
212352 return;
213353 }
354 if (item.event.type === "thread.reverted") {
355 // A revert rewrites loaded history (whole turns disappear), so an
356 // older-page fetch in flight may straddle the removed range; the epoch
357 // bump discards it. The stored page cursor stays valid: cursors are an
358 // (anchor, turnId) keyset derived from event content, which survives
359 // the revert projector's row rewrite, so no refresh is needed — the
360 // revert reducer's turn filtering fully handles loaded history.
361 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);
362 }
214363 const result = applyThreadDetailEvent(current.data.value, item.event);
215364 if (result.kind === "updated") {
216 yield* setThread(result.thread);
365 yield* setThread(result.thread, "keep");
217366 } else if (result.kind === "deleted") {
218367 yield* setDeleted();
219368 }
369 // The event may have advanced the live state past a parked page's
370 // watermark; merge it as soon as that happens.
371 yield* tryMergePendingOlderPage();
372 });
373
374 // Merges a parked older page once the live state has caught up to the
375 // page's thread watermark, or discards it if history was rewritten
376 // (epoch advanced) while it waited. Must run under applyLock.
377 const tryMergePendingOlderPage = Effect.fn("EnvironmentThreadState.tryMergePendingOlderPage")(
378 function* () {
379 const pending = yield* Ref.get(pendingOlderPage);
380 if (pending === null) {
381 return;
382 }
383 const epochNow = yield* Ref.get(historyEpoch);
384 if (epochNow !== pending.epoch) {
385 yield* Ref.set(pendingOlderPage, null);
386 yield* SubscriptionRef.update(state, (value) => ({
387 ...value,
388 page: Option.map(value.page, (existing) => ({ ...existing, loadingOlder: false })),
389 }));
390 return;
391 }
392 const watermark = pending.snapshot.page?.threadSequence;
393 const loadedSequence = yield* SubscriptionRef.get(lastSequence);
394 if (watermark !== undefined && watermark > loadedSequence) {
395 return;
396 }
397 yield* Ref.set(pendingOlderPage, null);
398 yield* mergeOlderPage(pending.snapshot);
399 },
400 );
401
402 const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (
403 item: OrchestrationThreadStreamItem,
404 ) {
405 yield* applyLock.withPermits(1)(applyItemLocked(item));
406 });
407
408 // Merges an older disjoint page below the currently loaded window. All four
409 // windowed collections prepend; identity dedupe guards the (server-bug or
410 // cursor-misuse) case of overlapping pages so a row never renders twice.
411 const mergeOlderPage = Effect.fn("EnvironmentThreadState.mergeOlderPage")(function* (
412 snapshot: OrchestrationThreadDetailSnapshot,
413 ) {
414 // The merge is built inside the update callback so it composes with
415 // whatever thread value is current at commit time. The applyLock already
416 // serializes this against event application; the atomic build is defense
417 // in depth against future callers outside the lock.
418 let merged: OrchestrationThread | null = null;
419 yield* SubscriptionRef.update(state, (value) => {
420 if (Option.isNone(value.data)) {
421 return value;
422 }
423 const loaded = value.data.value;
424 const older = snapshot.thread;
425 const mergeById = <T extends { readonly id: string }>(
426 olderRows: ReadonlyArray<T>,
427 loadedRows: ReadonlyArray<T>,
428 ): ReadonlyArray<T> => {
429 const seen = new Set(loadedRows.map((row) => row.id));
430 return [...olderRows.filter((row) => !seen.has(row.id)), ...loadedRows];
431 };
432 const seenCheckpoints = new Set(loaded.checkpoints.map((row) => row.turnId));
433 merged = {
434 // Thread metadata stays the loaded (newer) snapshot's; only the
435 // windowed collections gain rows from the older page.
436 ...loaded,
437 messages: mergeById(older.messages, loaded.messages),
438 activities: mergeById(older.activities, loaded.activities),
439 proposedPlans: mergeById(older.proposedPlans, loaded.proposedPlans),
440 checkpoints: [
441 ...older.checkpoints.filter((row) => !seenCheckpoints.has(row.turnId)),
442 ...loaded.checkpoints,
443 ],
444 };
445 return {
446 ...value,
447 data: Option.some(merged),
448 page: pageStateFromSnapshot(snapshot.page),
449 };
450 });
451 // Persist the widened window under the *loaded* watermark: the merged
452 // content is only known consistent with the state it merged into, not
453 // with the page's own (possibly newer) sequence.
454 if (merged !== null && shouldPersistThread(merged)) {
455 const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
456 yield* Queue.offer(persistence, {
457 snapshotSequence,
458 thread: merged,
459 ...(snapshot.page === undefined ? {} : { page: { ...snapshot.page, snapshotSequence } }),
460 });
461 }
462 });
463
464 const loadOlderTurns = Effect.fn("EnvironmentThreadState.loadOlderTurns")(function* () {
465 // Gated on the connected server's capability: a reconnect to a
466 // pre-pagination server must never receive window parameters.
467 if (!(yield* Ref.get(paginationSupported))) {
468 return;
469 }
470 const current = yield* SubscriptionRef.get(state);
471 const page = Option.getOrNull(current.page);
472 if (page === null || page.loadingOlder || !page.hasMore || page.beforeCursor === null) {
473 return;
474 }
475 const prepared = Option.getOrNull(yield* SubscriptionRef.get(supervisor.prepared));
476 if (prepared === null) {
477 return;
478 }
479 const epochAtStart = yield* Ref.get(historyEpoch);
480 yield* SubscriptionRef.update(state, (value) => ({
481 ...value,
482 page: Option.map(value.page, (existing) => ({ ...existing, loadingOlder: true })),
483 }));
484 const window: ThreadSnapshotWindow = {
485 turnLimit: OLDER_THREAD_PAGE_USER_TURN_LIMIT,
486 beforeCursor: page.beforeCursor,
487 };
488 const response = yield* snapshotLoader.load(prepared, threadId, window);
489 // Staleness check and merge run under the same lock as stream-item
490 // application, so a revert/snapshot cannot land between them (TOCTOU
491 // review finding) — anything that rewrites history bumps the epoch
492 // before this permit is acquired.
493 yield* applyLock.withPermits(1)(
494 Effect.gen(function* () {
495 const epochNow = yield* Ref.get(historyEpoch);
496 const loadedSequence = yield* SubscriptionRef.get(lastSequence);
497 // A page carrying a sequence older than the loaded state was read
498 // from a projection behind what we render; merging it could
499 // resurrect turns a newer snapshot or revert already removed.
500 const stale =
501 epochNow !== epochAtStart ||
502 Option.match(response, {
503 onNone: () => false,
504 onSome: (snapshot) => snapshot.snapshotSequence < loadedSequence,
505 });
506 if (Option.isNone(response) || stale) {
507 yield* SubscriptionRef.update(state, (value) => ({
508 ...value,
509 page: Option.map(value.page, (existing) => ({ ...existing, loadingOlder: false })),
510 }));
511 return;
512 }
513 // A page read AHEAD of the live state may include content (e.g.
514 // streaming deltas of an out-of-window turn) the subscription has
515 // not delivered yet; merging now and then replaying those events
516 // would duplicate them. Park the page until the live state reaches
517 // the page's thread-scoped watermark; loadingOlder stays true so
518 // the UI shows progress and no second fetch starts. Pages from
519 // pre-watermark servers (threadSequence absent) merge immediately,
520 // preserving the old behavior.
521 const watermark = response.value.page?.threadSequence;
522 if (watermark !== undefined && watermark > loadedSequence) {
523 yield* Ref.set(pendingOlderPage, {
524 snapshot: response.value,
525 epoch: epochNow,
526 });
527 return;
528 }
529 yield* mergeOlderPage(response.value);
530 }),
531 );
220532 });
221533
222534 yield* SubscriptionRef.changes(supervisor.state).pipe(
......@@ -244,14 +556,40 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
244556 subscribeDynamic(
245557 ORCHESTRATION_WS_METHODS.subscribeThread,
246558 Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) {
247 const supportsCompletionMarker = yield* session.initialConfig.pipe(
248 Effect.map((config) => config.threadResumeCompletionMarker === true),
249 Effect.orElseSucceed(() => false),
559 const config = yield* session.initialConfig.pipe(
560 Effect.orElseSucceed(
561 () =>
562 ({}) as {
563 threadResumeCompletionMarker?: boolean;
564 threadSnapshotPagination?: boolean;
565 },
566 ),
250567 );
568 const supportsCompletionMarker = config.threadResumeCompletionMarker === true;
569 // Windowed loads are gated on the server capability: pre-pagination
570 // servers reject unknown query params, and a windowed WS fallback to
571 // such a server would silently hide history.
572 const supportsPagination = config.threadSnapshotPagination === true;
573 yield* Ref.set(paginationSupported, supportsPagination);
251574 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
252575 yield* setSynchronizing;
253576
254577 let current = yield* SubscriptionRef.get(state);
578 // A windowed cache resuming against a server without pagination is a
579 // trap: afterSequence resume keeps only the window, and the missing
580 // older turns can never be loaded (the server has no cursor reads).
581 // Drop the window marker and treat the data as needing a full reload.
582 if (!supportsPagination && Option.isSome(current.page)) {
583 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);
584 yield* SubscriptionRef.update(state, (value) => ({
585 ...value,
586 data: Option.none(),
587 status: value.status === "deleted" ? value.status : ("empty" as const),
588 page: Option.none(),
589 }));
590 yield* SubscriptionRef.set(lastSequence, 0);
591 current = yield* SubscriptionRef.get(state);
592 }
255593 if (Option.isNone(current.data) && current.status !== "deleted") {
256594 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
257595 Effect.flatMap(
......@@ -267,7 +605,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
267605 }),
268606 ),
269607 );
270 const httpSnapshot = yield* snapshotLoader.load(prepared, threadId);
608 const httpSnapshot = yield* snapshotLoader.load(
609 prepared,
610 threadId,
611 supportsPagination ? { turnLimit: INITIAL_THREAD_USER_TURN_LIMIT } : undefined,
612 );
271613 if (Option.isSome(httpSnapshot)) {
272614 yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value });
273615 current = yield* SubscriptionRef.get(state);
......@@ -288,6 +630,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
288630 threadId,
289631 ...(canResume ? { afterSequence: sequence } : {}),
290632 ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
633 // The WS fallback snapshot (sent when afterSequence is missing or
634 // the gap is too large) should be windowed the same as the HTTP
635 // path; without this a resume failure re-downloads the full thread.
636 ...(supportsPagination ? { turnLimit: INITIAL_THREAD_USER_TURN_LIMIT } : {}),
291637 };
292638 }),
293639 {
......@@ -298,13 +644,47 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
298644 ).pipe(Stream.runForEach(applyItem)),
299645 );
300646
647 // Expose loadOlderTurns to UI actions through the request registry.
648 // Requests funnel through a sliding queue drained serially, so mashing
649 // "load earlier" coalesces (loadOlderTurns itself no-ops while a fetch is
650 // in flight).
651 const olderTurnRequestRegistry = yield* ThreadOlderTurnRequests;
652 const olderTurnRequests = yield* Queue.sliding<void>(1);
653 yield* Stream.fromQueue(olderTurnRequests).pipe(
654 Stream.runForEach(() => loadOlderTurns()),
655 Effect.forkScoped,
656 );
657 const deregister = olderTurnRequestRegistry.register(
658 threadKey({ environmentId, threadId }),
659 () => {
660 Queue.offerUnsafe(olderTurnRequests, undefined);
661 },
662 );
663 yield* Effect.addFinalizer(() => Effect.sync(deregister));
664
301665 yield* Effect.addFinalizer(() =>
302666 Effect.all([SubscriptionRef.get(state), SubscriptionRef.get(lastSequence)]).pipe(
303667 Effect.flatMap(([current, snapshotSequence]) =>
304668 Option.match(current.data, {
305669 onNone: () => Effect.void,
306670 onSome: (thread) =>
307 shouldPersistThread(thread) ? persist({ snapshotSequence, thread }) : Effect.void,
671 shouldPersistThread(thread)
672 ? persist({
673 snapshotSequence,
674 thread,
675 ...Option.match(current.page, {
676 onNone: () => ({}),
677 onSome: (page) =>
678 ({
679 page: {
680 beforeCursor: page.beforeCursor,
681 hasMore: page.hasMore,
682 snapshotSequence,
683 },
684 }) as const,
685 }),
686 })
687 : Effect.void,
308688 }),
309689 ),
310690 ),
packages/contracts/src/environmentHttp.ts+11
......@@ -457,6 +457,16 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({
457457 threadId: ThreadId,
458458});
459459
460// Query-string window for windowed thread snapshots (GET payloads must encode
461// to strings). Both fields optional: omitting them keeps the full-snapshot
462// behavior, so pagination stays opt-in per request.
463const EnvironmentOrchestrationThreadSnapshotQuery = {
464 turnLimit: Schema.optional(
465 Schema.FiniteFromString.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(1)),
466 ),
467 beforeCursor: Schema.optional(TrimmedNonEmptyString),
468};
469
460470export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration")
461471 .add(
462472 HttpApiEndpoint.get("snapshot", "/api/orchestration/snapshot", {
......@@ -476,6 +486,7 @@ export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestr
476486 HttpApiEndpoint.get("threadSnapshot", "/api/orchestration/threads/:threadId", {
477487 headers: OptionalBearerHeaders,
478488 params: EnvironmentOrchestrationThreadSnapshotParams,
489 payload: EnvironmentOrchestrationThreadSnapshotQuery,
479490 success: OrchestrationThreadDetailSnapshot,
480491 error: EnvironmentOrchestrationThreadSnapshotErrors,
481492 }).middleware(EnvironmentAuthenticatedAuth),
packages/contracts/src/orchestration.ts+51
......@@ -13,6 +13,7 @@ import {
1313 IsoDateTime,
1414 MessageId,
1515 NonNegativeInt,
16 PositiveInt,
1617 ProjectId,
1718 ProviderItemId,
1819 ThreadId,
......@@ -525,12 +526,62 @@ export const OrchestrationSubscribeThreadInput = Schema.Struct({
525526 * snapshot or catch-up replay and before it begins emitting live events.
526527 */
527528 requestCompletionMarker: Schema.optionalKey(Schema.Boolean),
529 /**
530 * When provided, the fallback snapshot frame (sent when `afterSequence` is
531 * missing or the catch-up gap is too large) is windowed to the last
532 * `turnLimit` user-anchored turns and carries `page` metadata. Absent means
533 * the fallback snapshot is the full thread, preserving pre-pagination client
534 * behavior. Live events are unaffected either way.
535 */
536 turnLimit: Schema.optionalKey(PositiveInt),
528537});
529538export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type;
530539
540/**
541 * Bounds a thread detail read to a window of recent turns. `turnLimit` counts
542 * turns with a user pending message (subagent/fan-out turns between them ride
543 * along), so the window always contains the last N user prompts. `beforeCursor`
544 * requests the disjoint page of older turns strictly before a previously
545 * returned cursor. Requests without a window get the full thread; pagination is
546 * strictly opt-in so older clients keep today's behavior on both HTTP and the
547 * WebSocket fallback snapshot.
548 */
549export const OrchestrationThreadDetailWindow = Schema.Struct({
550 turnLimit: Schema.optionalKey(PositiveInt),
551 beforeCursor: Schema.optionalKey(TrimmedNonEmptyString),
552});
553export type OrchestrationThreadDetailWindow = typeof OrchestrationThreadDetailWindow.Type;
554
555/**
556 * Page metadata for a windowed thread detail read. `beforeCursor` is opaque and
557 * exclusive: passing it back returns the adjacent disjoint slice of older
558 * turns. `null` means the thread is fully loaded below this page. The
559 * `snapshotSequence` mirrors the top-level snapshot sequence so history pages
560 * can be sequence-checked against live state before merging.
561 */
562export const OrchestrationThreadDetailPage = Schema.Struct({
563 beforeCursor: Schema.NullOr(TrimmedNonEmptyString),
564 hasMore: Schema.Boolean,
565 snapshotSequence: NonNegativeInt,
566 /**
567 * Highest event sequence applied to THIS thread at page read time. The
568 * global `snapshotSequence` advances with every thread's events, so a
569 * client cannot wait for it via its per-thread subscription; this
570 * thread-scoped watermark is reachable. A client merging an older page
571 * must first have applied live events up to it — otherwise a streaming
572 * turn outside the loaded window could have deltas replayed on top of
573 * page content that already includes them, duplicating text.
574 */
575 threadSequence: Schema.optionalKey(NonNegativeInt),
576});
577export type OrchestrationThreadDetailPage = typeof OrchestrationThreadDetailPage.Type;
578
531579export const OrchestrationThreadDetailSnapshot = Schema.Struct({
532580 snapshotSequence: NonNegativeInt,
533581 thread: OrchestrationThread,
582 // Present only on windowed responses. Absent on full snapshots (and from
583 // pre-pagination servers), which clients treat as fully loaded.
584 page: Schema.optional(OrchestrationThreadDetailPage),
534585});
535586export type OrchestrationThreadDetailSnapshot = typeof OrchestrationThreadDetailSnapshot.Type;
536587
packages/contracts/src/server.ts+6
......@@ -434,6 +434,12 @@ export const ServerConfig = Schema.Struct({
434434 shellResumeCompletionMarker: Schema.optionalKey(Schema.Boolean),
435435 /** Whether thread subscriptions can emit an opt-in catch-up completion marker. */
436436 threadResumeCompletionMarker: Schema.optionalKey(Schema.Boolean),
437 /**
438 * Whether thread detail reads accept a turn window (`turnLimit`/
439 * `beforeCursor`) and return `page` metadata. Clients must not send window
440 * fields to servers that don't advertise this.
441 */
442 threadSnapshotPagination: Schema.optionalKey(Schema.Boolean),
437443});
438444export type ServerConfig = typeof ServerConfig.Type;
439445