Merge remote-tracking branch 'origin/master' into xtr/trajectory-virtual-scroll

# Conflicts: # apps/web/tests/chat-scroll-fixture.ts # packages/client/runtime/README.i18n.yaml
2026-08-05 10:41:50 +08:00
parent 92de5bb6d4 81c63bbf4d
commit d4908eeb14
823 changed files with 16467 additions and 30960 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: f5dd94bc1c937b69d74f450f11bf875b50af4c62
-README.zh.md: 270d2d41e266fe198f91a939eba946e2f86844d9
+README.md: 676b2a287ae08d840fd1dbabb4187f8ca5753a08
+README.zh.md: aa82944a957b7cfce0dc3b91821810db90d532e0
--- a/packages/client/runtime/README.md
+++ b/packages/client/runtime/README.md
@@ -8,6 +8,8 @@ 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 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: disconnect 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.

 `WorkspaceListState.archivedSessionIds` mirrors the Host's registry-global archive set (a `readonly SessionId[]` in Host order, replaced only when membership changes; consumers needing O(1) lookups build a transient Set). It is full-snapshot state: the `workspace.list` baseline, the `archiveSession` unary echo, and the `host/archived-sessions-changed` frame each install the complete set. `WorkspacesService.archiveSession(sessionId)` archives over the wire; the projection sweep clears the current selection into the New Session view state whenever it lands in the archive set — one rule covering the local echo, another tab's frame, and a reconnect baseline restoring a selection archived while this client was away. A set installed while a `workspace.list` request is in flight also supersedes that stale baseline's set. Grouping surfaces hide members everywhere while the session rows stay in the list store.
--- a/packages/client/runtime/README.zh.md
+++ b/packages/client/runtime/README.zh.md
@@ -8,6 +8,8 @@

 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 composer 二元呈现约束的请求才会保留独立的 `plan-review` 状态。该状态的作用域限定在连接代次内：断连时清除，mux 打开时的回放只恢复仍处于 pending 的请求。
+
 `WorkspacesService.delete(workspaceId)` 在一元响应成功后从客户端投影中移除注册记录；对应的 `host/workspace-removed` 帧具有幂等性，并负责同步其他标签页。Session 状态与当前 Session selection 相互独立，因此 Workspace 消失后，其已纳入客户端投影的 Session 会立即投影到 Ungrouped 下。

 `WorkspaceListState.archivedSessionIds` 镜像 Host 的注册表级全局归档集合（一个按 Host 顺序的 `readonly SessionId[]`，仅在成员变化时才替换；需要 O(1) 查询的消费方自建临时 Set）。它是全快照状态：`workspace.list` 基线、`archiveSession` 一元回声和 `host/archived-sessions-changed` 帧各自安装完整集合。`WorkspacesService.archiveSession(sessionId)` 通过 wire 归档；投影层在当前 selection 落入归档集合时统一清空为 New Session 视图状态——一条规则同时覆盖本地回声、其他标签页的帧、以及重连基线恢复出一个离线期间被归档的 selection。在 `workspace.list` 请求进行中安装的集合还会取代该过期基线携带的集合。各分组视图在所有位置隐藏集合成员，而会话行本身仍留在列表 store 中。
--- a/packages/client/runtime/src/client/index.ts
+++ b/packages/client/runtime/src/client/index.ts
@@ -59,7 +59,9 @@ export type {
 export type { ConversationHistoryProjection } from './session-history/history-fold.ts'
 export type { SessionHistoryInspection } from './sessions/history.ts'
 export { PendingWait } from './sessions/pending.ts'
-export type { PendingInteraction, PendingKind, PendingPayloads } from './sessions/pending.ts'
+export type {
+  PendingInteraction, PendingInteractionStatus, PendingKind, PendingPayloads,
+} from './sessions/pending.ts'
 // Projection value store (session-projection RFC, push model): host-computed
 // whole values per key; domains ship projection support with zero client code.
 export type {
--- a/packages/client/runtime/src/client/session-history/source.ts
+++ b/packages/client/runtime/src/client/session-history/source.ts
@@ -10,7 +10,7 @@ import {
  compactHistoryInspectionEntries, createHistoryInspection,
 } from '../sessions/history.ts'
 import { Notifier } from '../sessions/notifier.ts'
-import { PartialAccumulator } from '../sessions/partial.ts'
+import { isVisibleAssistantChunk, PartialAccumulator } from '../sessions/partial.ts'

 const HISTORY_PAGE_MESSAGES = 50

@@ -430,11 +430,3 @@ export class SessionHistorySource implements SessionHistoryFace {
    return this.inspectionCache.value
  }
 }
-
-function isVisibleAssistantChunk(type: string): boolean {
-  return type === 'block-start'
-    || type === 'text-delta'
-    || type === 'reasoning-delta'
-    || type === 'tool-call-delta'
-    || type === 'block-end'
-}
--- a/packages/client/runtime/src/client/sessions/conversation.ts
+++ b/packages/client/runtime/src/client/sessions/conversation.ts
@@ -331,6 +331,8 @@ export interface ConversationSnapshot {
  sessionId: SessionId
  /** Human transcript plus retry notices and interrupted-turn terminal nodes in event order. */
  nodes: readonly ConversationNode[]
+  /** Exact in-window `turn/start` time and optional matching `turn/end` time. */
+  turnTimings: ReadonlyMap<number, { readonly startTime: number; readonly endTime?: number }>
  /** In-window completed turn number -> its `turn/end` event seq. */
  turnEnds: ReadonlyMap<number, number>
  partial: PartialAssistant | null
--- a/packages/client/runtime/src/client/sessions/lineage.ts
+++ b/packages/client/runtime/src/client/sessions/lineage.ts
@@ -4,6 +4,7 @@

 import type { SessionId, SessionSummary } from '@deepseek-ai/dsh-client-connection/client'
 import type { SessionProjectionMap } from '@deepseek-ai/dsh-session-projection/types'
+import type { PendingInteractionStatus } from './pending.ts'

 /** Host list summary enriched with the latest mux-projected durable title. */
 export interface TitledSessionSummary extends SessionSummary {
@@ -12,7 +13,7 @@ export interface TitledSessionSummary extends SessionSummary {
  projectionValues?: Readonly<Partial<SessionProjectionMap>>
 }

-/** One flattened session-list row (summary + lineage indent depth + live pending-approval bit). */
+/** One flattened session-list row with lineage depth and live pending interaction. */
 export interface SessionListEntry {
  sessionId: SessionId
  title?: string
@@ -26,8 +27,8 @@ export interface SessionListEntry {
  cwd?: string
  /** Current host-computed projection values for list consumers. */
  projectionValues?: Readonly<Partial<SessionProjectionMap>>
-  /** An approval question is pending on this session (mux-frame derived; the sidebar's amber dot). */
-  waitingApproval: boolean
+  /** User interaction currently blocking this session, derived from live mux frames. */
+  pendingInteraction?: PendingInteractionStatus
  /** Lineage indent depth: root = 0; the UI just multiplies by the indent width. */
  depth: number
 }
@@ -37,10 +38,13 @@ export interface SessionListEntry {
 * follows the established input order; this projection never re-sorts a
 * hydrated list from mutable timestamps.
 * @param summaries - the host's session.list items.
- * @param waitingApproval - sessions with a pending approval question (manager-owned live fact; absent = false).
+ * @param pendingInteractions - current manager-owned interaction status by session.
 * @returns display rows in render order.
 */
-export function flattenLineage(summaries: readonly TitledSessionSummary[], waitingApproval?: ReadonlySet<SessionId>): SessionListEntry[] {
+export function flattenLineage(
+  summaries: readonly TitledSessionSummary[],
+  pendingInteractions?: ReadonlyMap<SessionId, PendingInteractionStatus>,
+): SessionListEntry[] {
  const byId = new Map<SessionId, TitledSessionSummary>()
  for (const s of summaries) byId.set(s.sessionId, s)

@@ -64,7 +68,12 @@ export function flattenLineage(summaries: readonly TitledSessionSummary[], waiti
      return
    }
    visited.add(s.sessionId)
-    out.push({ ...s, waitingApproval: waitingApproval?.has(s.sessionId) ?? false, depth })
+    const pendingInteraction = pendingInteractions?.get(s.sessionId)
+    out.push({
+      ...s,
+      ...(pendingInteraction === undefined ? {} : { pendingInteraction }),
+      depth,
+    })
    const kids = children.get(s.sessionId)
    if (kids === undefined) return
    for (const kid of kids) walk(kid, depth + 1)
--- a/packages/client/runtime/src/client/sessions/manager.ts
+++ b/packages/client/runtime/src/client/sessions/manager.ts
@@ -12,6 +12,7 @@ import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
 import { mergeOrderedBaseline } from '../ordered-baseline.ts'
 import type { SessionListEntry, TitledSessionSummary } from './lineage.ts'
 import { flattenLineage } from './lineage.ts'
+import type { PendingInteractionStatus } from './pending.ts'
 // Type-only merge edge: the title domain's client-namespace outlet declares
 // the 'title' projection key this manager projects into list rows (and any
 // useProjection('title') consumer reads). Zero value imports by construction.
@@ -70,23 +71,44 @@ type SessionListMutation =
  /** Local first-send flip: the sender clears blank without waiting for a host frame. */
  | { kind: 'engaged'; sessionId: SessionId }

-/** Per-session cap for pre-instantiation approval/question buffering (low-frequency frames; a few dozen covers any real backlog). */
-const PENDING_BUFFER_CAP = 32
+/** Stable identity of a frame retained until an uninstantiated Session can consume it. */
+function bufferedRequestKey(envelope: RpcRequest<MuxFrame>): string | undefined {
+  const frame = envelope.payload
+  switch (frame.type) {
+    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 {
  private readonly sessions = new Map<SessionId, Session>()
-  /** Approval/question frame buffer for uninstantiated sessions: pending interactions never hit
-   *  history (cannot be backfilled on open), the one frame class that must not take the
-   *  drop-and-backfill path; replayed and cleared on instantiation. Bounded per session (these
-   *  frames are low-frequency; overflow drops oldest) and dropped on session-removed (audit S7). */
+  /** Pre-instantiation buffer for answerable requests and the queued-turn snapshot, which history
+   *  cannot reconstruct on open. Live requests remain until resolution; queue and replay duplicates
+   *  compact by identity. Instantiation replays and clears it, while removal drops it (audit S7). */
  private readonly pendingBuffers = new Map<SessionId, RpcRequest<MuxFrame>[]>()
-  /** Outstanding approval questions per session, keyed by approvalId (idempotent under mux-open
-   *  replays of the same requested frame). Manager-owned rather than read off Session instances
-   *  because the sidebar must light up for sessions never instantiated. Cleared per connection
-   *  generation — the reopen replay re-adds still-pending questions — and on session-removed. */
-  private readonly waitingApprovals = new Map<SessionId, Set<string>>()
+  /** Outstanding answerable interactions per session, keyed by their stable request identity.
+   *  Manager-owned rather than read off Session instances because the sidebar must light up for
+   *  sessions never instantiated. Cleared per connection generation — the reopen replay re-adds
+   *  still-pending requests — and on session-removed. */
+  private readonly pendingInteractions = new Map<SessionId, Map<string, PendingInteractionStatus>>()
  /** Per-session projection value stores, retained independently of instance arrival (the
   *  title-snapshot precedent, generalized): push frames land here whether or not the Session
   *  is instantiated (list rows read the 'title' key), and an instantiated Session adopts the
@@ -567,6 +589,26 @@ export class SessionManager {
    return this.listSnapshotCache
  }

+  /** Add or refresh one stable pending-interaction identity. */
+  private trackPending(sessionId: SessionId, key: string, status: PendingInteractionStatus): void {
+    let interactions = this.pendingInteractions.get(sessionId)
+    if (interactions === undefined) {
+      interactions = new Map()
+      this.pendingInteractions.set(sessionId, interactions)
+    }
+    if (interactions.get(key) === status) return
+    interactions.set(key, status)
+    this.notifier.markDirty()
+  }
+
+  /** Settle one pending-interaction identity without disturbing sibling waits. */
+  private resolvePending(sessionId: SessionId, key: string): void {
+    const interactions = this.pendingInteractions.get(sessionId)
+    if (interactions === undefined || !interactions.delete(key)) return
+    if (interactions.size === 0) this.pendingInteractions.delete(sessionId)
+    this.notifier.markDirty()
+  }
+
  // ---- ConnectionController sinks (wired by boot) ----

  /**
@@ -592,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')
@@ -606,43 +647,54 @@ export class SessionManager {
        }
      }
    }
-    // List-level waiting-approval bit (the sidebar amber dot): tracked here for
-    // every session, instantiated or not; approvalId keys make replays idempotent.
+    // List-level pending-interaction status (the sidebar amber dot): tracked
+    // for every session, instantiated or not; stable keys make replays idempotent.
    if (frame.type === 'approval/requested') {
-      let ids = this.waitingApprovals.get(frame.sessionId)
-      if (ids === undefined) this.waitingApprovals.set(frame.sessionId, ids = new Set())
-      if (!ids.has(frame.approvalId)) {
-        ids.add(frame.approvalId)
-        this.notifier.markDirty()
-      }
+      this.trackPending(frame.sessionId, `a:${frame.approvalId}`, 'approval')
    } else if (frame.type === 'approval/resolved') {
-      const ids = this.waitingApprovals.get(frame.sessionId)
-      if (ids !== undefined && ids.delete(frame.approvalId)) {
-        if (ids.size === 0) this.waitingApprovals.delete(frame.sessionId)
-        this.notifier.markDirty()
-      }
+      this.resolvePending(frame.sessionId, `a:${frame.approvalId}`)
+    } else if (frame.type === 'question/requested') {
+      this.trackPending(
+        frame.sessionId,
+        `q:${envelope.rpcId}`,
+        questionInteractionStatus(frame.questions),
+      )
+    } else if (frame.type === 'question/resolved') {
+      this.resolvePending(frame.sessionId, `q:${frame.questionRpcId}`)
    }
    const session = this.sessions.get(frame.sessionId)
    if (session === undefined) {
-      // Approval/question/queue frames never hit history: buffer for replay on
-      // instantiation; everything else drops (not instantiated — history fully
-      // backfills on open).
+      // Answerable requests never hit history: retain each live identity until
+      // instantiation, compacting replay duplicates and resolutions so list
+      // status cannot outlive the PendingWait the user would need to answer.
+      // Queue is a latest-value snapshot; everything else drops because open
+      // backfills it from history.
      switch (frame.type) {
        case 'approval/requested':
-        case 'approval/resolved':
        case 'question/requested':
-        case 'question/resolved':
        case 'session/queue': {
          const buffer = this.pendingBuffers.get(frame.sessionId) ?? []
-          const prior = frame.type === 'session/queue'
-            ? buffer.findIndex(item => item.payload.type === 'session/queue')
-            : -1
-          if (prior !== -1) buffer.splice(prior, 1)
-          buffer.push(envelope)
-          if (buffer.length > PENDING_BUFFER_CAP) buffer.splice(0, buffer.length - PENDING_BUFFER_CAP)
+          const key = frame.type === 'approval/requested'
+            ? `a:${frame.approvalId}`
+            : frame.type === 'question/requested' ? `q:${envelope.rpcId}` : 'queue'
+          const prior = buffer.findIndex(item => bufferedRequestKey(item) === key)
+          if (prior === -1) buffer.push(envelope)
+          else buffer[prior] = envelope
          this.pendingBuffers.set(frame.sessionId, buffer)
          return
        }
+        case 'approval/resolved':
+        case 'question/resolved': {
+          const buffer = this.pendingBuffers.get(frame.sessionId)
+          if (buffer === undefined) return
+          const key = frame.type === 'approval/resolved'
+            ? `a:${frame.approvalId}`
+            : `q:${frame.questionRpcId}`
+          const prior = buffer.findIndex(item => bufferedRequestKey(item) === key)
+          if (prior !== -1) buffer.splice(prior, 1)
+          if (buffer.length === 0) this.pendingBuffers.delete(frame.sessionId)
+          return
+        }
        default:
          return
      }
@@ -689,7 +741,7 @@ export class SessionManager {
          this.sessions.get(frame.sessionId)?.handleRemoved()
        }
        this.pendingBuffers.delete(frame.sessionId) // a removed session's buffered frames must not replay on a future instantiation
-        this.waitingApprovals.delete(frame.sessionId) // a removed session cannot wait on anyone
+        this.pendingInteractions.delete(frame.sessionId) // a removed session cannot wait on anyone
        if (!durableSubagent) this.projectionStores.delete(frame.sessionId)
        // A pull already in flight was requested before this removal and can
        // carry the pre-removal parentAvailable:true, which would resurrect
@@ -735,20 +787,19 @@ export class SessionManager {
   * The moment a connection generation dies (before any next-generation frame
   * can arrive — onConnected waits for the readiness handshake while replayed
   * frames flow from stream open, so clearing there would race the replay):
-   * drop generation-scoped live state. Approvals resolved while disconnected
-   * send no frame, so the stale bits and the buffered answerable frames must
-   * not survive into the next generation — the mux-open replay re-adds every
-   * still-pending question with its live rpcId.
-   */
+   * drop generation-scoped live state. Interactions resolved while disconnected
+   * send no frame, so stale statuses and buffered answerable frames must not
+   * survive into the next generation — mux-open replay re-adds every still-pending
+   * request with its live rpcId.
+  */
  handleDisconnected(): void {
-    if (this.waitingApprovals.size > 0) {
-      this.waitingApprovals.clear()
+    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)
@@ -855,7 +906,15 @@ export class SessionManager {
        ...(projectionValues === undefined ? {} : { projectionValues }),
      }
    })
-    const fresh = flattenLineage(merged, new Set(this.waitingApprovals.keys()))
+    const pendingInteractions = new Map<SessionId, PendingInteractionStatus>()
+    for (const [sessionId, interactions] of this.pendingInteractions) {
+      const statuses = [...interactions.values()]
+      // The composer selects the first question ahead of approval. Mirror that
+      // answer order so the sidebar names the interaction the user can act on.
+      const status = statuses.find(candidate => candidate !== 'approval') ?? statuses[0]
+      if (status !== undefined) pendingInteractions.set(sessionId, status)
+    }
+    const fresh = flattenLineage(merged, pendingInteractions)
    const items = fresh.map((entry) => {
      const prev = this.entryCache.get(entry.sessionId)
      if (
@@ -863,7 +922,7 @@ export class SessionManager {
        && prev.blank === entry.blank
        && prev.parentSessionId === entry.parentSessionId && prev.cwd === entry.cwd
        && prev.origin === entry.origin && prev.title === entry.title && prev.depth === entry.depth
-        && prev.waitingApproval === entry.waitingApproval
+        && prev.pendingInteraction === entry.pendingInteraction
        && prev.projectionValues === entry.projectionValues
      ) return prev
      this.entryCache.set(entry.sessionId, entry)
--- a/packages/client/runtime/src/client/sessions/notifier.ts
+++ b/packages/client/runtime/src/client/sessions/notifier.ts
@@ -1,5 +1,6 @@
-// Notifier: subscription + microtask-batched notification primitive shared by Session and
-// SessionManager. Semantics: N markDirty calls collapse into one microtask flush;
+// Notifier: subscription + batched notification primitive shared by Session and
+// SessionManager. Semantics: N markDirty calls collapse into one microtask flush, while
+// N markFrameDirty calls collapse into one animation-frame flush;
 // the flush rebuilds the snapshot cache BEFORE notifying (useSyncExternalStore requires a stable
 // getSnapshot reference). With no listeners the rebuild is skipped and only the dirty bit is set
 // (keeps frame storms cheap); the next getSnapshot rebuilds lazily.
@@ -9,12 +10,13 @@
 // swallow the notification — push subscribers (object-layer watchers) would
 // otherwise starve whenever any reader pulls first.

-/** Subscription + microtask-batched notification primitive (shared by Session and SessionManager). */
+/** Subscription + batched notification primitive (shared by Session and SessionManager). */
 export class Notifier {
  private listeners = new Set<() => void>()
  private dirty = false
  private notifyPending = false
-  private scheduled = false
+  private scheduled: 'none' | 'microtask' | 'frame' = 'none'
+  private scheduleGeneration = 0

  /** @param rebuild - snapshot rebuild function injected by the owner (writes the owner's snapshotCache). */
  constructor(private readonly rebuild: () => void) {}
@@ -35,19 +37,16 @@ export class Notifier {
  markDirty(): void {
    this.dirty = true
    this.notifyPending = true
-    if (this.scheduled) return
-    this.scheduled = true
-    queueMicrotask(() => {
-      this.scheduled = false
-      if (!this.notifyPending) return
-      if (this.listeners.size === 0) return // lazy: no subscribers; dirty (if still set) rebuilds on next getSnapshot
-      this.notifyPending = false
-      if (this.dirty) {
-        this.dirty = false
-        this.rebuild()
-      }
-      for (const listener of this.listeners) listener()
-    })
+    if (this.scheduled === 'microtask') return
+    this.schedule('microtask')
+  }
+
+  /** Stream-change entry: mark dirty and publish the cumulative state at most once per frame. */
+  markFrameDirty(): void {
+    this.dirty = true
+    this.notifyPending = true
+    if (this.scheduled !== 'none') return
+    this.schedule(typeof globalThis.requestAnimationFrame === 'function' ? 'frame' : 'microtask')
  }

  /**
@@ -57,11 +56,8 @@ export class Notifier {
  notifyNow(): void {
    this.dirty = true
    this.notifyPending = true
-    if (this.listeners.size === 0) return // lazy: same as markDirty, next getSnapshot rebuilds
-    this.notifyPending = false
-    this.dirty = false
-    this.rebuild()
-    for (const listener of this.listeners) listener()
+    this.invalidateSchedule()
+    this.flush()
  }

  /**
@@ -73,4 +69,35 @@ export class Notifier {
    this.dirty = false
    this.rebuild()
  }
+
+  private schedule(kind: 'microtask' | 'frame'): void {
+    const generation = ++this.scheduleGeneration
+    this.scheduled = kind
+    const publish = () => {
+      if (generation !== this.scheduleGeneration) return
+      this.scheduled = 'none'
+      this.flush()
+    }
+    if (kind === 'frame') {
+      globalThis.requestAnimationFrame(publish)
+    } else {
+      queueMicrotask(publish)
+    }
+  }
+
+  private invalidateSchedule(): void {
+    this.scheduleGeneration++
+    this.scheduled = 'none'
+  }
+
+  private flush(): void {
+    if (!this.notifyPending) return
+    if (this.listeners.size === 0) return // lazy: dirty (if still set) rebuilds on next getSnapshot
+    this.notifyPending = false
+    if (this.dirty) {
+      this.dirty = false
+      this.rebuild()
+    }
+    for (const listener of this.listeners) listener()
+  }
 }
--- a/packages/client/runtime/src/client/sessions/partial.ts
+++ b/packages/client/runtime/src/client/sessions/partial.ts
@@ -6,6 +6,19 @@ import type { StreamChunk } from '@deepseek-ai/dsh-llm/types'
 import type { AssistantBlock, PartialAssistant } from './conversation.ts'
 import { toAssistantBlock } from './conversation.ts'

+/**
+ * Whether a stream chunk changes the partial assistant projection shown by the UI.
+ * @param type - Stream chunk discriminant.
+ * @returns Whether publishing the accumulated partial can change the visible snapshot.
+ */
+export function isVisibleAssistantChunk(type: string): boolean {
+  return type === 'block-start'
+    || type === 'text-delta'
+    || type === 'reasoning-delta'
+    || type === 'tool-call-delta'
+    || type === 'block-end'
+}
+
 /** assistant/chunk accumulator: folds StreamChunks into AssistantBlock[] with block-level immutability. */
 export class PartialAccumulator {
  // Sparse on purpose: block-start may arrive out of order, leaving holes until compaction.
--- a/packages/client/runtime/src/client/sessions/pending.ts
+++ b/packages/client/runtime/src/client/sessions/pending.ts
@@ -15,6 +15,9 @@ export interface PendingPayloads {
 /** Pending-interaction discriminant (the keys of PendingPayloads). */
 export type PendingKind = keyof PendingPayloads

+/** Session-list summary of the user action currently blocking progress. */
+export type PendingInteractionStatus = 'approval' | 'plan-review' | 'question'
+
 /** Kind-discriminated union of concrete waits: narrowing on `kind` types `payload`. */
 export type PendingInteraction = { [K in PendingKind]: PendingWait<K> }[PendingKind]

--- a/packages/client/runtime/src/client/sessions/service.ts
+++ b/packages/client/runtime/src/client/sessions/service.ts
@@ -33,6 +33,7 @@ import type { ISessions } from '../contract/sessions.ts'
 import { createScope, scopeOf as scopeTagOf } from '../agents/scope.ts'
 import { SessionManager } from './manager.ts'
 import type { SessionListPhase, SessionSearchResultItem, SubagentCatalogSnapshot } from './manager.ts'
+import type { PendingInteractionStatus } from './pending.ts'
 import { SessionProvideChannel } from './provide.ts'
 import type { Session } from './session.ts'

@@ -48,8 +49,8 @@ export interface SessionSummary {
  /** Coarse durable origin for navigation filtering; not a continuation capability. */
  origin?: 'subagent'
  running: boolean
-  /** An approval question is pending on this session (sidebar amber-dot state). */
-  waitingApproval: boolean
+  /** User interaction currently blocking this session (sidebar amber-dot state). */
+  pendingInteraction?: PendingInteractionStatus
  /**
   * Empty-log bit (host summary derivation mirror). New Session reuses a blank
   * one targeting the same workspace. Filtering stays with the consumer: the
@@ -613,9 +614,11 @@ export class SessionsService implements ISessions {
        id: entry.sessionId,
        displayTitle: displayTitleOf(entry.title, entry.cwd, entry.sessionId),
        running: entry.running,
-        waitingApproval: entry.waitingApproval,
        blank: entry.blank,
        updatedAt: entry.updatedAt,
+        ...(entry.pendingInteraction === undefined
+          ? {}
+          : { pendingInteraction: entry.pendingInteraction }),
        ...(entry.projectionValues === undefined
          ? {}
          : { projectionValues: entry.projectionValues }),
@@ -643,7 +646,6 @@ export class SessionsService implements ISessions {
            parentId: address.parentSessionId,
            origin: 'subagent',
            running: child.activity === 'running',
-            waitingApproval: false,
            blank: false,
            updatedAt: 0,
          }
--- a/packages/client/runtime/src/client/sessions/session.ts
+++ b/packages/client/runtime/src/client/sessions/session.ts
@@ -21,7 +21,7 @@ import { PendingWait } from './pending.ts'
 import { TranscriptAdapter } from './transcript-adapter.ts'
 import { displayFailureMessage } from './failure-display.ts'
 import { Notifier } from './notifier.ts'
-import { PartialAccumulator } from './partial.ts'
+import { isVisibleAssistantChunk, PartialAccumulator } from './partial.ts'
 import { ProjectionValueStore } from './projection-store.ts'
 import type { ProjectionsBaseline } from './projection-store.ts'

@@ -113,6 +113,11 @@ export class Session implements SessionFace {
  private pendingCache: { rev: number; value: PendingInteraction[] } | null = null
  private derivedRev = 0
  private nodesCache: { projected: readonly ConversationNode[]; derivedRev: number; value: readonly ConversationNode[] } | null = null
+  /** Exact turn timing retained from the raw window so presentation never
+   *  infers elapsed time from transcript content. */
+  private turnTimings = new Map<number, { startTime: number; endTime?: number }>()
+  private turnTimingsRev = 0
+  private turnTimingsCache: { rev: number; value: ConversationSnapshot['turnTimings'] } | null = null
  /** Completed turn boundaries retained from the raw window so presentation
   *  actions never infer a safe fork point from transcript content alone. */
  private turnEnds = new Map<number, number>()
@@ -687,6 +692,10 @@ export class Session implements SessionFace {
      return
    }
    this.appendLive(event, view)
+    if (event.type === 'assistant/chunk') {
+      if (isVisibleAssistantChunk(event.data.chunk.type)) this.notifier.markFrameDirty()
+      return
+    }
    this.notifier.markDirty()
  }

@@ -795,6 +804,8 @@ export class Session implements SessionFace {
    }
    switch (event.type) {
      case 'turn/start': {
+        this.turnTimings.set(event.data.turn, { startTime: event.time })
+        this.turnTimingsRev++
        if (event.data.trigger.kind === 'retry') this.settleScheduledRetry('started')
        return
      }
@@ -826,6 +837,11 @@ export class Session implements SessionFace {
        return
      }
      case 'turn/end': {
+        const timing = this.turnTimings.get(event.data.turn)
+        if (timing !== undefined) {
+          this.turnTimings.set(event.data.turn, { ...timing, endTime: event.time })
+          this.turnTimingsRev++
+        }
        this.turnEnds.set(event.data.turn, event.seq)
        this.turnEndsRev++
        if (event.data.reason.kind === 'aborted' || event.data.reason.kind === 'disposed') {
@@ -918,6 +934,8 @@ export class Session implements SessionFace {
    this.callsRev++
    this.derivedNodes = []
    this.derivedRev++
+    this.turnTimings = new Map()
+    this.turnTimingsRev++
    this.turnEnds = new Map()
    this.turnEndsRev++
    this.codeDispatches = new Map()
@@ -951,6 +969,9 @@ export class Session implements SessionFace {
    if (this.callsCache === null || this.callsCache.rev !== this.callsRev) {
      this.callsCache = { rev: this.callsRev, value: [...this.openCalls.values()] }
    }
+    if (this.turnTimingsCache === null || this.turnTimingsCache.rev !== this.turnTimingsRev) {
+      this.turnTimingsCache = { rev: this.turnTimingsRev, value: new Map(this.turnTimings) }
+    }
    if (this.turnEndsCache === null || this.turnEndsCache.rev !== this.turnEndsRev) {
      this.turnEndsCache = { rev: this.turnEndsRev, value: new Map(this.turnEnds) }
    }
@@ -967,6 +988,7 @@ export class Session implements SessionFace {
    return {
      sessionId: this.sessionId,
      nodes,
+      turnTimings: this.turnTimingsCache.value,
      turnEnds: this.turnEndsCache.value,
      partial,
      runningCalls: this.callsCache.value,
--- a/packages/client/runtime/tests/manager.spec.ts
+++ b/packages/client/runtime/tests/manager.spec.ts
@@ -40,6 +40,7 @@ describe('instances', () => {
    const manager = new SessionManager(api)
    // Uninstantiated: approval buffers, plain session/event drops.
    manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
+    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', payload: { approvalId: 'ap1' } }])
@@ -47,16 +48,26 @@ describe('instances', () => {
    expect(manager.get(S2).getSnapshot().pending).toEqual([])
  })

-  it('caps the pending buffer at 32 keeping the newest, and drops it on session-removed', () => {
+  it('retains every live answerable request and compacts resolutions before instantiation', () => {
    const api = new FakeApiClient()
    const manager = new SessionManager(api)
-    // 40 distinct question frames for an uninstantiated session: only the newest 32 survive.
+    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
    for (let i = 0; i < 40; i++) {
      manager.handleMuxEnvelope({ rpcId: `q${i}` as never, payload: { type: 'question/requested', sessionId: S1, questions: [] } })
    }
-    const pending = manager.get(S1).getSnapshot().pending
-    expect(pending).toHaveLength(32)
-    expect(pending.map(p => p.key)).toEqual(Array.from({ length: 32 }, (_, i) => `q:q${i + 8}`)) // oldest 8 dropped
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
+    for (let i = 0; i < 40; i++) {
+      manager.handleMuxEnvelope({
+        rpcId: `r${i}` as never,
+        payload: { type: 'question/resolved', sessionId: S1, questionRpcId: `q${i}` as never, outcome: 'answered' },
+      })
+    }
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
+    expect(manager.get(S1).getSnapshot().pending).toEqual([])
+  })
+
+  it('drops buffered answerable requests on session removal', () => {
+    const manager = new SessionManager(new FakeApiClient())
    // 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 } })
@@ -862,48 +873,102 @@ describe('connected generation', () => {
  })
 })

-describe('waiting-approval list bit', () => {
-  it('lights on requested, survives replay duplicates, and clears on resolved — without instantiation', () => {
+describe('pending-interaction list status', () => {
+  it('tracks approval requests through replay and resolution without instantiation', () => {
    const manager = new SessionManager(new FakeApiClient())
    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(false)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
    manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(true)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
    // Mux-open replay of the same question (same approvalId) is idempotent.
    manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(true)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
    manager.handleMuxEnvelope({ rpcId: 'rx' as never, payload: { type: 'approval/resolved', sessionId: S1, approvalId: 'ap1' as never, outcome: 'allowed-once' as never } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(false)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  })

-  it('clears only when the last outstanding question resolves; session-removed drops the bit', () => {
+  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({
+      rpcId: 'q1' as never,
+      payload: { type: 'question/requested', sessionId: S1, questions: [{ id: 'name', question: 'Name?' }] },
+    })
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
+    manager.handleMuxEnvelope({ rpcId: 'qx' as never, payload: { type: 'question/resolved', sessionId: S1, questionRpcId: 'q1' as never, outcome: 'answered' } })
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
+
+    manager.handleMuxEnvelope({
+      rpcId: 'q2' as never,
+      payload: {
+        type: 'question/requested',
+        sessionId: S1,
+        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')
+    manager.handleMuxEnvelope({ rpcId: 'qy' as never, payload: { type: 'question/resolved', sessionId: S1, questionRpcId: 'q2' as never, outcome: 'cancelled' } })
+    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 } })
    manager.handleMuxEnvelope({ rpcId: 'r1' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'a1' as never, toolName: 'rm' } })
-    manager.handleMuxEnvelope({ rpcId: 'r2' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'a2' as never, toolName: 'rm' } })
+    manager.handleMuxEnvelope({
+      rpcId: 'q1' as never,
+      payload: { type: 'question/requested', sessionId: S1, questions: [{ id: 'name', question: 'Name?' }] },
+    })
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
+    manager.handleMuxEnvelope({ rpcId: 'qy' as never, payload: { type: 'question/resolved', sessionId: S1, questionRpcId: 'q1' as never, outcome: 'answered' } })
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
    manager.handleMuxEnvelope({ rpcId: 'rx' as never, payload: { type: 'approval/resolved', sessionId: S1, approvalId: 'a1' as never, outcome: 'rejected' as never } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(true)
-    manager.handleMuxEnvelope({ rpcId: 'ry' as never, payload: { type: 'approval/resolved', sessionId: S1, approvalId: 'a2' as never, outcome: 'rejected' as never } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(false)
-    // Removed sessions drop their bit outright.
-    manager.handleMuxEnvelope({ rpcId: 'r3' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'a3' as never, toolName: 'rm' } })
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
+
+    manager.handleMuxEnvelope({ rpcId: 'r2' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'a2' as never, toolName: 'rm' } })
    manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-removed', sessionId: S1 } })
    expect(manager.getListSnapshot().items).toHaveLength(0)
  })

-  it('drops stale bits at generation death — BEFORE the reopen replay re-adds still-pending questions', () => {
+  it('drops stale status at generation death before replay re-adds live interactions', () => {
    const manager = new SessionManager(new FakeApiClient())
    manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
    manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(true)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
    // Generation death clears (resolved-while-disconnected questions send no frame)…
    manager.handleDisconnected()
-    expect(manager.getListSnapshot().items[0]?.waitingApproval).toBe(false)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
    // …and a replayed frame arriving before onConnected (stream open precedes
    // the readiness handshake) survives the later handleConnected 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]?.waitingApproval).toBe(true)
+    expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
  })

  it('generation death drops buffered answerable frames (a dead generation cannot be answered)', () => {
--- a/packages/client/runtime/tests/notifier.spec.ts
+++ b/packages/client/runtime/tests/notifier.spec.ts
@@ -1,13 +1,17 @@
 /**
- * Notifier: microtask batching, rebuild-before-notify ordering, no-listener
- * laziness, synchronous notifyNow, and unsubscribe.
+ * Notifier: microtask/frame batching, rebuild-before-notify ordering,
+ * no-listener laziness, synchronous notifyNow, and unsubscribe.
 */

-import { describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Notifier } from '../src/client/sessions/notifier.ts'

 const microtask = (): Promise<void> => new Promise((resolve) => { queueMicrotask(resolve) })

+afterEach(() => {
+  vi.unstubAllGlobals()
+})
+
 describe('Notifier', () => {
  it('collapses N markDirty calls into one flush, rebuilding before notifying', async () => {
    const order: string[] = []
@@ -60,6 +64,57 @@ describe('Notifier', () => {
    expect(rebuilds).toBe(1)
  })

+  it('collapses frame-dirty changes into one cumulative frame publication', () => {
+    const frames: FrameRequestCallback[] = []
+    vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) => {
+      frames.push(callback)
+      return frames.length
+    })
+    const order: string[] = []
+    const notifier = new Notifier(() => order.push('rebuild'))
+    notifier.subscribe(() => order.push('notify'))
+
+    notifier.markFrameDirty()
+    notifier.markFrameDirty()
+    notifier.markFrameDirty()
+
+    expect(order).toEqual([])
+    expect(frames).toHaveLength(1)
+    frames.shift()!(0)
+    expect(order).toEqual(['rebuild', 'notify'])
+  })
+
+  it('lets a structural microtask publication supersede a pending frame', async () => {
+    const frames: FrameRequestCallback[] = []
+    vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) => {
+      frames.push(callback)
+      return frames.length
+    })
+    let notifications = 0
+    const notifier = new Notifier(() => undefined)
+    notifier.subscribe(() => { notifications++ })
+
+    notifier.markFrameDirty()
+    notifier.markDirty()
+    await microtask()
+    expect(notifications).toBe(1)
+
+    frames.shift()!(0)
+    expect(notifications).toBe(1)
+  })
+
+  it('falls back to microtask batching when animation frames are unavailable', async () => {
+    let notifications = 0
+    const notifier = new Notifier(() => undefined)
+    notifier.subscribe(() => { notifications++ })
+
+    notifier.markFrameDirty()
+    notifier.markFrameDirty()
+    expect(notifications).toBe(0)
+    await microtask()
+    expect(notifications).toBe(1)
+  })
+
  it('unsubscribed listeners stop receiving notifications', async () => {
    let calls = 0
    const notifier = new Notifier(() => undefined)
--- a/packages/client/runtime/tests/session.spec.ts
+++ b/packages/client/runtime/tests/session.spec.ts
@@ -6,7 +6,7 @@
 * enough.
 */

-import { describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
@@ -20,6 +20,10 @@ const at = (seq: number, e: Record<string, unknown>): SessionEvent =>
 const SID = 'fk-s1' as SessionId
 const PARENT = 'fk-parent' as SessionId

+afterEach(() => {
+  vi.unstubAllGlobals()
+})
+
 function makeSession(api = new FakeApiClient()): { api: FakeApiClient; session: Session } {
  return { api, session: new Session(SID, api) }
 }
@@ -42,6 +46,11 @@ describe('open', () => {
    expect(snapshot.openState).toBe('open')
    expect(snapshot.hasMore).toBe(true)
    expect(snapshot.nodes.map(n => n.kind)).toEqual(['user', 'assistant'])
+    expect(snapshot.turnTimings.get(3)).toEqual({
+      startTime: 1_700_000_000_010,
+      endTime: 1_700_000_000_015,
+    })
+    expect(snapshot.turnEnds.get(3)).toBe(15)
  })

  it('is idempotent: concurrent opens share one history call, reopening when open is a no-op', async () => {
@@ -163,6 +172,40 @@ describe('live event path', () => {
    expect((last as { interrupted?: true }).interrupted).toBeUndefined()
  })

+  it('publishes cumulative chunks once per frame and lets finalization supersede the pending frame', async () => {
+    const frames: FrameRequestCallback[] = []
+    vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) => {
+      frames.push(callback)
+      return frames.length
+    })
+    const { session } = await opened()
+    const published: Array<string | null> = []
+    session.subscribe(() => {
+      const block = session.getSnapshot().partial?.blocks[0]
+      published.push(block?.kind === 'text' ? block.text : null)
+    })
+    const feed = (event: SessionEvent) => {
+      session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event })
+    }
+
+    feed(ev.chunkStart(6, 1))
+    feed(ev.chunkText(7, 1, '累'))
+    feed(ev.chunkText(8, 1, '计'))
+    expect(published).toEqual([])
+    expect(frames).toHaveLength(1)
+
+    frames.shift()!(0)
+    expect(published).toEqual(['累计'])
+
+    feed(ev.chunkText(9, 1, '完成'))
+    feed(ev.assistant(10, 1, '累计完成'))
+    await Promise.resolve()
+    expect(published).toEqual(['累计', null])
+
+    frames.shift()!(0)
+    expect(published).toEqual(['累计', null])
+  })
+
  it('retracts the failed step partial on retry and keeps a replayable notice before the recovered response', async () => {
    const { session } = await opened()
    const feed = (event: SessionEvent) => { session.handleMuxEnvelope('r' as never, { type: 'session/event', sessionId: SID, event }) }
@@ -215,11 +258,22 @@ describe('live event path', () => {
    expect(snapshot.nodes.some(node => node.kind === 'turn-error')).toBe(false)
    expect(snapshot.nodes.at(-2)).toMatchObject({ kind: 'model-retry', retryState: 'started' })
    expect(snapshot.nodes.at(-1)).toMatchObject({ kind: 'assistant', blocks: [{ kind: 'text', text: '完整回复' }] })
+    const retryStart = retryTurn.find(event =>
+      event.type === 'turn/start' && event.data.trigger.kind === 'retry')
+    if (retryStart?.type !== 'turn/start') throw new Error('test fixture must include a retry turn/start')
+    const retryEnd = retryTurn.find(event =>
+      event.type === 'turn/end' && event.data.turn === retryStart.data.turn)
+    if (retryEnd?.type !== 'turn/end') throw new Error('test fixture must complete the retry turn')
+    expect(snapshot.turnTimings.get(retryStart.data.turn)).toEqual({
+      startTime: retryStart.time,
+      endTime: retryEnd.time,
+    })

    const replay = makeSession()
    replay.api.onHistory = () => histResponse([...plainTurn(0, 0, 'a', 'b'), ...retryTurn])
    await replay.session.open()
    expect(replay.session.getSnapshot().nodes).toEqual(snapshot.nodes)
+    expect(replay.session.getSnapshot().turnTimings).toEqual(snapshot.turnTimings)
    expect(replay.session.getSnapshot().partial).toBeNull()
  })

@@ -1216,6 +1270,7 @@ describe('reference stability (the memo contract)', () => {
    expect(after).not.toBe(before)
    expect(after.runningCalls).toBe(before.runningCalls)
    expect(after.pending).toBe(before.pending)
+    expect(after.turnTimings).toBe(before.turnTimings)
    expect(after.turnEnds).toBe(before.turnEnds)
    // And a mutation on the tracked domain swaps that array.
    feed(ev.toolResult(11, 1, 'c1', 'ECHO'))