fix(acp): accept baseline resource links and release a failed prompt slot

ACP v1 requires every agent to accept text AND resource_link prompt
content; the automation rewrite dropped the resource_link half of that
baseline. Restore the old bracketed-reference flattening in the codec,
reject only beyond-baseline blocks, and update the package contract and
Agent Note.

Also release the per-session prompt slot when agent.send() throws
synchronously (an agent disposed outside the bridge would otherwise
wedge the session into permanent 'already in flight' rejections),
drop the tautological version-negotiation branch, prove the scenario
env layer reaches the snapshot subprocess, pin bridge-side fail-closed
permission errors, and correct two overpromising test names.

packages/acp/acp/README.md

@@ -19,10 +19,10 @@ Both fields are optional so another agent/request listener may supply the target
 
 | Method | Behavior |
 |---|---|
-| `initialize` | Negotiates the supported version and advertises text-only prompts. No session, editor, terminal, filesystem, or MCP capability is advertised. |
+| `initialize` | Negotiates the supported version and advertises baseline-only prompts (no image, audio, or embedded-context capability). No session, editor, terminal, filesystem, or MCP capability is advertised. |
 | `authenticate` | No-op because the server advertises no authentication methods. |
 | `session/new` | Creates a fresh agent with an absolute primary `cwd`; empty `additionalDirectories` and `mcpServers` are accepted, non-empty values reject. |
-| `session/prompt` | Concatenates text blocks, rejects empty or non-text input, permits one in-flight request per session, and settles from that request's owning durable `turn/end`. |
+| `session/prompt` | Concatenates text blocks, renders baseline resource links as bracketed textual references, rejects empty or beyond-baseline input, permits one in-flight request per session, and settles from that request's owning durable `turn/end`. |
 | `session/cancel` | Cancels only the addressed agent and settles its pending prompt as `cancelled`; unknown ids are no-ops. |
 | `session/update` | Emits one `agent_message_chunk` per non-empty text block in a committed `assistant/message`. Raw deltas and non-message events are omitted. |
 | `session/request_permission` | Offers one-shot allow/reject choices for bridge-owned approval requests carrying a tool call id. Clients may answer automatically. |
@@ -45,7 +45,7 @@ Client disconnect and Cordis disposal share one memoized teardown. The bridge fi
 
 #### What the model sees
 
-`session/prompt` text blocks are concatenated verbatim into one user message. Protocol metadata, client capabilities, permission choices, and session ids never enter the model request.
+`session/prompt` text blocks are concatenated verbatim into one user message; a baseline resource link appears in that message as a bracketed `[resource_link name=… uri=…]` reference the model may open with its own tools. Protocol metadata, client capabilities, permission choices, and session ids never enter the model request.
 
 #### Token effect
 
@@ -72,6 +72,6 @@ Append-only through the owning tool result.
 ## Known Limitations and Deferred Work
 
 - **Fresh sessions only** — load, list, resume, delete, and fork are unsupported.
-- **Text and one workspace only** — resource links, images, audio, embedded resources, non-empty additional directories, and MCP servers reject.
+- **Baseline prompts and one workspace only** — images, audio, embedded resources, non-empty additional directories, and MCP servers reject; resource links flatten to textual references rather than fetched content.
 - **Committed answers only** — live progress, reasoning, tool activity, plans, titles, and usage stay off the wire.
 - **Connection-owned lifetime** — one connection releases all of its sessions; per-session close is not implemented.

packages/acp/acp/src/codec.ts

@@ -31,19 +31,33 @@ export function turnEndToStopReason(reason: TurnEndReason): StopReason {
 }
 
 /**
- * Concatenate an ACP prompt's text blocks.
+ * Flatten an ACP prompt's baseline blocks to text. Text blocks concatenate
+ * verbatim; resource links become explicit textual references so a baseline
+ * client can point at files without the bridge silently dropping that context.
  * @param prompt - supported ACP prompt blocks.
- * @returns text in wire order.
+ * @returns text in wire order, with resource links rendered as bracketed references.
  */
 export function acpPromptToText(prompt: readonly AcpContentBlock[]): string {
-  return prompt.flatMap(block => block.type === 'text' ? [block.text] : []).join('')
+  return prompt.flatMap((block): string[] => {
+    switch (block.type) {
+      case 'text':
+        return [block.text]
+      case 'resource_link':
+        return [`\n[resource_link name=${JSON.stringify(block.name)} uri=${JSON.stringify(block.uri)}]\n`]
+      default:
+        return []
+    }
+  }).join('')
 }
 
 /**
- * Whether a prompt asks the automation bridge to carry non-text content.
+ * Whether a prompt carries content beyond the ACP baseline. The spec requires
+ * every agent to accept `text` and `resource_link`; richer inline payloads
+ * (image, audio, embedded resource) are optional capabilities this bridge does
+ * not advertise, so they are rejected rather than silently dropped.
  * @param prompt - ACP prompt blocks to inspect.
- * @returns `true` when any block is not text.
+ * @returns `true` when any block is neither `text` nor `resource_link`.
  */
 export function promptHasUnsupportedContent(prompt: readonly AcpContentBlock[]): boolean {
-  return prompt.some(block => block.type !== 'text')
+  return prompt.some(block => block.type !== 'text' && block.type !== 'resource_link')
 }

packages/acp/acp/src/index.ts

@@ -192,12 +192,11 @@ export function apply(ctx: Context, config: AcpConfig): void {
   const makeAgent = (connection: AgentSideConnection): AcpAgent => {
     conn = connection
     return {
-      initialize(params: InitializeRequest): Promise<InitializeResponse> {
-        const protocolVersion = params.protocolVersion === PROTOCOL_VERSION
-          ? params.protocolVersion
-          : PROTOCOL_VERSION
+      initialize(_params: InitializeRequest): Promise<InitializeResponse> {
+        // Single-version agent: per spec, answer the latest version this
+        // server supports regardless of the client's requested version.
         return Promise.resolve({
-          protocolVersion,
+          protocolVersion: PROTOCOL_VERSION,
           agentInfo: { name: 'deepseek-harness-acp', version: '0.0.1' },
           agentCapabilities: {
             promptCapabilities: { image: false, audio: false, embeddedContext: false },
@@ -239,14 +238,28 @@ export function apply(ctx: Context, config: AcpConfig): void {
           throw invalidParams('a prompt is already in flight for this session')
         }
         if (promptHasUnsupportedContent(params.prompt)) {
-          throw invalidParams('only text prompt content is supported')
+          throw invalidParams('only text and resource_link prompt content is supported')
         }
         const text = acpPromptToText(params.prompt)
         if (text.trim().length === 0) throw invalidParams('empty prompt')
 
         const stopReason = await new Promise<StopReason>((resolve, reject) => {
+          // Arm the slot before send() so a listener-driven synchronous turn
+          // cannot slip past correlation; a synchronous send() failure (an
+          // agent disposed outside the bridge, e.g. an agent-loop-only reload)
+          // must free the slot again or the session would reject every later
+          // prompt as already in flight.
           record.inflight = { resolve, reject, turn: undefined }
-          record.agent.send([{ type: 'text', text }])
+          try {
+            record.agent.send([{ type: 'text', text }])
+          } catch (error: unknown) {
+            record.inflight = undefined
+            // send() throws only Errors (disposed agent / invalid input); the
+            // String arm is a defensive fallback for a non-Error throw.
+            /* v8 ignore next */
+            const detail = error instanceof Error ? error.message : String(error)
+            throw internalError(`prompt was not queued: ${detail}`)
+          }
         })
         return { stopReason }
       },

packages/acp/acp/tests/approval.spec.ts

@@ -50,6 +50,13 @@ describe('ACP machine permission policy', () => {
     await expect(harness.ctx.approval.request(request)).resolves.toBe('rejected')
   })
 
+  it('fails closed when the client errors the permission request', async () => {
+    harness = await makeBridgeHarness()
+    const request = await ownedRequest()
+    harness.onPermission = () => { throw new Error('client gone') }
+    await expect(harness.ctx.approval.request(request)).resolves.toBe('unavailable')
+  })
+
   it('delegates a same-id foreign agent', async () => {
     harness = await makeBridgeHarness()
     const request = await ownedRequest()

packages/acp/acp/tests/bridge.spec.ts

@@ -107,7 +107,7 @@ describe('automation-only ACP bridge', () => {
     })).resolves.toHaveProperty('sessionId')
   })
 
-  it('rejects empty and non-text prompts before a turn starts', async () => {
+  it('rejects empty and beyond-baseline prompts before a turn starts', async () => {
     harness = await makeBridgeHarness()
     await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
     const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
@@ -116,11 +116,28 @@ describe('automation-only ACP bridge', () => {
       .rejects.toThrow(/empty prompt/)
     await expect(harness.client.prompt({
       sessionId,
-      prompt: [{ type: 'resource_link', name: 'file', uri: 'file:///tmp/a' }],
-    })).rejects.toThrow(/only text/)
+      prompt: [{ type: 'image', data: '', mimeType: 'image/png' }],
+    })).rejects.toThrow(/only text and resource_link/)
     expect(harness.ctx.agents.get(SessionId(sessionId))?.session.events.some(event => event.type === 'turn/start')).toBe(false)
   })
 
+  it('renders baseline resource links as textual references in the user message', async () => {
+    harness = await makeBridgeHarness({ script: [textResponse('done')] })
+    await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
+    const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })
+    await harness.client.prompt({
+      sessionId,
+      prompt: [
+        { type: 'text', text: 'summarize' },
+        { type: 'resource_link', name: 'notes.txt', uri: 'file:///tmp/notes.txt' },
+      ],
+    })
+    expect(harness.adapter.requests[0]?.messages.at(-1)?.content).toEqual([{
+      type: 'text',
+      text: 'summarize\n[resource_link name="notes.txt" uri="file:///tmp/notes.txt"]\n',
+    }])
+  })
+
   it('rejects prompts for unknown sessions and ignores unknown cancellation', async () => {
     harness = await makeBridgeHarness()
     await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })

packages/acp/acp/tests/codec.spec.ts

@@ -20,10 +20,19 @@ describe('ACP automation codec', () => {
     expect(turnEndToStopReason({ kind: 'future' } as unknown as TurnEndReason)).toBe('end_turn')
   })
 
-  it('concatenates text and rejects every non-text block', () => {
+  it('flattens baseline blocks and rejects everything richer', () => {
     expect(acpPromptToText([{ type: 'text', text: 'a' }, { type: 'text', text: 'b' }])).toBe('ab')
-    expect(acpPromptToText([{ type: 'resource_link', name: 'x', uri: 'file:///x' }])).toBe('')
-    expect(promptHasUnsupportedContent([{ type: 'text', text: 'ok' }])).toBe(false)
-    expect(promptHasUnsupportedContent([{ type: 'resource_link', name: 'x', uri: 'file:///x' }])).toBe(true)
+    expect(acpPromptToText([
+      { type: 'text', text: 'see' },
+      { type: 'resource_link', name: 'x', uri: 'file:///x' },
+    ])).toBe('see\n[resource_link name="x" uri="file:///x"]\n')
+    expect(acpPromptToText([{ type: 'image', data: '', mimeType: 'image/png' }])).toBe('')
+    expect(promptHasUnsupportedContent([
+      { type: 'text', text: 'ok' },
+      { type: 'resource_link', name: 'x', uri: 'file:///x' },
+    ])).toBe(false)
+    expect(promptHasUnsupportedContent([
+      { type: 'image', data: '', mimeType: 'image/png' },
+    ])).toBe(true)
   })
 })

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

@@ -54,7 +54,11 @@ describe('ACP automation output boundary', () => {
     expect(harness.updates).toHaveLength(0)
   })
 
-  it('contains client update failures without changing prompt settlement', async () => {
+  // `session/update` is a JSON-RPC notification, so a client-side handler
+  // failure never reaches the bridge; this pins that the prompt still settles
+  // normally with such a client. The bridge's own write-failure guard is
+  // transport-level and documented untestable at `notify`.
+  it('settles the prompt normally when the client rejects update notifications', async () => {
     harness = await makeBridgeHarness({ script: [textResponse('answer')] })
     await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} })
     const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] })

packages/acp/acp/tests/harness.ts

@@ -101,6 +101,8 @@ export interface BridgeHarness {
   closeClientTransport: () => Promise<void>
   abortClientTransport: () => Promise<void>
   acpFiber: Awaited<ReturnType<Context['plugin']>>
+  /** The AgentLoop fiber, so a test can reload the loop out from under the bridge. */
+  loopFiber: Awaited<ReturnType<Context['plugin']>>
   dispose: () => Promise<void>
 }
 
@@ -115,7 +117,7 @@ export async function makeBridgeHarness(options: {
   const adapter = new MockAdapter(options.script ?? [])
   const ctx = new Context()
   await mountAgentLoopTestDependencies(ctx, { systemPrompt: { persona: options.persona ?? '' } })
-  await ctx.plugin(AgentLoop, { agents: [] })
+  const loopFiber = await ctx.plugin(AgentLoop, { agents: [] })
   ctx.llm.registerAdapter(['mock'], adapter)
 
   const agentToClient = new TransformStream<Uint8Array, Uint8Array>()
@@ -140,6 +142,7 @@ export async function makeBridgeHarness(options: {
     onSessionUpdateError: undefined,
     client: undefined as unknown as ClientSideConnection,
     acpFiber: undefined as unknown as BridgeHarness['acpFiber'],
+    loopFiber,
     closeClientTransport: async () => { await clientToAgentWriter.close() },
     abortClientTransport: async () => { await clientToAgentWriter.abort(new Error('client transport failed')) },
     dispose: async () => { await ctx.fiber.dispose() },

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

@@ -109,6 +109,19 @@ describe('ACP prompt lifecycle', () => {
     await expect(prompt).resolves.toEqual({ stopReason: 'cancelled' })
   })
 
+  it('frees the prompt slot when the agent rejects the send synchronously', async () => {
+    harness = await makeBridgeHarness({ script: [] })
+    const sessionId = await newSession(harness)
+    // Reload the loop out from under the bridge: its agents dispose while the
+    // bridge record survives, so the next send() throws synchronously.
+    await harness.loopFiber.dispose()
+    await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'one' }] }))
+      .rejects.toThrow(/prompt was not queued/)
+    // The failed prompt must not wedge the session's single prompt slot.
+    await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'two' }] }))
+      .rejects.toThrow(/prompt was not queued/)
+  })
+
   it('permits only one in-flight prompt per session', async () => {
     harness = await makeBridgeHarness({ script: ['hang'] })
     const sessionId = await newSession(harness)

packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts

@@ -149,6 +149,8 @@ async function handlePrompt(id: number | string): Promise<void> {
       override: process.env.DSH_SNAPSHOT_OVERRIDE ?? null,
       childFiles: process.env.DSH_SNAPSHOT_CHILD_FILES ?? null,
       spillRoot: process.env.DSH_SNAPSHOT_SPILL_ROOT ?? null,
+      // Scenario-supplied deployment env (the `Scenario.env` layering seam).
+      permissionMode: process.env.DSH_PERMISSION_MODE ?? null,
     })}`)
   }
   if (behavior.echoWorkspace === true) {

packages/support/acp-snapshot/tests/suite.spec.ts

@@ -131,6 +131,8 @@ describe('defineAcpSnapshotSuite: refresh write-back', () => {
     expect(stdout).not.toContain('stale stdout')
     expect(stdout).toContain('env:{\\"mode\\":\\"replay\\"')
     expect(stdout).not.toContain('\\"mode\\":\\"refresh\\"')
+    // The scenario's own env layer reached the subprocess.
+    expect(stdout).toContain('\\"permissionMode\\":\\"never\\"')
 
     const blocked = readFileSync(join(refreshDir, 'blocked-log', 'session.jsonl'), 'utf8')
     expect(blocked).toContain('"decision":"block"')