Merge origin/master into worktree/skill-catalog-hot-refresh

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

@@ -0,0 +1,330 @@
+/**
+ * Approval pending registry over the proxy: an ask through `ctx.approval`
+ * becomes an answerable `approval/requested` mux frame (stable rpcId, replayed
+ * verbatim on a later mux open), `respond` routes by the echoed rpcId and
+ * validates the audit correlation, and the ask's abort signal withdraws the
+ * question with a broadcast `cancelled`.
+ */
+
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import SessionStore from '@deepseek-ai/dsh-session'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
+import ApprovalService from '@deepseek-ai/dsh-user-approval'
+import type { ApprovalRequestId } from '@deepseek-ai/dsh-user-approval'
+import type { ApiProxy, MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
+import type { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
+import { RpcId as mintRpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
+import { createApiProxy } from '../src/api-proxy.ts'
+
+async function harness(): Promise<{ ctx: Context; api: ApiProxy }> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(SystemPrompt, { persona: '' })
+  await ctx.plugin(UserInteractionService)
+  await ctx.plugin(AgentRegistry)
+  await ctx.plugin(ApprovalService)
+  const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
+  return { ctx, api }
+}
+
+/** A minimal agent stand-in inside an open turn (the service only reaches `.session`). */
+function agentOf(ctx: Context): Agent {
+  const session = ctx.sessions.create()
+  session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+  return { session } as unknown as Agent
+}
+
+/** Open a mux stream and capture frames into an array (returns an on-demand waiter). */
+function openMux(api: ApiProxy, abort: AbortController): { frames: MuxFrame[]; envelopes: RpcRequest<MuxFrame>[]; waitFor(type: MuxFrame['type']): Promise<MuxFrame> } {
+  const frames: MuxFrame[] = []
+  const envelopes: RpcRequest<MuxFrame>[] = []
+  const waiters: { type: MuxFrame['type']; resolve: (frame: MuxFrame) => void }[] = []
+  void (async () => {
+    for await (const envelope of api.events.mux({ rpcId: mintRpcId('t-mux'), payload: {} }, abort.signal)) {
+      frames.push(envelope.payload)
+      envelopes.push(envelope)
+      for (let i = waiters.length - 1; i >= 0; i -= 1) {
+        const waiter = waiters[i] as (typeof waiters)[number]
+        if (waiter.type === envelope.payload.type) {
+          waiters.splice(i, 1)
+          waiter.resolve(envelope.payload)
+        }
+      }
+    }
+  })()
+  return {
+    frames,
+    envelopes,
+    waitFor: (type) => {
+      const found = frames.find(frame => frame.type === type)
+      if (found !== undefined) return Promise.resolve(found)
+      return new Promise((resolve) => { waiters.push({ type, resolve }) })
+    },
+  }
+}
+
+function requestedOf(frame: MuxFrame): Extract<MuxFrame, { type: 'approval/requested' }> {
+  if (frame.type !== 'approval/requested') throw new Error(`expected approval/requested, got ${frame.type}`)
+  return frame
+}
+
+/** Wait until the stream delivered `count` frames of `type` (bounded poll; waitFor only covers the first). */
+async function waitForCount(mux: { frames: MuxFrame[] }, type: MuxFrame['type'], count: number): Promise<void> {
+  for (let i = 0; i < 200 && mux.frames.filter(frame => frame.type === type).length < count; i += 1) {
+    await new Promise(resolve => setTimeout(resolve, 5))
+  }
+  expect(mux.frames.filter(frame => frame.type === type).length).toBeGreaterThanOrEqual(count)
+}
+
+function answer(rpcId: RpcId, sessionId: unknown, approvalId: ApprovalRequestId, outcome: 'allowed-once' | 'rejected'): Parameters<ApiProxy['respond']>[0] {
+  return { type: 'client-response', rpcId, result: { ok: true, value: { sessionId, approvalId, outcome } } }
+}
+
+describe('approval pending registry', () => {
+  it('round-trips ask → requested frame → respond → outcome + resolved broadcast', async () => {
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const agent = agentOf(ctx)
+
+    const asked = ctx.approval.request({ agent, toolName: 'bash', reason: 'sandbox escalation' })
+    const requested = requestedOf(await mux.waitFor('approval/requested'))
+    expect(requested).toMatchObject({ toolName: 'bash', reason: 'sandbox escalation', sessionId: agent.session.id })
+
+    const envelope = mux.envelopes.find(e => e.payload.type === 'approval/requested') as RpcRequest<MuxFrame>
+    const receipt = await api.respond(answer(envelope.rpcId, requested.sessionId, requested.approvalId, 'allowed-once'))
+    expect(receipt).toEqual({ accepted: true })
+    await expect(asked).resolves.toBe('allowed-once')
+
+    const resolved = await mux.waitFor('approval/resolved')
+    expect(resolved).toMatchObject({ approvalId: requested.approvalId, outcome: 'allowed-once' })
+
+    // The question settled: a duplicate answer is late, not re-decidable.
+    const dup = await api.respond(answer(envelope.rpcId, requested.sessionId, requested.approvalId, 'rejected'))
+    expect(dup).toEqual({ accepted: false, reason: 'not-pending' })
+    abort.abort()
+  })
+
+  it('replays a still-pending requested frame (same rpcId) on a later mux open', async () => {
+    const { ctx, api } = await harness()
+    const first = new AbortController()
+    const firstMux = openMux(api, first)
+    const agent = agentOf(ctx)
+    const asked = ctx.approval.request({ agent, toolName: 'write' })
+    const requested = requestedOf(await firstMux.waitFor('approval/requested'))
+    const firstEnvelope = firstMux.envelopes.find(e => e.payload.type === 'approval/requested') as RpcRequest<MuxFrame>
+    first.abort()
+
+    // A fresh subscriber (refresh recovery) sees the same stable rpcId.
+    const second = new AbortController()
+    const secondMux = openMux(api, second)
+    const replayed = requestedOf(await secondMux.waitFor('approval/requested'))
+    const secondEnvelope = secondMux.envelopes.find(e => e.payload.type === 'approval/requested') as RpcRequest<MuxFrame>
+    expect(secondEnvelope.rpcId).toBe(firstEnvelope.rpcId)
+    expect(replayed.approvalId).toBe(requested.approvalId)
+
+    const receipt = await api.respond(answer(secondEnvelope.rpcId, replayed.sessionId, replayed.approvalId, 'rejected'))
+    expect(receipt).toEqual({ accepted: true })
+    await expect(asked).resolves.toBe('rejected')
+    second.abort()
+  })
+
+  it('rejects malformed and mismatched answers as bad-response, unknown ids as not-pending', async () => {
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const agent = agentOf(ctx)
+    void ctx.approval.request({ agent, toolName: 'bash' })
+    const requested = requestedOf(await mux.waitFor('approval/requested'))
+    const envelope = mux.envelopes.find(e => e.payload.type === 'approval/requested') as RpcRequest<MuxFrame>
+
+    // Unknown rpcId: not routed to any pending entry.
+    expect(await api.respond(answer(mintRpcId('ghost'), requested.sessionId, requested.approvalId, 'rejected')))
+      .toEqual({ accepted: false, reason: 'not-pending' })
+    // Error-branch result: the client can only answer with a value.
+    expect(await api.respond({ type: 'client-response', rpcId: envelope.rpcId, result: { ok: false, error: { code: 'internal', message: 'x', details: {} } } }))
+      .toEqual({ accepted: false, reason: 'bad-response' })
+    // Wrong audit correlation: the rpcId routed, but the payload disagrees.
+    expect(await api.respond(answer(envelope.rpcId, requested.sessionId, 'other-approval' as ApprovalRequestId, 'rejected')))
+      .toEqual({ accepted: false, reason: 'bad-response' })
+    // Malformed payload shape.
+    expect(await api.respond({ type: 'client-response', rpcId: envelope.rpcId, result: { ok: true, value: { nonsense: 1 } } }))
+      .toEqual({ accepted: false, reason: 'bad-response' })
+    abort.abort()
+  })
+
+  it('withdraws the question on the ask signal: cancelled outcome, resolved broadcast, late answer not-pending', async () => {
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const agent = agentOf(ctx)
+    const cancel = new AbortController()
+    const asked = ctx.approval.request({ agent, toolName: 'bash', signal: cancel.signal })
+    const requested = requestedOf(await mux.waitFor('approval/requested'))
+    const envelope = mux.envelopes.find(e => e.payload.type === 'approval/requested') as RpcRequest<MuxFrame>
+
+    cancel.abort()
+    await expect(asked).resolves.toBe('cancelled')
+    const resolved = await mux.waitFor('approval/resolved')
+    expect(resolved).toMatchObject({ approvalId: requested.approvalId, outcome: 'cancelled' })
+    expect(await api.respond(answer(envelope.rpcId, requested.sessionId, requested.approvalId, 'allowed-once')))
+      .toEqual({ accepted: false, reason: 'not-pending' })
+    abort.abort()
+  })
+
+  it('an ask whose signal aborted before dispatch settles cancelled without publishing', async () => {
+    // The service checks the signal, then dispatch rides a microtask: an
+    // abort in that window must not register a dead listener and strand the
+    // entry (zombie frame on every replay). Drive the waterfall directly
+    // with a pre-aborted signal to hit the answerer's register-path guard.
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const session = ctx.sessions.create()
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    session.append('approval/asked', { id: 'pre-aborted' as ApprovalRequestId, toolName: 'bash' })
+    const agent = { session } as unknown as Agent
+    const cancelled = new AbortController()
+    cancelled.abort()
+    const outcome = await ctx.waterfall(
+      'approval/request',
+      { agent, toolName: 'bash', signal: cancelled.signal },
+      () => Promise.resolve('unavailable' as const),
+    )
+    expect(outcome).toBe('cancelled')
+    // Nothing was published: a fresh mux open replays no approval frame.
+    const abort2 = new AbortController()
+    const mux2 = openMux(api, abort2)
+    await new Promise(resolve => setTimeout(resolve, 10))
+    expect(mux2.envelopes.some(e => e.payload.type === 'approval/requested')).toBe(false)
+    abort2.abort()
+    abort.abort()
+    void mux
+  })
+
+  it('gateway teardown settles pending approvals as cancelled (question-provider parity)', async () => {
+    // Mount the proxy on its own fiber so disposal exercises the teardown
+    // effect while an ask is still pending.
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(SystemPrompt, { persona: '' })
+    await ctx.plugin(UserInteractionService)
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(ApprovalService)
+    let api!: ApiProxy
+    const fiber = ctx.plugin(Object.assign((fiberCtx: Context) => {
+      api = createApiProxy(fiberCtx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
+    }, { inject: ['sessions', 'agents', 'userInteraction', 'approval'] }))
+    await fiber.await()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const asked = ctx.approval.request({ agent: agentOf(ctx), toolName: 'bash' })
+    const requested = requestedOf(await mux.waitFor('approval/requested'))
+    await fiber.dispose()
+    await expect(asked).resolves.toBe('cancelled')
+    const resolved = await mux.waitFor('approval/resolved')
+    expect(resolved).toMatchObject({ approvalId: requested.approvalId, outcome: 'cancelled' })
+    abort.abort()
+  })
+
+  it('carries callId on the frame and ignores a late abort after the answer settled', async () => {
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const agent = agentOf(ctx)
+    const cancel = new AbortController()
+    const asked = ctx.approval.request({ agent, toolName: 'bash', callId: 'call-9' as never, signal: cancel.signal })
+    const requested = requestedOf(await mux.waitFor('approval/requested'))
+    expect(requested.callId).toBe('call-9')
+    const envelope = mux.envelopes.find(e => e.payload.type === 'approval/requested') as RpcRequest<MuxFrame>
+    expect(await api.respond(answer(envelope.rpcId, requested.sessionId, requested.approvalId, 'allowed-once')))
+      .toEqual({ accepted: true })
+    await expect(asked).resolves.toBe('allowed-once')
+    // Late abort: the pending entry is gone; settle's delete-guard returns.
+    cancel.abort()
+    expect(mux.frames.filter(f => f.type === 'approval/resolved')).toHaveLength(1)
+    abort.abort()
+  })
+
+  it('pairs parallel asks by callId: each requested frame carries its own audit id', async () => {
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const agent = agentOf(ctx)
+    // Both asks append their approval/asked audit events before either
+    // answerer's microtask dispatch runs — the parallel tool-call window.
+    const askA = ctx.approval.request({ agent, toolName: 'bash', callId: 'call-a' as never })
+    const askB = ctx.approval.request({ agent, toolName: 'bash', callId: 'call-b' as never })
+    await waitForCount(mux, 'approval/requested', 2)
+    const frames = mux.envelopes.filter(e => e.payload.type === 'approval/requested')
+    const frameA = frames.find(e => requestedOf(e.payload).callId === 'call-a') as RpcRequest<MuxFrame>
+    const frameB = frames.find(e => requestedOf(e.payload).callId === 'call-b') as RpcRequest<MuxFrame>
+    // Each frame claimed the asked event with its own callId, not merely the newest.
+    const askedIdByCall = new Map(agent.session.events
+      .filter(event => event.type === 'approval/asked')
+      .map(event => [String(event.data.callId), event.data.id]))
+    expect(requestedOf(frameA.payload).approvalId).toBe(askedIdByCall.get('call-a'))
+    expect(requestedOf(frameB.payload).approvalId).toBe(askedIdByCall.get('call-b'))
+    // Answers route back to the right ask through the pairing.
+    expect(await api.respond(answer(frameB.rpcId, agent.session.id, requestedOf(frameB.payload).approvalId, 'rejected')))
+      .toEqual({ accepted: true })
+    expect(await api.respond(answer(frameA.rpcId, agent.session.id, requestedOf(frameA.payload).approvalId, 'allowed-once')))
+      .toEqual({ accepted: true })
+    await expect(askA).resolves.toBe('allowed-once')
+    await expect(askB).resolves.toBe('rejected')
+    abort.abort()
+  })
+
+  it('gives parallel callId-less asks distinct audit ids (claimed-entry skip); both stay answerable', async () => {
+    const { ctx, api } = await harness()
+    const abort = new AbortController()
+    const mux = openMux(api, abort)
+    const agent = agentOf(ctx)
+    const askA = ctx.approval.request({ agent, toolName: 'alpha' })
+    const askB = ctx.approval.request({ agent, toolName: 'beta' })
+    await waitForCount(mux, 'approval/requested', 2)
+    const frames = mux.envelopes.filter(e => e.payload.type === 'approval/requested')
+    const frameA = frames.find(e => requestedOf(e.payload).toolName === 'alpha') as RpcRequest<MuxFrame>
+    const frameB = frames.find(e => requestedOf(e.payload).toolName === 'beta') as RpcRequest<MuxFrame>
+    // Without a callId the pairing is heuristic, but never shared: the second
+    // dispatch skips the id the first pending entry already claimed.
+    expect(requestedOf(frameA.payload).approvalId).not.toBe(requestedOf(frameB.payload).approvalId)
+    expect(await api.respond(answer(frameA.rpcId, agent.session.id, requestedOf(frameA.payload).approvalId, 'allowed-once')))
+      .toEqual({ accepted: true })
+    expect(await api.respond(answer(frameB.rpcId, agent.session.id, requestedOf(frameB.payload).approvalId, 'rejected')))
+      .toEqual({ accepted: true })
+    await expect(askA).resolves.toBe('allowed-once')
+    await expect(askB).resolves.toBe('rejected')
+    abort.abort()
+  })
+
+  it('delegates a dispatch whose only asked candidate is already decided (stale re-dispatch)', async () => {
+    const { ctx, api } = await harness()
+    void api // the answerer is registered; the fake below bypasses the service
+    // Bypass ApprovalService: a log whose sole asked event already has its
+    // decided partner must not be re-claimed — the answerer delegates.
+    const session = ctx.sessions.create()
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    session.append('approval/asked', { id: 'stale-ask' as ApprovalRequestId, toolName: 'bash' })
+    session.append('approval/decided', { id: 'stale-ask' as ApprovalRequestId, outcome: 'rejected' })
+    const agent = { session } as unknown as Agent
+    const outcome = await ctx.waterfall('approval/request', { agent, toolName: 'bash' }, () => Promise.resolve('unavailable' as const))
+    expect(outcome).toBe('unavailable')
+  })
+
+  it('delegates an ask whose session log carries no asked audit event (foreign channel)', async () => {
+    const { ctx, api } = await harness()
+    void api // the answerer is registered; the fake below bypasses the audit path
+    // Bypass ApprovalService: dispatch the waterfall directly with a session
+    // that has no approval/asked event — the proxy answerer must call next().
+    const session = ctx.sessions.create()
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    const agent = { session } as unknown as Agent
+    const outcome = await ctx.waterfall('approval/request', { agent, toolName: 'x' }, () => Promise.resolve('unavailable' as const))
+    expect(outcome).toBe('unavailable')
+  })
+})

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

@@ -0,0 +1,85 @@
+/**
+ * The summary blank bit means "conversation not started" (no turn has run),
+ * not "log empty": standalone plugin events — command lifecycle records,
+ * plan/mode, permission knob events, session titles — never flip it, so running /plan or /goal on a
+ * fresh session keeps it list-hidden and reusable, while the first accepted
+ * prompt's turn/start clears it. The host/session-added frame shares the
+ * same predicate function (covered by the workspace spec's frame assertion).
+ */
+
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import SessionStore from '@deepseek-ai/dsh-session'
+import type { Session } from '@deepseek-ai/dsh-session'
+import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
+import { CommandId } from '@deepseek-ai/dsh-commands/brand'
+// Side-effect type imports: the knob-event SessionEventMap merges.
+import type {} from '@deepseek-ai/dsh-permission'
+import type {} from '@deepseek-ai/dsh-sandbox-policy'
+import type {} from '@deepseek-ai/dsh-user-approval'
+import type { ApiProxy, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
+import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
+import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
+
+let nextRpc = 1
+function request<P>(payload: P): RpcRequest<P> {
+  return { rpcId: RpcId(`blank-${String(nextRpc++)}`), payload }
+}
+
+async function harness(): Promise<{ ctx: Context; api: ApiProxy; attach: (session: Session) => void }> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(UserInteractionService)
+  await ctx.plugin(AgentRegistry)
+  return {
+    ctx,
+    api: createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' }),
+    attach: (session) => {
+      ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent)
+    },
+  }
+}
+
+/** Append the standalone (non-conversation) event family a fresh session can accumulate. */
+function appendStandalone(session: Session): void {
+  session.append('command/run', {
+    commandId: CommandId('blank-cmd-1'), name: 'plan', args: '', source: { kind: 'user' },
+  })
+  session.append('plan/mode', { active: true })
+  session.append('command/done', { commandId: CommandId('blank-cmd-1'), kind: 'success', text: 'Plan mode on.' })
+  session.append('session/title', {
+    title: 'standalone title', messageSeqs: [], source: { kind: 'fallback' },
+  })
+  // The three permission knob events (a /permission switch on a fresh session).
+  session.append('permission/preset', { preset: 'danger-full-access' })
+  session.append('sandbox/mode', { mode: 'danger-full-access' })
+  session.append('approval/policy', { policy: 'never' })
+}
+
+async function listBlank(api: ApiProxy, id: string): Promise<boolean | undefined> {
+  const response = await api.sessions.list(request({}))
+  if (!response.result.ok) throw new Error('list failed')
+  return response.result.value.items.find(item => item.sessionId === id)?.blank
+}
+
+describe('summary blank = conversation not started', () => {
+  it('standalone events (command lifecycle, plan/mode, title) keep the session blank', async () => {
+    const { ctx, api, attach } = await harness()
+    const session = ctx.sessions.create()
+    attach(session)
+    expect(await listBlank(api, session.id)).toBe(true)
+    appendStandalone(session)
+    expect(await listBlank(api, session.id)).toBe(true)
+  })
+
+  it('the first turn clears blank', async () => {
+    const { ctx, api, attach } = await harness()
+    const session = ctx.sessions.create()
+    attach(session)
+    appendStandalone(session)
+    session.append('turn/start', { turn: 0, trigger: { kind: 'message', source: { kind: 'user' } } })
+    expect(await listBlank(api, session.id)).toBe(false)
+  })
+})

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'
@@ -115,8 +116,16 @@ describe('command.execute', () => {
     const api = createApiProxy(ctx, DEFAULTS)
     const agent = stubAgent(ctx)
     const value = expectOk(await api.commands.execute(request({ sessionId: agent.id, line: '/goal ship it' }), new AbortController().signal))
-    expect(value).toEqual({ matched: true, result: { kind: 'success', text: `goal:${agent.id}` } })
+    expect(value).toMatchObject({ matched: true })
+    expect(value.commandId).toBeTruthy()
     expect(received).toBe(' ship it')
+    // Pure admission on the wire: the outcome rides the durably logged
+    // lifecycle pair instead of the response.
+    const lifecycle = agent.session.events.filter(e => e.type === 'command/run' || e.type === 'command/done')
+    expect(lifecycle).toMatchObject([
+      { type: 'command/run', data: { commandId: value.commandId, name: 'goal', args: ' ship it' } },
+      { type: 'command/done', data: { commandId: value.commandId, kind: 'success', text: `goal:${agent.id}` } },
+    ])
   })
 
   it('returns matched:false when syntax or name does not resolve', async () => {
@@ -238,9 +247,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) },
   })
@@ -263,8 +273,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.
@@ -282,8 +292,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>(
@@ -291,6 +301,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)
@@ -306,6 +334,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 })
   })
 })

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

@@ -0,0 +1,261 @@
+/**
+ * Projection carrier paths of the host ApiProxy: the history tail page's
+ * projections block reads the registry's watermark snapshot (asOfSeq = last
+ * event seq, one consistent cut); loadOlder pages never carry the block; a
+ * composition without the registry serves histories without it; a disposed
+ * registration's key leaves subsequent responses; and every unit change is
+ * pushed to mux consumers as a session/projection frame minted here.
+ */
+
+import { describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import { z } from 'zod'
+import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+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'
+import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
+import type { MuxFrame, RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api'
+import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
+import { createApiProxy } from '@deepseek-ai/dsh-host-apiproxy'
+
+declare module '@deepseek-ai/dsh-session-projection/types' {
+  interface SessionProjectionMap {
+    'test/last-user': { text: string } | null
+  }
+}
+
+let nextRpc = 1
+function request<P>(payload: P): RpcRequest<P> {
+  return { rpcId: RpcId(`proj-${String(nextRpc++)}`), payload }
+}
+
+/** Whole-value unit folding the latest user/message text; null before the first. */
+type LastUserState = { text: string } | null
+const lastUserUnit = (): ProjectionDefinition<'test/last-user', LastUserState> => ({
+  key: 'test/last-user',
+  schema: z.union([z.object({ text: z.string() }), z.null()]),
+  init: () => null,
+  apply: (state, event) => (event.type === 'user/message'
+    ? { text: (event.data.content[0] as { text?: string }).text ?? '' }
+    : state),
+  view: state => state,
+  stateVersion: 1,
+})
+
+async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session }> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(UserInteractionService)
+  await ctx.plugin(AgentRegistry)
+  if (withRegistry) await ctx.plugin(SessionProjectionRegistry)
+  const session = ctx.sessions.create()
+  // history resolves the agent first; a live structural stub is enough (only
+  // .session is read on this path).
+  ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent)
+  return { ctx, 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', createUserMessage({
+      content: [{ type: 'text', text: `m${i}` }],
+      source: { kind: 'user' },
+    }), { surfaceOp: 'append' })
+  }
+}
+
+const api = (ctx: Context) => createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
+
+describe('session.history projections block', () => {
+  it('serves the unit value on the tail page with asOfSeq = last event seq', async () => {
+    const { ctx, session } = await harness(true)
+    ctx.sessionProjections.register(lastUserUnit())
+    seedMessages(session, 3)
+    const response = await api(ctx).sessions.history(request({ sessionId: session.id }))
+    expect(response.result.ok).toBe(true)
+    if (!response.result.ok) throw new Error('unreachable')
+    const { events, projections } = response.result.value
+    expect(projections).toBeDefined()
+    expect(projections?.asOfSeq).toBe(session.seq - 1)
+    expect(projections?.values['test/last-user']).toEqual({ text: 'm2' })
+    // asOfSeq IS the window tail: the last served event carries it.
+    expect(events.at(-1)?.event.seq).toBe(projections?.asOfSeq)
+  })
+
+  it('never carries the block on loadOlder pages (beforeSeq present)', async () => {
+    const { ctx, session } = await harness(true)
+    ctx.sessionProjections.register(lastUserUnit())
+    seedMessages(session, 5)
+    const older = await api(ctx).sessions.history(request({ sessionId: session.id, beforeSeq: 3, maxMessages: 2 }))
+    expect(older.result.ok).toBe(true)
+    if (!older.result.ok) throw new Error('unreachable')
+    expect('projections' in older.result.value).toBe(false)
+  })
+
+  it('serves no block when the composition has no projection registry', async () => {
+    const { ctx, session } = await harness(false)
+    seedMessages(session, 2)
+    const response = await api(ctx).sessions.history(request({ sessionId: session.id }))
+    expect(response.result.ok).toBe(true)
+    if (!response.result.ok) throw new Error('unreachable')
+    expect('projections' in response.result.value).toBe(false)
+  })
+
+  it('drops a disposed registration from subsequent tail pages (empty block, key absent)', async () => {
+    const { ctx, session } = await harness(true)
+    const dispose = ctx.sessionProjections.register(lastUserUnit())
+    seedMessages(session, 1)
+    const proxy = api(ctx)
+    const before = await proxy.sessions.history(request({ sessionId: session.id }))
+    if (!before.result.ok) throw new Error('unreachable')
+    expect(before.result.value.projections?.values['test/last-user']).toEqual({ text: 'm0' })
+
+    dispose()
+    const after = await proxy.sessions.history(request({ sessionId: session.id }))
+    if (!after.result.ok) throw new Error('unreachable')
+    // The registry is still mounted, so the block itself stays (asOfSeq cut
+    // with zero keys); the disposed key reads as capability absence.
+    expect(after.result.value.projections?.asOfSeq).toBe(session.seq - 1)
+    expect(after.result.value.projections?.values).toEqual({})
+  })
+})
+
+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[]> {
+    const frames: MuxFrame[] = []
+    for await (const envelope of iterable) {
+      frames.push(envelope.payload)
+      if (frames.filter(f => f.type === 'session/projection').length >= count) abort.abort()
+    }
+    return frames
+  }
+
+  it('broadcasts a frame per changed unit with the causing seq, and none for same-reference applies', async () => {
+    const { ctx, session } = await harness(true)
+    ctx.sessionProjections.register(lastUserUnit())
+    const proxy = api(ctx)
+    // The gateway's onChanged subscription lives in an inject child whose
+    // fiber activates asynchronously; yield until it lands before appending.
+    await new Promise(resolve => setTimeout(resolve, 0))
+    const abort = new AbortController()
+    const stream = proxy.events.mux({ rpcId: RpcId('t-proj-mux'), payload: {} }, abort.signal)
+    const collected = collect(stream, 2, abort)
+
+    seedMessages(session, 1)
+    // Same-reference apply: turn/start does not concern the unit — no frame.
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    seedMessages(session, 1)
+
+    const frames = await collected
+    const pushes = frames.filter(
+      (f): f is Extract<MuxFrame, { type: 'session/projection' }> => f.type === 'session/projection',
+    )
+    expect(pushes).toEqual([
+      { type: 'session/projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 0 },
+      { type: 'session/projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 2 },
+    ])
+    // Frame seq aligns with the tail block's asOfSeq vocabulary (higher-seq-wins compatible).
+    const tail = await proxy.sessions.history(request({ sessionId: session.id }))
+    if (!tail.result.ok) throw new Error('unreachable')
+    expect(tail.result.value.projections?.asOfSeq).toBe(pushes.at(-1)?.seq)
+  })
+
+  it('emits no projection frames when the composition has no registry', async () => {
+    const { ctx, session } = await harness(false)
+    const proxy = api(ctx)
+    const abort = new AbortController()
+    const stream = proxy.events.mux({ rpcId: RpcId('t-noproj-mux'), payload: {} }, abort.signal)
+    const frames: MuxFrame[] = []
+    const drained = (async () => {
+      for await (const envelope of stream) {
+        frames.push(envelope.payload)
+        if (frames.filter(f => f.type === 'session/event').length >= 2) abort.abort()
+      }
+    })()
+    seedMessages(session, 2)
+    await drained
+    expect(frames.some(f => f.type === 'session/projection')).toBe(false)
+  })
+})

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)
@@ -154,39 +207,6 @@ describe('mux live view computation', () => {
     expect('view' in (byKey.get('tool/result:h-plain') ?? {})).toBe(false)
   })
 
-  it('tail page carries the full-log todo projection; older pages and todo-less sessions omit it', async () => {
-    const { ctx } = await harness()
-    const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
-    const session = ctx.sessions.create()
-    ctx.agents.register({ id: session.id, session, status: 'idle', ctx } as Agent)
-    // Superseded write early in the log, latest write later; enough messages to page.
-    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('turn/end', { turn, reason: { kind: 'completed' } })
-    }
-    session.append('todo/write', { todos: [{ content: 'current', status: 'in_progress' }] })
-
-    // Tail page limited to 2 messages: the latest todo/write may or may not sit
-    // in the window — the projection must come from the FULL log either way.
-    const tail = await api.sessions.history({ rpcId: RpcId('t-todos'), payload: { sessionId: session.id, maxMessages: 2 } })
-    if (!tail.result.ok) throw new Error('history failed')
-    expect(tail.result.value.todos).toEqual([{ content: 'current', status: 'in_progress' }])
-    // An older page omits the projection (session-level, tail-page-only).
-    const boundary = tail.result.value.events[0]?.event.seq ?? 0
-    const older = await api.sessions.history({ rpcId: RpcId('t-todos-2'), payload: { sessionId: session.id, beforeSeq: boundary, maxMessages: 2 } })
-    if (!older.result.ok) throw new Error('older failed')
-    expect('todos' in older.result.value).toBe(false)
-    // A session with no todo/write anywhere omits the field.
-    const bare = ctx.sessions.create()
-    ctx.agents.register({ id: bare.id, session: bare, status: 'idle', ctx } as Agent)
-    const bareTail = await api.sessions.history({ rpcId: RpcId('t-todos-3'), payload: { sessionId: bare.id } })
-    if (!bareTail.result.ok) throw new Error('bare failed')
-    expect('todos' in bareTail.result.value).toBe(false)
-  })
-
   it('drops a disposed session from the live open-call table (result after dispose gets no view)', async () => {
     const { ctx } = await harness()
     const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp', workspaceRoot: '/tmp' })
@@ -221,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')

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

@@ -3,13 +3,15 @@ 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'
 import Storage from '@deepseek-ai/dsh-storage'
 import { DomainFacility } from '@deepseek-ai/dsh-storage-domain'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
+import { DirectoryPickerError } from '@deepseek-ai/dsh-host-directory-picker'
+import type { DirectoryPickerCapability } from '@deepseek-ai/dsh-host-directory-picker'
 import WorkspaceRegistry from '@deepseek-ai/dsh-workspace'
 import type { HostFrame, WorkspaceId } from '@deepseek-ai/dsh-host-apiproxy/api'
 import type { RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
@@ -45,10 +47,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(),
   }
@@ -57,7 +59,8 @@ function stubAgent(session: Session): Agent {
 /** Compose the API over real Session, Agent, Storage, Domain, and Workspace services. */
 async function harness(
   workspaceRoot = realpathSync(mkdtempSync(join(tmpdir(), 'dsh-apiproxy-workspace-'))),
-  pickDirectory?: (signal: AbortSignal) => Promise<string | null>,
+  picker: DirectoryPickerCapability = { kind: 'native', pick: async () => null },
+  extras: { openPath?: (path: string, signal: AbortSignal) => Promise<void> } = {},
 ) {
   const ctx = new Context()
   await ctx.plugin(SessionStore)
@@ -92,36 +95,151 @@ async function harness(
     },
   }
   ctx.agents.setFactory(factory)
+  // Structural picker fake: the gateway only reads capability(); a stable
+  // object per harness mirrors the seam's stability contract.
+  ctx.provide('directoryPicker', { capability: () => picker } as never)
   const api = createApiProxy(ctx, {
     provider: 'test',
     model: 'test-model',
     cwd: workspaceRoot,
     workspaceRoot,
-    ...pickDirectory === undefined ? {} : { pickDirectory },
+    ...extras.openPath === undefined ? {} : { openPath: extras.openPath },
   })
   return { api, ctx, storageDomain, workspaceRoot }
 }
 
 describe('host.pickDirectory', () => {
-  it('returns a selected path or explicit cancellation from the injected native boundary', async () => {
-    const selected = await harness(undefined, async () => '/tmp/project')
+  it('returns a selected path or explicit cancellation from the native capability', async () => {
+    const selected = await harness(undefined, { kind: 'native', pick: async () => '/tmp/project' })
     expect((await selected.api.host.pickDirectory(request({}), new AbortController().signal)).result)
       .toEqual({ ok: true, value: { path: '/tmp/project' } })
 
-    const cancelled = await harness(undefined, async () => null)
+    const cancelled = await harness(undefined, { kind: 'native', pick: async () => null })
     expect((await cancelled.api.host.pickDirectory(request({}), new AbortController().signal)).result)
       .toEqual({ ok: true, value: { path: null } })
   })
 
-  it('propagates abort into the native boundary as a cancelled RPC error', async () => {
-    const { api } = await harness(undefined, signal => new Promise((_resolve, reject) => {
-      signal.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
-    }))
+  it('propagates abort into the native capability as a cancelled RPC error', async () => {
+    const { api } = await harness(undefined, {
+      kind: 'native',
+      pick: signal => new Promise((_resolve, reject) => {
+        signal.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
+      }),
+    })
     const abort = new AbortController()
     const pending = api.host.pickDirectory(request({}), abort.signal)
     abort.abort()
     expect((await pending).result).toMatchObject({ ok: false, error: { code: 'cancelled' } })
   })
+
+  it('folds a non-abort native-chooser failure into an internal error', async () => {
+    const { api } = await harness(undefined, { kind: 'native', pick: async () => { throw new Error('no chooser installed') } })
+    const response = await api.host.pickDirectory(request({}), new AbortController().signal)
+    expect(response.result).toMatchObject({ ok: false, error: { code: 'internal' } })
+  })
+
+  it('refuses the native RPC under a browse composition', async () => {
+    const { api } = await harness(undefined, BROWSE_STUB)
+    const response = await api.host.pickDirectory(request({}), new AbortController().signal)
+    expect(response.result).toMatchObject({
+      ok: false,
+      error: { code: 'directory-picker-unavailable', details: { capability: 'browse' } },
+    })
+  })
+})
+
+/** Canned browse capability: one listing, one created path, typed failures on demand. */
+const BROWSE_STUB: DirectoryPickerCapability = {
+  kind: 'browse',
+  list: async (path) => {
+    if (path === '/denied') throw new DirectoryPickerError('directory-unreadable', '/denied', 'cannot list /denied')
+    const target = path ?? '/home/user'
+    return {
+      path: target,
+      home: '/home/user',
+      crumbs: [{ name: '/', path: '/', hidden: false }],
+      entries: [{ name: 'projects', path: `${target}/projects`, hidden: false }],
+      truncated: false,
+    }
+  },
+  createDirectory: async (path, name) => {
+    if (name === 'taken') throw new DirectoryPickerError('directory-exists', `${path}/${name}`, 'already exists')
+    if (name === 'unwritable') throw new Error('disk detached')
+    return `${path}/${name}`
+  },
+}
+
+describe('host.listDirectory / host.createDirectory', () => {
+  it('serves listings and creation through the browse capability, defaulting to home', async () => {
+    const { api } = await harness(undefined, BROWSE_STUB)
+    const home = await api.host.listDirectory(request({}), new AbortController().signal)
+    expect(home.result).toMatchObject({ ok: true, value: { path: '/home/user', home: '/home/user' } })
+    const listed = await api.host.listDirectory(request({ path: '/home/user/projects' }), new AbortController().signal)
+    expect(listed.result).toMatchObject({ ok: true, value: { path: '/home/user/projects' } })
+    const created = await api.host.createDirectory(request({ path: '/home/user', name: 'fresh' }))
+    expect(created.result).toEqual({ ok: true, value: { path: '/home/user/fresh' } })
+  })
+
+  it('maps typed picker failures onto the wire error codes and folds unknown throws to internal', async () => {
+    const { api } = await harness(undefined, BROWSE_STUB)
+    expect((await api.host.listDirectory(request({ path: '/denied' }), new AbortController().signal)).result).toMatchObject({
+      ok: false, error: { code: 'directory-unreadable', details: { path: '/denied' } },
+    })
+    expect((await api.host.createDirectory(request({ path: '/home/user', name: 'taken' }))).result).toMatchObject({
+      ok: false, error: { code: 'directory-exists' },
+    })
+    expect((await api.host.createDirectory(request({ path: '/home/user', name: 'unwritable' }))).result).toMatchObject({
+      ok: false, error: { code: 'internal' },
+    })
+  })
+
+  it('reports an aborted listing as cancelled, like the other signal-following RPCs', async () => {
+    const { api } = await harness(undefined, {
+      kind: 'browse',
+      list: (_path, signal) => new Promise((_resolve, reject) => {
+        signal?.addEventListener('abort', () => { reject(new Error('scan aborted')) }, { once: true })
+      }),
+      createDirectory: async () => '/never',
+    })
+    const abort = new AbortController()
+    const pending = api.host.listDirectory(request({}), abort.signal)
+    abort.abort()
+    expect((await pending).result).toMatchObject({ ok: false, error: { code: 'cancelled' } })
+  })
+
+  it('refuses the browse RPCs under a native composition', async () => {
+    const { api } = await harness()
+    expect((await api.host.listDirectory(request({}), new AbortController().signal)).result).toMatchObject({
+      ok: false, error: { code: 'directory-picker-unavailable', details: { capability: 'native' } },
+    })
+    expect((await api.host.createDirectory(request({ path: '/x', name: 'y' }))).result).toMatchObject({
+      ok: false, error: { code: 'directory-picker-unavailable', details: { capability: 'native' } },
+    })
+  })
+})
+
+describe('host.openPath', () => {
+  it('opens through the injected native boundary', async () => {
+    const opened: string[] = []
+    const { api } = await harness(undefined, undefined, {
+      openPath: async (path) => { opened.push(path) },
+    })
+    expect((await api.host.openPath(request({ path: '/tmp/a.txt' }), new AbortController().signal)).result)
+      .toEqual({ ok: true, value: { opened: true } })
+    expect(opened).toEqual(['/tmp/a.txt'])
+  })
+
+  it('propagates abort into the native boundary as a cancelled RPC error', async () => {
+    const { api } = await harness(undefined, undefined, {
+      openPath: (_path, signal) => new Promise((_resolve, reject) => {
+        signal.addEventListener('abort', () => { reject(new Error('aborted')) }, { once: true })
+      }),
+    })
+    const abort = new AbortController()
+    const pending = api.host.openPath(request({ path: '/tmp/a.txt' }), abort.signal)
+    abort.abort()
+    expect((await pending).result).toMatchObject({ ok: false, error: { code: 'cancelled' } })
+  })
 })
 
 describe('workspace.create', () => {

packages/host/apiproxy/tests/client-handler.spec.ts

@@ -7,7 +7,7 @@
 
 import { describe, expect, it, vi } from 'vitest'
 import type { SessionId } from '@deepseek-ai/dsh-session'
-import type { ApiProxy, HostFrame, MuxFrame, RpcMessage, RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy'
+import type { ApiProxy, GoalRef, HostFrame, MuxFrame, RpcMessage, RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy'
 import { InProcessApiClient, RpcId, toFetchHandler } from '@deepseek-ai/dsh-host-apiproxy'
 
 const sid = (id: string): SessionId => id as SessionId
@@ -23,9 +23,12 @@ function scriptedApi(overrides: {
   commands?: Partial<ApiProxy['commands']>
   skills?: Partial<ApiProxy['skills']>
   events?: Partial<ApiProxy['events']>
+  goals?: Partial<ApiProxy['goals']>
   respond?: ApiProxy['respond']
 } = {}): ApiProxy {
   async function *empty<F>(): AsyncGenerator<RpcRequest<F>> { /* no frames */ }
+  const err = <T>(r: RpcRequest<unknown>): Promise<RpcResponse<T>> =>
+    Promise.resolve({ rpcId: r.rpcId, result: { ok: false, error: { code: 'internal' as const, message: 'stub', details: {} } } })
   return {
     sessions: {
       list: r => ok(r, { items: [] }),
@@ -50,6 +53,9 @@ function scriptedApi(overrides: {
     host: {
       describe: r => ok(r, { version: '0-test', cwd: '/t', attachedSessions: 0 }),
       pickDirectory: r => ok(r, { path: null }),
+      listDirectory: r => ok(r, { path: '/t', home: '/t', crumbs: [], entries: [], truncated: false }),
+      createDirectory: r => ok(r, { path: '/t/new' }),
+      openPath: r => ok(r, { opened: true as const }),
       ...overrides.host,
     },
     workspace: {
@@ -65,6 +71,15 @@ function scriptedApi(overrides: {
       ...overrides.commands,
     },
     skills: { list: r => ok(r, { skills: [] }), ...overrides.skills },
+    goals: {
+      create: err,
+      edit: err,
+      pause: err,
+      resume: err,
+      complete: err,
+      clear: err,
+      ...overrides.goals,
+    },
     events: { mux: () => empty<MuxFrame>(), host: () => empty<HostFrame>(), ...overrides.events },
     respond: overrides.respond ?? (() => Promise.resolve({ accepted: false as const, reason: 'not-pending' as const })),
   }
@@ -138,7 +153,7 @@ describe('unary round trip', () => {
   it('rejects a method/path mismatch as bad-request', async () => {
     const handler = toFetchHandler(scriptedApi())
     const body = { type: 'client-request', rpcId: 'r1', method: 'session.create', payload: {} }
-    const response = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', body: JSON.stringify(body) })
+    const response = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify(body) })
     expect(response.status).toBe(200)
     const parsed = await response.json() as { result: { ok: boolean; error?: { code: string; message: string } } }
     expect(parsed.result.ok).toBe(false)
@@ -149,13 +164,13 @@ describe('unary round trip', () => {
   it('rejects a malformed envelope as bad-request, salvaging the rpcId or falling back to the sentinel', async () => {
     const handler = toFetchHandler(scriptedApi())
     // No salvageable rpcId → the fixed invalid-request sentinel keeps the response a valid ServerResponse.
-    const noId = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', body: JSON.stringify({ nonsense: true }) })
+    const noId = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ nonsense: true }) })
     expect(noId.status).toBe(200)
     const noIdParsed = await noId.json() as { rpcId: string; result: { ok: boolean } }
     expect(noIdParsed.result.ok).toBe(false)
     expect(noIdParsed.rpcId).toBe('invalid-request')
     // A string rpcId in the otherwise-bad body is salvaged for correlation.
-    const withId = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', body: JSON.stringify({ rpcId: 'salvage-me', nonsense: true }) })
+    const withId = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ rpcId: 'salvage-me', nonsense: true }) })
     const withIdParsed = await withId.json() as { rpcId: string; result: { ok: boolean } }
     expect(withIdParsed.result.ok).toBe(false)
     expect(withIdParsed.rpcId).toBe('salvage-me')
@@ -164,16 +179,34 @@ describe('unary round trip', () => {
   it('maps carrier failures to HTTP statuses and the client throws transport failure', async () => {
     const handler = toFetchHandler(scriptedApi())
     // Unknown method → 404.
-    const notFound = await handler.fetch('http://dsh.internal/api/no.such', { method: 'POST', body: '{}' })
+    const notFound = await handler.fetch('http://dsh.internal/api/no.such', { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })
     expect(notFound.status).toBe(404)
     // Non-JSON body → 400.
-    const badBody = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', body: '{oops' })
+    const badBody = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{oops' })
     expect(badBody.status).toBe(400)
     // Impl crash → 500, and through the client that is a throw, not an err result.
     const crashing = scriptedApi({ sessions: { list: () => { throw new Error('impl exploded') } } })
     await expect(client(crashing).sessions.list({})).rejects.toThrow(/transport failure .*500/)
   })
 
+  it('rejects non-JSON media types before executing anything (cross-site simple-request fence)', async () => {
+    const list = vi.fn((r: RpcRequest<{}>) => ok(r, { items: [] }))
+    const handler = toFetchHandler(scriptedApi({ sessions: { list } }))
+    const body = JSON.stringify({ type: 'client-request', rpcId: 'r1', method: 'session.list', payload: {} })
+    // A "simple" browser POST (text/plain — sent with no CORS preflight) is
+    // refused at the carrier before the impl runs.
+    const plain = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', headers: { 'content-type': 'text/plain' }, body })
+    expect(plain.status).toBe(415)
+    // A string body with no explicit header defaults to text/plain — same fence.
+    const unlabelled = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', body })
+    expect(unlabelled.status).toBe(415)
+    expect(list).not.toHaveBeenCalled()
+    // Media-type parameters pass: the fence checks the type, not the exact string.
+    const charset = await handler.fetch('http://dsh.internal/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json; charset=utf-8' }, body })
+    expect(charset.status).toBe(200)
+    expect(list).toHaveBeenCalledTimes(1)
+  })
+
   it('rejects when the transport never resolves within timeoutMs', async () => {
     // AbortSignal.timeout is immune to fake timers; a short real timeout keeps this fast.
     const never = new InProcessApiClient({
@@ -404,6 +437,67 @@ describe('SSE stream path', () => {
   })
 })
 
+describe('goals unary surface', () => {
+  const ref: GoalRef = { id: 'goal-1' as GoalRef['id'], revision: 1 }
+  /** The `{ ref }` acknowledgement every non-clear mutation answers (state travels on the projection). */
+  const ack = { ref: { id: 'goal-1' as GoalRef['id'], revision: 2 } }
+
+  it('round-trips every goal method with its own payload and value shape', async () => {
+    const seen: { method: string; payload: unknown }[] = []
+    const record = <P, V>(method: string, respond: (r: RpcRequest<P>) => Promise<RpcResponse<V>>) =>
+      (r: RpcRequest<P>): Promise<RpcResponse<V>> => {
+        seen.push({ method, payload: r.payload })
+        return respond(r)
+      }
+    const api = scriptedApi({
+      goals: {
+        create: record('goal.create', r => ok(r, ack)),
+        edit: record('goal.edit', r => ok(r, { ref: { ...ack.ref, revision: 3 } })),
+        pause: record('goal.pause', r => ok(r, ack)),
+        resume: record('goal.resume', r => ok(r, ack)),
+        complete: record('goal.complete', r => ok(r, ack)),
+        clear: record('goal.clear', r => ok(r, { cleared: true as const })),
+      },
+    })
+    const c = client(api)
+
+    const created = await c.goals.create({ sessionId: sid('s1'), objective: 'ship it', maxGoalRounds: 4 })
+    expect(created.result).toEqual({ ok: true, value: ack })
+    const edited = await c.goals.edit({ sessionId: sid('s1'), ref, objective: 'ship v2' })
+    expect(edited.result).toEqual({ ok: true, value: { ref: { ...ack.ref, revision: 3 } } })
+    expect((await c.goals.pause({ sessionId: sid('s1'), ref })).result).toEqual({ ok: true, value: ack })
+    expect((await c.goals.resume({ sessionId: sid('s1'), ref })).result).toEqual({ ok: true, value: ack })
+    expect((await c.goals.complete({ sessionId: sid('s1'), ref })).result).toEqual({ ok: true, value: ack })
+    const cleared = await c.goals.clear({ sessionId: sid('s1'), ref })
+    expect(cleared.result).toEqual({ ok: true, value: { cleared: true } })
+
+    // The handler dispatched each call through its own route row: payload parsed per method.
+    expect(seen.map(s => s.method)).toEqual(['goal.create', 'goal.edit', 'goal.pause', 'goal.resume', 'goal.complete', 'goal.clear'])
+    expect(seen[0]?.payload).toEqual({ sessionId: 's1', objective: 'ship it', maxGoalRounds: 4 })
+    expect(seen[1]?.payload).toEqual({ sessionId: 's1', ref, objective: 'ship v2' })
+  })
+
+  it('passes business errors through as results, not throws', async () => {
+    // Default scripted goals impl answers an err result: it must arrive as a result, not a throw.
+    const failed = await client(scriptedApi()).goals.pause({ sessionId: sid('s1'), ref })
+    expect(failed.result.ok).toBe(false)
+    if (!failed.result.ok) expect(failed.result.error.code).toBe('internal')
+  })
+
+  it('rejects an invalid goal payload at the handler as bad-request', async () => {
+    const response = await client(scriptedApi()).goals.create({ sessionId: sid('s1'), objective: '' })
+    expect(response.result.ok).toBe(false)
+    if (!response.result.ok) expect(response.result.error.code).toBe('bad-request')
+
+    let editCalls = 0
+    const api = scriptedApi({ goals: { edit: (r) => { editCalls++; return ok(r, ack) } } })
+    const emptyEdit = await client(api).goals.edit({ sessionId: sid('s1'), ref })
+    expect(emptyEdit.result.ok).toBe(false)
+    if (!emptyEdit.result.ok) expect(emptyEdit.result.error.code).toBe('bad-request')
+    expect(editCalls).toBe(0)
+  })
+})
+
 describe('respond path', () => {
   it('round-trips a client-response to a receipt', async () => {
     const seen: unknown[] = []
@@ -421,7 +515,7 @@ describe('respond path', () => {
   it('returns bad-response for a malformed client-response without reaching the impl', async () => {
     const respond = vi.fn()
     const handler = toFetchHandler(scriptedApi({ respond }))
-    const response = await handler.fetch('http://dsh.internal/api/respond', { method: 'POST', body: JSON.stringify({ type: 'client-response' }) })
+    const response = await handler.fetch('http://dsh.internal/api/respond', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ type: 'client-response' }) })
     expect(await response.json()).toEqual({ accepted: false, reason: 'bad-response' })
     expect(respond).not.toHaveBeenCalled()
   })

packages/host/apiproxy/tests/fetch-carrier.spec.ts

@@ -1,3 +1,4 @@
+import { CommandId } from '@deepseek-ai/dsh-commands/brand'
 import { describe, expect, it, vi } from 'vitest'
 import type { ApiProxy, HostFrame, MuxFrame } from '../src/api/index.ts'
 import type { ClientResponse, RpcMessage, RpcReceipt, RpcRequest } from '../src/api/rpc.ts'
@@ -25,10 +26,10 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra
         return { rpcId: request.rpcId, result: { ok: true, value: { sessionId: 's-new' as never } } }
       },
       async history(request) {
-        if (request.payload.sessionId === ('with-todos' as never)) {
+        if (request.payload.sessionId === ('with-projections' as never)) {
           return {
             rpcId: request.rpcId,
-            result: { ok: true, value: { events: [], hasMore: false, todos: [{ content: 'current', status: 'in_progress' as const }] } },
+            result: { ok: true, value: { events: [], hasMore: false, projections: { asOfSeq: 9, values: { todos: [{ content: 'current', status: 'in_progress' as const }] } } } },
           }
         }
         return {
@@ -80,6 +81,15 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra
       async pickDirectory(request) {
         return { rpcId: request.rpcId, result: { ok: true, value: { path: null } } }
       },
+      async listDirectory(request) {
+        return { rpcId: request.rpcId, result: { ok: true, value: { path: '/w', home: '/w', crumbs: [{ name: '/', path: '/', hidden: false }], entries: [], truncated: false } } }
+      },
+      async createDirectory(request) {
+        return { rpcId: request.rpcId, result: { ok: true, value: { path: '/w/new' } } }
+      },
+      async openPath(request) {
+        return { rpcId: request.rpcId, result: { ok: true, value: { opened: true as const } } }
+      },
     },
     workspace: {
       async list(request) {
@@ -121,7 +131,7 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra
           return { rpcId: request.rpcId, result: { ok: false, error: { code: 'cancelled', message: 'aborted', details: {} } } }
         }
         if (request.payload.line.startsWith('/plan')) {
-          return { rpcId: request.rpcId, result: { ok: true, value: { matched: true, result: { kind: 'success' as const, text: 'plan set' } } } }
+          return { rpcId: request.rpcId, result: { ok: true, value: { matched: true, commandId: CommandId('cmd-x') } } }
         }
         return { rpcId: request.rpcId, result: { ok: true, value: { matched: false } } }
       },
@@ -131,6 +141,26 @@ function fakeApi(overrides: Partial<{ muxFrames: MuxFrame[]; hostFrames: HostFra
         return { rpcId: request.rpcId, result: { ok: true, value: { skills: [{ name: 'commit-helper', description: 'Git commits' }] } } }
       },
     },
+    goals: {
+      async create(request) {
+        return { rpcId: request.rpcId, result: { ok: false, error: { code: 'internal', message: 'stub', details: {} } } }
+      },
+      async edit(request) {
+        return { rpcId: request.rpcId, result: { ok: false, error: { code: 'internal', message: 'stub', details: {} } } }
+      },
+      async pause(request) {
+        return { rpcId: request.rpcId, result: { ok: false, error: { code: 'internal', message: 'stub', details: {} } } }
+      },
+      async resume(request) {
+        return { rpcId: request.rpcId, result: { ok: false, error: { code: 'internal', message: 'stub', details: {} } } }
+      },
+      async complete(request) {
+        return { rpcId: request.rpcId, result: { ok: false, error: { code: 'internal', message: 'stub', details: {} } } }
+      },
+      async clear(request) {
+        return { rpcId: request.rpcId, result: { ok: false, error: { code: 'internal', message: 'stub', details: {} } } }
+      },
+    },
     events: {
       mux: (_request, signal) => stream(muxFrames, signal),
       host: (_request, signal) => stream(hostFrames, signal),
@@ -158,10 +188,14 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
     expect(response.rpcId).toMatch(/[0-9a-f-]{36}/)
   })
 
-  it('carries the tail-page todos projection through the wire schema (Zod must not strip it)', async () => {
-    const response = await client().sessions.history({ sessionId: 'with-todos' as never })
+  it('carries the tail-page projections block through the wire schema (Zod must not strip it)', async () => {
+    const response = await client().sessions.history({ sessionId: 'with-projections' as never })
     expect(response.result.ok).toBe(true)
-    if (response.result.ok) expect(response.result.value.todos).toEqual([{ content: 'current', status: 'in_progress' }])
+    if (response.result.ok) {
+      expect(response.result.value.projections).toEqual(
+        { asOfSeq: 9, values: { todos: [{ content: 'current', status: 'in_progress' }] } },
+      )
+    }
   })
 
   it('carries a business error as 200 + error result', async () => {
@@ -205,12 +239,37 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
     expect(response.result).toEqual({ ok: true, value: { path: '/tmp/project' } })
   })
 
+  it('round-trips the browse listing and creation calls through the wire form', async () => {
+    const c = client()
+    const listed = await c.host.listDirectory({ path: '/w' })
+    expect(listed.result).toEqual({
+      ok: true,
+      value: { path: '/w', home: '/w', crumbs: [{ name: '/', path: '/', hidden: false }], entries: [], truncated: false },
+    })
+    const home = await c.host.listDirectory({})
+    expect(home.result).toMatchObject({ ok: true, value: { home: '/w' } })
+    const created = await c.host.createDirectory({ path: '/w', name: 'fresh' })
+    expect(created.result).toEqual({ ok: true, value: { path: '/w/new' } })
+  })
+
+  it('round-trips host.openPath through the wire form', async () => {
+    const api = fakeApi()
+    let opened: string | undefined
+    api.host.openPath = async (request) => {
+      opened = request.payload.path
+      return { rpcId: request.rpcId, result: { ok: true, value: { opened: true as const } } }
+    }
+    const response = await client(api).host.openPath({ path: '/tmp/a.txt' })
+    expect(opened).toBe('/tmp/a.txt')
+    expect(response.result).toEqual({ ok: true, value: { opened: true } })
+  })
+
   it('round-trips command.list / command.execute / skill.list through the wire form', async () => {
     const c = client()
     const list = await c.commands.list({ sessionId: 's' as never })
     expect(list.result).toEqual({ ok: true, value: { commands: [{ name: 'plan', description: 'Toggle plan mode', input: { hint: 'on|off' } }] } })
     const hit = await c.commands.execute({ sessionId: 's' as never, line: '/plan off' })
-    expect(hit.result).toEqual({ ok: true, value: { matched: true, result: { kind: 'success', text: 'plan set' } } })
+    expect(hit.result).toEqual({ ok: true, value: { matched: true, commandId: 'cmd-x' } })
     const miss = await c.commands.execute({ sessionId: 's' as never, line: '/nope' })
     expect(miss.result).toEqual({ ok: true, value: { matched: false } })
     const skills = await c.skills.list({ sessionId: 's' as never })
@@ -223,7 +282,7 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
     const body = JSON.stringify({ type: 'client-request', rpcId: 'r-sig', method: 'command.execute', payload: { sessionId: 's', line: '/hang' } })
     // The fake's /hang settles only when the invoke-level signal aborts: a
     // completed response with the cancelled error proves req.signal reached it.
-    const pending = handler.fetch(new Request('http://x/api/command.execute', { method: 'POST', body, signal: controller.signal }))
+    const pending = handler.fetch(new Request('http://x/api/command.execute', { method: 'POST', headers: { 'content-type': 'application/json' }, body, signal: controller.signal }))
     controller.abort()
     const response = await pending
     const parsed = await response.json() as { rpcId: string; result: { ok: boolean; error?: { code: string } } }
@@ -248,7 +307,7 @@ describe('unary round trip (handler ⇄ client, no network)', () => {
     const controller = new AbortController()
     const body = JSON.stringify({ type: 'client-request', rpcId: 'r-picker', method: 'host.pickDirectory', payload: {} })
     const pending = handler.fetch(new Request('http://x/api/host.pickDirectory', {
-      method: 'POST', body, signal: controller.signal,
+      method: 'POST', headers: { 'content-type': 'application/json' }, body, signal: controller.signal,
     }))
     controller.abort()
     const parsed = await (await pending).json() as { result: { error?: { code: string } } }
@@ -260,18 +319,18 @@ describe('handler carrier-layer statuses', () => {
   const handler = toFetchHandler(fakeApi())
 
   it('404s unknown paths and non-POST non-stream methods', async () => {
-    expect((await handler.fetch(new Request('http://x/other', { method: 'POST', body: '{}' }))).status).toBe(404)
+    expect((await handler.fetch(new Request('http://x/other', { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' }))).status).toBe(404)
     expect((await handler.fetch(new Request('http://x/api/session.list', { method: 'GET' }))).status).toBe(404)
-    expect((await handler.fetch(new Request('http://x/api/no.such', { method: 'POST', body: JSON.stringify({ type: 'client-request', rpcId: 'r', method: 'no.such', payload: {} }) }))).status).toBe(404)
+    expect((await handler.fetch(new Request('http://x/api/no.such', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ type: 'client-request', rpcId: 'r', method: 'no.such', payload: {} }) }))).status).toBe(404)
   })
 
   it('400s a non-JSON body', async () => {
-    const response = await handler.fetch(new Request('http://x/api/session.list', { method: 'POST', body: 'not json' }))
+    const response = await handler.fetch(new Request('http://x/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body: 'not json' }))
     expect(response.status).toBe(400)
   })
 
   it('rejects a malformed envelope with bad-request and the invalid-request sentinel rpcId', async () => {
-    const response = await handler.fetch(new Request('http://x/api/session.list', { method: 'POST', body: JSON.stringify({ nope: true }) }))
+    const response = await handler.fetch(new Request('http://x/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ nope: true }) }))
     expect(response.status).toBe(200)
     const body = await response.json() as { rpcId: string; result: { ok: boolean; error?: { code: string } } }
     expect(body.rpcId).toBe('invalid-request')
@@ -280,7 +339,7 @@ describe('handler carrier-layer statuses', () => {
 
   it('rejects a method/path mismatch echoing the envelope rpcId', async () => {
     const body = JSON.stringify({ type: 'client-request', rpcId: 'r-9', method: 'session.cancel', payload: {} })
-    const response = await handler.fetch(new Request('http://x/api/session.list', { method: 'POST', body }))
+    const response = await handler.fetch(new Request('http://x/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body }))
     const parsed = await response.json() as { rpcId: string; result: { error?: { message: string } } }
     expect(parsed.rpcId).toBe('r-9')
     expect(parsed.result.error?.message).toContain('does not match path')
@@ -288,7 +347,7 @@ describe('handler carrier-layer statuses', () => {
 
   it('rejects an invalid payload with the zod issues attached', async () => {
     const body = JSON.stringify({ type: 'client-request', rpcId: 'r-10', method: 'session.cancel', payload: {} })
-    const response = await handler.fetch(new Request('http://x/api/session.cancel', { method: 'POST', body }))
+    const response = await handler.fetch(new Request('http://x/api/session.cancel', { method: 'POST', headers: { 'content-type': 'application/json' }, body }))
     const parsed = await response.json() as { result: { error?: { code: string; details: { issues: unknown[] } } } }
     expect(parsed.result.error?.code).toBe('bad-request')
     expect(parsed.result.error?.details.issues.length).toBeGreaterThan(0)
@@ -297,23 +356,23 @@ describe('handler carrier-layer statuses', () => {
   it('500s when the impl itself throws', async () => {
     const crashing = toFetchHandler(fakeApi({ crashOn: 'session.list' }))
     const body = JSON.stringify({ type: 'client-request', rpcId: 'r-11', method: 'session.list', payload: {} })
-    const response = await crashing.fetch(new Request('http://x/api/session.list', { method: 'POST', body }))
+    const response = await crashing.fetch(new Request('http://x/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body }))
     expect(response.status).toBe(500)
     expect(await response.text()).toContain('impl crashed')
   })
 
   it('routes /api/respond, rejecting malformed client-responses as a receipt', async () => {
     const good = JSON.stringify({ type: 'client-response', rpcId: 'known', result: { ok: true, value: null } })
-    const goodReceipt: unknown = await (await handler.fetch(new Request('http://x/api/respond', { method: 'POST', body: good }))).json()
+    const goodReceipt: unknown = await (await handler.fetch(new Request('http://x/api/respond', { method: 'POST', headers: { 'content-type': 'application/json' }, body: good }))).json()
     expect(goodReceipt).toEqual({ accepted: true })
     const bad = JSON.stringify({ type: 'client-request', rpcId: 'r', method: 'x', payload: {} })
-    const badReceipt: unknown = await (await handler.fetch(new Request('http://x/api/respond', { method: 'POST', body: bad }))).json()
+    const badReceipt: unknown = await (await handler.fetch(new Request('http://x/api/respond', { method: 'POST', headers: { 'content-type': 'application/json' }, body: bad }))).json()
     expect(badReceipt).toEqual({ accepted: false, reason: 'bad-response' })
   })
 
   it('accepts (url, init) form fetch invocation', async () => {
     const body = JSON.stringify({ type: 'client-request', rpcId: 'r-12', method: 'session.list', payload: {} })
-    const response = await handler.fetch('http://x/api/session.list', { method: 'POST', body })
+    const response = await handler.fetch('http://x/api/session.list', { method: 'POST', headers: { 'content-type': 'application/json' }, body })
     expect(response.status).toBe(200)
   })
 })

packages/host/apiproxy/tests/native-directory-picker.spec.ts

@@ -1,139 +0,0 @@
-type ExecFileCallback = (
-  error: (Error & { code?: string | number }) | null,
-  stdout: string,
-  stderr: string,
-) => void
-type ExecFileMock = (
-  command: string,
-  args: readonly string[],
-  options: { encoding: string; signal: AbortSignal; windowsHide: boolean },
-  callback: ExecFileCallback,
-) => void
-
-const { execFileMock } = vi.hoisted(() => ({ execFileMock: vi.fn<ExecFileMock>() }))
-
-vi.mock('node:child_process', () => ({ execFile: execFileMock }))
-
-import { describe, expect, it, vi } from 'vitest'
-import { pickNativeDirectory, type DirectoryPickerRunner } from '../src/native-directory-picker.ts'
-
-function failure(code: string | number, stderr = ''): Error {
-  return Object.assign(new Error(`command failed: ${String(code)}`), { code, stderr })
-}
-
-const signal = () => new AbortController().signal
-
-describe('native directory picker', () => {
-  it('uses the macOS folder chooser and maps user cancellation to null', async () => {
-    const run = vi.fn<DirectoryPickerRunner>(async () => ({ stdout: '/Users/test/project/\n', stderr: '' }))
-    await expect(pickNativeDirectory(signal(), { platform: 'darwin', run })).resolves.toBe('/Users/test/project/')
-    expect(run).toHaveBeenCalledWith('osascript', expect.arrayContaining(['POSIX path of selectedFolder']), expect.any(AbortSignal))
-
-    run.mockRejectedValueOnce(failure(1, 'execution error: User canceled. (-128)'))
-    await expect(pickNativeDirectory(signal(), { platform: 'darwin', run })).resolves.toBeNull()
-
-    run.mockRejectedValueOnce(failure(2, 'permission denied'))
-    await expect(pickNativeDirectory(signal(), { platform: 'darwin', run })).rejects.toThrow('command failed')
-  })
-
-  it.each([
-    ['a primitive error', 'failed'],
-    ['an invalid code type', { code: true }],
-    ['a missing stderr property', { code: 1 }],
-    ['a non-string stderr property', { code: 1, stderr: 42 }],
-  ])('does not mistake %s for macOS cancellation', async (_label, reason) => {
-    const run = vi.fn<DirectoryPickerRunner>(async () => { throw reason })
-    await expect(pickNativeDirectory(signal(), { platform: 'darwin', run })).rejects.toBe(reason)
-  })
-
-  it('uses the Windows STA folder dialog and maps empty output to cancellation', async () => {
-    const run = vi.fn<DirectoryPickerRunner>(async () => ({ stdout: 'C:\\work\\project\r\n', stderr: '' }))
-    await expect(pickNativeDirectory(signal(), { platform: 'win32', run })).resolves.toBe('C:\\work\\project')
-    expect(run).toHaveBeenCalledWith(
-      'powershell.exe',
-      expect.arrayContaining(['-NoProfile', '-STA', '-Command']),
-      expect.any(AbortSignal),
-    )
-    expect(run.mock.calls[0]?.[1].at(-1)).toContain("$ErrorActionPreference = 'Stop'")
-    run.mockResolvedValueOnce({ stdout: '', stderr: '' })
-    await expect(pickNativeDirectory(signal(), { platform: 'win32', run })).resolves.toBeNull()
-    run.mockRejectedValueOnce(failure(1, 'Add-Type failed'))
-    await expect(pickNativeDirectory(signal(), { platform: 'win32', run })).rejects.toThrow('command failed')
-  })
-
-  it('runs the default command adapter without a shell and preserves command failures', async () => {
-    execFileMock.mockImplementationOnce((_command, _args, _options, callback) => {
-      callback(null, 'C:\\work\\default\r\n', '')
-    })
-    await expect(pickNativeDirectory(signal(), { platform: 'win32' })).resolves.toBe('C:\\work\\default')
-    const [command, args, options] = execFileMock.mock.calls[0]!
-    expect(command).toBe('powershell.exe')
-    expect(args).toEqual(expect.arrayContaining(['-NoProfile', '-STA', '-Command']))
-    expect(options.encoding).toBe('utf8')
-    expect(options.windowsHide).toBe(true)
-    expect(options.signal).toBeInstanceOf(AbortSignal)
-
-    const commandError = Object.assign(new Error('powershell failed'), { code: 7 })
-    execFileMock.mockImplementationOnce((_command, _args, _options, callback) => {
-      callback(commandError, 'partial output', 'failure details')
-    })
-    await expect(pickNativeDirectory(signal(), { platform: 'win32' })).rejects.toMatchObject({
-      message: 'powershell failed', cause: commandError, code: 7,
-      stdout: 'partial output', stderr: 'failure details',
-    })
-  })
-
-  it('uses the current process platform when no platform override is supplied', async () => {
-    const run = vi.fn<DirectoryPickerRunner>(async () => ({ stdout: '/default/platform\n', stderr: '' }))
-    await expect(pickNativeDirectory(signal(), { run })).resolves.toBe('/default/platform')
-  })
-
-  it('uses Zenity on Linux and falls back to KDialog only when Zenity is missing', async () => {
-    const run = vi.fn<DirectoryPickerRunner>()
-      .mockRejectedValueOnce(failure('ENOENT'))
-      .mockResolvedValueOnce({ stdout: '/home/test/project\n', stderr: '' })
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run })).resolves.toBe('/home/test/project')
-    expect(run.mock.calls.map(call => call[0])).toEqual(['zenity', 'kdialog'])
-
-    const zenity = vi.fn<DirectoryPickerRunner>(async () => ({ stdout: '/home/test/direct\n', stderr: '' }))
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run: zenity }))
-      .resolves.toBe('/home/test/direct')
-    expect(zenity).toHaveBeenCalledOnce()
-  })
-
-  it('maps Linux cancellation to null and reports a missing desktop picker', async () => {
-    const cancelled = vi.fn<DirectoryPickerRunner>(async () => { throw failure(1) })
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run: cancelled })).resolves.toBeNull()
-
-    const missing = vi.fn<DirectoryPickerRunner>(async () => { throw failure('ENOENT') })
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run: missing }))
-      .rejects.toThrow('install zenity or kdialog')
-
-    const kdialogCancelled = vi.fn<DirectoryPickerRunner>()
-      .mockRejectedValueOnce(failure('ENOENT'))
-      .mockRejectedValueOnce(failure(1))
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run: kdialogCancelled }))
-      .resolves.toBeNull()
-
-    const zenityFailed = vi.fn<DirectoryPickerRunner>(async () => { throw failure(2) })
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run: zenityFailed }))
-      .rejects.toThrow('command failed')
-
-    const kdialogFailed = vi.fn<DirectoryPickerRunner>()
-      .mockRejectedValueOnce(failure('ENOENT'))
-      .mockRejectedValueOnce(failure(2))
-    await expect(pickNativeDirectory(signal(), { platform: 'linux', run: kdialogFailed }))
-      .rejects.toThrow('command failed')
-  })
-
-  it('does not convert caller aborts into user cancellation', async () => {
-    const abort = new AbortController()
-    abort.abort(new Error('closed'))
-    const run = vi.fn<DirectoryPickerRunner>(async () => { throw failure('ABORT_ERR') })
-    await expect(pickNativeDirectory(abort.signal, { platform: 'linux', run })).rejects.toThrow('command failed')
-  })
-
-  it('reports unsupported platforms', async () => {
-    await expect(pickNativeDirectory(signal(), { platform: 'aix' })).rejects.toThrow('unsupported on aix')
-  })
-})

packages/host/apiproxy/tests/native-path-opener.spec.ts

@@ -0,0 +1,82 @@
+type ExecFileCallback = (
+  error: (Error & { code?: string | number }) | null,
+  stdout: string,
+  stderr: string,
+) => void
+type ExecFileMock = (
+  command: string,
+  args: readonly string[],
+  options: { encoding: string; signal: AbortSignal; windowsHide: boolean },
+  callback: ExecFileCallback,
+) => void
+
+const { execFileMock } = vi.hoisted(() => ({ execFileMock: vi.fn<ExecFileMock>() }))
+
+vi.mock('node:child_process', () => ({ execFile: execFileMock }))
+
+import { describe, expect, it, vi } from 'vitest'
+import { openNativePath, type PathOpenerRunner } from '../src/native-path-opener.ts'
+
+const signal = () => new AbortController().signal
+
+describe('native path opener', () => {
+  it('opens with macOS open(1)', async () => {
+    const run = vi.fn<PathOpenerRunner>(async () => ({ stdout: '', stderr: '' }))
+    await openNativePath('/Users/test/file.txt', signal(), { platform: 'darwin', run })
+    expect(run).toHaveBeenCalledWith('open', ['/Users/test/file.txt'], expect.any(AbortSignal))
+  })
+
+  it('opens with Windows Invoke-Item and escapes single quotes', async () => {
+    const run = vi.fn<PathOpenerRunner>(async () => ({ stdout: '', stderr: '' }))
+    await openNativePath("C:\\work\\o'reilly.txt", signal(), { platform: 'win32', run })
+    expect(run).toHaveBeenCalledWith(
+      'powershell.exe',
+      ['-NoProfile', '-Command', "Invoke-Item -LiteralPath 'C:\\work\\o''reilly.txt'"],
+      expect.any(AbortSignal),
+    )
+  })
+
+  it('opens with Linux xdg-open', async () => {
+    const run = vi.fn<PathOpenerRunner>(async () => ({ stdout: '', stderr: '' }))
+    await openNativePath('/tmp/a.txt', signal(), { platform: 'linux', run })
+    expect(run).toHaveBeenCalledWith('xdg-open', ['/tmp/a.txt'], expect.any(AbortSignal))
+  })
+
+  it('rejects unsupported platforms', async () => {
+    await expect(openNativePath('/x', signal(), { platform: 'freebsd' as NodeJS.Platform }))
+      .rejects.toThrow('unsupported on freebsd')
+  })
+
+  it('uses the current process platform when no platform override is supplied', async () => {
+    const run = vi.fn<PathOpenerRunner>(async () => ({ stdout: '', stderr: '' }))
+    await openNativePath('/tmp/platform-default.txt', signal(), { run })
+    const expected = process.platform === 'win32'
+      ? 'powershell.exe'
+      : process.platform === 'linux'
+        ? 'xdg-open'
+        : 'open'
+    expect(run.mock.calls[0]?.[0]).toBe(expected)
+  })
+
+  it('runs the default command adapter without a shell and preserves command failures', async () => {
+    execFileMock.mockImplementationOnce((_command, _args, _options, callback) => {
+      callback(null, '', '')
+    })
+    await openNativePath('/tmp/default.txt', signal(), { platform: 'darwin' })
+    const [command, args, options] = execFileMock.mock.calls[0]!
+    expect(command).toBe('open')
+    expect(args).toEqual(['/tmp/default.txt'])
+    expect(options.encoding).toBe('utf8')
+    expect(options.windowsHide).toBe(true)
+    expect(options.signal).toBeInstanceOf(AbortSignal)
+
+    const commandError = Object.assign(new Error('open failed'), { code: 1 })
+    execFileMock.mockImplementationOnce((_command, _args, _options, callback) => {
+      callback(commandError, 'partial output', 'failure details')
+    })
+    await expect(openNativePath('/tmp/missing.txt', signal(), { platform: 'darwin' })).rejects.toMatchObject({
+      message: 'open failed', cause: commandError, code: 1,
+      stdout: 'partial output', stderr: 'failure details',
+    })
+  })
+})

packages/host/apiproxy/tests/rpc-schemas.spec.ts

@@ -12,7 +12,11 @@ import {
   sessionModelsValueSchema, sessionPromptRequestSchema, sessionPromptValueSchema,
   sessionSelectModelRequestSchema, sessionSelectModelValueSchema, sessionSummarySchema,
 } from '../src/api/sessions.schema.ts'
-import { hostDescribeRequestSchema, hostDescribeValueSchema } from '../src/api/host.schema.ts'
+import {
+  hostCreateDirectoryRequestSchema, hostCreateDirectoryValueSchema,
+  hostDescribeRequestSchema, hostDescribeValueSchema,
+  hostListDirectoryRequestSchema, hostListDirectoryValueSchema,
+} from '../src/api/host.schema.ts'
 import {
   workspaceCreateRequestSchema, workspaceCreateValueSchema, workspaceIdSchema,
   workspaceDeleteRequestSchema, workspaceDeleteValueSchema,
@@ -28,6 +32,7 @@ import { skillEntrySchema, skillListRequestSchema, skillListValueSchema } from '
 import { hostFrameSchema, muxFrameSchema, askUserQuestionItemSchema } from '../src/api/events.schema.ts'
 import { approvalRequestIdSchema, approvalResponsePayloadSchema } from '../src/api/approvals.schema.ts'
 import { askUserQuestionAnswerSchema, questionResponsePayloadSchema } from '../src/api/questions.schema.ts'
+import { goalEditRequestSchema } from '../src/api/goals.schema.ts'
 
 describe('RpcId', () => {
   it('brands a raw string at zero runtime cost', () => {
@@ -63,11 +68,14 @@ describe('rpcErrorSchema', () => {
       details: { provider: 'p', model: 'm' },
     }).code).toBe('model-unavailable')
     expect(rpcErrorSchema.parse({ code: 'agent-busy', message: 'm', details: { reason: 'r' } }).code).toBe('agent-busy')
+    expect(rpcErrorSchema.parse({ code: 'command-error', message: 'm', details: {} }).code).toBe('command-error')
+    expect(rpcErrorSchema.parse({ code: 'unknown-command', message: 'm', details: {} }).code).toBe('unknown-command')
     expect(rpcErrorSchema.parse({ code: 'internal', message: 'm', details: {} }).code).toBe('internal')
   })
 
   it('rejects a known code with missing details', () => {
     expect(() => rpcErrorSchema.parse({ code: 'agent-busy', message: 'm', details: {} })).toThrow()
+    expect(() => rpcErrorSchema.parse({ code: 'command-error', message: 'm' })).toThrow()
     expect(() => rpcErrorSchema.parse({ code: 'nope', message: 'm', details: {} })).toThrow()
   })
 })
@@ -119,9 +127,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', () => {
@@ -195,6 +213,11 @@ describe('sessions domain schemas', () => {
     expect(prompt.mode).toBe('queue')
     expect(() => sessionPromptRequestSchema.parse({ sessionId: 's1', mode: 'inject', content: [] })).toThrow()
     expect(sessionPromptValueSchema.parse({ accepted: true }).accepted).toBe(true)
+    // The command slot appears only when the prompt dispatched a slash command.
+    const dispatched = sessionPromptValueSchema.parse({ accepted: true, command: { kind: 'success', text: 'Goal set' } })
+    expect(dispatched.command?.text).toBe('Goal set')
+    expect(sessionPromptValueSchema.parse({ accepted: true, command: { kind: 'success' } }).command).toEqual({ kind: 'success' })
+    expect(() => sessionPromptValueSchema.parse({ accepted: true, command: { kind: 'failure' } })).toThrow()
     expect(sessionCancelRequestSchema.parse({ sessionId: 's1' }).sessionId).toBe('s1')
     expect(sessionCancelValueSchema.parse({ accepted: true }).accepted).toBe(true)
     expect(contentBlockSchema.parse({ type: 'text', text: 'x', extra: 1 })).toMatchObject({ extra: 1 })
@@ -208,6 +231,26 @@ describe('host domain schemas', () => {
     expect(value.attachedSessions).toBe(2)
     expect(hostDescribeValueSchema.parse({ version: '1', cwd: '/x', attachedSessions: 0 }).provider).toBeUndefined()
   })
+
+  it('validates the browse listing/creation payloads', () => {
+    expect(hostListDirectoryRequestSchema.parse({})).toEqual({})
+    expect(hostListDirectoryRequestSchema.parse({ path: '/x' })).toEqual({ path: '/x' })
+    const listing = hostListDirectoryValueSchema.parse({
+      path: '/home/u/p',
+      home: '/home/u',
+      crumbs: [{ name: '/', path: '/', hidden: false }, { name: 'p', path: '/home/u/p', hidden: false }],
+      entries: [{ name: '.dot', path: '/home/u/p/.dot', hidden: true }],
+      truncated: false,
+    })
+    expect(listing.entries[0]?.hidden).toBe(true)
+    // The flag is part of the wire value, not an optional decoration.
+    expect(() => hostListDirectoryValueSchema.parse({ path: '/x', home: '/x', crumbs: [], entries: [] })).toThrow()
+    expect(hostCreateDirectoryRequestSchema.parse({ path: '/x', name: 'new' })).toEqual({ path: '/x', name: 'new' })
+    for (const name of ['', ' ', '.', '..', 'a/b', 'a\\b']) {
+      expect(() => hostCreateDirectoryRequestSchema.parse({ path: '/x', name })).toThrow()
+    }
+    expect(hostCreateDirectoryValueSchema.parse({ path: '/x/new' })).toEqual({ path: '/x/new' })
+  })
 })
 
 describe('workspace domain schemas', () => {
@@ -276,10 +319,13 @@ describe('commands domain schemas', () => {
     expect(() => commandExecuteRequestSchema.parse({ line: '/compact' })).toThrow()
     expect(() => commandExecuteRequestSchema.parse({ sessionId: 's1' })).toThrow()
     expect(commandExecuteValueSchema.parse({ matched: false })).toEqual({ matched: false })
-    const matched = commandExecuteValueSchema.parse({ matched: true, result: { kind: 'success', text: 'done' } })
-    expect(matched.result?.kind).toBe('success')
-    expect(commandExecuteValueSchema.parse({ matched: true, result: { kind: 'error', text: 'bad' } }).result?.kind).toBe('error')
-    expect(() => commandExecuteValueSchema.parse({ matched: true, result: { kind: 'other' } })).toThrow()
+    // Pure admission: matched plus the optional lifecycle pairing id
+    // (outcomes ride the logged lifecycle events, never this response).
+    expect(commandExecuteValueSchema.parse({ matched: true, commandId: 'cmd-1' }))
+      .toEqual({ matched: true, commandId: 'cmd-1' })
+    expect(commandExecuteValueSchema.parse({ matched: true })).toEqual({ matched: true })
+    expect(() => commandExecuteValueSchema.parse({ matched: true, commandId: '' })).toThrow()
+    expect(() => commandExecuteValueSchema.parse({})).toThrow()
   })
 })
 
@@ -299,28 +345,35 @@ describe('skills domain schemas', () => {
   })
 })
 
+describe('goals domain schemas', () => {
+  it('requires at least one replacement field for goal.edit', () => {
+    const ref = { id: 'g1', revision: 1 }
+    expect(goalEditRequestSchema.parse({ sessionId: 's1', ref, objective: 'updated' }).objective).toBe('updated')
+    expect(goalEditRequestSchema.parse({ sessionId: 's1', ref, maxGoalRounds: 3 }).maxGoalRounds).toBe(3)
+    expect(() => goalEditRequestSchema.parse({ sessionId: 's1', ref })).toThrow()
+  })
+})
+
 describe('events frame schemas', () => {
   it('accepts every mux frame branch', () => {
     const frames = [
       { type: 'session/event', sessionId: 's', event: { type: 't', seq: 0, time: 1, data: null } },
       { type: 'session/subscribed', sessionId: 's', lastSeq: -1 },
-      { type: 'session/title', sessionId: 's', title: 'Durable title', eventSeq: 2, updatedAt: 3 },
       { type: 'approval/requested', sessionId: 's', approvalId: 'a', toolName: 'bash', callId: 'c', reason: 'r' },
       { 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: {} } },
     ]
     for (const frame of frames) expect(muxFrameSchema.parse(frame)).toMatchObject({ type: frame.type })
     expect(() => muxFrameSchema.parse({ type: 'unknown/frame' })).toThrow()
     for (const invalid of [
-      { type: 'session/title', sessionId: 's', title: '', eventSeq: 0, updatedAt: 1 },
-      { type: 'session/title', sessionId: 's', title: 'x', eventSeq: -1, updatedAt: 1 },
-      { type: 'session/title', sessionId: 's', title: 'x', eventSeq: 0.5, updatedAt: 1 },
-      { type: 'session/title', sessionId: 's', title: 'x', eventSeq: 0, updatedAt: 'now' },
-      { type: 'session/title', sessionId: 's', title: 'x', eventSeq: 0, updatedAt: Number.NaN },
+      { type: 'session/projection', sessionId: 's', key: '', value: null, seq: 0 },
+      { type: 'session/projection', sessionId: 's', key: 'todos', value: null, seq: -1 },
+      { type: 'session/projection', sessionId: 's', key: 'todos', value: null, seq: 0.5 },
     ]) expect(() => muxFrameSchema.parse(invalid)).toThrow()
     expect(askUserQuestionItemSchema.parse({ id: 'q', question: 'Q?' }).id).toBe('q')
   })
@@ -330,9 +383,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', () => {