diff --git a/apps/web/tests/seeded-history.e2e.ts b/apps/web/tests/seeded-history.e2e.ts index e593352d6b..2b2d194657 100644 --- a/apps/web/tests/seeded-history.e2e.ts +++ b/apps/web/tests/seeded-history.e2e.ts @@ -182,6 +182,7 @@ describe('web e2e: seeded history renders through cold resume', () => { let browser: Browser let page: Page let tripwire: ReturnType + let seededThroughSeq = -1 beforeAll(async () => { scaffold = await launchWebScaffold({}) @@ -201,6 +202,7 @@ describe('web e2e: seeded history renders through cold resume', () => { const meter = scaffold.ctx.get('tokenMeter') if (meter === undefined) throw new Error('seeded-history requires the host token meter') const realizedWithCompaction = withCompaction(realizeSeedFixture(scaffold, raw, SEED_ID), meter) + seededThroughSeq = parseSeedFixture(realizedWithCompaction).events.at(-1)?.seq ?? -1 await seedSession(scaffold, realizedWithCompaction, SEED_ID) } browser = await chromium.launch() @@ -238,7 +240,10 @@ describe('web e2e: seeded history renders through cold resume', () => { body: JSON.stringify({ type: 'client-request', rpcId: 'seeded-projections', method: 'session/page', payload: { - args: { request: { address: { kind: 'session', sessionId: SEED_ID } } }, + args: { request: { + address: { kind: 'session', sessionId: SEED_ID }, + throughSeq: seededThroughSeq, + } }, }, }), }) diff --git a/apps/web/tests/smoke-real.e2e.ts b/apps/web/tests/smoke-real.e2e.ts index 60c6814824..d541eaba86 100644 --- a/apps/web/tests/smoke-real.e2e.ts +++ b/apps/web/tests/smoke-real.e2e.ts @@ -70,6 +70,85 @@ async function remoteRpc(baseUrl: string, endpoint: string, args: object): Pr return body.result.value } +/** Read the explicit page cut from a freshly opened Session follow stream. */ +async function sessionCursor(baseUrl: string, sessionId: string): Promise { + const socket = new WebSocket(`${baseUrl.replace(/^http/, 'ws')}/api/remote.mux`) + const streamId = `smoke-history-${randomUUID()}` + try { + await new Promise((resolve, reject) => { + const cleanup = (): void => { + socket.removeEventListener('open', opened) + socket.removeEventListener('error', failed) + socket.removeEventListener('close', closed) + } + const opened = (): void => { + cleanup() + resolve() + } + const failed = (): void => { + cleanup() + reject(new Error('session/follow carrier failed before opening')) + } + const closed = (): void => { + cleanup() + reject(new Error('session/follow carrier closed before opening')) + } + socket.addEventListener('open', opened) + socket.addEventListener('error', failed) + socket.addEventListener('close', closed) + }) + return await new Promise((resolve, reject) => { + const timer = setTimeout(() => { finish(new Error('session/follow did not publish an opening cursor')) }, 10_000) + const cleanup = (): void => { + clearTimeout(timer) + socket.removeEventListener('message', message) + socket.removeEventListener('error', failed) + socket.removeEventListener('close', closed) + } + const finish = (error: Error | undefined, cursor?: number): void => { + cleanup() + if (error !== undefined) reject(error) + else resolve(cursor ?? -1) + } + const message = (event: MessageEvent): void => { + try { + if (typeof event.data !== 'string') throw new Error('session/follow published a non-text frame') + const frame: unknown = JSON.parse(event.data) + if (!isRecord(frame) || frame.streamId !== streamId) return + if (frame.type === 'error') { + finish(new Error(`session/follow failed: ${JSON.stringify(frame.error)}`)) + return + } + if (frame.type === 'end') { + finish(new Error('session/follow ended before its opening cursor')) + return + } + const value = frame.value + if (frame.type === 'item' && isRecord(value) + && value.type === 'opened' && Number.isSafeInteger(value.cursor)) { + finish(undefined, value.cursor as number) + } + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))) + } + } + const failed = (): void => { finish(new Error('session/follow carrier failed before its opening cursor')) } + const closed = (): void => { finish(new Error('session/follow carrier closed before its opening cursor')) } + socket.addEventListener('message', message) + socket.addEventListener('error', failed) + socket.addEventListener('close', closed) + socket.send(JSON.stringify({ + type: 'open', + streamId, + endpoint: 'session/follow', + payload: { args: { request: { address: { kind: 'session', sessionId } } } }, + })) + }) + } finally { + socket.close() + } +} + interface HistoryPage { events: { event: { type: string; data: unknown } }[] hasMore: boolean @@ -102,8 +181,9 @@ function hasAssistantMarker(page: HistoryPage, marker: string): boolean { } async function history(baseUrl: string, sessionId: string): Promise { + const throughSeq = await sessionCursor(baseUrl, sessionId) return remoteRpc(baseUrl, 'session/page', { - request: { address: { kind: 'session', sessionId }, maxMessages: 10 }, + request: { address: { kind: 'session', sessionId }, throughSeq, maxMessages: 10 }, }) } @@ -390,7 +470,7 @@ describe('dsh web keyless CLI smoke', () => { await new Promise(resolveClose => provider.close(() => { resolveClose() })) rmSync(workspace, { recursive: true, force: true }) } - }, 30_000) + }, 120_000) it('DSH_TOOLS_MODE=code collapses the provider wire tools to run_code with the SDK prompt section', async () => { requireDist() diff --git a/packages/client/connection/src/client/fixture.ts b/packages/client/connection/src/client/fixture.ts index 729c518095..5dedcee80b 100644 --- a/packages/client/connection/src/client/fixture.ts +++ b/packages/client/connection/src/client/fixture.ts @@ -82,6 +82,7 @@ interface FixtureFollowRequest { interface FixturePageRequest { readonly address: FixtureSessionAddress + readonly throughSeq: number readonly beforeSeq?: number readonly maxMessages?: number } @@ -193,6 +194,7 @@ interface FixtureSessionApi { fork(request: { readonly sessionId: SessionId; readonly atSeq?: number }): Promise> history(request: { readonly sessionId: SessionId + readonly throughSeq?: number readonly beforeSeq?: number readonly maxMessages?: number }): Promise> @@ -2698,13 +2700,15 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld { }, history: async (request) => { const log = logs.get(request.sessionId) ?? [] + const throughSeq = request.throughSeq ?? log.length - 1 + const boundedLog = log.slice(0, throughSeq + 1) // Snapshot at request time, then deliver after the transit delay. - const page = pageOf(log, request.beforeSeq, request.maxMessages ?? 50) + const page = pageOf(boundedLog, request.beforeSeq, request.maxMessages ?? 50) // Tail page carries the projections block (host parallel: one consistent // cut over the registered units; asOfSeq = window tail seq, -1 on an // empty log — the host's session.seq-1 convention). const projections = request.beforeSeq === undefined - ? { asOfSeq: log.length - 1, values: projectionValuesOf(log) } + ? { asOfSeq: throughSeq, values: projectionValuesOf(boundedLog) } : undefined const doomed = failNextHistory failNextHistory = false @@ -3520,6 +3524,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld { : page.address.childSessionId return sessionApi.history({ sessionId: pageSessionId, + throughSeq: page.throughSeq, ...page.beforeSeq === undefined ? {} : { beforeSeq: page.beforeSeq }, ...page.maxMessages === undefined ? {} : { maxMessages: page.maxMessages }, })