fix(daemon): ACP-HTTP review round 3 — PR bot fold-ins (QwenLM#4472)

Addresses the automated PR reviews:
- B1 (P0): abort in-flight prompt on session/cancel + teardown (promptAbort
  on SessionBinding) — stops orphaned agent runs burning model quota.
- B2 (P0): thread fromLoopback (kernel remoteAddress) into sessionCtx so
  the local-only permission policy works for loopback ACP clients.
- B3 (P0): SSE write failure logs + closes the stream (no zombie streams).
- B4 (P0): sweep logs reaps; pumpSessionEvents touch()es; maxConnections
  cap (64) -> 503 on flood.
- B5 (P1): throw on missing bridge-stamped clientId (no silent fallback);
  FakeBridge stamps one.
- B6/B7 (P1): validate cwd (string/<=4096/absolute) + prompt (non-empty
  object array), mirroring the REST surface.
- B8 (P1): toRpcError maps known bridge errors to coded, client-safe shapes.
- B9 (P1): daemon-originated JSON-RPC ids are strings (_qwen_perm_N),
  collision-proof vs client ids.
- B10 (P2): resolveClientResponse param type; conn-stream onClose log;
  DELETE missing-header 400; SseStream onClose try/catch; load/resume/close
  + DELETE-400 tests.

Suite 18 -> 22 tests, all green. Re-verified live (16 session/update ->
end_turn). Base-branch acpAgent.ts typecheck findings are out of scope
(introduced by QwenLM#4353; tracked separately).

Generated with AI

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

Author: 秦奇

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

@@ -392,3 +392,31 @@ All fixes verified by the expanded vitest suite (**18 tests**) + a fresh live sm
   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.
+
+---
+
+## 12. Review round 3 — PR bot fold-ins (#4472)
+
+Two automated PR reviewers plus the summary bot.
+All fixes verified by the suite (now **22 tests**) + a fresh live run (16 `session/update` → `end_turn`).
+
+| # | Severity | Finding | Fix |
+|---|----------|---------|-----|
+| B1 | **P0** | `handlePrompt`'s `AbortController` was never aborted — a disconnecting/cancelling client left the agent running (burned model quota, blocked the session FIFO). Flagged by both bots + 5 sub-agents. | `promptAbort` parked on `SessionBinding`; aborted by `session/cancel` and by session/connection teardown (`closeSessionStream`/`destroy`). |
+| B2 | **P0** | `sessionCtx` missing `fromLoopback` → every ACP permission vote treated as remote; `local-only` policy would reject loopback clients. | Capture loopback at `initialize` (kernel `remoteAddress`, not forgeable headers) → `AcpConnection.fromLoopback` → threaded through `sessionCtx`. |
+| B3 | **P0** | SSE write failures silently swallowed → zombie streams (heartbeats fire, zero events delivered, no logs). | First write failure logs + closes the stream. |
+| B4 | **P0** | Idle sweep destroyed connections with no log + no connection cap (initialize-flood). | Sweep logs each reap; `pumpSessionEvents` calls `touch()` (long quiet prompts aren't reaped); `maxConnections` cap (64) → `503`. |
+| B5 | **P1** | `sessionCtx` silently fell back to the connection's unregistered clientId when the binding lacked one (untested, always-fired in `FakeBridge`). | Throw on missing stamped clientId (invariant violation); `FakeBridge` now stamps one. |
+| B6 | **P1** | `session/new|load|resume` accepted `cwd` unvalidated (REST validates string/length/absolute — amplification DoS). | Shared `parseOptionalWorkspaceCwd` (string, ≤4096, absolute). |
+| B7 | **P1** | `session/prompt` forwarded an unvalidated `prompt` to the bridge. | `validatePrompt` (non-empty array of objects), mirroring REST. |
+| B8 | **P1** | Raw bridge error messages echoed to the client. | `toRpcError` maps known bridge errors to coded, client-safe shapes; unknown → generic `Internal error` (full detail still to stderr). |
+| B9 | **P1** | `nextId` used sequential negatives — a client legally using negative ids could collide in `pending`. | Daemon-originated ids are now strings (`_qwen_perm_N`), disjoint from any client id. |
+| B10 | **P2** | `resolveClientResponse` param type excluded `JsonRpcError`; conn-scoped SSE stream had no `onClose`; `DELETE` with no header was a silent 202; `SseStream.close` ran `onClose` outside try/catch; `session/load`·`resume`·`close` untested. | Widened param to `JsonRpcResponse`; conn stream logs on close; `DELETE` missing header → `400`; `onClose` wrapped in try/catch; added load/resume/close + DELETE-400 tests. |
+
+**Out of scope (base-branch `daemon_mode_b_main`, not this diff)** — the second reviewer flagged
+typecheck errors in `acpAgent.ts` (`entryCount`/`entrySummary`/`sessionClose`) and other pre-existing
+items it explicitly attributed to the base branch (introduced by #4353). Tracked separately; not
+touched here.
+
+**Still deferred** (documented): per-connection secret for `DELETE`/connection ownership (token remains
+the boundary); WebSocket + HTTP/2 (§7); strict prompt-result vs trailing-update barrier (§11).

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

@@ -15,6 +15,13 @@ import type { SseStream } from './sseStream.js';
  */
 const MAX_BUFFERED_FRAMES = 256;
 
+/** Default cap on concurrent live connections (mirrors a bounded resource). */
+const DEFAULT_MAX_CONNECTIONS = 64;
+
+function logStderr(line: string): void {
+  process.stderr.write(`${line}\n`);
+}
+
 /**
  * Invoked when a session/connection tears down while an agent→client
  * request (e.g. a permission prompt) is still outstanding, so the bridge
@@ -53,6 +60,13 @@ export interface SessionBinding {
    * event-starved.
    */
   abort: AbortController;
+  /**
+   * Aborts the in-flight `session/prompt` for this session. Set by
+   * `handlePrompt` while a prompt runs; aborted on `session/cancel` and on
+   * session/connection teardown so a disconnecting client doesn't leave
+   * the agent burning model quota on a result nobody will read.
+   */
+  promptAbort?: AbortController;
 }
 
 /** An agent→client request awaiting the client's JSON-RPC response. */
@@ -77,27 +91,40 @@ export class AcpConnection {
    * connection can't drive or eavesdrop on a session it never claimed.
    */
   readonly ownedSessions = new Set<string>();
-  readonly pending = new Map<number, PendingClientRequest>();
+  /** Agent→client requests awaiting a client response, keyed by JSON-RPC id. */
+  readonly pending = new Map<string, PendingClientRequest>();
   /** Daemon-issued client id reused across this connection's bridge calls. */
   readonly clientId: string;
+  /**
+   * True when the `initialize` POST arrived from a kernel-stamped loopback
+   * peer. Threaded into per-session bridge contexts so the `local-only`
+   * permission policy can gate votes by transport — mirrors the REST
+   * surface's `detectFromLoopback(req)`. NOT derived from forgeable
+   * headers (`X-Forwarded-For` etc).
+   */
+  readonly fromLoopback: boolean;
   lastActiveMs: number = Date.now();
   private idCounter = 0;
 
   constructor(
     connectionId: string | undefined,
+    fromLoopback: boolean,
     private readonly onAbandonPending?: AbandonPendingFn,
   ) {
     this.connectionId = connectionId ?? randomUUID();
     this.clientId = randomUUID();
+    this.fromLoopback = fromLoopback;
   }
 
-  /** Allocate a fresh JSON-RPC id for an agent→client request. */
-  nextId(): number {
-    // Negative ids keep our outbound request ids disjoint from the
-    // client's (clients conventionally use positive ids), so a client
-    // that echoes ids can't collide with our permission requests.
-    this.idCounter -= 1;
-    return this.idCounter;
+  /**
+   * Allocate a fresh JSON-RPC id for an agent→client request. STRING-typed
+   * (`_qwen_perm_N`) so it can never collide with a client-originated id —
+   * JSON-RPC 2.0 permits clients to use any number (incl. negatives) or
+   * string, so a numeric namespace wasn't actually safe.
+   */
+  nextId(): string {
+    this.idCounter += 1;
+    return `_qwen_perm_${this.idCounter}`;
   }
 
   touch(): void {
@@ -173,6 +200,7 @@ export class AcpConnection {
     const binding = this.sessions.get(sessionId);
     if (!binding) return;
     binding.abort.abort();
+    binding.promptAbort?.abort();
     binding.stream?.close();
     this.abandonPendingForSession(sessionId, binding.clientId);
     this.sessions.delete(sessionId);
@@ -182,6 +210,7 @@ export class AcpConnection {
   destroy(): void {
     for (const binding of this.sessions.values()) {
       binding.abort.abort();
+      binding.promptAbort?.abort();
       binding.stream?.close();
       this.abandonPendingForSession(binding.sessionId, binding.clientId);
     }
@@ -220,14 +249,23 @@ export class ConnectionRegistry {
 
   constructor(
     private readonly onAbandonPending?: AbandonPendingFn,
+    private readonly maxConnections = DEFAULT_MAX_CONNECTIONS,
     private readonly idleTtlMs = 30 * 60_000,
   ) {
     this.sweepTimer = setInterval(() => this.sweep(), 60_000);
     this.sweepTimer.unref();
   }
 
-  create(): AcpConnection {
-    const conn = new AcpConnection(undefined, this.onAbandonPending);
+  /**
+   * Mint a connection, or return `undefined` when the live-connection cap
+   * is reached (the caller answers `503`). Bounds an `initialize` flood from
+   * growing the registry without limit through the full TTL window.
+   */
+  create(fromLoopback: boolean): AcpConnection | undefined {
+    if (this.maxConnections > 0 && this.byId.size >= this.maxConnections) {
+      return undefined;
+    }
+    const conn = new AcpConnection(undefined, fromLoopback, this.onAbandonPending);
     this.byId.set(conn.connectionId, conn);
     return conn;
   }
@@ -258,7 +296,17 @@ export class ConnectionRegistry {
   private sweep(): void {
     const cutoff = Date.now() - this.idleTtlMs;
     for (const [id, conn] of this.byId) {
-      if (conn.lastActiveMs < cutoff) this.delete(id);
+      if (conn.lastActiveMs >= cutoff) continue;
+      // Observability: a reaped connection silently dropping its SSE
+      // streams is otherwise invisible to operators chasing "my client
+      // froze". Note that `touch()` fires on inbound HTTP AND on event
+      // delivery (pumpSessionEvents), so a long quiet prompt isn't reaped.
+      logStderr(
+        `qwen serve: /acp reaping idle connection ${id} ` +
+          `(idle > ${Math.round(this.idleTtlMs / 60_000)}m, ` +
+          `${conn.sessions.size} session(s))`,
+      );
+      this.delete(id);
     }
   }
 }