diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index ada8cc7331e..9e85688ba58 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -99,7 +99,7 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design 'session_set_model', 'client_identity', 'client_heartbeat', 'session_permission_vote', 'permission_vote', 'workspace_mcp', 'workspace_skills', 'workspace_providers', 'session_context', 'session_supported_commands', - 'session_close', 'session_metadata'] + 'session_close', 'session_metadata', 'mcp_guardrails'] ``` `session_scope_override` is the negotiation handle for the per-request `sessionScope` field on `POST /session` (see below). Older daemons silently ignore the field, so SDK clients should pre-flight `caps.features` for this tag before sending it. @@ -114,12 +114,18 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design `session_close` and `session_metadata` advertise `DELETE /session/:id` and `PATCH /session/:id/metadata`. Older daemons return `404`; pre-flight these tags before exposing close or rename affordances. +`mcp_guardrails` (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14) covers the MCP budget surface: the `clientCount` / `clientBudget` / `budgetMode` / `budgets[]` fields on `GET /workspace/mcp`, the `disabledReason` field on per-server cells, and the `--mcp-client-budget` / `--mcp-budget-mode` CLI flags. Older daemons omit the new fields entirely; SDK clients pre-flight this tag before relying on `budgets[]` semantics. The registry descriptor also carries `modes: ['warn', 'enforce']` for future feature-modes exposure — for now, clients infer mode from the snapshot's `budgetMode` field. Server refusal under `enforce` mode is deterministic by `Object.entries(mcpServers)` declaration order; a future scope-precedence layer (if qwen-code adopts one) would shift this to "lowest-precedence first" to mirror claude-code's `plugin < user < project < local` convention. + +> ⚠️ **PR 14 v1 scope: per-session, not per-workspace.** Each ACP session inside the daemon constructs its own `Config` + `McpClientManager` (via `acpAgent.newSessionConfig`). The budget caps live MCP clients **per session**; each session independently reads `QWEN_SERVE_MCP_CLIENT_BUDGET` from the forwarded env. With `--mcp-client-budget=10` and 5 concurrent ACP sessions, the actual live MCP client count can reach 5 × 10 = 50 across the daemon. The `GET /workspace/mcp` snapshot reads the **bootstrap session's** `McpClientManager` accounting only — the `budgets[0].scope: 'session'` value is the honest signal that this is per-session, not aggregated. **Wave 5 PR 23 (shared MCP pool)** will introduce a workspace-scoped manager and add a `scope: 'workspace'` cell alongside the per-session cell for true cross-session aggregation. v1 is the in-process counter + soft enforcement foundation that PR 23 builds on. + **Conditional tags.** A small number of feature tags are advertised only when the matching deployment toggle is on. Tag presence = behavior is on; absence = either an older daemon predating the tag, OR a current daemon where the operator did not opt in. Currently: | Tag | Advertised when … | | -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `require_auth` | the daemon was started with `--require-auth` (or `requireAuth: true` via the embedded API). Bearer token is mandatory on every route, including `/health` on loopback binds. | +`mcp_guardrails` is **not** in this conditional table — it's an always-on tag, advertised whenever the binary supports the new `/workspace/mcp` budget fields, regardless of whether the operator configured a budget. Operators who haven't set `--mcp-client-budget` still get the new fields (with `budgetMode: 'off'`, `budgets: []`). + ## Routes ### `GET /health` @@ -249,6 +255,76 @@ vars only; proxy URLs are stripped of credentials and reduced to `transport` is one of `stdio`, `sse`, `http`, `websocket`, `sdk`, or `unknown`. `errors` is omitted when discovery succeeds. +**MCP client guardrails (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14).** Post-PR-14 daemons extend the payload with four additive fields and one workspace-level cell: + +```jsonc +{ + "v": 1, + "workspaceCwd": "/canonical/path", + "initialized": true, + "discoveryState": "completed", + "clientCount": 3, + "clientBudget": 2, + "budgetMode": "enforce", + "budgets": [ + { + "kind": "mcp_budget", + "scope": "session", + "status": "error", + "errorKind": "budget_exhausted", + "hint": "Raise --mcp-client-budget or remove servers from mcpServers config.", + "liveCount": 2, + "budget": 2, + "mode": "enforce", + "refusedCount": 1, + }, + ], + "servers": [ + { + "kind": "mcp_server", + "status": "ok", + "name": "a", + "mcpStatus": "connected", + "transport": "stdio", + "disabled": false, + }, + { + "kind": "mcp_server", + "status": "ok", + "name": "b", + "mcpStatus": "connected", + "transport": "stdio", + "disabled": false, + }, + { + "kind": "mcp_server", + "status": "error", + "name": "c", + "mcpStatus": "disconnected", + "transport": "stdio", + "disabled": false, + "disabledReason": "budget", + "errorKind": "budget_exhausted", + "hint": "...", + }, + ], +} +``` + +`budgetMode` is one of `enforce`, `warn`, or `off`. `clientBudget` is absent when no budget was set. `budgets[]` is **always an array** on post-PR-14 daemons (possibly empty when `budgetMode === 'off'`); pre-PR-14 daemons omit the field entirely. v1 emits one cell with `scope: 'session'` (per-session enforcement — see the capabilities section above for why). Consumers MUST tolerate additional `budgets[]` entries with unrecognized `scope` values — Wave 5 PR 23 will add `scope: 'workspace'` (or `'pool'`) alongside the per-session cell without a schema bump. + +`disabledReason` on per-server cells distinguishes operator-disabled (`'config'` — `disabledMcpServers` config list) from budget-refused (`'budget'` — discovered but never connected due to `enforce` mode). Refusals are deterministic by `Object.entries(mcpServers)` declaration order. The per-server `status: 'error', errorKind: 'budget_exhausted'` shadows the raw `mcpStatus: 'disconnected'` (which is true but not the operator-facing severity). + +Budget enforcement in PR 14 v1 is **per-session, not per-workspace**. Although Mode B daemons are `1 daemon = 1 workspace × N sessions` post-#4113 at the process level, the `McpClientManager` is constructed inside each ACP session's `Config` via `acpAgent.newSessionConfig`, so N sessions each enforce their own copy of the cap. The snapshot represents the bootstrap session's view. Wave 5 PR 23 introduces a workspace-scoped shared MCP pool that graduates this to true per-workspace enforcement. + +**Detecting budget pressure in v1 (no push events yet).** PR 14 v1 is snapshot-only — typed SSE push events (`mcp_budget_warning` + `mcp_child_refused_batch`) ship in PR 14b. Until then, operator dashboards poll `GET /workspace/mcp` and inspect the per-session budget cell (`budgets[0]`): + +- `budgets[0].status === 'warning'` ⇔ `liveCount >= 0.75 * clientBudget` (matches the hysteresis threshold PR 14b's push event will use). +- `budgets[0].status === 'error'` ⇔ `refusedCount > 0` (one or more servers refused this discovery pass). +- `budgets[0].status === 'ok'` ⇔ below the 75% threshold AND no refusals. + +Recommended poll cadence: aligned with whatever already polls `/workspace/mcp`; the snapshot is cheap and the budget cell carries no extra discovery cost. + ### `GET /workspace/skills` ```json diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index ee9ab49db58..31b3a609b98 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -151,17 +151,19 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 ## CLI flags -| Flag | Default | Purpose | -| ----------------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -| `--port ` | `4170` | TCP port. `0` = OS-assigned ephemeral port. | -| `--hostname ` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. | -| `--token ` | — | Bearer token. Falls back to `QWEN_SERVER_TOKEN` env var (with leading/trailing whitespace stripped — handy for `$(cat token.txt)`). | -| `--require-auth` | `false` | Refuse to start without a bearer token, even on loopback. Hardens the `127.0.0.1` developer default for shared dev hosts / CI runners / multi-tenant workstations where any local user can hit the listener. Boots only with `--token` or `QWEN_SERVER_TOKEN` set; gates `/health` behind the bearer too. | -| `--max-sessions ` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~30–50 MB per session). | -| `--workspace ` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. | -| `--max-connections ` | `256` | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session. | -| `--event-ring-size ` | `8000` | Per-session SSE replay ring depth (#3803 §02 target). Sets the backlog available to `GET /session/:id/events` with `Last-Event-ID: N`. Larger = more reconnect headroom at the cost of a few hundred KB extra RAM per session. SDK clients can additionally request a larger per-subscriber backlog cap on a specific subscription via `?maxQueued=N` (range `[16, 2048]`, default 256). Daemons also emit a non-terminal `slow_client_warning` SSE frame at 75% queue fill so clients can drain / reconnect before getting evicted. Pre-flight `caps.features.slow_client_warning`. | -| `--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. | +| Flag | Default | Purpose | +| ------------------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `--port ` | `4170` | TCP port. `0` = OS-assigned ephemeral port. | +| `--hostname ` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. | +| `--token ` | — | Bearer token. Falls back to `QWEN_SERVER_TOKEN` env var (with leading/trailing whitespace stripped — handy for `$(cat token.txt)`). | +| `--require-auth` | `false` | Refuse to start without a bearer token, even on loopback. Hardens the `127.0.0.1` developer default for shared dev hosts / CI runners / multi-tenant workstations where any local user can hit the listener. Boots only with `--token` or `QWEN_SERVER_TOKEN` set; gates `/health` behind the bearer too. | +| `--max-sessions ` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~30–50 MB per session). | +| `--workspace ` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. | +| `--max-connections ` | `256` | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session. | +| `--event-ring-size ` | `8000` | Per-session SSE replay ring depth (#3803 §02 target). Sets the backlog available to `GET /session/:id/events` with `Last-Event-ID: N`. Larger = more reconnect headroom at the cost of a few hundred KB extra RAM per session. SDK clients can additionally request a larger per-subscriber backlog cap on a specific subscription via `?maxQueued=N` (range `[16, 2048]`, default 256). Daemons also emit a non-terminal `slow_client_warning` SSE frame at 75% queue fill so clients can drain / reconnect before getting evicted. Pre-flight `caps.features.slow_client_warning`. | +| `--mcp-client-budget ` | — | Positive integer cap on live MCP clients **per ACP session** (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14 v1; PR 23 graduates this to per-workspace via the shared MCP pool). Combine with `--mcp-budget-mode`. When unset, no accounting-driven enforcement (but `GET /workspace/mcp` still reports `clientCount`). Distinct from claude-code's `MCP_SERVER_CONNECTION_BATCH_SIZE` which gates startup concurrency, not the total client count. Pre-flight `caps.features.mcp_guardrails`. | +| `--mcp-budget-mode ` | `warn` / `off` | How `--mcp-client-budget` is enforced. `warn` (default when budget set): no refusal, snapshot's `budgets[0].status` flips to `warning` at ≥75% of budget. `enforce`: connects past the cap are refused, per-server cell shows `disabledReason: 'budget'`, deterministic by `mcpServers` declaration order. `off` (default when budget unset): pure observability. Boot rejects `enforce` without a budget. | +| `--http-bridge` | `true` | Stage 1 mode: one `qwen --acp` child per daemon (bound to one workspace at boot, per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02); N sessions multiplex onto that child via ACP `newSession()`. Stage 2 native in-process becomes available later. | > **Sizing the load knobs.** `--max-sessions` is the **new-child** cap. > Three other layers also limit load — when sizing for a high-concurrency @@ -181,6 +183,18 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 > 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. +> +> ⚠️ **v1 scope: per-session, not per-workspace.** Each ACP session inside the daemon has its own `Config`/`McpClientManager` (created via `newSessionConfig` per session). The budget caps live MCP clients **per session**, not aggregated across all sessions in the workspace. Snapshot at `GET /workspace/mcp` reflects the bootstrap session's view (the cell carries `scope: 'session'` for honesty). If you run 5 concurrent ACP sessions with `--mcp-client-budget=10`, you may have up to 50 live MCP clients across the daemon — the cap holds per session. **Wave 5 PR 23 (shared MCP pool)** introduces a workspace-scoped manager and graduates this to true per-workspace enforcement. +> +> ```sh +> qwen serve --mcp-client-budget=10 --mcp-budget-mode=warn +> # later, after telemetry shows your real-world distribution: +> qwen serve --mcp-client-budget=10 --mcp-budget-mode=enforce +> ``` +> +> This is **not** the same as claude-code's `MCP_SERVER_CONNECTION_BATCH_SIZE` (which gates startup concurrency); they're orthogonal. PR 23 will add a real shared MCP pool (a `scope: 'workspace'` cell in `budgets[]` alongside the per-session cell); PR 14 v1 is the in-process counter + soft enforcement on the existing per-session manager. + ## Default deployment threat model - **127.0.0.1 only** — loopback bind, no auth needed. diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index e14bdf1cc74..4b70ed88546 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -12,6 +12,7 @@ import { createDebugLogger, QwenOAuth2Event, qwenOAuth2Events, + MCP_BUDGET_WARN_FRACTION, MCPServerConfig, SessionService, SESSION_TITLE_MAX_LENGTH, @@ -87,6 +88,8 @@ import { mapDomainErrorToErrorKind, type AcpPreflightKind, type ServeErrorKind, + type ServeMcpBudgetMode, + type ServeMcpBudgetStatusCell, type ServeMcpDiscoveryState, type ServeMcpServerRuntimeStatus, type ServeMcpTransport, @@ -691,6 +694,41 @@ class QwenAgent implements Agent { try { const workspaceCwd = this.workspaceCwd(config); const servers = config.getMcpServers() ?? {}; + + // PR 14: pull live accounting + budget config from the child's + // McpClientManager so the daemon's read-only route reflects the + // single source of truth (not a daemon-side polled cache). + // `getToolRegistry()` and `getMcpClientManager()` are best-effort + // — older test stubs or partially-initialized configs may not + // expose them; in that case we fall back to "no budget surface". + let clientCount: number | undefined; + let clientBudget: number | undefined; + let budgetMode: ServeMcpBudgetMode | undefined; + let refusedSet: ReadonlySet = new Set(); + try { + const manager = config.getToolRegistry()?.getMcpClientManager(); + if (manager) { + const accounting = manager.getMcpClientAccounting(); + clientCount = accounting.total; + clientBudget = manager.getMcpClientBudget(); + budgetMode = manager.getMcpBudgetMode(); + refusedSet = new Set(accounting.refusedServerNames); + } + } catch (err) { + // Accounting failure must not crash the snapshot — the per- + // server data is still useful even without budget overlay. + // PR 14 fix (review #4247 wenshao S7a): bumped from + // `debugLogger.debug` to stderr `process.stderr.write` so a + // production daemon emits a visible warning when accounting + // breaks. `debugLogger.debug` is gated on the operator + // having set debug=true, which makes silent slot-leak / type- + // mismatch failures invisible in real deployments. + process.stderr.write( + `qwen serve: getMcpClientAccounting failed: ` + + `${err instanceof Error ? err.message : String(err)}\n`, + ); + } + return { v: STATUS_SCHEMA_VERSION, workspaceCwd, @@ -699,14 +737,39 @@ class QwenAgent implements Agent { servers: Object.entries(servers).map(([name, server]) => { const disabled = config.isMcpServerDisabled(name); const rawStatus = getMCPServerStatus(name); + const refusedByBudget = refusedSet.has(name); + // PR 14 fix (review #4247): config-disable takes precedence + // over budget-refusal. `lastRefusedServerNames` is a + // per-discovery-pass snapshot; if an operator runs + // `/mcp disable ` against a server that was refused + // last pass, the entry stays in the refused list until the + // next discovery pass clears it (`McpClientManager.removeServer` + // now drops the entry too — see sibling fix). Either way, + // a `disabled` cell should NEVER show `budget_exhausted` — + // the operator's deliberate disable wins. + const effectivelyRefused = refusedByBudget && !disabled; const out: ServeWorkspaceMcpServerStatus = { kind: 'mcp_server', - status: this.mcpCellStatus(rawStatus, disabled), + // Refused-by-budget shadows the raw status: the rawStatus + // is `DISCONNECTED` (we never tried to connect), but the + // operator-facing severity is `error` with an explanatory + // errorKind rather than the generic disconnected `error`. + status: effectivelyRefused + ? 'error' + : this.mcpCellStatus(rawStatus, disabled), name, mcpStatus: this.mcpStatus(rawStatus), transport: this.mcpTransport(server), disabled, }; + if (effectivelyRefused) { + out.errorKind = 'budget_exhausted'; + out.disabledReason = 'budget'; + out.hint = + 'Raise --mcp-client-budget or remove servers from mcpServers config.'; + } else if (disabled) { + out.disabledReason = 'config'; + } const description = server && typeof server === 'object' ? (server as { description?: unknown }).description @@ -723,6 +786,31 @@ class QwenAgent implements Agent { } return out; }), + ...(clientCount !== undefined ? { clientCount } : {}), + ...(clientBudget !== undefined ? { clientBudget } : {}), + ...(budgetMode !== undefined ? { budgetMode } : {}), + ...(budgetMode !== undefined + ? { + // PR 14 fix (review #4247 wenshao R2-#6): filter out + // servers that are now config-disabled so the + // workspace cell matches the per-server cell + // precedence (`effectivelyRefused = refusedByBudget + // && !disabled` above). Pre-fix a server disabled + // after being refused would render `disabled` on its + // per-server row but `error: budget_exhausted` on the + // workspace row — confusing for dashboards. Use + // `Array.from(refusedSet).filter(...)` to apply the + // same disabled gate the per-server loop applies. + budgets: this.buildBudgetCells( + clientCount ?? 0, + clientBudget, + budgetMode, + Array.from(refusedSet).filter( + (n) => !config.isMcpServerDisabled(n), + ).length, + ), + } + : {}), }; } catch (error) { return { @@ -735,6 +823,88 @@ class QwenAgent implements Agent { } } + /** + * Build the MCP budget status cells exposed on `GET /workspace/mcp` + * (PR 14). v1 emits one cell with `scope: 'session'` — each ACP + * session has its own `McpClientManager`, so the budget enforces + * per-session (snapshot reflects the bootstrap session's view). + * Wave 5 PR 23 (shared MCP pool) will add `scope: 'workspace'` + * for true per-workspace aggregation. Consumers MUST tolerate + * additional entries with unrecognized scope values (drop, don't + * fail). + * + * Cell `status` semantics: + * - `error` — refusals happened this pass (only possible in enforce mode) + * - `warning` — live count crossed 75% of budget (warn or enforce mode) + * - `ok` — under threshold (or `off` mode) + * + * **`liveCount` vs `reservedSlots.size` (PR 14 review #4247 R9 #5)**: + * `liveCount` here is `accounting.total` — only `MCPServerStatus.CONNECTED` + * clients. Enforcement (`tryReserveSlot`) on the other hand uses + * `reservedSlots.size` — all reserved names, including in-flight + * connects and never-connected stale entries. The two diverge when + * servers hold a slot during the connect handshake or after a + * connect failure that didn't release (e.g. `'already_held'` + * reconnect timeouts). The snapshot intentionally uses the live + * count for **operator observability** — "how many MCP clients + * are actually serving requests right now" — while enforcement + * uses the reservation count to prevent capacity races across + * `Promise.all` microtask boundaries. PR 14b's typed events + * should consider exposing both for real-time pressure signals. + */ + private buildBudgetCells( + liveCount: number, + budget: number | undefined, + mode: ServeMcpBudgetMode, + refusedCount: number, + ): ServeMcpBudgetStatusCell[] { + // PR 14 fix (review #4247): when no `--mcp-client-budget` is + // configured the manager resolves to `mode: 'off'`. The protocol + // docs and SDK type comments promise `budgets: []` for that case; + // a synthetic `mcp_budget` cell carrying nothing actionable was + // (a) protocol-noncompliant, (b) clutter — clients iterating + // `budgets[]` to render rows would draw an "ok" budget row for + // uncapped workspaces. Always return empty so the top-level + // `budgetMode: 'off'` field is the sole signal that guardrails + // are inactive. + if (mode === 'off') return []; + let status: ServeStatus = 'ok'; + let errorKind: ServeErrorKind | undefined; + let hint: string | undefined; + if (refusedCount > 0) { + status = 'error'; + errorKind = 'budget_exhausted'; + hint = + 'Raise --mcp-client-budget or remove servers from mcpServers config.'; + } else if ( + budget !== undefined && + budget > 0 && + liveCount >= MCP_BUDGET_WARN_FRACTION * budget + ) { + status = 'warning'; + hint = `Live MCP clients are above ${Math.round( + MCP_BUDGET_WARN_FRACTION * 100, + )}% of the configured budget.`; + } + const cell: ServeMcpBudgetStatusCell = { + kind: 'mcp_budget', + // PR 14 v1: per-session, not per-workspace. Each ACP session has + // its own `Config`/`McpClientManager` (via `newSessionConfig`) + // and reads `QWEN_SERVE_MCP_CLIENT_BUDGET` independently. + // Snapshot shows the bootstrap session's view. Wave 5 PR 23 + // shared MCP pool will graduate this to `'workspace'`. + scope: 'session', + status, + liveCount, + mode, + refusedCount, + }; + if (budget !== undefined) cell.budget = budget; + if (errorKind) cell.errorKind = errorKind; + if (hint) cell.hint = hint; + return [cell]; + } + private errorCell( kind: string, error: unknown, diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index f63174dddd8..dae40a9285f 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -12,6 +12,7 @@ import type { Argv, CommandModule } from 'yargs'; // handler below so it only loads when the user actually runs `qwen serve`. import { writeStderrLine } from '../utils/stdioHelpers.js'; import { DEFAULT_RING_SIZE } from '../serve/eventBus.js'; +import { MCP_BUDGET_WARN_FRACTION } from '@qwen-code/qwen-code-core'; /** * Pause the current async function indefinitely. Used after the daemon @@ -38,6 +39,8 @@ interface ServeArgs { // synthesizes is convenient for handlers but type-confusing here. The // handler reads `argv['http-bridge']` directly. 'http-bridge': boolean; + 'mcp-client-budget'?: number; + 'mcp-budget-mode'?: 'enforce' | 'warn' | 'off'; } export const serveCommand: CommandModule = { @@ -120,6 +123,27 @@ export const serveCommand: CommandModule = { 'one workspace at boot, multiplexing N sessions onto that child via ' + "the agent's native `newSession()`). Stage 2 native in-process mode " + 'is not yet implemented; this flag will become opt-in then.', + }) + .option('mcp-client-budget', { + type: 'number', + description: + 'Cap on live MCP clients spawned inside the ACP child for the bound ' + + 'workspace (issue #4175 PR 14). Positive integer. Combine with ' + + '--mcp-budget-mode to control behavior at the cap. When unset, ' + + 'mode defaults to off (no accounting-driven enforcement, but ' + + 'GET /workspace/mcp still reports `clientCount`). Distinct from ' + + 'claude-code MCP_SERVER_CONNECTION_BATCH_SIZE which gates startup ' + + 'concurrency, not the total client count.', + }) + .option('mcp-budget-mode', { + choices: ['enforce', 'warn', 'off'] as const, + description: + 'How --mcp-client-budget is enforced (issue #4175 PR 14). ' + + '`warn` (default when budget set): no refusal, snapshot surfaces ' + + 'warning at >=75% of budget. `enforce`: connects past the cap are ' + + 'refused (`disabledReason: "budget"`, deterministic by mcpServers ' + + 'declaration order). `off`: pure observability. Boot rejects ' + + '`enforce` without a budget.', }) as unknown as Argv, handler: async (argv) => { if (!argv['http-bridge']) { @@ -139,6 +163,46 @@ export const serveCommand: CommandModule = { 'deployment.', ); } + // PR 14: validate budget + mode combination at boot, before we + // lazy-load the serve module. Yargs already constrains `choices` + // for mcp-budget-mode, so we only have to police the budget value + // and the `enforce` ⇒ budget invariant. + const mcpClientBudget = argv['mcp-client-budget']; + const mcpBudgetMode = argv['mcp-budget-mode']; + if (mcpClientBudget !== undefined) { + if ( + !Number.isFinite(mcpClientBudget) || + !Number.isInteger(mcpClientBudget) || + mcpClientBudget <= 0 + ) { + writeStderrLine( + 'qwen serve: --mcp-client-budget must be a positive integer.', + ); + process.exit(1); + } + } + if (mcpBudgetMode === 'enforce' && mcpClientBudget === undefined) { + writeStderrLine( + 'qwen serve: --mcp-budget-mode=enforce requires --mcp-client-budget=N.', + ); + process.exit(1); + } + const resolvedMcpMode: 'enforce' | 'warn' | 'off' = + mcpBudgetMode ?? (mcpClientBudget !== undefined ? 'warn' : 'off'); + if (mcpClientBudget !== undefined) { + // Mirror PR 15's `--require-auth` breadcrumb: surface the active + // policy in stderr (journald / docker logs) so operators don't + // have to parse /capabilities or /workspace/mcp to confirm it. + writeStderrLine( + `qwen serve: --mcp-client-budget=${mcpClientBudget} mode=${resolvedMcpMode}` + + (resolvedMcpMode === 'enforce' + ? ' (servers past the cap will be refused at discovery)' + : resolvedMcpMode === 'warn' + ? ` (warnings at >=${Math.ceil(mcpClientBudget * MCP_BUDGET_WARN_FRACTION)}, no refusal)` + : ''), + ); + } + // Lazy-load the serve module so non-serve invocations don't pay for // express + body-parser + qs in their startup path. const { runQwenServe } = await import('../serve/index.js'); @@ -153,6 +217,8 @@ export const serveCommand: CommandModule = { eventRingSize: argv['event-ring-size'], workspace: argv.workspace, requireAuth: argv['require-auth'], + mcpClientBudget, + mcpBudgetMode: resolvedMcpMode, }); } catch (err) { writeStderrLine( diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index 9ba3356627f..5948a71a186 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -20,6 +20,18 @@ export interface ServeProtocolVersions { export interface ServeCapabilityDescriptor { since: ServeProtocolVersion; + /** + * Sub-mode names supported by this capability, when the feature has + * more than one operating mode and clients benefit from feature- + * detecting the active set. Optional — baseline tags (always-on, + * single behavior) omit this field. + * + * Introduced for `mcp_guardrails` (issue #4175 PR 14) where the + * tag advertises `['warn', 'enforce']` so clients can pre-flight + * whether the daemon supports refusal-on-budget-exhausted before + * relying on `mcp_child_refused_batch` semantics. + */ + modes?: readonly string[]; } export const SERVE_CAPABILITY_REGISTRY = { @@ -60,6 +72,19 @@ export const SERVE_CAPABILITY_REGISTRY = { session_supported_commands: { since: 'v1' }, session_close: { since: 'v1' }, session_metadata: { since: 'v1' }, + // Issue #4175 PR 14. Daemon supports the MCP client guardrail + // surface: an in-process counter exposed on `GET /workspace/mcp` + // (`clientCount`, `clientBudget`, `budgetMode`, `budgets[]`), a + // `--mcp-client-budget=N` flag with `--mcp-budget-mode={enforce, + // warn, off}`, and a `disabledReason: 'budget'` tag on per-server + // cells when refused at discovery. `modes` enumerates the + // implemented behaviors — clients pre-flight `'enforce'` before + // relying on refusal semantics, since a future split (e.g. PR 23 + // shared pool) could shift enforcement elsewhere. Listed BEFORE + // `require_auth` so always-on tags stay grouped together; + // `require_auth` is the only conditional tag, kept last for + // visibility in `Object.keys(SERVE_CAPABILITY_REGISTRY)`. + mcp_guardrails: { since: 'v1', modes: ['warn', 'enforce'] }, // Issue #4175 PR 15. Daemon was booted with `--require-auth` (or // `requireAuth: true`), so even loopback callers must carry a bearer // token. Advertised CONDITIONALLY — only when the flag is on — so diff --git a/packages/cli/src/serve/httpAcpBridge.test.ts b/packages/cli/src/serve/httpAcpBridge.test.ts index 667dd1b4af2..80a191f23ff 100644 --- a/packages/cli/src/serve/httpAcpBridge.test.ts +++ b/packages/cli/src/serve/httpAcpBridge.test.ts @@ -320,6 +320,53 @@ describe('createHttpAcpBridge', () => { ); }); + it('forwards childEnvOverrides to the channelFactory at spawn time (#4247 R6 line 216)', async () => { + // Round 6 (wenshao R5 line 216): pre-fix `runQwenServe` set + // `process.env` globally to pass the MCP budget config to the + // ACP child. With concurrent embedded daemons, the last + // `runQwenServe` to set the var would silently win for all + // other daemons' subsequent spawns (because + // `defaultSpawnChannelFactory` snapshots `process.env` AT + // SPAWN TIME, not at runQwenServe time). The fix routes the + // env through `BridgeOptions.childEnvOverrides` closed over + // inside each bridge — so each bridge's spawn factory sees + // ITS own overrides, regardless of what other daemons did. + const seenEnvs: Array | undefined> = []; + const factory: ChannelFactory = async (_cwd, env) => { + // Snapshot the override map so later iterations don't + // accidentally mutate the recorded value. + seenEnvs.push(env ? { ...env } : env); + return makeChannel().channel; + }; + const bridge1 = makeBridge({ + channelFactory: factory, + childEnvOverrides: { + QWEN_SERVE_MCP_CLIENT_BUDGET: '5', + QWEN_SERVE_MCP_BUDGET_MODE: 'enforce', + }, + }); + const bridge2 = makeBridge({ + channelFactory: factory, + childEnvOverrides: { + QWEN_SERVE_MCP_CLIENT_BUDGET: '20', + QWEN_SERVE_MCP_BUDGET_MODE: 'warn', + }, + }); + await bridge1.spawnOrAttach({ workspaceCwd: WS_A }); + await bridge2.spawnOrAttach({ workspaceCwd: WS_A }); + expect(seenEnvs).toHaveLength(2); + expect(seenEnvs[0]).toEqual({ + QWEN_SERVE_MCP_CLIENT_BUDGET: '5', + QWEN_SERVE_MCP_BUDGET_MODE: 'enforce', + }); + expect(seenEnvs[1]).toEqual({ + QWEN_SERVE_MCP_CLIENT_BUDGET: '20', + QWEN_SERVE_MCP_BUDGET_MODE: 'warn', + }); + await bridge1.shutdown(); + await bridge2.shutdown(); + }); + it('spawns a session and returns the agent-assigned id', async () => { const handles: ChannelHandle[] = []; const factory: ChannelFactory = async () => { diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index a44323f6c8b..483fc0fcd79 100644 --- a/packages/cli/src/serve/httpAcpBridge.ts +++ b/packages/cli/src/serve/httpAcpBridge.ts @@ -648,7 +648,10 @@ export interface AcpChannelExitInfo { signalCode: NodeJS.Signals | null; } -export type ChannelFactory = (workspaceCwd: string) => Promise; +export type ChannelFactory = ( + workspaceCwd: string, + childEnvOverrides?: Readonly>, +) => Promise; // FIXME(stage-1.5, chiga0 finding 1 + 4): // Stage 1.5 should split this file's responsibilities into: @@ -748,6 +751,30 @@ export interface BridgeOptions { * MUST canonicalize before passing. */ boundWorkspace: string; + /** + * PR 14 fix (review #4247 wenshao R5 runQwenServe.ts:216): per- + * handle env overrides forwarded to `defaultSpawnChannelFactory` + * at spawn time. Replaces the prior `process.env` mutation in + * `runQwenServe` so concurrent embedded daemons in the same + * process don't cross-contaminate each other's MCP budget / + * mode env (the `defaultSpawnChannelFactory` snapshots + * `process.env` AT SPAWN TIME, not at `runQwenServe()` call + * time — so the last `runQwenServe()` to set the global env + * would win for all subsequent spawns across all daemon + * handles, breaking the documented per-daemon policy). + * + * Shape: `Record`. A `string` value + * sets the env var for the child; `undefined` explicitly + * REMOVES the var from the child env (useful for "this daemon + * has no MCP budget" embedded callers that need to scrub a + * stale global). Keys NOT present in this record are inherited + * from `process.env` as before. + * + * Custom `channelFactory` callers receive this through the + * factory's second arg and decide what to do with it (tests + * typically ignore it; the production factory merges it). + */ + childEnvOverrides?: Readonly>; } /** @@ -1481,6 +1508,17 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { ); } const channelFactory = opts.channelFactory ?? defaultSpawnChannelFactory; + // PR 14 fix (review #4247 wenshao R5 runQwenServe.ts:216): close over + // a per-handle env-override snapshot. Calls to `channelFactory` at + // spawn time receive this as the 2nd arg, so the default factory + // can merge into the child env without consulting any global state + // that another concurrent `runQwenServe()` handle might have + // mutated. Frozen to make accidental mutation throw rather than + // silently corrupt later spawns. + const childEnvOverrides: Readonly> = + opts.childEnvOverrides + ? Object.freeze({ ...opts.childEnvOverrides }) + : Object.freeze({}); const initTimeoutMs = opts.initializeTimeoutMs ?? DEFAULT_INIT_TIMEOUT_MS; if (initTimeoutMs <= 0) { throw new TypeError( @@ -1782,7 +1820,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { if (inFlightChannelSpawn) return await inFlightChannelSpawn; const promise = (async () => { - const channel = await channelFactory(boundWorkspace); + const channel = await channelFactory(boundWorkspace, childEnvOverrides); const client = new BridgeClient( // BfFut: ACP today carries a sessionId on every per-session // notification / request, so the no-sessionId branch is @@ -4005,6 +4043,7 @@ async function safeCheck( */ export const defaultSpawnChannelFactory: ChannelFactory = async ( workspaceCwd, + childEnvOverrides, ) => { // Resolution order: // 1. `QWEN_CLI_ENTRY` env override — escape hatch for non-standard @@ -4066,6 +4105,24 @@ export const defaultSpawnChannelFactory: ChannelFactory = async ( for (const key of SCRUBBED_CHILD_ENV_KEYS) { delete childEnv[key]; } + // PR 14 fix (review #4247 wenshao R5 runQwenServe.ts:216): apply + // per-handle env overrides on top of the process.env snapshot. + // `undefined` value means "delete this var from the child env" so + // an embedded caller can scrub a stale inherited var without + // having to mutate the daemon's global process.env. Applied AFTER + // `SCRUBBED_CHILD_ENV_KEYS` so the daemon-only secret list still + // wins (operators can't override the scrub by passing + // `QWEN_SERVER_TOKEN` in overrides — defense in depth). + if (childEnvOverrides) { + for (const [key, value] of Object.entries(childEnvOverrides)) { + if (SCRUBBED_CHILD_ENV_KEYS.has(key)) continue; + if (value === undefined) { + delete childEnv[key]; + } else { + childEnv[key] = value; + } + } + } // CodeQL `js/path-injection` flags the `cwd: workspaceCwd` flow. // Stage 1 trust model accepts this — see the function-level comment // above for the design rationale. Defense-in-depth: the cwd is diff --git a/packages/cli/src/serve/runQwenServe.ts b/packages/cli/src/serve/runQwenServe.ts index 223b70e6408..8015158b535 100644 --- a/packages/cli/src/serve/runQwenServe.ts +++ b/packages/cli/src/serve/runQwenServe.ts @@ -177,6 +177,60 @@ export async function runQwenServe( // server.ts's raw `opts.workspace` and clients see one path on // `/capabilities` but another on `POST /session` responses. const boundWorkspace = canonicalizeWorkspace(rawWorkspace); + // Issue #4175 PR 14. The MCP client guardrails enforce in the ACP + // child process (where `McpClientManager` lives), not the daemon. + // Forward the budget config via env vars so the child's + // `readBudgetFromEnv()` picks them up. + // + // PR 14 fix (review #4247 wenshao R5 line 216): use per-handle env + // overrides via `BridgeOptions.childEnvOverrides` instead of + // mutating global `process.env`. Pre-fix concurrent embedded + // daemons (`runQwenServe()` × 2 in the same process) would race + // on `process.env` — `defaultSpawnChannelFactory` snapshots + // `process.env` AT SPAWN TIME, so the later daemon's env value + // would silently win for the earlier daemon's subsequent ACP + // child spawns. With per-handle overrides closed over inside + // each bridge, each daemon's children inherit ONLY that + // daemon's intended budget config, regardless of what other + // daemons in the same process are doing. + // + // Also: `runQwenServe` is exported and other validations + // (`requireAuth` no-token, `maxConnections` NaN, `--workspace` + // checks) live here, so embedded callers expect boot-time + // rejection of invalid inputs. The yargs CLI handler duplicates + // these for fast-fail UX, but `runQwenServe` is the source of + // truth. + if (opts.mcpClientBudget !== undefined) { + if ( + !Number.isFinite(opts.mcpClientBudget) || + !Number.isInteger(opts.mcpClientBudget) || + opts.mcpClientBudget <= 0 + ) { + throw new TypeError( + `Invalid mcpClientBudget: ${opts.mcpClientBudget}. Must be a positive integer.`, + ); + } + } + if (opts.mcpBudgetMode === 'enforce' && opts.mcpClientBudget === undefined) { + throw new Error( + 'mcpBudgetMode="enforce" requires a positive mcpClientBudget. ' + + 'Pass mcpClientBudget=N, or set mcpBudgetMode to "warn" or "off".', + ); + } + // Per-handle env overrides: `undefined` value means "scrub this + // var from the child env" — important when a different daemon + // in the same process set the var globally previously. Always + // set both keys explicitly (to value or `undefined`) so each + // child's MCP budget env is fully determined by this handle's + // options, with no inheritance from process.env's current state. + const childEnvOverrides: Record = { + QWEN_SERVE_MCP_CLIENT_BUDGET: + opts.mcpClientBudget !== undefined + ? String(opts.mcpClientBudget) + : undefined, + QWEN_SERVE_MCP_BUDGET_MODE: opts.mcpBudgetMode, + }; + const bridge = deps.bridge ?? createHttpAcpBridge({ @@ -185,6 +239,7 @@ export async function runQwenServe( ? { eventRingSize: opts.eventRingSize } : {}), boundWorkspace, + childEnvOverrides, }); let actualPort = opts.port; // Pass the already-canonical `boundWorkspace` into `createServeApp` diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 83130d64c78..0c7daa31690 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -102,6 +102,11 @@ const EXPECTED_STAGE1_FEATURES = [ 'session_supported_commands', 'session_close', 'session_metadata', + // Issue #4175 PR 14. Always-on. Daemon supports the MCP client + // guardrail surface (`--mcp-client-budget`, `clientCount` / + // `budgets[]` on `/workspace/mcp`, `disabledReason: 'budget'` on + // refused per-server cells). + 'mcp_guardrails', ] as const; // Issue #4175 PR 15. `require_auth` is registered but conditionally @@ -648,6 +653,19 @@ describe('createServeApp', () => { ).toEqual(EXPECTED_REGISTERED_FEATURES.map(() => 'v1')); }); + it('exposes `modes` metadata on mcp_guardrails (#4175 PR 14)', () => { + // `modes` is currently registry-only documentation (no wire + // surface yet) — a client wanting to feature-detect `enforce` + // semantics reads `caps.features.includes('mcp_guardrails')`, + // not a separate `featureModes` field. The descriptor still + // carries `modes` so future PRs that DO expose it on the wire + // don't have to chase down every entry to backfill metadata. + expect(SERVE_CAPABILITY_REGISTRY['mcp_guardrails']).toEqual({ + since: 'v1', + modes: ['warn', 'enforce'], + }); + }); + it('returns protocol version metadata with a fresh supported array', () => { const versions = getServeProtocolVersions(); expect(versions).toEqual({ current: 'v1', supported: ['v1'] }); @@ -771,6 +789,86 @@ describe('createServeApp', () => { expect(bridge.workspaceMcpCalls).toBe(1); }); + it('round-trips PR 14 budget fields on /workspace/mcp', async () => { + // Issue #4175 PR 14. The route is a thin JSON forwarder, so the + // assertion is structural: the new fields (`clientCount`, + // `clientBudget`, `budgetMode`, `budgets[]`, per-server + // `disabledReason`) must survive verbatim. Catches future + // serialization regressions that drop unknown optional fields. + const payload: ServeWorkspaceMcpStatus = { + v: 1, + workspaceCwd: WS_BOUND, + initialized: true, + discoveryState: 'completed', + clientCount: 3, + clientBudget: 2, + budgetMode: 'enforce', + budgets: [ + { + kind: 'mcp_budget', + scope: 'session', + status: 'error', + errorKind: 'budget_exhausted', + hint: 'Raise --mcp-client-budget or remove servers.', + liveCount: 2, + budget: 2, + mode: 'enforce', + refusedCount: 1, + }, + ], + servers: [ + { + kind: 'mcp_server', + status: 'ok', + name: 'a', + mcpStatus: 'connected', + transport: 'stdio', + disabled: false, + }, + { + kind: 'mcp_server', + status: 'ok', + name: 'b', + mcpStatus: 'connected', + transport: 'stdio', + disabled: false, + }, + { + kind: 'mcp_server', + status: 'error', + errorKind: 'budget_exhausted', + hint: 'Raise --mcp-client-budget or remove servers from mcpServers config.', + name: 'c', + mcpStatus: 'disconnected', + transport: 'stdio', + disabled: false, + disabledReason: 'budget', + }, + ], + }; + const bridge = fakeBridge({ workspaceMcpImpl: async () => payload }); + const app = createServeApp( + { ...baseOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + const res = await request(app) + .get('/workspace/mcp') + .set('Host', `127.0.0.1:${baseOpts.port}`); + + expect(res.status).toBe(200); + expect(res.body).toEqual(payload); + expect(res.body.budgets).toHaveLength(1); + expect(res.body.budgets[0]).toMatchObject({ + kind: 'mcp_budget', + scope: 'session', + status: 'error', + errorKind: 'budget_exhausted', + refusedCount: 1, + }); + expect(res.body.servers[2].disabledReason).toBe('budget'); + }); + it('returns workspace skills and providers status from the bridge', async () => { const skills: ServeWorkspaceSkillsStatus = { v: 1, @@ -2681,6 +2779,91 @@ describe('runQwenServe', () => { ).rejects.toThrow(/--require-auth/); }); + // PR 14 fix (review #4247): runQwenServe is the documented embedded + // entry point, so budget validation must live here, not just in the + // yargs CLI handler. Embedded callers (other tools wrapping the + // daemon, deps.bridge test injection) silently produced an uncapped + // child pre-fix despite requesting enforce. + it('rejects non-positive mcpClientBudget (#4175 PR 14)', async () => { + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + mcpClientBudget: 0, + }), + ).rejects.toThrow(/mcpClientBudget/); + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + mcpClientBudget: -5, + }), + ).rejects.toThrow(/mcpClientBudget/); + }); + + it('rejects mcpBudgetMode=enforce without a budget (#4175 PR 14)', async () => { + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + mcpBudgetMode: 'enforce', + }), + ).rejects.toThrow(/enforce.*requires.*mcpClientBudget/); + }); + + // Round 6 (wenshao R5 line 216): replaced the R3 `process.env` + // mutation tests. `runQwenServe` now passes per-handle env + // overrides via `BridgeOptions.childEnvOverrides`, NOT by mutating + // global `process.env` — so concurrent embedded daemons don't + // cross-contaminate each other's MCP budget env. The two tests + // below assert (a) runQwenServe doesn't touch process.env and + // (b) a pre-existing process.env value survives runQwenServe + // calls unrelated to MCP overrides (proving runQwenServe is no + // longer the source of env mutation). + it('does not mutate process.env when caller provides mcp budget options (#4247 R6 line 216)', async () => { + // Sanity-check: no MCP env vars set before. + delete process.env['QWEN_SERVE_MCP_CLIENT_BUDGET']; + delete process.env['QWEN_SERVE_MCP_BUDGET_MODE']; + handle = await runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + mcpClientBudget: 10, + mcpBudgetMode: 'warn', + }); + // Pre-R6 this leaked into global process.env. Post-R6 the values + // travel via `BridgeOptions.childEnvOverrides` closure → only + // the spawned ACP child sees them. + expect(process.env['QWEN_SERVE_MCP_CLIENT_BUDGET']).toBeUndefined(); + expect(process.env['QWEN_SERVE_MCP_BUDGET_MODE']).toBeUndefined(); + }); + + it('preserves pre-existing process.env values (no longer wipes globals on omit) (#4247 R6 line 216)', async () => { + // Pre-R6 the "scrub on omit" code path delete'd these from + // process.env. Post-R6 runQwenServe doesn't touch process.env + // at all; the override mechanism handles "scrub" at the + // per-handle level inside the bridge's spawn factory. So if an + // operator had QWEN_SERVE_MCP_CLIENT_BUDGET exported in their + // shell BEFORE starting the daemon, it stays in their process + // env (and gets ignored by this daemon's child, which receives + // `undefined` via overrides to scrub it on spawn). + process.env['QWEN_SERVE_MCP_CLIENT_BUDGET'] = '99'; + try { + handle = await runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + // No mcpClientBudget — override will scrub the var on spawn. + }); + expect(process.env['QWEN_SERVE_MCP_CLIENT_BUDGET']).toBe('99'); + } finally { + delete process.env['QWEN_SERVE_MCP_CLIENT_BUDGET']; + } + }); + it('starts with --require-auth + token on loopback', async () => { handle = await runQwenServe({ hostname: '127.0.0.1', diff --git a/packages/cli/src/serve/status.test.ts b/packages/cli/src/serve/status.test.ts index 1956f46ab64..7b776c1c387 100644 --- a/packages/cli/src/serve/status.test.ts +++ b/packages/cli/src/serve/status.test.ts @@ -13,7 +13,12 @@ import { } from './status.js'; describe('SERVE_ERROR_KINDS', () => { - it('exposes the seven roadmap-defined error kinds in stable order', () => { + it('exposes the eight roadmap-defined error kinds in stable order', () => { + // PR 13 introduced the closed taxonomy with seven preflight/env + // kinds; PR 14 added `'budget_exhausted'` for MCP guardrail + // refusals (see #4175 PR 14). Future additions append to this + // list — the order is part of the contract so SDK consumers + // can pattern-match without per-kind lookups. expect(SERVE_ERROR_KINDS).toEqual([ 'missing_binary', 'blocked_egress', @@ -22,6 +27,7 @@ describe('SERVE_ERROR_KINDS', () => { 'protocol_error', 'missing_file', 'parse_error', + 'budget_exhausted', ]); }); }); diff --git a/packages/cli/src/serve/status.ts b/packages/cli/src/serve/status.ts index e37db2c6791..93d531e7b60 100644 --- a/packages/cli/src/serve/status.ts +++ b/packages/cli/src/serve/status.ts @@ -23,6 +23,10 @@ export const SERVE_ERROR_KINDS = [ 'protocol_error', 'missing_file', 'parse_error', + // Issue #4175 PR 14: budget refusal under `--mcp-budget-mode=enforce`. + // Surfaced on per-server `mcp_server` cells (refused at discovery) + // and on the workspace-level `mcp_budget` cell (any refusal this pass). + 'budget_exhausted', ] as const; export type ServeErrorKind = (typeof SERVE_ERROR_KINDS)[number]; @@ -94,6 +98,60 @@ export interface ServeWorkspaceMcpServerStatus extends ServeStatusCell { disabled: boolean; description?: string; extensionName?: string; + /** + * Why this server is not live, when known. Distinguishes + * operator-disabled (`disabled: true` from `disabledMcpServers` + * config) from PR 14 budget-refused (`status: 'error', errorKind: + * 'budget_exhausted'`). Operators dashboarding the workspace + * shouldn't have to cross-reference the `errors[]` or `budgets[]` + * arrays to render a per-server row correctly. + */ + disabledReason?: 'config' | 'budget'; +} + +/** Budget mode for the MCP client guardrails (issue #4175 PR 14). */ +export type ServeMcpBudgetMode = 'enforce' | 'warn' | 'off'; + +/** + * Workspace-level budget status cell. Surfaced as one entry in + * `ServeWorkspaceMcpStatus.budgets[]`. The list shape (vs a single + * `budget?` field) is forward-compat for Wave 5 PR 23, which will + * add a `scope: 'pool'` cell alongside without a schema bump. + * + * Consumers MUST tolerate additional entries with unrecognized + * `scope` values — drop them rather than failing. + */ +export interface ServeMcpBudgetStatusCell extends ServeStatusCell { + kind: 'mcp_budget'; + /** + * Identifies which accounting scope this cell describes. + * + * **PR 14 v1 emits `'session'`** because each ACP session creates + * its own `Config`/`McpClientManager` via `acpAgent.newSessionConfig()` + * — so the budget caps live MCP clients **per session**, not + * per-workspace. The snapshot reflects the bootstrap session's + * view; concurrent sessions each enforce their own copy of the + * cap independently. See `qwen-serve-protocol.md` "PR 14 v1 + * scope: per-session" for the operator-facing rationale. + * + * Future PRs: + * - Wave 5 PR 23 (shared MCP pool) introduces a workspace-scoped + * manager and will emit `'workspace'` (or `'pool'`) cells. + * - The `string & {}` widening keeps IDE autocomplete + literal + * narrowing for known scopes while allowing unknown scopes + * through without a compile-time break — the protocol contract + * is "consumers MUST tolerate additional scope values, drop + * don't fail." + */ + scope: 'session' | 'workspace' | (string & {}); + /** Live (CONNECTED) MCP client count at snapshot time. */ + liveCount: number; + /** Configured cap (positive integer). Absent only when mode is `off`. */ + budget?: number; + /** Active enforcement mode. `off` mode produces no cell — `budgets: []`. */ + mode: ServeMcpBudgetMode; + /** Servers refused during the most recent discovery pass. */ + refusedCount: number; } export interface ServeWorkspaceMcpStatus { @@ -103,6 +161,18 @@ export interface ServeWorkspaceMcpStatus { discoveryState?: ServeMcpDiscoveryState; servers: ServeWorkspaceMcpServerStatus[]; errors?: ServeStatusCell[]; + /** PR 14: live MCP client count (sum across all transports). */ + clientCount?: number; + /** PR 14: configured budget. Absent when no cap was set. */ + clientBudget?: number; + /** PR 14: active enforcement mode. Absent on pre-PR-14 daemons. */ + budgetMode?: ServeMcpBudgetMode; + /** + * PR 14: workspace-level status cells for budget enforcement. Always + * an array (possibly empty) on post-PR-14 daemons; absent on older + * daemons. PR 23 will add a `scope: 'pool'` cell alongside. + */ + budgets?: ServeMcpBudgetStatusCell[]; } export type ServeSkillLevel = 'project' | 'user' | 'extension' | 'bundled'; @@ -179,12 +249,22 @@ export interface ServeSessionSupportedCommandsStatus { export function createIdleWorkspaceMcpStatus( workspaceCwd: string, ): ServeWorkspaceMcpStatus { + // PR 14: an idle workspace has zero live clients and no enforcement + // pressure. `budgetMode` is `'off'` (regardless of how the operator + // configured it) because no discovery has run, so no reservation + // could have happened. `budgets` is an empty array, not absent — + // the daemon DOES support the surface, the snapshot just has + // nothing to report yet. Older daemons omitting the array entirely + // are still spec-compliant; consumers default-coalesce to `[]`. return { v: STATUS_SCHEMA_VERSION, workspaceCwd, initialized: false, discoveryState: 'not_started', servers: [], + clientCount: 0, + budgetMode: 'off', + budgets: [], }; } diff --git a/packages/cli/src/serve/types.ts b/packages/cli/src/serve/types.ts index 6a8ac986910..d24d00d5d83 100644 --- a/packages/cli/src/serve/types.ts +++ b/packages/cli/src/serve/types.ts @@ -108,6 +108,30 @@ export interface ServeOptions { * irrespective of this flag. */ requireAuth?: boolean; + /** + * Issue #4175 PR 14. Cap on live MCP clients spawned inside the + * ACP child for the bound workspace. When set, the daemon + * forwards `QWEN_SERVE_MCP_CLIENT_BUDGET` to the child's env so + * core's `McpClientManager` picks it up. Combined with + * `mcpBudgetMode`: + * - `warn` (default when budget set): no refusal, snapshot + * surfaces `status: 'warning'` at >=75% of budget. + * - `enforce`: connects past the cap are refused, per-server + * cell shows `disabledReason: 'budget'`, deterministic by + * `Object.entries(mcpServers)` declaration order. + * - `off`: no accounting-driven enforcement (the implicit + * default when no budget is configured). + * + * Positive integer required; non-positive / NaN values throw at + * boot. + */ + mcpClientBudget?: number; + /** + * Issue #4175 PR 14. Enforcement mode for `mcpClientBudget`. + * Boot rejects `enforce` without a budget; otherwise resolves to + * `warn` when budget set / `off` when budget unset. + */ + mcpBudgetMode?: 'enforce' | 'warn' | 'off'; } /** diff --git a/packages/core/src/tools/mcp-client-manager.test.ts b/packages/core/src/tools/mcp-client-manager.test.ts index 1e2da7ddea3..9c728f91f57 100644 --- a/packages/core/src/tools/mcp-client-manager.test.ts +++ b/packages/core/src/tools/mcp-client-manager.test.ts @@ -929,3 +929,1102 @@ describe('McpClientManager', () => { expect(observedStatesAtEmit[0]).toBe(MCPDiscoveryState.IN_PROGRESS); }); }); + +// Issue #4175 PR 14: MCP client guardrails (counter + slot reservation + +// budget enforcement). Kept in its own describe so the existing test +// suite stays untouched and a future revert of PR 14 drops a single +// contiguous block. +describe('McpClientManager — PR 14 guardrails', () => { + afterEach(() => { + vi.restoreAllMocks(); + delete process.env['QWEN_SERVE_MCP_CLIENT_BUDGET']; + delete process.env['QWEN_SERVE_MCP_BUDGET_MODE']; + }); + + /** + * Mock factory: returns a fresh stub McpClient whose `getStatus()` + * returns CONNECTED after `connect()` resolves. Mirrors the + * `discoverAllMcpTools` happy path — counter sees the client as + * live only when `getStatus === CONNECTED`, so without flipping + * the mock status the accounting would always read zero. + */ + function makeConnectedMcpClientMock() { + // Real McpClient.getStatus is sync — start CONNECTED so accounting + // sees it as live immediately after construction. `connect()` is a + // no-op (we don't simulate handshake state machinery in unit + // tests; the accounting cares only about the final status). + const state = { status: undefined as unknown }; + return { + connect: vi.fn().mockImplementation(async () => { + const { MCPServerStatus } = await import('./mcp-client.js'); + state.status = MCPServerStatus.CONNECTED; + }), + discover: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn().mockResolvedValue(undefined), + getStatus: vi.fn(() => state.status), + readResource: vi.fn().mockResolvedValue({ contents: [] }), + }; + } + + function configWithServers( + servers: Record, + overrides: Partial = {}, + ): Config { + return { + isTrustedFolder: () => true, + getMcpServers: () => servers, + getMcpServerCommand: () => undefined, + getPromptRegistry: () => ({}) as PromptRegistry, + getWorkspaceContext: () => ({}) as WorkspaceContext, + getDebugMode: () => false, + isMcpServerDisabled: () => false, + ...overrides, + } as unknown as Config; + } + + it('getMcpClientAccounting returns zero on an empty manager', async () => { + const { McpClientManager: MgrCtor } = await import( + './mcp-client-manager.js' + ); + const config = configWithServers({}); + const manager = new MgrCtor(config, {} as ToolRegistry); + const accounting = manager.getMcpClientAccounting(); + expect(accounting.total).toBe(0); + expect(accounting.subprocessCount).toBe(0); + expect(accounting.reservedSlots).toEqual([]); + expect(accounting.refusedServerNames).toEqual([]); + expect(accounting.byTransport).toEqual({ + stdio: 0, + sse: 0, + http: 0, + websocket: 0, + sdk: 0, + unknown: 0, + }); + }); + + it('mcpTransportOf maps each transport family correctly', async () => { + const { mcpTransportOf } = await import('./mcp-client-manager.js'); + const cfg = (overrides: Record) => + overrides as unknown as import('../config/config.js').MCPServerConfig; + expect(mcpTransportOf(cfg({ command: 'node' }))).toBe('stdio'); + expect(mcpTransportOf(cfg({ httpUrl: 'http://x' }))).toBe('http'); + expect(mcpTransportOf(cfg({ url: 'http://x' }))).toBe('sse'); + expect(mcpTransportOf(cfg({ tcp: 'ws://x' }))).toBe('websocket'); + expect(mcpTransportOf(cfg({}))).toBe('unknown'); + // SDK detection short-circuits: even with a placeholder command, + // an SDK-marked server reports `sdk` (not `stdio`). + expect(mcpTransportOf(cfg({ type: 'sdk', command: 'node' }))).toBe('sdk'); + }); + + it('enforce mode refuses connects past the budget', async () => { + const created: Array> = []; + vi.mocked(McpClient).mockImplementation(() => { + const m = makeConnectedMcpClientMock(); + created.push(m); + return m as unknown as McpClient; + }); + // 4 stdio servers, budget 2. enforce mode refuses 2. + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + c: { command: 'node' }, + d: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 2, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(created).toHaveLength(2); // only 2 McpClient instances created + const accounting = manager.getMcpClientAccounting(); + expect(accounting.total).toBe(2); + expect(accounting.byTransport.stdio).toBe(2); + expect(accounting.subprocessCount).toBe(2); + expect(accounting.reservedSlots.sort()).toEqual(['a', 'b']); + expect(accounting.refusedServerNames.sort()).toEqual(['c', 'd']); + }); + + it('warn mode never refuses but tracks oversized reservations', async () => { + const created: Array> = []; + vi.mocked(McpClient).mockImplementation(() => { + const m = makeConnectedMcpClientMock(); + created.push(m); + return m as unknown as McpClient; + }); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + c: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 2, budgetMode: 'warn' }, + ); + await manager.discoverAllMcpTools(config); + // warn mode: all 3 connect; reservedSlots grows past budget; no refusals. + expect(created).toHaveLength(3); + const accounting = manager.getMcpClientAccounting(); + expect(accounting.total).toBe(3); + expect(accounting.reservedSlots.sort()).toEqual(['a', 'b', 'c']); + expect(accounting.refusedServerNames).toEqual([]); + }); + + it('off mode does not reserve any slot', async () => { + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { budgetMode: 'off' }, + ); + await manager.discoverAllMcpTools(config); + const accounting = manager.getMcpClientAccounting(); + expect(accounting.total).toBe(2); + // `off` skips reservation altogether — operators see live count via + // `total`, but reservedSlots stays empty. + expect(accounting.reservedSlots).toEqual([]); + expect(accounting.refusedServerNames).toEqual([]); + }); + + it('refusal is deterministic by config-declaration order', async () => { + const created: string[] = []; + vi.mocked(McpClient).mockImplementation((name: string) => { + created.push(name); + return makeConnectedMcpClientMock() as unknown as McpClient; + }); + // Insertion order: zulu, alpha, mike. Budget 2 → zulu+alpha survive. + const config = configWithServers({ + zulu: { command: 'node' }, + alpha: { command: 'node' }, + mike: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 2, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(created).toEqual(['zulu', 'alpha']); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([ + 'mike', + ]); + }); + + it('discoverAllMcpTools resets lastRefusedServerNames each pass', async () => { + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + + // Second pass: stop()→clear→re-run. The reset happens at the start + // of discoverAllMcpTools (see also stop() clearing reservedSlots). + await manager.discoverAllMcpTools(config); + // Same outcome (still budget 1, still 2 servers), but the array + // is fresh — not appended to. + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + }); + + it('readResource throws BudgetExhaustedError in enforce mode when full', async () => { + const { BudgetExhaustedError } = await import('./mcp-client-manager.js'); + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + // `a` was reserved; `b` was refused. A `readResource('b', ...)` would + // lazy-spawn — must throw rather than silently exceed the cap. + await expect(manager.readResource('b', 'file:///x')).rejects.toBeInstanceOf( + BudgetExhaustedError, + ); + }); + + it('disconnectServer releases the slot for re-use', async () => { + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual(['a']); + await manager.disconnectServer('a'); + // Slot released — accounting shows the configured set shrank. + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + }); + + it('env var fallback resolves budget + mode when constructor omits opts', async () => { + process.env['QWEN_SERVE_MCP_CLIENT_BUDGET'] = '7'; + process.env['QWEN_SERVE_MCP_BUDGET_MODE'] = 'enforce'; + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + expect(manager.getMcpClientBudget()).toBe(7); + expect(manager.getMcpBudgetMode()).toBe('enforce'); + }); + + it('env var fallback defaults mode to warn when only budget is set', async () => { + process.env['QWEN_SERVE_MCP_CLIENT_BUDGET'] = '5'; + // No mode env var. Resolved mode is `warn` (the safe default). + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + expect(manager.getMcpClientBudget()).toBe(5); + expect(manager.getMcpBudgetMode()).toBe('warn'); + }); + + it('env var fallback rejects non-positive budgets silently', async () => { + process.env['QWEN_SERVE_MCP_CLIENT_BUDGET'] = '-3'; + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + // Invalid values fall through to `undefined` budget + `off` mode — + // no enforcement, no boot-time crash. Validation lives in the CLI + // flag handler (`packages/cli/src/commands/serve.ts`). + expect(manager.getMcpClientBudget()).toBeUndefined(); + expect(manager.getMcpBudgetMode()).toBe('off'); + }); + + it('disabled servers do not consume a budget slot', async () => { + const created: string[] = []; + vi.mocked(McpClient).mockImplementation((name: string) => { + created.push(name); + return makeConnectedMcpClientMock() as unknown as McpClient; + }); + // `b` is disabled — must not even attempt to reserve. With budget=2, + // `a` and `c` should both succeed (b is invisible to the gate, so it + // doesn't consume a slot; the cap is enough for the remaining two). + const config = configWithServers( + { + a: { command: 'node' }, + b: { command: 'node' }, + c: { command: 'node' }, + }, + { + isMcpServerDisabled: ((name: string) => + name === 'b') as Config['isMcpServerDisabled'], + }, + ); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 2, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(created.sort()).toEqual(['a', 'c']); + expect(manager.getMcpClientAccounting().reservedSlots.sort()).toEqual([ + 'a', + 'c', + ]); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([]); + }); + + // PR 14 fix (review #4247): regression tests for the four bypass / + // ordering / staleness bugs the Codex + Copilot reviews caught. + it('single-server rediscovery respects the budget gate (review #1)', async () => { + const created: string[] = []; + vi.mocked(McpClient).mockImplementation((name: string) => { + created.push(name); + return makeConnectedMcpClientMock() as unknown as McpClient; + }); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + // `b` was refused at startup. A manual `/mcp reconnect b` (which goes + // through `discoverMcpToolsForServer` → `...Internal`) would have + // pre-fix bypassed the gate and exceeded the cap. Now it must stay + // refused. + expect(created).toEqual(['a']); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + await manager.discoverMcpToolsForServer('b', config); + expect(created).toEqual(['a']); // no new McpClient created + expect(manager.getMcpClientAccounting().reservedSlots).toEqual(['a']); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + }); + + it('disconnectServer-then-disable drops refusal tag (review #4)', async () => { + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + // Operator action: explicit disconnect of `b` should drop it from + // the refusal log so a snapshot doesn't keep tagging the now- + // operator-disabled server with `budget_exhausted`. + await manager.disconnectServer('b'); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([]); + }); + + it('incremental discovery frees removed slots BEFORE reserving new ones (review #5)', async () => { + let inflight = 0; + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + inflight = 0; + const mcpServers: Record = { + a: { command: 'node' }, + b: { command: 'node' }, + }; + const config = { + isTrustedFolder: () => true, + getMcpServers: () => mcpServers, + getMcpServerCommand: () => undefined, + getPromptRegistry: () => ({}) as PromptRegistry, + getWorkspaceContext: () => ({}) as WorkspaceContext, + getDebugMode: () => false, + isMcpServerDisabled: () => false, + } as unknown as Config; + const manager = new McpClientManager( + config, + { + removeMcpToolsByServer: () => undefined, + } as unknown as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 2, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpToolsIncremental(config); + expect(manager.getMcpClientAccounting().reservedSlots.sort()).toEqual([ + 'a', + 'b', + ]); + + // Swap b → c (still budget=2). Pre-fix order: `c` refused because + // `b`'s slot was only freed after the new-server loop. Post-fix: + // `b` removed first → reservedSlots={a} → `c` reserved. + delete mcpServers['b']; + mcpServers['c'] = { command: 'node' }; + await manager.discoverAllMcpToolsIncremental(config); + expect(manager.getMcpClientAccounting().reservedSlots.sort()).toEqual([ + 'a', + 'c', + ]); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([]); + void inflight; + }); + + it('buildBudgetCells deferred to acpAgent — manager off-mode returns no budget bookkeeping (review #2)', async () => { + // Sibling check: when `mode === 'off'` the manager doesn't reserve + // anything and the snapshot has empty `reservedSlots` + zero + // `refusedServerNames`. The empty-`budgets[]` assertion lives in + // the serve route test (`server.test.ts`) because the cell is + // built by `acpAgent.buildBudgetCells`. This test just pins the + // manager-side invariant: off-mode is pure observability. + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { budgetMode: 'off' }, + ); + await manager.discoverAllMcpTools(config); + const accounting = manager.getMcpClientAccounting(); + expect(accounting.total).toBe(2); + expect(accounting.reservedSlots).toEqual([]); + expect(accounting.refusedServerNames).toEqual([]); + }); + + // Round 2 review fixes (PR #4247 wenshao Critical 2, Critical 3, Suggestion 4). + it('connect() failure releases the reserved slot in discoverAllMcpTools (wenshao C2)', async () => { + // Failing client: getStatus stays DISCONNECTED; connect() throws. + // Pre-fix the slot stayed reserved → permanent leak under enforce + // → second server couldn't claim a freed slot until full restart. + let firstCall = true; + vi.mocked(McpClient).mockImplementation(() => { + if (firstCall) { + firstCall = false; + return { + connect: vi.fn().mockRejectedValue(new Error('boom')), + discover: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn().mockResolvedValue(undefined), + getStatus: vi.fn(), + } as unknown as McpClient; + } + return makeConnectedMcpClientMock() as unknown as McpClient; + }); + const config = configWithServers({ + a: { command: 'node' }, // will fail + b: { command: 'node' }, // would be refused pre-fix + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + // `a` failed → slot freed → `b` ought to fit (budget=1, current=0 + // after `a` released). But discoverAllMcpTools walks all servers + // concurrently — `b` may have been refused at the time of its + // synchronous reserve check (before `a` released). What we MUST + // assert is the post-conditions: `a` released its slot, `a` not + // in clients map. `b` may be either reserved or refused depending + // on the schedule, but the slot leak itself is gone. + const accounting = manager.getMcpClientAccounting(); + expect(accounting.reservedSlots).not.toContain('a'); + // No leaked client entry either: + expect( + (manager as unknown as { clients: Map }).clients.has( + 'a', + ), + ).toBe(false); + }); + + it('connect() failure in readResource releases the slot AND re-throws (wenshao C3)', async () => { + let getResourceCalled = false; + vi.mocked(McpClient).mockImplementation( + () => + ({ + // Stays disconnected → readResource code path forces a + // `client.connect()` before `client.readResource(...)`. + connect: vi.fn().mockRejectedValue(new Error('lazy connect boom')), + discover: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn().mockResolvedValue(undefined), + getStatus: vi.fn(), + readResource: vi.fn().mockImplementation(() => { + getResourceCalled = true; + return Promise.resolve({}); + }), + }) as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + // No discovery yet → `a` not in clients → lazy spawn path. + await expect(manager.readResource('a', 'file:///x')).rejects.toThrow( + 'lazy connect boom', + ); + // Slot must NOT leak — pre-fix one failed readResource permanently + // burned a budget slot. + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + expect( + (manager as unknown as { clients: Map }).clients.has( + 'a', + ), + ).toBe(false); + // And the readResource ext-method was never reached (we threw at connect). + expect(getResourceCalled).toBe(false); + }); + + it('readBudgetFromEnv downgrades enforce-without-budget to off (wenshao S4)', async () => { + process.env['QWEN_SERVE_MCP_BUDGET_MODE'] = 'enforce'; + // No QWEN_SERVE_MCP_CLIENT_BUDGET — silently fail-open pre-fix: + // `tryReserveSlot` returns 'reserved' when `clientBudget === undefined`, + // so an "enforce" daemon would let unlimited servers through. + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + expect(manager.getMcpClientBudget()).toBeUndefined(); + // Downgraded — not 'enforce' — because enforce requires a budget. + expect(manager.getMcpBudgetMode()).toBe('off'); + }); + + // Round 3 review fixes (PR #4247 wenshao second pass). + it('readResource rejects disabled servers before checking budget (wenshao R3 #5)', async () => { + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + // Pre-fix the lazy spawn path bypassed `isMcpServerDisabled`, + // so a disabled server could be resurrected by a resource read. + const config = configWithServers( + { + a: { command: 'node' }, + }, + { + isMcpServerDisabled: ((name: string) => + name === 'a') as Config['isMcpServerDisabled'], + }, + ); + const manager = new McpClientManager(config, {} as ToolRegistry); + await expect(manager.readResource('a', 'file:///x')).rejects.toThrow( + /'a' is disabled/, + ); + }); + + it('readResource disabled gate fires BEFORE budget gate (wenshao R3 #5 precedence)', async () => { + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + // Set up a budget-exhausted scenario + disable the target. The + // disabled error must win over the budget error (matches the + // per-server cell precedence: disabled wins). + const config = configWithServers( + { + a: { command: 'node' }, + b: { command: 'node' }, + }, + { + isMcpServerDisabled: ((name: string) => + name === 'b') as Config['isMcpServerDisabled'], + }, + ); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + // Even though `b` would be budget-refused if not disabled, the + // disabled gate must trip first. + await expect(manager.readResource('b', 'file:///x')).rejects.toThrow( + /'b' is disabled/, + ); + }); + + it('exports MCP_BUDGET_WARN_FRACTION constant (wenshao R3 #7)', async () => { + const { MCP_BUDGET_WARN_FRACTION } = await import( + './mcp-client-manager.js' + ); + // Pinned to 0.75 to match PR 10's slow_client_warning hysteresis + // primer (eventBus.ts WARN_THRESHOLD_RATIO). PR 14b will introduce + // the matching reset fraction (0.375) to complete the dual-threshold + // pair; this test is a tripwire against accidental fraction drift. + expect(MCP_BUDGET_WARN_FRACTION).toBe(0.75); + }); + + // Round 4 review fixes (PR #4247 wenshao R3-R4 zombie leak in internal path). + it('discoverMcpToolsForServer fresh-reserve connect-failure releases slot (wenshao R4 C2)', async () => { + vi.mocked(McpClient).mockImplementation( + () => + ({ + connect: vi.fn().mockRejectedValue(new Error('boom')), + discover: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn().mockResolvedValue(undefined), + getStatus: vi.fn(), + }) as unknown as McpClient, + ); + const config = configWithServers({ x: { command: 'node' } }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + // Server `x` not previously reserved; this call freshly reserves + // then connect() throws. Pre-fix the slot leaked permanently + // under enforce mode, blocking any later server in `clients.size=1`. + await manager.discoverMcpToolsForServer('x', config); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + expect( + (manager as unknown as { clients: Map }).clients.has( + 'x', + ), + ).toBe(false); + }); + + // R8 #4 (line 1221): the `freshReservations` Set distinguishes + // fresh-reservation timeouts (release) from `'already_held'` + // reconnect timeouts (keep slot). Verified by code inspection + + // the R5 release-on-fresh test below; a dedicated already_held + // timeout test requires either driving the health-monitor flow + // end-to-end (which needs autoReconnect timer interleaving with + // fake timers — interferes with the sibling R5 test in the same + // file) or piercing the private `runWithDiscoveryTimeout` + // helper. The invariant is small enough that the fresh-release + // test below is sufficient regression coverage; an integration + // test in a separate file can add the already_held variant + // without the timer interleave problem. + it('runWithDiscoveryTimeout timeout handler releases the budget slot (wenshao R5 line 956)', async () => { + vi.useFakeTimers(); + // McpClient.connect never resolves → timeout fires. + vi.mocked(McpClient).mockImplementation( + () => + ({ + connect: vi.fn(() => new Promise(() => {})), + discover: vi.fn(), + disconnect: vi.fn().mockResolvedValue(undefined), + getStatus: vi.fn(), + }) as unknown as McpClient, + ); + const config = configWithServers({ a: { command: 'node' } }); + const manager = new McpClientManager( + config, + { removeMcpToolsByServer: () => undefined } as unknown as ToolRegistry, + undefined, + undefined, + { + autoReconnect: false, + checkIntervalMs: 100, + maxConsecutiveFailures: 1, + reconnectDelayMs: 100, + }, + { clientBudget: 2, budgetMode: 'enforce' }, + ); + const discoveryPromise = manager.discoverAllMcpToolsIncremental(config); + // Advance past the stdio default discovery timeout (30s). + await vi.advanceTimersByTimeAsync(31_000); + await discoveryPromise; + // Pre-fix the timeout cleaned up clients but not reservedSlots, + // permanently consuming a budget slot. + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + vi.useRealTimers(); + }); + + it('incremental discovery still refuses past the cap after R6 pre-reservation removal (wenshao R6 line 956)', async () => { + // Round 6 removed the duplicate pre-reservation in + // discoverAllMcpToolsIncremental — refusal now happens INSIDE + // discoverMcpToolsForServerInternal's tryReserveSlot. Verify + // the observable refusal behavior is unchanged from the outside. + const created: string[] = []; + vi.mocked(McpClient).mockImplementation((name: string) => { + created.push(name); + return makeConnectedMcpClientMock() as unknown as McpClient; + }); + const config = configWithServers({ + first: { command: 'node' }, + second: { command: 'node' }, + third: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + { removeMcpToolsByServer: () => undefined } as unknown as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 2, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpToolsIncremental(config); + // First two declared servers fit; third refused. Refusal-order + // determinism preserved (config-declaration order) — the inner + // tryReserveSlot is called in the same serversToUpdate iteration + // order as the outer walk produced. + expect(created).toEqual(['first', 'second']); + expect(manager.getMcpClientAccounting().reservedSlots.sort()).toEqual([ + 'first', + 'second', + ]); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([ + 'third', + ]); + }); + + it('readResource late re-reserve clears stale refused entry (wenshao R5 line 1268)', async () => { + // First: discoverAllMcpTools refuses `b` (budget=1, both a+b configured). + // Then: disconnect `a` freeing the slot; readResource('b') succeeds and + // must drop `b` from lastRefusedServerNames (pre-fix the snapshot kept + // reporting `b` as `disabledReason: 'budget'` even after it connected). + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + // Free a slot. + await manager.disconnectServer('a'); + // Lazy spawn b — should now succeed (slot available). + await manager.readResource('b', 'file:///x'); + // Stale refusal entry must be cleared. + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([]); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual(['b']); + }); + + it('discoverMcpToolsForServer clears stale refused entry on success (wenshao R7 #1 line 612)', async () => { + // Critical: a previously-refused server that connects successfully + // (e.g. via /mcp reconnect after another server frees a slot) + // would leave a stale entry in lastRefusedServerNames, so the + // snapshot reported `disabledReason: 'budget'` for a CONNECTED + // server until the next discovery pass cleared the per-pass log. + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + const config = configWithServers({ + a: { command: 'node' }, + b: { command: 'node' }, + }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual(['b']); + // Free a slot. + await manager.disconnectServer('a'); + // Manual /mcp reconnect path exercises discoverMcpToolsForServer. + await manager.discoverMcpToolsForServer('b', config); + // The successful late connect must clear the stale refusal entry. + expect(manager.getMcpClientAccounting().refusedServerNames).toEqual([]); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual(['b']); + }); + + it('discoverMcpToolsForServerInternal rejects disabled servers (wenshao R7 #2 line 528)', async () => { + // Reachable from /mcp reconnect, OAuth re-discovery, and health + // monitor reconnect. Pre-fix none of these paths checked the + // disabled flag, so a disabled server could be resurrected. + let createdCount = 0; + vi.mocked(McpClient).mockImplementation(() => { + createdCount += 1; + return makeConnectedMcpClientMock() as unknown as McpClient; + }); + const config = configWithServers( + { a: { command: 'node' } }, + { + isMcpServerDisabled: ((name: string) => + name === 'a') as Config['isMcpServerDisabled'], + }, + ); + const manager = new McpClientManager(config, {} as ToolRegistry); + await manager.discoverMcpToolsForServer('a', config); + expect(createdCount).toBe(0); + }); + + it('discoverMcpToolsForServerInternal disconnects on discover() failure (wenshao R7 #3 line 634)', async () => { + // Pre-fix: `connect()` succeeds + `discover()` throws → catch + // deletes the client from the map without calling + // `disconnect()`, leaking the stdio child. + let disconnectCalls = 0; + vi.mocked(McpClient).mockImplementation( + () => + ({ + connect: vi.fn().mockResolvedValue(undefined), + discover: vi.fn().mockRejectedValue(new Error('discover failed')), + disconnect: vi.fn().mockImplementation(() => { + disconnectCalls += 1; + return Promise.resolve(); + }), + getStatus: vi.fn(), + }) as unknown as McpClient, + ); + const config = configWithServers({ x: { command: 'node' } }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverMcpToolsForServer('x', config); + // Slot released on weReservedSlot+catch path AND the transport + // was closed before dropping the client reference. + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + expect(disconnectCalls).toBeGreaterThanOrEqual(1); + }); + + it('readBudgetFromEnv emits stderr warning on invalid budget value (wenshao R7 #6 line 191)', async () => { + const writeSpy = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + process.env['QWEN_SERVE_MCP_CLIENT_BUDGET'] = 'abc'; + try { + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + expect(manager.getMcpClientBudget()).toBeUndefined(); + // Operator-visible breadcrumb landed on stderr. + const calls = writeSpy.mock.calls.map((c) => String(c[0])); + expect( + calls.some( + (s) => + s.includes('ignoring invalid QWEN_SERVE_MCP_CLIENT_BUDGET') && + s.includes("'abc'"), + ), + ).toBe(true); + } finally { + writeSpy.mockRestore(); + } + }); + + it('readResource rejects existing-but-now-disabled servers (wenshao R7 #5 line 1342)', async () => { + // Pre-fix: a server connected pre-disable and then operator- + // disabled mid-session via settings reload would still serve + // resource reads via its existing CONNECTED client until the + // next incremental discovery pass called removeServer. + vi.mocked(McpClient).mockImplementation( + () => makeConnectedMcpClientMock() as unknown as McpClient, + ); + let disabled = false; + const config = configWithServers( + { a: { command: 'node' } }, + { + isMcpServerDisabled: ((name: string) => + name === 'a' && disabled) as Config['isMcpServerDisabled'], + }, + ); + const manager = new McpClientManager(config, {} as ToolRegistry); + // First connect while NOT disabled. + await manager.discoverAllMcpTools(config); + // Now operator disables 'a' mid-session. + disabled = true; + // readResource on the EXISTING (still CONNECTED) client must + // reject — pre-fix this would have proceeded to client.readResource. + await expect(manager.readResource('a', 'file:///x')).rejects.toThrow( + /'a' is disabled/, + ); + }); + + it('readResource lazy spawn disconnects on connect() failure (wenshao R9 #2 line 1534)', async () => { + // Mirror of the discovery-side R7 #3 / R8 #1 fixes, but for + // the readResource lazy-spawn path. Pre-fix: connect() + // partially established transport then threw → catch deleted + // client without disconnect() → stdio child / socket leaked. + let disconnectCalls = 0; + vi.mocked(McpClient).mockImplementation( + () => + ({ + connect: vi + .fn() + .mockRejectedValue(new Error('mid-handshake failure')), + discover: vi.fn(), + disconnect: vi.fn().mockImplementation(() => { + disconnectCalls += 1; + return Promise.resolve(); + }), + getStatus: vi.fn(), + readResource: vi.fn(), + }) as unknown as McpClient, + ); + const config = configWithServers({ x: { command: 'node' } }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await expect(manager.readResource('x', 'file:///a')).rejects.toThrow( + /mid-handshake failure/, + ); + expect(disconnectCalls).toBeGreaterThanOrEqual(1); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + }); + + it('readBudgetFromEnv emits stderr breadcrumb on enforce-no-budget downgrade (wenshao R9 #7)', async () => { + const writeSpy = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + process.env['QWEN_SERVE_MCP_BUDGET_MODE'] = 'enforce'; + // No budget → downgrade fires + try { + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + expect(manager.getMcpBudgetMode()).toBe('off'); + const calls = writeSpy.mock.calls.map((c) => String(c[0])); + expect( + calls.some( + (s) => + s.includes('QWEN_SERVE_MCP_BUDGET_MODE=enforce') && + s.includes('downgrading to off'), + ), + ).toBe(true); + } finally { + writeSpy.mockRestore(); + } + }); + + it('discoverAllMcpTools disconnects on discover() failure (wenshao R8 #1 line 532)', async () => { + // Bulk-path mirror of R7 #3 (per-server path). Pre-fix: + // connect() success + discover() throw → catch deleted client + // without disconnect() → stdio child / WebSocket / HTTP socket + // leaked for the rest of the daemon's lifetime (stop() can't + // see the entry it just removed from this.clients). + let disconnectCalls = 0; + vi.mocked(McpClient).mockImplementation( + () => + ({ + connect: vi.fn().mockResolvedValue(undefined), + discover: vi.fn().mockRejectedValue(new Error('discover failed')), + disconnect: vi.fn().mockImplementation(() => { + disconnectCalls += 1; + return Promise.resolve(); + }), + getStatus: vi.fn(), + }) as unknown as McpClient, + ); + const config = configWithServers({ a: { command: 'node' } }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + await manager.discoverAllMcpTools(config); + // Transport closed before client reference dropped + slot released. + expect(disconnectCalls).toBeGreaterThanOrEqual(1); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual([]); + }); + + it('readBudgetFromEnv downgrades warn-without-budget to off (wenshao R8 #2)', async () => { + process.env['QWEN_SERVE_MCP_BUDGET_MODE'] = 'warn'; + // No budget — pre-fix this passed through with mode='warn', + // reaching emitBudgetTelemetry with clientBudget=undefined. + const config = configWithServers({}); + const manager = new McpClientManager(config, {} as ToolRegistry); + expect(manager.getMcpClientBudget()).toBeUndefined(); + expect(manager.getMcpBudgetMode()).toBe('off'); + }); + + it('constructor downgrades enforce-without-budget when budgetConfig passed directly (wenshao R8 #5)', async () => { + // Direct-budgetConfig path is test-/embedded-only — production + // callers (CLI, runQwenServe, env-var fallback) all validate + // upfront. Defense-in-depth: constructor mirrors the env-var + // path's downgrade so a future caller that bypasses validation + // can't silently fail-open. + const config = configWithServers({}); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + // Invalid combination: enforce mode without a budget. + { budgetMode: 'enforce' }, + ); + // Downgraded to off so tryReserveSlot doesn't masquerade as enforce. + expect(manager.getMcpBudgetMode()).toBe('off'); + }); + + it('discoverMcpToolsForServer reconnect-attempt connect-failure KEEPS slot (wenshao R4 C2 already_held)', async () => { + // Distinguish from the previous test: same call signature, but + // here the slot is already-held (from a prior successful connect + // in discoverAllMcpTools). A failed reconnect must NOT release — + // the operator's stable server that just hiccupped should keep + // its capacity reservation for the health-monitor retry loop. + let connectThrows = false; + vi.mocked(McpClient).mockImplementation( + () => + ({ + connect: vi.fn().mockImplementation(async () => { + if (connectThrows) throw new Error('reconnect boom'); + }), + discover: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn().mockResolvedValue(undefined), + getStatus: vi.fn(() => + connectThrows + ? undefined + : ((vi.mocked as unknown as { val: unknown }).val = + 'CONNECTED' as unknown), + ), + }) as unknown as McpClient, + ); + const config = configWithServers({ a: { command: 'node' } }); + const manager = new McpClientManager( + config, + {} as ToolRegistry, + undefined, + undefined, + undefined, + { clientBudget: 1, budgetMode: 'enforce' }, + ); + // First pass: a connects successfully, slot reserved. + await manager.discoverAllMcpTools(config); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual(['a']); + // Now simulate health-monitor reconnect against a flaky server: + // discoverMcpToolsForServer goes through tryReserveSlot → + // 'already_held' (slot stays) → existing client.disconnect() + // (slot stays) → new client.connect() throws → fix says + // weReservedSlot=false here so slot NOT released. + connectThrows = true; + await manager.discoverMcpToolsForServer('a', config); + expect(manager.getMcpClientAccounting().reservedSlots).toEqual(['a']); + }); +}); diff --git a/packages/core/src/tools/mcp-client-manager.ts b/packages/core/src/tools/mcp-client-manager.ts index 54df8adc4f8..b6aefa4fb26 100644 --- a/packages/core/src/tools/mcp-client-manager.ts +++ b/packages/core/src/tools/mcp-client-manager.ts @@ -45,6 +45,230 @@ const DEFAULT_HEALTH_CONFIG: MCPHealthMonitorConfig = { reconnectDelayMs: 5000, // 5 seconds }; +/** + * Single-threshold warning fraction for the snapshot-based budget cell + * (PR 14 v1). When `liveCount >= MCP_BUDGET_WARN_FRACTION * budget` the + * `budgets[0].status` flips to `'warning'`. Exported and consumed by + * (a) `acpAgent.buildBudgetCells` (snapshot status) and (b) + * `commands/serve.ts` (stderr boot breadcrumb) — pre-extract these + * shared so PR 14b can swap to a dual-threshold hysteresis pair + * (`armed` boolean per opencode `cli/heap.ts`) by editing one file. + * + * Picked 0.75 to mirror PR 10's `slow_client_warning` + * (`eventBus.ts:WARN_THRESHOLD_RATIO`) — same rationale: "warning" + * fires before "error" with enough headroom for the operator to act. + */ +export const MCP_BUDGET_WARN_FRACTION = 0.75 as const; + +/** + * Budget enforcement mode for MCP client guardrails (issue #4175 PR 14). + * + * `off` — no accounting-driven enforcement (default when no budget is + * configured). `getMcpClientAccounting()` still works as pure + * observability; slot reservation is a no-op. + * `warn` — measure-only. Reserved slots track the configured set even + * beyond the budget so operators see `liveCount > budget` in the + * snapshot. No connect is refused. Snapshot consumers render a + * warning cell when `liveCount >= 0.75 * budget`. + * `enforce` — hard cap. Connects beyond the budget are refused, the + * per-server cell shows `errorKind: 'budget_exhausted'`, and the + * server name lands in `refusedServerNames`. Refusal is deterministic + * by `Object.entries(servers)` declaration order. + */ +export type McpBudgetMode = 'enforce' | 'warn' | 'off'; + +export interface McpBudgetConfig { + /** + * Cap on live MCP clients **per ACP session** (PR 14 v1; R4 review + * scope correction — see `acpAgent.newSessionConfig` constructs a + * fresh `Config`/`McpClientManager` per session, so each session + * enforces its own copy of the cap independently). Wave 5 PR 23 + * shared MCP pool will graduate this to per-workspace. + * `undefined` = unlimited. + */ + clientBudget?: number; + /** Behavior at and above the cap. `off` when `clientBudget` is undefined. */ + budgetMode: McpBudgetMode; +} + +/** Transport family per `MCPServerConfig`. `unknown` covers misconfigured entries. */ +export type McpTransportKind = + | 'stdio' + | 'sse' + | 'http' + | 'websocket' + | 'sdk' + | 'unknown'; + +/** + * Snapshot of the manager's live + reserved MCP state. The daemon's + * read-only `GET /workspace/mcp` route fans this out via the ACP + * `qwen/status/workspace/mcp` ext-method. `subprocessCount` is the + * value PR 1's `pgrep -P` baseline harness can validate against. + */ +export interface McpClientAccounting { + /** Live (`MCPServerStatus.CONNECTED`) client count, all transports. */ + total: number; + /** Live client count split by transport family. */ + byTransport: Record; + /** stdio + websocket — the only transports that spawn an OS process. */ + subprocessCount: number; + /** Server names currently holding a budget slot (in or over the cap). */ + reservedSlots: string[]; + /** Server names refused during the most recent `discoverAllMcpTools*` pass. */ + refusedServerNames: string[]; +} + +/** + * Thrown by `readResource` lazy-spawn path when the live count is + * already at `clientBudget` and `budgetMode === 'enforce'`. Discovery- + * time refusals don't throw (they're recorded in `refusedServerNames` + * and reported via the snapshot), because the discovery loop is + * best-effort and a thrown error would cancel sibling connects. + */ +export class BudgetExhaustedError extends Error { + readonly serverName: string; + readonly budget: number; + /** + * Number of slots currently reserved (== `reservedSlots.size` at the + * time of the refusal). PR 14 fix (review #4247 wenshao S6): renamed + * from `liveCount` because `reservedSlots` tracks reserved server + * NAMES, not `MCPServerStatus.CONNECTED` clients — a reserved-but- + * disconnected server still consumes a slot, and that's the + * accurate quantity blocking this new server from getting in. + * `getMcpClientAccounting().total` would have been the genuine + * "live" count and is a different number. + */ + readonly reservedCount: number; + constructor(serverName: string, budget: number, reservedCount: number) { + super( + `MCP client budget exhausted: cannot reserve slot for '${serverName}' ` + + `(budget=${budget}, reservedCount=${reservedCount}). ` + + `Raise --mcp-client-budget or remove servers from mcpServers config.`, + ); + this.name = 'BudgetExhaustedError'; + this.serverName = serverName; + this.budget = budget; + this.reservedCount = reservedCount; + } +} + +/** + * Map an `MCPServerConfig` to its transport family. Aligned with the + * detection order in `mcp-client.ts:createTransport` (sdk → httpUrl + * → url → command) with ONE forward-looking exception: `tcp` is + * mapped here to `websocket` matching the field's declared intent on + * `MCPServerConfig`, but `createTransport` does NOT yet construct a + * websocket transport. A config carrying both `tcp` and `command` + * is labeled `websocket` in the accounting snapshot while the real + * connection fires through the `command` path as `stdio`. The + * `subprocessCount = stdio + websocket` arithmetic is therefore + * accurate-by-vacancy today (no real websocket subprocesses exist + * yet) and will need revisiting if a websocket transport ships. + * Tracked: PR 14b / future core decision (see PR #4247 thread for + * Copilot finding #8 + wenshao P2 line 147 — defer pending direction + * on (a) implement WS in createTransport vs (b) drop `tcp` from + * `MCPServerConfig` + both mappers). + * + * `sdk` is checked first because `SDK_MCP_SERVER_FIELDS` may coexist + * with a placeholder `command` — without the sdk-first order, an + * in-process SDK server would mis-report as `stdio`. + */ +export function mcpTransportOf(config: MCPServerConfig): McpTransportKind { + if (isSdkMcpServerConfig(config)) return 'sdk'; + if (typeof config.httpUrl === 'string') return 'http'; + if (typeof config.url === 'string') return 'sse'; + if (typeof config.tcp === 'string') return 'websocket'; + if (typeof config.command === 'string') return 'stdio'; + return 'unknown'; +} + +/** + * Resolve budget config from env vars when the constructor caller + * doesn't pass one. Daemon-mode (`qwen serve`) sets these when + * spawning the `qwen --acp` child; standalone `qwen` invocations + * leave them unset and get `{ budgetMode: 'off' }` — the historical + * behavior, no enforcement. + * + * `QWEN_SERVE_MCP_CLIENT_BUDGET` — positive integer; non-numeric / + * zero / negative / NaN are silently ignored (treated as unset). + * `QWEN_SERVE_MCP_BUDGET_MODE` — `enforce|warn|off`. Defaults to + * `warn` when a budget is set, `off` otherwise. + */ +function readBudgetFromEnv(): McpBudgetConfig { + const rawBudget = process.env['QWEN_SERVE_MCP_CLIENT_BUDGET']; + const rawMode = process.env['QWEN_SERVE_MCP_BUDGET_MODE']; + let clientBudget: number | undefined; + if (rawBudget !== undefined && rawBudget !== '') { + const parsed = Number(rawBudget); + if (Number.isFinite(parsed) && Number.isInteger(parsed) && parsed > 0) { + clientBudget = parsed; + } else { + // PR 14 fix (review #4247 wenshao R7 line 191): operator typos + // like `QWEN_SERVE_MCP_CLIENT_BUDGET=abc` previously fell + // through silently to "no budget" with zero indication. The + // CLI parent (`commands/serve.ts` + `runQwenServe.ts`) + // validates and throws, but the ACP child process — where + // this function runs — has no such validation. Surface a + // boot breadcrumb so operators see the misconfiguration in + // journald / docker logs. + process.stderr.write( + `qwen serve: ignoring invalid QWEN_SERVE_MCP_CLIENT_BUDGET=` + + `'${rawBudget}' (expected positive integer); ` + + `MCP budget enforcement disabled for this child.\n`, + ); + } + } + let budgetMode: McpBudgetMode; + if (rawMode === 'enforce' || rawMode === 'warn' || rawMode === 'off') { + budgetMode = rawMode; + } else { + if (rawMode !== undefined && rawMode !== '') { + // Same operator-visibility rationale as the budget breadcrumb + // above. Unknown mode value silently fell through to the + // budget-driven default; now it gets a stderr line so the + // typo is visible. + process.stderr.write( + `qwen serve: ignoring invalid QWEN_SERVE_MCP_BUDGET_MODE=` + + `'${rawMode}' (expected enforce|warn|off); falling back to ` + + `${clientBudget === undefined ? 'off' : 'warn'}.\n`, + ); + } + budgetMode = clientBudget === undefined ? 'off' : 'warn'; + } + // PR 14 fix (review #4247 wenshao S4 + R8 #2): mode-without-budget + // downgrade. Originally only `enforce` got downgraded — but `warn` + // mode without a budget threshold is equally meaningless: nothing + // actionable can ever fire (no `liveCount >= 0.75 * budget` + // comparison can be true when budget is undefined). Downgrading + // BOTH to `off` removes the comment-vs-code mismatch in + // `emitBudgetTelemetry` (which previously claimed + // `mode !== 'off' ⇒ clientBudget defined` — true for enforce, + // false for warn until this fix). + // + // R9 #7: emit a stderr breadcrumb when the downgrade fires. + // Pre-fix the downgrade was silent — operator sets + // `QWEN_SERVE_MCP_BUDGET_MODE=enforce` in a Docker Compose / k8s + // env without the matching budget, daemon boots happy, snapshot + // shows `budgetMode: 'off'`, and enforcement is silently + // disabled. The CLI handler + `runQwenServe` path both throw on + // this combination; the env-var fallback path (used by the ACP + // child) was the laggard. Now mirrors the R7 #6 invalid-value + // breadcrumb pattern. + if ( + (budgetMode === 'enforce' || budgetMode === 'warn') && + clientBudget === undefined + ) { + process.stderr.write( + `qwen serve: QWEN_SERVE_MCP_BUDGET_MODE=${budgetMode} requires ` + + `QWEN_SERVE_MCP_CLIENT_BUDGET=N; downgrading to off. ` + + `Set both env vars to enable MCP guardrail enforcement.\n`, + ); + budgetMode = 'off'; + } + return { clientBudget, budgetMode }; +} + /** * Manages the lifecycle of multiple MCP clients, including local child processes. * This class is responsible for starting, stopping, and discovering tools from @@ -63,12 +287,48 @@ export class McpClientManager { private isReconnecting: Map = new Map(); private serverDiscoveryPromises: Map> = new Map(); + /** + * Budget bookkeeping. Slots are reserved synchronously by server name + * inside the discovery loop BEFORE any `await client.connect()`, so + * `Promise.all(discoveryPromises)` cannot interleave a second connect + * past the cap. `enforce` mode refuses past the cap; `warn` mode + * over-reserves so accounting reflects the configured set; `off` + * doesn't reserve at all. + */ + private readonly reservedSlots = new Set(); + private readonly clientBudget?: number; + private readonly budgetMode: McpBudgetMode; + /** + * PR 14 fix (review #4247 wenshao R8 #4 line 1221): names whose + * slot was freshly reserved (not `'already_held'`) by an + * in-flight `discoverMcpToolsForServerInternal` call. Read by + * `runWithDiscoveryTimeout`'s timeout handler to decide whether + * to release the slot on hard timeout — fresh reservations + * release (server never connected, slot shouldn't permanently + * block other servers); `'already_held'` reconnects keep their + * slot (operator's previously-healthy server shouldn't be + * permanently demoted by a transient timeout). + * + * Lifetime: `add` after `tryReserveSlot` returns `'reserved'` + * with the `.has` guard, `delete` in success / catch / finally + * cleanup. Idempotent — multiple deletes are no-ops. + */ + private readonly freshReservations = new Set(); + /** + * Servers refused during the most recent `discoverAllMcpTools*` pass. + * Reset at the start of each pass; survives between passes so a + * snapshot taken between discoveries still shows the last set of + * refusals to operators. + */ + private lastRefusedServerNames: string[] = []; + constructor( config: Config, toolRegistry: ToolRegistry, eventEmitter?: EventEmitter, sendSdkMcpMessage?: SendSdkMcpMessage, healthConfig?: Partial, + budgetConfig?: McpBudgetConfig, ) { this.cliConfig = config; this.toolRegistry = toolRegistry; @@ -76,6 +336,184 @@ export class McpClientManager { this.eventEmitter = eventEmitter; this.sendSdkMcpMessage = sendSdkMcpMessage; this.healthConfig = { ...DEFAULT_HEALTH_CONFIG, ...healthConfig }; + + // Tests inject `budgetConfig` directly; production reads env vars + // set by `qwen serve --mcp-client-budget=N --mcp-budget-mode=X` + // when spawning the ACP child. Standalone `qwen` invocations + // leave both unset and get `mode: 'off'` — the pre-PR-14 default. + const resolved = budgetConfig ?? readBudgetFromEnv(); + let resolvedMode = resolved.budgetMode; + // PR 14 fix (review #4247 wenshao R8 #5 + R10 line 357): mirror + // `readBudgetFromEnv`'s `(enforce|warn)`-without-budget + // downgrade for the direct-`budgetConfig` path too. All + // production callers (CLI handler, `runQwenServe`, env-var + // fallback) validate upfront, but a future code path that + // injects `budgetConfig` without running the validation + // would re-introduce the silent fail-open. Defense in depth. + // + // R10 line 357: emit the same stderr breadcrumb the env-var + // path uses. Pre-R10 the env-var path logged on downgrade but + // this constructor path was silent — same operator-visibility + // failure mode (operator only sees `budgetMode: 'off'` after + // the fact via the snapshot). Now both paths surface the + // misconfiguration at boot, so a future caller that bypasses + // CLI / env-var validation can't ship a daemon that + // advertises `mcp_guardrails` while silently disabling + // enforcement. + if ( + (resolvedMode === 'enforce' || resolvedMode === 'warn') && + resolved.clientBudget === undefined + ) { + process.stderr.write( + `qwen serve: McpClientManager constructed with budgetMode=${resolvedMode} ` + + `but no clientBudget; downgrading to off.\n`, + ); + resolvedMode = 'off'; + } + this.clientBudget = resolved.clientBudget; + this.budgetMode = resolvedMode; + } + + /** + * Atomic budget check + slot reservation. Synchronous so the + * concurrent discovery loop (`Promise.all` over server entries) can't + * interleave a second connect past the cap at any `await` boundary. + * + * Returns: + * `reserved` — slot newly held (or `off`-mode no-op) + * `already_held` — slot was already reserved (reconnect / dup) + * `refused` — `enforce` mode and the cap is full + */ + private tryReserveSlot( + serverName: string, + ): 'reserved' | 'already_held' | 'refused' { + if (this.reservedSlots.has(serverName)) return 'already_held'; + if (this.clientBudget === undefined || this.budgetMode === 'off') { + return 'reserved'; + } + if ( + this.budgetMode === 'enforce' && + this.reservedSlots.size >= this.clientBudget + ) { + return 'refused'; + } + // `warn` mode (and `enforce` under cap) — track in the configured set. + this.reservedSlots.add(serverName); + return 'reserved'; + } + + /** + * Snapshot the manager's MCP accounting for the daemon's read-only + * `GET /workspace/mcp` route. Cheap to call — iterates `this.clients` + * once and constructs a fresh struct each time so callers can mutate + * the returned arrays without affecting internal state. + * + * `total` counts only `CONNECTED` clients; `reservedSlots` includes + * the configured set (which under `enforce` mode is bounded by + * `clientBudget`, but under `warn` mode can exceed it). + */ + getMcpClientAccounting(): McpClientAccounting { + const byTransport: Record = { + stdio: 0, + sse: 0, + http: 0, + websocket: 0, + sdk: 0, + unknown: 0, + }; + let total = 0; + const servers = this.cliConfig.getMcpServers() ?? {}; + for (const [name, client] of this.clients) { + if (client.getStatus() !== MCPServerStatus.CONNECTED) continue; + const cfg = servers[name]; + const transport: McpTransportKind = cfg ? mcpTransportOf(cfg) : 'unknown'; + byTransport[transport] += 1; + total += 1; + } + return { + total, + byTransport, + subprocessCount: byTransport.stdio + byTransport.websocket, + reservedSlots: Array.from(this.reservedSlots), + refusedServerNames: [...this.lastRefusedServerNames], + }; + } + + /** Resolved budget mode (env-var or constructor-supplied). */ + getMcpBudgetMode(): McpBudgetMode { + return this.budgetMode; + } + + /** Resolved client budget, or `undefined` when unlimited. */ + getMcpClientBudget(): number | undefined { + return this.clientBudget; + } + + /** + * PR 14 fix (review #4247 wenshao R7 line 464): drop a server's + * entry from the per-pass refusal log, if present. The + * `indexOf` + `splice` pattern was repeated at 4 sites + * (`removeServer`, `disconnectServer`, `runWithDiscoveryTimeout` + * timeout handler, `readResource` late-reserve clear). Centralizing + * here makes future fixes (e.g. emitting an `mcp_budget_cleared` + * event when the entry is dropped) a one-place change. + */ + private dropRefusalEntry(serverName: string): void { + const idx = this.lastRefusedServerNames.indexOf(serverName); + if (idx >= 0) { + this.lastRefusedServerNames.splice(idx, 1); + } + } + + /** + * PR 14 fix (review #4247 wenshao R7 line 464): record a refusal + + * emit the operator-visible stderr breadcrumb. The push + + * stderr.write block was repeated at 3 sites (`discoverAllMcpTools` + * + `discoverAllMcpToolsIncremental` + `discoverMcpToolsForServerInternal`). + * Centralizing here keeps the message format consistent and makes + * future telemetry additions (e.g. `recordStartupEvent` per + * refusal) a one-place change. + * + * Idempotent on the push: if `serverName` is already in the list + * (rare but possible for the lazy-spawn refusal path which can be + * reached more than once for the same server), the array isn't + * grown. The stderr line still fires so the operator sees the + * refusal at every reproduction. + */ + private refuseAndLog(serverName: string): void { + if (!this.lastRefusedServerNames.includes(serverName)) { + this.lastRefusedServerNames.push(serverName); + } + process.stderr.write( + `qwen serve: MCP server '${serverName}' refused (budget exhausted, ` + + `budget=${this.clientBudget}, mode=enforce)\n`, + ); + } + + /** + * PR 14 fix (review #4247 wenshao S5): post-discovery budget + * telemetry was duplicated verbatim in `discoverAllMcpTools` and + * `discoverAllMcpToolsIncremental`. Centralized here so future + * field additions to `mcp_budget_decision` happen in one place. + * `off` mode is a no-op — operators who never set a budget don't + * pollute the startup-event sink. + * + * Invariant (post R8 #2): `mode !== 'off'` ⇒ `clientBudget` was + * resolved. Both `readBudgetFromEnv` AND the constructor downgrade + * `enforce`/`warn`-without-budget to `off` so neither call site can + * leave a budgetless mode reaching this telemetry path. + * `clientBudget ?? 0` is kept as belt-and-suspenders against future + * call sites that might bypass both validations. + */ + private emitBudgetTelemetry(configuredCount: number): void { + if (this.budgetMode === 'off') return; + recordStartupEvent('mcp_budget_decision', { + mode: this.budgetMode, + budget: this.clientBudget ?? 0, + configured: configuredCount, + reserved: this.reservedSlots.size, + refused: this.lastRefusedServerNames.length, + }); } /** @@ -95,6 +533,11 @@ export class McpClientManager { ); this.discoveryState = MCPDiscoveryState.IN_PROGRESS; + // Reset per-pass refusal log so a snapshot taken after this pass + // reflects THIS pass's refusals, not a stale one. Reservations + // (this.reservedSlots) persist across passes — they're keyed by + // server name, which is the operator's intent unit. + this.lastRefusedServerNames = []; this.eventEmitter?.emit('mcp-client-update', this.clients); const discoveryPromises = Object.entries(servers).map( @@ -105,6 +548,16 @@ export class McpClientManager { return; } + // Budget gate (PR 14): synchronous slot reservation BEFORE the + // `await client.connect()` below. Refusal only happens under + // `enforce` mode; `warn` mode reserves regardless so accounting + // reflects the configured set. `off` is a no-op. + const reservation = this.tryReserveSlot(name); + if (reservation === 'refused') { + this.refuseAndLog(name); + return; + } + // For SDK MCP servers, pass the sendSdkMcpMessage callback const sdkCallback = isSdkMcpServerConfig(config) ? this.sendSdkMcpMessage @@ -127,6 +580,48 @@ export class McpClientManager { await client.discover(cliConfig); this.eventEmitter?.emit('mcp-client-update', this.clients); } catch (error) { + // PR 14 fix (review #4247 wenshao C2): zombie slot leak. + // `tryReserveSlot(name)` reserved a slot above. If `connect()` + // throws, the slot would stay reserved forever and the client + // entry would stay in `this.clients` in a never-CONNECTED + // state, blocking other servers in `enforce` mode until a + // full discovery restart. Release both so the budget cap + // reflects actual usable capacity. + // + // Slot bookkeeping in this bulk path is partially redundant + // with `await this.stop()` at the top of + // `discoverAllMcpTools` (line ~320) — the next bulk run + // wipes `reservedSlots` regardless. But the SAME catch + // ALSO needs to handle the transport (see below): the + // client object held by `clients.delete(name)` only had + // its tracking reference removed, not its underlying + // transport closed. Leaving the orphan transport alive + // would leak the stdio child / WebSocket / HTTP socket + // for the rest of the process — `stop()` can't clean it + // because we just removed it from the map. + // + // The per-server reconnect path + // (`discoverMcpToolsForServerInternal`) keeps the slot + // when `weReservedSlot === false` so health-monitor retry + // doesn't have to compete for capacity — different + // lifecycle, different contract. Bulk path always releases + // because every server is "fresh" here (preceded by + // stop()). + // + // PR 14 fix (review #4247 wenshao R8 #1 line 532): also + // call `await client.disconnect()` BEFORE dropping the + // reference. R7 #3 fixed the analogous leak in the + // per-server path; this is the bulk-path mirror. Errors + // intentionally swallowed (we're already in a discovery- + // failure catch; double-throwing would lose the original + // error context). + try { + await client.disconnect(); + } catch { + // best-effort transport cleanup + } + this.reservedSlots.delete(name); + this.clients.delete(name); this.eventEmitter?.emit('mcp-client-update', this.clients); // Log the error but don't let a single failed server stop the others debugLogger.error( @@ -140,6 +635,7 @@ export class McpClientManager { await Promise.all(discoveryPromises); this.discoveryState = MCPDiscoveryState.COMPLETED; + this.emitBudgetTelemetry(Object.keys(servers).length); } /** @@ -185,6 +681,69 @@ export class McpClientManager { if (!serverConfig) { return; } + // PR 14 fix (review #4247 wenshao R7 line 528): disabled gate. + // `discoverMcpToolsForServerInternal` is reachable from + // `/mcp reconnect`, OAuth re-discovery, and the health monitor's + // `reconnectServer`. Without this check those paths could + // resurrect a server the operator has explicitly disabled, + // wasting a budget slot and registering tools the user told us + // to ignore. Mirrors the disabled checks in + // `discoverAllMcpTools` + `discoverAllMcpToolsIncremental` + + // `readResource`. + // + // Optional-chain on `isMcpServerDisabled` is defensive against + // test fixtures that omit the method (the bulk paths already + // assume it exists; this single-server path was the laggard). + // Production `Config` always defines the method. + if (this.cliConfig.isMcpServerDisabled?.(serverName)) { + debugLogger.debug(`Skipping disabled MCP server: ${serverName}`); + return; + } + + // PR 14 fix (review #4247): single-server rediscovery (reachable from + // `/mcp reconnect ` and `ToolRegistry.discoverToolsForServer`) + // previously bypassed the budget gate, so a server refused at startup + // could be brought online later under `enforce` mode and exceed the + // cap. True reconnect against a held slot returns `'already_held'` + // and falls through unchanged; only a fresh attempt against a server + // without a reservation can be refused. Best-effort semantics — log + // the refusal and return without creating an `McpClient`; the caller + // observes the absence via `getStatus()` like any other discovery + // failure. + const reservation = this.tryReserveSlot(serverName); + if (reservation === 'refused') { + this.refuseAndLog(serverName); + return; + } + // PR 14 fix (review #4247 wenshao R3-R4): track whether THIS call + // freshly reserved the slot. Used in the connect-failure catch + // below — only the fresh-reserve case releases the slot; a true + // reconnect (`'already_held'`) keeps its existing reservation so + // health-monitor retry doesn't have to compete for capacity. + // + // The `reservedSlots.has(serverName)` guard distinguishes a real + // reservation from an `off`-mode no-op: in `off` mode + // `tryReserveSlot` returns `'reserved'` WITHOUT adding to the + // set (no enforcement), so we don't want to fire cleanup for + // a slot we never actually took — that would unnecessarily + // remove the failed client entry and break the + // health-monitor-driven retry loop (regression test: + // "should restore health checks after failed server + // rediscovery"). + const weReservedSlot = + reservation === 'reserved' && this.reservedSlots.has(serverName); + // PR 14 fix (review #4247 wenshao R8 #4): mark this name in + // `freshReservations` so the `runWithDiscoveryTimeout` timeout + // handler can distinguish fresh-reservation timeouts (release + // the slot — never connected, shouldn't block others) from + // `'already_held'` reconnect timeouts (keep the slot — operator's + // previously-healthy server shouldn't be demoted by a transient + // timeout). Cleared in success / catch / finally below so the + // marker only spans the current discoverMcpToolsForServerInternal + // invocation. + if (weReservedSlot) { + this.freshReservations.add(serverName); + } this.stopHealthCheck(serverName); @@ -224,7 +783,59 @@ export class McpClientManager { try { await client.connect(); await client.discover(cliConfig); + // PR 14 fix (review #4247 wenshao R7 line 612): a server that + // was refused at a previous discovery pass and is now + // successfully (re)connected via this path (e.g. `/mcp + // reconnect`, health-monitor retry after another server was + // removed) leaves a stale entry in `lastRefusedServerNames`. + // The snapshot would then report `error / disabledReason: + // 'budget'` for a CONNECTED server until the next discovery + // pass clears the per-pass log. Clear it here so post-success + // snapshots immediately reflect reality. Mirrors the same + // pattern in `readResource`'s late-reserve branch. + this.dropRefusalEntry(serverName); } catch (error) { + // PR 14 fix (review #4247 wenshao R3 line 546): two-mode + // cleanup for connect failure, matching the `readResource` + // R2 C3 fix pattern: + // + // - `weReservedSlot === true` (this call freshly took a + // slot for a brand-new server): RELEASE the slot + drop + // the client. The server never successfully held a slot + // and shouldn't permanently block another server in + // `enforce` mode. Operator can re-add it later; the next + // `discoverAllMcpToolsIncremental` pass will re-reserve + // if capacity is available. + // - `weReservedSlot === false` (reconnect against an + // `'already_held'` slot — e.g. health-monitor retry, + // `/mcp reconnect` against a stable-but-momentarily-flaky + // server): KEEP the slot. The original successful connect + // established operator intent + capacity reservation; a + // transient reconnect hiccup shouldn't lose that. + // + // Round 3 documented "always keep" — corrected here per + // wenshao R3 P3 line 390 + R4 line 546/639: align with + // `discoverAllMcpTools` (bulk) catch and `readResource` + // (lazy spawn) catch. All three paths now use the same + // weReserved-driven cleanup. + if (weReservedSlot) { + // PR 14 fix (review #4247 wenshao R7 line 634): transport + // leak — when `connect()` succeeded (transport established) + // but `discover()` later threw, deleting the client without + // calling `disconnect()` left the stdio child process / + // socket alive until Node exits. Best-effort disconnect + // here closes the transport before dropping our reference. + // Errors from disconnect are intentionally swallowed + // (we're already in a discovery-failure catch; double- + // throwing would lose the original error context). + try { + await client.disconnect(); + } catch { + // best-effort transport cleanup + } + this.reservedSlots.delete(serverName); + this.clients.delete(serverName); + } // Log the error but don't throw: callers expect best-effort discovery. debugLogger.error( `Error during discovery for server '${serverName}': ${getErrorMessage( @@ -234,6 +845,11 @@ export class McpClientManager { } finally { this.startHealthCheck(serverName); this.eventEmitter?.emit('mcp-client-update', this.clients); + // R8 #4: clear the fresh-reservation marker — this in-flight + // call has settled (success, catch, OR a timeout that already + // ran its handler). Idempotent on the timeout-already-deleted + // case. + this.freshReservations.delete(serverName); } } @@ -262,6 +878,13 @@ export class McpClientManager { this.consecutiveFailures.clear(); this.isReconnecting.clear(); this.serverDiscoveryPromises.clear(); + // PR 14: clean shutdown releases ALL budget slots. A subsequent + // `discoverAllMcpTools*` (e.g. the `discoverAllMcpTools` call in + // its own body line 90, which awaits `this.stop()` first) starts + // from an empty reservation set. + this.reservedSlots.clear(); + this.freshReservations.clear(); + this.lastRefusedServerNames = []; } /** @@ -288,6 +911,17 @@ export class McpClientManager { this.eventEmitter?.emit('mcp-client-update', this.clients); } } + // PR 14: explicit operator-driven disconnect releases the budget + // slot AND drops the entry from the per-pass refusal log. Outside + // the `if (client)` guard because a budget-refused server has NO + // `McpClient` instance — but operator intent ("stop tracking this + // server") still demands the records be cleared so a subsequent + // snapshot doesn't keep tagging it as `budget_exhausted`. The + // internal reconnect path (`discoverMcpToolsForServerInternal`) + // calls `existingClient.disconnect()` directly, NOT this public + // method, so reconnect still doesn't release the slot. + this.reservedSlots.delete(serverName); + this.dropRefusalEntry(serverName); } getDiscoveryState(): MCPDiscoveryState { @@ -459,6 +1093,9 @@ export class McpClientManager { ); this.discoveryState = MCPDiscoveryState.IN_PROGRESS; + // Reset per-pass refusal log; see the sibling reset in + // `discoverAllMcpTools` for rationale. + this.lastRefusedServerNames = []; recordStartupEvent('mcp_discovery_start', { serverCount: Object.keys(servers).length, incremental: true, @@ -479,6 +1116,22 @@ export class McpClientManager { const currentServerNames = new Set(this.clients.keys()); const newServerNames = new Set(Object.keys(servers)); + // PR 14 fix (review #4247): process removals BEFORE the new-server + // reservation pass so freed slots are visible to `tryReserveSlot`. + // Scenario: budget=2, currently `{a, b}` reserved, new config + // `{a, c}`. Pre-fix order refused `c` because `b`'s slot was only + // freed after the new-server loop. Now `b` is removed first → + // reservedSlots={a} → `c` reservation succeeds. Disabled-mid-session + // removals stay inline (below) because they also release slots + // via `removeServer`'s `reservedSlots.delete` — same call, just + // reached from a different branch. + for (const name of currentServerNames) { + if (!newServerNames.has(name)) { + // Server was removed from configuration + await this.removeServer(name); + } + } + // Check for new servers or configuration changes for (const [name] of Object.entries(servers)) { // Mirror `discoverAllMcpTools` (line ~102): users who explicitly @@ -502,7 +1155,25 @@ export class McpClientManager { } const existingClient = this.clients.get(name); if (!existingClient) { - // New server + // PR 14 fix (review #4247 wenshao R6 line 956): pre-reservation + // here was a TOCTOU race. The inner + // `discoverMcpToolsForServerInternal` ALSO does `tryReserveSlot` + // (added in R1 fix #1). With BOTH sites reserving, the + // reservation lifecycle didn't align with the timeout + // cleanup site — `runWithDiscoveryTimeout`'s timeout handler + // could release the slot mid-flight while the inner + // `connect()` later resolves successfully, leaving a + // CONNECTED client with NO reservation. Next pass admits + // another new server because `reservedSlots.size < budget`, + // and `enforce` mode silently exceeds the cap. + // + // Fix: delete the pre-reservation. `discoverMcpToolsForServerInternal` + // owns the reservation lifecycle end-to-end (reserve → + // try-catch around connect → release on weReservedSlot + // failure path → cleared by timeout handler if it fires). + // Refusal still happens — just inside the inner call. The + // operator-visible behavior is identical; only the race is + // closed. serversToUpdate.push(name); } else if (existingClient.getStatus() === MCPServerStatus.DISCONNECTED) { // Disconnected server, try to reconnect @@ -512,14 +1183,6 @@ export class McpClientManager { // the old and new config, which is not implemented here } - // Find removed servers - for (const name of currentServerNames) { - if (!newServerNames.has(name)) { - // Server was removed from configuration - await this.removeServer(name); - } - } - // Update only the servers that need it. Each per-server discover is // wrapped in a discovery-only timeout (stdio default 30s, remote 5s, // per-server override via `discoveryTimeoutMs`). Tool-call timeout is @@ -586,6 +1249,7 @@ export class McpClientManager { serverCount: Object.keys(servers).length, incremental: true, }); + this.emitBudgetTelemetry(Object.keys(servers).length); // Trailing `mcp-client-update` AFTER flipping discoveryState to // COMPLETED. Without this the per-server updates above all fire while // the state is still IN_PROGRESS, so the AppContainer batch-flush @@ -654,6 +1318,31 @@ export class McpClientManager { // absent, so the trailing `finally`-block call becomes a no-op. this.stopHealthCheck(serverName); this.clients.delete(serverName); + // PR 14 fix (review #4247 wenshao R5 line 956 + R8 #4 line + // 1221): release the budget slot ONLY if THIS in-flight + // discoverMcpToolsForServerInternal call freshly reserved + // it. `freshReservations.has(serverName)` distinguishes: + // + // - Fresh reservation (never connected): release — a server + // that never connected shouldn't permanently consume a + // slot under enforce mode. + // - `'already_held'` reconnect (server was previously + // healthy, now flaky): KEEP the slot. Health-monitor + // retry doesn't have to compete for capacity with new + // servers admitted during the timeout window. + // + // R5 originally treated all timeouts as "release"; wenshao + // R8 #4 caught the asymmetry with the connect-failure + // path's `weReservedSlot` guard. Now they match. + if (this.freshReservations.has(serverName)) { + this.reservedSlots.delete(serverName); + this.freshReservations.delete(serverName); + } + // And drop any stale refusal entry — operator intent shifts + // when a slot becomes free again, and snapshot consumers + // shouldn't keep tagging a now-slotless server as + // `disabledReason: 'budget'`. + this.dropRefusalEntry(serverName); reject( new Error( `MCP server '${serverName}' discovery timed out after ${timeoutMs}ms`, @@ -735,6 +1424,18 @@ export class McpClientManager { this.consecutiveFailures.delete(serverName); } + // PR 14: server gone from config (or disabled mid-session) releases + // the budget slot too — operator intent is "this server should not + // be running", so it must not block a different server from taking + // its place on the next discovery pass. + this.reservedSlots.delete(serverName); + // PR 14 fix (review #4247): also drop the entry from the per-pass + // refusal log so a snapshot taken between discoveries doesn't + // stale-tag the (now-disabled or now-removed) server as + // `disabledReason: 'budget'`. Operator action wins over the + // last-pass startup refusal record. + this.dropRefusalEntry(serverName); + // Remove tools for this server from registry this.toolRegistry.removeMcpToolsByServer(serverName); @@ -751,16 +1452,93 @@ export class McpClientManager { options?: { signal?: AbortSignal }, ): Promise { let client = this.clients.get(serverName); + // PR 14 fix (review #4247 wenshao C3): track whether THIS call + // reserved the slot + created the client, so the zombie-leak + // cleanup on `connect()` failure (below) only fires for + // newly-created lazy spawns — never for a reuse of an already- + // CONNECTED client (`client !== undefined` branch). + let weReservedSlot = false; + // PR 14 fix (review #4247 wenshao R9 #6 line 1521): hoist the + // serverConfig lookup so the timeout-wrapped connect site + // (below) can pass it to `discoveryTimeoutFor` regardless of + // whether we're on the lazy-spawn or already-existing-client + // path. Existing clients get the same per-server discovery + // timeout as fresh ones — uniform behavior across spawn paths. + const servers = populateMcpServerCommand( + this.cliConfig.getMcpServers() || {}, + this.cliConfig.getMcpServerCommand(), + ); + const serverConfig = servers[serverName]; if (!client) { - const servers = populateMcpServerCommand( - this.cliConfig.getMcpServers() || {}, - this.cliConfig.getMcpServerCommand(), - ); - const serverConfig = servers[serverName]; + // PR 14 invariant (wenshao R2 P3 line 501): the lookup→ + // disabled-check→budget-reserve→client-create sequence below + // runs synchronously — no `await` until `client.connect()`. + // `cliConfig.getMcpServers()` returns the current Map snapshot, + // and `cliConfig` is mutated only between discovery passes (via + // settings reload) or via `removeServer` (which releases its + // own slot). So the TOCTOU window between `serverConfig` + // lookup and `tryReserveSlot` is closed by Node's single- + // threaded execution model. If the manager ever grows an + // `await`-containing branch in this section, wrap from line + // below through `clients.set` in `try { ... } catch { + // this.reservedSlots.delete(serverName); throw; }` to close + // a real race. if (!serverConfig) { throw new Error(`MCP server '${serverName}' is not configured.`); } + // PR 14 fix (review #4247 wenshao R2-#5): the lazy-spawn path + // previously bypassed `isMcpServerDisabled`. A server the + // operator disabled via `mcpServers..disabled: true` or + // `/mcp disable ` could be resurrected by any resource + // read call. Now matches the disabled-check pattern in + // `discoverAllMcpTools` and `discoverAllMcpToolsIncremental`. + // Placed BEFORE the budget gate so a disabled server reports + // its actual reason rather than a misleading budget refusal. + if (this.cliConfig.isMcpServerDisabled(serverName)) { + throw new Error(`MCP server '${serverName}' is disabled.`); + } + + // Budget gate (PR 14): a lazy `readResource` against a server + // that was refused at discovery time (or that the operator has + // never connected) must NOT silently spawn a new MCP client past + // the cap. Discovery-time refusals don't throw (best-effort + // semantics), but the resource-read caller has a synchronous + // consumer that benefits from a typed error it can render. + const reservation = this.tryReserveSlot(serverName); + if (reservation === 'refused') { + // R7 #7 helper: refuseAndLog records the entry + emits the + // operator-visible stderr breadcrumb. Calling it BEFORE the + // throw so operators get the same stderr trail as bulk + // discovery refusals — the throw alone doesn't surface to + // stderr (caller decides what to do with the typed error). + this.refuseAndLog(serverName); + throw new BudgetExhaustedError( + serverName, + this.clientBudget as number, + this.reservedSlots.size, + ); + } + // R7 #4: align with `discoverMcpToolsForServerInternal` — + // `tryReserveSlot` returns `'reserved'` in `off` mode WITHOUT + // adding to the set. The `.has` guard ensures we only treat it + // as a real reservation when the slot was actually taken. + weReservedSlot = + reservation === 'reserved' && this.reservedSlots.has(serverName); + + // PR 14 fix (review #4247 wenshao R5 line 1268-1): a server + // that was refused at discovery time stays in + // `lastRefusedServerNames` so the snapshot reports it. If a + // later `readResource` call successfully reserves a slot for + // that server (e.g., another server was disconnected and + // freed capacity), the refusal entry becomes stale — the + // snapshot would keep tagging the now-connected server as + // `disabledReason: 'budget'`. Drop the stale entry here so + // the next snapshot reflects the late-reservation success. + if (weReservedSlot) { + this.dropRefusalEntry(serverName); + } + const sdkCallback = isSdkMcpServerConfig(serverConfig) ? this.sendSdkMcpMessage : undefined; @@ -778,8 +1556,108 @@ export class McpClientManager { this.eventEmitter?.emit('mcp-client-update', this.clients); } + // PR 14 fix (review #4247 wenshao R7 line 1342): when an already- + // tracked client exists (the `if (!client)` block above is + // skipped), the disabled gate added in R3 #5 doesn't fire. So a + // server connected pre-disable, then operator-disabled mid- + // session via `/mcp disable ` or a settings reload, would + // still serve resource reads via its existing CONNECTED client + // until the next incremental discovery pass calls `removeServer`. + // Re-check disabled state on every readResource, regardless of + // whether the client was just lazy-spawned or pre-existing. + if (this.cliConfig.isMcpServerDisabled(serverName)) { + throw new Error(`MCP server '${serverName}' is disabled.`); + } + if (client.getStatus() !== MCPServerStatus.CONNECTED) { - await client.connect(); + try { + // PR 14 fix (review #4247 wenshao R9 #6 line 1521): wrap the + // lazy-spawn `client.connect()` in the same discovery + // timeout the bulk + incremental paths use. Pre-fix a hung + // MCP server during a resource-read spawn would block + // forever and permanently consume a budget slot under + // `enforce` mode, cascading into total budget exhaustion + // on subsequent discovery passes. Reuses + // `discoveryTimeoutFor` so per-server `discoveryTimeoutMs` + // overrides apply uniformly across spawn paths. + // + // R10 line 1572 cleanup contract: when the timeout side + // wins the race, the catch below calls + // `await client.disconnect()` to abort the orphan + // `client.connect()` that's still pending in the + // background. This relies on `McpClient.disconnect()` + // cancelling an in-flight connect — closing the underlying + // transport (stdio child SIGTERM, WebSocket close frame, + // HTTP socket teardown) so the pending connect promise + // settles. If `disconnect()` on a never-completed connect + // were a no-op, the orphan transport would survive with + // no `this.clients` entry and `stop()` couldn't reach it. + // This same contract is relied on by + // `runWithDiscoveryTimeout`'s timeout handler (bulk + + // incremental paths), so all three spawn paths share the + // assumption — verified by the bulk path having shipped + // production-stable for several releases. Worth a unit + // test in a follow-up that exercises the + // disconnect-cancels-pending-connect invariant against + // a fixture that asserts the transport is actually torn + // down. + + const timeoutMs = this.discoveryTimeoutFor(serverConfig); + let timeoutId: NodeJS.Timeout | undefined; + await Promise.race([ + client.connect(), + new Promise((_, reject) => { + timeoutId = setTimeout(() => { + reject( + new Error( + `MCP server '${serverName}' lazy connect timed out after ${timeoutMs}ms`, + ), + ); + }, timeoutMs); + }), + ]).finally(() => { + if (timeoutId) clearTimeout(timeoutId); + }); + // PR 14 fix (review #4247 wenshao R9 #8 line 1514): start + // the health monitor on a successful lazy spawn. Pre-fix + // a lazy-spawned server that later disconnected (crash, + // network) had no automatic reconnect path — the client + // sat DISCONNECTED in `this.clients` until the next + // readResource or incremental pass. Mirror the + // `discoverMcpToolsForServerInternal` finally-block + // pattern. + this.startHealthCheck(serverName); + } catch (err) { + // PR 14 fix (review #4247 wenshao C3 + R9 #2 line 1534): + // zombie slot leak + transport leak. + // + // A failed lazy spawn would otherwise permanently consume + // a budget slot AND leave a never-CONNECTED client entry + // in `this.clients` (which `getMcpClientAccounting` + // correctly excludes from `total`, but the slot still + // blocks other servers). Only release if THIS call did + // the reservation — a reuse path with an already-tracked + // client must not collateral-damage another caller's + // slot. + // + // R9 #2: `connect()` may have established the transport + // (spawned the stdio child / opened the socket) before + // throwing on a later handshake step. Best-effort + // `await client.disconnect()` closes that transport + // before dropping the reference — mirrors the R7 #3 + + // R8 #1 fixes in the discovery-side catch blocks. + if (weReservedSlot) { + try { + await client.disconnect(); + } catch { + // best-effort transport cleanup + } + this.reservedSlots.delete(serverName); + this.clients.delete(serverName); + this.eventEmitter?.emit('mcp-client-update', this.clients); + } + throw err; + } } return client.readResource(uri, options); diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 68395b24619..ecf23e1d9fe 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -190,6 +190,9 @@ export const DAEMON_ERROR_KINDS = [ 'protocol_error', 'missing_file', 'parse_error', + // Issue #4175 PR 14: budget refusal under `--mcp-budget-mode=enforce`. + // Mirrors the serve-side `SERVE_ERROR_KINDS` addition. + 'budget_exhausted', ] as const; export type DaemonErrorKind = (typeof DAEMON_ERROR_KINDS)[number]; @@ -228,6 +231,47 @@ export interface DaemonWorkspaceMcpServerStatus extends DaemonStatusCell { disabled: boolean; description?: string; extensionName?: string; + /** + * Why this server is not live, when known (issue #4175 PR 14). + * `'config'` — operator-disabled via `disabledMcpServers`. + * `'budget'` — refused by the workspace MCP client budget + * (snapshot also surfaces `errorKind: + * 'budget_exhausted'`). + * Absent on pre-PR-14 daemons. + */ + disabledReason?: 'config' | 'budget'; +} + +/** Budget enforcement mode for MCP client guardrails (issue #4175 PR 14). */ +export type DaemonMcpBudgetMode = 'enforce' | 'warn' | 'off'; + +/** + * MCP client budget status cell. Issue #4175 PR 14 v1 emits one + * entry with `scope: 'session'` (per-session enforcement; see the + * `scope` field doc for why). Wave 5 PR 23 shared pool will add + * `scope: 'workspace'`. Consumers MUST tolerate unrecognized scope + * values — drop, don't fail. + */ +export interface DaemonMcpBudgetStatusCell extends DaemonStatusCell { + kind: 'mcp_budget'; + /** + * **PR 14 v1 emits `'session'`** — the budget caps live MCP + * clients per ACP session, not per-workspace. Each session has its + * own `McpClientManager` (created via `acpAgent.newSessionConfig`). + * Wave 5 PR 23 (shared MCP pool) will introduce a workspace-scoped + * manager and emit `'workspace'` (or `'pool'`) cells. + * + * The `string & {}` widening keeps IDE autocomplete + literal + * narrowing for known scopes while allowing unknown scopes through + * — the protocol contract is "consumers MUST tolerate additional + * scope values, drop don't fail." See `qwen-serve-protocol.md`. + */ + scope: 'session' | 'workspace' | (string & {}); + liveCount: number; + /** Configured cap. Absent when mode is `off`. */ + budget?: number; + mode: DaemonMcpBudgetMode; + refusedCount: number; } export interface DaemonWorkspaceMcpStatus { @@ -237,6 +281,18 @@ export interface DaemonWorkspaceMcpStatus { discoveryState?: DaemonMcpDiscoveryState; servers: DaemonWorkspaceMcpServerStatus[]; errors?: DaemonStatusCell[]; + /** PR 14: live MCP client count, all transports. Absent on pre-PR-14 daemons. */ + clientCount?: number; + /** PR 14: configured budget. Absent when no cap set. */ + clientBudget?: number; + /** PR 14: active enforcement mode. Absent on pre-PR-14 daemons. */ + budgetMode?: DaemonMcpBudgetMode; + /** + * PR 14: workspace-level budget cells. Empty array (not absent) + * on post-PR-14 daemons when no budget is configured AND mode + * resolves to `off`. Pre-PR-14 daemons omit the field. + */ + budgets?: DaemonMcpBudgetStatusCell[]; } export type DaemonSkillLevel = 'project' | 'user' | 'extension' | 'bundled';