Skip to content

Commit c3d0f99

Browse files
fix(resume): cap transcript hydration in rows after fold (#1257)
A 600-block mapping window under-fills the retained row tail once call+result pairs fold, so retention never evicts and the dropped-rows notice stays unpainted. Resume now reads a turn window sized for that fold and lets the host cap in rows.
1 parent 9535d0a commit c3d0f99

10 files changed

Lines changed: 341 additions & 56 deletions

‎src/session/optimized-context-store.test.ts‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -369,7 +369,8 @@ describe("loadRecentTurns", () => {
369369
);
370370

371371
const loaded = await loadRecentTurns(dir, 2);
372-
expect(turnTexts(loaded)).toEqual(["d", "e"]);
372+
expect(turnTexts(loaded.turns)).toEqual(["d", "e"]);
373+
expect(loaded.truncated).toBe(true);
373374
});
374375

375376
test("walks back into older segments when the window exceeds the newest one", async () => {
@@ -388,12 +389,27 @@ describe("loadRecentTurns", () => {
388389
// window of 4, so the walk continues into segment 0 (2 turns) whole —
389390
// reads are segment-granular, not turn-exact.
390391
const loaded = await loadRecentTurns(dir, 4);
391-
expect(turnTexts(loaded)).toEqual(["a", "b", "c", "d", "e"]);
392+
expect(turnTexts(loaded.turns)).toEqual(["a", "b", "c", "d", "e"]);
393+
expect(loaded.truncated).toBe(false);
392394
});
393395

394396
test("returns an empty list when no segments exist", async () => {
395397
const dir = tempDir();
396-
expect(await loadRecentTurns(dir, 10)).toEqual([]);
398+
expect(await loadRecentTurns(dir, 10)).toEqual({
399+
turns: [],
400+
truncated: false,
401+
});
402+
});
403+
404+
test("a single segment that fills the window is not truncated", async () => {
405+
const dir = tempDir();
406+
fs.writeFileSync(
407+
path.join(dir, TURNS_FILE),
408+
jsonl([turn("a"), turn("b"), turn("c")]),
409+
);
410+
const loaded = await loadRecentTurns(dir, 2);
411+
expect(turnTexts(loaded.turns)).toEqual(["a", "b", "c"]);
412+
expect(loaded.truncated).toBe(false);
397413
});
398414

399415
test("reactor load skips mid-file garbage in an extra segment", async () => {

‎src/session/optimized-context-store.ts‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -287,6 +287,17 @@ async function unlinkExtraSegmentsFrom(
287287
return removed;
288288
}
289289

290+
/**
291+
* Display-only tail of the turn history. Older segments are unread when
292+
* `truncated` is set — callers that paint a bounded transcript can still
293+
* notice that more history exists even if the loaded window hydrates to
294+
* exactly the retention cap.
295+
*/
296+
export type RecentTurns = {
297+
readonly turns: ConversationTurn[];
298+
readonly truncated: boolean;
299+
};
300+
290301
/**
291302
* Read only the tail of the turn history needed to satisfy `minTurns`, walking
292303
* segments from newest to oldest and stopping as soon as enough turns have
@@ -302,12 +313,13 @@ async function unlinkExtraSegmentsFrom(
302313
export async function loadRecentTurns(
303314
dir: string,
304315
minTurns: number,
305-
): Promise<ConversationTurn[]> {
316+
): Promise<RecentTurns> {
306317
const segments = await listSegmentFiles(dir, TURNS_FILE);
307-
if (segments.length === 0) return [];
318+
if (segments.length === 0) return { turns: [], truncated: false };
308319

309320
const collectedNewestFirst: ConversationTurn[][] = [];
310321
let total = 0;
322+
let truncated = false;
311323
for (let i = segments.length - 1; i >= 0; i--) {
312324
const name = segments[i];
313325
if (name === undefined) continue;
@@ -325,15 +337,18 @@ export async function loadRecentTurns(
325337
);
326338
collectedNewestFirst.push(turns);
327339
total += turns.length;
328-
if (total >= minTurns) break;
340+
if (total >= minTurns) {
341+
truncated = i > 0;
342+
break;
343+
}
329344
}
330345

331346
const turns: ConversationTurn[] = [];
332347
for (let i = collectedNewestFirst.length - 1; i >= 0; i--) {
333348
const chunk = collectedNewestFirst[i];
334349
if (chunk !== undefined) turns.push(...chunk);
335350
}
336-
return turns;
351+
return { turns, truncated };
337352
}
338353

339354
/**

‎src/tui/history-hydrate.test.ts‎

Lines changed: 64 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import {
77
EMPTY_VIEW_DETAIL,
88
hydrateHistoryRows,
99
MISSING_ERROR_DETAIL,
10+
parseHistoryHydratePayload,
1011
rowFromHistoryBlock,
1112
type HistoryBlock,
1213
} from "./history-hydrate.js";
@@ -410,4 +411,67 @@ describe("resume pipeline end to end (turns-to-blocks into hydrate)", () => {
410411
const rows = hydrateHistoryRows(JSON.parse(JSON.stringify(blocks)));
411412
expect(rows).toEqual([]);
412413
});
414+
415+
test("spawn_agent pairs fold to one row each, including a pair at the old 600-block splice", () => {
416+
const turns: ConversationTurn[] = [];
417+
for (let i = 0; i < 400; i++) {
418+
turns.push({
419+
role: "assistant",
420+
model: "test",
421+
timestamp: i,
422+
content: [
423+
{
424+
type: "tool_call",
425+
id: `c${i}`,
426+
name: "spawn_agent",
427+
arguments: { description: `job-${i}` },
428+
},
429+
],
430+
} as unknown as ConversationTurn);
431+
turns.push({
432+
role: "assistant",
433+
model: "test",
434+
timestamp: i,
435+
content: [
436+
{
437+
type: "tool_result",
438+
callId: `c${i}`,
439+
content: `done c${i}`,
440+
isError: false,
441+
},
442+
],
443+
} as unknown as ConversationTurn);
444+
}
445+
turns.push({
446+
role: "user",
447+
content: [{ type: "text", text: "trailing" }],
448+
timestamp: 400,
449+
} as unknown as ConversationTurn);
450+
const blocks = turnsToContentBlocks(turns);
451+
expect(blocks.length).toBe(801);
452+
const rows = hydrateHistoryRows(blocks);
453+
expect(rows.length).toBe(401);
454+
expect(rows[0]?.pending).not.toBe(true);
455+
expect(rows[0]?.text).toBe("done c0");
456+
expect(rows[399]?.text).toBe("done c399");
457+
expect(rows[400]).toEqual({ role: "user", text: "trailing" });
458+
});
459+
});
460+
461+
describe("parseHistoryHydratePayload", () => {
462+
test("keeps a raw block array and treats a wrapper as truncated only when flagged", () => {
463+
const blocks = [{ type: "user", content: "hi" }];
464+
expect(parseHistoryHydratePayload(blocks)).toEqual({
465+
blocks,
466+
truncated: false,
467+
});
468+
expect(parseHistoryHydratePayload({ blocks, truncated: true })).toEqual({
469+
blocks,
470+
truncated: true,
471+
});
472+
expect(parseHistoryHydratePayload({ blocks })).toEqual({
473+
blocks,
474+
truncated: false,
475+
});
476+
});
413477
});

‎src/tui/history-hydrate.ts‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -203,6 +203,31 @@ export function hydrateHistoryRows(blocks: unknown): StreamRow[] {
203203
return rows;
204204
}
205205

206+
/**
207+
* Resume hydrate events are either a raw block array (tests, load errors) or
208+
* `{ blocks, truncated }` when loadRecentTurns left older segments unread.
209+
*/
210+
export function parseHistoryHydratePayload(payload: unknown): {
211+
blocks: unknown;
212+
truncated: boolean;
213+
} {
214+
if (
215+
Array.isArray(payload) ||
216+
payload === null ||
217+
typeof payload !== "object"
218+
) {
219+
return { blocks: payload, truncated: false };
220+
}
221+
const record = payload as Record<string, unknown>;
222+
if (!("blocks" in record)) {
223+
return { blocks: payload, truncated: false };
224+
}
225+
return {
226+
blocks: record.blocks,
227+
truncated: record.truncated === true,
228+
};
229+
}
230+
206231
/** Argument payload of a tool_call block, wherever the block carries it. */
207232
function callArguments(block: HistoryBlock): string | undefined {
208233
if (block.content !== undefined) return block.content;

‎src/tui/product-host.test.ts‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import { EventEmitter } from "node:events";
66
import { describe, expect, test } from "bun:test";
77
import type { KeyEvent } from "@opentui/core";
8+
import type { ConversationTurn } from "@intx/types/runtime";
89
import { createHarness, type Harness } from "./harness.js";
910
import { acceptOverlaySelection } from "./shell/overlay-host.js";
1011
import {
@@ -16,6 +17,7 @@ import { mountProductHost, type ProductHostConfig } from "./product-host.js";
1617
import { buildModelsFirstCatalog, modelOptionId } from "./model-catalog.js";
1718
import { hydrateHistoryRows } from "./history-hydrate.js";
1819
import { MAX_RETAINED_STREAM_ROWS } from "./long-log.js";
20+
import { turnsToContentBlocks } from "./turns-to-blocks.js";
1921
import { enterSubagentObserve } from "./shell/observe.js";
2022
import { transcriptMarker } from "./shell/transcript.js";
2123

@@ -97,6 +99,46 @@ function composedKey(glyph: string): KeyEvent {
9799
} as KeyEvent;
98100
}
99101

102+
function userTextTurn(text: string): ConversationTurn {
103+
return {
104+
role: "user",
105+
content: [{ type: "text", text }],
106+
timestamp: 0,
107+
} as unknown as ConversationTurn;
108+
}
109+
110+
function spawnCallTurn(id: string): ConversationTurn {
111+
return {
112+
role: "assistant",
113+
model: "test",
114+
timestamp: 0,
115+
content: [
116+
{
117+
type: "tool_call",
118+
id,
119+
name: "spawn_agent",
120+
arguments: { description: `job-${id}` },
121+
},
122+
],
123+
} as unknown as ConversationTurn;
124+
}
125+
126+
function spawnResultTurn(id: string): ConversationTurn {
127+
return {
128+
role: "assistant",
129+
model: "test",
130+
timestamp: 0,
131+
content: [
132+
{
133+
type: "tool_result",
134+
callId: id,
135+
content: `done ${id}`,
136+
isError: false,
137+
},
138+
],
139+
} as unknown as ConversationTurn;
140+
}
141+
100142
describe("mountProductHost", () => {
101143
test("stream events emitted on the event emitter paint rows into the shell", async () => {
102144
const { host, emitter } = await mountHeadless();
@@ -260,6 +302,113 @@ describe("mountProductHost", () => {
260302
}
261303
});
262304

305+
test("resume pipeline of long text history paints the eviction marker", async () => {
306+
const { host, emitter } = await mountHeadless();
307+
try {
308+
const total = MAX_RETAINED_STREAM_ROWS + 200;
309+
const turns = Array.from({ length: total }, (_, i) =>
310+
userTextTurn(`row-${i}`),
311+
);
312+
const blocks = turnsToContentBlocks(turns);
313+
emitter.emit("history.hydrate", blocks);
314+
expect(host.shell.streamLog.length).toBe(MAX_RETAINED_STREAM_ROWS);
315+
expect(host.shell.streamLogBase).toBe(200);
316+
expect(host.shell.streamLog[0]).toEqual({
317+
role: "user",
318+
text: "row-200",
319+
});
320+
expect(transcriptMarker(host.shell)).toBeDefined();
321+
} finally {
322+
host.dispose();
323+
}
324+
});
325+
326+
test("resume pipeline of tool pairs fills the retained row cap", async () => {
327+
const { host, emitter } = await mountHeadless();
328+
try {
329+
const pairCount = MAX_RETAINED_STREAM_ROWS + 200;
330+
const turns: ConversationTurn[] = [];
331+
for (let i = 0; i < pairCount; i++) {
332+
turns.push(spawnCallTurn(`p${i}`), spawnResultTurn(`p${i}`));
333+
}
334+
const blocks = turnsToContentBlocks(turns);
335+
emitter.emit("history.hydrate", blocks);
336+
expect(host.shell.streamLog.length).toBe(MAX_RETAINED_STREAM_ROWS);
337+
expect(host.shell.streamLogBase).toBe(200);
338+
const first = host.shell.streamLog[0];
339+
expect(first?.pending).not.toBe(true);
340+
expect(first?.text).toBe("done p200");
341+
const last = host.shell.streamLog[host.shell.streamLog.length - 1];
342+
expect(last?.pending).not.toBe(true);
343+
expect(last?.text).toBe(`done p${pairCount - 1}`);
344+
expect(transcriptMarker(host.shell)).toBeDefined();
345+
} finally {
346+
host.dispose();
347+
}
348+
});
349+
350+
test("resume pipeline keeps a pair merged when it straddles the old block splice", async () => {
351+
const { host, emitter } = await mountHeadless();
352+
try {
353+
const turns: ConversationTurn[] = [];
354+
for (let i = 0; i < 400; i++) {
355+
turns.push(spawnCallTurn(`c${i}`), spawnResultTurn(`c${i}`));
356+
}
357+
turns.push(userTextTurn("trailing"));
358+
const blocks = turnsToContentBlocks(turns);
359+
emitter.emit("history.hydrate", { blocks, truncated: false });
360+
expect(host.shell.streamLog.length).toBe(401);
361+
expect(host.shell.streamLog[0]?.pending).not.toBe(true);
362+
expect(host.shell.streamLog[0]?.text).toBe("done c0");
363+
expect(host.shell.streamLog[host.shell.streamLog.length - 1]).toEqual({
364+
role: "user",
365+
text: "trailing",
366+
});
367+
expect(host.shell.streamLogBase).toBe(0);
368+
} finally {
369+
host.dispose();
370+
}
371+
});
372+
373+
test("resume pipeline keeps a pair merged across the retained row cap", async () => {
374+
const { host, emitter } = await mountHeadless();
375+
try {
376+
const turns = [
377+
...Array.from({ length: 605 }, (_, i) => userTextTurn(`row-${i}`)),
378+
spawnCallTurn("cut-1"),
379+
spawnResultTurn("cut-1"),
380+
];
381+
const blocks = turnsToContentBlocks(turns);
382+
emitter.emit("history.hydrate", blocks);
383+
const allRows = hydrateHistoryRows(blocks);
384+
const expected = allRows.slice(-MAX_RETAINED_STREAM_ROWS);
385+
expect(host.shell.streamLog).toEqual(expected);
386+
const last = host.shell.streamLog[host.shell.streamLog.length - 1];
387+
expect(last?.pending).not.toBe(true);
388+
expect(last?.text).toBe("done cut-1");
389+
expect(host.shell.streamLogBase).toBeGreaterThan(0);
390+
expect(transcriptMarker(host.shell)).toBeDefined();
391+
} finally {
392+
host.dispose();
393+
}
394+
});
395+
396+
test("truncated load of an exact-cap tail still paints the eviction marker", async () => {
397+
const { host, emitter } = await mountHeadless();
398+
try {
399+
const turns = Array.from({ length: MAX_RETAINED_STREAM_ROWS }, (_, i) =>
400+
userTextTurn(`kept-${i}`),
401+
);
402+
const blocks = turnsToContentBlocks(turns);
403+
emitter.emit("history.hydrate", { blocks, truncated: true });
404+
expect(host.shell.streamLog.length).toBe(MAX_RETAINED_STREAM_ROWS);
405+
expect(host.shell.streamLogBase).toBeGreaterThan(0);
406+
expect(transcriptMarker(host.shell)).toBeDefined();
407+
} finally {
408+
host.dispose();
409+
}
410+
});
411+
263412
test("session.title updates the shell header", async () => {
264413
const { host, emitter } = await mountHeadless();
265414
try {

0 commit comments

Comments
 (0)