diff --git a/packages/client/ui-approval/src/client/contract/slots.ts b/packages/client/ui-approval/src/client/contract/slots.ts index 3f8d481767..0c12b119e7 100644 --- a/packages/client/ui-approval/src/client/contract/slots.ts +++ b/packages/client/ui-approval/src/client/contract/slots.ts @@ -72,6 +72,7 @@ export class PendingApproval { readonly #reject: (reason: unknown) => void readonly #signal: AbortSignal | undefined readonly #onAbort: (() => void) | undefined + readonly #delegated = Symbol('pending approval delegated') #settled = false /** @@ -111,6 +112,21 @@ export class PendingApproval { }, 'pending approval settlement failed') } + /** Delegate an unanswered request to the next waterfall listener. */ + delegate(): void { + if (this.#settled) return + this.finish(() => { this.#reject(this.#delegated) }) + } + + /** + * Test whether a rejection requests waterfall delegation. + * @param reason - rejection received from {@link PendingApproval.result}. + * @returns whether {@link PendingApproval.delegate} produced it. + */ + isDelegation(reason: unknown): boolean { + return reason === this.#delegated + } + /** * End an unanswered presentation when its transport, scope, or plugin lifetime ends. * @param reason - rejection exposed to the waiting Remote Event listener. diff --git a/packages/client/ui-approval/src/client/index.ts b/packages/client/ui-approval/src/client/index.ts index 04efe273c6..aeb34a433c 100644 --- a/packages/client/ui-approval/src/client/index.ts +++ b/packages/client/ui-approval/src/client/index.ts @@ -4,7 +4,7 @@ import type {} from '@deepseek-ai/dsh-api-remotes/client' import type {} from '@deepseek-ai/dsh-api-session-controller/client' import type { ComposerChainProps } from '@deepseek-ai/dsh-client-ui-conversation/client' import type {} from '@deepseek-ai/dsh-client-ui-renderer/client' -import type {} from '@deepseek-ai/dsh-client-ui-session/client' +import type { PendingInteractionPublisher } from '@deepseek-ai/dsh-client-ui-session/client' import type { TypertClientEventListener } from '@deepseek-ai/dsh-typert-protocol' import type {} from '@deepseek-ai/dsh-client-locale/client' import { ApprovalPanel } from './ApprovalPanel.tsx' @@ -36,7 +36,7 @@ async function answerApproval( owner: ClientContext, request: ClientApprovalRequest, next: ClientApprovalNext, - registerPendingInteraction: (pending: PendingApproval) => () => void, + registerPendingInteraction: PendingInteractionPublisher, ): Promise { const sessionId = ctx.sessions.scopeOf(owner) if (sessionId === undefined) return next() @@ -48,11 +48,21 @@ async function answerApproval( ...(request.reason === undefined ? {} : { reason: request.reason }), ...(request.signal === undefined ? {} : { signal: request.signal }), }) - const remove = registerPendingInteraction(pending) + const completed = Promise.withResolvers() + const remove = registerPendingInteraction(pending, async () => { + pending.delegate() + await completed.promise + }) try { - return await pending.result + try { + return await pending.result + } catch (error) { + if (pending.isDelegation(error)) return await next() + throw error + } } finally { remove() + completed.resolve() } } diff --git a/packages/client/ui-approval/tests/ui-approval.client.spec.tsx b/packages/client/ui-approval/tests/ui-approval.client.spec.tsx index 9e326f18f0..665ae928a4 100644 --- a/packages/client/ui-approval/tests/ui-approval.client.spec.tsx +++ b/packages/client/ui-approval/tests/ui-approval.client.spec.tsx @@ -33,6 +33,7 @@ interface PluginBench { readonly disposeLocale: ReturnType readonly register: ReturnType readonly injectSlot: ReturnType + releasePending(): Promise registration(): { options: { select(props: { pendingInteraction: PendingApproval | undefined }): PendingApproval | null @@ -52,13 +53,14 @@ function setupPlugin(): PluginBench { } | undefined const disposeSlot = vi.fn() const disposeLocale = vi.fn() - let pending: readonly PendingApproval[] = [] + const pending = new Map Promise>() const registerPendingInteraction = vi.fn((_precedence: (value: PendingApproval) => number) => ( value: PendingApproval, + delegate: () => Promise, ) => { _precedence(value) - pending = [...pending, value] - return () => { pending = pending.filter(candidate => candidate !== value) } + pending.set(value, delegate) + return () => { pending.delete(value) } }) const register = vi.fn(( options: NonNullable['options'], @@ -89,12 +91,17 @@ function setupPlugin(): PluginBench { return { ctx, listener, - pending: { getSnapshot: () => pending }, + pending: { getSnapshot: () => [...pending.keys()] }, registerPendingInteraction, disposeSlot, disposeLocale, register, injectSlot, + async releasePending() { + const delegates = [...pending.values()] + pending.clear() + await Promise.allSettled(delegates.map(delegate => delegate())) + }, registration: () => { if (registration === undefined) throw new Error('approval slot was not registered') return registration @@ -129,6 +136,7 @@ describe('PendingApproval', () => { expect(pending.reason).toBe('needs access') expect(remove).toHaveBeenCalledWith('abort', expect.any(Function)) expect(() => { pending.abort(new Error('late')) }).not.toThrow() + expect(() => { pending.delegate() }).not.toThrow() await expect(pending.answer('rejected')).rejects.toThrow(/already settled/) }) @@ -258,6 +266,22 @@ describe('approval Remote Event consumer', () => { await scope.fiber.dispose() }) + it('delegates an active request when its interaction domain unloads', async () => { + const bench = setupPlugin() + const scope = createScope(bench.ctx, id('s1')) + await scope.fiber.await() + const next = vi.fn(() => Promise.resolve<'unavailable'>('unavailable')) + const result = bench.listener.call(scope.ctx, { toolName: 'bash' }, next) + expect(bench.pending.getSnapshot()).toHaveLength(1) + + await bench.releasePending() + + await expect(result).resolves.toBe('unavailable') + expect(next).toHaveBeenCalledOnce() + expect(bench.pending.getSnapshot()).toEqual([]) + await scope.fiber.dispose() + }) + it('publishes a scoped request without optional request metadata', async () => { const bench = setupPlugin() const scope = createScope(bench.ctx, id('s1')) diff --git a/packages/client/ui-session/src/client/index.ts b/packages/client/ui-session/src/client/index.ts index 0f0c521b78..b720041f4d 100644 --- a/packages/client/ui-session/src/client/index.ts +++ b/packages/client/ui-session/src/client/index.ts @@ -55,8 +55,19 @@ export type SessionPendingInteractionSnapshot = ReadonlyMap +/** Publish one pending interaction and define how plugin teardown delegates it. */ +export type PendingInteractionPublisher = ( + interaction: T, + delegate: () => Promise, +) => () => void + +interface PendingInteractionEntry { + readonly interaction: T + readonly delegate: () => Promise +} + class PendingInteractionDomain { - private readonly values = new Map() + private readonly values = new Map>() constructor( readonly precedence: (interaction: T) => number, @@ -64,23 +75,30 @@ class PendingInteractionDomain { ) {} valuesSnapshot(): readonly T[] { - return [...this.values.values()] + return [...this.values.values()].map(entry => entry.interaction) } - publish(interaction: T): () => void { + publish(interaction: T, delegate: () => Promise): () => void { if (this.values.has(interaction.key)) { throw new Error(`ui-session: duplicate pending interaction key '${interaction.key}'`) } - this.values.set(interaction.key, interaction) + this.values.set(interaction.key, { interaction, delegate }) this.changed() let active = true return () => { if (!active) return active = false - this.values.delete(interaction.key) + if (!this.values.delete(interaction.key)) return this.changed() } } + + /** Remove every pending value and return the operations that settle their owners. */ + release(): readonly (() => Promise)[] { + const delegates = [...this.values.values()].map(entry => entry.delegate) + this.values.clear() + return delegates + } } declare module '@deepseek-ai/dsh-client-ui-slots' { @@ -278,12 +296,14 @@ export class UiSession extends Service { /** * Register one pending-interaction domain and return its publication function. + * Domain teardown first removes its visible values, then delegates and awaits + * every still-active owner request. * @param precedence - deterministic cross-domain precedence; larger values win. - * @returns a function that publishes one exact interaction until its disposer runs. + * @returns a function that publishes one interaction and its teardown delegation. */ registerPendingInteraction( precedence: (interaction: T) => number, - ): (interaction: T) => () => void { + ): PendingInteractionPublisher { const domain = new PendingInteractionDomain(precedence, () => { this.publishPendingInteractions() }) @@ -291,13 +311,15 @@ export class UiSession extends Service { this.ctx.effect(() => { this.pendingDomains.push(runtimeDomain) this.publishPendingInteractions() - return () => { + return async () => { + const delegates = domain.release() const index = this.pendingDomains.indexOf(runtimeDomain) this.pendingDomains.splice(index, 1) this.publishPendingInteractions() + await Promise.allSettled(delegates.map(delegate => Promise.resolve().then(delegate))) } }, 'uiSession.registerPendingInteraction()') - return interaction => domain.publish(interaction) + return (interaction, delegate) => domain.publish(interaction, delegate) } private rebuildBindings(): void { diff --git a/packages/client/ui-session/tests/ui-session.client.spec.ts b/packages/client/ui-session/tests/ui-session.client.spec.ts index 5b9ebe6f3d..97946f4c1d 100644 --- a/packages/client/ui-session/tests/ui-session.client.spec.ts +++ b/packages/client/ui-session/tests/ui-session.client.spec.ts @@ -444,15 +444,16 @@ describe('UiSession pending interactions', () => { const question = { key: 'question:1', kind: 'question', sessionId: id } const plan = { key: 'question:2', kind: 'plan-review', sessionId: id } const background = { key: 'background:1', kind: 'background', sessionId: id } - const removeApproval = registerApproval(approval) + const delegate = (): Promise => Promise.resolve() + const removeApproval = registerApproval(approval, delegate) expect(service.pendingInteractions.getSnapshot().get(id)).toBe(approval) - const removeDuplicate = registerApproval(duplicate) + const removeDuplicate = registerApproval(duplicate, delegate) expect(service.pendingInteractions.getSnapshot().get(id)).toBe(duplicate) - const removeQuestion = registerQuestion(question) + const removeQuestion = registerQuestion(question, delegate) expect(service.pendingInteractions.getSnapshot().get(id)).toBe(question) - const removePlan = registerQuestion(plan) + const removePlan = registerQuestion(plan, delegate) expect(service.pendingInteractions.getSnapshot().get(id)).toBe(plan) - const removeBackground = registerBackground(background) + const removeBackground = registerBackground(background, delegate) expect(service.pendingInteractions.getSnapshot().get(id)).toBe(plan) removeBackground() @@ -477,8 +478,9 @@ describe('UiSession pending interactions', () => { () => 1, ) const interaction = { key: 'question:1', kind: 'question', sessionId: sessionId('s1') } - const remove = registerPendingInteraction(interaction) - expect(() => { registerPendingInteraction(interaction) }) + const delegate = () => Promise.resolve() + const remove = registerPendingInteraction(interaction, delegate) + expect(() => { registerPendingInteraction(interaction, delegate) }) .toThrow("ui-session: duplicate pending interaction key 'question:1'") const failure = new Error('pending subscriber failed') @@ -495,6 +497,31 @@ describe('UiSession pending interactions', () => { failure, ) }) + + it('removes active values before awaiting their teardown delegation', async () => { + const ctx = new Context() + const bench = createSessionsBench(ctx) + const service = createUiSession(ctx, bench) + const gate = Promise.withResolvers() + const delegate = vi.fn(() => gate.promise) + const publish = service.registerPendingInteraction(() => 1) + const remove = publish( + { key: 'question:1', kind: 'question', sessionId: sessionId('s1') }, + delegate, + ) + + let disposed = false + const disposal = ctx.fiber.dispose().then(() => { disposed = true }) + await vi.waitFor(() => { expect(delegate).toHaveBeenCalledOnce() }) + expect(service.pendingInteractions.getSnapshot()).toEqual(new Map()) + expect(disposed).toBe(false) + remove() + remove() + + gate.resolve(undefined) + await disposal + expect(disposed).toBe(true) + }) }) describe('ui-session apply', () => { diff --git a/packages/client/ui-user-questions/src/client/contract/slots.ts b/packages/client/ui-user-questions/src/client/contract/slots.ts index cec6e85fb8..0a65ef79eb 100644 --- a/packages/client/ui-user-questions/src/client/contract/slots.ts +++ b/packages/client/ui-user-questions/src/client/contract/slots.ts @@ -107,6 +107,7 @@ export class PendingQuestion { readonly #reject: (reason: unknown) => void readonly #signal: AbortSignal | undefined readonly #onAbort: (() => void) | undefined + readonly #delegated = Symbol('pending question delegated') #settled = false /** @@ -150,6 +151,21 @@ export class PendingQuestion { }, 'pending question settlement failed') } + /** Delegate an unanswered request to the next waterfall listener. */ + delegate(): void { + if (this.#settled) return + this.finish(() => { this.#reject(this.#delegated) }) + } + + /** + * Test whether a rejection requests waterfall delegation. + * @param reason - rejection received from {@link PendingQuestion.result}. + * @returns whether {@link PendingQuestion.delegate} produced it. + */ + isDelegation(reason: unknown): boolean { + return reason === this.#delegated + } + /** Reject the Host waterfall because the user closed the question. */ cancel(): Promise { return settlePendingComposer(() => { diff --git a/packages/client/ui-user-questions/src/client/index.ts b/packages/client/ui-user-questions/src/client/index.ts index 6efc7fa956..5344c80ed7 100644 --- a/packages/client/ui-user-questions/src/client/index.ts +++ b/packages/client/ui-user-questions/src/client/index.ts @@ -16,7 +16,7 @@ import type { Context as ClientContext } from '@deepseek-ai/cordis' import type {} from '@deepseek-ai/dsh-api-remotes/client' import type { ComposerChainProps } from '@deepseek-ai/dsh-client-ui-conversation/client' import type {} from '@deepseek-ai/dsh-client-ui-renderer/client' -import type {} from '@deepseek-ai/dsh-client-ui-session/client' +import type { PendingInteractionPublisher } from '@deepseek-ai/dsh-client-ui-session/client' import type { TypertClientEventListener } from '@deepseek-ai/dsh-typert-protocol' // Type-only: pulls the locale plugin's Context merge (ctx.locale). import type {} from '@deepseek-ai/dsh-client-locale/client' @@ -54,16 +54,26 @@ async function answerQuestion( owner: ClientContext, request: ClientQuestionRequest, next: ClientQuestionNext, - registerPendingInteraction: (pending: PendingQuestion) => () => void, + registerPendingInteraction: PendingInteractionPublisher, ): Promise { const sessionId = ctx.sessions.scopeOf(owner) if (sessionId === undefined) return next() const pending = new PendingQuestion(sessionId, request.questions, request.signal) - const remove = registerPendingInteraction(pending) + const completed = Promise.withResolvers() + const remove = registerPendingInteraction(pending, async () => { + pending.delegate() + await completed.promise + }) try { - return await pending.result + try { + return await pending.result + } catch (error) { + if (pending.isDelegation(error)) return await next() + throw error + } } finally { remove() + completed.resolve() } } diff --git a/packages/client/ui-user-questions/tests/browser-plugin.client.spec.ts b/packages/client/ui-user-questions/tests/browser-plugin.client.spec.ts index 866ac883e0..881e6fa193 100644 --- a/packages/client/ui-user-questions/tests/browser-plugin.client.spec.ts +++ b/packages/client/ui-user-questions/tests/browser-plugin.client.spec.ts @@ -49,13 +49,14 @@ async function bench(declare = true) { candidate as Context & { [SESSION_SCOPE]?: SessionId } )[SESSION_SCOPE]) ctx.provide('sessions', { scopeOf } as never) - let pending: readonly PendingQuestion[] = [] + const pending = new Map Promise>() const registerPendingInteraction = vi.fn((_precedence: (value: PendingQuestion) => number) => ( value: PendingQuestion, + delegate: () => Promise, ) => { _precedence(value) - pending = [...pending, value] - return () => { pending = pending.filter(candidate => candidate !== value) } + pending.set(value, delegate) + return () => { pending.delete(value) } }) ctx.provide('uiSession', { registerPendingInteraction } as never) let listener: QuestionListener | undefined @@ -81,11 +82,16 @@ async function bench(declare = true) { locale, agent, scopeOf, - pending: { getSnapshot: () => pending }, + pending: { getSnapshot: () => [...pending.keys()] }, registerPendingInteraction, on, fiber, invoke, + async releasePending() { + const delegates = [...pending.values()] + pending.clear() + await Promise.allSettled(delegates.map(delegate => delegate())) + }, } } @@ -174,6 +180,20 @@ describe('apply', () => { expect(b.slots.entries('conversation.composer')).toHaveLength(1) }) + it('delegates an active request when its interaction domain unloads', async () => { + const b = await bench() + const next = vi.fn(async () => ANSWER) + const result = b.invoke(b.agent, { questions: QUESTIONS }, next) + await Promise.resolve() + expect(b.pending.getSnapshot()).toHaveLength(1) + + await b.releasePending() + + await expect(result).resolves.toBe(ANSWER) + expect(next).toHaveBeenCalledOnce() + expect(b.pending.getSnapshot()).toEqual([]) + }) + it('removes the stable composer with the plugin lifetime', async () => { const b = await bench() expect(b.slots.entries('conversation.composer')).toHaveLength(1) @@ -216,6 +236,7 @@ describe('PendingQuestion', () => { await pending.answer(ANSWER) await expect(pending.result).resolves.toBe(ANSWER) pending.abort(new Error('late disposal')) + pending.delegate() }) it('rejects an unanswered request with its caller-owned lifecycle reason', async () => {