From c5fe3bc59d9fdbaa2d9b0801e9e0ef307bf6409c Mon Sep 17 00:00:00 2001 From: ytahdn Date: Fri, 17 Jul 2026 09:55:28 +0800 Subject: [PATCH 1/5] feat(web-shell): paginate restored session history --- docs/design/web-shell-history-pagination.md | 155 +++++++++ packages/acp-bridge/src/bridge.test.ts | 17 +- packages/acp-bridge/src/bridge.ts | 39 ++- packages/acp-bridge/src/bridgeTypes.ts | 7 + .../acp-bridge/src/compactionEngine.test.ts | 33 ++ packages/acp-bridge/src/compactionEngine.ts | 37 +- .../cli/src/acp-integration/acpAgent.test.ts | 109 ++++++ packages/cli/src/acp-integration/acpAgent.ts | 78 ++++- .../session/history-replay-page.ts | 32 +- .../session/history-replayer.ts | 7 +- .../cli/src/acp-integration/session/types.ts | 2 + packages/cli/src/serve/capabilities.ts | 1 + packages/cli/src/serve/routes/session.ts | 73 ++++ packages/cli/src/serve/server.test.ts | 51 +++ .../session-transcript-reader.test.ts | 124 +++++++ .../src/services/session-transcript-reader.ts | 142 +++++++- .../sdk-typescript/src/daemon/DaemonClient.ts | 30 +- .../src/daemon/DaemonSessionClient.ts | 6 + packages/sdk-typescript/src/daemon/types.ts | 4 + packages/web-shell/client/App.tsx | 8 + .../web-shell/client/components/ChatPane.tsx | 6 + .../components/MessageList.dom.test.tsx | 117 ++++++- .../client/components/MessageList.module.css | 6 + .../client/components/MessageList.tsx | 92 ++++- .../web-shell/client/components/SplitView.tsx | 4 +- .../components/WorkspaceSessionProvider.tsx | 8 +- .../web-shell/client/constants/sessions.ts | 6 +- packages/web-shell/client/i18n.tsx | 5 + packages/webui/src/daemon-react-sdk.ts | 4 + packages/webui/src/daemon/index.ts | 2 + .../session/DaemonSessionProvider.test.tsx | 325 ++++++++++++++++++ .../daemon/session/DaemonSessionProvider.tsx | 286 ++++++++++++++- packages/webui/src/daemon/session/index.ts | 2 + packages/webui/src/daemon/session/types.ts | 2 + 34 files changed, 1764 insertions(+), 56 deletions(-) create mode 100644 docs/design/web-shell-history-pagination.md diff --git a/docs/design/web-shell-history-pagination.md b/docs/design/web-shell-history-pagination.md new file mode 100644 index 00000000000..05feb76ae55 --- /dev/null +++ b/docs/design/web-shell-history-pagination.md @@ -0,0 +1,155 @@ +# Web Shell History Pagination + +## Problem + +Web Shell restores an existing session through `POST /session/:id/load`. The +daemon currently converts the complete persisted conversation into replay +updates, bounds the resulting in-memory snapshot by bytes, and returns the +retained snapshot as one `compactedReplay` response. Web Shell normalizes that +entire response before painting the transcript. + +The byte bound prevents unbounded daemon memory, but a long session can still +produce a multi-megabyte response and a large synchronous normalization pass. +Older events are also represented only by a truncation marker; the persisted +transcript pager is not connected to Web Shell. + +The existing transcript API pages from the oldest record toward the newest. +That order is useful for exports but does not support chat history UX, where the +latest messages must paint first and earlier messages are requested on demand. + +## Goals + +- Let Web Shell request a bounded latest page during `session/load`. +- Preserve the atomic replay snapshot and SSE watermark produced by load. +- Load older persisted history with an opaque backward cursor. +- Keep pages chronological when rendered and preserve the viewport when older + content is prepended. +- Bound browser transcript state even when the persisted session is very long. +- Preserve existing load and forward transcript paging behavior for clients + that do not opt in. + +## Non-goals + +- Changing the model-facing conversation restored by core. The agent still + receives the complete resumed conversation required for continuation. +- Automatically loading the complete transcript. +- Paging archived transcripts. +- Replacing SSE replay or reconnect behavior. +- Making the first persisted-transcript index scan sublinear. The existing + frozen-snapshot index remains linear on a cache miss. + +## Design + +### Stable history record identity + +History replay already tracks the active persisted `ChatRecord` while emitting +ACP updates. The replay collector will stamp that record UUID into private +update metadata. This metadata is present only on replayed persisted history; +live events keep their current shape. + +The UUID gives the initial load page an exact exclusive boundary. Older paging +therefore does not rely on timestamps and cannot duplicate the first retained +record when timestamps collide. + +### Paged session load + +The TypeScript SDK adds an optional `historyPageSize` restore option. The serve +route validates it with the transcript pager's existing range of 1 through 500 +and forwards it through the bridge as private ACP load metadata. Servers that +support this contract advertise `session_transcript_pagination` through +`/capabilities.features`; Web Shell sends the option only when that feature is +present, so older servers retain their bulk-load behavior. + +When the option is absent, load is unchanged. When present, the ACP agent keeps +the complete resumed conversation in core but converts only the latest record +suffix for the UI replay envelope. The suffix begins at a normal user-turn +boundary. The envelope reports whether older active-chain records exist. + +The bridge seeds only that page into the session EventBus. Its response still +contains the replay events and the EventBus `lastEventId` from one load +operation, so events emitted after the snapshot remain available through SSE +without a `resume`/transcript-read race. + +### Backward persisted transcript pages + +`GET /session/:id/transcript` and its workspace-qualified form accept an +exclusive `beforeRecordId` only on the first backward request. The response +uses the existing `nextCursor` field; the signed cursor records that paging is +backward and freezes file identity, active leaf, byte size, position, and replay +direction. + +Backward pages are returned in chronological display order. Each selected page +starts at a normal user-turn boundary. The record limit is therefore a soft page +target: a long turn is returned intact even when it exceeds the requested record +count, so scrolling never reveals only the tail of the previous turn. The +workspace route retains its hard source-byte limit: if a complete turn exceeds +that limit, it returns `transcript_page_too_large` rather than returning a +partial turn. Forward cursors and responses remain byte-for-byte compatible. + +### WebUI transcript state + +The provider normalizes an older page in an isolated transcript store, +allocates non-conflicting block ordinals, and resets the active store with only +the resulting blocks prepended. Live side-channel state, active streaming +pointers, and the SSE event cursor are preserved. + +Older pages are prepended atomically: if a complete page would exceed +`maxBlocks`, the UI keeps the current transcript unchanged, stops paging, and +reports that earlier records remain persisted but cannot be displayed in the +current view. A complete initial page is never truncated mid-turn; when it +exceeds `maxBlocks`, the provider retains that page and prevents older paging. + +### Web Shell interaction + +Web Shell opts into a 100-record initial history page. The session provider +derives the exclusive boundary from the earliest replay record UUID and exposes +history state through a small hook: + +- whether an older page exists; +- whether a page request is in flight; +- an action to load the next page. + +When the message list reaches the top, it automatically requests the next page +and shows a loading status above the transcript. It records the scroll container +height before the request and restores the visual anchor after prepending. A +short page that does not fill the viewport automatically advances until the +transcript becomes scrollable or history is exhausted. A failed or partial page +leaves the current transcript intact, stops automatic retries, and surfaces the +existing daemon notice path. + +## Consistency and failure handling + +- Initial history and the SSE watermark remain coupled through `session/load`. +- The first older page excludes the exact earliest persisted record already in + the load snapshot. +- Continuation cursors freeze the transcript snapshot; later appends do not + change older pages. +- Session changes invalidate in-flight page results before they can modify the + new transcript. +- Invalid boundaries and cursors return the existing invalid-transcript-cursor + error family. +- Snapshot replacement, rewind, archive, or deletion returns + `transcript_snapshot_unavailable`; the provider stops paging and leaves + already-rendered content intact. +- A replay conversion failure does not prepend a partial page or advance its + boundary. The provider stops automatic paging and emits a warning notice. +- Existing clients that omit `historyPageSize` or `beforeRecordId` retain the + current bulk-load and forward-page contracts. + +## Affected areas + +| Layer | Change | +| ------------------------ | ------------------------------------------------------------ | +| Core transcript reader | Backward cursor and exclusive record boundary | +| ACP replay | Record UUID metadata and latest-suffix selection | +| ACP bridge / serve route | Paged-load metadata, validation, and `hasMore` propagation | +| TypeScript SDK | Restore option, backward page option, restored history state | +| WebUI provider | Isolated prepend, page state, stale-request protection | +| Web Shell | Opt-in page size and automatic top-loading behavior | + +## Open questions and follow-ups + +- A session created and kept live for its entire lifetime has no persisted + record UUID on its live EventBus frames. This change pages sessions restored + through the new load path; adding record identity to live emission is a + separate recording/emission coordination change. diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 5b211936434..418deea7606 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -1581,7 +1581,7 @@ describe('createAcpSessionBridge', () => { const result = await bridge.getSessionTranscriptPage({ sessionId: 'session-1', - cursor: 'opaque', + beforeRecordId: 'record-3', limit: 2, }); @@ -1597,7 +1597,7 @@ describe('createAcpSessionBridge', () => { params: { cwd: WS_A, sessionId: 'session-1', - cursor: 'opaque', + beforeRecordId: 'record-3', limit: 2, }, }, @@ -2775,12 +2775,14 @@ describe('createAcpSessionBridge', () => { loadSessionImpl: (p) => { expect(p._meta).toMatchObject({ 'qwen.session.loadReplayMode': 'bulk', + 'qwen.session.loadReplayPageSize': 100, }); return { _meta: { keep: 'state', 'qwen.session.loadReplay': { v: 1, + hasMore: true, partial: true, replayError: 'replay boom', updates: [ @@ -2808,17 +2810,22 @@ describe('createAcpSessionBridge', () => { sessionId: 'persisted-bulk-history', workspaceCwd: WS_A, historyReplay: 'response', + historyPageSize: 100, }); expect(handles[0]?.agent.loadSessionCalls[0]).toMatchObject({ sessionId: 'persisted-bulk-history', cwd: WS_A, mcpServers: [], - _meta: { 'qwen.session.loadReplayMode': 'bulk' }, + _meta: { + 'qwen.session.loadReplayMode': 'bulk', + 'qwen.session.loadReplayPageSize': 100, + }, }); expect(loaded.state).toEqual({ _meta: { keep: 'state' } }); expect(loaded.partial).toBe(true); expect(loaded.replayError).toBe('replay boom'); + expect(loaded.historyHasMore).toBe(true); expect(loaded.lastEventId).toBe(2); expect(loaded.compactedReplay).toHaveLength(2); expect(loaded.liveJournal).toEqual([]); @@ -2916,10 +2923,12 @@ describe('createAcpSessionBridge', () => { { sessionUpdate: 'user_message_chunk', content: { type: 'text', text: `old-${'x'.repeat(600)}` }, + _meta: { recordId: 'record-1' }, }, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: `new-${'y'.repeat(600)}` }, + _meta: { recordId: 'record-2' }, }, ], }, @@ -2935,6 +2944,7 @@ describe('createAcpSessionBridge', () => { sessionId: 'persisted-bounded-history', workspaceCwd: WS_A, historyReplay: 'response', + historyPageSize: 100, }); expect(loaded.lastEventId).toBe(2); @@ -2951,6 +2961,7 @@ describe('createAcpSessionBridge', () => { update?: { content?: { text?: string } }; }; expect(retained.update?.content?.text).toBe(`new-${'y'.repeat(600)}`); + expect(loaded.historyHasMore).toBe(true); await bridge.shutdown(); }); diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index ef9050ab99c..8521e63219e 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -88,6 +88,7 @@ import { LOAD_REPLAY_BULK_MODE, LOAD_REPLAY_META_KEY, LOAD_REPLAY_MODE_META_KEY, + LOAD_REPLAY_PAGE_SIZE_META_KEY, LOAD_REPLAY_VERSION, } from './bridgeTypes.js'; import type { @@ -218,6 +219,7 @@ function extractLoadReplayResponse(state: BridgeSessionState): { updates: SessionUpdate[]; partial?: true; replayError?: string; + hasMore?: boolean; } { const meta = isRecord(state._meta) ? state._meta : undefined; const replay = meta?.[LOAD_REPLAY_META_KEY]; @@ -256,6 +258,13 @@ function extractLoadReplayResponse(state: BridgeSessionState): { `(version=${LOAD_REPLAY_VERSION}, replayError=${describeLoadReplayValue(replayError)})`, ); } + const hasMore = replay['hasMore']; + if (hasMore !== undefined && typeof hasMore !== 'boolean') { + throw new Error( + `Invalid qwen.session.loadReplay hasMore ` + + `(version=${LOAD_REPLAY_VERSION}, hasMore=${describeLoadReplayValue(hasMore)})`, + ); + } const invalidUpdateIndex = rawUpdates.findIndex( (update) => !isBulkReplayUpdate(update), ); @@ -284,6 +293,7 @@ function extractLoadReplayResponse(state: BridgeSessionState): { updates: rawUpdates, ...(partial === true ? { partial: true as const } : {}), ...(typeof replayError === 'string' ? { replayError } : {}), + ...(hasMore === true ? { hasMore: true } : {}), }; } @@ -588,6 +598,7 @@ interface SessionEntry { /** Response-mode `session/load` can return a partial replay prefix. */ restoreReplayPartial?: true; restoreReplayError?: string; + restoreHistoryHasMore?: true; /** * Most recent heartbeat across any client on this session (Date.now() * epoch ms). Set on every `recordHeartbeat` call regardless of whether @@ -3637,7 +3648,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { const replayFieldsFor = ( entry: Pick< SessionEntry, - 'events' | 'restoreReplayPartial' | 'restoreReplayError' + | 'events' + | 'restoreReplayPartial' + | 'restoreReplayError' + | 'restoreHistoryHasMore' >, action: 'load' | 'resume', ): Pick< @@ -3647,6 +3661,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { | 'lastEventId' | 'partial' | 'replayError' + | 'historyHasMore' > => { const replayStatus = action === 'load' && entry.restoreReplayPartial === true @@ -3667,6 +3682,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { liveJournal: snapshot.liveJournal, lastEventId: snapshot.lastEventId, ...replayStatus, + ...(entry.restoreHistoryHasMore === true + ? { historyHasMore: true } + : {}), }; } return { lastEventId: snapshot.lastEventId, ...replayStatus }; @@ -3890,6 +3908,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { let replayUpdates: SessionUpdate[] = []; let replayPartial: true | undefined; let replayError: string | undefined; + let replayHasMore: true | undefined; try { if (action === 'load') { state = await Promise.race([ @@ -3907,6 +3926,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ? { _meta: { [LOAD_REPLAY_MODE_META_KEY]: LOAD_REPLAY_BULK_MODE, + ...(req.historyPageSize !== undefined + ? { + [LOAD_REPLAY_PAGE_SIZE_META_KEY]: + req.historyPageSize, + } + : {}), }, } : {}), @@ -3936,6 +3961,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { replayUpdates = extracted.updates; replayPartial = extracted.partial; replayError = extracted.replayError; + replayHasMore = extracted.hasMore === true ? true : undefined; } } catch (err) { restoreEvents.close(); @@ -4031,6 +4057,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (replayError !== undefined) { entry.restoreReplayError = replayError; } + if (replayHasMore === true) { + entry.restoreHistoryHasMore = true; + } seedSnapshotCaches(entry, publicState); const artifactRestoreWarnings = await entry.artifacts.restore( restoredArtifactSnapshot, @@ -4050,6 +4079,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ingestArtifacts: restoredArtifactSnapshot === undefined || artifactRestoreFailed, }); + if ( + req.historyPageSize !== undefined && + entry.events + .snapshotReplay() + ?.compactedTurns.some((event) => event.type === 'history_truncated') + ) { + entry.restoreHistoryHasMore = true; + } ci.client.drainEarlyEvents(entry.sessionId, entry); } const clientId = registerClient(entry, req.clientId); diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 93cbfd06723..1cc1b448d87 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -137,6 +137,8 @@ export interface BridgeRestoreSessionRequest { clientId?: string; /** Internal replay transport for `session/load`; defaults to bulk response. */ historyReplay?: 'stream' | 'response'; + /** Optional newest persisted-record page requested for response replay. */ + historyPageSize?: number; approvalMode?: ApprovalMode; /** * Persisted parent lineage recovered from the transcript by the caller (the @@ -154,12 +156,14 @@ export interface BridgeRestoreSessionRequest { export const LOAD_REPLAY_MODE_META_KEY = 'qwen.session.loadReplayMode'; export const LOAD_REPLAY_META_KEY = 'qwen.session.loadReplay'; +export const LOAD_REPLAY_PAGE_SIZE_META_KEY = 'qwen.session.loadReplayPageSize'; export const LOAD_REPLAY_BULK_MODE = 'bulk'; export const LOAD_REPLAY_VERSION = 1 as const; export interface BridgeLoadReplayEnvelope { v: typeof LOAD_REPLAY_VERSION; updates: SessionUpdate[]; + hasMore?: boolean; partial?: true; replayError?: string; } @@ -185,6 +189,8 @@ export interface BridgeRestoredSession extends BridgeSession { compactedReplay?: BridgeEvent[]; /** Raw events since last turn boundary (current incomplete turn). */ liveJournal?: BridgeEvent[]; + /** True when persisted records exist before the returned replay page. */ + historyHasMore?: boolean; /** High-water mark event ID — client uses this as initial SSE cursor. */ lastEventId?: number; } @@ -192,6 +198,7 @@ export interface BridgeRestoredSession extends BridgeSession { export interface BridgeSessionTranscriptPageRequest { sessionId: string; cursor?: string; + beforeRecordId?: string; limit?: number; } diff --git a/packages/acp-bridge/src/compactionEngine.test.ts b/packages/acp-bridge/src/compactionEngine.test.ts index c83968f010e..1ab634dd5bb 100644 --- a/packages/acp-bridge/src/compactionEngine.test.ts +++ b/packages/acp-bridge/src/compactionEngine.test.ts @@ -932,6 +932,39 @@ describe('EventBus + CompactionEngine integration', () => { ).toBeUndefined(); }); + it('seedReplayEvents evicts whole persisted records', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); + const bus = new EventBus(100, undefined, engine); + const update = (recordId: string, text: string) => ({ + type: 'session_update' as const, + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: `${text}-${'x'.repeat(600)}` }, + _meta: { recordId }, + }, + }, + }); + + bus.seedReplayEvents([ + update('old-record', 'old-1'), + update('old-record', 'old-2'), + update('new-record', 'new-1'), + update('new-record', 'new-2'), + ]); + + const snapshot = bus.snapshotReplay()!; + expect(snapshot.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snapshot.compactedTurns)).toEqual([ + `new-1-${'x'.repeat(600)}`, + `new-2-${'x'.repeat(600)}`, + ]); + expect(snapshot.compactedTurns[0]?.data).toMatchObject({ + truncatedEvents: 2, + retainedEvents: 2, + }); + }); + it('seedReplayEvents replaces prior replay and truncation state', () => { const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); const bus = new EventBus(100, undefined, engine); diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index 47caed7df56..7922f6c92b5 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -65,6 +65,23 @@ interface ReplaySegment { turnCount: number; } +function replayRecordId(event: BridgeEvent): string | undefined { + if (event.type !== 'session_update') return undefined; + const data = event.data; + if (!data || typeof data !== 'object' || Array.isArray(data)) + return undefined; + const update = (data as Record)['update']; + if (!update || typeof update !== 'object' || Array.isArray(update)) { + return undefined; + } + const meta = (update as Record)['_meta']; + if (!meta || typeof meta !== 'object' || Array.isArray(meta)) { + return undefined; + } + const recordId = (meta as Record)['recordId']; + return typeof recordId === 'string' ? recordId : undefined; +} + export interface ReplayWindowEviction { droppedBytes: number; droppedEvents: number; @@ -168,11 +185,29 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { seedReplayEvents(events: BridgeEvent[]): void { if (this.closed) return; this.resetReplayWindow(); + let recordEvents: BridgeEvent[] = []; + let recordId: string | undefined; + const flushRecord = () => { + this.addReplaySegment(recordEvents, 0); + recordEvents = []; + recordId = undefined; + }; for (const event of events) { this.recordLastEventId(event); if (TRANSIENT_TYPES.has(event.type)) continue; - this.addReplaySegment([event], 0); + const nextRecordId = replayRecordId(event); + if (nextRecordId === undefined) { + flushRecord(); + this.addReplaySegment([event], 0); + continue; + } + if (recordId !== undefined && recordId !== nextRecordId) { + flushRecord(); + } + recordId = nextRecordId; + recordEvents.push(event); } + flushRecord(); this.liveJournal = []; this.slots = []; this.toolSlotIndex.clear(); diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index bcdd450b8d9..cf9e5208422 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -10888,6 +10888,115 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => { await agentPromise; }); + it('loadSession limits bulk replay to complete recent turns', async () => { + const makeMessage = ( + uuid: string, + parentUuid: string | null, + type: 'user' | 'assistant', + ) => ({ + uuid, + parentUuid, + sessionId: 'persisted-1', + timestamp: '2026-07-16T00:00:00.000Z', + type, + cwd: '/tmp', + version: 'test', + message: { role: type === 'user' ? 'user' : 'model', parts: [] }, + }); + const messages = [ + makeMessage('u1', null, 'user'), + makeMessage('a1', 'u1', 'assistant'), + makeMessage('u2', 'a1', 'user'), + makeMessage('a2', 'u2', 'assistant'), + makeMessage('u3', 'a2', 'user'), + makeMessage('a3', 'u3', 'assistant'), + ]; + bindRestoreMocks({ + sessionExists: true, + resumedConversation: { messages }, + }); + mockHistoryReplay.mockImplementation(async (_context, history) => { + expect(history).toEqual(messages.slice(4)); + }); + const { agent, agentPromise } = await spawnAgent(); + + const response = (await agent.loadSession({ + cwd: '/tmp', + sessionId: 'persisted-1', + mcpServers: [], + _meta: { + 'qwen.session.loadReplayMode': 'bulk', + 'qwen.session.loadReplayPageSize': 2, + }, + })) as { + _meta?: Record; + }; + + expect(response._meta?.['qwen.session.loadReplay']?.hasMore).toBe(true); + expect(lastSessionMock?.primeTurnFromHistory).toHaveBeenCalledWith( + messages, + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('loadSession keeps one long turn complete', async () => { + const makeMessage = ( + uuid: string, + parentUuid: string | null, + type: 'user' | 'assistant' | 'tool_result', + ) => ({ + uuid, + parentUuid, + sessionId: 'persisted-long-turn', + timestamp: '2026-07-16T00:00:00.000Z', + type, + cwd: '/tmp', + version: 'test', + message: { + role: type === 'assistant' ? ('model' as const) : ('user' as const), + parts: [], + }, + }); + const messages = [ + makeMessage('u1', null, 'user'), + makeMessage('a-tool', 'u1', 'assistant'), + makeMessage('t1', 'a-tool', 'tool_result'), + makeMessage('a-final', 't1', 'assistant'), + ]; + bindRestoreMocks({ + sessionExists: true, + resumedConversation: { messages }, + }); + mockHistoryReplay.mockImplementation(async (_context, history) => { + expect(history).toEqual(messages); + }); + const { agent, agentPromise } = await spawnAgent(); + + const response = (await agent.loadSession({ + cwd: '/tmp', + sessionId: 'persisted-long-turn', + mcpServers: [], + _meta: { + 'qwen.session.loadReplayMode': 'bulk', + 'qwen.session.loadReplayPageSize': 2, + }, + })) as { + _meta?: Record; + }; + + expect( + response._meta?.['qwen.session.loadReplay']?.hasMore, + ).toBeUndefined(); + expect(lastSessionMock?.primeTurnFromHistory).toHaveBeenCalledWith( + messages, + ); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('loadSession returns partial bulk replay updates when replay throws', async () => { const messages = [{ role: 'user', parts: [{ text: 'hi' }] }]; bindRestoreMocks({ diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 177d452a400..3fb5a744ad9 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -100,6 +100,7 @@ import { type SessionArtifactEventRecordPayload, type SessionArtifactSnapshotRecordPayload, type WorkspaceRememberContextMode, + type ChatRecord, } from '@qwen-code/qwen-code-core'; import { randomUUID } from 'node:crypto'; import { performance } from 'node:perf_hooks'; @@ -284,6 +285,7 @@ import { LOAD_REPLAY_BULK_MODE, LOAD_REPLAY_META_KEY, LOAD_REPLAY_MODE_META_KEY, + LOAD_REPLAY_PAGE_SIZE_META_KEY, LOAD_REPLAY_VERSION, type ClientMcpOverWsRuntimeConfig, type BridgeLoadReplayEnvelope, @@ -497,6 +499,47 @@ function isBulkLoadReplayRequest(params: LoadSessionRequest): boolean { return meta?.[LOAD_REPLAY_MODE_META_KEY] === LOAD_REPLAY_BULK_MODE; } +function getLoadReplayPageSize(params: LoadSessionRequest): number | undefined { + const meta = isObjectRecord(params._meta) ? params._meta : undefined; + const value = meta?.[LOAD_REPLAY_PAGE_SIZE_META_KEY]; + if (value === undefined) return undefined; + if ( + !Number.isSafeInteger(value) || + (value as number) < 1 || + (value as number) > SESSION_TRANSCRIPT_MAX_LIMIT + ) { + throw RequestError.invalidParams( + undefined, + `Invalid load replay page size; expected 1..${SESSION_TRANSCRIPT_MAX_LIMIT}`, + ); + } + return value as number; +} + +function isHistoryTurnStart(record: ChatRecord): boolean { + return record.type === 'user' && record.subtype !== 'mid_turn_user_message'; +} + +function selectRecentHistoryRecords( + records: ChatRecord[], + pageSize: number | undefined, +): { records: ChatRecord[]; hasMore: boolean } { + if (pageSize === undefined || records.length <= pageSize) { + return { records, hasMore: false }; + } + let start = records.length - pageSize; + for (let i = start; i < records.length; i++) { + if (isHistoryTurnStart(records[i]!)) { + start = i; + break; + } + } + while (start > 0 && !isHistoryTurnStart(records[start]!)) { + start--; + } + return { records: records.slice(start), hasMore: start > 0 }; +} + function createHiddenWorkspaceMemoryConfig(config: Config): Config { return new Proxy(config, { get(target, prop) { @@ -3441,6 +3484,9 @@ class QwenAgent implements Agent { const sessionData = config.getResumedSessionData(); const bulkReplay = isBulkLoadReplayRequest(params); + const replayPageSize = bulkReplay + ? getLoadReplayPageSize(params) + : undefined; const session = await this.createAndStoreSession( config, settings, @@ -3456,11 +3502,15 @@ class QwenAgent implements Agent { let replayUpdates: SessionUpdate[] = []; if (records) { session.primeTurnFromHistory(records); + const replayPage = selectRecentHistoryRecords( + records, + replayPageSize, + ); const replayUsage = createReplayCumulativeUsage(); const replay = await collectHistoryReplayUpdates({ sessionId: params.sessionId, config, - records, + records: replayPage.records, gaps: sessionData?.historyGaps, cumulativeUsage: replayUsage, logger: debugLogger, @@ -3473,8 +3523,14 @@ class QwenAgent implements Agent { updates: replayUpdates, partial: true, replayError: replay.replayError, + ...(replayPage.hasMore ? { hasMore: true } : {}), }; } + replayEnvelope ??= { + v: LOAD_REPLAY_VERSION, + updates: replayUpdates, + ...(replayPage.hasMore ? { hasMore: true } : {}), + }; } replayEnvelope ??= { v: LOAD_REPLAY_VERSION, @@ -6212,6 +6268,23 @@ class QwenAgent implements Agent { 'Invalid transcript cursor', ); } + const rawBeforeRecordId = params['beforeRecordId']; + if ( + rawBeforeRecordId !== undefined && + (typeof rawBeforeRecordId !== 'string' || + rawBeforeRecordId.length === 0) + ) { + throw RequestError.invalidParams( + undefined, + 'Invalid transcript record boundary', + ); + } + if (rawCursor !== undefined && rawBeforeRecordId !== undefined) { + throw RequestError.invalidParams( + undefined, + 'Transcript cursor and record boundary are mutually exclusive', + ); + } const rawLimit = params['limit']; if ( rawLimit !== undefined && @@ -6231,6 +6304,9 @@ class QwenAgent implements Agent { const reader = new SessionTranscriptReader(cwd); const page = await reader.readPage(sessionId, { ...(typeof rawCursor === 'string' ? { cursor: rawCursor } : {}), + ...(typeof rawBeforeRecordId === 'string' + ? { beforeRecordId: rawBeforeRecordId } + : {}), ...(typeof rawLimit === 'number' ? { limit: rawLimit } : {}), }); const config = await this.getTranscriptReplayConfig(cwd, settings); diff --git a/packages/cli/src/acp-integration/session/history-replay-page.ts b/packages/cli/src/acp-integration/session/history-replay-page.ts index 5e408b793af..b5e9d96bff6 100644 --- a/packages/cli/src/acp-integration/session/history-replay-page.ts +++ b/packages/cli/src/acp-integration/session/history-replay-page.ts @@ -109,10 +109,23 @@ function replayContext( cumulativeUsage: CumulativeUsage, config?: Config, ): SessionEmitterContext { + let activeRecordId: string | null = null; return { sessionId, sendUpdate: async (update) => { - updates.push(update); + if (activeRecordId === null) { + updates.push(update); + return; + } + const record = update as unknown as Record; + const meta = isObjectRecord(record['_meta']) ? record['_meta'] : {}; + updates.push({ + ...record, + _meta: { ...meta, recordId: activeRecordId }, + } as unknown as SessionUpdate); + }, + setActiveRecordId: (recordId: string | null) => { + activeRecordId = recordId; }, cumulativeUsage, ...(config ? { config } : {}), @@ -197,8 +210,9 @@ export async function replayTranscriptRecordPage({ let replayError: string | undefined; try { const replayState = await replayer.replayPage(page.records, { - pendingToolCalls: state.pendingToolCalls, - finalizeDangling: !page.hasMore, + pendingToolCalls: + page.direction === 'backward' ? [] : state.pendingToolCalls, + finalizeDangling: page.direction === 'backward' || !page.hasMore, gaps: page.gaps, }); pendingToolCalls = replayState.pendingToolCalls; @@ -217,10 +231,14 @@ export async function replayTranscriptRecordPage({ page.nextCursorState && replayError === undefined ? encodeCursor({ ...page.nextCursorState, - replay: { - pendingToolCalls, - cumulativeUsage: state.cumulativeUsage, - }, + ...(page.direction === 'backward' + ? {} + : { + replay: { + pendingToolCalls, + cumulativeUsage: state.cumulativeUsage, + }, + }), }) : undefined; diff --git a/packages/cli/src/acp-integration/session/history-replayer.ts b/packages/cli/src/acp-integration/session/history-replayer.ts index 3e542fa49d8..dd48950cc6b 100644 --- a/packages/cli/src/acp-integration/session/history-replayer.ts +++ b/packages/cli/src/acp-integration/session/history-replayer.ts @@ -453,11 +453,6 @@ export class HistoryReplayer { } private setActiveRecordId(recordId: string | null, timestamp?: string): void { - const context = this.ctx as unknown as { - setActiveRecordId?: (id: string | null, timestamp?: string) => void; - }; - if (typeof context.setActiveRecordId === 'function') { - context.setActiveRecordId(recordId, timestamp); - } + this.ctx.setActiveRecordId?.(recordId, timestamp); } } diff --git a/packages/cli/src/acp-integration/session/types.ts b/packages/cli/src/acp-integration/session/types.ts index 868df0a9100..64d40637cd3 100644 --- a/packages/cli/src/acp-integration/session/types.ts +++ b/packages/cli/src/acp-integration/session/types.ts @@ -48,6 +48,8 @@ export interface CumulativeUsage { export interface SessionEmitterContext extends SessionUpdateSender { readonly sessionId: string; + /** History replay hook used to correlate emitted updates with disk records. */ + setActiveRecordId?: (id: string | null, timestamp?: string) => void; /** Optional message rewrite middleware for ACP message transformation. * Installed after history replay to avoid rewriting historical messages. */ messageRewriter?: MessageRewriteMiddleware; diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index 595284966cc..26b715796c5 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -98,6 +98,7 @@ export const SERVE_CAPABILITY_REGISTRY = { session_organization: { since: 'v1' }, session_export: { since: 'v1' }, session_transcript: { since: 'v1' }, + session_transcript_pagination: { since: 'v1' }, // Daemon supports the MCP client guardrail surface: an in-process // counter exposed on `GET /workspace/mcp`, a `--mcp-client-budget=N` // flag with `--mcp-budget-mode={enforce, warn, off}`, and a diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index f258ec28881..52fa3f2906f 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -276,6 +276,46 @@ function parseTranscriptCursorQuery( return rawCursor; } +function parseTranscriptRecordBoundaryQuery( + rawBoundary: unknown, + res: Response, +): string | undefined | null { + if (rawBoundary === undefined) return undefined; + if ( + typeof rawBoundary !== 'string' || + rawBoundary.trim() === '' || + rawBoundary.length > 200 + ) { + res.status(400).json({ + error: '`beforeRecordId` must be a non-empty record id', + code: 'invalid_transcript_cursor', + }); + return null; + } + return rawBoundary; +} + +function parseHistoryPageSize( + body: Record, + res: Response, +): number | undefined | null { + const value = body['historyPageSize']; + if (value === undefined) return undefined; + if ( + !Number.isSafeInteger(value) || + (value as number) < 1 || + (value as number) > SESSION_TRANSCRIPT_MAX_LIMIT + ) { + res.status(400).json({ + error: `\`historyPageSize\` must be between 1 and ${SESSION_TRANSCRIPT_MAX_LIMIT}`, + code: 'invalid_transcript_limit', + maxLimit: SESSION_TRANSCRIPT_MAX_LIMIT, + }); + return null; + } + return value as number; +} + function workspaceTranscriptCursorExceedsLimit( cursor: string, maxBytes = WORKSPACE_TRANSCRIPT_CURSOR_MAX_BYTES, @@ -1274,6 +1314,12 @@ export function registerSessionRoutes( releaseRestoreOwner(); return; } + const historyPageSize = + action === 'load' ? parseHistoryPageSize(body ?? {}, res) : undefined; + if (historyPageSize === null) { + releaseRestoreOwner(); + return; + } const clientId = parseClientIdHeader(req, res); if (clientId === null) { releaseRestoreOwner(); @@ -1295,6 +1341,7 @@ export function registerSessionRoutes( sessionId, workspaceCwd, historyReplay: 'response', + ...(historyPageSize !== undefined ? { historyPageSize } : {}), ...(clientId !== undefined ? { clientId } : {}), ...(approvalMode !== undefined ? { approvalMode } : {}), ...metadata, @@ -1512,6 +1559,18 @@ export function registerSessionRoutes( if (limit === null) return; const cursor = parseTranscriptCursorQuery(req.query['cursor'], res); if (cursor === null) return; + const beforeRecordId = parseTranscriptRecordBoundaryQuery( + req.query['beforeRecordId'], + res, + ); + if (beforeRecordId === null) return; + if (cursor !== undefined && beforeRecordId !== undefined) { + res.status(400).json({ + error: '`cursor` and `beforeRecordId` are mutually exclusive', + code: 'invalid_transcript_cursor', + }); + return; + } try { const result = await archiveCoordinator.runSharedMany( @@ -1528,6 +1587,7 @@ export function registerSessionRoutes( sessionId, ...(limit !== undefined ? { limit } : {}), ...(cursor !== undefined ? { cursor } : {}), + ...(beforeRecordId !== undefined ? { beforeRecordId } : {}), }); }, ); @@ -1563,6 +1623,18 @@ export function registerSessionRoutes( if (limit === null) return; const cursor = parseTranscriptCursorQuery(req.query['cursor'], res); if (cursor === null) return; + const beforeRecordId = parseTranscriptRecordBoundaryQuery( + req.query['beforeRecordId'], + res, + ); + if (beforeRecordId === null) return; + if (cursor !== undefined && beforeRecordId !== undefined) { + res.status(400).json({ + error: '`cursor` and `beforeRecordId` are mutually exclusive', + code: 'invalid_transcript_cursor', + }); + return; + } if (cursor !== undefined && workspaceTranscriptCursorExceedsLimit(cursor)) { res.status(400).json({ error: '`cursor` exceeds the maximum size', @@ -1588,6 +1660,7 @@ export function registerSessionRoutes( page = await reader.readPage(sessionId, { ...(limit !== undefined ? { limit } : {}), ...(cursor !== undefined ? { cursor } : {}), + ...(beforeRecordId !== undefined ? { beforeRecordId } : {}), maxBytes: WORKSPACE_TRANSCRIPT_PAGE_SOURCE_MAX_BYTES, }); } catch (error) { diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 6456a6a7b52..219d35a5953 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -269,6 +269,7 @@ const EXPECTED_STAGE1_FEATURES = [ 'session_organization', 'session_export', 'session_transcript', + 'session_transcript_pagination', // Issue #4175 PR 14. Always-on. Daemon supports the MCP client // guardrail surface (`--mcp-client-budget`, `clientCount` / // `budgets[]` on `/workspace/mcp`, `disabledReason: 'budget'` on @@ -8024,6 +8025,30 @@ describe('createServeApp', () => { }); describe('POST /session/:id/load and /resume', () => { + it('passes the requested initial history page size to load', async () => { + const bridge = fakeBridge(); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .post('/session/persisted-page/load') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ historyPageSize: 100 }); + + expect(res.status).toBe(200); + expect(bridge.loadCalls).toEqual([ + { + sessionId: 'persisted-page', + workspaceCwd: WS_BOUND, + historyReplay: 'response', + historyPageSize: 100, + }, + ]); + }); + it('falls back to bound workspace and uses the route session id', async () => { for (const action of ['load', 'resume'] as const) { const bridge = fakeBridge(); @@ -14788,6 +14813,32 @@ describe('createServeApp', () => { expect(bridge.resumeCalls).toHaveLength(0); }); + it('forwards an exclusive persisted-record boundary', async () => { + const sid = '55555555-bbbb-cccc-dddd-bbbbbbbbbbbb'; + const bridge = fakeBridge({ + sessionTranscriptImpl: async (req) => ({ + v: 1, + sessionId: req.sessionId, + events: [], + hasMore: false, + }), + }); + await writeTranscriptSession(sid); + const app = createServeApp({ ...baseOpts, workspace: wsDir }, undefined, { + bridge, + boundWorkspace: wsDir, + }); + + const res = await request(app) + .get(`/session/${sid}/transcript?limit=100&beforeRecordId=record-3`) + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(200); + expect(bridge.sessionTranscriptCalls).toEqual([ + { sessionId: sid, limit: 100, beforeRecordId: 'record-3' }, + ]); + }); + it('routes inactive active transcript pages through the owning workspace runtime', async () => { const sid = '55555555-bbbb-cccc-dddd-abababababab'; const secondaryDir = path.join(runtimeDir, 'secondary-workspace'); diff --git a/packages/core/src/services/session-transcript-reader.test.ts b/packages/core/src/services/session-transcript-reader.test.ts index 9f64fcfa0ca..dfbe32e9362 100644 --- a/packages/core/src/services/session-transcript-reader.test.ts +++ b/packages/core/src/services/session-transcript-reader.test.ts @@ -243,6 +243,130 @@ describe('SessionTranscriptReader', () => { expect(second.nextCursorState).toBeUndefined(); }); + it('pages backward before an exclusive active record boundary', async () => { + await writeRecords([ + record('u1', null, 'first prompt'), + record('a1', 'u1', 'first answer'), + record('u2', 'a1', 'second prompt'), + record('a2', 'u2', 'second answer'), + record('u3', 'a2', 'third prompt'), + record('a3', 'u3', 'third answer'), + ]); + + const reader = new SessionTranscriptReader(workspaceDir); + const first = await reader.readPage(sessionId, { + beforeRecordId: 'u3', + limit: 2, + }); + const second = await reader.readPage(sessionId, { + cursor: encodeCursor(first.nextCursorState!), + limit: 2, + }); + + expect(first.records.map((item) => item.uuid)).toEqual(['u2', 'a2']); + expect(first.direction).toBe('backward'); + expect(first.hasMore).toBe(true); + expect(first.nextCursorState).toMatchObject({ + position: 2, + direction: 'backward', + }); + expect(second.records.map((item) => item.uuid)).toEqual(['u1', 'a1']); + expect(second.hasMore).toBe(false); + }); + + it('keeps backward pages within a normal user turn boundary', async () => { + const toolCall = record('a-tool', 'u1', 'call tool'); + const toolResult = { + ...record('t1', 'a-tool', 'tool result'), + type: 'tool_result' as const, + }; + await writeRecords([ + record('u1', null, 'first prompt'), + toolCall, + toolResult, + record('a1', 't1', 'first answer'), + record('u2', 'a1', 'second prompt'), + record('a2', 'u2', 'second answer'), + ]); + + const page = await new SessionTranscriptReader(workspaceDir).readPage( + sessionId, + { beforeRecordId: 'u2', limit: 4 }, + ); + + expect(page.records.map((item) => item.uuid)).toEqual([ + 'u1', + 'a-tool', + 't1', + 'a1', + ]); + }); + + it('keeps a long user turn complete when it exceeds the record limit', async () => { + const toolCall = record('a-tool', 'u1', 'call tool'); + const toolResult = { + ...record('t1', 'a-tool', 'tool result'), + type: 'tool_result' as const, + }; + await writeRecords([ + record('u1', null, 'prompt'), + toolCall, + toolResult, + record('a-final', 't1', 'final answer'), + record('u2', 'a-final', 'next prompt'), + ]); + + const page = await new SessionTranscriptReader(workspaceDir).readPage( + sessionId, + { beforeRecordId: 'u2', limit: 2 }, + ); + + expect(page.records.map((item) => item.uuid)).toEqual([ + 'u1', + 'a-tool', + 't1', + 'a-final', + ]); + expect(page.hasMore).toBe(false); + }); + + it('rejects a backward turn that exceeds maxBytes after alignment', async () => { + const toolCall = record('a-tool', 'u1', 'call tool'); + const toolResult = { + ...record('t1', 'a-tool', 'tool result'), + type: 'tool_result' as const, + }; + const finalAnswer = record('a-final', 't1', 'final answer'); + await writeRecords([ + record('u1', null, 'prompt'), + toolCall, + toolResult, + finalAnswer, + record('u2', 'a-final', 'next prompt'), + ]); + + await expect( + new SessionTranscriptReader(workspaceDir).readPage(sessionId, { + beforeRecordId: 'u2', + limit: 2, + maxBytes: Buffer.byteLength(JSON.stringify(finalAnswer)), + }), + ).rejects.toBeInstanceOf(SessionTranscriptPageTooLargeError); + }); + + it('rejects a backward boundary outside the active chain', async () => { + await writeRecords([ + record('u1', null, 'root'), + record('a1', 'u1', 'answer'), + ]); + + await expect( + new SessionTranscriptReader(workspaceDir).readPage(sessionId, { + beforeRecordId: 'missing', + }), + ).rejects.toBeInstanceOf(InvalidSessionTranscriptCursorError); + }); + it('continues a frozen snapshot after new records are appended', async () => { const filePath = await writeRecords([ record('u1', null, 'root'), diff --git a/packages/core/src/services/session-transcript-reader.ts b/packages/core/src/services/session-transcript-reader.ts index eadc1bddf8a..ca6bdd3776e 100644 --- a/packages/core/src/services/session-transcript-reader.ts +++ b/packages/core/src/services/session-transcript-reader.ts @@ -66,6 +66,8 @@ export interface SessionTranscriptCursorState { fileIdentity: SessionTranscriptFileIdentity; snapshotSize: number; position: number; + /** Omitted for legacy oldest-to-newest cursors. */ + direction?: 'backward'; leafUuid: string; startTime: string; lastUpdated: string; @@ -74,6 +76,8 @@ export interface SessionTranscriptCursorState { export interface SessionTranscriptReadPageOptions { cursor?: string; + /** Start a newest-to-oldest snapshot immediately before this active record. */ + beforeRecordId?: string; limit?: number; maxBytes?: number; } @@ -84,6 +88,7 @@ export interface SessionTranscriptRecordPage { records: ChatRecord[]; gaps: HistoryGap[]; hasMore: boolean; + direction?: 'backward'; nextCursorState?: SessionTranscriptCursorState; replay?: unknown; startTime: string; @@ -104,6 +109,8 @@ interface RecordSegment { interface UuidIndexEntry { parentUuid: string | null; + type: ChatRecord['type']; + subtype?: ChatRecord['subtype']; segments: RecordSegment[]; } @@ -181,6 +188,7 @@ function cursorPayload( }, snapshotSize: state.snapshotSize, position: state.position, + ...(state.direction === 'backward' ? { direction: 'backward' } : {}), leafUuid: state.leafUuid, startTime: state.startTime, lastUpdated: state.lastUpdated, @@ -310,6 +318,8 @@ function decodeCursorState( !isFiniteNonNegativeInteger(fileIdentity['ino']) || !isFiniteNonNegativeInteger(parsed['snapshotSize']) || !isFiniteNonNegativeInteger(parsed['position']) || + (parsed['direction'] !== undefined && + parsed['direction'] !== 'backward') || typeof parsed['leafUuid'] !== 'string' || typeof parsed['startTime'] !== 'string' || typeof parsed['lastUpdated'] !== 'string' || @@ -327,6 +337,9 @@ function decodeCursorState( }, snapshotSize: parsed['snapshotSize'], position: parsed['position'], + ...(parsed['direction'] === 'backward' + ? { direction: 'backward' as const } + : {}), leafUuid: parsed['leafUuid'], startTime: parsed['startTime'], lastUpdated: parsed['lastUpdated'], @@ -447,6 +460,81 @@ function selectPageUuids( return selected; } +function isReplayTurnStart(index: TranscriptIndex, uuid: string): boolean { + const entry = index.byUuid.get(uuid); + return entry?.type === 'user' && entry.subtype !== 'mid_turn_user_message'; +} + +function selectBackwardPageUuids( + index: TranscriptIndex, + sessionId: string, + position: number, + limit: number, + maxBytes: number | undefined, +): { uuids: string[]; nextPosition: number } { + let start = Math.max(0, position - limit); + for (let i = start; i < position; i++) { + if (isReplayTurnStart(index, index.activeUuids[i]!)) { + start = i; + break; + } + } + while (start > 0 && !isReplayTurnStart(index, index.activeUuids[start]!)) { + start--; + } + + let selectedStart = position; + let selectedBytes = 0; + for (let i = position - 1; i >= start; i--) { + const uuid = index.activeUuids[i]!; + const bytes = recordSegmentBytes(index, uuid); + if ( + selectedStart === position && + maxBytes !== undefined && + bytes > maxBytes + ) { + throw new SessionTranscriptPageTooLargeError(sessionId, bytes, maxBytes); + } + if (maxBytes !== undefined && selectedBytes + bytes > maxBytes) break; + selectedStart = i; + selectedBytes += bytes; + } + + let alignedToReplayBoundary = false; + for (let i = selectedStart; i < position; i++) { + if (isReplayTurnStart(index, index.activeUuids[i]!)) { + selectedStart = i; + alignedToReplayBoundary = true; + break; + } + } + if (!alignedToReplayBoundary) { + while ( + selectedStart > 0 && + !isReplayTurnStart(index, index.activeUuids[selectedStart]!) + ) { + selectedStart--; + } + if (maxBytes !== undefined) { + const alignedBytes = index.activeUuids + .slice(selectedStart, position) + .reduce((total, uuid) => total + recordSegmentBytes(index, uuid), 0); + if (alignedBytes > maxBytes) { + throw new SessionTranscriptPageTooLargeError( + sessionId, + alignedBytes, + maxBytes, + ); + } + } + } + + return { + uuids: index.activeUuids.slice(selectedStart, position), + nextPosition: selectedStart, + }; +} + function fileIdentityFromStats(stats: fs.Stats): SessionTranscriptFileIdentity { return { dev: stats.dev, ino: stats.ino }; } @@ -783,6 +871,10 @@ async function buildIndex(params: { } else { byUuid.set(record.uuid, { parentUuid: record.parentUuid, + type: record.type, + ...(record.subtype !== undefined + ? { subtype: record.subtype } + : {}), segments: [segment], }); } @@ -944,6 +1036,9 @@ export class SessionTranscriptReader { ? (this.cursorCodec?.decode(options.cursor) ?? decodeSessionTranscriptCursor(options.cursor, this.workspaceCwd)) : undefined; + if (cursor && options.beforeRecordId !== undefined) { + throw new InvalidSessionTranscriptCursorError(); + } if (cursor && cursor.sessionId !== sessionId) { debugLogger.debug( `cursor session mismatch requested=${sessionId} cursor=${cursor.sessionId}`, @@ -983,7 +1078,26 @@ export class SessionTranscriptReader { throw new SessionTranscriptSnapshotUnavailableError(sessionId); } - const position = cursor?.position ?? 0; + const direction = + cursor?.direction ?? + (options.beforeRecordId !== undefined ? 'backward' : 'forward'); + let position = cursor?.position ?? 0; + if (!cursor && options.beforeRecordId !== undefined) { + if (options.beforeRecordId.length === 0) { + throw new InvalidSessionTranscriptCursorError(); + } + position = index.activeUuids.indexOf(options.beforeRecordId); + if (position < 0) { + throw new InvalidSessionTranscriptCursorError(); + } + } + if ( + !cursor && + direction === 'backward' && + options.beforeRecordId === undefined + ) { + position = index.activeUuids.length; + } if (position > index.activeUuids.length) { debugLogger.debug( `cursor position out of range session=${sessionId} ` + @@ -991,16 +1105,20 @@ export class SessionTranscriptReader { ); throw new InvalidSessionTranscriptCursorError(); } - const pageUuids = selectPageUuids( - index, - sessionId, - position, - limit, - maxBytes, - ); - const nextPosition = position + pageUuids.length; + const backwardPage = + direction === 'backward' + ? selectBackwardPageUuids(index, sessionId, position, limit, maxBytes) + : undefined; + const pageUuids = + backwardPage?.uuids ?? + selectPageUuids(index, sessionId, position, limit, maxBytes); + const nextPosition = + backwardPage?.nextPosition ?? position + pageUuids.length; const records = await readAggregatedRecords(index, pageUuids); - const hasMore = nextPosition < index.activeUuids.length; + const hasMore = + direction === 'backward' + ? nextPosition > 0 + : nextPosition < index.activeUuids.length; const nextCursorState: SessionTranscriptCursorState | undefined = hasMore ? { v: SESSION_TRANSCRIPT_CURSOR_VERSION, @@ -1008,6 +1126,9 @@ export class SessionTranscriptReader { fileIdentity, snapshotSize, position: nextPosition, + ...(direction === 'backward' + ? { direction: 'backward' as const } + : {}), leafUuid: index.leafUuid, startTime: index.startTime, lastUpdated: index.lastUpdated, @@ -1026,6 +1147,7 @@ export class SessionTranscriptReader { records, gaps: index.gaps, hasMore, + ...(direction === 'backward' ? { direction: 'backward' as const } : {}), ...(nextCursorState ? { nextCursorState } : {}), ...(cursor?.replay !== undefined ? { replay: cursor.replay } : {}), startTime: index.startTime, diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 5451bcd6655..036e29ca17a 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -300,6 +300,19 @@ const MCP_RESTART_DEFAULT_TIMEOUT_MS = const CLIENT_ID_HEADER = 'X-Qwen-Client-Id'; const urlEncode = encodeURIComponent; +function transcriptPageSuffix( + opts: DaemonSessionTranscriptPageOptions, +): string { + const query = new URLSearchParams(); + if (opts.cursor !== undefined) query.set('cursor', opts.cursor); + if (opts.beforeRecordId !== undefined) { + query.set('beforeRecordId', opts.beforeRecordId); + } + if (opts.limit !== undefined) query.set('limit', String(opts.limit)); + const value = query.toString(); + return value ? `?${value}` : ''; +} + function normalizePermissionRuleInput(rule: string): string { const trimmed = rule.trim(); if (!trimmed) { @@ -453,6 +466,8 @@ export interface RestoreSessionRequest { */ workspaceCwd?: string; approvalMode?: string; + /** Latest persisted records to include in the initial load replay. */ + historyPageSize?: number; } export interface PromptRequest { @@ -2093,12 +2108,8 @@ export class DaemonClient { sessionId: string, opts: DaemonSessionTranscriptPageOptions = {}, ): Promise { - const params = new URLSearchParams(); - if (opts.cursor !== undefined) params.set('cursor', opts.cursor); - if (opts.limit !== undefined) params.set('limit', String(opts.limit)); - const query = params.toString(); return await this.jsonRequest( - `/session/${urlEncode(sessionId)}/transcript${query ? `?${query}` : ''}`, + `/session/${urlEncode(sessionId)}/transcript${transcriptPageSuffix(opts)}`, 'GET /session/:id/transcript', { clientId: opts.clientId, @@ -2349,6 +2360,9 @@ export class DaemonClient { ...(req.approvalMode !== undefined ? { approvalMode: req.approvalMode } : {}), + ...(action === 'load' && req.historyPageSize !== undefined + ? { historyPageSize: req.historyPageSize } + : {}), }), }, async (res) => { @@ -4306,13 +4320,9 @@ export class WorkspaceDaemonClient { sessionId: string, opts: DaemonSessionTranscriptPageOptions = {}, ): Promise { - const query = new URLSearchParams(); - if (opts.cursor !== undefined) query.set('cursor', opts.cursor); - if (opts.limit !== undefined) query.set('limit', String(opts.limit)); - const suffix = query.size > 0 ? `?${query.toString()}` : ''; return this.client.workspaceJsonRequest( this.workspaceSelector, - `/session/${urlEncode(sessionId)}/transcript${suffix}`, + `/session/${urlEncode(sessionId)}/transcript${transcriptPageSuffix(opts)}`, 'GET /workspaces/:workspace/session/:id/transcript', { clientId: opts.clientId, mode: 'rest' }, ); diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index 3220ce5f1b3..d0e6e77da3f 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -70,6 +70,8 @@ export interface DaemonSessionClientOptions { lastEventId?: number; /** Compacted replay snapshot from daemon load response. */ replaySnapshot?: DaemonReplaySnapshot; + /** True when older persisted records precede the replay snapshot. */ + historyHasMore?: boolean; /** * Local per-session prompt cap. The counter is shared with the parent * `DaemonClient`; other session clients using the same parent instance @@ -105,6 +107,7 @@ export class DaemonSessionClient { readonly state: DaemonSessionState; readonly replaySnapshot: DaemonReplaySnapshot; readonly hasActivePrompt: boolean; + readonly historyHasMore: boolean; private lastSeenEventId: number | undefined; private subscriptionActive = false; /** In-flight `reattach()` so concurrent prompts re-register only once. */ @@ -123,6 +126,7 @@ export class DaemonSessionClient { this.session = { ...opts.session }; this.state = { ...(opts.state ?? {}) }; this.hasActivePrompt = opts.hasActivePrompt ?? false; + this.historyHasMore = opts.historyHasMore ?? false; this.replaySnapshot = opts.replaySnapshot ?? { compactedReplay: [], liveJournal: [], @@ -198,6 +202,7 @@ export class DaemonSessionClient { hasActivePrompt, compactedReplay, liveJournal, + historyHasMore, lastEventId: serverLastEventId, ...session } = await client.loadSession(sessionId, req, clientId); @@ -211,6 +216,7 @@ export class DaemonSessionClient { compactedReplay: compactedReplay ?? [], liveJournal: liveJournal ?? [], }, + historyHasMore, }); } diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 4267d83cae2..afcf5f51d72 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -497,6 +497,8 @@ export interface DaemonRestoredSession extends DaemonSession { compactedReplay?: DaemonEvent[]; /** Raw events since last turn boundary — current incomplete turn (load only). */ liveJournal?: DaemonEvent[]; + /** True when older persisted records precede this load replay page. */ + historyHasMore?: boolean; /** Event bus watermark — used as initial SSE cursor. */ lastEventId?: number; } @@ -616,6 +618,8 @@ export interface DaemonSessionExportResult { export interface DaemonSessionTranscriptPageOptions { cursor?: string; + /** Start a newest-to-oldest page before this persisted record UUID. */ + beforeRecordId?: string; limit?: number; clientId?: string; } diff --git a/packages/web-shell/client/App.tsx b/packages/web-shell/client/App.tsx index cad39004a5f..b7255e8d0e2 100644 --- a/packages/web-shell/client/App.tsx +++ b/packages/web-shell/client/App.tsx @@ -20,6 +20,7 @@ import { useSessionNotices, useStreamingState, useTranscriptBlocks, + useTranscriptHistory, useTranscriptStore, useWorkspace, useWorkspaceActions, @@ -1186,6 +1187,7 @@ export function App({ const store = useTranscriptStore(); const blocks = useTranscriptBlocks(); const connection = useConnection(); + const transcriptHistory = useTranscriptHistory(); const workspace = useWorkspace(); const workspaces = useMemo(() => { const capabilityWorkspaces = workspace.capabilities?.workspaces ?? []; @@ -6431,6 +6433,12 @@ export function App({ onShowContextDetail={handleShowContextDetail} loadingTranscript={connection.loadingTranscript} catchingUp={connection.catchingUp} + hasOlderHistory={transcriptHistory.hasMore} + loadingOlderHistory={transcriptHistory.loading} + historyCapacityReached={ + transcriptHistory.capacityReached + } + onLoadOlderHistory={transcriptHistory.loadMore} isResponding={streamingState !== 'idle'} activeTurnStartedAt={activeTurnStartedAt} workspaceCwd={connection.workspaceCwd || ''} diff --git a/packages/web-shell/client/components/ChatPane.tsx b/packages/web-shell/client/components/ChatPane.tsx index 088782e545c..521bcc04fae 100644 --- a/packages/web-shell/client/components/ChatPane.tsx +++ b/packages/web-shell/client/components/ChatPane.tsx @@ -12,6 +12,7 @@ import { useDaemonFollowupSuggestion, useStreamingState, useTranscriptBlocks, + useTranscriptHistory, useTranscriptStore, useWorkspace, useWorkspaceActions, @@ -123,6 +124,7 @@ export function ChatPane({ const workspace = useWorkspace(); const messages = useMessages(t); const blocks = useTranscriptBlocks(); + const transcriptHistory = useTranscriptHistory(); const store = useTranscriptStore(); const streamingState = useStreamingState(); const { artifacts } = useSessionArtifacts(); @@ -500,6 +502,10 @@ export function ChatPane({ pendingApproval={pendingToolApproval} loadingTranscript={connection.loadingTranscript} catchingUp={connection.catchingUp} + hasOlderHistory={transcriptHistory.hasMore} + loadingOlderHistory={transcriptHistory.loading} + historyCapacityReached={transcriptHistory.capacityReached} + onLoadOlderHistory={transcriptHistory.loadMore} isResponding={isResponding} workspaceCwd={connection.workspaceCwd || ''} hideSessionTimeline diff --git a/packages/web-shell/client/components/MessageList.dom.test.tsx b/packages/web-shell/client/components/MessageList.dom.test.tsx index 6a56ca161f4..aba548aa413 100644 --- a/packages/web-shell/client/components/MessageList.dom.test.tsx +++ b/packages/web-shell/client/components/MessageList.dom.test.tsx @@ -110,7 +110,7 @@ class ResizeObserverStub { unobserve() {} disconnect() {} } -(globalThis as { ResizeObserver?: unknown }).ResizeObserver ??= +(globalThis as { ResizeObserver?: unknown }).ResizeObserver = ResizeObserverStub; if (!Element.prototype.scrollIntoView) { Element.prototype.scrollIntoView = () => {}; @@ -199,6 +199,10 @@ function mount( hideSessionTimeline?: boolean; loadingTranscript?: boolean; catchingUp?: boolean; + hasOlderHistory?: boolean; + loadingOlderHistory?: boolean; + historyCapacityReached?: boolean; + onLoadOlderHistory?: () => Promise; isResponding?: boolean; onCanScrollToBottomChange?: (canScrollToBottom: boolean) => void; customization?: WebShellCustomization; @@ -218,6 +222,10 @@ function mount( hideSessionTimeline={opts.hideSessionTimeline} loadingTranscript={opts.loadingTranscript} catchingUp={opts.catchingUp} + hasOlderHistory={opts.hasOlderHistory} + loadingOlderHistory={opts.loadingOlderHistory} + historyCapacityReached={opts.historyCapacityReached} + onLoadOlderHistory={opts.onLoadOlderHistory} isResponding={opts.isResponding} onCanScrollToBottomChange={opts.onCanScrollToBottomChange} /> @@ -1134,6 +1142,113 @@ describe('MessageList — turn collapse (DOM)', () => { ).toBeNull(); }); + it('loads earlier history once when the transcript reaches the top', async () => { + Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { + configurable: true, + value: 1200, + }); + Object.defineProperty(HTMLElement.prototype, 'clientHeight', { + configurable: true, + value: 600, + }); + const onLoadOlderHistory = vi.fn().mockResolvedValue(undefined); + const c = mount([userMsg('u1')], undefined, { + hasOlderHistory: true, + onLoadOlderHistory, + }); + const list = c.querySelector('[data-web-shell-message-list]'); + Object.defineProperty(list, 'scrollTop', { + configurable: true, + writable: true, + value: 0, + }); + + await act(async () => { + list?.dispatchEvent(new Event('scroll')); + list?.dispatchEvent(new Event('scroll')); + await Promise.resolve(); + }); + + expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); + }); + + it('loads earlier history when the transcript does not overflow', async () => { + Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { + configurable: true, + value: 300, + }); + Object.defineProperty(HTMLElement.prototype, 'clientHeight', { + configurable: true, + value: 600, + }); + const onLoadOlderHistory = vi.fn().mockResolvedValue(undefined); + + mount([userMsg('u1')], undefined, { + hasOlderHistory: true, + onLoadOlderHistory, + }); + await act(async () => { + await Promise.resolve(); + }); + + expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); + }); + + it('loads earlier history when a resize removes the overflow', async () => { + let clientHeight = 600; + Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { + configurable: true, + value: 1200, + }); + Object.defineProperty(HTMLElement.prototype, 'clientHeight', { + configurable: true, + get: () => clientHeight, + }); + const onLoadOlderHistory = vi.fn().mockResolvedValue(undefined); + + const c = mount([userMsg('u1')], undefined, { + hasOlderHistory: true, + onLoadOlderHistory, + }); + const list = c.querySelector( + '[data-web-shell-message-list]', + ) as HTMLElement; + expect(onLoadOlderHistory).not.toHaveBeenCalled(); + expect(list.scrollHeight).toBe(1200); + expect(list.clientHeight).toBe(600); + + clientHeight = 1200; + expect(list.clientHeight).toBe(1200); + await act(async () => { + triggerResizeObservers(); + await Promise.resolve(); + }); + + expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); + }); + + it('shows a status while loading earlier history', () => { + const c = mount([userMsg('u1')], undefined, { + hasOlderHistory: true, + loadingOlderHistory: true, + }); + + expect(c.querySelector('[role="status"]')?.textContent).toBe( + 'Loading earlier messages…', + ); + expect(c.querySelector('button')).toBeNull(); + }); + + it('shows when the history display limit is reached', () => { + const c = mount([userMsg('u1')], undefined, { + historyCapacityReached: true, + }); + + expect(c.querySelector('[role="status"]')?.textContent).toBe( + 'History display limit reached. Earlier messages remain saved.', + ); + }); + it('does not smooth-scroll when existing session history loads after an empty render', () => { const scrollTo = vi.fn(); let scrollTop = 0; diff --git a/packages/web-shell/client/components/MessageList.module.css b/packages/web-shell/client/components/MessageList.module.css index 418887445b2..01a8a884e3e 100644 --- a/packages/web-shell/client/components/MessageList.module.css +++ b/packages/web-shell/client/components/MessageList.module.css @@ -29,6 +29,12 @@ border-radius: 3px; } +.historyStatus { + color: var(--muted-foreground); + text-align: center; + padding: 4px 0 12px; +} + .list > * { width: min( 100%, diff --git a/packages/web-shell/client/components/MessageList.tsx b/packages/web-shell/client/components/MessageList.tsx index dd40313eafb..7326dedadba 100644 --- a/packages/web-shell/client/components/MessageList.tsx +++ b/packages/web-shell/client/components/MessageList.tsx @@ -56,6 +56,10 @@ interface MessageListProps { onShowContextDetail?: () => void; loadingTranscript?: boolean; catchingUp?: boolean; + hasOlderHistory?: boolean; + loadingOlderHistory?: boolean; + historyCapacityReached?: boolean; + onLoadOlderHistory?: () => Promise; /** * True while the agent is still answering. The newest turn then stays * expanded and un-collapsible so streaming output is never hidden. @@ -1541,6 +1545,7 @@ const ESTIMATE_MESSAGE = 80; const ESTIMATE_TURN_COLLAPSE = 32; const ESTIMATE_TAIL = 240; const FOLLOW_BOTTOM_THRESHOLD_PX = 30; +const LOAD_OLDER_HISTORY_THRESHOLD_PX = 48; export const VIRTUAL_SCROLL_THRESHOLD = 200; const SESSION_TIMELINE_MIN_VISIBLE_ENTRIES = 4; @@ -2174,6 +2179,10 @@ export const MessageList = memo( onShowContextDetail, loadingTranscript, catchingUp, + hasOlderHistory = false, + loadingOlderHistory = false, + historyCapacityReached = false, + onLoadOlderHistory, isResponding = false, activeTurnStartedAt, welcomeHeader, @@ -2309,6 +2318,7 @@ export const MessageList = memo( const followPausedByUserRef = useRef(false); const userScrollIntentUntil = useRef(0); const lastScrollTop = useRef(0); + const olderHistoryLoadInFlight = useRef(false); const scrollCooldown = useRef(false); const scrollCooldownCount = useRef(0); const sessionTimelineFrame = useRef(null); @@ -2908,19 +2918,49 @@ export const MessageList = memo( performScrollToRow(idx + headerOffset, pending); }, [visibleItems, headerOffset, performScrollToRow]); + const loadOlderHistory = useCallback(async () => { + const el = containerRef.current; + if ( + !el || + !onLoadOlderHistory || + loadingOlderHistory || + olderHistoryLoadInFlight.current + ) { + return; + } + olderHistoryLoadInFlight.current = true; + const previousHeight = el.scrollHeight; + const previousTop = el.scrollTop; + followPausedByUserRef.current = true; + try { + await onLoadOlderHistory(); + window.requestAnimationFrame(() => { + const current = containerRef.current; + if (!current) return; + current.scrollTop = + previousTop + Math.max(0, current.scrollHeight - previousHeight); + }); + } finally { + olderHistoryLoadInFlight.current = false; + } + }, [loadingOlderHistory, onLoadOlderHistory]); + // Rules 2 & 3: detect scroll direction to toggle follow mode. // Runs synchronously in the scroll handler — no rAF needed since // the browser already coalesces scroll events. const handleScroll = useCallback(() => { const el = getScrollElement(); if (!el) return; + const curr = el.scrollTop; + if (hasOlderHistory && curr <= LOAD_OLDER_HISTORY_THRESHOLD_PX) { + void loadOlderHistory(); + } if (scrollCooldown.current) { - lastScrollTop.current = el.scrollTop; + lastScrollTop.current = curr; return; } scheduleSessionTimelineRangeUpdate(); const prev = lastScrollTop.current; - const curr = el.scrollTop; lastScrollTop.current = curr; const distanceFromBottom = el.scrollHeight - curr - el.clientHeight; scheduleScrollOverflowReport(); @@ -2951,6 +2991,8 @@ export const MessageList = memo( } }, [ getScrollElement, + hasOlderHistory, + loadOlderHistory, scheduleScrollOverflowReport, scheduleSessionTimelineRangeUpdate, setShouldFollow, @@ -2963,6 +3005,31 @@ export const MessageList = memo( return () => el.removeEventListener('scroll', handleScroll); }, [getScrollElement, handleScroll]); + const loadOlderHistoryIfUnderfilled = useCallback(() => { + if ( + !hasOlderHistory || + loadingOlderHistory || + catchingUp || + showLoadingSkeleton + ) { + return; + } + const el = getScrollElement(); + if (!el || el.scrollHeight > el.clientHeight + 1) return; + void loadOlderHistory(); + }, [ + catchingUp, + getScrollElement, + hasOlderHistory, + loadOlderHistory, + loadingOlderHistory, + showLoadingSkeleton, + ]); + + useEffect(() => { + loadOlderHistoryIfUnderfilled(); + }, [loadOlderHistoryIfUnderfilled, totalVirtualSize]); + useEffect(() => { const el = getScrollElement(); if (!el) return; @@ -3006,7 +3073,10 @@ export const MessageList = memo( useEffect(() => { const el = getScrollElement(); if (!el || typeof ResizeObserver === 'undefined') return; - const observer = new ResizeObserver(scheduleScrollOverflowReport); + const observer = new ResizeObserver(() => { + scheduleScrollOverflowReport(); + loadOlderHistoryIfUnderfilled(); + }); observer.observe(el); for (const child of Array.from(el.children)) { observer.observe(child); @@ -3028,7 +3098,11 @@ export const MessageList = memo( observer.disconnect(); mutationObserver.disconnect(); }; - }, [getScrollElement, scheduleScrollOverflowReport]); + }, [ + getScrollElement, + loadOlderHistoryIfUnderfilled, + scheduleScrollOverflowReport, + ]); // Clear screen (e.g. /clear) → reset to follow mode, drop stale per-turn // collapse overrides, and disarm any deferred scroll so it can't fire @@ -3443,6 +3517,16 @@ export const MessageList = memo( {showLoadingSkeleton && ( )} + {loadingOlderHistory && !showLoadingSkeleton && ( +
+ {t('history.loadingEarlier')} +
+ )} + {historyCapacityReached && !showLoadingSkeleton && ( +
+ {t('history.capacityReached')} +
+ )} `. */ DaemonSessionProviderProps, + DaemonTranscriptHistory, /** Streaming lifecycle: `'idle' | 'waiting' | 'responding' | 'thinking'`. */ DaemonStreamingState, /** Prompt submission status: `'idle' | 'waiting' | 'streaming'`. */ diff --git a/packages/webui/src/daemon/index.ts b/packages/webui/src/daemon/index.ts index a7c48a935e3..95583304561 100644 --- a/packages/webui/src/daemon/index.ts +++ b/packages/webui/src/daemon/index.ts @@ -16,6 +16,7 @@ export { useDaemonStreamingState, useDaemonSession, useDaemonTranscriptBlocks, + useDaemonTranscriptHistory, useDaemonTranscriptState, useDaemonTranscriptStore, useDaemonWorkspaceEventSignals, @@ -47,6 +48,7 @@ export type { DaemonSessionContextValue, DaemonSessionNotice, DaemonSessionProviderProps, + DaemonTranscriptHistory, DaemonTokenUsage, DaemonTodoItem, DaemonTodoList, diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index 60817cef97e..6a6da0a62c8 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -26,6 +26,7 @@ import { useDaemonPromptStatus, useDaemonStreamingState, useDaemonTranscriptBlocks, + useDaemonTranscriptHistory, useDaemonTranscriptState, useDaemonTranscriptStore, useDaemonWorkspaceEventSignals, @@ -50,6 +51,7 @@ interface MockSession { clientId: string; state?: Record; hasActivePrompt?: boolean; + historyHasMore?: boolean; client?: MockClient; lastEventId?: number; setLastEventId: (lastEventId: number | undefined) => void; @@ -136,6 +138,10 @@ interface MockClient { displayName: string; clientId?: string; }>; + getSessionTranscriptPage: ( + sessionId: string, + opts: unknown, + ) => Promise; } const sdkMocks = vi.hoisted(() => { @@ -165,6 +171,7 @@ const sdkMocks = vi.hoisted(() => { const getPendingPrompts = vi.fn(); const removePendingPrompt = vi.fn(); const branchSession = vi.fn(); + const getSessionTranscriptPage = vi.fn(); class MockDaemonClient { constructor(_opts: unknown) {} @@ -197,6 +204,7 @@ const sdkMocks = vi.hoisted(() => { getPendingPrompts = getPendingPrompts; removePendingPrompt = removePendingPrompt; branchSession = branchSession; + getSessionTranscriptPage = getSessionTranscriptPage; dispose = vi.fn(); } @@ -237,6 +245,7 @@ const sdkMocks = vi.hoisted(() => { getPendingPrompts, removePendingPrompt, branchSession, + getSessionTranscriptPage, reset() { sessions.length = 0; capabilities.mockReset(); @@ -340,6 +349,7 @@ const sdkMocks = vi.hoisted(() => { displayName: 'Branch Session', clientId: 'branch-client', }); + getSessionTranscriptPage.mockReset(); MockDaemonSessionClient.createOrAttach.mockReset(); MockDaemonSessionClient.createOrAttach.mockImplementation( async (client: unknown, _req: unknown): Promise => @@ -8472,6 +8482,320 @@ describe('DaemonSessionProvider', () => { ]); }); + it('keeps history pagination disabled when the server does not advertise it', async () => { + const session = createMockSession({ + sessionId: 'session-history-unsupported', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [ + { + id: 1, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: 'complete history' }, + _meta: { recordId: 'record-1' }, + }, + }, + }, + ], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + let history: ReturnType | undefined; + + function Harness() { + history = useDaemonTranscriptHistory(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + }); + + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledWith( + expect.anything(), + session.sessionId, + { workspaceCwd: '/mock-workspace' }, + expect.any(String), + ); + expect(history?.hasMore).toBe(false); + await act(async () => history?.loadMore()); + expect(sdkMocks.getSessionTranscriptPage).not.toHaveBeenCalled(); + }); + + it('prepends an older transcript page from the first replay record', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const replayEvent = ( + id: number, + text: string, + recordId?: string, + ): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + ...(recordId ? { _meta: { recordId } } : {}), + }, + }, + }); + const session = createMockSession({ + sessionId: 'session-history-page', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [replayEvent(2, 'recent prompt', 'record-2')], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + sdkMocks.getSessionTranscriptPage.mockResolvedValue({ + v: 1, + sessionId: session.sessionId, + events: [replayEvent(1, 'older prompt')], + hasMore: false, + }); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + }); + expect(sdkMocks.MockDaemonSessionClient.load).toHaveBeenCalledWith( + expect.anything(), + session.sessionId, + { workspaceCwd: '/mock-workspace', historyPageSize: 25 }, + expect.any(String), + ); + expect(history?.hasMore).toBe(true); + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + + expect(sdkMocks.getSessionTranscriptPage).toHaveBeenCalledWith( + session.sessionId, + { + beforeRecordId: 'record-2', + limit: 25, + clientId: session.clientId, + }, + ); + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toEqual(['older prompt', 'recent prompt']); + expect(history?.hasMore).toBe(false); + }); + + it('reports a partial older page without changing the transcript', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const replayEvent = ( + id: number, + text: string, + recordId?: string, + ): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + ...(recordId ? { _meta: { recordId } } : {}), + }, + }, + }); + const session = createMockSession({ + sessionId: 'session-partial-history-page', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [replayEvent(2, 'recent prompt', 'record-2')], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + sdkMocks.getSessionTranscriptPage.mockResolvedValue({ + v: 1, + sessionId: session.sessionId, + events: [replayEvent(1, 'partial older prompt')], + hasMore: false, + partial: true, + replayError: 'Replay conversion failed for this page', + }); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let notices: readonly DaemonSessionNotice[] = []; + + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + notices = useDaemonSessionNotices().notices; + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + }); + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toEqual(['recent prompt']); + expect(history?.hasMore).toBe(false); + expect(notices.at(-1)).toMatchObject({ + code: 'daemon.transcript_history.failed', + message: 'Failed to load earlier session history', + debugMessage: 'Replay conversion failed for this page', + }); + }); + + it('keeps an oversized initial replay intact and stops older pagination', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const event = ( + id: number, + kind: 'user_message_chunk' | 'agent_message_chunk', + text: string, + recordId: string, + ): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: kind, + content: { type: 'text', text }, + _meta: { recordId }, + }, + }, + }); + const session = createMockSession({ + sessionId: 'session-oversized-initial-page', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [ + event(1, 'user_message_chunk', 'recent prompt', 'record-1'), + event(2, 'agent_message_chunk', 'recent answer', 'record-2'), + ], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + maxBlocks: 1, + }); + + expect(blocks).toMatchObject([ + { kind: 'user', text: 'recent prompt' }, + { kind: 'assistant', text: 'recent answer' }, + ]); + expect(history).toMatchObject({ + hasMore: false, + loading: false, + capacityReached: true, + }); + expect(sdkMocks.getSessionTranscriptPage).not.toHaveBeenCalled(); + }); + + it('rejects a terminal older page that does not fit atomically', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const replayEvent = ( + id: number, + text: string, + recordId?: string, + ): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + ...(recordId ? { _meta: { recordId } } : {}), + }, + }, + }); + const session = createMockSession({ + sessionId: 'session-history-capacity', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [replayEvent(2, 'recent prompt', 'record-2')], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + sdkMocks.getSessionTranscriptPage.mockResolvedValue({ + v: 1, + sessionId: session.sessionId, + events: [replayEvent(1, 'older prompt')], + hasMore: false, + }); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + maxBlocks: 1, + }); + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + + expect(history).toMatchObject({ + hasMore: false, + loading: false, + capacityReached: true, + }); + expect(blocks).toMatchObject([{ kind: 'user', text: 'recent prompt' }]); + }); + async function renderWithProvider( children: ReactNode, props: Partial = {}, @@ -8542,6 +8866,7 @@ function createMockSession(opts: Partial = {}): MockSession { clientId: opts.clientId ?? 'client-1', state: opts.state ?? {}, hasActivePrompt: opts.hasActivePrompt ?? false, + historyHasMore: opts.historyHasMore ?? false, lastEventId: opts.lastEventId, setLastEventId: opts.setLastEventId ?? diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index a60ba52c4d7..f5d0008f1c3 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -115,6 +115,15 @@ export type { SendPromptOptions, } from './types.js'; +export interface DaemonTranscriptHistory { + hasMore: boolean; + loading: boolean; + capacityReached: boolean; + loadMore(): Promise; +} + +const SESSION_TRANSCRIPT_PAGINATION_FEATURE = 'session_transcript_pagination'; + function assistantDoneFromTurnEvent( event: DaemonEvent, reason: string, @@ -128,6 +137,51 @@ function assistantDoneFromTurnEvent( }; } +function getPersistedReplayRecordId(event: DaemonEvent): string | undefined { + if (event.type !== 'session_update') { + return undefined; + } + try { + if (!isRecord(event.data)) return undefined; + const update = event.data['update']; + const meta = isRecord(update) ? update['_meta'] : event.data['_meta']; + return isRecord(meta) ? getString(meta, 'recordId') : undefined; + } catch { + return undefined; + } +} + +function prependTranscriptHistory( + store: DaemonTranscriptStore, + events: DaemonUiEvent[], + maxBlocks: number, +): boolean { + const current = store.getSnapshot(); + const historyStore = createDaemonTranscriptStore({ + maxBlocks: Number.MAX_SAFE_INTEGER, + nextOrdinal: current.nextOrdinal, + }); + historyStore.dispatch(events); + const history = historyStore.getSnapshot(); + if (history.blocks.length + current.blocks.length > maxBlocks) { + return false; + } + store.reset({ + ...current, + blocks: [...history.blocks, ...current.blocks], + nextOrdinal: history.nextOrdinal, + toolBlockByCallId: { + ...history.toolBlockByCallId, + ...current.toolBlockByCallId, + }, + permissionBlockByRequestId: { + ...history.permissionBlockByRequestId, + ...current.permissionBlockByRequestId, + }, + }); + return true; +} + const DaemonStoreContext = createContext( undefined, ); @@ -137,6 +191,9 @@ const DaemonConnectionContext = createContext< const DaemonActionsContext = createContext( undefined, ); +const DaemonTranscriptHistoryContext = createContext< + DaemonTranscriptHistory | undefined +>(undefined); const DaemonPromptStatusContext = createContext( undefined, ); @@ -208,6 +265,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { createSessionRequest, maxQueued = 1024, maxBlocks = DEFAULT_MAX_BLOCKS, + historyPageSize, suppressOwnUserEcho = true, includeRawEvent = false, autoConnect = true, @@ -249,6 +307,19 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { [maxBlocks], ); const sessionRef = useRef(undefined); + const transcriptHistoryRef = useRef<{ + sessionId?: string; + beforeRecordId?: string; + cursor?: string; + hasMore: boolean; + loading: boolean; + capacityReached: boolean; + }>({ hasMore: false, loading: false, capacityReached: false }); + const [transcriptHistoryState, setTranscriptHistoryState] = useState({ + hasMore: false, + loading: false, + capacityReached: false, + }); const lastSessionIdRef = useRef(undefined); const activePromptsRef = useRef>(new Map()); const settledPromptsRef = useRef>(new Map()); @@ -273,6 +344,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { const eventOptionsRef = useRef({ suppressOwnUserEcho, includeRawEvent }); const reconnectConfigRef = useRef({ reconnectDelayMs, maxReconnectDelayMs }); const loadWarningsRef = useRef(loadWarnings); + const historyPageSizeRef = useRef(historyPageSize); const clientIdRef = useRef(undefined); if (!clientIdRef.current || clientId) { clientIdRef.current = getStableClientId(clientId); @@ -280,6 +352,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { eventOptionsRef.current = { suppressOwnUserEcho, includeRawEvent }; reconnectConfigRef.current = { reconnectDelayMs, maxReconnectDelayMs }; loadWarningsRef.current = loadWarnings; + historyPageSizeRef.current = historyPageSize; const modelServiceId = createSessionRequest?.modelServiceId; const sessionScope = createSessionRequest?.sessionScope; const createSessionRequestRef = useRef(createSessionRequest); @@ -514,6 +587,9 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { : await client.capabilities()); if (disposed || abort.signal.aborted) return; capabilities = caps; + const historyPaginationSupported = + Array.isArray(caps.features) && + caps.features.includes(SESSION_TRANSCRIPT_PAGINATION_FEATURE); heartbeatSupportedRef.current = Array.isArray(caps.features) && caps.features.includes('client_heartbeat'); @@ -666,14 +742,27 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ? await restoreMethod( client, restoreSessionId, - { workspaceCwd: effectWorkspaceCwd }, + { + workspaceCwd: effectWorkspaceCwd, + ...(historyPaginationSupported && + restoreMode === 'load' && + historyPageSizeRef.current !== undefined + ? { historyPageSize: historyPageSizeRef.current } + : {}), + }, requestClientId, ) : reconnectSessionId ? await DaemonSessionClient.load( client, reconnectSessionId, - { workspaceCwd: effectWorkspaceCwd }, + { + workspaceCwd: effectWorkspaceCwd, + ...(historyPaginationSupported && + historyPageSizeRef.current !== undefined + ? { historyPageSize: historyPageSizeRef.current } + : {}), + }, requestClientId, ) : await DaemonSessionClient.createOrAttach( @@ -814,6 +903,30 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { // only fires once with the fully-populated state. const { compactedReplay, liveJournal } = activeSession.replaySnapshot; const replayEvents = [...compactedReplay, ...liveJournal]; + const firstPersistedRecordId = replayEvents + .map(getPersistedReplayRecordId) + .find((recordId): recordId is string => recordId !== undefined); + const historyHasMore = + Array.isArray(capabilities?.features) && + capabilities.features.includes( + SESSION_TRANSCRIPT_PAGINATION_FEATURE, + ) && + activeSession.historyHasMore && + firstPersistedRecordId !== undefined; + transcriptHistoryRef.current = { + sessionId: activeSession.sessionId, + ...(firstPersistedRecordId !== undefined + ? { beforeRecordId: firstPersistedRecordId } + : {}), + hasMore: historyHasMore, + loading: false, + capacityReached: false, + }; + setTranscriptHistoryState({ + hasMore: historyHasMore, + loading: false, + capacityReached: false, + }); const replayInjected = shouldInjectReplaySnapshot && replayEvents.length > 0; if (needsStoreReset && !replayInjected) { @@ -874,13 +987,33 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ); } } - if (needsStoreReset) { - store.reset(); + let replayExceededCapacity = false; + if (needsStoreReset || store.getSnapshot().blocks.length === 0) { + const replayStore = createDaemonTranscriptStore({ + maxBlocks: Number.MAX_SAFE_INTEGER, + }); + replayStore.dispatch(allUiEvents); + const replayState = replayStore.getSnapshot(); + replayExceededCapacity = replayState.blocks.length > maxBlocks; + store.reset({ + ...replayState, + maxBlocks: Math.max(maxBlocks, replayState.blocks.length), + }); + } else if (allUiEvents.length > 0) { + store.dispatch(allUiEvents); } if (allUiEvents.length > 0) { - store.dispatch(allUiEvents); bumpWorkspaceEventSignals(allUiEvents, setWorkspaceEventSignals); } + if (replayExceededCapacity && historyHasMore) { + transcriptHistoryRef.current.hasMore = false; + transcriptHistoryRef.current.capacityReached = true; + setTranscriptHistoryState({ + hasMore: false, + loading: false, + capacityReached: true, + }); + } for (const replayEvent of replayEvents) { settleActivePromptFromTurnEvent( activePromptsRef.current, @@ -1679,6 +1812,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { modelServiceId, sessionScope, maxQueued, + maxBlocks, store, restoreSessionId, restoreWorkspaceCwd, @@ -1897,6 +2031,132 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { }), [addNotice, clientId, resolvedBaseUrl, resolvedToken, store], ); + const loadMoreTranscript = useCallback(async () => { + const history = transcriptHistoryRef.current; + const activeSession = sessionRef.current; + if ( + !history.hasMore || + history.loading || + !activeSession || + activeSession.sessionId !== history.sessionId + ) { + return; + } + + history.loading = true; + setTranscriptHistoryState({ + hasMore: true, + loading: true, + capacityReached: false, + }); + try { + const page = await activeSession.client.getSessionTranscriptPage( + activeSession.sessionId, + { + ...(history.cursor !== undefined + ? { cursor: history.cursor } + : history.beforeRecordId !== undefined + ? { beforeRecordId: history.beforeRecordId } + : {}), + limit: historyPageSizeRef.current ?? 100, + clientId: activeSession.clientId, + }, + ); + if ( + sessionRef.current !== activeSession || + transcriptHistoryRef.current !== history + ) { + return; + } + if (page.partial || page.replayError) { + throw new Error( + page.replayError ?? 'Earlier session history was only partially read', + ); + } + + const replayOpts = { + ...eventOptionsRef.current, + suppressOwnUserEcho: false, + }; + const uiEvents: DaemonUiEvent[] = []; + for (const replayEvent of page.events) { + uiEvents.push( + ...filterDaemonUiEventsForTranscript( + replayEvent, + normalizeAndFilterEvent( + replayEvent, + activeSession.clientId, + replayOpts, + setConnection, + { updateConnection: false }, + ), + addNotice, + dismissNotice, + ), + ); + } + if ( + uiEvents.length > 0 && + !prependTranscriptHistory(store, uiEvents, maxBlocks) + ) { + history.hasMore = false; + history.loading = false; + history.capacityReached = true; + setTranscriptHistoryState({ + hasMore: false, + loading: false, + capacityReached: true, + }); + return; + } + const hasCapacity = store.getSnapshot().blocks.length < maxBlocks; + history.capacityReached = page.hasMore && !hasCapacity; + history.cursor = page.nextCursor; + history.beforeRecordId = undefined; + history.hasMore = page.hasMore && hasCapacity; + history.loading = false; + setTranscriptHistoryState({ + hasMore: history.hasMore, + loading: false, + capacityReached: history.capacityReached, + }); + } catch (error) { + if ( + sessionRef.current !== activeSession || + transcriptHistoryRef.current !== history + ) { + return; + } + history.hasMore = false; + history.loading = false; + history.capacityReached = false; + setTranscriptHistoryState({ + hasMore: false, + loading: false, + capacityReached: false, + }); + addNotice({ + severity: 'warning', + category: 'user_action', + operation: 'load_session', + code: 'daemon.transcript_history.failed', + message: 'Failed to load earlier session history', + debugMessage: error instanceof Error ? error.message : String(error), + recoverable: true, + }); + } + }, [addNotice, dismissNotice, maxBlocks, store]); + const transcriptHistoryValue = useMemo(() => { + const active = + connection.sessionId === transcriptHistoryRef.current.sessionId && + sessionRef.current?.sessionId === transcriptHistoryRef.current.sessionId; + return { + hasMore: active && transcriptHistoryState.hasMore, + loading: active && transcriptHistoryState.loading, + capacityReached: active && transcriptHistoryState.capacityReached, + loadMore: loadMoreTranscript, + }; + }, [connection.sessionId, loadMoreTranscript, transcriptHistoryState]); const lastHandledSessionIdRef = useRef< string | undefined | typeof UNHANDLED_SESSION >(UNHANDLED_SESSION); @@ -1933,7 +2193,11 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { value={workspaceEventSignals} > - {children} + + {children} + @@ -2209,6 +2473,16 @@ export function useDaemonTranscriptStore(): DaemonTranscriptStore { return store; } +export function useDaemonTranscriptHistory(): DaemonTranscriptHistory { + const history = useContext(DaemonTranscriptHistoryContext); + if (!history) { + throw new Error( + 'useDaemonTranscriptHistory must be used within DaemonSessionProvider', + ); + } + return history; +} + export function useDaemonTranscriptState(): DaemonTranscriptState { const store = useDaemonTranscriptStore(); return useSyncExternalStore( diff --git a/packages/webui/src/daemon/session/index.ts b/packages/webui/src/daemon/session/index.ts index 1721d62cb4f..44f4675e8cd 100644 --- a/packages/webui/src/daemon/session/index.ts +++ b/packages/webui/src/daemon/session/index.ts @@ -17,9 +17,11 @@ export { useDaemonStreamingState, useDaemonSession, useDaemonTranscriptBlocks, + useDaemonTranscriptHistory, useDaemonTranscriptState, useDaemonTranscriptStore, } from './DaemonSessionProvider.js'; +export type { DaemonTranscriptHistory } from './DaemonSessionProvider.js'; export type { DaemonCommandInfo, DaemonConnectionState, diff --git a/packages/webui/src/daemon/session/types.ts b/packages/webui/src/daemon/session/types.ts index ead22bce643..40d382ebe7e 100644 --- a/packages/webui/src/daemon/session/types.ts +++ b/packages/webui/src/daemon/session/types.ts @@ -113,6 +113,8 @@ export interface DaemonSessionProviderProps { maxQueued?: number; /** Maximum normalized transcript blocks retained in memory. */ maxBlocks?: number; + /** Latest persisted records requested during an existing-session load. */ + historyPageSize?: number; /** Hide this client's own user prompt echo when the daemon replays events. */ suppressOwnUserEcho?: boolean; /** Attach raw daemon events to normalized transcript blocks for debugging. */ From fe42aaf486978669cb135b437100f9ba8af9ea94 Mon Sep 17 00:00:00 2001 From: ytahdn Date: Fri, 17 Jul 2026 10:16:06 +0800 Subject: [PATCH 2/5] test(web-shell): align workspace visual assertion --- .../client/e2e/visuals/screenshots.spec.ts | 15 +++++---------- 1 file changed, 5 insertions(+), 10 deletions(-) diff --git a/packages/web-shell/client/e2e/visuals/screenshots.spec.ts b/packages/web-shell/client/e2e/visuals/screenshots.spec.ts index 29abe49c4b8..cc8a370a101 100644 --- a/packages/web-shell/client/e2e/visuals/screenshots.spec.ts +++ b/packages/web-shell/client/e2e/visuals/screenshots.spec.ts @@ -342,11 +342,8 @@ for (const theme of THEMES) { }); test(`workspace sidebar`, async ({ page }, testInfo) => { - // Two workspaces make the sidebar group sessions per workspace and tag the - // primary one — the surface the "primary workspace" label/badge lives on. - // Every other scenario here is single-workspace, where that tag never - // renders (it is gated on more than one displayed workspace), so this is - // the only scenario that can surface a change to the workspace labels. + // Two workspaces make the sidebar group sessions per workspace, so this + // scenario captures both the primary and secondary workspace headings. // // Pin the primary workspace cwd and its loaded session name explicitly, // rather than leaning on createWebShellDaemonScenario's defaults: the @@ -381,10 +378,9 @@ for (const theme of THEMES) { resolveBaseURL(testInfo), ); await gotoSession(page, scenario, daemon, theme); - // Each workspace renders a section headed by its basename; the primary one - // also carries a "Primary" tag. Assert both workspace names and the tag so - // a regression in the grouping or the (removable) primary label fails an - // assertion, not only the visually-reviewed screenshot. + // Each workspace renders a section headed by its basename. Assert both + // workspace names so a regression in the grouping fails an assertion, + // not only the visually-reviewed screenshot. const sidebar = page.getByRole('complementary'); await expect( sidebar.getByText('qwen-web-shell-e2e', { exact: true }), @@ -392,7 +388,6 @@ for (const theme of THEMES) { await expect( sidebar.getByText('qwen-api-service', { exact: true }), ).toBeVisible(); - await expect(sidebar.getByText('Primary', { exact: true })).toBeVisible(); // The primary workspace auto-expands and streams its session rows in via a // per-workspace fetch. Wait for the loaded session's row before capturing // so the async load has settled — otherwise the row list races the From 8c1c31140acda163fe87309c2ff3e05db06db089 Mon Sep 17 00:00:00 2001 From: ytahdn Date: Fri, 17 Jul 2026 11:43:04 +0800 Subject: [PATCH 3/5] fix(web-shell): harden history pagination --- packages/acp-bridge/src/bridge.test.ts | 4 +- .../acp-bridge/src/compactionEngine.test.ts | 2 +- packages/acp-bridge/src/compactionEngine.ts | 2 +- .../cli/src/acp-integration/acpAgent.test.ts | 50 ++++++++++++++++++- packages/cli/src/acp-integration/acpAgent.ts | 11 ++++ .../session/history-replay-page.ts | 2 +- packages/cli/src/serve/routes/session.ts | 4 +- packages/cli/src/serve/server.test.ts | 34 +++++++++++++ .../cli/src/serve/server/error-response.ts | 17 +++++++ packages/core/src/index.ts | 1 + .../src/services/session-transcript-reader.ts | 8 +-- packages/web-shell/client/App.test.tsx | 6 +++ .../client/components/ChatPane.test.tsx | 6 +++ .../components/MessageList.dom.test.tsx | 46 +++++++++++++++++ .../client/components/MessageList.tsx | 27 +++++++--- .../session/DaemonSessionProvider.test.tsx | 10 ++-- .../daemon/session/DaemonSessionProvider.tsx | 4 +- 17 files changed, 207 insertions(+), 27 deletions(-) diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 418deea7606..1702bf3be35 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -2923,12 +2923,12 @@ describe('createAcpSessionBridge', () => { { sessionUpdate: 'user_message_chunk', content: { type: 'text', text: `old-${'x'.repeat(600)}` }, - _meta: { recordId: 'record-1' }, + _meta: { 'qwen.session.recordId': 'record-1' }, }, { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: `new-${'y'.repeat(600)}` }, - _meta: { recordId: 'record-2' }, + _meta: { 'qwen.session.recordId': 'record-2' }, }, ], }, diff --git a/packages/acp-bridge/src/compactionEngine.test.ts b/packages/acp-bridge/src/compactionEngine.test.ts index 1ab634dd5bb..23674fb7429 100644 --- a/packages/acp-bridge/src/compactionEngine.test.ts +++ b/packages/acp-bridge/src/compactionEngine.test.ts @@ -941,7 +941,7 @@ describe('EventBus + CompactionEngine integration', () => { update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: `${text}-${'x'.repeat(600)}` }, - _meta: { recordId }, + _meta: { 'qwen.session.recordId': recordId }, }, }, }); diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index 7922f6c92b5..583e50e90ed 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -78,7 +78,7 @@ function replayRecordId(event: BridgeEvent): string | undefined { if (!meta || typeof meta !== 'object' || Array.isArray(meta)) { return undefined; } - const recordId = (meta as Record)['recordId']; + const recordId = (meta as Record)['qwen.session.recordId']; return typeof recordId === 'string' ? recordId : undefined; } diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index cf9e5208422..0ecf3d96067 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -219,6 +219,7 @@ vi.mock('@qwen-code/qwen-code-core', () => ({ maxImagesPerTurn: 4, }), SESSION_TRANSCRIPT_MAX_LIMIT: 500, + SESSION_TRANSCRIPT_MAX_PAGE_BYTES: 4 * 1024 * 1024, InvalidSessionTranscriptCursorError: class InvalidSessionTranscriptCursorError extends Error {}, SessionTranscriptSnapshotUnavailableError: class SessionTranscriptSnapshotUnavailableError extends Error {}, SessionTranscriptTooLargeError: class SessionTranscriptTooLargeError extends Error { @@ -230,6 +231,15 @@ vi.mock('@qwen-code/qwen-code-core', () => ({ super('Transcript snapshot is too large'); } }, + SessionTranscriptPageTooLargeError: class SessionTranscriptPageTooLargeError extends Error { + constructor( + readonly sessionId: string, + readonly pageBytes: number, + readonly maxBytes: number, + ) { + super('Transcript page is too large'); + } + }, encodeSessionTranscriptCursor: vi.fn((state: unknown) => Buffer.from(JSON.stringify(state), 'utf8').toString('base64url'), ), @@ -735,6 +745,7 @@ import { InvalidSessionTranscriptCursorError, SessionTranscriptSnapshotUnavailableError, SessionTranscriptTooLargeError, + SessionTranscriptPageTooLargeError, encodeSessionTranscriptCursor, unregisterGoalHook, startEventLoopLagMonitor, @@ -7290,10 +7301,12 @@ describe('QwenAgent MCP SSE/HTTP support', () => { expect(readPage).toHaveBeenCalledWith(VALID_SESSION_ID, { cursor: 'cursor-1', limit: 2, + maxBytes: 4 * 1024 * 1024, }); expect(readPage).toHaveBeenCalledWith(VALID_SESSION_ID, { cursor: 'cursor-2', limit: 2, + maxBytes: 4 * 1024 * 1024, }); expect(result.events).toEqual([ { @@ -7611,7 +7624,42 @@ describe('QwenAgent MCP SSE/HTTP support', () => { maxBytes: 200, }, }); - expect(readPage).toHaveBeenCalledWith(VALID_SESSION_ID, {}); + expect(readPage).toHaveBeenCalledWith(VALID_SESSION_ID, { + maxBytes: 4 * 1024 * 1024, + }); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('qwen/status/session/transcript maps oversized pages to structured errors', async () => { + const settings = makeCoreSettings(); + const readPage = vi + .fn() + .mockRejectedValue( + new SessionTranscriptPageTooLargeError(VALID_SESSION_ID, 300, 200), + ); + vi.mocked(SessionTranscriptReader).mockImplementation( + () => + ({ + readPage, + }) as unknown as InstanceType, + ); + const { agent, agentPromise } = await bootCoreSettingsAgent(settings); + + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.sessionTranscript, { + sessionId: VALID_SESSION_ID, + }), + ).rejects.toMatchObject({ + code: -32012, + data: { + errorKind: 'transcript_page_too_large', + sessionId: VALID_SESSION_ID, + pageBytes: 300, + maxBytes: 200, + }, + }); mockConnectionState.resolve(); await agentPromise; diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index 3fb5a744ad9..af0dc7dcd04 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -58,7 +58,9 @@ import { MCPOAuthTokenStorage, InvalidSessionTranscriptCursorError, SESSION_TRANSCRIPT_MAX_LIMIT, + SESSION_TRANSCRIPT_MAX_PAGE_BYTES, SessionTranscriptReader, + SessionTranscriptPageTooLargeError, SessionTranscriptSnapshotUnavailableError, SessionTranscriptTooLargeError, encodeSessionTranscriptCursor, @@ -6308,6 +6310,7 @@ class QwenAgent implements Agent { ? { beforeRecordId: rawBeforeRecordId } : {}), ...(typeof rawLimit === 'number' ? { limit: rawLimit } : {}), + maxBytes: SESSION_TRANSCRIPT_MAX_PAGE_BYTES, }); const config = await this.getTranscriptReplayConfig(cwd, settings); const replay = await replayTranscriptRecordPage({ @@ -6367,6 +6370,14 @@ class QwenAgent implements Agent { maxBytes: error.maxBytes, }); } + if (error instanceof SessionTranscriptPageTooLargeError) { + throw new RequestError(-32012, error.message, { + errorKind: 'transcript_page_too_large', + sessionId, + pageBytes: error.pageBytes, + maxBytes: error.maxBytes, + }); + } if ((error as NodeJS.ErrnoException).code === 'ENOENT') { if (typeof rawCursor === 'string') { throw new RequestError( diff --git a/packages/cli/src/acp-integration/session/history-replay-page.ts b/packages/cli/src/acp-integration/session/history-replay-page.ts index b5e9d96bff6..c681b09058f 100644 --- a/packages/cli/src/acp-integration/session/history-replay-page.ts +++ b/packages/cli/src/acp-integration/session/history-replay-page.ts @@ -121,7 +121,7 @@ function replayContext( const meta = isObjectRecord(record['_meta']) ? record['_meta'] : {}; updates.push({ ...record, - _meta: { ...meta, recordId: activeRecordId }, + _meta: { ...meta, 'qwen.session.recordId': activeRecordId }, } as unknown as SessionUpdate); }, setActiveRecordId: (recordId: string | null) => { diff --git a/packages/cli/src/serve/routes/session.ts b/packages/cli/src/serve/routes/session.ts index 52fa3f2906f..617e4c555db 100644 --- a/packages/cli/src/serve/routes/session.ts +++ b/packages/cli/src/serve/routes/session.ts @@ -13,6 +13,7 @@ import { SessionService, SessionOrganizationError, SESSION_TRANSCRIPT_MAX_LIMIT, + SESSION_TRANSCRIPT_MAX_PAGE_BYTES, SessionTranscriptPageTooLargeError, SessionTranscriptCursorCodec, SessionTranscriptReader, @@ -99,7 +100,6 @@ interface RegisterSessionRoutesDeps { languageCodes: string[]; } -const WORKSPACE_TRANSCRIPT_PAGE_SOURCE_MAX_BYTES = 4 * 1024 * 1024; const WORKSPACE_TRANSCRIPT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024; const WORKSPACE_TRANSCRIPT_CURSOR_MAX_BYTES = 64 * 1024; const TRANSCRIPT_CURSOR_TOO_LARGE_REPLAY_ERROR = @@ -1661,7 +1661,7 @@ export function registerSessionRoutes( ...(limit !== undefined ? { limit } : {}), ...(cursor !== undefined ? { cursor } : {}), ...(beforeRecordId !== undefined ? { beforeRecordId } : {}), - maxBytes: WORKSPACE_TRANSCRIPT_PAGE_SOURCE_MAX_BYTES, + maxBytes: SESSION_TRANSCRIPT_MAX_PAGE_BYTES, }); } catch (error) { if ((error as NodeJS.ErrnoException).code !== 'ENOENT') { diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 219d35a5953..869d52a2068 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -15284,6 +15284,40 @@ describe('createServeApp', () => { }); expect(bridge.sessionTranscriptCalls).toEqual([{ sessionId: sid }]); }); + + it('maps oversized transcript pages to 413', async () => { + const sid = '55555555-bbbb-cccc-dddd-aaaaaaaaaaaa'; + const bridge = fakeBridge({ + sessionTranscriptImpl: async () => { + throw Object.assign(new Error('Transcript page is too large'), { + data: { + errorKind: 'transcript_page_too_large', + sessionId: sid, + pageBytes: 300, + maxBytes: 200, + }, + }); + }, + }); + await writeTranscriptSession(sid); + const app = createServeApp({ ...baseOpts, workspace: wsDir }, undefined, { + bridge, + boundWorkspace: wsDir, + }); + + const res = await request(app) + .get(`/session/${sid}/transcript`) + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(413); + expect(res.body).toMatchObject({ + code: 'transcript_page_too_large', + sessionId: sid, + pageBytes: 300, + maxBytes: 200, + }); + expect(bridge.sessionTranscriptCalls).toEqual([{ sessionId: sid }]); + }); }); describe('POST /sessions/delete', () => { diff --git a/packages/cli/src/serve/server/error-response.ts b/packages/cli/src/serve/server/error-response.ts index 01205a87193..b55fddc39d7 100644 --- a/packages/cli/src/serve/server/error-response.ts +++ b/packages/cli/src/serve/server/error-response.ts @@ -652,6 +652,23 @@ export function sendBridgeError( }); return; } + if (kind === 'transcript_page_too_large') { + const d = data as { + sessionId?: string; + pageBytes?: number; + maxBytes?: number; + }; + res.status(413).json({ + error: errorMessage(err), + code: 'transcript_page_too_large', + ...(d.sessionId ? { sessionId: d.sessionId } : {}), + ...(typeof d.pageBytes === 'number' + ? { pageBytes: d.pageBytes } + : {}), + ...(typeof d.maxBytes === 'number' ? { maxBytes: d.maxBytes } : {}), + }); + return; + } } } // 5xx is the kind of error operators need to see in their daemon log diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 61bba145063..1ed430b287a 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -242,6 +242,7 @@ export { SESSION_TRANSCRIPT_DEFAULT_LIMIT, SESSION_TRANSCRIPT_MAX_INDEX_BYTES, SESSION_TRANSCRIPT_MAX_LIMIT, + SESSION_TRANSCRIPT_MAX_PAGE_BYTES, SessionTranscriptCursorCodec, SessionTranscriptReader, SessionTranscriptPageTooLargeError, diff --git a/packages/core/src/services/session-transcript-reader.ts b/packages/core/src/services/session-transcript-reader.ts index ca6bdd3776e..09d63d91b7e 100644 --- a/packages/core/src/services/session-transcript-reader.ts +++ b/packages/core/src/services/session-transcript-reader.ts @@ -19,6 +19,7 @@ export const SESSION_TRANSCRIPT_DEFAULT_LIMIT = 100; export const SESSION_TRANSCRIPT_MAX_LIMIT = 500; export const SESSION_TRANSCRIPT_CURSOR_VERSION = 1 as const; export const SESSION_TRANSCRIPT_MAX_INDEX_BYTES = 256 * 1024 * 1024; +export const SESSION_TRANSCRIPT_MAX_PAGE_BYTES = 4 * 1024 * 1024; export class InvalidSessionTranscriptCursorError extends Error { constructor(message = 'Invalid transcript cursor') { @@ -1091,13 +1092,6 @@ export class SessionTranscriptReader { throw new InvalidSessionTranscriptCursorError(); } } - if ( - !cursor && - direction === 'backward' && - options.beforeRecordId === undefined - ) { - position = index.activeUuids.length; - } if (position > index.activeUuids.length) { debugLogger.debug( `cursor position out of range session=${sessionId} ` + diff --git a/packages/web-shell/client/App.test.tsx b/packages/web-shell/client/App.test.tsx index f35ff18c98f..9fa77e527d2 100644 --- a/packages/web-shell/client/App.test.tsx +++ b/packages/web-shell/client/App.test.tsx @@ -189,6 +189,12 @@ vi.mock('@qwen-code/webui/daemon-react-sdk', () => ({ }), useStreamingState: () => testState.streamingState, useTranscriptBlocks: () => testState.blocks, + useTranscriptHistory: () => ({ + hasMore: false, + loading: false, + capacityReached: false, + loadMore: vi.fn(), + }), useTranscriptStore: () => mockStore, useWorkspace: () => mockWorkspace, useWorkspaceActions: () => mockWorkspaceActions, diff --git a/packages/web-shell/client/components/ChatPane.test.tsx b/packages/web-shell/client/components/ChatPane.test.tsx index 38162456892..9d0f42cf8b4 100644 --- a/packages/web-shell/client/components/ChatPane.test.tsx +++ b/packages/web-shell/client/components/ChatPane.test.tsx @@ -68,6 +68,12 @@ vi.mock('@qwen-code/webui/daemon-react-sdk', () => ({ }, useStreamingState: () => streamingStateValue, useTranscriptBlocks: () => [], + useTranscriptHistory: () => ({ + hasMore: false, + loading: false, + capacityReached: false, + loadMore: vi.fn(), + }), useTranscriptStore: () => ({ dispatch: transcriptDispatch, }), diff --git a/packages/web-shell/client/components/MessageList.dom.test.tsx b/packages/web-shell/client/components/MessageList.dom.test.tsx index aba548aa413..10f701a2eb4 100644 --- a/packages/web-shell/client/components/MessageList.dom.test.tsx +++ b/packages/web-shell/client/components/MessageList.dom.test.tsx @@ -1172,6 +1172,52 @@ describe('MessageList — turn collapse (DOM)', () => { expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); }); + it('preserves the scroll anchor after prepending earlier history', async () => { + let scrollHeight = 1200; + let scrollTop = 40; + Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { + configurable: true, + get: () => scrollHeight, + }); + Object.defineProperty(HTMLElement.prototype, 'clientHeight', { + configurable: true, + value: 600, + }); + Object.defineProperty(HTMLElement.prototype, 'scrollTop', { + configurable: true, + get: () => scrollTop, + set: (value: number) => { + scrollTop = value; + }, + }); + const onLoadOlderHistory = vi.fn(async () => { + scrollHeight = 1800; + }); + const c = mount([userMsg('u1')], undefined, { + hasOlderHistory: true, + onLoadOlderHistory, + }); + const list = c.querySelector( + '[data-web-shell-message-list]', + ) as HTMLElement; + Object.defineProperty(list, 'scrollTop', { + configurable: true, + get: () => scrollTop, + set: (value: number) => { + scrollTop = value; + }, + }); + scrollTop = 40; + + await act(async () => { + list.dispatchEvent(new Event('scroll')); + await Promise.resolve(); + }); + + expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); + expect(scrollTop).toBe(640); + }); + it('loads earlier history when the transcript does not overflow', async () => { Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { configurable: true, diff --git a/packages/web-shell/client/components/MessageList.tsx b/packages/web-shell/client/components/MessageList.tsx index 7326dedadba..741e4d3ee8a 100644 --- a/packages/web-shell/client/components/MessageList.tsx +++ b/packages/web-shell/client/components/MessageList.tsx @@ -2344,6 +2344,22 @@ export const MessageList = memo( const pendingOverflowFrame = useRef(undefined); catchingUpRef.current = catchingUp; const containerRef = useRef(null); + const [olderHistoryAnchor, setOlderHistoryAnchor] = useState<{ + scrollHeight: number; + scrollTop: number; + } | null>(null); + + useLayoutEffect(() => { + if (!olderHistoryAnchor) return; + const current = containerRef.current; + if (current) { + current.scrollTop = + olderHistoryAnchor.scrollTop + + Math.max(0, current.scrollHeight - olderHistoryAnchor.scrollHeight); + } + olderHistoryLoadInFlight.current = false; + setOlderHistoryAnchor(null); + }, [olderHistoryAnchor]); const reportCanScrollToBottom = useCallback(() => { const el = containerRef.current; @@ -2934,14 +2950,13 @@ export const MessageList = memo( followPausedByUserRef.current = true; try { await onLoadOlderHistory(); - window.requestAnimationFrame(() => { - const current = containerRef.current; - if (!current) return; - current.scrollTop = - previousTop + Math.max(0, current.scrollHeight - previousHeight); + setOlderHistoryAnchor({ + scrollHeight: previousHeight, + scrollTop: previousTop, }); - } finally { + } catch (error) { olderHistoryLoadInFlight.current = false; + throw error; } }, [loadingOlderHistory, onLoadOlderHistory]); diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index 6a6da0a62c8..e87c9feff83 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -8496,7 +8496,7 @@ describe('DaemonSessionProvider', () => { update: { sessionUpdate: 'user_message_chunk', content: { type: 'text', text: 'complete history' }, - _meta: { recordId: 'record-1' }, + _meta: { 'qwen.session.recordId': 'record-1' }, }, }, }, @@ -8545,7 +8545,7 @@ describe('DaemonSessionProvider', () => { update: { sessionUpdate: 'user_message_chunk', content: { type: 'text', text }, - ...(recordId ? { _meta: { recordId } } : {}), + ...(recordId ? { _meta: { 'qwen.session.recordId': recordId } } : {}), }, }, }); @@ -8620,7 +8620,7 @@ describe('DaemonSessionProvider', () => { update: { sessionUpdate: 'user_message_chunk', content: { type: 'text', text }, - ...(recordId ? { _meta: { recordId } } : {}), + ...(recordId ? { _meta: { 'qwen.session.recordId': recordId } } : {}), }, }, }); @@ -8690,7 +8690,7 @@ describe('DaemonSessionProvider', () => { update: { sessionUpdate: kind, content: { type: 'text', text }, - _meta: { recordId }, + _meta: { 'qwen.session.recordId': recordId }, }, }, }); @@ -8750,7 +8750,7 @@ describe('DaemonSessionProvider', () => { update: { sessionUpdate: 'user_message_chunk', content: { type: 'text', text }, - ...(recordId ? { _meta: { recordId } } : {}), + ...(recordId ? { _meta: { 'qwen.session.recordId': recordId } } : {}), }, }, }); diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index f5d0008f1c3..ca53b75b00f 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -145,7 +145,9 @@ function getPersistedReplayRecordId(event: DaemonEvent): string | undefined { if (!isRecord(event.data)) return undefined; const update = event.data['update']; const meta = isRecord(update) ? update['_meta'] : event.data['_meta']; - return isRecord(meta) ? getString(meta, 'recordId') : undefined; + return isRecord(meta) + ? getString(meta, 'qwen.session.recordId') + : undefined; } catch { return undefined; } From bd6540c01eff245c8e232f13b7e61d123b4d7b8f Mon Sep 17 00:00:00 2001 From: ytahdn Date: Fri, 17 Jul 2026 13:35:35 +0800 Subject: [PATCH 4/5] fix(web-shell): keep history paging retryable --- .../cli/qwen-serve-routes.test.ts | 1 + .../session/history-replay-page.test.ts | 33 +++++++ .../components/MessageList.dom.test.tsx | 40 ++++++++ .../client/components/MessageList.tsx | 95 +++++++++++++------ .../session/DaemonSessionProvider.test.tsx | 71 +++++++++++++- .../daemon/session/DaemonSessionProvider.tsx | 15 ++- 6 files changed, 220 insertions(+), 35 deletions(-) diff --git a/integration-tests/cli/qwen-serve-routes.test.ts b/integration-tests/cli/qwen-serve-routes.test.ts index 3c5056dbd31..8dca9862458 100644 --- a/integration-tests/cli/qwen-serve-routes.test.ts +++ b/integration-tests/cli/qwen-serve-routes.test.ts @@ -344,6 +344,7 @@ describe('qwen serve — capabilities envelope', () => { 'session_organization', 'session_export', 'session_transcript', + 'session_transcript_pagination', 'mcp_guardrails', 'workspace_mcp_manage', 'mcp_guardrail_events', diff --git a/packages/cli/src/acp-integration/session/history-replay-page.test.ts b/packages/cli/src/acp-integration/session/history-replay-page.test.ts index 90fff918988..371885108a9 100644 --- a/packages/cli/src/acp-integration/session/history-replay-page.test.ts +++ b/packages/cli/src/acp-integration/session/history-replay-page.test.ts @@ -144,6 +144,39 @@ describe('history replay page', () => { ); }); + it('replays backward pages without forward replay state', async () => { + const replayPage = vi + .spyOn(HistoryReplayer.prototype, 'replayPage') + .mockResolvedValueOnce({ pendingToolCalls: [] }); + const encodeCursor = vi.fn(() => 'next-cursor'); + + await replayTranscriptRecordPage({ + sessionId: SESSION_ID, + page: recordPage({ + direction: 'backward', + hasMore: true, + nextCursorState: cursorState(), + replay: { + pendingToolCalls: [ + { + callId: 'stale-call', + toolName: 'Read', + recordId: 'stale-record', + }, + ], + }, + }), + encodeCursor, + }); + + expect(replayPage).toHaveBeenCalledWith([], { + pendingToolCalls: [], + finalizeDangling: true, + gaps: [], + }); + expect(encodeCursor).toHaveBeenCalledWith(cursorState()); + }); + it('terminates pagination when replay conversion fails', async () => { vi.spyOn(HistoryReplayer.prototype, 'replayPage').mockRejectedValueOnce( new Error('replay failed'), diff --git a/packages/web-shell/client/components/MessageList.dom.test.tsx b/packages/web-shell/client/components/MessageList.dom.test.tsx index 10f701a2eb4..fd06d99519c 100644 --- a/packages/web-shell/client/components/MessageList.dom.test.tsx +++ b/packages/web-shell/client/components/MessageList.dom.test.tsx @@ -1240,6 +1240,46 @@ describe('MessageList — turn collapse (DOM)', () => { expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); }); + it('waits for another upward scroll intent before retrying a failed underfill load', async () => { + Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { + configurable: true, + value: 300, + }); + Object.defineProperty(HTMLElement.prototype, 'clientHeight', { + configurable: true, + value: 600, + }); + const onLoadOlderHistory = vi + .fn() + .mockRejectedValueOnce(new Error('temporary failure')) + .mockResolvedValueOnce(undefined); + + const c = mount([userMsg('u1')], undefined, { + hasOlderHistory: true, + onLoadOlderHistory, + }); + const list = c.querySelector( + '[data-web-shell-message-list]', + ) as HTMLElement; + Object.defineProperty(list, 'scrollTop', { + configurable: true, + writable: true, + value: 0, + }); + await act(async () => { + await Promise.resolve(); + await Promise.resolve(); + }); + expect(onLoadOlderHistory).toHaveBeenCalledTimes(1); + + await act(async () => { + list.dispatchEvent(new WheelEvent('wheel', { deltaY: -1 })); + await Promise.resolve(); + }); + + expect(onLoadOlderHistory).toHaveBeenCalledTimes(2); + }); + it('loads earlier history when a resize removes the overflow', async () => { let clientHeight = 600; Object.defineProperty(HTMLElement.prototype, 'scrollHeight', { diff --git a/packages/web-shell/client/components/MessageList.tsx b/packages/web-shell/client/components/MessageList.tsx index 741e4d3ee8a..96afec5cd2d 100644 --- a/packages/web-shell/client/components/MessageList.tsx +++ b/packages/web-shell/client/components/MessageList.tsx @@ -2344,6 +2344,7 @@ export const MessageList = memo( const pendingOverflowFrame = useRef(undefined); catchingUpRef.current = catchingUp; const containerRef = useRef(null); + const olderHistoryRetryBlocked = useRef(false); const [olderHistoryAnchor, setOlderHistoryAnchor] = useState<{ scrollHeight: number; scrollTop: number; @@ -2361,6 +2362,10 @@ export const MessageList = memo( setOlderHistoryAnchor(null); }, [olderHistoryAnchor]); + useEffect(() => { + if (!hasOlderHistory) olderHistoryRetryBlocked.current = false; + }, [hasOlderHistory]); + const reportCanScrollToBottom = useCallback(() => { const el = containerRef.current; const distanceFromBottom = el @@ -2934,31 +2939,36 @@ export const MessageList = memo( performScrollToRow(idx + headerOffset, pending); }, [visibleItems, headerOffset, performScrollToRow]); - const loadOlderHistory = useCallback(async () => { - const el = containerRef.current; - if ( - !el || - !onLoadOlderHistory || - loadingOlderHistory || - olderHistoryLoadInFlight.current - ) { - return; - } - olderHistoryLoadInFlight.current = true; - const previousHeight = el.scrollHeight; - const previousTop = el.scrollTop; - followPausedByUserRef.current = true; - try { - await onLoadOlderHistory(); - setOlderHistoryAnchor({ - scrollHeight: previousHeight, - scrollTop: previousTop, - }); - } catch (error) { - olderHistoryLoadInFlight.current = false; - throw error; - } - }, [loadingOlderHistory, onLoadOlderHistory]); + const loadOlderHistory = useCallback( + async (allowRetry = false) => { + const el = containerRef.current; + if ( + !el || + !onLoadOlderHistory || + loadingOlderHistory || + olderHistoryLoadInFlight.current || + (olderHistoryRetryBlocked.current && !allowRetry) + ) { + return; + } + olderHistoryRetryBlocked.current = false; + olderHistoryLoadInFlight.current = true; + const previousHeight = el.scrollHeight; + const previousTop = el.scrollTop; + followPausedByUserRef.current = true; + try { + await onLoadOlderHistory(); + setOlderHistoryAnchor({ + scrollHeight: previousHeight, + scrollTop: previousTop, + }); + } catch { + olderHistoryRetryBlocked.current = true; + olderHistoryLoadInFlight.current = false; + } + }, + [loadingOlderHistory, onLoadOlderHistory], + ); // Rules 2 & 3: detect scroll direction to toggle follow mode. // Runs synchronously in the scroll handler — no rAF needed since @@ -2968,7 +2978,7 @@ export const MessageList = memo( if (!el) return; const curr = el.scrollTop; if (hasOlderHistory && curr <= LOAD_OLDER_HISTORY_THRESHOLD_PX) { - void loadOlderHistory(); + void loadOlderHistory(true); } if (scrollCooldown.current) { lastScrollTop.current = curr; @@ -3048,6 +3058,22 @@ export const MessageList = memo( useEffect(() => { const el = getScrollElement(); if (!el) return; + const retryOlderHistoryAtTop = () => { + if ( + olderHistoryRetryBlocked.current && + el.scrollTop <= LOAD_OLDER_HISTORY_THRESHOLD_PX + ) { + void loadOlderHistory(true); + } + }; + const markFromWheel = (event: WheelEvent) => { + markUserScrollIntent(); + if (event.deltaY < 0) retryOlderHistoryAtTop(); + }; + const markFromTouch = () => { + markUserScrollIntent(); + retryOlderHistoryAtTop(); + }; const markFromPointer = (event: PointerEvent) => { const rect = el.getBoundingClientRect(); const scrollbarEdge = 20; @@ -3069,21 +3095,28 @@ export const MessageList = memo( event.key === ' ' ) { markUserScrollIntent(); + if ( + event.key === 'ArrowUp' || + event.key === 'PageUp' || + event.key === 'Home' + ) { + retryOlderHistoryAtTop(); + } } }; - el.addEventListener('wheel', markUserScrollIntent, { passive: true }); - el.addEventListener('touchstart', markUserScrollIntent, { + el.addEventListener('wheel', markFromWheel, { passive: true }); + el.addEventListener('touchstart', markFromTouch, { passive: true, }); el.addEventListener('pointerdown', markFromPointer, { passive: true }); el.addEventListener('keydown', markFromKey, { passive: true }); return () => { - el.removeEventListener('wheel', markUserScrollIntent); - el.removeEventListener('touchstart', markUserScrollIntent); + el.removeEventListener('wheel', markFromWheel); + el.removeEventListener('touchstart', markFromTouch); el.removeEventListener('pointerdown', markFromPointer); el.removeEventListener('keydown', markFromKey); }; - }, [getScrollElement, markUserScrollIntent]); + }, [getScrollElement, loadOlderHistory, markUserScrollIntent]); useEffect(() => { const el = getScrollElement(); diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index e87c9feff83..a40d9936f05 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -8603,6 +8603,73 @@ describe('DaemonSessionProvider', () => { expect(history?.hasMore).toBe(false); }); + it('keeps transient transcript page failures retryable', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const replayEvent = (id: number, text: string): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + _meta: { 'qwen.session.recordId': `record-${id}` }, + }, + }, + }); + const session = createMockSession({ + sessionId: 'session-retry-history-page', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [replayEvent(2, 'recent prompt')], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + sdkMocks.getSessionTranscriptPage + .mockRejectedValueOnce(new Error('temporary network failure')) + .mockResolvedValueOnce({ + v: 1, + sessionId: session.sessionId, + events: [replayEvent(1, 'older prompt')], + hasMore: false, + }); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + }); + await act(async () => { + await expect(history?.loadMore()).rejects.toThrow( + 'temporary network failure', + ); + await flushPromises(); + }); + expect(history?.hasMore).toBe(true); + + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + + expect(sdkMocks.getSessionTranscriptPage).toHaveBeenCalledTimes(2); + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toEqual(['older prompt', 'recent prompt']); + expect(history?.hasMore).toBe(false); + }); + it('reports a partial older page without changing the transcript', async () => { sdkMocks.capabilities.mockResolvedValue({ workspaceCwd: '/mock-workspace', @@ -8657,7 +8724,9 @@ describe('DaemonSessionProvider', () => { historyPageSize: 25, }); await act(async () => { - await history?.loadMore(); + await expect(history?.loadMore()).rejects.toThrow( + 'Replay conversion failed for this page', + ); await flushPromises(); }); diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index ca53b75b00f..ba91680a82d 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -2051,6 +2051,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { loading: true, capacityReached: false, }); + let terminalFailure = false; try { const page = await activeSession.client.getSessionTranscriptPage( activeSession.sessionId, @@ -2071,6 +2072,7 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { return; } if (page.partial || page.replayError) { + terminalFailure = true; throw new Error( page.replayError ?? 'Earlier session history was only partially read', ); @@ -2129,11 +2131,17 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { ) { return; } - history.hasMore = false; + const retryable = + !terminalFailure && + (!(error instanceof DaemonHttpError) || + error.status >= 500 || + error.status === 408 || + error.status === 429); + history.hasMore = retryable; history.loading = false; history.capacityReached = false; setTranscriptHistoryState({ - hasMore: false, + hasMore: retryable, loading: false, capacityReached: false, }); @@ -2144,8 +2152,9 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { code: 'daemon.transcript_history.failed', message: 'Failed to load earlier session history', debugMessage: error instanceof Error ? error.message : String(error), - recoverable: true, + recoverable: retryable, }); + throw error; } }, [addNotice, dismissNotice, maxBlocks, store]); const transcriptHistoryValue = useMemo(() => { From 73ddfdebc11eb8d1ee22bb09b3bfd43e233599a8 Mon Sep 17 00:00:00 2001 From: ytahdn Date: Sat, 18 Jul 2026 02:58:47 +0800 Subject: [PATCH 5/5] fix(webui): skip malformed transcript page events --- .../session/DaemonSessionProvider.test.tsx | 99 +++++++++++++++++++ .../daemon/session/DaemonSessionProvider.tsx | 42 +++++--- 2 files changed, 129 insertions(+), 12 deletions(-) diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index 0a0c12f0f40..79b22c1475b 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -8824,6 +8824,105 @@ describe('DaemonSessionProvider', () => { expect(history?.hasMore).toBe(false); }); + it('skips malformed older-page events and advances the cursor', async () => { + sdkMocks.capabilities.mockResolvedValue({ + workspaceCwd: '/mock-workspace', + features: ['session_transcript_pagination'], + }); + const replayEvent = (id: number, text: string): DaemonEvent => ({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + _meta: { 'qwen.session.recordId': `record-${id}` }, + }, + }, + }); + const malformedEvent = { + id: 1, + v: 1, + type: 'session_update', + } as DaemonEvent; + Object.defineProperty(malformedEvent, 'data', { + get() { + throw new Error('malformed history event'); + }, + }); + const session = createMockSession({ + sessionId: 'session-malformed-history-page', + historyHasMore: true, + replaySnapshot: { + compactedReplay: [replayEvent(3, 'recent prompt')], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + sdkMocks.getSessionTranscriptPage + .mockResolvedValueOnce({ + v: 1, + sessionId: session.sessionId, + events: [malformedEvent, replayEvent(2, 'older prompt')], + nextCursor: 'next-page', + hasMore: true, + }) + .mockResolvedValueOnce({ + v: 1, + sessionId: session.sessionId, + events: [replayEvent(1, 'oldest prompt')], + hasMore: false, + }); + let history: ReturnType | undefined; + let blocks: readonly DaemonTranscriptBlock[] = []; + let notices: readonly DaemonSessionNotice[] = []; + + function Harness() { + history = useDaemonTranscriptHistory(); + blocks = useDaemonTranscriptBlocks(); + notices = useDaemonSessionNotices().notices; + return null; + } + + await renderWithProvider(, { + autoConnect: true, + historyPageSize: 25, + }); + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + + expect(history?.hasMore).toBe(true); + await act(async () => { + await history?.loadMore(); + await flushPromises(); + }); + + expect(sdkMocks.getSessionTranscriptPage).toHaveBeenCalledTimes(2); + expect(sdkMocks.getSessionTranscriptPage).toHaveBeenNthCalledWith( + 2, + session.sessionId, + { + cursor: 'next-page', + limit: 25, + clientId: session.clientId, + }, + ); + expect( + blocks.map((block) => ('text' in block ? block.text : undefined)), + ).toEqual(['oldest prompt', 'older prompt', 'recent prompt']); + expect(history?.hasMore).toBe(false); + expect(notices).toContainEqual( + expect.objectContaining({ + code: 'daemon.replay_event_malformed', + message: 'Skipped malformed history event', + debugMessage: 'malformed history event', + }), + ); + }); + it('reports a partial older page without changing the transcript', async () => { sdkMocks.capabilities.mockResolvedValue({ workspaceCwd: '/mock-workspace', diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index 839b5c1026e..40911f4faa7 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -2157,20 +2157,38 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { }; const uiEvents: DaemonUiEvent[] = []; for (const replayEvent of page.events) { - uiEvents.push( - ...filterDaemonUiEventsForTranscript( - replayEvent, - normalizeAndFilterEvent( + try { + uiEvents.push( + ...filterDaemonUiEventsForTranscript( replayEvent, - activeSession.clientId, - replayOpts, - setConnection, - { updateConnection: false }, + normalizeAndFilterEvent( + replayEvent, + activeSession.clientId, + replayOpts, + setConnection, + { updateConnection: false }, + ), + addNotice, + dismissNotice, ), - addNotice, - dismissNotice, - ), - ); + ); + } catch (error) { + const message = + error instanceof Error ? error.message : String(error); + addNotice({ + severity: 'warning', + category: 'protocol', + operation: 'normalize_event', + code: 'daemon.replay_event_malformed', + message: 'Skipped malformed history event', + debugMessage: message, + recoverable: true, + }); + console.warn( + '[DaemonSessionProvider] skipped malformed history event:', + error, + ); + } } if ( uiEvents.length > 0 &&