From 408cdf743645e1aa1a132c15517dc59351a1f673 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:20:46 +0800 Subject: [PATCH] refactor(gui): mint kind-agnostic PendingWait carriers for pending interactions --- packages/client/runtime/src/client/index.ts | 5 +- .../src/client/sessions/conversation.ts | 8 +- .../runtime/src/client/sessions/pending.ts | 79 +++++++++++++++++++ .../runtime/src/client/sessions/session.ts | 48 +++++++---- packages/client/runtime/src/client/slots.ts | 4 + packages/client/runtime/tests/fake-api.ts | 8 +- packages/client/runtime/tests/manager.spec.ts | 4 +- packages/client/runtime/tests/session.spec.ts | 48 ++++++++++- 8 files changed, 174 insertions(+), 30 deletions(-) create mode 100644 packages/client/runtime/src/client/sessions/pending.ts diff --git a/packages/client/runtime/src/client/index.ts b/packages/client/runtime/src/client/index.ts index b90812916a..93d2e2846d 100644 --- a/packages/client/runtime/src/client/index.ts +++ b/packages/client/runtime/src/client/index.ts @@ -34,9 +34,12 @@ export type { } from './contract/store.ts' export type { AssistantBlock, AssistantMessageNode, ContextMessageNode, ConversationNode, ConversationSnapshot, - PendingInteraction, RunningToolCall, SteeringMessageNode, + RunningToolCall, SteeringMessageNode, ToolResultNode, UnknownSurfaceNode, UserMessageNode, } from './sessions/conversation.ts' +// PendingWait is a value export: tests construct fixture waits directly. +export { PendingWait } from './sessions/pending.ts' +export type { PendingInteraction, PendingKind, PendingPayloads } from './sessions/pending.ts' export type { SessionId } from '@deepseek-ai/dsh-client-connection/client' // ---- Narrowed aliases (the single narrowing point of the slot type chain: diff --git a/packages/client/runtime/src/client/sessions/conversation.ts b/packages/client/runtime/src/client/sessions/conversation.ts index 18c5501972..1e06fad70d 100644 --- a/packages/client/runtime/src/client/sessions/conversation.ts +++ b/packages/client/runtime/src/client/sessions/conversation.ts @@ -4,7 +4,8 @@ // string here (narrow to real brands when convenient). import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' -import type { RpcError, RpcId, SessionId, ToolCallView, ToolResultView } from '@deepseek-ai/dsh-client-connection/client' +import type { RpcError, SessionId, ToolCallView, ToolResultView } from '@deepseek-ai/dsh-client-connection/client' +import type { PendingInteraction } from './pending.ts' /** Assistant content blocks sorted by what the UI cares about * (text body / collapsible reasoning / tool-call card head / other fallback). */ @@ -121,11 +122,6 @@ export interface RunningToolCall { callView: ToolCallView | null } -/** Approval/question placeholder cards (visible, not answerable; - * rpcId = the requested frame's envelope id, the future respond backfill key). */ -export type PendingInteraction = - | { kind: 'approval'; rpcId: RpcId; approvalId: string; toolName: string; callId?: string; reason?: string } - | { kind: 'question'; rpcId: RpcId; questions: readonly unknown[] } /** In-progress assistant output (chunk accumulator product). */ export interface PartialAssistant { diff --git a/packages/client/runtime/src/client/sessions/pending.ts b/packages/client/runtime/src/client/sessions/pending.ts new file mode 100644 index 0000000000..ba69a6951e --- /dev/null +++ b/packages/client/runtime/src/client/sessions/pending.ts @@ -0,0 +1,79 @@ +// PendingWait: the carrier-protocol half of a pending host interaction. The runtime owns only +// envelope knowledge (rpcId backfill into a client-response); domain result encoding belongs to +// the interaction's consumer package. + +import type { + ClientResponse, MuxFrame, RpcId, RpcReceipt, SessionId, +} from '@deepseek-ai/dsh-client-connection/client' + +/** Kind-keyed payload map: the requested frame's domain fields (envelope fields stripped). */ +export interface PendingPayloads { + approval: Omit, 'type' | 'sessionId'> + question: Omit, 'type' | 'sessionId'> +} + +/** Pending-interaction discriminant (the keys of PendingPayloads). */ +export type PendingKind = keyof PendingPayloads + +/** Kind-discriminated union of concrete waits: narrowing on `kind` types `payload`. */ +export type PendingInteraction = { [K in PendingKind]: PendingWait }[PendingKind] + +/** Key prefixes, one per kind (the key doubles as the Session pending-map key). */ +const KEY_PREFIX: Record = { approval: 'a', question: 'q' } + +/** + * One pending host-owned interaction wait: an immutable render face + * (kind/key/sessionId/payload) plus the response carrier. respond() backfills + * the requested frame's rpcId into a client-response envelope — no consumer + * ever sees the raw rpcId. Settlement is expressed only by pending-list + * membership (the settled flag is a fail-loud guard, not a render input). + */ +export class PendingWait { + /** Interaction kind (union discriminant). */ + readonly kind: K + /** Opaque render identity, `:` — stable across baseline replay, usable as a React key. */ + readonly key: string + /** Owning session. */ + readonly sessionId: SessionId + /** The requested frame's domain fields, verbatim. */ + readonly payload: PendingPayloads[K] + #settled = false + readonly #rpcId: RpcId + readonly #respond: (message: ClientResponse) => Promise + + /** + * Minted by Session on a requested frame (public construction is the test-fixture path). + * @param kind - interaction kind. + * @param rpcId - the requested frame's stable envelope id (kept private; respond echoes it). + * @param sessionId - owning session. + * @param payload - the requested frame's domain fields. + * @param respond - the client-response carrier (api.respond). + */ + constructor( + kind: K, rpcId: RpcId, sessionId: SessionId, payload: PendingPayloads[K], + respond: (message: ClientResponse) => Promise, + ) { + this.kind = kind + this.key = `${KEY_PREFIX[kind]}:${rpcId}` + this.sessionId = sessionId + this.payload = payload + this.#rpcId = rpcId + this.#respond = respond + } + + /** + * Send a result for this wait: wraps it into the client-response envelope + * with the rpcId backfilled. Throws synchronously once settled. + * @param result - the result shell (ok value / error envelope), domain-encoded by the caller. + * @returns the carrier receipt. + */ + respond(result: ClientResponse['result']): Promise { + if (this.#settled) throw new Error(`pending wait ${this.key} is already settled`) + return this.#respond({ type: 'client-response', rpcId: this.#rpcId, result }) + } + + /** Session-only settlement mark (the authoritative resolved frame arrived); respond() throws afterwards. */ + markSettled(): void { + this.#settled = true + } +} diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/client/runtime/src/client/sessions/session.ts index 761aca96c2..6251391df0 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -5,12 +5,17 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types' import type { SessionEvent } from '@deepseek-ai/dsh-session/types' -import type { HistoryEntry, IApiClient, MuxFrame, RpcError, RpcId, RpcResult, SessionId, ToolEventView } from '@deepseek-ai/dsh-client-connection/client' +import type { + HistoryEntry, IApiClient, MuxFrame, RpcError, RpcId, RpcResult, + SessionId, ToolEventView, +} from '@deepseek-ai/dsh-client-connection/client' import { transportError } from '@deepseek-ai/dsh-client-connection/client' import type { ObservableSnapshot } from '../contract/store.ts' import type { - ConversationNode, ConversationSnapshot, OpenState, PendingInteraction, PromptError, RunningToolCall, + ConversationNode, ConversationSnapshot, OpenState, PromptError, RunningToolCall, } from './conversation.ts' +import type { PendingInteraction } from './pending.ts' +import { PendingWait } from './pending.ts' import { FoldAdapter } from './fold-adapter.ts' import { Notifier } from './notifier.ts' import { PartialAccumulator } from './partial.ts' @@ -183,7 +188,9 @@ export class Session implements ObservableSnapshot { this.events = [] this.views = [] this.baseSeq = 0 - this.pending.clear() // the subscribed baseline replay re-sends still-pending requested frames verbatim + // Superseded, not settled: the baseline replay re-sends still-pending requested frames verbatim + // (same rpcId), re-minting fresh waits; a stale reference's respond() still reaches the host. + this.pending.clear() this.pendingRev++ this.subscribedLastSeq = null this.liveBuffer = [] @@ -229,33 +236,27 @@ export class Session implements ObservableSnapshot { return // pure baseline bookkeeping, no visible change } case 'approval/requested': { - this.pending.set(`a:${rpcId}`, { - kind: 'approval', rpcId, approvalId: frame.approvalId, toolName: frame.toolName, - ...(frame.callId !== undefined ? { callId: frame.callId } : {}), - ...(frame.reason !== undefined ? { reason: frame.reason } : {}), - }) - this.pendingRev++ + const { type: _type, sessionId: _sid, ...payload } = frame + this.mint(new PendingWait('approval', rpcId, this.sessionId, payload, m => this.api.respond(m))) this.notifier.markDirty() return } case 'approval/resolved': { - for (const [key, item] of this.pending) { - if (item.kind === 'approval' && item.approvalId === frame.approvalId) { - this.pending.delete(key) - this.pendingRev++ - } + for (const item of this.pending.values()) { + if (item.kind === 'approval' && item.payload.approvalId === frame.approvalId) this.settle(item) } this.notifier.markDirty() return } case 'question/requested': { - this.pending.set(`q:${rpcId}`, { kind: 'question', rpcId, questions: frame.questions }) - this.pendingRev++ + const { type: _type, sessionId: _sid, ...payload } = frame + this.mint(new PendingWait('question', rpcId, this.sessionId, payload, m => this.api.respond(m))) this.notifier.markDirty() return } case 'question/resolved': { - if (this.pending.delete(`q:${frame.questionRpcId}`)) this.pendingRev++ + const item = this.pending.get(`q:${frame.questionRpcId}`) + if (item !== undefined) this.settle(item) this.notifier.markDirty() return } @@ -295,6 +296,19 @@ export class Session implements ObservableSnapshot { // ---- 私有 ---- + /** Requested-frame arrival: the wait enters the pending map under its own key. */ + private mint(wait: PendingInteraction): void { + this.pending.set(wait.key, wait) + this.pendingRev++ + } + + /** Authoritative resolved-frame settlement: mark, then drop from the pending map. */ + private settle(wait: PendingInteraction): void { + wait.markSettled() + this.pending.delete(wait.key) + this.pendingRev++ + } + /** @param generation - openGeneration at launch; every await re-checks it and a stale pass * drops all writes (resync superseded this open — its outcome belongs to a dead connection). */ private async doOpen(generation: number): Promise { diff --git a/packages/client/runtime/src/client/slots.ts b/packages/client/runtime/src/client/slots.ts index 700026df52..0930787e0a 100644 --- a/packages/client/runtime/src/client/slots.ts +++ b/packages/client/runtime/src/client/slots.ts @@ -66,6 +66,10 @@ interface ErasedRegisterOptions { id?: string order?: number label?: string + /** Chain-slot routing selector (pure; the core validates presence for chain targets). */ + select?: (owner: never) => unknown + /** Chain-slot explicit ordering override (ascending; registration order otherwise). */ + priority?: number registrant?: string } diff --git a/packages/client/runtime/tests/fake-api.ts b/packages/client/runtime/tests/fake-api.ts index c13ef09fcb..25f12c2b70 100644 --- a/packages/client/runtime/tests/fake-api.ts +++ b/packages/client/runtime/tests/fake-api.ts @@ -2,7 +2,7 @@ // data source on a real clock; behavior tests need per-case responses and // deferred-controlled timing). Streams are hand pumps: pushMux/pushHost. import type { - HostFrame, IApiClient, MuxFrame, RpcError, RpcRequest, RpcResponse, SessionId, + ClientResponse, HostFrame, IApiClient, MuxFrame, RpcError, RpcReceipt, RpcRequest, RpcResponse, SessionId, } from '@deepseek-ai/dsh-client-connection/client' import { RpcId } from '@deepseek-ai/dsh-client-connection/client' @@ -93,8 +93,10 @@ export class FakeApiClient implements IApiClient { host: (_payload: unknown, signal: AbortSignal, onOpen?: () => void) => this.openStream(this.hostConns, signal, onOpen), } - respond(): Promise<{ accepted: false; reason: 'not-pending' }> { - return Promise.resolve({ accepted: false, reason: 'not-pending' }) + onRespond: (message: ClientResponse) => Promise = () => Promise.resolve({ accepted: true }) + + respond(message: ClientResponse): Promise { + return this.record('respond', message, this.onRespond(message)) } /** Push one mux frame to every open mux stream (rpcId minted unless pinned by the case). */ diff --git a/packages/client/runtime/tests/manager.spec.ts b/packages/client/runtime/tests/manager.spec.ts index af7bc60fd2..e44df24310 100644 --- a/packages/client/runtime/tests/manager.spec.ts +++ b/packages/client/runtime/tests/manager.spec.ts @@ -34,7 +34,7 @@ describe('instances', () => { manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } }) manager.handleMuxEnvelope({ rpcId: 're' as never, payload: { type: 'session/event', sessionId: S1, event: plainTurn(0, 0, 'x', 'y')[0] as never } }) const session = manager.get(S1) - expect(session.getSnapshot().pending).toMatchObject([{ kind: 'approval', approvalId: 'ap1' }]) + expect(session.getSnapshot().pending).toMatchObject([{ kind: 'approval', payload: { approvalId: 'ap1' } }]) // Buffer cleared: a second instantiation of another id gets nothing. expect(manager.get(S2).getSnapshot().pending).toEqual([]) }) @@ -48,7 +48,7 @@ describe('instances', () => { } const pending = manager.get(S1).getSnapshot().pending expect(pending).toHaveLength(32) - expect(pending.map(p => p.rpcId)).toEqual(Array.from({ length: 32 }, (_, i) => `q${i + 8}`)) // oldest 8 dropped + expect(pending.map(p => p.key)).toEqual(Array.from({ length: 32 }, (_, i) => `q:q${i + 8}`)) // oldest 8 dropped // Removed session: buffered frames must not replay on a future instantiation. manager.handleMuxEnvelope({ rpcId: 'qz' as never, payload: { type: 'question/requested', sessionId: S2, questions: [] } }) manager.handleHostEnvelope({ rpcId: 'hz' as never, payload: { type: 'host/session-removed', sessionId: S2 } }) diff --git a/packages/client/runtime/tests/session.spec.ts b/packages/client/runtime/tests/session.spec.ts index 149f73c1fc..b980a674fe 100644 --- a/packages/client/runtime/tests/session.spec.ts +++ b/packages/client/runtime/tests/session.spec.ts @@ -251,6 +251,36 @@ describe('pending interactions', () => { session.handleMuxEnvelope('ry' as never, { type: 'question/resolved', sessionId: SID, questionRpcId: 'rq' as never, outcome: 'answered' }) expect(session.getSnapshot().pending).toEqual([]) }) + + it('mints waits whose respond() backfills the requested rpcId into the client-response envelope', async () => { + const { api, session } = makeSession() + session.handleMuxEnvelope('rq-answer' as never, { type: 'question/requested', sessionId: SID, questions: [] }) + const wait = session.getSnapshot().pending[0]! + expect(wait).toMatchObject({ kind: 'question', key: 'q:rq-answer', sessionId: SID, payload: { questions: [] } }) + const receipt = await wait.respond({ + ok: true, + value: { sessionId: SID, answer: { answers: [{ id: 'mode', selected: ['Fast'] }] } }, + }) + expect(receipt).toEqual({ accepted: true }) + expect(api.callsOf('respond')).toEqual([{ + type: 'client-response', rpcId: 'rq-answer', + result: { + ok: true, + value: { sessionId: SID, answer: { answers: [{ id: 'mode', selected: ['Fast'] }] } }, + }, + }]) + }) + + it('settles the wait on the authoritative resolved frame: respond() then throws synchronously', async () => { + const { api, session } = makeSession() + session.handleMuxEnvelope('rq1' as never, { type: 'question/requested', sessionId: SID, questions: [] }) + const wait = session.getSnapshot().pending[0]! + session.handleMuxEnvelope('ry' as never, { type: 'question/resolved', sessionId: SID, questionRpcId: 'rq1' as never, outcome: 'answered' }) + expect(session.getSnapshot().pending).toEqual([]) + expect(() => wait.respond({ ok: false, error: { code: 'internal', message: 'x', details: {} } })) + .toThrow('already settled') + expect(api.callsOf('respond')).toEqual([]) + }) }) describe('remaining branches', () => { @@ -355,7 +385,7 @@ describe('remaining branches', () => { session.handleMuxEnvelope('ra' as never, { type: 'approval/requested', sessionId: SID, approvalId: 'ap2' as never, toolName: 'rm', callId: 'c1' as never, reason: '危险', }) - expect(session.getSnapshot().pending[0]).toMatchObject({ kind: 'approval', callId: 'c1', reason: '危险' }) + expect(session.getSnapshot().pending[0]).toMatchObject({ kind: 'approval', payload: { callId: 'c1', reason: '危险' } }) session.handleMuxEnvelope('rx' as never, { type: 'approval/resolved', sessionId: SID, approvalId: 'ap2' as never, outcome: 'approved' as never }) session.handleMuxEnvelope('rx2' as never, { type: 'approval/resolved', sessionId: SID, approvalId: 'ap2' as never, outcome: 'approved' as never }) session.handleMuxEnvelope('ry2' as never, { type: 'question/resolved', sessionId: SID, questionRpcId: 'never-was' as never, outcome: 'cancelled' }) @@ -568,6 +598,22 @@ describe('resync', () => { expect(cold.api.calls).toEqual([]) // never opened: no traffic }) + it('re-mints a replayed requested frame as a fresh wait with the same key (old reference superseded)', async () => { + const { api, session } = makeSession() + api.onHistory = () => histResponse(plainTurn(0, 0, 'a', 'b')) + await session.open() + session.handleMuxEnvelope('rq-replay' as never, { type: 'question/requested', sessionId: SID, questions: [] }) + const before = session.getSnapshot().pending[0]! + await session.resync() + session.handleMuxEnvelope('rq-replay' as never, { type: 'question/requested', sessionId: SID, questions: [] }) + const after = session.getSnapshot().pending[0]! + expect(after).not.toBe(before) + expect(after.key).toBe(before.key) + // Superseded ≠ settled: an in-flight respond on the stale reference still reaches the host. + await before.respond({ ok: false, error: { code: 'internal', message: 'x', details: {} } }) + expect(api.callsOf('respond')).toMatchObject([{ rpcId: 'rq-replay' }]) + }) + it('drops a stale in-flight open superseded by resync (generation guard)', async () => { const { api, session } = makeSession() const stale = deferred>>()