fix(client): satisfy stream lifecycle contracts

This commit is contained in:
imccyu
2026-08-23 16:16:04 +08:00
parent 9eb3747ffc
commit e38982adc8
6 changed files with 21 additions and 19 deletions
@@ -261,7 +261,7 @@ describe('RemoteJournalStream', () => {
], ],
[ [
signal => new Promise<Page>((_resolve, reject) => { signal => new Promise<Page>((_resolve, reject) => {
const aborted = (): void => { reject(signal.reason) } const aborted = (): void => { reject(new Error('page aborted')) }
signal.addEventListener('abort', aborted, { once: true }) signal.addEventListener('abort', aborted, { once: true })
if (signal.aborted) aborted() if (signal.aborted) aborted()
}), }),
@@ -120,10 +120,10 @@ describe('session.history projections block', () => {
if (!response.ok) throw new Error('history failed') if (!response.ok) throw new Error('history failed')
expect(response.value.events.map(entry => entry.event.seq)).toEqual([0]) expect(response.value.events.map(entry => entry.event.seq)).toEqual([0])
expect(response.value.projections).toEqual({ expect(response.value.projections?.asOfSeq).toBe(0)
asOfSeq: 0, expect(response.value.projections?.values).toEqual(
values: expect.objectContaining({ 'test/last-user': { text: 'm0' } }), expect.objectContaining({ 'test/last-user': { text: 'm0' } }),
}) )
}) })
it('projects an empty log at cursor -1', async () => { it('projects an empty log at cursor -1', async () => {
@@ -134,10 +134,10 @@ describe('session.history projections block', () => {
if (!response.ok) throw new Error('history failed') if (!response.ok) throw new Error('history failed')
expect(response.value.events).toEqual([]) expect(response.value.events).toEqual([])
expect(response.value.projections).toEqual({ expect(response.value.projections?.asOfSeq).toBe(-1)
asOfSeq: -1, expect(response.value.projections?.values).toEqual(
values: expect.objectContaining({ 'test/last-user': null }), expect.objectContaining({ 'test/last-user': null }),
}) )
}) })
it('publishes the attachments imageLimits as a constant unit while both seams are composed', async () => { it('publishes the attachments imageLimits as a constant unit while both seams are composed', async () => {
@@ -205,22 +205,23 @@ export function apply(ctx: Context): void {
const source = generationSource const source = generationSource
if (source === undefined) throw new Error('connection: no generation source is registered') if (source === undefined) throw new Error('connection: no generation source is registered')
const token = {} const token = {}
const ownsGeneration = (): boolean => owner?.token === token
const controller = new ConnectionController(api, source, { const controller = new ConnectionController(api, source, {
...sinks, ...sinks,
onConnected: (next) => { onConnected: (next) => {
if (owner?.token !== token) return if (!ownsGeneration()) return
publishDescription(next) publishDescription(next)
// A description subscriber may synchronously stop the loop. In that // A description subscriber may synchronously stop the loop. In that
// case publishDescription(undefined) has already retracted this // case publishDescription(undefined) has already retracted this
// generation, so do not leak its stale connected notification to // generation, so do not leak its stale connected notification to
// the consumer sink afterward. // the consumer sink afterward.
if (owner?.token !== token || !Object.is(description, next)) return if (!ownsGeneration() || !Object.is(description, next)) return
sinks.onConnected?.(next) sinks.onConnected?.(next)
}, },
onStateChange: (state) => { onStateChange: (state) => {
if (owner?.token !== token) return if (!ownsGeneration()) return
if (state === 'reconnecting') publishDescription(undefined) if (state === 'reconnecting') publishDescription(undefined)
if (owner?.token !== token) return if (!ownsGeneration()) return
sinks.onStateChange?.(state) sinks.onStateChange?.(state)
}, },
}, config ?? {}) }, config ?? {})
@@ -816,9 +816,9 @@ describe('remaining branches', () => {
expect(session.getSnapshot().promptError).toBeNull() expect(session.getSnapshot().promptError).toBeNull()
}) })
it('dispose is a reserved no-op on resident instances', () => { it('dispose is a reserved no-op on resident instances', async () => {
const { session } = makeSession() const { session } = makeSession()
expect(() => { session.dispose() }).not.toThrow() await expect(session.dispose()).resolves.toBeUndefined()
}) })
it('carries history-entry and follow-frame views into the business-neutral Event input', async () => { it('carries history-entry and follow-frame views into the business-neutral Event input', async () => {
@@ -1342,7 +1342,7 @@ describe('ChatView', () => {
const h = makeHarness({ const h = makeHarness({
pending: [ pending: [
new PendingWait('approval', 'r1', SID, new PendingWait('approval', 'r1', SID,
{ approvalId: 'ap1', toolName: 'bash' } as PendingWait<'approval'>['payload'], vi.fn()), { approvalId: 'ap1', toolName: 'bash' }, vi.fn()),
new PendingWait('question', 'r2', SID, new PendingWait('question', 'r2', SID,
{ questions: [{ id: 'q1', question: '选择' }] }, vi.fn()), { questions: [{ id: 'q1', question: '选择' }] }, vi.fn()),
], ],
@@ -88,18 +88,19 @@ describe('UserQuestionService', () => {
const pending = Promise.withResolvers<never>() const pending = Promise.withResolvers<never>()
ctx.userQuestions.registerProvider({ ask: () => pending.promise }) ctx.userQuestions.registerProvider({ ask: () => pending.promise })
const controller = new AbortController() const controller = new AbortController()
const abortReason = new DOMException('This operation was aborted', 'AbortError')
const answer = ctx.userQuestions.ask({ const answer = ctx.userQuestions.ask({
questions: [{ id: 'confirm', question: 'Proceed?' }], questions: [{ id: 'confirm', question: 'Proceed?' }],
signal: controller.signal, signal: controller.signal,
}) })
controller.abort() controller.abort(abortReason)
pending.reject(controller.signal.reason) pending.reject(abortReason)
await expect(answer).rejects.toMatchObject({ await expect(answer).rejects.toMatchObject({
name: 'UserQuestionError', name: 'UserQuestionError',
code: 'ASK_ABORTED', code: 'ASK_ABORTED',
cause: controller.signal.reason, cause: abortReason,
}) })
}) })