feat(serve): MCP guardrail push events + hysteresis (#4175 Wave 3 PR 14b)

docs/developers/qwen-serve-protocol.md

@@ -101,6 +101,7 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design
  'workspace_providers', 'workspace_env', 'workspace_preflight',
  'session_context', 'session_supported_commands',
  'session_close', 'session_metadata', 'mcp_guardrails',
+ 'mcp_guardrail_events',
  'workspace_file_read', 'workspace_file_bytes', 'workspace_file_write',
  'session_approval_mode_control', 'workspace_tool_toggle',
  'workspace_init', 'workspace_mcp_restart']
@@ -141,6 +142,13 @@ routes and require a configured bearer token even on loopback.
 
 `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: []`).
 
+`mcp_guardrail_events` (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14b) advertises the typed SSE push events that surface MCP budget state crossings without a poll loop. Two frame types arrive on `GET /session/:id/events`:
+
+- `mcp_budget_warning` — fires once on the upward 75% crossing of `reservedSlots.size / clientBudget`. Re-arms only after the ratio drops below 37.5% (`MCP_BUDGET_REARM_FRACTION`). Mirrors PR 10's `slow_client_warning` hysteresis, but at the manager level rather than the per-subscriber backlog level. Payload: `{ liveCount, reservedCount, budget, thresholdRatio: 0.75, mode: 'warn' | 'enforce' }`. Fires under both `warn` and `enforce` modes; never under `off`.
+- `mcp_child_refused_batch` — fires at end of each `discoverAllMcpTools*` pass when one or more servers were refused, AND as a length-1 batch on the `readResource` lazy-spawn refusal path. Payload: `{ refusedServers: [{ name, transport, reason: 'budget_exhausted' }, ...], budget, liveCount, reservedCount, mode: 'enforce' }`. `mode` is the literal `'enforce'` because `warn` mode never refuses.
+
+Both events live in the per-session SSE replay ring (they carry an `id`) so a client reconnecting with `Last-Event-ID` resumes through them; the snapshot at `GET /workspace/mcp` is still the source-of-truth for state-after-extended-disconnect. Always-on once advertised — there is no conditional toggle. SDK reducer state (`DaemonSessionViewState`) exposes `mcpBudgetWarningCount`, `lastMcpBudgetWarning`, `mcpChildRefusedBatchCount`, `lastMcpChildRefusedBatch` for adapters that want simple lag-style UI.
+
 ## Routes
 
 ### `GET /health`
@@ -332,13 +340,16 @@ vars only; proxy URLs are stripped of credentials and reduced to
 
 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]`):
+**Detecting budget pressure.** Two surfaces, both populated post-PR-14b:
+
+- **Push events** (advertised via `mcp_guardrail_events`): subscribe to `GET /session/:id/events` and narrow `mcp_budget_warning` / `mcp_child_refused_batch` frames through `KnownDaemonEvent`. The state machine fires once per upward 75% crossing (re-armed below 37.5%); refusals are coalesced once per discovery pass under `enforce` mode.
+- **Snapshot poll** (advertised via `mcp_guardrails`): `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.
+Recommended poll cadence: aligned with whatever already polls `/workspace/mcp`; the snapshot is cheap and the budget cell carries no extra discovery cost. SDK clients that subscribe to push events still benefit from the snapshot for state-after-extended-disconnect (the SSE replay ring depth is finite — `--event-ring-size`, default 8000 — so a client offline longer than the ring's coverage falls back to snapshot resync).
 
 ### `GET /workspace/skills`
 

docs/users/qwen-serve.md

@@ -209,6 +209,8 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401
 > ```
 >
 > 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.
+>
+> **Push events (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14b).** SDK clients subscribed to `GET /session/:id/events` receive typed frames when budget thresholds cross — `mcp_budget_warning` (synthetic, fires once per upward 75% crossing with hysteresis re-arm at 37.5%, advertised via `mcp_guardrail_events`) and `mcp_child_refused_batch` (coalesced once per discovery pass under `enforce` mode; length-1 from `readResource` lazy-spawn refusal). The snapshot at `GET /workspace/mcp` is still the source-of-truth for state-after-reconnect; events are change-edges. Useful when dashboarding in real-time without polling.
 
 ## Default deployment threat model
 

integration-tests/cli/qwen-serve-baseline.test.ts

@@ -462,6 +462,78 @@ async function measureRssAtSessionCount(sessionCount: number): Promise<{
           fs.rmSync(ws, { recursive: true, force: true });
         }
       }, 120_000);
+
+      // PR 14b cross-check: validate the daemon's in-process MCP
+      // accounting against external `pgrep -P` measurement. The
+      // snapshot at `GET /workspace/mcp` exposes `clientCount`
+      // (live CONNECTED clients, `getMcpClientAccounting().total`)
+      // — that's the field SDK consumers and dashboards actually
+      // see, and it's the same source the push-event channel
+      // (`mcp_budget_warning` / `mcp_child_refused_batch`) reads.
+      // If `clientCount` diverges from what `pgrep -P` observes for
+      // the daemon's MCP grandchildren, the events lie.
+      //
+      // (Codex round 3 doc fix — codex/copilot finding: this test
+      // was named "in-process subprocessCount matches external pgrep"
+      // but actually asserts on `clientCount`. The snapshot's
+      // `clientCount` and the manager-internal `subprocessCount`
+      // (`stdio + websocket`) match for stdio-only fixtures, so the
+      // assertion is numerically correct — but the test name now
+      // matches the field it actually validates.)
+      //
+      // Skip-gated like the parent describe (POSIX, non-sandbox);
+      // idle MCP fixtures are stdio-only so `clientCount` should
+      // equal `mcpGrandchildren.length` exactly (no amplification
+      // slack required).
+      it('clientCount matches external pgrep observation', async () => {
+        const ws = makeTempWorkspace('mcp-counter');
+        let daemon: SpawnedDaemon | undefined;
+        try {
+          writeWorkspaceSettings(ws, {
+            mcpServers: {
+              idle1: { command: 'node', args: [IDLE_MCP_PATH] },
+              idle2: { command: 'node', args: [IDLE_MCP_PATH] },
+            },
+          });
+          daemon = await spawnDaemon({ workspaceCwd: ws });
+          await daemon.client.createOrAttachSession({ workspaceCwd: ws });
+
+          // Wait for MCP grandchildren to be observable via pgrep,
+          // then read both numbers atomically (pgrep first to lock
+          // the comparison floor, snapshot second so the daemon
+          // can't sneak in a new connect between the two reads).
+          const observed = await waitForMcpGrandchildren(
+            daemon.daemon.pid!,
+            MCP_SERVERS_CONFIGURED,
+          );
+          const snapshot = await daemon.client.workspaceMcp();
+
+          // PR 14b invariant: stdio-only fixtures →
+          // `clientCount === mcpGrandchildren.length`. The PR 14
+          // amplification slack
+          // (`MCP_SERVERS_CONFIGURED * mcpAmplificationFactor`) is
+          // for connect-storm transient overhead, not steady-state
+          // counter drift. At idle the daemon's accounting MUST
+          // match `pgrep -P` exactly (no zombies, no orphans).
+          //
+          // `clientCount` is the snapshot's authoritative live
+          // count; validating it against pgrep closes the loop on
+          // PR 14b's event-source assumption (the push events read
+          // the same accounting).
+          expect(snapshot.clientCount).toBe(MCP_SERVERS_CONFIGURED);
+          expect(observed.mcpGrandchildren.length).toBe(MCP_SERVERS_CONFIGURED);
+          // Defense-in-depth: even if a future race lets the OS
+          // observe a process the daemon already considers
+          // disconnected, `clientCount` must NEVER exceed the
+          // observed pgrep count. Equality at idle, `<=` always.
+          expect(snapshot.clientCount).toBeLessThanOrEqual(
+            observed.mcpGrandchildren.length,
+          );
+        } finally {
+          if (daemon) await daemon.dispose();
+          fs.rmSync(ws, { recursive: true, force: true });
+        }
+      }, 120_000);
     });
 
     describe('SSE backpressure (unit)', () => {

packages/cli/src/acp-integration/acpAgent.test.ts

@@ -2095,6 +2095,155 @@ describe('QwenAgent MCP SSE/HTTP support', () => {
     mockConnectionState.resolve();
     await agentPromise;
   });
+
+  // PR 14b: budget-event push channel. After codex review fix #2, the
+  // callback is wired via `Config.setMcpBudgetEventCallback` BEFORE
+  // `config.initialize()`, so MCP discovery (which can fire events
+  // synchronously in legacy blocking mode and races with background
+  // discovery in progressive mode) sees the callback wired from the
+  // first pass. The Config-level shim stashes the callback and applies
+  // it inside `createToolRegistry` to the freshly-constructed manager.
+  it('newSession wires Config.setMcpBudgetEventCallback BEFORE initialize() (codex fix #2)', async () => {
+    const sessionId = 'session-budget-events';
+    const innerConfig = await setupSessionMocks(sessionId);
+    // Stub `setMcpBudgetEventCallback` on the inner Config. The
+    // production path delegates the manager apply to Config; the test
+    // captures the callback at the Config boundary and verifies the
+    // ordering vs `initialize()`.
+    let capturedCallback:
+      | ((event: Record<string, unknown>) => void)
+      | undefined;
+    const callOrder: string[] = [];
+    (innerConfig as unknown as Record<string, unknown>)[
+      'setMcpBudgetEventCallback'
+    ] = vi.fn((cb: (event: Record<string, unknown>) => void) => {
+      callOrder.push('setMcpBudgetEventCallback');
+      capturedCallback = cb;
+    });
+    // Wrap `initialize` to record its position in `callOrder`. The
+    // critical invariant codex review fix #2 enforces: setter runs
+    // BEFORE initialize.
+    const originalInitialize = innerConfig.initialize;
+    innerConfig.initialize = vi.fn().mockImplementation(async () => {
+      callOrder.push('initialize');
+      return originalInitialize();
+    });
+
+    const agentPromise = runAcpAgent(
+      mockConfig,
+      makeSessionSettings(),
+      mockArgv,
+    );
+    await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined());
+
+    // Spy connection: only `extNotification` is exercised here, but
+    // the AgentSideConnection contract is wide. Stubbing only what the
+    // PR 14b code path touches keeps the test focused.
+    const extNotification = vi.fn().mockResolvedValue(undefined);
+    const fakeConn = {
+      get closed() {
+        return mockConnectionState.promise;
+      },
+      extNotification,
+    };
+    const agent = capturedAgentFactory!(
+      fakeConn as unknown as AgentSideConnectionLike,
+    ) as AgentLike;
+
+    await agent.newSession({ cwd: '/tmp', mcpServers: [] });
+
+    // Strict ordering invariant — codex review fix #2.
+    expect(callOrder).toEqual(['setMcpBudgetEventCallback', 'initialize']);
+    expect(typeof capturedCallback).toBe('function');
+
+    // Fire a synthetic budget_warning through the captured callback —
+    // the wired extNotification must receive the same shape with
+    // `sessionId` inserted and `v: 1` envelope.
+    const warningEvent = {
+      kind: 'budget_warning' as const,
+      liveCount: 4,
+      reservedCount: 4,
+      budget: 4,
+      thresholdRatio: 0.75 as const,
+      mode: 'warn' as const,
+    };
+    capturedCallback!(warningEvent);
+
+    expect(extNotification).toHaveBeenCalledTimes(1);
+    expect(extNotification).toHaveBeenCalledWith(
+      'qwen/notify/session/mcp-budget-event',
+      {
+        v: 1,
+        sessionId,
+        ...warningEvent,
+      },
+    );
+
+    // Fire a refused_batch through the same callback — same routing,
+    // discriminated union shape preserved verbatim.
+    const refusedEvent = {
+      kind: 'refused_batch' as const,
+      refusedServers: [
+        { name: 'b', transport: 'stdio', reason: 'budget_exhausted' },
+      ],
+      budget: 1,
+      liveCount: 1,
+      reservedCount: 1,
+      mode: 'enforce' as const,
+    };
+    capturedCallback!(refusedEvent);
+
+    expect(extNotification).toHaveBeenCalledTimes(2);
+    expect(extNotification).toHaveBeenLastCalledWith(
+      'qwen/notify/session/mcp-budget-event',
+      {
+        v: 1,
+        sessionId,
+        ...refusedEvent,
+      },
+    );
+
+    mockConnectionState.resolve();
+    await agentPromise;
+  });
+
+  it('newSession is a no-op for budget wiring when setMcpBudgetEventCallback is absent (defensive)', async () => {
+    // Codex review fix #2: the wiring path now goes through
+    // `Config.setMcpBudgetEventCallback`, not the manager directly.
+    // Older / stubbed `Config` shapes may omit it; the `typeof check`
+    // in newSessionConfig keeps the absence silent.
+    const innerConfig = await setupSessionMocks('session-no-cb-setter');
+    // `setupSessionMocks`/`makeInnerConfig` returns a Config without
+    // `setMcpBudgetEventCallback` defined — that's the defensive case.
+
+    const agentPromise = runAcpAgent(
+      mockConfig,
+      makeSessionSettings(),
+      mockArgv,
+    );
+    await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined());
+
+    const extNotification = vi.fn().mockResolvedValue(undefined);
+    const agent = capturedAgentFactory!({
+      get closed() {
+        return mockConnectionState.promise;
+      },
+      extNotification,
+    } as unknown as AgentSideConnectionLike) as AgentLike;
+
+    await agent.newSession({ cwd: '/tmp', mcpServers: [] });
+
+    // No setter on Config → no wiring → no extNotification fires.
+    expect(
+      (innerConfig as unknown as Record<string, unknown>)[
+        'setMcpBudgetEventCallback'
+      ],
+    ).toBeUndefined();
+    expect(extNotification).not.toHaveBeenCalled();
+
+    mockConnectionState.resolve();
+    await agentPromise;
+  });
 });
 
 // Regression coverage for the MR-review finding that ACP renameSession

packages/cli/src/acp-integration/acpAgent.ts

@@ -1812,6 +1812,66 @@ class QwenAgent implements Agent {
         projectHooks: this.settings.getProjectHooks(),
       },
     );
+    // PR 14b fix #2 (codex review round 1): register the MCP guardrail
+    // budget-event callback BEFORE `config.initialize()`. Pre-fix the
+    // registration ran AFTER initialize, which (a) missed end-of-pass
+    // events under `QWEN_CODE_LEGACY_MCP_BLOCKING=1` (synchronous
+    // discovery completes inside initialize, before our setter runs)
+    // and (b) raced against background-discovery completion under the
+    // default progressive mode. `Config.setMcpBudgetEventCallback`
+    // stashes the callback and `createToolRegistry` applies it to the
+    // manager BEFORE `discoverAllTools` / `startMcpDiscoveryInBackground`
+    // fires, closing both windows.
+    //
+    // sessionId source: `config.getSessionId()` reads the Config's own
+    // session id (auto-assigned via `randomUUID()` in the Config
+    // constructor when no override is passed — see `config.ts:849`),
+    // so the value is available immediately after `loadCliConfig`
+    // returns. The closure pins it for the manager's whole lifetime.
+    //
+    // Defensive `typeof` checks tolerate stub Configs / ToolRegistries
+    // in older tests (older fixtures may omit `setMcpBudgetEventCallback`
+    // or `getSessionId`).
+    const wiredSessionId =
+      typeof config.getSessionId === 'function'
+        ? config.getSessionId()
+        : undefined;
+    if (
+      typeof config.setMcpBudgetEventCallback === 'function' &&
+      wiredSessionId !== undefined
+    ) {
+      const sid = wiredSessionId;
+      config.setMcpBudgetEventCallback((event) => {
+        // Fire-and-forget: `extNotification` returns Promise<void> but
+        // the manager's call site doesn't await. `.catch` suppresses
+        // unhandled rejections — a mid-flight ACP disconnect would
+        // otherwise crash the child. Snapshot still carries the state
+        // for clients that reconnect.
+        //
+        // PR 14b fix (codex round 3 — DeepSeek): pre-fix the catch
+        // handler was `() => {}`, silently dropping every error
+        // including "real" ones (serialization bugs, protocol
+        // violations) — operators had no debug trail. Now logs at
+        // `debug` level: ACP channel closure during shutdown is the
+        // expected case and would spam at higher levels, but `debug`
+        // is opt-in so when an oncall engineer DOES turn it on for
+        // an MCP guardrail incident, they see exactly which event
+        // dropped and why.
+        void this.connection
+          .extNotification('qwen/notify/session/mcp-budget-event', {
+            v: 1,
+            sessionId: sid,
+            ...event,
+          })
+          .catch((err: unknown) => {
+            debugLogger.debug(
+              `MCP budget extNotification dropped ` +
+                `(session=${sid}, kind=${event.kind}): ` +
+                `${err instanceof Error ? err.message : String(err)}`,
+            );
+          });
+      });
+    }
     await config.initialize();
     // Same reasoning as the top-level runAcpAgent path: ACP feeds session
     // messages to the model immediately, so we cannot return a Config whose

packages/cli/src/serve/capabilities.ts

@@ -96,6 +96,18 @@ export const SERVE_CAPABILITY_REGISTRY = {
   // `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 14b. Daemon emits typed push events for MCP budget
+  // state crossings: `mcp_budget_warning` (synthetic, fires once per
+  // upward 75% crossing with hysteresis re-arm at 37.5%) and
+  // `mcp_child_refused_batch` (coalesced, one per discovery pass /
+  // length-1 per readResource refusal, only in `enforce` mode). SDK
+  // reducer narrows both via `KnownDaemonEvent` (`DaemonSessionViewState`
+  // exposes `mcpBudgetWarningCount`, `lastMcpBudgetWarning`,
+  // `mcpChildRefusedBatchCount`, `lastMcpChildRefusedBatch`). Always-on once
+  // PR 14b lands; orthogonal to `mcp_guardrails` (the snapshot
+  // surface). Listed alongside `mcp_guardrails` to keep the MCP-related
+  // tags grouped.
+  mcp_guardrail_events: { since: 'v1' },
   // Issue #4175 PR 19. Daemon supports the read-only workspace file
   // surface: `GET /file`, `GET /list`, `GET /glob`, `GET /stat`. The
   // four routes are gated as a single feature because they share the