feat(coder): v2.6 Phase 2 — warm ACP backend for goose/qwen

WarmAcpBackend (AgentBackend) holds one persistent goose acp / qwen --acp child + ClientSideConnection + ACP session per (chat,agent); initialize+session/new once, reused across turns. Abort = session/cancel the prompt only (never kills the child); child exit -> agent_sessions.status='crashed' -> re-spawn next turn. Dispatcher routes goose/qwen chat-tab tasks to the pooled warm backend via pure shouldUseWarmBackend (needs session_id+chat_id); one-shot runExternalAgent kept as fallback for arena/MCP/new_task. handleSessionUpdate extracted to a shared pure acp-event-map.ts (one-shot path byte-identical). SDK: installed @agentclientprotocol/sdk@^0.22.1 has stable resumeSession/loadSession; resume moot in the warm hot path, deferred to Phase 3. 15 new tests (warm-acp-routing, acp-event-map); 180 coder tests pass; tsc + build clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/coder/src/services/acp-event-map.ts

@@ -0,0 +1,68 @@
+/**
+ * Shared ACP session-update → normalized AgentEvent mapping.
+ *
+ * Extracted verbatim (v2.6 Phase 2) from `AcpStreamContext.handleSessionUpdate`
+ * in `acp-dispatch.ts` so the warm ACP backend (`backends/warm-acp.ts`) and the
+ * one-shot dispatch share ONE mapping. The one-shot path translates the returned
+ * events into broker frames itself (preserving its prior behavior byte-for-byte);
+ * the warm backend forwards them to the dispatcher's `ctx.onEvent` exactly like
+ * the opencode-server backend does. No I/O, no broker — pure, so it's unit-testable.
+ *
+ * Spec: openspec/changes/v2-6-persistent-agent-sessions/design.md §2b.
+ */
+import type { SessionNotification } from '@agentclientprotocol/sdk';
+import type { AgentEvent } from './agent-backend.js';
+import { type AcpToolSnapshot, mergeToolSnapshot } from './acp-tool-snapshot.js';
+
+/**
+ * Map one ACP `session/update` notification to zero-or-more normalized AgentEvents.
+ *
+ * `priorSnapshots` is the caller-owned tool-call snapshot accumulator (toolCallId →
+ * snapshot). For `tool_call` / `tool_call_update` the merged snapshot is written
+ * back into it (mutated in place, mirroring `AcpStreamContext.handleToolUpdate`)
+ * so a later `tool_call_update` merges over the earlier `tool_call`. Pass an empty
+ * Map for a stateless single call.
+ *
+ * Returns an array (never throws) so the caller can splat it onto `onEvent`.
+ */
+export function mapSessionUpdate(
+  params: SessionNotification,
+  priorSnapshots: Map<string, AcpToolSnapshot> = new Map(),
+): AgentEvent[] {
+  const update = params.update;
+  switch (update.sessionUpdate) {
+    case 'agent_message_chunk': {
+      const content = update.content;
+      if (content.type === 'text' && 'text' in content) {
+        return [{ type: 'text', text: (content as { text: string }).text }];
+      }
+      return [];
+    }
+    case 'agent_thought_chunk': {
+      const content = update.content;
+      if (content.type === 'text' && 'text' in content) {
+        return [{ type: 'reasoning', text: (content as { text: string }).text }];
+      }
+      return [];
+    }
+    case 'tool_call': {
+      const snapshot = mergeToolSnapshot(update.toolCallId, update, priorSnapshots.get(update.toolCallId));
+      priorSnapshots.set(update.toolCallId, snapshot);
+      return [{ type: 'tool_call', toolCall: snapshot }];
+    }
+    case 'tool_call_update': {
+      const snapshot = mergeToolSnapshot(update.toolCallId, update, priorSnapshots.get(update.toolCallId));
+      priorSnapshots.set(update.toolCallId, snapshot);
+      return [{ type: 'tool_update', toolCall: snapshot }];
+    }
+    case 'available_commands_update': {
+      const commands = update.availableCommands.map((cmd) => ({
+        name: cmd.name,
+        description: cmd.description ?? undefined,
+      }));
+      return [{ type: 'commands', commands }];
+    }
+    default:
+      return [];
+  }
+}