fix(agent-loop): route next-step input during admission

2026-07-27 17:55:55 +08:00
parent 52174e32cb
commit a59ce0367c
18 changed files with 207 additions and 85 deletions
--- a/packages/host/apiproxy/src/api-proxy.ts
+++ b/packages/host/apiproxy/src/api-proxy.ts
@@ -380,7 +380,7 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
   * `agent/inbox/dequeue` OR `agent/inbox/discard` (the inbox contract), so
   * the mirror needs no consumption heuristics or sweeps beyond disposal.
   */
-  const queuedMirror = new Map<SessionId, Map<AgentMessageId, AgentMessage>>()
+  const queuedMirror = new Map<SessionId, Map<AgentMessageId, { message: AgentMessage; steering: boolean }>>()
  ctx.effect(() => {
    const retire = (agent: Agent, id: AgentMessageId): void => {
      const entries = queuedMirror.get(agent.id)
@@ -389,11 +389,21 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
      if (entries.size === 0) queuedMirror.delete(agent.id)
    }
    const disposers = [
-      ctx.on('agent/inbox/enqueue', (agent: Agent, message: AgentMessage) => {
+      ctx.on('agent/inbox/enqueue', (agent: Agent, message: AgentMessage, placement) => {
        let entries = queuedMirror.get(agent.id)
-        if (entries === undefined) queuedMirror.set(agent.id, entries = new Map<AgentMessageId, AgentMessage>())
-        entries.set(message.id, message)
-        broadcast({ type: 'session/queued', sessionId: agent.id, content: message.content, source: message.source })
+        if (entries === undefined) {
+          entries = new Map<AgentMessageId, { message: AgentMessage; steering: boolean }>()
+          queuedMirror.set(agent.id, entries)
+        }
+        const steering = placement === 'steering'
+        entries.set(message.id, { message, steering })
+        broadcast({
+          type: 'session/queued',
+          sessionId: agent.id,
+          content: message.content,
+          source: message.source,
+          steering,
+        })
      }),
      ctx.on('agent/inbox/dequeue', (agent: Agent, message: AgentMessage) => {
        retire(agent, message.id)
@@ -929,7 +939,13 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
        // queue view from these alone.
        for (const [sessionId, entries] of queuedMirror) {
          for (const entry of entries.values()) {
-            queue.push(frame({ type: 'session/queued', sessionId, content: entry.content, source: entry.source }))
+            queue.push(frame({
+              type: 'session/queued',
+              sessionId,
+              content: entry.message.content,
+              source: entry.message.source,
+              steering: entry.steering,
+            }))
          }
        }
        // Per-session open-call table for result-view pairing. Bounded by the
--- a/packages/host/apiproxy/src/api/events.schema.ts
+++ b/packages/host/apiproxy/src/api/events.schema.ts
@@ -36,7 +36,7 @@ export const muxFrameSchema = z.discriminatedUnion('type', [
  z.object({ type: z.literal('question/requested'), sessionId: sessionIdSchema, questions: z.array(askUserQuestionItemSchema).min(1) }),
  z.object({ type: z.literal('question/resolved'), sessionId: sessionIdSchema, questionRpcId: rpcIdSchema, outcome: z.union([z.literal('answered'), z.literal('cancelled')]) }),
  // content/source reuse the wide passthroughs (both are merge-extensible in core).
-  z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }) }),
+  z.object({ type: z.literal('session/queued'), sessionId: sessionIdSchema, content: z.array(contentBlockSchema), source: z.looseObject({ kind: z.string() }), steering: z.boolean() }),
  z.object({ type: z.literal('stream/error'), error: rpcErrorSchema }),
 ]) as unknown as z.ZodType<MuxFrame>

--- a/packages/host/apiproxy/src/api/events.ts
+++ b/packages/host/apiproxy/src/api/events.ts
@@ -69,10 +69,12 @@ export type MuxFrame =
   * host replays the current queue snapshot for every attached session (same
   * refresh-recovery baseline as pending questions); queue clearing on cancel
   * has no dedicated frame — clients fold it from the status flip.
-   * source carries the prompt's rpcId when the message came over this wire
-   * (the client's provisional-echo reconciliation key).
+   * `steering` is the host's acceptance-time queue classification and remains
+   * authoritative in reconnect snapshots. `source` carries the prompt's rpcId
+   * when the message came over this wire (the client's provisional-echo
+   * reconciliation key).
   */
-  | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource }
+  | { type: 'session/queued'; sessionId: SessionId; content: ContentBlock[]; source: MessageSource; steering: boolean }
  | { type: 'stream/error'; error: RpcError }

 /**
--- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts
+++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts
@@ -258,20 +258,20 @@ describe('session/queued frames', () => {

    const queued = inboxMessage('m-1', 'queued prompt')
    const steering = inboxMessage('m-2', 'queued prompt')
-    ctx.emit('agent/inbox/enqueue', agent, queued)
-    ctx.emit('agent/inbox/enqueue', agent, steering)
+    ctx.emit('agent/inbox/enqueue', agent, queued, 'queued')
+    ctx.emit('agent/inbox/enqueue', agent, steering, 'steering')

    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' } },
-      { type: 'session/queued', sessionId: agent.id, content: steering.content, source: { kind: 'user' } },
+      { 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 },
    ])

    // A fresh mux connection replays the still-pending entries as its baseline.
    const replay = new AbortController()
    const replayFrames = await collect<MuxFrame>(
      api.events.mux({ rpcId: RpcId('t-mux-replay'), payload: {} }, replay.signal), 3, replay)
-    expect(replayFrames.filter(f => f.type === 'session/queued')).toHaveLength(2)
+    expect(replayFrames.filter(f => f.type === 'session/queued')).toEqual(liveFrames)
  })

  it('retires mirror entries on their terminal dequeue', async () => {
@@ -280,8 +280,8 @@ describe('session/queued frames', () => {
    const agent = stubAgent(ctx)
    const queued = inboxMessage('m-3', 'x')
    const steering = inboxMessage('m-4', 'x', 'r-1')
-    ctx.emit('agent/inbox/enqueue', agent, queued)
-    ctx.emit('agent/inbox/enqueue', agent, steering)
+    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)

@@ -297,8 +297,8 @@ describe('session/queued frames', () => {
    const agent = stubAgent(ctx)
    const doomed = inboxMessage('m-5', 'doomed')
    const survivor = inboxMessage('m-6', 'survivor')
-    ctx.emit('agent/inbox/enqueue', agent, doomed)
-    ctx.emit('agent/inbox/enqueue', agent, survivor)
+    ctx.emit('agent/inbox/enqueue', agent, doomed, 'queued')
+    ctx.emit('agent/inbox/enqueue', agent, survivor, 'queued')
    ctx.emit('agent/inbox/discard', agent, [doomed])

    const abort = new AbortController()
--- a/packages/host/apiproxy/tests/rpc-schemas.spec.ts
+++ b/packages/host/apiproxy/tests/rpc-schemas.spec.ts
@@ -248,8 +248,8 @@ describe('events frame schemas', () => {
      { 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' } },
-      { type: 'session/queued', sessionId: 's', content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' } },
+      { 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: 'stream/error', error: { code: 'internal', message: 'm', details: {} } },
    ]
    for (const frame of frames) expect(muxFrameSchema.parse(frame)).toMatchObject({ type: frame.type })
@@ -270,7 +270,8 @@ 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: {} })).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()
  })

  it('accepts every host frame branch', () => {