fix(daemon): ACP-HTTP review round 2 — reconnect, ownership, leaks, negotiation

Folds in two independent reviews (correctness/concurrency + conformance/
security):
- R1 (P0): session-stream reconnect was permanently dead (reused aborted
  AbortController) — fresh controller per stream + close prior stream.
- R2 (P0): unhandled rejection from handle()/pumpSessionEvents after
  res.end(202) — wrapped in catch; isResponse path now try/caught.
- R3 (P1): connection→session ownership (ownedSessions) — session stream
  403s and per-session POSTs INVALID_PARAMS for unowned ids.
- R4 (P1): dispose() wired into runQwenServe shutdown (was discarded).
- R5 (P1): cancel pending permissions on session/connection teardown.
- R6 (P1): cap pre-attach SSE buffers at 256 (drop-oldest).
- R7/R8/R9 (P2): protocolVersion negotiation; Acp-Session-Id vs
  params.sessionId cross-check; session/cancel request-form reply; de-dup
  _meta.qwen.

Tests expanded to 18 (ownership 403, unowned-prompt, header mismatch).
Re-verified live: 21 session/update frames -> stopReason=end_turn.

Generated with AI

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

Author: 秦奇

docs/design/daemon-acp-http/README.md

@@ -360,3 +360,35 @@ clientId the bridge has never issued and stamp a fresh one (returned in
 (unregistered) id on `sendPrompt`. Fix: persist the bridge-stamped id on the
 `SessionBinding` and echo it on every per-session call (`sessionCtx`). Re-verified
 green above.
+
+---
+
+## 11. Review round 2 — fold-ins
+
+Two independent reviews (correctness/concurrency + protocol-conformance/security) plus a self-read.
+All fixes verified by the expanded vitest suite (**18 tests**) + a fresh live smoke run
+(21 `session/update` frames → `stopReason=end_turn`).
+
+| # | Severity | Finding | Fix |
+|---|----------|---------|-----|
+| R1 | **P0** | Session-stream **reconnect was permanently dead**: `SessionBinding.abort` was created once and reused; on stream close it was aborted forever, so a reconnect's `subscribeEvents(signal)` got an already-aborted signal and received zero events. | `attachSessionStream` now installs a **fresh** `AbortController` per stream (and closes any prior stream); `index.ts` pumps on that fresh signal. |
+| R2 | **P0** | `await dispatcher.handle()` ran **after** `res.end(202)`; a throwing bridge call (notably the un-try/caught `isResponse` path) would reject and surface as an unhandled rejection → possible daemon crash. | Wrapped the `isResponse` path in try/catch; `.catch()` on the awaited `handle(...)` and on `pumpSessionEvents(...)`. |
+| R3 | **P1** | **No connection→session ownership**: any authenticated connection could open the session SSE for, or prompt, *any* sessionId in the workspace (read-eavesdrop; prompt was only blocked incidentally by the unregistered-clientId error). | `AcpConnection.ownedSessions` populated by `session/new`/`load`/`resume`; session stream returns `403` and per-session POSTs return `INVALID_PARAMS` for unowned ids (`requireOwned`). |
+| R4 | **P1** | `mountAcpHttp` handle was discarded → TTL sweep timer + live SSE streams leaked on shutdown. | Handle parked on `app.locals`; `runQwenServe` close hook calls `dispose()` before `bridge.shutdown()` (mirrors the device-flow registry). |
+| R5 | **P1** | **Pending permission leak**: closing a session/connection with a permission outstanding left the bridge blocked awaiting a vote. | `closeSessionStream`/`destroy` cancel matching pending requests via an injected `onAbandonPending` → `cancelAbandonedPermission`. |
+| R6 | **P1** | Pre-attach frame buffers (`connBuffer`/`binding.buffer`) were unbounded. | Capped at 256 frames (drop-oldest), matching the EventBus `maxQueued`. |
+| R7 | **P2** | `initialize` ignored the client's requested `protocolVersion`. | Negotiates `min(requested, 1)`. |
+| R8 | **P2** | No `Acp-Session-Id` ↔ `params.sessionId` cross-check (RFD §2.3). | POST asserts they agree; mismatch → `INVALID_PARAMS`. |
+| R9 | **P2** | `session/cancel` request-form (with id) never answered; duplicate top-level `_meta.qwen`. | Reply when an id is present; single `agentCapabilities._meta.qwen`. |
+
+### Accepted / documented (not fixed in v1)
+
+- **Prompt-result vs trailing `session/update` ordering** (P2): `handlePrompt` awaits `sendPrompt` then
+  writes the result frame, while updates stream concurrently. In practice the bridge publishes all
+  `session/update`s to the bus before `sendPrompt` resolves and both share one ordered SSE write
+  chain, so the result lands last (confirmed: 21 updates then result). A strict barrier is a possible
+  later hardening if a client reducer proves sensitive.
+- **Browser `EventSource` can't set `Authorization`** — `/acp` GET streams require the bearer header,
+  so browsers need the deferred WebSocket path (§7); CLI/Node clients are unaffected.
+- The daemon's real trust boundary remains the **bearer token + single-workspace bind** (same as the
+  REST surface); R3's ownership check is defense-in-depth + contract correctness, not a tenant boundary.

packages/cli/src/serve/acpHttp/connectionRegistry.ts

@@ -7,6 +7,24 @@
 import { randomUUID } from 'node:crypto';
 import type { SseStream } from './sseStream.js';
 
+/**
+ * Per-stream cap on frames buffered before the client attaches its SSE
+ * stream. Mirrors the EventBus's `maxQueued` backpressure cap so a client
+ * that drives requests without ever opening a stream can't grow daemon
+ * memory without bound. Oldest frames are dropped past the cap.
+ */
+const MAX_BUFFERED_FRAMES = 256;
+
+/**
+ * Invoked when a session/connection tears down while an agent→client
+ * request (e.g. a permission prompt) is still outstanding, so the bridge
+ * isn't left blocked awaiting a vote that will never arrive.
+ */
+export type AbandonPendingFn = (
+  req: PendingClientRequest,
+  clientId: string | undefined,
+) => void;
+
 /**
  * Tracks one logical ACP-over-HTTP connection (RFD #721). A connection is
  * minted at `initialize`, keyed by `Acp-Connection-Id`, and may host many
@@ -27,7 +45,13 @@ export interface SessionBinding {
   stream?: SseStream;
   /** Frames emitted before the session stream attached, flushed on attach. */
   buffer: unknown[];
-  /** Aborts the bridge event subscription when the stream/connection closes. */
+  /**
+   * Aborts the bridge event subscription tied to the CURRENT session
+   * stream. Replaced with a fresh controller on every re-attach — a
+   * controller, once aborted (on stream close), can never resume, so
+   * reusing it across reconnects would leave the new stream permanently
+   * event-starved.
+   */
   abort: AbortController;
 }
 
@@ -46,13 +70,23 @@ export class AcpConnection {
   /** Frames emitted before the connection stream attached, flushed on attach. */
   private readonly connBuffer: unknown[] = [];
   readonly sessions = new Map<string, SessionBinding>();
+  /**
+   * Sessions this connection created (`session/new`) or explicitly
+   * attached to (`session/load`/`resume`). Per-session operations
+   * (subscribe, prompt, cancel, …) are gated on membership here so one
+   * connection can't drive or eavesdrop on a session it never claimed.
+   */
+  readonly ownedSessions = new Set<string>();
   readonly pending = new Map<number, PendingClientRequest>();
   /** Daemon-issued client id reused across this connection's bridge calls. */
   readonly clientId: string;
   lastActiveMs: number = Date.now();
   private idCounter = 0;
 
-  constructor(connectionId?: string) {
+  constructor(
+    connectionId: string | undefined,
+    private readonly onAbandonPending?: AbandonPendingFn,
+  ) {
     this.connectionId = connectionId ?? randomUUID();
     this.clientId = randomUUID();
   }
@@ -70,6 +104,14 @@ export class AcpConnection {
     this.lastActiveMs = Date.now();
   }
 
+  ownSession(sessionId: string): void {
+    this.ownedSessions.add(sessionId);
+  }
+
+  ownsSession(sessionId: string): boolean {
+    return this.ownedSessions.has(sessionId);
+  }
+
   getOrCreateSession(sessionId: string): SessionBinding {
     let binding = this.sessions.get(sessionId);
     if (!binding) {
@@ -84,12 +126,15 @@ export class AcpConnection {
     if (this.connStream && !this.connStream.isClosed) {
       void this.connStream.send(frame);
     } else {
-      this.connBuffer.push(frame);
+      pushCapped(this.connBuffer, frame);
     }
   }
 
   /** Attach the connection-scoped stream and flush any buffered frames. */
   attachConnStream(stream: SseStream): void {
+    // Close any prior connection stream so its heartbeat interval + socket
+    // don't leak when a client reconnects the connection-scoped GET.
+    if (this.connStream && this.connStream !== stream) this.connStream.close();
     this.connStream = stream;
     for (const frame of this.connBuffer.splice(0)) void stream.send(frame);
   }
@@ -100,13 +145,25 @@ export class AcpConnection {
     if (binding.stream && !binding.stream.isClosed) {
       void binding.stream.send(frame);
     } else {
-      binding.buffer.push(frame);
+      pushCapped(binding.buffer, frame);
     }
   }
 
-  /** Attach a session-scoped stream and flush any buffered frames. */
-  attachSessionStream(sessionId: string, stream: SseStream): SessionBinding {
+  /**
+   * Attach a session-scoped stream: close any prior stream, abort the prior
+   * subscription, install the caller's FRESH AbortController (the old one is
+   * aborted and can never resume — reusing it would leave the new stream
+   * event-starved), flush buffered frames, and return the binding.
+   */
+  attachSessionStream(
+    sessionId: string,
+    stream: SseStream,
+    abort: AbortController,
+  ): SessionBinding {
     const binding = this.getOrCreateSession(sessionId);
+    if (binding.stream && binding.stream !== stream) binding.stream.close();
+    binding.abort.abort();
+    binding.abort = abort;
     binding.stream = stream;
     for (const frame of binding.buffer.splice(0)) void stream.send(frame);
     return binding;
@@ -117,18 +174,39 @@ export class AcpConnection {
     if (!binding) return;
     binding.abort.abort();
     binding.stream?.close();
+    this.abandonPendingForSession(sessionId, binding.clientId);
     this.sessions.delete(sessionId);
+    this.ownedSessions.delete(sessionId);
   }
 
   destroy(): void {
     for (const binding of this.sessions.values()) {
       binding.abort.abort();
       binding.stream?.close();
+      this.abandonPendingForSession(binding.sessionId, binding.clientId);
     }
     this.sessions.clear();
+    this.ownedSessions.clear();
     this.pending.clear();
     this.connStream?.close();
   }
+
+  /** Cancel + drop any pending agent→client requests for a closing session. */
+  private abandonPendingForSession(
+    sessionId: string,
+    clientId: string | undefined,
+  ): void {
+    for (const [id, req] of this.pending) {
+      if (req.sessionId !== sessionId) continue;
+      this.pending.delete(id);
+      this.onAbandonPending?.(req, clientId);
+    }
+  }
+}
+
+function pushCapped(buf: unknown[], frame: unknown): void {
+  if (buf.length >= MAX_BUFFERED_FRAMES) buf.shift();
+  buf.push(frame);
 }
 
 /**
@@ -140,13 +218,16 @@ export class ConnectionRegistry {
   private readonly byId = new Map<string, AcpConnection>();
   private readonly sweepTimer: ReturnType<typeof setInterval>;
 
-  constructor(private readonly idleTtlMs = 30 * 60_000) {
+  constructor(
+    private readonly onAbandonPending?: AbandonPendingFn,
+    private readonly idleTtlMs = 30 * 60_000,
+  ) {
     this.sweepTimer = setInterval(() => this.sweep(), 60_000);
     this.sweepTimer.unref();
   }
 
   create(): AcpConnection {
-    const conn = new AcpConnection();
+    const conn = new AcpConnection(undefined, this.onAbandonPending);
     this.byId.set(conn.connectionId, conn);
     return conn;
   }