feat(agent): unify send(target × wakeup), coalesce context/message into user/message

Replace send/steer/inject with one Agent.send primitive over the (target × wakeup) matrix; followup/steer/inject become fixed-preset alias methods on the now-abstract Agent class. Coalesce context/message into user/message (injected context is a non-user source). Replace agent/queued with agent/inbox/enqueue/dequeue/discard, add cancel keepInbox, and add a FIFO-conservation invariant.
2026-07-23 19:15:45 +08:00
parent 7c0c516f60
commit 44fd93fd06
117 changed files with 1249 additions and 728 deletions
--- a/packages/core/agent-loop/README.md
+++ b/packages/core/agent-loop/README.md
@@ -52,7 +52,7 @@ Configured agents start automatically. A model call requires both `provider` and

 The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.

-Each concrete `send()` materializes content, resolved source, and attached contexts once as a detached, deeply frozen lossless-JSON FIFO item. If claimed, it is the sole ordinary message in its turn; its contexts are the prompt waterfall's default additional contexts and therefore materialize only after admission. Absent or `separate` placement appends an independent `context/message`; `prompt-prefix` placement bakes the context, the stable `## My request:` delimiter, and effective request into one `user/message`, whose model-hidden envelope retains display content and context descriptors. The waterfall's returned allow is authoritative, so a listener wrapping `next()` preserves downstream `content` and `additionalContexts` unless it intentionally replaces them. A successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, a prompt block, or a pre-start failure may drop its contexts with the message. Running `steer()` enters the same record shape in the steering FIFO without dispatching `agent/prompt-submit`; its next checkpoint applies the same separate-or-prefix placement to `steering/message`, while policy can still stop before another step. Steering left after turn close and its checkpoint becomes later queued input with contexts intact unless terminal turn policy, cancellation, or disposal discards it. Open-turn `inject()` uses the same accepted-value boundary but defers in a FIFO while the current step executes assistant tool calls; successful batches place it after all results, and interrupted batches drain it before turn close. Malformed data throws before enqueue or append.
+The unified `send()` primitive materializes content, resolved source, and attached contexts once as a detached, deeply frozen lossless-JSON record, then routes it by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO (waking the driver unless `wakeup: false`); if claimed, it is the sole ordinary message in its turn, and its contexts are the prompt waterfall's default additional contexts that materialize only after admission. Absent or `separate` placement appends an independent injected `user/message`; `prompt-prefix` placement bakes the context, the stable `## My request:` delimiter, and effective request into one `user/message`, whose model-hidden envelope retains display content and context descriptors. The waterfall's returned allow is authoritative, so a listener wrapping `next()` preserves downstream `content` and `additionalContexts` unless it intentionally replaces them. A successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, a prompt block, or a pre-start failure may drop its contexts with the message. A running `next-step`/wakeup `steer()` enters the same record shape in the steering FIFO without dispatching `agent/prompt-submit`; its next checkpoint applies the same separate-or-prefix placement to `steering/message`, while policy can still stop before another step. Steering left after turn close and its checkpoint becomes later queued input with contexts intact unless terminal turn policy, cancellation, or disposal discards it. `next-step`/no-wakeup `inject()` bypasses the FIFOs and appends durable context directly: an open-turn injection uses the same accepted-value boundary but defers in a FIFO while the current step executes assistant tool calls (successful batches place it after all results, interrupted batches drain it before turn close), and an idle injection wraps a one-shot `injection` turn. Every FIFO enqueue publishes `agent/inbox/enqueue`; the driver's claims publish `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`. Malformed data throws before enqueue or append.

 ### Loop lifecycle (`loop.ts`)

--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -8,8 +8,8 @@

 import type { Context } from 'cordis'
 import { agentEvents } from '@deepseek-ai/dsh-agent'
-import type { AgentCancelCause, AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
-import type { Agent } from '@deepseek-ai/dsh-agent'
+import { Agent } from '@deepseek-ai/dsh-agent'
+import type { AgentCancelCause, AgentOptions, AgentStatus, CancelOptions, HookContext, InboxItemInfo, SendOptions } from '@deepseek-ai/dsh-agent'
 import { deepFreeze, errorChain } from '@deepseek-ai/dsh-llm'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import { snapshotJsonValue, type Session, type SessionId } from '@deepseek-ai/dsh-session'
@@ -100,7 +100,7 @@ export function bindReactLoopAgentContext(agent: ReactLoopAgent, ctx: Context):
 * the loop driver. Everything observable happens through session events and
 * the agent/* event taxonomy — plugins never need this class.
 */
-export class ReactLoopAgent implements Agent {
+export class ReactLoopAgent extends Agent {
  /** Queued + steering FIFOs; native-private so callers cannot bypass the public driving verbs. */
  readonly #inbox = new Inbox()

@@ -161,6 +161,7 @@ export class ReactLoopAgent implements Agent {
    public readonly session: Session,
    maxParallelToolCalls: number,
  ) {
+    super()
    this.maxParallelToolCalls = maxParallelToolCalls
    const { promise, resolve } = Promise.withResolvers<void>()
    this.disposed = promise
@@ -190,25 +191,25 @@ export class ReactLoopAgent implements Agent {
    for (const resolve of waiters) resolve()
  }

-  private resolveSource(options?: SendOptions): MessageSource {
-    return options?.source ?? { kind: 'user' }
-  }
-
  /**
   * Accept one public message payload as a detached record. Lossless-JSON
   * materialization reads every nested field once; deep freeze prevents later
   * caller mutation before an inbox or deferred-injection queue drains it.
   */
-  private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
-    const source = this.resolveSource(options)
+  private acceptMessage(content: ContentBlock[], source: MessageSource, wakeup: boolean, options?: SendOptions): InboxMessage {
    const contexts = options?.contexts ?? []
-    const accepted = snapshotJsonValue({ content, source, contexts })
+    const accepted = snapshotJsonValue({ content, source, contexts, wakeup })
    if (accepted === undefined) {
      throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
    }
    return deepFreeze(accepted)
  }

+  /** Build the `agent/inbox/*` payload for one accepted item. */
+  private inboxInfo(message: InboxMessage, steering: boolean): InboxItemInfo {
+    return { content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup }
+  }
+
  /** Detach one context before it can outlive its caller in the active-batch FIFO. */
  private acceptContext(context: HookContext): HookContext {
    const accepted = snapshotJsonValue(context)
@@ -225,24 +226,26 @@ export class ReactLoopAgent implements Agent {

  send(content: ContentBlock[], options?: SendOptions): void {
    this.assertNotDisposed()
-    const accepted = this.acceptMessage(content, options)
-    this.#inbox.enqueue(accepted)
-    const info = { source: accepted.source, contexts: accepted.contexts, steering: false } as const
-    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
+    const target = options?.target ?? 'next-turn'
+    const wakeup = options?.wakeup ?? true
+    // next-step/no-wakeup is injection: durable context without running the model.
+    if (target === 'next-step' && !wakeup) { this.injectContext(content, options); return }
+    // next-step/wakeup is steering into the running turn; idle falls back to a
+    // woken follow-up turn (there is no active turn to attach to).
+    const steering = target === 'next-step' && this._status === 'running'
+    const source = options?.source ?? { kind: 'user' }
+    const accepted = this.acceptMessage(content, source, wakeup, options)
+    if (steering) {
+      this.#inbox.steer(accepted)
+    } else {
+      this.#inbox.enqueue(accepted, wakeup)
+    }
+    agentEvents(this.loopCtx, this).emit('agent/inbox/enqueue', this.inboxInfo(accepted, steering))
  }

-  steer(content: ContentBlock[], options?: SendOptions): void {
-    this.assertNotDisposed()
-    if (this._status !== 'running') { this.send(content, options); return }
-    const accepted = this.acceptMessage(content, options)
-    this.#inbox.steer(accepted)
-    const info = { source: accepted.source, contexts: accepted.contexts, steering: true } as const
-    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
-  }
-
-  inject(content: ContentBlock[], options?: InjectOptions): void {
-    this.assertNotDisposed()
-    const source = this.resolveSource(options)
+  /** The `next-step`/no-wakeup injection path: durable context, no FIFO, no run. */
+  private injectContext(content: ContentBlock[], options?: SendOptions): void {
+    const source = options?.source ?? { kind: 'plugin', plugin: '' }
    const context = {
      content,
      source,
@@ -257,7 +260,7 @@ export class ReactLoopAgent implements Agent {
        this.deferredInjections.push(accepted)
        return
      }
-      this.session.append('context/message', accepted, { surfaceOp: 'append' })
+      this.session.append('user/message', accepted, { surfaceOp: 'append' })
      return
    }
    // No turn open: wrap the injection in a one-shot turn so every event stays
@@ -269,7 +272,7 @@ export class ReactLoopAgent implements Agent {
    // 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('context/message', context, { surfaceOp: 'append' })
+      this.session.append('user/message', context, { 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.
@@ -301,7 +304,7 @@ export class ReactLoopAgent implements Agent {
  private drainDeferredInjections(): void {
    const pending = this.deferredInjections.splice(0)
    for (const accepted of pending) {
-      this.session.append('context/message', accepted, { surfaceOp: 'append' })
+      this.session.append('user/message', accepted, { surfaceOp: 'append' })
    }
  }

@@ -325,10 +328,14 @@ export class ReactLoopAgent implements Agent {
    }
  }

-  cancel(cause?: AgentCancelCause): void {
+  cancel(cause?: AgentCancelCause, options?: CancelOptions): void {
    const resolvedCause = cause ?? { kind: 'user' }
+    const keepInbox = options?.keepInbox ?? false
    const cancellation = this.turnCancellation
-    const preRun = cancellation === undefined && (this.#inbox.hasQueued || this.#inbox.hasSteering)
+    // keepInbox preserves pending work, so un-started items must not arm the
+    // pre-run cancel path that would otherwise drop the next queued turn.
+    const preRun = !keepInbox && cancellation === undefined
+      && (this.#inbox.hasQueued || this.#inbox.hasSteering)
    if (cancellation !== undefined || preRun) {
      if (preRun) this.preRunCancelled = true
      // Coordination consumers must update their own state before this call
@@ -336,9 +343,18 @@ export class ReactLoopAgent implements Agent {
      // contained by the fused dispatcher and cannot veto cancellation.
      agentEvents(this.loopCtx, this).emit('agent/cancel-requested', resolvedCause)
    }
-    // Clear work already present before abort observers run. A replacement
-    // synchronously enqueued by an observer belongs to the next turn.
-    this.#inbox.clear()
+    if (!keepInbox) {
+      // Snapshot before clearing so the discard notification carries the exact
+      // dropped items; a replacement synchronously enqueued by an
+      // `agent/cancel-requested` observer belongs to the next turn, not here.
+      const discarded = this.#inbox.pending()
+      // Clear work already present before abort observers run.
+      this.#inbox.clear()
+      if (discarded.length > 0) {
+        const items = discarded.map(({ message, steering }) => this.inboxInfo(message, steering))
+        agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
+      }
+    }
    cancellation?.request(resolvedCause)
  }

--- a/packages/core/agent-loop/src/inbox.ts
+++ b/packages/core/agent-loop/src/inbox.ts
@@ -1,7 +1,7 @@
 /**
 * Per-agent message inbox: queued and steering FIFOs. Purely an in-memory
- * mechanism of the loop driver — the public surface is `Agent.send()` and
- * `Agent.steer()`.
+ * mechanism of the loop driver — the public surface is `Agent.send()` and its
+ * fixed-preset aliases.
 *
 * @module dsh-agent-loop/inbox
 */
@@ -14,12 +14,14 @@ export interface InboxMessage {
  content: ContentBlock[]
  source: MessageSource
  contexts: HookContext[]
+  /** Whether the item is marked to wake the driver or force a continuation. */
+  wakeup: boolean
 }

 /**
 * Per-agent inbox: a queued FIFO (dequeued once per turn start) and a steering FIFO
 * (drained between steps of a running turn). Purely an in-memory mechanism of
- * the loop — the public surface is `Agent.send()` / `Agent.steer()`.
+ * the loop — the public surface is `Agent.send()` and its aliases.
 */
 export class Inbox {
  private queuedMessages: InboxMessage[] = []
@@ -37,18 +39,21 @@ export class Inbox {
  }

  /**
-   * Add a message to the queued FIFO and wake a parked {@link waitForQueued}.
+   * Add a message to the queued FIFO, waking a parked {@link waitForQueued}
+   * unless the item opted out. A non-waking item still runs once any woken
+   * item or later wakeup drives the parked loop.
   * @param message - the message to queue for the next turn start.
+   * @param wake - whether to wake a parked idle wait (default true).
   */
-  enqueue(message: InboxMessage): void {
+  enqueue(message: InboxMessage, wake = true): void {
    this.queuedMessages.push(message)
-    this.wakeup?.()
+    if (wake) this.wakeup?.()
  }

  /**
   * Add a message to the steering FIFO. Deliberately no wakeup: steering is
   * drained between steps of a running turn, never by the idle wait —
-   * `Agent.steer()` on an idle agent falls back to `send()` instead.
+   * `Agent.steer()` on an idle agent falls back to a woken follow-up instead.
   * @param message - the message to inject between steps of the running turn.
   */
  steer(message: InboxMessage): void {
@@ -71,6 +76,18 @@ export class Inbox {
    return this.steeringMessages.splice(0)
  }

+  /**
+   * Snapshot the pending items (queued then steering, FIFO order) without
+   * removing them — the discard notification's payload source.
+   * @returns the pending items paired with whether each is steering.
+   */
+  pending(): { message: InboxMessage; steering: boolean }[] {
+    return [
+      ...this.queuedMessages.map(message => ({ message, steering: false })),
+      ...this.steeringMessages.map(message => ({ message, steering: true })),
+    ]
+  }
+
  /**
   * Discard all pending messages (queued + steering) without delivering them —
   * used by `cancel()`, which drops un-started work rather than draining it into
--- a/packages/core/agent-loop/src/loop.ts
+++ b/packages/core/agent-loop/src/loop.ts
@@ -10,7 +10,7 @@ import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFai
 import { isDeepStrictEqual } from 'node:util'
 import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm'
 import { agentEvents, agentInterruptReasonOf, assembleContextFor } from '@deepseek-ai/dsh-agent'
-import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
+import type { AgentEventDispatch, ContinuationDecision, HookContext, InboxItemInfo, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
 import { canonicalHeader } from '@deepseek-ai/dsh-session'
 import type { PromptMessageData, Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
 import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
@@ -19,9 +19,14 @@ import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
 import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
 import type {} from '@deepseek-ai/dsh-tools'
 import { executeToolCalls } from './tool-calls.ts'
-import type { Inbox } from './inbox.ts'
+import type { Inbox, InboxMessage } from './inbox.ts'
 import type { TurnCancellation } from './cancellation.ts'

+/** Build the `agent/inbox/dequeue` payload for one claimed inbox item. */
+function inboxInfo(message: InboxMessage, steering: boolean): InboxItemInfo {
+  return { content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup }
+}
+
 /** Normalize thrown values while preserving an existing error code. */
 function toError(error: unknown): RequestError {
  return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
@@ -279,10 +284,11 @@ async function runTurn(
  const drainSteering = (): boolean => {
    const messages = handle.inbox.drainSteering()
    for (const message of messages) {
+      events.emit('agent/inbox/dequeue', inboxInfo(message, true))
      const prepared = preparePromptMessage(message.content, message.source, message.contexts)
      session.append('steering/message', { turn, ...prepared.data }, { surfaceOp: 'append' })
      for (const context of prepared.separateContexts) {
-        session.append('context/message', {
+        session.append('user/message', {
          content: context.content,
          source: context.source,
          ...context.meta === undefined ? {} : { meta: context.meta },
@@ -296,6 +302,7 @@ async function runTurn(
  const message = handle.inbox.dequeueQueued()
  /* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
  if (!message) throw new Error('runTurn invariant violated: no queued message at turn start')
+  events.emit('agent/inbox/dequeue', inboxInfo(message, false))
  const trigger: TurnTrigger = { kind: 'message', source: message.source }

  let reason: TurnEndReason = { kind: 'completed' }
@@ -538,7 +545,7 @@ async function runTurn(

      // A continuation reason becomes next-step steering.
      if (decision.action === 'continue' && decision.reason) {
-        handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source, contexts: [] })
+        handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source, contexts: [], wakeup: true })
      }
      let shouldContinue = decision.action === 'continue'

--- a/packages/core/agent-loop/tests/agent.spec.ts
+++ b/packages/core/agent-loop/tests/agent.spec.ts
@@ -139,7 +139,7 @@ describe('Agent', () => {
    agent.session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
    agent.inject([{ type: 'text', text: 'mid' }], { source: { kind: 'plugin', plugin: 'p' } })
    expect(agent.session.events.filter(e => e.type === 'turn/start')).toHaveLength(1)
-    expect(agent.session.events.at(-1)!.type).toBe('context/message')
+    expect(agent.session.events.at(-1)!.type).toBe('user/message')

    // Close the turn; now inject must wrap its own one-shot injection turn.
    agent.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
@@ -151,6 +151,16 @@ describe('Agent', () => {
    expect(agent.session.events.at(-1)!.type).toBe('turn/end') // turn-enclosed
  })

+  it('inject() defaults its source to an empty plugin, never user', async () => {
+    const adapter = new MockAdapter([textResponse('ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+    agent.session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    agent.inject([{ type: 'text', text: 'no explicit source' }])
+    const injected = agent.session.events.at(-1)!
+    expect(injected.type === 'user/message' && injected.data.source).toEqual({ kind: 'plugin', plugin: '' })
+  })
+
  it('idle inject() contains a failing flush (logs, does not throw into the caller)', async () => {
    const adapter = new MockAdapter([textResponse('ok')])
    const ctx = await harness(adapter)
@@ -202,7 +212,7 @@ describe('Agent', () => {

    expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow()
    const types = agent.session.events.map(e => e.type)
-    expect(types).toEqual(['turn/start', 'context/message', 'turn/end']) // balanced
+    expect(types).toEqual(['turn/start', 'user/message', 'turn/end']) // balanced
    await new Promise(r => setTimeout(r, 10))
    expect(flushes).toBe(1) // checkpoint fired despite the throwing turn/end listener
  })
--- a/packages/core/agent-loop/tests/cancel.spec.ts
+++ b/packages/core/agent-loop/tests/cancel.spec.ts
@@ -98,6 +98,25 @@ describe('Agent.cancel()', () => {
    expect(agent.session.events.some(e => e.type === 'turn/end')).toBe(true)
  })

+  it('cancel({ keepInbox: true }) preserves queued work and emits no discard', async () => {
+    const adapter = new MockAdapter([textResponse('reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+    const discards: unknown[] = []
+    ctx.on('agent/inbox/discard', (subject, items) => { if (subject === agent) discards.push(items) })
+
+    // Queue a turn WITHOUT waking the driver, so it sits in the inbox.
+    agent.send([{ type: 'text', text: 'preserved' }], { target: 'next-turn', wakeup: false })
+    // keepInbox cancel: no active turn, work preserved, no discard event.
+    agent.cancel({ kind: 'user' }, { keepInbox: true })
+    expect(discards).toEqual([])
+
+    // The preserved item still runs once the driver is woken by a later send.
+    send(agent, 'wake it')
+    await waitForIdle(ctx, agent)
+    expect(userTexts(agent)).toEqual(['preserved', 'wake it'])
+  })
+
  it('pre-step cancel drops the about-to-start turn (no turn is opened)', async () => {
    const adapter = new MockAdapter([textResponse('should not run')])
    const ctx = await harness(adapter)
--- a/packages/core/agent-loop/tests/contract-regressions.spec.ts
+++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts
@@ -275,7 +275,9 @@ describe('abort during tool execution ends the turn', () => {
          order.push(`tool/result:${event.data.callId}:${outcome}`)
          break
        }
-        case 'context/message': order.push('context/message'); break
+        // Injected context is a plugin-sourced user/message; the direct human
+        // prompt (user source) is not tracked in this ordering.
+        case 'user/message': if (event.data.source.kind !== 'user') order.push('context/message'); break
        case 'steering/message': order.push('steering/message'); break
        case 'step/end': order.push('step/end'); break
        case 'turn/end': {
@@ -354,13 +356,14 @@ describe('abort during tool execution ends the turn', () => {
    await waitForIdle(ctx, agent)

    const events = [...agent.session.events]
+    const isInjected = (e: SessionEvent): e is SessionEvent<'user/message'> => e.type === 'user/message' && e.data.source.kind !== 'user'
    expect(events
-      .filter(event => event.type === 'tool/result' || event.type === 'context/message'
+      .filter(event => event.type === 'tool/result' || isInjected(event)
        || event.type === 'step/end' || event.type === 'turn/end')
-      .map(event => event.type))
+      .map(event => isInjected(event) ? 'context/message' : event.type))
      .toEqual(['tool/result', 'context/message', 'context/message', 'step/end', 'turn/end'])
    expect(events
-      .filter(event => event.type === 'context/message')
+      .filter(isInjected)
      .map(event => event.data.content))
      .toEqual([
        [{ type: 'text', text: 'accepted before abort' }],
@@ -410,12 +413,13 @@ describe('abort during tool execution ends the turn', () => {
    await waitForIdle(ctx, agent)

    const events = [...agent.session.events]
+    const isInjected = (e: SessionEvent): e is SessionEvent<'user/message'> => e.type === 'user/message' && e.data.source.kind !== 'user'
    expect(events
-      .filter(event => event.type === 'tool/result' || event.type === 'context/message'
+      .filter(event => event.type === 'tool/result' || isInjected(event)
        || event.type === 'step/end' || event.type === 'turn/end')
-      .map(event => event.type))
+      .map(event => isInjected(event) ? 'context/message' : event.type))
      .toEqual(['tool/result', 'tool/result', 'context/message', 'step/end', 'turn/end'])
-    expect(events.find(event => event.type === 'context/message')?.data.content)
+    expect(events.find(isInjected)?.data.content)
      .toEqual([{ type: 'text', text: 'accepted after first result' }])
  })

@@ -456,7 +460,7 @@ describe('abort during tool execution ends the turn', () => {
    await fiber.dispose()

    expect(agent.session.events
-      .filter(event => event.type === 'context/message')
+      .filter((event): event is SessionEvent<'user/message'> => event.type === 'user/message' && event.data.source.kind !== 'user')
      .map(event => event.data.content))
      .toEqual([
        [{ type: 'text', text: 'accepted before disposal' }],
@@ -507,7 +511,7 @@ describe('abort during tool execution ends the turn', () => {
    send(agent, 'start a text-only turn')
    await waitForIdle(ctx, agent)

-    expect(agent.session.events.find(event => event.type === 'context/message')?.data.content)
+    expect(agent.session.events.find((event): event is SessionEvent<'user/message'> => event.type === 'user/message' && event.data.source.kind !== 'user')?.data.content)
      .toEqual([{ type: 'text', text: 'new turn context' }])
    expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('new turn context')
  })
@@ -763,7 +767,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    expect(agent.session.deriveMessages().at(-1)?.content).toEqual([{ type: 'text', text: 'routed' }])
  })

-  it('agent/queued carries the resolved source; steering/message records its source', async () => {
+  it('agent/inbox/enqueue carries the resolved source; steering/message records its source', async () => {
    const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -778,7 +782,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    }))

    const queuedSources: { source: MessageSource; contexts: HookContext[]; steering: boolean }[] = []
-    ctx.on('agent/queued', (_agent, _content, info) => void queuedSources.push(info))
+    ctx.on('agent/inbox/enqueue', (_agent, info) => void queuedSources.push({ source: info.source, contexts: info.contexts, steering: info.steering }))

    send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
    await waitForIdle(ctx, agent)
@@ -800,11 +804,11 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    let notifiedContent: ContentBlock[] | undefined
    let notifiedSource: MessageSource | undefined
    let notifiedContexts: HookContext[] | undefined
-    ctx.on('agent/queued', (subject, acceptedContent, info) => {
+    ctx.on('agent/inbox/enqueue', (subject, 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
+      notifiedContent = info.content
      notifiedSource = info.source
      notifiedContexts = info.contexts
    })
@@ -863,9 +867,9 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    let notifiedContent: ContentBlock[] | undefined
    let notifiedSource: MessageSource | undefined
    let notifiedContexts: HookContext[] | undefined
-    ctx.on('agent/queued', (subject, acceptedContent, info) => {
+    ctx.on('agent/inbox/enqueue', (subject, info) => {
      if (subject !== agent || !info.steering) return
-      notifiedContent = acceptedContent
+      notifiedContent = info.content
      notifiedSource = info.source
      notifiedContexts = info.contexts
    })
@@ -951,7 +955,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    expect(request).not.toContain('caller-mutated-steering-context-without-meta')

    const steeringIndex = agent.session.events.findIndex(event => event.type === 'steering/message')
-    const contextIndex = agent.session.events.findIndex(event => event.type === 'context/message'
+    const contextIndex = agent.session.events.findIndex(event => event.type === 'user/message'
      && event.data.source.kind === 'plugin' && event.data.source.plugin === 'steering-context')
    expect(steeringIndex).toBeGreaterThanOrEqual(0)
    expect(contextIndex).toBe(steeringIndex + 1)
--- a/packages/core/agent-loop/tests/coverage-edges.spec.ts
+++ b/packages/core/agent-loop/tests/coverage-edges.spec.ts
@@ -47,7 +47,7 @@ describe('inbox acceptance', () => {
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
    let queued = 0
-    ctx.on('agent/queued', () => { queued += 1 })
+    ctx.on('agent/inbox/enqueue', () => { queued += 1 })

    expect(() => {
      agent.send([{ type: 'text', text: 'first', bad: 1n } as never])
--- a/packages/core/agent-loop/tests/inbox.spec.ts
+++ b/packages/core/agent-loop/tests/inbox.spec.ts
@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
 import { Inbox } from '../src/inbox.ts'

 function message(text: string) {
-  return { content: [{ type: 'text' as const, text }], source: { kind: 'user' as const }, contexts: [] }
+  return { content: [{ type: 'text' as const, text }], source: { kind: 'user' as const }, contexts: [], wakeup: true }
 }

 function resolverPair() {
@@ -25,6 +25,32 @@ describe('Inbox', () => {
    expect(inbox.dequeueQueued()).toBeUndefined()
  })

+  it('enqueue(msg, false) queues without waking a parked waiter', async () => {
+    const inbox = new Inbox()
+    let woke = false
+    const waiter = inbox.waitForQueued(new Promise(() => {})).then(() => { woke = true })
+    inbox.enqueue(message('quiet'), false)
+    // The item is queued, but the parked waiter was not resolved by it.
+    expect(inbox.hasQueued).toBe(true)
+    await Promise.resolve()
+    expect(woke).toBe(false)
+    // A later waking enqueue resolves the same waiter.
+    inbox.enqueue(message('loud'))
+    await waiter
+    expect(woke).toBe(true)
+  })
+
+  it('pending() snapshots queued then steering without removing them', () => {
+    const inbox = new Inbox()
+    inbox.enqueue(message('q'))
+    inbox.steer(message('s'))
+    const pending = inbox.pending()
+    expect(pending.map(p => p.steering)).toEqual([false, true])
+    // Snapshot does not drain the FIFOs.
+    expect(inbox.hasQueued).toBe(true)
+    expect(inbox.hasSteering).toBe(true)
+  })
+
  it('pushes and drains steering messages separately from queued', () => {
    const inbox = new Inbox()
    inbox.steer(message('steer'))
--- a/packages/core/agent-loop/tests/interception.spec.ts
+++ b/packages/core/agent-loop/tests/interception.spec.ts
@@ -87,7 +87,7 @@ describe('agent/prompt-submit', () => {
    expect(JSON.stringify(adapter.requests[0]!.messages)).not.toContain('original')
  })

-  it('allow with additionalContexts injects separate context/message events into the turn', async () => {
+  it('allow with additionalContexts injects separate injected-context user messages into the turn', async () => {
    const adapter = new MockAdapter([textResponse('ok')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -107,12 +107,12 @@ describe('agent/prompt-submit', () => {
    await waitForIdle(ctx, agent)

    const log = events(agent)
-    const userMsg = log.find(e => e.type === 'user/message')
-    const ctxMsg = log.find(e => e.type === 'context/message')
+    const userMsg = log.find(e => e.type === 'user/message' && e.data.source.kind === 'user')
+    const ctxMsg = log.find(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
    expect(userMsg).toBeDefined()
-    expect(ctxMsg?.type === 'context/message' && ctxMsg.data.content).toEqual([{ type: 'text', text: '<system-reminder>extra ctx</system-reminder>' }])
-    expect(ctxMsg?.type === 'context/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
-    expect(ctxMsg?.type === 'context/message' && ctxMsg.data.meta).toEqual(meta)
+    expect(ctxMsg?.type === 'user/message' && ctxMsg.data.content).toEqual([{ type: 'text', text: '<system-reminder>extra ctx</system-reminder>' }])
+    expect(ctxMsg?.type === 'user/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
+    expect(ctxMsg?.type === 'user/message' && ctxMsg.data.meta).toEqual(meta)
    const sent = JSON.stringify(adapter.requests[0]!.messages)
    expect(sent).toContain('extra ctx')
  })
@@ -155,7 +155,7 @@ describe('agent/prompt-submit', () => {
        }],
      },
    })
-    expect(log.some(event => event.type === 'context/message')).toBe(false)
+    expect(log.some(event => event.type === 'user/message' && event.data.source.kind === 'plugin')).toBe(false)
    expect(adapter.requests[0]?.messages.at(-1)).toEqual({
      role: 'user',
      content: [
@@ -215,7 +215,6 @@ describe('agent/prompt-submit', () => {
    expect(log.some(e => e.type === 'turn/start')).toBe(true)
    expect(log.some(e => e.type === 'turn/end')).toBe(true)
    expect(log.some(e => e.type === 'user/message')).toBe(false)
-    expect(log.some(e => e.type === 'context/message')).toBe(false)
    expect(log.some(e => e.type === 'step/start')).toBe(false)
    // the veto is recorded durably as a prompt/blocked in the open turn
    const blocked = log.find(e => e.type === 'prompt/blocked')
@@ -340,8 +339,8 @@ describe('agent/session-start', () => {
    // the injected context reached the model on the first (only) request
    expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('session preamble')
    // and is recorded with the plugin source, never mislabeled as a user prompt
-    const ctxMsg = events(agent).find(e => e.type === 'context/message')
-    expect(ctxMsg?.type === 'context/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
+    const ctxMsg = events(agent).find(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
+    expect(ctxMsg?.type === 'user/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
  })

  it('a throwing session-start listener does not abort agent construction', async () => {
@@ -624,23 +623,22 @@ describe('tool additionalContexts buffering across a step', () => {
    send(agent, 'go')
    await waitForIdle(ctx, agent)

-    // Event order in the log: both tool/results, THEN both context/messages —
+    // Event order in the log: both tool/results, THEN both injected contexts —
    // never interleaved (which would break tool-call/result adjacency).
-    const types = events(agent).map(e => e.type)
-    const firstResult = types.indexOf('tool/result')
-    const lastResult = types.lastIndexOf('tool/result')
-    const firstCtx = types.indexOf('context/message')
+    const injected = events(agent).filter(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
+    const seqs = events(agent)
+    const firstResult = seqs.findIndex(e => e.type === 'tool/result')
+    const lastResult = seqs.map(e => e.type).lastIndexOf('tool/result')
+    const firstCtx = seqs.findIndex(e => e === injected[0])
    expect(firstResult).toBeGreaterThanOrEqual(0)
    expect(lastResult).toBeGreaterThan(firstResult) // two results
    expect(firstCtx).toBeGreaterThan(lastResult)    // context only after ALL results
    // both contexts present
-    const ctxTexts = events(agent)
-      .filter(e => e.type === 'context/message')
-      .flatMap(e => (e.type === 'context/message' ? e.data.content : []))
+    const ctxTexts = injected
+      .flatMap(e => (e.type === 'user/message' ? e.data.content : []))
      .map(b => (b.type === 'text' ? b.text : ''))
    expect(ctxTexts).toEqual(['ctx-c1', 'ctx-c2'])
-    const contextEvents = events(agent).filter(e => e.type === 'context/message')
-    expect(contextEvents.map(e => e.type === 'context/message' && e.data.meta)).toEqual([{ callId: 'c1' }, { callId: 'c2' }])
+    expect(injected.map(e => e.type === 'user/message' && e.data.meta)).toEqual([{ callId: 'c1' }, { callId: 'c2' }])
  })

  it('appends multiple contexts deferred by one composite tool after its outer result', async () => {
@@ -661,14 +659,14 @@ describe('tool additionalContexts buffering across a step', () => {

    const log = events(agent)
    const resultIndex = log.findIndex(event => event.type === 'tool/result')
-    const contextEvents = log.filter(event => event.type === 'context/message')
+    const contextEvents = log.filter(event => event.type === 'user/message' && event.data.source.kind === 'plugin')
    expect(resultIndex).toBeGreaterThanOrEqual(0)
    expect(log.findIndex(event => event === contextEvents[0])).toBeGreaterThan(resultIndex)
-    expect(contextEvents.map(event => event.type === 'context/message' && event.data.source)).toEqual([
+    expect(contextEvents.map(event => event.type === 'user/message' && event.data.source)).toEqual([
      { kind: 'plugin', plugin: 'a' },
      { kind: 'plugin', plugin: 'b' },
    ])
-    expect(contextEvents.map(event => event.type === 'context/message' && event.data.meta)).toEqual([{ order: 1 }, { order: 2 }])
+    expect(contextEvents.map(event => event.type === 'user/message' && event.data.meta)).toEqual([{ order: 1 }, { order: 2 }])
  })
 })

@@ -750,13 +748,13 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se

    const log = events(agent)
    // session-start preamble injected
-    expect(log.some(e => e.type === 'context/message'
+    expect(log.some(e => e.type === 'user/message' && e.data.source.kind === 'plugin'
      && e.data.content.some(b => b.type === 'text' && b.text.includes('policy active (started: startup)')))).toBe(true)
-    // prompt allowed → user/message recorded
-    expect(log.some(e => e.type === 'user/message')).toBe(true)
+    // prompt allowed → user-sourced user/message recorded
+    expect(log.some(e => e.type === 'user/message' && e.data.source.kind === 'user')).toBe(true)
    // tool ran (echo allowed) and post-execute attached "audited" context
    expect(log.some(e => e.type === 'tool/result' && !e.data.isError)).toBe(true)
-    expect(log.some(e => e.type === 'context/message'
+    expect(log.some(e => e.type === 'user/message' && e.data.source.kind === 'plugin'
      && e.data.content.some(b => b.type === 'text' && b.text === 'audited'))).toBe(true)
    // NO hook/* events — a native plugin needs none
    expect(log.some(e => e.type.startsWith('hook/'))).toBe(false)
--- a/packages/core/agent-loop/tests/invariant.spec.ts
+++ b/packages/core/agent-loop/tests/invariant.spec.ts
@@ -42,7 +42,7 @@ describe('request-reconstruction invariant', () => {

  it('uses the step boundary rather than content appended afterward', async () => {
    const { ctx, session, boundary } = await requestSetup()
-    session.append('context/message', { content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' } }, { surfaceOp: 'append' })
+    session.append('user/message', { content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' } }, { surfaceOp: 'append' })
    const options = loopRequest({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
    expect(() => { dispatch(ctx, options) }).not.toThrow()
  })
--- a/packages/core/agent-loop/tests/loop.spec.ts
+++ b/packages/core/agent-loop/tests/loop.spec.ts
@@ -380,7 +380,7 @@ describe('agent loop', () => {
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })

    agent.inject([{ type: 'text', text: 'file changed: a.ts' }], { source: { kind: 'plugin', plugin: 'watcher' } })
-    // The idle inject records a self-contained turn (turn/start → context/message
+    // The idle inject records a self-contained turn (turn/start → user/message
    // → turn/end) so the event stays turn-enclosed, but does NOT run the model.
    await new Promise(r => setTimeout(r, 20))
    expect(agent.status).toBe('idle')
@@ -416,8 +416,8 @@ describe('agent loop', () => {
    send(agent, 'go')
    await waitForIdle(ctx, agent)

-    const contextEvent = agent.session.events.find(event => event.type === 'context/message')
-    expect(contextEvent?.type === 'context/message' && contextEvent.data).toMatchObject({ meta })
+    const contextEvent = agent.session.events.find(event => event.type === 'user/message' && event.data.source.kind === 'plugin')
+    expect(contextEvent?.type === 'user/message' && contextEvent.data).toMatchObject({ meta })
    const requestText = JSON.stringify(adapter.requests[0]!.messages)
    expect(requestText).toContain('Additional instructions from: pkg/AGENTS.md')
    expect(requestText).not.toContain('<context source=')
@@ -445,7 +445,7 @@ describe('agent loop', () => {
        })
        first.text = 'mutated after inject'
        agent.inject([{ type: 'text', text: 'second notice' }], { source: { kind: 'plugin', plugin: 'x' } })
-        visibleDuringTool = agent.session.events.some(e => e.type === 'context/message')
+        visibleDuringTool = agent.session.events.some(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
        return [{ type: 'text', text: 'ok' }]
      },
    }))
@@ -462,13 +462,13 @@ describe('agent loop', () => {
    const ts0 = turnStarts[0]!
    expect(ts0.type === 'turn/start' && ts0.data.trigger.kind).toBe('message')
    const result = agent.session.events.find(e => e.type === 'tool/result')!
-    const contexts = agent.session.events.filter(e => e.type === 'context/message')
+    const contexts = agent.session.events.filter(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
    expect(contexts).toHaveLength(2)
    expect(result.seq).toBeLessThan(contexts[0]!.seq)
-    expect(contexts[0]?.type === 'context/message' && contexts[0].data).toMatchObject({
+    expect(contexts[0]?.type === 'user/message' && contexts[0].data).toMatchObject({
      meta,
    })
-    expect(contexts.flatMap(event => event.type === 'context/message' ? event.data.content : []))
+    expect(contexts.flatMap(event => event.type === 'user/message' ? event.data.content : []))
      .toEqual([
        { type: 'text', text: 'mid-turn notice' },
        { type: 'text', text: 'second notice' },
@@ -512,7 +512,7 @@ describe('agent loop', () => {
    send(agent, 'go')
    await waitForIdle(ctx, agent)

-    expect(agent.session.events.some(event => event.type === 'context/message')).toBe(false)
+    expect(agent.session.events.some(event => event.type === 'user/message' && event.data.source.kind === 'plugin')).toBe(false)
  })

  it('agent/turn-continuation can force-continue (/loop pattern) and force-stop', async () => {
@@ -621,7 +621,7 @@ describe('agent loop', () => {
    ctx.on('agent/pre-step', (subject) => {
      if (subject === agent && !injected) {
        injected = true
-        subject.session.append('context/message', {
+        subject.session.append('user/message', {
          content: [{ type: 'text', text: 'INJECTED-IN-PRE-STEP' }],
          source: { kind: 'plugin', plugin: 'test' },
        }, { surfaceOp: 'append' })
@@ -639,7 +639,7 @@ describe('agent loop', () => {
    // And the injected event sits BEFORE the first step/start in the log —
    // the seam fired outside the step.
    const events = agent.session.events
-    const injectedSeq = events.find(e => e.type === 'context/message')!.seq
+    const injectedSeq = events.find(e => e.type === 'user/message' && e.data.source.kind === 'plugin')!.seq
    const firstStepStartSeq = events.find(e => e.type === 'step/start')!.seq
    expect(injectedSeq).toBeLessThan(firstStepStartSeq)
  })
@@ -1017,13 +1017,13 @@ describe('agent loop', () => {
    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('turn-end listener message')
  })

-  it('keeps a reentrant agent/queued send as the next independent turn', async () => {
+  it('keeps a reentrant agent/inbox/enqueue send as the next independent turn', async () => {
    const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })

    let nested = false
-    ctx.on('agent/queued', (subject) => {
+    ctx.on('agent/inbox/enqueue', (subject) => {
      if (subject !== agent || nested) return
      nested = true
      send(agent, 'queued listener message')
--- a/packages/core/agent-loop/tests/request-reconstruction.spec.ts
+++ b/packages/core/agent-loop/tests/request-reconstruction.spec.ts
@@ -118,7 +118,7 @@ describe('request stability across the loop', () => {
      preStep()
      const session = agent.session
      const nodes = session.surface.nodes
-      session.append('context/message', {
+      session.append('user/message', {
        content: [{ type: 'text', text: '[summary of turn 1]' }],
        source: { kind: 'plugin', plugin: 'test-compact' },
      }, {
@@ -180,7 +180,7 @@ describe('request stability across the loop', () => {
    const first = adapter.requests[0]!
    // The inject landed in the log after the boundary: not in THIS request…
    expect(first.messages.some(m => m.content.some(b => b.type === 'text' && b.text.includes('[late context]')))).toBe(false)
-    expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
+    expect(agent.session.events.some(e => e.type === 'user/message' && e.data.source.kind === 'plugin')).toBe(true)

    send(agent, 'second')
    await waitForIdle(ctx, agent)
--- a/packages/core/agent-loop/tests/request-recovery.spec.ts
+++ b/packages/core/agent-loop/tests/request-recovery.spec.ts
@@ -154,12 +154,15 @@ describe('agent post-step and request-error lifecycle', () => {
    const agent = ctx.agentLoop.create(SessionId('post-step-order'), { provider: 'mock', model: 'mock' })
    const order: string[] = []
    ctx.on('session/event', (_session, event) => {
+      // Injected context is a plugin-sourced user/message; the direct human
+      // prompt (user source) stays untracked as before.
+      const isInjected = event.type === 'user/message' && event.data.source.kind !== 'user'
      if (
        event.type === 'assistant/message' || event.type === 'tool/call'
-        || event.type === 'tool/result' || event.type === 'context/message'
+        || event.type === 'tool/result' || isInjected
        || event.type === 'steering/message' || event.type === 'step/end'
      ) {
-        if (!('step' in event.data) || event.data.step === 1) order.push(event.type)
+        if (!('step' in event.data) || event.data.step === 1) order.push(isInjected ? 'context/message' : event.type)
      }
    })
    ctx.on('agent/post-step', (subject, turn, step, signal) => {
@@ -271,7 +274,7 @@ describe('agent post-step and request-error lifecycle', () => {
      expect({ turn, step, code: error.code }).toEqual({ turn: 1, step: 1, code: CONTEXT_WINDOW_EXCEEDED_CODE })
      expect(facts.code).toBe(CONTEXT_WINDOW_EXCEEDED_CODE)
      attempts.push(history.length)
-      subject.session.append('context/message', {
+      subject.session.append('user/message', {
        content: [{ type: 'text', text: 'RECOVERY SURFACE MUTATION' }],
        source: { kind: 'plugin', plugin: 'test-recovery' },
      }, { surfaceOp: 'append' })
@@ -288,7 +291,7 @@ describe('agent post-step and request-error lifecycle', () => {
    const ends = agent.session.events.filter(event => event.type === 'step/end')
    expect(starts.map(event => event.data.step)).toEqual([1, 2])
    expect(ends.map(event => event.data.step)).toEqual([1, 2])
-    const recovery = agent.session.events.find(event => event.type === 'context/message')!
+    const recovery = agent.session.events.find(event => event.type === 'user/message' && event.data.source.kind === 'plugin')!
    expect(ends[0]!.seq).toBeLessThan(recovery.seq)
    expect(recovery.seq).toBeLessThan(starts[1]!.seq)
  })
--- a/packages/core/agent-loop/tests/tool-calls.spec.ts
+++ b/packages/core/agent-loop/tests/tool-calls.spec.ts
@@ -403,11 +403,11 @@ describe('tool-call scheduler: ordered middleware and additional contexts', () =
    await waitForIdle(ctx, agent)

    const log = events(agent)
-    const contextTexts = log.filter(e => e.type === 'context/message')
-      .map(e => (e.data.content[0] as { text: string }).text)
+    const contextTexts = log.filter(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
+      .map(e => ((e.data as { content: { text: string }[] }).content[0]!).text)
    expect(contextTexts).toEqual(['ctx-c1', 'ctx-c2'])
    const lastResult = log.findLastIndex(e => e.type === 'tool/result')
-    const firstContext = log.findIndex(e => e.type === 'context/message')
+    const firstContext = log.findIndex(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
    expect(lastResult).toBeLessThan(firstContext)
  })

@@ -544,10 +544,11 @@ describe('tool-call scheduler: abort handling', () => {
        expect.objectContaining({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } }),
        expect.objectContaining({ callId: CallId('c4'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } }),
      ])
-    const settled = events(agent).filter(e => e.type === 'tool/result' || e.type === 'context/message')
+    const settled = events(agent).filter(e => e.type === 'tool/result'
+      || (e.type === 'user/message' && e.data.source.kind === 'plugin'))
    expect(settled.map(e => e.type))
-      .toEqual(['tool/result', 'tool/result', 'tool/result', 'tool/result', 'context/message', 'context/message'])
-    expect(settled.filter(e => e.type === 'context/message')
+      .toEqual(['tool/result', 'tool/result', 'tool/result', 'tool/result', 'user/message', 'user/message'])
+    expect(settled.filter(e => e.type === 'user/message')
      .map(e => (e.data.content[0] as { text: string }).text))
      .toEqual(['ctx-c1', 'ctx-c2'])
  })