fix(agent-loop): own queued message input

packages/core/agent-loop/README.md

@@ -44,7 +44,7 @@ Agents listed in config are auto-created at startup. `cwd` applies only to fresh
 
 - `ReactLoopAgent` — the concrete `Agent` implementation. Its inbox is a JavaScript native-private field, and one prepared session can be claimed by only one concrete driver. Everything observable happens through session events and the `agent/*` event taxonomy.
 
-`Inbox`, `runLoop`, and the instance-bound enable/start controls are package-internal. The package root does not export them, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than constructing or starting the driver internals.
+`Inbox`, `runLoop`, and the instance-bound enable/start controls are package-internal. The package root does not export them, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than constructing or starting the driver internals. `ReactLoopAgent.send()` and running `steer()` materialize content plus resolved source once as detached, deeply frozen lossless JSON, then share that accepted record between `agent/queued` and the inbox; malformed data throws before either boundary.
 
 ### Loop lifecycle (`loop.ts`)
 

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

@@ -12,8 +12,8 @@ import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { deepFreeze } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
-import type { Session } from '@deepseek-ai/dsh-session'
-import { Inbox } from './inbox.ts'
+import { snapshotJsonValue, type Session } from '@deepseek-ai/dsh-session'
+import { Inbox, type InboxMessage } from './inbox.ts'
 import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
 
 /** Agents whose rollback-covered publication enabled driving. */
@@ -213,6 +213,26 @@ export class ReactLoopAgent implements Agent {
     return options?.source ?? { kind: 'user' }
   }
 
+  /**
+   * Accept one public send/steer payload as the exact detached record shared by
+   * the live notification and inbox. Lossless-JSON materialization reads every
+   * nested field once; deep freeze prevents an observer from rewriting queued
+   * work before the loop drains it.
+   */
+  private acceptInboxMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
+    const source = this.resolveSource(options)
+    const accepted = snapshotJsonValue({ content, source })
+    if (accepted === undefined) {
+      throw new TypeError('agent message content and source must be losslessly JSON-serializable')
+    }
+    return deepFreeze(accepted)
+  }
+
+  /** Reject a driving operation once teardown has synchronously closed the agent. */
+  private assertNotDisposed(): void {
+    if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
+  }
+
   /** Reject every driving verb while creation setup still owns the agent. */
   private assertDriveEnabled(action: string): void {
     if (driveEnabledAgents.has(this)) return
@@ -221,24 +241,29 @@ export class ReactLoopAgent implements Agent {
 
   send(content: ContentBlock[], options?: SendOptions): void {
     this.assertDriveEnabled('send')
-    if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
-    const source = this.resolveSource(options)
-    this.#inbox.enqueue({ content, source })
-    agentEvents(this.loopCtx, this).emit('agent/queued', content, { source, steering: false })
+    this.assertNotDisposed()
+    const accepted = this.acceptInboxMessage(content, options)
+    // Materialization invokes caller getters, which may reenter handle disposal.
+    this.assertNotDisposed()
+    this.#inbox.enqueue(accepted)
+    const info = deepFreeze({ source: accepted.source, steering: false })
+    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
   }
 
   steer(content: ContentBlock[], options?: SendOptions): void {
     this.assertDriveEnabled('steer')
-    if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
+    this.assertNotDisposed()
     if (this._status !== 'running') { this.send(content, options); return }
-    const source = this.resolveSource(options)
-    this.#inbox.steer({ content, source })
-    agentEvents(this.loopCtx, this).emit('agent/queued', content, { source, steering: true })
+    const accepted = this.acceptInboxMessage(content, options)
+    this.assertNotDisposed()
+    this.#inbox.steer(accepted)
+    const info = deepFreeze({ source: accepted.source, steering: true })
+    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
   }
 
   inject(content: ContentBlock[], options?: SendOptions): void {
     this.assertDriveEnabled('inject')
-    if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
+    this.assertNotDisposed()
     const source = this.resolveSource(options)
     if (isTurnOpen(this.session)) {
       // A turn is open in the LOG (decided from the log, not agent status —

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

@@ -291,11 +291,15 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
       // the previous turn/end), where the persistence backend drops it as a
       // crash tail (the turn-enclosure RFC). Report via agent/error + the logger only; the
       // driver survives and moves on.
+      /* v8 ignore start -- defensive internal-corruption backstop: public
+       * send/steer input is accepted as lossless JSON before enqueue, and
+       * runTurn contains every failure after turn/start. */
       const err = toError(error)
       ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
       try {
         events.emit('agent/error', turn, 0, err)
       } catch { /* contained: a throwing agent/error listener must not kill the driver */ }
+      /* v8 ignore stop */
     }
 
     // Reset the cancel marker UNCONDITIONALLY here, after the turn returns and
@@ -730,9 +734,10 @@ async function runTurn(
     // `closeStep()` IS idempotent (guarded by `stepOpen`) — it may have run
     // already in a step branch, so running it again is a safe no-op. Absent
     // turn/start means the append threw BEFORE its push (a non-serializable
-    // trigger — impossible for our fixed trigger); nothing was opened, so rethrow
+    // trigger outside the public lossless-JSON boundary); nothing was opened, so rethrow
     // to the runLoop backstop.
     const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
+    /* v8 ignore next -- defensive internal-corruption path; public inbox input is lossless JSON */
     if (!turnStartLogged) throw error
     closeStep()
     // Choose the close reason. Disposal wins only if no error was already

packages/core/agent-loop/tests/coverage-edges.spec.ts

@@ -35,31 +35,24 @@ function send(agent: ReactLoopAgent, text: string) {
   agent.send([{ type: 'text', text }])
 }
 
-describe('turn boundary listener throws (handled in-turn, loop survives)', () => {
-  it('a pre-push turn/start failure (non-serializable source) is rethrown to the runLoop backstop', async () => {
-    // A non-serializable message source makes the turn/start append throw BEFORE
-    // the event is pushed (Session.append validates before push), so turn/start
-    // never enters the log. runTurn sees no logged turn/start and rethrows; the
-    // runLoop backstop reports via agent/error (step 0) + the logger and the
-    // driver survives. This is the ONLY path that reaches the backstop.
-    const adapter = new MockAdapter([textResponse('turn 2')])
+describe('inbox acceptance', () => {
+  it('rejects non-serializable content or source synchronously before notification or enqueue', async () => {
+    const adapter = new MockAdapter([textResponse('turn 1')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
+    let queued = 0
+    ctx.on('agent/queued', () => { queued += 1 })
 
-    const errors: { turn: number; step: number; message: string }[] = []
-    ctx.on('agent/error', (_a, turn, step, error) => void errors.push({ turn, step, message: error.message }))
+    expect(() => {
+      agent.send([{ type: 'text', text: 'first', bad: 1n } as never])
+    }).toThrow(/losslessly JSON-serializable/)
+    expect(() => {
+      agent.send([{ type: 'text', text: 'first' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
+    }).toThrow(/losslessly JSON-serializable/)
+    expect(queued).toBe(0)
+    expect(agent.session.events).toHaveLength(0)
 
-    // A non-serializable source (BigInt) on the queued message.
-    agent.send([{ type: 'text', text: 'first' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
-    await waitForIdle(ctx, agent)
-
-    expect(errors).toHaveLength(1)
-    expect(errors[0]!.step).toBe(0)
-    expect(errors[0]!.message).toMatch(/non-JSON-serializable/)
-    // No turn boundary was written (the turn/start append threw before push).
-    expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
-
-    // loop survives: a well-formed second turn runs normally.
+    // The rejected value never woke or poisoned the loop; a valid message runs.
     send(agent, 'second')
     await waitForIdle(ctx, agent)
     expect(adapter.requests).toHaveLength(1)

packages/core/agent-loop/tests/review-fixes.spec.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
-import LlmService, { CallId, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
+import LlmService, { CallId, ContentBlock, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
@@ -437,6 +437,127 @@ describe('MEDIUM: misc registry and config fixes', () => {
     const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.source] : [])
     expect(steeringSources).toEqual([{ kind: 'plugin', plugin: 'goal' }])
   })
+
+  it('send() owns content and source before notification and delivery', async () => {
+    const adapter = new MockAdapter([textResponse('done')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('owned-send'), { model: 'mock' })
+    const content = [{ type: 'text' as const, text: 'accepted-send' }]
+    const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
+    let notifiedContent: ContentBlock[] | undefined
+    let notifiedSource: MessageSource | undefined
+    let notifiedInfoFrozen = false
+    ctx.on('agent/queued', (subject, acceptedContent, info) => {
+      if (subject !== agent || info.steering) return
+      // Retain the exact notification references: cloning here would test the
+      // listener's copy rather than the event/inbox ownership boundary.
+      notifiedContent = acceptedContent
+      notifiedSource = info.source
+      notifiedInfoFrozen = Object.isFrozen(info)
+    })
+
+    agent.send(content, { source })
+    content[0]!.text = 'caller-mutated-send'
+    source.plugin = 'caller-mutated-source'
+    await waitForIdle(ctx, agent)
+
+    expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-send' }])
+    expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
+    expect(Object.isFrozen(notifiedContent)).toBe(true)
+    expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
+    expect(Object.isFrozen(notifiedSource)).toBe(true)
+    expect(notifiedInfoFrozen).toBe(true)
+    const recorded = agent.session.events.flatMap(event => event.type === 'user/message' ? [event.data] : [])
+    expect(recorded).toContainEqual({
+      content: [{ type: 'text', text: 'accepted-send' }],
+      source: { kind: 'plugin', plugin: 'accepted-source' },
+    })
+    const request = JSON.stringify(adapter.requests[0]!.messages)
+    expect(request).toContain('accepted-send')
+    expect(request).not.toContain('caller-mutated-send')
+  })
+
+  it('send() rechecks disposal after materializing caller getters', async () => {
+    const adapter = new MockAdapter([textResponse('unused')])
+    const ctx = await harness(adapter)
+    const handle = await ctx.agents.create({
+      agentId: AgentId('reentrant-send-dispose'),
+      sessionId: SessionId('reentrant-send-dispose-session'),
+      agentOptions: { model: 'mock' },
+    })
+    const { agent } = handle
+    let queued = 0
+    ctx.on('agent/queued', subject => void (queued += Number(subject === agent)))
+    const content = [{
+      type: 'text' as const,
+      get text() {
+        void handle.dispose()
+        return 'accepted-after-dispose'
+      },
+    }]
+
+    expect(() => { agent.send(content) }).toThrow(/agent "reentrant-send-dispose" is disposed/)
+    await handle.dispose()
+
+    expect(queued).toBe(0)
+    expect(agent.session.events).toHaveLength(0)
+    expect(adapter.requests).toHaveLength(0)
+  })
+
+  it('running steer() owns content and source before notification and delivery', async () => {
+    const adapter = new MockAdapter([toolCallResponse('c1', 'gate', {}), textResponse('done')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(AgentId('owned-steer'), { model: 'mock' })
+    const entered = Promise.withResolvers<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    ctx.tools.register(defineTool({
+      name: 'gate',
+      description: '',
+      parameters: {},
+      async execute() {
+        entered.resolve(undefined)
+        await release.promise
+        return [{ type: 'text', text: 'tool done' }]
+      },
+    }))
+    let notifiedContent: ContentBlock[] | undefined
+    let notifiedSource: MessageSource | undefined
+    let notifiedInfoFrozen = false
+    ctx.on('agent/queued', (subject, acceptedContent, info) => {
+      if (subject !== agent || !info.steering) return
+      notifiedContent = acceptedContent
+      notifiedSource = info.source
+      notifiedInfoFrozen = Object.isFrozen(info)
+    })
+
+    agent.send([{ type: 'text', text: 'start' }])
+    await entered.promise
+    expect(agent.status).toBe('running')
+    const content = [{ type: 'text' as const, text: 'accepted-steer' }]
+    const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
+    agent.steer(content, { source })
+    content[0]!.text = 'caller-mutated-steer'
+    source.plugin = 'caller-mutated-source'
+    const idle = waitForIdle(ctx, agent)
+    release.resolve(undefined)
+    await idle
+
+    expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-steer' }])
+    expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
+    expect(Object.isFrozen(notifiedContent)).toBe(true)
+    expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
+    expect(Object.isFrozen(notifiedSource)).toBe(true)
+    expect(notifiedInfoFrozen).toBe(true)
+    const recorded = agent.session.events.flatMap(event => event.type === 'steering/message' ? [event.data] : [])
+    expect(recorded).toContainEqual({
+      turn: 1,
+      content: [{ type: 'text', text: 'accepted-steer' }],
+      source: { kind: 'plugin', plugin: 'accepted-source' },
+    })
+    const request = JSON.stringify(adapter.requests[1]!.messages)
+    expect(request).toContain('accepted-steer')
+    expect(request).not.toContain('caller-mutated-steer')
+  })
 })
 
 describe('MEDIUM: turn numbering continues across seeded (forked) sessions', () => {