Merge latest master into skill invocation controls

# Conflicts: # docs/config-catalog.md # docs/cordis-catalog/services.md # examples/acp-agent/tests/snapshots/skill-load/session.jsonl # packages/host/apiproxy/README.i18n.yaml # packages/host/apiproxy/README.md # packages/host/apiproxy/README.zh.md # packages/host/apiproxy/src/api-proxy.ts # packages/skill/tool-skill/src/index.ts # packages/ui/tui/README.i18n.yaml
2026-07-29 01:29:39 +08:00
parent 1fd06eae89 b6b13441ac
commit 07d61ee992
665 changed files with 15945 additions and 6555 deletions
--- a/packages/host/apiproxy/README.i18n.yaml
+++ b/packages/host/apiproxy/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/host/apiproxy/README.md
-README.md: ab732499af6e5af2d49f68f9ce96d2ae05243dab
-README.zh.md: 9154ff2eb2787fd7bc9c4b41aca6ee82af460869
+README.md: f4ca58a6187f308406b9edad58256191e0b16764
+README.zh.md: 0722c4f1f76dd229612bebfda2c457b9be113116
--- a/packages/host/apiproxy/README.md
+++ b/packages/host/apiproxy/README.md
@@ -22,8 +22,6 @@ Workspace and Session lists are separate reconnect baselines. `workspace.create`

 `host.openPath` opens a filesystem path with the operating system's default application (`open` on macOS, `Invoke-Item` on Windows, `xdg-open` on Linux). The opener is injectable for tests. The browser carrier applies the same loopback, same-origin restriction as `host.pickDirectory`.

-`session.history` pages on message boundaries, and its tail page (no `beforeSeq`) carries two session-level extras the page window cannot supply: the in-flight partial's chunk events, and `todos` — the latest `todo/write` whole-list projection over the full log. Older pages omit `todos` because the projection is session-level, not per-page; a tail response that omits it means the whole log holds no `todo/write`, so clients read the absent field as the empty plan rather than as unchanged state.
-
 The `command.*` and `skill.*` domains expose the host command registry and skill catalog to clients. Every method addresses one session's agent by `sessionId` (a served session always has an Agent; `command.*` resumes cold sessions through the same path as `session.*`, while `skill.list` resolves the project root from the session header without touching the Agent registry). `skill.list` serves the browser's user-selected model-reference path, so it returns only skills that are both model-invocable and user-invocable; this domain has no direct skill-loading RPC. `command.execute` runs a slash-command line host-side with pure admission semantics: the response reports whether the line resolved to a handler plus the minted lifecycle `commandId` when it did (correlating the acknowledgment with the flow node), while the outcome rides the durably logged `command/run`/`command/done` lifecycle pair broadcast on the mux stream; the carrier's request signal cancels the running handler. `host/commands-changed` is the catalog invalidation frame: clients refetch `command.list` instead of diffing.

 ## Carrier layer (`/client` + root)
--- a/packages/host/apiproxy/README.zh.md
+++ b/packages/host/apiproxy/README.zh.md
@@ -22,8 +22,6 @@ Workspace 列表与 Session 列表是相互独立的重连基线。`workspace.cr

 `host.openPath` 会用操作系统的默认应用打开一个文件系统路径（macOS 为 `open`，Windows 为 `Invoke-Item`，Linux 为 `xdg-open`）。打开器可在测试中注入。浏览器载体对其施加与 `host.pickDirectory` 相同的回环、同源限制。

-`session.history` 按消息边界分页，其尾页（不带 `beforeSeq`）额外携带两项页窗口本身无法提供的会话级数据：进行中局部消息的 chunk 事件，以及 `todos`——整份日志上最后一次 `todo/write` 的整表投影。较早的页面不带 `todos`，因为该投影是会话级而非分页级的；尾页响应缺少该字段意味着整份日志中没有任何 `todo/write`，因此客户端要把缺失字段读作空计划，而不是读作「状态未变」。
-
 `command.*` 与 `skill.*` 领域向客户端暴露宿主命令注册表和技能目录。每个方法都通过 `sessionId` 寻址一个会话的 Agent（被服务的会话必有 Agent；`command.*` 经由与 `session.*` 相同的路径恢复冷会话，而 `skill.list` 从会话头解析项目根目录，不触碰 Agent 注册表）。`skill.list` 服务于浏览器中由用户选择的模型引用路径，因此仅返回模型和用户均可调用的 skill；该领域没有直接加载 skill 的 RPC。`command.execute` 在宿主侧运行一条斜杠命令行，语义为纯准入：响应报告该行是否解析到处理器，并在解析到时回带铸造的生命周期 `commandId`（将本次确认与流节点关联）；结局经由持久落账并在 mux 流广播的 `command/run`/`command/done` 生命周期事件对承载；载体的请求信号可取消正在运行的处理器。`host/commands-changed` 是目录失效帧：客户端重新拉取 `command.list` 而不是做差分。

 ## 载体层（`/client` + 根路径）
--- a/packages/host/apiproxy/package.json
+++ b/packages/host/apiproxy/package.json
@@ -47,6 +47,7 @@
    "@deepseek-ai/dsh-session": "workspace:^",
    "@deepseek-ai/dsh-session-persistence": "workspace:^",
    "@deepseek-ai/dsh-session-projection": "workspace:^",
+    "@deepseek-ai/dsh-session-projection-cache": "workspace:^",
    "@deepseek-ai/dsh-skill": "workspace:^",
    "@deepseek-ai/dsh-tools": "workspace:^",
    "@deepseek-ai/dsh-user-approval": "workspace:^",
--- a/packages/host/apiproxy/src/api-proxy.ts
+++ b/packages/host/apiproxy/src/api-proxy.ts
@@ -9,12 +9,12 @@ import { join } from 'node:path'
 import type { Context } from 'cordis'
 import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent'
 import type {
-  Agent, AgentLlmTarget, AgentLlmTargetRef, AgentMessage, AgentMessageId, AgentStatus,
+  Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxPlacement,
 } from '@deepseek-ai/dsh-agent'
-import { ReasoningEffortId } from '@deepseek-ai/dsh-llm'
+import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
 import { errorChain } from '@deepseek-ai/dsh-llm'
-import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
-import type { JsonValue, Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
+import type { MessageId, MessageSource } from '@deepseek-ai/dsh-llm'
+import type { Session, SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
 import { isModelInvocable, isUserInvocable } from '@deepseek-ai/dsh-skill'
 import type { Workspace, WorkspaceRecord } from '@deepseek-ai/dsh-workspace'
@@ -31,6 +31,8 @@ import type {
 } from './api/index.ts'
 // Type-only: resolves `ctx.get('sessionProjections')` to the projection registry.
 import type {} from '@deepseek-ai/dsh-session-projection'
+// Type-only: resolves `ctx.get('sessionProjectionCache')` (the cold listing column).
+import type {} from '@deepseek-ai/dsh-session-projection-cache'
 // Type-only edge: resolves `ctx.get('commands')` and the `commands/change` event.
 import type {} from '@deepseek-ai/dsh-commands'
 import { questionResponsePayloadSchema } from './api/questions.schema.ts'
@@ -194,9 +196,6 @@ export interface ApiProxyDefaults {

 /** The tool/call payload fields the presenter path reads. */
 interface ToolCallData { callId: string; name: string; arguments: string }
-/** The tool/result payload fields the presenter path reads. */
-interface ToolResultData { callId: string; content: ContentBlock[]; isError: boolean; meta?: JsonValue }
-
 /** One host-owned question wait, addressed by the stable server-request id. */
 interface PendingQuestion {
  rpcId: RpcId
@@ -243,10 +242,16 @@ function viewFor(ctx: Context, event: SessionEvent, argsFor: (callId: string) =>
      return view === undefined ? undefined : { for: 'call', view }
    }
    if (event.type === 'tool/result') {
-      const { callId, content, isError, meta } = event.data as ToolResultData
+      const { message, meta } = event.data
+      const [result] = message.content
+      const callId = message.source.callId
      const call = argsFor(callId) as { name: string; args: unknown } | undefined
      if (call === undefined) return undefined
-      const view = ctx.tools.get(call.name)?.presentResult?.(call.args, { content, isError, ...meta === undefined ? {} : { meta } })
+      const view = ctx.tools.get(call.name)?.presentResult?.(call.args, {
+        content: result.content,
+        isError: result.isError === true,
+        ...meta === undefined ? {} : { meta },
+      })
      return view === undefined ? undefined : { for: 'result', view }
    }
  } catch (error: unknown) {
@@ -294,6 +299,28 @@ function projectionsFor(ctx: Context, agent: Agent): SessionProjectionsBlock | u
  return registry.snapshot(agent.session)
 }

+/**
+ * The projection baseline of one session.list row, fail-soft: attached
+ * sessions cut the registry's live watermark cache; cold sessions view the
+ * persisted projection cache's identity-checked stored rows (zero log loads
+ * either way — the listing use case the cache exists for). The block shape
+ * (values + asOfSeq) matches the history tail's, so a client seeds its
+ * value store under the same higher-seq-wins rule. Any failure — and an
+ * empty value set — yields an absent block: a listing without projections
+ * is degraded, never broken.
+ */
+function listProjectionsFor(ctx: Context, meta: SessionHeader, session: Session | undefined): SessionProjectionsBlock | undefined {
+  try {
+    const block = session !== undefined
+      ? ctx.get('sessionProjections')?.snapshot(session)
+      : ctx.get('sessionProjectionCache')?.cachedSnapshot(meta)
+    return block !== undefined && Object.keys(block.values).length > 0 ? block : undefined
+  } catch (error) {
+    ctx.logger.warn(`session.list: projection column for "${meta.id}" failed (serving the row without it): ${String(error)}`)
+    return undefined
+  }
+}
+
 /**
 * Thrown by the cold-resume path when the id names no servable session
 * (absent from the store, or a pre-project legacy log without a cwd).
@@ -421,41 +448,42 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
  })

  /**
-   * Per-session inbox mirror serving the mux-open queue snapshot (the same
-   * refresh-recovery baseline as pending questions). Keyed by the stable
-   * AgentMessageId: every enqueued id receives exactly one terminal
-   * `agent/inbox/dequeue` OR `agent/inbox/discard` (the inbox contract), so
-   * the mirror needs no consumption heuristics or sweeps beyond disposal.
+   * Per-session inbox occurrence mirror serving the mux-open queue snapshot
+   * (the same refresh-recovery baseline as pending questions). Each terminal
+   * inbox event retires one matching occurrence, so repeated sends of the same
+   * identified message remain visible until every occurrence is claimed.
   */
-  const queuedMirror = new Map<SessionId, Map<AgentMessageId, { message: AgentMessage; steering: boolean }>>()
+  const queuedMirror = new Map<SessionId, { message: UserMessage; steering: boolean }[]>()
  ctx.effect(() => {
-    const retire = (agent: Agent, id: AgentMessageId): void => {
+    const retire = (agent: Agent, id: MessageId, placement?: InboxPlacement): void => {
      const entries = queuedMirror.get(agent.id)
      if (entries === undefined) return
-      entries.delete(id)
-      if (entries.size === 0) queuedMirror.delete(agent.id)
+      const index = entries.findIndex(entry =>
+        entry.message.id === id
+        && (placement === undefined || entry.steering === (placement === 'steering')))
+      if (index !== -1) entries.splice(index, 1)
+      if (entries.length === 0) queuedMirror.delete(agent.id)
    }
    const disposers = [
-      ctx.on('agent/inbox/enqueue', (agent: Agent, message: AgentMessage, placement) => {
+      ctx.on('agent/inbox/enqueue', (agent: Agent, message: UserMessage, placement) => {
        let entries = queuedMirror.get(agent.id)
        if (entries === undefined) {
-          entries = new Map<AgentMessageId, { message: AgentMessage; steering: boolean }>()
+          entries = []
          queuedMirror.set(agent.id, entries)
        }
        const steering = placement === 'steering'
-        entries.set(message.id, { message, steering })
+        entries.push({ message, steering })
        broadcast({
          type: 'session/queued',
          sessionId: agent.id,
-          content: message.content,
-          source: message.source,
+          message,
          steering,
        })
      }),
-      ctx.on('agent/inbox/dequeue', (agent: Agent, message: AgentMessage) => {
-        retire(agent, message.id)
+      ctx.on('agent/inbox/dequeue', (agent: Agent, message: UserMessage, placement) => {
+        retire(agent, message.id, placement)
      }),
-      ctx.on('agent/inbox/discard', (agent: Agent, messages: AgentMessage[]) => {
+      ctx.on('agent/inbox/discard', (agent: Agent, messages: UserMessage[]) => {
        for (const message of messages) retire(agent, message.id)
      }),
      ctx.on('session/disposed', (session: Session) => {
@@ -650,13 +678,25 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
      async list(request) {
        const items = ctx.sessions.list().map((session) => {
          const agent = ctx.agents.get(session.id)
-          return summarize(session, agent?.status === 'running')
+          const projections = listProjectionsFor(ctx, session.header, session)
+          return {
+            ...summarize(session, agent?.status === 'running'),
+            ...projections === undefined ? {} : { projections },
+          }
        })
        const attached = new Set(items.map(item => item.sessionId))
        const persistence = ctx.get('sessionPersistence')
        if (persistence !== undefined) {
          const cold = (await persistence.list()).filter(meta => !attached.has(meta.id) && meta.cwd !== undefined)
-          items.push(...await Promise.all(cold.map(meta => summarizeCold(persistence, meta))))
+          items.push(...await Promise.all(cold.map(async (meta) => {
+            // Cold rows read the persisted projection cache only — never a
+            // log load; a session without a cache row simply has no column.
+            const projections = listProjectionsFor(ctx, meta, undefined)
+            return {
+              ...await summarizeCold(persistence, meta),
+              ...projections === undefined ? {} : { projections },
+            }
+          })))
        }
        items.sort((a, b) => b.updatedAt - a.updatedAt)
        return ok(request, { items })
@@ -843,8 +883,9 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
        // The rpcId rides MessageSource into user/message (merge declaration in api/sessions.ts; provisional correlation).
        const source: MessageSource = { kind: 'user', rpcId: request.rpcId }
        try {
-          if (mode === 'steer') agent.steer({ content, source })
-          else agent.followup({ content, source })
+          const message: UserMessage = createUserMessage({ content, source })
+          if (mode === 'steer') agent.steer(message)
+          else agent.followup(message)
        } catch (error: unknown) {
          // A synchronous throw from steer/followup means disposed or invalid input; surface as agent-busy with the reason attached.
          return err(request, { code: 'agent-busy', message: 'prompt rejected', details: { reason: String(error) } })
@@ -1150,12 +1191,11 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
        // in arrival order per session; a reconnecting client rebuilds its
        // queue view from these alone.
        for (const [sessionId, entries] of queuedMirror) {
-          for (const entry of entries.values()) {
+          for (const entry of entries) {
            queue.push(frame({
              type: 'session/queued',
              sessionId,
-              content: entry.message.content,
-              source: entry.message.source,
+              message: entry.message,
              steering: entry.steering,
            }))
          }
--- a/packages/host/apiproxy/src/api/events.schema.ts
+++ b/packages/host/apiproxy/src/api/events.schema.ts
@@ -23,6 +23,14 @@ export const askUserQuestionItemSchema = z.object({
  multiSelect: z.boolean().optional(),
 }) satisfies z.ZodType<Wire<AskUserQuestionItem>>

+/** Unified message envelope carried by transient queue frames. */
+const messageSchema = z.object({
+  id: z.string().min(1),
+  role: z.union([z.literal('system'), z.literal('user'), z.literal('assistant')]),
+  content: z.array(contentBlockSchema),
+  source: z.looseObject({ kind: z.string() }),
+})
+
 /** MuxFrame union (payload slot of a mux-stream ServerRequest). */
 export const muxFrameSchema = z.discriminatedUnion('type', [
  z.object({ type: z.literal('session/event'), sessionId: sessionIdSchema, event: sessionEventSchema, view: toolEventViewSchema.optional() }),
@@ -34,8 +42,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
  // and must fail loud here, not reach the composer.
  z.object({ type: z.literal('question/requested'), sessionId: sessionIdSchema, questions: z.array(askUserQuestionItemSchema).min(1) }),
  z.object({ type: z.literal('question/resolved'), sessionId: sessionIdSchema, questionRpcId: rpcIdSchema, outcome: z.union([z.literal('answered'), z.literal('cancelled')]) }),
-  // content/source reuse the wide passthroughs (both are merge-extensible in core).
-  z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }), steering: z.boolean() }),
+  z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, message: messageSchema, steering: z.boolean() }),
  // value stays wide: it already passed its unit's own schema on the host,
  // and deep-validating here would import every domain's schema into the carrier.
  z.object({ type: z.literal('session/projection'), sessionId: sessionIdSchema, key: z.string().min(1), value: z.unknown(), seq: z.number().int().nonnegative() }),
--- a/packages/host/apiproxy/src/api/events.ts
+++ b/packages/host/apiproxy/src/api/events.ts
@@ -8,7 +8,7 @@

 import type { AskUserQuestionItem } from '@deepseek-ai/dsh-user-interaction/types'
 import type { ApprovalOutcome, ApprovalRequestId } from '@deepseek-ai/dsh-user-approval/types'
-import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm/types'
+import type { Message } from '@deepseek-ai/dsh-llm/types'
 import type { CallId } from '@deepseek-ai/dsh-llm/brand'
 import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
 import type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
@@ -69,11 +69,11 @@ export type MuxFrame =
   * refresh-recovery baseline as pending questions); queue clearing on cancel
   * has no dedicated frame — clients fold it from the status flip.
   * `steering` is the host's acceptance-time queue classification and remains
-   * authoritative in reconnect snapshots. `source` carries the prompt's rpcId
+   * authoritative in reconnect snapshots. `message.source` carries the prompt's rpcId
   * when the message came over this wire (the client's provisional-echo
   * reconciliation key).
   */
-  | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource; steering: boolean }
+  | { type: 'session/queued'; sessionId: SessionId; message: Message; steering: boolean }
  /**
   * One projection unit's finished value changed (session-projection RFC).
   * Live push state, never logged — replay recomputes on the host (the
--- a/packages/host/apiproxy/src/api/sessions.schema.ts
+++ b/packages/host/apiproxy/src/api/sessions.schema.ts
@@ -37,7 +37,7 @@ export const sessionEventSchema = z.object({
  surfaceOp: z.unknown().optional(),
 }) as unknown as z.ZodType<SessionEvent>

-/** SessionSummary row of session.list. */
+/** SessionSummary row of session.list (`projections` reuses the history block's shape and schema). */
 export const sessionSummarySchema = z.object({
  sessionId: sessionIdSchema,
  updatedAt: z.number(),
@@ -45,7 +45,8 @@ export const sessionSummarySchema = z.object({
  blank: z.boolean(),
  parentSessionId: sessionIdSchema.optional(),
  cwd: z.string().optional(),
-}) satisfies z.ZodType<Wire<SessionSummary>>
+  projections: z.lazy(() => sessionProjectionsBlockSchema).optional(),
+}) as unknown as z.ZodType<Wire<SessionSummary>>

 /** session.list request payload (cursor is a reserved seat, unimplemented in v1). */
 export const sessionListRequestSchema = z.object({
@@ -53,9 +54,9 @@ export const sessionListRequestSchema = z.object({
 }) satisfies z.ZodType<Wire<RequestPayload<'session.list'>>>

 /** session.list response value. */
-export const sessionListValueSchema = z.object({
+export const sessionListValueSchema: z.ZodType<Wire<ResponseValue<'session.list'>>> = z.object({
  items: z.array(sessionSummarySchema),
-}) satisfies z.ZodType<Wire<ResponseValue<'session.list'>>>
+})

 /** session.create request payload (at most one of workspaceId / cwd). */
 export const sessionCreateRequestSchema = z.object({
--- a/packages/host/apiproxy/src/api/sessions.ts
+++ b/packages/host/apiproxy/src/api/sessions.ts
@@ -143,6 +143,18 @@ export interface SessionSummary {
  parentSessionId?: SessionId
  /** Session working directory (header.cwd passthrough); absent when unrecorded. */
  cwd?: string
+  /**
+   * Projection baseline for this row, with zero log loads: attached sessions
+   * read the registry's live watermark cut; cold sessions read the persisted
+   * projection cache's stored rows — as stale as that session's last durable
+   * checkpoint (`asOfSeq` says exactly how stale), never wrong, and directly
+   * seedable into the client's per-session value store under its
+   * higher-seq-wins rule (a list baseline can never overwrite a newer push
+   * frame). Absent when no value is available (no registry, no cache row for
+   * a cold session, or a fail-soft cache read miss); a listing client treats
+   * absence as "no title yet", exactly like a blank session.
+   */
+  projections?: SessionProjectionsBlock
 }

 /** Session-domain unary methods (the map keys session.* of RpcMethodMap). */
--- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts
+++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts
@@ -1,3 +1,4 @@
+import { MessageId, freezeMessage } from '@deepseek-ai/dsh-llm'
 /**
 * Command/skill RPC handlers and the two new frames over createApiProxy:
 * command.list serves the addressed agent's effective catalog (missing
@@ -10,10 +11,10 @@

 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
-import AgentRegistry, { AgentMessageId } from '@deepseek-ai/dsh-agent'
-import type { Agent, AgentMessage } from '@deepseek-ai/dsh-agent'
+import AgentRegistry, {} from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
 import SessionStore from '@deepseek-ai/dsh-session'
-import type { SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionId, UserMessage } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry from '@deepseek-ai/dsh-tools'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
@@ -263,9 +264,10 @@ describe('host/commands-changed frame', () => {
 })

 /** Build one frozen inbox message for the live `agent/inbox/*` events. */
-function inboxMessage(id: string, text: string, rpcId?: string): AgentMessage {
-  return Object.freeze({
-    id: AgentMessageId(id),
+function inboxMessage(id: string, text: string, rpcId?: string): UserMessage {
+  return freezeMessage({
+    id: MessageId(id),
+    role: 'user',
    content: [{ type: 'text' as const, text }],
    source: rpcId === undefined ? { kind: 'user' as const } : { kind: 'user' as const, rpcId: RpcId(rpcId) },
  })
@@ -288,8 +290,8 @@ describe('session/queued frames', () => {

    const liveFrames = (await liveCollected).filter(f => f.type === 'session/queued')
    expect(liveFrames).toEqual([
-      { type: 'session/queued', sessionId: agent.id, content: queued.content, source: { kind: 'user' }, steering: false },
-      { type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' }, steering: true },
+      { type: 'session/queued', sessionId: agent.id, message: queued, steering: false },
+      { type: 'session/queued', sessionId: agent.id, message: steering, steering: true },
    ])

    // A fresh mux connection replays the still-pending entries as its baseline.
@@ -307,8 +309,8 @@ describe('session/queued frames', () => {
    const steering = inboxMessage('m-4', 'x', 'r-1')
    ctx.emit('agent/inbox/enqueue', agent, queued, 'queued')
    ctx.emit('agent/inbox/enqueue', agent, steering, 'steering')
-    ctx.emit('agent/inbox/dequeue', agent, queued)
-    ctx.emit('agent/inbox/dequeue', agent, steering)
+    ctx.emit('agent/inbox/dequeue', agent, queued, 'queued')
+    ctx.emit('agent/inbox/dequeue', agent, steering, 'steering')

    const abort = new AbortController()
    const frames = await collect<MuxFrame>(
@@ -316,6 +318,24 @@ describe('session/queued frames', () => {
    expect(frames.filter(f => f.type === 'session/queued')).toHaveLength(0)
  })

+  it('retires the matching placement when one message identity is queued and steering', async () => {
+    const ctx = await harness()
+    const api = createApiProxy(ctx, DEFAULTS)
+    const agent = stubAgent(ctx)
+    const repeated = inboxMessage('m-repeat', 'same prompt')
+    ctx.emit('agent/inbox/enqueue', agent, repeated, 'queued')
+    ctx.emit('agent/inbox/enqueue', agent, repeated, 'steering')
+    ctx.emit('agent/inbox/dequeue', agent, inboxMessage('unknown', 'not queued'), 'queued')
+    ctx.emit('agent/inbox/dequeue', agent, repeated, 'steering')
+
+    const abort = new AbortController()
+    const frames = await collect<MuxFrame>(
+      api.events.mux({ rpcId: RpcId('t-mux-repeat'), payload: {} }, abort.signal), 2, abort)
+    expect(frames.filter(f => f.type === 'session/queued')).toEqual([
+      { type: 'session/queued', sessionId: agent.id, message: repeated, steering: false },
+    ])
+  })
+
  it('retires mirror entries on a batch discard (cancel path)', async () => {
    const ctx = await harness()
    const api = createApiProxy(ctx, DEFAULTS)
@@ -331,6 +351,6 @@ describe('session/queued frames', () => {
      api.events.mux({ rpcId: RpcId('t-mux-swept'), payload: {} }, abort.signal), 2, abort)
    const remaining = frames.filter(f => f.type === 'session/queued')
    expect(remaining).toHaveLength(1)
-    expect(remaining[0]).toMatchObject({ content: survivor.content })
+    expect(remaining[0]).toMatchObject({ message: survivor })
  })
 })
--- a/packages/host/apiproxy/tests/api-proxy-projections.spec.ts
+++ b/packages/host/apiproxy/tests/api-proxy-projections.spec.ts
@@ -12,7 +12,8 @@ import { Context } from 'cordis'
 import { z } from 'zod'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
-import SessionStore from '@deepseek-ai/dsh-session'
+import { createUserMessage } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session } from '@deepseek-ai/dsh-session'
 import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
 import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
@@ -61,7 +62,10 @@ async function harness(withRegistry: boolean): Promise<{ ctx: Context; session:
 /** Append `count` user messages so the log has paginable message boundaries. */
 function seedMessages(session: Session, count: number): void {
  for (let i = 0; i < count; i++) {
-    session.append('user/message', { content: [{ type: 'text', text: `m${i}` }], source: { kind: 'user' } }, { surfaceOp: 'append' })
+    session.append('user/message', createUserMessage({
+      content: [{ type: 'text', text: `m${i}` }],
+      source: { kind: 'user' },
+    }), { surfaceOp: 'append' })
  }
 }

@@ -121,6 +125,82 @@ describe('session.history projections block', () => {
  })
 })

+describe('session.list projections column', () => {
+  it('serves attached rows from the live registry cut, watermarked for client seeding', async () => {
+    const { ctx, session } = await harness(true)
+    ctx.sessionProjections.register(lastUserUnit())
+    seedMessages(session, 1)
+    const response = await api(ctx).sessions.list(request({}))
+    if (!response.result.ok) throw new Error('unreachable')
+    const row = response.result.value.items.find(item => item.sessionId === session.id)
+    expect(row?.projections?.values['test/last-user']).toEqual({ text: 'm0' })
+    expect(row?.projections?.asOfSeq).toBe(session.seq - 1)
+  })
+
+  it('omits the column entirely when no registry is mounted', async () => {
+    const { ctx, session } = await harness(false)
+    seedMessages(session, 1)
+    const response = await api(ctx).sessions.list(request({}))
+    if (!response.result.ok) throw new Error('unreachable')
+    const row = response.result.value.items.find(item => item.sessionId === session.id)
+    expect(row).toBeDefined()
+    expect(row !== undefined && 'projections' in row).toBe(false)
+  })
+
+  it('serves cold rows from the persisted projection cache with zero log loads', async () => {
+    const { ctx } = await harness(true)
+    const coldId = SessionId('session-cold-listing')
+    const load = () => { throw new Error('list must not load event logs') }
+    ctx.provide('sessionPersistence', {
+      list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
+      locate: () => undefined,
+      load,
+      inspect: load,
+      readFrom: load,
+    } as never)
+    ctx.provide('sessionProjectionCache', {
+      // The carrier hands the listed header through as the identity witness.
+      cachedSnapshot: (meta: { id: unknown; createdAt: number }) =>
+        (meta.id === coldId && meta.createdAt === 5
+          ? { asOfSeq: 7, values: { 'test/last-user': { text: 'cached' } } }
+          : undefined),
+    } as never)
+    const response = await api(ctx).sessions.list(request({}))
+    if (!response.result.ok) throw new Error('unreachable')
+    const row = response.result.value.items.find(item => item.sessionId === coldId)
+    expect(row?.running).toBe(false)
+    expect(row?.projections).toEqual({ asOfSeq: 7, values: { 'test/last-user': { text: 'cached' } } })
+  })
+
+  it('cold rows without a cache plugin (or without a stored row) just lack the column', async () => {
+    const { ctx } = await harness(true)
+    const coldId = SessionId('session-cold-uncached')
+    ctx.provide('sessionPersistence', {
+      list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
+      locate: () => undefined,
+    } as never)
+    const response = await api(ctx).sessions.list(request({}))
+    if (!response.result.ok) throw new Error('unreachable')
+    const row = response.result.value.items.find(item => item.sessionId === coldId)
+    expect(row).toBeDefined()
+    expect(row !== undefined && 'projections' in row).toBe(false)
+  })
+
+  it('a throwing column read degrades that row, never the listing', async () => {
+    const { ctx, session } = await harness(true)
+    ctx.sessionProjections.register({
+      ...lastUserUnit(),
+      view: () => { throw new Error('unit exploded') },
+    })
+    seedMessages(session, 1)
+    const response = await api(ctx).sessions.list(request({}))
+    if (!response.result.ok) throw new Error('unreachable')
+    const row = response.result.value.items.find(item => item.sessionId === session.id)
+    expect(row).toBeDefined()
+    expect(row !== undefined && 'projections' in row).toBe(false)
+  })
+})
+
 describe('session/projection push frame', () => {
  /** Drain frames until `count` session/projection frames arrived. */
  async function collect(iterable: AsyncIterable<RpcRequest<MuxFrame>>, count: number, abort: AbortController): Promise<MuxFrame[]> {
--- a/packages/host/apiproxy/tests/api-proxy-view.spec.ts
+++ b/packages/host/apiproxy/tests/api-proxy-view.spec.ts
@@ -14,9 +14,9 @@ import type { Agent } from '@deepseek-ai/dsh-agent'
 import SessionStore from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
+import { CallId, createToolResultMessage } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock } from '@deepseek-ai/dsh-llm'
-import { CallId } from '@deepseek-ai/dsh-llm'
-import type { Session, SessionId } from '@deepseek-ai/dsh-session'
+import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
 import type { ToolDefinition } from '@deepseek-ai/dsh-tools'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import type { MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
@@ -88,16 +88,35 @@ describe('mux live view computation', () => {
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-term'), name: 'term', arguments: '{"cmd":"echo hi"}' })
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-diff'), name: 'diffy', arguments: '{}' })
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-call-only'), name: 'call-only', arguments: '{}' })
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('c-call-only'), content: [{ type: 'text', text: rawResult }], isError: false }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('c-call-only'),
+        content: [{ type: 'text', text: rawResult }],
+        isError: false,
+      }),
+    }, { surfaceOp: 'append' })
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-plain'), name: 'plain', arguments: '{}' })
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('c-boom'), name: 'boom', arguments: '{}' })
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('c-gen'), content: [{ type: 'text', text: 'ok' }], isError: false }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('c-gen'),
+        content: [{ type: 'text', text: 'ok' }],
+        isError: false,
+      }),
+    }, { surfaceOp: 'append' })

    const frames = await collected
    const events = frames.filter(f => f.type === 'session/event')
    const byCall = new Map(events
      .filter(f => f.event.type === 'tool/call' || f.event.type === 'tool/result')
-      .map(f => [`${f.event.type}:${(f.event.data as { callId: string }).callId}`, f]))
+      .map(f => [
+        `${f.event.type}:${f.event.type === 'tool/call'
+          ? f.event.data.callId
+          : (f.event.data as SessionEvent<'tool/result'>['data']).message.source.callId}`,
+        f,
+      ]))

    expect(byCall.get('tool/call:c-gen')?.view).toEqual({ for: 'call', view: { card: 'generic', title: 'gen call' } })
    expect(byCall.get('tool/call:c-term')?.view).toEqual({ for: 'call', view: { card: 'terminal', title: 'echo hi' } })
@@ -130,15 +149,44 @@ describe('mux live view computation', () => {
    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('h-term'), name: 'term', arguments: '{"cmd":"ls"}' })
    // meta rides through to presentResult's ToolResult (the spread arm).
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-term'), content: [{ type: 'text', text: 'ok' }], isError: false, meta: { n: 1 } }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('h-term'),
+        content: [{ type: 'text', text: 'ok' }],
+        isError: false,
+      }),
+      meta: { n: 1 },
+    }, { surfaceOp: 'append' })
    // Unpaired result: no tool/call with this id anywhere in the page.
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-orphan'), content: [{ type: 'text', text: 'x' }], isError: false }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('h-orphan'),
+        content: [{ type: 'text', text: 'x' }],
+        isError: false,
+      }),
+    }, { surfaceOp: 'append' })
    // Paired, but the call's stored arguments do not parse: backscan soft-falls.
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('h-bad'), name: 'term', arguments: '{broken' })
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-bad'), content: [{ type: 'text', text: 'y' }], isError: false }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('h-bad'),
+        content: [{ type: 'text', text: 'y' }],
+        isError: false,
+      }),
+    }, { surfaceOp: 'append' })
    // Presenterless tool: pairing succeeds but presentResult is absent.
    session.append('tool/call', { turn: 1, step: 1, callId: CallId('h-plain'), name: 'plain', arguments: '{}' })
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('h-plain'), content: [{ type: 'text', text: 'z' }], isError: false }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('h-plain'),
+        content: [{ type: 'text', text: 'z' }],
+        isError: false,
+      }),
+    }, { surfaceOp: 'append' })

    const response = await api.sessions.history({ rpcId: RpcId('t-hist'), payload: { sessionId: session.id } })
    expect(response.result.ok).toBe(true)
@@ -146,7 +194,12 @@ describe('mux live view computation', () => {
    const entries = response.result.value.events
    const byKey = new Map(entries
      .filter(entry => entry.event.type === 'tool/call' || entry.event.type === 'tool/result')
-      .map(entry => [`${entry.event.type}:${(entry.event.data as { callId: string }).callId}`, entry]))
+      .map(entry => [
+        `${entry.event.type}:${entry.event.type === 'tool/call'
+          ? entry.event.data.callId
+          : (entry.event.data as SessionEvent<'tool/result'>['data']).message.source.callId}`,
+        entry,
+      ]))
    expect(byKey.get('tool/call:h-term')?.view).toEqual({ for: 'call', view: { card: 'terminal', title: 'ls' } })
    expect(byKey.get('tool/result:h-term')?.view).toEqual({ for: 'result', view: { card: 'terminal', output: 'done' } })
    expect('view' in (byKey.get('tool/result:h-orphan') ?? {})).toBe(false)
@@ -188,7 +241,14 @@ describe('mux live view computation', () => {
    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
    // The turn/end above cleared the live table; pairing must fall back to
    // scanning the session's in-memory events.
-    session.append('tool/result', { turn: 1, step: 1, callId: CallId('c-late'), content: [{ type: 'text', text: 'ok' }], isError: false }, { surfaceOp: 'append' })
+    session.append('tool/result', {
+      turn: 1, step: 1,
+      message: createToolResultMessage({
+        callId: CallId('c-late'),
+        content: [{ type: 'text', text: 'ok' }],
+        isError: false,
+      }),
+    }, { surfaceOp: 'append' })

    const frames = await collected
    const result = frames.find(f => f.type === 'session/event' && f.event.type === 'tool/result')
--- a/packages/host/apiproxy/tests/api-proxy-workspace.spec.ts
+++ b/packages/host/apiproxy/tests/api-proxy-workspace.spec.ts
@@ -3,7 +3,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
-import AgentRegistry, { AgentMessageId } from '@deepseek-ai/dsh-agent'
+import AgentRegistry, {} from '@deepseek-ai/dsh-agent'
 import type { Agent, AgentFactory } from '@deepseek-ai/dsh-agent'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session } from '@deepseek-ai/dsh-session'
@@ -45,10 +45,10 @@ function stubAgent(session: Session): Agent {
    status: 'idle',
    acceptsNextStep: false,
    ctx: new Context(),
-    followup: () => AgentMessageId('stub'),
-    steer: () => AgentMessageId('stub'),
-    inject: () => AgentMessageId('stub'),
-    send: () => AgentMessageId('stub'),
+    followup: () => {},
+    steer: () => {},
+    inject: () => {},
+    send: () => {},
    cancel() {},
    whenIdle: () => Promise.resolve(),
  }
--- a/packages/host/apiproxy/tests/rpc-schemas.spec.ts
+++ b/packages/host/apiproxy/tests/rpc-schemas.spec.ts
@@ -119,9 +119,19 @@ describe('sessions domain schemas', () => {
    expect(sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: true, blank: false, parentSessionId: 'p', cwd: '/x' }).cwd).toBe('/x')
    // blank is mandatory: a summary without it fails the parse.
    expect(() => sessionSummarySchema.parse({ sessionId: 's1', updatedAt: 1, running: false })).toThrow()
-    const event = sessionEventSchema.parse({ type: 'user/message', seq: 0, time: 1, data: { any: true } })
+    const event = sessionEventSchema.parse({
+      type: 'user/message',
+      seq: 0,
+      time: 1,
+      data: { any: true },
+    })
    expect(event).toMatchObject({ type: 'user/message' })
-    expect(() => sessionEventSchema.parse({ type: 'user/message', seq: -1, time: 1, data: {} })).toThrow()
+    expect(() => sessionEventSchema.parse({
+      type: 'user/message',
+      seq: -1,
+      time: 1,
+      data: {},
+    })).toThrow()
  })

  it('validates the per-method request/value pairs', () => {
@@ -311,8 +321,8 @@ describe('events frame schemas', () => {
      { type: 'approval/resolved', sessionId: 's', approvalId: 'a', outcome: 'allowed-once' },
      { type: 'question/requested', sessionId: 's', questions: [{ id: 'q', question: 'Q?', options: [{ label: 'L' }], multiSelect: true }] },
      { type: 'question/resolved', sessionId: 's', questionRpcId: 'r', outcome: 'answered' },
-      { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' }, steering: false },
-      { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' }, steering: true },
+      { type: 'session/queued', sessionId: 's', message: { id: 'm1', role: 'user', content: [{ type: 'text', text: 'queued prompt' }], source: { kind: 'user', rpcId: 'r9' } }, steering: false },
+      { type: 'session/queued', sessionId: 's', message: { id: 'm2', role: 'user', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' } }, steering: true },
      { type: 'session/projection', sessionId: 's', key: 'todos', value: [{ content: 'x', status: 'pending' }], seq: 7 },
      { type: 'stream/error', error: { code: 'internal', message: 'm', details: {} } },
    ]
@@ -331,9 +341,9 @@ describe('events frame schemas', () => {
  })

  it('rejects a queued frame missing its members', () => {
-    expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: 'x', source: { kind: 'user' } })).toThrow()
-    expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: { kind: 'user' } })).toThrow()
-    expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', content: [], source: {}, steering: false })).toThrow()
+    expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', message: 'x', steering: false })).toThrow()
+    expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', message: { id: 'm', role: 'user', content: [], source: { kind: 'user' } } })).toThrow()
+    expect(() => muxFrameSchema.parse({ type: 'session/queued', sessionId: 's', message: { id: 'm', role: 'user', content: [], source: {} }, steering: false })).toThrow()
  })

  it('accepts every host frame branch', () => {
--- a/packages/host/apiproxy/tsconfig.json
+++ b/packages/host/apiproxy/tsconfig.json
@@ -35,6 +35,9 @@
    {
      "path": "../../session-projection/session-projection"
    },
+    {
+      "path": "../../session-projection/session-projection-cache"
+    },
    {
      "path": "../../skill/skill"
    },