From 26086c9666c1fe1714e3f6dc6d7eae7296a574ab Mon Sep 17 00:00:00 2001 From: testikun <320479488+testikun@users.noreply.github.com> Date: Fri, 25 Sep 2026 13:18:19 +0900 Subject: [PATCH 01/14] perf(desktop): bound session-switch history and skip unrelated resource replay Read a small recent transcript range, load earlier Turns on upward input, and avoid projecting root-session history for ShellRun resource queries. Generated-by: Codex --- .../runtime-host-session-observer.test.ts | 60 ++++ .../__tests__/transcript-history-read.test.ts | 4 +- .../src/main/desktop-transcript-replica.ts | 3 +- .../src/main/runtime-host-session-observer.ts | 12 +- .../src/preload/transcript-contract.ts | 6 +- apps/desktop/stories/app-shell.stories.tsx | 43 ++- apps/desktop/stories/workhub.stories.tsx | 8 +- .../session-manager-shell-resources.test.ts | 321 ++++++++++++++++++ packages/runtime/src/session-manager.ts | 16 +- .../__tests__/chat-view-load-earlier.test.tsx | 91 +++-- packages/ui/src/chat-view.tsx | 57 ++-- packages/ui/src/conversation-copy.ts | 7 +- .../ui/src/transcript-scroll-authority.tsx | 10 + packages/ui/src/use-chat-scroll.ts | 4 + 14 files changed, 552 insertions(+), 90 deletions(-) create mode 100644 packages/runtime/src/__tests__/session-manager-shell-resources.test.ts diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index e92339ef85f..895391ac632 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -1331,6 +1331,66 @@ test('coalesces transcript changes into one bounded delta while renderer deliver await observer.close(); }); +test('opens a small recent range and reads larger earlier ranges without losing history', async (t) => { + const rows = Array.from({ length: 24 }, (_, index) => + turnRow((index + 1) * 10, `turn-${index}`, 'x'.repeat(64 * 1024)), + ); + const host = historyHost(rows, { pageRows: 2 }); + const requests: number[] = []; + const events = new AsyncFrameQueue(); + const observer = new RuntimeHostSessionObserver({ + client: { + openSession: async () => runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), events, + transcriptBootstrap: host.bootstrap, + loadTranscriptPage: async (request) => { + requests.push(request.maxBytes); + return host.loadTranscriptPage(request); + }, + decodeTranscriptPage: host.decodeTranscriptPage, + async close() { events.end(); }, + }), + }, + emitSessionsChanged() {}, + }); + t.after(() => observer.close()); + const batches: DesktopTranscriptBatch[] = []; + const consumerId = 'consumer-bounded-open'; + await observer.openTranscript('session-1', consumerId, ackingTranscriptTarget(observer, consumerId, 26, batches), 'history'); + const takeAnswer = () => { + const answer = batches.splice(0); + assert.equal(answer.filter((batch) => batch.ready).length, 1); + assert.equal(answer.at(-1)!.ready, true); + return { + rows: answer.flatMap(durableSequences).sort((left, right) => left - right), + hasOlder: answer.at(-1)!.hasOlder, + earlierThan: answer[0]!.earlierThan, + }; + }; + const initial = takeAnswer(); + assert.deepEqual(initial.rows, [230, 240], 'opening must not replay the entire history'); + assert.equal(initial.hasOlder, true); + assert.ok(requests.every((bytes) => bytes <= 128 * 1024), 'bound Host reads as well as the final answer'); + requests.splice(0); + + await observer.loadEarlierTranscript(consumerId, 26); + const earlier = takeAnswer(); + assert.equal(earlier.earlierThan, initial.rows[0]); + assert.ok(earlier.rows.length > initial.rows.length, 'reading earlier uses its own larger budget'); + assert.equal(earlier.hasOlder, true, 'one earlier read does not automatically fetch everything'); + assert.ok(requests.every((bytes) => bytes <= 512 * 1024)); + const delivered = [...earlier.rows, ...initial.rows]; + let hasOlder: boolean | undefined = earlier.hasOlder; + while (hasOlder) { + await observer.loadEarlierTranscript(consumerId, 26); + const answer = takeAnswer(); + assert.equal(answer.earlierThan, delivered[0]); + delivered.unshift(...answer.rows); + hasOlder = answer.hasOlder; + } + assert.deepEqual(delivered, rows.map((row) => row.identity), 'paging must retain every row exactly once'); +}); + test('delivers history in whole Turns within the budget and continues exactly on load earlier', async () => { const events = new AsyncFrameQueue(); // Oldest first: a (3 rows), b (1 huge row), c (2 rows), d (1 huge row). The budget is one and a half diff --git a/apps/desktop/src/main/__tests__/transcript-history-read.test.ts b/apps/desktop/src/main/__tests__/transcript-history-read.test.ts index e83b3189668..7e42c5f3ae8 100644 --- a/apps/desktop/src/main/__tests__/transcript-history-read.test.ts +++ b/apps/desktop/src/main/__tests__/transcript-history-read.test.ts @@ -77,7 +77,9 @@ test('keeps both Turns reachable when an oversized ledger Turn is followed by a const history: { sequence: number; message: StoredMessage }[] = []; let cursor: string | null = null; do { - const page = await replica.readOlderPage(completeThrough, cursor); + // Opening uses a smaller Host page than later reading, including when a + // single tool result needs several pages to finish decoding. + const page = await replica.readOlderPage(completeThrough, cursor, PAGE_BYTES); history.unshift(...page.durable); cursor = page.nextCursor; } while (cursor !== null); diff --git a/apps/desktop/src/main/desktop-transcript-replica.ts b/apps/desktop/src/main/desktop-transcript-replica.ts index 3e70272a5a1..f4dd3b524c4 100644 --- a/apps/desktop/src/main/desktop-transcript-replica.ts +++ b/apps/desktop/src/main/desktop-transcript-replica.ts @@ -262,6 +262,7 @@ export class DesktopTranscriptReplica { async readOlderPage( throughSequence: number, cursor: string | null, + maxBytes = SESSION_TRANSCRIPT_PAGE_MAX_BYTES, ): Promise { this.#assertLive(); const page = await this.#handle.loadTranscriptPage({ @@ -269,7 +270,7 @@ export class DesktopTranscriptReplica { throughSequence, cursor, anchorSequence: null, - maxBytes: SESSION_TRANSCRIPT_PAGE_MAX_BYTES, + maxBytes: Math.min(maxBytes, SESSION_TRANSCRIPT_PAGE_MAX_BYTES), }); return this.#withDecodedPage(page, (decoded) => { this.#assertLive(); diff --git a/apps/desktop/src/main/runtime-host-session-observer.ts b/apps/desktop/src/main/runtime-host-session-observer.ts index d9ee8d6d3ae..4c0e30e467a 100644 --- a/apps/desktop/src/main/runtime-host-session-observer.ts +++ b/apps/desktop/src/main/runtime-host-session-observer.ts @@ -44,6 +44,7 @@ import { DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, DESKTOP_TRANSCRIPT_GLOBAL_CACHE_MAX_BYTES, DESKTOP_TRANSCRIPT_HISTORY_MAX_BYTES, + DESKTOP_TRANSCRIPT_INITIAL_HISTORY_MAX_BYTES, DESKTOP_TRANSCRIPT_MESSAGE_MAX_BYTES, type DesktopTranscriptBatch, type DesktopTranscriptBatchPayload, @@ -239,6 +240,7 @@ export class RuntimeHostSessionObserver { readonly #emitSubscriptionRecovered: (sessionId: string) => void; readonly #recoverConnectionClosed: boolean; readonly #transcriptHistoryBytes: number; + readonly #transcriptInitialHistoryBytes: number; readonly #transcriptGlobalCacheMaxBytes: number; readonly #now: () => number; #closed = false; @@ -264,6 +266,7 @@ export class RuntimeHostSessionObserver { deps.emitSubscriptionRecovered ?? (() => undefined); this.#recoverConnectionClosed = deps.recoverConnectionClosed ?? false; this.#transcriptHistoryBytes = deps.transcriptHistoryBytes ?? DESKTOP_TRANSCRIPT_HISTORY_MAX_BYTES; + this.#transcriptInitialHistoryBytes = deps.transcriptHistoryBytes ?? DESKTOP_TRANSCRIPT_INITIAL_HISTORY_MAX_BYTES; this.#transcriptGlobalCacheMaxBytes = deps.transcriptGlobalCacheMaxBytes ?? DESKTOP_TRANSCRIPT_GLOBAL_CACHE_MAX_BYTES; this.#now = deps.now ?? Date.now; @@ -1436,7 +1439,12 @@ export class RuntimeHostSessionObserver { if (history.oldestSequence === null || !historyHasOlder(history)) return; earlierThan = history.oldestSequence; } - const budget = this.#transcriptHistoryBytes; + // Opening near the live tail needs a small answer. A recovery must still + // restore everything already delivered, so it uses the earlier-read budget + // and continues through the saved floor even when that exceeds the budget. + const budget = reset && floor === null + ? this.#transcriptInitialHistoryBytes + : this.#transcriptHistoryBytes; const identity = { sessionId: replica.sessionId, generation: replica.generation, hostEpoch: replica.hostEpoch }; const isCurrent = () => state.replica === replica && @@ -1465,7 +1473,7 @@ export class RuntimeHostSessionObserver { while (history.throughSequence !== null && historyHasOlder(history)) { let page: Awaited>; try { - page = await replica.readOlderPage(history.throughSequence, history.cursor); + page = await replica.readOlderPage(history.throughSequence, history.cursor, budget); } catch (error) { if (!isCurrent()) return; throw error; diff --git a/apps/desktop/src/preload/transcript-contract.ts b/apps/desktop/src/preload/transcript-contract.ts index 3b3042f4673..40221752b79 100644 --- a/apps/desktop/src/preload/transcript-contract.ts +++ b/apps/desktop/src/preload/transcript-contract.ts @@ -23,8 +23,10 @@ export const DESKTOP_TRANSCRIPT_TAIL_MAX_BYTES = 512 * 1024; export const DESKTOP_TRANSCRIPT_TAIL_MAX_TURNS = 10; export const DESKTOP_TRANSCRIPT_MESSAGE_MAX_BYTES = 16 * 1024 * 1024; export const DESKTOP_TRANSCRIPT_GLOBAL_CACHE_MAX_BYTES = 64 * 1024 * 1024; -/** Projected message bytes of whole Turns one history read delivers: the first read and each "load earlier". */ -export const DESKTOP_TRANSCRIPT_HISTORY_MAX_BYTES = 64 * 1024 * 1024; +/** Recent whole Turns needed to open a reader; older history remains available on demand. */ +export const DESKTOP_TRANSCRIPT_INITIAL_HISTORY_MAX_BYTES = 128 * 1024; +/** Projected message bytes per earlier read; restoring a held range also respects its oldest row. */ +export const DESKTOP_TRANSCRIPT_HISTORY_MAX_BYTES = 512 * 1024; export interface DesktopTranscriptFragment { readonly sequence: number; diff --git a/apps/desktop/stories/app-shell.stories.tsx b/apps/desktop/stories/app-shell.stories.tsx index 510daf470fd..33199bdbf4b 100644 --- a/apps/desktop/stories/app-shell.stories.tsx +++ b/apps/desktop/stories/app-shell.stories.tsx @@ -2514,14 +2514,18 @@ function SettledTranscriptHarness({ * `HISTORY_BATCH` Turns. The loader settles a frame later, the way an answer * delivered over IPC does. */ +let releaseHistoryPage: (() => void) | undefined; + function HistoryHarness({ turns, olderTurns = 0, mixed = false, -}: { turns: number; olderTurns?: number; mixed?: boolean }) { + holdFirstPage = false, +}: { turns: number; olderTurns?: number; mixed?: boolean; holdFirstPage?: boolean }) { const [from, setFrom] = useState(0); useEffect(() => { historyLoads.length = 0; + releaseHistoryPage = undefined; }, []); return ( -olderTurns, onLoadEarlierHistory: async () => { historyLoads.push(firstResidentTurnId() ?? '(none)'); + // A pending remote page lets the story measure the reading anchor + // before delivering the same production history update. + if (holdFirstPage && historyLoads.length === 1) { + await new Promise((resolve) => { releaseHistoryPage = resolve; }); + } setFrom((current) => Math.max(-olderTurns, current - HISTORY_BATCH)); await painted(2); }, @@ -2672,21 +2681,27 @@ export const DockAffordanceReturnsToTail: Story = { }, }; +// Real path: scroll a tool's nested output near the beginning of loaded conversation history. export const NestedScrollerNearHistoryBoundaryAsksForNothing: Story = { render: () => , play: async () => { await tailSettled(); const nested = injectNestedScroller(messageList()); - scrollAsReader(tailScroller(), 0); + const root = tailScroller(); + scrollAsReader(root, root.clientHeight * 2); + root.dispatchEvent(new Event('scrollend')); + await painted(2); + root.scrollTop = 0; + root.dispatchEvent(new Event('scroll')); await painted(6); wheelUp(nested); - wheelUp(tailScroller()); await painted(6); - // Scrolling never loads history; only the explicit control does. expect(historyLoads).toEqual([]); expect(nested.scrollTop).toBe(600); + wheelUp(root); + await waitFor(() => expect(historyLoads).toHaveLength(1)); }, }; @@ -3019,8 +3034,9 @@ export const UpwardTraversalHoldsTurnGeometry: Story = { }, }; +// Real path: open a long conversation, scroll upward, and receive an older history page. export const HistoryAtTheTopStillLandsAboveTheReader: Story = { - render: () => , + render: () => , play: async () => { const root = tailScroller(); // Measure history publication against rendered content, not the cold @@ -3034,8 +3050,8 @@ export const HistoryAtTheTopStillLandsAboveTheReader: Story = { expect(settled.scrollTop, JSON.stringify(settled)).toBeGreaterThan(0); expect(settled.distance, JSON.stringify(settled)).toBeLessThanOrEqual(4); }); - // The one position where the browser declines to anchor, and the one the - // load-earlier control sits at. Rows mounted by the scroll still carry the + expect(historyLoads).toEqual([]); + // The one position where the browser declines to anchor. Rows mounted by the scroll still carry the // cold Markdown layout — measure the anchor only after they settle, the // same rule as the tail state above. scrollAsReader(root, 0); @@ -3043,13 +3059,16 @@ export const HistoryAtTheTopStillLandsAboveTheReader: Story = { await painted(4); const before = firstResidentTurnId(); const reading = anchorInView(); - within(document.body).getByRole('button', { name: '载入更早的记录' }).click(); + await waitFor(() => expect(releaseHistoryPage).toBeDefined()); + releaseHistoryPage!(); await waitFor(() => expect(firstResidentTurnId()).not.toBe(before)); await waitFor(() => expect(document.querySelector('.maka-markdown-pending')).toBeNull()); await painted(6); - expect(historyLoads).toEqual([before]); + // The request can precede the virtualizer mounting the rows at the new + // scroll offset; its previous first DOM row is not the history boundary. + expect(historyLoads).toHaveLength(1); expect(Math.abs(turnTop(reading.turnId) - reading.top)).toBeLessThanOrEqual(1); expect(within(document.body).queryByRole('button', { name: '载入更早的记录' })).toBeNull(); }, @@ -3098,11 +3117,12 @@ async function measureEveryRow(): Promise { * batch along: the document's extent goes wrong, and rows the reader has * already been past push them when they come back. */ +// Real path: scroll into older history, then revisit already measured Turns in both directions. export const PrependedHistoryKeepsMeasuredHeights: Story = { // Shallower than the other history stories: this one walks the whole // transcript twice, and a slid measurement shows on the first row past the // prepend as well as on the hundredth. - render: () => , + render: () => , play: async () => { const root = tailScroller(); await document.fonts.ready; @@ -3114,7 +3134,8 @@ export const PrependedHistoryKeepsMeasuredHeights: Story = { await measureEveryRow(); const before = firstResidentTurnId(); - within(document.body).getByRole('button', { name: '载入更早的记录' }).click(); + await waitFor(() => expect(releaseHistoryPage).toBeDefined()); + releaseHistoryPage!(); await waitFor(() => expect(firstResidentTurnId()).not.toBe(before)); await waitFor(() => expect(document.querySelector('.maka-markdown-pending')).toBeNull()); await painted(6); diff --git a/apps/desktop/stories/workhub.stories.tsx b/apps/desktop/stories/workhub.stories.tsx index b42fb388478..dad36f95f80 100644 --- a/apps/desktop/stories/workhub.stories.tsx +++ b/apps/desktop/stories/workhub.stories.tsx @@ -593,13 +593,19 @@ function PagedWorkConversation() { ; } +// Real path: filter a WorkHub conversation to one Work and scroll upward into its older Turns. export const FilterWorkHistoryPages: Story = { render: () => , play: async ({ canvasElement }) => { const canvas = within(canvasElement); await waitFor(() => expect(canvas.getByText('继续补充异常场景。')).toBeInTheDocument()); expect(canvas.queryByText('请检查支付回调幂等性。')).toBeNull(); - await userEvent.click(canvas.getByRole('button', { name: '载入更早的记录' })); + expect(canvas.queryByRole('button', { name: '载入更早的记录' })).toBeNull(); + const scroller = canvasElement.querySelector('[data-chat-scroll-container]'); + expect(scroller).not.toBeNull(); + scroller!.dispatchEvent(new WheelEvent('wheel', { bubbles: true, deltaY: -120 })); + scroller!.scrollTop = 0; + scroller!.dispatchEvent(new Event('scroll')); await waitFor(() => expect(canvas.getByText('请检查支付回调幂等性。')).toBeInTheDocument()); expect(canvas.queryByText('先讨论一下整体计划。')).toBeNull(); expect(canvas.queryByRole('button', { name: '载入更早的记录' })).toBeNull(); diff --git a/packages/runtime/src/__tests__/session-manager-shell-resources.test.ts b/packages/runtime/src/__tests__/session-manager-shell-resources.test.ts new file mode 100644 index 00000000000..94279c73b70 --- /dev/null +++ b/packages/runtime/src/__tests__/session-manager-shell-resources.test.ts @@ -0,0 +1,321 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { randomUUID } from 'node:crypto'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test, { type TestContext } from 'node:test'; +import type { ShellRunRecord } from '@maka/core/shell-run'; +import type { SessionHeaderPatch } from '@maka/core/session'; +import type { RuntimeEvent } from '@maka/core/runtime-event'; +import { openInteractiveExecutionStoresForWrite } from '@maka/storage/execution-stores'; +import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '@maka/storage/root-authority'; +import { openInteractiveShellRunStoreForWrite } from '@maka/storage/shell-run-authority'; +import { BackendRegistry, SessionManager } from '../session-manager.js'; +import { ShellRunProcessManager } from '../shell-run-manager.js'; +import { shellRunContent, shellRunUpdate } from '../shell-run-tool-result.js'; +import { seedInvocation } from './invocation-fixture.js'; + +test('root resource reads preserve own states without reading transcript history', async (t) => { + const f = await fixture(t); + const session = await f.session(); + const records = ['starting', 'running', 'completed', 'cancelled'].map((status, index) => + record(session.id, `own-${index}`, { + status: status as ShellRunRecord['status'], + ...(status === 'completed' ? { completedAt: 2, exitCode: 0 } : {}), + ...(status === 'cancelled' ? { completedAt: 2, exitCode: 130 } : {}), + }), + ); + records[0]!.visibility = 'user'; + for (const item of records) await f.shellStore.createShellRun(item); + const noHistory = t.mock.method(f.manager, 'getMessages', async () => { + throw new Error('Resource query must not read transcript history'); + }); + + assert.deepEqual(await f.manager.listShellRunUpdates(session.id), records.map(shellRunUpdate)); + for (const item of records) { + assert.deepEqual( + await f.manager.getShellRunUpdate(session.id, shellRunContent(item).ref), + shellRunUpdate(item), + ); + } + assert.equal( + await f.manager.getShellRunUpdate( + session.id, + shellRunContent(record(session.id, 'missing')).ref, + ), + null, + ); + assert.deepEqual(await f.manager.listShellRunUpdates((await f.session()).id), []); + assert.equal(noHistory.mock.callCount(), 0); +}); + +test('a failed lineage read is not an empty resource answer, while own lookup stays independent', async (t) => { + const f = await fixture(t); + const session = await f.session(); + const own = record(session.id, 'own'); + await f.shellStore.createShellRun(own); + const manager = new SessionManager({ + ...f.options, + store: { + ...f.stores.sessionStore, + readHeader: async () => { + throw new Error('header unavailable'); + }, + }, + }); + assert.deepEqual( + await manager.getShellRunUpdate(session.id, shellRunContent(own).ref), + shellRunUpdate(own), + ); + await assert.rejects(manager.listShellRunUpdates(session.id), /header unavailable/); + await assert.rejects( + manager.getShellRunUpdate(session.id, shellRunContent(record(session.id, 'missing')).ref), + /header unavailable/, + ); +}); + +test('root resources do not declare staged transcript history ready', async (t) => { + const f = await fixture(t); + const session = await f.session({ transcriptLedgerVersion: 0 }); + const own = record(session.id, 'staged-own'); + await f.shellStore.createShellRun(own); + const noHistory = t.mock.method(f.manager, 'getMessages', async () => { + throw new Error('Resource query must not read transcript history'); + }); + assert.deepEqual(await f.manager.listShellRunUpdates(session.id), [shellRunUpdate(own)]); + assert.equal( + await f.manager.getShellRunUpdate( + session.id, + shellRunContent(record(session.id, 'missing')).ref, + ), + null, + ); + assert.equal(noHistory.mock.callCount(), 0); + assert.equal((await f.stores.sessionStore.readHeader(session.id)).transcriptLedgerVersion, 0); + await assert.rejects( + f.manager.ensureTranscriptLedgerForRead(session.id), + /Imported Session history is still being prepared/, + ); +}); + +for (const lineage of ['parentSessionId', 'revisionParentSessionId'] as const) { + test(`${lineage} resources retain ancestor ownership, own priority and copied-history scope`, async (t) => { + const f = await fixture(t); + const ancestor = await f.session(); + const parent = await f.session({ parentSessionId: ancestor.id }); + const decoy = await f.session(); + const child = await f.session({ + parentSessionId: lineage === 'parentSessionId' ? parent.id : decoy.id, + ...(lineage === 'revisionParentSessionId' + ? { + revisionParentSessionId: parent.id, + revisionRootSessionId: ancestor.id, + revisionOfTurnId: 'history-turn', + revisionIndex: 2, + revisionState: 'committed' as const, + } + : {}), + }); + const inherited = record(ancestor.id, 'inherited'); + const shadowed = record(ancestor.id, 'shadowed'); + const outsideCopy = record(ancestor.id, 'outside-copy'); + for (const item of [inherited, shadowed, outsideCopy]) await f.shellStore.createShellRun(item); + const own = record(child.id, 'own', { sourceToolCallId: shadowed.sourceToolCallId }); + await f.shellStore.createShellRun(own); + await f.history(child.id, [inherited, shadowed]); + // The inherited snapshot says running, while its real owner has since finished. + await f.shellStore.updateShellRun(ancestor.id, inherited.shellRunId, { + status: 'completed', + exitCode: 0, + completedAt: 2, + updatedAt: 2, + }); + const expected = { + sessionId: child.id, + ownership: { kind: 'source_owned', sourceSessionId: parent.id, ownerSessionId: ancestor.id }, + sourceTurnId: 'history-turn', + sourceToolCallId: inherited.sourceToolCallId, + result: await f.shellRuns.inspectResource(ancestor.id, shellRunContent(inherited).ref), + }; + assert.deepEqual(await f.manager.listShellRunUpdates(child.id), [ + shellRunUpdate(own), + expected, + ]); + assert.deepEqual( + await f.manager.getShellRunUpdate(child.id, shellRunContent(inherited).ref), + expected, + ); + assert.equal( + await f.manager.getShellRunUpdate(child.id, shellRunContent(outsideCopy).ref), + null, + ); + }); +} + +test('missing inherited owners and cyclic ancestry preserve source-unavailable snapshots', async (t) => { + const f = await fixture(t); + const parent = await f.session(); + const child = await f.session({ parentSessionId: parent.id }); + await f.stores.sessionStore.updateHeader(parent.id, { parentSessionId: child.id }); + const missing = record(parent.id, 'gone'); + await f.history(child.id, [missing]); + const expected = { + sessionId: child.id, + ownership: { kind: 'source_unavailable', sourceSessionId: parent.id }, + sourceTurnId: 'history-turn', + sourceToolCallId: missing.sourceToolCallId, + result: shellRunContent(missing), + }; + assert.deepEqual(await f.manager.listShellRunUpdates(child.id), [expected]); + assert.deepEqual(await f.manager.getShellRunUpdate(child.id, expected.result.ref), expected); +}); + +test('inherited discovery excludes non-Bash and terminal snapshots', async (t) => { + const f = await fixture(t); + const parent = await f.session(); + const child = await f.session({ parentSessionId: parent.id }); + const terminal = record(parent.id, 'terminal', { + status: 'completed', + completedAt: 2, + exitCode: 0, + }); + const otherTool = record(parent.id, 'other-tool'); + await f.history(child.id, [terminal, otherTool], ['Bash', 'Other']); + assert.deepEqual(await f.manager.listShellRunUpdates(child.id), []); + for (const item of [terminal, otherTool]) { + assert.equal(await f.manager.getShellRunUpdate(child.id, shellRunContent(item).ref), null); + } +}); + +async function fixture(t: TestContext) { + const base = await mkdtemp(join(tmpdir(), 'maka-shell-resource-read-')); + const capability = await resolveStorageRoot({ path: join(base, 'root'), kind: 'interactive' }); + const owner = await tryAcquireInteractiveRootOwner(capability); + assert.ok(owner); + const stores = await openInteractiveExecutionStoresForWrite(owner.lease); + const shellStore = await openInteractiveShellRunStoreForWrite(owner.lease); + const shellRuns = new ShellRunProcessManager({ + store: shellStore, + newId: randomUUID, + now: Date.now, + }); + t.after(async () => { + await shellRuns.terminateAll(); + shellStore.close(); + await stores.sessionStore.close?.(); + await owner.close(); + await rm(base, { recursive: true, force: true }); + }); + const options = { + store: stores.sessionStore, + runStore: stores.agentRunStore, + runtimeEventStore: stores.runtimeEventStore, + shellRuns, + backends: new BackendRegistry(), + newId: randomUUID, + now: Date.now, + }; + const manager = new SessionManager(options); + return { + stores, + shellStore, + shellRuns, + manager, + options, + session: async (patch: SessionHeaderPatch = {}) => { + const session = await stores.sessionStore.create({ + cwd: base, + llmConnectionId: 'fake-connection', + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + }); + return stores.sessionStore.updateHeader(session.id, { transcriptLedgerVersion: 1, ...patch }); + }, + history: async (sessionId: string, records: ShellRunRecord[], names: string[] = []) => { + const identity = await seedInvocation(stores.runtimeEventStore, { + sessionId, + runId: 'history-run', + turnId: 'history-turn', + openedAt: 1, + }); + let sequence = 1; + const append = (patch: Partial) => + stores.runtimeEventStore.appendRuntimeEvent(sessionId, identity.runId, { + ...identity, + id: `event-${++sequence}`, + ts: sequence, + partial: false, + role: 'system', + author: 'system', + ...patch, + }); + for (const [index, item] of records.entries()) { + const name = names[index] ?? 'Bash'; + await append({ + role: 'model', + author: 'agent', + content: { kind: 'function_call', id: item.sourceToolCallId, name, args: {} }, + }); + await append({ + role: 'tool', + author: 'tool', + content: { + kind: 'function_response', + id: item.sourceToolCallId, + name, + result: shellRunContent(item), + }, + }); + } + await append({ status: 'completed', actions: { endInvocation: true } }); + }, + }; +} + +function record( + sessionId: string, + shellRunId: string, + patch: Partial = {}, +): ShellRunRecord { + return { + sessionId, + shellRunId, + sourceTurnId: 'resource-turn', + sourceToolCallId: `call-${shellRunId}`, + cwd: '/tmp', + command: 'fixture', + status: 'running', + startedAt: 1, + updatedAt: 1, + revision: 1, + output: { + mode: 'pipes', + stdout: '', + stderr: '', + stdoutTruncated: false, + stderrTruncated: false, + redacted: false, + }, + ...patch, + }; +} diff --git a/packages/runtime/src/session-manager.ts b/packages/runtime/src/session-manager.ts index 40b020fc83c..a62d8c38a8b 100644 --- a/packages/runtime/src/session-manager.ts +++ b/packages/runtime/src/session-manager.ts @@ -1382,6 +1382,12 @@ export class SessionManager { if (!shellRuns) return []; const ownUpdates = await shellRuns.listSessionUpdates(sessionId); + // Only a branched or revised Session can inherit resources from its history. + // Ordinary Sessions already have their complete answer in the ShellRun store. + const inheritedFrom = await this.deps.store.readHeader(sessionId); + const parentSessionId = inheritedFrom.revisionParentSessionId ?? inheritedFrom.parentSessionId; + if (!parentSessionId) return ownUpdates; + const ownToolCalls = new Set(ownUpdates.map((update) => update.sourceToolCallId)); const messages = await this.readShellRunProjectionMessages(sessionId); if (!messages) return ownUpdates; @@ -1414,9 +1420,6 @@ export class SessionManager { } if (inherited.size === 0) return ownUpdates; - const inheritedFrom = await this.deps.store.readHeader(sessionId); - const parentSessionId = inheritedFrom.revisionParentSessionId ?? inheritedFrom.parentSessionId; - if (!parentSessionId) return ownUpdates; const inheritedUpdates = await Promise.all( [...inherited.values()].map(async (candidate) => { const owner = await this.resolveShellRunOwner(parentSessionId, candidate.ref); @@ -1444,6 +1447,10 @@ export class SessionManager { const own = await shellRuns.getSessionUpdate(sessionId, ref); if (own) return own; + const inheritedFrom = await this.deps.store.readHeader(sessionId); + const parentSessionId = inheritedFrom.revisionParentSessionId ?? inheritedFrom.parentSessionId; + if (!parentSessionId) return null; + const messages = await this.readShellRunProjectionMessages(sessionId); if (!messages) return null; const bashToolCalls = shellRunBashToolCallIds(messages); @@ -1468,9 +1475,6 @@ export class SessionManager { } if (!candidate) return null; - const inheritedFrom = await this.deps.store.readHeader(sessionId); - const parentSessionId = inheritedFrom.revisionParentSessionId ?? inheritedFrom.parentSessionId; - if (!parentSessionId) return null; const owner = await this.resolveShellRunOwner(parentSessionId, ref); return { sessionId, diff --git a/packages/ui/src/__tests__/chat-view-load-earlier.test.tsx b/packages/ui/src/__tests__/chat-view-load-earlier.test.tsx index f55b86ad876..9d7cbf2dd67 100644 --- a/packages/ui/src/__tests__/chat-view-load-earlier.test.tsx +++ b/packages/ui/src/__tests__/chat-view-load-earlier.test.tsx @@ -110,18 +110,15 @@ function harness() { }, readerScrollTo(scroller: HTMLElement, top: number): void { const wheel = new current.window.Event('wheel', { bubbles: true }); - Object.defineProperty(wheel, 'deltaY', { value: top < scroller.scrollTop ? -120 : 120 }); + Object.defineProperty(wheel, 'deltaY', { value: top <= scroller.scrollTop ? -120 : 120 }); scroller.dispatchEvent(wheel); scroller.scrollTop = top; scroller.dispatchEvent(new current.window.Event('scroll')); }, - click(element: Element): void { - element.dispatchEvent(new current.window.Event('click', { bubbles: true })); - }, }; } -test('the load-earlier button exists only while earlier history does, and waits for its load', async () => { +test('upward reading near the beginning loads one range, without a button or mount-time prefetch', async () => { const view = harness(); await view.render({ messages: turnMessages(4, 8), onLoadEarlierHistory: () => {} }); assert.equal(view.loadButton(), null, 'no button without earlier history'); @@ -132,27 +129,74 @@ test('the load-earlier button exists only while earlier history does, and waits loads += 1; return new Promise((resolve) => { finish = resolve; }); }; - await view.render({ messages: turnMessages(4, 8), hasEarlierHistory: true, onLoadEarlierHistory }); - const button = view.loadButton(); - assert.ok(button); - await act(async () => { view.click(button); }); + const scroller = await view.render({ messages: turnMessages(4, 8), hasEarlierHistory: true, onLoadEarlierHistory }); + assert.equal(view.loadButton(), null); + assert.equal(loads, 0, 'opening a conversation must not read all older history'); + scroller.scrollTop = 1000; + await act(async () => { view.readerScrollTo(scroller, 800); }); + assert.equal(loads, 0, 'reading far from the boundary does not load'); + await act(async () => { view.readerScrollTo(scroller, 400); }); assert.equal(loads, 1); - assert.equal(view.loadButton()?.disabled, true, 'a pending load cannot be requested again'); + await act(async () => { view.readerScrollTo(scroller, 200); }); + assert.equal(loads, 1, 'a pending load cannot be requested again'); await act(async () => { finish(); }); - assert.equal(view.loadButton()?.disabled, false); + assert.equal(loads, 1, 'settling a page does not recursively prefetch the rest'); + await act(async () => { view.readerScrollTo(scroller, 300); }); + assert.equal(loads, 1, 'downward reading does not load older history'); + await act(async () => { view.readerScrollTo(scroller, 100); }); + assert.equal(loads, 2); + await act(async () => { finish(); }); }); /** * A visible transcript with nothing in it is exactly when the reader most needs - * the control: WorkHub filters the list to one Work, and a Work whose Turns are - * all in unloaded history filters it down to nothing. The control is the only - * way to load those Turns, so it cannot be inside the non-empty branch. + * an upward gesture: WorkHub may filter every loaded Turn out of the view. + * Even when there is no scroll distance, the reader can request older rows. */ -test('offers to load earlier history even with nothing to show', async () => { +test('upward input reaches earlier history even with nothing to show', async () => { + const view = harness(); + let loads = 0; + const scroller = await view.render({ messages: [], hasEarlierHistory: true, onLoadEarlierHistory: () => { loads++; } }); + assert.equal(loads, 0); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(loads, 1); + assert.equal(view.loadButton(), null); +}); + +test('a previous conversation finishing does not unlock another conversation pending page', async () => { const view = harness(); - await view.render({ messages: [], hasEarlierHistory: true, onLoadEarlierHistory: () => {} }); - assert.ok(view.loadButton(), 'an empty message list hid the only way to load the rest'); + const finishes: Array<() => void> = []; + const onLoadEarlierHistory = () => new Promise((resolve) => { finishes.push(resolve); }); + const props = { messages: turnMessages(4, 8), hasEarlierHistory: true, onLoadEarlierHistory }; + const scroller = await view.render(props); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(finishes.length, 1); + await view.render({ ...props, activeSession: { ...activeSession, id: 'other-session' } }); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(finishes.length, 2); + await act(async () => { finishes[0]!(); }); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(finishes.length, 2, 'the old completion must not release the new pending request'); + await act(async () => { finishes[1]!(); }); + await view.render({ ...props, activeSession: { ...activeSession, id: 'other-session' }, hasEarlierHistory: false }); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(finishes.length, 2, 'exhausted history does not load again'); +}); + +test('a failed automatic read retries only after another upward gesture', async () => { + const view = harness(); + let loads = 0; + const scroller = await view.render({ + messages: turnMessages(4, 8), hasEarlierHistory: true, + onLoadEarlierHistory: async () => { loads++; throw new Error('disconnected'); }, + }); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(loads, 1); + await act(async () => {}); + assert.equal(loads, 1, 'failure must not start a retry loop'); + await act(async () => { view.readerScrollTo(scroller, 0); }); + assert.equal(loads, 2); }); test('prepended Turns keep a released reader on their Turn without re-pinning', async () => { @@ -163,10 +207,7 @@ test('prepended Turns keep a released reader on their Turn without re-pinning', assert.equal(view.authority.getSnapshot().pinned, false); assert.equal(view.anchors.at(-1), 'turn-5'); - // The reader asks for history, which is the only way it arrives. - const button = view.loadButton(); - assert.ok(button); - await act(async () => { view.click(button); }); + // The upward gesture requested history; its eventual arrival is not input. await view.render({ messages: turnMessages(2, 8), hasEarlierHistory: false, onLoadEarlierHistory: () => {} }); await act(() => { scroller.dispatchEvent(new dom!.window.Event('scroll')); }); assert.equal(view.loadButton(), null); @@ -185,10 +226,12 @@ test('a selection from outside every Turn keeps every Turn it spans mounted', as const rows = mounted(); assert.ok(rows.length < 20, 'the fixture mounts only some of the Turns'); - const button = view.loadButton(); + const selectionStart = dom!.document.createElement('p'); + selectionStart.textContent = 'Text before the transcript'; + scroller.prepend(selectionStart); const last = dom!.container.querySelector(`[data-transcript-turn-id="${rows.at(-1)}"]`); - assert.ok(button && last); - const selection = { isCollapsed: false, anchorNode: button, focusNode: last }; + assert.ok(last); + const selection = { isCollapsed: false, anchorNode: selectionStart, focusNode: last }; dom!.document.getSelection = () => selection as unknown as Selection; await act(async () => { dom!.document.dispatchEvent(new dom!.window.Event('selectionchange')); }); await act(() => { view.readerScrollTo(scroller, 16 * TURN_HEIGHT); }); diff --git a/packages/ui/src/chat-view.tsx b/packages/ui/src/chat-view.tsx index 8a093726f7f..89cee68e606 100644 --- a/packages/ui/src/chat-view.tsx +++ b/packages/ui/src/chat-view.tsx @@ -551,6 +551,23 @@ export function ChatView(props: { turns, ); const { startMargin, listRef, measureStartMargin } = useTranscriptStartMargin(scrollRef); + const earlierReader = useRef({ sessionId: props.activeSession?.id, pending: false }); + if (earlierReader.current.sessionId !== props.activeSession?.id) { + earlierReader.current = { sessionId: props.activeSession?.id, pending: false }; + } + const loadEarlierHistory = (): boolean => { + if (!props.hasEarlierHistory || !props.onLoadEarlierHistory) return false; + const reader = earlierReader.current; + if (reader.pending) return true; + reader.pending = true; + const settled = () => { reader.pending = false; }; + try { + void Promise.resolve(props.onLoadEarlierHistory()).then(settled, settled); + } catch { + settled(); + } + return true; + }; const { highlightedTurnId, placed, commandTurnId, revealTurnAtStart, measurement } = useChatScroll({ scrollRef, measureStartMargin, @@ -561,6 +578,7 @@ export function ChatView(props: { restoreTarget: props.restoreTargetTurn, viewportNavigation: props.viewportNavigation, onReadingAnchorChange: props.onReadingAnchorChange, + onReadEarlier: loadEarlierHistory, behavior: props.scrollBehavior, }); const onLoadTranscriptTurnRef = useRef(props.onLoadTranscriptTurn); @@ -589,16 +607,6 @@ export function ChatView(props: { keepMountedIndexes.add(index); } } - const [loadingEarlierHistory, setLoadingEarlierHistory] = useState(false); - const loadEarlierHistory = (): void => { - const pending = props.onLoadEarlierHistory?.(); - if (!pending) return; - setLoadingEarlierHistory(true); - void pending.then( - () => setLoadingEarlierHistory(false), - () => setLoadingEarlierHistory(false), - ); - }; const { quote: selectionQuote, clear: clearSelectionQuote } = useMessageSelectionQuote( scrollRef, Boolean(props.onQuoteSelection || props.onAskAboutSelection), @@ -728,27 +736,6 @@ export function ChatView(props: { : props.emptyOverride ?? ( ); - /** - * Nothing to show is exactly when this matters most: WorkHub filters the - * transcript to one Work, and a Work whose Turns are all still in unloaded - * history filters it down to nothing. So the control rides along with the - * empty state rather than sitting in the branch that replaces it — which - * also keeps the list's only child a single `null`, the shape that lets - * `ChatMessageList` render an empty state at all. - */ - const loadEarlierHistoryControl = props.hasEarlierHistory && props.onLoadEarlierHistory ? ( - -