refactor(agent): unify sourced message delivery

packages/ui/acp/tests/config-options.spec.ts

@@ -406,7 +406,7 @@ describe('acp bridge — session config options', () => {
     const { sessionId } = await h.client.newSession({ cwd: process.cwd(), mcpServers: [] })
     const agent = h.ctx.agents.list()[0]
     if (agent === undefined) throw new Error('expected an agent')
-    agent.inject([{ type: 'text', text: 'checkpoint' }], { source: { kind: 'plugin', plugin: 'test' } })
+    agent.inject({ content: [{ type: 'text', text: 'checkpoint' }], source: { kind: 'plugin', plugin: 'test' } })
     await agent.whenIdle()
     await h.dispose()
     h = undefined

packages/ui/acp/tests/dispose.spec.ts

@@ -264,7 +264,7 @@ describe('acp bridge — disposal & HMR safety', () => {
     const handle = await harness.ctx.agents.create({
       sessionId: SessionId('guard-a'), agentOptions: { provider: 'mock', model: 'mock' },
     })
-    handle.agent.followup([{ type: 'text', text: 'go' }])
+    handle.agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
     await handle.agent.whenIdle()
     expect(harness.ctx.sessions.get(SessionId('guard-a'))).toBeDefined()
 
@@ -288,7 +288,7 @@ describe('acp bridge — disposal & HMR safety', () => {
     // Drive a turn that hangs in the model stream, so the loop is mid-turn when
     // disposed — its exit runs a final session/flush we can gate to hold the
     // teardown observably in-flight.
-    handle.agent.followup([{ type: 'text', text: 'go' }])
+    handle.agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
     await new Promise(r => setTimeout(r, 30))
     expect(handle.agent.status).toBe('running')
     // Both callers join the same teardown and observe registry removal.

packages/ui/acp/tests/edges.spec.ts

@@ -30,7 +30,7 @@ describe('acp bridge — demux & config edges', () => {
     const before = harness.updates.length
 
     const { agent: foreign } = await harness.ctx.agents.create({ sessionId: SessionId('foreign-session'), agentOptions: { provider: 'mock', model: 'mock' } })
-    foreign.followup([{ type: 'text', text: 'hi' }])
+    foreign.followup({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })
     await foreign.whenIdle()
     await new Promise(r => setTimeout(r, 10))
 

packages/ui/acp/tests/turns.spec.ts

@@ -51,11 +51,50 @@ describe('acp bridge — turn outcomes', () => {
 
   it('rejects an ordinary plugin turn failure through the same ACP boundary', async () => {
     harness = await makeBridgeHarness({ storageDir, script: [textResponse('must not run')] })
-    harness.ctx.on('agent/step', () => { throw new Error('plugin pre-step failed') })
+    harness.ctx.on('agent/step', () => { throw new Error('plugin step failed') })
     const sessionId = await newSession(harness)
 
     await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }))
-      .rejects.toThrow(/turn failed: plugin pre-step failed/)
+      .rejects.toThrow(/turn failed: plugin step failed/)
+  })
+
+  it('settles a prompt rejected during admission without opening a turn', async () => {
+    harness = await makeBridgeHarness({ storageDir, script: [textResponse('must not run')] })
+    harness.ctx.on('agent/prompt-submit', async () => ({ kind: 'block', reason: 'policy veto' }))
+    const sessionId = await newSession(harness)
+
+    await expect(harness.client.prompt({
+      sessionId,
+      prompt: [{ type: 'text', text: 'blocked' }],
+    })).resolves.toEqual({ stopReason: 'cancelled' })
+    const agent = harness.ctx.agents.get(SessionId(sessionId))
+    expect(agent?.session.events.some(event => event.type === 'turn/start')).toBe(false)
+  })
+
+  it('does not classify an asynchronous allowed admission as a no-turn rejection', async () => {
+    harness = await makeBridgeHarness({ storageDir, script: [textResponse('ok')] })
+    const entered = Promise.withResolvers<true>()
+    const release = Promise.withResolvers<true>()
+    harness.ctx.on('agent/prompt-submit', async () => {
+      entered.resolve(true)
+      await release.promise
+      return { kind: 'allow' }
+    })
+    const sessionId = await newSession(harness)
+    let settled = false
+    const prompt = harness.client.prompt({
+      sessionId,
+      prompt: [{ type: 'text', text: 'allowed' }],
+    }).then((result) => {
+      settled = true
+      return result
+    })
+
+    await entered.promise
+    await Promise.resolve()
+    expect(settled).toBe(false)
+    release.resolve(true)
+    await expect(prompt).resolves.toEqual({ stopReason: 'end_turn' })
   })
 
   it('streams a tool call as tool_call then tool_call_update', async () => {
@@ -291,7 +330,7 @@ describe('acp bridge — turn outcomes', () => {
     harness.ctx.on('agent/inbox/enqueue', (subject) => {
       if (subject === agent && !injected) {
         injected = true
-        agent.inject([{ type: 'text', text: 'ctx note' }], { source: { kind: 'plugin', plugin: 'test' } })
+        agent.inject({ content: [{ type: 'text', text: 'ctx note' }], source: { kind: 'plugin', plugin: 'test' } })
       }
     })
     const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })