Merge remote-tracking branch 'origin/master' into worktree/web-multimodal-image-input

# Conflicts:
#	apps/cli/src/web.ts
#	apps/web/tests/smoke-fixture.e2e.ts
#	docs/architecture.i18n.yaml
#	packages/client/connection/src/client/fixture.ts
#	packages/client/runtime/src/client/sessions/conversation.ts
#	packages/client/runtime/src/client/sessions/session.ts
#	packages/client/ui-conversation/package.json
#	packages/client/ui-conversation/src/client/contract/slots.ts
#	packages/client/ui-conversation/src/client/index.ts
#	packages/client/ui-conversation/src/client/service.ts
#	packages/client/ui-conversation/src/client/skeleton/ConversationRoot.tsx
#	packages/client/ui-conversation/tests/apply-inject.spec.tsx
#	packages/client/ui-conversation/tests/service-orchestration.spec.ts
#	packages/host/runtime/src/api-proxy.ts
#	packages/host/runtime/src/boot.ts
#	packages/host/webserver/tests/webserver.spec.ts
#	pnpm-lock.yaml

packages/host/runtime/src/api-proxy.ts

@@ -1,8 +1,6 @@
 /**
- * Host-side ApiProxy implementation (minimal-first —
- * describe/list/create/history/prompt/cancel and both streams are real,
- * respond is a stub). Signature discipline: unary takes the narrow
- * RpcRequest<P> and echoes request.rpcId on the RpcResponse<T>.
+ * Host-side ApiProxy implementation. Signature discipline: unary takes the
+ * narrow RpcRequest<P> and echoes request.rpcId on the RpcResponse<T>.
  */
 
 import { randomUUID } from 'node:crypto'
@@ -14,9 +12,17 @@ import type { ImageAttachmentRef } from '@deepseek-ai/dsh-attachment-local'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import type { JsonValue, Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
-import type { ApiProxy, HistoryEntry, HostFrame, MuxFrame, PromptContentPart, SessionSummary, ToolEventView } from '@deepseek-ai/dsh-host-apiproxy/api'
+import { foldSessionTitle } from '@deepseek-ai/dsh-session-title'
+import type {
+  ApiProxy, HistoryEntry, HostFrame, MuxFrame, PromptContentPart, QuestionResponsePayload, SessionSummary, ToolEventView,
+} from '@deepseek-ai/dsh-host-apiproxy/api'
+import { questionResponsePayloadSchema } from '@deepseek-ai/dsh-host-apiproxy/api/questions.schema'
 import type { ClientResponse, RpcError, RpcReceipt, RpcRequest, RpcResponse } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
 import { RpcId } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
+import type {
+  AskUserQuestionAnswer, AskUserQuestionItem, AskUserQuestionRequest,
+} from '@deepseek-ai/dsh-user-interaction'
+import { UserInteractionError } from '@deepseek-ai/dsh-user-interaction'
 
 /** Page size when history is called without maxMessages. */
 const DEFAULT_MAX_MESSAGES = 50
@@ -29,7 +35,9 @@ function decodeBase64(data: string): Uint8Array {
     throw new AttachmentError('Image upload is not canonical base64.', 'INVALID_IMAGE_BASE64')
   }
   const decoded = Buffer.from(data, 'base64')
-  if (decoded.toString('base64') !== data) throw new AttachmentError('Image upload is not canonical base64.', 'INVALID_IMAGE_BASE64')
+  if (decoded.toString('base64') !== data) {
+    throw new AttachmentError('Image upload is not canonical base64.', 'INVALID_IMAGE_BASE64')
+  }
   return new Uint8Array(decoded)
 }
 
@@ -169,6 +177,28 @@ function frame<F>(payload: F): RpcRequest<F> {
   return { rpcId: RpcId(randomUUID()), payload }
 }
 
+type SessionTitleFrame = Extract<MuxFrame, { type: 'session/title' }>
+
+/** Project the latest durable title without exposing title-generation policy. */
+function titleFrame(session: Session): SessionTitleFrame | undefined {
+  const title = foldSessionTitle(session.events)
+  if (title === undefined) return undefined
+  return {
+    type: 'session/title',
+    sessionId: session.id,
+    title: title.title,
+    eventSeq: title.eventSeq,
+    updatedAt: title.updatedAt,
+  }
+}
+
+/** Queue the subscription baseline followed by its optional title snapshot. */
+function subscribeSession(queue: FrameQueue<RpcRequest<MuxFrame>>, session: Session): void {
+  queue.push(frame({ type: 'session/subscribed', sessionId: session.id, lastSeq: session.seq - 1 }))
+  const title = titleFrame(session)
+  if (title !== undefined) queue.push(frame(title))
+}
+
 /** SessionSummary projection for attached (in-memory) sessions. */
 function summarize(session: Session, running: boolean): SessionSummary {
   return {
@@ -220,6 +250,35 @@ interface ToolCallData { callId: string; name: string; arguments: string }
 /** The tool/result payload fields the presenter path reads. */
 interface ToolResultData { callId: string; content: ContentBlock[]; isError: boolean; meta?: JsonValue }
 
+/** One host-owned question wait, addressed by the stable server-request id. */
+interface PendingQuestion {
+  rpcId: RpcId
+  sessionId: SessionId
+  questions: AskUserQuestionItem[]
+  resolve: (answer: AskUserQuestionAnswer) => void
+  reject: (error: UserInteractionError) => void
+  signal?: AbortSignal
+  onAbort?: () => void
+}
+
+/** Validate one answer batch against the exact question request it resolves. */
+function matchesQuestions(payload: QuestionResponsePayload, pending: PendingQuestion): boolean {
+  if (payload.sessionId !== pending.sessionId) return false
+  const answers = payload.answer.answers
+  if (answers.length !== pending.questions.length) return false
+  return answers.every((answer, index) => {
+    const question = pending.questions[index] as AskUserQuestionItem
+    if (answer.id !== question.id) return false
+    if (new Set(answer.selected).size !== answer.selected.length) return false
+    const custom = answer.custom?.trim()
+    if (custom !== undefined && custom === '') return false
+    if (custom !== undefined && answer.selected.length > 0) return false
+    if (question.multiSelect !== true && answer.selected.length > 1) return false
+    const labels = new Set(question.options?.map(option => option.label) ?? [])
+    return answer.selected.every(label => labels.has(label))
+  })
+}
+
 /**
  * Compute the render intent for a tool/call or tool/result event through the
  * presenters registered at this moment; every other event type gets none. A
@@ -284,12 +343,70 @@ class SessionNotFound extends Error {}
  * @param ctx - the root context returned by bootHost (sessions/agents services mounted).
  * @param defaults - host-level default provider/model: injected as
  * agentOptions on create/resume, reported by describe from the same source.
- * @returns the ApiProxy implementation (minimal-first; stubs noted per method).
+ * @returns the ApiProxy implementation.
  */
 export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiProxy {
   const agentOptions = { provider: defaults.provider, model: defaults.model }
   /** Implicit resume of cold sessions, deduplicating concurrent calls (follows the jsonrpc sessionCreations precedent). */
   const resumes = new Map<SessionId, Promise<Agent>>()
+  const pendingQuestions = new Map<RpcId, PendingQuestion>()
+  const muxQueues = new Set<FrameQueue<RpcRequest<MuxFrame>>>()
+
+  /** Send one transient frame to every connected mux consumer. */
+  function broadcast(payload: MuxFrame): void {
+    const envelope = frame(payload)
+    for (const queue of muxQueues) queue.push(envelope)
+  }
+
+  /** Remove a wait before settling it: synchronous deletion makes the first claimant win. */
+  function claimQuestion(pending: PendingQuestion, outcome: 'answered' | 'cancelled'): void {
+    pendingQuestions.delete(pending.rpcId)
+    if (pending.signal !== undefined && pending.onAbort !== undefined) {
+      pending.signal.removeEventListener('abort', pending.onAbort)
+    }
+    broadcast({
+      type: 'question/resolved', sessionId: pending.sessionId,
+      questionRpcId: pending.rpcId, outcome,
+    })
+  }
+
+  const disposeProvider = ctx.userInteraction.registerProvider({
+    ask(request: AskUserQuestionRequest): Promise<AskUserQuestionAnswer> {
+      const sessionId = request.agent?.id
+      if (sessionId === undefined) {
+        return Promise.reject(new UserInteractionError(
+          'web user interaction requires an agent-owned session', 'ASK_MISSING_AGENT'))
+      }
+      return new Promise<AskUserQuestionAnswer>((resolve, reject) => {
+        const rpcId = RpcId(randomUUID())
+        const pending: PendingQuestion = {
+          rpcId, sessionId, questions: request.questions, resolve, reject,
+          ...(request.signal === undefined ? {} : { signal: request.signal }),
+        }
+        const onAbort = (): void => {
+          claimQuestion(pending, 'cancelled')
+          reject(new UserInteractionError(
+            'ask_user_question was aborted before the user answered', 'ASK_ABORTED'))
+        }
+        pending.onAbort = onAbort
+        pendingQuestions.set(rpcId, pending)
+        request.signal?.addEventListener('abort', onAbort, { once: true })
+        const envelope: RpcRequest<MuxFrame> = {
+          rpcId,
+          payload: { type: 'question/requested', sessionId, questions: request.questions },
+        }
+        for (const queue of muxQueues) queue.push(envelope)
+      })
+    },
+  })
+  ctx.effect(() => () => {
+    disposeProvider()
+    for (const pending of [...pendingQuestions.values()]) {
+      claimQuestion(pending, 'cancelled')
+      pending.reject(new UserInteractionError(
+        'web user-interaction provider was disposed', 'ASK_ABORTED'))
+    }
+  }, 'api-proxy: user-interaction provider')
 
   /**
    * Gate the cold path on the store: an id absent from it, or naming a legacy
@@ -404,7 +521,11 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
           else agent.send(durable, { source })
         } catch (error: unknown) {
           if (error instanceof AttachmentError) {
-            return err(request, { code: 'attachment-error', message: error.message, details: { reason: error.code } })
+            return err(request, {
+              code: 'attachment-error',
+              message: error.message,
+              details: { reason: error.code },
+            })
           }
           // A synchronous throw from send/steer means disposed or invalid input; surface as agent-busy with the reason attached.
           return err(request, { code: 'agent-busy', message: 'prompt rejected', details: { reason: String(error) } })
@@ -426,12 +547,23 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
         }
         try {
           const stored = await ctx.attachments.readImage(ref)
-          return ok(request, { attachment: stored.ref, data: Buffer.from(stored.data).toString('base64') })
+          return ok(request, {
+            attachment: stored.ref,
+            data: Buffer.from(stored.data).toString('base64'),
+          })
         } catch (error: unknown) {
           if (error instanceof AttachmentError) {
-            return err(request, { code: 'attachment-error', message: error.message, details: { reason: error.code } })
+            return err(request, {
+              code: 'attachment-error',
+              message: error.message,
+              details: { reason: error.code },
+            })
           }
-          return err(request, { code: 'internal', message: 'Unable to read image attachment.', details: {} })
+          return err(request, {
+            code: 'internal',
+            message: 'Unable to read image attachment.',
+            details: {},
+          })
         }
       },
 
@@ -452,7 +584,8 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
 
     host: {
       async describe(request) {
-        const activeModel = (await ctx.llm.listModels(defaults.provider)).find(model => model.id === defaults.model)
+        const activeModel = (await ctx.llm.listModels(defaults.provider))
+          .find(model => model.id === defaults.model)
         // TODO(step2): version should read apps/cli's package.json; placeholder for now.
         return ok(request, {
           version: '0.0.1',
@@ -472,8 +605,18 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
     events: {
       mux(_request, signal) {
         const queue = new FrameQueue<RpcRequest<MuxFrame>>()
+        muxQueues.add(queue)
         for (const session of ctx.sessions.list()) {
-          queue.push(frame({ type: 'session/subscribed', sessionId: session.id, lastSeq: session.seq - 1 }))
+          subscribeSession(queue, session)
+        }
+        for (const pending of pendingQuestions.values()) {
+          queue.push({
+            rpcId: pending.rpcId,
+            payload: {
+              type: 'question/requested', sessionId: pending.sessionId,
+              questions: pending.questions,
+            },
+          })
         }
         // Per-session open-call table for result-view pairing. Bounded by the
         // per-turn call count: entries clear on turn/end; a table miss (stream
@@ -496,15 +639,22 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             const view = viewFor(ctx, event, callId =>
               openCalls.get(session.id)?.get(callId) ?? backscanArgs(session.events, callId))
             queue.push(frame({ type: 'session/event', sessionId: session.id, event, ...view === undefined ? {} : { view } }))
+            if (event.type === 'session/title') {
+              // The accepted raw event is already in session.events, so the fold must find it.
+              queue.push(frame(titleFrame(session) as SessionTitleFrame))
+            }
           }),
           ctx.on('session/created', (session: Session) => {
-            queue.push(frame({ type: 'session/subscribed', sessionId: session.id, lastSeq: session.seq - 1 }))
+            subscribeSession(queue, session)
           }),
           ctx.on('session/disposed', (session: Session) => {
             openCalls.delete(session.id)
           }),
         ]
-        return queue.iterate(signal, () => { for (const dispose of disposers) dispose() })
+        return queue.iterate(signal, () => {
+          muxQueues.delete(queue)
+          for (const dispose of disposers) dispose()
+        })
       },
 
       host(_request, signal) {
@@ -532,9 +682,38 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
       },
     },
 
-    // TODO(step2): approval/question pending registry (wire answerer + proxy provider).
-    respond(_message: ClientResponse): Promise<RpcReceipt> {
-      return Promise.resolve({ accepted: false, reason: 'not-pending' })
+    respond(message: ClientResponse): Promise<RpcReceipt> {
+      const pending = pendingQuestions.get(message.rpcId)
+      if (pending === undefined) return Promise.resolve({ accepted: false, reason: 'not-pending' })
+      if (!message.result.ok) {
+        if (message.result.error.code !== 'cancelled') {
+          return Promise.resolve({ accepted: false, reason: 'bad-response' })
+        }
+        claimQuestion(pending, 'cancelled')
+        pending.reject(new UserInteractionError(
+          'the user cancelled ask_user_question', 'ASK_CANCELLED'))
+        return Promise.resolve({ accepted: true })
+      }
+      const parsed = questionResponsePayloadSchema.safeParse(message.result.value)
+      if (!parsed.success) {
+        return Promise.resolve({ accepted: false, reason: 'bad-response' })
+      }
+      const payload: QuestionResponsePayload = {
+        sessionId: parsed.data.sessionId,
+        answer: {
+          answers: parsed.data.answer.answers.map(answer => ({
+            id: answer.id,
+            selected: answer.selected,
+            ...(answer.custom === undefined ? {} : { custom: answer.custom }),
+          })),
+        },
+      }
+      if (!matchesQuestions(payload, pending)) {
+        return Promise.resolve({ accepted: false, reason: 'bad-response' })
+      }
+      claimQuestion(pending, 'answered')
+      pending.resolve(payload.answer)
+      return Promise.resolve({ accepted: true })
     },
   }
 }