Merge branch 'codex/tool-json-schema-dsl' into codex/canonical-tool-output

# Conflicts: # docs/config-catalog.md # docs/persistence-catalog.md # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/stdout.expected.jsonl # packages/cordis/tool-cordis/src/api-catalog.ts # packages/core/agent-loop/tests/contract-regressions.spec.ts # packages/ui/acp/src/index.ts # packages/ui/tui/src/index.ts
2026-07-22 23:50:41 +08:00
parent b8aa1507cd 065d20c249
commit 8d42d3c979
159 changed files with 3918 additions and 362 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 plus resolved source once as a detached, deeply frozen lossless-JSON FIFO item. If claimed, it is the sole ordinary message in its turn; a successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, or a pre-start failure may drop it without a turn. Running `steer()` enters the steering FIFO: an open turn records it at the next steering checkpoint before a request or continuation decision, but policy can still stop before another step; steering left after turn close and its checkpoint becomes later queued input 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.
+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.

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

--- a/packages/core/agent-loop/src/agent.ts
+++ b/packages/core/agent-loop/src/agent.ts
@@ -201,9 +201,10 @@ export class ReactLoopAgent implements Agent {
   */
  private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
    const source = this.resolveSource(options)
-    const accepted = snapshotJsonValue({ content, source })
+    const contexts = options?.contexts ?? []
+    const accepted = snapshotJsonValue({ content, source, contexts })
    if (accepted === undefined) {
-      throw new TypeError('agent message content and source must be losslessly JSON-serializable')
+      throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
    }
    return deepFreeze(accepted)
  }
@@ -226,7 +227,7 @@ export class ReactLoopAgent implements Agent {
    this.assertNotDisposed()
    const accepted = this.acceptMessage(content, options)
    this.#inbox.enqueue(accepted)
-    const info = { source: accepted.source, steering: false } as const
+    const info = { source: accepted.source, contexts: accepted.contexts, steering: false } as const
    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
  }

@@ -235,7 +236,7 @@ export class ReactLoopAgent implements Agent {
    if (this._status !== 'running') { this.send(content, options); return }
    const accepted = this.acceptMessage(content, options)
    this.#inbox.steer(accepted)
-    const info = { source: accepted.source, steering: true } as const
+    const info = { source: accepted.source, contexts: accepted.contexts, steering: true } as const
    agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
  }

--- a/packages/core/agent-loop/src/inbox.ts
+++ b/packages/core/agent-loop/src/inbox.ts
@@ -7,11 +7,13 @@
 */

 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
+import type { HookContext } from '@deepseek-ai/dsh-agent'

 /** One message waiting in an agent's inbox. */
 export interface InboxMessage {
  content: ContentBlock[]
  source: MessageSource
+  contexts: HookContext[]
 }

 /**
--- a/packages/core/agent-loop/src/loop.ts
+++ b/packages/core/agent-loop/src/loop.ts
@@ -12,7 +12,7 @@ import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorC
 import { agentEvents, agentInterruptReasonOf, assembleContextFor } from '@deepseek-ai/dsh-agent'
 import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
 import { canonicalHeader } from '@deepseek-ai/dsh-session'
-import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
+import type { PromptMessageData, Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
 import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
 import type { TransmissionLog } from './request-log.ts'
 import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
@@ -92,6 +92,45 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
 /** Internal control-flow sentinel; durable classification comes only from the turn signal. */
 const TURN_INTERRUPTED = new Error('turn interrupted')

+const PROMPT_PREFIX_REQUEST_DELIMITER: ContentBlock = {
+  type: 'text',
+  text: '\n\n## My request:\n',
+}
+
+interface PreparedPromptMessage {
+  data: PromptMessageData
+  separateContexts: HookContext[]
+}
+
+/** Bake declared prefix contexts into one reconstructable prompt message. */
+function preparePromptMessage(
+  content: ContentBlock[],
+  source: PromptMessageData['source'],
+  contexts: readonly HookContext[],
+): PreparedPromptMessage {
+  const prefixContexts = contexts.filter(context => context.placement === 'prompt-prefix')
+  const separateContexts = contexts.filter(context => context.placement !== 'prompt-prefix')
+  if (prefixContexts.length === 0) return { data: { content, source }, separateContexts }
+  return {
+    data: {
+      content: [
+        ...prefixContexts.flatMap(context => context.content),
+        PROMPT_PREFIX_REQUEST_DELIMITER,
+        ...content,
+      ],
+      source,
+      envelope: {
+        displayContent: content,
+        prefixContexts: prefixContexts.map(context => ({
+          source: context.source,
+          ...context.meta === undefined ? {} : { meta: context.meta },
+        })),
+      },
+    },
+    separateContexts,
+  }
+}
+
 /** Stop at an explicit cooperative boundary without stringifying the runtime reason. */
 function interruptionCheckpoint(signal: AbortSignal): void {
  if (signal.aborted) throw TURN_INTERRUPTED
@@ -240,7 +279,15 @@ async function runTurn(
  const drainSteering = (): boolean => {
    const messages = handle.inbox.drainSteering()
    for (const message of messages) {
-      session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
+      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', {
+          content: context.content,
+          source: context.source,
+          ...context.meta === undefined ? {} : { meta: context.meta },
+        }, { surfaceOp: 'append' })
+      }
    }
    return messages.length > 0
  }
@@ -301,7 +348,10 @@ async function runTurn(
    // throws) is caught below and the turn still closes.
    const promptDecision = await events.waterfall(
      'agent/prompt-submit', message.content, message.source, signal,
-      () => Promise.resolve<PromptDecision>({ kind: 'allow' }),
+      () => Promise.resolve<PromptDecision>({
+        kind: 'allow',
+        ...message.contexts.length === 0 ? {} : { additionalContexts: message.contexts },
+      }),
    )
    interruptionCheckpoint(signal)
    if (promptDecision.kind === 'block') {
@@ -310,11 +360,12 @@ async function runTurn(
    } else {
      // `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
      const content = promptDecision.content ?? message.content
-      session.append('user/message', { content, source: message.source }, { surfaceOp: 'append' })
-      // Every `allow.additionalContexts` entry is a separate context/message the
-      // next request also sees. The turn is open, so inject() appends each one
-      // into THIS turn without flattening provenance or metadata.
-      for (const context of promptDecision.additionalContexts ?? []) {
+      const prepared = preparePromptMessage(content, message.source, promptDecision.additionalContexts ?? [])
+      session.append('user/message', prepared.data, { surfaceOp: 'append' })
+      // Separate contexts still enter THIS turn through inject(). Prefix
+      // contexts are already baked into the user/message with their durable
+      // display envelope, so appending them again would duplicate model input.
+      for (const context of prepared.separateContexts) {
        agent.inject(context.content, {
          source: context.source,
          ...context.meta !== undefined ? { meta: context.meta } : {},
@@ -487,7 +538,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 })
+        handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source, contexts: [] })
      }
      let shouldContinue = decision.action === 'continue'

--- a/packages/core/agent-loop/tests/contract-regressions.spec.ts
+++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts
@@ -4,7 +4,7 @@ import LlmService, { CallId, ContentBlock, MessageSource, ProviderRequestId, Str
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineContentToolFixture, TOOL_ABORTED, TOOL_ABORTED_BEFORE_DISPATCH, type PostToolDecision } from '@deepseek-ai/dsh-tools'
-import AgentRegistry, { type Agent, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
+import AgentRegistry, { type Agent, type ContinuationDecision, type HookContext } from '@deepseek-ai/dsh-agent'
 import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from '@deepseek-ai/dsh-agent-loop'
 import { prepareReactLoopAgent } from '../src/agent.ts'
 import InvariantService from '@deepseek-ai/dsh-invariants'
@@ -777,14 +777,14 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
      },
    }))

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

    send(agent, 'go') // no explicit source → default {kind:'user'} must be visible
    await waitForIdle(ctx, agent)

-    expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, steering: false })
-    expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, steering: true })
+    expect(queuedSources[0]).toEqual({ source: { kind: 'user' }, contexts: [], steering: false })
+    expect(queuedSources[1]).toEqual({ source: { kind: 'plugin', plugin: 'goal' }, contexts: [], steering: true })
    // The drain appends the durable steering/message with the caller's source
    // intact — the log, not a transient emit, is where consumers read it.
    const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.source] : [])
@@ -799,24 +799,39 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    const source = { kind: 'plugin' as const, plugin: 'accepted-source' }
    let notifiedContent: ContentBlock[] | undefined
    let notifiedSource: MessageSource | undefined
+    let notifiedContexts: HookContext[] | undefined
    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
+      notifiedContexts = info.contexts
    })

-    agent.send(content, { source })
+    const contexts: HookContext[] = [{
+      content: [{ type: 'text', text: 'accepted-context' }],
+      source: { kind: 'plugin', plugin: 'context-source' },
+      meta: { version: 1 },
+    }]
+    agent.send(content, { source, contexts })
    content[0]!.text = 'caller-mutated-send'
    source.plugin = 'caller-mutated-source'
+    contexts[0]!.content[0] = { type: 'text', text: 'caller-mutated-context' }
    await waitForIdle(ctx, agent)

    expect(notifiedContent).toEqual([{ type: 'text', text: 'accepted-send' }])
    expect(notifiedSource).toEqual({ kind: 'plugin', plugin: 'accepted-source' })
+    expect(notifiedContexts).toEqual([{
+      content: [{ type: 'text', text: 'accepted-context' }],
+      source: { kind: 'plugin', plugin: 'context-source' },
+      meta: { version: 1 },
+    }])
    expect(Object.isFrozen(notifiedContent)).toBe(true)
    expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
    expect(Object.isFrozen(notifiedSource)).toBe(true)
+    expect(Object.isFrozen(notifiedContexts)).toBe(true)
+    expect(Object.isFrozen(notifiedContexts?.[0]?.content)).toBe(true)
    const recorded = agent.session.events.flatMap(event => event.type === 'user/message' ? [event.data] : [])
    expect(recorded).toContainEqual({
      content: [{ type: 'text', text: 'accepted-send' }],
@@ -824,7 +839,9 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    })
    const request = JSON.stringify(adapter.requests[0]!.messages)
    expect(request).toContain('accepted-send')
+    expect(request).toContain('accepted-context')
    expect(request).not.toContain('caller-mutated-send')
+    expect(request).not.toContain('caller-mutated-context')
  })

  it('running steer() owns content and source before notification and delivery', async () => {
@@ -845,10 +862,12 @@ 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) => {
      if (subject !== agent || !info.steering) return
      notifiedContent = acceptedContent
      notifiedSource = info.source
+      notifiedContexts = info.contexts
    })

    agent.send([{ type: 'text', text: 'start' }])
@@ -856,27 +875,86 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
    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 })
+    const contexts: HookContext[] = [
+      {
+        content: [{ type: 'text', text: 'accepted-steering-prefix' }],
+        source: { kind: 'plugin', plugin: 'steering-prefix' },
+        placement: 'prompt-prefix',
+      },
+      {
+        content: [{ type: 'text', text: 'accepted-steering-context' }],
+        source: { kind: 'plugin', plugin: 'steering-context' },
+        meta: { kind: 'separate-card' },
+      },
+      {
+        content: [{ type: 'text', text: 'accepted-steering-context-without-meta' }],
+        source: { kind: 'plugin', plugin: 'steering-context-without-meta' },
+      },
+    ]
+    agent.steer(content, { source, contexts })
    content[0]!.text = 'caller-mutated-steer'
    source.plugin = 'caller-mutated-source'
+    contexts[0]!.content[0] = { type: 'text', text: 'caller-mutated-steering-prefix' }
+    contexts[0]!.placement = 'separate'
+    contexts[1]!.content[0] = { type: 'text', text: 'caller-mutated-steering-context' }
+    contexts[2]!.content[0] = { type: 'text', text: 'caller-mutated-steering-context-without-meta' }
    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(notifiedContexts).toEqual([
+      {
+        content: [{ type: 'text', text: 'accepted-steering-prefix' }],
+        source: { kind: 'plugin', plugin: 'steering-prefix' },
+        placement: 'prompt-prefix',
+      },
+      {
+        content: [{ type: 'text', text: 'accepted-steering-context' }],
+        source: { kind: 'plugin', plugin: 'steering-context' },
+        meta: { kind: 'separate-card' },
+      },
+      {
+        content: [{ type: 'text', text: 'accepted-steering-context-without-meta' }],
+        source: { kind: 'plugin', plugin: 'steering-context-without-meta' },
+      },
+    ])
    expect(Object.isFrozen(notifiedContent)).toBe(true)
    expect(Object.isFrozen(notifiedContent?.[0])).toBe(true)
    expect(Object.isFrozen(notifiedSource)).toBe(true)
+    expect(Object.isFrozen(notifiedContexts)).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' }],
+      content: [
+        { type: 'text', text: 'accepted-steering-prefix' },
+        { type: 'text', text: '\n\n## My request:\n' },
+        { type: 'text', text: 'accepted-steer' },
+      ],
      source: { kind: 'plugin', plugin: 'accepted-source' },
+      envelope: {
+        displayContent: [{ type: 'text', text: 'accepted-steer' }],
+        prefixContexts: [{
+          source: { kind: 'plugin', plugin: 'steering-prefix' },
+        }],
+      },
    })
    const request = JSON.stringify(adapter.requests[1]!.messages)
    expect(request).toContain('accepted-steer')
+    expect(request).toContain('accepted-steering-prefix')
+    expect(request).toContain('accepted-steering-context')
+    expect(request).toContain('accepted-steering-context-without-meta')
    expect(request).not.toContain('caller-mutated-steer')
+    expect(request).not.toContain('caller-mutated-steering-prefix')
+    expect(request).not.toContain('caller-mutated-steering-context')
+    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'
+      && 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/inbox.spec.ts
+++ b/packages/core/agent-loop/tests/inbox.spec.ts
@@ -1,6 +1,10 @@
 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: [] }
+}
+
 function resolverPair() {
  let r!: () => void
  const p = new Promise<void>((resolve) => { r = resolve })
@@ -10,8 +14,8 @@ function resolverPair() {
 describe('Inbox', () => {
  it('dequeues one queued message at a time in FIFO order', () => {
    const inbox = new Inbox()
-    inbox.enqueue({ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } })
-    inbox.enqueue({ content: [{ type: 'text', text: 'second' }], source: { kind: 'user' } })
+    inbox.enqueue(message('first'))
+    inbox.enqueue(message('second'))
    expect(inbox.hasQueued).toBe(true)

    expect(inbox.dequeueQueued()?.content[0]).toMatchObject({ text: 'first' })
@@ -23,7 +27,7 @@ describe('Inbox', () => {

  it('pushes and drains steering messages separately from queued', () => {
    const inbox = new Inbox()
-    inbox.steer({ content: [{ type: 'text', text: 'steer' }], source: { kind: 'user' } })
+    inbox.steer(message('steer'))
    expect(inbox.hasQueued).toBe(false)
    expect(inbox.hasSteering).toBe(true)

@@ -34,7 +38,7 @@ describe('Inbox', () => {

  it('waitForQueued returns immediately when a queued message is already present', async () => {
    const inbox = new Inbox()
-    inbox.enqueue({ content: [{ type: 'text', text: 'ready' }], source: { kind: 'user' } })
+    inbox.enqueue(message('ready'))

    const started = Date.now()
    await inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
@@ -45,7 +49,7 @@ describe('Inbox', () => {
    const inbox = new Inbox()
    const waiter = inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
    // enqueue after starting the wait
-    setTimeout(() => { inbox.enqueue({ content: [{ type: 'text', text: 'wake' }], source: { kind: 'user' } }) }, 5)
+    setTimeout(() => { inbox.enqueue(message('wake')) }, 5)
    await waiter
  })

@@ -69,7 +73,7 @@ describe('Inbox', () => {
    r1()
    await p1

-    inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
+    inbox.enqueue(message('hey'))
  })

  it('clears wakeup in finally handler when enqueue resolves', async () => {
@@ -77,7 +81,7 @@ describe('Inbox', () => {
    void inbox.waitForQueued(new Promise(() => {})) // never-resolving cancel
    // The wakeup is set. Now trigger it via enqueue → wakeup() calls resolve,
    // promise resolves, finally clears wakeup because wakeup === resolve.
-    inbox.enqueue({ content: [{ type: 'text', text: 'wake' }], source: { kind: 'user' } })
+    inbox.enqueue(message('wake'))
    // No explicit await needed — enqueue is synchronous, and the microtask
    // (finally) runs. The key coverage hit is finally with wakeup === resolve.
  })
@@ -94,6 +98,6 @@ describe('Inbox', () => {
    await c1

    // The replacement remains registered and is resolved by enqueue.
-    inbox.enqueue({ content: [{ type: 'text', text: 'hey' }], source: { kind: 'user' } })
+    inbox.enqueue(message('hey'))
  })
 })
--- a/packages/core/agent-loop/tests/interception.spec.ts
+++ b/packages/core/agent-loop/tests/interception.spec.ts
@@ -117,6 +117,55 @@ describe('agent/prompt-submit', () => {
    expect(sent).toContain('extra ctx')
  })

+  it('bakes prompt-prefix contexts and a request delimiter into one durable user message', async () => {
+    const adapter = new MockAdapter([textResponse('ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('prefixed'), { provider: 'mock', model: 'mock' })
+
+    ctx.on('agent/prompt-submit', async (_agent, _content, _source, _signal, next): Promise<PromptDecision> => {
+      const downstream = await next()
+      return downstream.kind === 'block'
+        ? downstream
+        : { ...downstream, content: [{ type: 'text', text: 'rewritten request' }] }
+    })
+    agent.send([{ type: 'text', text: 'original request' }], {
+      contexts: [{
+        content: [{ type: 'text', text: 'untrusted prefix' }],
+        source: { kind: 'plugin', plugin: 'prefix' },
+        placement: 'prompt-prefix',
+        meta: { kind: 'prefix-card' },
+      }],
+    })
+    await waitForIdle(ctx, agent)
+
+    const log = events(agent)
+    const user = log.find(event => event.type === 'user/message')
+    expect(user?.type === 'user/message' && user.data).toEqual({
+      content: [
+        { type: 'text', text: 'untrusted prefix' },
+        { type: 'text', text: '\n\n## My request:\n' },
+        { type: 'text', text: 'rewritten request' },
+      ],
+      source: { kind: 'user' },
+      envelope: {
+        displayContent: [{ type: 'text', text: 'rewritten request' }],
+        prefixContexts: [{
+          source: { kind: 'plugin', plugin: 'prefix' },
+          meta: { kind: 'prefix-card' },
+        }],
+      },
+    })
+    expect(log.some(event => event.type === 'context/message')).toBe(false)
+    expect(adapter.requests[0]?.messages.at(-1)).toEqual({
+      role: 'user',
+      content: [
+        { type: 'text', text: 'untrusted prefix' },
+        { type: 'text', text: '\n\n## My request:\n' },
+        { type: 'text', text: 'rewritten request' },
+      ],
+    })
+  })
+
  it('runs pre-step after prompt rewrites and injected context become durable', async () => {
    const adapter = new MockAdapter([textResponse('ok')])
    const ctx = await harness(adapter)
@@ -154,7 +203,9 @@ describe('agent/prompt-submit', () => {
    const reasons: TurnEndReason[] = []
    ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })

-    send(agent, 'do something')
+    agent.send([{ type: 'text', text: 'do something' }], {
+      contexts: [{ content: [{ type: 'text', text: 'must be dropped' }], source: { kind: 'plugin', plugin: 'test' } }],
+    })
    await waitForIdle(ctx, agent)

    // the model was never called
@@ -164,6 +215,7 @@ 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')