Merge remote-tracking branch 'origin/feat/send-unify' into worktree/agent-message-intents

# Conflicts:
#	.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.i18n.yaml
#	.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md
#	.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.zh.md
#	docs/cordis-catalog/events.md
#	docs/event-producer-consumer.md
#	packages/core/agent-loop/src/agent.ts

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

@@ -298,13 +298,14 @@ export class ReactLoopAgent implements Agent {
   /** The `next-step`/no-wakeup injection path: durable context, no FIFO, no run. */
   private injectContext(input: Extract<ResolvedAgentInput, { target: 'next-step'; wakeup: false }>): void {
     const { content, source, meta } = input
-    const context = {
+    // Detach and validate the payload before any append, so malformed input
+    // cannot open a one-shot turn or otherwise mutate the session.
+    const accepted = this.acceptContext({
       content,
       source,
       ...meta !== undefined ? { meta } : {},
-    }
+    })
     if (isTurnOpen(this.session)) {
-      const accepted = this.acceptContext(context)
       // Provider protocols require every assistant tool-call batch to be
       // followed only by its tool results. Historical interrupted batches do
       // not own new context; only the currently executing batch may defer it.
@@ -316,23 +317,25 @@ export class ReactLoopAgent implements Agent {
       return
     }
     // No turn open: wrap the injection in a one-shot turn so every event stays
-    // turn-enclosed (the durability/replay boundary is the turn).
+    // turn-enclosed (the durability/replay boundary is the turn). The payload is
+    // validated above, but `Session.append` can still reject a turn/start
+    // pre-commit (append re-entrancy from a session/event listener, or an
+    // internal-dispatch veto), so the finally owes a turn/end only when
+    // turn/start actually committed.
     const turn = lastTurnNumber(this.session) + 1
-    // Once turn/start enters the log, a turn/end is owed even if the message
-    // append fails acceptance or pre-commit validation. The finally re-checks
-    // the log and closes only a turn that actually opened; post-commit observers
-    // are contained by Session and cannot create a false append failure.
     try {
       this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
-      this.session.append('user/message', context, { surfaceOp: 'append' })
+      this.session.append('user/message', accepted, { surfaceOp: 'append' })
     } finally {
       // Close the turn if turn/start made it into the log. A pre-commit veto
       // must escape rather than being mistaken for a committed turn/end.
       if (isTurnOpen(this.session)) {
         this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
       }
-      // Decide the durability checkpoint from the log: an accepted one-shot
-      // turn must be flushed even when its message append was the failing step.
+      // Checkpoint only an accepted one-shot turn: a turn/start rejected
+      // pre-commit recorded nothing, so it owes no flush (and a spurious flush
+      // would emit a phantom-turn agent/error). The payload is validated up
+      // front, so a committed turn/start is always followed by its user/message.
       const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
       // Keep inject() synchronous: report checkpoint failures live instead of
       // rejecting the caller, and track the task so disposal still drains it.
@@ -483,8 +486,21 @@ export class ReactLoopAgent implements Agent {
    */
   private [stopDriver](): Promise<void> | void {
     if (this._status !== 'disposed') {
+      // Snapshot any still-pending inbox items, then CLEAR and mark disposed
+      // BEFORE emitting the discard — mirroring cancel()'s snapshot→clear→emit
+      // order so a re-entrant send()/cancel() from a discard listener throws
+      // `disposed` (or finds an empty inbox) instead of leaking or double-
+      // discarding an id. `send()` emits enqueue unconditionally, so the discard
+      // is unconditional too (even on an unpublished rollback) to keep every
+      // enqueued id matched.
+      const discarded = this.#inbox.pending()
+      this.#inbox.clear()
       this._status = 'disposed'
       this.resolveDisposed()
+      if (discarded.length > 0) {
+        const items = discarded.map(({ message, steering }) => agentMessage(message, steering))
+        agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
+      }
       // Release whenIdle waiters BEFORE the (guarded) event emit — they are
       // internal state that must settle even if a listener throws below. Each
       // waiter chains `done`, so it resolves only once the loop actually exits.

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

@@ -29,7 +29,14 @@ export interface InboxMessage {
  * @returns the live-event message for enqueue/dequeue/discard.
  */
 export function agentMessage(message: InboxMessage, steering: boolean): AgentMessage {
-  return { id: message.id, content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup }
+  // Frozen: the fused emitter passes this exact object to every listener in
+  // turn, so one listener must not be able to mutate a field (`id`, `steering`,
+  // `content`, …) a later listener then observes. `message` is already a frozen
+  // inbox record, so its nested fields need no re-clone.
+  return Object.freeze({
+    id: message.id, content: message.content, source: message.source,
+    contexts: message.contexts, steering, wakeup: message.wakeup,
+  })
 }
 
 /**

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

@@ -132,6 +132,27 @@ describe('Agent', () => {
     expect(() => { agent.send([{ type: 'text', text: 'too late' }]) }).toThrow('disposed')
   })
 
+  it('disposal discards still-pending inbox items so every id gets a terminal event', async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+    let agent!: Agent
+    const discarded: string[] = []
+    ctx.on('agent/inbox/discard', (subject, messages) => {
+      if (subject === agent) discarded.push(...messages.map(m => m.id))
+    })
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      agent = inner.agentLoop.create(SessionId('scoped'), { provider: 'mock', model: 'mock' })
+    }, { inject: ['agentLoop'] }))
+
+    // A quiet (non-waking) item stays parked in the inbox; disposal must drop it
+    // WITH a discard so its enqueued id is not left dangling forever.
+    const id = agent.queue([{ type: 'text', text: 'never runs' }])
+    await fiber.dispose()
+    await driverDone(agent)
+
+    expect(discarded).toEqual([id])
+  })
+
   it('steer() throws after disposal', async () => {
     const adapter = new MockAdapter(['hang'])
     const ctx = await harness(adapter)
@@ -211,24 +232,54 @@ describe('Agent', () => {
     warn.mockRestore()
   })
 
-  it('idle inject() closes its one-shot turn AND still checkpoints even if the append throws', async () => {
+  it('idle inject() validates its payload BEFORE opening a turn, so invalid input appends nothing', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
     let flushes = 0
     ctx.on('session/flush', () => { flushes += 1 })
 
-    // Non-serializable injected content makes Session.append throw AFTER
-    // turn/start was recorded. The turn/end must still be appended (finally),
-    // AND the durability checkpoint must still fire — the balanced turn is in
-    // memory and a crash before the next turn/dispose would otherwise lose it.
+    // Non-serializable injected content is rejected by the up-front snapshot
+    // BEFORE any append (the unified send contract: invalid input throws before
+    // mutating the log). No one-shot turn opens and no durability checkpoint fires.
     expect(() => {
       agent.inject([{ type: 'text', text: 'x', bad: 1n } as never], { source: { kind: 'plugin', plugin: 'p' } })
-    }).toThrow(/non-JSON-serializable/)
-    const types = agent.session.events.map(e => e.type)
-    expect(types).toEqual(['turn/start', 'turn/end']) // balanced, no open turn
-    await new Promise(r => setTimeout(r, 10)) // let the fire-and-forget flush run
-    expect(flushes).toBe(1) // checkpoint fired despite the throw
+    }).toThrow(/losslessly JSON-serializable/)
+    expect(agent.session.events).toHaveLength(0)
+    await new Promise(r => setTimeout(r, 10)) // give any (erroneous) flush a chance
+    expect(flushes).toBe(0) // nothing was appended, so no checkpoint
+  })
+
+  it('idle inject() re-entered from a session/event listener is rejected pre-commit and opens no turn', async () => {
+    const adapter = new MockAdapter([textResponse('ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+    // Injecting from inside a session/event listener re-enters Session.append,
+    // which rejects pre-commit — so turn/start never commits. The finally sees
+    // no open turn (closes nothing) and no recorded turn (no checkpoint), and
+    // the reentrant throw is contained by Session's post-commit dispatch.
+    // Fire on turn/end: at that instant the outer one-shot turn is closed (no
+    // turn open), so the reentrant inject takes the idle one-shot-turn path and
+    // its turn/start append re-enters Session and is rejected pre-commit.
+    let reentered = false
+    ctx.on('session/event', (_s, event) => {
+      if (!reentered && event.type === 'turn/end') {
+        reentered = true
+        agent.inject([{ type: 'text', text: 'reentrant' }], { source: { kind: 'plugin', plugin: 'p' } })
+      }
+    })
+
+    agent.inject([{ type: 'text', text: 'outer' }], { source: { kind: 'plugin', plugin: 'p' } })
+    // The outer injection's own one-shot turn is balanced; the reentrant one
+    // opened no turn (its turn/start was rejected pre-commit).
+    const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
+    expect(turnStarts).toHaveLength(1)
+    const injected = agent.session.events.filter(e => e.type === 'user/message')
+    expect(injected).toHaveLength(1) // the reentrant user/message never committed
+    await new Promise(r => setTimeout(r, 10))
+    expect(flushes).toBe(1) // only the outer accepted turn checkpointed
   })
 
   it('idle inject() still checkpoints when a listener throws on the synthetic turn/end', async () => {
@@ -278,13 +329,11 @@ describe('Agent', () => {
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
 
-    // A non-serializable source makes the turn/start append throw BEFORE the
-    // event is pushed (Session.append validates before push), so NO turn opens.
-    // The finally's isTurnOpen() guard sees no open turn and appends nothing —
-    // the log stays empty, not left with a dangling turn/start.
+    // A non-serializable source is rejected by the up-front snapshot BEFORE any
+    // append, so NO turn opens and the log stays empty.
     expect(() => {
       agent.inject([{ type: 'text', text: 'x' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
-    }).toThrow(/non-JSON-serializable/)
+    }).toThrow(/losslessly JSON-serializable/)
     expect(agent.session.events).toHaveLength(0)
   })
 

packages/core/agent-loop/tests/inbox.spec.ts

@@ -1,11 +1,20 @@
 import { describe, expect, it } from 'vitest'
 import { AgentMessageId } from '@deepseek-ai/dsh-agent'
-import { Inbox } from '../src/inbox.ts'
+import { Inbox, agentMessage } from '../src/inbox.ts'
 
 function message(text: string) {
   return { id: AgentMessageId(text), content: [{ type: 'text' as const, text }], source: { kind: 'user' as const }, contexts: [], wakeup: true }
 }
 
+describe('agentMessage', () => {
+  it('returns a frozen payload so a listener cannot mutate it for later listeners', () => {
+    const payload = agentMessage(message('m'), false)
+    expect(Object.isFrozen(payload)).toBe(true)
+    expect(() => { (payload as { id: string }).id = 'mutated' }).toThrow()
+    expect(payload.id).toBe(AgentMessageId('m'))
+  })
+})
+
 function resolverPair() {
   let r!: () => void
   const p = new Promise<void>((resolve) => { r = resolve })

packages/core/agent/src/types.ts

@@ -325,11 +325,15 @@ declare module 'cordis' {
      */
     'agent/inbox/dequeue'(this: Scoped<Agent>, agent: Agent, message: AgentMessage): void
     /**
-     * `cancel()` (without `keepInbox`) dropped pending inbox items without
-     * delivering them. Fires once per effective clearing call with every
-     * discarded item, after `agent/cancel-requested` and before the abort.
-     * @param agent - the agent whose inbox was cleared.
-     * @param messages - the discarded messages in FIFO order (queued then steering); empty when nothing was pending.
+     * Pending inbox items were dropped without delivering them, so every
+     * enqueued id receives exactly one terminal `agent/inbox/dequeue` OR
+     * `agent/inbox/discard`. Emitters: `cancel()` without `keepInbox` (after
+     * `agent/cancel-requested`, before the abort); a terminal `agent/turn-stop`
+     * dropping pending steering (in-turn and on the post-turn late-steering
+     * drain); and disposal of any still-pending items (before
+     * `agent/status('disposed')`). Fires once per drop with every dropped item.
+     * @param agent - the agent whose inbox items were dropped.
+     * @param messages - the discarded messages in FIFO order (queued then steering); never empty.
      * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
      * @mode emit
      */