From 331c6dce7f6745863752b1b423fc76d24014deec Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 20:13:10 -0400 Subject: [PATCH 1/8] fix(server): skip origin fetch when creating worktrees in repos without an origin remote (#5556) Co-authored-by: Claude Fable 5 --- apps/server/src/git/GitWorkflowService.ts | 8 ++ apps/server/src/server.test.ts | 113 ++++++++++++++++++++++ apps/server/src/vcs/GitVcsDriver.ts | 6 ++ apps/server/src/vcs/GitVcsDriverCore.ts | 8 +- apps/server/src/ws.ts | 11 ++- 5 files changed, 143 insertions(+), 3 deletions(-) diff --git a/apps/server/src/git/GitWorkflowService.ts b/apps/server/src/git/GitWorkflowService.ts index 100b9beadba..da22794951f 100644 --- a/apps/server/src/git/GitWorkflowService.ts +++ b/apps/server/src/git/GitWorkflowService.ts @@ -69,6 +69,10 @@ export class GitWorkflowService extends Context.Service< readonly cwd: string; readonly remoteName: string; }) => Effect.Effect; + readonly remoteExists: (input: { + readonly cwd: string; + readonly remoteName: string; + }) => Effect.Effect; readonly resolveRemoteTrackingCommit: (input: { readonly cwd: string; readonly refName: string; @@ -303,6 +307,10 @@ export const make = Effect.gen(function* () { ensureGitCommand("GitWorkflowService.fetchRemote", input.cwd).pipe( Effect.andThen(git.fetchRemote(input)), ), + remoteExists: (input) => + ensureGitCommand("GitWorkflowService.remoteExists", input.cwd).pipe( + Effect.andThen(git.remoteExists(input)), + ), resolveRemoteTrackingCommit: (input) => ensureGitCommand("GitWorkflowService.resolveRemoteTrackingCommit", input.cwd).pipe( Effect.andThen(git.resolveRemoteTrackingCommit(input)), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 8628aeef314..a403e228b06 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7134,6 +7134,13 @@ it.layer(NodeServices.layer)("server router seam", (it) => { pr: null, }), ); + const remoteExists = vi.fn( + (_: Parameters[0]) => + Effect.sync(() => { + bootstrapGitOperations.push("remote-exists"); + return true; + }), + ); const fetchRemote = vi.fn( (_: Parameters[0]) => Effect.sync(() => { @@ -7181,6 +7188,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { yield* buildAppUnderTest({ layers: { gitVcsDriver: { + remoteExists, fetchRemote, resolveRemoteTrackingCommit, createWorktree, @@ -7271,6 +7279,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { fallbackRemoteName: "origin", }); assert.deepEqual(bootstrapGitOperations, [ + "remote-exists", "fetch", "resolve-remote-commit", "create-worktree", @@ -7299,6 +7308,110 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect( + "falls back to the local base branch when startFromOrigin is set but no origin remote exists", + () => + Effect.gen(function* () { + const dispatchedCommands: Array = []; + const remoteExists = vi.fn( + (_: Parameters[0]) => + Effect.succeed(false), + ); + const fetchRemote = vi.fn( + (_: Parameters[0]) => Effect.void, + ); + const resolveRemoteTrackingCommit = vi.fn( + (_: Parameters[0]) => + Effect.succeed({ + commitSha: "0123456789abcdef0123456789abcdef01234567", + remoteRefName: "origin/main", + }), + ); + const createWorktree = vi.fn( + (_: Parameters[0]) => + Effect.succeed({ + worktree: { + refName: "t3code/bootstrap-refName", + path: "/tmp/bootstrap-worktree", + }, + }), + ); + + yield* buildAppUnderTest({ + layers: { + gitVcsDriver: { + remoteExists, + fetchRemote, + resolveRemoteTrackingCommit, + createWorktree, + }, + orchestrationEngine: { + dispatch: (command) => + Effect.sync(() => { + dispatchedCommands.push(command); + return { sequence: dispatchedCommands.length }; + }), + readEvents: () => Stream.empty, + }, + }, + }); + + const createdAt = "2026-01-01T00:00:00.000Z"; + const wsUrl = yield* getWsServerUrl("/ws"); + yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ + type: "thread.turn.start", + commandId: CommandId.make("cmd-bootstrap-turn-start-no-origin"), + threadId: ThreadId.make("thread-bootstrap-no-origin"), + message: { + messageId: MessageId.make("msg-bootstrap-no-origin"), + role: "user", + text: "hello", + attachments: [], + }, + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + bootstrap: { + createThread: { + projectId: defaultProjectId, + title: "Bootstrap Thread", + modelSelection: defaultModelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: null, + createdAt, + }, + prepareWorktree: { + projectCwd: "/tmp/project", + baseBranch: "main", + branch: "t3code/bootstrap-refName", + startFromOrigin: true, + }, + }, + createdAt, + }), + ), + ); + + assert.deepEqual(remoteExists.mock.calls[0]?.[0], { + cwd: "/tmp/project", + remoteName: "origin", + }); + assert.equal(fetchRemote.mock.calls.length, 0); + assert.equal(resolveRemoteTrackingCommit.mock.calls.length, 0); + assert.deepEqual(createWorktree.mock.calls[0]?.[0], { + cwd: "/tmp/project", + refName: "main", + newRefName: "t3code/bootstrap-refName", + baseRefName: "main", + path: null, + }); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("records setup-script failures without aborting bootstrap turn start", () => Effect.gen(function* () { const dispatchedCommands: Array = []; diff --git a/apps/server/src/vcs/GitVcsDriver.ts b/apps/server/src/vcs/GitVcsDriver.ts index 192efe5a7d0..f256a7dd4e1 100644 --- a/apps/server/src/vcs/GitVcsDriver.ts +++ b/apps/server/src/vcs/GitVcsDriver.ts @@ -168,6 +168,11 @@ export interface GitFetchRemoteInput { remoteName: string; } +export interface GitRemoteExistsInput { + cwd: string; + remoteName: string; +} + export interface GitResolveRemoteTrackingCommitInput { cwd: string; refName: string; @@ -243,6 +248,7 @@ export class GitVcsDriver extends Context.Service< readonly ensureRemote: (input: GitEnsureRemoteInput) => Effect.Effect; readonly resolvePrimaryRemoteName: (cwd: string) => Effect.Effect; readonly fetchRemote: (input: GitFetchRemoteInput) => Effect.Effect; + readonly remoteExists: (input: GitRemoteExistsInput) => Effect.Effect; readonly resolveRemoteTrackingCommit: ( input: GitResolveRemoteTrackingCommitInput, ) => Effect.Effect; diff --git a/apps/server/src/vcs/GitVcsDriverCore.ts b/apps/server/src/vcs/GitVcsDriverCore.ts index abcb10a8c9a..d39817c0ee1 100644 --- a/apps/server/src/vcs/GitVcsDriverCore.ts +++ b/apps/server/src/vcs/GitVcsDriverCore.ts @@ -1284,11 +1284,14 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* }, ).pipe(Effect.map((result) => result.exitCode === 0)); - const originRemoteExists = (cwd: string): Effect.Effect => - executeGit("GitVcsDriver.originRemoteExists", cwd, ["remote", "get-url", "origin"], { + const remoteExists: GitVcsDriver.GitVcsDriver["Service"]["remoteExists"] = (input) => + executeGit("GitVcsDriver.remoteExists", input.cwd, ["remote", "get-url", input.remoteName], { allowNonZeroExit: true, }).pipe(Effect.map((result) => result.exitCode === 0)); + const originRemoteExists = (cwd: string): Effect.Effect => + remoteExists({ cwd, remoteName: "origin" }); + const listRemoteNames = (cwd: string): Effect.Effect, GitCommandError> => runGitStdout("GitVcsDriver.listRemoteNames", cwd, ["remote"]).pipe( Effect.map(parseRemoteNamesInGitOrder), @@ -3071,6 +3074,7 @@ export const makeGitVcsDriverCore = Effect.fn("makeGitVcsDriverCore")(function* ensureRemote: (input) => withListRefsInvalidation(input.cwd, ensureRemote(input)), resolvePrimaryRemoteName, fetchRemote: (input) => withListRefsInvalidation(input.cwd, fetchRemote(input)), + remoteExists, resolveRemoteTrackingCommit, fetchRemoteBranch: (input) => withListRefsInvalidation(input.cwd, fetchRemoteBranch(input)), fetchRemoteTrackingBranch: (input) => diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index fc65602679a..a04fce3fd2c 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -908,7 +908,16 @@ const makeWsRpcLayer = ( if (bootstrap?.prepareWorktree) { let worktreeBaseRef = bootstrap.prepareWorktree.baseBranch; - if (bootstrap.prepareWorktree.startFromOrigin) { + // "Start from origin" is a stored default; repos without an + // origin remote fall back to the local base branch instead of + // failing the whole bootstrap on `git fetch origin`. + const startFromOrigin = + bootstrap.prepareWorktree.startFromOrigin === true && + (yield* gitWorkflow.remoteExists({ + cwd: bootstrap.prepareWorktree.projectCwd, + remoteName: "origin", + })); + if (startFromOrigin) { yield* gitWorkflow.fetchRemote({ cwd: bootstrap.prepareWorktree.projectCwd, remoteName: "origin", From ea50b695a749d6a0d44ef96b479b6dfceb8881e3 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 20:31:12 -0400 Subject: [PATCH 2/8] fix(web): update tooltip no longer dismisses when scrolling release notes (#5547) Co-authored-by: Claude Fable 5 --- apps/web/src/components/sidebar/SidebarUpdatePill.tsx | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/web/src/components/sidebar/SidebarUpdatePill.tsx b/apps/web/src/components/sidebar/SidebarUpdatePill.tsx index 06a0e714a6e..89120850f20 100644 --- a/apps/web/src/components/sidebar/SidebarUpdatePill.tsx +++ b/apps/web/src/components/sidebar/SidebarUpdatePill.tsx @@ -44,7 +44,7 @@ function SidebarUpdateReleaseNotesTooltip({
{tooltip}
-
+
{state.releaseNotes.map((releaseNote, index) => (
{index > 0 && } @@ -203,7 +203,9 @@ export function SidebarUpdatePill() { align="start" className={ state?.channel === "nightly" && state.releaseNotes.length > 0 - ? "max-w-none text-balance" + ? // pointer-events-auto overrides the positioner's pointer-events-none so the + // release notes stay open (and scrollable) when the cursor moves into them. + "pointer-events-auto max-w-none text-balance" : undefined } side="top" From 0ec4fbc4a376cbf6465e59b05506ce9d89b3d078 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 20:52:29 -0400 Subject: [PATCH 3/8] fix(server): stop showing commit/push/PR notices as errors in the work log (#5559) Co-authored-by: Claude Fable 5 --- .../src/provider/Layers/ClaudeAdapter.test.ts | 18 ++++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.ts | 12 +++++++++--- 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 8697505ef24..24b8429a391 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -2021,6 +2021,24 @@ describe("ClaudeAdapterLive", () => { session_id: "session", uuid: "roster", }, + { + type: "system", + subtype: "vcs_state_changed", + kind: "push", + cwd: "/tmp/worktree", + session_id: "session", + uuid: "vcs", + }, + { + type: "system", + subtype: "code_change_published", + provider: "github", + url: "https://github.com/pingdotgg/t3code/pull/1", + repo: "pingdotgg/t3code", + identifier: "1", + session_id: "session", + uuid: "ccp", + }, { type: "system", subtype: "task_updated", diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 812a7310928..27acedc383a 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -3034,9 +3034,15 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( // error rows in client work logs. `background_tasks_changed` is a roster // snapshot ({tasks: [...]}) — the task_* lifecycle events carry the // authoritative per-agent data and the typed background_tasks control - // request is the reconciliation source. - if ((message.subtype as string) === "background_tasks_changed") { - return; + // request is the reconciliation source. `vcs_state_changed` + // ({kind: commit|push|rebase}) and `code_change_published` + // ({provider, url, repo}) are informational CLI notices; the work log + // already shows the underlying git/gh tool calls. + switch (message.subtype as string) { + case "background_tasks_changed": + case "vcs_state_changed": + case "code_change_published": + return; } switch (message.subtype) { From 64a991ad455234e6fdd808a5a3caadc51f18a152 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 20:55:00 -0400 Subject: [PATCH 4/8] fix(web): show remote environment for non-Git projects (#5555) --- .../components/BranchToolbar.logic.test.ts | 33 +++++++++++ .../web/src/components/BranchToolbar.logic.ts | 8 +++ apps/web/src/components/BranchToolbar.tsx | 56 +++++++++++-------- apps/web/src/components/ChatView.tsx | 22 +++++++- 4 files changed, 93 insertions(+), 26 deletions(-) diff --git a/apps/web/src/components/BranchToolbar.logic.test.ts b/apps/web/src/components/BranchToolbar.logic.test.ts index 76336f1ef1f..36d42a60fa8 100644 --- a/apps/web/src/components/BranchToolbar.logic.test.ts +++ b/apps/web/src/components/BranchToolbar.logic.test.ts @@ -17,6 +17,7 @@ import { resolvePreviousWorktreeLabel, resolvePreviousWorktreeSeed, shouldIncludeBranchPickerItem, + shouldShowComposerContextStrip, shouldShowEnvironmentIndicator, } from "./BranchToolbar.logic"; @@ -421,6 +422,38 @@ describe("shouldShowEnvironmentIndicator", () => { }); }); +describe("shouldShowComposerContextStrip", () => { + it("keeps the environment indicator visible for a non-Git project", () => { + expect( + shouldShowComposerContextStrip({ + hasActiveProject: true, + isGitRepo: false, + showEnvironmentIndicator: true, + }), + ).toBe(true); + }); + + it("hides the strip when a non-Git project has no environment indicator", () => { + expect( + shouldShowComposerContextStrip({ + hasActiveProject: true, + isGitRepo: false, + showEnvironmentIndicator: false, + }), + ).toBe(false); + }); + + it("shows Git controls without requiring an environment indicator", () => { + expect( + shouldShowComposerContextStrip({ + hasActiveProject: true, + isGitRepo: true, + showEnvironmentIndicator: false, + }), + ).toBe(true); + }); +}); + describe("resolveEffectiveEnvMode", () => { it("treats draft threads already attached to a worktree as current-checkout mode", () => { expect( diff --git a/apps/web/src/components/BranchToolbar.logic.ts b/apps/web/src/components/BranchToolbar.logic.ts index d9737f17a32..485ffbf8d37 100644 --- a/apps/web/src/components/BranchToolbar.logic.ts +++ b/apps/web/src/components/BranchToolbar.logic.ts @@ -54,6 +54,14 @@ export function shouldShowEnvironmentIndicator(input: { return input.activeEnvironment !== null && !input.activeEnvironment.isPrimary; } +export function shouldShowComposerContextStrip(input: { + hasActiveProject: boolean; + isGitRepo: boolean; + showEnvironmentIndicator: boolean; +}): boolean { + return input.hasActiveProject && (input.isGitRepo || input.showEnvironmentIndicator); +} + export function resolveEnvModeLabel(mode: EnvMode): string { return mode === "worktree" ? "New worktree" : "Current checkout"; } diff --git a/apps/web/src/components/BranchToolbar.tsx b/apps/web/src/components/BranchToolbar.tsx index a3f043c6536..440f48d7c90 100644 --- a/apps/web/src/components/BranchToolbar.tsx +++ b/apps/web/src/components/BranchToolbar.tsx @@ -44,6 +44,7 @@ import { Separator } from "./ui/separator"; interface BranchToolbarProps { environmentId: EnvironmentId; threadId: ThreadId; + showGitControls: boolean; draftId?: DraftId; onEnvModeChange: (mode: EnvMode) => void; effectiveEnvModeOverride?: EnvMode; @@ -309,6 +310,7 @@ function useLabelsOverflow(element: HTMLDivElement | null): boolean { export const BranchToolbar = memo(function BranchToolbar({ environmentId, threadId, + showGitControls, draftId, onEnvModeChange, effectiveEnvModeOverride, @@ -403,7 +405,7 @@ export const BranchToolbar = memo(function BranchToolbar({ data-compact={labelsOverflow ? "" : undefined} className="chat-composer-context-strip group/composer-context -mt-4 mx-auto flex w-[calc(100%-2.75rem)] max-w-[calc(48rem-2.75rem)] items-center gap-2 ps-1 pe-2 pt-5 pb-1" > - {isMobile ? ( + {isMobile && showGitControls ? ( - + {showGitControls ? ( + + ) : null} )} - + {showGitControls ? ( + + ) : null}
)} - + {showGitControls ? ( + + ) : null}
); }); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 7b59530c955..6c2dc1478e6 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -244,7 +244,12 @@ import { ChatHeader } from "./chat/ChatHeader"; import { PanelLayoutControls, RightPanelMaximizeControl } from "./chat/PanelLayoutControls"; import { type ExpandedImagePreview } from "./chat/ExpandedImagePreview"; import { NoActiveThreadState } from "./NoActiveThreadState"; -import { resolveEffectiveEnvMode, resolveLocalCheckoutBranchMismatch } from "./BranchToolbar.logic"; +import { + resolveEffectiveEnvMode, + resolveLocalCheckoutBranchMismatch, + shouldShowComposerContextStrip, + shouldShowEnvironmentIndicator, +} from "./BranchToolbar.logic"; import { getProviderStatusBannerKey, ProviderStatusBanner, @@ -1745,6 +1750,14 @@ function ChatViewContent(props: ChatViewProps) { return envs; }, [activeProject, allProjects, projectGroupingSettings, primaryEnvironmentId, environmentById]); const hasMultipleEnvironments = logicalProjectEnvironments.length > 1; + const activeEnvironmentOption = + logicalProjectEnvironments.find( + (environment) => environment.environmentId === activeThread?.environmentId, + ) ?? null; + const showComposerEnvironmentIndicator = shouldShowEnvironmentIndicator({ + activeEnvironment: activeEnvironmentOption, + canPickEnvironment: hasMultipleEnvironments, + }); const openPullRequestDialog = useCallback( (reference?: string) => { @@ -2514,7 +2527,11 @@ function ChatViewContent(props: ChatViewProps) { terminalUiLaunchContext?.threadId === activeThreadId ? terminalUiLaunchContext : null; // Default true while loading to avoid toolbar flicker. const isGitRepo = gitStatusQuery.data?.isRepo ?? true; - const showComposerContextStrip = isGitRepo && activeProject !== null; + const showComposerContextStrip = shouldShowComposerContextStrip({ + hasActiveProject: activeProject !== null, + isGitRepo, + showEnvironmentIndicator: showComposerEnvironmentIndicator, + }); const initialDiffPanelGitScope = gitStatusQuery.data?.hasWorkingTreeChanges === true ? "unstaged" : "branch"; const diffPanelGitStatusResolutionKey = gitStatusQuery.data ? "resolved" : "pending"; @@ -6180,6 +6197,7 @@ function ChatViewContent(props: ChatViewProps) { Date: Thu, 6 Aug 2026 20:55:20 -0400 Subject: [PATCH 5/8] fix(server): stopping a Claude thread no longer shows an ede_diagnostic error (#5557) Co-authored-by: Claude Fable 5 --- .../src/provider/Layers/ClaudeAdapter.test.ts | 61 +++++++++++++++++++ .../src/provider/Layers/ClaudeAdapter.ts | 24 +++++++- 2 files changed, 84 insertions(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 24b8429a391..afa65ea39d6 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -1450,6 +1450,67 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("treats aborted_tools results as interrupted and hides ede_diagnostic errors", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 6).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + const turn = yield* adapter.sendTurn({ + threadId: session.threadId, + input: "hello", + attachments: [], + }); + + // Exact shape the CLI emits when Stop lands mid-tool-call: is_error + // is true and the only error is internal diagnostic telemetry. + harness.query.emit({ + type: "result", + subtype: "error_during_execution", + is_error: true, + errors: ["[ede_diagnostic] result_type=user last_content_type=n/a stop_reason=tool_use"], + stop_reason: "tool_use", + terminal_reason: "aborted_tools", + session_id: "sdk-session-abort-tools", + uuid: "result-abort-tools", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + assert.deepEqual( + runtimeEvents.map((event) => event.type), + [ + "session.started", + "session.configured", + "session.state.changed", + "turn.started", + "thread.started", + "turn.completed", + ], + ); + + const turnCompleted = runtimeEvents[runtimeEvents.length - 1]; + assert.equal(turnCompleted?.type, "turn.completed"); + if (turnCompleted?.type === "turn.completed") { + assert.equal(String(turnCompleted.turnId), String(turn.turnId)); + assert.equal(turnCompleted.payload.state, "interrupted"); + assert.equal(turnCompleted.payload.errorMessage, undefined); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("interruptTurn stops every live task before interrupting the turn", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 27acedc383a..f6f1c14420d 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -348,7 +348,29 @@ function resultErrorsText(result: SDKResultMessage): string { : ""; } +/** + * First user-facing error from a non-success result. "[ede_diagnostic] ..." + * entries are CLI-internal telemetry (the CLI hides them from its own UI too), + * so they must never become the error banner. + */ +function resultUserFacingError(result: SDKResultMessage): string | undefined { + if (result.subtype === "success" || !Array.isArray(result.errors)) { + return undefined; + } + return result.errors.find((error) => !error.startsWith("[ede_diagnostic]")); +} + function isInterruptedResult(result: SDKResultMessage): boolean { + // The CLI stamps user aborts explicitly: interrupting mid-tool-call yields + // "aborted_tools" (with an internal "[ede_diagnostic] ..." error and + // is_error: true), interrupting mid-stream yields "aborted_streaming". + if ( + result.terminal_reason === "aborted_tools" || + result.terminal_reason === "aborted_streaming" + ) { + return true; + } + const errors = resultErrorsText(result); if (errors.includes("interrupt")) { return true; @@ -2919,7 +2941,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } const status = turnStatusFromResult(message); - const errorMessage = message.subtype === "success" ? undefined : message.errors[0]; + const errorMessage = resultUserFacingError(message); if (status === "failed") { yield* emitRuntimeError(context, errorMessage ?? "Claude turn failed."); From 6da92244cc2a7438703be95a0fcfaca0b73502a7 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 21:16:36 -0400 Subject: [PATCH 6/8] fix(web): show one toast when snoozing threads in bulk (#5560) --- apps/web/src/components/SidebarV2.tsx | 150 ++++++++++++++++++-------- 1 file changed, 106 insertions(+), 44 deletions(-) diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index 462bfde13b7..003bec64d0f 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -2182,6 +2182,40 @@ export default function SidebarV2() { ); // One snooze per thread at a time — same double-dispatch guard as settle. const snoozingThreadKeysRef = useRef(new Set()); + const performSnooze = useCallback( + async ( + threadRef: ScopedThreadRef, + preset: SnoozePreset, + opts: { coSnoozingKeys?: ReadonlySet } = {}, + ) => { + const threadKey = scopedThreadKey(threadRef); + if (snoozingThreadKeysRef.current.has(threadKey)) { + return { status: "skipped" } as const; + } + snoozingThreadKeysRef.current.add(threadKey); + try { + // Snoozing the open thread moves you forward, same as settle — + // both park the thread you're done with for now. + const navigateAfterSnooze = planForwardNavigation(threadKey, opts.coSnoozingKeys); + const result = await snoozeThread(threadRef, preset.snoozedUntil); + if (result._tag === "Failure") { + // Never navigate away from a thread that did not snooze. + return isAtomCommandInterrupted(result) + ? ({ status: "interrupted" } as const) + : ({ status: "failure", error: squashAtomCommandFailure(result) } as const); + } + // Only move forward if the user is still on the snoozed thread — + // a navigation made during the await wins over ours. + if (routeThreadKeyRef.current === threadKey) { + navigateAfterSnooze?.(); + } + return { status: "success" } as const; + } finally { + snoozingThreadKeysRef.current.delete(threadKey); + } + }, + [planForwardNavigation, snoozeThread], + ); const attemptSnooze = useCallback( ( threadRef: ScopedThreadRef, @@ -2189,52 +2223,35 @@ export default function SidebarV2() { opts: { coSnoozingKeys?: ReadonlySet } = {}, ) => { void (async () => { - const threadKey = scopedThreadKey(threadRef); - if (snoozingThreadKeysRef.current.has(threadKey)) return; - snoozingThreadKeysRef.current.add(threadKey); - try { - // Snoozing the open thread moves you forward, same as settle — - // both park the thread you're done with for now. - const navigateAfterSnooze = planForwardNavigation(threadKey, opts.coSnoozingKeys); - const result = await snoozeThread(threadRef, preset.snoozedUntil); - if (result._tag === "Failure") { - // Never navigate away from a thread that did not snooze. - if (!isAtomCommandInterrupted(result)) { - const error = squashAtomCommandFailure(result); - toastManager.add( - stackedThreadToast({ - type: "error", - title: "Failed to snooze thread", - description: error instanceof Error ? error.message : "An error occurred.", - }), - ); - } - return; - } - // Snooze hides the row, so the toast is the only confirmation — - // and the Undo is the escape hatch for a mis-click. + const outcome = await performSnooze(threadRef, preset, opts); + if (outcome.status === "failure") { toastManager.add( stackedThreadToast({ - type: "success", - title: `Snoozed until ${snoozeWakeDescription(preset.snoozedUntil, new Date(), timestampFormat)}`, - timeout: 5_000, - actionProps: { - children: "Undo", - onClick: () => attemptUnsnooze(threadRef), - }, + type: "error", + title: "Failed to snooze thread", + description: + outcome.error instanceof Error ? outcome.error.message : "An error occurred.", }), ); - // Only move forward if the user is still on the snoozed thread — - // a navigation made during the await wins over ours. - if (routeThreadKeyRef.current === threadKey) { - navigateAfterSnooze?.(); - } - } finally { - snoozingThreadKeysRef.current.delete(threadKey); + return; } + if (outcome.status !== "success") return; + // Snooze hides the row, so the toast is the only confirmation — + // and the Undo is the escape hatch for a mis-click. + toastManager.add( + stackedThreadToast({ + type: "success", + title: `Snoozed until ${snoozeWakeDescription(preset.snoozedUntil, new Date(), timestampFormat)}`, + timeout: 5_000, + actionProps: { + children: "Undo", + onClick: () => attemptUnsnooze(threadRef), + }, + }), + ); })(); }, - [attemptUnsnooze, planForwardNavigation, snoozeThread, timestampFormat], + [attemptUnsnooze, performSnooze, timestampFormat], ); const removeFromSelection = useThreadSelectionStore((s) => s.removeFromSelection); @@ -2308,12 +2325,55 @@ export default function SidebarV2() { // Post-snooze navigation must skip threads snoozing in this same // batch — they are all leaving the card block together. const coSnoozingKeys = new Set(threadKeys); - for (const thread of selectedThreads) { - attemptSnooze(scopeThreadRef(thread.environmentId, thread.id), preset, { - coSnoozingKeys, - }); - } clearSelection(); + const outcomes = await Promise.all( + selectedThreads.map(async (thread) => { + const threadRef = scopeThreadRef(thread.environmentId, thread.id); + const outcome = await performSnooze(threadRef, preset, { coSnoozingKeys }); + return { outcome, threadRef }; + }), + ); + const snoozedThreadRefs = outcomes.flatMap(({ outcome, threadRef }) => + outcome.status === "success" ? [threadRef] : [], + ); + const failures = outcomes.flatMap(({ outcome }) => + outcome.status === "failure" ? [outcome.error] : [], + ); + + if (snoozedThreadRefs.length > 0) { + const snoozedCount = snoozedThreadRefs.length; + const failedCount = failures.length; + toastManager.add( + stackedThreadToast({ + type: failedCount > 0 ? "warning" : "success", + title: + failedCount > 0 + ? `Snoozed ${snoozedCount} of ${selectedThreads.length} threads` + : `Snoozed ${snoozedCount} thread${snoozedCount === 1 ? "" : "s"}`, + description: + failedCount > 0 + ? `${failedCount} thread${failedCount === 1 ? "" : "s"} couldn't be snoozed.` + : undefined, + timeout: 5_000, + actionProps: { + children: "Undo", + onClick: () => { + for (const threadRef of snoozedThreadRefs) attemptUnsnooze(threadRef); + }, + }, + }), + ); + } else if (failures.length > 0) { + const firstError = failures[0]; + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Failed to snooze threads", + description: + firstError instanceof Error ? firstError.message : "An error occurred.", + }), + ); + } } return; } @@ -2409,8 +2469,10 @@ export default function SidebarV2() { confirmThreadDelete, deleteThread, markThreadUnread, + performSnooze, removeFromSelection, serverConfigs, + attemptUnsnooze, updateThreadMetadata, timestampFormat, ], From 7aad7911f66c2fecba1cfd6601ea783a3fe2bf31 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 21:43:13 -0400 Subject: [PATCH 7/8] fix(server): let stopped threads settle immediately (#5553) --- .../src/orchestration/Layers/ProjectionPipeline.test.ts | 7 +++++++ apps/server/src/orchestration/Layers/ProjectionPipeline.ts | 3 ++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 9c4caf4c97d..09d7573f5d8 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1430,6 +1430,13 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { assert.deepEqual(settledRows, [ { state: "completed", completedAt: "2026-01-01T00:01:00.000Z" }, ]); + + const threadRows = yield* sql<{ readonly latestTurnId: string | null }>` + SELECT latest_turn_id AS "latestTurnId" + FROM projection_threads + WHERE thread_id = ${threadId} + `; + assert.deepEqual(threadRows, [{ latestTurnId: turnId }]); }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index fe683b08a3c..7776e374ee2 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -850,7 +850,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti } yield* projectionThreadRepository.upsert({ ...existingRow.value, - latestTurnId: event.payload.session.activeTurnId, + // activeTurnId describes current work; a terminal session must not erase history. + latestTurnId: event.payload.session.activeTurnId ?? existingRow.value.latestTurnId, updatedAt: event.occurredAt, }); yield* refreshThreadShellSummary(event.payload.threadId); From 6b73b3defe1dfb365de3b7bbb97ca56a26b50a43 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 6 Aug 2026 22:33:49 -0400 Subject: [PATCH 8/8] feat: paginate thread loading with user-anchored turn windows (#5493) Co-authored-by: Claude Fable 5 --- .../src/connection/environment-cache-store.ts | 5 +- .../features/threads/ThreadDetailScreen.tsx | 3 + .../src/features/threads/ThreadFeed.tsx | 20 +- .../features/threads/ThreadRouteScreen.tsx | 19 + .../Layers/ProjectionSnapshotQuery.test.ts | 405 +++++++++++++ .../Layers/ProjectionSnapshotQuery.ts | 369 +++++++++++- .../Services/ProjectionSnapshotQuery.ts | 8 + apps/server/src/orchestration/http.ts | 12 +- .../orchestration/threadDetailCursor.test.ts | 44 ++ .../src/orchestration/threadDetailCursor.ts | 62 ++ apps/server/src/persistence/Migrations.ts | 2 + .../037_ProjectionTurnsKeysetIndex.ts | 17 + apps/server/src/ws.ts | 10 +- apps/web/src/components/ChatView.tsx | 24 +- .../src/components/chat/MessagesTimeline.tsx | 44 +- apps/web/src/connection/storage.ts | 9 +- .../client-runtime/src/state/entities.test.ts | 2 + .../src/state/threadSnapshotHttp.ts | 27 +- .../client-runtime/src/state/threadState.ts | 24 + .../src/state/threads-pagination.test.ts | 543 ++++++++++++++++++ packages/client-runtime/src/state/threads.ts | 406 ++++++++++++- packages/contracts/src/environmentHttp.ts | 11 + packages/contracts/src/orchestration.ts | 51 ++ packages/contracts/src/server.ts | 6 + 24 files changed, 2093 insertions(+), 30 deletions(-) create mode 100644 apps/server/src/orchestration/threadDetailCursor.test.ts create mode 100644 apps/server/src/orchestration/threadDetailCursor.ts create mode 100644 apps/server/src/persistence/Migrations/037_ProjectionTurnsKeysetIndex.ts create mode 100644 packages/client-runtime/src/state/threads-pagination.test.ts 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;