diff --git a/packages/api/session-controller/README.i18n.yaml b/packages/api/session-controller/README.i18n.yaml index 544c5f0228..01a5e7002a 100644 --- a/packages/api/session-controller/README.i18n.yaml +++ b/packages/api/session-controller/README.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write packages/api/session-controller/README.md -README.md: f67355be0cb9737ebc67bf589528878bccc26e6d -README.zh.md: e52501c145c3789213aec517b0801e5e7ee2a4cb +README.md: fe6ed9200d520dee41945b1fc946a5a44472bdc8 +README.zh.md: 0c2fc764eb20fcdf37c84f7d3b37ca8543e86b7d diff --git a/packages/api/session-controller/README.md b/packages/api/session-controller/README.md index f67355be0c..fe6ed9200d 100644 --- a/packages/api/session-controller/README.md +++ b/packages/api/session-controller/README.md @@ -44,6 +44,8 @@ Fork copies history through the selected completed turn, including its `turn/end A resume blocked by an existing write handle returns `session/writer-held` with the Session id; other resume failures retain `gateway/internal`. +`loadThrough(seq)` retains older pages privately until its shared target is covered or loading ends, then publishes the successful pages as one ordered prepend. Live events remain visible while history loads. A later page failure retains the successful prefix; replacing the history window discards buffered pages from the replaced window. Ordinary `loadOlder()` still publishes its single page directly. + ## Client references diff --git a/packages/api/session-controller/README.zh.md b/packages/api/session-controller/README.zh.md index e52501c145..0c2fc764eb 100644 --- a/packages/api/session-controller/README.zh.md +++ b/packages/api/session-controller/README.zh.md @@ -44,6 +44,8 @@ Session 对象还承载本地提交回显:`session.beginSubmission` 在调用 恢复会话时若已有写句柄占用,返回 `session/writer-held`,并携带会话 id;其他恢复失败仍返回 `gateway/internal`。 +`loadThrough(seq)` 在共享目标被覆盖或加载结束前私下保留较早页面,随后把成功取得的页面按顺序作为一次前插发布。历史加载期间实时事件仍然可见。后续页面失败时保留已成功取得的部分;历史窗口被替换时丢弃被替换窗口的暂存页面。普通 `loadOlder()` 仍直接发布其单页结果。 + ## Client 引用 diff --git a/packages/api/session-controller/src/client/contract/session.ts b/packages/api/session-controller/src/client/contract/session.ts index 20c2985c80..246d2c8a92 100644 --- a/packages/api/session-controller/src/client/contract/session.ts +++ b/packages/api/session-controller/src/client/contract/session.ts @@ -127,6 +127,8 @@ export interface ISession { * turn-jump loader. Repeated calls while a jump is paging lower its shared * target and return the in-flight completion; `snapshot.loadingOlder` is * the busy signal for the whole jump. + * Older pages publish together at completion, including successful pages + * before a later failure; live events remain independently visible. * @param seq - durable event seq the window must reach (a turn's `turn/start` seq). * @returns completion once covered, exhausted, superseded, or failed soft. */ diff --git a/packages/api/session-controller/src/client/sessions/session.ts b/packages/api/session-controller/src/client/sessions/session.ts index 3c401cc3e7..ad5b426a40 100644 --- a/packages/api/session-controller/src/client/sessions/session.ts +++ b/packages/api/session-controller/src/client/sessions/session.ts @@ -53,6 +53,12 @@ export const PAGE_MESSAGES = 50 /** Messages requested per page while a turn jump loops backwards (fewer, larger round trips). */ export const JUMP_PAGE_MESSAGES = 200 +interface PendingHistory { + beforeSeq: SessionLogOffset + hasMore: boolean + readonly pages: (readonly SessionEventLikeEntry[])[] +} + /** Manager-owned observers of a Session object's local state edges. */ export interface SessionOptions { /** Catalog-discovered address selecting non-activating subagent transport. */ @@ -97,6 +103,7 @@ export class Session implements SessionFace { private jumpTargetSeq: SessionSeq | null = null /** The running jump loop's completion, shared by retargeting callers. */ private jumpPromise: Promise | null = null + private pendingHistory: PendingHistory | null = null private readonly stopObservingInbox: () => void private readonly assistantStream = new ClientAssistantStream() private running = false @@ -419,6 +426,14 @@ export class Session implements SessionFace { // target behind — only the loop's finally clears that field, and no // loop starts here. if (this.loadingOlder) return Promise.resolve() + const events = this.events + if (events === undefined) return Promise.resolve() + const pending: PendingHistory = { + beforeSeq: this.baseSeq, + hasMore: this.hasMore, + pages: [], + } + this.pendingHistory = pending this.jumpTargetSeq = seq this.loadingOlder = true this.notifier.markDirty() @@ -428,15 +443,13 @@ export class Session implements SessionFace { const generation = this.openGeneration this.jumpPromise = (async () => { try { - while (this.hasMore && this.jumpTargetSeq !== null && this.baseSeq > this.jumpTargetSeq) { + while (pending.hasMore && this.jumpTargetSeq !== null && pending.beforeSeq > this.jumpTargetSeq) { if (generation !== this.openGeneration) return - const events = this.events - if (events === undefined) return - const before = this.baseSeq - await events.prepend({ beforeSeq: this.baseSeq, maxMessages: JUMP_PAGE_MESSAGES }) + const before = pending.beforeSeq + await events.prepend({ beforeSeq: before, maxMessages: JUMP_PAGE_MESSAGES }) // No-progress guard: an empty or dropped page that still claims more // history must end the loop, not spin it. - if (this.baseSeq >= before) return + if (pending.beforeSeq >= before) return } } catch (error) { if (!isRemoteFailure(error)) { @@ -445,7 +458,11 @@ export class Session implements SessionFace { } finally { this.jumpTargetSeq = null this.jumpPromise = null + this.pendingHistory = null this.loadingOlder = false + if (generation === this.openGeneration && pending.pages.length > 0) { + this.prependWindow(pending.pages.reverse().flat(), pending.hasMore) + } this.notifier.markDirty() } })() @@ -648,6 +665,11 @@ export class Session implements SessionFace { const visible = this.assistantStream.replace(entries, assistantStream) this.baseSeq = SessionLogOffset(entries[0]?.event.seq ?? 0) this.hasMore = hasMore + if (this.pendingHistory !== null) { + this.pendingHistory.beforeSeq = this.baseSeq + this.pendingHistory.hasMore = hasMore + this.pendingHistory.pages.length = 0 + } if (visible.some(entry => entry.event.type === 'turn/start')) this.firstPromptPendingTurn = false if (projections !== undefined) this.projections.seed(projections) this.eventSource.replace(visible, hasMore) @@ -684,6 +706,13 @@ export class Session implements SessionFace { /** Prepend one stream-validated history page. */ private prependWindow(entries: readonly SessionEventLikeEntry[], hasMore: boolean): void { + if (this.pendingHistory !== null) { + const pending = this.pendingHistory + pending.beforeSeq = entries[0] === undefined ? pending.beforeSeq : SessionLogOffset(entries[0].event.seq) + pending.hasMore = hasMore + pending.pages.push(entries) + return + } this.baseSeq = entries[0] === undefined ? this.baseSeq : SessionLogOffset(entries[0].event.seq) this.hasMore = hasMore this.eventSource.prepend(entries, hasMore) diff --git a/packages/api/session-controller/tests/session.client.spec.ts b/packages/api/session-controller/tests/session.client.spec.ts index 42ee4b94ff..0372e005a5 100644 --- a/packages/api/session-controller/tests/session.client.spec.ts +++ b/packages/api/session-controller/tests/session.client.spec.ts @@ -262,6 +262,88 @@ describe('paging', () => { expect(mock.log.requests()).toHaveLength(sent) }) + it('publishes jump pages together while live events remain visible', async ({ mock, start }) => { + const oldest = plainTurn(SessionSeq(0), 0, 'old', 'answer') + const middle = plainTurn(SessionSeq(6), 1, 'middle', 'answer') + const newest = plainTurn(SessionSeq(12), 2, 'new', 'answer') + const session = await sessionBench(mock, start, SID) + mock.stream(FOLLOW, followScript(history(newest, true))) + await session.open() + const secondRequested = Promise.withResolvers() + const lastPage = Promise.withResolvers>() + mock.remote.session.page.mockImplementation(pageRule((request) => { + if (request.beforeSeq === 12) return history(middle, true) + secondRequested.resolve(undefined) + return lastPage.promise + })) + const changes: string[] = [] + onTestFinished(session.eventSource.subscribe(() => { changes.push(session.eventSource.getSnapshot().change.kind) })) + const jump = session.loadThrough(SessionSeq(0)) + await secondRequested.promise + expect(changes).toEqual([]) + expect(eventSeqs(session)).toEqual(newest.map(event => event.seq)) + expect(session.getSnapshot().loadingOlder).toBe(true) + + await pushEvent(mock, ev.turnStart(SessionSeq(18), 3)) + expect(changes).toEqual(['append']) + expect(eventSeqs(session)).toEqual([...newest.map(event => event.seq), 18]) + const nearerJump = session.loadThrough(SessionSeq(6)) + expect(nearerJump).toBe(jump) + + lastPage.resolve(history(oldest)) + await jump + expect(changes).toEqual(['append', 'prepend']) + expect(eventSeqs(session)).toEqual(Array.from({ length: 19 }, (_, index) => index)) + expect(session.eventSource.getSnapshot().change).toMatchObject({ + kind: 'prepend', entries: entries([...oldest, ...middle]), + }) + expect(session.getSnapshot()).toMatchObject({ loadingOlder: false, hasMore: false }) + }) + + it('publishes the successful jump prefix once when a later page fails', async ({ mock, start }) => { + const middle = plainTurn(SessionSeq(6), 1, 'middle', 'answer') + const newest = plainTurn(SessionSeq(12), 2, 'new', 'answer') + const session = await sessionBench(mock, start, SID) + mock.stream(FOLLOW, followScript(history(newest, true))) + await session.open() + const changed = vi.fn() + onTestFinished(session.eventSource.subscribe(changed)) + mock.remote.session.page.mockImplementation(pageRule(request => request.beforeSeq === 12 + ? history(middle, true) + : err(new RemoteError('gateway/internal', 'page unavailable', {})))) + + await session.loadThrough(SessionSeq(0)) + + expect(changed).toHaveBeenCalledOnce() + expect(eventSeqs(session)).toEqual([...middle, ...newest].map(event => event.seq)) + expect(session.getSnapshot()).toMatchObject({ loadingOlder: false, hasMore: true }) + }) + + it('discards buffered jump pages when the Session replaces its stream', async ({ mock, start }) => { + const newest = plainTurn(SessionSeq(12), 2, 'new', 'answer') + const session = await sessionBench(mock, start, SID) + mock.stream(FOLLOW, followScript(history(newest, true))) + await session.open() + const secondRequested = Promise.withResolvers() + const lastPage = Promise.withResolvers>() + mock.remote.session.page.mockImplementation(pageRule((request) => { + if (request.beforeSeq === 12) return history(plainTurn(SessionSeq(6), 1, 'middle', 'answer'), true) + secondRequested.resolve(undefined) + return lastPage.promise + })) + const changes: string[] = [] + onTestFinished(session.eventSource.subscribe(() => { changes.push(session.eventSource.getSnapshot().change.kind) })) + const jump = session.loadThrough(SessionSeq(0)) + await secondRequested.promise + const replacement = session.resync() + lastPage.resolve(history(plainTurn(SessionSeq(0), 0, 'old', 'answer'))) + await Promise.all([jump, replacement]) + + expect(changes).toEqual(['replace']) + expect(eventSeqs(session)).toEqual(newest.map(event => event.seq)) + expect(session.getSnapshot().loadingOlder).toBe(false) + }) + it('loadThrough retargets a running jump to the lowest requested seq and shares its completion', async ({ mock, start }) => { const oldest = plainTurn(SessionSeq(0), 0, 'a', 'b') const middle = plainTurn(SessionSeq(6), 1, 'c', 'd')