fix(agent): address initial review findings

docs/persistence-catalog.md

@@ -87,7 +87,11 @@ Sources: [`packages/core/session/src/types.ts:258`](../packages/core/session/src
 #### `agent/inbox/spliced` — log-only
 
 ```ts persistence-catalog
-/** One normalized mutation of an agent's durable pending-message lists. */
+/**
+ * One normalized mutation of an agent's durable pending-message lists.
+ * Live dispatch precedes projection mutation, so synchronous observers may
+ * read the pre-splice inbox to recover the removed messages.
+ */
 'agent/inbox/spliced': {
   target: InboxTarget
   start: number
@@ -97,7 +101,7 @@ Sources: [`packages/core/session/src/types.ts:258`](../packages/core/session/src
 }
 ```
 
-Source: [`packages/core/agent/src/types.ts:268`](../packages/core/agent/src/types.ts)
+Source: [`packages/core/agent/src/types.ts:272`](../packages/core/agent/src/types.ts)
 
 ### `approval/*`
 

packages/core/agent-loop/src/agent.ts

@@ -144,10 +144,10 @@ export class ReactLoopAgent implements Agent {
   }
 
   private async admit(onTurnBoundary: boolean): Promise<Admission> {
-    if (this.phase.kind !== 'running') throw new Error()
+    if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": admit outside running phase`)
     const signal = this.phase.abort.signal
     const claimed = [...this.inbox.nextStep]
-    const outboxLength = this.inbox.nextStep.length
+    const outboxLength = claimed.length
     const queued = onTurnBoundary ? this.inbox.nextTurn[0] : undefined
     if (queued !== undefined) claimed.push(queued)
     if (claimed.length === 0) return { kind: 'empty' }
@@ -167,7 +167,7 @@ export class ReactLoopAgent implements Agent {
 
   /** Admitted input stays unowned until `turn/start` commits. */
   private async turn(): Promise<boolean> {
-    if (this.phase.kind === 'idle') throw new Error()
+    if (this.phase.kind === 'idle') throw new Error(`agent "${this.id}": turn without driver reservation`)
     const abort = this.phase.kind === 'collecting' ? this.phase.abort : new AbortController()
     const { signal } = abort
     const lastTurn = this.phase.kind === 'collecting' ? this.phase.lastTurn : this.phase.turn
@@ -227,7 +227,7 @@ export class ReactLoopAgent implements Agent {
   }
 
   private async step(): Promise<TurnEndReason | null> {
-    if (this.phase.kind !== 'running') throw new Error()
+    if (this.phase.kind !== 'running') throw new Error(`agent "${this.id}": step outside running phase`)
     const { turn, step, abort: { signal } } = this.phase
     signal.throwIfAborted()
     await this.loopCtx.serial(agentCarrier(this), 'agent/step', this, turn, step, signal)

packages/core/agent/src/inbox.ts

@@ -44,6 +44,9 @@ export class Inbox {
 
   /**
    * Apply standard splice semantics and durably record the normalized result.
+   * The durable event commits before the live projection mutates, so synchronous
+   * `session/event` observers see the pre-splice lists and can reconstruct the
+   * removed messages from the normalized coordinates.
    * @param target - pending list to mutate.
    * @param start - splice position.
    * @param deleteCount - maximum number of messages to remove.
@@ -59,12 +62,14 @@ export class Inbox {
     outcome?: 'admitted' | 'canceled',
   ): UserMessage[] {
     const inbox = this.state[target]
-    const offset = Math.trunc(start) || 0
+    const truncatedStart = Math.trunc(start)
+    const offset = Number.isNaN(truncatedStart) ? 0 : truncatedStart
     const actualStart = offset < 0
       ? Math.max(inbox.length + offset, 0)
       : Math.min(offset, inbox.length)
+    const truncatedDeleteCount = Math.trunc(deleteCount)
     const actualDeleteCount = Math.min(
-      Math.max(Math.trunc(deleteCount) || 0, 0),
+      Math.max(Number.isNaN(truncatedDeleteCount) ? 0 : truncatedDeleteCount, 0),
       inbox.length - actualStart,
     )
     if (actualDeleteCount === 0 && inserted.length === 0) return []

packages/core/agent/src/types.ts

@@ -264,7 +264,11 @@ declare module 'cordis' {
 
 declare module '@deepseek-ai/dsh-session' {
   interface SessionEventMap {
-    /** One normalized mutation of an agent's durable pending-message lists. */
+    /**
+     * One normalized mutation of an agent's durable pending-message lists.
+     * Live dispatch precedes projection mutation, so synchronous observers may
+     * read the pre-splice inbox to recover the removed messages.
+     */
     'agent/inbox/spliced': {
       target: InboxTarget
       start: number

packages/goal/goal-session/src/index.ts

@@ -418,6 +418,7 @@ export function apply(ctx: Context): void {
           attempt.stale = true
           if (attempt.phase === 'admitted' && state.agent.status === 'running') {
             state.agent.cancel({ kind: 'parent' })
+            waits.push(state.agent.whenIdle())
           }
         }
         if (state.run !== undefined) waits.push(state.run)

packages/goal/goal-session/tests/goal-session.spec.ts

@@ -735,7 +735,7 @@ describe('same-session goal driving', () => {
       activation: 'disarmed',
       roundsStarted: 1,
     })
-    await test.agent.whenIdle()
+    expect(test.agent.status).toBe('idle')
     expect(test.adapter.requests).toHaveLength(1)
   })
 

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

@@ -1502,6 +1502,20 @@ 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 === 'agent/inbox/spliced' && event.data.target === 'next-turn') {
+              const agent = ctx.agents.get(session.id)
+              if (agent?.session === session) {
+                queue.push(frame({
+                  type: 'session/queue',
+                  sessionId: session.id,
+                  items: agent.inbox.nextTurn.toSpliced(
+                    event.data.start,
+                    event.data.removedCount ?? 0,
+                    ...event.data.inserted,
+                  ),
+                }))
+              }
+            }
           }),
           ctx.on('session/created', (session: Session) => {
             subscribeSession(queue, session)

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

@@ -62,9 +62,9 @@ export type MuxFrame =
   | { type: 'question/requested'; sessionId: SessionId; questions: AskUserQuestionItem[] }
   | { type: 'question/resolved'; sessionId: SessionId; questionRpcId: RpcId; outcome: 'answered' | 'cancelled' }
   /**
-   * 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.
+   * Complete next-turn queue snapshot emitted when a mux stream opens and
+   * after every live next-turn mutation. Pending next-step input is outside
+   * this Web queue projection.
    */
   | { type: 'session/queue'; sessionId: SessionId; items: UserMessage[] }
   /**

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

@@ -20,7 +20,7 @@ import ToolRegistry from '@deepseek-ai/dsh-tools'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import CommandService from '@deepseek-ai/dsh-commands'
 import SkillService from '@deepseek-ai/dsh-skill'
-import type { HostFrame, MuxFrame } from '../src/api/index.ts'
+import type { HostFrame } from '../src/api/index.ts'
 import type { RpcRequest, RpcResponse } from '../src/api/rpc.ts'
 import { RpcId } from '../src/api/rpc.ts'
 import { createApiProxy } from '../src/api-proxy.ts'
@@ -85,6 +85,13 @@ async function collect<F>(iterable: AsyncIterable<RpcRequest<F>>, count: number,
   return frames
 }
 
+/** Read the next payload from an open stream. */
+async function nextFrame<F>(iterator: AsyncIterator<RpcRequest<F>>): Promise<F> {
+  const result = await iterator.next()
+  if (result.done) throw new Error('stream ended')
+  return result.value.payload
+}
+
 describe('command.list', () => {
   it('serves the addressed agent\'s name-sorted catalog', async () => {
     const ctx = await harness()
@@ -332,24 +339,41 @@ describe('session.updateQueue', () => {
 })
 
 describe('session/queue frames', () => {
-  it('publishes the durable next-turn baseline without duplicating message identity', async () => {
+  it('publishes authoritative next-turn snapshots without duplicating message identity', async () => {
     const ctx = await harness()
     const api = createApiProxy(ctx, DEFAULTS)
     const agent = stubAgent(ctx)
     const queued = inboxMessage('m-1', 'queued prompt')
+    const edited = inboxMessage('m-1', 'edited prompt')
     const steering = inboxMessage('m-2', 'steering prompt')
     agent.inbox.splice('next-turn', 0, 0, [queued])
     agent.inbox.splice('next-step', 0, 0, [steering])
 
     const abort = new AbortController()
-    const frames = await collect<MuxFrame>(
-      api.events.mux({ rpcId: RpcId('t-mux-baseline'), payload: {} }, abort.signal), 2, abort)
+    const iterator = api.events.mux({
+      rpcId: RpcId('t-mux-baseline'),
+      payload: {},
+    }, abort.signal)[Symbol.asyncIterator]()
+    const frames = [
+      await nextFrame(iterator),
+      await nextFrame(iterator),
+    ]
+    agent.inbox.splice('next-turn', 0, 1, [edited])
+    frames.push(await nextFrame(iterator), await nextFrame(iterator))
+    abort.abort()
+    await iterator.return?.()
+
     expect(frames.filter(frame => frame.type === 'session/queue')).toEqual([
       {
         type: 'session/queue',
         sessionId: agent.id,
         items: [queued],
       },
+      {
+        type: 'session/queue',
+        sessionId: agent.id,
+        items: [edited],
+      },
     ])
   })
 })

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

@@ -10,7 +10,7 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import { z } from 'zod'
-import AgentRegistry from '@deepseek-ai/dsh-agent'
+import AgentRegistry, { Inbox } 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'
@@ -53,9 +53,8 @@ async function harness(withRegistry: boolean): Promise<{ ctx: Context; session:
   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)
+  // The gateway reads both the session and durable inbox baseline.
+  ctx.agents.register({ id: session.id, session, inbox: new Inbox(session), status: 'idle', ctx } as Agent)
   return { ctx, session }
 }