Merge branch 'master' into worktree/web-session-titles

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:

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 {

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<Extract<MuxFrame, { type: 'approval/requested' }>, 'type' | 'sessionId'>
+  question: Omit<Extract<MuxFrame, { type: 'question/requested' }>, '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<K> }[PendingKind]
+
+/** Key prefixes, one per kind (the key doubles as the Session pending-map key). */
+const KEY_PREFIX: Record<PendingKind, string> = { 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<K extends PendingKind = PendingKind> {
+  /** Interaction kind (union discriminant). */
+  readonly kind: K
+  /** Opaque render identity, `<prefix>:<rpcId>` — 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<RpcReceipt>
+
+  /**
+   * 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<RpcReceipt>,
+  ) {
+    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<RpcReceipt> {
+    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
+  }
+}

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<ConversationSnapshot> {
     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<ConversationSnapshot> {
         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<ConversationSnapshot> {
 
   // ---- 私有 ----
 
+  /** 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<void> {

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
 }
 

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<RpcReceipt> = () => Promise.resolve({ accepted: true })
+
+  respond(message: ClientResponse): Promise<RpcReceipt> {
+    return this.record('respond', message, this.onRespond(message))
   }
 
   /** Push one mux frame to every open mux stream (rpcId minted unless pinned by the case). */

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 } })

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<Awaited<ReturnType<FakeApiClient['onHistory']>>>()