vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-08-16 04:05:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 24

fix(core): bound backward transcript pages in long single-turn sessions (#8553)

Browse source 
* fix(core): bound backward transcript pages in long single-turn sessions

* fix(core): keep byte budget when backward turn alignment fails

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(core): keep tool call/result pairs intact on backward pages

* refactor(core): share replay turn-start predicate with ACP history selection

* fix(core): bound backward pair extension for long tool_result runs

* fix(core): bound backward pair extension by the page byte budget

* fix(core): cap backward page expansion at a shared byte ceiling

Turn-alignment expansion was record-bounded only, so a byte-heavy turn
whose start sat within the expansion window could absorb records past
the workspace route's response cap and dead-end backward pagination at
that anchor on every retry. Gate both turn-alignment and pair extension
at a shared hard page ceiling (SESSION_TRANSCRIPT_MAX_EXPANDED_PAGE_BYTES);
the route derives its serialized-response cap as twice that value so the
caps cannot drift.

Also accept the expansion floor only when it is a real turn boundary, so
pages inside a long turn stay `limit` records instead of `2 * limit`,
and exclude interleaved realtime conversation records from the
pair-boundary predicate so the pair walk passes through them to the
owning call. Extension skips are logged with a reason, and backward
chaining under maxBytes is covered by tests.

* fix(core): bound backward pagination expansion and mid-turn turn starts

* fix(core): pair tool calls with results across backward pagination and bulk replay

Exempt the force-joined owner from the pair-extension byte budget so an
oversized tool-call record no longer splits its pair by construction, and
mirror the same pair-extension guard onto the ACP bulk-replay selector.
Deduplicate the bounded boundary walk into a shared findBoundaryAtOrBefore,
harden the ceiling-clamp byte-budget tests, pin mid-turn notification
classification, and refresh the stale bulk-load contract sentence.

---------

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
Co-authored-by: qwen-code-dev-bot <qwen-code-dev@service.alibaba.com>
Co-authored-by: Qwen Autofix <autofix@qwen-code.dev>
This commit is contained in:
Shaojin Wen 2026-08-07 12:21:45 +08:00 • committed by GitHub
parent 7edc16ba11
commit 650e085fec
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 1386 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

36
docs/design/web-shell/web-shell-history-pagination.md
View file

@ -63,7 +63,11 @@ 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.
boundary when one is reachable within one extra window of expansion;
otherwise it keeps the requested window and may start mid-turn (extending to
the owning assistant record when the window would otherwise start mid tool
call/result pair), with older records paged backward. 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
@ -79,12 +83,30 @@ 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.
starts at a normal user-turn boundary when one is reachable within one extra
window (`limit` records) of expansion and the expanded page still fits one
extra page byte budget (a bounded multiple of the soft page budget, clamped
to the hard page ceiling): a small turn rides over the soft page budget
whole, while a byte-heavy turn that would exceed that budget keeps the
bounded selection so the route can always serialize the page. Inside a
single long turn no boundary is reachable within that budget, so pages stay
the requested window (`limit` records) near the anchor, may start mid-turn,
and chain until the turn start surfaces. A page boundary avoids landing
between a tool call and its persisted result: when the selection starts
mid-pair the page extends to the owning assistant record if that owner is
reachable within one further window (`limit` records) below the selection
and the added records fit the same extra byte budget, so independently
replayed backward pages do not split the pair; an owner beyond either
budget keeps the bounded selection and the page starts mid-pair. Worst
case a page holds three windows (`3 * limit` records). The workspace route
caps serialized responses at twice the core page ceiling — a chosen cap
with headroom for the response envelope, not a derived guarantee, since one
oversized aggregated record can exceed any page budget; a page that still
exceeds it returns `transcript_page_too_large`. Backward page replay
carries no state across records beyond tool call/result pairing and
per-page goal-state reseeding — assistant text and thought parts are
self-contained within a record — so mid-turn boundaries break nothing
else. Forward cursors and responses remain byte-for-byte compatible.
### WebUI transcript state

184
packages/cli/src/acp-integration/acpAgent.test.ts
View file

@ -276,6 +276,12 @@ vi.mock('@qwen-code/qwen-code-core', async (importOriginal) => ({
Buffer.from(JSON.stringify(state), 'utf8').toString('base64url'),
),
SessionTranscriptReader: vi.fn(),
isReplayTurnStartType: (
await importOriginal<typeof import('@qwen-code/qwen-code-core')>()
).isReplayTurnStartType,
findBoundaryAtOrBefore: (
await importOriginal<typeof import('@qwen-code/qwen-code-core')>()
).findBoundaryAtOrBefore,
ALL_PROVIDERS: [
{
id: 'deepseek',
@ -15145,6 +15151,184 @@ describe('QwenAgent loadSession / unstable_resumeSession', () => {
await agentPromise;
});
it('loadSession bounds bulk replay expansion inside one long turn', async () => {
const makeMessage = (
uuid: string,
parentUuid: string | null,
type: 'user' | 'assistant',
) => ({
uuid,
parentUuid,
sessionId: 'persisted-bounded',
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')];
let parent = 'u1';
for (let i = 1; i <= 20; i++) {
messages.push(makeMessage(`a${i}`, parent, 'assistant'));
parent = `a${i}`;
}
bindRestoreMocks({
sessionExists: true,
resumedConversation: { messages },
});
mockHistoryReplay.mockImplementation(async (_context, history) => {
// The alignment walk stays bounded to one extra window instead of
// expanding the single long turn into the whole history, and the
// client can still page backward.
expect(history).toEqual(messages.slice(19));
});
const { agent, agentPromise } = await spawnAgent();
const response = (await agent.loadSession({
cwd: '/tmp',
sessionId: 'persisted-bounded',
mcpServers: [],
_meta: {
'qwen.session.loadReplayMode': 'bulk',
'qwen.session.loadReplayPageSize': 2,
},
})) as {
_meta?: Record<string, { hasMore?: boolean }>;
};
expect(response._meta?.['qwen.session.loadReplay']?.hasMore).toBe(true);
mockConnectionState.resolve();
await agentPromise;
});
it('loadSession bulk replay extends a mid-pair window to the owning call', async () => {
const makeMessage = (
uuid: string,
parentUuid: string | null,
type: 'user' | 'assistant' | 'tool_result',
) => ({
uuid,
parentUuid,
sessionId: 'persisted-midpair',
timestamp: '2026-07-16T00:00:00.000Z',
type,
cwd: '/tmp',
version: 'test',
message: {
role: type === 'assistant' ? ('model' as const) : ('user' as const),
parts: [],
},
});
// One call owning a result run; the requested window lands mid-run, so
// without pair extension the page would start on an orphaned
// tool_result (replaying the completed call as failed).
const messages = [
makeMessage('u1', null, 'user'),
makeMessage('ac1', 'u1', 'assistant'),
makeMessage('ar1', 'ac1', 'tool_result'),
makeMessage('ar2', 'ar1', 'tool_result'),
makeMessage('ar3', 'ar2', 'tool_result'),
];
bindRestoreMocks({
sessionExists: true,
resumedConversation: { messages },
});
let capturedHistory: unknown;
mockHistoryReplay.mockImplementation(async (_context, history) => {
capturedHistory = history;
});
const { agent, agentPromise } = await spawnAgent();
const response = (await agent.loadSession({
cwd: '/tmp',
sessionId: 'persisted-midpair',
mcpServers: [],
_meta: {
'qwen.session.loadReplayMode': 'bulk',
'qwen.session.loadReplayPageSize': 2,
},
})) as {
_meta?: Record<
string,
{ hasMore?: boolean; partial?: boolean; replayError?: string }
>;
};
// The pageSize-2 window lands on ar2/ar3; pair extension pulls the page
// down to the owning call ac1 so it does not start mid-pair.
expect(capturedHistory).toEqual(messages.slice(1));
expect(
response._meta?.['qwen.session.loadReplay']?.partial,
).toBeUndefined();
expect(response._meta?.['qwen.session.loadReplay']?.hasMore).toBe(true);
mockConnectionState.resolve();
await agentPromise;
});
it('loadSession bulk replay passes over mid-turn notification records', async () => {
const makeMessage = (
uuid: string,
parentUuid: string | null,
type: 'user' | 'assistant',
subtype?: string,
) => ({
uuid,
parentUuid,
sessionId: 'persisted-notification',
timestamp: '2026-07-16T00:00:00.000Z',
type,
...(subtype !== undefined ? { subtype } : {}),
cwd: '/tmp',
version: 'test',
message: { role: type === 'user' ? 'user' : 'model', parts: [] },
});
const messages = [
makeMessage('u1', null, 'user'),
makeMessage('a1', 'u1', 'assistant'),
makeMessage('notif', 'a1', 'user', 'notification'),
makeMessage('a2', 'notif', 'assistant'),
makeMessage('a3', 'a2', 'assistant'),
];
bindRestoreMocks({
sessionExists: true,
resumedConversation: { messages },
});
let capturedHistory: unknown;
mockHistoryReplay.mockImplementation(async (_context, history) => {
capturedHistory = history;
});
const { agent, agentPromise } = await spawnAgent();
const response = (await agent.loadSession({
cwd: '/tmp',
sessionId: 'persisted-notification',
mcpServers: [],
_meta: {
'qwen.session.loadReplayMode': 'bulk',
'qwen.session.loadReplayPageSize': 2,
},
})) as {
_meta?: Record<
string,
{ hasMore?: boolean; partial?: boolean; replayError?: string }
>;
};
// A notification is a mid-turn record, not a turn start: the bounded
// alignment passes over it and keeps the trailing window instead of
// realigning the page onto it.
expect(capturedHistory).toEqual(messages.slice(3));
expect(
response._meta?.['qwen.session.loadReplay']?.partial,
).toBeUndefined();
expect(response._meta?.['qwen.session.loadReplay']?.hasMore).toBe(true);
mockConnectionState.resolve();
await agentPromise;
});
it('loadSession returns partial bulk replay updates when replay throws', async () => {
const messages = [{ role: 'user', parts: [{ text: 'hi' }] }];
bindRestoreMocks({

69
packages/cli/src/acp-integration/acpAgent.ts
View file

@ -72,6 +72,8 @@ import {
SessionTranscriptSnapshotUnavailableError,
SessionTranscriptTooLargeError,
encodeSessionTranscriptCursor,
findBoundaryAtOrBefore,
isReplayTurnStartType,
subagentGenerator,
redactUrlCredentials,
computeUniqueBranchTitle,
@ -717,7 +719,37 @@ function getLoadReplayPageSize(params: LoadSessionRequest): number | undefined {
}
function isHistoryTurnStart(record: ChatRecord): boolean {
return record.type === 'user' && record.subtype !== 'mid_turn_user_message';
return isReplayTurnStartType(record.type, record.subtype);
}
// A bulk page can safely start at a turn start or at the assistant record
// owning any following tool results, mirroring the core reader's page-start
// rule. Realtime records interleave at wall-clock time and own no tool
// results, so the pair walk passes through them instead of splitting them.
function isHistoryPageStart(record: ChatRecord): boolean {
return (
record.subtype !== 'realtime_message' &&
(record.type === 'assistant' || isHistoryTurnStart(record))
);
}
// True when the first tool_result in records[start, end) lost its owning
// call below `start`, i.e. the selection begins mid-pair. Only the first
// result needs checking: later results belong to calls at or after it, all
// inside the page once the first pair is whole.
function historyOrphansToolResult(
records: ChatRecord[],
start: number,
end: number,
): boolean {
for (let i = start; i < end; i++) {
if (records[i]!.type !== 'tool_result') continue;
for (let owner = i - 1; owner >= start; owner--) {
if (isHistoryPageStart(records[owner]!)) return false;
}
return true;
}
return false;
}
function selectRecentHistoryRecords(
@ -734,8 +766,39 @@ function selectRecentHistoryRecords(
break;
}
}
while (start > 0 && !isHistoryTurnStart(records[start]!)) {
start--;
// Turn-boundary alignment may expand the page past the requested window,
// but never without bound: a history dominated by a single long in-flight
// turn would otherwise replay the WHOLE history in one payload and report
// hasMore=false, leaving the client unable to page backward. Allow at
// most one extra window of expansion, and only when it reaches a real
// turn start; otherwise keep the requested window so pages inside a long
// turn stay bounded and chainable.
const expansionFloor = Math.max(0, records.length - 2 * pageSize);
const aligned = findBoundaryAtOrBefore(
records,
start,
expansionFloor,
isHistoryTurnStart,
);
if (isHistoryTurnStart(records[aligned]!)) {
start = aligned;
}
// Backward replay renders each page independently, so a page that starts
// mid-pair (on a tool_result whose owning call lies below) would show the
// completed call as failed and its result as an orphan block. When the
// bounded selection starts on an orphaned tool_result, extend down to the
// owning record within one further window, mirroring the core reader.
if (start > 0 && historyOrphansToolResult(records, start, records.length)) {
const pairFloor = Math.max(0, start - pageSize);
const owner = findBoundaryAtOrBefore(
records,
start,
pairFloor,
isHistoryPageStart,
);
if (isHistoryPageStart(records[owner]!)) {
start = owner;
}
}
return { records: records.slice(start), hasMore: start > 0 };
}

11
packages/cli/src/serve/routes/session.ts
View file

@ -15,6 +15,7 @@ import {
SessionOrganizationError,
SessionService,
SESSION_TRANSCRIPT_MAX_LIMIT,
SESSION_TRANSCRIPT_MAX_EXPANDED_PAGE_BYTES,
SESSION_TRANSCRIPT_MAX_PAGE_BYTES,
SessionTranscriptPageTooLargeError,
SessionTranscriptCursorCodec,
@ -175,7 +176,15 @@ interface RegisterSessionRoutesDeps {
isLiveSessionActive?: (sessionId: string) => boolean;
}
const WORKSPACE_TRANSCRIPT_RESPONSE_MAX_BYTES = 32 * 1024 * 1024;
// Chosen cap for one serialized transcript response, kept proportional to
// the core expanded-page ceiling so the two cannot drift arbitrarily. This
// is not a derived guarantee: a single aggregated record can exceed any
// page budget (the reader always takes at least one record so pagination
// cannot dead-end), and replayed SessionUpdate objects are not a fixed
// multiple of their source records. A page this route cannot serialize
// returns transcript_page_too_large for that anchor.
const WORKSPACE_TRANSCRIPT_RESPONSE_MAX_BYTES =
2 * SESSION_TRANSCRIPT_MAX_EXPANDED_PAGE_BYTES;
const WORKSPACE_TRANSCRIPT_CURSOR_MAX_BYTES = 64 * 1024;
const TRANSCRIPT_CURSOR_TOO_LARGE_REPLAY_ERROR =
'Transcript pagination state exceeds the safe limit';

3
packages/core/src/index.ts
View file

@ -288,9 +288,12 @@ export * from './services/session-writer-lease.js';
export {
decodeSessionTranscriptCursor,
encodeSessionTranscriptCursor,
findBoundaryAtOrBefore,
InvalidSessionTranscriptCursorError,
isReplayTurnStartType,
SESSION_TRANSCRIPT_CURSOR_VERSION,
SESSION_TRANSCRIPT_DEFAULT_LIMIT,
SESSION_TRANSCRIPT_MAX_EXPANDED_PAGE_BYTES,
SESSION_TRANSCRIPT_MAX_INDEX_BYTES,
SESSION_TRANSCRIPT_MAX_LIMIT,
SESSION_TRANSCRIPT_MAX_PAGE_BYTES,

831
packages/core/src/services/session-transcript-reader.test.ts
View file

@ -44,9 +44,11 @@ import {
encodeSessionTranscriptCursor,
getSessionTranscriptIndexCacheStatsForTest,
InvalidSessionTranscriptCursorError,
isReplayTurnStartType,
SESSION_TRANSCRIPT_MAX_INDEX_BYTES,
SESSION_TRANSCRIPT_MAX_LIMIT,
resetSessionTranscriptIndexCacheForTest,
setSessionTranscriptExpandedPageBytesForTest,
setSessionTranscriptIndexCacheMaxBytesForTest,
SessionTranscriptCursorCodec,
SessionTranscriptSnapshotUnavailableError,
@ -131,6 +133,46 @@ describe('SessionTranscriptReader', () => {
};
}
function toolCallRecord(
uuid: string,
parentUuid: string,
callId: string,
): ChatRecord {
return {
...record(uuid, parentUuid, ''),
message: {
role: 'model',
parts: [
{ functionCall: { name: 'run_shell_command', id: callId, args: {} } },
],
},
};
}
function toolResultRecord(
uuid: string,
parentUuid: string,
callId: string,
): ChatRecord {
return {
...record(uuid, parentUuid, ''),
type: 'tool_result',
message: {
role: 'user',
parts: [
{
functionResponse: {
name: 'run_shell_command',
id: callId,
response: { output: 'ok' },
},
},
],
},
toolCallResult: { callId, status: 'success' },
};
}
function encodeCursor(
state: Parameters<typeof encodeSessionTranscriptCursor>[0],
): string {
@ -614,30 +656,34 @@ describe('SessionTranscriptReader', () => {
});
it('returns a backward turn that exceeds maxBytes after alignment', async () => {
const prompt = record('u1', null, 'prompt');
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');
const turnRecords = [prompt, toolCall, toolResult, finalAnswer];
await writeRecords([
record('u1', null, 'prompt'),
toolCall,
toolResult,
finalAnswer,
...turnRecords,
record('u2', 'a-final', 'next prompt'),
]);
const turnBytes = turnRecords.reduce(
(total, item) => total + Buffer.byteLength(JSON.stringify(item)),
0,
);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{
beforeRecordId: 'u2',
limit: 2,
maxBytes: Buffer.byteLength(JSON.stringify(finalAnswer)),
// The turn exceeds the soft budget, but it still fits one extra
// budget (2 * maxBytes), so alignment admits it whole.
maxBytes: Math.ceil(turnBytes / 2),
},
);
// The turn cannot be split across pages, so it rides over budget whole.
expect(page.records.map((item) => item.uuid)).toEqual([
'u1',
'a-tool',
@ -1360,5 +1406,778 @@ describe('SessionTranscriptReader', () => {
expect(page.records.map((item) => item.uuid)).toEqual(['a1']);
expect(page.hasMore).toBe(false);
});
it('bounds backward pages inside a single long turn and still chains to the start', async () => {
// One prompt followed by a single long in-flight turn (the concurrent
// /review shape): the only turn start sits at the file head, so the
// turn-alignment walk must NOT expand every backward page to the whole
// transcript — pages stay bounded near the tail and chaining still
// reaches the turn start.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
for (let i = 1; i <= 300; i++) {
records.push(
record(`a${i}`, i === 1 ? 'u1' : `a${i - 1}`, `step ${i}`),
);
}
await writeRecords(records);
const reader = new SessionTranscriptReader(workspaceDir);
const first = await reader.readPage(sessionId, {
direction: 'backward',
limit: 50,
});
// No turn boundary is reachable, so the page stays the requested
// window (`limit` records, not `2 * limit`).
expect(first.records.length).toBe(50);
expect(first.records.at(-1)?.uuid).toBe('a300');
expect(first.records.at(0)?.uuid).toBe('a251');
expect(first.hasMore).toBe(true);
// Chain backward with beforeRecordId anchors (the client's pagination
// shape) until the turn start surfaces.
const seen = new Set(first.records.map((item) => item.uuid));
let boundary: string | undefined = first.records.at(0)?.uuid;
let pages = 1;
while (boundary !== undefined) {
const next = await reader.readPage(sessionId, {
beforeRecordId: boundary,
limit: 50,
});
pages += 1;
expect(next.records.length).toBeLessThanOrEqual(100);
for (const item of next.records) seen.add(item.uuid);
boundary = next.hasMore ? next.records.at(0)?.uuid : undefined;
expect(pages).toBeLessThan(20);
}
expect(seen.size).toBe(301);
expect(seen.has('u1')).toBe(true);
});
it('bounds backward turn expansion under a byte budget in a single long turn', async () => {
const records: ChatRecord[] = [record('u1', null, 'prompt')];
for (let i = 1; i <= 150; i++) {
records.push(
record(`a${i}`, i === 1 ? 'u1' : `a${i - 1}`, `x`.repeat(2000)),
);
}
await writeRecords(records);
const reader = new SessionTranscriptReader(workspaceDir);
const page = await reader.readPage(sessionId, {
direction: 'backward',
limit: 50,
maxBytes: 5000,
});
// The byte budget stops selection two records from the tail; the
// turn-alignment walk must not drag the page back toward the file
// head once alignment proves unreachable within the expansion budget.
expect(page.records.map((item) => item.uuid)).toEqual(['a149', 'a150']);
expect(page.hasMore).toBe(true);
});
it('caps turn-alignment expansion at the hard byte ceiling', async () => {
// A byte-heavy turn whose start sits within the expansion window
// below the selection. The caller's maxBytes is above half the
// (test-only) ceiling, so the expansion budget is the ceiling itself
// rather than 2 * maxBytes: the ~56 KB turn fits 2 * maxBytes
// (64 KB) but not the 48 KB ceiling, so alignment must keep the
// bounded selection. A mutant that dropped the ceiling clamp (or
// ignored the override) would admit the whole turn and fail the u1
// assertion below.
setSessionTranscriptExpandedPageBytesForTest(48 * 1024);
const records: ChatRecord[] = [record('u1', null, 'prompt')];
for (let i = 1; i <= 20; i++) {
records.push(
record(`a${i}`, i === 1 ? 'u1' : `a${i - 1}`, 'x'.repeat(2560)),
);
}
await writeRecords(records);
const maxBytes = 32 * 1024;
const reader = new SessionTranscriptReader(workspaceDir);
const page = await reader.readPage(sessionId, {
direction: 'backward',
limit: 50,
maxBytes,
});
// The soft budget admits a handful of records from the tail; the
// ~56 KB turn exceeds the 48 KB ceiling clamp, so the bounded
// selection stands instead of the whole turn.
expect(page.records.length).toBeLessThan(20);
expect(page.records.at(-1)?.uuid).toBe('a20');
expect(page.records.some((item) => item.uuid === 'u1')).toBe(false);
expect(page.hasMore).toBe(true);
expect(
mockDebugLogger.debug.mock.calls.some(
(args) =>
String(args[0]).includes('backward turn expansion skipped') &&
String(args[0]).includes('reason=byte-budget'),
),
).toBe(true);
// Chaining still reaches the turn start: once the remaining turn
// fits the expansion budget, alignment admits it whole.
const seen = new Set(page.records.map((item) => item.uuid));
let boundary: string | undefined = page.records.at(0)?.uuid;
let pages = 1;
while (boundary !== undefined) {
const next = await reader.readPage(sessionId, {
beforeRecordId: boundary,
limit: 50,
maxBytes,
});
pages += 1;
for (const item of next.records) seen.add(item.uuid);
boundary = next.hasMore ? next.records.at(0)?.uuid : undefined;
expect(pages).toBeLessThan(40);
}
expect(seen.size).toBe(records.length);
expect(seen.has('u1')).toBe(true);
});
it('keeps chained backward pages within a bounded multiple of the byte budget', async () => {
// Turn-alignment expansion is capped at a bounded multiple of the
// caller's maxBytes, not at the hard ceiling: with the nominal route
// budget a chained page must not balloon straight to the ceiling.
// The ceiling clamp itself is exercised by the test above.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
for (let i = 1; i <= 60; i++) {
records.push(
record(`a${i}`, i === 1 ? 'u1' : `a${i - 1}`, 'x'.repeat(3 * 1024)),
);
}
await writeRecords(records);
const maxBytes = 16 * 1024;
const reader = new SessionTranscriptReader(workspaceDir);
let boundary: string | undefined;
let pages = 0;
do {
const page = await reader.readPage(sessionId, {
...(boundary === undefined
? { direction: 'backward' as const }
: { beforeRecordId: boundary }),
limit: 50,
maxBytes,
});
pages += 1;
expect(page.records.length).toBeGreaterThan(0);
// Selection respects maxBytes and alignment may add at most one
// extra budget on top — never the whole ceiling.
const pageBytes = page.records.reduce(
(total, item) => total + Buffer.byteLength(JSON.stringify(item)),
0,
);
expect(pageBytes).toBeLessThanOrEqual(2 * maxBytes + 4 * 1024);
boundary = page.hasMore ? page.records.at(0)?.uuid : undefined;
expect(pages).toBeLessThan(40);
} while (boundary !== undefined);
});
it('walks past mid-turn user records to the owning call', async () => {
// notification, cron and goal_runtime records are persisted mid-turn
// as user-role records. They are not turn boundaries and own no tool
// results, so pair extension must pass through them exactly like
// realtime records instead of starting the page on an orphan result.
const subtypes = ['notification', 'cron', 'goal_runtime'] as const;
for (let variant = 0; variant < subtypes.length; variant++) {
const subtype = subtypes[variant]!;
const targetSessionId = `550e8400-e29b-41d4-a716-44665544000${variant}`;
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push(record('af0', 'u1', 'filler'));
records.push(toolCallRecord('ac1', 'af0', 'call-1'));
records.push(toolResultRecord('ar0', 'ac1', 'call-1'));
records.push({
...record('usyn', 'ar0', 'interjection'),
subtype,
});
records.push(toolResultRecord('ar1', 'usyn', 'call-1'));
records.push(record('af1', 'ar1', 'filler'));
// A distinct session per variant: rewriting one file in place can
// keep the inode and byte length, which the index cache keys on.
await writeRecords(records, targetSessionId);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
targetSessionId,
{ direction: 'backward', limit: 3 },
);
expect(page.records.map((item) => item.uuid)).toEqual([
'ac1',
'ar0',
'usyn',
'ar1',
'af1',
]);
expect(page.records.at(0)?.type).not.toBe('tool_result');
}
});
it('extends to the owning call when one tool_result exceeds the byte budget', async () => {
// A tool_result larger than 2 * maxBytes is force-taken by the
// always-take-one-record rule; the pair-extension budget must count
// only the records the extension adds, or the oversized result it is
// joining would fail the check by construction and the page would
// start on an orphan result (replaying the successful call as
// failed "result missing").
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push(toolCallRecord('ac1', 'u1', 'call-1'));
records.push({
...toolResultRecord('ar1', 'ac1', 'call-1'),
message: {
role: 'user',
parts: [
{
functionResponse: {
name: 'run_shell_command',
id: 'call-1',
response: { output: 'x'.repeat(3 * 1024 * 1024) },
},
},
],
},
});
await writeRecords(records);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 50, maxBytes: 1024 * 1024 },
);
expect(page.records.map((item) => item.uuid)).toEqual(['ac1', 'ar1']);
expect(page.records.at(0)?.type).toBe('assistant');
expect(page.hasMore).toBe(true);
});
it('keeps the pair together when the owning call exceeds the expansion budget', async () => {
// The owner assistant record can itself exceed the expansion byte
// budget (e.g. one write_file call carrying a large file body). Pair
// extension exempts the force-joined owner the way the selection loop
// exempts its forced first record, instead of failing the byte check
// by construction and splitting the pair — which would replay the
// successful call as failed ("result missing") on the older page.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push({
...toolCallRecord('ac1', 'u1', 'call-1'),
message: {
role: 'model',
parts: [
{
functionCall: {
name: 'write_file',
id: 'call-1',
args: { content: 'x'.repeat(40 * 1024) },
},
},
],
},
});
records.push(toolResultRecord('ar1', 'ac1', 'call-1'));
await writeRecords(records);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 50, maxBytes: 16 * 1024 },
);
expect(page.records.map((item) => item.uuid)).toEqual(['ac1', 'ar1']);
expect(page.records.at(0)?.type).toBe('assistant');
expect(page.hasMore).toBe(true);
});
it('does not expand a backward page when no tool pair is split', async () => {
// Pair extension exists to keep tool call/result pairs on one page.
// When the selection holds no orphaned tool_result, walking further
// down gains nothing — even though system records are not page
// starts — and only inflates the page past `limit`.
const records: ChatRecord[] = [record('u0', null, 'prompt')];
let parent = 'u0';
for (let i = 0; i < 10; i++) {
records.push(record(`a${i}`, parent, `step ${i}`));
parent = `a${i}`;
}
records.push({
...record('sys1', 'a9', 'goal state'),
type: 'system',
subtype: 'goal_state',
message: undefined,
});
records.push(record('a10', 'sys1', 'step 10'));
records.push(record('a11', 'a10', 'step 11'));
await writeRecords(records);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 3 },
);
expect(page.records.map((item) => item.uuid)).toEqual([
'sys1',
'a10',
'a11',
]);
expect(page.hasMore).toBe(true);
});
it('does not walk past an anchored realtime record without a pair', async () => {
// Turn alignment may legitimately anchor a page on a realtime user
// record. With no orphaned tool_result in the selection the pair
// walk must not refuse that anchor and drag the page back to an
// earlier assistant, burning the whole expansion window.
const records: ChatRecord[] = [];
for (let i = 0; i < 10; i++) {
records.push(
record(`af${i}`, i === 0 ? null : `af${i - 1}`, `filler ${i}`),
);
}
let parent = 'af9';
for (let i = 0; i < 3; i++) {
records.push({
...record(`urt${i}`, parent, `user speech ${i}`),
subtype: 'realtime_message',
});
records.push({
...record(`art${i}`, `urt${i}`, `assistant speech ${i}`),
subtype: 'realtime_message',
});
parent = `art${i}`;
}
await writeRecords(records);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 4 },
);
expect(page.records.map((item) => item.uuid)).toEqual([
'urt1',
'art1',
'urt2',
'art2',
]);
expect(page.hasMore).toBe(true);
});
it('bounds the leading prefix absorbed before the first turn', async () => {
// Sessions can persist a long run of system records ahead of the
// first turn. Aligning to the first turn must not absorb an
// arbitrarily long leading prefix: worst case a page holds
// 3 * limit records.
const records: ChatRecord[] = [];
for (let i = 0; i < 200; i++) {
records.push({
...record(`sys${i}`, i === 0 ? null : `sys${i - 1}`, `event ${i}`),
type: 'system',
subtype: 'ui_telemetry',
message: undefined,
});
}
records.push(record('u1', 'sys199', 'prompt'));
let parent = 'u1';
for (let i = 1; i <= 5; i++) {
records.push(record(`a${i}`, parent, `step ${i}`));
parent = `a${i}`;
}
await writeRecords(records);
const reader = new SessionTranscriptReader(workspaceDir);
const first = await reader.readPage(sessionId, {
direction: 'backward',
limit: 10,
});
expect(first.records.map((item) => item.uuid)).toEqual([
'u1',
'a1',
'a2',
'a3',
'a4',
'a5',
]);
expect(first.hasMore).toBe(true);
expect(
mockDebugLogger.debug.mock.calls.some(
(args) =>
String(args[0]).includes('backward turn expansion skipped') &&
String(args[0]).includes('reason=record-budget'),
),
).toBe(true);
// Chaining still covers the whole leading prefix.
const seen = new Set(first.records.map((item) => item.uuid));
let boundary: string | undefined = first.records.at(0)?.uuid;
let pages = 1;
while (boundary !== undefined) {
const next = await reader.readPage(sessionId, {
beforeRecordId: boundary,
limit: 10,
});
pages += 1;
for (const item of next.records) seen.add(item.uuid);
boundary = next.hasMore ? next.records.at(0)?.uuid : undefined;
expect(pages).toBeLessThan(40);
}
expect(seen.size).toBe(records.length);
});
it('keeps tool call/result pairs on the same backward page', async () => {
// A single long turn of assistant tool calls and persisted results.
// Call 30 has two results (a parallel batch) so backward page
// boundaries land mid-run on tool_result records. Backward replay
// finalizes each page independently, so a boundary between a call and
// its result would render the completed call as failed on the older
// page and the result as an orphan block on the newer one.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
const resultsByCall = new Map<string, string[]>();
let parent = 'u1';
for (let i = 1; i <= 120; i++) {
const callUuid = `ac${i}`;
const callId = `call-${i}`;
records.push(toolCallRecord(callUuid, parent, callId));
const resultUuids: string[] = [];
let resultParent = callUuid;
for (let r = 0; r < (i === 30 ? 2 : 1); r++) {
const resultUuid = i === 30 ? `ar30-${r}` : `ar${i}`;
records.push(toolResultRecord(resultUuid, resultParent, callId));
resultUuids.push(resultUuid);
resultParent = resultUuid;
}
resultsByCall.set(callUuid, resultUuids);
parent = resultUuids[resultUuids.length - 1]!;
}
await writeRecords(records);
const reader = new SessionTranscriptReader(workspaceDir);
const pages: ChatRecord[][] = [];
let page = await reader.readPage(sessionId, {
direction: 'backward',
limit: 50,
});
for (;;) {
pages.push(page.records);
expect(page.records.length).toBeGreaterThan(0);
if (!page.hasMore) break;
const boundary = page.records.at(0)?.uuid;
expect(boundary).toBeDefined();
page = await reader.readPage(sessionId, {
beforeRecordId: boundary,
limit: 50,
});
expect(pages.length).toBeLessThan(20);
}
let sawMidTurnCallStart = false;
for (const pageRecords of pages) {
// A page starting at a tool_result would replay that result without
// its call.
expect(pageRecords.at(0)?.type).not.toBe('tool_result');
const uuids = new Set(pageRecords.map((item) => item.uuid));
for (const item of pageRecords) {
if (item.type === 'tool_result') {
expect(item.parentUuid).not.toBeNull();
expect(uuids.has(item.parentUuid!)).toBe(true);
}
const results = resultsByCall.get(item.uuid);
if (results) {
for (const resultUuid of results) {
expect(uuids.has(resultUuid)).toBe(true);
}
if (pageRecords.at(0)?.uuid === item.uuid) {
sawMidTurnCallStart = true;
}
}
}
// Symmetric end invariant: the page must not end on a call whose
// results live on the newer page (without a byte budget the pair
// extension always succeeds, so no accepted mid-pair edge exists).
const lastResults = resultsByCall.get(pageRecords.at(-1)?.uuid ?? '');
if (lastResults) {
for (const resultUuid of lastResults) {
expect(uuids.has(resultUuid)).toBe(true);
}
}
}
// The chain really exercised mid-turn page starts, not just
// turn-aligned pages.
expect(sawMidTurnCallStart).toBe(true);
const flat = pages.flat();
expect(flat.length).toBe(records.length);
expect(new Set(flat.map((item) => item.uuid)).size).toBe(records.length);
expect(flat.some((item) => item.uuid === 'u1')).toBe(true);
});
it('extends a byte-limited backward page to the owning tool call', async () => {
const records: ChatRecord[] = [record('u1', null, 'prompt')];
let parent = 'u1';
for (let i = 1; i <= 3; i++) {
records.push(toolCallRecord(`ac${i}`, parent, `call-${i}`));
records.push(toolResultRecord(`ar${i}`, `ac${i}`, `call-${i}`));
parent = `ar${i}`;
}
await writeRecords(records);
// Budget admits only the trailing tool_result; the page must still
// extend to the owning call rather than split the pair.
const resultBytes = Buffer.byteLength(JSON.stringify(records.at(-1)));
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{
direction: 'backward',
limit: 2,
maxBytes: resultBytes,
},
);
expect(page.records.map((item) => item.uuid)).toEqual(['ac3', 'ar3']);
expect(page.hasMore).toBe(true);
});
it('extends the page to an owning call several records below the selection', async () => {
// One call owning a tool_result run long enough that the natural
// selection starts mid-run: the page must extend through the whole
// result run to the owning call instead of starting mid-pair on a
// tool_result.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push(toolCallRecord('ac1', 'u1', 'call-1'));
let parent = 'ac1';
for (let i = 0; i < 5; i++) {
records.push(toolResultRecord(`ar${i}`, parent, 'call-1'));
parent = `ar${i}`;
}
await writeRecords(records);
// limit 3 lands the natural selection start on the third result
// (ar2); the owning call sits exactly three records below it, at the
// edge of the one-window pair-extension budget.
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 3 },
);
expect(page.records.map((item) => item.uuid)).toEqual([
'ac1',
'ar0',
'ar1',
'ar2',
'ar3',
'ar4',
]);
expect(page.records.at(0)?.type).not.toBe('tool_result');
expect(page.hasMore).toBe(true);
});
it('walks past interleaved realtime records to the owning call', async () => {
// Realtime conversation records persist at wall-clock time and can
// land between a call and its results. They own no tool results, so
// pair extension must pass through them instead of splitting the
// pair at the interjection.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push(record('af0', 'u1', 'filler'));
records.push(toolCallRecord('ac1', 'af0', 'call-1'));
records.push(toolResultRecord('ar0', 'ac1', 'call-1'));
records.push({
...record('a-live', 'ar0', 'live interjection'),
subtype: 'realtime_message',
});
records.push(toolResultRecord('ar1', 'a-live', 'call-1'));
records.push(record('af1', 'ar1', 'filler'));
await writeRecords(records);
// limit 3 lands the natural selection start on the realtime record;
// the owning call sits three records below it, at the edge of the
// one-window pair-extension budget.
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 3 },
);
expect(page.records.map((item) => item.uuid)).toEqual([
'ac1',
'ar0',
'a-live',
'ar1',
'af1',
]);
expect(page.records.at(0)?.type).not.toBe('tool_result');
expect(page.hasMore).toBe(true);
});
it('caps pair extension at the byte budget for a large result batch', async () => {
// A long single turn whose tail holds a parallel batch of large
// tool_result records owned by one call. The byte budget stops the
// selection mid-batch; pair extension toward the owner would absorb
// the rest of the batch and balloon the page far past the budget
// (toward the route's hard response cap), so the bounded selection
// must stand and chaining continues from the mid-batch anchor.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
let parent = 'u1';
for (let i = 1; i <= 150; i++) {
records.push(record(`af${i}`, parent, `step ${i}`));
parent = `af${i}`;
}
records.push(toolCallRecord('ac1', parent, 'call-1'));
parent = 'ac1';
for (let i = 0; i < 50; i++) {
const resultUuid = `ar${i}`;
records.push({
...toolResultRecord(resultUuid, parent, 'call-1'),
message: {
role: 'user',
parts: [
{
functionResponse: {
name: 'run_shell_command',
id: 'call-1',
response: { output: 'x'.repeat(4000) },
},
},
],
},
});
parent = resultUuid;
}
await writeRecords(records);
const reader = new SessionTranscriptReader(workspaceDir);
const page = await reader.readPage(sessionId, {
direction: 'backward',
limit: 100,
maxBytes: 10000,
});
// The budget admits two large results; the 49-record extension to the
// owning call does not fit one extra budget, so the page stays at the
// bounded selection (a mid-batch boundary) instead of 51 records.
expect(page.records.map((item) => item.uuid)).toEqual(['ar48', 'ar49']);
expect(page.hasMore).toBe(true);
expect(
mockDebugLogger.debug.mock.calls.some(
(args) =>
String(args[0]).includes('backward pair extension skipped') &&
String(args[0]).includes('reason=byte-budget'),
),
).toBe(true);
// Chaining continues from the mid-batch anchor under the same byte
// budget: every page stays bounded, chaining terminates, and the full
// record set is covered exactly once.
const seen = new Set(page.records.map((item) => item.uuid));
let boundary: string | undefined = page.records.at(0)?.uuid;
let pages = 1;
while (boundary !== undefined) {
const next = await reader.readPage(sessionId, {
beforeRecordId: boundary,
limit: 100,
maxBytes: 10000,
});
pages += 1;
expect(next.records.length).toBeGreaterThan(0);
expect(next.records.length).toBeLessThanOrEqual(200);
for (const item of next.records) seen.add(item.uuid);
boundary = next.hasMore ? next.records.at(0)?.uuid : undefined;
expect(pages).toBeLessThan(40);
}
expect(seen.size).toBe(records.length);
expect(seen.has('u1')).toBe(true);
});
it('caps pair extension for a long tool_result run', async () => {
// One assistant record owning a long contiguous tool_result run (a
// persisted parallel batch): the walk toward the owning call must
// stay bounded instead of absorbing the whole run, and chaining must
// still reach the owner and the turn start.
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push(toolCallRecord('ac1', 'u1', 'call-1'));
let parent = 'ac1';
for (let i = 0; i < 400; i++) {
const resultUuid = `ar${i}`;
records.push(toolResultRecord(resultUuid, parent, 'call-1'));
parent = resultUuid;
}
await writeRecords(records);
const reader = new SessionTranscriptReader(workspaceDir);
const first = await reader.readPage(sessionId, {
direction: 'backward',
limit: 50,
});
expect(first.records.length).toBeLessThanOrEqual(100);
// The owner lies below the expansion budget, so the bounded selection
// stands and the page starts mid-run on a tool_result record.
expect(first.records.at(0)?.type).toBe('tool_result');
expect(first.records.at(-1)?.uuid).toBe('ar399');
expect(first.hasMore).toBe(true);
expect(
mockDebugLogger.debug.mock.calls.some(
(args) =>
String(args[0]).includes('backward pair extension skipped') &&
String(args[0]).includes('reason=record-budget'),
),
).toBe(true);
const seen = new Set(first.records.map((item) => item.uuid));
let boundary: string | undefined = first.records.at(0)?.uuid;
let pages = 1;
while (boundary !== undefined) {
const next = await reader.readPage(sessionId, {
beforeRecordId: boundary,
limit: 50,
});
pages += 1;
// Contract bound: requested window + one alignment window + one
// pair-extension window (the page absorbing the owning call).
expect(next.records.length).toBeLessThanOrEqual(150);
for (const item of next.records) seen.add(item.uuid);
boundary = next.hasMore ? next.records.at(0)?.uuid : undefined;
expect(pages).toBeLessThan(20);
}
expect(seen.size).toBe(records.length);
expect(seen.has('ac1')).toBe(true);
expect(seen.has('u1')).toBe(true);
});
it('keeps a byte-limited page bounded against a long tool_result run', async () => {
const records: ChatRecord[] = [record('u1', null, 'prompt')];
records.push(toolCallRecord('ac1', 'u1', 'call-1'));
let parent = 'ac1';
for (let i = 0; i < 400; i++) {
const resultUuid = `ar${i}`;
records.push(toolResultRecord(resultUuid, parent, 'call-1'));
parent = resultUuid;
}
await writeRecords(records);
const page = await new SessionTranscriptReader(workspaceDir).readPage(
sessionId,
{ direction: 'backward', limit: 50, maxBytes: 5000 },
);
// The budget admits only a few records; pair extension must not drag
// the page back through the 400-record run toward the owning call.
expect(page.records.length).toBeGreaterThan(0);
expect(page.records.length).toBeLessThanOrEqual(100);
expect(page.records.at(0)?.type).toBe('tool_result');
expect(page.records.at(-1)?.uuid).toBe('ar399');
expect(page.hasMore).toBe(true);
});
});
});
describe('isReplayTurnStartType', () => {
it('treats only non-mid-turn user records as turn starts', () => {
expect(isReplayTurnStartType('user', undefined)).toBe(true);
expect(isReplayTurnStartType('user', 'slash_command')).toBe(true);
expect(isReplayTurnStartType('user', 'realtime_message')).toBe(true);
expect(isReplayTurnStartType('user', 'mid_turn_user_message')).toBe(false);
expect(isReplayTurnStartType('user', 'notification')).toBe(false);
expect(isReplayTurnStartType('user', 'cron')).toBe(false);
expect(isReplayTurnStartType('user', 'goal_runtime')).toBe(false);
expect(isReplayTurnStartType('assistant', undefined)).toBe(false);
expect(isReplayTurnStartType(undefined, undefined)).toBe(false);
});
});

289
packages/core/src/services/session-transcript-reader.ts
View file

@ -28,6 +28,16 @@ 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;
// Hard source-byte ceiling for one backward page, counting everything the
// turn-alignment and pair extensions add above the soft `maxBytes`
// selection budget. It is a backstop, not the only bound: each expansion is
// also capped at a bounded multiple of the caller's `maxBytes`. The
// workspace route caps serialized responses at twice this value, leaving
// headroom for the envelope; a single aggregated record can still exceed
// both caps (the always-take-one-record rule admits it so pagination
// cannot dead-end), in which case that anchor reports
// transcript_page_too_large.
export const SESSION_TRANSCRIPT_MAX_EXPANDED_PAGE_BYTES = 16 * 1024 * 1024;
export class InvalidSessionTranscriptCursorError extends Error {
constructor(message = 'Invalid transcript cursor') {
@ -168,6 +178,11 @@ const indexCache = new Map<string, CacheEntry>();
// who can already read the key file next to the transcripts it signs.
const cursorHmacKeys = new Map<string, Buffer>();
let indexCacheMaxBytesForTest: number | undefined;
let expandedPageBytesForTest: number | undefined;
function getExpandedPageBytes(): number {
return expandedPageBytesForTest ?? SESSION_TRANSCRIPT_MAX_EXPANDED_PAGE_BYTES;
}
function makeSessionTranscriptNotFoundError(
sessionId: string,
@ -480,17 +495,139 @@ 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';
// User-role records the turn loop persists mid-turn. Replay renders them
// as inline messages, not turn boundaries (see projectUserRecord in
// transcript-replay), so turn alignment and page starts must pass over
// them. realtime_message is deliberately absent: a realtime user record is
// a genuine user turn start even though it is not a page start (see
// isReplayPageStart).
const REPLAY_MID_TURN_USER_SUBTYPES: ReadonlySet<string> = new Set([
'goal_runtime',
'notification',
'cron',
'mid_turn_user_message',
] satisfies ReadonlyArray<NonNullable<ChatRecord['subtype']>>);
export function isReplayTurnStartType(
type: ChatRecord['type'] | undefined,
subtype: string | undefined,
): boolean {
return (
type === 'user' &&
(subtype === undefined || !REPLAY_MID_TURN_USER_SUBTYPES.has(subtype))
);
}
function isReplayTurnStart(index: TranscriptIndex, uuid: string): boolean {
const entry = index.byUuid.get(uuid);
return isReplayTurnStartType(entry?.type, entry?.subtype);
}
// A backward page can safely start at a replay turn start or at the
// assistant record owning any following tool results. The turn loop
// persists one assistant record per model response and records each tool
// run's results as one contiguous batch before the next assistant record,
// so the nearest matching assistant below a tool_result run owns it.
// Realtime conversation records are the exception: they interleave at
// wall-clock time and own no tool results, so the walk must pass through
// them instead of splitting the pair.
function isReplayPageStart(index: TranscriptIndex, uuid: string): boolean {
const entry = index.byUuid.get(uuid);
return (
entry?.subtype !== 'realtime_message' &&
(entry?.type === 'assistant' ||
isReplayTurnStartType(entry?.type, entry?.subtype))
);
}
// Walk backward from `from` toward the nearest item matching `isBoundary`,
// never below `floor`. The returned index is a boundary only if one exists
// within the bound; otherwise it is `floor` itself, so callers must re-check
// the result. Shared by the uuid-indexed reader and the record-array
// selectors (ACP bulk replay) so the walk/floor/accept policy lives in one
// place.
export function findBoundaryAtOrBefore<T>(
items: ArrayLike<T>,
from: number,
floor: number,
isBoundary: (item: T) => boolean,
): number {
let candidate = from;
while (candidate > floor && !isBoundary(items[candidate]!)) {
candidate--;
}
return candidate;
}
function findReplayBoundaryAtOrBefore(
index: TranscriptIndex,
from: number,
floor: number,
isBoundary: (index: TranscriptIndex, uuid: string) => boolean,
): number {
return findBoundaryAtOrBefore(index.activeUuids, from, floor, (uuid) =>
isBoundary(index, uuid),
);
}
function backwardPageBytesFit(
index: TranscriptIndex,
start: number,
end: number,
budget: number,
): boolean {
let total = 0;
for (let i = start; i < end; i++) {
total += recordSegmentBytes(index, index.activeUuids[i]!);
if (total > budget) return false;
}
return true;
}
// True when the first tool_result in [start, end) lost its owning call
// below `start`, i.e. the selection begins mid-pair. Only the first result
// needs checking: later results belong to calls at or after it, all inside
// the page once the first pair is whole.
function selectionOrphansToolResult(
index: TranscriptIndex,
start: number,
end: number,
): boolean {
for (let i = start; i < end; i++) {
if (index.byUuid.get(index.activeUuids[i]!)?.type !== 'tool_result') {
continue;
}
for (let owner = i - 1; owner >= start; owner--) {
if (isReplayPageStart(index, index.activeUuids[owner]!)) return false;
}
return true;
}
return false;
}
// Selects one backward page. Worst case the page holds 3 * limit records:
// the requested window, one turn-alignment window, and one pair-extension
// window. Each expansion is additionally capped at one extra byte budget —
// a bounded multiple of the soft `maxBytes` budget, clamped to the hard
// page ceiling — so chained pages stay bounded relative to the caller's
// budget instead of jumping straight to the ceiling.
function selectBackwardPageUuids(
index: TranscriptIndex,
sessionId: string,
position: number,
limit: number,
maxBytes: number | undefined,
): { uuids: string[]; nextPosition: number } {
if (position === 0) return { uuids: [], nextPosition: 0 };
// One extra byte budget per expansion: enough to admit a small
// over-budget turn or absorb a result batch whole, but bounded relative
// to the caller's soft budget so chained pages cannot balloon to the
// absolute ceiling.
const expansionByteBudget =
maxBytes === undefined
? getExpandedPageBytes()
: Math.min(2 * maxBytes, getExpandedPageBytes());
let start = Math.max(0, position - limit);
for (let i = start; i < position; i++) {
if (isReplayTurnStart(index, index.activeUuids[i]!)) {
@ -498,8 +635,23 @@ function selectBackwardPageUuids(
break;
}
}
while (start > 0 && !isReplayTurnStart(index, index.activeUuids[start]!)) {
start--;
// Turn-boundary alignment may expand the page past the requested window,
// but never without bound: a transcript dominated by a single long turn
// (e.g. one in-flight prompt with thousands of records) would otherwise
// turn EVERY backward page into the whole transcript — ignoring `limit`
// and making anchor-based pagination dead-end at the file head. Allow at
// most one extra window (`limit` records) of expansion, and only when it
// reaches a real boundary: otherwise keep the requested window so pages
// inside a long turn stay `limit` records, not `2 * limit`.
const expansionFloor = Math.max(0, position - 2 * limit);
const expandedStart = findReplayBoundaryAtOrBefore(
index,
start,
expansionFloor,
isReplayTurnStart,
);
if (isReplayTurnStart(index, index.activeUuids[expandedStart]!)) {
start = expandedStart;
}
let selectedStart = position;
@ -507,8 +659,8 @@ function selectBackwardPageUuids(
for (let i = position - 1; i >= start; i--) {
const uuid = index.activeUuids[i]!;
const bytes = recordSegmentBytes(index, uuid);
// A turn cannot be split across pages; always take at least one record
// so an oversized turn cannot dead-end backward pagination.
// Always take at least one record so backward pagination cannot
// dead-end.
if (
selectedStart < position &&
maxBytes !== undefined &&
@ -520,6 +672,16 @@ function selectBackwardPageUuids(
selectedBytes += bytes;
}
// Turn-alignment expansion admits a whole turn even when it overshoots
// the soft `maxBytes` budget, but never past the expansion byte budget:
// a page the workspace route cannot serialize would fail at its response
// cap and dead-end backward pagination at this anchor on every retry.
const logTurnExpansionSkipped = (reason: string): void => {
debugLogger.debug(
`backward turn expansion skipped session=${sessionId} ` +
`start=${index.activeUuids[selectedStart]!} reason=${reason}`,
);
};
let alignedToReplayBoundary = false;
for (let i = selectedStart; i < position; i++) {
if (isReplayTurnStart(index, index.activeUuids[i]!)) {
@ -529,22 +691,102 @@ function selectBackwardPageUuids(
}
}
if (alignedToReplayBoundary && selectedStart > 0) {
let previousTurnStart = selectedStart - 1;
while (
previousTurnStart >= 0 &&
!isReplayTurnStart(index, index.activeUuids[previousTurnStart]!)
) {
previousTurnStart--;
}
const previousTurnStart = findReplayBoundaryAtOrBefore(
index,
selectedStart - 1,
-1,
isReplayTurnStart,
);
if (previousTurnStart < 0) {
selectedStart = 0;
// No earlier turn start anywhere: the file head is the only boundary
// below. Absorb the leading prefix only when it lies inside the same
// record and byte budgets as every other expansion, so a long
// synthetic prefix cannot balloon the page past the 3 * limit worst
// case.
if (
expansionFloor === 0 &&
backwardPageBytesFit(index, 0, position, expansionByteBudget)
) {
selectedStart = 0;
} else {
logTurnExpansionSkipped(
expansionFloor === 0 ? 'byte-budget' : 'record-budget',
);
}
}
} else if (!alignedToReplayBoundary) {
while (
selectedStart > 0 &&
!isReplayTurnStart(index, index.activeUuids[selectedStart]!)
// Expansion only pays off when it reaches a turn boundary; otherwise
// keep the limit/maxBytes-respecting selection.
const candidate = findReplayBoundaryAtOrBefore(
index,
selectedStart,
expansionFloor,
isReplayTurnStart,
);
if (isReplayTurnStart(index, index.activeUuids[candidate]!)) {
if (
backwardPageBytesFit(index, candidate, position, expansionByteBudget)
) {
selectedStart = candidate;
} else {
logTurnExpansionSkipped('byte-budget');
}
}
}
// Backward replay finalizes each page independently, so a page boundary
// between a tool call and its persisted result would render the completed
// call as failed ("result missing") on the older page and the result as an
// orphan block on the newer one. When the selection starts mid-pair,
// extend the page down to the owning assistant record (or turn boundary)
// so the pair stays on a single page. The extension runs only when a
// tool_result in the selection actually lost its call: system records and
// mid-turn user records are not page starts either, but walking further
// down gains nothing when there is no pair to keep together. The walk is
// bounded to one window below the selection: one assistant record can own
// an arbitrarily long contiguous tool_result run (a persisted parallel
// batch), and an uncapped walk would balloon the page far past `limit` —
// reintroducing the unbounded growth this function exists to cap. The
// budget covers only the records the extension adds beyond the owner —
// the selection above already respected `maxBytes`, and the owner itself
// is exempt the way the selection loop exempts its forced first record,
// so a single oversized owner (which the next page would force-take
// anyway) cannot fail the check by construction and split the pair.
// Records between the owner and the selection — a result batch — still
// count against the budget; an extension that would absorb more than the
// budget keeps the bounded selection, accepting a mid-pair boundary in
// that edge. The skip is logged so such a report stays diagnosable
// without re-deriving the budget arithmetic.
if (
selectedStart > 0 &&
selectionOrphansToolResult(index, selectedStart, position)
) {
const pairFloor = Math.max(0, selectedStart - limit);
const pairStart = findReplayBoundaryAtOrBefore(
index,
selectedStart,
pairFloor,
isReplayPageStart,
);
if (!isReplayPageStart(index, index.activeUuids[pairStart]!)) {
debugLogger.debug(
`backward pair extension skipped session=${sessionId} ` +
`start=${index.activeUuids[selectedStart]!} reason=record-budget`,
);
} else if (
!backwardPageBytesFit(
index,
pairStart + 1,
selectedStart,
expansionByteBudget,
)
) {
selectedStart--;
debugLogger.debug(
`backward pair extension skipped session=${sessionId} ` +
`start=${index.activeUuids[selectedStart]!} reason=byte-budget`,
);
} else {
selectedStart = pairStart;
}
}
@ -1124,7 +1366,7 @@ export class SessionTranscriptReader {
}
const backwardPage =
direction === 'backward'
? selectBackwardPageUuids(index, position, limit, maxBytes)
? selectBackwardPageUuids(index, sessionId, position, limit, maxBytes)
: undefined;
const pageUuids =
backwardPage?.uuids ?? selectPageUuids(index, position, limit, maxBytes);
@ -1184,6 +1426,7 @@ export function resetSessionTranscriptIndexCacheForTest(): void {
indexCache.clear();
cursorHmacKeys.clear();
indexCacheMaxBytesForTest = undefined;
expandedPageBytesForTest = undefined;
}
export function setSessionTranscriptIndexCacheMaxBytesForTest(
@ -1193,6 +1436,12 @@ export function setSessionTranscriptIndexCacheMaxBytesForTest(
pruneCache();
}
export function setSessionTranscriptExpandedPageBytesForTest(
maxBytes: number,
): void {
expandedPageBytesForTest = maxBytes;
}
export function getSessionTranscriptIndexCacheStatsForTest(): {
entries: number;
byteSize: number;