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 710d39c4c29..9331400f8ed 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 405103450e8..0f1739a1d8b 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -201,6 +201,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 9a136b06872..216b374a02a 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -342,6 +342,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { createdAt: "2026-02-24T00:00:06.000Z", }, ], + hasMoreActivities: false, checkpoints: [ { turnId: asTurnId("turn-1"), @@ -974,6 +975,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 ?? [], [ @@ -1153,6 +1156,254 @@ 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("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 32210436e67..bd2d494e5ab 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,31 @@ 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; + +// `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: NonNegativeInt, + limit: NonNegativeInt, +}); +const ThreadActivitiesBeforeActivityInput = Schema.Struct({ + threadId: ThreadId, + beforeCreatedAt: IsoDateTime, + beforeActivityId: EventId, + limit: NonNegativeInt, +}); const ProjectionProjectLookupRowSchema = ProjectionProjectDbRowSchema; const ProjectionThreadIdLookupRowSchema = Schema.Struct({ threadId: ThreadId, @@ -254,6 +280,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) @@ -808,6 +851,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, @@ -823,8 +871,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, @@ -832,6 +889,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, @@ -1076,16 +1208,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); } @@ -1190,6 +1313,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, })); @@ -1998,21 +2123,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, @@ -2063,6 +2181,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, @@ -2078,6 +2233,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 166a88ef17b..06769b45984 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -210,6 +210,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 1a331bc717f..9f679e3e119 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, @@ -203,6 +205,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, @@ -252,6 +255,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 fafdbfa6814..52e3d9b7273 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 ProjectEntriesFailure, @@ -279,6 +280,7 @@ const PROVIDER_STATUS_DEBOUNCE_MS = 200; 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], @@ -1024,6 +1026,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/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f5ea5bb1eba..196e3bd7e24 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -62,6 +62,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"; @@ -188,6 +192,7 @@ import { primaryServerKeybindingsAtom, serverEnvironment, } from "../state/server"; +import { orchestrationEnvironment } from "../state/orchestration"; import { terminalEnvironment } from "../state/terminal"; import { threadEnvironment } from "../state/threads"; import { vcsEnvironment } from "../state/vcs"; @@ -1725,7 +1730,170 @@ 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 [olderHistoryCursorVersion, setOlderHistoryCursorVersion] = useState(0); + 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 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); + setOlderHistoryCursorVersion((version) => version + 1); + }, [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); + setOlderHistoryCursorVersion((version) => version + 1); + }) + .finally(() => { + if (olderActivitiesGenRef.current === gen) { + inFlightOlderKeyRef.current = null; + setLoadingOlderActivities(false); + } + }); + }, [ + activeThread, + activeThreadActivityRequestKey, + hasMoreOlderActivities, + threadActivities, + loadThreadActivities, + ]); + const workLogEntries = useMemo(() => deriveWorkLogEntries(threadActivities), [threadActivities]); const pendingApprovals = useMemo( () => derivePendingApprovals(threadActivities), @@ -5076,6 +5244,10 @@ function ChatViewContent(props: ChatViewProps) { activeTurnStartedAt={activeWorkStartedAt} listRef={legendListRef} timelineEntries={timelineEntries} + hasMoreOlder={hasMoreOlderActivities} + loadingOlder={loadingOlderActivities} + olderHistoryCursorVersion={olderHistoryCursorVersion} + onLoadOlder={loadOlderActivities} latestTurn={activeLatestTurn} runningTurnId={ activeThread.session?.status === "running" @@ -5179,7 +5351,11 @@ function ChatViewContent(props: ChatViewProps) { providerStatuses={providerStatuses as ServerProvider[]} activeProjectDefaultModelSelection={activeProject?.defaultModelSelection} activeThreadModelSelection={activeThread?.modelSelection} - activeThreadActivities={activeThread?.activities} + // Merged set (older pages + live window), not the windowed + // live slice: the context meter scans for the latest + // context-window.updated event, which can sit in a + // lazy-loaded page on long threads. + activeThreadActivities={threadActivities} resolvedTheme={resolvedTheme} settings={settings} keybindings={keybindings} diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index bb6b2752a17..3adac0a148f 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -3494,7 +3494,11 @@ export default function Sidebar() { : EMPTY_THREAD_JUMP_LABELS; const orderedSidebarThreadKeys = visibleSidebarThreadKeys; const prewarmedSidebarThreadKeys = useMemo( - () => getSidebarThreadIdsToPrewarm(visibleSidebarThreadKeys), + // Browser clients can sit behind constrained remote links. Prewarming every + // visible thread hydrates several full detail windows before the user opens + // any of them, so keep the eager cache warm-up desktop-only. The active + // route still subscribes to its selected thread normally in either mode. + () => (isElectron ? getSidebarThreadIdsToPrewarm(visibleSidebarThreadKeys) : []), [visibleSidebarThreadKeys], ); const prewarmedSidebarThreadRefs = useMemo( diff --git a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts index 6d74204bc1c..350278734cb 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.test.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.test.ts @@ -4,9 +4,97 @@ import { computeMessageDurationStart, deriveMessagesTimelineRows, normalizeCompactToolLabel, + resolveOlderHistoryAutoLoad, resolveAssistantMessageCopyState, } from "./MessagesTimeline.logic"; +describe("resolveOlderHistoryAutoLoad", () => { + it("does not retry continuously while a failed request leaves the viewport at the start", () => { + const first = resolveOlderHistoryAutoLoad({ + armed: true, + hasMore: true, + isAtStart: true, + loading: false, + observedProgressVersion: 0, + progressVersion: 0, + }); + expect(first).toEqual({ + armed: false, + observedProgressVersion: 0, + shouldLoad: true, + }); + + const afterFailure = resolveOlderHistoryAutoLoad({ + armed: first.armed, + hasMore: true, + isAtStart: true, + loading: false, + observedProgressVersion: first.observedProgressVersion, + progressVersion: 0, + }); + expect(afterFailure).toEqual({ + armed: false, + observedProgressVersion: 0, + shouldLoad: false, + }); + + const afterLeavingStart = resolveOlderHistoryAutoLoad({ + armed: afterFailure.armed, + hasMore: true, + isAtStart: false, + loading: false, + observedProgressVersion: afterFailure.observedProgressVersion, + progressVersion: 0, + }); + expect(afterLeavingStart).toEqual({ + armed: true, + observedProgressVersion: 0, + shouldLoad: false, + }); + + expect( + resolveOlderHistoryAutoLoad({ + armed: afterLeavingStart.armed, + hasMore: true, + isAtStart: true, + loading: false, + observedProgressVersion: afterLeavingStart.observedProgressVersion, + progressVersion: 0, + }), + ).toEqual({ + armed: false, + observedProgressVersion: 0, + shouldLoad: true, + }); + }); + + it("rearms at the start only after a page successfully advances the cursor", () => { + const afterFirstAttempt = resolveOlderHistoryAutoLoad({ + armed: true, + hasMore: true, + isAtStart: true, + loading: false, + observedProgressVersion: 0, + progressVersion: 0, + }); + + expect( + resolveOlderHistoryAutoLoad({ + armed: afterFirstAttempt.armed, + hasMore: true, + isAtStart: true, + loading: false, + observedProgressVersion: afterFirstAttempt.observedProgressVersion, + progressVersion: 1, + }), + ).toEqual({ + armed: false, + observedProgressVersion: 1, + shouldLoad: true, + }); + }); +}); + 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 c6e277cce08..0522f9f2b09 100644 --- a/apps/web/src/components/chat/MessagesTimeline.logic.ts +++ b/apps/web/src/components/chat/MessagesTimeline.logic.ts @@ -21,6 +21,51 @@ export interface TimelineEndState { readonly isNearEnd?: boolean; } +export interface OlderHistoryAutoLoadDecision { + readonly armed: boolean; + readonly observedProgressVersion: number; + readonly shouldLoad: boolean; +} + +/** + * Treat reaching the start as an edge, not a continuously-true condition. + * A failed request leaves the viewport at the start, so level-triggered loading + * would immediately retry on every render. Leaving the start OR observing a + * successfully advanced page cursor rearms one future automatic request. The + * visible header control remains available for explicit retries while the edge + * is disarmed. + */ +export function resolveOlderHistoryAutoLoad(input: { + readonly armed: boolean; + readonly hasMore: boolean; + readonly isAtStart: boolean; + readonly loading: boolean; + readonly observedProgressVersion: number; + readonly progressVersion: number; +}): OlderHistoryAutoLoadDecision { + const progressed = input.progressVersion !== input.observedProgressVersion; + const armed = input.armed || progressed; + if (!input.isAtStart) { + return { + armed: true, + observedProgressVersion: input.progressVersion, + shouldLoad: false, + }; + } + if (!armed || !input.hasMore || input.loading) { + return { + armed, + observedProgressVersion: input.progressVersion, + shouldLoad: false, + }; + } + return { + armed: false, + observedProgressVersion: input.progressVersion, + shouldLoad: true, + }; +} + export function resolveTimelineIsAtEnd(state: TimelineEndState | undefined): boolean | undefined { return state?.isNearEnd ?? state?.isAtEnd; } diff --git a/apps/web/src/components/chat/MessagesTimeline.test.tsx b/apps/web/src/components/chat/MessagesTimeline.test.tsx index 0957e025311..09ba59f5190 100644 --- a/apps/web/src/components/chat/MessagesTimeline.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.test.tsx @@ -564,4 +564,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 61d7855844d..49b2fd5a49d 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -69,6 +69,7 @@ import { computeStableMessagesTimelineRows, deriveMessagesTimelineRows, normalizeCompactToolLabel, + resolveOlderHistoryAutoLoad, resolveAssistantMessageCopyState, resolveTimelineIsAtEnd, resolveTimelineMinimapHasPersistentGutter, @@ -179,6 +180,12 @@ interface MessagesTimelineProps { contentInsetEndAdjustment: number; onIsAtEndChange: (isAtEnd: boolean) => void; onManualNavigation: () => void; + /** Older history beyond the live activity window can be lazy-loaded. */ + hasMoreOlder?: boolean; + loadingOlder?: boolean; + /** Increments after the older-history cursor advances or is reset. */ + olderHistoryCursorVersion?: number; + onLoadOlder?: () => void; } // --------------------------------------------------------------------------- @@ -212,6 +219,10 @@ export const MessagesTimeline = memo(function MessagesTimeline({ contentInsetEndAdjustment, onIsAtEndChange, onManualNavigation, + hasMoreOlder = false, + loadingOlder = false, + olderHistoryCursorVersion = 0, + onLoadOlder, }: MessagesTimelineProps) { const [expandedTurnIds, setExpandedTurnIds] = useState>(new Set()); const [expandedWorkGroupIds, setExpandedWorkGroupIds] = useState>(new Set()); @@ -322,6 +333,19 @@ export const MessagesTimeline = memo(function MessagesTimeline({ null, ); const [minimapHasPersistentGutter, setMinimapHasPersistentGutter] = useState(false); + const olderHistoryAutoLoadArmedRef = useRef(true); + const olderHistoryObservedProgressVersionRef = useRef(olderHistoryCursorVersion); + const requestOlderHistory = useCallback(() => { + // Disarm before both automatic and explicit requests. If a request fails, + // prop changes while the viewport remains at the start must not trigger an + // immediate retry loop; the header button still permits a deliberate retry. + olderHistoryAutoLoadArmedRef.current = false; + onLoadOlder?.(); + }, [onLoadOlder]); + useEffect(() => { + olderHistoryAutoLoadArmedRef.current = true; + olderHistoryObservedProgressVersionRef.current = olderHistoryCursorVersion; + }, [routeThreadKey, olderHistoryCursorVersion]); const handleAnchorReady = useCallback( (info: { anchorIndex: number | undefined }) => { if (anchorMessageId !== null && info.anchorIndex !== undefined) { @@ -353,6 +377,21 @@ 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. + const olderHistoryDecision = resolveOlderHistoryAutoLoad({ + armed: olderHistoryAutoLoadArmedRef.current, + hasMore: hasMoreOlder, + isAtStart: state?.isAtStart ?? false, + loading: loadingOlder, + observedProgressVersion: olderHistoryObservedProgressVersionRef.current, + progressVersion: olderHistoryCursorVersion, + }); + olderHistoryAutoLoadArmedRef.current = olderHistoryDecision.armed; + olderHistoryObservedProgressVersionRef.current = olderHistoryDecision.observedProgressVersion; + if (olderHistoryDecision.shouldLoad) { + requestOlderHistory(); + } if (!state || minimapItems.length === 0) { return; } @@ -375,7 +414,16 @@ export const MessagesTimeline = memo(function MessagesTimeline({ strip.dataset.inView = inView ? "true" : "false"; } - }, [listRef, minimapItems, minimapStripMap, onIsAtEndChange]); + }, [ + listRef, + minimapItems, + minimapStripMap, + onIsAtEndChange, + hasMoreOlder, + loadingOlder, + olderHistoryCursorVersion, + requestOlderHistory, + ]); useEffect(() => { const frame = requestAnimationFrame(handleScroll); @@ -406,6 +454,28 @@ export const MessagesTimeline = memo(function MessagesTimeline({ }; }, [timelineViewportElement, rows.length]); + const listHeader = useMemo(() => { + if (loadingOlder) { + return ( +
+ Loading older history… +
+ ); + } + if (hasMoreOlder) { + return ( + + ); + } + return TIMELINE_LIST_HEADER; + }, [loadingOlder, hasMoreOlder, requestOlderHistory]); + const sharedState = useMemo( () => ({ timestampFormat, @@ -457,7 +527,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) { return (

@@ -499,7 +573,7 @@ export const MessagesTimeline = memo(function MessagesTimeline({ }} onScroll={handleScroll} className="scrollbar-gutter-both h-full min-h-0 overflow-x-hidden overscroll-y-contain px-3 [overflow-anchor:none] sm:px-5" - ListHeaderComponent={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, diff --git a/packages/client-runtime/src/state/threadReducer.test.ts b/packages/client-runtime/src/state/threadReducer.test.ts index 94eb1c65370..379897870de 100644 --- a/packages/client-runtime/src/state/threadReducer.test.ts +++ b/packages/client-runtime/src/state/threadReducer.test.ts @@ -9,9 +9,25 @@ 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"), @@ -697,4 +713,61 @@ 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 670540fee70..37ae19b44d7 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 diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 7e9c421b5df..ac3cb18dd08 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", @@ -362,6 +363,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), }); @@ -1222,6 +1227,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, @@ -1249,6 +1285,10 @@ export const OrchestrationRpcSchemas = { input: OrchestrationGetTurnDiffInput, output: OrchestrationGetTurnDiffResult, }, + getThreadActivities: { + input: OrchestrationGetThreadActivitiesInput, + output: OrchestrationGetThreadActivitiesResult, + }, getFullThreadDiff: { input: OrchestrationGetFullThreadDiffInput, output: OrchestrationGetFullThreadDiffResult, @@ -1295,6 +1335,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 48c5d9a774d..7553190a85c 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, @@ -605,6 +607,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, { @@ -745,6 +756,7 @@ export const WsRpcGroup = RpcGroup.make( WsSubscribeAuthAccessRpc, WsOrchestrationDispatchCommandRpc, WsOrchestrationGetTurnDiffRpc, + WsOrchestrationGetThreadActivitiesRpc, WsOrchestrationGetFullThreadDiffRpc, WsOrchestrationReplayEventsRpc, WsOrchestrationGetArchivedShellSnapshotRpc,