Merge master into fix/subagent-stack-end-result

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

@@ -776,10 +776,11 @@ 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.
+   * Per-session pending-occurrence mirror serving live and mux-open
+   * `session/queue` snapshots. It carries both queued and steering placements.
+   * Each terminal inbox event retires one matching occurrence, so repeated
+   * sends of the same identified message remain visible until every occurrence
+   * is claimed.
    */
   const queuedMirror = new Map<SessionId, InboxItem[]>()
   type UnseenQueueEvent =
@@ -818,29 +819,28 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
       sessionId,
       items: items.map(item => ({
         id: item.id,
+        placement: item.placement,
         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
-      }
+    const retireKnown = (sessionId: SessionId, itemId: InboxItemId): boolean => {
+      const entries = queuedMirror.get(sessionId)
+      if (entries === undefined) return false
+      const index = entries.findIndex(entry => entry.id === itemId)
+      if (index === -1) return false
       entries.splice(index, 1)
-      if (entries.length === 0) queuedMirror.delete(agent.id)
+      if (entries.length === 0) queuedMirror.delete(sessionId)
       return true
     }
+    const retire = (agent: Agent, item: InboxItem): boolean => {
+      if (retireKnown(agent.id, item.id)) return true
+      rememberUnseen(agent.id, item.id, { kind: 'terminal' })
+      return false
+    }
     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)
@@ -866,7 +866,24 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
         publishQueue(agent.id)
       }),
       ctx.on('agent/inbox/dequeue', (agent: Agent, item: InboxItem) => {
-        if (retire(agent, item)) publishQueue(agent.id)
+        if (item.placement === 'steering') {
+          // AgentLoop appends the durable steering/message synchronously after
+          // this claim. Retain and retire the mirror row in the following
+          // microtask so any re-entrant snapshot and the Host's linear mux
+          // stream keep it visible until the durable event exists.
+          const present = queuedMirror.get(agent.id)?.some(entry => entry.id === item.id) === true
+          if (!present) {
+            retire(agent, item)
+            return
+          }
+          queueMicrotask(() => {
+            if (retireKnown(agent.id, item.id)) publishQueue(agent.id)
+          })
+        } else if (retire(agent, item)) {
+          // Queued claims have no durable same-message handoff to order.
+          // Publish retirement synchronously as before.
+          publishQueue(agent.id)
+        }
       }),
       ctx.on('agent/inbox/discard', (agent: Agent, items: InboxItem[]) => {
         let changed = false
@@ -1882,13 +1899,28 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
         if (agent !== undefined && hasSubagentOwner(agent.session, agent)) {
           return Promise.resolve(err(request, subagentOwnershipError(sessionId)))
         }
-        if (agent === undefined || agent.updateInbox(itemId, action) === 'not-found') {
+        if (agent === undefined) {
           return Promise.resolve(err(request, {
             code: 'queue-item-not-found',
             message: 'queued item is no longer pending',
             details: { itemId },
           }))
         }
+        const result = agent.updateInbox(itemId, action)
+        if (result === 'not-found') {
+          return Promise.resolve(err(request, {
+            code: 'queue-item-not-found',
+            message: 'queued item is no longer pending',
+            details: { itemId },
+          }))
+        }
+        if (result === 'steer-unavailable') {
+          return Promise.resolve(err(request, {
+            code: 'steer-unavailable',
+            message: 'current turn no longer accepts steering',
+            details: { itemId },
+          }))
+        }
         return Promise.resolve(ok(request, { accepted: true as const }))
       },
 
@@ -2531,6 +2563,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
             sessionId,
             items: items.map(item => ({
               id: item.id,
+              placement: item.placement,
               message: item.message,
             })),
           }))

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

@@ -54,6 +54,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
     sessionId: sessionIdSchema,
     items: z.array(z.object({
       id: inboxItemIdSchema,
+      placement: z.union([z.literal('queued'), z.literal('steering')]),
       message: messageSchema,
     })),
   }),

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

@@ -32,10 +32,12 @@ export type ToolEventView =
   | { for: 'call'; view: ToolCallView }
   | { for: 'result'; view: ToolResultView }
 
-/** One pending queued occurrence in an authoritative queue snapshot. */
+/** One pending inbox occurrence in the authoritative `session/queue` snapshot. */
 export interface QueuedInboxItem {
-  /** Agent-owned occurrence identity used by queue mutations. */
+  /** Agent-owned occurrence identity; queue mutations address only `queued` items. */
   id: InboxItemId
+  /** Agent-resolved FIFO placement; clients render queued and steering items on different surfaces. */
+  placement: 'queued' | 'steering'
   /** Complete pending message; it is not durable until the Agent claims it. */
   message: Message
 }
@@ -71,11 +73,12 @@ 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
+   * Complete transient inbox 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.
+   * reconnect converge through one authoritative signal. `session/queue`
+   * covers both resolved placements: queued items render
+   * in QueueDock, while pending steering renders at the conversation tail.
    */
   | { type: 'session/queue'; sessionId: SessionId; items: QueuedInboxItem[] }
   /**

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

@@ -48,6 +48,7 @@ export const rpcErrorSchema: z.ZodType<RpcError> = z.discriminatedUnion('code',
   z.object({ code: z.literal('directory-picker-unavailable'), message: z.string(), details: z.object({ capability: z.string() }) }),
   z.object({ code: z.literal('agent-busy'), message: z.string(), details: z.object({ reason: z.string() }) }),
   z.object({ code: z.literal('queue-item-not-found'), message: z.string(), details: z.object({ itemId: z.string() }) }),
+  z.object({ code: z.literal('steer-unavailable'), message: z.string(), details: z.object({ itemId: z.string() }) }),
   z.object({ code: z.literal('command-error'), message: z.string(), details: z.object({}) }),
   z.object({ code: z.literal('unknown-command'), message: z.string(), details: z.object({}) }),
   z.object({ code: z.literal('settings-rejected'), message: z.string(), details: z.object({ ns: z.string() }) }),

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

@@ -46,6 +46,7 @@ export interface RpcErrorDetailsMap {
   'directory-picker-unavailable': { capability: string }
   'agent-busy': { reason: string }
   'queue-item-not-found': { itemId: InboxItemId }
+  'steer-unavailable': { itemId: InboxItemId }
   /** 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

@@ -269,6 +269,7 @@ export const sessionUpdateQueueRequestSchema = z.object({
   action: z.discriminatedUnion('kind', [
     z.object({ kind: z.literal('edit'), content: z.array(contentBlockSchema) }),
     z.object({ kind: z.literal('remove') }),
+    z.object({ kind: z.literal('steer') }),
   ]),
 }) as unknown as z.ZodType<RequestPayload<'session.updateQueue'>>
 

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

@@ -129,6 +129,7 @@ export interface SessionModels {
 export type QueueAction =
   | { kind: 'edit'; content: ContentBlock[] }
   | { kind: 'remove' }
+  | { kind: 'steer' }
 
 /** Session list entry (v1 builds no index: list does readdir+stat). */
 export interface SessionSummary {
@@ -285,7 +286,7 @@ export interface SessionsApi {
   Promise<RpcResponse<{ accepted: true; command?: { kind: 'success'; text?: string } }>>
 
   /**
-   * Edits or removes one pending queued occurrence on an ordinary session.
+   * Edits, removes, or strictly steers one pending queued occurrence on an ordinary session.
    * Session-backed subagents reject with `agent-busy`.
    */
   updateQueue(request: RpcRequest<{ sessionId: SessionId; itemId: InboxItemId; action: QueueAction }>):