diff --git a/docs/design/daemon-acp-http/sse-resumable-stream.md b/docs/design/daemon-acp-http/sse-resumable-stream.md new file mode 100644 index 00000000000..290e5cec8e9 --- /dev/null +++ b/docs/design/daemon-acp-http/sse-resumable-stream.md @@ -0,0 +1,229 @@ +# ACP-over-HTTP — Resumable session event stream (`Last-Event-ID`) + +> Status: design + implementation in this PR. +> Closes the resumability gap tracked as RFD Phase 4 in +> [`README.md`](./README.md) §7 / row "Resume cursor (ring `Last-Event-ID`)". + +## Problem + +The `/acp` Streamable-HTTP session event stream (`GET /acp` with an +`Acp-Session-Id` header) is **live-only**: it neither emits an SSE `id:` +sequence nor honours a `Last-Event-ID` request header on reconnect. + +When a control-plane proxy idle-closes the long-lived SSE connection +mid-turn (the daemon itself sends `retry: 3000`, and ingress proxies cut +long SSE frequently), the client reconnects and re-claims ownership, but +**every content frame the daemon produced during the gap is lost** — +`session/update` notifications carrying `agent_thought_chunk` / +`agent_message_chunk`. The turn still reaches a terminal state (a +`turn_complete` is produced / synthesised), so the UI shows "done" with an +empty or truncated body. Re-sending the same prompt works, which is the +tell: the loss is in the transport gap, not the model. + +Symptom and field evidence are catalogued in the integration notes as +**§1.8** (`sdk-known-issues.md`). + +## What already exists (and why this is small) + +The replay engine is **already built and battle-tested** — the gap is only +that the `/acp` transport is not wired to it. + +`packages/acp-bridge/src/eventBus.ts`: + +- Monotonic per-session `id`, starting at 1 (`nextId`, assigned in + `publish()`). +- Bounded ring buffer per session (`DEFAULT_RING_SIZE = 8000`, operator + override `qwen serve --event-ring-size`). +- `subscribeEvents(sessionId, { lastEventId, signal })` replays ring frames + with `id > lastEventId` before live events flow, and emits the synthetic + control frames `replay_complete`, `state_resync_required` (ring-evicted / + epoch reset on daemon restart), `client_evicted`, `slow_client_warning`. + +The **REST** surface `GET /session/:id/events` already consumes all of +this: it reads `last-event-id` (`server.ts` → `parseLastEventId`), passes +it to `subscribeEvents`, and serialises each frame with an SSE `id:` line +(`formatSseFrame`). The bug is that the **`/acp` transport** does none of +this: + +| Layer | REST `/session/:id/events` | `/acp` GET (today) | +| ----------------------------------------- | -------------------------- | --------------------------------------------- | +| reads `Last-Event-ID` header | yes | **no** | +| passes `lastEventId` to `subscribeEvents` | yes | **no** (`dispatch.ts pumpSessionEvents`) | +| emits SSE `id:` line | yes (`formatSseFrame`) | **no** (`SseStream.send` writes `data:` only) | + +`acp-http/sse-stream.ts` even says so in a comment: _"no ring-buffer `id:` +sequencing — resumability is RFD Phase 4, deferred."_ This PR removes that +deferral. + +## Wire decision — SSE `id:` line (not in-payload `_meta`) + +The two SSE surfaces carry **different payloads**: + +- REST streams **`BridgeEvent` envelopes** (`{ id, v, type, data, _meta }`). + The SDK parser (`sdk-typescript/src/daemon/sse.ts`) extracts the cursor + from the **JSON envelope's `id` field** (it only reads `data:` lines). +- `/acp` streams **raw JSON-RPC 2.0 objects** (`session/update` + notifications, `session/request_permission` requests, responses). These + have no envelope `id` to carry a bus cursor, and a JSON-RPC `id` means + something else (request id). + +So for `/acp` the resume cursor is the **standard SSE `id:` line**: + +- It is EventSource-native — a spec-compliant SSE client (incl. the + vendored `AcpHttpTransport`) auto-tracks the last `id:` and auto-sends it + back as the `Last-Event-ID` header on reconnect. +- It keeps the JSON-RPC payload clean (no non-standard `_meta.qwen.eventId` + injection into protocol frames). +- It mirrors what `formatSseFrame` already emits on REST, so both surfaces + share the **same** `eventBus` ids and the same `Last-Event-ID` semantics. + +Only **bus-originated** frames carry an `id:` (`session/update`, +`session/request_permission`, daemon-pushed notifies). JSON-RPC +**responses/replies** that ride the session stream are _not_ bus events and +carry **no** `id:` — they are not in the ring and are intentionally not +replay-tracked (a lost in-flight prompt _response_ is the separately-tracked +§1.7 concern, out of scope here; §1.8 is about lost _content_ frames, which +are all bus `session/update` events). + +Synthetic terminal frames (`client_evicted`, `stream_error`, …) have no bus +`id` and so emit no `id:` line — matching REST, so they don't burn a slot in +the monotonic sequence the client resumes from. + +## Changes + +1. **`transport-stream.ts`** — `send(message, id?: number)`. The optional + `id` is the bus event id for SSE cursor tracking. +2. **`sse-stream.ts`** — `send(message, id?)` prepends `id: ${id}\n` before + the `data:` line when `id !== undefined` (mirrors `formatSseFrame`). +3. **`ws-stream.ts`** — `send(message, id?)` accepts and **ignores** `id`: + WebSocket is a stateful connection, no SSE replay (consistent with + `AcpWsTransport.supportsReplay = false`). +4. **`connection-registry.ts`** — `sendSession(sessionId, frame, id?)` + threads `id` to `stream.send`. The per-session pre-attach **buffer** + stores `{ frame, id? }` pairs so a buffered frame keeps its cursor when + flushed on attach. (The connection-scoped buffer is unchanged — those + frames are JSON-RPC responses with no bus id.) +5. **`dispatch.ts`** + - `translateEvent` passes `event.id` through every `sendSession` / + `binding.stream.send` call for bus events. + - `pumpSessionEvents(conn, sessionId, signal, lastEventId?)` forwards + `lastEventId` to `subscribeEvents` — directly reusing the existing + ring replay. +6. **`index.ts`** — the `GET /acp` session-stream branch reads the + `Last-Event-ID` header (via a strict `parseLastEventId`, same accept-only- + decimal-digits rule as REST) and passes it to `pumpSessionEvents`. + +No `eventBus`/bridge changes — the engine is reused verbatim. + +## Making resume actually engage (session-stream grace/reclaim) + +The `id:`/`Last-Event-ID` plumbing above is necessary but **not sufficient** — +on its own it never fires in the real flow. Previously, when a session SSE +stream closed at the transport level, the GET handler ran the **full** +`closeSessionStream` teardown: it removed the session from `ownedSessions`, +aborted the in-flight prompt, and detached the bridge client. In the real +EventSource/proxy order (old socket closes _first_, then the client +reconnects), that means a reconnect carrying `Last-Event-ID` is rejected +**403** by the ownership check before the cursor is ever read — and the prompt +producing the content was already aborted. The replay engine would have +nothing to reconnect to. + +So a transport-level session-stream close now **detaches** instead of tears +down (`AcpConnection.detachSessionStream`): it stops only the stream + its +event subscription and **keeps the binding, ownership, the in-flight prompt, +and the bridge-client registration** alive for a grace window +(`SESSION_GRACE_MS`, mirroring `CONN_GRACE_MS`). A reconnect within the window +re-attaches (`attachSessionStream` clears the grace timer — reclaim) and the +ring replay backfills the gap. If no reconnect arrives, the grace timer runs +the full teardown — bounding the runaway-prompt cost. Full teardown remains +immediate for an explicit `session/close` and for connection teardown +(`destroy`). The GET handler branches on `stream.isClosed`: a transport close +→ detach-with-grace; a pump that ends while the stream is still open +(subprocess done / iterator error) → full close (zombie stream). + +### Two replay-correctness guards this unlocks + +Both are latent until resume actually runs; the grace/reclaim above makes them +reachable, so they ship together: + +- **No double-delivery AND no silent loss (buffer ↔ ring).** A buffered bus + event is _also_ in the EventBus ring (it was published there to get its id). + So on a resume (`Last-Event-ID` present), `attachSessionStream` is given the + cursor and **does not flush id-bearing buffered frames at all** — the ring + replay (started at the client's cursor) is the single delivery path for every + bus event after the cursor. This is deliberately _not_ "flush the buffer, then + advance the replay cursor past it": a frame sent to the now-dead socket but + never received by the client has an id _below_ the buffer's ids yet _above_ the + client's cursor, so advancing the cursor past the buffer would **silently drop + it**. Letting the ring own all bus events delivers each exactly once with no + gap. Id-_less_ frames (JSON-RPC replies routed via `replySession`) are not ring + events, so the ring won't redeliver them — but they must not be flushed at + attach either: a buffered `session/prompt` _result_ flushed before replay would + arrive ahead of the content chunks that preceded it (client sees "done" before + the body — the exact truncated-body failure §1.8 fixes). So on resume the + id-less frames are **deferred**: left in the buffer, and the event pump releases + them (`flushBufferedSessionFrames`) once the replay boundary passes + (`replay_complete` / `state_resync_required`), preserving original stream order. + (A fresh connect with no `Last-Event-ID` has no ring anchor, so it flushes the + whole buffer immediately, in order, as before.) +- **Idempotent `permission_request` under replay.** A `permission_request` is + an id-bearing ring event, so a reconnect whose cursor precedes a still- + unanswered permission replays it. `translateEvent` now reuses the existing + `conn.pending` entry for that `bridgeRequestId` (re-sending the same outbound + JSON-RPC id for catch-up) instead of minting a second id + entry — no orphan + pending, no double-prompt for a client that dedupes on `_meta.requestId`. + +`parseLastEventId` is extracted to a shared `serve/sse-last-event-id.ts` used +by both the REST and `/acp` surfaces, so their strict accept/reject rules and +operator logging can't drift. + +## Backward compatibility + +- **Old clients that don't send `Last-Event-ID`** → `lastEventId` is + `undefined` → `subscribeEvents` starts live, exactly as today. +- **Adding `id:` lines is backward-compatible SSE** — a client that ignores + the field is unaffected; an EventSource-based one starts tracking it for + free. +- **The vendored `AcpHttpTransport` keeps `supportsReplay = false`** until + it opts in; the daemon change is inert for it until then. Once it flips + `supportsReplay = true` and resends `Last-Event-ID`, gap frames are + replayed from the ring and the §1.8 content loss is closed — **no further + daemon change needed**. +- The REST surface is untouched. + +## Test plan + +- `sse-stream.test.ts` — `send(msg, 7)` emits `id: 7\n` before `data:`; + `send(msg)` (no id) omits the `id:` line; ordering `id:` → `data:` → + blank line. +- `transport.test.ts` (end-to-end over the `/acp` transport): + - live `session/update` frames now arrive with an `id:` line; + - a `GET /acp` carrying `Last-Event-ID: N` flows the cursor to + `subscribeEvents`; a fresh stream with no header behaves as today; + - an overflow `Last-Event-ID` (> `MAX_SAFE_INTEGER`) → live-only; + - **real close-then-reconnect order**: close the old SSE _first_, then + reconnect with `Last-Event-ID` — assert **200 not 403** (ownership kept) + and the prompt is **not** aborted (grace/reclaim); + - a replayed `permission_request` reuses the pending entry (same outbound id). +- `connection-registry.test.ts` — a non-resume attach flushes the whole buffer + threading each frame's `id`; a **resume** attach (cursor present) skips the + id-bearing frames (ring replay owns them) but still flushes id-less JSON-RPC + replies; `detachSessionStream` keeps ownership/prompt across the grace window + then tears down on expiry; a reconnect within the window reclaims (cancels the + pending teardown). +- `ws-stream.test.ts` — `send(msg, id)` ignores the id: the WS wire frame is the + bare JSON, no SSE `id:` framing leaks in. + +## Out of scope (still deferred) + +- WebSocket / HTTP/2 transports. +- §1.7 cross-connection permission resolve (a vote POSTed on a different + `Acp-Connection-Id` than the one that streamed the prompt) — a separate, + security-sensitive concern tracked as its own follow-up. This PR does make + `permission_request` translation idempotent under replay (above), but does + not add the session-global requestId resolve. +- The lost in-flight _prompt response_ on the session stream — recovered + content frames all flow through the `eventBus` ring; a JSON-RPC response is + not a ring event. +- Consumer-side `supportsReplay` flip in the external `agent-web` + `AcpHttpTransport` (lives in a different repo; unblocked by this PR). diff --git a/packages/cli/src/serve/acp-http/connection-registry.test.ts b/packages/cli/src/serve/acp-http/connection-registry.test.ts index 2aee872a153..2269d3c5cbe 100644 --- a/packages/cli/src/serve/acp-http/connection-registry.test.ts +++ b/packages/cli/src/serve/acp-http/connection-registry.test.ts @@ -4,16 +4,20 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { ConnectionRegistry } from './connection-registry.js'; import type { TransportStream } from './transport-stream.js'; class FakeStream implements TransportStream { isClosed = false; + /** Records every send so tests can assert the bus `id` is threaded. */ + readonly sent: Array<{ message: unknown; id?: number }> = []; constructor(readonly kind: 'sse' | 'ws') {} - async send(_message: unknown): Promise {} + async send(message: unknown, id?: number): Promise { + this.sent.push({ message, id }); + } close(): void { this.isClosed = true; @@ -91,6 +95,265 @@ describe('ConnectionRegistry.getSnapshot', () => { } }); + it('finds and clears pending permissions across connections', () => { + const registry = new ConnectionRegistry(); + try { + const connA = registry.create(true); + const connB = registry.create(true); + expect(connA).toBeDefined(); + expect(connB).toBeDefined(); + if (!connA || !connB) return; + + const idA = connA.nextId(); + const idB = connB.nextId(); + expect(idA).not.toBe(idB); + + connA.pending.set(idA, { + sessionId: 'sess-1', + bridgeRequestId: 'perm-1', + kind: 'permission', + }); + expect(registry.findPendingClientRequest(idA)?.conn).toBe(connA); + expect(registry.findPendingPermission('perm-1', 'sess-1')?.id).toBe(idA); + + registry.deletePendingPermission('sess-1', 'perm-1'); + expect(registry.findPendingClientRequest(idA)).toBeUndefined(); + } finally { + registry.dispose(); + } + }); + + it('non-resume attach flushes all pre-attach buffered frames WITH their bus id', () => { + const registry = new ConnectionRegistry(); + try { + const conn = registry.create(true); + if (!conn) return; + conn.ownSession('sess-1'); + conn.getOrCreateSession('sess-1'); // binding exists, no stream yet + // Buffered before any stream attaches (id-bearing + an id-less frame). + conn.sendSession('sess-1', { a: 1 }, 5); + conn.sendSession('sess-1', { b: 2 }); // response frame, no bus id + conn.sendSession('sess-1', { c: 3 }, 8); + + const stream = new FakeStream('sse'); + const binding = conn.attachSessionStream( + 'sess-1', + stream, + new AbortController(), + ); + + // Non-resume attach (no Last-Event-ID): flush EVERYTHING, each frame + // keeping its id across the buffer → stream handoff (a regression to + // `send(frame)` would drop the cursor for early §1.8 frames). + expect(stream.sent).toEqual([ + { message: { a: 1 }, id: 5 }, + { message: { b: 2 }, id: undefined }, + { message: { c: 3 }, id: 8 }, + ]); + // The binding no longer carries a `lastFlushedEventId` — the resume cursor + // is the client's Last-Event-ID verbatim (see the resume test below). + expect( + (binding as unknown as { lastFlushedEventId?: number }) + .lastFlushedEventId, + ).toBeUndefined(); + } finally { + registry.dispose(); + } + }); + + it('on resume, skips id-bearing buffered frames (ring owns them) AND defers id-less replies until flushBufferedSessionFrames (post-replay order)', () => { + // Two regressions in one path: + // (1) silent-frame-loss: a frame sent to the dead socket (id below the + // buffer's ids, above the client cursor) must come back via ring + // replay — so the buffer must NOT flush bus events on resume. + // (2) out-of-order completion: an id-less `session/prompt` result buffered + // during the gap must NOT be flushed at attach (it would arrive BEFORE + // the ring replays the content chunks that preceded it). It's deferred + // and released by the pump after `replay_complete`. + const registry = new ConnectionRegistry(); + try { + const conn = registry.create(true); + if (!conn) return; + conn.ownSession('sess-1'); + conn.getOrCreateSession('sess-1'); + // Gap buffer holds two bus events (ids 6, 7) and one id-less reply. + conn.sendSession('sess-1', { a: 1 }, 6); + conn.sendSession('sess-1', { reply: true }); // JSON-RPC reply, no bus id + conn.sendSession('sess-1', { c: 3 }, 7); + + const stream = new FakeStream('sse'); + // Client resumes from id 3 (it never saw frame 4, lost in-flight). + conn.attachSessionStream('sess-1', stream, new AbortController(), 3); + + // At attach: NOTHING is sent. Bus events (6,7) belong to the ring replay; + // the id-less reply is deferred so it can't jump ahead of replayed content. + expect(stream.sent).toEqual([]); + + // The pump calls this once the replay boundary passes → the deferred + // reply is released, after the (replayed) content chunks. + conn.flushBufferedSessionFrames('sess-1'); + expect(stream.sent).toEqual([ + { message: { reply: true }, id: undefined }, + ]); + } finally { + registry.dispose(); + } + }); + + it('detachSessionStream is a no-op for a stale stream after reclaim (identity guard)', () => { + // The CONTRACT at the attach site marks this guard load-bearing: once a + // reclaim installs s2, the OLD stream s1 closing must NOT tear down or + // re-arm grace on the fresh binding — that would be frame loss + // indistinguishable from a network drop. + vi.useFakeTimers(); + const detached: string[] = []; + const registry = new ConnectionRegistry(undefined, (sid) => + detached.push(sid), + ); + try { + const conn = registry.create(true); + if (!conn) return; + conn.ownSession('sess-1'); + const s1 = new FakeStream('sse'); + conn.attachSessionStream('sess-1', s1, new AbortController()); + conn.detachSessionStream('sess-1', s1, 10_000); // grace armed for s1 + const s2 = new FakeStream('sse'); + conn.attachSessionStream('sess-1', s2, new AbortController()); // reclaim + const graceAfterReclaim = conn.sessions.get('sess-1')?.graceTimer; + expect(graceAfterReclaim).toBeUndefined(); // reclaim cleared the timer + + // The stale s1 close arrives late — must be a pure no-op. + conn.detachSessionStream('sess-1', s1, 10_000); + expect(conn.sessions.get('sess-1')?.stream).toBe(s2); // s2 still bound + expect(conn.sessions.get('sess-1')?.graceTimer).toBeUndefined(); // no re-arm + expect(conn.ownsSession('sess-1')).toBe(true); + + // And no teardown fires from the stale close. + vi.advanceTimersByTime(20_000); + expect(detached).not.toContain('sess-1'); + expect(conn.sessions.get('sess-1')?.stream).toBe(s2); + } finally { + registry.dispose(); + vi.useRealTimers(); + } + }); + + it('detachSessionStream keeps ownership/prompt across the grace window, then tears down on expiry', () => { + vi.useFakeTimers(); + const detached: string[] = []; + const registry = new ConnectionRegistry(undefined, (sid) => + detached.push(sid), + ); + try { + const conn = registry.create(true); + if (!conn) return; + conn.ownSession('sess-1'); + const stream = new FakeStream('sse'); + const binding = conn.attachSessionStream( + 'sess-1', + stream, + new AbortController(), + ); + const promptAbort = new AbortController(); + binding.promptAbort = promptAbort; + + // Transport-level close → detach with grace (NOT teardown). + conn.detachSessionStream('sess-1', stream, 10_000); + expect(conn.ownsSession('sess-1')).toBe(true); + expect(conn.sessions.has('sess-1')).toBe(true); + expect(promptAbort.signal.aborted).toBe(false); // prompt survives + expect(binding.stream).toBeUndefined(); // frames buffer until reconnect + + // No reconnect within the window → full teardown. + vi.advanceTimersByTime(10_000); + expect(conn.ownsSession('sess-1')).toBe(false); + expect(conn.sessions.has('sess-1')).toBe(false); + expect(promptAbort.signal.aborted).toBe(true); + expect(detached).toContain('sess-1'); + } finally { + registry.dispose(); + vi.useRealTimers(); + } + }); + + it('attachSessionStream within the grace window reclaims (cancels the pending teardown)', () => { + vi.useFakeTimers(); + const detached: string[] = []; + const registry = new ConnectionRegistry(undefined, (sid) => + detached.push(sid), + ); + try { + const conn = registry.create(true); + if (!conn) return; + conn.ownSession('sess-1'); + const s1 = new FakeStream('sse'); + const binding = conn.attachSessionStream( + 'sess-1', + s1, + new AbortController(), + ); + const promptAbort = new AbortController(); + binding.promptAbort = promptAbort; + + conn.detachSessionStream('sess-1', s1, 10_000); + // Reconnect within grace. + const s2 = new FakeStream('sse'); + conn.attachSessionStream('sess-1', s2, new AbortController()); + + // Past the original grace — teardown must NOT fire (timer cleared). + vi.advanceTimersByTime(20_000); + expect(conn.ownsSession('sess-1')).toBe(true); + expect(promptAbort.signal.aborted).toBe(false); + expect(detached).not.toContain('sess-1'); + expect(conn.sessions.get('sess-1')?.stream).toBe(s2); + } finally { + registry.dispose(); + vi.useRealTimers(); + } + }); + + it('buffers events produced during the detach gap and flushes them exactly once on reattach', () => { + // End-to-end of the PR's core value prop at the registry layer: detach → + // produce gap events (no stream attached → buffered) → reattach → the gap + // events flush exactly once, in order. (A resuming reattach instead leaves + // id-bearing frames to the ring replay — covered by the resume test above.) + vi.useFakeTimers(); + const registry = new ConnectionRegistry(); + try { + const conn = registry.create(true); + if (!conn) return; + conn.ownSession('sess-1'); + const s1 = new FakeStream('sse'); + conn.attachSessionStream('sess-1', s1, new AbortController()); + + // Transport-level close → detach with grace; stream is gone, ownership + // and the binding survive so subsequent frames buffer. + conn.detachSessionStream('sess-1', s1, 10_000); + expect(conn.sessions.get('sess-1')?.stream).toBeUndefined(); + + // Gap events arrive while detached — they must buffer, not drop. + conn.sendSession('sess-1', { chunk: 'a' }, 10); + conn.sendSession('sess-1', { chunk: 'b' }, 11); + expect(s1.sent).toEqual([]); // old stream is gone — nothing leaks to it + + // Non-resume reattach (no Last-Event-ID) → flush the whole gap buffer once. + const s2 = new FakeStream('sse'); + conn.attachSessionStream('sess-1', s2, new AbortController()); + expect(s2.sent).toEqual([ + { message: { chunk: 'a' }, id: 10 }, + { message: { chunk: 'b' }, id: 11 }, + ]); + + // The buffer is drained — a second reattach delivers nothing again. + const s3 = new FakeStream('sse'); + conn.attachSessionStream('sess-1', s3, new AbortController()); + expect(s3.sent).toEqual([]); + } finally { + registry.dispose(); + vi.useRealTimers(); + } + }); + it('aborts the connection signal when the connection is deleted', () => { const registry = new ConnectionRegistry(); try { diff --git a/packages/cli/src/serve/acp-http/connection-registry.ts b/packages/cli/src/serve/acp-http/connection-registry.ts index 80a551db126..c4c0d1a3f20 100644 --- a/packages/cli/src/serve/acp-http/connection-registry.ts +++ b/packages/cli/src/serve/acp-http/connection-registry.ts @@ -42,6 +42,12 @@ export type DetachSessionFn = ( clientId: string | undefined, ) => void; +/** A pre-attach session frame plus its optional bus event id (SSE cursor). */ +interface BufferedSessionFrame { + frame: unknown; + id?: number; +} + /** * Tracks one logical ACP-over-HTTP connection (RFD #721). A connection is * minted at `initialize`, keyed by `Acp-Connection-Id`, and may host many @@ -60,8 +66,12 @@ export interface SessionBinding { clientId?: string; /** Session-scoped SSE stream (the client's `GET /acp` with both headers). */ stream?: TransportStream; - /** Frames emitted before the session stream attached, flushed on attach. */ - buffer: unknown[]; + /** + * Frames emitted before the session stream attached, flushed on attach. + * Each keeps its bus event id (when it has one) so the SSE `id:` resume + * cursor survives the buffer → live-stream handoff. + */ + buffer: BufferedSessionFrame[]; /** * Aborts the bridge event subscription tied to the CURRENT session * stream. Replaced with a fresh controller on every re-attach — a @@ -77,6 +87,16 @@ export interface SessionBinding { * the agent burning model quota on a result nobody will read. */ promptAbort?: AbortController; + /** + * Armed by `detachSessionStream` when the session stream closes at the + * transport level (proxy idle-close, network blip) WITHOUT an explicit + * `session/close`. The binding — ownership, prompt, bridge-client — is kept + * alive across the window so a reconnect (`attachSessionStream`) can resume + * (ring replay backfills the gap, §1.8). If no reconnect arrives the timer + * fires the full teardown, bounding the runaway-prompt cost. Cleared on + * reconnect and on teardown. + */ + graceTimer?: ReturnType; } /** An agent→client request awaiting the client's JSON-RPC response. */ @@ -87,6 +107,12 @@ export interface PendingClientRequest { kind: 'permission'; } +export interface PendingClientRequestRef { + conn: AcpConnection; + id: string; + req: PendingClientRequest; +} + export interface AcpConnectionDiagnostic { connectionIdPrefix: string; fromLoopback: boolean; @@ -186,7 +212,7 @@ export class AcpConnection { */ nextId(): string { this.idCounter += 1; - return `_qwen_perm_${this.idCounter}`; + return `_qwen_perm_${this.connectionId}_${this.idCounter}`; } touch(): void { @@ -293,13 +319,13 @@ export class AcpConnection { * creating here would resurrect a ghost binding (no stream, no owner) that * buffers up to 256 late pump/reply frames forever. */ - sendSession(sessionId: string, frame: unknown): void { + sendSession(sessionId: string, frame: unknown, id?: number): void { const binding = this.sessions.get(sessionId); if (!binding) return; if (binding.stream && !binding.stream.isClosed) { - void binding.stream.send(frame); + void binding.stream.send(frame, id); } else { - pushCapped(binding.buffer, frame, `session ${sessionId}`); + pushCapped(binding.buffer, { frame, id }, `session ${sessionId}`); } } @@ -313,26 +339,125 @@ export class AcpConnection { sessionId: string, stream: TransportStream, abort: AbortController, + resumeFromEventId?: number, ): SessionBinding { const binding = this.getOrCreateSession(sessionId); + // Reclaim: a reconnect within the grace window cancels the pending + // teardown so ownership/prompt survive the transport-level blip. Log it so + // an operator can tell "reclaimed within grace" apart from a first attach + // (the detach + grace-expiry paths already log; this completes the trail). + if (binding.graceTimer) { + clearTimeout(binding.graceTimer); + binding.graceTimer = undefined; + writeStderrLine( + `qwen serve: /acp session reclaimed within grace (${logSafe(sessionId)})`, + ); + } const prevStream = binding.stream; binding.abort.abort(); binding.abort = abort; - // Install the NEW stream BEFORE closing the old one. The old stream's - // `onClose` is identity-guarded on `binding.stream` (see the session-GET - // handler in `index.ts` — `if (conn.sessions.get(sessionId)?.stream === - // stream) ...promptAbort?.abort()`), so installing first means a - // reconnect's close can't abort the in-flight prompt (the client is - // reconnecting, not leaving — the prompt must survive). CONTRACT: that - // identity guard and this ordering must stay in lockstep. + // Install the NEW stream BEFORE closing the old one. Each stream's event + // pump has its OWN abort controller, and the post-pump teardown in the + // session-GET handler (`index.ts` `onPumpSettled`) is identity-guarded on + // `binding.stream`: a settling stream only acts if it is STILL the bound + // stream (`conn.sessions.get(sessionId)?.stream === stream`). Installing + // first means the old stream settles against a binding that already points + // at the new stream, so it falls into detach-with-grace instead of tearing + // down the in-flight prompt — the client is reconnecting, not leaving, and + // the prompt must survive. CONTRACT: that identity guard and this ordering + // must stay in lockstep. binding.stream = stream; if (prevStream && prevStream !== stream && prevStream !== this.connStream) { prevStream.close(); } - for (const frame of binding.buffer.splice(0)) void stream.send(frame); + // Flush buffered pre-attach frames produced during the detach gap. + // + // FRESH CONNECT (`resumeFromEventId === undefined`, no `Last-Event-ID`): + // there's no ring replay, so the buffer is the only delivery path — flush + // everything now, in order. + // + // RESUME (`resumeFromEventId !== undefined`): the ring replay the event pump + // starts at that cursor already redelivers every BUS event (`id !== + // undefined`) after the cursor — including frames lost in-flight to the dead + // socket — so we do NOT flush id-bearing frames here (flushing would + // double-deliver, and advancing a cursor past them to dedupe would silently + // drop an in-flight-lost frame whose id sits below the buffer's ids). + // + // Id-LESS frames are JSON-RPC replies (`replySession`), NOT ring events, so + // the ring won't redeliver them. But flushing them HERE — before replay — + // would deliver e.g. a `session/prompt` result BEFORE the ring replays the + // content chunks that preceded it, so the client would see "prompt complete" + // ahead of the body (the exact truncated-body failure §1.8 fixes). So on + // resume we DEFER id-less frames: leave them in the buffer for the pump to + // flush after `replay_complete` (`flushBufferedSessionFrames`), preserving + // original stream order. + for (const entry of binding.buffer.splice(0)) { + if (resumeFromEventId === undefined) { + void stream.send(entry.frame, entry.id); // fresh connect: flush all now + } else if (entry.id !== undefined) { + continue; // resume: ring replay owns bus events + } else { + binding.buffer.push(entry); // resume: defer id-less past replay + } + } return binding; } + /** + * Flush any frames still buffered for a session to its live stream, in order. + * On resume, `attachSessionStream` defers id-less JSON-RPC replies (e.g. a + * `session/prompt` result that landed during the detach gap) into the buffer; + * the event pump calls this once the ring replay boundary + * (`replay_complete` / `state_resync_required`) has passed, so those replies + * are delivered AFTER the content chunks they followed in the original stream. + * No-op if the session has no live stream (the frames stay buffered for the + * next attach). + */ + flushBufferedSessionFrames(sessionId: string): void { + const binding = this.sessions.get(sessionId); + if (!binding?.stream || binding.buffer.length === 0) return; + for (const { frame, id } of binding.buffer.splice(0)) { + void binding.stream.send(frame, id); + } + } + + /** + * Transport-level session-stream close (proxy idle-close / network blip) — + * as opposed to an explicit `session/close`. Detaches ONLY the stream and + * its event subscription while KEEPING the binding, ownership, the in-flight + * prompt, and the bridge-client registration, so a reconnect within + * `graceMs` can resume (ring replay backfills the gap — §1.8). If no + * reconnect arrives, the grace timer runs the full `closeSessionStream` + * teardown, bounding the runaway-prompt cost. Identity-guarded: a stale + * stream's close can't detach a newer reconnect's stream. + */ + detachSessionStream( + sessionId: string, + stream: TransportStream, + graceMs: number, + ): void { + const binding = this.sessions.get(sessionId); + if (!binding || binding.stream !== stream) return; + // Stop the closing stream's event pump; the prompt + ownership live on. + binding.abort.abort(); + // Drop the stream ref so frames produced during the gap buffer until the + // reconnect re-attaches and flushes them. + binding.stream = undefined; + if (binding.graceTimer) clearTimeout(binding.graceTimer); + binding.graceTimer = setTimeout(() => { + // Grace expired with no reconnect → full teardown (aborts the prompt, + // releases ownership, detaches the bridge client). Log it so an operator + // debugging a vanished session can tell grace-expiry teardown apart from + // an explicit `session/close` or connection drop. + writeStderrLine( + `qwen serve: /acp session grace expired (${logSafe(sessionId)}), ` + + `no reconnect within ${graceMs}ms — tearing down`, + ); + this.closeSessionStream(sessionId); + }, graceMs); + binding.graceTimer.unref?.(); + } + closeSessionStream(sessionId: string): void { const binding = this.sessions.get(sessionId); if (!binding) return; @@ -361,6 +486,10 @@ export class AcpConnection { } private teardownBinding(binding: SessionBinding): void { + if (binding.graceTimer) { + clearTimeout(binding.graceTimer); + binding.graceTimer = undefined; + } binding.abort.abort(); binding.promptAbort?.abort(); // Don't close the stream if it's the shared connStream (WS reuses @@ -399,7 +528,7 @@ export class AcpConnection { } } -function pushCapped(buf: unknown[], frame: unknown, label = 'stream'): void { +function pushCapped(buf: T[], frame: T, label = 'stream'): void { if (buf.length >= MAX_BUFFERED_FRAMES) { buf.shift(); writeStderrLine( @@ -454,6 +583,38 @@ export class ConnectionRegistry { return conn; } + findPendingClientRequest(id: string): PendingClientRequestRef | undefined { + for (const conn of this.byId.values()) { + const req = conn.pending.get(id); + if (req) return { conn, id, req }; + } + return undefined; + } + + findPendingPermission( + requestId: string, + sessionId?: string, + ): PendingClientRequestRef | undefined { + for (const conn of this.byId.values()) { + for (const [id, req] of conn.pending) { + if (req.bridgeRequestId !== requestId) continue; + if (sessionId !== undefined && req.sessionId !== sessionId) continue; + return { conn, id, req }; + } + } + return undefined; + } + + deletePendingPermission(sessionId: string, requestId: string): void { + for (const conn of this.byId.values()) { + for (const [id, req] of conn.pending) { + if (req.sessionId === sessionId && req.bridgeRequestId === requestId) { + conn.pending.delete(id); + } + } + } + } + delete(connectionId: string): boolean { const conn = this.byId.get(connectionId); if (!conn) return false; diff --git a/packages/cli/src/serve/acp-http/dispatch.ts b/packages/cli/src/serve/acp-http/dispatch.ts index ca62b724764..57523df4fe3 100644 --- a/packages/cli/src/serve/acp-http/dispatch.ts +++ b/packages/cli/src/serve/acp-http/dispatch.ts @@ -72,7 +72,12 @@ import { WorkspacePermissionRulesSessionRequiredError, WorkspaceSettingsPartialPersistError, } from '../workspace-service/types.js'; -import type { AcpConnection } from './connection-registry.js'; +import type { + AcpConnection, + ConnectionRegistry, + PendingClientRequest, + PendingClientRequestRef, +} from './connection-registry.js'; import { QWEN_META_KEY, QWEN_METHOD_NS, @@ -178,6 +183,7 @@ const CONN_ROUTED_METHODS = new Set([ 'session/list', 'session/close', 'session/fork', + 'session/permission', ...ALL_QWEN_VENDOR_METHODS, ]); @@ -275,6 +281,30 @@ function validatePrompt(params: Record): void { } } +function parsePermissionResponse( + params: Record, +): Parameters[2] { + const outcome = params['outcome']; + if ( + typeof outcome !== 'object' || + outcome === null || + !( + (outcome as Record)['outcome'] === 'cancelled' || + ((outcome as Record)['outcome'] === 'selected' && + typeof (outcome as Record)['optionId'] === 'string' && + ((outcome as Record)['optionId'] as string).length > 0) + ) + ) { + throw new AcpParamError( + '`outcome` must be cancelled or selected with a non-empty optionId', + ); + } + const response: Record = { ...params, outcome }; + delete response['sessionId']; + delete response['requestId']; + return response as Parameters[2]; +} + /** * Map a thrown error to a JSON-RPC error code + a client-safe message. * Param-validation errors are echoed (they describe the client's own bad @@ -431,6 +461,7 @@ export class AcpDispatcher { private readonly fsFactory?: WorkspaceFileSystemFactory, private readonly deviceFlowRegistry?: DeviceFlowRegistry, private readonly sessionShellCommandEnabled: boolean = false, + private readonly registry?: ConnectionRegistry, ) { this.agentManager = createDaemonSubagentManager(boundWorkspace); } @@ -485,6 +516,37 @@ export class AcpDispatcher { return { clientId, fromLoopback }; } + private ensurePermissionPending( + conn: AcpConnection, + sessionId: string, + bridgeRequestId: string, + ): string { + for (const [existingId, req] of conn.pending) { + if ( + req.kind === 'permission' && + req.sessionId === sessionId && + req.bridgeRequestId === bridgeRequestId + ) { + return existingId; + } + } + const id = conn.nextId(); + conn.pending.set(id, { sessionId, bridgeRequestId, kind: 'permission' }); + return id; + } + + private dropResolvedPermission( + conn: AcpConnection, + id: string, + req: PendingClientRequest, + ): void { + if (this.registry) { + this.registry.deletePendingPermission(req.sessionId, req.bridgeRequestId); + } else { + conn.pending.delete(id); + } + } + /** * The session's ACP-shaped config options (model/mode/…), read from the * child's own session state. Returned in `session/new` and as the result @@ -1005,6 +1067,77 @@ export class AcpDispatcher { return; } + case 'session/permission': { + const requestId = + typeof params['requestId'] === 'string' ? params['requestId'] : ''; + if (!requestId) { + if (id !== undefined) { + conn.sendConn( + error(id, RPC.INVALID_PARAMS, '`requestId` is required'), + ); + } + return; + } + const response = parsePermissionResponse(params); + let sessionId = + typeof params['sessionId'] === 'string' + ? params['sessionId'] + : undefined; + if (sessionId !== undefined) { + if (!this.requireOwned(conn, sessionId, id)) return; + } else { + const pending = this.registry?.findPendingPermission(requestId); + sessionId = pending?.req.sessionId; + if (sessionId === undefined) { + if (id !== undefined) { + conn.sendConn( + error( + id, + RPC.INVALID_PARAMS, + 'No pending permission request', + { httpStatus: 404, requestId }, + ), + ); + } + return; + } + if (!conn.ownsSession(sessionId)) { + if (id !== undefined) { + conn.sendConn( + error( + id, + RPC.INVALID_PARAMS, + 'Permission request is not owned by this connection', + ), + ); + } + return; + } + } + const accepted = this.bridge.respondToSessionPermission( + sessionId, + requestId, + response, + this.sessionCtx(conn, sessionId, loopback), + ); + if (!accepted) { + if (id !== undefined) { + conn.sendConn( + error( + id, + RPC.INVALID_PARAMS, + 'No pending permission request for session', + { httpStatus: 404, sessionId, requestId }, + ), + ); + } + return; + } + this.registry?.deletePendingPermission(sessionId, requestId); + this.replyConn(conn, id, {}); + return; + } + // STANDARD method (SDK 0.14.1, non-`unstable_`): model + mode live // here under categories `model`/`mode`, routed to the existing bridge // setters. Replaces the old vendor `_qwen/session/set_model`. @@ -2738,16 +2871,52 @@ export class AcpDispatcher { conn: AcpConnection, sessionId: string, signal: AbortSignal, + lastEventId?: number, ): Promise { try { - const iterable = this.bridge.subscribeEvents(sessionId, { signal }); + // `lastEventId` (from the `Last-Event-ID` reconnect header) drives the + // EventBus ring replay: events with `id > lastEventId` still buffered + // are replayed before live events flow, recovering content frames lost + // in a mid-turn proxy gap (§1.8). `undefined` ⇒ live-only, as before. + const iterable = this.bridge.subscribeEvents(sessionId, { + signal, + ...(lastEventId !== undefined ? { lastEventId } : {}), + }); + // On resume, `attachSessionStream` defers id-less buffered replies (e.g. a + // `session/prompt` result produced during the detach gap) so they land + // AFTER the ring replays the content chunks that preceded them. Release + // them once the replay boundary passes — `replay_complete` (caught up) or + // `state_resync_required` (ring couldn't replay; client reloads, but the + // pending request still needs its reply). Once-guarded. + let deferredFlushed = false; + const flushDeferred = () => { + if (deferredFlushed) return; + deferredFlushed = true; + conn.flushBufferedSessionFrames(sessionId); + }; for await (const event of iterable) { if (signal.aborted) break; // Count event delivery as connection activity so a long, quiet prompt // (no inbound HTTP) isn't reaped by the idle-TTL sweep. conn.touch(); this.translateEvent(conn, sessionId, event); + if ( + event.type === 'replay_complete' || + event.type === 'state_resync_required' + ) { + flushDeferred(); + } } + // Safety: a live-only subscription (no cursor → no replay boundary) or a + // clean end without a boundary frame still releases anything deferred — + // but NOT if this pump was aborted. An abort means the stream was + // detached/reclaimed; flushing here could drain the deferred reply onto a + // RECLAIMING stream ahead of its own replay (reintroducing the very out-of- + // order delivery the deferral prevents). The reclaiming pump owns the + // buffer then and will flush after its replay boundary. On an iterator + // error mid-replay (the catch below) the deferred frames likewise stay + // buffered for the next attach — never lost, just delivered next round. + if (!signal.aborted) flushDeferred(); } catch (err) { // Symmetric for the SYNC `subscribeEvents` throw and a MID-STREAM // iterator error: surface a `stream_error` to the client, then re-throw @@ -2777,7 +2946,13 @@ export class AcpDispatcher { switch (event.type) { case 'session_update': { // `event.data` is the ACP `SessionNotification` (params shape). - conn.sendSession(sessionId, notification('session/update', event.data)); + // `event.id` is the bus cursor → SSE `id:` line for `Last-Event-ID` + // resume (the content frames §1.8 recovers all flow through here). + conn.sendSession( + sessionId, + notification('session/update', event.data), + event.id, + ); return; } case 'permission_request': { @@ -2787,15 +2962,29 @@ export class AcpDispatcher { toolCall: unknown; options: unknown; }; - // A permission request MUST reach a LIVE session stream. Going - // through `sendSession` would (a) silently drop the frame if the - // session was torn down (lookup-only), or (b) buffer it pre-attach - // where `pushCapped` could evict it under event throughput — either - // way the `pending` entry is orphaned and the agent's prompt blocks - // on a vote forever. So deliver DIRECTLY to a live stream, and if - // there is none, cancel (deny-safe) rather than register+stall. + const frameFor = (id: string) => + request(id, 'session/request_permission', { + sessionId: data.sessionId, + toolCall: data.toolCall, + options: data.options, + _meta: { [QWEN_META_KEY]: { requestId: data.requestId } }, + }); + // A permission request normally MUST reach a LIVE session stream. + // During reconnect grace (or the tiny close-before-detach window) the + // binding is intentionally alive with no usable stream, so defer + // through the session buffer/ring replay instead of auto-denying. + // Outside that reconnect path, keep the deny-safe cancel. const binding = conn.sessions.get(sessionId); if (!binding?.stream || binding.stream.isClosed) { + if (binding?.graceTimer || binding?.stream?.isClosed) { + const id = this.ensurePermissionPending( + conn, + sessionId, + data.requestId, + ); + conn.sendSession(sessionId, frameFor(id), event.id); + return; + } const cancelled = this.cancelAbandonedPermission( { sessionId, bridgeRequestId: data.requestId }, // Pass the bridge-stamped clientId when the binding still exists @@ -2816,19 +3005,22 @@ export class AcpDispatcher { } return; } - const id = conn.nextId(); - conn.pending.set(id, { + const id = this.ensurePermissionPending( + conn, sessionId, - bridgeRequestId: data.requestId, - kind: 'permission', - }); + data.requestId, + ); + // INVARIANT: this sends straight to `binding.stream` (not via + // `conn.sendSession`) and is safe ONLY because `translateEvent` runs + // synchronously from the pump — `binding.stream` was checked non-null + // above and cannot be detached mid-call. Do NOT introduce an `await` + // between that check and this send: a detach during the gap would set + // `binding.stream = undefined` and this would throw `TypeError`. void binding.stream.send( - request(id, 'session/request_permission', { - sessionId: data.sessionId, - toolCall: data.toolCall, - options: data.options, - _meta: { [QWEN_META_KEY]: { requestId: data.requestId } }, - }), + frameFor(id), + // Carry the bus cursor: a permission request is a real sequenced + // event, so the client must resume past it. + event.id, ); return; } @@ -2841,18 +3033,25 @@ export class AcpDispatcher { ...(event.data as object), kind: 'stream_error', }), + // Pass the bus cursor through if present; a synthetic terminal frame + // has no bus id (event.id undefined) so no SSE `id:` line is written. + event.id, ); return; } default: { // client_evicted / slow_client_warning / state_resync_required / // model_switched / approval_mode_changed / … → opaque qwen notify. + // `event.id` is undefined for the synthetic control frames (no SSE + // `id:` line, so they don't burn a slot in the resume sequence) and + // set for ring-backed daemon events. conn.sendSession( sessionId, notification(`${QWEN_METHOD_NS}notify`, { kind: event.type, data: event.data, }), + event.id, ); } } @@ -2868,12 +3067,17 @@ export class AcpDispatcher { msg: JsonRpcResponse, fromLoopback: boolean, ): void { - // Our outbound request ids are strings (`_qwen_perm_N`); a client echoes - // the same id verbatim. Anything else can't match a pending entry. + // Our outbound request ids are strings (`_qwen_perm__N`); a client + // echoes the same id verbatim. Anything else can't match a pending entry. const id = msg.id; if (typeof id !== 'string') return; - const pending = conn.pending.get(id); - if (!pending) return; + const localPending = conn.pending.get(id); + const pendingRef: PendingClientRequestRef | undefined = localPending + ? { conn, id, req: localPending } + : this.registry?.findPendingClientRequest(id); + if (!pendingRef) return; + const { conn: pendingConn, req: pending } = pendingRef; + if (pendingConn !== conn && !conn.ownsSession(pending.sessionId)) return; // NOTE: do NOT delete the pending entry yet. Keep it until either the // bridge vote OR the cancel fallback runs — if both somehow fail, the // entry survives so a later session/connection teardown @@ -2899,7 +3103,7 @@ export class AcpDispatcher { >[2], this.sessionCtx(conn, pending.sessionId, fromLoopback), ); - conn.pending.delete(id); // vote landed — safe to drop + this.dropResolvedPermission(pendingConn, id, pending); } catch (err) { writeStderrLine( `qwen serve: /acp permission vote failed (${logSafe(pending.sessionId)}): ${logSafe(errMsg(err))}`, @@ -2910,9 +3114,9 @@ export class AcpDispatcher { // permanently stuck with no recovery path. const cancelled = this.cancelAbandonedPermission( pending, - conn.sessions.get(pending.sessionId)?.clientId, + pendingConn.sessions.get(pending.sessionId)?.clientId, ); - if (cancelled) conn.pending.delete(id); + if (cancelled) pendingConn.pending.delete(id); } } diff --git a/packages/cli/src/serve/acp-http/index.ts b/packages/cli/src/serve/acp-http/index.ts index 336b5fba9ec..9111c55246d 100644 --- a/packages/cli/src/serve/acp-http/index.ts +++ b/packages/cli/src/serve/acp-http/index.ts @@ -23,6 +23,7 @@ import { SseStream } from './sse-stream.js'; import { WsStream } from './ws-stream.js'; import type { RateLimitTier } from '../rate-limit.js'; import { RPC, error as rpcError, isRequest, parseInbound } from './json-rpc.js'; +import { parseLastEventId } from '../sse-last-event-id.js'; export const ACP_CONNECTION_HEADER = 'acp-connection-id'; export const ACP_SESSION_HEADER = 'acp-session-id'; @@ -75,6 +76,16 @@ function extractUpgradeBearer(req: IncomingMessage): string | undefined { */ const CONN_GRACE_MS = 10_000; +/** + * Grace window after a SESSION-scoped SSE stream closes at the transport level + * (proxy idle-close, network blip) without an explicit `session/close`. The + * binding — ownership, in-flight prompt, bridge-client — is kept alive so a + * reconnect within the window resumes via ring replay (§1.8) instead of being + * rejected (403) and re-spawning. Short enough to bound the runaway-prompt + * cost if the client never returns. Mirrors `CONN_GRACE_MS`. + */ +const SESSION_GRACE_MS = 10_000; + const WS_EXEMPT_METHODS = new Set([ '_qwen/session/heartbeat', '_qwen/session/update_metadata', @@ -170,18 +181,16 @@ export function mountAcpHttp( if (!enabled) return undefined; const path = opts.path ?? '/acp'; - const dispatcher = new AcpDispatcher( - bridge, - opts.boundWorkspace, - opts.workspace, - opts.fsFactory, - opts.deviceFlowRegistry, - opts.sessionShellCommandEnabled === true, - ); + const dispatcherRef: { current?: AcpDispatcher } = {}; // When a session/connection tears down with a permission still pending, // cancel it on the bridge so the agent's prompt isn't left blocked. const registry = new ConnectionRegistry( - (req, clientId) => dispatcher.cancelAbandonedPermission(req, clientId), + (req, clientId) => { + if (!dispatcherRef.current) { + throw new Error('ACP dispatcher not initialized'); + } + return dispatcherRef.current.cancelAbandonedPermission(req, clientId); + }, // Best-effort bridge detach so a torn-down connection's bridge-stamped // client ids don't linger in the bridge's voter/known-client sets. (sessionId, clientId) => { @@ -195,6 +204,16 @@ export function mountAcpHttp( }, opts.maxConnections, ); + const dispatcher = new AcpDispatcher( + bridge, + opts.boundWorkspace, + opts.workspace, + opts.fsFactory, + opts.deviceFlowRegistry, + opts.sessionShellCommandEnabled === true, + registry, + ); + dispatcherRef.current = dispatcher; // ── POST /acp ────────────────────────────────────────────────────── app.post(path, async (req: Request, res: Response) => { @@ -415,46 +434,57 @@ export function mountAcpHttp( const stream = new SseStream( res, () => { - // Stream closed (tab close / network drop / crash): stop the event - // pump AND abort any in-flight prompt for this session — otherwise - // the agent keeps running (quota, FIFO) until idle TTL. + // Transport-level close (tab close / network drop / proxy idle-close): + // stop THIS stream's event pump only. The prompt + ownership are NOT + // torn down here — `detachSessionStream` (below) keeps them across a + // grace window so a reconnect can resume (§1.8). Only an expired grace, + // an explicit `session/close`, or connection teardown aborts the prompt. ac.abort(); - // BUT only abort the prompt when THIS is still the session's live - // stream. A reconnect already installed a newer stream — the prompt - // must survive the old stream's close. CONTRACT: this identity guard - // pairs with `attachSessionStream`'s install-before-close ordering - // (connection-registry.ts) — keep both in lockstep. - if (conn.sessions.get(sessionId)?.stream === stream) { - conn.sessions.get(sessionId)?.promptAbort?.abort(); - } }, () => conn.touch(), ); // Open (write SSE headers + `retry:`) BEFORE attaching, so the protocol // handshake precedes any buffered frames the attach flushes. stream.open(); - conn.attachSessionStream(sessionId, stream, ac); - // Identity-guarded close: only tear down if THIS stream is still the - // session's current one (a reconnect between settle and this microtask - // would otherwise kill the fresh stream). - const closeIfCurrent = () => { - if (conn.sessions.get(sessionId)?.stream === stream) { + // Resume cursor: an EventSource/SSE client auto-resends the last `id:` it + // saw as `Last-Event-ID` on reconnect. Drives the EventBus ring replay so + // content frames produced during a mid-turn proxy gap are recovered (§1.8). + const lastEventId = parseLastEventId( + headerOf(req, 'last-event-id'), + '/acp ', + ); + // Pass the resume cursor INTO attach: when resuming, attach skips flushing + // id-bearing buffered frames because the ring replay below redelivers every + // bus event after `lastEventId` exactly once — including any frame lost + // in-flight to the dead socket (whose id sits below the buffer's ids but + // above the client's cursor). Advancing the cursor past the buffer instead + // would silently drop that frame; flushing AND replaying would double-send. + // Id-less JSON-RPC replies are still flushed (they aren't ring events). + conn.attachSessionStream(sessionId, stream, ac, lastEventId); + // When the pump settles, branch on WHY: + // • the transport closed the stream (proxy idle-close / tab close) → + // DETACH with a grace window: keep ownership + the in-flight prompt so a + // reconnect resumes (§1.8); full teardown only if no reconnect arrives. + // • the pump ended while the stream is still open (subprocess done / + // iterator error) → the stream is a zombie; full close now. + // Both are identity-guarded so a stale stream can't act on a newer one. + const onPumpSettled = () => { + if (stream.isClosed) { + conn.detachSessionStream(sessionId, stream, SESSION_GRACE_MS); + } else if (conn.sessions.get(sessionId)?.stream === stream) { conn.closeSessionStream(sessionId); } }; - void dispatcher.pumpSessionEvents(conn, sessionId, ac.signal).then( - // NORMAL completion (iterator returned `done` — subprocess ended): close - // so the stream isn't a zombie heartbeating with nothing left to deliver. - closeIfCurrent, - (err: unknown) => { + void dispatcher + .pumpSessionEvents(conn, sessionId, ac.signal, lastEventId) + .then(onPumpSettled, (err: unknown) => { writeStderrLine( - `qwen serve: /acp event pump error (${sessionId}): ${ - err instanceof Error ? err.message : String(err) - }`, + `qwen serve: /acp event pump error (${sessionId}, lastEventId=${ + lastEventId ?? 'none' + }): ${err instanceof Error ? err.message : String(err)}`, ); - closeIfCurrent(); - }, - ); + onPumpSettled(); + }); }); // ── DELETE /acp ──────────────────────────────────────────────────── diff --git a/packages/cli/src/serve/acp-http/sse-stream.test.ts b/packages/cli/src/serve/acp-http/sse-stream.test.ts index fe125d8390c..3a851571434 100644 --- a/packages/cli/src/serve/acp-http/sse-stream.test.ts +++ b/packages/cli/src/serve/acp-http/sse-stream.test.ts @@ -63,6 +63,36 @@ describe('SseStream', () => { ); }); + it('send(message, id) prepends an `id:` line before `data:` (resume cursor)', async () => { + const res = mockRes(); + const s = new SseStream(res); + s.open(); + await s.send({ jsonrpc: '2.0', method: 'session/update', params: {} }, 7); + const joined = (res as unknown as { chunks: string[] }).chunks.join(''); + expect(joined).toContain( + 'id: 7\ndata: {"jsonrpc":"2.0","method":"session/update","params":{}}\n\n', + ); + }); + + it('send(message) without an id omits the `id:` line (synthetic/response frames)', async () => { + const res = mockRes(); + const s = new SseStream(res); + s.open(); + await s.send({ jsonrpc: '2.0', id: 1, result: {} }); + const joined = (res as unknown as { chunks: string[] }).chunks.join(''); + expect(joined).toContain('data: {"jsonrpc":"2.0","id":1,"result":{}}\n\n'); + expect(joined).not.toMatch(/(^|\n)id: /); + }); + + it('send(message, 0) emits `id: 0` (id 0 is a real cursor, not "absent")', async () => { + const res = mockRes(); + const s = new SseStream(res); + s.open(); + await s.send({ ping: true }, 0); + const joined = (res as unknown as { chunks: string[] }).chunks.join(''); + expect(joined).toContain('id: 0\ndata: {"ping":true}\n\n'); + }); + it('close() ends the response once and is idempotent', () => { const res = mockRes(); const s = new SseStream(res); diff --git a/packages/cli/src/serve/acp-http/sse-stream.ts b/packages/cli/src/serve/acp-http/sse-stream.ts index f6bc7f7dbda..d8d0d4b836d 100644 --- a/packages/cli/src/serve/acp-http/sse-stream.ts +++ b/packages/cli/src/serve/acp-http/sse-stream.ts @@ -18,9 +18,9 @@ import { writeStderrLine } from '../../utils/stdioHelpers.js'; * - respect backpressure (`res.write` → false ⇒ await `drain`), * - emit periodic comment heartbeats to keep NAT/proxies alive. * - * This mirrors the battle-tested pattern in `server.ts`'s SSE handler but - * trimmed to what the ACP transport needs (no ring-buffer `id:` sequencing — - * resumability is RFD Phase 4, deferred per the design doc §7). + * This mirrors the battle-tested pattern in `server.ts`'s SSE handler, + * including the optional ring-buffer `id:` sequencing that drives + * `Last-Event-ID` resume (see `docs/design/daemon-acp-http/sse-resumable-stream.md`). */ export class SseStream { readonly kind = 'sse' as const; @@ -63,9 +63,16 @@ export class SseStream { this.res.on('error', this.cleanupFn); } - /** Serialize a JSON-RPC message as one SSE frame. */ - send(message: unknown): Promise { - return this.writeRaw(`data: ${JSON.stringify(message)}\n\n`); + /** + * Serialize a JSON-RPC message as one SSE frame. When `id` is supplied + * (a bus event id) prepend an `id:` line so an EventSource/SSE client + * tracks it and resends it as `Last-Event-ID` on reconnect — the resume + * cursor for ring replay. Omitted for JSON-RPC responses and synthetic + * terminal frames (no bus id), matching REST `formatSseFrame`. + */ + send(message: unknown, id?: number): Promise { + const idLine = id !== undefined ? `id: ${id}\n` : ''; + return this.writeRaw(`${idLine}data: ${JSON.stringify(message)}\n\n`); } get isClosed(): boolean { diff --git a/packages/cli/src/serve/acp-http/transport-stream.ts b/packages/cli/src/serve/acp-http/transport-stream.ts index e0408f6d251..f3958afeb0a 100644 --- a/packages/cli/src/serve/acp-http/transport-stream.ts +++ b/packages/cli/src/serve/acp-http/transport-stream.ts @@ -10,7 +10,13 @@ */ export interface TransportStream { readonly kind: 'sse' | 'ws'; - send(message: unknown): Promise; + /** + * Serialize one frame. `id` is the bus event id (`BridgeEvent.id`) used as + * the SSE `id:` resume cursor — present only for ring-backed session events, + * omitted for JSON-RPC responses and synthetic terminal frames. The + * WebSocket transport ignores it (stateful connection, no SSE replay). + */ + send(message: unknown, id?: number): Promise; close(): void; readonly isClosed: boolean; } diff --git a/packages/cli/src/serve/acp-http/transport.test.ts b/packages/cli/src/serve/acp-http/transport.test.ts index f864e0cfa74..2f43c297231 100644 --- a/packages/cli/src/serve/acp-http/transport.test.ts +++ b/packages/cli/src/serve/acp-http/transport.test.ts @@ -47,6 +47,9 @@ import { MAX_TRUST_REASON_LENGTH, MAX_VOICE_MODEL_LENGTH, } from '../validation-limits.js'; +import { AcpDispatcher } from './dispatch.js'; +import { AcpConnection } from './connection-registry.js'; +import type { TransportStream } from './transport-stream.js'; const stdioMocks = vi.hoisted(() => ({ writeStderrLine: vi.fn(), @@ -120,6 +123,21 @@ function pushQueue(signal?: AbortSignal): PushIterable { }; } +class FakeTransportStream implements TransportStream { + isClosed = false; + readonly sent: Array<{ message: unknown; id?: number }> = []; + + constructor(readonly kind: 'sse' | 'ws' = 'sse') {} + + async send(message: unknown, id?: number): Promise { + this.sent.push({ message, id }); + } + + close(): void { + this.isClosed = true; + } +} + // A controllable fake bridge: tests register what `sendPrompt` should do. class FakeBridge { queues = new Map(); @@ -184,9 +202,19 @@ class FakeBridge { } subscribeThrows = false; + /** Records every subscribeEvents call so tests can assert the resume cursor. */ + subscribeCalls: Array<{ sessionId: string; lastEventId?: number }> = []; + /** Parallel to `subscribeCalls`: each subscription's abort signal, so a test + * can detect when a closed stream's pump has actually stopped server-side. */ + subscribeSignals: Array = []; - subscribeEvents(sessionId: string, opts?: { signal?: AbortSignal }) { + subscribeEvents( + sessionId: string, + opts?: { signal?: AbortSignal; lastEventId?: number }, + ) { if (this.subscribeThrows) throw new Error('subscribe failed'); + this.subscribeCalls.push({ sessionId, lastEventId: opts?.lastEventId }); + this.subscribeSignals.push(opts?.signal); const q = pushQueue(opts?.signal); this.queues.set(sessionId, q); return q.iterable; @@ -529,6 +557,53 @@ async function* readSse( } } +/** + * Like `readSse` but yields the RAW frame text (so the `id:` resume-cursor + * line is visible — `readSse` only keeps the parsed `data:` payload). + */ +async function* readSseRaw( + res: Response, + signal: AbortSignal, +): AsyncGenerator { + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + signal.addEventListener('abort', () => void reader.cancel().catch(() => {})); + while (true) { + const { value, done } = await reader.read(); + if (done) return; + buf += decoder.decode(value, { stream: true }); + let idx: number; + while ((idx = buf.indexOf('\n\n')) !== -1) { + const frame = buf.slice(0, idx); + buf = buf.slice(idx + 2); + // Skip the `retry:` hint and comment-only heartbeats (no `data:` line). + if (frame.split('\n').some((l) => l.startsWith('data: '))) yield frame; + } + } +} + +/** Read the next N RAW data frames (with `id:` lines) from an SSE response. */ +async function takeRawFrames( + res: Response, + n: number, + timeoutMs = 2000, +): Promise { + const out: string[] = []; + const ac = new AbortController(); + const timer = setTimeout(() => ac.abort(), timeoutMs); + try { + for await (const f of readSseRaw(res, ac.signal)) { + out.push(f); + if (out.length >= n) break; + } + } finally { + clearTimeout(timer); + ac.abort(); + } + return out; +} + /** Read the next N data frames from an SSE response, then abort. */ async function takeFrames( res: Response, @@ -853,6 +928,195 @@ describe('ACP Streamable HTTP transport (over the wire)', () => { ).toBe('end_turn'); }); + it('live session/update frames carry an SSE `id:` resume cursor', async () => { + bridge.promptBehavior = async (_s, q) => { + q.push({ + type: 'session_update', + data: { + sessionId: 'sess-1', + update: { sessionUpdate: 'agent_message_chunk' }, + }, + }); + await new Promise((r) => setTimeout(r, 20)); + return { stopReason: 'end_turn' }; + }; + const connId = await initialize(); + await newSession(connId); + const sessStream = await openStream(connId, 'sess-1'); + const got = takeRawFrames(sessStream, 1); + await new Promise((r) => setTimeout(r, 50)); + await post(connId, { + jsonrpc: '2.0', + id: 5, + method: 'session/prompt', + params: { sessionId: 'sess-1', prompt: [{ type: 'text', text: 'hi' }] }, + }); + const frames = await got; + // `pushQueue` stamps bus ids from 1, so the first session_update is id 1. + // The frame MUST carry `id: 1` before its `data:` line — that is the + // cursor an SSE client echoes as `Last-Event-ID` on reconnect. + expect(frames[0]).toMatch(/(^|\n)id: 1\ndata: /); + expect(frames[0]).toContain('"method":"session/update"'); + }); + + it('GET Last-Event-ID flows to subscribeEvents as the resume cursor', async () => { + const connId = await initialize(); + await newSession(connId); + + // Reconnect carrying a cursor → subscribeEvents gets lastEventId=42. + const resumed = await fetch(`${base}/acp`, { + headers: { + accept: 'text/event-stream', + 'acp-connection-id': connId, + 'acp-session-id': 'sess-1', + 'last-event-id': '42', + }, + }); + await waitUntil(() => bridge.subscribeCalls.length >= 1); + expect(bridge.subscribeCalls.at(-1)).toEqual({ + sessionId: 'sess-1', + lastEventId: 42, + }); + await resumed.body?.cancel().catch(() => {}); + + // A non-numeric header is rejected (logged) → live-only (undefined). + const bad = await fetch(`${base}/acp`, { + headers: { + accept: 'text/event-stream', + 'acp-connection-id': connId, + 'acp-session-id': 'sess-1', + 'last-event-id': 'not-a-number', + }, + }); + await waitUntil(() => bridge.subscribeCalls.length >= 2); + expect(bridge.subscribeCalls.at(-1)).toEqual({ + sessionId: 'sess-1', + lastEventId: undefined, + }); + await bad.body?.cancel().catch(() => {}); + + // A fresh stream with no header → live-only (undefined), as before. + const fresh = await openStream(connId, 'sess-1'); + await waitUntil(() => bridge.subscribeCalls.length >= 3); + expect(bridge.subscribeCalls.at(-1)).toEqual({ + sessionId: 'sess-1', + lastEventId: undefined, + }); + await fresh.body?.cancel().catch(() => {}); + }); + + it('real close-then-reconnect order keeps ownership (no 403) + prompt alive, resumes via Last-Event-ID', async () => { + let promptSignal: AbortSignal | undefined; + bridge.promptBehavior = async (_s, q, signal) => { + promptSignal = signal; + q.push({ + type: 'session_update', + data: { + sessionId: 'sess-1', + update: { sessionUpdate: 'agent_message_chunk' }, + }, + }); + // Keep the prompt running across the disconnect + reconnect. + await new Promise((r) => setTimeout(r, 1000)); + return { stopReason: 'end_turn' }; + }; + const connId = await initialize(); + await newSession(connId); + const s1 = await openStream(connId, 'sess-1'); + await waitUntil(() => bridge.subscribeCalls.length >= 1); // pump subscribed + const r1 = frameReader(s1); + const ack = await post(connId, { + jsonrpc: '2.0', + id: 5, + method: 'session/prompt', + params: { sessionId: 'sess-1', prompt: [{ type: 'text', text: 'hi' }] }, + }); + expect(ack.status).toBe(202); + await r1.next(); // first content frame (bus id 1) + + // Close the OLD stream FIRST — the real EventSource/proxy order (the + // existing reconnect tests overlap streams, hiding this). + r1.close(); + await s1.body?.cancel().catch(() => {}); + // Wait until the daemon has PROCESSED the close (old pump's signal aborted). + await waitUntil(() => bridge.subscribeSignals[0]?.aborted === true); + + // Detach-with-grace, NOT teardown: the in-flight prompt must survive. + expect(promptSignal?.aborted).toBe(false); + + // Reconnect carrying the cursor — must be 200 (ownership kept), not 403. + const s2 = await fetch(`${base}/acp`, { + headers: { + accept: 'text/event-stream', + 'acp-connection-id': connId, + 'acp-session-id': 'sess-1', + 'last-event-id': '1', + }, + }); + expect(s2.status).toBe(200); + await waitUntil(() => bridge.subscribeCalls.length >= 2); + expect(bridge.subscribeCalls.at(-1)).toEqual({ + sessionId: 'sess-1', + lastEventId: 1, + }); + expect(promptSignal?.aborted).toBe(false); // still alive after reconnect + await s2.body?.cancel().catch(() => {}); + }); + + it('GET Last-Event-ID past MAX_SAFE_INTEGER → live-only (undefined)', async () => { + const connId = await initialize(); + await newSession(connId); + const overflow = await fetch(`${base}/acp`, { + headers: { + accept: 'text/event-stream', + 'acp-connection-id': connId, + 'acp-session-id': 'sess-1', + 'last-event-id': '9007199254740992', // MAX_SAFE_INTEGER + 1 + }, + }); + await waitUntil(() => bridge.subscribeCalls.length >= 1); + expect(bridge.subscribeCalls.at(-1)).toEqual({ + sessionId: 'sess-1', + lastEventId: undefined, + }); + await overflow.body?.cancel().catch(() => {}); + }); + + it('a replayed permission_request reuses the pending entry (idempotent, same outbound id)', async () => { + bridge.promptBehavior = async (_s, q) => { + const perm = { + requestId: 'perm-1', + sessionId: 'sess-1', + toolCall: { name: 'shell' }, + options: [{ optionId: 'allow', name: 'Allow' }], + }; + q.push({ type: 'permission_request', data: perm }); + // Simulate a ring replay re-delivering the SAME bridge request (a + // reconnect whose Last-Event-ID precedes the still-pending permission). + q.push({ type: 'permission_request', data: perm }); + await new Promise((r) => setTimeout(r, 50)); + return { stopReason: 'end_turn' }; + }; + const connId = await initialize(); + await newSession(connId); + const sess = await openStream(connId, 'sess-1'); + await waitUntil(() => bridge.subscribeCalls.length >= 1); + const reader = frameReader(sess); + await post(connId, { + jsonrpc: '2.0', + id: 5, + method: 'session/prompt', + params: { sessionId: 'sess-1', prompt: [{ type: 'text', text: 'go' }] }, + }); + const f1 = (await reader.next()) as { method: string; id: unknown }; + const f2 = (await reader.next()) as { method: string; id: unknown }; + expect(f1.method).toBe('session/request_permission'); + expect(f2.method).toBe('session/request_permission'); + // SAME outbound JSON-RPC id ⇒ one pending entry reused, not a 2nd orphan. + expect(f1.id).toBe(f2.id); + reader.close(); + }); + it('permission request round-trips agent→client→agent', async () => { let resolvedWith: unknown; bridge.respondToSessionPermission = (( @@ -913,6 +1177,269 @@ describe('ACP Streamable HTTP transport (over the wire)', () => { } }); + it('cross-connection permission response resolves for a co-owned session', async () => { + let resolvedWith: unknown; + bridge.respondToSessionPermission = (( + sessionId: string, + requestId: string, + resp: unknown, + ) => { + resolvedWith = { sessionId, requestId, resp }; + return true; + }) as never; + bridge.promptBehavior = async (_s, q) => { + q.push({ + type: 'permission_request', + data: { + requestId: 'perm-cross', + sessionId: 'sess-1', + toolCall: { name: 'shell' }, + options: [{ optionId: 'allow', name: 'Allow' }], + }, + }); + await new Promise((r) => setTimeout(r, 30)); + return { stopReason: 'end_turn' }; + }; + const streamConnId = await initialize(); + await newSession(streamConnId); + const voterConnId = await initialize(); + await newSession(voterConnId, 100); + const sessStream = await openStream(streamConnId, 'sess-1'); + const reader = frameReader(sessStream); + try { + await post(streamConnId, { + jsonrpc: '2.0', + id: 7, + method: 'session/prompt', + params: { + sessionId: 'sess-1', + prompt: [{ type: 'text', text: 'rm' }], + }, + }); + const reqFrame = (await reader.next()) as { id: string }; + await post(voterConnId, { + jsonrpc: '2.0', + id: reqFrame.id, + result: { outcome: { outcome: 'selected', optionId: 'allow' } }, + }); + await waitUntil(() => resolvedWith !== undefined); + expect(resolvedWith).toEqual({ + sessionId: 'sess-1', + requestId: 'perm-cross', + resp: { outcome: { outcome: 'selected', optionId: 'allow' } }, + }); + } finally { + reader.close(); + } + }); + + it('cross-connection permission response is ignored without session ownership', async () => { + let resolvedWith: unknown; + bridge.respondToSessionPermission = (( + _sessionId: string, + _requestId: string, + resp: unknown, + ) => { + resolvedWith = resp; + return true; + }) as never; + bridge.promptBehavior = async (_s, q) => { + q.push({ + type: 'permission_request', + data: { + requestId: 'perm-unauthorized', + sessionId: 'sess-1', + toolCall: { name: 'shell' }, + options: [{ optionId: 'allow', name: 'Allow' }], + }, + }); + await new Promise((r) => setTimeout(r, 80)); + return { stopReason: 'end_turn' }; + }; + const streamConnId = await initialize(); + await newSession(streamConnId); + const voterConnId = await initialize(); + const sessStream = await openStream(streamConnId, 'sess-1'); + const reader = frameReader(sessStream); + try { + await post(streamConnId, { + jsonrpc: '2.0', + id: 7, + method: 'session/prompt', + params: { + sessionId: 'sess-1', + prompt: [{ type: 'text', text: 'rm' }], + }, + }); + const reqFrame = (await reader.next()) as { id: string }; + await post(voterConnId, { + jsonrpc: '2.0', + id: reqFrame.id, + result: { outcome: { outcome: 'selected', optionId: 'allow' } }, + }); + await new Promise((r) => setTimeout(r, 50)); + expect(resolvedWith).toBeUndefined(); + } finally { + reader.close(); + } + }); + + it('session/permission resolves by bridge request id and replies on the connection stream', async () => { + let resolvedWith: unknown; + bridge.respondToSessionPermission = (( + sessionId: string, + requestId: string, + resp: unknown, + ) => { + resolvedWith = { sessionId, requestId, resp }; + return true; + }) as never; + bridge.promptBehavior = async (_s, q) => { + q.push({ + type: 'permission_request', + data: { + requestId: 'perm-route', + sessionId: 'sess-1', + toolCall: { name: 'shell' }, + options: [{ optionId: 'allow', name: 'Allow' }], + }, + }); + await new Promise((r) => setTimeout(r, 30)); + return { stopReason: 'end_turn' }; + }; + const connId = await initialize(); + await newSession(connId); + const connStream = await openStream(connId); + const connReader = frameReader(connStream); + const sessStream = await openStream(connId, 'sess-1'); + const sessReader = frameReader(sessStream); + try { + await connReader.next(); // buffered session/new response + await post(connId, { + jsonrpc: '2.0', + id: 7, + method: 'session/prompt', + params: { + sessionId: 'sess-1', + prompt: [{ type: 'text', text: 'rm' }], + }, + }); + await sessReader.next(); + await post(connId, { + jsonrpc: '2.0', + id: 8, + method: 'session/permission', + params: { + sessionId: 'sess-1', + requestId: 'perm-route', + outcome: { outcome: 'selected', optionId: 'allow' }, + }, + }); + const ack = (await connReader.next()) as { + id: number; + result?: unknown; + }; + expect(ack).toEqual({ jsonrpc: '2.0', id: 8, result: {} }); + expect(resolvedWith).toEqual({ + sessionId: 'sess-1', + requestId: 'perm-route', + resp: { outcome: { outcome: 'selected', optionId: 'allow' } }, + }); + } finally { + connReader.close(); + sessReader.close(); + } + }); + + it('session/permission rejects an unowned session', async () => { + const ownerConnId = await initialize(); + await newSession(ownerConnId); + const otherConnId = await initialize(); + const otherStream = await openStream(otherConnId); + const otherReader = frameReader(otherStream); + try { + await post(otherConnId, { + jsonrpc: '2.0', + id: 9, + method: 'session/permission', + params: { + sessionId: 'sess-1', + requestId: 'perm-nope', + outcome: { outcome: 'selected', optionId: 'allow' }, + }, + }); + const ack = (await otherReader.next()) as { + id: number; + error?: { message?: string }; + }; + expect(ack.id).toBe(9); + expect(ack.error?.message).toContain('not owned'); + } finally { + otherReader.close(); + } + }); + + it('permission request during session-stream grace is buffered, not cancelled', () => { + vi.useFakeTimers(); + try { + const cancelCalls: unknown[] = []; + bridge.respondToSessionPermission = (( + sessionId: string, + requestId: string, + resp: unknown, + ) => { + cancelCalls.push({ sessionId, requestId, resp }); + return true; + }) as never; + const dispatcher = new AcpDispatcher( + bridge as unknown as HttpAcpBridge, + '/ws', + fakeWorkspace as unknown as DaemonWorkspaceService, + ); + const conn = new AcpConnection(undefined, true); + conn.ownSession('sess-1'); + conn.getOrCreateSession('sess-1').clientId = 'client-1'; + const stream = new FakeTransportStream(); + conn.attachSessionStream('sess-1', stream, new AbortController()); + conn.detachSessionStream('sess-1', stream, 10_000); + + type DispatcherWithTranslate = { + translateEvent( + targetConn: AcpConnection, + targetSessionId: string, + event: BridgeEvent, + ): void; + }; + (dispatcher as unknown as DispatcherWithTranslate).translateEvent( + conn, + 'sess-1', + { + v: 1, + id: 12, + type: 'permission_request', + data: { + requestId: 'perm-grace', + sessionId: 'sess-1', + toolCall: { name: 'shell' }, + options: [{ optionId: 'allow', name: 'Allow' }], + }, + }, + ); + + const binding = conn.sessions.get('sess-1'); + expect(cancelCalls).toEqual([]); + expect(conn.pending.size).toBe(1); + expect(binding?.buffer).toHaveLength(1); + expect(binding?.buffer[0]?.id).toBe(12); + expect( + (binding?.buffer[0]?.frame as { method?: string } | undefined)?.method, + ).toBe('session/request_permission'); + conn.closeSessionStream('sess-1'); + } finally { + vi.useRealTimers(); + } + }); + it('standard session/set_config_option (model) routes to the bridge', async () => { const connId = await initialize(); await newSession(connId); diff --git a/packages/cli/src/serve/acp-http/ws-stream.test.ts b/packages/cli/src/serve/acp-http/ws-stream.test.ts index fd1afb17831..a9dd09428e6 100644 --- a/packages/cli/src/serve/acp-http/ws-stream.test.ts +++ b/packages/cli/src/serve/acp-http/ws-stream.test.ts @@ -51,6 +51,19 @@ describe('WsStream', () => { stream.close(); }); + it('send() ignores the bus event id — no SSE `id:` framing on the WS wire', async () => { + // WsStream.send accepts `id` only for TransportStream parity; WebSocket is + // stateful and has no Last-Event-ID replay. The wire payload must be the + // bare JSON message — if a refactor ever let the id leak into the frame it + // would corrupt the WS protocol with SSE-specific framing. + const stream = new WsStream(ws as never); + await stream.send({ data: 1 }, 42); + expect(ws.sent).toEqual(['{"data":1}']); + expect(ws.sent[0]).not.toContain('id:'); + expect(ws.sent[0]).not.toContain('42'); + stream.close(); + }); + it('send() serializes writes sequentially (no interleaving)', async () => { const stream = new WsStream(ws as never); const p1 = stream.send({ seq: 1 }); diff --git a/packages/cli/src/serve/acp-http/ws-stream.ts b/packages/cli/src/serve/acp-http/ws-stream.ts index b62992cd381..67a0dda2380 100644 --- a/packages/cli/src/serve/acp-http/ws-stream.ts +++ b/packages/cli/src/serve/acp-http/ws-stream.ts @@ -52,7 +52,10 @@ export class WsStream implements TransportStream { this.heartbeat.unref(); } - send(message: unknown): Promise { + // `_id` (bus event id) is accepted for `TransportStream` parity but ignored: + // WebSocket is a stateful connection with no SSE `Last-Event-ID` replay + // (matches `AcpWsTransport.supportsReplay = false`). + send(message: unknown, _id?: number): Promise { const data = JSON.stringify(message); const next = this.writeChain.then( () => diff --git a/packages/cli/src/serve/sse-last-event-id.ts b/packages/cli/src/serve/sse-last-event-id.ts new file mode 100644 index 00000000000..92cee139ee0 --- /dev/null +++ b/packages/cli/src/serve/sse-last-event-id.ts @@ -0,0 +1,62 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { writeStderrLine } from '../utils/stdioHelpers.js'; + +/** Truncate + sanitize an untrusted header value for a single log line. */ +function safeLogValue(raw: unknown): string { + const s = typeof raw === 'string' ? raw : String(raw); + const clipped = s.length > 64 ? `${s.slice(0, 64)}…` : s; + // Strip ALL C0 control chars + DEL (covers CR/LF log-forging AND ANSI ESC + // `\x1b` / null bytes a crafted header could use to manipulate an operator's + // terminal when this value is written to stderr). Matching control chars in + // the regex is the intent here, so the lint rule is deliberately disabled. + // eslint-disable-next-line no-control-regex + return clipped.replace(/[\x00-\x1f\x7f]+/g, ' '); +} + +/** + * Parse a `Last-Event-ID` header into a bus event id for the ACP `GET /acp` + * SSE surface. + * + * NOTE: the REST `GET /session/:id/events` surface still has its own copy in + * `server/request-helpers.ts` (the two implement the same accept/reject rule). + * Unifying them onto this util is a worthwhile cleanup but is deliberately + * deferred: it would change the REST surface, and this PR keeps REST untouched + * (no behavioural side effects). Tracked as a follow-up. + * + * Stricter than `Number.parseInt`: accept ONLY pure decimal digits (so + * "1abc" / "1.5" don't silently parse to 1) and reject values past + * `Number.MAX_SAFE_INTEGER` (the EventBus's monotonic ids are bounded by it). + * Returns `undefined` for missing/invalid headers ⇒ live-only subscription. + * Rejections are logged with the offending value for operators; the common + * "first connect, no resume" case (missing/empty header) is silent. + * + * @param logPrefix distinguishes the surface in logs, e.g. `'/acp '` vs `''`. + */ +export function parseLastEventId( + raw: unknown, + logPrefix = '', +): number | undefined { + if (typeof raw !== 'string' || !/^\d+$/.test(raw)) { + if (typeof raw === 'string' && raw.length > 0) { + writeStderrLine( + `qwen serve: ${logPrefix}rejected Last-Event-ID ${safeLogValue(raw)} ` + + `(not a decimal integer)`, + ); + } + return undefined; + } + const n = Number.parseInt(raw, 10); + if (!Number.isFinite(n) || n > Number.MAX_SAFE_INTEGER) { + writeStderrLine( + `qwen serve: ${logPrefix}rejected Last-Event-ID ${safeLogValue(raw)} ` + + `(exceeds Number.MAX_SAFE_INTEGER)`, + ); + return undefined; + } + return n; +} diff --git a/packages/sdk-typescript/package.json b/packages/sdk-typescript/package.json index 867faccba01..574540848ee 100644 --- a/packages/sdk-typescript/package.json +++ b/packages/sdk-typescript/package.json @@ -17,6 +17,11 @@ "import": "./dist/daemon/index.js", "require": "./dist/daemon/index.cjs" }, + "./daemon/transports": { + "types": "./dist/daemon/transports.d.ts", + "import": "./dist/daemon/transports.js", + "require": "./dist/daemon/transports.cjs" + }, "./package.json": "./package.json" }, "bin": { diff --git a/packages/sdk-typescript/scripts/build.js b/packages/sdk-typescript/scripts/build.js index ecf6201a06d..a4e92c29570 100755 --- a/packages/sdk-typescript/scripts/build.js +++ b/packages/sdk-typescript/scripts/build.js @@ -37,6 +37,11 @@ const rootDir = join(__dirname, '..'); // Bumped from 130KB to 131KB for the workspace MCP resources drill-down // (workspaceMcpResources client method + route + resource status types). const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 131 * 1024; +// The opt-in `daemon/transports` browser bundle legitimately ships the concrete +// ACP transports (AcpHttpTransport/AcpWsTransport/AutoReconnect + negotiate), so +// it's larger than the default barrel — but still budgeted so a future PR can't +// silently bloat what browser consumers (agent-web) pull in. Current size ~29KB. +const MAX_TRANSPORTS_BROWSER_BUNDLE_BYTES = 48 * 1024; rmSync(join(rootDir, 'dist'), { recursive: true, force: true }); mkdirSync(join(rootDir, 'dist'), { recursive: true }); @@ -141,6 +146,47 @@ await esbuild.build({ treeShaking: true, }); +// Opt-in transports subpath (`@qwen-code/sdk/daemon/transports`): the concrete +// ACP transports + negotiateTransport. Kept out of the default daemon barrel +// (and its byte budget) so REST-only consumers stay tree-shaken; consumers who +// want resumable ACP-over-HTTP import this entry explicitly. Built as its own +// bundle for both browser (esm) and node (cjs) targets. +await esbuild.build({ + entryPoints: [join(rootDir, 'src', 'daemon', 'transports.ts')], + bundle: true, + format: 'esm', + platform: 'browser', + target: 'es2022', + outfile: join(rootDir, 'dist', 'daemon', 'transports.js'), + sourcemap: false, + minify: true, + minifyWhitespace: true, + minifyIdentifiers: true, + minifySyntax: true, + legalComments: 'none', + keepNames: false, + treeShaking: true, +}); + +assertTransportsBundle(join(rootDir, 'dist', 'daemon', 'transports.js')); + +await esbuild.build({ + entryPoints: [join(rootDir, 'src', 'daemon', 'transports.ts')], + bundle: true, + format: 'cjs', + platform: 'node', + target: 'node22', + outfile: join(rootDir, 'dist', 'daemon', 'transports.cjs'), + sourcemap: false, + minify: true, + minifyWhitespace: true, + minifyIdentifiers: true, + minifySyntax: true, + legalComments: 'none', + keepNames: false, + treeShaking: true, +}); + // Build serve-bridge CLI bin entry await esbuild.build({ entryPoints: [join(rootDir, 'src', 'daemon-mcp', 'serve-bridge', 'bin.ts')], @@ -172,10 +218,30 @@ function assertBrowserSafeBundle(filePath) { `Browser daemon SDK bundle is ${size} bytes; expected <= ${MAX_DAEMON_BROWSER_BUNDLE_BYTES}`, ); } + assertNoNodeBuiltins(filePath, 'Browser daemon SDK bundle'); +} + +// Browser-safety + size budget for the opt-in `daemon/transports` bundle. +// Larger budget than the default barrel (it ships the concrete transports), but +// still bounded so a future PR can't silently bloat what browser consumers pull. +function assertTransportsBundle(filePath) { + const size = statSync(filePath).size; + if (size > MAX_TRANSPORTS_BROWSER_BUNDLE_BYTES) { + throw new Error( + `Browser daemon transports bundle is ${size} bytes; expected <= ${MAX_TRANSPORTS_BROWSER_BUNDLE_BYTES}`, + ); + } + assertNoNodeBuiltins(filePath, 'Browser daemon transports bundle'); +} +// Node-builtin guard, shared by the budget-checked default daemon barrel and +// the opt-in `daemon/transports` bundle. The transports bundle is allowed to +// be larger (it ships the concrete ACP transports), but must still be +// browser-safe — agent-web consumes it in the browser. +function assertNoNodeBuiltins(filePath, label) { const contents = readFileSync(filePath, 'utf8'); if (contents.includes('node:')) { - throw new Error('Browser daemon SDK bundle contains Node-only token node:'); + throw new Error(`${label} contains Node-only token node:`); } const forbiddenBuiltins = [ 'assert', @@ -206,8 +272,6 @@ function assertBrowserSafeBundle(filePath) { ); const found = contents.match(requirePattern); if (found) { - throw new Error( - `Browser daemon SDK bundle contains Node-only token ${found[0]}`, - ); + throw new Error(`${label} contains Node-only token ${found[0]}`); } } diff --git a/packages/sdk-typescript/src/daemon/AcpHttpTransport.ts b/packages/sdk-typescript/src/daemon/AcpHttpTransport.ts index 43185af94a3..fc0982c5321 100644 --- a/packages/sdk-typescript/src/daemon/AcpHttpTransport.ts +++ b/packages/sdk-typescript/src/daemon/AcpHttpTransport.ts @@ -11,8 +11,11 @@ import type { DaemonTransportSubscribeOptions, } from './DaemonTransport.js'; import { DaemonTransportClosedError } from './DaemonTransport.js'; -import { parseSseStream } from './sse.js'; -import type { JsonRpcNotification } from './AcpEventDenormalizer.js'; +import { consumeFrames } from './sse.js'; +import { + denormalizeAcpNotification, + type JsonRpcNotification, +} from './AcpEventDenormalizer.js'; import { matchRoute, synthesizeResponse, @@ -22,6 +25,14 @@ import { mergeHeaders, } from './acpTransportUtils.js'; +/** + * Cap the unread SSE buffer of the session-stream parser. Mirrors + * `parseSseStream`'s `MAX_BUF_CHARS` — an unbounded buffer is a memory-pressure + * vector (a tab crash for browser consumers) if a server/proxy never emits a + * frame boundary or serves a non-SSE body. + */ +const MAX_SSE_BUF_CHARS = 16 * 1024 * 1024; + // --------------------------------------------------------------------------- // JSON-RPC types // --------------------------------------------------------------------------- @@ -49,6 +60,42 @@ interface PendingRequest { reject: (error: Error) => void; } +/** + * Map a `session/request_permission` JSON-RPC request (as the daemon sends it + * on the session-scoped `/acp` stream) to a `permission_request` DaemonEvent, + * mirroring what the REST surface emits so consumers handle it identically. + * The agent-stamped `requestId` (in `_meta.qwen.requestId`) is the correlator + * the eventual vote must echo (§1.7). Returns `undefined` if it can't be read. + */ +function permissionRequestToEvent( + msg: Record, + busId: number | undefined, +): DaemonEvent | undefined { + const params = isRecord(msg['params']) ? msg['params'] : {}; + const meta = isRecord(params['_meta']) ? params['_meta'] : undefined; + const qwenMeta = meta && isRecord(meta['qwen']) ? meta['qwen'] : undefined; + const requestId = + qwenMeta && typeof qwenMeta['requestId'] === 'string' + ? qwenMeta['requestId'] + : undefined; + if (!requestId) return undefined; + return { + id: busId, + v: 1, + type: 'permission_request', + data: { + requestId, + sessionId: + typeof params['sessionId'] === 'string' + ? params['sessionId'] + : undefined, + toolCall: params['toolCall'], + options: params['options'], + }, + _meta: meta, + }; +} + // --------------------------------------------------------------------------- // AcpHttpTransport // --------------------------------------------------------------------------- @@ -63,11 +110,20 @@ interface PendingRequest { * connection-scoped SSE stream at `GET /acp` for subsequent responses. * * Subsequent `POST /acp` requests return 202 (ack); the real JSON-RPC - * response rides the connection-scoped SSE stream. Responses are - * correlated by `id` using a `Map`. + * response rides an SSE stream. Responses are correlated by `id` using a + * `Map` shared across both streams. * - * Session events are received via a session-scoped SSE stream at - * `GET /acp` with appropriate headers (session filtering). + * Session events AND session-scoped JSON-RPC responses are received via the + * session-scoped SSE stream at `GET /acp` (with `Acp-Session-Id`), which is + * the resumable §1.8 stream the daemon's `replySession` routes session replies + * onto. `subscribeEvents` reads it and dispatches each frame: a JSON-RPC + * response resolves its pending request (so e.g. `session/prompt` doesn't hang + * waiting on a reply it would otherwise never observe), a notification becomes + * a `DaemonEvent`, and a `session/request_permission` request is surfaced as a + * `permission_request` event. Consumers answer with the normal REST-like SDK + * permission methods, which this transport maps to `session/permission` over + * `/acp`. The connection-scoped stream still carries replies to + * connection-level requests (e.g. `initialize`, `session/new`). */ export class AcpHttpTransport implements DaemonTransport { private readonly baseUrl: string; @@ -182,15 +238,21 @@ export class AcpHttpTransport implements DaemonTransport { await this.ensureInitialized(); - // Open a session-scoped SSE stream. For ACP HTTP, we use - // the daemon's per-session SSE endpoint — same URL as REST - // because ACP HTTP sessions still expose SSE for events. + // Open the SESSION-scoped `/acp` stream (GET /acp + Acp-Session-Id), NOT + // REST `/session/:id/events`. This is the resumable §1.8 stream and — the + // reason for this routing — the stream the daemon's `replySession` puts + // session-scoped JSON-RPC *responses* on. Reading it here is what lets a + // `session/prompt` reply resolve its pending request instead of hanging. const headers: Record = { Accept: 'text/event-stream', }; if (this.token) { headers['Authorization'] = `Bearer ${this.token}`; } + if (this.connectionId) { + headers['Acp-Connection-Id'] = this.connectionId; + } + headers['Acp-Session-Id'] = sessionId; if (opts.lastEventId !== undefined) { headers['Last-Event-ID'] = String(opts.lastEventId); } @@ -219,14 +281,12 @@ export class AcpHttpTransport implements DaemonTransport { ? composeAbortSignals([opts.signal, connectCtrl.signal]) : connectCtrl.signal; - let url = `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/events`; - if (opts.maxQueued !== undefined) { - url += `?maxQueued=${encodeURIComponent(String(opts.maxQueued))}`; - } - let res: Response; try { - res = await this._fetch(url, { headers, signal: fetchSignal }); + res = await this._fetch(`${this.baseUrl}/acp`, { + headers, + signal: fetchSignal, + }); } finally { if (connectTimer !== undefined) clearTimeout(connectTimer); } @@ -247,7 +307,7 @@ export class AcpHttpTransport implements DaemonTransport { body && typeof body === 'object' && 'error' in body ? String((body as { error: unknown }).error) : `HTTP ${res.status}`; - throw Object.assign(new Error(`GET /session/:id/events: ${detail}`), { + throw Object.assign(new Error(`GET /acp (session stream): ${detail}`), { status: res.status, body, }); @@ -262,7 +322,7 @@ export class AcpHttpTransport implements DaemonTransport { } throw Object.assign( new Error( - `GET /session/:id/events: expected content-type text/event-stream, got "${ct}"`, + `GET /acp (session stream): expected content-type text/event-stream, got "${ct}"`, ), { status: res.status, body: ct }, ); @@ -272,7 +332,126 @@ export class AcpHttpTransport implements DaemonTransport { throw new Error('SSE response has no body'); } - yield* parseSseStream(res.body, opts.signal); + // The `/acp` session stream carries RAW JSON-RPC frames (not REST + // `BridgeEvent` envelopes), so parse them directly and dispatch by shape. + // Each SSE frame may carry an `id:` line — the EventBus cursor we stamp + // onto yielded events so the consumer resumes from the REAL daemon id + // (the denormalizer's synthetic id is not resume-compatible); frames with + // no `id:` (synthetic terminals) yield `id: undefined`, which the consumer + // ignores for Last-Event-ID tracking. + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + const signal = opts.signal; + // `reader.read()` doesn't observe `signal` on its own — race it against an + // abort rejection so dispose()/caller-abort can unblock a hanging read. + const abortPromise = new Promise((_, reject) => { + if (signal?.aborted) { + reject(signal.reason ?? new DOMException('Aborted', 'AbortError')); + return; + } + signal?.addEventListener( + 'abort', + () => + reject(signal.reason ?? new DOMException('Aborted', 'AbortError')), + { once: true }, + ); + }); + + try { + while (!signal?.aborted) { + const { value, done } = await Promise.race([ + reader.read(), + abortPromise, + ]); + if (done) break; + buf += decoder.decode(value, { stream: true }); + if (buf.length > MAX_SSE_BUF_CHARS) { + throw new Error( + `AcpHttpTransport: unread SSE buffer exceeded ${MAX_SSE_BUF_CHARS} ` + + `bytes without a frame boundary`, + ); + } + + // Reuse the shared CRLF-aware frame splitter (handles both `\n\n` and + // `\r\n\r\n`) instead of reimplementing it. + const { frames, tail } = consumeFrames(buf); + buf = tail; + for (const rawFrame of frames) { + let busId: number | undefined; + const dataParts: string[] = []; + for (const rawLine of rawFrame.split('\n')) { + // Strip a trailing CR so CRLF line endings don't corrupt JSON.parse. + const line = rawLine.endsWith('\r') + ? rawLine.slice(0, -1) + : rawLine; + if (line.startsWith('id:')) { + const n = Number(line.slice(3).trim()); + if (Number.isInteger(n)) busId = n; + } else if (line.startsWith('data:')) { + // Per the SSE spec, multiple `data:` lines in one event join with + // a newline. + dataParts.push(line.slice('data:'.length).replace(/^ /, '')); + } + } + if (dataParts.length === 0) continue; + const dataLine = dataParts.join('\n'); + + let msg: unknown; + try { + msg = JSON.parse(dataLine); + } catch { + continue; // heartbeat / non-JSON + } + if (!isRecord(msg)) continue; + + const hasId = 'id' in msg; + const method = (msg as { method?: unknown }).method; + + // (1) JSON-RPC response (id, no method) → resolve the pending request. + // THIS is the W2 fix: a `session/prompt` reply routed here by the + // daemon's `replySession` now settles its promise instead of hanging. + if (hasId && typeof method !== 'string') { + const rid = (msg as { id: unknown }).id; + if (typeof rid === 'number') { + const pending = this.pending.get(rid); + if (pending) { + this.pending.delete(rid); + pending.resolve(msg as unknown as JsonRpcResponse); + } + } + continue; + } + + // (2) Agent→client permission request → surface as an event so the + // consumer can show it and answer via respondToSessionPermission(). + if (method === 'session/request_permission') { + const ev = permissionRequestToEvent(msg, busId); + if (ev) yield ev; + continue; + } + + // (3) Notification → DaemonEvent, stamped with the real bus cursor. + if (typeof method === 'string' && !hasId) { + const ev = denormalizeAcpNotification( + msg as unknown as JsonRpcNotification, + ); + if (ev) { + ev.id = busId; // authoritative cursor (or undefined → ignored) + yield ev; + } + continue; + } + // else: unrecognized frame → ignore + } + } + } finally { + try { + reader.cancel().catch(() => {}); + } catch { + /* already closed */ + } + } } dispose(): void { diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index 4d3ba9b03bd..ceafe8d4fcb 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -28,16 +28,17 @@ export type { DaemonTransportType, } from './DaemonTransport.js'; export type { RestSseTransport } from './RestSseTransport.js'; -// negotiateTransport + ACP transport classes live in their own files to -// break the static import chain from this barrel, keeping the browser -// bundle under budget. Monorepo consumers import from source paths: -// import { negotiateTransport } from '../../sdk-typescript/src/daemon/negotiateTransport.js'; -// import { AcpWsTransport } from '../../sdk-typescript/src/daemon/AcpWsTransport.js'; -// import { AcpHttpTransport } from '../../sdk-typescript/src/daemon/AcpHttpTransport.js'; -// import { AutoReconnectTransport } from '../../sdk-typescript/src/daemon/AutoReconnectTransport.js'; -// Deep package exports are intentionally omitted: the SDK barrel does not -// re-export these classes, and the package's `files` field ships only -// `dist/` which does not include per-module entry points for them. +// negotiateTransport + the concrete ACP transport classes are intentionally +// NOT re-exported here: a static import chain from this barrel would pull +// their framing/SSE code into the budget-checked browser bundle (see +// `scripts/build.js` MAX_DAEMON_BROWSER_BUNDLE_BYTES). They ship instead +// behind the opt-in `@qwen-code/sdk/daemon/transports` subpath +// (`./transports.ts`), so REST-only consumers stay tree-shaken while +// consumers who want resumable ACP-over-HTTP get a first-class import: +// import { negotiateTransport, AcpHttpTransport } +// from '@qwen-code/sdk/daemon/transports'; +// The `NegotiateTransportOptions` *type* stays available from this barrel +// for backward compatibility (type-only, no bundle cost). export type { NegotiateTransportOptions } from './negotiateTransport.js'; export type { JsonRpcNotification } from './AcpEventDenormalizer.js'; export type { TransportFactory } from './AutoReconnectTransport.js'; diff --git a/packages/sdk-typescript/src/daemon/negotiateTransport.ts b/packages/sdk-typescript/src/daemon/negotiateTransport.ts index 6af580f98fb..08021f6579f 100644 --- a/packages/sdk-typescript/src/daemon/negotiateTransport.ts +++ b/packages/sdk-typescript/src/daemon/negotiateTransport.ts @@ -14,6 +14,13 @@ import type { DaemonTransport } from './DaemonTransport.js'; export interface NegotiateTransportOptions { /** Timeout for the capabilities probe and WS handshake. Default 5000ms. */ probeTimeoutMs?: number; + /** + * `fetch` implementation used for the capabilities probe and threaded + * into the constructed REST / ACP-HTTP transport. Defaults to the + * global `fetch`. Supply this to inject auth headers, a proxy agent, or + * a test double in environments where the global isn't what you want. + */ + fetchFn?: typeof globalThis.fetch; } /** @@ -41,7 +48,7 @@ export async function negotiateTransport( token?: string, opts?: NegotiateTransportOptions, ): Promise { - const fetchFn = globalThis.fetch.bind(globalThis); + const fetchFn = opts?.fetchFn ?? globalThis.fetch.bind(globalThis); const probeTimeoutMs = opts?.probeTimeoutMs ?? 5_000; // Lazy imports to avoid circular module initialization. These diff --git a/packages/sdk-typescript/src/daemon/sse.ts b/packages/sdk-typescript/src/daemon/sse.ts index 70320403e78..99156d5abea 100644 --- a/packages/sdk-typescript/src/daemon/sse.ts +++ b/packages/sdk-typescript/src/daemon/sse.ts @@ -186,8 +186,14 @@ export async function* parseSseStream( /** * Walk `buf` and pull off every complete frame (either `\n\n` or * `\r\n\r\n` separator). Returns the frames + the unconsumed tail. + * + * Exported so other SSE readers (e.g. the ACP transport's raw JSON-RPC frame + * parser) reuse this CRLF-aware boundary scan instead of reimplementing it. */ -function consumeFrames(buf: string): { frames: string[]; tail: string } { +export function consumeFrames(buf: string): { + frames: string[]; + tail: string; +} { const frames: string[] = []; let cursor = 0; // BX9_a + BeFHR + BeFId: scan for `\n\n` first; on hit, look for diff --git a/packages/sdk-typescript/src/daemon/transports.ts b/packages/sdk-typescript/src/daemon/transports.ts new file mode 100644 index 00000000000..0b91e72eff4 --- /dev/null +++ b/packages/sdk-typescript/src/daemon/transports.ts @@ -0,0 +1,42 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Opt-in transport surface: `@qwen-code/sdk/daemon/transports`. + * + * The default `@qwen-code/sdk/daemon` barrel deliberately ships only the + * `DaemonTransport` interface and the lightweight `RestSseTransport` *type* + * so its browser bundle stays under budget (see `scripts/build.js` + * `MAX_DAEMON_BROWSER_BUNDLE_BYTES`). The concrete ACP transports — + * `AcpHttpTransport` (native `supportsReplay` + `Last-Event-ID` resume), + * `AcpWsTransport`, the `AutoReconnectTransport` wrapper, and the + * `negotiateTransport` factory — pull in their own framing/SSE code, so + * they live behind this separate subpath. Consumers that only need REST + * never pay for them; consumers that want resumable ACP-over-HTTP opt in + * with one import: + * + * ```ts + * import { + * negotiateTransport, + * AcpHttpTransport, + * } from '@qwen-code/sdk/daemon/transports'; + * + * const transport = await negotiateTransport(baseUrl, token); + * const client = new DaemonClient({ baseUrl, token, transport }); + * ``` + */ + +export { AcpHttpTransport } from './AcpHttpTransport.js'; +export { AcpWsTransport } from './AcpWsTransport.js'; +export { + AutoReconnectTransport, + type TransportFactory, +} from './AutoReconnectTransport.js'; +export { RestSseTransport } from './RestSseTransport.js'; +export { + negotiateTransport, + type NegotiateTransportOptions, +} from './negotiateTransport.js'; diff --git a/packages/sdk-typescript/test/unit/AcpHttpTransport.test.ts b/packages/sdk-typescript/test/unit/AcpHttpTransport.test.ts index e48b151eed2..020c634bbc5 100644 --- a/packages/sdk-typescript/test/unit/AcpHttpTransport.test.ts +++ b/packages/sdk-typescript/test/unit/AcpHttpTransport.test.ts @@ -380,6 +380,34 @@ describe('AcpHttpTransport', () => { transport.dispose(); }); + + it('POST /session/:id/permission/:requestId sends session/permission', async () => { + const { fetch, calls } = initAwareFetch(); + const transport = new AcpHttpTransport('http://d', undefined, fetch); + + const res = await transport.fetch('http://d/session/s1/permission/p1', { + method: 'POST', + body: JSON.stringify({ + outcome: { outcome: 'selected', optionId: 'allow' }, + }), + }); + + expect(res.status).toBe(200); + const permissionCall = calls + .filter((c) => c.url.endsWith('/acp') && c.method === 'POST') + .find((c) => { + if (!c.body) return false; + return JSON.parse(c.body).method === 'session/permission'; + }); + expect(permissionCall).toBeDefined(); + expect(JSON.parse(permissionCall!.body!).params).toEqual({ + sessionId: 's1', + requestId: 'p1', + outcome: { outcome: 'selected', optionId: 'allow' }, + }); + + transport.dispose(); + }); }); // ---- Error handling --------------------------------------------------- @@ -571,3 +599,131 @@ describe('AcpHttpTransport', () => { }); }); }); + +// A streamed text/event-stream Response that emits the given raw SSE frames +// then closes, so subscribeEvents' read loop ends and the generator returns. +function sseResponse(frames: string[]): Response { + const enc = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + for (const f of frames) controller.enqueue(enc.encode(f)); + controller.close(); + }, + }); + return new Response(stream, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }); +} + +describe('AcpHttpTransport — subscribeEvents (session-scoped /acp stream)', () => { + // One SSE frame: optional `id:` (bus cursor) + a `data:` JSON-RPC payload. + function frame(id: number | undefined, msg: unknown): string { + const idLine = id !== undefined ? `id: ${id}\n` : ''; + return `${idLine}data: ${JSON.stringify(msg)}\n\n`; + } + + function sessionStreamFetch(frames: string[]) { + return initAwareFetch({ + connectionIdHeader: 'conn-1', + subsequentReply: (req) => + req.method === 'GET' && req.headers['acp-session-id'] + ? sseResponse(frames) + : jsonResponse(200, { jsonrpc: '2.0', id: 1, result: { ok: true } }), + }); + } + + async function collect( + t: AcpHttpTransport, + sessionId: string, + ): Promise> { + const out: Array<{ id?: number; type: string; data: unknown }> = []; + for await (const e of t.subscribeEvents(sessionId)) { + out.push(e as { id?: number; type: string; data: unknown }); + } + return out; + } + + it('opens GET /acp with Acp-Session-Id + Acp-Connection-Id (not REST /session/:id/events)', async () => { + const { fetch, calls } = sessionStreamFetch([]); + const t = new AcpHttpTransport('http://d', undefined, fetch); + await collect(t, 'sess-1'); + + const getCall = calls.find( + (c) => c.method === 'GET' && c.url.endsWith('/acp'), + ); + expect(getCall).toBeDefined(); + expect(getCall?.headers['acp-session-id']).toBe('sess-1'); + expect(getCall?.headers['acp-connection-id']).toBe('conn-1'); + // Must NOT use the REST session-events endpoint. + expect(calls.some((c) => c.url.includes('/session/'))).toBe(false); + }); + + it('yields a session/update notification as a DaemonEvent stamped with the bus id from the `id:` line', async () => { + const { fetch } = sessionStreamFetch([ + frame(42, { + jsonrpc: '2.0', + method: 'session/update', + params: { + sessionId: 'sess-1', + update: { sessionUpdate: 'agent_message_chunk', text: 'hi' }, + }, + }), + ]); + const t = new AcpHttpTransport('http://d', undefined, fetch); + const events = await collect(t, 'sess-1'); + + expect(events).toHaveLength(1); + expect(events[0].type).toBe('agent_message_chunk'); + expect(events[0].id).toBe(42); // real bus cursor, not the synthetic id + }); + + it('consumes a JSON-RPC response frame (routes to pending) WITHOUT yielding it as an event — the W2 no-hang dispatch', async () => { + // A response frame (id, no method) is dispatched to pending-resolution + // (same path as the connection stream), not surfaced as a DaemonEvent. + // The following notification proves the stream keeps flowing past it. + const { fetch } = sessionStreamFetch([ + frame(undefined, { + jsonrpc: '2.0', + id: 999, + result: { stopReason: 'end_turn' }, + }), + frame(7, { + jsonrpc: '2.0', + method: 'session/update', + params: { + sessionId: 'sess-1', + update: { sessionUpdate: 'agent_thought_chunk', text: 't' }, + }, + }), + ]); + const t = new AcpHttpTransport('http://d', undefined, fetch); + const events = await collect(t, 'sess-1'); + + expect(events).toHaveLength(1); + expect(events[0].type).toBe('agent_thought_chunk'); + }); + + it('surfaces a session/request_permission request as a permission_request event', async () => { + const { fetch } = sessionStreamFetch([ + frame(9, { + jsonrpc: '2.0', + id: 5, + method: 'session/request_permission', + params: { + sessionId: 'sess-1', + toolCall: { name: 'write_file' }, + options: [{ optionId: 'allow' }], + _meta: { qwen: { requestId: 'req-1' } }, + }, + }), + ]); + const t = new AcpHttpTransport('http://d', undefined, fetch); + const events = await collect(t, 'sess-1'); + + expect(events).toHaveLength(1); + expect(events[0].type).toBe('permission_request'); + expect((events[0].data as { requestId: string }).requestId).toBe('req-1'); + expect(events[0].id).toBe(9); + }); +}); diff --git a/packages/sdk-typescript/test/unit/daemon-transports-surface.test.ts b/packages/sdk-typescript/test/unit/daemon-transports-surface.test.ts new file mode 100644 index 00000000000..76d0265b26a --- /dev/null +++ b/packages/sdk-typescript/test/unit/daemon-transports-surface.test.ts @@ -0,0 +1,69 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, it, expect, expectTypeOf } from 'vitest'; +import { readFileSync } from 'node:fs'; +import { fileURLToPath } from 'node:url'; +import { dirname, join } from 'node:path'; +import * as Transports from '../../src/daemon/transports.js'; +import type { + NegotiateTransportOptions, + TransportFactory, +} from '../../src/daemon/transports.js'; + +const here = dirname(fileURLToPath(import.meta.url)); +const pkgPath = join(here, '..', '..', 'package.json'); + +describe('@qwen-code/sdk/daemon/transports — opt-in transport surface', () => { + it('exports the concrete ACP transports + negotiateTransport at runtime', () => { + // Locks the consumer-facing contract that lets agent-web (and any + // external SDK consumer) get resumable ACP-over-HTTP without forking + // or reaching into source paths. These deliberately live OFF the + // default `./daemon` barrel to keep its browser bundle under budget; + // if a future barrel reshuffle drops them here, this fails loudly. + expect(typeof Transports.AcpHttpTransport).toBe('function'); + expect(typeof Transports.AcpWsTransport).toBe('function'); + expect(typeof Transports.AutoReconnectTransport).toBe('function'); + expect(typeof Transports.RestSseTransport).toBe('function'); + expect(typeof Transports.negotiateTransport).toBe('function'); + }); + + it('AcpHttpTransport advertises native replay (supportsReplay)', () => { + // The whole reason to expose this transport: it natively sends + // Last-Event-ID on reconnect, which is what closes the §1.8 + // mid-turn content-loss gap against the resumable daemon stream. + const t = new Transports.AcpHttpTransport( + 'http://localhost:0', + undefined, + globalThis.fetch.bind(globalThis), + ); + expect(t.type).toBe('acp-http'); + expect(t.supportsReplay).toBe(true); + }); + + it('exposes the transport option types at the subpath (compile-time)', () => { + expectTypeOf().not.toBeNever(); + expectTypeOf().not.toBeNever(); + // The fetchFn injection point must stay on the negotiate options. + expectTypeOf< + NonNullable + >().toEqualTypeOf(); + }); + + it('declares the ./daemon/transports subpath in package.json exports', () => { + // The runtime imports above resolve via the bundler's source mapping; + // this pins the *published* contract so the subpath actually ships. + const pkg = JSON.parse(readFileSync(pkgPath, 'utf8')) as { + exports: Record>; + }; + const entry = pkg.exports['./daemon/transports']; + expect(entry).toBeDefined(); + // Bracket access: `entry` is typed via an index signature. + expect(entry['types']).toBe('./dist/daemon/transports.d.ts'); + expect(entry['import']).toBe('./dist/daemon/transports.js'); + expect(entry['require']).toBe('./dist/daemon/transports.cjs'); + }); +});