mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-29 02:44:45 +00:00
fix(serve): Add prompt queue backpressure (#5033)
* fix(serve): add prompt queue backpressure Add per-session prompt admission limits across the bridge, REST and ACP entrypoints, and SDK clients. The server now rejects full prompt queues before returning accepted semantics, advertises the active limit through capabilities, and documents the behavior with focused tests and design artifacts. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * test(sdk): stabilize pending prompt cleanup Close the mocked SSE stream explicitly in the pending prompt cap test so cleanup does not rely on abort-driven stream cancellation timing in CI. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * test(sdk): stabilize subscription prompt race Reject accepted subscription prompts if the event stream has already ended, and make the prompt-cap tests wait for the pending registration before closing or injecting SSE frames. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk): map prompt queue full responses Map server-side prompt_queue_full responses to DaemonPendingPromptLimitError for both blocking and non-blocking prompt calls, include the session id in the local limit error, and cross-reference the duplicated default prompt cap constants. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * test: keep qwen planning docs ignored Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(serve): address prompt backpressure review Log synchronous prompt queue rejections, document the sync admission contract, clean up SDK prompt-slot release, and cover the reviewed backpressure edge cases. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> * fix(sdk): restore daemon bundle budget headroom Reduce the generated daemon client bundle slightly and raise the browser daemon SDK bundle budget to 116 KiB so the PR merge ref has practical headroom. Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> --------- Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> Co-authored-by: Shaojin Wen <shaojin.wensj@alibaba-inc.com>
This commit is contained in:
jinye
GitHub
parent
84d01e7070
commit
acb0275ecd
24 changed files with 1677 additions and 89 deletions
83
.qwen/design/prompt-queue-backpressure.md
Normal file
83
.qwen/design/prompt-queue-backpressure.md
Normal file
|
|
@ -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.
|
||||
55
.qwen/e2e-tests/prompt-queue-backpressure.md
Normal file
55
.qwen/e2e-tests/prompt-queue-backpressure.md
Normal file
|
|
@ -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
|
||||
```
|
||||
|
|
@ -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 <n>` | `4170` | TCP port. `0` = OS-assigned ephemeral port. |
|
||||
| `--hostname <addr>` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. |
|
||||
| `--token <str>` | — | 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 <n>` | `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 <path>` | `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 <n>` | `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 <n>` | `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 <n>` | — | 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 <m>` | `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 <pat>` | — | 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 (`<scheme>://<host>[:<port>]`, 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 <n>` | `4170` | TCP port. `0` = OS-assigned ephemeral port. |
|
||||
| `--hostname <addr>` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. |
|
||||
| `--token <str>` | — | 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 <n>` | `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 <n>` | `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 <path>` | `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 <n>` | `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 <n>` | `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 <n>` | — | 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 <m>` | `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 <pat>` | — | 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 (`<scheme>://<host>[:<port>]`, 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 <pattern>`** (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 (`<scheme>://<host>[:<port>]`, no trailing slash / path / userinfo). Matched origins receive proper CORS response headers (`Access-Control-Allow-Origin: <echoed>`, `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.
|
||||
|
|
|
|||
|
|
@ -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<void>((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<void>((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<void>((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<void>((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<void>((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)', () => {
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
/** 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;
|
||||
},
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<unknown, ServeArgs> = {
|
|||
'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<unknown, ServeArgs> = {
|
|||
}
|
||||
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<unknown, ServeArgs> = {
|
|||
token: argv.token,
|
||||
mode: 'http-bridge',
|
||||
maxSessions: argv['max-sessions'],
|
||||
maxPendingPromptsPerSession,
|
||||
maxConnections: argv['max-connections'],
|
||||
eventRingSize: argv['event-ring-size'],
|
||||
workspace: argv.workspace,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> };
|
||||
}>;
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -82,6 +82,7 @@ export {
|
|||
RestoreInProgressError,
|
||||
InvalidSessionScopeError,
|
||||
SessionLimitExceededError,
|
||||
PromptQueueFullError,
|
||||
WorkspaceMismatchError,
|
||||
InvalidClientIdError,
|
||||
InvalidPermissionOptionError,
|
||||
|
|
|
|||
|
|
@ -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 }
|
||||
: {}),
|
||||
|
|
|
|||
|
|
@ -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<PromptResponse>;
|
||||
) => Promise<PromptResponse> | 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],
|
||||
|
|
|
|||
|
|
@ -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<AcpSessionBridge['sendPrompt']>;
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
|
|
|
|||
|
|
@ -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<string, number> = 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<DaemonHttpError> {
|
||||
): Promise<DaemonHttpError>;
|
||||
private async failOnError(
|
||||
res: Response,
|
||||
label: string,
|
||||
sessionId: string,
|
||||
): Promise<DaemonHttpError | DaemonPendingPromptLimitError>;
|
||||
private async failOnError(
|
||||
res: Response,
|
||||
label: string,
|
||||
sessionId?: string,
|
||||
): Promise<DaemonHttpError | DaemonPendingPromptLimitError> {
|
||||
// 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<PromptResult> {
|
||||
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) {
|
||||
|
|
|
|||
|
|
@ -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<PromptResult> {
|
||||
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<PromptResult>((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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@
|
|||
export {
|
||||
DaemonClient,
|
||||
DaemonHttpError,
|
||||
DaemonPendingPromptLimitError,
|
||||
isDaemonTurnError,
|
||||
isNonBlockingAccepted,
|
||||
matchTurnEvent,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -11,6 +11,7 @@ export {
|
|||
DaemonCapabilityMissingError,
|
||||
DaemonClient,
|
||||
DaemonHttpError,
|
||||
DaemonPendingPromptLimitError,
|
||||
DaemonSessionClient,
|
||||
asKnownDaemonEvent,
|
||||
createDaemonSessionViewState,
|
||||
|
|
|
|||
|
|
@ -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<Uint8Array>({
|
||||
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<unknown>([
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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<Uint8Array>) => void,
|
||||
): Response {
|
||||
const encoder = new TextEncoder();
|
||||
const body = new ReadableStream<Uint8Array>({
|
||||
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<string, unknown>;
|
||||
}
|
||||
)._pendingPrompts.keys(),
|
||||
];
|
||||
}
|
||||
|
||||
async function waitForPendingPrompt(
|
||||
session: DaemonSessionClient,
|
||||
promptId: string,
|
||||
): Promise<void> {
|
||||
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<Uint8Array>
|
||||
| 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<unknown>([
|
||||
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<Uint8Array>
|
||||
| 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<Uint8Array>
|
||||
| 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<Uint8Array>
|
||||
| undefined;
|
||||
const encoder = new TextEncoder();
|
||||
const { fetch, calls } = recordingFetch((req) => {
|
||||
if (req.url.endsWith('/session/s-1/events')) {
|
||||
return new Response(
|
||||
new ReadableStream<Uint8Array>({
|
||||
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<Uint8Array>
|
||||
| undefined;
|
||||
let resolvePrompt: ((response: Response) => void) | undefined;
|
||||
const promptResponse = new Promise<Response>((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<Uint8Array>({
|
||||
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<unknown>([
|
||||
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);
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue