From a2b447cd8bbabbe78ee3d937c009cc6289cf0809 Mon Sep 17 00:00:00 2001 From: olafura Date: Mon, 22 Jun 2026 20:52:53 +0200 Subject: [PATCH 01/13] fix(server): bound + paginate thread activity reads to stop heap OOM A busy SQLite database materialised hundreds of MB of activity payloads into the Node heap and crashed the server. Bound the reads and add an on-demand pagination path so deep history is still recoverable. - /api/orchestration/snapshot and the `t3` CLI offline path called getSnapshot(), loading every thread's full activity/message/checkpoint history even though they only read `.projects`. Point both at getCommandReadModel() (same shape, without the heavy per-thread tables). - getThreadDetailById windows activities to the most recent 500 (it fetches WINDOW+1 to detect truncation and sets `hasMoreActivities` so clients can lazy-load). Live activities still stream in via the event subscription. - New orchestration.getThreadActivities RPC pages older activities on demand: cursor is {beforeSequence} for sequenced rows or {beforeCreatedAt, beforeActivityId} for legacy unsequenced (NULL) rows, output {activities, hasMore}. The sequenced query orders by `sequence DESC` with `(sequence < ? OR sequence IS NULL)` so the (thread_id, sequence, created_at, activity_id) index satisfies the ORDER BY instead of a filesort. Unsequenced rows page by a (created_at, activity_id) cursor consistent with the window's ordering. Co-Authored-By: Claude Opus 4.8 --- .../checkpointing/CheckpointDiffQuery.test.ts | 10 + apps/server/src/cli/project.ts | 4 +- .../Layers/OrchestrationEngine.test.ts | 1 + .../Layers/ProjectionSnapshotQuery.test.ts | 156 ++++++++++++++ .../Layers/ProjectionSnapshotQuery.ts | 204 +++++++++++++++--- .../Services/ProjectionSnapshotQuery.ts | 12 ++ apps/server/src/orchestration/http.ts | 6 +- .../project/ProjectSetupScriptRunner.test.ts | 1 + .../Layers/ProviderSessionReaper.test.ts | 1 + apps/server/src/serverRuntimeStartup.test.ts | 4 + apps/server/src/ws.ts | 16 ++ packages/contracts/src/orchestration.ts | 48 +++++ packages/contracts/src/rpc.ts | 12 ++ 13 files changed, 446 insertions(+), 29 deletions(-) diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index 8e0e5fb74d5..6e1328210b0 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -77,6 +77,8 @@ describe("CheckpointDiffQuery.layer", () => { Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), + getThreadActivitiesPage: () => + Effect.die("CheckpointDiffQuery should not request thread activities"), getSnapshot: () => Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => @@ -185,6 +187,8 @@ describe("CheckpointDiffQuery.layer", () => { Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), + getThreadActivitiesPage: () => + Effect.die("CheckpointDiffQuery should not request thread activities"), getSnapshot: () => Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => @@ -268,6 +272,8 @@ describe("CheckpointDiffQuery.layer", () => { Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), + getThreadActivitiesPage: () => + Effect.die("CheckpointDiffQuery should not request thread activities"), getSnapshot: () => Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => @@ -336,6 +342,8 @@ describe("CheckpointDiffQuery.layer", () => { Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), + getThreadActivitiesPage: () => + Effect.die("CheckpointDiffQuery should not request thread activities"), getSnapshot: () => Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => @@ -389,6 +397,8 @@ describe("CheckpointDiffQuery.layer", () => { Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("CheckpointDiffQuery should not request the command read model"), + getThreadActivitiesPage: () => + Effect.die("CheckpointDiffQuery should not request thread activities"), getSnapshot: () => Effect.die("CheckpointDiffQuery should not request the full orchestration snapshot"), getShellSnapshot: () => diff --git a/apps/server/src/cli/project.ts b/apps/server/src/cli/project.ts index 25733a5e35b..8b7acdc3103 100644 --- a/apps/server/src/cli/project.ts +++ b/apps/server/src/cli/project.ts @@ -337,7 +337,9 @@ const dispatchLiveOrchestrationCommand = ( const getOfflineSnapshot = Effect.fn("getOfflineSnapshot")(function* () { const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; - return yield* projectionSnapshotQuery.getSnapshot(); + // Project resolution only reads `.projects`; the command read model returns the + // same shape without loading the heavy per-thread activity/message tables. + return yield* projectionSnapshotQuery.getCommandReadModel(); }); const tryResolveLiveProjectExecutionMode = Effect.fn("tryResolveLiveProjectExecutionMode")( diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index 731002ea783..b1d600aa706 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -203,6 +203,7 @@ describe("OrchestrationEngine", () => { getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), + getThreadActivitiesPage: () => Effect.die("unused"), }), ), Layer.provide( diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index d4a24a209ad..1a7c579922c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -346,6 +346,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { createdAt: "2026-02-24T00:00:06.000Z", }, ], + hasMoreActivities: false, checkpoints: [ { turnId: asTurnId("turn-1"), @@ -1091,6 +1092,8 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { assert.equal(threadDetail._tag, "Some"); if (threadDetail._tag === "Some") { assert.deepEqual(threadDetail.value.activities, snapshot.threads[0]?.activities ?? []); + // Well under the window — nothing older to lazy-load. + assert.equal(threadDetail.value.hasMoreActivities, false); } assert.deepEqual(snapshot.threads[0]?.activities ?? [], [ @@ -1270,6 +1273,159 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); + it.effect( + "windows thread-detail activities to the most recent 500 and pages older on demand", + () => + Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_thread_activities`; + yield* sql`DELETE FROM projection_state`; + + yield* sql` + INSERT INTO projection_projects ( + project_id, title, workspace_root, default_model_selection_json, + scripts_json, created_at, updated_at, deleted_at + ) + VALUES ( + 'project-1', 'Project 1', '/tmp/project-1', + '{"provider":"codex","model":"gpt-5-codex"}', '[]', + '2026-04-01T00:00:00.000Z', '2026-04-01T00:00:01.000Z', NULL + ) + `; + + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, branch, worktree_path, latest_turn_id, + latest_user_message_at, pending_approval_count, pending_user_input_count, + has_actionable_proposed_plan, created_at, updated_at, archived_at, deleted_at + ) + VALUES ( + 'thread-1', 'project-1', 'Thread 1', + '{"provider":"codex","model":"gpt-5-codex"}', 'full-access', 'default', + NULL, NULL, NULL, NULL, 0, 0, 0, + '2026-04-01T00:00:02.000Z', '2026-04-01T00:00:03.000Z', NULL, NULL + ) + `; + + // 600 activities (sequence 1..600); the detail load must return only the + // most recent 500 (sequence 101..600), re-sorted ascending for display. + const total = 600; + yield* Effect.forEach( + Array.from({ length: total }, (_unused, index) => index + 1), + (seq) => + sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, + sequence, created_at + ) + VALUES ( + ${`activity-${String(seq).padStart(4, "0")}`}, 'thread-1', NULL, + 'info', 'runtime.note', ${`act-${seq}`}, '{}', ${seq}, + '2026-04-01T00:01:00.000Z' + ) + `, + { discard: true }, + ); + + const threadDetail = yield* snapshotQuery.getThreadDetailById(ThreadId.make("thread-1")); + assert.equal(threadDetail._tag, "Some"); + if (threadDetail._tag === "Some") { + const activities = threadDetail.value.activities; + assert.equal(activities.length, 500); + assert.equal(activities[0]?.summary, "act-101"); + assert.equal(activities[0]?.sequence, 101); + assert.equal(activities.at(-1)?.summary, "act-600"); + // 600 > window, so the client is told older history can be lazy-loaded. + assert.equal(threadDetail.value.hasMoreActivities, true); + } + + // Lazy-load the page immediately older than the windowed view (cursor = + // oldest loaded sequence, 101): sequences 1..100, ascending, no more left. + const olderPage = yield* snapshotQuery.getThreadActivitiesPage({ + threadId: ThreadId.make("thread-1"), + beforeSequence: 101, + limit: 500, + }); + assert.equal(olderPage.activities.length, 100); + assert.equal(olderPage.activities[0]?.summary, "act-1"); + assert.equal(olderPage.activities.at(-1)?.summary, "act-100"); + assert.equal(olderPage.hasMore, false); + + // A bounded page returns the newest `limit` of the older set and reports + // that more remain (sequences 401..600, with 1..400 still older). + const boundedPage = yield* snapshotQuery.getThreadActivitiesPage({ + threadId: ThreadId.make("thread-1"), + beforeSequence: 601, + limit: 200, + }); + assert.equal(boundedPage.activities.length, 200); + assert.equal(boundedPage.activities[0]?.summary, "act-401"); + assert.equal(boundedPage.activities.at(-1)?.summary, "act-600"); + assert.equal(boundedPage.hasMore, true); + + yield* sql`DELETE FROM projection_thread_activities`; + + // Legacy rows may not have a sequence. They are still windowed in the + // detail load and must remain pageable by the deterministic created/id + // ordering used by the snapshot query. + yield* Effect.forEach( + Array.from({ length: total }, (_unused, index) => index + 1), + (seq) => + sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, + sequence, created_at + ) + VALUES ( + ${`unsequenced-${String(seq).padStart(4, "0")}`}, 'thread-1', NULL, + 'info', 'runtime.note', ${`legacy-act-${seq}`}, '{}', NULL, + '2026-04-01T00:01:00.000Z' + ) + `, + { discard: true }, + ); + + const legacyThreadDetail = yield* snapshotQuery.getThreadDetailById( + ThreadId.make("thread-1"), + ); + assert.equal(legacyThreadDetail._tag, "Some"); + if (legacyThreadDetail._tag === "Some") { + const activities = legacyThreadDetail.value.activities; + assert.equal(activities.length, 500); + assert.equal(activities[0]?.summary, "legacy-act-101"); + assert.equal(activities[0]?.sequence, undefined); + assert.equal(activities.at(-1)?.summary, "legacy-act-600"); + + const legacyOlderPage = yield* snapshotQuery.getThreadActivitiesPage({ + threadId: ThreadId.make("thread-1"), + beforeCreatedAt: activities[0]?.createdAt ?? "2026-04-01T00:01:00.000Z", + beforeActivityId: activities[0]?.id ?? asEventId("unsequenced-0101"), + limit: 500, + }); + assert.equal(legacyOlderPage.activities.length, 100); + assert.equal(legacyOlderPage.activities[0]?.summary, "legacy-act-1"); + assert.equal(legacyOlderPage.activities.at(-1)?.summary, "legacy-act-100"); + assert.equal(legacyOlderPage.hasMore, false); + } + + const legacyBoundedPage = yield* snapshotQuery.getThreadActivitiesPage({ + threadId: ThreadId.make("thread-1"), + beforeCreatedAt: "2026-04-01T00:01:00.000Z", + beforeActivityId: asEventId("unsequenced-0601"), + limit: 200, + }); + assert.equal(legacyBoundedPage.activities.length, 200); + assert.equal(legacyBoundedPage.activities[0]?.summary, "legacy-act-401"); + assert.equal(legacyBoundedPage.activities.at(-1)?.summary, "legacy-act-600"); + assert.equal(legacyBoundedPage.hasMore, true); + }), + ); + it.effect("uses projection_threads.latest_turn_id for bulk command and shell snapshots", () => Effect.gen(function* () { const snapshotQuery = yield* ProjectionSnapshotQuery; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 3d05bef4bdf..bd33cad20e2 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -1,6 +1,7 @@ import { ChatAttachment, CheckpointRef, + EventId, IsoDateTime, MessageId, NonNegativeInt, @@ -117,6 +118,25 @@ const ProjectIdLookupInput = Schema.Struct({ const ThreadIdLookupInput = Schema.Struct({ threadId: ThreadId, }); + +/** + * Maximum number of most-recent activities loaded into a thread-detail snapshot. + * Bounds peak memory when opening a long-lived thread; older activities are + * fetched on demand (lazy-load, planned) and live ones stream in via events. + */ +const THREAD_DETAIL_ACTIVITY_WINDOW = 500; + +const ThreadActivitiesBeforeSequenceInput = Schema.Struct({ + threadId: ThreadId, + beforeSequence: Schema.Number, + limit: Schema.Number, +}); +const ThreadActivitiesBeforeActivityInput = Schema.Struct({ + threadId: ThreadId, + beforeCreatedAt: IsoDateTime, + beforeActivityId: EventId, + limit: Schema.Number, +}); const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema; const ProjectionThreadIdLookupRowSchema = Schema.Struct({ threadId: ThreadId, @@ -254,6 +274,23 @@ function mapProposedPlanRow( }; } +function mapThreadActivityRow( + row: Schema.Schema.Type, +): OrchestrationThreadActivity { + const activity = { + id: row.activityId, + tone: row.tone, + kind: row.kind, + summary: row.summary, + payload: row.payload, + turnId: row.turnId, + createdAt: row.createdAt, + }; + // `sequence` is the pagination cursor; omit it when the (legacy) row is + // unsequenced so the optional contract field stays absent. + return row.sequence !== null ? Object.assign(activity, { sequence: row.sequence }) : activity; +} + function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { return (cause: unknown): ProjectionRepositoryError => Schema.isSchemaError(cause) @@ -824,6 +861,11 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + // The thread-detail timeline loads only the most recent window of activities so a + // long-lived thread (tens of thousands of activities, hundreds of MB of payload) + // can't blow the server heap. New activities still stream in live via the event + // subscription; older history will be fetched on demand once lazy-load lands. + // Select the most recent N (sequence DESC) then re-sort ascending for display. const listThreadActivityRowsByThread = SqlSchema.findAll({ Request: ThreadIdLookupInput, Result: ProjectionThreadActivityDbRowSchema, @@ -839,8 +881,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { payload_json AS "payload", sequence, created_at AS "createdAt" - FROM projection_thread_activities - WHERE thread_id = ${threadId} + FROM ( + SELECT * + FROM projection_thread_activities + WHERE thread_id = ${threadId} + ORDER BY + sequence DESC, + created_at DESC, + activity_id DESC + -- One extra beyond the window so the caller can report hasMoreActivities. + LIMIT ${THREAD_DETAIL_ACTIVITY_WINDOW + 1} + ) ORDER BY sequence ASC, created_at ASC, @@ -848,6 +899,81 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + // Older-than-cursor page for lazy-load. Returns rows newest-first (DESC) so a + // simple LIMIT yields the page adjacent to the cursor; the caller reverses to + // ascending. `sequence IS NULL` (legacy unsequenced) rows sort last under + // `sequence DESC` (SQLite orders NULLs last in DESC) — the very oldest — so + // paging eventually reaches them. `beforeSequence` is a NonNegativeInt, so + // `(sequence < beforeSequence OR sequence IS NULL)` is equivalent to the old + // `COALESCE(sequence, -1) < beforeSequence` but lets the + // (thread_id, sequence, created_at, activity_id) index satisfy the ORDER BY + // directly instead of forcing a filesort over the whole thread. + const listThreadActivityRowsBeforeSequence = SqlSchema.findAll({ + Request: ThreadActivitiesBeforeSequenceInput, + Result: ProjectionThreadActivityDbRowSchema, + execute: ({ threadId, beforeSequence, limit }) => + 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 (sequence < ${beforeSequence} OR sequence IS NULL) + ORDER BY + sequence DESC, + created_at DESC, + activity_id DESC + LIMIT ${limit} + `, + }); + + // Legacy unsequenced (sequence NULL) rows are paged by a (created_at, + // activity_id) cursor. created_at is compared lexicographically as TEXT, which + // equals chronological order only because timestamps are canonical ISO-8601 + // (always UTC `Z`, fixed millisecond precision) — the same invariant every + // `ORDER BY created_at` in this layer (including the detail window above) + // already relies on, so the cursor stays consistent with how rows are + // displayed. activity_id breaks created_at ties; its ordering is arbitrary but + // matches the window's `activity_id` tiebreak, so pages never skip or repeat. + const listUnsequencedThreadActivityRowsBeforeActivity = SqlSchema.findAll({ + Request: ThreadActivitiesBeforeActivityInput, + Result: ProjectionThreadActivityDbRowSchema, + execute: ({ threadId, beforeCreatedAt, beforeActivityId, limit }) => + 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 sequence IS NULL + AND ( + created_at < ${beforeCreatedAt} + OR ( + created_at = ${beforeCreatedAt} + AND activity_id < ${beforeActivityId} + ) + ) + ORDER BY + created_at DESC, + activity_id DESC + LIMIT ${limit} + `, + }); + const getThreadSessionRowByThread = SqlSchema.findOneOption({ Request: ThreadIdLookupInput, Result: ProjectionThreadSessionDbRowSchema, @@ -1092,16 +1218,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { for (const row of activityRows) { updatedAt = maxIso(updatedAt, row.createdAt); const threadActivities = activitiesByThread.get(row.threadId) ?? []; - threadActivities.push({ - id: row.activityId, - tone: row.tone, - kind: row.kind, - summary: row.summary, - payload: row.payload, - turnId: row.turnId, - ...(row.sequence !== null ? { sequence: row.sequence } : {}), - createdAt: row.createdAt, - }); + threadActivities.push(mapThreadActivityRow(row)); activitiesByThread.set(row.threadId, threadActivities); } @@ -1210,6 +1327,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { messages: messagesByThread.get(row.threadId) ?? [], proposedPlans: proposedPlansByThread.get(row.threadId) ?? [], activities: activitiesByThread.get(row.threadId) ?? [], + // The full snapshot is unwindowed, so there is never more to load. + hasMoreActivities: false, checkpoints: checkpointsByThread.get(row.threadId) ?? [], session: sessionsByThread.get(row.threadId) ?? null, })); @@ -2038,21 +2157,14 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { return message; }), proposedPlans: proposedPlanRows.map(mapProposedPlanRow), - activities: activityRows.map((row) => { - const activity = { - id: row.activityId, - tone: row.tone, - kind: row.kind, - summary: row.summary, - payload: row.payload, - turnId: row.turnId, - createdAt: row.createdAt, - }; - if (row.sequence !== null) { - return Object.assign(activity, { sequence: row.sequence }); - } - return activity; - }), + // The query fetches WINDOW+1 ascending rows; if it returned the extra + // one, older activities exist beyond the window — drop that oldest row + // and flag it so clients can lazy-load older history. + activities: (activityRows.length > THREAD_DETAIL_ACTIVITY_WINDOW + ? activityRows.slice(activityRows.length - THREAD_DETAIL_ACTIVITY_WINDOW) + : activityRows + ).map(mapThreadActivityRow), + hasMoreActivities: activityRows.length > THREAD_DETAIL_ACTIVITY_WINDOW, checkpoints: checkpointRows.map((row) => ({ turnId: row.turnId, checkpointTurnCount: row.checkpointTurnCount, @@ -2103,6 +2215,43 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { ), ); + const getThreadActivitiesPage: ProjectionSnapshotQueryShape["getThreadActivitiesPage"] = ( + input, + ) => + Effect.gen(function* () { + const limit = Math.min( + Math.max(1, input.limit ?? THREAD_DETAIL_ACTIVITY_WINDOW), + THREAD_DETAIL_ACTIVITY_WINDOW, + ); + // Fetch one extra to detect whether older activities remain. + const rowsEffect = + "beforeSequence" in input + ? listThreadActivityRowsBeforeSequence({ + threadId: input.threadId, + beforeSequence: input.beforeSequence, + limit: limit + 1, + }) + : listUnsequencedThreadActivityRowsBeforeActivity({ + threadId: input.threadId, + beforeCreatedAt: input.beforeCreatedAt, + beforeActivityId: input.beforeActivityId, + limit: limit + 1, + }); + const rows = yield* rowsEffect.pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.getThreadActivitiesPage:query", + "ProjectionSnapshotQuery.getThreadActivitiesPage:decodeRows", + ), + ), + ); + const hasMore = rows.length > limit; + // Rows are newest-first; keep the page closest to the cursor, then reverse + // to ascending for display. + const page = (hasMore ? rows.slice(0, limit) : rows).map(mapThreadActivityRow).toReversed(); + return { activities: page, hasMore }; + }); + return { getCommandReadModel, getSnapshot, @@ -2118,6 +2267,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { getThreadShellById, getThreadDetailById, getThreadDetailSnapshot, + getThreadActivitiesPage, } satisfies ProjectionSnapshotQueryShape; }); diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 23b291d8778..d235f460ce6 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -9,6 +9,8 @@ import type { CheckpointRef, OrchestrationCheckpointSummary, + OrchestrationGetThreadActivitiesInput, + OrchestrationGetThreadActivitiesResult, OrchestrationProject, OrchestrationProjectShell, OrchestrationReadModel, @@ -168,6 +170,16 @@ export interface ProjectionSnapshotQueryShape { readonly getThreadDetailSnapshot: ( threadId: ThreadId, ) => Effect.Effect, ProjectionRepositoryError>; + + /** + * Cursor-paginated load of a thread's older activities (lazy-load / infinite + * scroll). Returns the page of activities immediately older than the provided + * sequence or unsequenced activity cursor, ascending, plus whether older ones + * remain. + */ + readonly getThreadActivitiesPage: ( + input: OrchestrationGetThreadActivitiesInput, + ) => Effect.Effect; } /** diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts index 016c3d508ec..f57aa7d5711 100644 --- a/apps/server/src/orchestration/http.ts +++ b/apps/server/src/orchestration/http.ts @@ -31,8 +31,12 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group( Effect.fn("environment.orchestration.snapshot")(function* (args) { yield* annotateEnvironmentRequest(args.endpoint.name); yield* requireEnvironmentScope(AuthOrchestrationReadScope); + // The only consumer (the `t3` CLI project resolver) reads just + // `.projects`, so use the command read model — it returns the same + // OrchestrationReadModel shape but never materialises the per-thread + // activity/message/checkpoint tables (490MB+ on a busy DB → heap OOM). return yield* projectionSnapshotQuery - .getSnapshot() + .getCommandReadModel() .pipe( Effect.catch((cause) => failEnvironmentInternal("orchestration_snapshot_failed", cause), diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index 15612908079..077860a6944 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -27,6 +27,7 @@ const makeProject = (scripts: OrchestrationProject["scripts"]): OrchestrationPro const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => Layer.succeed(ProjectionSnapshotQuery.ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("unused"), + getThreadActivitiesPage: () => Effect.die("unused"), getSnapshot: () => Effect.die("unused"), getShellSnapshot: () => Effect.die("unused"), getArchivedShellSnapshot: () => Effect.die("unused"), diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 3843c8acbcd..8ddb7406953 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -212,6 +212,7 @@ describe("ProviderSessionReaper", () => { ), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), + getThreadActivitiesPage: () => Effect.die("unused"), }), ), Layer.provideMerge(NodeServices.layer), diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index b8102bda9ad..89c371a814e 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -97,6 +97,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa getThreadShellById: () => Effect.succeed(Option.none()), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), + getThreadActivitiesPage: () => Effect.die("unused"), }), Effect.provideService(AnalyticsService.AnalyticsService, { record: () => Effect.void, @@ -160,6 +161,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), + getThreadActivitiesPage: () => Effect.die("unused"), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -204,6 +206,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), + getThreadActivitiesPage: () => Effect.die("unused"), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, @@ -254,6 +257,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa getThreadShellById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), + getThreadActivitiesPage: () => Effect.die("unused"), }), Effect.provideService(OrchestrationEngine.OrchestrationEngineService, { readEvents: () => Stream.empty, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index f6f46d1e76e..4146b0c955b 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -34,6 +34,7 @@ import { type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, + OrchestrationGetThreadActivitiesError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, type ProjectId, @@ -288,6 +289,7 @@ const SHELL_RESUME_MAX_GAP = 1_000; const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], + [ORCHESTRATION_WS_METHODS.getThreadActivities, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getFullThreadDiff, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.replayEvents, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope], @@ -1186,6 +1188,20 @@ const makeWsRpcLayer = ( ), { "rpc.aggregate": "orchestration" }, ), + [ORCHESTRATION_WS_METHODS.getThreadActivities]: (input) => + observeRpcEffect( + ORCHESTRATION_WS_METHODS.getThreadActivities, + projectionSnapshotQuery.getThreadActivitiesPage(input).pipe( + Effect.mapError( + (cause) => + new OrchestrationGetThreadActivitiesError({ + message: "Failed to load thread activities page", + cause, + }), + ), + ), + { "rpc.aggregate": "orchestration" }, + ), [ORCHESTRATION_WS_METHODS.getFullThreadDiff]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.getFullThreadDiff, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 84b7a8fa07f..8e89bd28cb3 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -25,6 +25,7 @@ import { ProviderInstanceId } from "./providerInstance.ts"; export const ORCHESTRATION_WS_METHODS = { dispatchCommand: "orchestration.dispatchCommand", getTurnDiff: "orchestration.getTurnDiff", + getThreadActivities: "orchestration.getThreadActivities", getFullThreadDiff: "orchestration.getFullThreadDiff", replayEvents: "orchestration.replayEvents", getArchivedShellSnapshot: "orchestration.getArchivedShellSnapshot", @@ -373,6 +374,10 @@ export const OrchestrationThread = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed([])), ), activities: Schema.Array(OrchestrationThreadActivity), + // The detail snapshot windows `activities` to the most recent page; this is + // true when older activities exist beyond the window and can be lazy-loaded + // via the getThreadActivities RPC. Absent on lightweight (shell) threads. + hasMoreActivities: Schema.optional(Schema.Boolean), checkpoints: Schema.Array(OrchestrationCheckpointSummary), session: Schema.NullOr(OrchestrationSession), }); @@ -1353,6 +1358,37 @@ export type OrchestrationGetTurnDiffInput = typeof OrchestrationGetTurnDiffInput export const OrchestrationGetTurnDiffResult = ThreadTurnDiff; export type OrchestrationGetTurnDiffResult = typeof OrchestrationGetTurnDiffResult.Type; +/** + * Cursor-paginated load of a thread's OLDER activities (lazy-load / infinite + * scroll). Sequenced activity uses `beforeSequence`, the `sequence` of the + * oldest activity the client currently holds. Legacy unsequenced activity uses + * the `(beforeCreatedAt, beforeActivityId)` pair from the oldest loaded + * activity. The server returns the page of activities immediately older than the + * cursor (chronological ascending) plus whether any remain beyond that. + */ +export const OrchestrationGetThreadActivitiesInput = Schema.Union([ + Schema.Struct({ + threadId: ThreadId, + beforeSequence: NonNegativeInt, + limit: Schema.optional(NonNegativeInt), + }), + Schema.Struct({ + threadId: ThreadId, + beforeCreatedAt: IsoDateTime, + beforeActivityId: EventId, + limit: Schema.optional(NonNegativeInt), + }), +]); +export type OrchestrationGetThreadActivitiesInput = + typeof OrchestrationGetThreadActivitiesInput.Type; + +export const OrchestrationGetThreadActivitiesResult = Schema.Struct({ + activities: Schema.Array(OrchestrationThreadActivity), + hasMore: Schema.Boolean, +}); +export type OrchestrationGetThreadActivitiesResult = + typeof OrchestrationGetThreadActivitiesResult.Type; + export const OrchestrationGetFullThreadDiffInput = Schema.Struct({ threadId: ThreadId, toTurnCount: NonNegativeInt, @@ -1380,6 +1416,10 @@ export const OrchestrationRpcSchemas = { input: OrchestrationGetTurnDiffInput, output: OrchestrationGetTurnDiffResult, }, + getThreadActivities: { + input: OrchestrationGetThreadActivitiesInput, + output: OrchestrationGetThreadActivitiesResult, + }, getFullThreadDiff: { input: OrchestrationGetFullThreadDiffInput, output: OrchestrationGetFullThreadDiffResult, @@ -1426,6 +1466,14 @@ export class OrchestrationGetTurnDiffError extends Schema.TaggedErrorClass()( + "OrchestrationGetThreadActivitiesError", + { + message: TrimmedNonEmptyString, + cause: Schema.optional(Schema.Defect()), + }, +) {} + export class OrchestrationGetFullThreadDiffError extends Schema.TaggedErrorClass()( "OrchestrationGetFullThreadDiffError", { diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index fa2d23b8ef2..cf81d6234ba 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -52,6 +52,8 @@ import { OrchestrationGetFullThreadDiffError, OrchestrationGetFullThreadDiffInput, OrchestrationGetSnapshotError, + OrchestrationGetThreadActivitiesError, + OrchestrationGetThreadActivitiesInput, OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, OrchestrationReplayEventsError, @@ -622,6 +624,15 @@ export const WsOrchestrationGetTurnDiffRpc = Rpc.make(ORCHESTRATION_WS_METHODS.g error: Schema.Union([OrchestrationGetTurnDiffError, EnvironmentAuthorizationError]), }); +export const WsOrchestrationGetThreadActivitiesRpc = Rpc.make( + ORCHESTRATION_WS_METHODS.getThreadActivities, + { + payload: OrchestrationGetThreadActivitiesInput, + success: OrchestrationRpcSchemas.getThreadActivities.output, + error: Schema.Union([OrchestrationGetThreadActivitiesError, EnvironmentAuthorizationError]), + }, +); + export const WsOrchestrationGetFullThreadDiffRpc = Rpc.make( ORCHESTRATION_WS_METHODS.getFullThreadDiff, { @@ -764,6 +775,7 @@ export const WsRpcGroup = RpcGroup.make( WsSubscribeAuthAccessRpc, WsOrchestrationDispatchCommandRpc, WsOrchestrationGetTurnDiffRpc, + WsOrchestrationGetThreadActivitiesRpc, WsOrchestrationGetFullThreadDiffRpc, WsOrchestrationReplayEventsRpc, WsOrchestrationGetArchivedShellSnapshotRpc, From 4401d0749b4c4fe0afa2f31e08453d410d970bd6 Mon Sep 17 00:00:00 2001 From: olafura Date: Mon, 22 Jun 2026 20:52:53 +0200 Subject: [PATCH 02/13] feat(web): lazy-load older thread history on scroll-up The server windows thread detail to the most recent 500 activities and reports `hasMoreActivities`. The web timeline now fetches older pages on demand via the getThreadActivities RPC: - client-runtime: loadThreadActivities command on orchestrationEnvironment. - ChatView keeps per-thread older pages in local state, prepends them ahead of the live window, and derives hasMore from the server flag (no client-side window-size constant). A thread-keyed in-flight ref coalesces the duplicate dispatches a fast scroll-to-top would otherwise fire, and a request-key guard discards results after a thread switch. - MessagesTimeline triggers a load on reaching the top (maintainVisibleContentPosition anchors the viewport on prepend) with a "Load older history" header and loading indicator. Co-Authored-By: Claude Opus 4.8 --- apps/web/src/components/ChatView.tsx | 104 +++++++++++++++++- .../components/chat/MessagesTimeline.test.tsx | 33 ++++++ .../src/components/chat/MessagesTimeline.tsx | 38 ++++++- .../client-runtime/src/state/orchestration.ts | 7 +- 4 files changed, 178 insertions(+), 4 deletions(-) diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index ab1256cddb3..2bd9b915e3f 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -201,6 +201,7 @@ import { primaryServerSettingsAtom, serverEnvironment, } from "../state/server"; +import { orchestrationEnvironment } from "../state/orchestration"; import { terminalEnvironment } from "../state/terminal"; import { threadEnvironment } from "../state/threads"; import { vcsEnvironment } from "../state/vcs"; @@ -1924,7 +1925,105 @@ function ChatViewContent(props: ChatViewProps) { ); const selectedProvider: ProviderDriverKind = lockedProvider ?? unlockedSelectedProvider; const phase = derivePhase(activeThread?.session ?? null); - const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; + + // ── Older-history lazy-load ──────────────────────────────────────────────── + // The detail snapshot windows activities to the most recent page (the server + // sets `hasMoreActivities` when older ones exist); older pages are fetched on + // demand (infinite scroll-up) and prepended. Messages aren't windowed + // server-side, so this just back-fills the older tool activity. + const [olderActivities, setOlderActivities] = useState< + ReadonlyArray + >([]); + const [olderLoaded, setOlderLoaded] = useState(false); + const [olderHasMore, setOlderHasMore] = useState(false); + const [loadingOlderActivities, setLoadingOlderActivities] = useState(false); + const loadThreadActivities = useAtomCommand(orchestrationEnvironment.loadThreadActivities, { + reportFailure: false, + }); + const activeThreadActivityRequestKey = activeThread + ? `${activeThread.environmentId}\u0000${activeThread.id}` + : null; + const activeThreadActivityRequestKeyRef = useRef(activeThreadActivityRequestKey); + activeThreadActivityRequestKeyRef.current = activeThreadActivityRequestKey; + useEffect(() => { + setOlderActivities([]); + setOlderLoaded(false); + setOlderHasMore(false); + setLoadingOlderActivities(false); + }, [activeThreadActivityRequestKey]); + + const liveThreadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; + const threadActivities = useMemo( + () => + olderActivities.length > 0 + ? [...olderActivities, ...liveThreadActivities] + : liveThreadActivities, + [olderActivities, liveThreadActivities], + ); + // Before any page is loaded, the server tells us whether older history exists + // beyond the windowed snapshot; afterwards the page `hasMore` is authoritative. + const hasMoreOlderActivities = olderLoaded + ? olderHasMore + : (activeThread?.hasMoreActivities ?? false); + // Tracks the request key of an in-flight older-history load. The scroll + // handler fires onLoadOlder on every frame while at the top, but the loading + // *state* only updates on the next render — without a synchronous guard a fast + // scroll-to-top dispatches several duplicate requests for the same cursor. + const inFlightOlderKeyRef = useRef(null); + const loadOlderActivities = useCallback(() => { + if (!activeThread || !hasMoreOlderActivities) { + return; + } + const oldestActivity = threadActivities[0]; + if (!oldestActivity || !activeThreadActivityRequestKey) { + return; + } + if (inFlightOlderKeyRef.current === activeThreadActivityRequestKey) { + return; // a load for this thread is already in flight + } + const cursorInput = + oldestActivity.sequence !== undefined + ? { beforeSequence: oldestActivity.sequence } + : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id }; + const requestKey = activeThreadActivityRequestKey; + inFlightOlderKeyRef.current = requestKey; + setLoadingOlderActivities(true); + void loadThreadActivities({ + environmentId: activeThread.environmentId, + input: { threadId: activeThread.id, ...cursorInput }, + }) + .then((result) => { + if (activeThreadActivityRequestKeyRef.current !== requestKey) { + return; + } + if (result._tag !== "Success") { + return; + } + const page = result.value; + setOlderActivities((prev) => { + const seen = new Set(prev.map((activity) => activity.id)); + const fresh = page.activities.filter((activity) => !seen.has(activity.id)); + return [...fresh, ...prev]; + }); + setOlderLoaded(true); + setOlderHasMore(page.hasMore); + }) + .finally(() => { + if (inFlightOlderKeyRef.current === requestKey) { + inFlightOlderKeyRef.current = null; + } + if (activeThreadActivityRequestKeyRef.current === requestKey) { + setLoadingOlderActivities(false); + } + }); + }, [ + activeThread, + activeThreadActivityRequestKey, + hasMoreOlderActivities, + threadActivities, + loadThreadActivities, + ]); + const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); const pendingApprovals = useMemo( () => derivePendingApprovals(threadActivities), @@ -5680,6 +5779,9 @@ function ChatViewContent(props: ChatViewProps) { activeTurnStartedAt={activeWorkStartedAt} listRef={legendListRef} timelineEntries={timelineEntries} + hasMoreOlder={hasMoreOlderActivities} + loadingOlder={loadingOlderActivities} + onLoadOlder={loadOlderActivities} latestTurn={activeLatestTurn} runningTurnId={ activeThread.session?.status === "running" diff --git a/apps/web/src/components/chat/MessagesTimeline.test.tsx b/apps/web/src/components/chat/MessagesTimeline.test.tsx index 296a54457a2..68adda13dd8 100644 --- a/apps/web/src/components/chat/MessagesTimeline.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.test.tsx @@ -654,4 +654,37 @@ describe("MessagesTimeline", () => { expect(markup).toContain("lucide-x"); expect(markup).toContain('aria-label="Tool call failed"'); }); + + it("offers a 'Load older history' control when older activity remains", async () => { + const { MessagesTimeline } = await import("./MessagesTimeline"); + const markup = renderToStaticMarkup( + , + ); + expect(markup).toContain("Load older history"); + }); + + it("shows a loading indicator while older history is being fetched", async () => { + const { MessagesTimeline } = await import("./MessagesTimeline"); + const markup = renderToStaticMarkup( + , + ); + expect(markup).toContain("Loading older history"); + }); + + it("renders no older-history control when none remains", async () => { + const { MessagesTimeline } = await import("./MessagesTimeline"); + const markup = renderToStaticMarkup( + , + ); + expect(markup).not.toContain("older history"); + }); }); diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index ae190e00068..139a3f5908c 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -184,6 +184,10 @@ interface MessagesTimelineProps { onManualNavigation: () => void; hideEmptyPlaceholder?: boolean; topFadeEnabled?: boolean; + /** Older history beyond the live activity window can be lazy-loaded. */ + hasMoreOlder?: boolean; + loadingOlder?: boolean; + onLoadOlder?: () => void; } // --------------------------------------------------------------------------- @@ -219,6 +223,9 @@ export const MessagesTimeline = memo(function MessagesTimeline({ onManualNavigation, hideEmptyPlaceholder = false, topFadeEnabled = false, + hasMoreOlder = false, + loadingOlder = false, + onLoadOlder, }: MessagesTimelineProps) { const [expandedTurnIds, setExpandedTurnIds] = useState>(new Set()); const [expandedWorkGroupIds, setExpandedWorkGroupIds] = useState>(new Set()); @@ -361,6 +368,11 @@ export const MessagesTimeline = memo(function MessagesTimeline({ if (isAtEnd !== undefined) { onIsAtEndChange(isAtEnd); } + // Reaching the top lazy-loads older history; maintainVisibleContentPosition + // (set on the list) keeps the viewport anchored when rows prepend. + if (state?.isAtStart && hasMoreOlder && !loadingOlder) { + onLoadOlder?.(); + } if (!state || minimapItems.length === 0) { return; } @@ -383,7 +395,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ strip.dataset.inView = inView ? "true" : "false"; } - }, [listRef, minimapItems, minimapStripMap, onIsAtEndChange]); + }, [listRef, minimapItems, minimapStripMap, onIsAtEndChange, hasMoreOlder, loadingOlder, onLoadOlder]); useEffect(() => { const frame = requestAnimationFrame(handleScroll); @@ -415,6 +427,28 @@ export const MessagesTimeline = memo(function MessagesTimeline({ }; }, [timelineViewportElement, rows.length]); + const listHeader = useMemo(() => { + if (loadingOlder) { + return ( +
+ Loading older history… +
+ ); + } + if (hasMoreOlder) { + return ( + + ); + } + return topFadeEnabled ? TIMELINE_LIST_FADE_HEADER : TIMELINE_LIST_HEADER; + }, [loadingOlder, hasMoreOlder, onLoadOlder, topFadeEnabled]); + const sharedState = useMemo( () => ({ timestampFormat, @@ -515,7 +549,7 @@ 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={listHeader} ListFooterComponent={TIMELINE_LIST_FOOTER} /> ( @@ -12,6 +12,11 @@ export function createOrchestrationEnvironmentAtoms( label: "environment-data:orchestration:turn-diff", tag: ORCHESTRATION_WS_METHODS.getTurnDiff, }), + // Imperative lazy-load of older thread activities (infinite scroll-up). + loadThreadActivities: createEnvironmentRpcCommand(runtime, { + label: "environment-data:orchestration:thread-activities", + tag: ORCHESTRATION_WS_METHODS.getThreadActivities, + }), fullThreadDiff: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:full-thread-diff", tag: ORCHESTRATION_WS_METHODS.getFullThreadDiff, From 4d68dcec80fe89c2f64cdc4e55ae79b4873de104 Mon Sep 17 00:00:00 2001 From: olafura Date: Mon, 22 Jun 2026 20:52:53 +0200 Subject: [PATCH 03/13] feat(mobile): lazy-load older thread history on scroll-up MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mobile consumed the windowed thread detail with no way to fetch older activities, silently truncating history. Mirror the web lazy-load: - useThreadComposerState: older-activity state + a thread-keyed in-flight ref guard, a reset effect on thread switch, and loadThreadActivities with the same union cursor. Older pages are deduped (against prior pages and the live window) and prepended into the feed; initial hasMore comes from the server's hasMoreActivities flag. - ThreadFeed: onStartReached triggers the load, maintainVisibleContentPosition anchors the viewport on prepend, and the header shows a spinner while loading. - Forward the values through ThreadRouteScreen → ThreadDetailScreen. Co-Authored-By: Claude Opus 4.8 --- .../features/threads/ThreadDetailScreen.tsx | 6 + .../src/features/threads/ThreadFeed.tsx | 22 +++- .../features/threads/ThreadRouteScreen.tsx | 3 + .../src/state/use-thread-composer-state.ts | 119 +++++++++++++++++- 4 files changed, 146 insertions(+), 4 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 5cb04290f66..e68661a2d1f 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -62,6 +62,9 @@ export interface ThreadDetailScreenProps { /** Message sync status for the selected thread (drives the composer status pill). */ readonly threadSyncStatus?: EnvironmentThreadStatus; readonly activeThreadBusy: boolean; + readonly hasMoreOlderActivities: boolean; + readonly loadingOlderActivities: boolean; + readonly onLoadOlderActivities: () => void; readonly environmentId: EnvironmentId; readonly projectWorkspaceRoot: string | null; readonly threadCwd: string | null; @@ -371,6 +374,9 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread usesAutomaticContentInsets={props.usesAutomaticContentInsets} onHeaderMaterialVisibilityChange={props.onHeaderMaterialVisibilityChange} skills={selectedProviderSkills} + hasMoreOlder={props.hasMoreOlderActivities} + loadingOlder={props.loadingOlderActivities} + onLoadOlder={props.onLoadOlderActivities} /> ) : ( diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 37a8639fdbd..4b5b352b9ef 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -141,6 +141,10 @@ export interface ThreadFeedProps { readonly usesAutomaticContentInsets?: boolean; readonly onHeaderMaterialVisibilityChange?: (visible: boolean) => void; readonly skills?: ReadonlyArray; + /** Older history beyond the live activity window can be lazy-loaded on scroll-up. */ + readonly hasMoreOlder?: boolean; + readonly loadingOlder?: boolean; + readonly onLoadOlder?: () => void; } function MessageAttachmentImage(props: { @@ -1498,6 +1502,16 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { ? props.latestTurn.turnId : null; + // Reaching the top (oldest) lazy-loads older history. The hook keys an + // in-flight guard by thread, so repeated fires during scroll coalesce. + const { hasMoreOlder, loadingOlder, onLoadOlder } = props; + const onStartReachedOlderHistory = useCallback(() => { + if (hasMoreOlder && !loadingOlder) { + onLoadOlder?.(); + } + }, [hasMoreOlder, loadingOlder, onLoadOlder]); + + useEffect(() => { const previous = previousLatestTurnRef.current; previousLatestTurnRef.current = props.latestTurn; @@ -1790,9 +1804,15 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { estimatedItemSize={180} initialScrollAtEnd onScroll={handleScroll} + onStartReached={onStartReachedOlderHistory} + onStartReachedThreshold={0.5} scrollEventThrottle={16} ListHeaderComponent={ - usesNativeAutomaticInsets ? null : + usesNativeAutomaticInsets && !loadingOlder ? null : ( + + {loadingOlder ? : null} + + ) } contentContainerStyle={{ paddingTop: 12, diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index 7fb4740ddce..ae63c3406b2 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -767,6 +767,9 @@ function ThreadRouteContent( connectionStateLabel={routeConnectionState} threadSyncStatus={selectedThreadDetailState.status} activeThreadBusy={composer.activeThreadBusy} + hasMoreOlderActivities={composer.hasMoreOlderActivities} + loadingOlderActivities={composer.loadingOlderActivities} + onLoadOlderActivities={composer.onLoadOlderActivities} environmentId={selectedThread.environmentId} projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null} threadCwd={selectedThreadCwd} diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 90831f8437a..1639b777dc6 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -1,11 +1,12 @@ import { useAtomValue } from "@effect/atom-react"; -import { useCallback, useEffect, useMemo } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { CommandId, MessageId, type EnvironmentId, type ModelSelection, + type OrchestrationThreadActivity, type ProviderInteractionMode, type RuntimeMode, type ThreadId, @@ -36,11 +37,15 @@ import { useComposerDraft, } from "./use-composer-drafts"; import { setPendingConnectionError } from "../state/use-remote-environment-registry"; +import { orchestrationEnvironment } from "../state/orchestration"; import { useSelectedThreadDetail } from "../state/use-thread-detail"; import { useThreadSelection } from "../state/use-thread-selection"; +import { useAtomCommand } from "./use-atom-command"; import { enqueueThreadOutboxMessage } from "./thread-outbox"; import { useThreadOutboxMessages } from "./use-thread-outbox"; +const EMPTY_ACTIVITIES: ReadonlyArray = []; + export function appendReviewCommentToDraft(input: { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; @@ -89,9 +94,114 @@ export function useThreadComposerState() { () => (selectedThreadKey ? (queuedMessagesByThreadKey[selectedThreadKey] ?? []) : []), [queuedMessagesByThreadKey, selectedThreadKey], ); + + // ── Older-history lazy-load (mirrors web ChatView) ────────────────────────── + // The detail snapshot windows activities to the most recent page (the server + // sets `hasMoreActivities`); older pages are fetched on demand and prepended. + const [olderActivities, setOlderActivities] = useState< + ReadonlyArray + >([]); + const [olderLoaded, setOlderLoaded] = useState(false); + const [olderHasMore, setOlderHasMore] = useState(false); + const [loadingOlderActivities, setLoadingOlderActivities] = useState(false); + const loadThreadActivities = useAtomCommand(orchestrationEnvironment.loadThreadActivities, { + reportFailure: false, + }); + + const activityRequestKey = selectedThreadShell + ? `${selectedThreadShell.environmentId}\u0000${selectedThreadShell.id}` + : null; + const activityRequestKeyRef = useRef(activityRequestKey); + activityRequestKeyRef.current = activityRequestKey; + useEffect(() => { + setOlderActivities([]); + setOlderLoaded(false); + setOlderHasMore(false); + setLoadingOlderActivities(false); + }, [activityRequestKey]); + + const liveActivities = selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES; + const mergedActivities = useMemo( + () => + olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities, + [olderActivities, liveActivities], + ); + // Before any page is loaded, the server tells us whether older history exists. + const hasMoreOlderActivities = olderLoaded + ? olderHasMore + : (selectedThreadDetail?.hasMoreActivities ?? false); + + // Synchronous in-flight guard keyed by thread: the list fires onLoadOlder + // repeatedly while pinned at the top, but loading *state* only updates next + // render, so without this a fast scroll dispatches duplicate same-cursor calls. + const inFlightOlderKeyRef = useRef(null); + const onLoadOlderActivities = useCallback(() => { + if (!selectedThreadShell || !hasMoreOlderActivities) { + return; + } + const oldestActivity = mergedActivities[0]; + if (!oldestActivity || !activityRequestKey) { + return; + } + if (inFlightOlderKeyRef.current === activityRequestKey) { + return; + } + const cursorInput = + oldestActivity.sequence !== undefined + ? { beforeSequence: oldestActivity.sequence } + : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id }; + const requestKey = activityRequestKey; + inFlightOlderKeyRef.current = requestKey; + setLoadingOlderActivities(true); + void loadThreadActivities({ + environmentId: selectedThreadShell.environmentId, + input: { threadId: selectedThreadShell.id, ...cursorInput }, + }) + .then((result) => { + if (activityRequestKeyRef.current !== requestKey) { + return; + } + if (result._tag !== "Success") { + return; + } + const page = result.value; + setOlderActivities((prev) => { + // Dedup against both already-loaded older pages and the live window, + // since mobile merges everything into one array (duplicate ids would + // produce duplicate React keys in the feed). + const seen = new Set(prev.map((activity) => activity.id)); + for (const activity of liveActivities) { + seen.add(activity.id); + } + const fresh = page.activities.filter((activity) => !seen.has(activity.id)); + return [...fresh, ...prev]; + }); + setOlderLoaded(true); + setOlderHasMore(page.hasMore); + }) + .finally(() => { + if (inFlightOlderKeyRef.current === requestKey) { + inFlightOlderKeyRef.current = null; + } + if (activityRequestKeyRef.current === requestKey) { + setLoadingOlderActivities(false); + } + }); + }, [ + selectedThreadShell, + hasMoreOlderActivities, + mergedActivities, + activityRequestKey, + liveActivities, + loadThreadActivities, + ]); + const selectedThreadFeed = useMemo( - () => (selectedThreadDetail ? buildThreadFeed(selectedThreadDetail) : []), - [selectedThreadDetail], + () => + selectedThreadDetail + ? buildThreadFeed({ ...selectedThreadDetail, activities: mergedActivities }) + : [], + [selectedThreadDetail, mergedActivities], ); const selectedDraft = selectedThreadKey ? composerDrafts[selectedThreadKey] : null; @@ -299,6 +409,9 @@ export function useThreadComposerState() { runtimeMode, interactionMode, activeThreadBusy, + hasMoreOlderActivities, + loadingOlderActivities, + onLoadOlderActivities, onChangeDraftMessage, onPickDraftImages, onPasteIntoDraft, From 5bdff6106b1e9f5d1a548f8bd06aec45b2f7805b Mon Sep 17 00:00:00 2001 From: olafura Date: Mon, 22 Jun 2026 21:14:07 +0200 Subject: [PATCH 04/13] fix: invalidate lazy-loaded older history when the live window reshapes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review (PR #3510, Cursor Bugbot): - Reconnect / checkpoint revert can re-snapshot or filter the live activity window, but the prepended `olderActivities` weren't invalidated — leaving gaps or showing reverted history. Reset the lazy-load state when the live window's oldest activity id changes (it's stable while activities only append, so this doesn't fire during a normal turn), in addition to on thread switch. Web + mobile. - Web only deduped a new older page against already-loaded older pages, not the live window (mobile already did both). Dedup against both so a boundary overlap can't produce duplicate ids / React keys. Not changed — the "unsequenced cursor hides sequenced history" finding is a false positive: NULL-sequence (legacy) rows always sort oldest in the window/cursor ordering, so the oldest-loaded row is only unsequenced once every sequenced row is already loaded; the unsequenced cursor can never strand sequenced rows. Co-Authored-By: Claude Opus 4.8 --- .../src/state/use-thread-composer-state.ts | 11 ++++++++--- apps/web/src/components/ChatView.tsx | 16 ++++++++++++++-- 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 1639b777dc6..4e77fc668ab 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -113,14 +113,19 @@ export function useThreadComposerState() { : null; const activityRequestKeyRef = useRef(activityRequestKey); activityRequestKeyRef.current = activityRequestKey; + + const liveActivities = selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES; + // The live window's oldest activity is stable while new activities only + // append; it changes when the window is re-snapshotted (reconnect) or rows are + // removed (checkpoint revert), either of which can make prepended older pages + // stale or gappy — so reset the lazy-load state when it (or the thread) changes. + const liveOldestActivityId = liveActivities[0]?.id ?? null; useEffect(() => { setOlderActivities([]); setOlderLoaded(false); setOlderHasMore(false); setLoadingOlderActivities(false); - }, [activityRequestKey]); - - const liveActivities = selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES; + }, [activityRequestKey, liveOldestActivityId]); const mergedActivities = useMemo( () => olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 2bd9b915e3f..bb36e9ab9ca 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1945,14 +1945,19 @@ function ChatViewContent(props: ChatViewProps) { : null; const activeThreadActivityRequestKeyRef = useRef(activeThreadActivityRequestKey); activeThreadActivityRequestKeyRef.current = activeThreadActivityRequestKey; + const liveThreadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; + // The live window's oldest activity is stable while new activities only + // append; it changes when the window is re-snapshotted (reconnect) or rows are + // removed (checkpoint revert), either of which can make prepended older pages + // stale or gappy — so reset the lazy-load state when it (or the thread) changes. + const liveOldestActivityId = liveThreadActivities[0]?.id ?? null; useEffect(() => { setOlderActivities([]); setOlderLoaded(false); setOlderHasMore(false); setLoadingOlderActivities(false); - }, [activeThreadActivityRequestKey]); + }, [activeThreadActivityRequestKey, liveOldestActivityId]); - const liveThreadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; const threadActivities = useMemo( () => olderActivities.length > 0 @@ -2001,7 +2006,13 @@ function ChatViewContent(props: ChatViewProps) { } const page = result.value; setOlderActivities((prev) => { + // Dedup against both already-loaded older pages and the live window so + // an overlap at the window boundary can't leave duplicate ids (which + // would break timeline keys and work-log derivation). const seen = new Set(prev.map((activity) => activity.id)); + for (const activity of liveThreadActivities) { + seen.add(activity.id); + } const fresh = page.activities.filter((activity) => !seen.has(activity.id)); return [...fresh, ...prev]; }); @@ -2021,6 +2032,7 @@ function ChatViewContent(props: ChatViewProps) { activeThreadActivityRequestKey, hasMoreOlderActivities, threadActivities, + liveThreadActivities, loadThreadActivities, ]); From 28f9a953aee248083067125b8df28fa1fba43306 Mon Sep 17 00:00:00 2001 From: olafura Date: Mon, 22 Jun 2026 22:46:54 +0200 Subject: [PATCH 05/13] fix: harden lazy-load reset and dedup against window reshape (PR review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detect a live-window reshape from an order-independent oldest boundary (`liveWindowOldestActivityId`: min by createdAt/id) instead of `activities[0]`. The reducer sorts unsequenced rows to the end while the server snapshot lists legacy unsequenced rows first, so the first live append re-sorts the array and shifts index 0 — which previously looked like a reshape and wrongly cleared the user's scrolled-up history. Run the older-pages reset in useLayoutEffect so the cleared state commits before paint; otherwise a thread switch renders one frame with the previous thread's lazy-loaded pages still merged in, flashing stale work-log and approval rows. Applied to both web (ChatView) and mobile (use-thread-composer-state); the shared sentinel helper lives in client-runtime with unit tests. Co-Authored-By: Claude Opus 4.8 --- .../features/threads/ThreadRouteScreen.tsx | 4 +- .../src/state/use-selected-thread-requests.ts | 30 +++- .../src/state/use-thread-composer-state.ts | 129 +++++++++++++----- .../Layers/ProjectionSnapshotQuery.test.ts | 97 +++++++++++++ .../Layers/ProjectionSnapshotQuery.ts | 12 +- apps/web/src/components/ChatView.tsx | 120 +++++++++++----- .../src/components/chat/MessagesTimeline.tsx | 6 +- packages/client-runtime/package.json | 4 + .../src/state/threadReducer.test.ts | 84 +++++++++++- .../client-runtime/src/state/threadReducer.ts | 39 ++++++ 10 files changed, 448 insertions(+), 77 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index ae63c3406b2..ecd803ffe58 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -194,7 +194,9 @@ function ThreadRouteContent( const composer = useThreadComposerState(); const gitState = useSelectedThreadGitState(); const gitActions = useSelectedThreadGitActions(); - const requests = useSelectedThreadRequests(); + // Derive pending requests from the FULL loaded set (older pages + live + // window) so a prompt the user scrolled back to load still surfaces. + const requests = useSelectedThreadRequests(composer.mergedActivities); const interruptThreadTurn = useAtomCommand(threadEnvironment.interruptTurn, "thread interrupt"); const navigation = useNavigation(); const params = props.route.params; diff --git a/apps/mobile/src/state/use-selected-thread-requests.ts b/apps/mobile/src/state/use-selected-thread-requests.ts index c9e9db12530..8c934547c07 100644 --- a/apps/mobile/src/state/use-selected-thread-requests.ts +++ b/apps/mobile/src/state/use-selected-thread-requests.ts @@ -1,7 +1,11 @@ import { useAtomValue } from "@effect/atom-react"; import { useCallback, useMemo, useState } from "react"; -import { ApprovalRequestId, type ProviderApprovalDecision } from "@t3tools/contracts"; +import { + ApprovalRequestId, + type OrchestrationThreadActivity, + type ProviderApprovalDecision, +} from "@t3tools/contracts"; import { Atom } from "effect/unstable/reactivity"; import { threadEnvironment } from "../state/threads"; @@ -53,7 +57,20 @@ function setUserInputDraftCustomAnswer( }); } -export function useSelectedThreadRequests() { +/** + * Pending approval / user-input requests for the selected thread. + * + * `activities` should be the FULL loaded set (lazy-loaded older pages + the + * windowed live view, i.e. `useThreadComposerState().mergedActivities`): the + * detail snapshot windows activities to the most recent page, so deriving from + * `selectedThread.activities` alone would hide a prompt the user scrolled back + * to load. Falls back to the live window when not provided. Deriving from the + * merged set is sound — resolutions are always newer than their requests, so a + * loaded request whose resolution exists always has that resolution loaded too. + */ +export function useSelectedThreadRequests( + activities?: ReadonlyArray, +) { const respondToApproval = useAtomCommand( threadEnvironment.respondToApproval, "thread approval response", @@ -70,14 +87,15 @@ export function useSelectedThreadRequests() { null, ); + const requestActivities = activities ?? selectedThread?.activities ?? null; const activePendingApprovals = useMemo( - () => (selectedThread ? derivePendingApprovals(selectedThread.activities) : []), - [selectedThread], + () => (requestActivities ? derivePendingApprovals(requestActivities) : []), + [requestActivities], ); const activePendingApproval = activePendingApprovals[0] ?? null; const activePendingUserInputs = useMemo( - () => (selectedThread ? derivePendingUserInputs(selectedThread.activities) : []), - [selectedThread], + () => (requestActivities ? derivePendingUserInputs(requestActivities) : []), + [requestActivities], ); const activePendingUserInput = activePendingUserInputs[0] ?? null; const activePendingUserInputDrafts = diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 4e77fc668ab..2666434ba36 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -1,5 +1,5 @@ import { useAtomValue } from "@effect/atom-react"; -import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; import { CommandId, @@ -12,6 +12,11 @@ import { type ThreadId, } from "@t3tools/contracts"; import { safeErrorLogAttributes } from "@t3tools/client-runtime/errors"; +import { isAtomCommandInterrupted } from "@t3tools/client-runtime/state/runtime"; +import { + liveWindowOldestActivityId, + oldestActivityByChronology, +} from "@t3tools/client-runtime/state/thread-reducer"; import { deriveActiveWorkStartedAt } from "@t3tools/shared/orchestrationTiming"; import { makeQueuedMessageMetadata } from "../lib/commandMetadata"; @@ -111,40 +116,83 @@ export function useThreadComposerState() { const activityRequestKey = selectedThreadShell ? `${selectedThreadShell.environmentId}\u0000${selectedThreadShell.id}` : null; - const activityRequestKeyRef = useRef(activityRequestKey); - activityRequestKeyRef.current = activityRequestKey; - const liveActivities = selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES; - // The live window's oldest activity is stable while new activities only - // append; it changes when the window is re-snapshotted (reconnect) or rows are - // removed (checkpoint revert), either of which can make prepended older pages - // stale or gappy — so reset the lazy-load state when it (or the thread) changes. - const liveOldestActivityId = liveActivities[0]?.id ?? null; - useEffect(() => { + // Order-independent oldest boundary: `activities[0]` shifts when the reducer + // re-sorts unsequenced rows on the first live append, which would otherwise + // make a plain append look like a window reshape. See helper docs. + const liveOldestActivityId = useMemo( + () => liveWindowOldestActivityId(liveActivities), + [liveActivities], + ); + const liveActivityCount = liveActivities.length; + // Bumps on every lazy-load reset so a late in-flight load can't repopulate the + // freshly-cleared state (the thread key alone doesn't change on a same-thread + // window reshape). + const olderActivitiesGenRef = useRef(0); + // The request key of an in-flight older-history load — coalesces the duplicate + // dispatches the list fires before the loading state updates. + const inFlightOlderKeyRef = useRef(null); + // The oldest row we've paged past. Advancing this (not re-deriving from the + // merged set) lets an all-overlap page keep paging when the server still + // reports `hasMore`, without re-requesting the same cursor. Reset on reshape. + const olderCursorRef = useRef(null); + // Reset the lazy-loaded older pages when the live window is *reshaped* rather + // than purely appended-to: a different thread or a re-snapshot (reconnect) + // changes its oldest row, and a checkpoint revert removes rows so the count + // shrinks. A pure append (same thread, same oldest, larger count) keeps them. + const olderWindowRef = useRef({ + key: activityRequestKey, + oldest: liveOldestActivityId, + count: liveActivityCount, + }); + // useLayoutEffect (not useEffect) so the cleared state commits before the new + // thread paints; otherwise the previous thread's lazy-loaded pages stay merged + // in for one frame, flashing stale feed rows. + useLayoutEffect(() => { + const prev = olderWindowRef.current; + olderWindowRef.current = { + key: activityRequestKey, + oldest: liveOldestActivityId, + count: liveActivityCount, + }; + const reshaped = + activityRequestKey !== prev.key || + liveOldestActivityId !== prev.oldest || + liveActivityCount < prev.count; + if (!reshaped) { + return; + } + olderActivitiesGenRef.current += 1; + inFlightOlderKeyRef.current = null; + olderCursorRef.current = null; setOlderActivities([]); setOlderLoaded(false); setOlderHasMore(false); setLoadingOlderActivities(false); - }, [activityRequestKey, liveOldestActivityId]); + }, [activityRequestKey, liveOldestActivityId, liveActivityCount]); const mergedActivities = useMemo( () => olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities, [olderActivities, liveActivities], ); + // Latest merged set, read inside the async load handler so dedup runs against + // the current state, not the snapshot captured when the load was dispatched. + const mergedActivitiesRef = useRef(mergedActivities); + mergedActivitiesRef.current = mergedActivities; // Before any page is loaded, the server tells us whether older history exists. const hasMoreOlderActivities = olderLoaded ? olderHasMore : (selectedThreadDetail?.hasMoreActivities ?? false); - // Synchronous in-flight guard keyed by thread: the list fires onLoadOlder - // repeatedly while pinned at the top, but loading *state* only updates next - // render, so without this a fast scroll dispatches duplicate same-cursor calls. - const inFlightOlderKeyRef = useRef(null); const onLoadOlderActivities = useCallback(() => { if (!selectedThreadShell || !hasMoreOlderActivities) { return; } - const oldestActivity = mergedActivities[0]; + // Page from the explicit cursor (oldest row already paged past) or, before + // any page, the chronologically-oldest loaded row (matches the reshape + // sentinel): the reducer sorts unsequenced rows to the end, so index 0 can be + // a newer sequenced row whose cursor would skip older unsequenced history. + const oldestActivity = olderCursorRef.current ?? oldestActivityByChronology(mergedActivities); if (!oldestActivity || !activityRequestKey) { return; } @@ -156,6 +204,7 @@ export function useThreadComposerState() { ? { beforeSequence: oldestActivity.sequence } : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id }; const requestKey = activityRequestKey; + const gen = olderActivitiesGenRef.current; inFlightOlderKeyRef.current = requestKey; setLoadingOlderActivities(true); void loadThreadActivities({ @@ -163,32 +212,45 @@ export function useThreadComposerState() { input: { threadId: selectedThreadShell.id, ...cursorInput }, }) .then((result) => { - if (activityRequestKeyRef.current !== requestKey) { + // Window/thread reset while in flight — drop the page so it can't + // repopulate state cleared by the reset. + if (olderActivitiesGenRef.current !== gen) { return; } if (result._tag !== "Success") { + // Keep `hasMore` true — the history still exists and scrolling back to + // the top retries — but tell the user the fetch failed rather than + // silently showing a spinner that quietly gave up. + if (!isAtomCommandInterrupted(result)) { + setPendingConnectionError("Could not load older thread history."); + } return; } const page = result.value; - setOlderActivities((prev) => { - // Dedup against both already-loaded older pages and the live window, - // since mobile merges everything into one array (duplicate ids would - // produce duplicate React keys in the feed). - const seen = new Set(prev.map((activity) => activity.id)); - for (const activity of liveActivities) { - seen.add(activity.id); - } - const fresh = page.activities.filter((activity) => !seen.has(activity.id)); - return [...fresh, ...prev]; - }); + // Advance the cursor to this page's oldest row (pages are ascending) even + // if every row dedupes away — the server cursor is strict, so the cursor + // strictly decreases and paging can't loop, while an all-overlap page no + // longer terminates paging the server says has more. + const pageOldest = page.activities[0]; + if (pageOldest) { + olderCursorRef.current = pageOldest; + } + // Dedup against the LATEST merged set (via ref) so a live append or a + // prior prepend that settled mid-flight can't leave duplicate ids. + const seen = new Set(mergedActivitiesRef.current.map((activity) => activity.id)); + const fresh = page.activities.filter((activity) => !seen.has(activity.id)); + if (fresh.length === 0) { + setOlderLoaded(true); + setOlderHasMore(page.hasMore); + return; + } + setOlderActivities((prev) => [...fresh, ...prev]); setOlderLoaded(true); setOlderHasMore(page.hasMore); }) .finally(() => { - if (inFlightOlderKeyRef.current === requestKey) { + if (olderActivitiesGenRef.current === gen) { inFlightOlderKeyRef.current = null; - } - if (activityRequestKeyRef.current === requestKey) { setLoadingOlderActivities(false); } }); @@ -197,7 +259,6 @@ export function useThreadComposerState() { hasMoreOlderActivities, mergedActivities, activityRequestKey, - liveActivities, loadThreadActivities, ]); @@ -414,6 +475,10 @@ export function useThreadComposerState() { runtimeMode, interactionMode, activeThreadBusy, + // Lazy-loaded older pages + the live window — the full loaded activity set. + // Request derivations must run over this (not the windowed live set alone) + // so prompts pulled in by scroll-up still surface, matching web. + mergedActivities, hasMoreOlderActivities, loadingOlderActivities, onLoadOlderActivities, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 1a7c579922c..302dcf31ff9 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -1426,6 +1426,103 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); + it.effect( + "unsequenced cursor reaches all older rows without stranding sequenced ones", + () => + // Regression for the "unsequenced cursor hides sequenced history" concern: + // sequenced rows always sort newer than NULL-sequence (legacy) rows, so when + // the oldest loaded row is unsequenced every sequenced row is already in the + // window — the `sequence IS NULL` cursor can't strand sequenced rows. + Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_thread_activities`; + yield* sql`DELETE FROM projection_state`; + yield* sql` + INSERT INTO projection_projects ( + project_id, title, workspace_root, default_model_selection_json, + scripts_json, created_at, updated_at, deleted_at + ) VALUES ( + 'project-1', 'Project 1', '/tmp/project-1', + '{"provider":"codex","model":"gpt-5-codex"}', '[]', + '2026-04-01T00:00:00.000Z', '2026-04-01T00:00:01.000Z', NULL + ) + `; + yield* sql` + INSERT INTO projection_threads ( + thread_id, project_id, title, model_selection_json, runtime_mode, + interaction_mode, branch, worktree_path, latest_turn_id, + latest_user_message_at, pending_approval_count, pending_user_input_count, + has_actionable_proposed_plan, created_at, updated_at, archived_at, deleted_at + ) VALUES ( + 'thread-1', 'project-1', 'Thread 1', + '{"provider":"codex","model":"gpt-5-codex"}', 'full-access', 'default', + NULL, NULL, NULL, NULL, 0, 0, 0, + '2026-04-01T00:00:02.000Z', '2026-04-01T00:00:03.000Z', NULL, NULL + ) + `; + // 600 legacy unsequenced rows (older) + 3 sequenced rows (newer). The + // window keeps the 3 sequenced + the most-recent 497 unsequenced, so the + // oldest loaded row is unsequenced and 103 older unsequenced remain. + yield* Effect.forEach( + Array.from({ length: 600 }, (_u, index) => index + 1), + (n) => + sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, + sequence, created_at + ) VALUES ( + ${`unseq-${String(n).padStart(4, "0")}`}, 'thread-1', NULL, + 'info', 'runtime.note', ${`unseq-${n}`}, '{}', NULL, + ${`2026-04-01T00:00:01.${String(n).padStart(3, "0")}Z`} + ) + `, + { discard: true }, + ); + yield* Effect.forEach( + [1, 2, 3], + (seq) => + sql` + INSERT INTO projection_thread_activities ( + activity_id, thread_id, turn_id, tone, kind, summary, payload_json, + sequence, created_at + ) VALUES ( + ${`seq-${seq}`}, 'thread-1', NULL, 'info', 'runtime.note', + ${`seq-${seq}`}, '{}', ${seq}, ${`2026-04-01T09:00:0${seq}.000Z`} + ) + `, + { discard: true }, + ); + + const detail = yield* snapshotQuery.getThreadDetailById(ThreadId.make("thread-1")); + assert.equal(detail._tag, "Some"); + if (detail._tag !== "Some") return; + const windowed = detail.value.activities; + assert.equal(windowed.length, 500); + // Sequenced rows are the newest (end of the ascending window); the oldest + // loaded row is unsequenced — exactly the case the concern is about. + assert.equal(windowed.at(-1)?.summary, "seq-3"); + assert.equal(windowed[0]?.sequence, undefined); + + // The client pages with the unsequenced cursor of the oldest loaded row. + const oldest = windowed[0]; + assert.ok(oldest); + const olderPage = yield* snapshotQuery.getThreadActivitiesPage({ + threadId: ThreadId.make("thread-1"), + beforeCreatedAt: oldest.createdAt, + beforeActivityId: oldest.id, + limit: 500, + }); + // The 103 older unsequenced rows come back, none are sequenced, and no + // sequenced row was stranded (all 3 are already in the window). + assert.equal(olderPage.activities.length, 103); + assert.equal(olderPage.hasMore, false); + assert.ok(olderPage.activities.every((a) => a.sequence === undefined)); + }), + ); + it.effect("uses projection_threads.latest_turn_id for bulk command and shell snapshots", () => Effect.gen(function* () { const snapshotQuery = yield* ProjectionSnapshotQuery; diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index bd33cad20e2..7881bd7b3ad 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -126,16 +126,22 @@ const ThreadIdLookupInput = Schema.Struct({ */ const THREAD_DETAIL_ACTIVITY_WINDOW = 500; +// `beforeSequence`/`limit` are NonNegativeInt (not bare Number) to match the +// contract: the WHERE clause `(sequence < beforeSequence OR sequence IS NULL)` +// is only equivalent to the old `COALESCE(sequence, -1) < beforeSequence` when +// `beforeSequence` is non-negative — a negative cursor would silently match no +// sequenced rows and return only unsequenced ones. Validating here (not just at +// the RPC boundary) keeps any future non-RPC caller honest. const ThreadActivitiesBeforeSequenceInput = Schema.Struct({ threadId: ThreadId, - beforeSequence: Schema.Number, - limit: Schema.Number, + beforeSequence: NonNegativeInt, + limit: NonNegativeInt, }); const ThreadActivitiesBeforeActivityInput = Schema.Struct({ threadId: ThreadId, beforeCreatedAt: IsoDateTime, beforeActivityId: EventId, - limit: Schema.Number, + limit: NonNegativeInt, }); const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema; const ProjectionThreadIdLookupRowSchema = Schema.Struct({ diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index bb36e9ab9ca..ea1d2629cf3 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -64,6 +64,10 @@ import { squashAtomCommandFailure, type AtomCommandResult, } from "@t3tools/client-runtime/state/runtime"; +import { + liveWindowOldestActivityId, + oldestActivityByChronology, +} from "@t3tools/client-runtime/state/thread-reducer"; import * as Cause from "effect/Cause"; import { AsyncResult } from "effect/unstable/reactivity"; import { isElectron } from "../env"; @@ -1943,20 +1947,61 @@ function ChatViewContent(props: ChatViewProps) { const activeThreadActivityRequestKey = activeThread ? `${activeThread.environmentId}\u0000${activeThread.id}` : null; - const activeThreadActivityRequestKeyRef = useRef(activeThreadActivityRequestKey); - activeThreadActivityRequestKeyRef.current = activeThreadActivityRequestKey; const liveThreadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; - // The live window's oldest activity is stable while new activities only - // append; it changes when the window is re-snapshotted (reconnect) or rows are - // removed (checkpoint revert), either of which can make prepended older pages - // stale or gappy — so reset the lazy-load state when it (or the thread) changes. - const liveOldestActivityId = liveThreadActivities[0]?.id ?? null; - useEffect(() => { + // Order-independent oldest boundary: `activities[0]` shifts when the reducer + // re-sorts unsequenced rows on the first live append, which would otherwise + // make a plain append look like a window reshape. See helper docs. + const liveOldestActivityId = useMemo( + () => liveWindowOldestActivityId(liveThreadActivities), + [liveThreadActivities], + ); + const liveActivityCount = liveThreadActivities.length; + // Bumps on every lazy-load reset so a late in-flight load can't repopulate the + // freshly-cleared state (the thread key alone doesn't change on a same-thread + // window reshape). + const olderActivitiesGenRef = useRef(0); + // The request key of an in-flight older-history load — coalesces the duplicate + // dispatches a fast scroll-to-top fires before the loading state updates. + const inFlightOlderKeyRef = useRef(null); + // The oldest row we've paged past. Advancing this (not re-deriving from the + // merged set) lets an all-overlap page keep paging when the server still + // reports `hasMore` — without it a page that dedupes to nothing would either + // terminate paging early or re-request the same cursor forever. Reset on reshape. + const olderCursorRef = useRef(null); + // Reset the lazy-loaded older pages when the live window is *reshaped* rather + // than purely appended-to: a different thread or a re-snapshot (reconnect) + // changes its oldest row, and a checkpoint revert removes rows so the count + // shrinks. A pure append (same thread, same oldest, larger count) keeps them. + const olderWindowRef = useRef({ + key: activeThreadActivityRequestKey, + oldest: liveOldestActivityId, + count: liveActivityCount, + }); + // useLayoutEffect (not useEffect) so the cleared state commits before paint: + // otherwise the new thread renders one frame with the previous thread's + // lazy-loaded pages still merged in, flashing stale work-log/approval rows. + useLayoutEffect(() => { + const prev = olderWindowRef.current; + olderWindowRef.current = { + key: activeThreadActivityRequestKey, + oldest: liveOldestActivityId, + count: liveActivityCount, + }; + const reshaped = + activeThreadActivityRequestKey !== prev.key || + liveOldestActivityId !== prev.oldest || + liveActivityCount < prev.count; + if (!reshaped) { + return; + } + olderActivitiesGenRef.current += 1; + inFlightOlderKeyRef.current = null; + olderCursorRef.current = null; setOlderActivities([]); setOlderLoaded(false); setOlderHasMore(false); setLoadingOlderActivities(false); - }, [activeThreadActivityRequestKey, liveOldestActivityId]); + }, [activeThreadActivityRequestKey, liveOldestActivityId, liveActivityCount]); const threadActivities = useMemo( () => @@ -1965,21 +2010,25 @@ function ChatViewContent(props: ChatViewProps) { : liveThreadActivities, [olderActivities, liveThreadActivities], ); + // Latest merged set, read inside the async load handler so dedup runs against + // the current state, not the snapshot captured when the load was dispatched. + const threadActivitiesRef = useRef(threadActivities); + threadActivitiesRef.current = threadActivities; // Before any page is loaded, the server tells us whether older history exists // beyond the windowed snapshot; afterwards the page `hasMore` is authoritative. const hasMoreOlderActivities = olderLoaded ? olderHasMore : (activeThread?.hasMoreActivities ?? false); - // Tracks the request key of an in-flight older-history load. The scroll - // handler fires onLoadOlder on every frame while at the top, but the loading - // *state* only updates on the next render — without a synchronous guard a fast - // scroll-to-top dispatches several duplicate requests for the same cursor. - const inFlightOlderKeyRef = useRef(null); const loadOlderActivities = useCallback(() => { if (!activeThread || !hasMoreOlderActivities) { return; } - const oldestActivity = threadActivities[0]; + // Page from the explicit cursor (the oldest row we've already paged past) or, + // before any page, the chronologically-oldest loaded row — not + // `threadActivities[0]`: the reducer sorts unsequenced rows to the end, so + // index 0 can be a newer sequenced row whose `beforeSequence` cursor would + // skip older unsequenced history. This matches the reshape sentinel. + const oldestActivity = olderCursorRef.current ?? oldestActivityByChronology(threadActivities); if (!oldestActivity || !activeThreadActivityRequestKey) { return; } @@ -1991,6 +2040,7 @@ function ChatViewContent(props: ChatViewProps) { ? { beforeSequence: oldestActivity.sequence } : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id }; const requestKey = activeThreadActivityRequestKey; + const gen = olderActivitiesGenRef.current; inFlightOlderKeyRef.current = requestKey; setLoadingOlderActivities(true); void loadThreadActivities({ @@ -1998,32 +2048,37 @@ function ChatViewContent(props: ChatViewProps) { input: { threadId: activeThread.id, ...cursorInput }, }) .then((result) => { - if (activeThreadActivityRequestKeyRef.current !== requestKey) { + // The window/thread was reset while this was in flight — drop the page + // so it can't repopulate state cleared by the reset. + if (olderActivitiesGenRef.current !== gen) { return; } if (result._tag !== "Success") { return; } const page = result.value; - setOlderActivities((prev) => { - // Dedup against both already-loaded older pages and the live window so - // an overlap at the window boundary can't leave duplicate ids (which - // would break timeline keys and work-log derivation). - const seen = new Set(prev.map((activity) => activity.id)); - for (const activity of liveThreadActivities) { - seen.add(activity.id); - } - const fresh = page.activities.filter((activity) => !seen.has(activity.id)); - return [...fresh, ...prev]; - }); + // Advance the cursor to this page's oldest row (pages are ascending, so + // [0] is oldest) even if every row dedupes away — the server cursor is + // strict, so the cursor strictly decreases and paging can't loop, while + // an all-overlap page no longer terminates paging that the server says + // has more. + const pageOldest = page.activities[0]; + if (pageOldest) { + olderCursorRef.current = pageOldest; + } + // Dedup against the LATEST merged set (via ref) so a live append or a + // prior prepend that settled mid-flight can't leave duplicate ids. + const seen = new Set(threadActivitiesRef.current.map((activity) => activity.id)); + const fresh = page.activities.filter((activity) => !seen.has(activity.id)); + if (fresh.length > 0) { + setOlderActivities((prev) => [...fresh, ...prev]); + } setOlderLoaded(true); setOlderHasMore(page.hasMore); }) .finally(() => { - if (inFlightOlderKeyRef.current === requestKey) { + if (olderActivitiesGenRef.current === gen) { inFlightOlderKeyRef.current = null; - } - if (activeThreadActivityRequestKeyRef.current === requestKey) { setLoadingOlderActivities(false); } }); @@ -2032,7 +2087,6 @@ function ChatViewContent(props: ChatViewProps) { activeThreadActivityRequestKey, hasMoreOlderActivities, threadActivities, - liveThreadActivities, loadThreadActivities, ]); @@ -5938,7 +5992,9 @@ function ChatViewContent(props: ChatViewProps) { activeProject?.defaultModelSelection } activeThreadModelSelection={activeThread?.modelSelection} - activeThreadActivities={activeThread?.activities} + // Use the merged set: the latest context-window + // update can live in a lazy-loaded page. + activeThreadActivities={threadActivities} resolvedTheme={resolvedTheme} settings={settings} keybindings={keybindings} diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 139a3f5908c..fd299242bf6 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -501,7 +501,11 @@ export const MessagesTimeline = memo(function MessagesTimeline({ [], ); - if (rows.length === 0 && !isWorking) { + // Only short-circuit to the empty state when there is genuinely nothing to + // fetch: the window can derive zero VISIBLE rows (e.g. only tool-neutral work + // entries) while older history still exists — the list must render then so + // its "Load older history" header stays reachable. + if (rows.length === 0 && !isWorking && !hasMoreOlder && !loadingOlder) { if (hideEmptyPlaceholder) { return null; } diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index 4fa05f850e5..19682c4d5f6 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -123,6 +123,10 @@ "types": "./src/state/threads.ts", "default": "./src/state/threads.ts" }, + "./state/thread-reducer": { + "types": "./src/state/threadReducer.ts", + "default": "./src/state/threadReducer.ts" + }, "./state/thread-sort": { "types": "./src/state/threadSort.ts", "default": "./src/state/threadSort.ts" diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 211f8748f4e..99dc63ed031 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -9,9 +9,29 @@ import { ThreadId, TurnId, } from "@t3tools/contracts"; -import type { OrchestrationThread } from "@t3tools/contracts"; +import type { OrchestrationThread, OrchestrationThreadActivity } from "@t3tools/contracts"; -import { applyThreadDetailEvent } from "./threadReducer.ts"; +import { + applyThreadDetailEvent, + liveWindowOldestActivityId, + oldestActivityByChronology, +} from "./threadReducer.ts"; + +const activity = ( + id: string, + createdAt: string, + sequence?: number, +): OrchestrationThreadActivity => + ({ + id, + tone: "tool", + kind: "command", + summary: id, + payload: {}, + turnId: TurnId.make("turn-1"), + createdAt, + ...(sequence !== undefined ? { sequence } : {}), + }) as unknown as OrchestrationThreadActivity; const baseEventFields = { eventId: EventId.make("event-1"), @@ -755,4 +775,64 @@ describe("applyThreadDetailEvent", () => { expect(result.kind).toBe("unchanged"); }); }); + + describe("liveWindowOldestActivityId", () => { + it("returns null for an empty window", () => { + expect(liveWindowOldestActivityId([])).toBeNull(); + }); + + it("returns the chronologically-oldest id regardless of array position", () => { + // Reducer order places the unsequenced legacy row (oldest) LAST while the + // server snapshot would list it first; the helper picks it either way. + const window = [ + activity("seq-1", "2026-04-01T10:00:01.000Z", 1), + activity("seq-2", "2026-04-01T10:00:02.000Z", 2), + activity("legacy", "2026-04-01T09:00:00.000Z"), + ]; + expect(liveWindowOldestActivityId(window)).toBe("legacy"); + }); + + it("breaks createdAt ties by id", () => { + const window = [ + activity("b", "2026-04-01T10:00:00.000Z", 2), + activity("a", "2026-04-01T10:00:00.000Z", 1), + ]; + expect(liveWindowOldestActivityId(window)).toBe("a"); + }); + + it("is stable when a newer activity is appended (no false reshape)", () => { + const before = [ + activity("legacy", "2026-04-01T09:00:00.000Z"), + activity("seq-1", "2026-04-01T10:00:01.000Z", 1), + ]; + // A live append is the newest activity and is unsequenced in the payload; + // it must not change the detected oldest boundary. + const after = [ + ...before, + activity("appended", "2026-04-01T11:00:00.000Z"), + ]; + expect(liveWindowOldestActivityId(after)).toBe(liveWindowOldestActivityId(before)); + expect(liveWindowOldestActivityId(after)).toBe("legacy"); + }); + }); + + describe("oldestActivityByChronology", () => { + it("returns null for an empty set", () => { + expect(oldestActivityByChronology([])).toBeNull(); + }); + + it("returns the unsequenced legacy row so the pagination cursor agrees with the sentinel", () => { + // The reducer placed the legacy (unsequenced, oldest) row at the END; paging + // must cursor from it (createdAt cursor), not from index 0's sequenced row. + const merged = [ + activity("seq-5", "2026-04-01T10:00:05.000Z", 5), + activity("seq-6", "2026-04-01T10:00:06.000Z", 6), + activity("legacy", "2026-04-01T09:00:00.000Z"), + ]; + const oldest = oldestActivityByChronology(merged); + expect(oldest?.id).toBe("legacy"); + expect(oldest?.sequence).toBeUndefined(); // → drives the unsequenced cursor + expect(liveWindowOldestActivityId(merged)).toBe(oldest?.id); // sentinel agrees + }); + }); }); diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index ce0dca52f5a..ee6c580418f 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -35,6 +35,45 @@ const activityOrder = O.combineAll([ O.mapInput(O.String, (a) => a.id), ]); +/** + * The oldest activity in a set, by chronology (`createdAt`, then `id`) rather + * than array position. + * + * `activities[0]` is not a stable "oldest": {@link activityOrder} sorts + * unsequenced rows to the end (a missing `sequence` is treated as newest) while + * the server snapshot lists legacy unsequenced rows first, so the first live + * append re-sorts the array and shifts index 0. Both the lazy-load *reshape* + * sentinel ({@link liveWindowOldestActivityId}) and the lazy-load *pagination + * cursor* derive from this so they agree on which row is oldest regardless of + * the reducer's placement of unsequenced rows. Returns `null` when empty. + */ +export function oldestActivityByChronology( + activities: ReadonlyArray, +): OrchestrationThreadActivity | null { + let oldest: OrchestrationThreadActivity | null = null; + for (const activity of activities) { + if ( + oldest === null || + activity.createdAt < oldest.createdAt || + (activity.createdAt === oldest.createdAt && activity.id < oldest.id) + ) { + oldest = activity; + } + } + return oldest; +} + +/** + * The id of {@link oldestActivityByChronology}, used as the lazy-load reshape + * sentinel (a reconnect re-snapshot or checkpoint revert changes it; a plain + * append does not). Returns `null` when empty. + */ +export function liveWindowOldestActivityId( + activities: ReadonlyArray, +): OrchestrationThreadActivity["id"] | null { + return oldestActivityByChronology(activities)?.id ?? null; +} + /** * Apply a single orchestration event to an `OrchestrationThread`, returning * the updated thread, a deletion signal, or an "unchanged" marker when the From f1b0ed6c3cd9b89a5ad590309c269a5b8bcd1265 Mon Sep 17 00:00:00 2001 From: olafura Date: Thu, 2 Jul 2026 20:49:42 +0200 Subject: [PATCH 06/13] refactor(client): extract the shared older-history lazy-load engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The lazy-load state machine (~160 lines each) was hand-copied in the web ChatView and the mobile composer, and every review-round fix had to be applied to both — half the review churn on this PR was the copies drifting. One shared hook, `useOlderThreadActivities` in client-runtime, now owns all of the hardened behavior: reshape-sentinel reset, generation guard, synchronous in-flight coalescing, the advancing cursor (all-overlap pages keep paging), dedup against the latest merged set, and hasMore-preserving failure handling. Clients pass a `loadPage` transport wrapper and keep only their platform scroll trigger and error surfacing. The decision kernel (reshape detection, cursor selection, page dedup) is exported as pure functions with unit tests; react becomes an optional peer dependency of client-runtime, scoped to the new subpath. Co-Authored-By: Claude Opus 4.8 --- .../src/features/threads/ThreadFeed.tsx | 18 ++ .../src/state/use-selected-thread-requests.ts | 7 +- .../src/state/use-thread-composer-state.ts | 198 +++---------- apps/web/src/components/ChatView.tsx | 187 +++--------- packages/client-runtime/package.json | 14 + .../src/state/olderThreadActivities.test.ts | Bin 0 -> 3825 bytes .../src/state/olderThreadActivities.ts | 271 ++++++++++++++++++ 7 files changed, 381 insertions(+), 314 deletions(-) create mode 100644 packages/client-runtime/src/state/olderThreadActivities.test.ts create mode 100644 packages/client-runtime/src/state/olderThreadActivities.ts diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 4b5b352b9ef..19686858070 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -1820,7 +1820,25 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { }} /> + {props.feed.length === 0 && hasMoreOlder ? ( + // The window can derive zero visible entries while older history + // exists — without scrollable content `onStartReached` can never + // fire, so give the user an explicit affordance instead of the + // empty-state placeholder. + + + {loadingOlder ? ( + + ) : ( + onLoadOlder?.()}> + Load older history + + )} + + + ) : null} {props.feed.length === 0 && + !hasMoreOlder && props.activeWorkStartedAt === null && props.contentPresentation.kind === "ready" ? ( diff --git a/apps/mobile/src/state/use-selected-thread-requests.ts b/apps/mobile/src/state/use-selected-thread-requests.ts index 8c934547c07..2e52c780b9c 100644 --- a/apps/mobile/src/state/use-selected-thread-requests.ts +++ b/apps/mobile/src/state/use-selected-thread-requests.ts @@ -92,12 +92,15 @@ export function useSelectedThreadRequests( () => (requestActivities ? derivePendingApprovals(requestActivities) : []), [requestActivities], ); - const activePendingApproval = activePendingApprovals[0] ?? null; + // The derivations sort ascending by createdAt; surface the NEWEST open + // request. With lazy-loaded older pages in the set, index 0 could be an + // ancient dangling request hijacking the prompt for the current one. + const activePendingApproval = activePendingApprovals.at(-1) ?? null; const activePendingUserInputs = useMemo( () => (requestActivities ? derivePendingUserInputs(requestActivities) : []), [requestActivities], ); - const activePendingUserInput = activePendingUserInputs[0] ?? null; + const activePendingUserInput = activePendingUserInputs.at(-1) ?? null; const activePendingUserInputDrafts = activePendingUserInput && selectedThreadShell ? (userInputDraftsByRequestKey[ diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 2666434ba36..50a07a1f328 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -1,5 +1,5 @@ import { useAtomValue } from "@effect/atom-react"; -import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from "react"; +import { useCallback, useEffect, useMemo, useState } from "react"; import { CommandId, @@ -14,9 +14,9 @@ import { import { safeErrorLogAttributes } from "@t3tools/client-runtime/errors"; import { isAtomCommandInterrupted } from "@t3tools/client-runtime/state/runtime"; import { - liveWindowOldestActivityId, - oldestActivityByChronology, -} from "@t3tools/client-runtime/state/thread-reducer"; + useOlderThreadActivities, + type OlderActivitiesCursor, +} from "@t3tools/client-runtime/state/older-thread-activities"; import { deriveActiveWorkStartedAt } from "@t3tools/shared/orchestrationTiming"; import { makeQueuedMessageMetadata } from "../lib/commandMetadata"; @@ -100,167 +100,49 @@ export function useThreadComposerState() { [queuedMessagesByThreadKey, selectedThreadKey], ); - // ── Older-history lazy-load (mirrors web ChatView) ────────────────────────── + // ── Older-history lazy-load (shared engine; see useOlderThreadActivities) ── // The detail snapshot windows activities to the most recent page (the server // sets `hasMoreActivities`); older pages are fetched on demand and prepended. - const [olderActivities, setOlderActivities] = useState< - ReadonlyArray - >([]); - const [olderLoaded, setOlderLoaded] = useState(false); - const [olderHasMore, setOlderHasMore] = useState(false); - const [loadingOlderActivities, setLoadingOlderActivities] = useState(false); const loadThreadActivities = useAtomCommand(orchestrationEnvironment.loadThreadActivities, { reportFailure: false, }); - - const activityRequestKey = selectedThreadShell - ? `${selectedThreadShell.environmentId}\u0000${selectedThreadShell.id}` - : null; - const liveActivities = selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES; - // Order-independent oldest boundary: `activities[0]` shifts when the reducer - // re-sorts unsequenced rows on the first live append, which would otherwise - // make a plain append look like a window reshape. See helper docs. - const liveOldestActivityId = useMemo( - () => liveWindowOldestActivityId(liveActivities), - [liveActivities], - ); - const liveActivityCount = liveActivities.length; - // Bumps on every lazy-load reset so a late in-flight load can't repopulate the - // freshly-cleared state (the thread key alone doesn't change on a same-thread - // window reshape). - const olderActivitiesGenRef = useRef(0); - // The request key of an in-flight older-history load — coalesces the duplicate - // dispatches the list fires before the loading state updates. - const inFlightOlderKeyRef = useRef(null); - // The oldest row we've paged past. Advancing this (not re-deriving from the - // merged set) lets an all-overlap page keep paging when the server still - // reports `hasMore`, without re-requesting the same cursor. Reset on reshape. - const olderCursorRef = useRef(null); - // Reset the lazy-loaded older pages when the live window is *reshaped* rather - // than purely appended-to: a different thread or a re-snapshot (reconnect) - // changes its oldest row, and a checkpoint revert removes rows so the count - // shrinks. A pure append (same thread, same oldest, larger count) keeps them. - const olderWindowRef = useRef({ - key: activityRequestKey, - oldest: liveOldestActivityId, - count: liveActivityCount, - }); - // useLayoutEffect (not useEffect) so the cleared state commits before the new - // thread paints; otherwise the previous thread's lazy-loaded pages stay merged - // in for one frame, flashing stale feed rows. - useLayoutEffect(() => { - const prev = olderWindowRef.current; - olderWindowRef.current = { - key: activityRequestKey, - oldest: liveOldestActivityId, - count: liveActivityCount, - }; - const reshaped = - activityRequestKey !== prev.key || - liveOldestActivityId !== prev.oldest || - liveActivityCount < prev.count; - if (!reshaped) { - return; - } - olderActivitiesGenRef.current += 1; - inFlightOlderKeyRef.current = null; - olderCursorRef.current = null; - setOlderActivities([]); - setOlderLoaded(false); - setOlderHasMore(false); - setLoadingOlderActivities(false); - }, [activityRequestKey, liveOldestActivityId, liveActivityCount]); - const mergedActivities = useMemo( - () => - olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities, - [olderActivities, liveActivities], - ); - // Latest merged set, read inside the async load handler so dedup runs against - // the current state, not the snapshot captured when the load was dispatched. - const mergedActivitiesRef = useRef(mergedActivities); - mergedActivitiesRef.current = mergedActivities; - // Before any page is loaded, the server tells us whether older history exists. - const hasMoreOlderActivities = olderLoaded - ? olderHasMore - : (selectedThreadDetail?.hasMoreActivities ?? false); - - const onLoadOlderActivities = useCallback(() => { - if (!selectedThreadShell || !hasMoreOlderActivities) { - return; - } - // Page from the explicit cursor (oldest row already paged past) or, before - // any page, the chronologically-oldest loaded row (matches the reshape - // sentinel): the reducer sorts unsequenced rows to the end, so index 0 can be - // a newer sequenced row whose cursor would skip older unsequenced history. - const oldestActivity = olderCursorRef.current ?? oldestActivityByChronology(mergedActivities); - if (!oldestActivity || !activityRequestKey) { - return; - } - if (inFlightOlderKeyRef.current === activityRequestKey) { - return; - } - const cursorInput = - oldestActivity.sequence !== undefined - ? { beforeSequence: oldestActivity.sequence } - : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id }; - const requestKey = activityRequestKey; - const gen = olderActivitiesGenRef.current; - inFlightOlderKeyRef.current = requestKey; - setLoadingOlderActivities(true); - void loadThreadActivities({ - environmentId: selectedThreadShell.environmentId, - input: { threadId: selectedThreadShell.id, ...cursorInput }, - }) - .then((result) => { - // Window/thread reset while in flight — drop the page so it can't - // repopulate state cleared by the reset. - if (olderActivitiesGenRef.current !== gen) { - return; - } - if (result._tag !== "Success") { - // Keep `hasMore` true — the history still exists and scrolling back to - // the top retries — but tell the user the fetch failed rather than - // silently showing a spinner that quietly gave up. - if (!isAtomCommandInterrupted(result)) { - setPendingConnectionError("Could not load older thread history."); - } - return; - } - const page = result.value; - // Advance the cursor to this page's oldest row (pages are ascending) even - // if every row dedupes away — the server cursor is strict, so the cursor - // strictly decreases and paging can't loop, while an all-overlap page no - // longer terminates paging the server says has more. - const pageOldest = page.activities[0]; - if (pageOldest) { - olderCursorRef.current = pageOldest; - } - // Dedup against the LATEST merged set (via ref) so a live append or a - // prior prepend that settled mid-flight can't leave duplicate ids. - const seen = new Set(mergedActivitiesRef.current.map((activity) => activity.id)); - const fresh = page.activities.filter((activity) => !seen.has(activity.id)); - if (fresh.length === 0) { - setOlderLoaded(true); - setOlderHasMore(page.hasMore); - return; - } - setOlderActivities((prev) => [...fresh, ...prev]); - setOlderLoaded(true); - setOlderHasMore(page.hasMore); - }) - .finally(() => { - if (olderActivitiesGenRef.current === gen) { - inFlightOlderKeyRef.current = null; - setLoadingOlderActivities(false); - } + const selectedEnvironmentIdForActivities = selectedThreadShell?.environmentId ?? null; + const selectedThreadIdForActivities = selectedThreadShell?.id ?? null; + const loadOlderActivitiesPage = useCallback( + async (cursor: OlderActivitiesCursor) => { + if (selectedEnvironmentIdForActivities === null || selectedThreadIdForActivities === null) { + return null; + } + const result = await loadThreadActivities({ + environmentId: selectedEnvironmentIdForActivities, + input: { threadId: selectedThreadIdForActivities, ...cursor }, }); - }, [ - selectedThreadShell, - hasMoreOlderActivities, + if (result._tag !== "Success") { + // Surface real failures (a spinner that quietly gives up reads as + // missing history); keep `hasMore` so scrolling back retries. + if (!isAtomCommandInterrupted(result)) { + setPendingConnectionError("Could not load older thread history."); + } + return null; + } + return result.value; + }, + [selectedEnvironmentIdForActivities, selectedThreadIdForActivities, loadThreadActivities], + ); + const { mergedActivities, - activityRequestKey, - loadThreadActivities, - ]); + hasMoreOlder: hasMoreOlderActivities, + loadingOlder: loadingOlderActivities, + loadOlder: onLoadOlderActivities, + } = useOlderThreadActivities({ + threadKey: selectedThreadShell + ? `${selectedThreadShell.environmentId}\u0000${selectedThreadShell.id}` + : null, + liveActivities: selectedThreadDetail?.activities ?? EMPTY_ACTIVITIES, + hasMoreLiveActivities: selectedThreadDetail?.hasMoreActivities ?? false, + loadPage: loadOlderActivitiesPage, + }); + const selectedThreadFeed = useMemo( () => diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index ea1d2629cf3..9166fb0c258 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -65,9 +65,9 @@ import { type AtomCommandResult, } from "@t3tools/client-runtime/state/runtime"; import { - liveWindowOldestActivityId, - oldestActivityByChronology, -} from "@t3tools/client-runtime/state/thread-reducer"; + useOlderThreadActivities, + type OlderActivitiesCursor, +} from "@t3tools/client-runtime/state/older-thread-activities"; import * as Cause from "effect/Cause"; import { AsyncResult } from "effect/unstable/reactivity"; import { isElectron } from "../env"; @@ -1933,162 +1933,41 @@ function ChatViewContent(props: ChatViewProps) { // ── Older-history lazy-load ──────────────────────────────────────────────── // The detail snapshot windows activities to the most recent page (the server // sets `hasMoreActivities` when older ones exist); older pages are fetched on - // demand (infinite scroll-up) and prepended. Messages aren't windowed - // server-side, so this just back-fills the older tool activity. - const [olderActivities, setOlderActivities] = useState< - ReadonlyArray - >([]); - const [olderLoaded, setOlderLoaded] = useState(false); - const [olderHasMore, setOlderHasMore] = useState(false); - const [loadingOlderActivities, setLoadingOlderActivities] = useState(false); + // demand (infinite scroll-up) and prepended by the shared engine. Messages + // aren't windowed server-side, so this just back-fills the older tool + // activity. const loadThreadActivities = useAtomCommand(orchestrationEnvironment.loadThreadActivities, { reportFailure: false, }); - const activeThreadActivityRequestKey = activeThread - ? `${activeThread.environmentId}\u0000${activeThread.id}` - : null; - const liveThreadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; - // Order-independent oldest boundary: `activities[0]` shifts when the reducer - // re-sorts unsequenced rows on the first live append, which would otherwise - // make a plain append look like a window reshape. See helper docs. - const liveOldestActivityId = useMemo( - () => liveWindowOldestActivityId(liveThreadActivities), - [liveThreadActivities], + const activeThreadEnvironmentIdForActivities = activeThread?.environmentId ?? null; + const activeThreadIdForActivities = activeThread?.id ?? null; + const loadOlderActivitiesPage = useCallback( + async (cursor: OlderActivitiesCursor) => { + if (activeThreadEnvironmentIdForActivities === null || activeThreadIdForActivities === null) { + return null; + } + const result = await loadThreadActivities({ + environmentId: activeThreadEnvironmentIdForActivities, + input: { threadId: activeThreadIdForActivities, ...cursor }, + }); + // Failures stay silent on web (the "Load older history" affordance itself + // is the retry surface); returning null keeps `hasMore` for the retry. + return result._tag === "Success" ? result.value : null; + }, + [activeThreadEnvironmentIdForActivities, activeThreadIdForActivities, loadThreadActivities], ); - const liveActivityCount = liveThreadActivities.length; - // Bumps on every lazy-load reset so a late in-flight load can't repopulate the - // freshly-cleared state (the thread key alone doesn't change on a same-thread - // window reshape). - const olderActivitiesGenRef = useRef(0); - // The request key of an in-flight older-history load — coalesces the duplicate - // dispatches a fast scroll-to-top fires before the loading state updates. - const inFlightOlderKeyRef = useRef(null); - // The oldest row we've paged past. Advancing this (not re-deriving from the - // merged set) lets an all-overlap page keep paging when the server still - // reports `hasMore` — without it a page that dedupes to nothing would either - // terminate paging early or re-request the same cursor forever. Reset on reshape. - const olderCursorRef = useRef(null); - // Reset the lazy-loaded older pages when the live window is *reshaped* rather - // than purely appended-to: a different thread or a re-snapshot (reconnect) - // changes its oldest row, and a checkpoint revert removes rows so the count - // shrinks. A pure append (same thread, same oldest, larger count) keeps them. - const olderWindowRef = useRef({ - key: activeThreadActivityRequestKey, - oldest: liveOldestActivityId, - count: liveActivityCount, + const { + mergedActivities: threadActivities, + hasMoreOlder: hasMoreOlderActivities, + loadingOlder: loadingOlderActivities, + loadOlder: loadOlderActivities, + } = useOlderThreadActivities({ + threadKey: activeThread ? `${activeThread.environmentId}\u0000${activeThread.id}` : null, + liveActivities: activeThread?.activities ?? EMPTY_ACTIVITIES, + hasMoreLiveActivities: activeThread?.hasMoreActivities ?? false, + loadPage: loadOlderActivitiesPage, }); - // useLayoutEffect (not useEffect) so the cleared state commits before paint: - // otherwise the new thread renders one frame with the previous thread's - // lazy-loaded pages still merged in, flashing stale work-log/approval rows. - useLayoutEffect(() => { - const prev = olderWindowRef.current; - olderWindowRef.current = { - key: activeThreadActivityRequestKey, - oldest: liveOldestActivityId, - count: liveActivityCount, - }; - const reshaped = - activeThreadActivityRequestKey !== prev.key || - liveOldestActivityId !== prev.oldest || - liveActivityCount < prev.count; - if (!reshaped) { - return; - } - olderActivitiesGenRef.current += 1; - inFlightOlderKeyRef.current = null; - olderCursorRef.current = null; - setOlderActivities([]); - setOlderLoaded(false); - setOlderHasMore(false); - setLoadingOlderActivities(false); - }, [activeThreadActivityRequestKey, liveOldestActivityId, liveActivityCount]); - - const threadActivities = useMemo( - () => - olderActivities.length > 0 - ? [...olderActivities, ...liveThreadActivities] - : liveThreadActivities, - [olderActivities, liveThreadActivities], - ); - // Latest merged set, read inside the async load handler so dedup runs against - // the current state, not the snapshot captured when the load was dispatched. - const threadActivitiesRef = useRef(threadActivities); - threadActivitiesRef.current = threadActivities; - // Before any page is loaded, the server tells us whether older history exists - // beyond the windowed snapshot; afterwards the page `hasMore` is authoritative. - const hasMoreOlderActivities = olderLoaded - ? olderHasMore - : (activeThread?.hasMoreActivities ?? false); - const loadOlderActivities = useCallback(() => { - if (!activeThread || !hasMoreOlderActivities) { - return; - } - // Page from the explicit cursor (the oldest row we've already paged past) or, - // before any page, the chronologically-oldest loaded row — not - // `threadActivities[0]`: the reducer sorts unsequenced rows to the end, so - // index 0 can be a newer sequenced row whose `beforeSequence` cursor would - // skip older unsequenced history. This matches the reshape sentinel. - const oldestActivity = olderCursorRef.current ?? oldestActivityByChronology(threadActivities); - if (!oldestActivity || !activeThreadActivityRequestKey) { - return; - } - if (inFlightOlderKeyRef.current === activeThreadActivityRequestKey) { - return; // a load for this thread is already in flight - } - const cursorInput = - oldestActivity.sequence !== undefined - ? { beforeSequence: oldestActivity.sequence } - : { beforeCreatedAt: oldestActivity.createdAt, beforeActivityId: oldestActivity.id }; - const requestKey = activeThreadActivityRequestKey; - const gen = olderActivitiesGenRef.current; - inFlightOlderKeyRef.current = requestKey; - setLoadingOlderActivities(true); - void loadThreadActivities({ - environmentId: activeThread.environmentId, - input: { threadId: activeThread.id, ...cursorInput }, - }) - .then((result) => { - // The window/thread was reset while this was in flight — drop the page - // so it can't repopulate state cleared by the reset. - if (olderActivitiesGenRef.current !== gen) { - return; - } - if (result._tag !== "Success") { - return; - } - const page = result.value; - // Advance the cursor to this page's oldest row (pages are ascending, so - // [0] is oldest) even if every row dedupes away — the server cursor is - // strict, so the cursor strictly decreases and paging can't loop, while - // an all-overlap page no longer terminates paging that the server says - // has more. - const pageOldest = page.activities[0]; - if (pageOldest) { - olderCursorRef.current = pageOldest; - } - // Dedup against the LATEST merged set (via ref) so a live append or a - // prior prepend that settled mid-flight can't leave duplicate ids. - const seen = new Set(threadActivitiesRef.current.map((activity) => activity.id)); - const fresh = page.activities.filter((activity) => !seen.has(activity.id)); - if (fresh.length > 0) { - setOlderActivities((prev) => [...fresh, ...prev]); - } - setOlderLoaded(true); - setOlderHasMore(page.hasMore); - }) - .finally(() => { - if (olderActivitiesGenRef.current === gen) { - inFlightOlderKeyRef.current = null; - setLoadingOlderActivities(false); - } - }); - }, [ - activeThread, - activeThreadActivityRequestKey, - hasMoreOlderActivities, - threadActivities, - loadThreadActivities, - ]); + const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); const pendingApprovals = useMemo( diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index 19682c4d5f6..8c54f53ed16 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -127,6 +127,10 @@ "types": "./src/state/threadReducer.ts", "default": "./src/state/threadReducer.ts" }, + "./state/older-thread-activities": { + "types": "./src/state/olderThreadActivities.ts", + "default": "./src/state/olderThreadActivities.ts" + }, "./state/thread-sort": { "types": "./src/state/threadSort.ts", "default": "./src/state/threadSort.ts" @@ -149,8 +153,18 @@ "@t3tools/shared": "workspace:*", "effect": "catalog:" }, + "peerDependencies": { + "react": "^19.0.0" + }, + "peerDependenciesMeta": { + "react": { + "optional": true + } + }, "devDependencies": { "@effect/vitest": "catalog:", + "@types/react": "~19.2.14", + "react": "19.2.6", "vite-plus": "catalog:" } } diff --git a/packages/client-runtime/src/state/olderThreadActivities.test.ts b/packages/client-runtime/src/state/olderThreadActivities.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..041b5bd142434c311a6900301c0ced9918f847a3 GIT binary patch literal 3825 zcmb_f;cnYD5Z>=R#ZA9tz;;x(SXUQr*A*RBplgOKK!FwcK}#2luuQ5XWydHC>>>7q zdy?Iel58oCn+*YiK%&Sy-gozXcb2PC8wXz@Lz@~o$06WtiK!ccbg+XPqgN2ErNdEK zRQBAVbs24V|&_6W9RD`Kqj*fa*dy)%Jk+EZ7xb2()x{|=ZhjkQ`?kid0pYH>n*>njMe5+ zZ}_|J9tq!m(PntoU@$)SdpZ{;<6|G;8HH;dP`n|I8N_5%W>cU*rCP9ylp=MQy>W+& z7Qa-eQhYrHRjuYo4hPed5Y1kZt(Y-j2W%L?X@ygW82KoyED0D|N%d+aR8}wAir&n2 z+MX-Lw$Oq$e%;lEd^V>LvN~RgCB~7X#nF>!5Vo{ogxB$S95+G0_p=#PD#IJ8FoV~W z?Do?R*o(KZ8-TF1Z>jV~@#Be#pi*lpGmefWJsLthfY~cxL_!wk!XnJb`4TDC2-W&K z=gW+O82U_G%0VRjYpN?nZ%<)It_V>VM*_;qAc(R=l|gL93fYv;za?V$Bd{=fV+SBD ze7v{<0bF=ggq{09O{VzR&11gjkgiD(hR6CY&tNwgjGg`&;~P;}9K7^N4!k^Nimlu| zxuf%KWB~}U8|l)VFaIbPzXS9n+Q;Bb_3#i5(C|i8RTM`|bz(9|c~_)al!K8~qO`eo z5HnqglawY%bD1AsV|JNu*XQtsrGdXfgI5nl^MUGnO@r?*F4YdeOr*d2xvA z{b9XrZb2M6NK5y42)M~ZU0d7t+}H$r_E`wo;QthXK!|2 zItn*vdSTY(Gil0aQa8u_sX*!8v2J+&%ylcyrNJ9C7C1>%T7^t$0%LY$qf(HNrG8!k zt3goLEQ*mP_7|cI*Xvrs&lBmOl&Ren7Um*d(n#ZhNOPl=F7!gC^ldu|umVpllq?dEA&qu~ z1JJyKw~+MWK0k*mg41ADrL>9WJ;#$rk1(|_>h?p1w>AYiIpYTT@!h9)m%b>-T<0j{ zNQ1&ZfBju2ht$KU94^`tNfZPX;VpmsHSAZU1>hpCy7knhk zaSMBoF~^4I4xry)iXZ6wRCPb6?+_&%n)GU~tIs{AWJXi1&uMbYXq;_%;M<8P_y@(7 zlIZKFE|j|>7ee7mb?{tU6wSOb$IMQ-~w>nSjdI_jXL5K zcz~+4uEW7&>GIrbONR#dGLDgciEaIi?4Ov5%*D{_K1W = []; + +/** + * Pagination cursor for a thread's older activities. Sequenced rows page by + * `beforeSequence`; legacy/unsequenced rows (the common case — `sequence` is + * absent on most real rows) page by the `(createdAt, activityId)` keyset. + */ +export type OlderActivitiesCursor = + | { readonly beforeSequence: number } + | { + readonly beforeCreatedAt: OrchestrationThreadActivity["createdAt"]; + readonly beforeActivityId: OrchestrationThreadActivity["id"]; + }; + +export interface OlderActivitiesPage { + readonly activities: ReadonlyArray; + readonly hasMore: boolean; +} + +export interface UseOlderThreadActivitiesOptions { + /** + * Identity of the thread the live window belongs to (e.g. + * `${environmentId}\0${threadId}`); null when no thread is selected. + * Changing it resets the lazy-loaded pages. + */ + readonly threadKey: string | null; + /** The server-windowed live activity set from the thread detail. */ + readonly liveActivities: ReadonlyArray; + /** The server's `hasMoreActivities` flag from the detail snapshot. */ + readonly hasMoreLiveActivities: boolean; + /** + * Fetch the page immediately older than the cursor. Resolve `null` to skip + * the page silently (a failure the caller already surfaced, or an + * interrupted command) — `hasMore` is left true so the user can retry. + * MUST be referentially stable (useCallback) for the load callback to be. + */ + readonly loadPage: (cursor: OlderActivitiesCursor) => Promise; +} + +export interface UseOlderThreadActivitiesResult { + /** Lazy-loaded older pages + the live window, oldest first. */ + readonly mergedActivities: ReadonlyArray; + /** Whether older history exists beyond everything loaded. */ + readonly hasMoreOlder: boolean; + readonly loadingOlder: boolean; + /** Dispatch a load of the next older page (no-op while one is in flight). */ + readonly loadOlder: () => void; +} + +// ── Pure decision kernel (exported for unit tests) ────────────────────────── + +export interface LiveWindowShape { + readonly key: string | null; + /** Chronological-oldest activity id (an identity sentinel, not a lookup key). */ + readonly oldest: string | null; + readonly count: number; +} + +/** + * Whether the live window was RESHAPED rather than purely appended-to: a + * different thread, a re-snapshot (reconnect) that changes the window's + * chronological-oldest row, or a checkpoint revert that shrinks it. A pure + * append (same thread, same oldest, count not smaller) is NOT a reshape. + */ +export function didLiveWindowReshape(previous: LiveWindowShape, next: LiveWindowShape): boolean { + return ( + next.key !== previous.key || next.oldest !== previous.oldest || next.count < previous.count + ); +} + +/** + * The cursor for the page immediately older than `oldest`: sequenced rows page + * by `beforeSequence`; unsequenced rows (the common case) by the + * `(createdAt, activityId)` keyset. + */ +export function olderActivitiesCursorFor( + oldest: OrchestrationThreadActivity, +): OlderActivitiesCursor { + return oldest.sequence !== undefined + ? { beforeSequence: oldest.sequence } + : { beforeCreatedAt: oldest.createdAt, beforeActivityId: oldest.id }; +} + +/** + * The row the NEXT load should cursor from: the explicit cursor row already + * paged past when one exists (so an all-overlap page keeps advancing), else + * the chronologically-oldest loaded row — never index 0, which the reducer + * can fill with a newer row (unsequenced rows sort to the end). + */ +export function nextOlderActivitiesCursorRow( + pagedPast: OrchestrationThreadActivity | null, + merged: ReadonlyArray, +): OrchestrationThreadActivity | null { + return pagedPast ?? oldestActivityByChronology(merged); +} + +/** + * The page rows not already present in the loaded set (older pages + live + * window) — boundary overlap and mid-flight appends must never produce + * duplicate ids in the merged timeline. + */ +export function freshOlderActivities( + page: OlderActivitiesPage, + merged: ReadonlyArray, +): ReadonlyArray { + const seen = new Set(merged.map((activity) => activity.id)); + return page.activities.filter((activity) => !seen.has(activity.id)); +} + +/** + * The older-history lazy-load engine, shared by every client (web ChatView, + * the mobile composer, the TUI ChatView). The thread-detail snapshot windows + * activities to the most recent page; older pages are fetched on demand and + * prepended. + * + * One implementation holds all the hardening the per-client copies kept + * drifting on: + * - reset on live-window RESHAPE, not just thread switch: a reconnect + * re-snapshot changes the window's chronological-oldest row and a checkpoint + * revert shrinks it, but a plain append does neither (the reducer re-sorts + * unsequenced rows, so index 0 is not a stable boundary — the sentinel is + * {@link liveWindowOldestActivityId}); + * - a generation guard so a load resolving after a reset can't repopulate the + * cleared state; + * - a synchronous in-flight key so scroll-triggered duplicate dispatches + * coalesce before the loading state commits; + * - an explicit advancing cursor (the oldest row paged PAST), so an + * all-overlap page keeps paging instead of dead-ending while the server + * still reports more — the server cursor is strict, so it strictly + * decreases and paging cannot loop; + * - dedup against the LATEST merged set via a ref, so a live append or a + * prior prepend settling mid-flight can't produce duplicate ids; + * - `hasMore` stays true on a failed/skipped page (the history still exists; + * scrolling back retries). + */ +export function useOlderThreadActivities( + options: UseOlderThreadActivitiesOptions, +): UseOlderThreadActivitiesResult { + const { threadKey, liveActivities, hasMoreLiveActivities, loadPage } = options; + + const [olderActivities, setOlderActivities] = useState< + ReadonlyArray + >([]); + const [olderLoaded, setOlderLoaded] = useState(false); + const [olderHasMore, setOlderHasMore] = useState(false); + const [loadingOlder, setLoadingOlder] = useState(false); + + // Order-independent oldest boundary: `liveActivities[0]` shifts when the + // reducer re-sorts unsequenced rows on the first live append, which would + // otherwise make a plain append look like a window reshape. + const liveOldestActivityId = useMemo( + () => liveWindowOldestActivityId(liveActivities), + [liveActivities], + ); + const liveActivityCount = liveActivities.length; + + // Bumps on every reset so a late in-flight load can't repopulate the + // freshly-cleared state (the thread key alone doesn't change on a + // same-thread window reshape). + const generationRef = useRef(0); + // The thread key of an in-flight load — coalesces the duplicate dispatches a + // fast scroll fires before the loading state updates. + const inFlightKeyRef = useRef(null); + // The oldest row we've paged past; advances even when a page dedupes to + // nothing. Reset on reshape. + const cursorRef = useRef(null); + const windowRef = useRef({ + key: threadKey, + oldest: liveOldestActivityId, + count: liveActivityCount, + }); + + // useLayoutEffect (not useEffect) so the cleared state commits before paint: + // otherwise a thread switch renders one frame with the previous thread's + // lazy-loaded pages still merged in, flashing stale rows. + useLayoutEffect(() => { + const previous = windowRef.current; + windowRef.current = { + key: threadKey, + oldest: liveOldestActivityId, + count: liveActivityCount, + }; + if (!didLiveWindowReshape(previous, windowRef.current)) { + return; + } + generationRef.current += 1; + inFlightKeyRef.current = null; + cursorRef.current = null; + setOlderActivities([]); + setOlderLoaded(false); + setOlderHasMore(false); + setLoadingOlder(false); + }, [threadKey, liveOldestActivityId, liveActivityCount]); + + const mergedActivities = useMemo( + () => + olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities, + [olderActivities, liveActivities], + ); + // Latest merged set, read inside the async load handler so dedup runs + // against current state, not the snapshot captured at dispatch time. + const mergedActivitiesRef = useRef(mergedActivities); + mergedActivitiesRef.current = mergedActivities; + + // Before any page is loaded the server flag is authoritative; afterwards + // the latest page's `hasMore` is. + const hasMoreOlder = olderLoaded + ? olderHasMore + : threadKey !== null && hasMoreLiveActivities; + + const loadOlder = useCallback(() => { + if (threadKey === null || !hasMoreOlder) { + return; + } + const oldest = nextOlderActivitiesCursorRow(cursorRef.current, mergedActivitiesRef.current); + if (!oldest) { + return; + } + if (inFlightKeyRef.current === threadKey) { + return; // a load for this thread is already in flight + } + const cursor = olderActivitiesCursorFor(oldest); + const generation = generationRef.current; + inFlightKeyRef.current = threadKey; + setLoadingOlder(true); + void loadPage(cursor) + .then((page) => { + // The window/thread was reset while this was in flight — drop the page + // so it can't repopulate state cleared by the reset. + if (generationRef.current !== generation) { + return; + } + if (page === null) { + // Failed or interrupted (already surfaced by the caller). Keep + // `hasMore` — the history still exists and retrying is valid. + return; + } + // Advance the cursor even when every row dedupes away — the server + // cursor is strict, so it strictly decreases and paging can't loop. + const pageOldest = page.activities[0]; + if (pageOldest) { + cursorRef.current = pageOldest; + } + const fresh = freshOlderActivities(page, mergedActivitiesRef.current); + if (fresh.length > 0) { + setOlderActivities((previous) => [...fresh, ...previous]); + } + setOlderLoaded(true); + setOlderHasMore(page.hasMore); + }) + .finally(() => { + if (generationRef.current === generation) { + inFlightKeyRef.current = null; + setLoadingOlder(false); + } + }); + }, [threadKey, hasMoreOlder, loadPage]); + + return { + mergedActivities: threadKey === null ? EMPTY_ACTIVITIES : mergedActivities, + hasMoreOlder, + loadingOlder, + loadOlder, + }; +} From 4739847ee2d116fa153cb383bd95b676a6d37cce Mon Sep 17 00:00:00 2001 From: olafura Date: Thu, 16 Jul 2026 15:22:09 +0200 Subject: [PATCH 07/13] fix(server): bound subscription event catch-up --- .../Services/OrchestrationEngine.ts | 5 +- apps/server/src/server.test.ts | 46 +++++++++++++ apps/server/src/ws.ts | 67 +++++++++++++++++-- 3 files changed, 110 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts index f8bcfd76ac0..b224887e465 100644 --- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts +++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts @@ -27,9 +27,8 @@ export interface OrchestrationEngineShape { * * @param fromSequenceExclusive - Sequence cursor (exclusive). * @param limit - Maximum number of events to read. Defaults to the event - * store's page-bounded default; pass a higher value when the caller must - * read every event after the cursor (e.g. per-thread catch-up that filters - * a small subset out of a potentially larger global range). + * store's page-bounded default. Callers must keep this bounded; use a + * projection snapshot instead of replaying an arbitrarily stale cursor. * @returns Stream containing ordered events. */ readonly readEvents: ( diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 871b79eca90..5c55ae68966 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5890,6 +5890,52 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("subscribeThread replaces a stale cursor with a fresh snapshot", () => + Effect.gen(function* () { + const snapshotSequence = 5_000; + const thread = makeDefaultOrchestrationReadModel().threads[0]!; + let replayCalls = 0; + + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence }), + getThreadDetailSnapshot: () => + Effect.succeed( + Option.some({ + snapshotSequence, + thread, + }), + ), + }, + orchestrationEngine: { + readEvents: () => { + replayCalls += 1; + return Stream.empty; + }, + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 1, + }).pipe(Stream.take(1), Stream.runCollect), + ), + ); + + assert.equal(replayCalls, 0); + assert.equal(result[0]?.kind, "snapshot"); + if (result[0]?.kind === "snapshot") { + assert.equal(result[0].snapshot.snapshotSequence, snapshotSequence); + assert.equal(result[0].snapshot.thread.id, defaultThreadId); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("subscribeShell coalesces a per-thread burst without stalling other threads", () => Effect.gen(function* () { const busyThreadId = ThreadId.make("thread-busy"); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4146b0c955b..88c880a30c5 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -117,6 +117,24 @@ import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; + +const ORCHESTRATION_SUBSCRIPTION_REPLAY_LIMIT = 1_000; + +const subscriptionReplayLimit = ( + afterSequence: number, + snapshotSequence: number, +): number | null => { + const eventCount = snapshotSequence - afterSequence; + if ( + !Number.isSafeInteger(eventCount) || + eventCount < 0 || + eventCount > ORCHESTRATION_SUBSCRIPTION_REPLAY_LIMIT + ) { + return null; + } + return eventCount; +}; + const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); @@ -1401,14 +1419,53 @@ const makeWsRpcLayer = ( // catch-up followed by the buffered/ongoing live events. Overlapping // events are deduped by sequence on the client. // - // Read the full range after the cursor (not the store's default - // page-bounded limit): the range is normally tiny (a fresh HTTP - // snapshot sequence) and the per-thread filter runs after reading, - // so a global cap could otherwise omit this thread's events. + // A recent cursor replays the exact global range through the + // per-thread filter. A stale cursor receives a current thread + // snapshot instead, keeping catch-up bounded even when the global + // event log is very large. if (input.afterSequence !== undefined) { const afterSequence = input.afterSequence; + const { snapshotSequence } = yield* projectionSnapshotQuery + .getSnapshotSequence() + .pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to load orchestration snapshot sequence", + cause, + }), + ), + ); + const replayLimit = subscriptionReplayLimit(afterSequence, snapshotSequence); + + if (replayLimit === null) { + const snapshot = yield* projectionSnapshotQuery + .getThreadDetailSnapshot(input.threadId) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to load thread ${input.threadId}`, + cause, + }), + ), + ); + + if (Option.isNone(snapshot)) { + return yield* new OrchestrationGetSnapshotError({ + message: `Thread ${input.threadId} was not found`, + cause: input.threadId, + }); + } + + return Stream.concat( + Stream.make({ kind: "snapshot" as const, snapshot: snapshot.value }), + bufferedLiveStream, + ); + } + const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) + .readEvents(afterSequence, replayLimit) .pipe( Stream.filter(isThisThreadDetailEvent), Stream.map((event) => ({ kind: "event" as const, event })), From 506bdff6917f306764100613fab9575dce116665 Mon Sep 17 00:00:00 2001 From: olafura Date: Thu, 16 Jul 2026 15:39:35 +0200 Subject: [PATCH 08/13] chore: format rebased pagination changes --- .../src/features/threads/ThreadFeed.tsx | 1 - .../src/state/use-selected-thread-requests.ts | 4 +- .../src/state/use-thread-composer-state.ts | 3 +- .../Layers/ProjectionSnapshotQuery.test.ts | 110 +++++++++--------- apps/web/src/components/ChatView.tsx | 1 - .../src/components/chat/MessagesTimeline.tsx | 10 +- packages/client-runtime/package.json | 12 +- .../src/state/olderThreadActivities.test.ts | Bin 3825 -> 3816 bytes .../src/state/olderThreadActivities.ts | 7 +- .../src/state/threadReducer.test.ts | 11 +- 10 files changed, 75 insertions(+), 84 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 19686858070..ea910141269 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -1511,7 +1511,6 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { } }, [hasMoreOlder, loadingOlder, onLoadOlder]); - useEffect(() => { const previous = previousLatestTurnRef.current; previousLatestTurnRef.current = props.latestTurn; diff --git a/apps/mobile/src/state/use-selected-thread-requests.ts b/apps/mobile/src/state/use-selected-thread-requests.ts index 2e52c780b9c..0386c4e9574 100644 --- a/apps/mobile/src/state/use-selected-thread-requests.ts +++ b/apps/mobile/src/state/use-selected-thread-requests.ts @@ -68,9 +68,7 @@ function setUserInputDraftCustomAnswer( * merged set is sound — resolutions are always newer than their requests, so a * loaded request whose resolution exists always has that resolution loaded too. */ -export function useSelectedThreadRequests( - activities?: ReadonlyArray, -) { +export function useSelectedThreadRequests(activities?: ReadonlyArray) { const respondToApproval = useAtomCommand( threadEnvironment.respondToApproval, "thread approval response", diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 50a07a1f328..be83fe3ea31 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -1,5 +1,5 @@ import { useAtomValue } from "@effect/atom-react"; -import { useCallback, useEffect, useMemo, useState } from "react"; +import { useCallback, useEffect, useMemo } from "react"; import { CommandId, @@ -143,7 +143,6 @@ export function useThreadComposerState() { loadPage: loadOlderActivitiesPage, }); - const selectedThreadFeed = useMemo( () => selectedThreadDetail diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 302dcf31ff9..85242f16ed6 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -1426,21 +1426,19 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { }), ); - it.effect( - "unsequenced cursor reaches all older rows without stranding sequenced ones", - () => - // Regression for the "unsequenced cursor hides sequenced history" concern: - // sequenced rows always sort newer than NULL-sequence (legacy) rows, so when - // the oldest loaded row is unsequenced every sequenced row is already in the - // window — the `sequence IS NULL` cursor can't strand sequenced rows. - Effect.gen(function* () { - const snapshotQuery = yield* ProjectionSnapshotQuery; - const sql = yield* SqlClient.SqlClient; - yield* sql`DELETE FROM projection_projects`; - yield* sql`DELETE FROM projection_threads`; - yield* sql`DELETE FROM projection_thread_activities`; - yield* sql`DELETE FROM projection_state`; - yield* sql` + it.effect("unsequenced cursor reaches all older rows without stranding sequenced ones", () => + // Regression for the "unsequenced cursor hides sequenced history" concern: + // sequenced rows always sort newer than NULL-sequence (legacy) rows, so when + // the oldest loaded row is unsequenced every sequenced row is already in the + // window — the `sequence IS NULL` cursor can't strand sequenced rows. + Effect.gen(function* () { + const snapshotQuery = yield* ProjectionSnapshotQuery; + const sql = yield* SqlClient.SqlClient; + yield* sql`DELETE FROM projection_projects`; + yield* sql`DELETE FROM projection_threads`; + yield* sql`DELETE FROM projection_thread_activities`; + yield* sql`DELETE FROM projection_state`; + yield* sql` INSERT INTO projection_projects ( project_id, title, workspace_root, default_model_selection_json, scripts_json, created_at, updated_at, deleted_at @@ -1450,7 +1448,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { '2026-04-01T00:00:00.000Z', '2026-04-01T00:00:01.000Z', NULL ) `; - yield* sql` + yield* sql` INSERT INTO projection_threads ( thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, branch, worktree_path, latest_turn_id, @@ -1463,13 +1461,13 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { '2026-04-01T00:00:02.000Z', '2026-04-01T00:00:03.000Z', NULL, NULL ) `; - // 600 legacy unsequenced rows (older) + 3 sequenced rows (newer). The - // window keeps the 3 sequenced + the most-recent 497 unsequenced, so the - // oldest loaded row is unsequenced and 103 older unsequenced remain. - yield* Effect.forEach( - Array.from({ length: 600 }, (_u, index) => index + 1), - (n) => - sql` + // 600 legacy unsequenced rows (older) + 3 sequenced rows (newer). The + // window keeps the 3 sequenced + the most-recent 497 unsequenced, so the + // oldest loaded row is unsequenced and 103 older unsequenced remain. + yield* Effect.forEach( + Array.from({ length: 600 }, (_u, index) => index + 1), + (n) => + sql` INSERT INTO projection_thread_activities ( activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at @@ -1479,12 +1477,12 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ${`2026-04-01T00:00:01.${String(n).padStart(3, "0")}Z`} ) `, - { discard: true }, - ); - yield* Effect.forEach( - [1, 2, 3], - (seq) => - sql` + { discard: true }, + ); + yield* Effect.forEach( + [1, 2, 3], + (seq) => + sql` INSERT INTO projection_thread_activities ( activity_id, thread_id, turn_id, tone, kind, summary, payload_json, sequence, created_at @@ -1493,34 +1491,34 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { ${`seq-${seq}`}, '{}', ${seq}, ${`2026-04-01T09:00:0${seq}.000Z`} ) `, - { discard: true }, - ); + { discard: true }, + ); - const detail = yield* snapshotQuery.getThreadDetailById(ThreadId.make("thread-1")); - assert.equal(detail._tag, "Some"); - if (detail._tag !== "Some") return; - const windowed = detail.value.activities; - assert.equal(windowed.length, 500); - // Sequenced rows are the newest (end of the ascending window); the oldest - // loaded row is unsequenced — exactly the case the concern is about. - assert.equal(windowed.at(-1)?.summary, "seq-3"); - assert.equal(windowed[0]?.sequence, undefined); - - // The client pages with the unsequenced cursor of the oldest loaded row. - const oldest = windowed[0]; - assert.ok(oldest); - const olderPage = yield* snapshotQuery.getThreadActivitiesPage({ - threadId: ThreadId.make("thread-1"), - beforeCreatedAt: oldest.createdAt, - beforeActivityId: oldest.id, - limit: 500, - }); - // The 103 older unsequenced rows come back, none are sequenced, and no - // sequenced row was stranded (all 3 are already in the window). - assert.equal(olderPage.activities.length, 103); - assert.equal(olderPage.hasMore, false); - assert.ok(olderPage.activities.every((a) => a.sequence === undefined)); - }), + const detail = yield* snapshotQuery.getThreadDetailById(ThreadId.make("thread-1")); + assert.equal(detail._tag, "Some"); + if (detail._tag !== "Some") return; + const windowed = detail.value.activities; + assert.equal(windowed.length, 500); + // Sequenced rows are the newest (end of the ascending window); the oldest + // loaded row is unsequenced — exactly the case the concern is about. + assert.equal(windowed.at(-1)?.summary, "seq-3"); + assert.equal(windowed[0]?.sequence, undefined); + + // The client pages with the unsequenced cursor of the oldest loaded row. + const oldest = windowed[0]; + assert.ok(oldest); + const olderPage = yield* snapshotQuery.getThreadActivitiesPage({ + threadId: ThreadId.make("thread-1"), + beforeCreatedAt: oldest.createdAt, + beforeActivityId: oldest.id, + limit: 500, + }); + // The 103 older unsequenced rows come back, none are sequenced, and no + // sequenced row was stranded (all 3 are already in the window). + assert.equal(olderPage.activities.length, 103); + assert.equal(olderPage.hasMore, false); + assert.ok(olderPage.activities.every((a) => a.sequence === undefined)); + }), ); it.effect("uses projection_threads.latest_turn_id for bulk command and shell snapshots", () => diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 9166fb0c258..76dbec3cedb 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -1968,7 +1968,6 @@ function ChatViewContent(props: ChatViewProps) { loadPage: loadOlderActivitiesPage, }); - const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); const pendingApprovals = useMemo( () => derivePendingApprovals(threadActivities), diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index fd299242bf6..26d3c3dce1e 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -395,7 +395,15 @@ export const MessagesTimeline = memo(function MessagesTimeline({ strip.dataset.inView = inView ? "true" : "false"; } - }, [listRef, minimapItems, minimapStripMap, onIsAtEndChange, hasMoreOlder, loadingOlder, onLoadOlder]); + }, [ + listRef, + minimapItems, + minimapStripMap, + onIsAtEndChange, + hasMoreOlder, + loadingOlder, + onLoadOlder, + ]); useEffect(() => { const frame = requestAnimationFrame(handleScroll); diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index 8c54f53ed16..718bae7d84e 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -153,6 +153,12 @@ "@t3tools/shared": "workspace:*", "effect": "catalog:" }, + "devDependencies": { + "@effect/vitest": "catalog:", + "@types/react": "~19.2.14", + "react": "19.2.6", + "vite-plus": "catalog:" + }, "peerDependencies": { "react": "^19.0.0" }, @@ -160,11 +166,5 @@ "react": { "optional": true } - }, - "devDependencies": { - "@effect/vitest": "catalog:", - "@types/react": "~19.2.14", - "react": "19.2.6", - "vite-plus": "catalog:" } } diff --git a/packages/client-runtime/src/state/olderThreadActivities.test.ts b/packages/client-runtime/src/state/olderThreadActivities.test.ts index 041b5bd142434c311a6900301c0ced9918f847a3..756a12cbfbcfe17f9fd3c21ea4a9039bfd789683 100644 GIT binary patch delta 61 zcmew;`$Bd@6r(|Aij_ifNl|8Ax{g9}QEFmIYKmhCLZ~>kurxI - olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities, + () => (olderActivities.length > 0 ? [...olderActivities, ...liveActivities] : liveActivities), [olderActivities, liveActivities], ); // Latest merged set, read inside the async load handler so dedup runs @@ -210,9 +209,7 @@ export function useOlderThreadActivities( // Before any page is loaded the server flag is authoritative; afterwards // the latest page's `hasMore` is. - const hasMoreOlder = olderLoaded - ? olderHasMore - : threadKey !== null && hasMoreLiveActivities; + const hasMoreOlder = olderLoaded ? olderHasMore : threadKey !== null && hasMoreLiveActivities; const loadOlder = useCallback(() => { if (threadKey === null || !hasMoreOlder) { diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 99dc63ed031..2df8efd4405 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -17,11 +17,7 @@ import { oldestActivityByChronology, } from "./threadReducer.ts"; -const activity = ( - id: string, - createdAt: string, - sequence?: number, -): OrchestrationThreadActivity => +const activity = (id: string, createdAt: string, sequence?: number): OrchestrationThreadActivity => ({ id, tone: "tool", @@ -807,10 +803,7 @@ describe("applyThreadDetailEvent", () => { ]; // A live append is the newest activity and is unsequenced in the payload; // it must not change the detected oldest boundary. - const after = [ - ...before, - activity("appended", "2026-04-01T11:00:00.000Z"), - ]; + const after = [...before, activity("appended", "2026-04-01T11:00:00.000Z")]; expect(liveWindowOldestActivityId(after)).toBe(liveWindowOldestActivityId(before)); expect(liveWindowOldestActivityId(after)).toBe("legacy"); }); From cfdd0ffd3c7d8f82349f1e8660198ac0890e9fa5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 20 Jul 2026 17:45:45 +0200 Subject: [PATCH 09/13] fix(web): bound older-history auto-loading Co-authored-by: codex --- .../chat/MessagesTimeline.logic.test.ts | 60 +++++++++++++++++++ .../components/chat/MessagesTimeline.logic.ts | 36 +++++++++++ .../src/components/chat/MessagesTimeline.tsx | 31 ++++++++-- .../src/state/olderThreadActivities.ts | 4 +- 4 files changed, 124 insertions(+), 7 deletions(-) diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 6d74204bc1c..75f6d5472c4 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -5,8 +5,68 @@ import { deriveMessagesTimelineRows, normalizeCompactToolLabel, resolveAssistantMessageCopyState, + resolveOlderHistoryAutoLoad, } from "./MessagesTimeline.logic"; +describe("resolveOlderHistoryAutoLoad", () => { + it("dispatches only once while a failed or successful load leaves the viewport at the top", () => { + const first = resolveOlderHistoryAutoLoad({ + isAtStart: true, + hasMoreOlder: true, + loadingOlder: false, + requestedAtStart: false, + }); + expect(first).toEqual({ requestedAtStart: true, shouldLoad: true }); + + expect( + resolveOlderHistoryAutoLoad({ + isAtStart: true, + hasMoreOlder: true, + loadingOlder: false, + requestedAtStart: first.requestedAtStart, + }), + ).toEqual({ requestedAtStart: true, shouldLoad: false }); + expect( + resolveOlderHistoryAutoLoad({ + isAtStart: true, + hasMoreOlder: true, + loadingOlder: true, + requestedAtStart: first.requestedAtStart, + }), + ).toEqual({ requestedAtStart: true, shouldLoad: false }); + }); + + it("rearms automatic loading only after leaving the top", () => { + const away = resolveOlderHistoryAutoLoad({ + isAtStart: false, + hasMoreOlder: true, + loadingOlder: false, + requestedAtStart: true, + }); + expect(away).toEqual({ requestedAtStart: false, shouldLoad: false }); + + expect( + resolveOlderHistoryAutoLoad({ + isAtStart: true, + hasMoreOlder: true, + loadingOlder: false, + requestedAtStart: away.requestedAtStart, + }), + ).toEqual({ requestedAtStart: true, shouldLoad: true }); + }); + + it("does not reset the latch before list state is available", () => { + expect( + resolveOlderHistoryAutoLoad({ + isAtStart: undefined, + hasMoreOlder: true, + loadingOlder: false, + requestedAtStart: true, + }), + ).toEqual({ requestedAtStart: true, shouldLoad: false }); + }); +}); + describe("computeMessageDurationStart", () => { it("returns message createdAt when there is no preceding user message", () => { const result = computeMessageDurationStart([ diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.ts b/apps/web/src/components/chat/MessagesTimeline.logic.ts index 3227bac2413..ecb19933ca0 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -21,6 +21,42 @@ export interface TimelineEndState { readonly isNearEnd?: boolean; } +export interface OlderHistoryAutoLoadInput { + readonly isAtStart: boolean | undefined; + readonly hasMoreOlder: boolean; + readonly loadingOlder: boolean; + readonly requestedAtStart: boolean; +} + +export interface OlderHistoryAutoLoadResult { + readonly requestedAtStart: boolean; + readonly shouldLoad: boolean; +} + +/** + * Permit at most one automatic older-history request per visit to the top. + * Loading completion and prepended rows both retrigger timeline measurements; + * neither should recursively request another page while the viewport remains + * pinned at the start. Leaving the top rearms automatic loading, while the + * explicit header control remains available for an immediate retry. + */ +export function resolveOlderHistoryAutoLoad( + input: OlderHistoryAutoLoadInput, +): OlderHistoryAutoLoadResult { + if (input.isAtStart === false) { + return { requestedAtStart: false, shouldLoad: false }; + } + if ( + input.isAtStart !== true || + input.requestedAtStart || + !input.hasMoreOlder || + input.loadingOlder + ) { + return { requestedAtStart: input.requestedAtStart, shouldLoad: false }; + } + return { requestedAtStart: true, shouldLoad: true }; +} + export function resolveTimelineIsAtEnd(state: TimelineEndState | undefined): boolean | undefined { return state?.isNearEnd ?? state?.isAtEnd; } diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 26d3c3dce1e..760735c2b5b 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -69,6 +69,7 @@ import { deriveMessagesTimelineRows, normalizeCompactToolLabel, resolveAssistantMessageCopyState, + resolveOlderHistoryAutoLoad, resolveTimelineIsAtEnd, resolveTimelineMinimapHasPersistentGutter, resolveTimelineMinimapHeightStyle, @@ -337,6 +338,14 @@ export const MessagesTimeline = memo(function MessagesTimeline({ ); const [minimapHasPersistentGutter, setMinimapHasPersistentGutter] = useState(false); const [minimapHitStripWidth, setMinimapHitStripWidth] = useState(0); + const olderHistoryRequestedAtStartRef = useRef(false); + useEffect(() => { + olderHistoryRequestedAtStartRef.current = false; + }, [routeThreadKey]); + const requestOlderHistoryAtStart = useCallback(() => { + olderHistoryRequestedAtStartRef.current = true; + onLoadOlder?.(); + }, [onLoadOlder]); const handleAnchorReady = useCallback( (info: { anchorIndex: number | undefined }) => { if (anchorMessageId !== null && info.anchorIndex !== undefined) { @@ -368,10 +377,19 @@ export const MessagesTimeline = memo(function MessagesTimeline({ if (isAtEnd !== undefined) { onIsAtEndChange(isAtEnd); } - // Reaching the top lazy-loads older history; maintainVisibleContentPosition - // (set on the list) keeps the viewport anchored when rows prepend. - if (state?.isAtStart && hasMoreOlder && !loadingOlder) { - onLoadOlder?.(); + // Reaching the top lazy-loads one older page. Measurements rerun when the + // request settles and when rows prepend, so keep the visit latched until + // the user leaves the top; otherwise failures retry forever and successful + // pages chain-load the whole history back into memory. + const olderHistoryAutoLoad = resolveOlderHistoryAutoLoad({ + isAtStart: state?.isAtStart, + hasMoreOlder: hasMoreOlder && onLoadOlder !== undefined, + loadingOlder, + requestedAtStart: olderHistoryRequestedAtStartRef.current, + }); + olderHistoryRequestedAtStartRef.current = olderHistoryAutoLoad.requestedAtStart; + if (olderHistoryAutoLoad.shouldLoad) { + requestOlderHistoryAtStart(); } if (!state || minimapItems.length === 0) { return; @@ -403,6 +421,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ hasMoreOlder, loadingOlder, onLoadOlder, + requestOlderHistoryAtStart, ]); useEffect(() => { @@ -447,7 +466,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ return (