diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index 6573c9e1187..ad5ef13b62d 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -17,7 +17,10 @@ import * as Schema from "effect/Schema"; import * as MobileDatabase from "../persistence/mobile-database"; const SHELL_SNAPSHOT_CACHE_SCHEMA_VERSION = 1; -const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 2; +// v3 adds windowed (paginated) snapshots carrying `page` metadata; the bump +// makes pre-pagination clients discard the record instead of decoding a +// partial thread as complete (rollback safety). +const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 3; const SERVER_CONFIG_CACHE_SCHEMA_VERSION = 1; const VCS_REFS_CACHE_SCHEMA_VERSION = 1; diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 5cb04290f66..3d83c837500 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -61,6 +61,8 @@ export interface ThreadDetailScreenProps { readonly connectionStateLabel: EnvironmentConnectionPhase; /** Message sync status for the selected thread (drives the composer status pill). */ readonly threadSyncStatus?: EnvironmentThreadStatus; + /** Non-null when older turns exist beyond the loaded window. */ + readonly loadEarlier?: { readonly loading: boolean; readonly onLoadEarlier: () => void } | null; readonly activeThreadBusy: boolean; readonly environmentId: EnvironmentId; readonly projectWorkspaceRoot: string | null; @@ -371,6 +373,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread usesAutomaticContentInsets={props.usesAutomaticContentInsets} onHeaderMaterialVisibilityChange={props.onHeaderMaterialVisibilityChange} skills={selectedProviderSkills} + loadEarlier={props.loadEarlier ?? null} /> ) : ( diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 8ad117c8635..fd8ffb270cb 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -164,6 +164,11 @@ export interface ThreadFeedProps { readonly usesAutomaticContentInsets?: boolean; readonly onHeaderMaterialVisibilityChange?: (visible: boolean) => void; readonly skills?: ReadonlyArray; + /** Non-null when older turns exist beyond the loaded window. */ + readonly loadEarlier?: { + readonly loading: boolean; + readonly onLoadEarlier: () => void; + } | null; } function MessageAttachmentImage(props: { @@ -1893,7 +1898,20 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { onScroll={handleScroll} scrollEventThrottle={16} ListHeaderComponent={ - usesNativeAutomaticInsets ? null : + <> + {usesNativeAutomaticInsets ? null : } + {props.loadEarlier != null ? ( + + + {props.loadEarlier.loading ? "Loading earlier turns…" : "Load earlier turns"} + + + ) : null} + } contentContainerStyle={{ paddingTop: 12, diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 7fb4740ddce..d7754b7d78f 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -8,6 +8,10 @@ import { import { useCallback, useEffect, useMemo, useRef, useState, type ReactNode } from "react"; import * as Option from "effect/Option"; import { EnvironmentId, ThreadId, type ProjectScript } from "@t3tools/contracts"; +import { + requestOlderThreadTurns, + threadHasOlderTurns, +} from "@t3tools/client-runtime/state/threads"; import { projectScriptCwd, projectScriptRuntimeEnv } from "@t3tools/shared/projectScripts"; import { Platform, ScrollView, View } from "react-native"; import { useSafeAreaInsets } from "react-native-safe-area-context"; @@ -190,6 +194,20 @@ function ThreadRouteContent( useThreadSelection(); const selectedThreadDetailState = props.selectedThreadDetailState; const selectedThreadDetail = Option.getOrNull(selectedThreadDetailState.data); + // "Load earlier turns" header state for windowed (paginated) thread loads. + const loadEarlierTurns = useMemo(() => { + if (selectedThread === null || !threadHasOlderTurns(selectedThreadDetailState)) { + return null; + } + return { + loading: + selectedThreadDetailState.page._tag === "Some" && + selectedThreadDetailState.page.value.loadingOlder, + onLoadEarlier: () => { + requestOlderThreadTurns(selectedThread.environmentId, selectedThread.id); + }, + }; + }, [selectedThread, selectedThreadDetailState]); const { selectedThreadCwd } = useSelectedThreadWorktree(); const composer = useThreadComposerState(); const gitState = useSelectedThreadGitState(); @@ -766,6 +784,7 @@ function ThreadRouteContent( draftAttachments={composer.draftAttachments} connectionStateLabel={routeConnectionState} threadSyncStatus={selectedThreadDetailState.status} + loadEarlier={loadEarlierTurns} activeThreadBusy={composer.activeThreadBusy} environmentId={selectedThread.environmentId} projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null} diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index b7b630a16fd..92c87ebdc04 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -19,6 +19,7 @@ import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "./ProjectionSnapshotQuery.ts"; import * as ThreadBackgroundLiveness from "../ThreadBackgroundLiveness.ts"; import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; +import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value); const asTurnId = (value: string): TurnId => TurnId.make(value); @@ -1917,3 +1918,407 @@ it.effect( }).pipe(Effect.provide(layer)); }, ); + +projectionSnapshotLayer("ProjectionSnapshotQuery windowed thread detail", (it) => { + // A thread shaped like real fan-out usage: user turns interleaved with + // subagent turns (no user pending message), plus a turnless straggler user + // message and a turnless activity anchored between turns. + // + // row turn pending msg anchor (requested_at) + // 1 turn-1 user-msg-1 T00 + // 2 turn-2 (subagent) T01 + // 3 turn-3 (subagent) T02 + // 4 turn-4 user-msg-4 T03 + // 5 turn-5 user-msg-5 T04 + // + // Straggler user message at T03.5 (turn_id NULL, not any pending_message_id) + // and a turnless activity at T03.6 — both belong to the page containing T03+. + const seedFanOutThread = Effect.fnUntraced(function* () { + const sql = yield* SqlClient.SqlClient; + + // Tests in this block share one in-memory database; reset before seeding. + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_turns`; + yield* sql`DELETE FROM projection_thread_messages`; + yield* sql`DELETE FROM projection_thread_activities`; + yield* sql`DELETE FROM projection_state`; + + yield* sql` + INSERT INTO projection_projects ( + project_id, title, workspace_root, scripts_json, created_at, updated_at, deleted_at + ) + VALUES ('project-w', 'Windowed', '/tmp/project-w', '[]', + '2026-03-01T00:00:00.000Z', '2026-03-01T00:00:00.000Z', NULL) + `; + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, + latest_turn_id, pending_approval_count, pending_user_input_count, + has_actionable_proposed_plan, created_at, updated_at, deleted_at + ) + VALUES ('thread-w', 'project-w', 'Windowed thread', + '{"provider":"codex","model":"gpt-5-codex"}', 'full-access', 'default', + 'turn-5', 0, 0, 0, '2026-03-01T00:00:00.000Z', '2026-03-01T00:00:10.000Z', NULL) + `; + + const turns: ReadonlyArray<{ + turn: string; + pendingMessage: string | null; + at: string; + }> = [ + { turn: "turn-1", pendingMessage: "user-msg-1", at: "2026-03-01T00:00:00.000Z" }, + { turn: "turn-2", pendingMessage: null, at: "2026-03-01T00:01:00.000Z" }, + { turn: "turn-3", pendingMessage: null, at: "2026-03-01T00:02:00.000Z" }, + { turn: "turn-4", pendingMessage: "user-msg-4", at: "2026-03-01T00:03:00.000Z" }, + { turn: "turn-5", pendingMessage: "user-msg-5", at: "2026-03-01T00:04:00.000Z" }, + ]; + for (const { turn, pendingMessage, at } of turns) { + yield* sql` + INSERT INTO projection_turns ( + thread_id, turn_id, pending_message_id, state, requested_at, started_at, completed_at, + checkpoint_files_json + ) + VALUES ('thread-w', ${turn}, ${pendingMessage}, 'completed', ${at}, ${at}, ${at}, '[]') + `; + if (pendingMessage !== null) { + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at + ) + VALUES (${pendingMessage}, 'thread-w', NULL, 'user', ${"prompt for " + turn}, 0, ${at}, ${at}) + `; + } + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at + ) + VALUES (${turn + "-reply"}, 'thread-w', ${turn}, 'assistant', ${"reply from " + turn}, 0, ${at}, ${at}) + `; + yield* sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at + ) + VALUES (${turn + "-activity"}, 'thread-w', ${turn}, 'tool', 'tool.completed', + 'ran tool', '{"ok":true}', ${at}) + `; + } + + // Straggler user message sent while turn-4 ran: turn_id NULL and not any + // turn's pending_message_id. + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at + ) + VALUES ('user-msg-straggler', 'thread-w', NULL, 'user', 'while you are at it', + 0, '2026-03-01T00:03:30.000Z', '2026-03-01T00:03:30.000Z') + `; + // Turnless activity in the same time range. + yield* sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, created_at + ) + VALUES ('turnless-activity', 'thread-w', NULL, 'info', 'context-window.updated', + 'usage', '{"usedTokens":1}', '2026-03-01T00:03:36.000Z') + `; + + for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) { + yield* sql` + INSERT INTO projection_state (projector, last_applied_sequence, updated_at) + VALUES (${projector}, 42, '2026-03-01T00:00:10.000Z') + `; + } + }); + + const threadW = ThreadId.make("thread-w"); + const messageIds = (snapshot: { thread: { messages: ReadonlyArray<{ id: string }> } }) => + snapshot.thread.messages.map((message) => message.id).toSorted(); + const activityIds = (snapshot: { thread: { activities: ReadonlyArray<{ id: string }> } }) => + snapshot.thread.activities.map((activity) => activity.id).toSorted(); + + it.effect("returns the full thread with no page metadata when no window is requested", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag === "Some") { + assert.equal(snapshot.value.page, undefined); + assert.equal(snapshot.value.thread.messages.length, 9); + assert.equal(snapshot.value.thread.activities.length, 6); + assert.equal(snapshot.value.snapshotSequence, 42); + } + }), + ); + + it.effect("windows to the last N user-anchored turns with subagent turns riding along", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + // turnLimit 2 walks back: turn-5 (user), turn-4 (user) -> window is + // rows 4..5. Subagent turns 2-3 are older than the 2nd user turn and + // stay out; the straggler message and turnless activity (T03.5/T03.6, + // after turn-4's anchor) ride along. + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 }); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag === "Some") { + assert.deepEqual(messageIds(snapshot.value), [ + "turn-4-reply", + "turn-5-reply", + "user-msg-4", + "user-msg-5", + "user-msg-straggler", + ]); + assert.deepEqual(activityIds(snapshot.value), [ + "turn-4-activity", + "turn-5-activity", + "turnless-activity", + ]); + assert.equal(snapshot.value.page?.hasMore, true); + assert.notEqual(snapshot.value.page?.beforeCursor, null); + assert.equal(snapshot.value.page?.snapshotSequence, 42); + } + }), + ); + + it.effect("subagent turns between user turns ride along inside the window", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + // turnLimit 3 reaches user turn-1, dragging subagent turns 2-3 along: + // the full thread, so no further pages. + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 3 }); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag === "Some") { + assert.equal(snapshot.value.thread.messages.length, 9); + assert.equal(snapshot.value.thread.activities.length, 6); + assert.equal(snapshot.value.page?.hasMore, false); + assert.equal(snapshot.value.page?.beforeCursor, null); + } + }), + ); + + it.effect("cursors survive a projection rewrite that reassigns turn row ids", () => + Effect.gen(function* () { + // The revert projector (and any projection rebuild) deletes and + // re-upserts projection_turns, assigning fresh autoincrement row ids. + // The keyset cursor is derived from event content, so a page cursor + // minted before the rewrite must keep working after it. + yield* seedFanOutThread(); + const sql = yield* SqlClient.SqlClient; + const snapshotQuery = yield* ProjectionSnapshotQuery; + + const firstPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 }); + assert.equal(firstPage._tag, "Some"); + if (firstPage._tag !== "Some") return; + const cursor = firstPage.value.page?.beforeCursor; + assert.notEqual(cursor, null); + if (cursor === null || cursor === undefined) return; + + // Simulate the rewrite: delete and re-insert every turn row with the + // same content, which reassigns all row ids. + const turnRows = yield* sql` + SELECT thread_id, turn_id, pending_message_id, state, requested_at, started_at, + completed_at, checkpoint_files_json + FROM projection_turns WHERE thread_id = 'thread-w' ORDER BY row_id + `; + yield* sql`DELETE FROM projection_turns WHERE thread_id = 'thread-w'`; + for (const row of turnRows) { + yield* sql` + INSERT INTO projection_turns ( + thread_id, turn_id, pending_message_id, state, requested_at, started_at, + completed_at, checkpoint_files_json + ) + VALUES (${row.thread_id as string}, ${row.turn_id as string}, + ${row.pending_message_id as string | null}, ${row.state as string}, + ${row.requested_at as string}, ${row.started_at as string}, + ${row.completed_at as string}, ${row.checkpoint_files_json as string}) + `; + } + + const olderPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { + turnLimit: 1, + beforeCursor: cursor, + }); + assert.equal(olderPage._tag, "Some"); + if (olderPage._tag === "Some") { + // Identical older slice to what the pre-rewrite cursor would return. + assert.deepEqual(messageIds(olderPage.value), [ + "turn-1-reply", + "turn-2-reply", + "turn-3-reply", + "user-msg-1", + ]); + assert.equal(olderPage.value.page?.hasMore, false); + } + }), + ); + + it.effect("beforeCursor returns the disjoint adjacent older slice", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + const firstPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 }); + assert.equal(firstPage._tag, "Some"); + if (firstPage._tag !== "Some") return; + const cursor = firstPage.value.page?.beforeCursor; + assert.notEqual(cursor, null); + assert.notEqual(cursor, undefined); + if (cursor === null || cursor === undefined) return; + + // Older page: user turn-1 plus subagent turns 2-3 riding along. Disjoint + // from the first page: no turn-4/5 rows, no straggler. + const olderPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { + turnLimit: 1, + beforeCursor: cursor, + }); + assert.equal(olderPage._tag, "Some"); + if (olderPage._tag === "Some") { + assert.deepEqual(messageIds(olderPage.value), [ + "turn-1-reply", + "turn-2-reply", + "turn-3-reply", + "user-msg-1", + ]); + assert.deepEqual(activityIds(olderPage.value), [ + "turn-1-activity", + "turn-2-activity", + "turn-3-activity", + ]); + assert.equal(olderPage.value.page?.hasMore, false); + assert.equal(olderPage.value.page?.beforeCursor, null); + } + }), + ); + + it.effect("a cursor for a different thread degrades to the first page", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + const firstPage = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { turnLimit: 2 }); + assert.equal(firstPage._tag, "Some"); + if (firstPage._tag !== "Some") return; + + const foreign = encodeThreadDetailPageCursor({ + threadId: ThreadId.make("thread-other"), + beforeAnchorAt: "2026-03-01T00:01:00.000Z", + beforeTurnId: "turn-2", + }); + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { + turnLimit: 2, + beforeCursor: foreign, + }); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag === "Some") { + assert.deepEqual(messageIds(snapshot.value), messageIds(firstPage.value)); + } + }), + ); + + it.effect("a malformed cursor degrades to the first page instead of failing", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { + turnLimit: 2, + beforeCursor: "not-a-cursor", + }); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag === "Some") { + assert.equal(snapshot.value.page?.hasMore, true); + assert.equal(snapshot.value.thread.messages.length, 5); + } + }), + ); + + it.effect("windows never split below the raw-turn ceiling boundary contiguously", () => + Effect.gen(function* () { + yield* seedFanOutThread(); + const snapshotQuery = yield* ProjectionSnapshotQuery; + + // Page repeatedly with turnLimit 1 and assert the union of all pages is + // exactly the full thread with no duplicates (disjointness + coverage). + const seenMessages: string[] = []; + const seenActivities: string[] = []; + let cursor: string | undefined; + for (let page = 0; page < 10; page += 1) { + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(threadW, { + turnLimit: 1, + ...(cursor !== undefined ? { beforeCursor: cursor } : {}), + }); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag !== "Some") return; + seenMessages.push(...snapshot.value.thread.messages.map((message) => message.id)); + seenActivities.push(...snapshot.value.thread.activities.map((activity) => activity.id)); + const next = snapshot.value.page?.beforeCursor; + if (next === null || next === undefined) break; + cursor = next; + } + assert.equal(new Set(seenMessages).size, seenMessages.length); + assert.equal(new Set(seenActivities).size, seenActivities.length); + assert.equal(seenMessages.length, 9); + assert.equal(seenActivities.length, 6); + }), + ); + + it.effect("a thread with no turns returns its content unwindowed on the first page", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const snapshotQuery = yield* ProjectionSnapshotQuery; + + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_turns`; + yield* sql`DELETE FROM projection_thread_messages`; + yield* sql`DELETE FROM projection_thread_activities`; + yield* sql`DELETE FROM projection_state`; + + yield* sql` + INSERT INTO projection_projects ( + project_id, title, workspace_root, scripts_json, created_at, updated_at, deleted_at + ) + VALUES ('project-e', 'Empty', '/tmp/project-e', '[]', + '2026-03-02T00:00:00.000Z', '2026-03-02T00:00:00.000Z', NULL) + `; + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, + pending_approval_count, pending_user_input_count, has_actionable_proposed_plan, + created_at, updated_at, deleted_at + ) + VALUES ('thread-e', 'project-e', 'Turnless thread', + '{"provider":"codex","model":"gpt-5-codex"}', 'full-access', 'default', + 0, 0, 0, '2026-03-02T00:00:00.000Z', '2026-03-02T00:00:00.000Z', NULL) + `; + yield* sql` + INSERT INTO projection_thread_messages ( + message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at + ) + VALUES ('pre-turn-msg', 'thread-e', NULL, 'user', 'first prompt', 0, + '2026-03-02T00:00:01.000Z', '2026-03-02T00:00:01.000Z') + `; + for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) { + yield* sql` + INSERT INTO projection_state (projector, last_applied_sequence, updated_at) + VALUES (${projector}, 7, '2026-03-02T00:00:01.000Z') + `; + } + + const snapshot = yield* snapshotQuery.getThreadDetailSnapshot(ThreadId.make("thread-e"), { + turnLimit: 5, + }); + assert.equal(snapshot._tag, "Some"); + if (snapshot._tag === "Some") { + assert.deepEqual(messageIds(snapshot.value), ["pre-turn-msg"]); + assert.equal(snapshot.value.page?.hasMore, false); + assert.equal(snapshot.value.page?.beforeCursor, null); + } + }), + ); +}); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 2d8a98d8c6f..f036198fe49 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -51,6 +51,10 @@ import { ProjectionThreadMessage } from "../../persistence/Services/ProjectionTh import { ProjectionThreadProposedPlan } from "../../persistence/Services/ProjectionThreadProposedPlans.ts"; import { ProjectionThreadSession } from "../../persistence/Services/ProjectionThreadSessions.ts"; import { ProjectionThread } from "../../persistence/Services/ProjectionThreads.ts"; +import { + decodeThreadDetailPageCursor, + encodeThreadDetailPageCursor, +} from "../threadDetailCursor.ts"; import * as RepositoryIdentityResolver from "../../project/RepositoryIdentityResolver.ts"; import { ORCHESTRATION_PROJECTOR_NAMES } from "./ProjectionPipeline.ts"; import { @@ -130,6 +134,36 @@ const ProjectIdLookupInput = Schema.Struct({ const ThreadIdLookupInput = Schema.Struct({ threadId: ThreadId, }); +// Windowed reads order turns by the stable keyset (anchor, turn key), where +// anchor is requested_at and turn key is +// COALESCE(turn_id, ''). Both are event-derived, so cursors survive the +// revert projector's row-id rewrite and full projection rebuilds. +const ThreadTurnWindowLookupInput = Schema.Struct({ + threadId: ThreadId, + // Exclusive keyset upper bound. Sentinels "~"/"" mean unbounded ("~" sorts + // after every ISO timestamp). + beforeAnchorAt: Schema.String, + beforeTurnKey: Schema.String, + userTurnLimit: Schema.Number, + maxRawTurns: Schema.Number, +}); +const ProjectionTurnWindowRowSchema = Schema.Struct({ + // The turn's timeline anchor, used to bound rows that have no turn linkage + // (user messages and turnless activities) to the same page window. + anchorAt: Schema.String, + turnKey: Schema.String, +}); +const ThreadTurnRangeLookupInput = Schema.Struct({ + threadId: ThreadId, + // Turn-linked rows are bounded by the keyset range [min, before) over + // (anchor, turn key); turnless rows by the matching [minAnchorAt, + // beforeAnchorAt) time range. Unbounded ends use sentinels: "" for the + // lower bound, "~" (sorts after ISO dates) for the upper bound. + minAnchorAt: Schema.String, + minTurnKey: Schema.String, + beforeAnchorAt: Schema.String, + beforeTurnKey: Schema.String, +}); const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema; const ProjectionThreadIdLookupRowSchema = Schema.Struct({ threadId: ThreadId, @@ -1043,6 +1077,197 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + // Resolves a page of recent turns for a windowed thread detail read. Walks + // back from the exclusive (beforeAnchorAt, beforeTurnKey) keyset boundary + // (sentinels "~"/"" mean unbounded, i.e. the first page) until it has seen + // `userTurnLimit` user-anchored turns — turns whose pending message is a + // user message; subagent/fan-out turns between them ride along — or hits the + // `maxRawTurns` ceiling that bounds pathological fan-out. The `candidates` + // CTE applies the keyset bound and LIMIT before the window functions run; + // its ORDER BY uses raw columns so the migration-037 + // (thread_id, requested_at, turn_id) index serves both range and order with + // no temp B-tree — the scan is genuinely bounded by the LIMIT. (Raw + // turn_id DESC places NULLs exactly where COALESCE-to-'' would, below every + // real id.) The caller derives the continuation cursor from the oldest + // returned row. + // Highest thread-DETAIL event sequence for this thread that the projection + // has applied (bounded by the global snapshot sequence read in the same + // transaction). This is the thread-scoped watermark a windowed page carries + // so clients can defer merging until their live subscription has caught up; + // the global sequence is not waitable per-thread. The event_type filter + // must match ws.ts's isThreadDetailEvent exactly: the subscription only + // delivers these types, so a watermark counting any other event could + // never be reached by the client and would park the page forever. Served + // by the event store's (aggregate_kind, stream_id, sequence) index. + const getThreadEventWatermarkRow = SqlSchema.findOneOption({ + Request: Schema.Struct({ threadId: ThreadId, maxSequence: Schema.Number }), + Result: Schema.Struct({ threadSequence: Schema.NullOr(Schema.Number) }), + execute: ({ threadId, maxSequence }) => + sql` + SELECT MAX(sequence) AS "threadSequence" + FROM orchestration_events + WHERE aggregate_kind = 'thread' + AND stream_id = ${threadId} + AND sequence <= ${maxSequence} + AND event_type IN ( + 'thread.message-sent', + 'thread.proposed-plan-upserted', + 'thread.activity-appended', + 'thread.turn-diff-completed', + 'thread.reverted', + 'thread.session-set' + ) + `, + }); + + const listTurnWindowRows = SqlSchema.findAll({ + Request: ThreadTurnWindowLookupInput, + Result: ProjectionTurnWindowRowSchema, + execute: ({ threadId, beforeAnchorAt, beforeTurnKey, userTurnLimit, maxRawTurns }) => + sql` + WITH candidates AS ( + SELECT + turns.requested_at AS anchor_at, + COALESCE(turns.turn_id, '') AS turn_key, + turns.pending_message_id + FROM projection_turns AS turns + WHERE turns.thread_id = ${threadId} + AND ( + turns.requested_at < ${beforeAnchorAt} + OR ( + turns.requested_at = ${beforeAnchorAt} + AND COALESCE(turns.turn_id, '') < ${beforeTurnKey} + ) + ) + ORDER BY turns.requested_at DESC, turns.turn_id DESC + LIMIT ${maxRawTurns} + ), + walked AS ( + SELECT + candidates.anchor_at, + candidates.turn_key, + CASE WHEN messages.role = 'user' THEN 1 ELSE 0 END AS is_user_turn, + SUM(CASE WHEN messages.role = 'user' THEN 1 ELSE 0 END) OVER ( + ORDER BY candidates.anchor_at DESC, candidates.turn_key DESC + ) AS user_turns_seen + FROM candidates + LEFT JOIN projection_thread_messages AS messages + ON messages.message_id = candidates.pending_message_id + ) + SELECT + anchor_at AS "anchorAt", + turn_key AS "turnKey" + FROM walked + WHERE user_turns_seen < ${userTurnLimit} + OR (user_turns_seen = ${userTurnLimit} AND is_user_turn = 1) + ORDER BY anchor_at ASC, turn_key ASC + `, + }); + + // Windowed variants of the two heavy collections. Turn-linked rows are + // bounded by the page's (anchor, turn key) keyset range over + // projection_turns; rows with no turn linkage (user messages always, and + // turnless activities like pre-turn context-window updates) are bounded by + // the matching turn-anchor time range so they land on the same page as the + // turns around them. Proposed plans and checkpoints stay unwindowed: they + // are metadata-scale. + const listThreadMessageRowsByThreadWindow = SqlSchema.findAll({ + Request: ThreadTurnRangeLookupInput, + Result: ProjectionThreadMessageDbRowSchema, + execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) => + sql` + SELECT + message_id AS "messageId", + thread_id AS "threadId", + turn_id AS "turnId", + role, + text, + attachments_json AS "attachments", + is_streaming AS "isStreaming", + created_at AS "createdAt", + updated_at AS "updatedAt" + FROM projection_thread_messages + WHERE thread_id = ${threadId} + AND ( + turn_id IN ( + SELECT turn_id FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id IS NOT NULL + AND ( + requested_at > ${minAnchorAt} + OR ( + requested_at = ${minAnchorAt} + AND turn_id >= ${minTurnKey} + ) + ) + AND ( + requested_at < ${beforeAnchorAt} + OR ( + requested_at = ${beforeAnchorAt} + AND turn_id < ${beforeTurnKey} + ) + ) + ) + OR ( + turn_id IS NULL + AND created_at >= ${minAnchorAt} + AND created_at < ${beforeAnchorAt} + ) + ) + ORDER BY created_at ASC, message_id ASC + `, + }); + + const listThreadActivityRowsByThreadWindow = SqlSchema.findAll({ + Request: ThreadTurnRangeLookupInput, + Result: ProjectionThreadActivityDbRowSchema, + execute: ({ threadId, minAnchorAt, minTurnKey, beforeAnchorAt, beforeTurnKey }) => + sql` + SELECT + activity_id AS "activityId", + thread_id AS "threadId", + turn_id AS "turnId", + tone, + kind, + summary, + payload_json AS "payload", + sequence, + created_at AS "createdAt" + FROM projection_thread_activities + WHERE thread_id = ${threadId} + AND ( + turn_id IN ( + SELECT turn_id FROM projection_turns + WHERE thread_id = ${threadId} + AND turn_id IS NOT NULL + AND ( + requested_at > ${minAnchorAt} + OR ( + requested_at = ${minAnchorAt} + AND turn_id >= ${minTurnKey} + ) + ) + AND ( + requested_at < ${beforeAnchorAt} + OR ( + requested_at = ${beforeAnchorAt} + AND turn_id < ${beforeTurnKey} + ) + ) + ) + OR ( + turn_id IS NULL + AND created_at >= ${minAnchorAt} + AND created_at < ${beforeAnchorAt} + ) + ) + ORDER BY + sequence ASC, + created_at ASC, + activity_id ASC + `, + }); + const getFullThreadDiffContextRow = SqlSchema.findOneOption({ Request: FullThreadDiffContextLookupInput, Result: ProjectionFullThreadDiffContextRowSchema, @@ -2104,7 +2329,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { } satisfies OrchestrationThreadShell); }); - const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (threadId) => + // Contiguous turn range bounding a windowed detail read; undefined loads the + // full thread. Resolved from a window request inside the snapshot + // transaction (see getThreadDetailSnapshot). + interface ThreadDetailBounds { + readonly minAnchorAt: string; + readonly minTurnKey: string; + readonly beforeAnchorAt: string; + readonly beforeTurnKey: string; + } + + const getThreadDetailByIdBounded = (threadId: ThreadId, bounds: ThreadDetailBounds | undefined) => Effect.gen(function* () { const [ threadRow, @@ -2123,7 +2358,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), - listThreadMessageRowsByThread({ threadId }).pipe( + (bounds === undefined + ? listThreadMessageRowsByThread({ threadId }) + : listThreadMessageRowsByThreadWindow({ threadId, ...bounds }) + ).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getThreadDetailById:listMessages:query", @@ -2139,7 +2377,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ), ), - listThreadActivityRowsByThread({ threadId }).pipe( + (bounds === undefined + ? listThreadActivityRowsByThread({ threadId }) + : listThreadActivityRowsByThreadWindow({ threadId, ...bounds }) + ).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProjectionSnapshotQuery.getThreadDetailById:listActivities:query", @@ -2249,23 +2490,139 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ); }); + const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (threadId) => + getThreadDetailByIdBounded(threadId, undefined); + + // Bounds pathological fan-out: one user turn that spawned hundreds of + // subagent turns still pages in bounded chunks, at the cost of splitting the + // fan-out group across pages (the cursor continues the same group). Also + // structurally bounds the window scan via the candidates CTE's LIMIT. + const THREAD_DETAIL_MAX_RAW_TURNS_PER_PAGE = 150; + // Sentinels for unbounded keyset ends; "~" sorts after any ISO timestamp. + const ANCHOR_UNBOUNDED = "~"; + const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = ( threadId, + window, ) => // Read the thread detail and the snapshot sequence within a single // transaction so the sequence is consistent with the returned state; a // projector update landing between two separate reads could otherwise return // a sequence ahead of the thread detail, causing the client to resume from - // too far and drop events. + // too far and drop events. Window resolution runs inside the same + // transaction so the page boundary is consistent with the returned rows. sql .withTransaction( Effect.gen(function* () { - const thread = yield* getThreadDetailById(threadId); + if (window?.turnLimit === undefined) { + const thread = yield* getThreadDetailById(threadId); + if (Option.isNone(thread)) { + return Option.none(); + } + const { snapshotSequence } = yield* getSnapshotSequence(); + return Option.some({ snapshotSequence, thread: thread.value }); + } + + // A malformed or foreign-thread cursor falls back to the first page + // rather than failing: the client's stale cursor after a revert or + // reconnect should degrade to "reload recent history", not error. + const decodedCursor = + window.beforeCursor === undefined + ? null + : decodeThreadDetailPageCursor(window.beforeCursor); + const cursor = decodedCursor?.threadId === threadId ? decodedCursor : null; + + const windowRows = yield* listTurnWindowRows({ + threadId, + beforeAnchorAt: cursor?.beforeAnchorAt ?? ANCHOR_UNBOUNDED, + beforeTurnKey: cursor?.beforeTurnId ?? "", + userTurnLimit: window.turnLimit, + maxRawTurns: THREAD_DETAIL_MAX_RAW_TURNS_PER_PAGE, + }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadDetailSnapshot:listTurnWindow:query", + "ProjectionSnapshotQuery.getThreadDetailSnapshot:listTurnWindow:decodeRows", + ), + ), + ); + + const oldest = windowRows[0]; + // An empty window (no turns before the cursor, or a thread with no + // turns at all) still returns thread metadata with empty collections + // for turn-linked rows; turnless rows are bounded to the same empty + // range. The first page of a turnless thread stays unwindowed so + // pre-turn content (e.g. a just-created thread) is not hidden. + const bounds: ThreadDetailBounds | undefined = + oldest === undefined && cursor === null + ? undefined + : { + minAnchorAt: oldest?.anchorAt ?? "", + minTurnKey: oldest?.turnKey ?? "", + beforeAnchorAt: cursor?.beforeAnchorAt ?? ANCHOR_UNBOUNDED, + beforeTurnKey: cursor?.beforeTurnId ?? "", + }; + // Empty window behind a cursor: nothing older remains. + const emptyBounds = + oldest === undefined && cursor !== null + ? { minAnchorAt: "", minTurnKey: "", beforeAnchorAt: "", beforeTurnKey: "" } + : undefined; + + const thread = yield* getThreadDetailByIdBounded(threadId, emptyBounds ?? bounds); if (Option.isNone(thread)) { return Option.none(); } + + const hasMore = + oldest !== undefined && + (yield* listTurnWindowRows({ + threadId, + beforeAnchorAt: oldest.anchorAt, + beforeTurnKey: oldest.turnKey, + userTurnLimit: 1, + maxRawTurns: 1, + }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadDetailSnapshot:probeOlder:query", + "ProjectionSnapshotQuery.getThreadDetailSnapshot:probeOlder:decodeRows", + ), + ), + )).length > 0; + const { snapshotSequence } = yield* getSnapshotSequence(); - return Option.some({ snapshotSequence, thread: thread.value }); + const watermarkRow = yield* getThreadEventWatermarkRow({ + threadId, + maxSequence: snapshotSequence, + }).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadDetailSnapshot:threadWatermark:query", + "ProjectionSnapshotQuery.getThreadDetailSnapshot:threadWatermark:decodeRow", + ), + ), + ); + const threadSequence = Option.match(watermarkRow, { + onNone: () => 0, + onSome: (row) => row.threadSequence ?? 0, + }); + return Option.some({ + snapshotSequence, + thread: thread.value, + page: { + beforeCursor: + hasMore && oldest !== undefined + ? encodeThreadDetailPageCursor({ + threadId, + beforeAnchorAt: oldest.anchorAt, + beforeTurnId: oldest.turnKey, + }) + : null, + hasMore, + snapshotSequence, + threadSequence, + }, + }); }), ) .pipe( diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 64138fb7559..0a00253a228 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -17,6 +17,7 @@ import type { OrchestrationShellSnapshot, OrchestrationThread, OrchestrationThreadDetailSnapshot, + OrchestrationThreadDetailWindow, OrchestrationThreadShell, ProjectId, ThreadId, @@ -174,9 +175,16 @@ export interface ProjectionSnapshotQueryShape { * sequence in one consistent transaction, so the returned `snapshotSequence` * exactly matches the state reflected in `thread` (no interleaving projector * update between the two reads). + * + * When `window` is provided, the thread's messages, activities, proposed + * plans, and checkpoints are bounded to a page of recent turns and the + * response carries `page` metadata (see `OrchestrationThreadDetailWindow`). + * Without a window the full thread is returned with no `page` field — + * pagination is strictly opt-in. */ readonly getThreadDetailSnapshot: ( threadId: ThreadId, + window?: OrchestrationThreadDetailWindow, ) => Effect.Effect, ProjectionRepositoryError>; } diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 9a5c8c0be39..04d54ea8eff 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -66,7 +66,17 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); const snapshot = yield* projectionSnapshotQuery - .getThreadDetailSnapshot(args.params.threadId) + .getThreadDetailSnapshot( + args.params.threadId, + args.payload.turnLimit === undefined + ? undefined + : { + turnLimit: args.payload.turnLimit, + ...(args.payload.beforeCursor !== undefined + ? { beforeCursor: args.payload.beforeCursor } + : {}), + }, + ) .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_thread_snapshot_failed", cause), diff --git a/apps/server/src/orchestration/threadDetailCursor.test.ts b/apps/server/src/orchestration/threadDetailCursor.test.ts new file mode 100644 index 00000000000..434d83e86b1 --- /dev/null +++ b/apps/server/src/orchestration/threadDetailCursor.test.ts @@ -0,0 +1,44 @@ +import { ThreadId } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; + +import { + decodeThreadDetailPageCursor, + encodeThreadDetailPageCursor, +} from "./threadDetailCursor.ts"; + +describe("threadDetailCursor", () => { + it("round-trips a cursor", () => { + const cursor = { + threadId: ThreadId.make("thread-1"), + beforeAnchorAt: "2026-08-01T00:00:00.000Z", + beforeTurnId: "turn-9", + }; + expect(decodeThreadDetailPageCursor(encodeThreadDetailPageCursor(cursor))).toEqual(cursor); + }); + + it("round-trips empty boundary values", () => { + // The anchor is COALESCE(requested_at, started_at, '') and the turn key + // is COALESCE(turn_id, ''), so a server-minted cursor can legitimately + // carry empty strings; rejecting them would degrade a valid cursor to a + // first-page request that repeats recent history (review finding). + const cursor = { + threadId: ThreadId.make("thread-1"), + beforeAnchorAt: "", + beforeTurnId: "", + }; + expect(decodeThreadDetailPageCursor(encodeThreadDetailPageCursor(cursor))).toEqual(cursor); + }); + + it("rejects malformed input", () => { + expect(decodeThreadDetailPageCursor("not-base64-json")).toBeNull(); + expect(decodeThreadDetailPageCursor(Buffer.from("[]").toString("base64url"))).toBeNull(); + expect( + decodeThreadDetailPageCursor(Buffer.from(JSON.stringify({ t: "" })).toString("base64url")), + ).toBeNull(); + expect( + decodeThreadDetailPageCursor( + Buffer.from(JSON.stringify({ t: "thread-1", a: 5, i: "x" })).toString("base64url"), + ), + ).toBeNull(); + }); +}); diff --git a/apps/server/src/orchestration/threadDetailCursor.ts b/apps/server/src/orchestration/threadDetailCursor.ts new file mode 100644 index 00000000000..a7dcf231ee6 --- /dev/null +++ b/apps/server/src/orchestration/threadDetailCursor.ts @@ -0,0 +1,62 @@ +import type { ThreadId } from "@t3tools/contracts"; + +/** + * Opaque, exclusive cursor for windowed thread detail reads. Encodes the thread + * id and the keyset boundary of an already-delivered page: the boundary turn's + * anchor timestamp (`COALESCE(requested_at, started_at, '')`) and turn id. + * Passing it back requests the adjacent disjoint slice of strictly older turns + * under `(anchor, turn_id)` ordering. + * + * The boundary is deliberately NOT a `projection_turns.row_id`: row ids are + * rewritten by the revert projector (delete + re-upsert) and by projection + * rebuilds, which would silently invalidate every persisted cursor with no + * event emitted. The (anchor, turnId) pair is derived from event content, so + * cursors survive both and no client-side refresh machinery is needed. The + * anchor doubles as the time bound for rows with no turn linkage (straggler + * user messages, turnless activities). The thread id is embedded so a cursor + * can never be replayed against a different thread. Clients must treat the + * string as opaque. + */ +export interface ThreadDetailPageCursor { + readonly threadId: ThreadId; + readonly beforeAnchorAt: string; + /** Boundary turn id; "" for the rare turn row with a null turn_id. */ + readonly beforeTurnId: string; +} + +export function encodeThreadDetailPageCursor(cursor: ThreadDetailPageCursor): string { + return Buffer.from( + JSON.stringify({ t: cursor.threadId, a: cursor.beforeAnchorAt, i: cursor.beforeTurnId }), + ).toString("base64url"); +} + +/** + * Returns null for anything that is not a well-formed cursor. Callers degrade + * a malformed or foreign-thread cursor to a first-page request. + */ +export function decodeThreadDetailPageCursor(encoded: string): ThreadDetailPageCursor | null { + let parsed: unknown; + try { + parsed = JSON.parse(Buffer.from(encoded, "base64url").toString("utf8")); + } catch { + return null; + } + if (parsed === null || typeof parsed !== "object") { + return null; + } + const record = parsed as Record; + if (typeof record.t !== "string" || record.t.length === 0) { + return null; + } + // Empty strings are valid boundary values, not malformed input: the anchor + // is COALESCE(requested_at, started_at, ''), so a boundary turn with no + // timestamps encodes a: "" (and sorts before every real anchor, correctly + // ending the walk); the turn key is "" for a null turn_id. + if (typeof record.a !== "string") { + return null; + } + if (typeof record.i !== "string") { + return null; + } + return { threadId: record.t as ThreadId, beforeAnchorAt: record.a, beforeTurnId: record.i }; +} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 1309cd7ef59..1f335bdfda7 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -49,6 +49,7 @@ import Migration0033 from "./Migrations/033_ProjectionThreadsSettled.ts"; import Migration0034 from "./Migrations/034_ProjectionThreadsSnoozed.ts"; import Migration0035 from "./Migrations/035_ProjectionThreadTitleRegeneration.ts"; import Migration0036 from "./Migrations/036_ProjectionThreadsPinned.ts"; +import Migration0037 from "./Migrations/037_ProjectionTurnsKeysetIndex.ts"; /** * Migration loader with all migrations defined inline. @@ -97,6 +98,7 @@ export const migrationEntries = [ [34, "ProjectionThreadsSnoozed", Migration0034], [35, "ProjectionThreadTitleRegeneration", Migration0035], [36, "ProjectionThreadsPinned", Migration0036], + [37, "ProjectionTurnsKeysetIndex", Migration0037], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/037_ProjectionTurnsKeysetIndex.ts b/apps/server/src/persistence/Migrations/037_ProjectionTurnsKeysetIndex.ts new file mode 100644 index 00000000000..6b1ee7c0304 --- /dev/null +++ b/apps/server/src/persistence/Migrations/037_ProjectionTurnsKeysetIndex.ts @@ -0,0 +1,17 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +/** + * Composite index for windowed thread detail reads. Pagination orders turns by + * the stable keyset (requested_at, turn_id); the pre-existing + * (thread_id, requested_at) index cannot serve the tiebreak order, forcing a + * temp B-tree over all of a thread's turns before the page LIMIT applies. + * With this index the candidates scan is genuinely bounded by the page size. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_turns_thread_keyset + ON projection_turns(thread_id, requested_at, turn_id) + `; +}); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index a04fce3fd2c..6bafb9ec3ba 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1019,6 +1019,7 @@ const makeWsRpcLayer = ( settings, shellResumeCompletionMarker: true, threadResumeCompletionMarker: true, + threadSnapshotPagination: true, }; }); @@ -1351,7 +1352,14 @@ const makeWsRpcLayer = ( } const snapshot = yield* projectionSnapshotQuery - .getThreadDetailSnapshot(input.threadId) + .getThreadDetailSnapshot( + input.threadId, + // Windowing the fallback snapshot is opt-in per subscription: + // clients that don't send turnLimit (including all + // pre-pagination clients) get the full thread, since they + // have no way to load older pages. + input.turnLimit === undefined ? undefined : { turnLimit: input.turnLimit }, + ) .pipe( Effect.mapError( (cause) => diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 6c2dc1478e6..f17e7021c44 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -223,7 +223,11 @@ import { serverEnvironment, } from "../state/server"; import { terminalEnvironment } from "../state/terminal"; -import { threadEnvironment } from "../state/threads"; +import { threadEnvironment, useEnvironmentThread } from "../state/threads"; +import { + requestOlderThreadTurns, + threadHasOlderTurns, +} from "@t3tools/client-runtime/state/threads"; import { vcsEnvironment } from "../state/vcs"; import { useEnvironments, usePrimaryEnvironment } from "../state/environments"; import { @@ -1239,6 +1243,23 @@ function ChatViewContent(props: ChatViewProps) { [routeServerThreadShell, threadDetailLoading], ); const activeServerThread = serverThread ?? loadingServerThread; + // Pagination window state for the routed server thread: drives the + // "load earlier turns" header when the loaded window has older history. + const routeThreadState = useEnvironmentThread( + routeKind === "server" ? routeThreadRef.environmentId : null, + routeKind === "server" ? routeThreadRef.threadId : null, + ); + const loadEarlierTurns = useMemo(() => { + if (routeKind !== "server" || !threadHasOlderTurns(routeThreadState)) { + return null; + } + return { + loading: routeThreadState.page._tag === "Some" && routeThreadState.page.value.loadingOlder, + onLoadEarlier: () => { + requestOlderThreadTurns(routeThreadRef.environmentId, routeThreadRef.threadId); + }, + }; + }, [routeKind, routeThreadRef, routeThreadState]); const markThreadVisited = useUiStateStore((store) => store.markThreadVisited); const settings = useEnvironmentSettings(environmentId); // New-thread defaults live in the primary environment's settings.json (the @@ -6029,6 +6050,7 @@ function ChatViewContent(props: ChatViewProps) { onManualNavigation={cancelTimelineLiveFollowForUserNavigation} hideEmptyPlaceholder={isDraftHeroState || threadDetailLoading} topFadeEnabled={!hasTimelineTopBanner} + loadEarlier={loadEarlierTurns} /> {/* scroll to end pill — shown when user has scrolled away from the live edge */} diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index c952eb3d128..8e27b7b6962 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -158,6 +158,33 @@ const TimelineRowCtx = createContext(null!); const TimelineRowActivityCtx = createContext(null!); const TIMELINE_LIST_HEADER =
; const TIMELINE_LIST_FADE_HEADER =
; + +// Header row shown when older turns exist beyond the loaded window. Plain +// button, no spinner animation; the label change is the loading indicator. +function TimelineLoadEarlierHeader({ + loading, + onLoadEarlier, + fade, +}: { + loading: boolean; + onLoadEarlier: () => void; + fade: boolean; +}) { + return ( +
+
+ +
+
+ ); +} const TIMELINE_LIST_FOOTER =
; const EMPTY_TIMELINE_SKILLS: ReadonlyArray> = []; @@ -196,6 +223,8 @@ interface MessagesTimelineProps { onManualNavigation: () => void; hideEmptyPlaceholder?: boolean; topFadeEnabled?: boolean; + /** Non-null when older turns exist beyond the loaded window. */ + loadEarlier?: { readonly loading: boolean; readonly onLoadEarlier: () => void } | null; } // --------------------------------------------------------------------------- @@ -233,6 +262,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ onManualNavigation, hideEmptyPlaceholder = false, topFadeEnabled = false, + loadEarlier = null, }: MessagesTimelineProps) { const [expandedTurnIds, setExpandedTurnIds] = useState>(new Set()); const [expandedWorkGroupIds, setExpandedWorkGroupIds] = useState>(new Set()); @@ -533,7 +563,19 @@ export const MessagesTimeline = memo(function MessagesTimeline({ "scrollbar-gutter-both h-full min-h-0 overflow-x-hidden overscroll-y-contain px-3 [overflow-anchor:none] sm:px-5", topFadeEnabled && "chat-timeline-scroll-fade", )} - ListHeaderComponent={topFadeEnabled ? TIMELINE_LIST_FADE_HEADER : TIMELINE_LIST_HEADER} + ListHeaderComponent={ + loadEarlier !== null ? ( + + ) : topFadeEnabled ? ( + TIMELINE_LIST_FADE_HEADER + ) : ( + TIMELINE_LIST_HEADER + ) + } ListFooterComponent={TIMELINE_LIST_FOOTER} /> Effect.gen(function* () { const encoded = yield* encodeStoredThreadSnapshot({ - schemaVersion: 2, + schemaVersion: 3, environmentId, threadId: snapshot.thread.id, snapshot, diff --git a/packages/client-runtime/src/state/entities.test.ts b/packages/client-runtime/src/state/entities.test.ts index e08fd9e552f..d3bb6680208 100644 --- a/packages/client-runtime/src/state/entities.test.ts +++ b/packages/client-runtime/src/state/entities.test.ts @@ -333,6 +333,7 @@ describe("environment entity projections", () => { data: Option.some(detail), status: "live", error: Option.none(), + page: Option.none(), }), ); @@ -361,6 +362,7 @@ describe("environment entity projections", () => { }), status: "live", error: Option.none(), + page: Option.none(), }), ); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index 874bcc30ebd..6acc3b5d8a4 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -26,6 +26,16 @@ const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000; * WebSocket subscription's first frame. The response is gzip-compressible by * the transport and keeps the (potentially multi-KB) snapshot off the socket. */ +/** + * Optional turn window for a snapshot fetch. Only send a window to servers + * that advertise `threadSnapshotPagination`; older servers reject unknown + * query parameters. + */ +export interface ThreadSnapshotWindow { + readonly turnLimit: number; + readonly beforeCursor?: string; +} + export const fetchEnvironmentThreadSnapshot = Effect.fn( "clientRuntime.state.fetchEnvironmentThreadSnapshot", )(function* (input: { @@ -33,6 +43,7 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( readonly threadId: ThreadId; readonly signer: Option.Option; readonly timeoutMs?: number; + readonly window?: ThreadSnapshotWindow; }) { const requestUrl = environmentEndpointUrl( input.prepared.httpBaseUrl, @@ -52,6 +63,12 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( input.prepared.httpAuthorization, client.orchestration.threadSnapshot({ params: { threadId: input.threadId }, + payload: { + ...(input.window !== undefined ? { turnLimit: input.window.turnLimit } : {}), + ...(input.window?.beforeCursor !== undefined + ? { beforeCursor: input.window.beforeCursor } + : {}), + }, headers, }), ), @@ -72,6 +89,7 @@ export class ThreadSnapshotLoader extends Context.Service< readonly load: ( prepared: PreparedConnection, threadId: ThreadId, + window?: ThreadSnapshotWindow, ) => Effect.Effect>; } >()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {} @@ -89,8 +107,13 @@ export const threadSnapshotLoaderLayer: Layer.Layer< // connections work without one). const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner); return ThreadSnapshotLoader.of({ - load: (prepared: PreparedConnection, threadId: ThreadId) => - fetchEnvironmentThreadSnapshot({ prepared, threadId, signer }).pipe( + load: (prepared: PreparedConnection, threadId: ThreadId, window?: ThreadSnapshotWindow) => + fetchEnvironmentThreadSnapshot({ + prepared, + threadId, + signer, + ...(window !== undefined ? { window } : {}), + }).pipe( Effect.map(Option.some), Effect.provideService(HttpClient.HttpClient, httpClient), // A genuinely missing thread (404) is expected — the socket diff --git a/packages/client-runtime/src/state/threadState.ts b/packages/client-runtime/src/state/threadState.ts index 89be139e925..8ba9696ec57 100644 --- a/packages/client-runtime/src/state/threadState.ts +++ b/packages/client-runtime/src/state/threadState.ts @@ -3,14 +3,38 @@ import * as Option from "effect/Option"; export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted"; +/** + * Pagination state for a windowed thread. Present only when the loaded thread + * is a window (the server returned `page` metadata); absent means the thread is + * fully loaded — either the server predates pagination or the window reached + * the top. + */ +export interface EnvironmentThreadPageState { + /** Opaque exclusive cursor for the next older slice; null when fully loaded. */ + readonly beforeCursor: string | null; + readonly hasMore: boolean; + /** True while an older page fetch is in flight. */ + readonly loadingOlder: boolean; +} + export interface EnvironmentThreadState { readonly data: Option.Option; readonly status: EnvironmentThreadStatus; readonly error: Option.Option; + readonly page: Option.Option; } export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = { data: Option.none(), status: "empty", error: Option.none(), + page: Option.none(), }; + +/** Whether the thread has older turns that can be loaded with more pages. */ +export function threadHasOlderTurns(state: EnvironmentThreadState): boolean { + return Option.match(state.page, { + onNone: () => false, + onSome: (page) => page.hasMore, + }); +} diff --git a/packages/client-runtime/src/state/threads-pagination.test.ts b/packages/client-runtime/src/state/threads-pagination.test.ts new file mode 100644 index 00000000000..62cad18f89e --- /dev/null +++ b/packages/client-runtime/src/state/threads-pagination.test.ts @@ -0,0 +1,543 @@ +import { + EnvironmentId, + EventId, + ORCHESTRATION_WS_METHODS, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationMessage, + type OrchestrationThread, + type OrchestrationThreadDetailSnapshot, + type OrchestrationThreadStreamItem, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; + +import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; +import { + AVAILABLE_CONNECTION_STATE, + PrimaryConnectionTarget, + type PreparedConnection, + type SupervisorConnectionState, +} from "../connection/model.ts"; +import * as EnvironmentSupervisor from "../connection/supervisor.ts"; +import * as Persistence from "../platform/persistence.ts"; +import * as RpcSession from "../rpc/session.ts"; +import type { ThreadSnapshotWindow } from "./threadSnapshotHttp.ts"; +import { + INITIAL_THREAD_USER_TURN_LIMIT, + makeEnvironmentThreadState, + requestOlderThreadTurns, + ThreadSnapshotLoader, + type EnvironmentThreadState, +} from "./threads.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-1"), + label: "Test environment", + httpBaseUrl: "https://environment.example.test", + wsBaseUrl: "wss://environment.example.test", +}); +const THREAD_ID = ThreadId.make("thread-1"); +const PREPARED: PreparedConnection = { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, +}; + +function message(id: string, turnId: string, createdAt: string): OrchestrationMessage { + return { + id: id as OrchestrationMessage["id"], + role: "assistant", + text: `text of ${id}`, + turnId: TurnId.make(turnId), + streaming: false, + createdAt, + updatedAt: createdAt, + }; +} + +const OLDER_MESSAGE = message("message-old", "turn-1", "2026-04-01T00:00:00.000Z"); +const RECENT_MESSAGE = message("message-recent", "turn-2", "2026-04-01T01:00:00.000Z"); + +// Reverts retain turns via checkpoints with checkpointTurnCount <= the revert's +// turnCount, so both fixture turns carry one: reverting to turnCount 1 keeps +// turn-1 (the older page's turn) and discards turn-2 (the loaded window's). +function checkpoint(turnId: string, turnCount: number): OrchestrationThread["checkpoints"][number] { + return { + turnId: TurnId.make(turnId), + checkpointTurnCount: turnCount, + checkpointRef: + `checkpoint-${turnCount}` as OrchestrationThread["checkpoints"][number]["checkpointRef"], + status: "ready", + files: [], + assistantMessageId: null, + completedAt: "2026-04-01T01:00:00.000Z", + }; +} + +const BASE_THREAD: OrchestrationThread = { + id: THREAD_ID, + projectId: ProjectId.make("project-1"), + title: "Windowed thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: null, + latestTurn: null, + createdAt: "2026-04-01T00:00:00.000Z", + updatedAt: "2026-04-01T00:00:00.000Z", + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + messages: [RECENT_MESSAGE], + proposedPlans: [], + activities: [], + checkpoints: [checkpoint("turn-2", 2)], + session: null, +}; + +const WINDOWED_SNAPSHOT: OrchestrationThreadDetailSnapshot = { + snapshotSequence: 10, + thread: BASE_THREAD, + page: { beforeCursor: "cursor-1", hasMore: true, snapshotSequence: 10 }, +}; + +const OLDER_PAGE: OrchestrationThreadDetailSnapshot = { + snapshotSequence: 10, + thread: { + ...BASE_THREAD, + messages: [OLDER_MESSAGE], + checkpoints: [checkpoint("turn-1", 1)], + }, + page: { beforeCursor: null, hasMore: false, snapshotSequence: 10 }, +}; + +type LoaderResponse = Option.Option; + +const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (options?: { + readonly paginationCapability?: boolean; + readonly initialResponse?: LoaderResponse; + /** Cached snapshot returned by the cache store (simulates a warm cache). */ + readonly cached?: OrchestrationThreadDetailSnapshot; +}) { + const inputs = yield* Queue.unbounded(); + const observed = yield* Queue.unbounded(); + const loaderWindows = yield* Ref.make>([]); + const lastSubscribeInput = yield* Ref.make | undefined>(undefined); + const savedThreads = yield* Ref.make>([]); + // Older-page responses resolve through deferreds so tests can interleave + // live events with an in-flight page fetch. + const pendingPageResponses = yield* Queue.unbounded>(); + const supervisorState = yield* SubscriptionRef.make( + AVAILABLE_CONNECTION_STATE, + ); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: Record) => + Stream.unwrap(Ref.set(lastSubscribeInput, input).pipe(Effect.as(Stream.fromQueue(inputs)))), + } as unknown as WsRpcProtocolClient; + const session: RpcSession.RpcSession = { + client, + initialConfig: Effect.succeed({ + threadSnapshotPagination: options?.paginationCapability !== false, + } as never), + ready: Effect.void, + probe: Effect.void, + closed: Effect.never, + }; + const supervisorSession = yield* SubscriptionRef.make>( + Option.some(session), + ); + const prepared = yield* SubscriptionRef.make>( + Option.some(PREPARED), + ); + const snapshotLoader = ThreadSnapshotLoader.of({ + load: (_prepared, _threadId, window) => + Ref.update(loaderWindows, (current) => [...current, window]).pipe( + Effect.andThen( + window?.beforeCursor === undefined + ? Effect.succeed( + options?.initialResponse ?? Option.none(), + ) + : Deferred.make().pipe( + Effect.tap((deferred) => Queue.offer(pendingPageResponses, deferred)), + Effect.flatMap(Deferred.await), + ), + ), + ), + }); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: supervisorSession, + prepared, + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => + Effect.succeed(options?.cached !== undefined ? Option.some(options.cached) : Option.none()), + saveThread: (_environmentId, thread) => + Ref.update(savedThreads, (current) => [...current, thread]), + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ThreadSnapshotLoader, snapshotLoader), + ); + yield* SubscriptionRef.changes(threadState).pipe( + Stream.runForEach((state) => Queue.offer(observed, state)), + Effect.forkScoped, + ); + + const awaitState = (predicate: (state: EnvironmentThreadState) => boolean) => + Queue.take(observed).pipe(Effect.repeat({ until: predicate })); + const resolveNextPage = (response: LoaderResponse) => + Queue.take(pendingPageResponses).pipe( + Effect.flatMap((deferred) => Deferred.succeed(deferred, response)), + ); + + return { + inputs, + observed, + awaitState, + resolveNextPage, + loaderWindows, + lastSubscribeInput, + savedThreads, + threadState, + }; +}); + +const hasMessage = (state: EnvironmentThreadState, id: string): boolean => + Option.match(state.data, { + onNone: () => false, + onSome: (thread) => thread.messages.some((entry) => entry.id === id), + }); + +const titleEvent = (title: string, sequence: number): OrchestrationThreadStreamItem => ({ + kind: "event", + event: { + eventId: EventId.make(`event-title-${sequence}`), + sequence, + occurredAt: "2026-04-01T01:30:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.meta-updated", + payload: { + threadId: THREAD_ID, + title, + updatedAt: "2026-04-01T01:30:00.000Z", + }, + }, +}); + +// Reverting to turnCount 1 retains only turns whose checkpoint count is <= 1: +// turn-1 survives, turn-2 (the loaded window's newest turn) is discarded. +const revertEvent = (sequence: number): OrchestrationThreadStreamItem => ({ + kind: "event", + event: { + eventId: EventId.make(`event-revert-${sequence}`), + sequence, + occurredAt: "2026-04-01T02:00:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + aggregateKind: "thread", + aggregateId: THREAD_ID, + type: "thread.reverted", + payload: { + threadId: THREAD_ID, + turnCount: 1, + }, + }, +}); + +describe("thread pagination state", () => { + it.effect("windows the initial load when the server advertises pagination", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + const state = yield* harness.awaitState((value) => Option.isSome(value.page)); + expect(Option.getOrThrow(state.page)).toEqual({ + beforeCursor: "cursor-1", + hasMore: true, + loadingOlder: false, + }); + const windows = yield* Ref.get(harness.loaderWindows); + expect(windows[0]?.turnLimit).toBe(INITIAL_THREAD_USER_TURN_LIMIT); + const subscribeInput = yield* Ref.get(harness.lastSubscribeInput); + expect(subscribeInput?.turnLimit).toBe(INITIAL_THREAD_USER_TURN_LIMIT); + }), + ); + + it.effect("does not send a window to servers without the capability", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + paginationCapability: false, + initialResponse: Option.some({ snapshotSequence: 10, thread: BASE_THREAD }), + }); + const state = yield* harness.awaitState((value) => Option.isSome(value.data)); + expect(Option.isNone(state.page)).toBe(true); + const windows = yield* Ref.get(harness.loaderWindows); + expect(windows[0]).toBeUndefined(); + const subscribeInput = yield* Ref.get(harness.lastSubscribeInput); + expect(subscribeInput?.turnLimit).toBeUndefined(); + }), + ); + + it.effect("merges an older page below the loaded window and clears the cursor", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true); + yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }), + ); + yield* harness.resolveNextPage(Option.some(OLDER_PAGE)); + + const state = yield* harness.awaitState((value) => hasMessage(value, "message-old")); + const thread = Option.getOrThrow(state.data); + // Older rows land before the loaded window's rows. + expect(thread.messages.map((entry) => entry.id)).toEqual(["message-old", "message-recent"]); + expect(Option.getOrThrow(state.page)).toEqual({ + beforeCursor: null, + hasMore: false, + loadingOlder: false, + }); + }), + ); + + it.effect("discards an in-flight older page when a revert rewrites history", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }), + ); + // Revert lands while the page fetch is in flight and removes turn-2. + yield* Queue.offer(harness.inputs, revertEvent(11)); + yield* harness.awaitState((value) => !hasMessage(value, "message-recent")); + yield* harness.resolveNextPage(Option.some(OLDER_PAGE)); + + const state = yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => !page.loadingOlder }), + ); + // The stale page was dropped: no resurrected rows, cursor unchanged. + expect(hasMessage(state, "message-old")).toBe(false); + expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1"); + }), + ); + + it.effect("discards an in-flight older page when a fresh snapshot replaces the thread", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }), + ); + yield* Queue.offer(harness.inputs, { + kind: "snapshot", + snapshot: { + snapshotSequence: 20, + thread: { ...BASE_THREAD, title: "Replaced thread" }, + page: { beforeCursor: "cursor-2", hasMore: true, snapshotSequence: 20 }, + }, + }); + yield* harness.awaitState((value) => + Option.match(value.data, { + onNone: () => false, + onSome: (thread) => thread.title === "Replaced thread", + }), + ); + yield* harness.resolveNextPage(Option.some(OLDER_PAGE)); + + const state = yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => !page.loadingOlder }), + ); + expect(hasMessage(state, "message-old")).toBe(false); + // The replacement snapshot's cursor wins over the discarded page's. + expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-2"); + }), + ); + + it.effect("discards an older page read from a projection behind the loaded state", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }), + ); + yield* harness.resolveNextPage(Option.some({ ...OLDER_PAGE, snapshotSequence: 5 })); + + const state = yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => !page.loadingOlder }), + ); + expect(hasMessage(state, "message-old")).toBe(false); + expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1"); + }), + ); + + it.effect("a merged history page never advances the live-event dedupe sequence", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }), + ); + // The page was captured at a newer projection sequence (12) than the + // loaded state (10); merging it must not swallow events 11-12. + yield* harness.resolveNextPage( + Option.some({ + ...OLDER_PAGE, + snapshotSequence: 12, + page: { beforeCursor: null, hasMore: false, snapshotSequence: 12 }, + }), + ); + yield* harness.awaitState((value) => hasMessage(value, "message-old")); + + // Event at sequence 11 must still apply after the merge: the revert + // discards turn-2, so the loaded window's row disappears while the + // merged older turn-1 row survives. If the merge had advanced the + // dedupe sequence to the page's 12, this event would be swallowed. + yield* Queue.offer(harness.inputs, revertEvent(11)); + const state = yield* harness.awaitState( + (value) => !hasMessage(value, "message-recent") && hasMessage(value, "message-old"), + ); + expect(hasMessage(state, "message-old")).toBe(true); + }), + ); + + it.effect("parks a page read ahead of the live state until events catch up", () => + Effect.gen(function* () { + // A page whose thread watermark is ahead of the loaded state may + // contain streaming content the subscription has not delivered yet + // (e.g. an out-of-window subagent turn mid-stream); merging it + // immediately and then replaying those deltas would duplicate text. + // The page parks until the live state reaches the watermark. + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + requestOlderThreadTurns(TARGET.environmentId, THREAD_ID); + yield* harness.awaitState((value) => + Option.match(value.page, { onNone: () => false, onSome: (page) => page.loadingOlder }), + ); + // Page watermark 11 > loaded sequence 10: must park, not merge. + yield* harness.resolveNextPage( + Option.some({ + ...OLDER_PAGE, + snapshotSequence: 11, + page: { beforeCursor: null, hasMore: false, snapshotSequence: 11, threadSequence: 11 }, + }), + ); + + // A live event at sequence 11 arrives; only then does the page merge. + yield* Queue.offer(harness.inputs, titleEvent("Advanced past watermark", 11)); + const state = yield* harness.awaitState((value) => hasMessage(value, "message-old")); + expect(hasMessage(state, "message-recent")).toBe(true); + expect(Option.getOrThrow(state.page).loadingOlder).toBe(false); + }), + ); + + it.effect("a revert keeps the page cursor and triggers no refresh fetch", () => + Effect.gen(function* () { + // Cursors are an (anchor, turnId) keyset derived from event content, so + // they survive the revert projector's row rewrite: the machine keeps + // the stored cursor and performs no snapshot re-fetch. The revert + // reducer's turn filtering alone handles loaded history. + const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + yield* Queue.offer(harness.inputs, revertEvent(11)); + const state = yield* harness.awaitState((value) => !hasMessage(value, "message-recent")); + + expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1"); + const windows = yield* Ref.get(harness.loaderWindows); + // Only the initial load hit the loader — no post-revert refresh fetch. + expect(windows.length).toBe(1); + }), + ); + + it.effect("drops a windowed cache when the server lacks the pagination capability", () => + Effect.gen(function* () { + // Resuming a windowed cache via afterSequence against a pre-pagination + // server would render only the window forever with no way to load the + // rest: the machine must discard the cache and take a full snapshot. + const fullSnapshot: OrchestrationThreadDetailSnapshot = { + snapshotSequence: 20, + thread: { ...BASE_THREAD, title: "Full reload" }, + }; + const harness = yield* makeHarness({ + paginationCapability: false, + cached: WINDOWED_SNAPSHOT, + initialResponse: Option.some(fullSnapshot), + }); + + const state = yield* harness.awaitState((value) => + Option.match(value.data, { + onNone: () => false, + onSome: (thread) => thread.title === "Full reload", + }), + ); + expect(Option.isNone(state.page)).toBe(true); + // The subscription resumed from the fresh full snapshot, not the + // discarded windowed cache's watermark, and sent no window fields. + const subscribeInput = yield* Ref.get(harness.lastSubscribeInput); + expect(subscribeInput?.turnLimit).toBeUndefined(); + expect(subscribeInput?.afterSequence).toBe(20); + }), + ); + + it.effect("keeps a windowed cache when the server supports pagination", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: WINDOWED_SNAPSHOT }); + const state = yield* harness.awaitState((value) => Option.isSome(value.page)); + expect(Option.getOrThrow(state.page).beforeCursor).toBe("cursor-1"); + // Wait for the subscription (recorded when the WS method is invoked) + // before asserting its input. + const subscribeInput = yield* Ref.get(harness.lastSubscribeInput).pipe( + Effect.repeat({ until: (input) => input !== undefined }), + ); + expect(subscribeInput?.afterSequence).toBe(10); + }), + ); +}); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 06b5428ca58..4ba5a0e9df1 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -2,15 +2,18 @@ import { ORCHESTRATION_WS_METHODS, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, + type OrchestrationThreadDetailPage, type OrchestrationThreadDetailSnapshot, type OrchestrationThreadStreamItem, type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom } from "effect/unstable/reactivity"; @@ -21,13 +24,14 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribeDynamic } from "../rpc/client.ts"; -import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts"; +import { ThreadSnapshotLoader, type ThreadSnapshotWindow } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyThreadDetailEvent } from "./threadReducer.ts"; import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, + type EnvironmentThreadPageState, type EnvironmentThreadState, type EnvironmentThreadStatus, } from "./threadState.ts"; @@ -36,6 +40,85 @@ function statusWithoutLiveData(data: Option.Option): Enviro return Option.isSome(data) ? "cached" : "empty"; } +/** + * Turn window sizes for paginated thread loads: the initial page covers the + * last 10 user-anchored turns (subagent/fan-out turns ride along), each + * "load earlier" tap fetches 20 more. Sized so first paint on the heaviest + * observed threads stays around 100K gzipped while median threads load fully. + */ +export const INITIAL_THREAD_USER_TURN_LIMIT = 10; +export const OLDER_THREAD_PAGE_USER_TURN_LIMIT = 20; + +function pageStateFromSnapshot( + page: OrchestrationThreadDetailPage | undefined, +): Option.Option { + return page === undefined + ? Option.none() + : Option.some({ + beforeCursor: page.beforeCursor, + hasMore: page.hasMore, + loadingOlder: false, + }); +} + +interface ThreadOlderTurnRequestRegistry { + /** + * Registers the live state machine for a thread. Returns the deregistration + * cleanup; registration lives exactly as long as the machine's scope, and a + * successor machine for the same thread simply replaces the entry. + */ + readonly register: (key: string, handler: () => void) => () => void; + readonly request: (key: string) => boolean; +} + +function makeThreadOlderTurnRequestRegistry(): ThreadOlderTurnRequestRegistry { + const handlers = new Map void>(); + return { + register: (key, handler) => { + handlers.set(key, handler); + return () => { + if (handlers.get(key) === handler) { + handlers.delete(key); + } + }; + }, + request: (key) => { + const handler = handlers.get(key); + if (handler === undefined) { + return false; + } + handler(); + return true; + }, + }; +} + +const defaultOlderTurnRequestRegistry = makeThreadOlderTurnRequestRegistry(); + +/** + * Channel from UI actions to the live per-thread state machines. The machines + * resolve it from the Effect environment (overridable in tests); the default + * instance is shared with the sync `requestOlderThreadTurns` entry point so + * the apps get working wiring without providing anything. + */ +export class ThreadOlderTurnRequests extends Context.Reference( + "@t3tools/client-runtime/state/threads/ThreadOlderTurnRequests", + { defaultValue: () => defaultOlderTurnRequestRegistry }, +) {} + +/** + * Asks the live state machine for `threadId` to fetch the next older page. + * Returns false when no machine is live or no fetch was started (no cursor, + * already loading); callers render from `EnvironmentThreadState.page` and can + * treat false as "nothing to do". + */ +export function requestOlderThreadTurns( + environmentId: EnvironmentIdType, + threadId: ThreadIdType, +): boolean { + return defaultOlderTurnRequestRegistry.request(threadKey({ environmentId, threadId })); +} + function formatThreadError(cause: Cause.Cause): string { const error = Cause.squash(cause); return error instanceof Error && error.message.trim().length > 0 @@ -73,6 +156,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make data: cachedThread, status: statusWithoutLiveData(cachedThread), error: Option.none(), + // A cached windowed snapshot restores its page cursor so "load earlier" + // works while rendering from cache; a cached full snapshot has no page. + page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)), }); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. @@ -80,6 +166,25 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }), ); const awaitingCompletion = yield* Ref.make(false); + // Bumped whenever loaded history may have been rewritten out from under an + // in-flight older-page fetch (snapshot replacement, revert, deletion). A + // page response captured under an older epoch is discarded, not merged. + const historyEpoch = yield* Ref.make(0); + // Serializes stream-item application against older-page staleness checks + + // merges. Without it, a revert or snapshot processed between loadOlderTurns' + // epoch check and its merge could still slip resurrected history in. + const applyLock = yield* Semaphore.make(1); + // Whether the connected server accepts windowed reads; set per subscription + // from the session config. Gates loadOlderTurns so a reconnect to a + // pre-pagination server never sends unsupported window parameters. + const paginationSupported = yield* Ref.make(false); + // An older page whose thread watermark is ahead of the live state, parked + // until the subscription catches up (see mergeOlderPage's caller). At most + // one can exist because loadOlderTurns no-ops while loadingOlder is true. + const pendingOlderPage = yield* Ref.make<{ + readonly snapshot: OrchestrationThreadDetailSnapshot; + readonly epoch: number; + } | null>(null); const persistence = yield* Queue.sliding(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( @@ -124,6 +229,12 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); const setDisconnected = Effect.gen(function* () { yield* Ref.set(awaitingCompletion, false); + // The capability belongs to the session that advertised it. During a + // reconnect, a new prepared connection can exist before the new session's + // config arrives; leaving the old value would let loadOlderTurns send + // window parameters to a server that may not accept them (review + // finding). makeSubscribeInput re-sets it from the next session's config. + yield* Ref.set(paginationSupported, false); yield* SubscriptionRef.update(state, (current) => ({ ...current, status: current.status === "deleted" ? current.status : statusWithoutLiveData(current.data), @@ -143,28 +254,51 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationThread, + // "keep" preserves the current page state (live events touch only loaded + // recent turns); a snapshot or merged page passes its own page state. + page: Option.Option | "keep", ) { const waiting = yield* Ref.get(awaitingCompletion); - yield* SubscriptionRef.set(state, { + yield* SubscriptionRef.update(state, (current) => ({ data: Option.some(thread), - status: waiting ? "synchronizing" : "live", + status: waiting ? ("synchronizing" as const) : ("live" as const), error: Option.none(), - }); + page: page === "keep" ? current.page : page, + })); // Active threads can update many times per second and retain large tool // payloads. The server remains the source of truth while a turn is active; // persist once it settles so cache encoding stays off the streaming path. if (shouldPersistThread(thread)) { const snapshotSequence = yield* SubscriptionRef.get(lastSequence); - yield* Queue.offer(persistence, { snapshotSequence, thread }); + const currentPage = yield* SubscriptionRef.get(state).pipe(Effect.map((value) => value.page)); + yield* Queue.offer(persistence, { + snapshotSequence, + thread, + // Persist the window boundary with the window's content so a cache + // restore can keep paging from where the loaded history ends. + ...Option.match(currentPage, { + onNone: () => ({}), + onSome: (value) => + ({ + page: { + beforeCursor: value.beforeCursor, + hasMore: value.hasMore, + snapshotSequence, + }, + }) as const, + }), + }); } }); const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { yield* Ref.set(awaitingCompletion, false); + yield* Ref.update(historyEpoch, (epoch) => epoch + 1); yield* SubscriptionRef.set(state, { data: Option.none(), status: "deleted", error: Option.none(), + page: Option.none(), }); yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => @@ -179,7 +313,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); }); - const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( + // Body of applyItem, running under applyLock. + const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* ( item: OrchestrationThreadStreamItem, ) { if (item.kind === "synchronized") { @@ -193,8 +328,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } if (item.kind === "snapshot") { + // A fresh snapshot replaces all loaded history, including older + // pages: a turn reverted while disconnected would otherwise survive + // in the preserved history with no event left to remove it. The + // epoch bump discards any older-page fetch racing this snapshot. + yield* Ref.update(historyEpoch, (epoch) => epoch + 1); yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence); - yield* setThread(item.snapshot.thread); + yield* setThread(item.snapshot.thread, pageStateFromSnapshot(item.snapshot.page)); return; } @@ -211,12 +351,184 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } return; } + if (item.event.type === "thread.reverted") { + // A revert rewrites loaded history (whole turns disappear), so an + // older-page fetch in flight may straddle the removed range; the epoch + // bump discards it. The stored page cursor stays valid: cursors are an + // (anchor, turnId) keyset derived from event content, which survives + // the revert projector's row rewrite, so no refresh is needed — the + // revert reducer's turn filtering fully handles loaded history. + yield* Ref.update(historyEpoch, (epoch) => epoch + 1); + } const result = applyThreadDetailEvent(current.data.value, item.event); if (result.kind === "updated") { - yield* setThread(result.thread); + yield* setThread(result.thread, "keep"); } else if (result.kind === "deleted") { yield* setDeleted(); } + // The event may have advanced the live state past a parked page's + // watermark; merge it as soon as that happens. + yield* tryMergePendingOlderPage(); + }); + + // Merges a parked older page once the live state has caught up to the + // page's thread watermark, or discards it if history was rewritten + // (epoch advanced) while it waited. Must run under applyLock. + const tryMergePendingOlderPage = Effect.fn("EnvironmentThreadState.tryMergePendingOlderPage")( + function* () { + const pending = yield* Ref.get(pendingOlderPage); + if (pending === null) { + return; + } + const epochNow = yield* Ref.get(historyEpoch); + if (epochNow !== pending.epoch) { + yield* Ref.set(pendingOlderPage, null); + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + page: Option.map(value.page, (existing) => ({ ...existing, loadingOlder: false })), + })); + return; + } + const watermark = pending.snapshot.page?.threadSequence; + const loadedSequence = yield* SubscriptionRef.get(lastSequence); + if (watermark !== undefined && watermark > loadedSequence) { + return; + } + yield* Ref.set(pendingOlderPage, null); + yield* mergeOlderPage(pending.snapshot); + }, + ); + + const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( + item: OrchestrationThreadStreamItem, + ) { + yield* applyLock.withPermits(1)(applyItemLocked(item)); + }); + + // Merges an older disjoint page below the currently loaded window. All four + // windowed collections prepend; identity dedupe guards the (server-bug or + // cursor-misuse) case of overlapping pages so a row never renders twice. + const mergeOlderPage = Effect.fn("EnvironmentThreadState.mergeOlderPage")(function* ( + snapshot: OrchestrationThreadDetailSnapshot, + ) { + // The merge is built inside the update callback so it composes with + // whatever thread value is current at commit time. The applyLock already + // serializes this against event application; the atomic build is defense + // in depth against future callers outside the lock. + let merged: OrchestrationThread | null = null; + yield* SubscriptionRef.update(state, (value) => { + if (Option.isNone(value.data)) { + return value; + } + const loaded = value.data.value; + const older = snapshot.thread; + const mergeById = ( + olderRows: ReadonlyArray, + loadedRows: ReadonlyArray, + ): ReadonlyArray => { + const seen = new Set(loadedRows.map((row) => row.id)); + return [...olderRows.filter((row) => !seen.has(row.id)), ...loadedRows]; + }; + const seenCheckpoints = new Set(loaded.checkpoints.map((row) => row.turnId)); + merged = { + // Thread metadata stays the loaded (newer) snapshot's; only the + // windowed collections gain rows from the older page. + ...loaded, + messages: mergeById(older.messages, loaded.messages), + activities: mergeById(older.activities, loaded.activities), + proposedPlans: mergeById(older.proposedPlans, loaded.proposedPlans), + checkpoints: [ + ...older.checkpoints.filter((row) => !seenCheckpoints.has(row.turnId)), + ...loaded.checkpoints, + ], + }; + return { + ...value, + data: Option.some(merged), + page: pageStateFromSnapshot(snapshot.page), + }; + }); + // Persist the widened window under the *loaded* watermark: the merged + // content is only known consistent with the state it merged into, not + // with the page's own (possibly newer) sequence. + if (merged !== null && shouldPersistThread(merged)) { + const snapshotSequence = yield* SubscriptionRef.get(lastSequence); + yield* Queue.offer(persistence, { + snapshotSequence, + thread: merged, + ...(snapshot.page === undefined ? {} : { page: { ...snapshot.page, snapshotSequence } }), + }); + } + }); + + const loadOlderTurns = Effect.fn("EnvironmentThreadState.loadOlderTurns")(function* () { + // Gated on the connected server's capability: a reconnect to a + // pre-pagination server must never receive window parameters. + if (!(yield* Ref.get(paginationSupported))) { + return; + } + const current = yield* SubscriptionRef.get(state); + const page = Option.getOrNull(current.page); + if (page === null || page.loadingOlder || !page.hasMore || page.beforeCursor === null) { + return; + } + const prepared = Option.getOrNull(yield* SubscriptionRef.get(supervisor.prepared)); + if (prepared === null) { + return; + } + const epochAtStart = yield* Ref.get(historyEpoch); + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + page: Option.map(value.page, (existing) => ({ ...existing, loadingOlder: true })), + })); + const window: ThreadSnapshotWindow = { + turnLimit: OLDER_THREAD_PAGE_USER_TURN_LIMIT, + beforeCursor: page.beforeCursor, + }; + const response = yield* snapshotLoader.load(prepared, threadId, window); + // Staleness check and merge run under the same lock as stream-item + // application, so a revert/snapshot cannot land between them (TOCTOU + // review finding) — anything that rewrites history bumps the epoch + // before this permit is acquired. + yield* applyLock.withPermits(1)( + Effect.gen(function* () { + const epochNow = yield* Ref.get(historyEpoch); + const loadedSequence = yield* SubscriptionRef.get(lastSequence); + // A page carrying a sequence older than the loaded state was read + // from a projection behind what we render; merging it could + // resurrect turns a newer snapshot or revert already removed. + const stale = + epochNow !== epochAtStart || + Option.match(response, { + onNone: () => false, + onSome: (snapshot) => snapshot.snapshotSequence < loadedSequence, + }); + if (Option.isNone(response) || stale) { + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + page: Option.map(value.page, (existing) => ({ ...existing, loadingOlder: false })), + })); + return; + } + // A page read AHEAD of the live state may include content (e.g. + // streaming deltas of an out-of-window turn) the subscription has + // not delivered yet; merging now and then replaying those events + // would duplicate them. Park the page until the live state reaches + // the page's thread-scoped watermark; loadingOlder stays true so + // the UI shows progress and no second fetch starts. Pages from + // pre-watermark servers (threadSequence absent) merge immediately, + // preserving the old behavior. + const watermark = response.value.page?.threadSequence; + if (watermark !== undefined && watermark > loadedSequence) { + yield* Ref.set(pendingOlderPage, { + snapshot: response.value, + epoch: epochNow, + }); + return; + } + yield* mergeOlderPage(response.value); + }), + ); }); yield* SubscriptionRef.changes(supervisor.state).pipe( @@ -244,14 +556,40 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make subscribeDynamic( ORCHESTRATION_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { - const supportsCompletionMarker = yield* session.initialConfig.pipe( - Effect.map((config) => config.threadResumeCompletionMarker === true), - Effect.orElseSucceed(() => false), + const config = yield* session.initialConfig.pipe( + Effect.orElseSucceed( + () => + ({}) as { + threadResumeCompletionMarker?: boolean; + threadSnapshotPagination?: boolean; + }, + ), ); + const supportsCompletionMarker = config.threadResumeCompletionMarker === true; + // Windowed loads are gated on the server capability: pre-pagination + // servers reject unknown query params, and a windowed WS fallback to + // such a server would silently hide history. + const supportsPagination = config.threadSnapshotPagination === true; + yield* Ref.set(paginationSupported, supportsPagination); yield* Ref.set(awaitingCompletion, supportsCompletionMarker); yield* setSynchronizing; let current = yield* SubscriptionRef.get(state); + // A windowed cache resuming against a server without pagination is a + // trap: afterSequence resume keeps only the window, and the missing + // older turns can never be loaded (the server has no cursor reads). + // Drop the window marker and treat the data as needing a full reload. + if (!supportsPagination && Option.isSome(current.page)) { + yield* Ref.update(historyEpoch, (epoch) => epoch + 1); + yield* SubscriptionRef.update(state, (value) => ({ + ...value, + data: Option.none(), + status: value.status === "deleted" ? value.status : ("empty" as const), + page: Option.none(), + })); + yield* SubscriptionRef.set(lastSequence, 0); + current = yield* SubscriptionRef.get(state); + } if (Option.isNone(current.data) && current.status !== "deleted") { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( @@ -267,7 +605,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }), ), ); - const httpSnapshot = yield* snapshotLoader.load(prepared, threadId); + const httpSnapshot = yield* snapshotLoader.load( + prepared, + threadId, + supportsPagination ? { turnLimit: INITIAL_THREAD_USER_TURN_LIMIT } : undefined, + ); if (Option.isSome(httpSnapshot)) { yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); current = yield* SubscriptionRef.get(state); @@ -288,6 +630,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make threadId, ...(canResume ? { afterSequence: sequence } : {}), ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + // The WS fallback snapshot (sent when afterSequence is missing or + // the gap is too large) should be windowed the same as the HTTP + // path; without this a resume failure re-downloads the full thread. + ...(supportsPagination ? { turnLimit: INITIAL_THREAD_USER_TURN_LIMIT } : {}), }; }), { @@ -298,13 +644,47 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ).pipe(Stream.runForEach(applyItem)), ); + // Expose loadOlderTurns to UI actions through the request registry. + // Requests funnel through a sliding queue drained serially, so mashing + // "load earlier" coalesces (loadOlderTurns itself no-ops while a fetch is + // in flight). + const olderTurnRequestRegistry = yield* ThreadOlderTurnRequests; + const olderTurnRequests = yield* Queue.sliding(1); + yield* Stream.fromQueue(olderTurnRequests).pipe( + Stream.runForEach(() => loadOlderTurns()), + Effect.forkScoped, + ); + const deregister = olderTurnRequestRegistry.register( + threadKey({ environmentId, threadId }), + () => { + Queue.offerUnsafe(olderTurnRequests, undefined); + }, + ); + yield* Effect.addFinalizer(() => Effect.sync(deregister)); + yield* Effect.addFinalizer(() => Effect.all([SubscriptionRef.get(state), SubscriptionRef.get(lastSequence)]).pipe( Effect.flatMap(([current, snapshotSequence]) => Option.match(current.data, { onNone: () => Effect.void, onSome: (thread) => - shouldPersistThread(thread) ? persist({ snapshotSequence, thread }) : Effect.void, + shouldPersistThread(thread) + ? persist({ + snapshotSequence, + thread, + ...Option.match(current.page, { + onNone: () => ({}), + onSome: (page) => + ({ + page: { + beforeCursor: page.beforeCursor, + hasMore: page.hasMore, + snapshotSequence, + }, + }) as const, + }), + }) + : Effect.void, }), ), ), diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 2d40dad60cc..f385a2eff2c 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -457,6 +457,16 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({ threadId: ThreadId, }); +// Query-string window for windowed thread snapshots (GET payloads must encode +// to strings). Both fields optional: omitting them keeps the full-snapshot +// behavior, so pagination stays opt-in per request. +const EnvironmentOrchestrationThreadSnapshotQuery = { + turnLimit: Schema.optional( + Schema.FiniteFromString.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(1)), + ), + beforeCursor: Schema.optional(TrimmedNonEmptyString), +}; + export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration") .add( HttpApiEndpoint.get("snapshot", "/api/orchestration/snapshot", { @@ -476,6 +486,7 @@ export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestr HttpApiEndpoint.get("threadSnapshot", "/api/orchestration/threads/:threadId", { headers: OptionalBearerHeaders, params: EnvironmentOrchestrationThreadSnapshotParams, + payload: EnvironmentOrchestrationThreadSnapshotQuery, success: OrchestrationThreadDetailSnapshot, error: EnvironmentOrchestrationThreadSnapshotErrors, }).middleware(EnvironmentAuthenticatedAuth), diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index c9baa6ac670..7ccb3dc7cac 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -13,6 +13,7 @@ import { IsoDateTime, MessageId, NonNegativeInt, + PositiveInt, ProjectId, ProviderItemId, ThreadId, @@ -525,12 +526,62 @@ export const OrchestrationSubscribeThreadInput = Schema.Struct({ * snapshot or catch-up replay and before it begins emitting live events. */ requestCompletionMarker: Schema.optionalKey(Schema.Boolean), + /** + * When provided, the fallback snapshot frame (sent when `afterSequence` is + * missing or the catch-up gap is too large) is windowed to the last + * `turnLimit` user-anchored turns and carries `page` metadata. Absent means + * the fallback snapshot is the full thread, preserving pre-pagination client + * behavior. Live events are unaffected either way. + */ + turnLimit: Schema.optionalKey(PositiveInt), }); export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type; +/** + * Bounds a thread detail read to a window of recent turns. `turnLimit` counts + * turns with a user pending message (subagent/fan-out turns between them ride + * along), so the window always contains the last N user prompts. `beforeCursor` + * requests the disjoint page of older turns strictly before a previously + * returned cursor. Requests without a window get the full thread; pagination is + * strictly opt-in so older clients keep today's behavior on both HTTP and the + * WebSocket fallback snapshot. + */ +export const OrchestrationThreadDetailWindow = Schema.Struct({ + turnLimit: Schema.optionalKey(PositiveInt), + beforeCursor: Schema.optionalKey(TrimmedNonEmptyString), +}); +export type OrchestrationThreadDetailWindow = typeof OrchestrationThreadDetailWindow.Type; + +/** + * Page metadata for a windowed thread detail read. `beforeCursor` is opaque and + * exclusive: passing it back returns the adjacent disjoint slice of older + * turns. `null` means the thread is fully loaded below this page. The + * `snapshotSequence` mirrors the top-level snapshot sequence so history pages + * can be sequence-checked against live state before merging. + */ +export const OrchestrationThreadDetailPage = Schema.Struct({ + beforeCursor: Schema.NullOr(TrimmedNonEmptyString), + hasMore: Schema.Boolean, + snapshotSequence: NonNegativeInt, + /** + * Highest event sequence applied to THIS thread at page read time. The + * global `snapshotSequence` advances with every thread's events, so a + * client cannot wait for it via its per-thread subscription; this + * thread-scoped watermark is reachable. A client merging an older page + * must first have applied live events up to it — otherwise a streaming + * turn outside the loaded window could have deltas replayed on top of + * page content that already includes them, duplicating text. + */ + threadSequence: Schema.optionalKey(NonNegativeInt), +}); +export type OrchestrationThreadDetailPage = typeof OrchestrationThreadDetailPage.Type; + export const OrchestrationThreadDetailSnapshot = Schema.Struct({ snapshotSequence: NonNegativeInt, thread: OrchestrationThread, + // Present only on windowed responses. Absent on full snapshots (and from + // pre-pagination servers), which clients treat as fully loaded. + page: Schema.optional(OrchestrationThreadDetailPage), }); export type OrchestrationThreadDetailSnapshot = typeof OrchestrationThreadDetailSnapshot.Type; diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index 20b40dffa75..d7bc4c5c189 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -434,6 +434,12 @@ export const ServerConfig = Schema.Struct({ shellResumeCompletionMarker: Schema.optionalKey(Schema.Boolean), /** Whether thread subscriptions can emit an opt-in catch-up completion marker. */ threadResumeCompletionMarker: Schema.optionalKey(Schema.Boolean), + /** + * Whether thread detail reads accept a turn window (`turnLimit`/ + * `beforeCursor`) and return `page` metadata. Clients must not send window + * fields to servers that don't advertise this. + */ + threadSnapshotPagination: Schema.optionalKey(Schema.Boolean), }); export type ServerConfig = typeof ServerConfig.Type;