refactor(agent): expose mutable inbox state

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

@@ -9,9 +9,9 @@ import { join } from 'node:path'
 import type { Context } from 'cordis'
 import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent'
 import type {
-  Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxItem, InboxItemId,
+  Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus,
 } from '@deepseek-ai/dsh-agent'
-import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
+import { createUserMessage, freezeMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
 import { errorChain } from '@deepseek-ai/dsh-llm'
 import type { MessageSource } from '@deepseek-ai/dsh-llm'
 import type { Session, SessionEvent, SessionHeader, SessionId, UserMessage } from '@deepseek-ai/dsh-session'
@@ -507,112 +507,6 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
     })
   })
 
-  /**
-   * Per-session queued-occurrence mirror serving the mux-open queue snapshot
-   * (the same refresh-recovery baseline as pending questions). Each terminal
-   * queue event retires one matching occurrence, so repeated sends of the same
-   * identified message remain visible until every occurrence is claimed.
-   */
-  const queuedMirror = new Map<SessionId, InboxItem[]>()
-  type UnseenQueueEvent =
-    | { readonly kind: 'update'; readonly item: InboxItem }
-    | { readonly kind: 'terminal' }
-  const unseenQueueEvents = new Map<SessionId, Map<InboxItemId, UnseenQueueEvent>>()
-  const rememberUnseen = (sessionId: SessionId, itemId: InboxItemId, event: UnseenQueueEvent): void => {
-    let events = unseenQueueEvents.get(sessionId)
-    if (events === undefined) {
-      events = new Map()
-      unseenQueueEvents.set(sessionId, events)
-    }
-    events.set(itemId, event)
-    // Only synchronous re-entrancy may deliver a mutation before its outer
-    // enqueue observer. Drop unmatched protocol-invalid observations instead
-    // of retaining process-local ids indefinitely.
-    queueMicrotask(() => {
-      const current = unseenQueueEvents.get(sessionId)
-      if (current?.get(itemId) !== event) return
-      current.delete(itemId)
-      if (current.size === 0) unseenQueueEvents.delete(sessionId)
-    })
-  }
-  const takeUnseen = (sessionId: SessionId, itemId: InboxItemId): UnseenQueueEvent | undefined => {
-    const events = unseenQueueEvents.get(sessionId)
-    const event = events?.get(itemId)
-    if (event === undefined) return undefined
-    events?.delete(itemId)
-    if (events?.size === 0) unseenQueueEvents.delete(sessionId)
-    return event
-  }
-  const publishQueue = (sessionId: SessionId): void => {
-    const items = queuedMirror.get(sessionId) ?? []
-    broadcast({
-      type: 'session/queue',
-      sessionId,
-      items: items.map(item => ({
-        id: item.id,
-        message: item.message,
-      })),
-    })
-  }
-  ctx.effect(() => {
-    const retire = (agent: Agent, item: InboxItem): boolean => {
-      const entries = queuedMirror.get(agent.id)
-      if (entries === undefined) {
-        rememberUnseen(agent.id, item.id, { kind: 'terminal' })
-        return false
-      }
-      const index = entries.findIndex(entry => entry.id === item.id)
-      if (index === -1) {
-        rememberUnseen(agent.id, item.id, { kind: 'terminal' })
-        return false
-      }
-      entries.splice(index, 1)
-      if (entries.length === 0) queuedMirror.delete(agent.id)
-      return true
-    }
-    const disposers = [
-      ctx.on('agent/inbox/enqueue', (agent: Agent, item: InboxItem) => {
-        if (item.placement !== 'queued') return
-        const unseen = takeUnseen(agent.id, item.id)
-        if (unseen?.kind === 'terminal') return
-        let entries = queuedMirror.get(agent.id)
-        if (entries === undefined) {
-          entries = []
-          queuedMirror.set(agent.id, entries)
-        }
-        entries.push(unseen?.kind === 'update' ? unseen.item : item)
-        publishQueue(agent.id)
-      }),
-      ctx.on('agent/inbox/update', (agent: Agent, item: InboxItem) => {
-        const entries = queuedMirror.get(agent.id)
-        if (entries === undefined) {
-          rememberUnseen(agent.id, item.id, { kind: 'update', item })
-          return
-        }
-        const index = entries.findIndex(entry => entry.id === item.id)
-        if (index === -1) {
-          rememberUnseen(agent.id, item.id, { kind: 'update', item })
-          return
-        }
-        entries.splice(index, 1, item)
-        publishQueue(agent.id)
-      }),
-      ctx.on('agent/inbox/dequeue', (agent: Agent, item: InboxItem) => {
-        if (retire(agent, item)) publishQueue(agent.id)
-      }),
-      ctx.on('agent/inbox/discard', (agent: Agent, items: InboxItem[]) => {
-        let changed = false
-        for (const item of items) changed = retire(agent, item) || changed
-        if (changed) publishQueue(agent.id)
-      }),
-      ctx.on('session/disposed', (session: Session) => {
-        queuedMirror.delete(session.id)
-        unseenQueueEvents.delete(session.id)
-      }),
-    ]
-    return () => { for (const dispose of disposers) dispose() }
-  }, 'api-proxy: queued mirror')
-
   /** 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)
@@ -1162,13 +1056,21 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
       updateQueue(request) {
         const { sessionId, itemId, action } = request.payload
         const agent = ctx.agents.get(sessionId)
-        if (agent === undefined || agent.updateInbox(itemId, action) === 'not-found') {
+        const queued = agent?.inbox.nextTurn
+        const index = queued?.findIndex(message => message.id === itemId) ?? -1
+        const message = queued?.[index]
+        if (agent === undefined || message === undefined) {
           return Promise.resolve(err(request, {
             code: 'queue-item-not-found',
             message: 'queued item is no longer pending',
             details: { itemId },
           }))
         }
+        if (action.kind === 'edit') {
+          agent.inbox.splice('next-turn', index, 1, [freezeMessage({ ...message, content: action.content })])
+        } else {
+          agent.inbox.splice('next-turn', index, 1, [])
+        }
         return Promise.resolve(ok(request, { accepted: true as const }))
       },
 
@@ -1567,14 +1469,13 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
         // Queue snapshot baseline (pendingQuestions precedent): frames replayed
         // in arrival order per session; a reconnecting client rebuilds its
         // queue view from these alone.
-        for (const [sessionId, items] of queuedMirror) {
+        for (const agent of ctx.agents.list()) {
+          const items = agent.inbox.nextTurn
+          if (items.length === 0) continue
           queue.push(frame({
             type: 'session/queue',
-            sessionId,
-            items: items.map(item => ({
-              id: item.id,
-              message: item.message,
-            })),
+            sessionId: agent.id,
+            items: [...items],
           }))
         }
         // Per-session open-call table for result-view pairing. Bounded by the

packages/host/apiproxy/src/api/events.schema.ts

@@ -11,7 +11,7 @@ import type { Wire } from './rpc.schema.ts'
 import { rpcErrorSchema, rpcIdSchema } from './rpc.schema.ts'
 import { approvalRequestIdSchema } from './approvals.schema.ts'
 import {
-  contentBlockSchema, inboxItemIdSchema, sessionEventSchema, sessionIdSchema, toolEventViewSchema,
+  contentBlockSchema, messageIdSchema, sessionEventSchema, sessionIdSchema, toolEventViewSchema,
 } from './sessions.schema.ts'
 import { workspaceIdSchema, workspaceViewSchema } from './workspace.schema.ts'
 
@@ -25,10 +25,10 @@ export const askUserQuestionItemSchema = z.object({
   multiSelect: z.boolean().optional(),
 }) satisfies z.ZodType<Wire<AskUserQuestionItem>>
 
-/** Unified message envelope carried by transient queue frames. */
-const messageSchema = z.object({
-  id: z.string().min(1),
-  role: z.union([z.literal('system'), z.literal('user'), z.literal('assistant')]),
+/** User-message envelope carried by queue baselines. */
+const userMessageSchema = z.object({
+  id: messageIdSchema,
+  role: z.literal('user'),
   content: z.array(contentBlockSchema),
   source: z.looseObject({ kind: z.string() }),
 })
@@ -47,10 +47,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
   z.object({
     type: z.literal('session/queue'),
     sessionId: sessionIdSchema,
-    items: z.array(z.object({
-      id: inboxItemIdSchema,
-      message: messageSchema,
-    })),
+    items: z.array(userMessageSchema),
   }),
   // value stays wide: it already passed its unit's own schema on the host,
   // and deep-validating here would import every domain's schema into the carrier.

packages/host/apiproxy/src/api/events.ts

@@ -8,9 +8,8 @@
 
 import type { AskUserQuestionItem } from '@deepseek-ai/dsh-user-interaction/types'
 import type { ApprovalOutcome, ApprovalRequestId } from '@deepseek-ai/dsh-user-approval/types'
-import type { Message } from '@deepseek-ai/dsh-llm/types'
-import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand'
 import type { CallId } from '@deepseek-ai/dsh-llm/brand'
+import type { UserMessage } from '@deepseek-ai/dsh-llm/message'
 import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
 import type { ToolCallView, ToolResultView } from '@deepseek-ai/dsh-tools/presentation'
 import type { RpcError, RpcId, RpcRequest } from './rpc.ts'
@@ -32,14 +31,6 @@ export type ToolEventView =
   | { for: 'call'; view: ToolCallView }
   | { for: 'result'; view: ToolResultView }
 
-/** One pending queued occurrence in an authoritative queue snapshot. */
-export interface QueuedInboxItem {
-  /** Agent-owned occurrence identity used by queue mutations. */
-  id: InboxItemId
-  /** Complete pending message; it is not durable until the Agent claims it. */
-  message: Message
-}
-
 /** Streaming face of the contract: the two SSE stream openers (mux + host). */
 export interface EventsApi {
   /**
@@ -71,13 +62,11 @@ export type MuxFrame =
   | { type: 'question/requested'; sessionId: SessionId; questions: AskUserQuestionItem[] }
   | { type: 'question/resolved'; sessionId: SessionId; questionRpcId: RpcId; outcome: 'answered' | 'cancelled' }
   /**
-   * Complete transient queue state after every enqueue, mutation, claim, or
-   * discard. Pending work is not model-visible and therefore has no durable
-   * session event; the whole snapshot makes edit, deletion, cancel, and
-   * reconnect converge through one authoritative signal. Pending steering is
-   * outside this Web queue projection.
+   * Complete next-turn queue baseline emitted when a mux stream opens. Live
+   * mutations arrive through durable `agent/inbox/spliced` session events.
+   * Pending next-step input is outside this Web queue projection.
    */
-  | { type: 'session/queue'; sessionId: SessionId; items: QueuedInboxItem[] }
+  | { type: 'session/queue'; sessionId: SessionId; items: UserMessage[] }
   /**
    * One projection unit's finished value changed (session-projection RFC).
    * Live push state, never logged — replay recomputes on the host (the

packages/host/apiproxy/src/api/index.ts

@@ -35,7 +35,7 @@ export type { DirectoryEntry, DirectoryListing, HostApi } from './host.ts'
 export type { WorkspaceApi, WorkspaceId, WorkspaceView } from './workspace.ts'
 export type { CommandsApi, CommandDescriptor } from './commands.ts'
 export type { SkillsApi, SkillEntry } from './skills.ts'
-export type { EventsApi, MuxFrame, HostFrame, QueuedInboxItem, ToolCallView, ToolEventView, ToolResultView } from './events.ts'
+export type { EventsApi, MuxFrame, HostFrame, ToolCallView, ToolEventView, ToolResultView } from './events.ts'
 export type { GoalsApi, GoalId, GoalRef } from './goals.ts'
 export type { ApprovalResponsePayload } from './approvals.ts'
 export type { QuestionResponsePayload } from './questions.ts'
@@ -56,7 +56,5 @@ export type {
 // ---- Errors and ids ----
 export { RpcId, transportError } from './rpc.ts'
 export type { RpcError, RpcErrorCode, RpcErrorDetailsMap, RpcResult } from './rpc.ts'
-export type { InboxItemId } from '@deepseek-ai/dsh-agent/brand'
-
 // ---- Method registry and derived generics ----
 export type { RequestPayload, ResponseValue, RpcMethodMap } from './rpc-map.ts'

packages/host/apiproxy/src/api/rpc.ts

@@ -8,8 +8,8 @@
 import type { z as zCore } from 'zod'
 type ZodIssue = zCore.core.$ZodIssue
 import type { Branded } from '@deepseek-ai/dsh-brand'
+import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { SessionId } from '@deepseek-ai/dsh-session/types'
-import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand'
 
 /**
  * Message correlation id: the initiator mints it on a request; a response
@@ -45,7 +45,7 @@ export interface RpcErrorDetailsMap {
   'directory-create-failed': { path: string }
   'directory-picker-unavailable': { capability: string }
   'agent-busy': { reason: string }
-  'queue-item-not-found': { itemId: InboxItemId }
+  'queue-item-not-found': { itemId: MessageId }
   /** A known slash command reported a usage/state error; the message is the command's own text. */
   'command-error': {}
   /** A leading-/ prompt named no registered command; the message names the token. */

packages/host/apiproxy/src/api/sessions.schema.ts

@@ -7,7 +7,7 @@
 
 import { z } from 'zod'
 import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
-import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand'
+import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { RequestPayload, ResponseValue } from './rpc-map.ts'
 import type { Wire } from './rpc.schema.ts'
 import type {
@@ -20,8 +20,8 @@ import type { WorkspaceId } from './workspace.ts'
 /** SessionId: one brand cast after shape validation (the only cast point in this domain). */
 export const sessionIdSchema = z.string().min(1) as unknown as z.ZodType<SessionId>
 
-/** InboxItemId: one brand cast after non-empty string validation. */
-export const inboxItemIdSchema = z.string().min(1) as unknown as z.ZodType<InboxItemId>
+/** MessageId: one brand cast after non-empty string validation. */
+export const messageIdSchema = z.string().min(1) as unknown as z.ZodType<MessageId>
 
 /**
  * WorkspaceId: the workspace domain's one brand cast. Hosted here rather
@@ -221,7 +221,7 @@ export const sessionPromptValueSchema = z.object({
 /** session.updateQueue request payload. */
 export const sessionUpdateQueueRequestSchema = z.object({
   sessionId: sessionIdSchema,
-  itemId: inboxItemIdSchema,
+  itemId: messageIdSchema,
   action: z.discriminatedUnion('kind', [
     z.object({ kind: z.literal('edit'), content: z.array(contentBlockSchema) }),
     z.object({ kind: z.literal('remove') }),

packages/host/apiproxy/src/api/sessions.ts

@@ -4,8 +4,8 @@
  * else references RequestPayload<'session.*'> / ResponseValue<'session.*'>.
  */
 
+import type { MessageId } from '@deepseek-ai/dsh-llm/brand'
 import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
-import type { InboxItemId } from '@deepseek-ai/dsh-agent/brand'
 import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session/types'
 // The pure-type outlet: api/ is browser-importable, and the package root's
 // cordis Context merge (via dsh-agent) must not enter client aggregates.
@@ -238,7 +238,7 @@ export interface SessionsApi {
   /**
    * Edits or removes one pending queued occurrence.
    */
-  updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: InboxItemId; action: QueueAction }>):
+  updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: MessageId; action: QueueAction }>):
   Promise<RpcResponse<{ accepted: true }>>
 
   /** Stops: clears both FIFOs + aborts the current step (1:1 with agent.cancel). */