fix(client): preserve pending waits across reconnect

2026-08-04 18:52:11 +08:00
parent 00d4349eb4
commit 3a54694d28
17 changed files with 219 additions and 64 deletions
--- a/packages/client/runtime/README.i18n.yaml
+++ b/packages/client/runtime/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/client/runtime/README.md
-README.md: f24ab9f9c056b6c1136676146082e7aaaa23377f
-README.zh.md: af3d5bb42e62818b969c9e0de814c1bcff4d9f37
+README.md: 1d474c63f25f4aa25dc20c842f1968ea00ca3773
+README.zh.md: 6ed5d74042901f667fc3d6bccbb2304d1aab00a8
--- a/packages/client/runtime/README.md
+++ b/packages/client/runtime/README.md
@@ -8,7 +8,7 @@ Client cordis boot and React-free object services: SlotsService wraps SlotCore a

 Workspace and Session lists have independent monotone `pending` → `ready` baseline phases and separate refresh activity/error state. Incremental upsert/removal frames and unary mutation echoes arriving during a list request replay over its response. The first successful baseline establishes Host order; later refreshes update rows and membership without changing the relative order of identities already shown. Removed Workspace ids retain process-local tombstones so late changed frames cannot resurrect them; reconnect still takes `workspace.list` as the baseline. Workspace recency is derived only after both baselines are ready and never changes Workspace list order.

-`SessionSummary.pendingInteraction` classifies the live user action blocking a Session as `approval`, `plan-review`, or `question`. `SessionManager` tracks answerable requested/resolved mux frames by their stable request identities even before a Session object is instantiated; pre-instantiation buffering retains every live request, replaces replay duplicates, and removes resolved requests so the list status always has a matching answerable `PendingWait` when the Session is opened. The first pending question takes presentation priority over concurrent approvals to match composer routing, while a question carrying the `plan-review` intent keeps that distinct status. The state is connection-generation scoped: disconnect clears it, and mux-open replay restores only requests that remain pending.
+`SessionSummary.pendingInteraction` classifies the live user action blocking a Session as `approval`, `plan-review`, or `question`. `SessionManager` tracks answerable requested/resolved mux frames by their stable request identities even before a Session object is instantiated; pre-instantiation buffering retains every live request, replaces replay duplicates, and removes resolved requests so the list status always has a matching answerable `PendingWait` when the Session is opened. The first pending question takes presentation priority over concurrent approvals to match composer routing, while only a request that satisfies the plan-review composer's binary rendering constraints keeps the distinct `plan-review` status. The state is connection-generation scoped: every generation death clears it, and mux-open replay restores only requests that remain pending.

 `WorkspacesService.delete(workspaceId)` removes the registration from the client projection after the successful unary response; the matching `host/workspace-removed` frame is idempotent and synchronizes other tabs. Session state and the current Session selection are independent, so accounted Sessions immediately project under Ungrouped after their Workspace disappears.

--- a/packages/client/runtime/README.zh.md
+++ b/packages/client/runtime/README.zh.md
@@ -8,7 +8,7 @@

 Workspace 和 Session 列表各自具有单调的 `pending` → `ready` 基线阶段，也有各自的刷新活动／错误状态。列表请求期间到达的增量插入或更新／移除帧与一元变更回显会在其响应之上回放。第一次成功的基线建立 Host 顺序；后续刷新更新行和成员关系，但不改变已经显示的标识之间的相对顺序。已移除的 Workspace id 会保留进程本地删除标记，避免延迟到达的 changed 帧将其复活；重连仍以 `workspace.list` 作为基线。Workspace 新近程度只在两条基线都 ready 后派生，且绝不改变 Workspace 列表顺序。

-`SessionSummary.pendingInteraction` 将阻塞 Session 的实时用户操作分类为 `approval`、`plan-review` 或 `question`。`SessionManager` 依据稳定的请求标识跟踪可应答请求的 requested/resolved mux 帧，即使 `Session` 对象尚未实例化也不例外；实例化前的缓冲会保留每个仍有效的请求，替换回放产生的重复项，并移除已解决的请求，因此打开 Session 时，列表状态始终有一个对应的可应答 `PendingWait`。审批与问题并发时，第一个 pending 问题具有更高的呈现优先级，以匹配 composer 路由；带有 `plan-review` intent 的问题会保留这一独立状态。该状态的作用域限定在连接代次内：断连时清除，mux 打开时的回放只恢复仍处于 pending 的请求。
+`SessionSummary.pendingInteraction` 将阻塞 Session 的实时用户操作分类为 `approval`、`plan-review` 或 `question`。`SessionManager` 依据稳定的请求标识跟踪可应答请求的 requested/resolved mux 帧，即使 `Session` 对象尚未实例化也不例外；实例化前的缓冲会保留每个仍有效的请求，替换回放产生的重复项，并移除已解决的请求，因此打开 Session 时，列表状态始终有一个对应的可应答 `PendingWait`。审批与问题并发时，第一个 pending 问题具有更高的呈现优先级，以匹配 composer 路由；只有满足 plan-review composer 二元呈现约束的请求才会保留独立的 `plan-review` 状态。该状态的作用域限定在连接代次内：每个代次结束时清除，mux 打开时的回放只恢复仍处于 pending 的请求。

 `WorkspacesService.delete(workspaceId)` 在一元响应成功后从客户端投影中移除注册记录；对应的 `host/workspace-removed` 帧具有幂等性，并负责同步其他标签页。Session 状态与当前 Session selection 相互独立，因此 Workspace 消失后，其已纳入客户端投影的 Session 会立即投影到 Ungrouped 下。

--- a/packages/client/runtime/src/client/index.ts
+++ b/packages/client/runtime/src/client/index.ts
@@ -219,17 +219,14 @@ export function apply(ctx: Context): void {
        console.error('[web-runtime] history reconnect failed:', error)
      }
    },
-    onStateChange: (state) => {
-      // Generation death fires before any next-generation frame can arrive
-      // (reconnect replays flow from stream open, ahead of onConnected):
-      // the only safe moment to drop generation-scoped interaction state.
-      if (state === 'reconnecting') {
-        sessions.handleDisconnected()
-        try {
-          sessionHistory.handleDisconnected()
-        } catch (error) {
-          console.error('[web-runtime] history disconnect failed:', error)
-        }
+    onDisconnected: () => {
+      // Reconnect replays flow from stream open, ahead of onConnected, so each
+      // generation death is the only safe moment to drop generation-owned state.
+      sessions.handleDisconnected()
+      try {
+        sessionHistory.handleDisconnected()
+      } catch (error) {
+        console.error('[web-runtime] history disconnect failed:', error)
      }
    },
  })
--- a/packages/client/runtime/src/client/sessions/manager.ts
+++ b/packages/client/runtime/src/client/sessions/manager.ts
@@ -78,10 +78,24 @@ function bufferedRequestKey(envelope: RpcRequest<MuxFrame>): string | undefined
    case 'approval/requested': return `a:${frame.approvalId}`
    case 'question/requested': return `q:${envelope.rpcId}`
    case 'session/queue': return 'queue'
+    /* v8 ignore next -- pendingBuffers contains only the three frame types above. */
    default: return undefined
  }
 }

+/** Match ui-question's binary plan-review routing at the wire boundary. */
+function questionInteractionStatus(
+  questions: Extract<MuxFrame, { type: 'question/requested' }>['questions'],
+): PendingInteractionStatus {
+  if (questions.length !== 1) return 'question'
+  const question = questions[0] as typeof questions[number]
+  const intent = question.intent
+  if (intent?.kind !== 'plan-review' || question.detail === undefined) return 'question'
+  if (question.multiSelect === true) return 'question'
+  const options = question.options ?? []
+  if (options.length > 2) return 'question'
+  return options.some(option => option.label === intent.approve) ? 'plan-review' : 'question'
+}

 /** Instance cluster + frame entry + the session list (see the web client architecture RFC). */
 export class SessionManager {
@@ -620,11 +634,10 @@ export class SessionManager {
      // them so last-wins cannot pin a phantom value over recomputed truth.
      this.projectionStores.get(frame.sessionId)?.truncate(frame.lastSeq)
      this.notifier.markDirty()
-      // New mux-generation baseline: buffered session/queue frames belong to
-      // the previous generation and the host is about to resend the live
-      // snapshot — drop them, or every reconnect appends a duplicate batch
-      // (and enough reconnects push real approval/question frames past the
-      // cap). Same re-baseline signal Session uses for its own mirror.
+      // New mux-generation baseline: discard the previous queue snapshot.
+      // The host omits session/queue when the live queue is empty, so retaining
+      // it could replay stale work when the Session is instantiated later.
+      // This is the same re-baseline signal Session uses for its own mirror.
      const buffered = this.pendingBuffers.get(frame.sessionId)
      if (buffered !== undefined) {
        const kept = buffered.filter(item => item.payload.type !== 'session/queue')
@@ -644,9 +657,7 @@ export class SessionManager {
      this.trackPending(
        frame.sessionId,
        `q:${envelope.rpcId}`,
-        frame.questions.length === 1 && frame.questions[0]?.intent?.kind === 'plan-review'
-          ? 'plan-review'
-          : 'question',
+        questionInteractionStatus(frame.questions),
      )
    } else if (frame.type === 'question/resolved') {
      this.resolvePending(frame.sessionId, `q:${frame.questionRpcId}`)
@@ -782,14 +793,14 @@ export class SessionManager {
   * request with its live rpcId.
   */
  handleDisconnected(): void {
+    for (const session of this.sessions.values()) session.handleDisconnected()
    if (this.pendingInteractions.size > 0) {
      this.pendingInteractions.clear()
      this.notifier.markDirty()
    }
    for (const [sessionId, buffer] of [...this.pendingBuffers]) {
      const kept = buffer.filter(item =>
-        item.payload.type !== 'approval/requested' && item.payload.type !== 'approval/resolved'
-        && item.payload.type !== 'question/requested' && item.payload.type !== 'question/resolved')
+        item.payload.type !== 'approval/requested' && item.payload.type !== 'question/requested')
      if (kept.length === buffer.length) continue
      if (kept.length === 0) this.pendingBuffers.delete(sessionId)
      else this.pendingBuffers.set(sessionId, kept)
--- a/packages/client/runtime/src/client/sessions/session.ts
+++ b/packages/client/runtime/src/client/sessions/session.ts
@@ -393,8 +393,9 @@ export class Session implements SessionFace {
  }

  /** Reconnect rebuild (manager calls this on onConnected for instances that were opened):
-   *  reset the window and rerun open; pending waits for the baseline replay. Invalidates any
-   *  in-flight open first — its history request rode the dead connection and must not settle
+   *  reset the window and rerun open. Pending waits reset at generation death, before the next
+   *  baseline replay can arrive, so this method preserves freshly replayed waits. Invalidates
+   *  any in-flight open first — its history request rode the dead connection and must not settle
   *  the fresh generation into 'error' (audit S4). */
  async resync(): Promise<void> {
    // The queue mirror is NOT cleared here: onConnected (which drives resync)
@@ -410,10 +411,6 @@ export class Session implements SessionFace {
    this.events = []
    this.views = []
    this.baseSeq = 0
-    // 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 = []
    this.notifier.markDirty()
@@ -442,6 +439,16 @@ export class Session implements SessionFace {

  // ---- Manager-only entry points (@internal; never called by the UI) ----

+  /** Discard generation-scoped waits before a next-generation replay can arrive. */
+  handleDisconnected(): void {
+    if (this.pending.size === 0) return
+    // Superseded, not settled: a stale reference may already be responding,
+    // and the replay re-mints each still-pending request with the same rpcId.
+    this.pending.clear()
+    this.pendingRev++
+    this.notifier.markDirty()
+  }
+
  /**
   * Mux frame arrival (the dispatch switch).
   * @param rpcId - the frame envelope id (the respond backfill key for requested frames).
--- a/packages/client/runtime/tests/client-apply.spec.ts
+++ b/packages/client/runtime/tests/client-apply.spec.ts
@@ -76,8 +76,9 @@ describe('runtime client apply', () => {
    })
    await Promise.resolve()
    expect(workspaces.list.getSnapshot().items[0]?.workspaceId).toBe('w-new')
-    // Mux sink and onConnected route without throwing (manager semantics own the behavior).
+    // Mux and generation-lifecycle sinks route without throwing (manager semantics own the behavior).
    bench.sinks?.onMuxEnvelope?.({ rpcId: 'r2' as never, payload: { type: 'stream/error', message: 'x' } as never })
+    bench.sinks?.onDisconnected?.()
    bench.sinks?.onConnected?.()
  })

--- a/packages/client/runtime/tests/manager.spec.ts
+++ b/packages/client/runtime/tests/manager.spec.ts
@@ -887,7 +887,7 @@ describe('pending-interaction list status', () => {
    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  })

-  it('classifies ordinary questions and plan reviews, then clears by question rpcId', () => {
+  it('classifies ordinary questions and renderable plan reviews, then clears by question rpcId', () => {
    const manager = new SessionManager(new FakeApiClient())
    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
    manager.handleMuxEnvelope({
@@ -903,7 +903,11 @@ describe('pending-interaction list status', () => {
      payload: {
        type: 'question/requested',
        sessionId: S1,
-        questions: [{ id: 'plan', question: 'Approve?', detail: '# Plan', intent: { kind: 'plan-review', approve: 'Approve' } }],
+        questions: [{
+          id: 'plan', question: 'Approve?', detail: '# Plan',
+          options: [{ label: 'Approve' }, { label: 'Refuse' }],
+          intent: { kind: 'plan-review', approve: 'Approve' },
+        }],
      },
    })
    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('plan-review')
@@ -911,6 +915,28 @@ describe('pending-interaction list status', () => {
    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  })

+  it.each([
+    ['missing detail', {}],
+    ['multi-select', { detail: '# Plan', multiSelect: true }],
+    ['more than two options', { detail: '# Plan', options: [{ label: 'Approve' }, { label: 'Refuse' }, { label: 'Revise' }] }],
+    ['missing approve option', { detail: '# Plan', options: [{ label: 'Refuse' }] }],
+  ])('keeps an unrenderable %s plan intent on the ordinary question flow', (_name, over) => {
+    const manager = new SessionManager(new FakeApiClient())
+    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
+    manager.handleMuxEnvelope({
+      rpcId: 'q-plan' as never,
+      payload: {
+        type: 'question/requested', sessionId: S1,
+        questions: [{
+          id: 'plan', question: 'Approve?', options: [{ label: 'Approve' }],
+          intent: { kind: 'plan-review', approve: 'Approve' },
+          ...over,
+        }],
+      },
+    })
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
+  })
+
  it('the first question outranks sibling approvals and resolving it reveals the remaining wait', () => {
    const manager = new SessionManager(new FakeApiClient())
    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
@@ -930,19 +956,30 @@ describe('pending-interaction list status', () => {
    expect(manager.getListSnapshot().items).toHaveLength(0)
  })

-  it('drops stale status at generation death before replay re-adds live interactions', () => {
-    const manager = new SessionManager(new FakeApiClient())
+  it('drops stale waits at generation death and preserves replay arriving before onConnected', async () => {
+    const api = new FakeApiClient()
+    api.onHistory = () => Promise.resolve(ok({
+      events: entries(plainTurn(0, 0, 'a', 'b')) as never[],
+      hasMore: false,
+      modelTarget: { provider: 'deepseek-official', model: 'deepseek-chat' },
+    }))
+    const manager = new SessionManager(api)
    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
+    const session = manager.get(S1)
+    await session.open()
    manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
+    expect(session.getSnapshot().pending).toMatchObject([{ key: 'a:ra' }])
    // Generation death clears (resolved-while-disconnected questions send no frame)…
    manager.handleDisconnected()
    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
+    expect(session.getSnapshot().pending).toEqual([])
    // …and a replayed frame arriving before onConnected (stream open precedes
-    // the readiness handshake) survives the later handleConnected untouched.
+    // the readiness handshake) survives the later resync untouched.
    manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
    manager.handleConnected()
    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
+    expect(session.getSnapshot().pending).toMatchObject([{ key: 'a:ra' }])
  })

  it('generation death drops buffered answerable frames (a dead generation cannot be answered)', () => {
--- a/packages/client/runtime/tests/session.spec.ts
+++ b/packages/client/runtime/tests/session.spec.ts
@@ -1098,16 +1098,19 @@ describe('remaining branches', () => {
 })

 describe('resync', () => {
-  it('rebuilds the window and clears pending; cold instances no-op', async () => {
+  it('rebuilds the window without clearing a fresh-generation wait; cold instances no-op', async () => {
    const { api, session } = makeSession()
    api.onHistory = () => histResponse(plainTurn(0, 0, 'a', 'b'))
    await session.open()
    session.handleMuxEnvelope('ra' as never, { type: 'approval/requested', sessionId: SID, approvalId: 'ap1' as never, toolName: 'rm' })
+    session.handleDisconnected()
+    expect(session.getSnapshot().pending).toEqual([])
+    session.handleMuxEnvelope('ra' as never, { type: 'approval/requested', sessionId: SID, approvalId: 'ap1' as never, toolName: 'rm' })
    api.onHistory = () => histResponse([...plainTurn(0, 0, 'a', 'b'), ...plainTurn(6, 1, 'c', 'd')])
    await session.resync()
    const snapshot = session.getSnapshot()
    expect(snapshot.openState).toBe('open')
-    expect(snapshot.pending).toEqual([]) // baseline replay re-sends still-pending frames
+    expect(snapshot.pending).toMatchObject([{ key: 'a:ra', kind: 'approval' }])
    expect(snapshot.nodes).toHaveLength(4)

    const cold = makeSession()