Merge remote-tracking branch 'origin/master' into worktree/web-model-request-retry

# Conflicts:
#	apps/cli/README.i18n.yaml
#	apps/cli/README.md
#	apps/cli/README.zh.md
#	apps/web/tests/snapshots/live-interactions/cancel.expected.md
#	apps/web/tests/snapshots/live-interactions/error-auth.expected.md
#	apps/web/tests/snapshots/live-interactions/retry.expected.md
#	packages/client/runtime/README.i18n.yaml
#	packages/client/runtime/src/client/sessions/session.ts
#	packages/client/ui-conversation/README.i18n.yaml
#	packages/client/ui-conversation/README.zh.md

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: f62bce11beb746368683bd25ece61413e12b5387
-README.zh.md: 3cd3ac04b58a8663424cf459d6c882cb8c67aec4
+README.md: 8d95abb38c54a005efb3a7b383657c85d47736a7
+README.zh.md: e065f0c1e96eb87dd362eebf3688c6fed4af5d89

packages/client/runtime/README.md

@@ -16,6 +16,10 @@ SlotsService gives the renderer separate bare observables for `useSessions` and
 
 `WorkspacesService.connectWorkspace(workspaceId)` resolves the session a New Session flow lands in: it reuses the workspace's existing blank session from the list mirror (`blank && cwd == workspace.path`) or calls `session.create({workspaceId})`, returning the session id for the caller to open. `SessionSummary.blank` mirrors the host's derived empty-log bit and only ever lowers on the client: seeded by `session.list` / the `host/session-added` frame, flipped false by the first ACCEPTED local `prompt()` (on the RPC success response — acceptance proves the user message is in the host log; a rejected first prompt keeps the session blank and reusable) and by any `running: true` status frame, re-aligned by every list re-pull. List surfaces hide blank rows; the store carries every row. `SessionsService.create` accepts an optional caller-preallocated SessionId and throws `SessionCreateError` (carrying `requestedSessionId`) on failure.
 
+## Pending queue projection
+
+`ConversationSnapshot.queue` is the Host's authoritative transient Queue snapshot; pending steering stays outside this projection. Each row carries its `InboxItemId`, complete editable text when every content block is text, and a flattened preview. `session/queue` replaces the whole projection; reconnect buffering retains only the latest snapshot, and neither durable turn events nor running-status changes guess that an item was claimed. `Session.updateQueue()` sends edit/remove operations without optimistic mutation, so the next Host snapshot is the sole visible commit and a claim race can surface `queue-item-not-found`.
+
 ## Code Mode sub-dispatch index
 
 `ConversationSnapshot.codeDispatches` groups a `run_code` call's sub-dispatches under their parent callId, in start order, using the native call-block shapes: a `tool/code-dispatch-start` event lands the `RunningToolCall` form (rows derive the running ring from the shape) and its `tool/code-dispatch` settlement replaces it in place with the `ToolResultNode` form, `callTime` carrying the paired start's time. A settle whose start fell outside the replay window appends directly with `callTime: null` (duration unknown — never a fabricated zero). Live mux frames and history replay build the identical index; sub-calls never join the surface `nodes` flow; per-parent array and map references are memo-stable across unrelated snapshot swaps.

packages/client/runtime/README.zh.md

@@ -16,6 +16,10 @@ SlotsService 分别为 renderer 提供 `useSessions` 与 `useWorkspaces` 的裸
 
 `WorkspacesService.connectWorkspace(workspaceId)` 解析 New Session 流程最终落入的会话：先在列表镜像中复用该 workspace 的既有空会话（`blank && cwd == workspace.path`），未命中则调用 `session.create({workspaceId})`，返回会话 id 由调用方 open。`SessionSummary.blank` 镜像主机派生的空日志位，在客户端只降不升：由 `session.list`／`host/session-added` 帧播种，本地首次获 Host 接受的 `prompt()`（RPC 成功响应时——受理即证明用户消息已入主机日志；首讯被拒则会话保持 blank、保持可复用）与任何 `running: true` 状态帧翻为 false，每次列表重拉重新对齐。列表界面隐藏 blank 行；store 保留全部行。`SessionsService.create` 接受可选的、由调用方预先分配的 SessionId，失败时抛出 `SessionCreateError`（携带 `requestedSessionId`）。
 
+## 待处理队列投影
+
+`ConversationSnapshot.queue` 是 Host 提供的权威瞬态 Queue 快照；待处理 steering（中途引导）不进入此投影。每行都携带其 `InboxItemId`、所有内容块均为文本时的完整可编辑文本，以及扁平化预览。`session/queue` 会整体替换该投影；重连缓冲只保留最新快照，持久轮次事件和 running 状态变化都不会猜测某个项已被认领。`Session.updateQueue()` 发送编辑／移除操作，不进行乐观更新，因此下一份 Host 快照是唯一可见的提交结果，认领竞态则会返回 `queue-item-not-found`。
+
 ## Code Mode 子调用索引
 
 `ConversationSnapshot.codeDispatches` 按父调用的 callId 和启动顺序，用原生调用块形状组织一个 `run_code` 调用的子调用：`tool/code-dispatch-start` 事件落成 `RunningToolCall` 形状（行组件从该形状推导运行中的转圈状态），其 `tool/code-dispatch` 完结事件原位替换为 `ToolResultNode` 形状，`callTime` 携带成对 start 事件的时间。start 落在回放窗口之外的完结事件则直接追加，`callTime: null`（耗时未知——绝不伪造零耗时）。live mux 帧与历史回放构建相同的索引；子调用永不进入 surface `nodes` 流；无关快照交换不会改变每个父调用对应的数组引用和映射引用，两者均保持 memo 稳定。

packages/client/runtime/src/client/contract/session.ts

@@ -8,7 +8,9 @@
  * dispatch) stay on the class, invisible out here.
  */
 import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
-import type { RpcResult, SessionId } from '@deepseek-ai/dsh-client-connection/client'
+import type {
+  InboxItemId, QueueAction, RpcResult, SessionId,
+} from '@deepseek-ai/dsh-client-connection/client'
 import type { ConversationSnapshot } from '../sessions/conversation.ts'
 import type { ObservableSnapshot } from './store.ts'
 
@@ -36,6 +38,13 @@ export interface ISession {
    * @returns acceptance, or the business error (also mirrored into snapshot.promptError).
    */
   prompt(content: ContentBlock[], mode: 'queue' | 'steer'): Promise<RpcResult<{ accepted: true }>>
+  /**
+   * Apply one mutation to a still-pending queue occurrence.
+   * @param itemId - agent-owned inbox occurrence identity.
+   * @param action - edit or remove operation.
+   * @returns acceptance, or a business/transport error.
+   */
+  updateQueue(itemId: InboxItemId, action: QueueAction): Promise<RpcResult<{ accepted: true }>>
   /**
    * Cancel the running turn.
    * @returns acceptance, or the business error.

packages/client/runtime/src/client/sessions/conversation.ts

@@ -8,7 +8,7 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
 import type { LlmRetryEventData } from '@deepseek-ai/dsh-llm-retry/types'
 import type { TodoItem } from '@deepseek-ai/dsh-session/types'
 import type {
-  RpcError, SessionId, ToolCallView, ToolResultView,
+  InboxItemId, RpcError, SessionId, ToolCallView, ToolResultView,
 } from '@deepseek-ai/dsh-client-connection/client'
 import type { PendingInteraction } from './pending.ts'
 export type { TodoItem }
@@ -231,10 +231,12 @@ export interface RunningToolCall {
 }
 
 
-/** One queued-message row mirrored from `session/queued` frames (key: the enqueueing prompt's rpcId when wire-sourced). */
+/** One independently addressable row from the transient queue snapshot. */
 export interface QueuedMessage {
-  readonly key: string
+  readonly id: InboxItemId
   readonly preview: string
+  /** Complete editable text; null when the message contains non-text blocks. */
+  readonly text: string | null
 }
 
 /** In-progress assistant output (chunk accumulator product). */
@@ -292,7 +294,7 @@ export interface ConversationSnapshot {
    */
   codeDispatches: ReadonlyMap<string, readonly CodeSubCall[]>
   pending: readonly PendingInteraction[]
-  /** Read-only inbox mirror (session/queued frames + mux-open baseline; cleared by the leave-running flip). */
+  /** Authoritative transient inbox snapshot, replaced after every host-side change. */
   queue: readonly QueuedMessage[]
   running: boolean
   /** Input-area shape (see {@link ComposerPhase}); derived here, switched on by consumers. */

packages/client/runtime/src/client/sessions/manager.ts

@@ -351,14 +351,14 @@ 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/queued frames belong to
+      // 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.
       const buffered = this.pendingBuffers.get(frame.sessionId)
       if (buffered !== undefined) {
-        const kept = buffered.filter(item => item.payload.type !== 'session/queued')
+        const kept = buffered.filter(item => item.payload.type !== 'session/queue')
         if (kept.length !== buffered.length) {
           if (kept.length === 0) this.pendingBuffers.delete(frame.sessionId)
           else this.pendingBuffers.set(frame.sessionId, kept)
@@ -383,7 +383,7 @@ export class SessionManager {
     }
     const session = this.sessions.get(frame.sessionId)
     if (session === undefined) {
-      // Approval/question/queued frames never hit history: buffer for replay on
+      // Approval/question/queue frames never hit history: buffer for replay on
       // instantiation; everything else drops (not instantiated — history fully
       // backfills on open).
       switch (frame.type) {
@@ -391,8 +391,12 @@ export class SessionManager {
         case 'approval/resolved':
         case 'question/requested':
         case 'question/resolved':
-        case 'session/queued': {
+        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)
           this.pendingBuffers.set(frame.sessionId, buffer)

packages/client/runtime/src/client/sessions/session.ts

@@ -5,8 +5,8 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
 import type { LlmRetryEventData } from '@deepseek-ai/dsh-llm-retry/types'
 import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import type {
-  HistoryEntry, IApiClient, MuxFrame, RpcError, RpcId, RpcResult,
-  SessionId, ToolEventView,
+  HistoryEntry, IApiClient, InboxItemId, MuxFrame, QueueAction, RpcError,
+  RpcId, RpcResult, SessionId, ToolEventView,
 } from '@deepseek-ai/dsh-client-connection/client'
 // Value import from the inline-safe wire layer (not the connection plugin):
 // plugin-to-plugin value imports are a bundle purity error.
@@ -53,14 +53,6 @@ export interface SessionOptions {
 /** Queue-row preview cap: the dock renders one line, the full content never leaves the host mirror. */
 const QUEUE_PREVIEW_CHARS = 200
 
-/** Internal inbox-mirror entry: the snapshot row plus the retirement-matching fields the frames carry. */
-interface QueuedEntry {
-  row: QueuedMessage
-  steering: boolean
-  /** JSON-serialized MessageSource (steering retirement matches by source, the host-mirror precedent). */
-  sourceJson: string
-}
-
 /** Single-line queue-row preview: text blocks flattened, non-text as tags, capped by code point. */
 function queuePreviewOf(content: readonly ContentBlock[]): string {
   const flat = content
@@ -70,6 +62,12 @@ function queuePreviewOf(content: readonly ContentBlock[]): string {
   return chars.length > QUEUE_PREVIEW_CHARS ? `${chars.slice(0, QUEUE_PREVIEW_CHARS).join('')}…` : flat
 }
 
+/** Recover complete composer text only when editing cannot discard non-text blocks. */
+function queueTextOf(content: readonly ContentBlock[]): string | null {
+  if (!content.every(block => block.type === 'text')) return null
+  return content.map(block => block.text).join('')
+}
+
 /**
  * Owns a session's event window, derived conversation state, and observable
  * snapshot. React bindings remain outside this data layer. Features see only
@@ -109,9 +107,8 @@ export class Session implements SessionFace {
   private pendingCache: { rev: number; value: PendingInteraction[] } | null = null
   private derivedRev = 0
   private nodesCache: { folded: readonly ConversationNode[]; derivedRev: number; value: readonly ConversationNode[] } | null = null
-  /** Inbox mirror (session/queued frames + mux-open baseline). Queue frames never hit history,
-   *  so this is stream-only state: reconnect clears it and the fresh baseline re-populates. */
-  private queued: QueuedEntry[] = []
+  /** Authoritative stream-only inbox snapshot; pending work never hits history. */
+  private queued: QueuedMessage[] = []
   private queueRev = 0
   private queueCache: { rev: number; value: QueuedMessage[] } | null = null
   /** `run_code` sub-dispatches by parent callId (window-derived, like openCalls). Appends
@@ -239,6 +236,15 @@ export class Session implements SessionFace {
     return result
   }
 
+  /** Apply one operation to a still-pending queue occurrence. */
+  async updateQueue(itemId: InboxItemId, action: QueueAction): Promise<RpcResult<{ accepted: true }>> {
+    try {
+      return (await this.api.sessions.updateQueue({ sessionId: this.sessionId, itemId, action })).result
+    } catch (error) {
+      return transportError(error)
+    }
+  }
+
   /**
    * Stop: contract session.cancel 1:1; failures land in promptError (same error-strip display slot).
    * @returns the cancel result.
@@ -398,20 +404,15 @@ export class Session implements SessionFace {
   handleMuxEnvelope(rpcId: RpcId, frame: MuxFrame): void {
     switch (frame.type) {
       case 'session/event': {
-        this.retireQueued(frame.event)
         this.acceptLiveEvent(frame.event, frame.view)
         return
       }
-      case 'session/queued': {
-        const message = frame.message
-        // Row key: the enqueueing prompt's rpcId when it rode this wire (the
-        // provisional-echo reconciliation key); otherwise the frame envelope id.
-        const key = 'rpcId' in message.source ? String(message.source.rpcId) : `f:${rpcId}`
-        this.queued.push({
-          row: { key, preview: queuePreviewOf(message.content) },
-          steering: frame.steering,
-          sourceJson: JSON.stringify(message.source),
-        })
+      case 'session/queue': {
+        this.queued = frame.items.map(item => ({
+          id: item.id,
+          preview: queuePreviewOf(item.message.content),
+          text: queueTextOf(item.message.content),
+        }))
         this.queueRev++
         this.notifier.markDirty()
         return
@@ -464,15 +465,6 @@ export class Session implements SessionFace {
    * @param running - the new running state.
    */
   handleRunning(running: boolean): void {
-    // Leave-running sweep (host queuedMirror precedent): discard paths (cancel,
-    // terminal steering drop) have no per-entry frame, so ANY not-running signal
-    // with a nonempty mirror clears it — checked before the equality return so a
-    // stale replay on an already-idle session still sweeps.
-    if (!running && this.queued.length > 0) {
-      this.queued = []
-      this.queueRev++
-      this.notifier.markDirty()
-    }
     // Turn-start conversion: a blank session never runs, so the first
     // running:true proves another端's first message landed (设计稿 2.2).
     if (running && this.blankBit) {
@@ -637,27 +629,6 @@ export class Session implements SessionFace {
     }
   }
 
-  /** Consumption-event retirement, mirroring the host queuedMirror rules: a message-triggered
-   *  turn/start claims the oldest non-steering entry; a steering/message drains the oldest
-   *  steering entry with the same source (loop-authored steering matches nothing and drops none). */
-  private retireQueued(event: SessionEvent): void {
-    if (this.queued.length === 0) return
-    let index = -1
-    if (event.type === 'turn/start') {
-      if (event.data.trigger.kind !== 'message') return
-      index = this.queued.findIndex(entry => !entry.steering)
-    } else if (event.type === 'steering/message') {
-      const source = JSON.stringify(event.data.message.source)
-      index = this.queued.findIndex(entry => entry.steering && entry.sourceJson === source)
-    } else {
-      return
-    }
-    if (index < 0) return
-    this.queued.splice(index, 1)
-    this.queueRev++
-    this.notifier.markDirty()
-  }
-
   /** Per-event side effects (right column of the §A.9 dispatch table):
    *  chunk/retry projection and openCalls add-remove. */
   private applyEventSideEffects(event: SessionEvent, view?: ToolEventView): void {
@@ -885,7 +856,7 @@ export class Session implements SessionFace {
       this.dispatchesCache = { rev: this.dispatchesRev, value: new Map(this.codeDispatches) }
     }
     if (this.queueCache === null || this.queueCache.rev !== this.queueRev) {
-      this.queueCache = { rev: this.queueRev, value: this.queued.map(entry => entry.row) }
+      this.queueCache = { rev: this.queueRev, value: this.queued }
     }
     const partial = this.partial?.toPartial() ?? null
     return {

packages/client/runtime/src/client/slots.ts

@@ -18,7 +18,7 @@ import { Service } from 'cordis'
 import type { Context } from 'cordis'
 import { SlotCore } from '@deepseek-ai/dsh-client-ui-slots'
 import type {
-  OwnerOf, SlotEntryDef, SlotMap, SlotRenderer, SlotRendererHost,
+  LocaleFace, OwnerOf, SlotEntryDef, SlotMap, SlotRenderer, SlotRendererHost,
   SlotScope, SlotSpec, StoreDecl, StoredEntry, StoreInstanceLike,
 } from '@deepseek-ai/dsh-client-ui-slots'
 
@@ -70,6 +70,8 @@ interface ErasedRegisterOptions {
   select?: (owner: never) => unknown
   /** Chain-slot explicit ordering override (ascending; registration order otherwise). */
   priority?: number
+  /** Declared dictionary namespace (the renderer synthesizes the `t` seat from it). */
+  locale?: string
   registrant?: string
 }
 
@@ -82,6 +84,7 @@ export class SlotsService extends Service {
   /** Store-instance axis: handle -> mounted scope, refcount, resolved instances. */
   private readonly _stores = new Map<EngineStoreHandle, StoreAxisRecord>()
   private _renderer: SlotRenderer | undefined
+  private _locale: LocaleFace | undefined
   private _host: SlotRendererHost | undefined
 
   /**
@@ -127,6 +130,23 @@ export class SlotsService extends Service {
     }, 'slots.install()')
   }
 
+  /**
+   * Install the locale face backing the `t` standard seat (the locale
+   * plugin's product; same boot-once discipline as the renderer install).
+   * Runs through the caller's ctx.effect, so the installing fiber's unload
+   * uninstalls the face.
+   * @param face - namespace binder + revision observable.
+   */
+  installLocale(face: LocaleFace): void {
+    if (this._locale !== undefined) throw new Error('locale face already installed (installLocale() is boot-once)')
+    this.ctx.effect(() => {
+      this._locale = face
+      return () => {
+        if (this._locale === face) this._locale = undefined
+      }
+    }, 'slots.installLocale()')
+  }
+
   /**
    * The single ctx-level render entry: the shell renders 'root'; every other
    * key renders inside components through the props renderSlot face. All
@@ -246,6 +266,12 @@ export class SlotsService extends Service {
     if (workspaces === undefined) {
       throw new Error("renderSlot('root') before the workspaces service mounted — boot order puts runtime apply first")
     }
+    // `locale` is a live getter: the face installs (and, under HMR, swaps)
+    // on the locale plugin's own fiber lifetime, while this host object is
+    // built once — a captured value would strand renders on a dead face. The
+    // alias is required: `this` inside the getter is the host literal.
+    // oxlint-disable-next-line typescript/no-this-alias
+    const service = this
     this._host = {
       subscribe: (key, fn) => this._core.subscribe(key, fn),
       getVersion: key => this._core.getVersion(key),
@@ -259,6 +285,7 @@ export class SlotsService extends Service {
         provideInfo: sessions.currentProvideInfo,
       },
       workspaces: { list: workspaces.list },
+      get locale() { return service._locale },
     }
     return this._host
   }

packages/client/runtime/tests/fake-api.ts

@@ -81,6 +81,7 @@ export class FakeApiClient implements IApiClient {
   Promise<RpcResponse<{ selected: ModelTarget }>> =
     payload => Promise.resolve(ok({ selected: { provider: payload.provider, model: payload.model } }))
   onPrompt: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
+  onUpdateQueue: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
   onCancel: (payload: unknown) => Promise<RpcResponse<{ accepted: true }>> = () => Promise.resolve(ok({ accepted: true as const }))
 
   onDescribe: (payload: unknown) => Promise<RpcResponse<{ version: string; cwd: string; attachedSessions: number }>> =
@@ -118,6 +119,7 @@ export class FakeApiClient implements IApiClient {
       this.record('session.selectModel', payload, this.onSelectModel(payload)),
     rename: (payload: unknown) => this.record('session.rename', payload, this.onRename(payload)),
     prompt: (payload: unknown) => this.record('session.prompt', payload, this.onPrompt(payload)),
+    updateQueue: (payload: unknown) => this.record('session.updateQueue', payload, this.onUpdateQueue(payload)),
     cancel: (payload: unknown) => this.record('session.cancel', payload, this.onCancel(payload)),
   }
 

packages/client/runtime/tests/queue-store.spec.ts

@@ -1,32 +1,41 @@
 /**
- * Queue mirror semantics (web input-triggers queue cut 1): session/queued
- * intake, host-rule retirement (message turn/start claims oldest non-steering;
- * steering/message drains by source), leave-running sweep, reconnect reset,
- * pre-instantiation buffering, and snapshot reference stability.
+ * Queue snapshot semantics: authoritative replacement after every host-side
+ * change, reconnect re-baselining, pre-instantiation buffering, editable-text
+ * projection, and snapshot reference stability.
  */
 import { describe, expect, it } from 'vitest'
 import { createUserMessage } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
-import type { MuxFrame, RpcId, SessionId } from '@deepseek-ai/dsh-client-connection/client'
+import type {
+  InboxItemId, MuxFrame, RpcId, SessionId,
+} from '@deepseek-ai/dsh-client-connection/client'
 import { Session } from '../src/client/sessions/session.ts'
 import { SessionManager } from '../src/client/sessions/manager.ts'
 import { FakeApiClient } from './fake-api.ts'
-import { ev } from './event-script.ts'
 
 const SID = 'fk-q1' as SessionId
-const text = (t: string): ContentBlock[] => [{ type: 'text', text: t }]
+const text = (value: string): ContentBlock[] => [{ type: 'text', text: value }]
 const rid = (id: string): RpcId => id as RpcId
+const iid = (id: string): InboxItemId => id as InboxItemId
 
-/** session/queued frame with the wire-sourced rpcId key (the host prompt path). */
-function queuedFrame(body: string, rpcId: string, steering = false): MuxFrame {
+interface QueueFixture {
+  id: string
+  body: string
+  content?: ContentBlock[]
+}
+
+/** Build one authoritative queue snapshot. */
+function queueFrame(items: QueueFixture[]): MuxFrame {
   return {
-    type: 'session/queued',
+    type: 'session/queue',
     sessionId: SID,
-    message: createUserMessage({
-      content: text(body),
-      source: { kind: 'user', rpcId: rid(rpcId) } as never,
-    }),
-    steering,
+    items: items.map(item => ({
+      id: iid(item.id),
+      message: createUserMessage({
+        content: item.content ?? text(item.body),
+        source: { kind: 'user', rpcId: rid(`rpc-${item.id}`) } as never,
+      }),
+    })),
   }
 }
 
@@ -34,201 +43,131 @@ function makeSession(): Session {
   return new Session(SID, new FakeApiClient())
 }
 
-describe('queue intake', () => {
-  it('lands a queued frame as a row keyed by the source rpcId with a flat preview', () => {
+describe('queue snapshot intake', () => {
+  it('projects stable ids, flat previews, and complete text', () => {
     const session = makeSession()
-    session.handleMuxEnvelope(rid('env-1'), queuedFrame('第一条  排队\n消息', 'p-1'))
-    const queue = session.getSnapshot().queue
-    expect(queue).toEqual([{ key: 'p-1', preview: '第一条 排队 消息' }])
+    session.handleMuxEnvelope(rid('env-1'), queueFrame([
+      { id: 'q-1', body: '第一条  排队\n消息' },
+    ]))
+    expect(session.getSnapshot().queue).toEqual([
+      { id: 'q-1', preview: '第一条 排队 消息', text: '第一条  排队\n消息' },
+    ])
   })
 
-  it('falls back to the envelope rpcId when the source carries none, and tags non-text blocks', () => {
+  it('marks mixed-content messages non-editable while retaining their preview', () => {
     const session = makeSession()
-    session.handleMuxEnvelope(rid('env-2'), {
-      type: 'session/queued',
-      sessionId: SID,
-      message: createUserMessage({
-        content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
-        source: { kind: 'plugin', plugin: 'loop' },
-      }),
-      steering: false,
-    })
-    expect(session.getSnapshot().queue).toEqual([{ key: 'f:env-2', preview: 'hi [image]' }])
+    session.handleMuxEnvelope(rid('env-2'), queueFrame([{
+      id: 'q-image',
+      body: '',
+      content: [{ type: 'text', text: 'hi' }, { type: 'image', data: 'x' } as never],
+    }]))
+    expect(session.getSnapshot().queue).toEqual([
+      { id: 'q-image', preview: 'hi [image]', text: null },
+    ])
   })
 
-  it('caps the preview at 200 code points with an ellipsis', () => {
+  it('caps previews at 200 code points and preserves the full editable text', () => {
     const session = makeSession()
-    session.handleMuxEnvelope(rid('env-3'), queuedFrame('长'.repeat(201), 'p-cap'))
-    const preview = session.getSnapshot().queue[0]?.preview ?? ''
-    expect(Array.from(preview)).toHaveLength(201) // 200 + …
-    expect(preview.endsWith('…')).toBe(true)
+    const body = '长'.repeat(201)
+    session.handleMuxEnvelope(rid('env-3'), queueFrame([{ id: 'q-cap', body }]))
+    const row = session.getSnapshot().queue[0]
+    expect(Array.from(row?.preview ?? '')).toHaveLength(201)
+    expect(row?.preview.endsWith('…')).toBe(true)
+    expect(row?.text).toBe(body)
+  })
+
+  it('replaces content, order, and membership from each authoritative frame', () => {
+    const session = makeSession()
+    session.handleMuxEnvelope(rid('env-4'), queueFrame([
+      { id: 'q-1', body: 'one' },
+      { id: 'q-2', body: 'two' },
+    ]))
+    session.handleMuxEnvelope(rid('env-5'), queueFrame([
+      { id: 'q-2', body: 'two edited' },
+    ]))
+    expect(session.getSnapshot().queue).toEqual([
+      { id: 'q-2', preview: 'two edited', text: 'two edited' },
+    ])
+    session.handleMuxEnvelope(rid('env-6'), queueFrame([]))
+    expect(session.getSnapshot().queue).toEqual([])
   })
 
   it('keeps the queue array reference stable across unrelated snapshot swaps', () => {
     const session = makeSession()
-    session.handleMuxEnvelope(rid('env-4'), queuedFrame('稳定', 'p-s'))
+    session.handleMuxEnvelope(rid('env-7'), queueFrame([{ id: 'q-stable', body: '稳定' }]))
     const before = session.getSnapshot().queue
-    session.handleAgentError('unrelated') // dirties the snapshot without touching the queue
+    session.handleAgentError('unrelated')
     expect(session.getSnapshot().queue).toBe(before)
   })
 })
 
-describe('queue retirement (host queuedMirror rules)', () => {
-  it('a message-triggered turn/start claims the oldest non-steering row', () => {
-    const session = makeSession()
-    session.handleMuxEnvelope(rid('e1'), queuedFrame('先', 'p-1'))
-    session.handleMuxEnvelope(rid('e2'), queuedFrame('后', 'p-2'))
-    session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: ev.turnStart(0, 0) })
-    expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-2'])
-  })
+describe('queue operation transport', () => {
+  it('addresses the session.updateQueue RPC without optimistic local mutation', async () => {
+    const api = new FakeApiClient()
+    const session = new Session(SID, api)
+    session.handleMuxEnvelope(rid('env-op'), queueFrame([{ id: 'q-op', body: 'pending' }]))
+    const before = session.getSnapshot().queue
 
-  it('an injection-triggered turn/start claims nothing', () => {
-    const session = makeSession()
-    session.handleMuxEnvelope(rid('e1'), queuedFrame('留', 'p-1'))
-    const injection = {
-      ...ev.turnStart(0, 0),
-      data: { turn: 0, trigger: { kind: 'injection', source: { kind: 'plugin', plugin: 'x' } } },
-    } as never
-    session.handleMuxEnvelope(rid('e2'), { type: 'session/event', sessionId: SID, event: injection })
-    expect(session.getSnapshot().queue).toHaveLength(1)
-  })
-
-  it('steering/message drains the source-matched steering row only', () => {
-    const session = makeSession()
-    session.handleMuxEnvelope(rid('e1'), queuedFrame('普通', 'p-1')) // idle → non-steering
-    session.handleMuxEnvelope(rid('e3'), queuedFrame('插话', 'p-2', true))
-    // Loop-authored steering (different source) must not consume the user entry.
-    const foreignSteering = {
-      seq: 0, time: 1,
-      type: 'steering/message', surfaceOp: 'append',
-      data: {
-        turn: 0,
-        message: createUserMessage({
-          content: text('loop'),
-          source: { kind: 'plugin', plugin: 'loop' },
-        }),
-      },
-    } as never
-    session.handleMuxEnvelope(rid('e4'), { type: 'session/event', sessionId: SID, event: foreignSteering })
-    expect(session.getSnapshot().queue).toHaveLength(2)
-    const matchedSteering = {
-      seq: 1, time: 2,
-      type: 'steering/message', surfaceOp: 'append',
-      data: {
-        turn: 0,
-        message: createUserMessage({
-          content: text('插话'),
-          source: { kind: 'user', rpcId: rid('p-2') },
-        }),
-      },
-    } as never
-    session.handleMuxEnvelope(rid('e5'), { type: 'session/event', sessionId: SID, event: matchedSteering })
-    expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-1'])
-  })
-
-  it('a leave-running flip sweeps the whole mirror (cancel/terminal-drop cover)', () => {
-    const session = makeSession()
-    session.handleRunning(true)
-    session.handleMuxEnvelope(rid('e1'), queuedFrame('一', 'p-1'))
-    session.handleMuxEnvelope(rid('e2'), queuedFrame('二', 'p-2'))
-    session.handleRunning(false)
-    expect(session.getSnapshot().queue).toEqual([])
-  })
-
-  it('a stale not-running relay on an idle session still sweeps replayed rows', () => {
-    const session = makeSession()
-    session.handleMuxEnvelope(rid('e1'), queuedFrame('孤儿', 'p-1'))
-    session.handleRunning(false) // running already false: equality path must not skip the sweep
-    expect(session.getSnapshot().queue).toEqual([])
+    await expect(session.updateQueue(iid('q-op'), { kind: 'edit', content: text('next') }))
+      .resolves.toEqual({ ok: true, value: { accepted: true } })
+    expect(api.callsOf('session.updateQueue')).toEqual([{
+      sessionId: SID,
+      itemId: 'q-op',
+      action: { kind: 'edit', content: text('next') },
+    }])
+    expect(session.getSnapshot().queue).toBe(before)
   })
 })
 
 describe('queue reconnect semantics', () => {
-  it('session/subscribed re-baselines the mirror: stale rows drop, the following snapshot lands fresh', () => {
+  it('session/subscribed clears stale state before the fresh snapshot lands', () => {
     const session = makeSession()
-    session.handleMuxEnvelope(rid('e1'), queuedFrame('旧连接', 'p-old'))
-    // New mux generation: subscribed arrives first on the same stream...
+    session.handleMuxEnvelope(rid('e1'), queueFrame([{ id: 'q-old', body: '旧连接' }]))
     session.handleMuxEnvelope(rid('e2'), { type: 'session/subscribed', sessionId: SID, lastSeq: 10 })
     expect(session.getSnapshot().queue).toEqual([])
-    // ...then the queue snapshot replays the live inbox.
-    session.handleMuxEnvelope(rid('e3'), queuedFrame('新基线', 'p-new'))
-    expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-new'])
+    session.handleMuxEnvelope(rid('e3'), queueFrame([{ id: 'q-new', body: '新基线' }]))
+    expect(session.getSnapshot().queue.map(row => row.id)).toEqual(['q-new'])
   })
 
-  it('resync must NOT clear the mirror (regression: onConnected races the mux baseline)', async () => {
+  it('resync does not clear a baseline that raced ahead of the host connection signal', async () => {
     const session = makeSession()
-    // Reconnect ordering that broke: mux opened first and already delivered
-    // the fresh generation's baseline; host stream (and with it onConnected →
-    // resync) lands after. The host never resends — clearing here left the
-    // dock empty until the next enqueue.
     session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 })
-    session.handleMuxEnvelope(rid('e2'), queuedFrame('新基线', 'p-fresh'))
+    session.handleMuxEnvelope(rid('e2'), queueFrame([{ id: 'q-fresh', body: '新基线' }]))
     await session.resync()
-    expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-fresh'])
+    expect(session.getSnapshot().queue.map(row => row.id)).toEqual(['q-fresh'])
   })
 
-  it('replayed steering retires without a replayed turn/start', () => {
+  it('running-status changes never guess at queue retirement', () => {
     const session = makeSession()
-    session.handleMuxEnvelope(rid('e1'), { type: 'session/subscribed', sessionId: SID, lastSeq: 5 })
-    session.handleMuxEnvelope(rid('e2'), queuedFrame('重连插话', 'p-steer', true))
-    const committed = {
-      seq: 6, time: 2,
-      type: 'steering/message', surfaceOp: 'append',
-      data: {
-        turn: 1,
-        message: createUserMessage({
-          content: text('重连插话'),
-          source: { kind: 'user', rpcId: rid('p-steer') },
-        }),
-      },
-    } as never
-    session.handleMuxEnvelope(rid('e3'), { type: 'session/event', sessionId: SID, event: committed })
-    expect(session.getSnapshot().queue).toEqual([])
+    session.handleMuxEnvelope(rid('e1'), queueFrame([{ id: 'q-live', body: '保留' }]))
+    session.handleRunning(true)
+    session.handleRunning(false)
+    expect(session.getSnapshot().queue.map(row => row.id)).toEqual(['q-live'])
   })
 })
 
-describe('manager buffering of queued frames', () => {
-  it('buffers session/queued for uninstantiated sessions and replays before the running sync', () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(api)
-    manager.handleMuxEnvelope({ rpcId: rid('b1'), payload: queuedFrame('预热', 'p-b1') })
-    // Instantiation replays the buffer; no summary exists, so no running sweep runs.
-    const session = manager.get(SID)
-    expect(session.getSnapshot().queue.map(r => r.key)).toEqual(['p-b1'])
-    // The buffer is consumed: a second get must not double-replay.
-    expect(manager.get(SID).getSnapshot().queue).toHaveLength(1)
+describe('manager buffering of queue snapshots', () => {
+  it('replays only the latest snapshot for an uninstantiated session', () => {
+    const manager = new SessionManager(new FakeApiClient())
+    manager.handleMuxEnvelope({ rpcId: rid('b1'), payload: queueFrame([{ id: 'q-old', body: '旧' }]) })
+    manager.handleMuxEnvelope({ rpcId: rid('b2'), payload: queueFrame([{ id: 'q-new', body: '新' }]) })
+    expect(manager.get(SID).getSnapshot().queue.map(row => row.id)).toEqual(['q-new'])
   })
 
-  it('a not-running list summary sweeps replayed rows at instantiation', async () => {
-    const api = new FakeApiClient()
-    api.onList = () => Promise.resolve(ok([{ sessionId: SID, updatedAt: 1, running: false }]))
-    const manager = new SessionManager(api)
-    await manager.refreshList()
-    manager.handleMuxEnvelope({ rpcId: rid('b2'), payload: queuedFrame('该扫掉', 'p-b2') })
-    expect(manager.get(SID).getSnapshot().queue).toEqual([])
-  })
-
-  it('subscribed re-baselines the uninstantiated buffer: stale queued frames drop, non-queue frames survive (regression: reconnect duplication)', () => {
-    const api = new FakeApiClient()
-    const manager = new SessionManager(api)
-    // Generation 1 baseline lands while the session is uninstantiated, along
-    // with a pending approval (never re-derivable from history).
-    manager.handleMuxEnvelope({ rpcId: rid('g1a'), payload: queuedFrame('第一代', 'p-g1') })
+  it('subscribed drops the prior-generation snapshot while preserving answerable frames', () => {
+    const manager = new SessionManager(new FakeApiClient())
+    manager.handleMuxEnvelope({ rpcId: rid('g1a'), payload: queueFrame([{ id: 'q-g1', body: '第一代' }]) })
     manager.handleMuxEnvelope({
       rpcId: rid('g1b'),
       payload: { type: 'approval/requested', sessionId: SID, approvalId: 'ap-1' as never, toolName: 'bash' },
     })
-    // Reconnect: generation 2 replays subscribed + the SAME live queue entry.
-    manager.handleMuxEnvelope({ rpcId: rid('g2a'), payload: { type: 'session/subscribed', sessionId: SID, lastSeq: 3 } })
-    manager.handleMuxEnvelope({ rpcId: rid('g2b'), payload: queuedFrame('第一代', 'p-g1') })
+    manager.handleMuxEnvelope({
+      rpcId: rid('g2a'),
+      payload: { type: 'session/subscribed', sessionId: SID, lastSeq: 3 },
+    })
+    manager.handleMuxEnvelope({ rpcId: rid('g2b'), payload: queueFrame([{ id: 'q-g2', body: '第二代' }]) })
     const snapshot = manager.get(SID).getSnapshot()
-    // One queue row (no duplicate batch); the approval survived the re-baseline.
-    expect(snapshot.queue.map(r => r.key)).toEqual(['p-g1'])
-    expect(snapshot.pending.map(p => p.kind)).toEqual(['approval'])
+    expect(snapshot.queue.map(row => row.id)).toEqual(['q-g2'])
+    expect(snapshot.pending.map(pending => pending.kind)).toEqual(['approval'])
   })
 })
-
-/** ok wrapper with a typed items payload (the shared helper pins value to never[]). */
-function ok(items: { sessionId: SessionId; updatedAt: number; running: boolean }[]) {
-  return { rpcId: rid(`ok-${items.length}`), result: { ok: true as const, value: { items: items as never[] } } }
-}