perf(session-controller): publish historical jump pages as one batch

This commit is contained in:
imccyu committed 2026-09-18 14:13:47 +08:00
1 parent e22b0d75e8
commit 3d360e7fdb
6 files changed
+125 -8

No files matched your search

@@ -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
@@ -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.
<a id="client-references"></a>
## Client references
@@ -44,6 +44,8 @@ Session 对象还承载本地提交回显:`session.beginSubmission` 在调用
恢复会话时若已有写句柄占用,返回 `session/writer-held`,并携带会话 id;其他恢复失败仍返回 `gateway/internal`。
`loadThrough(seq)` 在共享目标被覆盖或加载结束前私下保留较早页面,随后把成功取得的页面按顺序作为一次前插发布。历史加载期间实时事件仍然可见。后续页面失败时保留已成功取得的部分;历史窗口被替换时丢弃被替换窗口的暂存页面。普通 `loadOlder()` 仍直接发布其单页结果。
<a id="client-references"></a>
## Client 引用
@@ -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.
*/
@@ -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<void> | 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)
@@ -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<undefined>()
const lastPage = Promise.withResolvers<RemoteResult<SessionPage>>()
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<undefined>()
const lastPage = Promise.withResolvers<RemoteResult<SessionPage>>()
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')