refactor: identify and freeze messages at creation

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,
 } 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, TodoItem } from '@deepseek-ai/dsh-session'
+import type { MessageId, MessageSource } from '@deepseek-ai/dsh-llm'
+import type { Session, SessionEvent, SessionHeader, SessionId, TodoItem, UserMessage } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
 import { foldSessionTitle } from '@deepseek-ai/dsh-session-title'
 import type { Workspace, WorkspaceRecord } from '@deepseek-ai/dsh-workspace'
@@ -207,9 +207,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
@@ -256,10 +253,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) {
@@ -420,23 +423,23 @@ 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
+   * MessageId: 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.
    */
-  const queuedMirror = new Map<SessionId, Map<AgentMessageId, { message: AgentMessage; steering: boolean }>>()
+  const queuedMirror = new Map<SessionId, Map<MessageId, { message: UserMessage; steering: boolean }>>()
   ctx.effect(() => {
-    const retire = (agent: Agent, id: AgentMessageId): void => {
+    const retire = (agent: Agent, id: MessageId): void => {
       const entries = queuedMirror.get(agent.id)
       if (entries === undefined) return
       entries.delete(id)
       if (entries.size === 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 = new Map<MessageId, { message: UserMessage; steering: boolean }>()
           queuedMirror.set(agent.id, entries)
         }
         const steering = placement === 'steering'
@@ -444,15 +447,14 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
         broadcast({
           type: 'session/queued',
           sessionId: agent.id,
-          content: message.content,
-          source: message.source,
+          message,
           steering,
         })
       }),
-      ctx.on('agent/inbox/dequeue', (agent: Agent, message: AgentMessage) => {
+      ctx.on('agent/inbox/dequeue', (agent: Agent, message: UserMessage) => {
         retire(agent, message.id)
       }),
-      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) => {
@@ -835,8 +837,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) } })
@@ -1120,8 +1123,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             queue.push(frame({
               type: 'session/queued',
               sessionId,
-              content: entry.message.content,
-              source: entry.message.source,
+              message: entry.message,
               steering: entry.steering,
             }))
           }

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() }),
@@ -35,8 +43,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() }),
   z.object({ type: z.literal('stream/error'), error: rpcErrorSchema }),
 ]) as unknown as z.ZodType<MuxFrame>
 

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'
@@ -70,11 +70,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 }
   | { type: 'stream/error'; error: RpcError }
 
 /**

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'
@@ -238,9 +239,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) },
   })

packages/host/apiproxy/tests/api-proxy-view.spec.ts

@@ -14,8 +14,8 @@ 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, createMessage, createToolResultMessage, createUserMessage } 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 { ToolDefinition } from '@deepseek-ai/dsh-tools'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
@@ -88,10 +88,24 @@ 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')
@@ -130,15 +144,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)
@@ -163,8 +206,20 @@ describe('mux live view computation', () => {
     session.append('todo/write', { todos: [{ content: 'old', status: 'pending' }] })
     for (let turn = 0; turn < 6; turn++) {
       session.append('turn/start', { turn, trigger: { kind: 'message', source: { kind: 'user' } } })
-      session.append('user/message', { content: [{ type: 'text', text: `q${turn}` }], source: { kind: 'user' } }, { surfaceOp: 'append' })
-      session.append('assistant/message', { turn, step: 0, content: [{ type: 'text', text: `a${turn}` }], provenance: { provider: 'p', model: 'm' } }, { surfaceOp: 'append' })
+      session.append('user/message', createUserMessage({
+        content: [{ type: 'text', text: `q${turn}` }], source: { kind: 'user' },
+      }), { surfaceOp: 'append' })
+      session.append('assistant/message', {
+        turn, step: 0,
+        message: createMessage({
+          role: 'assistant',
+          content: [{ type: 'text', text: `a${turn}` }],
+          source: {
+            kind: 'model',
+            ...{ provider: 'p', model: 'm' },
+          },
+        }),
+      }, { surfaceOp: 'append' })
       session.append('turn/end', { turn, reason: { kind: 'completed' } })
     }
     session.append('todo/write', { todos: [{ content: 'current', status: 'in_progress' }] })
@@ -221,7 +276,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')

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

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', () => {