diff --git a/packages/api/gateway/src/client/journal-stream.ts b/packages/api/gateway/src/client/journal-stream.ts index f21e79f241..306779ba94 100644 --- a/packages/api/gateway/src/client/journal-stream.ts +++ b/packages/api/gateway/src/client/journal-stream.ts @@ -7,9 +7,9 @@ import type { RemoteStreamOptions, } from './remote-stream.ts' -/** Transport-neutral opening cursor or journal entry. */ -export type RemoteJournalFrame = - | { readonly type: 'opened'; readonly cursor: Cursor } +/** Transport-neutral opening snapshot or journal entry. */ +export type RemoteJournalFrame = + | { readonly type: 'opened'; readonly cursor: Cursor; readonly page: Page } | { readonly type: 'entry'; readonly entry: Entry } /** One committed journal-window update. */ @@ -28,7 +28,7 @@ export type RemoteJournalChange = } | { readonly type: 'append'; readonly entry: Entry } -type JournalStreamItem = RemoteStreamItem> +type JournalStreamItem = RemoteStreamItem> /** Gateway capability used to create one reconnecting Remote stream. */ export interface RemoteStreamFactory { @@ -65,13 +65,13 @@ export interface RemoteJournalStreamOptions { } /** - * Owns follow-before-page opening, ordered live delivery, pagination, and repair. + * Owns snapshot-first opening, ordered live delivery, pagination, and repair. * * The domain retains its published window during reconnection. A replacement is - * published only after a tail page reaches the generation's opening cursor. + * published only after the opening page reaches the generation's cursor. */ export abstract class RemoteJournalStream { - private readonly stream: RemoteStream> + private readonly stream: RemoteStream> private initialRequest!: PageRequest private resumeCursor: Cursor | undefined private hasResumeCursor = false @@ -83,7 +83,7 @@ export abstract class RemoteJournalStream | undefined private closing: Promise | undefined - private pendingNext: Promise>> | undefined + private pendingNext: Promise>> | undefined /** * @param remote - Gateway factory for the reconnecting physical-generation stream. @@ -93,12 +93,9 @@ export abstract class RemoteJournalStream, ) { - this.stream = remote.$stream>({ + this.stream = remote.$stream>({ name: options.name, - open: signal => this.follow( - this.hasResumeCursor ? this.resumeCursor : undefined, - signal, - ), + open: signal => this.follow(this.initialRequest, signal), ended: accepted => accepted ? new RemoteStreamCarrierError(`${options.name} ended without a terminal result`) : new Error( @@ -111,15 +108,15 @@ export abstract class RemoteJournalStream> + ): AsyncIterable> /** * Read one journal page through the addressed domain source. @@ -143,7 +140,7 @@ export abstract class RemoteJournalStream>, + iterator: AsyncIterator>, ): Promise { try { while (true) { @@ -225,7 +222,7 @@ export abstract class RemoteJournalStream, - iterator: AsyncIterator>, + private replaceGeneration( + initial: JournalStreamItem, resumed: boolean, - ): Promise { - let item = initial - let isResumed = resumed - while (true) { - const cursor = this.opening(item, isResumed) - this.setResumeCursor(cursor) - const superseded = await this.replaceThrough( - request, - cursor, - item.generation, - item.signal, - iterator, - [], - ) - if (superseded === undefined) return - item = superseded - isResumed = true - } + ): void { + const opening = this.opening(initial, resumed) + this.replaceFromOpening(opening.page, opening.cursor) } private opening( - item: RemoteStreamItem>, + item: RemoteStreamItem>, resumed: boolean, - ): Cursor { + ): { readonly cursor: Cursor; readonly page: Page } { if (item.value.type !== 'opened') { throw new Error(`${resumed ? 'resumed ' : ''}${this.options.name} emitted an entry before its opening cursor`) } @@ -279,13 +259,30 @@ export abstract class RemoteJournalStream, - iterator: AsyncIterator>, + item: JournalStreamItem, + iterator: AsyncIterator>, ): Promise { const cursor = this.options.cursor(entry) const last = this.lastCursor as Cursor @@ -301,7 +298,7 @@ export abstract class RemoteJournalStream>, + iterator: AsyncIterator>, queued: Entry[], - ): Promise | undefined> { + ): Promise | undefined> { let read = await this.readPageWhileFollowing( request, requiredCursor, @@ -351,6 +348,7 @@ export abstract class RemoteJournalStream>, + iterator: AsyncIterator>, queued: Entry[], ): Promise< | { readonly type: 'page'; readonly page: Page } - | { readonly type: 'superseded'; readonly item: JournalStreamItem } + | { readonly type: 'superseded'; readonly item: JournalStreamItem } > { const page = this.readPage(request, through, signal).then( value => ({ type: 'page' as const, value }), @@ -410,12 +408,12 @@ export abstract class RemoteJournalStream>, - initial: Promise>>, - ): Promise<{ readonly type: 'superseded'; readonly item: JournalStreamItem }> { + iterator: AsyncIterator>, + initial: Promise>>, + ): Promise<{ readonly type: 'superseded'; readonly item: JournalStreamItem }> { let pending = initial while (true) { - let next: IteratorResult> + let next: IteratorResult> try { next = await pending } finally { @@ -461,15 +459,15 @@ export abstract class RemoteJournalStream>, - ): Promise>> { + iterator: AsyncIterator>, + ): Promise>> { this.pendingNext ??= iterator.next() return this.pendingNext } private async takeNext( - iterator: AsyncIterator>, - ): Promise>> { + iterator: AsyncIterator>, + ): Promise>> { const pending = this.nextResult(iterator) try { return await pending diff --git a/packages/api/gateway/tests/journal-stream.client.spec.ts b/packages/api/gateway/tests/journal-stream.client.spec.ts index 38362c5598..cc364bbf51 100644 --- a/packages/api/gateway/tests/journal-stream.client.spec.ts +++ b/packages/api/gateway/tests/journal-stream.client.spec.ts @@ -25,9 +25,12 @@ interface PageRequest { readonly limit?: number } +type JournalFrame = RemoteJournalFrame +type ScriptedFrame = JournalFrame + interface Generation { readonly frames: readonly ( - RemoteJournalFrame | Promise> + ScriptedFrame | Promise )[] readonly terminal?: Error readonly hold?: boolean @@ -67,7 +70,7 @@ class FixtureJournal extends RemoteJournalStream[], failed: (error: unknown) => void, factory: RemoteStreamFactory = STREAM_FACTORY, @@ -87,11 +90,11 @@ class FixtureJournal extends RemoteJournalStream> { + ): AsyncIterable { this.calls.push('follow') - this.followCursors.push(after) + this.followRequests.push(request) const generation = this.generations.shift() if (generation === undefined) throw new Error('no scripted journal generation') for (const [index, frame] of generation.frames.entries()) { @@ -138,12 +141,12 @@ function journalFixture( readonly calls: string[] readonly pageRequests: PageRequest[] readonly pageCursors: number[] - readonly followCursors: (number | undefined)[] + readonly followRequests: PageRequest[] } { const calls: string[] = [] const pageRequests: PageRequest[] = [] const pageCursors: number[] = [] - const followCursors: (number | undefined)[] = [] + const followRequests: PageRequest[] = [] const changes: RemoteJournalChange[] = [] const failed = vi.fn() const journal = new FixtureJournal( @@ -152,24 +155,28 @@ function journalFixture( calls, pageRequests, pageCursors, - followCursors, + followRequests, changes, failed, factory, ) - return { journal, changes, failed, calls, pageRequests, pageCursors, followCursors } + return { journal, changes, failed, calls, pageRequests, pageCursors, followRequests } +} + +function opened(cursor: number, value: Page): JournalFrame { + return { type: 'opened', cursor, page: value } } function remoteItem( generation: number, - value: RemoteJournalFrame, + value: ScriptedFrame, signal: AbortSignal, -): RemoteStreamItem> { +): RemoteStreamItem { return { generation, value, signal, accept: vi.fn() } } function controlledFactory( - next: () => Promise>>>, + next: () => Promise>>, ): RemoteStreamFactory { const lifetime = new AbortController() return { @@ -189,17 +196,17 @@ function controlledFactory( } describe('RemoteJournalStream', () => { - it('opens follow before page, removes overlap, appends live entries, and prepends history', async () => { + it('opens from the follow snapshot, removes overlap, appends live entries, and prepends history', async () => { const fixture = journalFixture( [{ frames: [ - { type: 'opened', cursor: 3 }, + opened(3, page('tail', [2, 3], true)), { type: 'entry', entry: { seq: 3 } }, { type: 'entry', entry: { seq: 4 } }, ], hold: true, }], - [page('tail', [2, 3], true), page('older', [0, 1])], + [page('older', [0, 1])], ) await fixture.journal.open({ limit: 2 }) @@ -207,8 +214,8 @@ describe('RemoteJournalStream', () => { await fixture.journal.prepend({ before: 2, limit: 2 }) expect(fixture.calls.slice(0, 2)).toEqual(['follow', 'page']) - expect(fixture.pageRequests).toEqual([{ limit: 2 }, { before: 2, limit: 2 }]) - expect(fixture.pageCursors).toEqual([3, 4]) + expect(fixture.pageRequests).toEqual([{ before: 2, limit: 2 }]) + expect(fixture.pageCursors).toEqual([4]) expect(fixture.changes).toEqual([ { type: 'replace', page: page('tail', [2, 3], true), entries: entries(2, 3), hasMore: true }, { type: 'append', entry: { seq: 4 } }, @@ -220,8 +227,8 @@ describe('RemoteJournalStream', () => { it('exposes its shared cancellation signal', async () => { const fixture = journalFixture( - [{ frames: [{ type: 'opened', cursor: -1 }], hold: true }], - [page('empty', [])], + [{ frames: [opened(-1, page('empty', []))], hold: true }], + [], ) expect(fixture.journal.signal.aborted).toBe(false) @@ -239,10 +246,10 @@ describe('RemoteJournalStream', () => { const finish = Promise.withResolvers() const resumed = journalFixture( [ - { frames: [{ type: 'opened', cursor: 0 }], waitAfterFrames: finish.promise }, + { frames: [opened(0, page('initial', [0]))], waitAfterFrames: finish.promise }, { frames: [] }, ], - [page('initial', [0])], + [], ) await resumed.journal.open({}) finish.resolve(undefined) @@ -255,8 +262,8 @@ describe('RemoteJournalStream', () => { it('prepends into an empty window and accepts its first live entry', async () => { const empty = journalFixture( - [{ frames: [{ type: 'opened', cursor: -1 }], hold: true }], - [page('empty', []), page('older', [0]), page('oldest', [])], + [{ frames: [opened(-1, page('empty', []))], hold: true }], + [page('older', [0]), page('oldest', [])], ) await empty.journal.open({}) await empty.journal.prepend({}) @@ -269,10 +276,10 @@ describe('RemoteJournalStream', () => { }) await empty.journal.dispose() - const live = Promise.withResolvers>() + const live = Promise.withResolvers() const followed = journalFixture( - [{ frames: [{ type: 'opened', cursor: -1 }, live.promise], hold: true }], - [page('empty', [])], + [{ frames: [opened(-1, page('empty', [])), live.promise], hold: true }], + [], ) await followed.journal.open({}) live.resolve({ type: 'entry', entry: { seq: 0 } }) @@ -281,61 +288,27 @@ describe('RemoteJournalStream', () => { await followed.journal.dispose() }) - it('publishes one sorted replacement from an exact page and live entries queued while it loads', async () => { - let resolvePage!: (value: Page) => void - const openingPage = new Promise((resolve) => { resolvePage = resolve }) - const fixture = journalFixture( - [{ - frames: [ - { type: 'opened', cursor: 15 }, - { type: 'entry', entry: { seq: 17 } }, - { type: 'entry', entry: { seq: 16 } }, - ], - hold: true, - }], - [openingPage], - ) - - const opening = fixture.journal.open({ limit: 6 }) - await vi.waitFor(() => { - expect(fixture.calls.filter(call => call === 'page')).toHaveLength(1) - }) - expect(fixture.changes).toEqual([]) - - resolvePage(page('opening', [10, 11, 12, 13, 14, 15])) - await opening - - expect(fixture.changes).toEqual([{ - type: 'replace', - page: page('opening', [10, 11, 12, 13, 14, 15]), - entries: entries(10, 11, 12, 13, 14, 15, 16, 17), - hasMore: false, - }]) - expect(fixture.pageCursors).toEqual([15]) - await fixture.journal.dispose() - }) - it('repairs a replacement generation through one tail page and drops replay overlap', async () => { const lost = new RemoteStreamCarrierError('carrier lost') const fixture = journalFixture( [ { frames: [ - { type: 'opened', cursor: 1 }, + opened(1, page('initial', [0, 1])), { type: 'entry', entry: { seq: 2 } }, ], terminal: lost, }, { frames: [ - { type: 'opened', cursor: 4 }, + opened(4, page('replacement', [0, 1, 2, 3, 4])), { type: 'entry', entry: { seq: 3 } }, { type: 'entry', entry: { seq: 4 } }, ], hold: true, }, ], - [page('initial', [0, 1]), page('repair', [0, 1, 2, 3, 4])], + [], ) await fixture.journal.open({ limit: 5 }) @@ -343,10 +316,10 @@ describe('RemoteJournalStream', () => { expect(fixture.changes.map(change => change.type)).toEqual(['replace', 'append', 'replace']) expect(fixture.changes[2]).toMatchObject({ - type: 'replace', page: { marker: 'repair' }, entries: entries(0, 1, 2, 3, 4), + type: 'replace', page: { marker: 'replacement' }, entries: entries(0, 1, 2, 3, 4), }) - expect(fixture.followCursors).toEqual([undefined, 2]) - expect(fixture.pageCursors).toEqual([1, 4]) + expect(fixture.followRequests).toEqual([{ limit: 5 }, { limit: 5 }]) + expect(fixture.pageCursors).toEqual([]) expect(fixture.failed).not.toHaveBeenCalled() await fixture.journal.dispose() }) @@ -355,11 +328,14 @@ describe('RemoteJournalStream', () => { const fixture = journalFixture( [ { - frames: [{ type: 'opened', cursor: 1 }], + frames: [ + opened(1, page('initial', [0, 1])), + { type: 'entry', entry: { seq: 3 } }, + ], terminal: new RemoteStreamCarrierError('carrier lost during page'), }, { - frames: [{ type: 'opened', cursor: 2 }], + frames: [opened(3, page('replacement', [0, 1, 2, 3]))], hold: true, }, ], @@ -369,20 +345,28 @@ describe('RemoteJournalStream', () => { signal.addEventListener('abort', aborted, { once: true }) if (signal.aborted) aborted() }), - page('replacement', [0, 1, 2]), ], ) await fixture.journal.open({ limit: 3 }) + await vi.waitFor(() => { expect(fixture.changes).toHaveLength(2) }) - expect(fixture.changes).toEqual([{ - type: 'replace', - page: page('replacement', [0, 1, 2]), - entries: entries(0, 1, 2), - hasMore: false, - }]) - expect(fixture.pageCursors).toEqual([1, 2]) - expect(fixture.followCursors).toEqual([undefined, 1]) + expect(fixture.changes).toEqual([ + { + type: 'replace', + page: page('initial', [0, 1]), + entries: entries(0, 1), + hasMore: false, + }, + { + type: 'replace', + page: page('replacement', [0, 1, 2, 3]), + entries: entries(0, 1, 2, 3), + hasMore: false, + }, + ]) + expect(fixture.pageCursors).toEqual([3]) + expect(fixture.followRequests).toEqual([{ limit: 3 }, { limit: 3 }]) expect(fixture.failed).not.toHaveBeenCalled() await fixture.journal.dispose() }) @@ -391,12 +375,12 @@ describe('RemoteJournalStream', () => { const fixture = journalFixture( [{ frames: [ - { type: 'opened', cursor: 1 }, + opened(1, page('initial', [0, 1])), { type: 'entry', entry: { seq: 4 } }, ], hold: true, }], - [page('initial', [0, 1]), page('repair', [0, 1, 2, 3, 4])], + [page('repair', [0, 1, 2, 3, 4])], ) await fixture.journal.open({}) @@ -404,24 +388,22 @@ describe('RemoteJournalStream', () => { expect(fixture.changes.map(change => change.type)).toEqual(['replace', 'replace']) expect(fixture.changes[1]).toMatchObject({ page: { marker: 'repair' } }) - expect(fixture.pageCursors).toEqual([1, 4]) + expect(fixture.pageCursors).toEqual([4]) await fixture.journal.dispose() }) it('replaces a superseded live-gap repair with the next generation', async () => { - const gap = Promise.withResolvers>() + const gap = Promise.withResolvers() const fixture = journalFixture( [ { - frames: [{ type: 'opened', cursor: 1 }, gap.promise], + frames: [opened(1, page('initial', [0, 1])), gap.promise], terminal: new RemoteStreamCarrierError('generation lost'), }, - { frames: [{ type: 'opened', cursor: 4 }], hold: true }, + { frames: [opened(4, page('replacement', [0, 1, 2, 3, 4]))], hold: true }, ], [ - page('initial', [0, 1]), () => new Promise(() => {}), - page('replacement', [0, 1, 2, 3, 4]), ], ) @@ -435,103 +417,173 @@ describe('RemoteJournalStream', () => { }) it('replaces a superseded second repair page with the next generation', async () => { - const live = Promise.withResolvers>() - const liveConsumed = Promise.withResolvers() - const openingPage = Promise.withResolvers() + const firstLive = Promise.withResolvers() + const secondLive = Promise.withResolvers() + const secondConsumed = Promise.withResolvers() + const firstRepair = Promise.withResolvers() const finish = Promise.withResolvers() const fixture = journalFixture( [ { - frames: [{ type: 'opened', cursor: 1 }, live.promise], + frames: [ + opened(1, page('initial', [0, 1])), + firstLive.promise, + secondLive.promise, + ], waitAfterFrames: finish.promise, terminal: new RemoteStreamCarrierError('generation lost'), - afterFrame: (index) => { if (index === 1) liveConsumed.resolve(undefined) }, + afterFrame: (index) => { if (index === 2) secondConsumed.resolve(undefined) }, }, - { frames: [{ type: 'opened', cursor: 4 }], hold: true }, + { frames: [opened(5, page('replacement', [0, 1, 2, 3, 4, 5]))], hold: true }, ], [ - openingPage.promise, - () => new Promise(() => {}), - page('replacement', [0, 1, 2, 3, 4]), + firstRepair.promise, + signal => new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { reject(new Error('page aborted')) }, { once: true }) + }), ], ) - const opening = fixture.journal.open({}) - await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([1]) }) - live.resolve({ type: 'entry', entry: { seq: 3 } }) - await liveConsumed.promise - openingPage.resolve(page('opening', [0, 1])) - await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([1, 3]) }) + await fixture.journal.open({}) + firstLive.resolve({ type: 'entry', entry: { seq: 3 } }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([3]) }) + secondLive.resolve({ type: 'entry', entry: { seq: 5 } }) + await secondConsumed.promise + firstRepair.resolve(page('first-repair', [0, 1, 2, 3])) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([3, 5]) }) finish.resolve(undefined) - await opening + await vi.waitFor(() => { expect(fixture.changes).toHaveLength(2) }) - expect(fixture.pageCursors).toEqual([1, 3, 4]) - expect(fixture.changes).toEqual([{ - type: 'replace', - page: page('replacement', [0, 1, 2, 3, 4]), - entries: entries(0, 1, 2, 3, 4), - hasMore: false, - }]) + expect(fixture.pageCursors).toEqual([3, 5]) + expect(fixture.changes).toEqual([ + { + type: 'replace', + page: page('initial', [0, 1]), + entries: entries(0, 1), + hasMore: false, + }, + { + type: 'replace', + page: page('replacement', [0, 1, 2, 3, 4, 5]), + entries: entries(0, 1, 2, 3, 4, 5), + hasMore: false, + }, + ]) await fixture.journal.dispose() }) - it('rereads the tail when queued entries advance beyond the opening page', async () => { - const live = Promise.withResolvers>() - const liveConsumed = Promise.withResolvers() - const openingPage = Promise.withResolvers() + it('rereads the tail when queued entries advance beyond the first repair page', async () => { + const firstLive = Promise.withResolvers() + const secondLive = Promise.withResolvers() + const secondConsumed = Promise.withResolvers() + const firstRepair = Promise.withResolvers() const fixture = journalFixture( [{ - frames: [{ type: 'opened', cursor: 1 }, live.promise], + frames: [ + opened(1, page('initial', [0, 1])), + firstLive.promise, + secondLive.promise, + ], hold: true, - afterFrame: (index) => { if (index === 1) liveConsumed.resolve(undefined) }, + afterFrame: (index) => { if (index === 2) secondConsumed.resolve(undefined) }, }], - [openingPage.promise, page('repair', [0, 1, 2, 3])], + [firstRepair.promise, page('repair', [0, 1, 2, 3, 4, 5])], ) - const opening = fixture.journal.open({ limit: 4 }) - await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([1]) }) - live.resolve({ type: 'entry', entry: { seq: 3 } }) - await liveConsumed.promise - openingPage.resolve(page('opening', [0, 1])) - await opening + await fixture.journal.open({ limit: 4 }) + firstLive.resolve({ type: 'entry', entry: { seq: 3 } }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([3]) }) + secondLive.resolve({ type: 'entry', entry: { seq: 5 } }) + await secondConsumed.promise + firstRepair.resolve(page('first-repair', [0, 1, 2, 3])) + await vi.waitFor(() => { expect(fixture.changes).toHaveLength(2) }) - expect(fixture.pageCursors).toEqual([1, 3]) - expect(fixture.changes).toEqual([{ - type: 'replace', page: page('repair', [0, 1, 2, 3]), entries: entries(0, 1, 2, 3), hasMore: false, - }]) + expect(fixture.pageCursors).toEqual([3, 5]) + expect(fixture.changes.at(-1)).toEqual({ + type: 'replace', + page: page('repair', [0, 1, 2, 3, 4, 5]), + entries: entries(0, 1, 2, 3, 4, 5), + hasMore: false, + }) + await fixture.journal.dispose() + }) + + it('merges contiguous entries that arrive while a replacement page is loading', async () => { + const firstLive = Promise.withResolvers() + const secondLive = Promise.withResolvers() + const secondConsumed = Promise.withResolvers() + const repair = Promise.withResolvers() + const fixture = journalFixture( + [{ + frames: [ + opened(1, page('initial', [0, 1])), + firstLive.promise, + secondLive.promise, + ], + hold: true, + afterFrame: (index) => { if (index === 2) secondConsumed.resolve(undefined) }, + }], + [repair.promise], + ) + + await fixture.journal.open({}) + firstLive.resolve({ type: 'entry', entry: { seq: 3 } }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([3]) }) + secondLive.resolve({ type: 'entry', entry: { seq: 4 } }) + await secondConsumed.promise + repair.resolve(page('repair', [0, 1, 2, 3])) + await vi.waitFor(() => { expect(fixture.changes).toHaveLength(2) }) + + expect(fixture.changes.at(-1)).toEqual({ + type: 'replace', + page: page('repair', [0, 1, 2, 3]), + entries: entries(0, 1, 2, 3, 4), + hasMore: false, + }) await fixture.journal.dispose() }) it('rejects when queued entries advance beyond the second repair page', async () => { - const firstLive = Promise.withResolvers>() - const secondLive = Promise.withResolvers>() - const firstConsumed = Promise.withResolvers() + const firstLive = Promise.withResolvers() + const secondLive = Promise.withResolvers() + const thirdLive = Promise.withResolvers() const secondConsumed = Promise.withResolvers() - const openingPage = Promise.withResolvers() - const repairPage = Promise.withResolvers() + const thirdConsumed = Promise.withResolvers() + const firstRepair = Promise.withResolvers() + const secondRepair = Promise.withResolvers() const fixture = journalFixture( [{ - frames: [{ type: 'opened', cursor: 1 }, firstLive.promise, secondLive.promise], + frames: [ + opened(1, page('initial', [0, 1])), + firstLive.promise, + secondLive.promise, + thirdLive.promise, + ], hold: true, afterFrame: (index) => { - if (index === 1) firstConsumed.resolve(undefined) if (index === 2) secondConsumed.resolve(undefined) + if (index === 3) thirdConsumed.resolve(undefined) }, }], - [openingPage.promise, repairPage.promise], + [firstRepair.promise, secondRepair.promise], ) - const opening = fixture.journal.open({}) - await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([1]) }) + await fixture.journal.open({}) firstLive.resolve({ type: 'entry', entry: { seq: 3 } }) - await firstConsumed.promise - openingPage.resolve(page('opening', [0, 1])) - await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([1, 3]) }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([3]) }) secondLive.resolve({ type: 'entry', entry: { seq: 5 } }) await secondConsumed.promise - repairPage.resolve(page('repair', [0, 1, 2, 3])) + firstRepair.resolve(page('first-repair', [0, 1, 2, 3])) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([3, 5]) }) + thirdLive.resolve({ type: 'entry', entry: { seq: 7 } }) + await thirdConsumed.promise + secondRepair.resolve(page('second-repair', [0, 1, 2, 3, 4, 5])) - await expect(opening).rejects.toThrow('page did not reach its opening cursor') + await vi.waitFor(() => { expect(fixture.failed).toHaveBeenCalledOnce() }) + expect(fixture.failed.mock.calls[0]?.[0]).toMatchObject({ + message: 'fixture journal page did not reach its opening cursor', + }) + await fixture.journal.dispose() }) it('reports a resumed generation that emits an entry before its cursor', async () => { @@ -539,13 +591,13 @@ describe('RemoteJournalStream', () => { const fixture = journalFixture( [ { - frames: [{ type: 'opened', cursor: 0 }], + frames: [opened(0, page('initial', [0]))], waitAfterFrames: finish.promise, terminal: new RemoteStreamCarrierError('lost'), }, { frames: [{ type: 'entry', entry: { seq: 1 } }] }, ], - [page('initial', [0])], + [], ) await fixture.journal.open({}) @@ -558,14 +610,14 @@ describe('RemoteJournalStream', () => { }) it('reports a duplicate opening cursor after the initial page is published', async () => { - const duplicate = Promise.withResolvers>() + const duplicate = Promise.withResolvers() const fixture = journalFixture( - [{ frames: [{ type: 'opened', cursor: 0 }, duplicate.promise], hold: true }], - [page('initial', [0])], + [{ frames: [opened(0, page('initial', [0])), duplicate.promise], hold: true }], + [], ) await fixture.journal.open({}) - duplicate.resolve({ type: 'opened', cursor: 0 }) + duplicate.resolve(opened(0, page('duplicate', [0]))) await vi.waitFor(() => { expect(fixture.failed).toHaveBeenCalledOnce() }) expect(fixture.failed.mock.calls[0]?.[0]).toMatchObject({ message: 'fixture journal emitted more than one opening cursor', @@ -573,47 +625,16 @@ describe('RemoteJournalStream', () => { await fixture.journal.dispose() }) - it('propagates follow failures and duplicate cursors while an opening page is pending', async () => { - const pendingPage = new Promise(() => {}) + it('reports a follow failure after publishing its opening snapshot', async () => { const failedFollow = journalFixture( - [{ frames: [{ type: 'opened', cursor: 0 }], terminal: new Error('follow failed') }], - [pendingPage], - ) - await expect(failedFollow.journal.open({})).rejects.toThrow('follow failed') - - const duplicate = Promise.withResolvers>() - const duplicatePage = new Promise(() => {}) - const duplicateOpening = journalFixture( - [{ frames: [{ type: 'opened', cursor: 0 }, duplicate.promise] }], - [duplicatePage], - ) - const opening = duplicateOpening.journal.open({}) - await vi.waitFor(() => { expect(duplicateOpening.pageCursors).toEqual([0]) }) - duplicate.resolve({ type: 'opened', cursor: 0 }) - await expect(opening).rejects.toThrow('more than one opening cursor') - }) - - it('rejects an iterator that ends while its opening page is pending', async () => { - const generation = new AbortController() - const results = [ - Promise.resolve>>>({ - done: false, - value: remoteItem(1, { type: 'opened', cursor: 0 }, generation.signal), - }), - Promise.resolve>>>({ - done: true, - value: undefined, - }), - ] - const fixture = journalFixture( + [{ frames: [opened(0, page('initial', [0]))], terminal: new Error('follow failed') }], [], - [new Promise(() => {})], - controlledFactory(() => results.shift() ?? Promise.resolve({ done: true, value: undefined })), - ) - - await expect(fixture.journal.open({})).rejects.toThrow( - 'ended while reading its replacement page', ) + await failedFollow.journal.open({}) + await vi.waitFor(() => { expect(failedFollow.failed).toHaveBeenCalledOnce() }) + expect(failedFollow.failed.mock.calls[0]?.[0]).toMatchObject({ message: 'follow failed' }) + expect(failedFollow.changes).toHaveLength(1) + await failedFollow.journal.dispose() }) it('rejects an iterator that ends before its opening cursor', async () => { @@ -627,17 +648,17 @@ describe('RemoteJournalStream', () => { it('suppresses a consumer failure after disposal begins', async () => { const generation = new AbortController() - const next = Promise.withResolvers>>>() + const next = Promise.withResolvers>>() const results = [ - Promise.resolve>>>({ + Promise.resolve>>({ done: false, - value: remoteItem(1, { type: 'opened', cursor: 0 }, generation.signal), + value: remoteItem(1, opened(0, page('initial', [0])), generation.signal), }), next.promise, ] const fixture = journalFixture( [], - [page('initial', [0])], + [], controlledFactory(() => results.shift() ?? Promise.resolve({ done: true, value: undefined })), ) @@ -645,7 +666,7 @@ describe('RemoteJournalStream', () => { const closing = fixture.journal.dispose() next.resolve({ done: false, - value: remoteItem(1, { type: 'opened', cursor: 0 }, generation.signal), + value: remoteItem(1, opened(0, page('duplicate', [0])), generation.signal), }) await closing expect(fixture.failed).not.toHaveBeenCalled() @@ -658,17 +679,17 @@ describe('RemoteJournalStream', () => { final: undefined, message: 'more than one opening cursor', }, - ])('rejects when an aborted page generation $name', async ({ final, message }) => { + ])('reports when an aborted repair generation $name', async ({ final, message }) => { const generation = new AbortController() - const pending = Promise.withResolvers>>>() - const nextPending = Promise.withResolvers>>>() + const gap = Promise.withResolvers>>() + const replacement = Promise.withResolvers>>() const results = [ - Promise.resolve>>>({ + Promise.resolve>>({ done: false, - value: remoteItem(1, { type: 'opened', cursor: 0 }, generation.signal), + value: remoteItem(1, opened(0, page('initial', [0])), generation.signal), }), - pending.promise, - nextPending.promise, + gap.promise, + replacement.promise, ] const fixture = journalFixture( [], @@ -678,47 +699,154 @@ describe('RemoteJournalStream', () => { controlledFactory(() => results.shift() ?? Promise.resolve({ done: true, value: undefined })), ) - const opening = fixture.journal.open({}) - await vi.waitFor(() => { expect(results).toHaveLength(1) }) + await fixture.journal.open({}) + gap.resolve({ + done: false, + value: remoteItem(1, { type: 'entry', entry: { seq: 2 } }, generation.signal), + }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([2]) }) generation.abort() if (final === undefined) { - pending.resolve({ + replacement.resolve({ done: false, - value: remoteItem(1, { type: 'entry', entry: { seq: 1 } }, generation.signal), - }) - await vi.waitFor(() => { expect(results).toHaveLength(0) }) - nextPending.resolve({ - done: false, - value: remoteItem(1, { type: 'opened', cursor: 1 }, generation.signal), + value: remoteItem(1, opened(2, page('duplicate', [0, 1, 2])), generation.signal), }) } else { - pending.resolve(final) + replacement.resolve(final) } - await expect(opening).rejects.toThrow(message) + await vi.waitFor(() => { expect(fixture.failed).toHaveBeenCalledOnce() }) + const failure: unknown = fixture.failed.mock.calls[0]?.[0] + expect(failure).toBeInstanceOf(Error) + if (!(failure instanceof Error)) throw new Error('journal failure was not an Error') + expect(failure.message).toContain(message) + await fixture.journal.dispose() + }) + + it('discards old-generation entries while waiting for the replacement opening', async () => { + const generation = new AbortController() + const gap = Promise.withResolvers>>() + const stale = Promise.withResolvers>>() + const replacement = Promise.withResolvers>>() + const results = [ + Promise.resolve>>({ + done: false, + value: remoteItem(1, opened(0, page('initial', [0])), generation.signal), + }), + gap.promise, + stale.promise, + replacement.promise, + ] + const fixture = journalFixture( + [], + [signal => new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { reject(new Error('page aborted')) }, { once: true }) + })], + controlledFactory(() => results.shift() ?? Promise.resolve({ done: true, value: undefined })), + ) + + await fixture.journal.open({}) + gap.resolve({ + done: false, + value: remoteItem(1, { type: 'entry', entry: { seq: 2 } }, generation.signal), + }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([2]) }) + generation.abort() + stale.resolve({ + done: false, + value: remoteItem(1, { type: 'entry', entry: { seq: 1 } }, generation.signal), + }) + replacement.resolve({ + done: false, + value: remoteItem(2, opened(2, page('replacement', [0, 1, 2])), new AbortController().signal), + }) + + await vi.waitFor(() => { expect(fixture.changes).toHaveLength(2) }) + expect(fixture.changes.at(-1)).toMatchObject({ page: { marker: 'replacement' } }) + await fixture.journal.dispose() + }) + + it.each([ + { + name: 'rejects', + settle: ( + _resolve: (value: IteratorResult>) => void, + reject: (reason?: unknown) => void, + ) => { reject(new Error('replacement follow failed')) }, + message: 'replacement follow failed', + }, + { + name: 'ends', + settle: (resolve: (value: IteratorResult>) => void) => { + resolve({ done: true, value: undefined }) + }, + message: 'ended while reading its replacement page', + }, + { + name: 'opens twice', + settle: (resolve: (value: IteratorResult>) => void) => { + resolve({ + done: false, + value: remoteItem(1, opened(2, page('duplicate', [0, 1, 2])), new AbortController().signal), + }) + }, + message: 'more than one opening cursor', + }, + ])('reports when a follow $name during live-gap repair', async ({ settle, message }) => { + const generation = new AbortController() + const gap = Promise.withResolvers>>() + const next = Promise.withResolvers>>() + const results = [ + Promise.resolve>>({ + done: false, + value: remoteItem(1, opened(0, page('initial', [0])), generation.signal), + }), + gap.promise, + next.promise, + ] + const fixture = journalFixture( + [], + [() => new Promise(() => {})], + controlledFactory(() => results.shift() ?? Promise.resolve({ done: true, value: undefined })), + ) + + await fixture.journal.open({}) + gap.resolve({ + done: false, + value: remoteItem(1, { type: 'entry', entry: { seq: 2 } }, generation.signal), + }) + await vi.waitFor(() => { expect(fixture.pageCursors).toEqual([2]) }) + settle(next.resolve, next.reject) + + await vi.waitFor(() => { expect(fixture.failed).toHaveBeenCalledOnce() }) + const failure: unknown = fixture.failed.mock.calls[0]?.[0] + expect(failure).toBeInstanceOf(Error) + if (!(failure instanceof Error)) throw new Error('journal failure was not an Error') + expect(failure.message).toContain(message) + await fixture.journal.dispose() }) it('rejects malformed opening and page sequences', async () => { const beforeOpening = journalFixture( [{ frames: [{ type: 'entry', entry: { seq: 0 } }] }], - [page('unused', [])], + [], ) await expect(beforeOpening.journal.open({})).rejects.toThrow('entry before its opening cursor') const discontinuousPage = journalFixture( - [{ frames: [{ type: 'opened', cursor: 3 }], hold: true }], - [page('bad', [0, 2, 3])], + [{ frames: [opened(3, page('bad', [0, 2, 3]))], hold: true }], + [], ) await expect(discontinuousPage.journal.open({})).rejects.toThrow('page contains discontinuous entries') const shortPage = journalFixture( - [{ frames: [{ type: 'opened', cursor: 3 }], hold: true }], - [page('short', [0, 1])], + [{ frames: [opened(3, page('short', [0, 1]))], hold: true }], + [], ) await expect(shortPage.journal.open({})).rejects.toThrow('page did not end at its requested cursor') const longPage = journalFixture( - [{ frames: [{ type: 'opened', cursor: 1 }], hold: true }], - [page('long', [0, 1, 2])], + [{ frames: [opened(1, page('long', [0, 1, 2]))], hold: true }], + [], ) await expect(longPage.journal.open({})).rejects.toThrow('page did not end at its requested cursor') }) @@ -726,9 +854,12 @@ describe('RemoteJournalStream', () => { it('reports duplicate and regressed generation cursors as terminal failures', async () => { const duplicate = journalFixture( [{ - frames: [{ type: 'opened', cursor: 1 }, { type: 'opened', cursor: 1 }], + frames: [ + opened(1, page('initial', [0, 1])), + opened(1, page('duplicate', [0, 1])), + ], }], - [page('initial', [0, 1])], + [], ) await duplicate.journal.open({}) await vi.waitFor(() => { expect(duplicate.failed).toHaveBeenCalledOnce() }) @@ -740,12 +871,12 @@ describe('RemoteJournalStream', () => { const regressed = journalFixture( [ { - frames: [{ type: 'opened', cursor: 1 }, { type: 'entry', entry: { seq: 2 } }], + frames: [opened(1, page('initial', [0, 1])), { type: 'entry', entry: { seq: 2 } }], terminal: new RemoteStreamCarrierError('lost'), }, - { frames: [{ type: 'opened', cursor: 1 }] }, + { frames: [opened(1, page('regressed', [0, 1]))] }, ], - [page('initial', [0, 1])], + [], ) await regressed.journal.open({}) await vi.waitFor(() => { expect(regressed.failed).toHaveBeenCalledOnce() }) @@ -757,8 +888,8 @@ describe('RemoteJournalStream', () => { it('rejects a discontinuous older page after publishing the fail-soft pagination state', async () => { const fixture = journalFixture( - [{ frames: [{ type: 'opened', cursor: 4 }], hold: true }], - [page('initial', [3, 4], true), page('older', [0, 1], true)], + [{ frames: [opened(4, page('initial', [3, 4], true))], hold: true }], + [page('older', [0, 1], true)], ) await fixture.journal.open({}) @@ -771,8 +902,8 @@ describe('RemoteJournalStream', () => { it('guards lifecycle operations before and after open', async () => { const fixture = journalFixture( - [{ frames: [{ type: 'opened', cursor: -1 }], hold: true }], - [page('empty', [])], + [{ frames: [opened(-1, page('empty', []))], hold: true }], + [], ) await expect(fixture.journal.prepend({})).rejects.toThrow('is not open') diff --git a/packages/api/session-controller/src/client/contract/snapshot.ts b/packages/api/session-controller/src/client/contract/snapshot.ts index 7304c21bb8..dff2317359 100644 --- a/packages/api/session-controller/src/client/contract/snapshot.ts +++ b/packages/api/session-controller/src/client/contract/snapshot.ts @@ -29,7 +29,11 @@ export interface SessionSnapshot { readonly sessionId: SessionId readonly queue: readonly QueuedMessage[] readonly running: boolean - readonly subagent: { readonly address: SubagentAddress; readonly parentAvailable: boolean } | null + readonly subagent: { + readonly address: SubagentAddress + /** Absent until the direct-parent catalog resolves. */ + readonly parentAvailable?: boolean + } | null readonly removed: boolean readonly openState: OpenState readonly openError: ClientFailure | null diff --git a/packages/api/session-controller/src/client/sessions/session.ts b/packages/api/session-controller/src/client/sessions/session.ts index d51d475474..7239e3149a 100644 --- a/packages/api/session-controller/src/client/sessions/session.ts +++ b/packages/api/session-controller/src/client/sessions/session.ts @@ -45,7 +45,7 @@ export const PAGE_MESSAGES = 50 export interface SessionOptions { /** Catalog-discovered address selecting non-activating subagent transport. */ address?: SubagentAddress - /** Whether the exact direct parent Agent was live at the latest catalog read. */ + /** Whether the exact direct parent Agent was live at the latest catalog read; absent before that read. */ parentAvailable?: boolean /** * First ACCEPTED prompt on a blank session (fires at most once, on the @@ -86,7 +86,7 @@ export class Session implements SessionFace { private readonly queueMirror = new SessionQueueMirror() private running = false private address: SubagentAddress | undefined - private parentAvailable = false + private parentAvailable: boolean | undefined /** * Sticky send marker, private input of the composerPhase derivation: set * synchronously before prompt()'s first await, never reset — the blank → @@ -144,7 +144,7 @@ export class Session implements SessionFace { ) { this.projections = options.projections ?? new ProjectionValueStore() this.address = options.address - this.parentAvailable = options.parentAvailable ?? false + this.parentAvailable = options.parentAvailable this.notifier = new Notifier(() => { this.snapshotCache = this.buildSnapshot() }) @@ -467,9 +467,9 @@ export class Session implements SessionFace { * Install or clear the catalog-discovered transport address. A changed * address rebuilds an already-open window through its new history route. * @param address - direct parent/child address, or undefined for ordinary transport. - * @param parentAvailable - latest exact-parent availability hint. + * @param parentAvailable - latest exact-parent availability hint, or undefined before a catalog read. */ - configureSubagent(address: SubagentAddress | undefined, parentAvailable = false): void { + configureSubagent(address: SubagentAddress | undefined, parentAvailable?: boolean): void { const same = this.address?.parentSessionId === address?.parentSessionId && this.address?.childSessionId === address?.childSessionId && this.address?.mode === address?.mode @@ -623,7 +623,10 @@ export class Session implements SessionFace { running: this.running, subagent: this.address === undefined ? null - : { address: this.address, parentAvailable: this.parentAvailable }, + : { + address: this.address, + ...(this.parentAvailable === undefined ? {} : { parentAvailable: this.parentAvailable }), + }, removed: this.removed, openState: this.openState, openError: this.openError, diff --git a/packages/api/session-controller/src/client/transport.ts b/packages/api/session-controller/src/client/transport.ts index 2b7452056c..0e8ca83d28 100644 --- a/packages/api/session-controller/src/client/transport.ts +++ b/packages/api/session-controller/src/client/transport.ts @@ -17,6 +17,7 @@ import type { SessionEventEntry, SessionPage, SessionPageRequest, + SessionProjectionBaseline, } from '../types.ts' export { @@ -30,8 +31,13 @@ export type ClientSessionPageRequest = Omit +export type SessionJournalChange = RemoteJournalChange type SessionControlBaselineFrame = Extract type SessionControlDeltaFrame = Exclude @@ -93,7 +99,7 @@ export function createSessionControlStream( /** Gateway-owned event journal bound to one ordinary or direct-subagent Session address. */ export class SessionEventStream extends RemoteJournalStream< - SessionPage, + SessionJournalPage, SessionEventEntry, number, ClientSessionPageRequest @@ -126,15 +132,23 @@ export class SessionEventStream extends RemoteJournalStream< /** @inheritdoc */ protected override async * follow( - afterSeq: number | undefined, + request: ClientSessionPageRequest, signal: AbortSignal, - ): AsyncIterable> { - const request = afterSeq === undefined - ? { address: this.address } - : { address: this.address, afterSeq } - for await (const frame of this.remote.session.follow(request, signal)) { - if (frame.type === 'opened') { - yield frame + ): AsyncIterable> { + for await (const frame of this.remote.session.follow({ + address: this.address, + ...(request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages }), + }, signal)) { + if (frame.type === 'snapshot') { + yield { + type: 'opened', + cursor: frame.cursor, + page: { + events: frame.events, + hasMore: frame.hasMore, + projections: frame.projections, + }, + } continue } const { type: _type, ...entry } = frame @@ -147,7 +161,7 @@ export class SessionEventStream extends RemoteJournalStream< request: ClientSessionPageRequest, throughSeq: number, signal: AbortSignal, - ): Promise { + ): Promise { const result = await this.remote.session.page( { address: this.address, throughSeq, ...request }, signal, diff --git a/packages/api/session-controller/src/history.ts b/packages/api/session-controller/src/history.ts index e30b4f1139..a1fb6b541b 100644 --- a/packages/api/session-controller/src/history.ts +++ b/packages/api/session-controller/src/history.ts @@ -2,9 +2,9 @@ import type { Context } from '@deepseek-ai/cordis' import { isAppendSurfaceEvent } from '@deepseek-ai/dsh-session' -import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' -import type { SessionInspection } from '@deepseek-ai/dsh-session-persistence' -import { foldSubagentDescriptor } from '@deepseek-ai/dsh-subagent' +import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session' +import { SessionQueryError, type SessionObservation } from '@deepseek-ai/dsh-session-query' +import type {} from '@deepseek-ai/dsh-subagent' import { TypertRemoteFailure } from '@deepseek-ai/dsh-typert-protocol' import type { SessionAddress, @@ -13,7 +13,7 @@ import type { SessionFollowFrame, SessionPage, SessionPageRequest, - SessionProjectionsBlock, + SessionProjectionBaseline, SessionProjectionValues, SessionWireEvent, } from './types.ts' @@ -21,16 +21,18 @@ import type { const DEFAULT_MAX_MESSAGES = 50 const MESSAGE_TYPES = new Set(['user/message', 'assistant/message']) -type SessionSource = - | { readonly kind: 'attached'; readonly session: Session } - | { readonly kind: 'detached'; readonly header: SessionHeader; readonly events: readonly SessionEvent[] } - /** Implements cold-safe history operations delegated by the Session Controller. */ export class SessionHistoryController { private readonly closeFollowers = new Set<() => void>() - /** @param ctx - Host context carrying Session, persistence, and projection services. */ - constructor(private readonly ctx: Context) { + /** + * @param ctx - Host context carrying Session query and projection services. + * @param promote - starts ordinary Session activation after snapshot delivery. + */ + constructor( + private readonly ctx: Context, + private readonly promote: (observation: SessionObservation) => void, + ) { ctx.effect(() => () => { for (const close of this.closeFollowers) close() this.closeFollowers.clear() @@ -41,13 +43,13 @@ export class SessionHistoryController { * Read one message-aligned history page without activating an Agent. * @param request - durable address and backwards-page cursor. * @param signal - caller cancellation for persistence reads. - * @returns a contiguous event page and a projection baseline on tail reads. + * @returns a contiguous event page. */ async page(request: SessionPageRequest, signal: AbortSignal): Promise { validatePageRequest(request) - const source = await this.sourceFor(request.address, signal) + using source = await this.sourceFor(request.address, signal, false) signal.throwIfAborted() - const sourceLog = sourceEvents(source) + const sourceLog = source.events const sourceCursor = sourceLog.at(-1)?.seq ?? -1 if (request.throughSeq > sourceCursor) { reject( @@ -56,19 +58,20 @@ export class SessionHistoryController { {}, ) } - const events = sourceLog.filter(event => event.seq <= request.throughSeq) - if ((events.at(-1)?.seq ?? -1) !== request.throughSeq) { + /* v8 ignore next -- Session and persistence validation guarantee a dense zero-based event prefix. */ + if (request.throughSeq >= 0 && sourceLog[request.throughSeq]?.seq !== request.throughSeq) { reject('internal', `session log does not contain through seq ${String(request.throughSeq)}`, {}) } - const page = paginate(events, request.beforeSeq, request.maxMessages ?? DEFAULT_MAX_MESSAGES) + const page = paginate( + sourceLog, + request.beforeSeq, + request.maxMessages ?? DEFAULT_MAX_MESSAGES, + request.throughSeq, + ) const entries = page.events.map(entryFor) - const projections = request.beforeSeq === undefined - ? this.projectionsFor(request.address, source, events) - : undefined return { events: entries, hasMore: page.hasMore, - ...(projections === undefined ? {} : { projections }), } } @@ -76,13 +79,14 @@ export class SessionHistoryController { * Follow events appended after an initial cursor on one durable address. * @param request - durable address and last committed sequence already held by the caller. * @param signal - stream cancellation owned by the Remote carrier. - * @returns an opened cursor followed by gap-free event frames. + * @returns a complete opening snapshot followed by gap-free event frames. */ async *follow(request: SessionFollowRequest, signal: AbortSignal): AsyncIterable { validateFollowRequest(request) - const { address, afterSeq } = request + const { address } = request const target = addressId(address) const buffered: SessionEvent[] = [] + let snapshotCursor: number | undefined let wake: (() => void) | undefined const notify = (): void => { const resume = wake @@ -102,35 +106,44 @@ export class SessionHistoryController { }, { global: true }) const disposeCreated = this.ctx.on('session/created', (session) => { if (session.id !== target) return - // Session construction appends session/end-seed before attachment, so the - // marker has no session/event notification. Earlier session/created listeners - // may publish later setup events first; this suffix must precede those notifications. - const suffix = session.events.slice(session.firstLiveSeq) + // Constructor seed events have no session/event notification. Normally + // only the end-seed suffix is new; if persistence advanced after the + // opening observation, replay everything beyond that snapshot cursor. + const suffix = session.events.slice(snapshotCursor === undefined + ? session.firstLiveSeq + : snapshotCursor + 1) buffered.unshift(...suffix) notify() }, { global: true }) const onAbort = (): void => { notify() } signal.addEventListener('abort', onAbort, { once: true }) try { - const source = await this.sourceFor(address, signal) - const events = [...sourceEvents(source)] + using source = await this.sourceFor(address, signal, true) + const events = source.events signal.throwIfAborted() - const cursor = events.at(-1)?.seq ?? -1 - if (afterSeq !== undefined && afterSeq > cursor) { - reject('bad-request', `session event resume seq ${String(afterSeq)} is past cursor ${String(cursor)}`, {}) + const cursor = source.cursor + snapshotCursor = cursor + const page = paginate(events, undefined, request.maxMessages ?? DEFAULT_MAX_MESSAGES) + yield { + type: 'snapshot', + header: source.header, + cursor, + events: page.events.map(entryFor), + hasMore: page.hasMore, + projections: source.projections === undefined + ? { asOfSeq: cursor, values: {} } + : projectionBlock(source.projections), } - let nextSeq = (afterSeq ?? cursor) + 1 - yield { type: 'opened', cursor } - if (afterSeq !== undefined) { - for (const event of events) { - if (event.seq < nextSeq) continue - if (event.seq !== nextSeq) { - reject('internal', `session event replay skipped seq ${String(nextSeq)}`, {}) - } - nextSeq++ - yield { type: 'event', ...entryFor(event) } + if (address.kind === 'session' && source.source === 'prepared') { + const promotion = source.retain() + try { + this.promote(promotion) + } catch (error: unknown) { + promotion[Symbol.dispose]() + throw error } } + let nextSeq = cursor + 1 while (!follower.closed && !signal.aborted) { const item = buffered.shift() if (item === undefined) { @@ -152,49 +165,44 @@ export class SessionHistoryController { } } - private async sourceFor(address: SessionAddress, signal: AbortSignal): Promise { + private async sourceFor( + address: SessionAddress, + signal: AbortSignal, + withProjections: boolean, + ): Promise { const sessionId = addressId(address) - const attached = this.ctx.sessions.get(sessionId) - if (attached !== undefined) { - validateAddress(address, attached.header, attached.events) - return { kind: 'attached', session: attached } + try { + const observation = await this.ctx.sessionQuery.observeSession(sessionId, { + signal, + projectionMode: withProjections || address.kind === 'subagent' ? 'all' : 'none', + }) + if (observation.header.cwd === undefined) { + observation[Symbol.dispose]() + rejectNotFound(address) + } + try { + validateAddress(address, observation.header, observation.projections) + } catch (error: unknown) { + observation[Symbol.dispose]() + throw error + } + return observation + } catch (error: unknown) { + if (error instanceof SessionQueryError + && error.code === 'SESSION_QUERY_SESSION_NOT_FOUND') rejectNotFound(address) + throw error } - const persistence = this.ctx.get('sessionPersistence') - if (persistence === undefined) { - reject('internal', 'session persistence is not configured', {}) - } - signal.throwIfAborted() - const header = (await persistence.list(signal)).find(candidate => candidate.id === sessionId) - if (header === undefined || header.cwd === undefined) rejectNotFound(address) - const inspected: SessionInspection = await persistence.inspect(sessionId, signal) - signal.throwIfAborted() - if (inspected.meta.cwd === undefined) rejectNotFound(address) - validateAddress(address, inspected.meta, inspected.events) - return { kind: 'detached', header: inspected.meta, events: inspected.events } } - private projectionsFor( - address: SessionAddress, - source: SessionSource, - events: readonly SessionEvent[], - ): SessionProjectionsBlock | undefined { - const registry = this.ctx.get('sessionProjections') - if (registry === undefined) return undefined - try { - const throughSeq = events.at(-1)?.seq ?? -1 - const snapshot = source.kind === 'attached' && source.session.seq - 1 === throughSeq - ? registry.snapshot(source.session) - : registry.restore({}, events, 0).snapshot - return { - asOfSeq: snapshot.asOfSeq, - // Projection definitions validate whole JSON values before snapshot publication. - values: snapshot.values as SessionProjectionValues, - } - } catch (error) { - if (address.kind === 'session') throw error - this.ctx.logger.warn(`session.page: projections for "${address.childSessionId}" failed: ${String(error)}`) - return undefined - } +} + +function projectionBlock( + snapshot: NonNullable, +): SessionProjectionBaseline { + return { + asOfSeq: snapshot.asOfSeq, + // Projection definitions validate whole JSON values before snapshot publication. + values: snapshot.values as SessionProjectionValues, } } @@ -213,9 +221,9 @@ function validatePageRequest(request: SessionPageRequest): void { } function validateFollowRequest(request: SessionFollowRequest): void { - if (request.afterSeq !== undefined - && (!Number.isSafeInteger(request.afterSeq) || request.afterSeq < -1)) { - reject('bad-request', 'afterSeq must be an integer greater than or equal to -1', {}) + if (request.maxMessages !== undefined + && (!Number.isSafeInteger(request.maxMessages) || request.maxMessages <= 0)) { + reject('bad-request', 'maxMessages must be a positive safe integer', {}) } } @@ -226,7 +234,7 @@ function addressId(address: SessionAddress): SessionId { function validateAddress( address: SessionAddress, header: SessionHeader, - events: readonly SessionEvent[], + projections: SessionObservation['projections'], ): void { if (address.kind === 'session') { if (header.origin === 'subagent') { @@ -241,24 +249,22 @@ function validateAddress( childSessionId: address.childSessionId, }) } - let descriptor - try { - descriptor = foldSubagentDescriptor(events.slice(header.seedLength ?? 0)) - } catch { + const identity = projections?.values.subagent + if (identity === null) { reject('subagent-catalog-diagnostic', 'subagent descriptor is corrupt', { parentSessionId: address.parentSessionId, childSessionId: address.childSessionId, reason: 'corrupt', }) } - if (descriptor === undefined) { + if (identity === undefined || identity.seq < (header.seedLength ?? 0)) { reject('subagent-catalog-diagnostic', 'subagent descriptor is unavailable', { parentSessionId: address.parentSessionId, childSessionId: address.childSessionId, reason: 'unsupported', }) } - if (descriptor.mode !== address.mode) { + if (identity.mode !== address.mode) { reject('subagent-unauthorized', 'subagent mode does not match the supplied address', { childSessionId: address.childSessionId, }) @@ -279,20 +285,17 @@ function reject(code: string, message: string, details: object): never { throw new TypertRemoteFailure({ code, message, details }) } -function sourceEvents(source: SessionSource): readonly SessionEvent[] { - return source.kind === 'attached' ? source.session.events : source.events -} - function paginate( events: readonly SessionEvent[], beforeSeq: number | undefined, maxMessages: number, + throughSeq = events.at(-1)?.seq ?? -1, ): { readonly events: SessionEvent[]; readonly hasMore: boolean } { - const window = beforeSeq === undefined ? [...events] : events.filter(event => event.seq < beforeSeq) + const end = Math.min(throughSeq + 1, beforeSeq ?? throughSeq + 1) let count = 0 let cut = 0 - for (let index = window.length - 1; index >= 0; index--) { - const event = window[index] as SessionEvent + for (let index = end - 1; index >= 0; index--) { + const event = events[index] as SessionEvent if (!MESSAGE_TYPES.has(event.type) || !isAppendSurfaceEvent(event)) continue count++ const sources = (event as { readonly sourceEventSeqs?: readonly number[] }).sourceEventSeqs @@ -305,7 +308,7 @@ function paginate( break } } - return { events: window.filter(event => event.seq >= cut), hasMore: cut > 0 } + return { events: events.slice(cut, end), hasMore: cut > 0 } } function entryFor(event: SessionEvent): SessionEventEntry { diff --git a/packages/api/session-controller/src/types.ts b/packages/api/session-controller/src/types.ts index adcf8f6c06..33e6d0ef46 100644 --- a/packages/api/session-controller/src/types.ts +++ b/packages/api/session-controller/src/types.ts @@ -6,7 +6,7 @@ import type { import type { Branded } from '@deepseek-ai/dsh-brand' import type { MessageId } from '@deepseek-ai/dsh-llm/brand' import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' -import type { JsonValue, SessionId, SurfaceOp } from '@deepseek-ai/dsh-session/types' +import type { JsonValue, SessionHeader, SessionId, SurfaceOp } from '@deepseek-ai/dsh-session/types' import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types' import type { JobId } from '@deepseek-ai/dsh-jobs/brand' import type { WorkspaceId } from '@deepseek-ai/dsh-workspace/types' @@ -17,12 +17,26 @@ declare module '@deepseek-ai/dsh-session-projection/types' { sessionListMetadata: SessionListMetadata /** Host state for the boot-constant image-limit view. */ imageLimits: null + /** Durable model selection already used by a request and still pending for a later request. */ + modelSelection: ModelSelectionProjectionState } interface SessionProjectionMap { /** Persisted facts used to summarize a Session without activating it. */ sessionListMetadata: SessionListMetadata /** Image-intake limits enforced by the Session prompt endpoint. */ imageLimits: ImageAttachmentLimits + /** Durable model selection already used and selected for the next request. */ + modelSelection: ModelSelectionProjection + } +} + +declare module '@deepseek-ai/dsh-session/types' { + interface SessionEventMap { + /** + * Complete validated model selection requested for subsequent prompt + * assembly. Log-only: it never enters derived model history. + */ + 'model/selection': ModelSelection } } @@ -34,10 +48,17 @@ export interface SessionListMetadata { readonly lastPromptAt: number | null } -/** Projection values and the durable event position they represent. */ -export interface SessionProjectionsBlock { +/** Every available cached wire value used as partial, possibly stale Session-list hints. */ +export interface SessionProjectionHints { readonly asOfSeq: number - /** Provider-validated values across the merge-extensible projection key space. */ + /** Provider-validated values present in the cache; omitted keys remain unknown. */ + readonly values: SessionProjectionValues +} + +/** Complete projection values at an exact Session event cursor. */ +export interface SessionProjectionBaseline { + readonly asOfSeq: number + /** Provider-validated values; omitted keys are absent capabilities at this cut. */ readonly values: SessionProjectionValues } @@ -62,6 +83,22 @@ export interface ModelSelection { readonly reasoningEffort?: string } +/** Host fold state for durable model selection. */ +export interface ModelSelectionProjectionState { + /** Selection consumed by the latest recorded model request. */ + readonly lastUsed: ModelSelection | null + /** Later user selection not yet consumed by a matching model request. */ + readonly pending: ModelSelection | null +} + +/** Client view of the durable model-selection fold. */ +export interface ModelSelectionProjection { + /** Selection consumed by the latest recorded model request. */ + readonly lastUsed: ModelSelection | null + /** Selection the next request should use, falling back to {@link lastUsed}. */ + readonly next: ModelSelection | null +} + /** One adapter-owned reasoning effort for an exact model route. */ export interface ModelReasoningEffort { readonly id: string @@ -97,10 +134,11 @@ export interface ModelCatalogFailure { readonly message: string } -/** Detached model-directory snapshot for one Session. */ -export interface SessionModels { - readonly current: ModelSelection - readonly routable: boolean +/** Host-generation model catalog and the default used by unconfigured Sessions. */ +export interface ModelCatalog { + readonly default: ModelSelection + /** Provider routes currently able to serve a request, including empty catalogs. */ + readonly routableProviders: readonly string[] readonly groups: readonly ModelProviderGroup[] readonly failures: readonly ModelCatalogFailure[] } @@ -120,8 +158,7 @@ export interface SessionSummary { readonly parentSessionId?: SessionId readonly origin?: 'subagent' readonly cwd?: string - readonly agentPreset?: string - readonly projections?: SessionProjectionsBlock + readonly projections?: SessionProjectionHints } /** One session-content search result. */ @@ -220,11 +257,6 @@ export interface SessionCreateValue { readonly agentPreset?: string } -/** Model-directory request. */ -export interface SessionModelsRequest { - readonly sessionId: SessionId -} - /** Session model-selection request. */ export interface SessionSelectModelRequest extends ModelSelection { readonly sessionId: SessionId @@ -352,22 +384,28 @@ export interface SessionPageRequest { readonly maxMessages?: number } -/** One live event request, optionally resuming after an already-applied event. */ +/** One live event request for a durable Session address. */ export interface SessionFollowRequest { readonly address: SessionAddress - readonly afterSeq?: number + readonly maxMessages?: number } /** One contiguous backwards page of a Session log. */ export interface SessionPage { readonly events: readonly SessionEventEntry[] readonly hasMore: boolean - readonly projections?: SessionProjectionsBlock } -/** Initial cursor followed by ordered events appended after that cursor. */ +/** Complete opening window followed by ordered events appended after its cursor. */ export type SessionFollowFrame = - | { readonly type: 'opened'; readonly cursor: number } + | { + readonly type: 'snapshot' + readonly header: SessionHeader + readonly cursor: number + readonly events: readonly SessionEventEntry[] + readonly hasMore: boolean + readonly projections: SessionProjectionBaseline + } | ({ readonly type: 'event' } & SessionEventEntry) /** One pending inbox occurrence in the authoritative queue snapshot. */ @@ -396,7 +434,7 @@ export interface SessionJob { export interface SessionControlBaseline { readonly queues: Readonly> readonly jobs: Readonly> - readonly projections: Readonly> + readonly projections: Readonly> } /** One finished projection value and its durable watermark. */ diff --git a/packages/api/session-controller/tests/session-history-journal.host.spec.ts b/packages/api/session-controller/tests/session-history-journal.host.spec.ts index b4c442d9f2..56b2fe5948 100644 --- a/packages/api/session-controller/tests/session-history-journal.host.spec.ts +++ b/packages/api/session-controller/tests/session-history-journal.host.spec.ts @@ -8,7 +8,7 @@ import { CallId, createMessage, createToolResultMessage, createUserMessage } fro import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session' import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts' import type { SessionFollowFrame } from '@deepseek-ai/dsh-api-session-controller/types' -import { createSessionTestRemote } from './test-remote.ts' +import { createSessionTestRemote, installSessionReadTestServices } from './test-remote.ts' /** Append a production-shaped human prompt to the session surface. */ function appendUserText(session: Session, text: string): SessionEvent { @@ -43,6 +43,7 @@ async function harness(): Promise<{ ctx: Context }> { const ctx = new Context() await ctx.plugin(SessionStore) await ctx.plugin(AgentRegistry) + installSessionReadTestServices(ctx) return { ctx } } @@ -71,7 +72,7 @@ async function openFollow( }, signal)[Symbol.asyncIterator]() await expect(iterator.next()).resolves.toMatchObject({ done: false, - value: { type: 'opened' }, + value: { type: 'snapshot' }, }) return { [Symbol.asyncIterator]: () => iterator } } @@ -79,8 +80,8 @@ async function openFollow( describe('Session history raw journal', () => { it('follows raw tool events and preserves result metadata without a Tools service', async () => { const { ctx } = await harness() - const session = ctx.sessions.create() - const history = new SessionHistoryController(ctx) + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) + const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() }) const abort = new AbortController() const stream = await openFollow(history, session.id, abort.signal) const collected = collect(stream, 2, abort) @@ -108,8 +109,8 @@ describe('Session history raw journal', () => { it('follows live results without rescanning Session history', async () => { const { ctx } = await harness() - const session = ctx.sessions.create() - const history = new SessionHistoryController(ctx) + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) + const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() }) const abort = new AbortController() const stream = await openFollow(history, session.id, abort.signal) const iterator = stream[Symbol.asyncIterator]() @@ -147,7 +148,7 @@ describe('Session history raw journal', () => { it('serves raw call and result entries without parsing tool arguments', async () => { const { ctx } = await harness() const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' }) - const session = ctx.sessions.create() + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) const start = session.append('turn/start', { turn: 1 }) const call = session.append('tool/call', { turn: 1, step: 1, callId: CallId('history-call'), name: 'custom', arguments: '{broken', @@ -178,7 +179,7 @@ describe('Session history raw journal', () => { it('counts only append-origin messages toward maxMessages and keeps each compaction summary with its replacement', async () => { const { ctx } = await harness() const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' }) - const session = ctx.sessions.create() + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) session.append('turn/start', { turn: 1 }) const first = appendUserText(session, 'first prompt') appendAssistantText(session, 'first reply', 1) @@ -227,7 +228,7 @@ describe('Session history raw journal', () => { it('paginates a message with many provenance sources without variadic argument expansion', async () => { const { ctx } = await harness() const remote = createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' }) - const session = ctx.sessions.create() + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) session.append('turn/start', { turn: 1 }) const sources = Array.from({ length: 128 }, (_unused, index) => session.append('assistant/chunk', { turn: 1, @@ -265,8 +266,8 @@ describe('Session history raw journal', () => { it('follows a result after turn/end without reading the addressed Session log', async () => { const { ctx } = await harness() - const session = ctx.sessions.create() - const history = new SessionHistoryController(ctx) + const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } }) + const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() }) const abort = new AbortController() const stream = await openFollow(history, session.id, abort.signal) const iterator = stream[Symbol.asyncIterator]() diff --git a/packages/api/session-controller/tests/session.client.spec.ts b/packages/api/session-controller/tests/session.client.spec.ts index aa8f26de24..79fac7b7b8 100644 --- a/packages/api/session-controller/tests/session.client.spec.ts +++ b/packages/api/session-controller/tests/session.client.spec.ts @@ -69,11 +69,12 @@ describe('Session open', () => { expect(session.eventSource.getSnapshot().change).toMatchObject({ kind: 'replace' }) }) - it('is idempotent: concurrent opens share one history call, reopening when open is a no-op', async () => { + it('is idempotent: concurrent opens share one follow, reopening when open is a no-op', async () => { const { api, session } = makeSession() await Promise.all([session.open(), session.open()]) await session.open() - expect(api.callsOf('session.history')).toHaveLength(1) + expect(api.callsOf('session.follow')).toHaveLength(1) + expect(api.callsOf('session.history')).toEqual([]) }) it('lands an error result in openState=error with the RpcError kept', async () => { @@ -98,7 +99,7 @@ describe('Session open', () => { const gate = deferred>>() api.onHistory = () => gate.promise const opening = session.open() - // Three live frames land mid-open; seq 15 overlaps the page tail (page covers 10..15). + // Three live frames land while the opening snapshot is pending; seq 15 overlaps its tail. const page = plainTurn(10, 0, '早', '安') const deliveries = [ follow(api, ev.turnStart(15, 1)), @@ -151,7 +152,7 @@ describe('live event path', () => { // seq 9 with tail 5 → gap; the event detours to the buffer and one history refetch fires. await follow(api, ev.assistant(9, 1, 'd')) await vi.waitFor(() => { - expect(api.callsOf('session.history').length).toBe(2) + expect(api.callsOf('session.history')).toHaveLength(1) }) await vi.waitFor(() => { expect(eventSeqs(session)).toEqual( @@ -172,8 +173,8 @@ describe('paging', () => { await session.open() await session.loadOlder() const snapshot = session.getSnapshot() + expect(api.callsOf('session.follow')).toHaveLength(1) expect(api.callsOf('session.history')).toMatchObject([ - { sessionId: SID, throughSeq: 11 }, { sessionId: SID, throughSeq: 11, beforeSeq: 6 }, ]) expect(snapshot.hasMore).toBe(false) @@ -231,7 +232,8 @@ describe('paging', () => { modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' }, })) await Promise.all([first, second]) - expect(api.callsOf('session.history')).toHaveLength(2) // open + one page, not two + expect(api.callsOf('session.follow')).toHaveLength(1) + expect(api.callsOf('session.history')).toHaveLength(1) }) }) @@ -248,9 +250,15 @@ describe('prompt and cancel errors', () => { expect(prompted).toEqual({ ok: true, value: { accepted: true } }) expect(cancelled).toEqual({ ok: true, value: { accepted: true } }) - expect(api.callsOf('subagent.history')).toEqual([ - { parentSessionId: PARENT, childSessionId: SID, mode: 'continuable', throughSeq: -1, maxMessages: 50 }, + expect(api.callsOf('session.follow')).toEqual([ + { + address: { + kind: 'subagent', parentSessionId: PARENT, childSessionId: SID, mode: 'continuable', + }, + maxMessages: 50, + }, ]) + expect(api.callsOf('subagent.history')).toEqual([]) expect(api.callsOf('subagent.prompt')).toEqual([ { parentSessionId: PARENT, childSessionId: SID, mode: 'continuable', @@ -300,9 +308,15 @@ describe('prompt and cancel errors', () => { expect(prompted).toMatchObject({ ok: false, error: { code: 'subagent-not-resumable' } }) expect(cancelled).toMatchObject({ ok: false, error: { code: 'subagent-delivery-unavailable' } }) - expect(api.callsOf('subagent.history')).toEqual([ - { parentSessionId: PARENT, childSessionId: SID, mode: 'one-shot', throughSeq: -1, maxMessages: 50 }, + expect(api.callsOf('session.follow')).toEqual([ + { + address: { + kind: 'subagent', parentSessionId: PARENT, childSessionId: SID, mode: 'one-shot', + }, + maxMessages: 50, + }, ]) + expect(api.callsOf('subagent.history')).toEqual([]) expect(api.callsOf('subagent.prompt')).toEqual([]) expect(api.callsOf('subagent.interrupt')).toEqual([]) expect(api.callsOf('session.cancel')).toEqual([]) @@ -572,7 +586,7 @@ describe('remaining branches', () => { const repairPull = deferred>>() api.onHistory = () => repairPull.promise const delivery = follow(api, ev.user(9, '洞')) - await vi.waitFor(() => { expect(api.callsOf('session.history')).toHaveLength(2) }) + await vi.waitFor(() => { expect(api.callsOf('session.history')).toHaveLength(1) }) api.onHistory = () => histResponse(plainTurn(6, 1, 'c', 'd')) const resynced = session.resync() // bumps the generation repairPull.resolve(ok({ @@ -626,7 +640,7 @@ describe('remaining branches', () => { }) describe('resync', () => { - it('keeps the old feed until one sorted page-and-live replacement is ready', async () => { + it('keeps the old feed until the reconnect snapshot, then repairs queued live gaps', async () => { const { api, session } = makeSession() api.onHistory = () => histResponse(plainTurn(0, 0, '旧', '窗')) await session.open() @@ -640,11 +654,16 @@ describe('resync', () => { }) const syncing = session.resync() - await vi.waitFor(() => { expect(api.callsOf('session.history')).toHaveLength(2) }) + await vi.waitFor(() => { expect(api.callsOf('session.follow')).toHaveLength(2) }) expect(session.eventSource.getSnapshot()).toBe(oldWindow) expect(publications).toEqual([]) - await Promise.all([ + api.onHistory = () => histResponse([ + ...plainTurn(10, 2, '终', '页'), + ev.user(16, '后到低位'), + ev.user(17, '后到高位'), + ]) + const liveDeliveries = Promise.all([ follow(api, ev.user(17, '后到高位')), follow(api, ev.user(16, '后到低位')), ]) @@ -654,12 +673,15 @@ describe('resync', () => { hasMore: false, modelSelection: { provider: 'deepseek-official', model: 'deepseek-v4-flash' }, })) - await syncing + await Promise.all([syncing, liveDeliveries]) + await vi.waitFor(() => { + expect(eventSeqs(session)).toEqual([10, 11, 12, 13, 14, 15, 16, 17]) + }) - expect(publications).toHaveLength(1) - expect(publications[0]?.entries).not.toHaveLength(0) - expect(publications[0]?.change.kind).toBe('replace') - expect(eventSeqs(session)).toEqual([10, 11, 12, 13, 14, 15, 16, 17]) + expect(publications).toHaveLength(2) + expect(publications.map(snapshot => snapshot.change.kind)).toEqual(['replace', 'replace']) + expect(publications[0]?.entries.map(entry => entry.event.seq)).toEqual([10, 11, 12, 13, 14, 15]) + expect(publications[1]?.entries.map(entry => entry.event.seq)).toEqual([10, 11, 12, 13, 14, 15, 16, 17]) off() }) diff --git a/packages/api/session-controller/tests/transport.client.spec.ts b/packages/api/session-controller/tests/transport.client.spec.ts index 46c2064c9e..3960e05285 100644 --- a/packages/api/session-controller/tests/transport.client.spec.ts +++ b/packages/api/session-controller/tests/transport.client.spec.ts @@ -43,6 +43,25 @@ function page(events: readonly SessionEventEntry[], hasMore = false): SessionPag return { events, hasMore } } +function snapshot( + cursor: number, + events: readonly SessionEventEntry[], + hasMore = false, +): SessionFollowFrame { + return { + type: 'snapshot', + header: { + version: 0, + id: ADDRESS.kind === 'session' ? ADDRESS.sessionId : ADDRESS.childSessionId, + createdAt: 0, + }, + cursor, + events, + hasMore, + projections: { asOfSeq: cursor, values: {} }, + } +} + function sessionClient(remote: SessionTransportRemote) { return { session: remote as SessionRemote, @@ -108,14 +127,13 @@ describe('Session Client stream adapters', () => { const remote = new ScriptedSessionRemote( [{ frames: [ - { type: 'opened', cursor: 3 }, + snapshot(3, [entry(2), entry(3)], true), { type: 'event', ...entry(3) }, { type: 'event', ...entry(4) }, ], hold: true, }], [ - { ok: true, value: page([entry(2), entry(3)], true) }, { ok: true, value: page([entry(0), entry(1)], false) }, ], ) @@ -129,9 +147,8 @@ describe('Session Client stream adapters', () => { await vi.waitFor(() => { expect(changes).toHaveLength(2) }) await stream.prepend({ beforeSeq: 2, maxMessages: 50 }) - expect(remote.followRequests).toEqual([{ address: ADDRESS }]) + expect(remote.followRequests).toEqual([{ address: ADDRESS, maxMessages: 50 }]) expect(remote.pageRequests).toEqual([ - { address: ADDRESS, throughSeq: 3, maxMessages: 50 }, { address: ADDRESS, throughSeq: 4, beforeSeq: 2, maxMessages: 50 }, ]) expect(changes).toMatchObject([ @@ -143,20 +160,17 @@ describe('Session Client stream adapters', () => { expect(remote.signals[0]?.aborted).toBe(true) }) - it('resumes after the applied cursor and repairs through the addressed tail page', async () => { + it('replaces the retained window from each reconnect snapshot', async () => { const lost = new RemoteStreamCarrierError('lost') const remote = new ScriptedSessionRemote( [ { - frames: [{ type: 'opened', cursor: 1 }, { type: 'event', ...entry(2) }], + frames: [snapshot(1, [entry(0), entry(1)]), { type: 'event', ...entry(2) }], terminal: lost, }, - { frames: [{ type: 'opened', cursor: 4 }], hold: true }, - ], - [ - { ok: true, value: page([entry(0), entry(1)]) }, - { ok: true, value: page([entry(0), entry(1), entry(2), entry(3), entry(4)]) }, + { frames: [snapshot(4, [entry(0), entry(1), entry(2), entry(3), entry(4)])], hold: true }, ], + [], ) const changes: SessionJournalChange[] = [] const carrierFailed = vi.fn() @@ -170,13 +184,10 @@ describe('Session Client stream adapters', () => { await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) }) expect(remote.followRequests).toEqual([ - { address: ADDRESS }, - { address: ADDRESS, afterSeq: 2 }, - ]) - expect(remote.pageRequests).toEqual([ - { address: ADDRESS, throughSeq: 1, maxMessages: 50 }, - { address: ADDRESS, throughSeq: 4, maxMessages: 50 }, + { address: ADDRESS, maxMessages: 50 }, + { address: ADDRESS, maxMessages: 50 }, ]) + expect(remote.pageRequests).toEqual([]) expect(changes.map(change => change.type)).toEqual(['replace', 'append', 'replace']) expect(carrierFailed).toHaveBeenCalledWith(lost) await stream.dispose() @@ -187,16 +198,13 @@ describe('Session Client stream adapters', () => { const remote = new ScriptedSessionRemote( [ { - frames: [{ type: 'opened', cursor: 0 }], + frames: [snapshot(0, [entry(0)])], waitAfterFrames: finish.promise, terminal: new RemoteStreamCarrierError('lost'), }, - { frames: [{ type: 'opened', cursor: 1 }], hold: true }, - ], - [ - { ok: true, value: page([entry(0)]) }, - { ok: true, value: page([entry(0), entry(1)]) }, + { frames: [snapshot(1, [entry(0), entry(1)])], hold: true }, ], + [], ) const stream = new SessionEventStream(sessionClient(remote), ADDRESS, { publish: vi.fn(), @@ -205,18 +213,33 @@ describe('Session Client stream adapters', () => { await stream.open({}) finish.resolve(undefined) - await vi.waitFor(() => { expect(remote.pageRequests).toHaveLength(2) }) - expect(remote.pageRequests).toEqual([ - { address: ADDRESS, throughSeq: 0 }, - { address: ADDRESS, throughSeq: 1 }, - ]) + await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) }) + expect(remote.followRequests).toEqual([{ address: ADDRESS }, { address: ADDRESS }]) + expect(remote.pageRequests).toEqual([]) await stream.dispose() }) - it('turns a page failure into a typed stream failure and closes follow', async () => { + it('repairs a live gap without adding an absent message limit', async () => { + const remote = new ScriptedSessionRemote( + [{ frames: [snapshot(0, [entry(0)]), { type: 'event', ...entry(2) }], hold: true }], + [{ ok: true, value: page([entry(0), entry(1), entry(2)]) }], + ) + const changes: SessionJournalChange[] = [] + const stream = new SessionEventStream(sessionClient(remote), ADDRESS, { + publish: (change) => { changes.push(change) }, + failed: vi.fn(), + }) + + await stream.open({}) + await vi.waitFor(() => { expect(changes).toHaveLength(2) }) + expect(remote.pageRequests).toEqual([{ address: ADDRESS, throughSeq: 2 }]) + await stream.dispose() + }) + + it('turns a pagination failure into a typed stream failure', async () => { const failure = { code: 'session-not-found', message: 'missing', details: { sessionId: 'session-1' } } as const const remote = new ScriptedSessionRemote( - [{ frames: [{ type: 'opened', cursor: -1 }], hold: true }], + [{ frames: [snapshot(-1, [])], hold: true }], [{ ok: false, error: failure }], ) const stream = new SessionEventStream(sessionClient(remote), ADDRESS, { @@ -224,13 +247,16 @@ describe('Session Client stream adapters', () => { failed: vi.fn(), }) - await expect(stream.open({})).rejects.toBeInstanceOf(RemoteStreamError) + await stream.open({}) + await expect(stream.prepend({})).rejects.toBeInstanceOf(RemoteStreamError) await expect(stream.open({})).rejects.toThrow('already opened') expect(sessionStreamFailure(new RemoteStreamError(failure.code, failure.message, failure.details))) .toEqual(failure) expect(sessionStreamFailure(new Error('local'))).toBeUndefined() - expect(remote.signals[0]?.aborted).toBe(true) + expect(remote.signals[0]?.aborted).toBe(false) expect(remote.pageRequests).toEqual([{ address: ADDRESS, throughSeq: -1 }]) + await stream.dispose() + expect(remote.signals[0]?.aborted).toBe(true) }) it('maps the Host-wide control baseline and deltas into one snapshot stream', async () => { diff --git a/packages/api/session-controller/tests/transport.host.spec.ts b/packages/api/session-controller/tests/transport.host.spec.ts index c30d3f095f..65338e27ae 100644 --- a/packages/api/session-controller/tests/transport.host.spec.ts +++ b/packages/api/session-controller/tests/transport.host.spec.ts @@ -2,9 +2,12 @@ import { Context } from '@deepseek-ai/cordis' import { createScope } from '@deepseek-ai/dsh-scope' import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' +import type { SessionObservation } from '@deepseek-ai/dsh-session-query' import { snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent' +import { subagentIdentityProjectionDefinition } from '@deepseek-ai/dsh-subagent/src/projection.ts' import { describe, expect, it, vi } from 'vitest' import { SessionHistoryController } from '../src/history.ts' +import { installSessionReadTestServices, testSessionPersistence } from './test-remote.ts' const signal = (): AbortSignal => new AbortController().signal @@ -22,7 +25,13 @@ function append( } function event(type: string, seq: number, data: unknown = {}): SessionEvent { - return { type, seq, time: seq + 1, data } as SessionEvent + return { + type, + seq, + time: seq + 1, + data, + ...type.startsWith('fixture/') ? { ignorable: true } : {}, + } as SessionEvent } function cold( @@ -30,10 +39,10 @@ function cold( header: SessionHeader, events: readonly SessionEvent[], ): void { - ctx.provide('sessionPersistence', { + ctx.provide('sessionPersistence', testSessionPersistence(ctx, { list: () => Promise.resolve([header]), inspect: () => Promise.resolve({ meta: header, events }), - } as never) + }) as never) } interface Deferred { @@ -50,7 +59,9 @@ function deferred(): Deferred { async function setup(): Promise<{ ctx: Context; transport: SessionHistoryController }> { const ctx = new Context() await ctx.plugin(SessionStore) - const transport = new SessionHistoryController(ctx) + installSessionReadTestServices(ctx) + ctx.sessionProjections.register(subagentIdentityProjectionDefinition) + const transport = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() }) return { ctx, transport } } @@ -65,7 +76,7 @@ describe('SessionHistoryController', () => { abort.signal, )[Symbol.asyncIterator]() - expect(await iterator.next()).toMatchObject({ done: false, value: { type: 'opened', cursor: 0 } }) + expect(await iterator.next()).toMatchObject({ done: false, value: { type: 'snapshot', cursor: 0 } }) session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) expect(await iterator.next()).toMatchObject({ done: false, @@ -85,10 +96,13 @@ describe('SessionHistoryController', () => { it('ends active followers when the owning Controller unloads', async () => { const ctx = new Context() await ctx.plugin(SessionStore) + installSessionReadTestServices(ctx) let transport!: SessionHistoryController const owner = ctx.plugin(Object.assign( - (inner: Context) => { transport = new SessionHistoryController(inner) }, - { inject: ['sessions'] }, + (inner: Context) => { + transport = new SessionHistoryController(inner, (observation) => { observation[Symbol.dispose]() }) + }, + { inject: ['sessions', 'sessionQuery'] }, )) await owner.await() const session = ctx.sessions.create(SessionId('controller-unload'), { meta: { cwd: '/workspace' } }) @@ -97,9 +111,9 @@ describe('SessionHistoryController', () => { new AbortController().signal, )[Symbol.asyncIterator]() - await expect(iterator.next()).resolves.toEqual({ + await expect(iterator.next()).resolves.toMatchObject({ done: false, - value: { type: 'opened', cursor: -1 }, + value: { type: 'snapshot', cursor: -1 }, }) const pending = iterator.next() await owner.dispose() @@ -107,7 +121,7 @@ describe('SessionHistoryController', () => { await ctx.fiber.dispose() }) - it('resumes from the last applied seq before delivering later live events', async () => { + it('reconnects with a complete replacement snapshot before later live events', async () => { const { ctx, transport } = await setup() const session = ctx.sessions.create(SessionId('resume'), { meta: { cwd: '/workspace' } }) session.append('turn/start', { turn: 1 }) @@ -116,12 +130,16 @@ describe('SessionHistoryController', () => { const abort = new AbortController() const iterator = transport.follow({ address: { kind: 'session', sessionId: session.id }, - afterSeq: 0, }, abort.signal)[Symbol.asyncIterator]() - expect(await iterator.next()).toEqual({ done: false, value: { type: 'opened', cursor: 2 } }) - expect(await iterator.next()).toMatchObject({ done: false, value: { type: 'event', event: { seq: 1 } } }) - expect(await iterator.next()).toMatchObject({ done: false, value: { type: 'event', event: { seq: 2 } } }) + expect(await iterator.next()).toMatchObject({ + done: false, + value: { + type: 'snapshot', + cursor: 2, + events: [{ event: { seq: 0 } }, { event: { seq: 1 } }, { event: { seq: 2 } }], + }, + }) session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }) expect(await iterator.next()).toMatchObject({ done: false, value: { type: 'event', event: { seq: 3 } } }) @@ -133,34 +151,74 @@ describe('SessionHistoryController', () => { const { ctx, transport } = await setup() const sessionId = SessionId('cold-race') const header = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' } - const listed = deferred() - ctx.provide('sessionPersistence', { - list: () => listed.promise, - inspect: () => Promise.resolve({ meta: header, events: [event('fixture/start', 0)] }), - } as never) + const inspected = deferred<{ meta: SessionHeader; events: readonly SessionEvent[] }>() + ctx.provide('sessionPersistence', testSessionPersistence(ctx, { + inspect: () => inspected.promise, + }) as never) const abort = new AbortController() const iterator = transport.follow({ address: { kind: 'session', sessionId } }, abort.signal) [Symbol.asyncIterator]() const opening = iterator.next() - ctx.emit('session/event', { id: SessionId('unrelated') } as Session, event('fixture/other', 0)) - ctx.emit('session/event', { id: sessionId } as Session, event('fixture/start', 0)) - listed.resolve([header]) - await expect(opening).resolves.toEqual({ done: false, value: { type: 'opened', cursor: 0 } }) + ctx.emit('session/event', { + id: SessionId('unrelated'), events: [event('fixture/other', 0)], + } as unknown as Session, event('fixture/other', 0)) + ctx.emit('session/event', { + id: sessionId, events: [event('fixture/start', 0)], + } as unknown as Session, event('fixture/start', 0)) + inspected.resolve({ meta: header, events: [event('fixture/start', 0)] }) + await expect(opening).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: 0 } }) const waiting = iterator.next() abort.abort() await expect(waiting).resolves.toMatchObject({ done: true }) }) + it('buffers creation while the opening observation is unresolved', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const sessionId = SessionId('created-during-observation') + const header = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' } + const observed = deferred() + ctx.provide('sessionQuery', { observeSession: () => observed.promise } as never) + const transport = new SessionHistoryController(ctx, vi.fn()) + const abort = new AbortController() + const iterator = transport.follow({ address: { kind: 'session', sessionId } }, abort.signal) + [Symbol.asyncIterator]() + const opening = iterator.next() + + const attached = ctx.sessions.create(sessionId, { meta: header, seed: [event('fixture/seed', 0)] }) + observed.resolve({ + source: 'live', + header: attached.header, + events: attached.events, + cursor: attached.seq - 1, + projections: { asOfSeq: attached.seq - 1, values: {} }, + retain: vi.fn(), + [Symbol.dispose]: vi.fn(), + } as unknown as SessionObservation) + await expect(opening).resolves.toMatchObject({ + done: false, + value: { + type: 'snapshot', cursor: 1, events: [{ event: { seq: 0 } }, { event: { seq: 1 } }], + }, + }) + expect(attached.id).toBe(sessionId) + abort.abort() + await expect(iterator.next()).resolves.toMatchObject({ done: true }) + }) + it('bridges the unpublished end-seed boundary when a cold source attaches', async () => { const ctx = new Context() await ctx.plugin(SessionStore) + installSessionReadTestServices(ctx) let transport!: SessionHistoryController let agentCtx!: Context await ctx.plugin(Object.assign( - (inner: Context) => { transport = new SessionHistoryController(inner) }, - { inject: ['sessions'] }, + (inner: Context) => { + transport = new SessionHistoryController(inner, (observation) => { observation[Symbol.dispose]() }) + }, + { inject: ['sessions', 'sessionQuery'] }, )) await ctx.plugin(Object.assign( (inner: Context) => { agentCtx = createScope(inner, { name: 'agent' }).ctx }, @@ -179,7 +237,7 @@ describe('SessionHistoryController', () => { const iterator = transport.follow({ address: { kind: 'session', sessionId } }, abort.signal) [Symbol.asyncIterator]() - await expect(iterator.next()).resolves.toEqual({ done: false, value: { type: 'opened', cursor: 0 } }) + await expect(iterator.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: 0 } }) agentCtx.sessions.create(SessionId('unrelated-created'), { meta: { cwd: '/workspace' } }) const attached = agentCtx.sessions.prepare(sessionId, { meta: header, seed }) agentCtx.sessions.enter(attached) @@ -212,11 +270,9 @@ describe('SessionHistoryController', () => { const replayHeader = { version: 0, id: replayId, createdAt: 1, cwd: '/workspace' } cold(replay.ctx, replayHeader, [event('fixture/start', 0), event('fixture/gap', 2)]) const replayed = replay.transport.follow({ - address: { kind: 'session', sessionId: replayId }, afterSeq: -1, + address: { kind: 'session', sessionId: replayId }, }, signal())[Symbol.asyncIterator]() - await expect(replayed.next()).resolves.toEqual({ done: false, value: { type: 'opened', cursor: 2 } }) - await expect(replayed.next()).resolves.toMatchObject({ done: false, value: { event: { seq: 0 } } }) - await expect(replayed.next()).rejects.toMatchObject({ failure: { code: 'internal' } }) + await expect(replayed.next()).rejects.toMatchObject({ code: 'SESSION_QUERY_CORRUPT_SESSION' }) const live = await setup() const session = live.ctx.sessions.create(SessionId('live-gap'), { meta: { cwd: '/workspace' } }) @@ -225,8 +281,13 @@ describe('SessionHistoryController', () => { const followed = live.transport.follow({ address: { kind: 'session', sessionId: session.id }, }, signal())[Symbol.asyncIterator]() - await expect(followed.next()).resolves.toEqual({ done: false, value: { type: 'opened', cursor: 0 } }) - live.ctx.emit('session/event', session, event('fixture/gap', 2)) + await expect(followed.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: 0 } }) + const skipped = event('fixture/skipped', 1) + const gap = event('fixture/gap', 2) + live.ctx.emit('session/event', { + id: session.id, + events: [event('fixture/start', 0), skipped, gap], + } as unknown as Session, gap) await expect(followed.next()).rejects.toMatchObject({ failure: { code: 'internal' } }) }) @@ -237,7 +298,7 @@ describe('SessionHistoryController', () => { const iterator = transport.follow({ address: { kind: 'session', sessionId: session.id }, }, abort.signal)[Symbol.asyncIterator]() - await expect(iterator.next()).resolves.toEqual({ done: false, value: { type: 'opened', cursor: -1 } }) + await expect(iterator.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: -1 } }) await expect(transport.page({ address: { kind: 'session', sessionId: session.id }, throughSeq: -1, }, signal())).resolves.toMatchObject({ events: [], hasMore: false }) @@ -245,6 +306,59 @@ describe('SessionHistoryController', () => { await expect(iterator.next()).resolves.toMatchObject({ done: true }) }) + it('publishes an empty projection baseline when the query has no registry', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const sessionId = SessionId('projectionless-follow') + const meta = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' } + ctx.provide('sessionQuery', { + observeSession: () => Promise.resolve({ + source: 'live', header: meta, events: [], cursor: -1, + retain: vi.fn(), [Symbol.dispose]: vi.fn(), + } satisfies SessionObservation), + } as never) + const history = new SessionHistoryController(ctx, vi.fn()) + const abort = new AbortController() + const iterator = history.follow({ address: { kind: 'session', sessionId } }, abort.signal) + [Symbol.asyncIterator]() + + await expect(iterator.next()).resolves.toMatchObject({ + value: { type: 'snapshot', projections: { asOfSeq: -1, values: {} } }, + }) + abort.abort() + await expect(iterator.next()).resolves.toMatchObject({ done: true }) + await ctx.fiber.dispose() + }) + + it('disposes a retained promotion when background activation rejects synchronously', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const sessionId = SessionId('promotion-failure') + const meta = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' } + const disposePromotion = vi.fn() + const promotion = { + source: 'prepared', header: meta, events: [], cursor: -1, + projections: { asOfSeq: -1, values: {} }, + retain: vi.fn(), [Symbol.dispose]: disposePromotion, + } as unknown as SessionObservation + const source = { + ...promotion, + retain: () => promotion, + [Symbol.dispose]: vi.fn(), + } as SessionObservation + ctx.provide('sessionQuery', { + observeSession: () => Promise.resolve(source), + } as never) + const history = new SessionHistoryController(ctx, () => { throw new Error('activation failed') }) + const iterator = history.follow({ address: { kind: 'session', sessionId } }, signal()) + [Symbol.asyncIterator]() + + await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'snapshot' } }) + await expect(iterator.next()).rejects.toThrow('activation failed') + expect(disposePromotion).toHaveBeenCalledOnce() + await ctx.fiber.dispose() + }) + it('requires the durable parent and mode for a direct subagent address', async () => { const { ctx, transport } = await setup() const parentSessionId = SessionId('parent') @@ -288,15 +402,18 @@ describe('SessionHistoryController', () => { const sessionId = SessionId('corrupt-cold') const failure = new Error('cold log is corrupt') const header = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' } - ctx.provide('sessionPersistence', { + ctx.provide('sessionPersistence', testSessionPersistence(ctx, { list: () => Promise.resolve([header]), inspect: () => Promise.reject(failure), - } as never) + }) as never) await expect(transport.page({ address: { kind: 'session', sessionId }, throughSeq: -1, - }, new AbortController().signal)).rejects.toBe(failure) + }, new AbortController().signal)).rejects.toMatchObject({ + code: 'SESSION_QUERY_PERSISTENCE_FAILED', + cause: failure, + }) }) it('rejects malformed page and follow cursors at the service boundary', async () => { @@ -325,25 +442,24 @@ describe('SessionHistoryController', () => { ) await expect(corrupt.transport.page({ address: { kind: 'session', sessionId: corruptId }, throughSeq: 1, - }, signal())).rejects.toMatchObject({ failure: { code: 'internal' } }) - for (const afterSeq of [-2, 0.5]) { - const iterator = transport.follow({ address, afterSeq }, signal())[Symbol.asyncIterator]() + }, signal())).rejects.toMatchObject({ code: 'SESSION_QUERY_CORRUPT_SESSION' }) + for (const maxMessages of [0, 0.5]) { + const iterator = transport.follow({ address, maxMessages }, signal())[Symbol.asyncIterator]() await expect(iterator.next()).rejects.toMatchObject({ failure: { code: 'bad-request' } }) } - const past = transport.follow({ address, afterSeq: 0 }, signal())[Symbol.asyncIterator]() - await expect(past.next()).rejects.toMatchObject({ failure: { code: 'bad-request' } }) }) it('reports missing ordinary and subagent sources without fabricating inspection failures', async () => { const { ctx, transport } = await setup() const ordinary = { kind: 'session' as const, sessionId: SessionId('missing') } await expect(transport.page({ address: ordinary, throughSeq: -1 }, signal())) - .rejects.toMatchObject({ failure: { code: 'internal' } }) + .rejects.toMatchObject({ failure: { code: 'session-not-found' } }) - ctx.provide('sessionPersistence', { + const inspect = vi.fn(() => Promise.resolve(undefined)) + ctx.provide('sessionPersistence', testSessionPersistence(ctx, { list: () => Promise.resolve([]), - inspect: () => Promise.reject(new Error('must not inspect')), - } as never) + inspect, + }) as never) await expect(transport.page({ address: ordinary, throughSeq: -1 }, signal())) .rejects.toMatchObject({ failure: { code: 'session-not-found' } }) await expect(transport.page({ @@ -355,25 +471,28 @@ describe('SessionHistoryController', () => { }, throughSeq: -1, }, signal())).rejects.toMatchObject({ failure: { code: 'subagent-not-found' } }) + expect(inspect).toHaveBeenCalledTimes(2) }) it('rejects incomplete cold metadata before serving a source', async () => { const first = await setup() const sessionId = SessionId('incomplete') const address = { kind: 'session' as const, sessionId } - first.ctx.provide('sessionPersistence', { - list: () => Promise.resolve([{ version: 0, id: sessionId, createdAt: 1 }]), - inspect: () => Promise.reject(new Error('must not inspect')), - } as never) + const firstHeader = { version: 0, id: sessionId, createdAt: 1 } + first.ctx.provide('sessionPersistence', testSessionPersistence(first.ctx, { + list: () => Promise.resolve([firstHeader]), + inspect: () => Promise.resolve({ meta: firstHeader, events: [] }), + }) as never) await expect(first.transport.page({ address, throughSeq: -1 }, signal())) .rejects.toMatchObject({ failure: { code: 'session-not-found' } }) const second = await setup() const listed = { version: 0, id: sessionId, createdAt: 1, cwd: '/workspace' } - second.ctx.provide('sessionPersistence', { + const inspected = { version: 0, id: sessionId, createdAt: 1 } + second.ctx.provide('sessionPersistence', testSessionPersistence(second.ctx, { list: () => Promise.resolve([listed]), - inspect: () => Promise.resolve({ meta: { ...listed, cwd: undefined }, events: [] }), - } as never) + inspect: () => Promise.resolve({ meta: inspected, events: [] }), + }) as never) await expect(second.transport.page({ address, throughSeq: -1 }, signal())) .rejects.toMatchObject({ failure: { code: 'session-not-found' } }) }) @@ -407,7 +526,7 @@ describe('SessionHistoryController', () => { const missing = await setup() cold(missing.ctx, childHeader, []) await expect(missing.transport.page({ address: childAddress, throughSeq: -1 }, signal())) - .rejects.toMatchObject({ failure: { code: 'subagent-catalog-diagnostic', details: { reason: 'unsupported' } } }) + .rejects.toMatchObject({ failure: { code: 'subagent-catalog-diagnostic', details: { reason: 'corrupt' } } }) const corrupt = await setup() cold(corrupt.ctx, childHeader, [event('subagent/descriptor', 0, { version: 'bad' })]) @@ -421,44 +540,48 @@ describe('SessionHistoryController', () => { .rejects.toMatchObject({ failure: { code: 'subagent-unauthorized' } }) }) - it('uses attached and detached projection cuts and isolates a child projection failure', async () => { - const attached = await setup() - const session = attached.ctx.sessions.create(SessionId('projected'), { meta: { cwd: '/workspace' } }) + it('reports an unavailable descriptor when an observed child has no projection value', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const parentSessionId = SessionId('missing-projection-parent') + const childSessionId = SessionId('missing-projection-child') + const meta: SessionHeader = { + version: 0, + id: childSessionId, + createdAt: 1, + cwd: '/workspace', + origin: 'subagent', + parentSession: parentSessionId, + } + ctx.provide('sessionQuery', { + observeSession: () => Promise.resolve({ + source: 'live', header: meta, events: [], cursor: -1, + projections: { asOfSeq: -1, values: {} }, + retain: vi.fn(), [Symbol.dispose]: vi.fn(), + } as unknown as SessionObservation), + } as never) + const history = new SessionHistoryController(ctx, vi.fn()) + + await expect(history.page({ + address: { kind: 'subagent', parentSessionId, childSessionId, mode: 'continuable' }, + throughSeq: -1, + }, signal())).rejects.toMatchObject({ + failure: { code: 'subagent-catalog-diagnostic', details: { reason: 'unsupported' } }, + }) + await ctx.fiber.dispose() + }) + + it('keeps pages projection-free and computes projections only for child authorization', async () => { + const ordinary = await setup() + const session = ordinary.ctx.sessions.create(SessionId('projected'), { meta: { cwd: '/workspace' } }) session.append('turn/start', { turn: 1 }) - const snapshot = vi.fn(() => ({ asOfSeq: 0, values: { title: 'attached' } })) - attached.ctx.provide('sessionProjections', { snapshot, restore: vi.fn() } as never) - await expect(attached.transport.page({ + const ordinarySnapshot = vi.spyOn(ordinary.ctx.sessionProjections, 'snapshot') + const ordinaryPage = await ordinary.transport.page({ address: { kind: 'session', sessionId: session.id }, throughSeq: 0, - }, signal())).resolves.toMatchObject({ projections: { asOfSeq: 0, values: { title: 'attached' } } }) - expect(snapshot).toHaveBeenCalledWith(session) - const older = await attached.transport.page({ - address: { kind: 'session', sessionId: session.id }, throughSeq: 0, beforeSeq: 1, }, signal()) - expect('projections' in older).toBe(false) - - const detached = await setup() - const coldId = SessionId('projected-cold') - const header = { version: 0, id: coldId, createdAt: 1, cwd: '/workspace' } - cold(detached.ctx, header, [event('turn/start', 0, { turn: 1 })]) - const restore = vi.fn(() => ({ snapshot: { asOfSeq: 0, values: { title: 'cold' } } })) - detached.ctx.provide('sessionProjections', { snapshot: vi.fn(), restore } as never) - await expect(detached.transport.page({ - address: { kind: 'session', sessionId: coldId }, - throughSeq: 0, - }, signal())).resolves.toMatchObject({ projections: { values: { title: 'cold' } } }) - expect(restore).toHaveBeenCalledWith({}, expect.any(Array), 0) - - const failed = await setup() - cold(failed.ctx, header, [event('turn/start', 0, { turn: 1 })]) - failed.ctx.provide('sessionProjections', { - snapshot: vi.fn(), - restore: () => { throw new Error('projection failed') }, - } as never) - await expect(failed.transport.page({ - address: { kind: 'session', sessionId: coldId }, - throughSeq: 0, - }, signal())).rejects.toThrow('projection failed') + expect('projections' in ordinaryPage).toBe(false) + expect(ordinarySnapshot).not.toHaveBeenCalled() const child = await setup() const parentSessionId = SessionId('projection-parent') @@ -469,17 +592,13 @@ describe('SessionHistoryController', () => { childSession.append('subagent/descriptor', snapshotSubagentDescriptor({ mode: 'continuable', provider: 'test', label: 'child', })) - const warn = vi.spyOn(child.ctx.logger, 'warn').mockImplementation(() => undefined) - child.ctx.provide('sessionProjections', { - snapshot: () => { throw new Error('child projection failed') }, - restore: vi.fn(), - } as never) + const childSnapshot = vi.spyOn(child.ctx.sessionProjections, 'snapshot') const page = await child.transport.page({ address: { kind: 'subagent', parentSessionId, childSessionId, mode: 'continuable' }, throughSeq: 0, }, signal()) expect('projections' in page).toBe(false) - expect(warn).toHaveBeenCalledWith(expect.stringContaining('child projection failed')) + expect(childSnapshot).toHaveBeenCalledWith(childSession) }) it('keeps message-aligned pagination contiguous across replacement provenance', async () => { diff --git a/packages/test-support/client-runtime/src/sessions.ts b/packages/test-support/client-runtime/src/sessions.ts index afb24fa287..cbd3b9a2e2 100644 --- a/packages/test-support/client-runtime/src/sessions.ts +++ b/packages/test-support/client-runtime/src/sessions.ts @@ -445,14 +445,6 @@ export class TestSessions implements ISessions { return Promise.resolve() } - /** Apply a confirmed preset switch into the fixture list, as production does. */ - noteAgentPreset(sessionId: SessionId, agentPreset: string): void { - this.list.update((draft) => { - const summary = draft.byId[sessionId] - if (summary !== undefined) draft.byId[sessionId] = { ...summary, agentPreset } - }) - } - /** Clear the current selection (recorded; the production no-session flow). */ clear(): void { this.calls.push({ method: 'clear', args: [] }) diff --git a/packages/test-support/client-runtime/tests/runtime.client.spec.tsx b/packages/test-support/client-runtime/tests/runtime.client.spec.tsx index f98b93bb0c..02d239b117 100644 --- a/packages/test-support/client-runtime/tests/runtime.client.spec.tsx +++ b/packages/test-support/client-runtime/tests/runtime.client.spec.tsx @@ -170,13 +170,6 @@ describe('sessions', () => { .toMatchObject({ displayTitle: 'renamed', running: true }) runtime.sessions.setSubagentCatalogOpen('s2' as SessionId, true) await runtime.sessions.refreshSubagents('s2' as SessionId) - // The confirmed-switch write-back lands on the row it names and ignores - // one the fixture never added, exactly as production's list upsert does. - runtime.sessions.noteAgentPreset('s1' as SessionId, 'minimal') - runtime.sessions.noteAgentPreset('missing' as SessionId, 'minimal') - await runtime.flush() - expect(runtime.sessions.list.getSnapshot().byId['s1' as SessionId]) - .toMatchObject({ agentPreset: 'minimal' }) runtime.sessions.open('s1' as SessionId) await runtime.flush() expect(runtime.sessions.list.getSnapshot().current).toBe('s1') diff --git a/packages/todo/tool-todo/package.json b/packages/todo/tool-todo/package.json index c1cc4c19de..e710fcfa30 100644 --- a/packages/todo/tool-todo/package.json +++ b/packages/todo/tool-todo/package.json @@ -54,7 +54,6 @@ "@deepseek-ai/dsh-agent": "workspace:^", "@deepseek-ai/dsh-agent-loop": "workspace:^", "@deepseek-ai/dsh-agent-loop-testkit": "workspace:^", - "@deepseek-ai/dsh-api-session-controller": "workspace:^", "@deepseek-ai/dsh-invariants": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", "@deepseek-ai/dsh-session": "workspace:^", diff --git a/packages/todo/tool-todo/tests/projection.spec.ts b/packages/todo/tool-todo/tests/projection.spec.ts index 592f158872..05a3662ef3 100644 --- a/packages/todo/tool-todo/tests/projection.spec.ts +++ b/packages/todo/tool-todo/tests/projection.spec.ts @@ -1,8 +1,8 @@ /** * The `todos` projection provider (session-projection RFC knife 4 — the "a * fourth domain is just its own registrations" acceptance probe): mounting - * tool-todo beside the registry serves the whole current list on the history - * tail page with a consistent asOfSeq (= last event seq); before any write the value is null; a + * tool-todo beside the registry serves the whole current list with a + * consistent asOfSeq (= last event seq); before any write the value is null; a * composition without tool-todo has no `todos` key; unmounting tool-todo * removes it (HMR safety). The carrier and framework are exercised unmodified. */ @@ -17,7 +17,6 @@ import type { Session } from '@deepseek-ai/dsh-session' import type { TodoItem } from '@deepseek-ai/dsh-tool-todo' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRuntime from '@deepseek-ai/dsh-tools' -import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts' import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection' import UserQuestionService from '@deepseek-ai/dsh-user-questions' import * as ToolTodo from '@deepseek-ai/dsh-tool-todo' @@ -39,16 +38,11 @@ async function harness(withTodoTool: boolean): Promise { if (withTodoTool) await ctx.plugin(ToolTodo, { allowParallelInProgress: true }) const session = ctx.sessions.create() ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent) - const history = new SessionHistoryController(ctx) return { ctx, session, async tailProjections() { - return (await history.page({ - address: { kind: 'session', sessionId: session.id }, - throughSeq: session.seq - 1, - }, new AbortController().signal)) - .projections + return ctx.sessionProjections.snapshot(session) }, } }