diff --git a/.qwen/design/prompt-queue-backpressure.md b/.qwen/design/prompt-queue-backpressure.md new file mode 100644 index 0000000000..1617cc01ac --- /dev/null +++ b/.qwen/design/prompt-queue-backpressure.md @@ -0,0 +1,83 @@ +# Prompt Queue Backpressure + +## Summary + +`qwen serve` now applies per-session prompt admission backpressure. The default limit is `5` pending prompts per session. A pending prompt is one that the daemon has accepted through `sendPrompt` and that has not settled yet, including prompts waiting in the per-session FIFO and the prompt currently executing. + +`branchSession` remains serialized behind the same per-session FIFO, but it is not a prompt and does not consume this prompt limit. + +## Semantics + +- Default: `maxPendingPromptsPerSession = 5`. +- Disabled: `0` or `Infinity` means unlimited. +- Invalid: negative numbers, fractions, and `NaN` are rejected by bridge construction and `runQwenServe`. The CLI flag accepts non-negative integers; `0` disables the cap. +- Authority: the bridge is the admission gate. SDK-side accounting is an early-fail guard, not a replacement for server enforcement. +- Prompt deadline: `--prompt-deadline-ms` still applies only to prompts that were already accepted. It is not a queue admission cap. + +## Bridge Behavior + +`SessionEntry` tracks `pendingPromptCount`. `sendPrompt` is intentionally not `async`, so the admission check can throw synchronously before HTTP routes return `202 Accepted`. + +Admission flow: + +1. Look up the session. +2. Reject pre-aborted signals before incrementing the counter. +3. If `pendingPromptCount >= maxPendingPromptsPerSession`, throw `PromptQueueFullError`. +4. Increment the counter and enqueue the prompt on the FIFO. +5. Release the slot exactly once when the caller-visible prompt promise settles. + +Failures do not poison the FIFO because the queue tail still swallows each prompt result. The original caller still receives the prompt rejection. + +## HTTP Behavior + +`POST /session/:id/prompt` catches synchronous `PromptQueueFullError` before emitting an accepted response. The route returns: + +- Status: `503` +- Header: `Retry-After: 5` +- Body: `{ code: 'prompt_queue_full', error, sessionId, limit, pendingCount }` + +No `promptId` is returned when admission fails. + +`/capabilities` advertises: + +```json +{ + "limits": { + "maxPendingPromptsPerSession": 5 + } +} +``` + +When the cap is disabled, the advertised value is `null`. + +## ACP HTTP Behavior + +The ACP JSON-RPC transport maps `PromptQueueFullError` to a stable error shape instead of falling through to an unstructured internal error: + +```json +{ + "data": { + "errorKind": "prompt_queue_full", + "sessionId": "...", + "limit": 5, + "pendingCount": 5 + } +} +``` + +## SDK Behavior + +`DaemonClient` has a local per-session reservation for `prompt()` calls. It reserves before sending the HTTP request and releases on: + +- legacy blocking `200` completion, +- non-blocking `202` turn completion, +- `turn_error`, +- caller abort, +- SSE end, +- fetch or response parsing failure. + +`DaemonPendingPromptLimitError` means the SDK rejected locally and did not send the prompt request. + +The SDK option accepts the numeric capability value directly; `null` disables the local cap to match `/capabilities.limits.maxPendingPromptsPerSession`. + +`DaemonSessionClient` applies the same local limit for the long-lived subscription path. Static `createOrAttach`, `load`, and `resume` keep their existing parameter positions; direct construction may override the local cap. diff --git a/.qwen/e2e-tests/prompt-queue-backpressure.md b/.qwen/e2e-tests/prompt-queue-backpressure.md new file mode 100644 index 0000000000..719f340a60 --- /dev/null +++ b/.qwen/e2e-tests/prompt-queue-backpressure.md @@ -0,0 +1,55 @@ +# Prompt Queue Backpressure E2E Test Plan + +## Scope + +Validate per-session prompt admission backpressure for `qwen serve`, REST clients, ACP HTTP clients, and the TypeScript SDK. + +## Baseline + +1. Start `qwen serve` with defaults. +2. Create a session. +3. Send one prompt. +4. Expected: prompt is accepted and the session emits normal turn events. + +## Full Queue + +1. Start `qwen serve` with defaults. +2. Create a session. +3. Hold one prompt active and enqueue four more prompts for the same session. +4. Send the sixth prompt. +5. Expected: the sixth request returns HTTP `503`, `Retry-After: 5`, and `code: "prompt_queue_full"`. The body includes `sessionId`, `limit: 5`, and `pendingCount: 5`. The response does not include `promptId`. + +## Release Then Recover + +1. Fill the default five pending prompt slots. +2. Let the active prompt complete or fail. +3. Send another prompt. +4. Expected: the new prompt is accepted after the previous slot is released. + +## ACP HTTP + +1. Send `session/prompt` through `/acp` while the same session has five pending prompts. +2. Expected: JSON-RPC returns stable error data with `errorKind: "prompt_queue_full"`, `limit`, `pendingCount`, and `sessionId`. + +## SDK Local Guard + +1. Construct `DaemonClient` with `maxPendingPromptsPerSession: 1`. +2. Use a daemon or fetch mock that accepts the first prompt with `202` and keeps its SSE stream pending. +3. Call `prompt()` again for the same session. +4. Expected: the SDK throws `DaemonPendingPromptLimitError` and does not issue the second fetch. + +## Disabled Cap + +1. Start `qwen serve --max-pending-prompts-per-session 0`. +2. Create a session. +3. Enqueue more than five prompts for the same session. +4. Expected: admission is not rejected by the prompt queue cap. `/capabilities.limits.maxPendingPromptsPerSession` is `null`. + +## Verification Commands + +```bash +cd packages/acp-bridge && npx vitest run src/bridge.test.ts +cd packages/cli && npx vitest run src/serve/server.test.ts src/serve/acpHttp/transport.test.ts +cd packages/sdk-typescript && npx vitest run test/unit/DaemonClient.test.ts test/unit/DaemonSessionClient.test.ts +npm run build && npm run typecheck +``` diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index 4026d5c1f3..6762659164 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -73,6 +73,7 @@ curl http://127.0.0.1:4170/capabilities ``` The `workspaceCwd` field surfaces the bound workspace so clients can pre-flight check + omit `cwd` on `POST /session`. +The `limits.maxPendingPromptsPerSession` field advertises the active per-session prompt admission cap; `null` means the cap is disabled. The daemon also exposes read-only runtime snapshots for client UIs: `GET /workspace/mcp`, `GET /workspace/skills`, `GET /workspace/providers`, @@ -202,20 +203,21 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 ## CLI flags -| Flag | Default | Purpose | -| ------------------------- | --------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `--port ` | `4170` | TCP port. `0` = OS-assigned ephemeral port. | -| `--hostname ` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. | -| `--token ` | — | Bearer token. Falls back to `QWEN_SERVER_TOKEN` env var (with leading/trailing whitespace stripped — handy for `$(cat token.txt)`). | -| `--require-auth` | `false` | Refuse to start without a bearer token, even on loopback. Hardens the `127.0.0.1` developer default for shared dev hosts / CI runners / multi-tenant workstations where any local user can hit the listener. Boots only with `--token` or `QWEN_SERVER_TOKEN` set; gates `/health` behind the bearer too. | -| `--max-sessions ` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~30–50 MB per session). | -| `--workspace ` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. | -| `--max-connections ` | `256` | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session. | -| `--event-ring-size ` | `8000` | Per-session SSE replay ring depth (#3803 §02 target). Sets the backlog available to `GET /session/:id/events` with `Last-Event-ID: N`. Larger = more reconnect headroom at the cost of a few hundred KB extra RAM per session. SDK clients can additionally request a larger per-subscriber backlog cap on a specific subscription via `?maxQueued=N` (range `[16, 2048]`, default 256). Daemons also emit a non-terminal `slow_client_warning` SSE frame at 75% queue fill so clients can drain / reconnect before getting evicted. Pre-flight `caps.features.slow_client_warning`. | -| `--mcp-client-budget ` | — | Positive integer cap on live MCP clients **per ACP session** (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14 v1; PR 23 graduates this to per-workspace via the shared MCP pool). Combine with `--mcp-budget-mode`. When unset, no accounting-driven enforcement (but `GET /workspace/mcp` still reports `clientCount`). Distinct from claude-code's `MCP_SERVER_CONNECTION_BATCH_SIZE` which gates startup concurrency, not the total client count. Pre-flight `caps.features.mcp_guardrails`. | -| `--mcp-budget-mode ` | `warn` / `off` | How `--mcp-client-budget` is enforced. `warn` (default when budget set): no refusal, snapshot's `budgets[0].status` flips to `warning` at ≥75% of budget. `enforce`: connects past the cap are refused, per-server cell shows `disabledReason: 'budget'`, deterministic by `mcpServers` declaration order. `off` (default when budget unset): pure observability. Boot rejects `enforce` without a budget. | -| `--http-bridge` | `true` | Stage 1 mode: one `qwen --acp` child per daemon (bound to one workspace at boot, per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02); N sessions multiplex onto that child via ACP `newSession()`. Stage 2 native in-process becomes available later. | -| `--allow-origin ` | — | T2.4 ([#4514](https://github.com/QwenLM/qwen-code/issues/4514)). Cross-origin allowlist for browser webui clients. Repeatable. Each value is `*` (any origin — boot refuses if no bearer token is configured; `--require-auth` on loopback is recommended so `/health` and `/demo` are also bearer-gated, since both are pre-auth on loopback by default) or a canonical URL origin (`://[:]`, no trailing slash / path / userinfo / query). **Subdomain wildcards (`https://*.example.com`) are intentionally unsupported** — list each subdomain explicitly, or use `*` with a configured token (and `--require-auth` for full hardening). Matched origins receive CORS response headers (`Access-Control-Allow-Origin`, `Vary: Origin`, methods, headers, max-age, and exposed `Retry-After`); unmatched origins still get a 403 with the same envelope as today's wall. `Origin: null` (sandboxed iframes, file:// docs) is always rejected, even under `*`. Pre-flight via `caps.features.allow_origin`. Loopback self-origin hits are unaffected. | +| Flag | Default | Purpose | +| --------------------------------------- | --------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `--port ` | `4170` | TCP port. `0` = OS-assigned ephemeral port. | +| `--hostname ` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. | +| `--token ` | — | Bearer token. Falls back to `QWEN_SERVER_TOKEN` env var (with leading/trailing whitespace stripped — handy for `$(cat token.txt)`). | +| `--require-auth` | `false` | Refuse to start without a bearer token, even on loopback. Hardens the `127.0.0.1` developer default for shared dev hosts / CI runners / multi-tenant workstations where any local user can hit the listener. Boots only with `--token` or `QWEN_SERVER_TOKEN` set; gates `/health` behind the bearer too. | +| `--max-sessions ` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~30–50 MB per session). | +| `--max-pending-prompts-per-session ` | `5` | Per-session cap on prompts accepted by `POST /session/:id/prompt` but not yet settled, including queued prompts and the active prompt. The bridge rejects overflow synchronously with `503`, `Retry-After: 5`, and `code: "prompt_queue_full"` before returning a `promptId`. Set to `0` to disable. `branchSession` serializes on the same FIFO but does not count against this prompt cap. | +| `--workspace ` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. | +| `--max-connections ` | `256` | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session. | +| `--event-ring-size ` | `8000` | Per-session SSE replay ring depth (#3803 §02 target). Sets the backlog available to `GET /session/:id/events` with `Last-Event-ID: N`. Larger = more reconnect headroom at the cost of a few hundred KB extra RAM per session. SDK clients can additionally request a larger per-subscriber backlog cap on a specific subscription via `?maxQueued=N` (range `[16, 2048]`, default 256). Daemons also emit a non-terminal `slow_client_warning` SSE frame at 75% queue fill so clients can drain / reconnect before getting evicted. Pre-flight `caps.features.slow_client_warning`. | +| `--mcp-client-budget ` | — | Positive integer cap on live MCP clients **per ACP session** (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14 v1; PR 23 graduates this to per-workspace via the shared MCP pool). Combine with `--mcp-budget-mode`. When unset, no accounting-driven enforcement (but `GET /workspace/mcp` still reports `clientCount`). Distinct from claude-code's `MCP_SERVER_CONNECTION_BATCH_SIZE` which gates startup concurrency, not the total client count. Pre-flight `caps.features.mcp_guardrails`. | +| `--mcp-budget-mode ` | `warn` / `off` | How `--mcp-client-budget` is enforced. `warn` (default when budget set): no refusal, snapshot's `budgets[0].status` flips to `warning` at ≥75% of budget. `enforce`: connects past the cap are refused, per-server cell shows `disabledReason: 'budget'`, deterministic by `mcpServers` declaration order. `off` (default when budget unset): pure observability. Boot rejects `enforce` without a budget. | +| `--http-bridge` | `true` | Stage 1 mode: one `qwen --acp` child per daemon (bound to one workspace at boot, per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02); N sessions multiplex onto that child via ACP `newSession()`. Stage 2 native in-process becomes available later. | +| `--allow-origin ` | — | T2.4 ([#4514](https://github.com/QwenLM/qwen-code/issues/4514)). Cross-origin allowlist for browser webui clients. Repeatable. Each value is `*` (any origin — boot refuses if no bearer token is configured; `--require-auth` on loopback is recommended so `/health` and `/demo` are also bearer-gated, since both are pre-auth on loopback by default) or a canonical URL origin (`://[:]`, no trailing slash / path / userinfo / query). **Subdomain wildcards (`https://*.example.com`) are intentionally unsupported** — list each subdomain explicitly, or use `*` with a configured token (and `--require-auth` for full hardening). Matched origins receive CORS response headers (`Access-Control-Allow-Origin`, `Vary: Origin`, methods, headers, max-age, and exposed `Retry-After`); unmatched origins still get a 403 with the same envelope as today's wall. `Origin: null` (sandboxed iframes, file:// docs) is always rejected, even under `*`. Pre-flight via `caps.features.allow_origin`. Loopback self-origin hits are unaffected. | > **Sizing the load knobs.** `--max-sessions` is the **new-child** cap. > Three other layers also limit load — when sizing for a high-concurrency @@ -226,14 +228,19 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 > - **per-session subscribers**: the EventBus caps SSE subscribers at > 64 per session by default; the 65th client gets a terminal > `stream_error` and is closed. +> - **per-session prompt admissions**: +> `--max-pending-prompts-per-session=5` bounds queued + active prompts +> accepted for one session. Overflow gets `503` with `Retry-After: 5`. > - **per-subscriber backlog**: a 256-frame queue per SSE client; an > over-capacity client gets a terminal `client_evicted` frame and is > closed (one slow consumer can't pin the daemon). > -> The four caps interact: `--max-sessions × 64 subscribers × 256 frames` -> is the worst-case in-flight memory at the EventBus layer. Default -> sizing assumes single-user / small-team load; raise progressively -> (and watch RSS) for multi-tenant deployments. +> These caps interact: `--max-sessions × 64 subscribers × 256 frames` +> is the worst-case in-flight memory at the EventBus layer, while +> `--max-sessions × --max-pending-prompts-per-session` bounds accepted +> prompt work at the admission layer. Default sizing assumes single-user / +> small-team load; raise progressively (and watch RSS) for multi-tenant +> deployments. > **MCP client guardrails (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14).** A workspace declaring 30 MCP servers in `mcpServers` will start 30 clients with no upstream cap unless you set one. `--mcp-client-budget=N` caps the live MCP client count; `--mcp-budget-mode={enforce,warn,off}` chooses the behavior. Default is `warn` when a budget is set (snapshot surfaces the warning but no client is refused — useful for measuring real-world fanout before flipping on enforcement). Refused servers under `enforce` mode get `disabledReason: 'budget'` on their per-server cell, and the `budgets[0]` cell shows `status: 'error'` + `errorKind: 'budget_exhausted'`. Slot reservation is by server name and survives reconnects / discovery timeouts — a refused server can't take a slot from a healthy one. > @@ -258,6 +265,7 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 - **CORS denies any browser Origin by default** — returns `403` JSON. Pass **`--allow-origin `** (repeatable, T2.4 #4514) to opt specific browser origins through. Each value is either the literal `*` (any origin — boot refuses if no bearer token is configured; `--require-auth` on loopback is recommended for full hardening since `/health` and `/demo` remain pre-auth on loopback by default) or a canonical URL origin (`://[:]`, no trailing slash / path / userinfo). Matched origins receive proper CORS response headers (`Access-Control-Allow-Origin: `, `Vary: Origin`, plus standard methods / headers / max-age and exposed `Retry-After`); unmatched origins still get a 403 with the same envelope as the default wall. `caps.features.allow_origin` is advertised conditionally so SDK / webui clients can pre-flight whether the daemon honors cross-origin hits before issuing them. Example: `qwen serve --allow-origin http://localhost:3000 --allow-origin http://localhost:5173`. Loopback self-origin hits (e.g. the `/demo` page) are unaffected — a separate Origin-strip shim handles them regardless of `--allow-origin`. **Browser webuis without `--allow-origin` configured** still fall back to the same Stage 1 options as before: package as a native shell (Electron/Tauri) so no `Origin` header is sent, or front the daemon with a same-origin reverse proxy. - **Spawned `qwen --acp` child inherits the daemon's environment** with one explicit scrub: `QWEN_SERVER_TOKEN` is removed before the child starts (the daemon's own bearer; the agent doesn't need it). Everything else — `OPENAI_API_KEY` / `ANTHROPIC_API_KEY` / `QWEN_*` / `DASHSCOPE_API_KEY` / your custom `modelProviders[].envKey` / etc. — passes through, because the agent legitimately needs those to authenticate to the LLM. **This is intentional, not a sandbox.** The agent runs as the same UID with shell-tool access, so anything in `~/.bashrc` / `~/.aws/credentials` / `~/.npmrc` is reachable by prompt injection regardless. The env passthrough is not the security boundary; the user-as-trust-root is. Don't run `qwen serve` under an identity that has env-resident credentials you wouldn't trust the agent with. - **Per-subscriber bounded SSE queues** — a slow client that overflows its queue gets a `client_evicted` terminal frame and is closed; one stuck consumer can't pin the daemon. +- **Per-session prompt admission cap** — defaults to 5 accepted-but-unsettled prompts per session. A buggy client cannot enqueue unbounded prompt promises or temporary SSE waits for one session. - **Graceful shutdown** — SIGINT/SIGTERM drain the agent children before closing the listener (10s deadline per child). > ⚠️ **Stage 1 known gap — permissions are daemon-global, not per-session (BUy4H).** `pendingPermissions` lives at daemon scope; any client holding the bearer token can vote on any `requestId` for any session it can see (and SSE `permission_request` events carry the requestId in their payload). This is acceptable under the single-user / small-team trust model where every authenticated client is the same human or collaborators they trust. Stage 1.5 will move to `POST /session/:id/permission/:requestId` + session-scoped pending map + per-client identity (must-have #3 from the downstream review); until then, don't run `qwen serve` behind a bearer shared with untrusted parties. diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 61a9616112..3d78697886 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -30,6 +30,7 @@ import { InvalidSessionMetadataError, InvalidSessionScopeError, NOT_CURRENTLY_GENERATING_CANCEL_MESSAGE, + PromptQueueFullError, RestoreInProgressError, SessionShellClientRequiredError, SessionShellDisabledError, @@ -2622,6 +2623,217 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('rejects prompts past the default per-session pending cap synchronously', async () => { + let releaseFirst: (() => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + promptImpl: async (p) => { + const text = + (p.prompt[0] as { text?: string } | undefined)?.text ?? ''; + if (text === 'hold') { + await new Promise((resolve) => { + releaseFirst = resolve; + }); + } + return { stopReason: 'end_turn' }; + }, + }).channel; + const bridge = makeBridge({ channelFactory: factory }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const accepted = Array.from({ length: 5 }, (_, i) => + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: i === 0 ? 'hold' : `queued-${i}` }], + }), + ); + + expect(() => + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'overflow' }], + }), + ).toThrow(PromptQueueFullError); + + await vi.waitFor(() => expect(releaseFirst).toBeDefined()); + releaseFirst!(); + await Promise.all(accepted); + await bridge.shutdown(); + }); + + it.each([[0], [Infinity]])( + 'does not cap pending prompts when maxPendingPromptsPerSession is %s', + async (maxPendingPromptsPerSession) => { + let releaseFirst: (() => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + promptImpl: async (p) => { + const text = + (p.prompt[0] as { text?: string } | undefined)?.text ?? ''; + if (text === 'hold') { + await new Promise((resolve) => { + releaseFirst = resolve; + }); + } + return { stopReason: 'end_turn' }; + }, + }).channel; + const bridge = makeBridge({ + channelFactory: factory, + maxPendingPromptsPerSession, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const accepted = Array.from({ length: 6 }, (_, i) => + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: i === 0 ? 'hold' : `queued-${i}` }], + }), + ); + + await vi.waitFor(() => expect(releaseFirst).toBeDefined()); + releaseFirst!(); + await expect(Promise.all(accepted)).resolves.toHaveLength(6); + await bridge.shutdown(); + }, + ); + + it('releases a pending prompt slot after a failed prompt settles', async () => { + let releaseFirst: (() => void) | undefined; + let calls = 0; + const factory: ChannelFactory = async () => + makeChannel({ + promptImpl: async () => { + calls += 1; + if (calls === 1) { + await new Promise((resolve) => { + releaseFirst = resolve; + }); + throw new Error('first prompt failed'); + } + return { stopReason: 'end_turn' }; + }, + }).channel; + const bridge = makeBridge({ + channelFactory: factory, + maxPendingPromptsPerSession: 1, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const failed = bridge + .sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'first' }], + }) + .catch((err: unknown) => err); + + expect(() => + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'overflow' }], + }), + ).toThrow(PromptQueueFullError); + + await vi.waitFor(() => expect(releaseFirst).toBeDefined()); + releaseFirst!(); + await expect(failed).resolves.toMatchObject({ code: -32603 }); + await expect( + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'after-failure' }], + }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + + await bridge.shutdown(); + }); + + it('does not count pre-aborted prompts against the pending cap', async () => { + let releaseFirst: (() => void) | undefined; + let calls = 0; + const factory: ChannelFactory = async () => + makeChannel({ + promptImpl: async () => { + calls += 1; + if (calls === 1) { + await new Promise((resolve) => { + releaseFirst = resolve; + }); + } + return { stopReason: 'end_turn' }; + }, + }).channel; + const bridge = makeBridge({ + channelFactory: factory, + maxPendingPromptsPerSession: 1, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const active = bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'active' }], + }); + + const aborted = new AbortController(); + aborted.abort(); + expect(() => + bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'aborted' }], + }, + aborted.signal, + ), + ).toThrow(/Prompt aborted/); + + await vi.waitFor(() => expect(releaseFirst).toBeDefined()); + releaseFirst!(); + await active; + await expect( + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'after-abort' }], + }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + await bridge.shutdown(); + }); + + it('does not count queued branchSession work against the prompt cap', async () => { + let releaseBranch: (() => void) | undefined; + const factory: ChannelFactory = async () => + makeChannel({ + extMethodImpl: async (method) => { + if (method !== 'qwen/control/session/branch') return {}; + await new Promise((resolve) => { + releaseBranch = resolve; + }); + return { newSessionId: 'branch-1', title: 'Branch 1' }; + }, + resumeSessionImpl: () => ({}), + }).channel; + const bridge = makeBridge({ + channelFactory: factory, + maxPendingPromptsPerSession: 1, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const branch = bridge.branchSession(session.sessionId, { + name: 'Branch 1', + }); + const prompt = bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'after-branch' }], + }); + + await vi.waitFor(() => expect(releaseBranch).toBeDefined()); + releaseBranch!(); + await expect(branch).resolves.toMatchObject({ + sessionId: 'branch-1', + title: 'Branch 1', + }); + await expect(prompt).resolves.toEqual({ stopReason: 'end_turn' }); + await bridge.shutdown(); + }); + it('a failed prompt does not poison the queue for subsequent prompts', async () => { let promptCount = 0; const handles: ChannelHandle[] = []; @@ -4284,6 +4496,25 @@ describe('createAcpSessionBridge', () => { expect(() => makeBridge({ maxSessions: 0 })).not.toThrow(); expect(() => makeBridge({ maxSessions: Infinity })).not.toThrow(); }); + + it.each([ + ['negative', -5], + ['float', 1.5], + ['NaN', Number.NaN], + ])('rejects invalid maxPendingPromptsPerSession (%s)', (_label, value) => { + expect(() => makeBridge({ maxPendingPromptsPerSession: value })).toThrow( + /maxPendingPromptsPerSession/, + ); + }); + + it('accepts disabled maxPendingPromptsPerSession sentinels', () => { + expect(() => + makeBridge({ maxPendingPromptsPerSession: 0 }), + ).not.toThrow(); + expect(() => + makeBridge({ maxPendingPromptsPerSession: Infinity }), + ).not.toThrow(); + }); }); describe('concurrent spawn coalescing (single scope)', () => { diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 91cb27a0d9..20bce325dd 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -51,6 +51,7 @@ import { RestoreInProgressError, InvalidSessionScopeError, SessionLimitExceededError, + PromptQueueFullError, WorkspaceMismatchError, InvalidClientIdError, SessionShellClientRequiredError, @@ -226,6 +227,8 @@ interface SessionEntry { * caller still observes the rejection on its own returned promise. */ promptQueue: Promise; + /** Accepted prompts that have not settled yet (queued + active). */ + pendingPromptCount: number; /** * Per-session model-change FIFO. Prevents two concurrent * `applyModelServiceId` calls (e.g. simultaneous attach-with-different- @@ -648,6 +651,8 @@ const SESSION_BTW_TIMEOUT_MS = 60_000; const SHELL_COMMAND_TIMEOUT_MS = 120_000; const MAX_SHELL_OUTPUT_FOR_HISTORY = 10_000; const DEFAULT_MAX_SESSIONS = 20; +// Keep in sync with CLI serve/server.ts and SDK DaemonClient.ts. +const DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION = 5; /** * Soft upper bound on `BridgeOptions.eventRingSize` to catch operator * typos before they OOM the daemon. At ~500 B per `BridgeEvent` an @@ -750,8 +755,8 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); } // Bd1yh + Bd1z5: per-permission deadline + per-session pending cap. - // 0 / Infinity / non-finite (NaN, -1) all disable — same sentinel - // convention as maxSessions / maxConnections. + // Permission caps keep the legacy sentinel behavior; prompt caps are + // stricter because they are an admission-control surface. const permissionTimeoutRaw = opts.permissionResponseTimeoutMs ?? DEFAULT_PERMISSION_TIMEOUT_MS; const permissionTimeoutMs = @@ -764,6 +769,25 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { maxPendingRaw > 0 && Number.isFinite(maxPendingRaw) ? maxPendingRaw : Infinity; + const maxPendingPromptsRaw = + opts.maxPendingPromptsPerSession ?? DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION; + let maxPendingPromptsPerSession: number; + if ( + maxPendingPromptsRaw === 0 || + maxPendingPromptsRaw === Number.POSITIVE_INFINITY + ) { + maxPendingPromptsPerSession = Infinity; + } else if ( + !Number.isInteger(maxPendingPromptsRaw) || + maxPendingPromptsRaw < 0 + ) { + throw new TypeError( + `Invalid maxPendingPromptsPerSession: ${maxPendingPromptsRaw}. ` + + `Must be a non-negative integer (0 / Infinity = unlimited).`, + ); + } else { + maxPendingPromptsPerSession = maxPendingPromptsRaw; + } // The bound path is the canonical form `spawnOrAttach` compares // incoming `workspaceCwd` against. The caller MUST pass an already- // canonical value (via `canonicalizeWorkspace`). `runQwenServe` @@ -1977,6 +2001,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { connection: ci.connection, events, promptQueue: Promise.resolve(), + pendingPromptCount: 0, modelChangeQueue: Promise.resolve(), approvalModeQueue: Promise.resolve(), modelPublishGeneration: 0, @@ -2654,7 +2679,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } }, - async sendPrompt(sessionId, req, signal, context) { + // Keep this method non-async: admission failures must throw before + // HTTP routes return 202. + sendPrompt(sessionId, req, signal, context) { opts.onDiagnosticLine?.( `qwen serve: bridge sendPrompt for session=${sessionId}`, 'info', @@ -2662,11 +2689,13 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { const capturedContext = telemetry.captureContext(); const queuedAt = Date.now(); const entry = byId.get(sessionId); - if (!entry) throw new SessionNotFoundError(sessionId); - const originatorClientId = resolveTrustedClientId( - entry, - context?.clientId, - ); + if (!entry) return Promise.reject(new SessionNotFoundError(sessionId)); + let originatorClientId: string | undefined; + try { + originatorClientId = resolveTrustedClientId(entry, context?.clientId); + } catch (err) { + return Promise.reject(err); + } // Pre-aborted: skip the queue entirely. Without this the prompt // chains onto promptQueue, waits its turn, and the FIFO worker // checks `signal.aborted` only AFTER reaching the head — wasted @@ -2675,6 +2704,20 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (signal?.aborted) { throw new DOMException('Prompt aborted', 'AbortError'); } + if (entry.pendingPromptCount >= maxPendingPromptsPerSession) { + throw new PromptQueueFullError( + maxPendingPromptsPerSession, + entry.pendingPromptCount, + sessionId, + ); + } + entry.pendingPromptCount += 1; + let promptSlotReleased = false; + const releasePromptSlot = () => { + if (promptSlotReleased) return; + promptSlotReleased = true; + entry.pendingPromptCount = Math.max(0, entry.pendingPromptCount - 1); + }; // Force the body's sessionId to match the routing id — a client that // sent a stale id in the body would otherwise be dispatched to the // wrong agent process. @@ -2898,6 +2941,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { () => undefined, () => undefined, ); + result.finally(releasePromptSlot).catch(() => {}); return result; }, diff --git a/packages/acp-bridge/src/bridgeErrors.ts b/packages/acp-bridge/src/bridgeErrors.ts index 792565db77..0c8edf9992 100644 --- a/packages/acp-bridge/src/bridgeErrors.ts +++ b/packages/acp-bridge/src/bridgeErrors.ts @@ -123,6 +123,30 @@ export class SessionLimitExceededError extends Error { } } +/** + * Thrown by `sendPrompt` when a session already has too many accepted + * prompts waiting or running. The REST route maps this to 503 with + * `Retry-After`; SDK clients can retry after observing a turn completion. + * The TypeScript SDK maps the same `prompt_queue_full` wire condition to + * `DaemonPendingPromptLimitError`. + */ +export class PromptQueueFullError extends Error { + readonly limit: number; + readonly pendingCount: number; + readonly sessionId: string; + + constructor(limit: number, pendingCount: number, sessionId: string) { + super( + `Prompt queue full for session "${sessionId}" ` + + `(${pendingCount}/${limit} pending)`, + ); + this.name = 'PromptQueueFullError'; + this.limit = limit; + this.pendingCount = pendingCount; + this.sessionId = sessionId; + } +} + /** * Thrown by `spawnOrAttach` when the requested `workspaceCwd` doesn't * canonicalize to the daemon's bound workspace. Every diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index 4268ee9a3c..87dc07c86c 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -185,6 +185,12 @@ export interface BridgeOptions { * cap. */ maxPendingPermissionsPerSession?: number; + /** + * Per-session cap on accepted prompts that have not settled yet, + * including the currently running prompt and queued prompts behind it. + * Defaults to 5. `0` / `Infinity` disable the cap. + */ + maxPendingPromptsPerSession?: number; /** * Absolute, **already-canonical** path this daemon is bound to (per * 1 daemon = 1 workspace). `spawnOrAttach` calls whose diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 9738ece5cc..43e1baddb1 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -218,8 +218,13 @@ export interface AcpSessionBridge { /** * Forward a prompt to the agent. Concurrent prompts against the same - * session FIFO-serialize through a per-session queue. Throws - * `SessionNotFoundError` when the id is unknown. + * session FIFO-serialize through a per-session queue. + * + * Admission contract: implementations must not be `async`. Admission + * failures such as `PromptQueueFullError` and pre-aborted signals throw + * synchronously so HTTP routes can reject before returning 202. Deferred + * failures such as `SessionNotFoundError` may be returned as rejected + * promises. */ sendPrompt( sessionId: string, diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 5c783e76b9..0df976c4e6 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -35,6 +35,7 @@ interface ServeArgs { hostname: string; token?: string; 'max-sessions': number; + 'max-pending-prompts-per-session': number; 'max-connections': number; 'event-ring-size': number; workspace?: string; @@ -90,6 +91,13 @@ export const serveCommand: CommandModule = { 'Cap on concurrent live sessions. New spawn requests beyond this return 503; ' + 'attach to existing sessions still works. Set to 0 to disable.', }) + .option('max-pending-prompts-per-session', { + type: 'number', + default: 5, + description: + 'Per-session cap on accepted prompts waiting or running. ' + + 'New prompts beyond this return 503. Set to 0 to disable.', + }) .option('workspace', { type: 'string', description: @@ -284,6 +292,18 @@ export const serveCommand: CommandModule = { } const resolvedMcpMode: 'enforce' | 'warn' | 'off' = mcpBudgetMode ?? (mcpClientBudget !== undefined ? 'warn' : 'off'); + const maxPendingPromptsPerSession = argv['max-pending-prompts-per-session']; + if ( + maxPendingPromptsPerSession !== Number.POSITIVE_INFINITY && + (!Number.isFinite(maxPendingPromptsPerSession) || + !Number.isInteger(maxPendingPromptsPerSession) || + maxPendingPromptsPerSession < 0) + ) { + writeStderrLine( + 'qwen serve: --max-pending-prompts-per-session must be a non-negative integer (0 / Infinity = unlimited).', + ); + process.exit(1); + } if (mcpClientBudget !== undefined) { // Mirror the `--require-auth` breadcrumb: surface the active // policy in stderr (journald / docker logs) so operators don't @@ -390,6 +410,7 @@ export const serveCommand: CommandModule = { token: argv.token, mode: 'http-bridge', maxSessions: argv['max-sessions'], + maxPendingPromptsPerSession, maxConnections: argv['max-connections'], eventRingSize: argv['event-ring-size'], workspace: argv.workspace, diff --git a/packages/cli/src/serve/acpHttp/dispatch.ts b/packages/cli/src/serve/acpHttp/dispatch.ts index f3dffaa25c..f41046345d 100644 --- a/packages/cli/src/serve/acpHttp/dispatch.ts +++ b/packages/cli/src/serve/acpHttp/dispatch.ts @@ -287,6 +287,23 @@ function toRpcError(err: unknown): { return { code: RPC.INVALID_PARAMS, message: errMsg(err) }; case 'SessionLimitExceededError': return { code: RPC.INTERNAL_ERROR, message: errMsg(err) }; + case 'PromptQueueFullError': { + const promptErr = err as { + sessionId?: unknown; + limit?: unknown; + pendingCount?: unknown; + }; + return { + code: RPC.INTERNAL_ERROR, + message: errMsg(err), + data: { + errorKind: 'prompt_queue_full', + sessionId: promptErr.sessionId, + limit: promptErr.limit, + pendingCount: promptErr.pendingCount, + }, + }; + } default: return { code: RPC.INTERNAL_ERROR, diff --git a/packages/cli/src/serve/acpHttp/transport.test.ts b/packages/cli/src/serve/acpHttp/transport.test.ts index 2df1de2c0a..8042c43430 100644 --- a/packages/cli/src/serve/acpHttp/transport.test.ts +++ b/packages/cli/src/serve/acpHttp/transport.test.ts @@ -13,6 +13,7 @@ import type { HttpAcpBridge } from '@qwen-code/acp-bridge/bridgeTypes'; import type { BridgeEvent } from '@qwen-code/acp-bridge/eventBus'; import { InvalidClientIdError, + PromptQueueFullError, SessionShellClientRequiredError, SessionShellDisabledError, } from '@qwen-code/acp-bridge/bridgeErrors'; @@ -149,11 +150,12 @@ class FakeBridge { return q.iterable; } - async sendPrompt(sessionId: string, _req: unknown, signal?: AbortSignal) { + sendPrompt(sessionId: string, _req: unknown, signal?: AbortSignal) { const q = this.queues.get(sessionId); - if (this.promptBehavior && q) - return this.promptBehavior(sessionId, q, signal); - return { stopReason: 'end_turn' }; + if (this.promptBehavior && q) { + return Promise.resolve(this.promptBehavior(sessionId, q, signal)); + } + return Promise.resolve({ stopReason: 'end_turn' }); } respondToSessionPermission() { @@ -1207,6 +1209,34 @@ describe('ACP Streamable HTTP transport (over the wire)', () => { expect(frame.error.code).toBe(-32602); }); + it('session/prompt queue cap error includes stable JSON-RPC data', async () => { + bridge.promptBehavior = () => { + throw new PromptQueueFullError(5, 5, 'sess-1'); + }; + const connId = await initialize(); + await newSession(connId); + const sessStream = await openStream(connId, 'sess-1'); + const got = takeFrames(sessStream, 1); + await new Promise((r) => setTimeout(r, 50)); + await post(connId, { + jsonrpc: '2.0', + id: 46, + method: 'session/prompt', + params: { sessionId: 'sess-1', prompt: [{ type: 'text', text: 'hi' }] }, + }); + + const [frame] = (await got) as Array<{ + error: { code: number; data: Record }; + }>; + expect(frame.error.code).toBe(-32603); + expect(frame.error.data).toMatchObject({ + errorKind: 'prompt_queue_full', + sessionId: 'sess-1', + limit: 5, + pendingCount: 5, + }); + }); + it('session/close runs local cleanup even if the bridge close throws', async () => { bridge.closeShouldThrow = true; const connId = await initialize(); diff --git a/packages/cli/src/serve/acpSessionBridge.ts b/packages/cli/src/serve/acpSessionBridge.ts index b60e7df47b..b899f45d34 100644 --- a/packages/cli/src/serve/acpSessionBridge.ts +++ b/packages/cli/src/serve/acpSessionBridge.ts @@ -82,6 +82,7 @@ export { RestoreInProgressError, InvalidSessionScopeError, SessionLimitExceededError, + PromptQueueFullError, WorkspaceMismatchError, InvalidClientIdError, InvalidPermissionOptionError, diff --git a/packages/cli/src/serve/runQwenServe.ts b/packages/cli/src/serve/runQwenServe.ts index 6893f68a9c..15fd627dd3 100644 --- a/packages/cli/src/serve/runQwenServe.ts +++ b/packages/cli/src/serve/runQwenServe.ts @@ -86,6 +86,13 @@ function isPositiveIntegerMs(value: number): boolean { return Number.isFinite(value) && Number.isInteger(value) && value > 0; } +function isNonNegativeIntegerOrInfinity(value: number): boolean { + return ( + value === Number.POSITIVE_INFINITY || + (Number.isFinite(value) && Number.isInteger(value) && value >= 0) + ); +} + const MAX_TIMEOUT_MS = 2_147_483_647; function assertTimerDelayInRange(name: string, value: number): void { @@ -653,6 +660,13 @@ export async function runQwenServe( } assertTimerDelayInRange('promptDeadlineMs', opts.promptDeadlineMs); } + if (opts.maxPendingPromptsPerSession !== undefined) { + if (!isNonNegativeIntegerOrInfinity(opts.maxPendingPromptsPerSession)) { + throw new TypeError( + `Invalid maxPendingPromptsPerSession: ${opts.maxPendingPromptsPerSession}. Must be a non-negative integer (0 / Infinity = unlimited).`, + ); + } + } if (opts.writerIdleTimeoutMs !== undefined) { if (!isPositiveIntegerMs(opts.writerIdleTimeoutMs)) { throw new TypeError( @@ -848,6 +862,9 @@ export async function runQwenServe( deps.bridge ?? createAcpSessionBridge({ maxSessions: opts.maxSessions, + ...(opts.maxPendingPromptsPerSession !== undefined + ? { maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession } + : {}), ...(opts.eventRingSize !== undefined ? { eventRingSize: opts.eventRingSize } : {}), diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 844773cb81..a3c25ca00e 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -52,6 +52,7 @@ import { McpServerRestartFailedError, PermissionForbiddenError, PermissionPolicyNotImplementedError, + PromptQueueFullError, RestoreInProgressError, SessionShellClientRequiredError, SessionShellDisabledError, @@ -87,6 +88,7 @@ import type { ServeWorkspaceToolsStatus, } from './status.js'; import { CAPABILITIES_SCHEMA_VERSION, type ServeOptions } from './types.js'; +import type { DaemonLogger } from './daemonLogger.js'; import { FsError, type WorkspaceFileSystemFactory } from './fs/index.js'; const baseOpts: ServeOptions = { @@ -95,6 +97,18 @@ const baseOpts: ServeOptions = { mode: 'http-bridge', }; +function fakeDaemonLog(): DaemonLogger { + return { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + raw: vi.fn(), + getLogPath: () => '', + getDaemonId: () => 'test-daemon', + flush: vi.fn(async () => {}), + }; +} + // Workspace fixtures must round-trip through `path.resolve` so the // expected values match the canonicalized form the route produces on // every platform. On Windows `path.resolve('/work/bound')` returns @@ -270,7 +284,7 @@ interface FakeBridgeOpts { req: PromptRequest, signal?: AbortSignal, context?: BridgeClientRequestContext, - ) => Promise; + ) => Promise | PromptResponse; cancelImpl?: ( sessionId: string, req?: CancelNotification, @@ -948,14 +962,14 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { resumeCalls.push(req); return result; }, - async sendPrompt(sessionId, req, signal, context) { + sendPrompt(sessionId, req, signal, context) { promptCalls.push({ sessionId, req, signal, ...(context ? { context } : {}), }); - return promptImpl(sessionId, req, signal, context); + return Promise.resolve(promptImpl(sessionId, req, signal, context)); }, async cancelSession(sessionId, req, context) { cancelCalls.push({ sessionId, req, ...(context ? { context } : {}) }); @@ -1581,6 +1595,39 @@ describe('createServeApp', () => { getAdvertisedServeFeatures(undefined, { mcpPoolActive: true }), ); expect(res.body.modelServices).toEqual([]); + expect(res.body.limits).toMatchObject({ + maxPendingPromptsPerSession: 5, + }); + }); + + it('reports disabled prompt queue cap as null in capabilities', async () => { + const app = createServeApp({ + ...baseOpts, + maxPendingPromptsPerSession: 0, + }); + const res = await request(app) + .get('/capabilities') + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(200); + expect(res.body.limits).toMatchObject({ + maxPendingPromptsPerSession: null, + }); + }); + + it('reports explicit prompt queue cap in capabilities', async () => { + const app = createServeApp({ + ...baseOpts, + maxPendingPromptsPerSession: 12, + }); + const res = await request(app) + .get('/capabilities') + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(200); + expect(res.body.limits).toMatchObject({ + maxPendingPromptsPerSession: 12, + }); }); it('omits mcp_workspace_pool / mcp_pool_restart when mcpPoolActive=false (F2 #4175 commit 5)', async () => { @@ -3129,6 +3176,38 @@ describe('createServeApp', () => { expect(res.body.promptId).toBeDefined(); }); + it('503 without promptId when bridge rejects prompt admission synchronously', async () => { + const bridge = fakeBridge({ + promptImpl: () => { + throw new PromptQueueFullError(5, 5, 'session-A'); + }, + }); + const daemonLog = fakeDaemonLog(); + const app = createServeApp(baseOpts, undefined, { bridge, daemonLog }); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ prompt: [{ type: 'text', text: 'hi' }] }); + + expect(res.status).toBe(503); + expect(res.headers['retry-after']).toBe('5'); + expect(res.body).toMatchObject({ + code: 'prompt_queue_full', + sessionId: 'session-A', + limit: 5, + pendingCount: 5, + }); + expect(res.body.promptId).toBeUndefined(); + expect(daemonLog.warn).toHaveBeenCalledWith( + 'prompt admission rejected: queue full', + expect.objectContaining({ + sessionId: 'session-A', + limit: 5, + pendingCount: 5, + }), + ); + }); + it('passes an AbortSignal into bridge.sendPrompt', async () => { let signalDefined = false; let abortedAtCall = false; @@ -6385,6 +6464,24 @@ describe('runQwenServe', () => { }, ); + it.each([ + ['negative', -5], + ['float', 1.5], + ['NaN', Number.NaN], + ])( + 'rejects invalid maxPendingPromptsPerSession (%s) at boot', + async (_label, value) => { + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + maxPendingPromptsPerSession: value, + }), + ).rejects.toThrow(/maxPendingPromptsPerSession/); + }, + ); + it.each([ ['zero', 0], ['negative', -5], diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index e45b590951..69108e8ede 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -69,6 +69,7 @@ import { McpServerRestartFailedError, PermissionForbiddenError, PermissionPolicyNotImplementedError, + PromptQueueFullError, RestoreInProgressError, SessionBusyError, InvalidRewindTargetError, @@ -859,6 +860,17 @@ export function resolvePromptDeadlineMs( return Math.min(serverMs, requestMs); } +// Keep in sync with acp-bridge bridge.ts and SDK DaemonClient.ts. +const DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION = 5; + +function advertisedMaxPendingPromptsPerSession( + value: number | undefined, +): number | null { + if (value === undefined) return DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION; + if (value === 0 || value === Number.POSITIVE_INFINITY) return null; + return value; +} + /** * Build the Express app for `qwen serve`. Pure function — no side effects on * the network or process; `runQwenServe` does the listen/signal handling. @@ -953,6 +965,7 @@ export function createServeApp( deps.bridge ?? createAcpSessionBridge({ maxSessions: opts.maxSessions, + maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession, eventRingSize: opts.eventRingSize, boundWorkspace, sessionShellCommandEnabled, @@ -1384,6 +1397,11 @@ export function createServeApp( workspaceCwd: boundWorkspace, // Active mediation policy under the `policy` namespace. policy: { permission: bridge.permissionPolicy }, + limits: { + maxPendingPromptsPerSession: advertisedMaxPendingPromptsPerSession( + opts.maxPendingPromptsPerSession, + ), + }, supportedLanguages: LANGUAGE_CODES, }; res.status(200).json(envelope); @@ -2235,8 +2253,9 @@ export function createServeApp( deadlineTimer.unref(); } - bridge - .sendPrompt( + let promptPromise: ReturnType; + try { + promptPromise = bridge.sendPrompt( sessionId, { ...forwardedBody, @@ -2248,7 +2267,26 @@ export function createServeApp( ...(clientId !== undefined ? { clientId } : {}), promptId, }, - ) + ); + } catch (err) { + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); + if (daemonLog && err instanceof PromptQueueFullError) { + daemonLog.warn('prompt admission rejected: queue full', { + sessionId, + promptId, + ...(clientId !== undefined ? { clientId } : {}), + limit: err.limit, + pendingCount: err.pendingCount, + }); + } + sendBridgeError(res, err, { + route: 'POST /session/:id/prompt', + sessionId, + }); + return; + } + + promptPromise .then( () => { if (daemonLog) { @@ -4469,6 +4507,17 @@ function sendBridgeErrorImpl( }); return; } + if (err instanceof PromptQueueFullError) { + res.set('Retry-After', '5'); + res.status(503).json({ + error: err.message, + code: 'prompt_queue_full', + sessionId: err.sessionId, + limit: err.limit, + pendingCount: err.pendingCount, + }); + return; + } if (err instanceof RestoreInProgressError) { // Match `SessionLimitExceededError`'s 5s hint (above) — the // underlying restore can take up to `initTimeoutMs` (default diff --git a/packages/cli/src/serve/types.ts b/packages/cli/src/serve/types.ts index e662f44c69..77ec74da36 100644 --- a/packages/cli/src/serve/types.ts +++ b/packages/cli/src/serve/types.ts @@ -53,6 +53,11 @@ export interface ServeOptions { * `0` or `Infinity` to disable. */ maxSessions?: number; + /** + * Per-session cap on accepted prompts that have not settled yet. + * Defaults to 5. `0` or `Infinity` disables the cap. + */ + maxPendingPromptsPerSession?: number; /** * Listener-level TCP connection cap (`server.maxConnections`). * Defaults to 256 — bounds the raw socket count regardless of @@ -249,6 +254,13 @@ export interface CapabilitiesEnvelope { */ permission?: PermissionPolicy; }; + /** + * Active daemon resource limits. Additive to v=1; older daemons may omit it. + * `null` means the operator explicitly disabled that cap. + */ + limits?: { + maxPendingPromptsPerSession?: number | null; + }; /** * Language codes accepted by `POST /session/:id/language`. * Additive — older daemons omit this field; clients should diff --git a/packages/sdk-typescript/scripts/build.js b/packages/sdk-typescript/scripts/build.js index b715dc5601..c5c9329a2a 100755 --- a/packages/sdk-typescript/scripts/build.js +++ b/packages/sdk-typescript/scripts/build.js @@ -20,7 +20,7 @@ import esbuild from 'esbuild'; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); const rootDir = join(__dirname, '..'); -const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 114 * 1024; +const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 116 * 1024; rmSync(join(rootDir, 'dist'), { recursive: true, force: true }); mkdirSync(join(rootDir, 'dist'), { recursive: true }); diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index d77412f75b..ce87988c1e 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -121,14 +121,37 @@ export interface DaemonClientOptions { * Defaults to 30s. Set to `0` or `Infinity` to disable. */ fetchTimeoutMs?: number; + /** + * Per-session cap on local `prompt()` calls that have been admitted but + * not completed. For 202 daemons the slot is held until the temporary + * SSE wait finishes. Defaults to 5. Set to `0` or `Infinity` to + * disable; `null` is accepted for direct + * `/capabilities.limits` passthrough. + */ + maxPendingPromptsPerSession?: number | null; } const DEFAULT_FETCH_TIMEOUT_MS = 30_000; +// Keep in sync with acp-bridge bridge.ts and CLI serve/server.ts. +const DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION = 5; // Server deadline + headroom so the client never races the daemon's own budget. const MCP_RESTART_DEFAULT_TIMEOUT_MS = MCP_RESTART_SERVER_DEADLINE_MS + MCP_RESTART_CLIENT_HEADROOM_MS; const CLIENT_ID_HEADER = 'X-Qwen-Client-Id'; +export function normalizePendingPromptLimit( + value: number | null | undefined, +): number { + if (value === undefined) return DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION; + if (value === null || value === 0 || value === Infinity) { + return Infinity; + } + if (!Number.isInteger(value) || value < 0) { + throw new TypeError('bad maxPendingPromptsPerSession'); + } + return value; +} + /** * Strip any trailing slashes from a base URL via plain string ops. The * obvious `replace(/\/+$/, '')` is technically linear here (the regex is @@ -198,6 +221,24 @@ export class DaemonHttpError extends Error { } } +/** + * SDK-side representation of the daemon's `prompt_queue_full` condition. + * Mirrors the bridge-side `PromptQueueFullError` wire data. + */ +export class DaemonPendingPromptLimitError extends Error { + declare readonly sessionId: string; + declare readonly limit: number; + declare readonly pendingCount: number; + + constructor(sessionId: string, limit: number, pendingCount: number) { + super(`Pending prompts full: "${sessionId}" (${pendingCount}/${limit})`); + this.name = 'DaemonPendingPromptLimitError'; + this.sessionId = sessionId; + this.limit = limit; + this.pendingCount = pendingCount; + } +} + export interface DaemonTurnError extends DaemonHttpError { _daemonTurnError: true; } @@ -302,6 +343,8 @@ export class DaemonClient { private readonly token: string | undefined; private readonly _fetch: typeof globalThis.fetch; private readonly fetchTimeoutMs: number; + private readonly promptLimit: number; + private readonly promptCounts: Record = Object.create(null); // Lazy singleton so clients that never touch auth pay no allocation cost. // Exposed via the readonly `auth` accessor below. private _authFlow?: DaemonAuthFlow; @@ -337,6 +380,34 @@ export class DaemonClient { // it instead of defending the math at every call site. const raw = opts.fetchTimeoutMs ?? DEFAULT_FETCH_TIMEOUT_MS; this.fetchTimeoutMs = Number.isFinite(raw) && raw > 0 ? raw : 0; + this.promptLimit = normalizePendingPromptLimit( + opts.maxPendingPromptsPerSession, + ); + } + + get maxPendingPromptsPerSession(): number { + return this.promptLimit; + } + + /** @internal */ + reservePromptSlot(sessionId: string, limit = this.promptLimit): () => void { + if (limit === Infinity) return () => {}; + const promptCounts = this.promptCounts; + const pendingCount = promptCounts[sessionId] ?? 0; + if (pendingCount >= limit) { + throw new DaemonPendingPromptLimitError(sessionId, limit, pendingCount); + } + promptCounts[sessionId] = pendingCount + 1; + let released: boolean | undefined; + return () => { + if (released) return; + released = true; + if ((promptCounts[sessionId] ?? 0) <= 1) { + delete promptCounts[sessionId]; + } else { + --promptCounts[sessionId]!; + } + }; } /** @@ -399,7 +470,7 @@ export class DaemonClient { // body consume callback, if any) settles. const ctrl = new AbortController(); const timer = setTimeout(() => { - ctrl.abort(new DOMException('The operation timed out', 'TimeoutError')); + ctrl.abort(new DOMException('timeout', 'TimeoutError')); }, effectiveTimeoutMs); if (typeof timer === 'object' && timer && 'unref' in timer) { (timer as { unref: () => void }).unref(); @@ -432,7 +503,17 @@ export class DaemonClient { private async failOnError( res: Response, label: string, - ): Promise { + ): Promise; + private async failOnError( + res: Response, + label: string, + sessionId: string, + ): Promise; + private async failOnError( + res: Response, + label: string, + sessionId?: string, + ): Promise { // Read the body exactly once. `res.json()` consumes the stream even on // parse-failure, leaving a subsequent `res.text()` empty — so go via // text() and attempt JSON parsing ourselves; raw text is a useful @@ -454,6 +535,21 @@ export class DaemonClient { body && typeof body === 'object' && 'error' in body ? String((body as { error: unknown }).error) : `HTTP ${res.status}`; + if (sessionId && res.status === 503 && body && typeof body === 'object') { + const data = body as { + code?: unknown; + limit?: unknown; + pendingCount?: unknown; + sessionId?: unknown; + }; + if (data.code === 'prompt_queue_full') { + return new DaemonPendingPromptLimitError( + typeof data.sessionId === 'string' ? data.sessionId : sessionId, + typeof data.limit === 'number' ? data.limit : 0, + typeof data.pendingCount === 'number' ? data.pendingCount : 0, + ); + } + } return new DaemonHttpError(res.status, body, `${label}: ${detail}`); } @@ -1759,29 +1855,50 @@ export class DaemonClient { signal?: AbortSignal, clientId?: string, ): Promise { - const res = await this._fetch( - `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/prompt`, - { - method: 'POST', - headers: this.headers({ 'Content-Type': 'application/json' }, clientId), - body: JSON.stringify(req), - signal, - }, - ); - - if (res.status === 202) { - const accept = (await res.json()) as NonBlockingPromptAccepted; - return this._awaitTurnComplete( - sessionId, - accept.promptId, - accept.lastEventId, - signal, - clientId, + signal?.throwIfAborted(); + const releasePromptSlot = this.reservePromptSlot(sessionId); + let releaseOnExit = true; + try { + const res = await this._fetch( + `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/prompt`, + { + method: 'POST', + headers: this.headers( + { 'Content-Type': 'application/json' }, + clientId, + ), + body: JSON.stringify(req), + signal, + }, ); - } - if (!res.ok) throw await this.failOnError(res, 'POST /session/:id/prompt'); - return (await res.json()) as PromptResult; + if (res.status === 202) { + const accept = (await res.json()) as NonBlockingPromptAccepted; + releaseOnExit = false; + try { + return await this._awaitTurnComplete( + sessionId, + accept.promptId, + accept.lastEventId, + signal, + clientId, + ); + } finally { + releasePromptSlot(); + } + } + + if (!res.ok) { + throw await this.failOnError( + res, + 'POST /session/:id/prompt', + sessionId, + ); + } + return (await res.json()) as PromptResult; + } finally { + if (releaseOnExit) releasePromptSlot(); + } } /** @@ -1797,6 +1914,10 @@ export class DaemonClient { * the temporary 202 fallback. * * Falls back to `prompt()` for legacy 200 daemons. + * + * Note: this method does not enforce the local pending-prompt cap. + * Callers that need early-fail behavior should use {@link prompt} or + * reserve a slot before calling this method. */ async promptNonBlocking( sessionId: string, @@ -1818,7 +1939,9 @@ export class DaemonClient { return (await res.json()) as NonBlockingPromptAccepted; } - if (!res.ok) throw await this.failOnError(res, 'POST /session/:id/prompt'); + if (!res.ok) { + throw await this.failOnError(res, 'POST /session/:id/prompt', sessionId); + } return (await res.json()) as PromptResult; } @@ -1843,7 +1966,7 @@ export class DaemonClient { const result = matchTurnEvent(event, promptId); if (result !== undefined) return result; } - throw new Error('SSE stream ended without turn completion'); + throw new Error('SSE stream ended'); } catch (err) { if ( signal?.aborted && @@ -1932,7 +2055,7 @@ export class DaemonClient { connectTimer = setTimeout( () => connectCtrl.abort( - new DOMException('Initial connect timed out', 'TimeoutError'), + new DOMException('connect timeout', 'TimeoutError'), ), this.fetchTimeoutMs, ); @@ -1987,11 +2110,11 @@ export class DaemonClient { throw new DaemonHttpError( res.status, ct, - `GET /session/:id/events: expected content-type text/event-stream, got "${ct}"`, + `Bad SSE content-type: "${ct}"`, ); } if (!res.body) { - throw new Error('SSE response has no body'); + throw new Error('No SSE body'); } // Forward the abort signal so post-200 aborts stop the iteration. // Without this, callers who `controller.abort()` after the response @@ -2350,8 +2473,7 @@ export function abortTimeout(ms: number): AbortSignal { // `if (err.name === 'TimeoutError')` would see the polyfill // differently from the native runtime. const handle = setTimeout( - () => - ctrl.abort(new DOMException('The operation timed out', 'TimeoutError')), + () => ctrl.abort(new DOMException('timeout', 'TimeoutError')), ms, ); if (typeof handle === 'object' && handle && 'unref' in handle) { diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index b3514b590b..a3674a2cd6 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -8,7 +8,9 @@ import type { DaemonClient } from './DaemonClient.js'; import { isNonBlockingAccepted, matchTurnEvent, + normalizePendingPromptLimit, type CreateSessionRequest, + type NonBlockingPromptAccepted, type PromptRequest, type RestoreSessionRequest, type SubscribeOptions, @@ -55,6 +57,13 @@ export interface DaemonSessionClientOptions { lastEventId?: number; /** Compacted replay snapshot from daemon load response. */ replaySnapshot?: DaemonReplaySnapshot; + /** + * Local per-session prompt cap. The counter is shared with the parent + * `DaemonClient`; other session clients using the same parent instance + * contend on the same count. Set to `null`, `0`, or `Infinity` to disable + * the local guard. Server-side admission still applies. + */ + maxPendingPromptsPerSession?: number | null; } export interface DaemonSessionSubscribeOptions extends SubscribeOptions { @@ -84,6 +93,7 @@ export class DaemonSessionClient { readonly replaySnapshot: DaemonReplaySnapshot; private lastSeenEventId: number | undefined; private subscriptionActive = false; + private readonly promptLimit: number; private readonly _pendingPrompts = new Map< string, { @@ -101,6 +111,10 @@ export class DaemonSessionClient { liveJournal: [], }; this.lastSeenEventId = validateLastEventId(opts.lastEventId); + this.promptLimit = + opts.maxPendingPromptsPerSession === undefined + ? opts.client.maxPendingPromptsPerSession + : normalizePendingPromptLimit(opts.maxPendingPromptsPerSession); } /** @@ -226,6 +240,7 @@ export class DaemonSessionClient { req: PromptRequest, signal?: AbortSignal, ): Promise { + signal?.throwIfAborted(); if (!this.subscriptionActive) { return await this.client.prompt( this.sessionId, @@ -235,35 +250,58 @@ export class DaemonSessionClient { ); } - const accepted = await this.client.promptNonBlocking( + const releaseAdmission = this.client.reservePromptSlot( this.sessionId, - req, - signal, - this.clientId, + this.promptLimit, ); - if (!isNonBlockingAccepted(accepted)) { - return accepted; + let accepted: NonBlockingPromptAccepted | PromptResult; + try { + accepted = await this.client.promptNonBlocking( + this.sessionId, + req, + signal, + this.clientId, + ); + if (!isNonBlockingAccepted(accepted)) { + releaseAdmission(); + return accepted; + } + if (!this.subscriptionActive) { + throw Error('SSE stream ended'); + } + } catch (err) { + releaseAdmission(); + throw err; } return new Promise((resolve, reject) => { const onAbort = () => { - if (this._pendingPrompts.delete(accepted.promptId)) { + const pending = this._pendingPrompts.get(accepted.promptId); + if (pending && this._pendingPrompts.delete(accepted.promptId)) { this.client.cancel(this.sessionId, this.clientId).catch(() => {}); - reject(signal!.reason ?? new DOMException('Aborted', 'AbortError')); + pending.reject( + signal!.reason ?? new DOMException('Aborted', 'AbortError'), + ); } }; const cleanup = () => signal?.removeEventListener('abort', onAbort); this._pendingPrompts.set(accepted.promptId, { resolve: (r) => { cleanup(); + releaseAdmission(); resolve(r); }, reject: (e) => { cleanup(); + releaseAdmission(); reject(e); }, }); - signal?.addEventListener('abort', onAbort, { once: true }); + if (signal?.aborted) { + onAbort(); + } else { + signal?.addEventListener('abort', onAbort, { once: true }); + } }); } @@ -456,10 +494,7 @@ export class DaemonSessionClient { const acquire = () => { if (started) return; if (this.subscriptionActive) { - throw new Error( - 'Another event subscription is already active on this session. ' + - 'Reuse the existing AsyncGenerator or create a separate DaemonSessionClient.', - ); + throw new Error('subscription active'); } this.subscriptionActive = true; started = true; @@ -559,7 +594,7 @@ function validateLastEventId( ): number | undefined { if (lastEventId === undefined) return undefined; if (!Number.isInteger(lastEventId) || lastEventId < 0) { - throw new TypeError('lastEventId must be a finite non-negative integer'); + throw new TypeError('invalid lastEventId'); } return lastEventId; } diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index 9eb66d7d36..d06aa25e45 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -7,6 +7,7 @@ export { DaemonClient, DaemonHttpError, + DaemonPendingPromptLimitError, isDaemonTurnError, isNonBlockingAccepted, matchTurnEvent, diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 91e88d6ff0..bdcf46c12d 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -20,6 +20,10 @@ export interface DaemonProtocolVersions { supported: string[]; } +export interface DaemonCapabilitiesLimits { + maxPendingPromptsPerSession?: number | null; +} + /** Capabilities envelope returned from `GET /capabilities`. */ export interface DaemonCapabilities { v: 1; @@ -39,6 +43,11 @@ export interface DaemonCapabilities { * `session_events`). Never gate UI off `mode`. */ features: string[]; + /** + * Numeric daemon limits. `null` means the daemon advertises the limit as + * disabled; absence means an older daemon did not advertise it. + */ + limits?: DaemonCapabilitiesLimits; modelServices: string[]; /** * Absolute canonical workspace path this daemon is bound to diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index e4eea74c37..c3f01dd460 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -11,6 +11,7 @@ export { DaemonCapabilityMissingError, DaemonClient, DaemonHttpError, + DaemonPendingPromptLimitError, DaemonSessionClient, asKnownDaemonEvent, createDaemonSessionViewState, diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index 4da24589eb..7a10a189df 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -8,8 +8,10 @@ import { describe, it, expect, vi, afterEach } from 'vitest'; import { DaemonClient, DaemonHttpError, + DaemonPendingPromptLimitError, abortTimeout, composeAbortSignals, + normalizePendingPromptLimit, } from '../../src/daemon/DaemonClient.js'; import { DaemonCapabilityMissingError, @@ -49,6 +51,19 @@ function sseResponse(frames: string): Response { }); } +function pendingSseResponse(): Response { + const encoder = new TextEncoder(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(': keepalive\n\n')); + }, + }); + return new Response(body, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }); +} + interface CapturedRequest { url: string; method: string; @@ -91,6 +106,26 @@ function recordingFetch( } describe('DaemonClient', () => { + describe('normalizePendingPromptLimit', () => { + it('defaults undefined to 5', () => { + expect(normalizePendingPromptLimit(undefined)).toBe(5); + }); + + it.each([[null], [0], [Infinity]])('disables cap for %s', (value) => { + expect(normalizePendingPromptLimit(value)).toBe(Infinity); + }); + + it('passes through positive integers', () => { + expect(normalizePendingPromptLimit(7)).toBe(7); + }); + + it.each([[-1], [1.5], [Number.NaN]])('throws for %s', (value) => { + expect(() => normalizePendingPromptLimit(value)).toThrow( + /bad maxPendingPromptsPerSession/, + ); + }); + }); + describe('health', () => { it('GETs /health and returns the body', async () => { const { fetch, calls } = recordingFetch(() => @@ -847,6 +882,323 @@ describe('DaemonClient', () => { ), ).rejects.toThrow(); }); + + it('rejects locally when a session reaches the pending prompt cap', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(202, { promptId: 'p-1', lastEventId: 0 }); + } + if (req.url.endsWith('/session/s-1/events')) { + return pendingSseResponse(); + } + if (req.url.endsWith('/session/s-1/cancel')) { + return new Response(null, { status: 204 }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 1, + }); + const ctrl = new AbortController(); + const first = client + .prompt( + 's-1', + { prompt: [{ type: 'text', text: 'first' }] }, + ctrl.signal, + ) + .catch((err: unknown) => err); + + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + const secondCtrl = new AbortController(); + const second = client + .prompt( + 's-1', + { prompt: [{ type: 'text', text: 'second' }] }, + secondCtrl.signal, + ) + .catch((err: unknown) => err); + try { + const secondResult = await Promise.race([ + second, + new Promise((resolve) => setTimeout(() => resolve('timed-out'), 50)), + ]); + expect(secondResult).toBeInstanceOf(DaemonPendingPromptLimitError); + expect((secondResult as Error).message).toContain('"s-1"'); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(1); + } finally { + ctrl.abort(); + secondCtrl.abort(); + await first; + await second; + } + }); + + it('maps server prompt queue full responses to the pending prompt limit error', async () => { + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(503, { + code: 'prompt_queue_full', + error: 'queue full', + sessionId: 's-1', + limit: 5, + pendingCount: 5, + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 0, + }); + + const result = await client + .prompt('s-1', { prompt: [{ type: 'text', text: 'hi' }] }) + .catch((err: unknown) => err); + + expect(result).toBeInstanceOf(DaemonPendingPromptLimitError); + expect(result).toMatchObject({ + sessionId: 's-1', + limit: 5, + pendingCount: 5, + }); + }); + + it('maps server prompt queue full responses on non-blocking prompts', async () => { + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(503, { + code: 'prompt_queue_full', + error: 'queue full', + sessionId: 's-1', + limit: 7, + pendingCount: 7, + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + }); + + const result = await client + .promptNonBlocking('s-1', { + prompt: [{ type: 'text', text: 'hi' }], + }) + .catch((err: unknown) => err); + + expect(result).toBeInstanceOf(DaemonPendingPromptLimitError); + expect(result).toMatchObject({ + sessionId: 's-1', + limit: 7, + pendingCount: 7, + }); + }); + + it('does not reserve a local prompt slot for a pre-aborted signal', async () => { + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(202, { promptId: 'p-1', lastEventId: 0 }); + } + if (req.url.endsWith('/session/s-1/events')) { + return pendingSseResponse(); + } + if (req.url.endsWith('/session/s-1/cancel')) { + return new Response(null, { status: 204 }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 1, + }); + const aborted = new AbortController(); + aborted.abort(); + + await expect( + client.prompt( + 's-1', + { prompt: [{ type: 'text', text: 'pre-aborted' }] }, + aborted.signal, + ), + ).rejects.toThrow(); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(0); + + const activeAbort = new AbortController(); + const active = client + .prompt( + 's-1', + { prompt: [{ type: 'text', text: 'active' }] }, + activeAbort.signal, + ) + .catch((err: unknown) => err); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(1); + + activeAbort.abort(); + await active; + }); + + it('releases the local pending prompt slot after turn completion', async () => { + let nextPromptId = 0; + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + nextPromptId += 1; + return jsonResponse(202, { + promptId: `p-${nextPromptId}`, + lastEventId: 0, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + const promptId = `p-${nextPromptId}`; + return sseResponse( + `id: 1\nevent: turn_complete\ndata: {"id":1,"v":1,"type":"turn_complete","data":{"promptId":"${promptId}","stopReason":"end_turn"}}\n\n`, + ); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 1, + }); + + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'first' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'second' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(2); + }); + + it('releases the local pending prompt slot after turn_error', async () => { + let nextPromptId = 0; + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + nextPromptId += 1; + return jsonResponse(202, { + promptId: `p-${nextPromptId}`, + lastEventId: 0, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + const promptId = `p-${nextPromptId}`; + const eventType = promptId === 'p-1' ? 'turn_error' : 'turn_complete'; + const data = + promptId === 'p-1' + ? { promptId, message: 'failed', code: 'turn_failed' } + : { promptId, stopReason: 'end_turn' }; + return sseResponse( + `id: 1\nevent: ${eventType}\ndata: ${JSON.stringify({ + id: 1, + v: 1, + type: eventType, + data, + })}\n\n`, + ); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 1, + }); + + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'first' }] }), + ).rejects.toThrow('failed'); + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'second' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(2); + }); + + it('releases the local pending prompt slot when SSE ends', async () => { + let nextPromptId = 0; + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + nextPromptId += 1; + return jsonResponse(202, { + promptId: `p-${nextPromptId}`, + lastEventId: 0, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + if (nextPromptId === 1) return sseResponse(''); + return sseResponse( + 'id: 1\nevent: turn_complete\ndata: {"id":1,"v":1,"type":"turn_complete","data":{"promptId":"p-2","stopReason":"end_turn"}}\n\n', + ); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 1, + }); + + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'first' }] }), + ).rejects.toThrow('SSE stream ended'); + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'second' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(2); + }); + + it('releases the local pending prompt slot after caller abort', async () => { + let nextPromptId = 0; + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/prompt')) { + nextPromptId += 1; + return jsonResponse(202, { + promptId: `p-${nextPromptId}`, + lastEventId: 0, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + if (nextPromptId === 1) return pendingSseResponse(); + return sseResponse( + 'id: 1\nevent: turn_complete\ndata: {"id":1,"v":1,"type":"turn_complete","data":{"promptId":"p-2","stopReason":"end_turn"}}\n\n', + ); + } + if (req.url.endsWith('/session/s-1/cancel')) { + return new Response(null, { status: 204 }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ + baseUrl: 'http://daemon', + fetch, + maxPendingPromptsPerSession: 1, + }); + const ctrl = new AbortController(); + const first = client.prompt( + 's-1', + { prompt: [{ type: 'text', text: 'first' }] }, + ctrl.signal, + ); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + + ctrl.abort(); + await expect(first).rejects.toThrow(); + await expect( + client.prompt('s-1', { prompt: [{ type: 'text', text: 'second' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(2); + }); }); describe('loadSession / resumeSession', () => { @@ -1927,7 +2279,7 @@ describe('DaemonClient', () => { ); const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); const iter = client.subscribeEvents('s-1'); - await expect(iter.next()).rejects.toThrow(/SSE response has no body/); + await expect(iter.next()).rejects.toThrow(/No SSE body/); }); it('throws DaemonHttpError when content-type is not text/event-stream', async () => { diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index 667847bc82..5c97818ee7 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -5,7 +5,10 @@ */ import { describe, it, expect, vi } from 'vitest'; -import { DaemonClient } from '../../src/daemon/DaemonClient.js'; +import { + DaemonClient, + DaemonPendingPromptLimitError, +} from '../../src/daemon/DaemonClient.js'; import { DaemonSessionClient } from '../../src/daemon/DaemonSessionClient.js'; function jsonResponse(status: number, body: unknown): Response { @@ -29,10 +32,14 @@ function sseResponse(frames: string): Response { }); } -function pendingSseResponse(onCancel: () => void): Response { +function pendingSseResponse( + onCancel: () => void, + onStart?: (controller: ReadableStreamDefaultController) => void, +): Response { const encoder = new TextEncoder(); const body = new ReadableStream({ start(controller) { + onStart?.(controller); controller.enqueue(encoder.encode(': keepalive\n\n')); }, cancel() { @@ -86,6 +93,29 @@ function recordingFetch( return { fetch: fetchImpl, calls }; } +function pendingPromptIds(session: DaemonSessionClient): string[] { + return [ + ...( + session as unknown as { + _pendingPrompts: Map; + } + )._pendingPrompts.keys(), + ]; +} + +async function waitForPendingPrompt( + session: DaemonSessionClient, + promptId: string, +): Promise { + await vi.waitFor(() => { + expect(pendingPromptIds(session)).toContain(promptId); + }); +} + +function turnCompleteFrame(promptId: string): string { + return `id: 1\nevent: turn_complete\ndata: {"id":1,"v":1,"type":"turn_complete","data":{"promptId":"${promptId}","stopReason":"end_turn"}}\n\n`; +} + describe('DaemonSessionClient', () => { it('creates or attaches a daemon session and exposes session metadata', async () => { const { fetch, calls } = recordingFetch(() => @@ -537,6 +567,346 @@ describe('DaemonSessionClient', () => { ]); }); + it('rejects locally in subscription mode when pending prompts reach the cap', async () => { + let eventsController: + | ReadableStreamDefaultController + | undefined; + const encoder = new TextEncoder(); + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/events')) { + return pendingSseResponse( + () => {}, + (controller) => { + eventsController = controller; + }, + ); + } + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(202, { promptId: 'p-1', lastEventId: 0 }); + } + if (req.url.endsWith('/session/s-1/cancel')) { + return new Response(null, { status: 204 }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = new DaemonSessionClient({ + client, + session: { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + }, + maxPendingPromptsPerSession: 1, + }); + const eventsAbort = new AbortController(); + const eventPump = (async () => { + for await (const _event of session.events({ + signal: eventsAbort.signal, + })) { + /* keep subscription active */ + } + })().catch(() => {}); + + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + const first = session + .prompt({ prompt: [{ type: 'text', text: 'first' }] }) + .catch((err: unknown) => err); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(1); + }); + await waitForPendingPrompt(session, 'p-1'); + + const secondCtrl = new AbortController(); + const second = session + .prompt({ prompt: [{ type: 'text', text: 'second' }] }, secondCtrl.signal) + .catch((err: unknown) => err); + try { + const secondResult = await Promise.race([ + second, + new Promise((resolve) => setTimeout(() => resolve('timed-out'), 50)), + ]); + expect(secondResult).toBeInstanceOf(DaemonPendingPromptLimitError); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(1); + } finally { + eventsController?.enqueue(encoder.encode(turnCompleteFrame('p-1'))); + eventsController?.close(); + secondCtrl.abort(); + eventsAbort.abort(); + await first; + await second; + await eventPump; + } + }); + + it('releases a subscription prompt slot after a non-202 result', async () => { + let eventsController: + | ReadableStreamDefaultController + | undefined; + const eventsAbort = new AbortController(); + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/events')) { + return pendingSseResponse( + () => {}, + (controller) => { + eventsController = controller; + }, + ); + } + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(200, { stopReason: 'end_turn' }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = new DaemonSessionClient({ + client, + session: { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + }, + maxPendingPromptsPerSession: 1, + }); + const eventPump = (async () => { + for await (const _event of session.events({ + signal: eventsAbort.signal, + })) { + /* keep subscription active */ + } + })().catch(() => {}); + + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'first' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'second' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(2); + + eventsController?.close(); + eventsAbort.abort(); + await eventPump; + }); + + it.each([[null], [0], [Infinity]])( + 'disables the subscription prompt cap for %s', + async (maxPendingPromptsPerSession) => { + let eventsController: + | ReadableStreamDefaultController + | undefined; + let nextPromptId = 0; + const encoder = new TextEncoder(); + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/events')) { + return pendingSseResponse( + () => {}, + (controller) => { + eventsController = controller; + }, + ); + } + if (req.url.endsWith('/session/s-1/prompt')) { + nextPromptId += 1; + return jsonResponse(202, { + promptId: `p-${nextPromptId}`, + lastEventId: 0, + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = new DaemonSessionClient({ + client, + session: { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + }, + maxPendingPromptsPerSession, + }); + const eventsAbort = new AbortController(); + const eventPump = (async () => { + for await (const _event of session.events({ + signal: eventsAbort.signal, + })) { + /* keep subscription active */ + } + })().catch(() => {}); + + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + const first = session.prompt({ + prompt: [{ type: 'text', text: 'first' }], + }); + const second = session.prompt({ + prompt: [{ type: 'text', text: 'second' }], + }); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(2); + }); + await waitForPendingPrompt(session, 'p-1'); + await waitForPendingPrompt(session, 'p-2'); + eventsController!.enqueue( + encoder.encode(turnCompleteFrame('p-1') + turnCompleteFrame('p-2')), + ); + + try { + await expect(first).resolves.toEqual({ stopReason: 'end_turn' }); + await expect(second).resolves.toEqual({ stopReason: 'end_turn' }); + } finally { + eventsController?.close(); + eventsAbort.abort(); + await eventPump; + } + }, + ); + + it('does not reserve a subscription prompt slot for a pre-aborted signal', async () => { + let eventsController: + | ReadableStreamDefaultController + | undefined; + const encoder = new TextEncoder(); + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/events')) { + return new Response( + new ReadableStream({ + start(controller) { + eventsController = controller; + controller.enqueue(encoder.encode(': keepalive\n\n')); + }, + }), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + } + if (req.url.endsWith('/session/s-1/prompt')) { + return jsonResponse(202, { promptId: 'p-1', lastEventId: 0 }); + } + if (req.url.endsWith('/session/s-1/cancel')) { + return new Response(null, { status: 204 }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = new DaemonSessionClient({ + client, + session: { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + }, + maxPendingPromptsPerSession: 1, + }); + const eventsAbort = new AbortController(); + const eventPump = (async () => { + for await (const _event of session.events({ + signal: eventsAbort.signal, + })) { + /* keep subscription active */ + } + })().catch(() => {}); + + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + const aborted = new AbortController(); + aborted.abort(); + await expect( + session.prompt( + { prompt: [{ type: 'text', text: 'pre-aborted' }] }, + aborted.signal, + ), + ).rejects.toThrow(); + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(0); + + const active = session + .prompt({ prompt: [{ type: 'text', text: 'active' }] }) + .catch((err: unknown) => err); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(1); + }); + await waitForPendingPrompt(session, 'p-1'); + eventsController!.enqueue(encoder.encode(turnCompleteFrame('p-1'))); + + try { + await expect(active).resolves.toEqual({ stopReason: 'end_turn' }); + } finally { + eventsController?.close(); + eventsAbort.abort(); + await eventPump; + } + }); + + it('rejects an accepted subscription prompt if the event stream has ended', async () => { + let eventsController: + | ReadableStreamDefaultController + | undefined; + let resolvePrompt: ((response: Response) => void) | undefined; + const promptResponse = new Promise((resolve) => { + resolvePrompt = resolve; + }); + const encoder = new TextEncoder(); + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/events')) { + return new Response( + new ReadableStream({ + start(controller) { + eventsController = controller; + controller.enqueue(encoder.encode(': keepalive\n\n')); + }, + }), + { status: 200, headers: { 'content-type': 'text/event-stream' } }, + ); + } + if (req.url.endsWith('/session/s-1/prompt')) { + return promptResponse; + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = new DaemonSessionClient({ + client, + session: { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + }, + maxPendingPromptsPerSession: 1, + }); + const eventPump = (async () => { + for await (const _event of session.events()) { + /* keep subscription active */ + } + })().catch(() => {}); + + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + const prompt = session + .prompt({ prompt: [{ type: 'text', text: 'late accept' }] }) + .catch((err: unknown) => err); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/prompt'))).toHaveLength(1); + }); + eventsController!.close(); + await eventPump; + resolvePrompt!(jsonResponse(202, { promptId: 'p-1', lastEventId: 0 })); + + const result = await Promise.race([ + prompt, + new Promise((resolve) => setTimeout(() => resolve('timed-out'), 50)), + ]); + expect(result).toBeInstanceOf(Error); + expect((result as Error).message).toBe('SSE stream ended'); + expect(pendingPromptIds(session)).toEqual([]); + }); + it('surfaces permission races and session operation failures', async () => { const { fetch } = recordingFetch((req) => { if (req.url.endsWith('/permission/missing-req')) { @@ -688,9 +1058,7 @@ describe('DaemonSessionClient', () => { }); const second = session.events(); - await expect(second.next()).rejects.toThrow( - 'Another event subscription is already active', - ); + await expect(second.next()).rejects.toThrow('subscription active'); await first.return(undefined);