feat(cli): Add serve env isolation and total admission (#6416)

* feat(cli): add serve env isolation and total admission

Add runtime-local serve env snapshots, explicit env injection for low-cost workspace-scoped consumers, and sourceEnv support for ACP child spawn.

Add a daemon-wide maxTotalSessions admission reservation hook for fresh session creation while keeping multi-workspace sessions gated.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* codex: address PR review feedback (#6416)

Reject fractional maxTotalSessions values so the daemon-wide session cap remains an integer count and matches the documented limit semantics.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): address PR review feedback (#6416)

Always pass the runtime env to A2UI stdio transports, keep daemon runtime env metadata coherent after env reload fallback, and tighten total-admission coverage.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): address total admission review feedback (#6416)

Add retryable ACP error data for total session limits, log total-admission REST rejections, and keep session-limit response scopes explicit.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): address env review feedback (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): restore scheduled task serve deps (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): isolate runtime env reload base (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* codex: address PR review feedback (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* codex: address PR review feedback (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* codex: address PR review feedback (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* codex: address PR review feedback (#6416)

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* codex: fix CI failure on PR #6416

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): address daemon admission review feedback

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): address runtime env review feedback

Scrub daemon bearer tokens from A2UI stdio MCP environments and prune reload-owned keys from the daemon runtime base before rebuilding runtime env snapshots.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): preserve daemon env base on reload

Keep runtime env rebuilds anchored to the boot-time daemon base snapshot, preventing reload-owned key pruning from dropping valid shell-exported values. Also carry env file read failure details into runtime metadata and daemon logs.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* fix(cli): satisfy env metadata lint rules

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

---------

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye 2026-07-08 08:52:36 +08:00 committed by GitHub
parent 394c1a289e
commit 27f8f2c95d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
45 changed files with 2422 additions and 299 deletions

View file

@ -2,17 +2,17 @@
## Summary
This document records the Phase 2a foundation contract for issue #6378 after
the Phase 1 `WorkspaceRegistry` PR. The current implementation batch combines
the Phase 1 repeated `--workspace` follow-up, the Phase 2a prep guardrails, and
the first internal registry/runtime contract needed by later multi-workspace
session work.
This document records the Phase 2a contract for issue #6378 after the Phase 1
`WorkspaceRegistry` PR and the Phase 2a foundation PR. Phase 2a is now split
into two implementation PRs: PR 1 lands env isolation and total-admission
guardrails while multi-workspace remains gated; PR 2 will wire non-primary live
session dispatch and publish the additive capabilities/status schema.
Phase 2a remains sessions-only. It does not add plural routes, a
`WorkspaceDaemonClient`, workspace-qualified ACP/WebSocket, file, memory, MCP,
settings, voice, channel-worker migration, env overlays, total-session
admission, capabilities `workspaces[]`, `multi_workspace_sessions`, route
dispatch, or non-primary runtime construction in this foundation batch.
settings, voice, or channel-worker migration. PR 1 does not add capabilities
`workspaces[]`, `multi_workspace_sessions`, route dispatch, or non-primary
runtime construction.
## Foundation Contract
@ -38,8 +38,10 @@ Phase 2a work:
- `primary`: true for the primary runtime.
- `trusted`: boot-time trust metadata; direct `createServeApp` fallback remains
false unless production passes an explicit trusted value.
- `env`: metadata only. This foundation batch records parent-process mode and
empty overlay keys; it does not compute runtime-local env overlays.
- `env`: runtime-local env source metadata. In single-workspace production,
the primary runtime now receives a computed effective env snapshot and a
mutable env source that can be refreshed after daemon env reload. Direct
`createServeApp` fallback remains parent-process metadata.
The internal `WorkspaceRegistry` supports exact cwd lookup, exact id lookup,
`resolveWorkspaceCwd(undefined)` primary fallback, and live session owner
@ -94,14 +96,35 @@ after tests prove they depend solely on the owning live bridge.
- Fail closed if more than one runtime reports the same live session id.
- Keep non-primary session listing live-only unless persisted entries are
explicitly marked non-resumable.
- Add runtime-local env overlays before non-primary child spawn.
- Add `maxTotalSessions` at the bridge fresh-creation seam so REST and primary
`/acp` cannot bypass it, while attach still bypasses admission.
- Publish `workspaces[]`, total limits, and `multi_workspace_sessions` only in
the final ungate PR.
- Reuse PR 1 runtime-local env overlays before non-primary child spawn.
- Reuse PR 1 `maxTotalSessions` admission at every future fresh-creation seam
so REST and primary `/acp` cannot bypass it, while attach still bypasses
admission.
- Publish `workspaces[]` and `multi_workspace_sessions` only in PR 2 when the
live session dispatch loop is complete.
- Update SDK capability types when the additive capabilities schema ships, but
do not add a workspace client in Phase 2a.
## PR 1 Guardrails
- Runtime env is computed from daemon base env plus workspace `.env`, settings
env, and Cloud Shell defaults without mutating parent `process.env` during
runtime initialization.
- The env helper intentionally does not virtualize `QWEN_HOME`, Storage, or
global config routing. Those remain daemon boot/base-env responsibilities.
- ACP child spawn accepts an explicit `sourceEnv`, and low-cost
workspace-scoped status/config readers use injected env instead of direct
`process.env` reads.
- `maxTotalSessions` is an optional daemon-wide fresh-session cap. It covers
spawn, persisted load/resume restore, and branch/fork session creation;
attach bypasses it.
- The bridge admission seam is a synchronous reservation hook. Failed fresh
creation releases the reservation, preventing concurrent oversell across
runtimes once non-primary bridges exist.
- `/daemon/status.limits.maxTotalSessions` is additive. `/capabilities` and SDK
capability types remain unchanged until PR 2 ungates multi-workspace
sessions.
## Audit Decisions
- The foundation PR must not create non-primary runtimes or relax any REST

View file

@ -91,10 +91,13 @@ Use this to detect mismatch pre-flight: read `workspaceCwd` off `/capabilities`
{
"error": "Session limit reached (20)",
"code": "session_limit_exceeded",
"limit": 20
"limit": 20,
"scope": "workspace"
}
```
When `--max-total-sessions` rejects a fresh session, the same response shape is returned with `"scope": "total"`.
Attaches to existing sessions are NOT counted toward the cap, so an idle daemon's reconnects keep working even when at-capacity.
`RestoreInProgressError` — only emitted by `POST /session/:id/load` and `POST /session/:id/resume` — returns `409` with a `Retry-After: 5` header (matching `session_limit_exceeded`) and:
@ -293,6 +296,7 @@ Response shape:
},
"limits": {
"maxSessions": 20,
"maxTotalSessions": null,
"maxPendingPromptsPerSession": 5,
"listenerMaxConnections": 256,
"eventRingSize": 8000,
@ -364,6 +368,8 @@ runtime routes return `503`.
`runtime.activity` reports daemon-wide prompt activity. `activePrompts` counts sessions with an in-flight prompt. `pendingPrompts` counts all accepted prompts that have not settled yet, including the running prompt and FIFO-waiting prompts. `queuedPrompts` counts FIFO-waiting prompts that have been accepted but not dispatched. `lastActivityAt` is the ISO 8601 timestamp of the last prompt start/end or session spawn; `null` when the daemon has never processed any activity since boot. `idleSinceMs` is computed from `lastActivityAt` at response generation time.
`limits.maxTotalSessions` is additive. `null` means the daemon-wide fresh-session cap is disabled. When set, it limits fresh session creation across the daemon and reports total-limit failures with the existing `session_limit_exceeded` error shape plus `scope: "total"`. It does not change `/capabilities`, does not advertise `workspaces[]`, and does not enable multi-workspace routing by itself.
`runtime.channel.live` reports the ACP bridge channel inside the daemon. It is
not the channel-adapter worker. Daemon-managed channels use
`runtime.channelWorker`, whose `state` is one of `disabled`, `starting`,

View file

@ -81,7 +81,7 @@ curl http://127.0.0.1:4170/daemon/status
```
The `workspaceCwd` field surfaces the bound workspace so clients can pre-flight check + omit `cwd` on `POST /session`.
The `limits.maxPendingPromptsPerSession` field advertises the active per-session prompt admission cap; `null` means the cap is disabled.
The `limits.maxPendingPromptsPerSession` field advertises the active per-session prompt admission cap; `null` means the cap is disabled. `limits.maxTotalSessions` advertises the optional daemon-wide fresh-session cap; `null` means unlimited.
### Run channels from the daemon
@ -303,6 +303,7 @@ Notes:
| `--tls-cert <path>` | — | Path to a PEM certificate file. Serve over **HTTPS** instead of HTTP. Must be paired with `--tls-key` (boot fails if only one is given). Unlocks secure-context browser APIs — voice input (`getUserMedia`), WebRTC — over a LAN IP, which browsers otherwise block on plain `http://`. TLS termination only; no auto-generation / ACME. See [HTTPS / TLS](#https--tls-for-mobile--cross-device-access) below. |
| `--tls-key <path>` | — | Path to a PEM private key file. Must be paired with `--tls-cert`. |
| `--max-sessions <n>` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~3050 MB per session). |
| `--max-total-sessions <n>` | unlimited | Optional non-negative integer daemon-wide cap on fresh session creation across the runtime. It applies to new child sessions, session restore, and branch/fork-created sessions; attaching to an existing live session does not consume a slot. Set to `0` or omit the flag for unlimited. This is a guardrail for future multi-workspace sessions and does not enable multi-workspace serving by itself. |
| `--max-pending-prompts-per-session <n>` | `5` | Per-session cap on prompts accepted by `POST /session/:id/prompt` but not yet settled, including queued prompts and the active prompt. The bridge rejects overflow synchronously with `503`, `Retry-After: 5`, and `code: "prompt_queue_full"` before returning a `promptId`. Set to `0` to disable. `branchSession` serializes on the same FIFO but does not count against this prompt cap. |
| `--workspace <path>` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. |
| `--channel <name\|all>` | — | Experimental daemon-managed channel worker. Repeat the flag to select multiple configured channels, or pass `all` to start every configured channel. `all` cannot be combined with named channels. Selected channel `cwd` values must resolve to the daemon workspace. The worker is owned by `qwen serve`; stop the daemon to stop serve-managed channels. |
@ -315,7 +316,7 @@ Notes:
| `--web` / `--no-web` | `true` | Serve the built Web Shell SPA at the daemon root (`GET /`, `/assets/*`, and SPA deep-link fallback). The static shell is registered **before** the bearer-auth gate — a browser can't attach a token to a `<script>` subresource or an address-bar navigation, the shell carries no secrets, and every API route stays token-gated regardless. On non-loopback binds a one-line stderr warning notes the UI is reachable without auth. Use `--no-web` for an API-only daemon. No effect when the build omits the Web Shell assets (the daemon logs a breadcrumb and runs API-only). |
| `--open` | `false` | After the listener is up, open the Web Shell in your default browser at the daemon URL (with `#token=` appended as a URL fragment when a token is configured — a fragment is never sent to the server, keeping the token out of access logs and Referer headers). No-op with `--no-web`, or in headless / CI / SSH environments where no browser is available. |
> **Sizing the load knobs.** `--max-sessions` is the **new-child** cap.
> **Sizing the load knobs.** `--max-sessions` is the per-workspace **new-child** cap. `--max-total-sessions`, when set, is the daemon-wide fresh-session cap.
> Three other layers also limit load — when sizing for a high-concurrency
> deployment, tune them together:
>
@ -327,11 +328,16 @@ Notes:
> - **per-session prompt admissions**:
> `--max-pending-prompts-per-session=5` bounds queued + active prompts
> accepted for one session. Overflow gets `503` with `Retry-After: 5`.
> - **daemon-wide fresh sessions**: `--max-total-sessions=N` bounds fresh
> session creation across the daemon. Overflow gets the same
> `session_limit_exceeded` shape with `scope: "total"`.
> - **per-subscriber backlog**: a 256-frame queue per SSE client; an
> over-capacity client gets a terminal `client_evicted` frame and is
> closed (one slow consumer can't pin the daemon).
>
> These caps interact: `--max-sessions × 64 subscribers × 256 frames`
> These caps interact: the lower of `--max-sessions` and
> `--max-total-sessions` bounds fresh sessions in today's single-workspace
> daemon. `--max-sessions × 64 subscribers × 256 frames`
> is the worst-case in-flight memory at the EventBus layer, while
> `--max-sessions × --max-pending-prompts-per-session` bounds accepted
> prompt work at the admission layer. Default sizing assumes single-user /

View file

@ -37,6 +37,7 @@ import {
SessionShellDisabledError,
SessionBusyError,
SessionNotFoundError,
TotalSessionLimitExceededError,
WorkspaceMismatchError,
} from './bridgeErrors.js';
import { MAX_WORKSPACE_PATH_LENGTH } from './workspacePaths.js';
@ -9468,6 +9469,259 @@ describe('createAcpSessionBridge', () => {
await bridge.shutdown();
});
it('calls freshSessionAdmission for fresh spawns and releases after registration', async () => {
const releases: string[] = [];
const contexts: Array<{ operation: string; workspaceCwd: string }> = [];
const bridge = makeBridge({
channelFactory: async () => makeChannel().channel,
sessionScope: 'thread',
freshSessionAdmission: (context) => {
contexts.push(context);
return {
release: () => releases.push(context.operation),
};
},
});
await bridge.spawnOrAttach({ workspaceCwd: WS_A });
expect(contexts).toMatchObject([
{ operation: 'spawn', workspaceCwd: WS_A },
]);
expect(releases).toEqual(['spawn']);
await bridge.shutdown();
});
it('releases fresh spawn admission once the session is live', async () => {
const modelSwitchGate = deferred<unknown>();
let releaseCount = 0;
const factory: ChannelFactory = async () => {
const { clientStream, agentStream } = createInMemoryChannel();
const fakeAgent = new FakeAgent();
const augmented = new Proxy(fakeAgent, {
get(target, prop) {
if (prop === 'unstable_setSessionModel') {
return async () => modelSwitchGate.promise;
}
// eslint-disable-next-line @typescript-eslint/no-explicit-any
return (target as any)[prop];
},
});
new AgentSideConnection(() => augmented as Agent, agentStream);
return {
stream: clientStream,
exited: new Promise<
| { exitCode: number | null; signalCode: NodeJS.Signals | null }
| undefined
>(() => {}),
kill: async () => {},
killSync: () => {},
};
};
const bridge = makeBridge({
channelFactory: factory,
sessionScope: 'thread',
freshSessionAdmission: () => ({
release: () => {
releaseCount++;
},
}),
});
const spawn = bridge.spawnOrAttach({
workspaceCwd: WS_A,
modelServiceId: 'qwen3-coder',
});
await vi.waitFor(() => {
expect(bridge.sessionCount).toBe(1);
expect(releaseCount).toBe(1);
});
modelSwitchGate.resolve({});
await spawn;
expect(releaseCount).toBe(1);
await bridge.shutdown();
});
it('reports freshSessionAdmission release failures without failing fresh spawns', async () => {
const diagnostics: Array<{ line: string; level?: string }> = [];
const bridge = makeBridge({
channelFactory: async () => makeChannel().channel,
sessionScope: 'thread',
freshSessionAdmission: () => ({
release: () => {
throw new Error('release broken');
},
}),
onDiagnosticLine: (line, level) => diagnostics.push({ line, level }),
});
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
expect(session.sessionId).toBeTruthy();
expect(diagnostics).toEqual([
{
line: 'qwen serve: fresh session admission release failed: release broken',
level: 'warn',
},
]);
await bridge.shutdown();
});
it('does not call freshSessionAdmission for single-scope attaches', async () => {
const freshSessionAdmission = vi.fn(() => ({
release: vi.fn(),
}));
const bridge = makeBridge({
channelFactory: async () => makeChannel().channel,
sessionScope: 'single',
freshSessionAdmission,
});
const spawned = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
freshSessionAdmission.mockClear();
const attached = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
expect(attached.sessionId).toBe(spawned.sessionId);
expect(attached.attached).toBe(true);
expect(freshSessionAdmission).not.toHaveBeenCalled();
await bridge.shutdown();
});
it('fails fresh spawns closed when freshSessionAdmission rejects', async () => {
const bridge = makeBridge({
channelFactory: async () => makeChannel().channel,
sessionScope: 'thread',
freshSessionAdmission: () => {
throw new TotalSessionLimitExceededError(1);
},
});
await expect(
bridge.spawnOrAttach({ workspaceCwd: WS_A }),
).rejects.toMatchObject({
name: 'TotalSessionLimitExceededError',
limit: 1,
scope: 'total',
});
expect(bridge.sessionCount).toBe(0);
await bridge.shutdown();
});
it('calls freshSessionAdmission for load and resume restores', async () => {
const contexts: Array<{ operation: string; workspaceCwd: string }> = [];
const releases: string[] = [];
const bridge = makeBridge({
channelFactory: async () => makeChannel().channel,
freshSessionAdmission: (context) => {
contexts.push(context);
return {
release: () => releases.push(context.operation),
};
},
});
await bridge.loadSession({
sessionId: 'load-1',
workspaceCwd: WS_A,
});
await bridge.resumeSession({
sessionId: 'resume-1',
workspaceCwd: WS_A,
});
expect(contexts).toMatchObject([
{ operation: 'load', workspaceCwd: WS_A },
{ operation: 'resume', workspaceCwd: WS_A },
]);
expect(releases).toEqual(['load', 'resume']);
await bridge.shutdown();
});
it('reserves branchSession before branch extMethod and skips a second restore reservation', async () => {
const contexts: Array<{
operation: string;
workspaceCwd: string;
sourceSessionId?: string;
}> = [];
const releases: string[] = [];
let branchExtMethodSawReservation = false;
const bridge = makeBridge({
channelFactory: async () =>
makeChannel({
extMethodImpl: (method) => {
if (method !== 'qwen/control/session/branch') return {};
branchExtMethodSawReservation =
contexts.at(-1)?.operation === 'branch';
return { newSessionId: 'branch-1', title: 'Branch 1' };
},
loadSessionImpl: () => ({}),
}).channel,
freshSessionAdmission: (context) => {
contexts.push(context);
return {
release: () => releases.push(context.operation),
};
},
});
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
contexts.length = 0;
releases.length = 0;
const branch = await bridge.branchSession(session.sessionId, {
name: 'Branch 1',
});
expect(branchExtMethodSawReservation).toBe(true);
expect(branch).toMatchObject({ sessionId: 'branch-1' });
expect(contexts).toEqual([
{
operation: 'branch',
workspaceCwd: WS_A,
sourceSessionId: session.sessionId,
},
]);
expect(releases).toEqual(['branch']);
await bridge.shutdown();
});
it('releases branchSession admission when the branch extMethod fails', async () => {
const contexts: Array<{ operation: string; workspaceCwd: string }> = [];
const releases: string[] = [];
const bridge = makeBridge({
channelFactory: async () =>
makeChannel({
extMethodImpl: (method) => {
if (method === 'qwen/control/session/branch') {
throw new Error('branch failed');
}
return {};
},
}).channel,
freshSessionAdmission: (context) => {
contexts.push(context);
return {
release: () => releases.push(context.operation),
};
},
});
const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A });
contexts.length = 0;
releases.length = 0;
await expect(
bridge.branchSession(session.sessionId, {
name: 'Branch 1',
}),
).rejects.toThrow();
expect(contexts).toMatchObject([
{ operation: 'branch', workspaceCwd: WS_A },
]);
expect(releases).toEqual(['branch']);
await bridge.shutdown();
});
it('Stage 1.5: killSession on one of N sessions does NOT kill the shared channel', async () => {
// Counterpart guarantee: tearing down one session must not take
// its siblings with it. The channel stays alive while

View file

@ -100,7 +100,12 @@ import type {
BridgeWorkspaceMemoryRememberRequest,
BridgeWorkspaceMemoryRememberResult,
} from './bridgeTypes.js';
import type { BridgeOptions, BridgeTelemetry } from './bridgeOptions.js';
import type {
BridgeFreshSessionAdmissionContext,
BridgeFreshSessionReservation,
BridgeOptions,
BridgeTelemetry,
} from './bridgeOptions.js';
import { MCP_RESTART_SERVER_DEADLINE_MS } from './mcpTimeouts.js';
import { defaultSpawnChannelFactory } from './spawnChannel.js';
import { writeStderrLine } from './internal/stderrLine.js';
@ -1084,6 +1089,25 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
} else {
maxSessions = opts.maxSessions;
}
const reserveFreshSession = (
context: BridgeFreshSessionAdmissionContext,
): BridgeFreshSessionReservation | undefined =>
opts.freshSessionAdmission?.(context);
const releaseFreshSessionReservation = (
reservation: BridgeFreshSessionReservation | undefined,
): void => {
if (!reservation) return;
try {
reservation.release();
} catch (err) {
opts.onDiagnosticLine?.(
`qwen serve: fresh session admission release failed: ${
err instanceof Error ? err.message : String(err)
}`,
'warn',
);
}
};
if (defaultSessionScope !== 'single' && defaultSessionScope !== 'thread') {
throw new TypeError(
`Invalid sessionScope: ${JSON.stringify(defaultSessionScope)}. ` +
@ -1870,6 +1894,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
modelServiceId: string | undefined,
effectiveScope: 'single' | 'thread',
requestedClientId?: string,
onSessionRegistered?: () => void,
): Promise<BridgeSession> {
// Get-or-create the daemon's single channel, then call
// `connection.newSession()` on it. Sessions share the child's
@ -1948,6 +1973,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
boundWorkspace,
);
sessionRegistered = true;
onSessionRegistered?.();
seedSnapshotCaches(entry, newSessionResp);
const clientId = registerClient(entry, requestedClientId);
// `defaultEntry` is the single-scope attach target — only sessions
@ -2767,6 +2793,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
async function restoreSession(
action: 'load' | 'resume',
req: BridgeRestoreSessionRequest,
options: { skipFreshSessionAdmission?: boolean } = {},
): Promise<BridgeRestoredSession> {
if (shuttingDown) {
throw new Error('AcpSessionBridge is shutting down');
@ -2864,6 +2891,20 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
// doc comment). Mutated synchronously by the coalesce branch above
// and read once by the IIFE when seeding `entry.attachCount`.
const coalesceState = { count: 0 };
const admission =
options.skipFreshSessionAdmission === true
? undefined
: reserveFreshSession({
operation: action,
workspaceCwd: workspaceKey,
sessionId: req.sessionId,
});
let admissionReleased = false;
const releaseAdmissionOnce = () => {
if (admissionReleased) return;
admissionReleased = true;
releaseFreshSessionReservation(admission);
};
const promise = (async (): Promise<BridgeRestoredSession> => {
pendingRestoreEvents.set(req.sessionId, restoreEvents);
ci = await ensureChannel();
@ -2998,6 +3039,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
restoreEvents,
{ drainEarlyEvents: replayUpdates.length === 0 },
);
releaseAdmissionOnce();
entry.restoreState = state;
if (replayPartial === true) {
entry.restoreReplayPartial = true;
@ -3037,6 +3079,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
...replayFieldsFor(entry, action),
};
})().finally(async () => {
releaseAdmissionOnce();
ci?.pendingRestoreIds.delete(req.sessionId);
// Pair with `markRestoreInFlight`. Once the IIFE settles, either
// `createSessionEntry` ran (`drainEarlyEvents` already cleared
@ -3452,7 +3495,22 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
throw new SessionLimitExceededError(maxSessions);
}
const promise = doSpawn(req.modelServiceId, effectiveScope, req.clientId);
const admission = reserveFreshSession({
operation: 'spawn',
workspaceCwd: workspaceKey,
});
let admissionReleased = false;
const releaseAdmissionOnce = () => {
if (admissionReleased) return;
admissionReleased = true;
releaseFreshSessionReservation(admission);
};
const promise = doSpawn(
req.modelServiceId,
effectiveScope,
req.clientId,
releaseAdmissionOnce,
);
// Track in-flight spawns regardless of scope. Under `single`
// this also serves the coalescing path above (a parallel
// `spawnOrAttach` finds the entry and waits for the same
@ -3473,6 +3531,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
try {
return await promise;
} finally {
releaseAdmissionOnce();
// Always clear the in-flight slot whether the spawn resolved
// or rejected — leaving a rejected promise behind would
// poison every future coalescing-path call for this
@ -4185,77 +4244,99 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge {
throw new SessionLimitExceededError(maxSessions);
}
const ci = await ensureChannel();
const result = (await withTimeout(
ci.connection.extMethod(SERVE_CONTROL_EXT_METHODS.sessionBranch, {
sessionId,
cwd: boundWorkspace,
name: req.name,
}),
initTimeoutMs,
'branchSession',
)) as { newSessionId: string; title?: string; displayName?: string };
if (!result || typeof result.newSessionId !== 'string') {
throw new Error(
`branchSession: agent returned invalid response: ${JSON.stringify(result)}`,
);
}
const rawBranchName = result.displayName ?? result.title;
const branchDisplayName =
typeof rawBranchName === 'string'
? rawBranchName
: result.newSessionId.slice(0, 8);
let restored;
const admission = reserveFreshSession({
operation: 'branch',
workspaceCwd: boundWorkspace,
sourceSessionId: sessionId,
});
let admissionReleased = false;
const releaseAdmissionOnce = () => {
if (admissionReleased) return;
admissionReleased = true;
releaseFreshSessionReservation(admission);
};
try {
restored = await restoreSession('load', {
sessionId: result.newSessionId,
workspaceCwd: boundWorkspace,
clientId: context?.clientId,
});
} catch (restoreErr) {
writeStderrLine(
`qwen serve: branchSession load failed for ${result.newSessionId}, attempting cleanup...`,
);
try {
await ci.connection.extMethod(
SERVE_CONTROL_EXT_METHODS.sessionClose,
{ sessionId: result.newSessionId, cwd: boundWorkspace },
);
} catch (cleanupErr) {
writeStderrLine(
`qwen serve: branchSession cleanup of ${result.newSessionId} failed: ${cleanupErr instanceof Error ? cleanupErr.message : cleanupErr}`,
const ci = await ensureChannel();
const result = (await withTimeout(
ci.connection.extMethod(SERVE_CONTROL_EXT_METHODS.sessionBranch, {
sessionId,
cwd: boundWorkspace,
name: req.name,
}),
initTimeoutMs,
'branchSession',
)) as { newSessionId: string; title?: string; displayName?: string };
if (!result || typeof result.newSessionId !== 'string') {
throw new Error(
`branchSession: agent returned invalid response: ${JSON.stringify(result)}`,
);
}
throw restoreErr;
const rawBranchName = result.displayName ?? result.title;
const branchDisplayName =
typeof rawBranchName === 'string'
? rawBranchName
: result.newSessionId.slice(0, 8);
let restored;
try {
restored = await restoreSession(
'load',
{
sessionId: result.newSessionId,
workspaceCwd: boundWorkspace,
clientId: context?.clientId,
},
{
skipFreshSessionAdmission: true,
},
);
releaseAdmissionOnce();
} catch (restoreErr) {
writeStderrLine(
`qwen serve: branchSession load failed for ${result.newSessionId}, attempting cleanup...`,
);
try {
await ci.connection.extMethod(
SERVE_CONTROL_EXT_METHODS.sessionClose,
{ sessionId: result.newSessionId, cwd: boundWorkspace },
);
} catch (cleanupErr) {
writeStderrLine(
`qwen serve: branchSession cleanup of ${result.newSessionId} failed: ${cleanupErr instanceof Error ? cleanupErr.message : cleanupErr}`,
);
}
throw restoreErr;
}
const newEntry = byId.get(result.newSessionId);
if (newEntry) newEntry.displayName = branchDisplayName;
const eventData = {
sourceSessionId: sessionId,
newSessionId: result.newSessionId,
displayName: branchDisplayName,
};
const branchEnvelope = {
type: 'session_branched' as const,
data: eventData,
...(originatorClientId ? { originatorClientId } : {}),
};
// The branch announcement belongs to the new session only. Publishing
// it on the source session would persist in that session's replay ring.
newEntry?.events.publish(branchEnvelope);
return {
...restored,
displayName: branchDisplayName,
forkedFrom: {
sessionId,
displayName: entry.displayName ?? sessionId.slice(0, 8),
},
};
} finally {
releaseAdmissionOnce();
}
const newEntry = byId.get(result.newSessionId);
if (newEntry) newEntry.displayName = branchDisplayName;
const eventData = {
sourceSessionId: sessionId,
newSessionId: result.newSessionId,
displayName: branchDisplayName,
};
const branchEnvelope = {
type: 'session_branched' as const,
data: eventData,
...(originatorClientId ? { originatorClientId } : {}),
};
// The branch announcement belongs to the new session only. Publishing
// it on the source session would persist in that session's replay ring.
newEntry?.events.publish(branchEnvelope);
return {
...restored,
displayName: branchDisplayName,
forkedFrom: {
sessionId,
displayName: entry.displayName ?? sessionId.slice(0, 8),
},
};
});
entry.promptQueue = branchResult.then(
() => undefined,

View file

@ -163,6 +163,16 @@ export class SessionLimitExceededError extends Error {
}
}
export class TotalSessionLimitExceededError extends Error {
readonly limit: number;
readonly scope = 'total' as const;
constructor(limit: number) {
super(`Total session limit reached (${limit})`);
this.name = 'TotalSessionLimitExceededError';
this.limit = limit;
}
}
/**
* Thrown by `sendPrompt` when a session already has too many accepted
* prompts waiting or running. The REST route maps this to 503 with

View file

@ -33,6 +33,21 @@ export type DiagnosticLineSink = (
level?: 'info' | 'warn' | 'error',
) => void;
export interface BridgeFreshSessionAdmissionContext {
readonly operation: 'spawn' | 'load' | 'resume' | 'branch';
readonly workspaceCwd: string;
readonly sessionId?: string;
readonly sourceSessionId?: string;
}
export interface BridgeFreshSessionReservation {
release(): void;
}
export type BridgeFreshSessionAdmission = (
context: BridgeFreshSessionAdmissionContext,
) => BridgeFreshSessionReservation | undefined;
/**
* Optional injection seam for daemon-host-specific status cells
* `process.env` snapshots and the daemon-side preflight checks
@ -147,6 +162,12 @@ export interface BridgeOptions {
* `ServeOptions.maxSessions` for the rationale.
*/
maxSessions?: number;
/**
* Host-level admission hook for fresh session creation across runtimes.
* Must be synchronous so callers can reserve before any async child or ACP
* side effect starts. Attaches bypass this hook.
*/
freshSessionAdmission?: BridgeFreshSessionAdmission;
/**
* Per-session SSE replay ring depth. Sets `ringSize` on every
* `new EventBus(...)` the bridge constructs (both fresh sessions
@ -398,6 +419,4 @@ export interface BridgeOptions {
*/
export type ClientMcpMessageSender = (
serverName: string,
) =>
| ((payload: unknown) => Promise<unknown>)
| undefined;
) => ((payload: unknown) => Promise<unknown>) | undefined;

View file

@ -69,11 +69,15 @@ describe('createSpawnChannelFactory env policy', () => {
const originalArgv1 = process.argv[1];
let originalSimple: string | undefined;
let originalServerToken: string | undefined;
let originalCliEntry: string | undefined;
let originalRuntimeOnlyForTest: string | undefined;
beforeEach(() => {
mockSpawn.mockReset();
originalSimple = process.env['QWEN_CODE_SIMPLE'];
originalServerToken = process.env['QWEN_SERVER_TOKEN'];
originalCliEntry = process.env['QWEN_CLI_ENTRY'];
originalRuntimeOnlyForTest = process.env['RUNTIME_ONLY_FOR_TEST'];
process.argv[1] = '/tmp/qwen.js';
process.env['QWEN_CODE_SIMPLE'] = '1';
process.env['QWEN_SERVER_TOKEN'] = 'secret';
@ -91,6 +95,16 @@ describe('createSpawnChannelFactory env policy', () => {
} else {
process.env['QWEN_SERVER_TOKEN'] = originalServerToken;
}
if (originalCliEntry === undefined) {
delete process.env['QWEN_CLI_ENTRY'];
} else {
process.env['QWEN_CLI_ENTRY'] = originalCliEntry;
}
if (originalRuntimeOnlyForTest === undefined) {
delete process.env['RUNTIME_ONLY_FOR_TEST'];
} else {
process.env['RUNTIME_ONLY_FOR_TEST'] = originalRuntimeOnlyForTest;
}
});
it('scrubs daemon-only env vars from the spawned ACP child', async () => {
@ -122,6 +136,30 @@ describe('createSpawnChannelFactory env policy', () => {
expect(args?.slice(-2)).toEqual(['--acp', '--experimental-lsp']);
});
it('builds child env and cli entry from sourceEnv when provided', async () => {
mockSpawn.mockReturnValue(createFakeChildProcess());
process.env['QWEN_CLI_ENTRY'] = '/process/qwen.js';
process.env['RUNTIME_ONLY_FOR_TEST'] = 'from-process';
const factory = createSpawnChannelFactory({
sourceEnv: {
QWEN_CLI_ENTRY: '/runtime/qwen.js',
RUNTIME_ONLY_FOR_TEST: 'from-runtime',
QWEN_SERVER_TOKEN: 'runtime-secret',
},
});
await factory('/tmp/project');
const args = mockSpawn.mock.calls[0]?.[1] as string[] | undefined;
const spawnOptions = mockSpawn.mock.calls[0]?.[2] as
| { env?: NodeJS.ProcessEnv }
| undefined;
expect(args).toContain('/runtime/qwen.js');
expect(args).not.toContain('/process/qwen.js');
expect(spawnOptions?.env?.['RUNTIME_ONLY_FOR_TEST']).toBe('from-runtime');
expect(spawnOptions?.env).not.toHaveProperty('QWEN_SERVER_TOKEN');
});
it('threads NDJSON pipe hooks through daemon-side spawned channels', async () => {
const child = createFakeChildProcess();
mockSpawn.mockReturnValue(child);

View file

@ -104,6 +104,7 @@ export interface SpawnChannelFactoryOptions {
onDiagnosticLine?: (line: string, level?: 'info' | 'warn' | 'error') => void;
extraArgs?: string[];
pipeHooks?: NdJsonStreamHooks;
sourceEnv?: Readonly<NodeJS.ProcessEnv>;
}
/**
@ -119,12 +120,13 @@ export function createSpawnChannelFactory(
options: SpawnChannelFactoryOptions = {},
): ChannelFactory {
return async (workspaceCwd, childEnvOverrides) => {
const cliEntry = process.env['QWEN_CLI_ENTRY'] || process.argv[1];
const sourceEnv = options.sourceEnv ?? process.env;
const cliEntry = sourceEnv['QWEN_CLI_ENTRY'] || process.argv[1];
if (!cliEntry) {
throw new MissingCliEntryError();
}
const childEnv = scrubChildEnv(
process.env,
sourceEnv,
SCRUBBED_CHILD_ENV_KEYS,
childEnvOverrides,
);

View file

@ -57,6 +57,11 @@ describe('serve command args', () => {
expect(parsed['permission-response-timeout-ms']).toBe(60000);
});
it('parses --max-total-sessions as a number', () => {
const parsed = buildParser().parseSync('--max-total-sessions 42');
expect(parsed['max-total-sessions']).toBe(42);
});
it('leaves --permission-response-timeout-ms unset by default', () => {
const parsed = buildParser().parseSync('');
expect(parsed['permission-response-timeout-ms']).toBeUndefined();
@ -235,6 +240,19 @@ describe('serve rate limit env parsing', () => {
);
});
it('passes --max-total-sessions to runQwenServe', async () => {
mockRunQwenServe.mockResolvedValueOnce({
url: 'http://127.0.0.1:4170/',
webShellMounted: false,
});
await startServeHandlerWithArgs('--no-web --max-total-sessions 42');
expect(mockRunQwenServe).toHaveBeenCalledWith(
expect.objectContaining({ maxTotalSessions: 42 }),
);
});
it('passes --channel all as an all-channel selection', async () => {
mockRunQwenServe.mockResolvedValueOnce({
url: 'http://127.0.0.1:4170/',

View file

@ -96,6 +96,7 @@ interface ServeArgs {
hostname: string;
token?: string;
'max-sessions': number;
'max-total-sessions'?: number;
'max-pending-prompts-per-session': number;
'max-connections': number;
'event-ring-size': number;
@ -165,6 +166,12 @@ export const serveCommand: CommandModule<unknown, ServeArgs> = {
'Cap on concurrent live sessions. New spawn requests beyond this return 503; ' +
'attach to existing sessions still works. Set to 0 to disable.',
})
.option('max-total-sessions', {
type: 'number',
description:
'Non-negative integer cap on concurrent live sessions across all ' +
'workspace runtimes. Set to 0 to disable.',
})
.option('max-pending-prompts-per-session', {
type: 'number',
default: 5,
@ -542,6 +549,9 @@ export const serveCommand: CommandModule<unknown, ServeArgs> = {
token: argv.token,
mode: 'http-bridge',
maxSessions: argv['max-sessions'],
...(argv['max-total-sessions'] !== undefined
? { maxTotalSessions: argv['max-total-sessions'] }
: {}),
maxPendingPromptsPerSession,
maxConnections: argv['max-connections'],
eventRingSize: argv['event-ring-size'],

View file

@ -0,0 +1,192 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import * as fs from 'node:fs';
import * as os from 'node:os';
import * as path from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { buildRuntimeEnvironment, loadEnvironment } from './environment.js';
import type { Settings } from './settingsSchema.js';
const TRACKED_ENV = [
'CLOUD_SHELL',
'GOOGLE_CLOUD_PROJECT',
'RUNTIME_DOTENV',
'RUNTIME_EMPTY',
'RUNTIME_EXCLUDED',
'RUNTIME_PARENT',
'RUNTIME_SETTINGS',
'RUNTIME_SETTINGS_ONLY',
'BASH_ENV',
'NODE_OPTIONS',
'QWEN_HOME',
'QWEN_RUNTIME_DIR',
'QWEN_SERVER_TOKEN',
] as const;
let tmpDirs: string[] = [];
const previousEnv = new Map<string, string | undefined>();
function makeWorkspace(): string {
const dir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qwen-runtime-env-')),
);
tmpDirs.push(dir);
return dir;
}
function testSettings(partial: Partial<Settings>): Settings {
return partial as Settings;
}
beforeEach(() => {
previousEnv.clear();
for (const key of TRACKED_ENV) {
previousEnv.set(key, process.env[key]);
delete process.env[key];
}
});
afterEach(() => {
for (const key of TRACKED_ENV) {
const value = previousEnv.get(key);
if (value === undefined) {
delete process.env[key];
} else {
process.env[key] = value;
}
}
for (const dir of tmpDirs) {
fs.rmSync(dir, { recursive: true, force: true });
}
tmpDirs = [];
});
describe('buildRuntimeEnvironment', () => {
it('computes a runtime overlay without mutating process.env or base env', () => {
const workspace = makeWorkspace();
fs.writeFileSync(
path.join(workspace, '.env'),
[
'RUNTIME_DOTENV=from-dotenv',
'RUNTIME_PARENT=dotenv-loses',
'RUNTIME_EMPTY=from-dotenv-empty',
'RUNTIME_SETTINGS=dotenv-wins',
'RUNTIME_EXCLUDED=excluded',
'NODE_OPTIONS=--require ./bad.js',
'QWEN_SERVER_TOKEN=dotenv-token',
'QWEN_HOME=/tmp/ignored-qwen-home',
'',
].join('\n'),
);
const baseEnv: NodeJS.ProcessEnv = {
RUNTIME_PARENT: 'from-parent',
RUNTIME_EMPTY: '',
};
const snapshot = buildRuntimeEnvironment(
testSettings({
advanced: {
excludedEnvVars: ['RUNTIME_EXCLUDED', 'RUNTIME_SETTINGS_EXCLUDED'],
},
env: {
RUNTIME_SETTINGS: 'settings-loses',
RUNTIME_SETTINGS_ONLY: 'from-settings',
RUNTIME_SETTINGS_EXCLUDED: 'settings-excluded',
BASH_ENV: '/tmp/bad-profile',
QWEN_RUNTIME_DIR: '/tmp/ignored-runtime-dir',
},
}),
workspace,
baseEnv,
);
expect(snapshot.effectiveEnv['RUNTIME_DOTENV']).toBe('from-dotenv');
expect(snapshot.effectiveEnv['RUNTIME_PARENT']).toBe('from-parent');
expect(snapshot.effectiveEnv['RUNTIME_EMPTY']).toBe('from-dotenv-empty');
expect(snapshot.effectiveEnv['RUNTIME_SETTINGS']).toBe('dotenv-wins');
expect(snapshot.effectiveEnv['RUNTIME_SETTINGS_ONLY']).toBe(
'from-settings',
);
expect(snapshot.effectiveEnv['RUNTIME_EXCLUDED']).toBeUndefined();
expect(snapshot.effectiveEnv['RUNTIME_SETTINGS_EXCLUDED']).toBeUndefined();
expect(snapshot.effectiveEnv['NODE_OPTIONS']).toBeUndefined();
expect(snapshot.effectiveEnv['BASH_ENV']).toBeUndefined();
expect(snapshot.effectiveEnv['QWEN_SERVER_TOKEN']).toBeUndefined();
expect(snapshot.effectiveEnv['QWEN_HOME']).toBeUndefined();
expect(snapshot.effectiveEnv['QWEN_RUNTIME_DIR']).toBeUndefined();
expect(snapshot.overlayKeys).toEqual([
'RUNTIME_DOTENV',
'RUNTIME_EMPTY',
'RUNTIME_SETTINGS',
'RUNTIME_SETTINGS_ONLY',
]);
expect(snapshot.envFilePaths).toContain(path.join(workspace, '.env'));
expect(snapshot.envFileReadFailed).toBe(false);
expect(snapshot.envFileReadFailures).toEqual([]);
expect(baseEnv).toEqual({
RUNTIME_PARENT: 'from-parent',
RUNTIME_EMPTY: '',
});
expect(process.env['RUNTIME_DOTENV']).toBeUndefined();
expect(process.env['RUNTIME_SETTINGS_ONLY']).toBeUndefined();
});
it('applies Cloud Shell project defaults to the runtime env only', () => {
const workspace = makeWorkspace();
const snapshot = buildRuntimeEnvironment(testSettings({}), workspace, {
CLOUD_SHELL: 'true',
});
expect(snapshot.effectiveEnv['GOOGLE_CLOUD_PROJECT']).toBe(
'cloudshell-gca',
);
expect(snapshot.overlayKeys).toContain('GOOGLE_CLOUD_PROJECT');
expect(process.env['GOOGLE_CLOUD_PROJECT']).toBeUndefined();
});
it('surfaces env file read failures in the runtime snapshot', () => {
const workspace = makeWorkspace();
const envPath = path.join(workspace, '.env');
fs.mkdirSync(envPath);
const snapshot = buildRuntimeEnvironment(testSettings({}), workspace, {});
expect(snapshot.envFilePaths).toContain(envPath);
expect(snapshot.envFileReadFailed).toBe(true);
expect(snapshot.envFileReadFailures).toEqual([
expect.objectContaining({
path: envPath,
error: expect.any(String),
}),
]);
expect(snapshot.effectiveEnv['RUNTIME_DOTENV']).toBeUndefined();
});
});
describe('loadEnvironment', () => {
it('filters reload-excluded keys from settings.env on initial load', () => {
const workspace = makeWorkspace();
loadEnvironment(
testSettings({
env: {
RUNTIME_SETTINGS_ONLY: 'from-settings',
BASH_ENV: '/tmp/bad-profile',
NODE_OPTIONS: '--require ./bad.js',
QWEN_SERVER_TOKEN: 'bad-token',
},
}),
workspace,
);
expect(process.env['RUNTIME_SETTINGS_ONLY']).toBe('from-settings');
expect(process.env['BASH_ENV']).toBeUndefined();
expect(process.env['NODE_OPTIONS']).toBeUndefined();
expect(process.env['QWEN_SERVER_TOKEN']).toBeUndefined();
});
});

View file

@ -306,20 +306,157 @@ export function setUpCloudShellEnvironment(envFilePath: string | null): void {
}
}
function setUpCloudShellEnvironmentFromFiles(envFilePaths: string[]): void {
for (const envFilePath of envFilePaths) {
if (!fs.existsSync(envFilePath)) {
continue;
}
const envFileContent = fs.readFileSync(envFilePath);
const parsedEnv = dotenv.parse(envFileContent);
if (parsedEnv['GOOGLE_CLOUD_PROJECT']) {
process.env['GOOGLE_CLOUD_PROJECT'] = parsedEnv['GOOGLE_CLOUD_PROJECT'];
function setUpCloudShellEnvironmentInEnv(
env: NodeJS.ProcessEnv,
envFiles: readonly ParsedEnvFile[],
): void {
for (const envFile of envFiles) {
if (envFile.parsedEnv['GOOGLE_CLOUD_PROJECT']) {
env['GOOGLE_CLOUD_PROJECT'] = envFile.parsedEnv['GOOGLE_CLOUD_PROJECT'];
return;
}
}
process.env['GOOGLE_CLOUD_PROJECT'] = 'cloudshell-gca';
env['GOOGLE_CLOUD_PROJECT'] = 'cloudshell-gca';
}
interface ParsedEnvFile {
readonly parsedEnv: Record<string, string>;
readonly isHomeScopedEnvFile: boolean;
readonly isQwenScopedEnvFile: boolean;
}
interface ParsedEnvFilesResult {
readonly files: readonly ParsedEnvFile[];
readonly readFailed: boolean;
readonly readFailures: readonly EnvFileReadFailure[];
}
export interface EnvFileReadFailure {
readonly path: string;
readonly error: string;
}
function parseEnvFiles(
envFilePaths: readonly string[],
userLevelPaths: ReadonlySet<string>,
): ParsedEnvFilesResult {
const files: ParsedEnvFile[] = [];
const readFailures: EnvFileReadFailure[] = [];
for (const envFilePath of envFilePaths) {
try {
const envFileContent = fs.readFileSync(envFilePath, 'utf-8');
const parsedEnv = dotenv.parse(envFileContent);
const normalizedEnvFilePath = path.normalize(envFilePath);
const isHomeScopedEnvFile = userLevelPaths.has(normalizedEnvFilePath);
const isQwenScopedEnvFile =
isHomeScopedEnvFile ||
path.basename(path.dirname(normalizedEnvFilePath)) === QWEN_DIR;
files.push({
parsedEnv,
isHomeScopedEnvFile,
isQwenScopedEnvFile,
});
} catch (err) {
readFailures.push({
path: envFilePath,
error: getErrorMessage(err),
});
}
}
return { files, readFailed: readFailures.length > 0, readFailures };
}
function canApplyParsedEnvKey(
envFile: ParsedEnvFile,
key: string,
excludedVars: readonly string[],
options: { readonly reload?: boolean } = {},
): boolean {
if (!Object.hasOwn(envFile.parsedEnv, key)) return false;
if (options.reload && RELOAD_EXCLUDED_KEYS.has(key)) return false;
if (
!envFile.isHomeScopedEnvFile &&
PROJECT_ENV_HARDCODED_EXCLUSIONS.includes(key)
) {
return false;
}
return envFile.isQwenScopedEnvFile || !excludedVars.includes(key);
}
export interface RuntimeEnvironmentSnapshot {
readonly effectiveEnv: Readonly<NodeJS.ProcessEnv>;
readonly overlayKeys: readonly string[];
readonly envFilePaths: readonly string[];
readonly envFileReadFailed: boolean;
readonly envFileReadFailures: readonly EnvFileReadFailure[];
}
function isEffectivelyUnset(env: NodeJS.ProcessEnv, key: string): boolean {
const existingValue = env[key];
return !Object.hasOwn(env, key) || existingValue === '';
}
function setRuntimeEnvIfUnset(
env: NodeJS.ProcessEnv,
key: string,
value: string,
): void {
if (isEffectivelyUnset(env, key)) {
env[key] = value;
}
}
export function buildRuntimeEnvironment(
settings: Settings,
startDir: string = process.cwd(),
baseEnv: Readonly<NodeJS.ProcessEnv> = process.env,
): RuntimeEnvironmentSnapshot {
const userLevelPaths = getUserLevelEnvPaths();
const envFilePaths = findEnvFiles(settings, startDir, userLevelPaths);
const parsedEnvFiles = parseEnvFiles(envFilePaths, userLevelPaths);
const effectiveEnv: NodeJS.ProcessEnv = { ...baseEnv };
if (baseEnv['CLOUD_SHELL'] === 'true') {
setUpCloudShellEnvironmentInEnv(effectiveEnv, parsedEnvFiles.files);
}
for (const envFile of parsedEnvFiles.files) {
const excludedVars =
settings?.advanced?.excludedEnvVars || DEFAULT_EXCLUDED_ENV_VARS;
for (const key in envFile.parsedEnv) {
if (!canApplyParsedEnvKey(envFile, key, excludedVars, { reload: true })) {
continue;
}
setRuntimeEnvIfUnset(effectiveEnv, key, envFile.parsedEnv[key]!);
}
}
if (settings.env) {
const excludedVars =
settings?.advanced?.excludedEnvVars || DEFAULT_EXCLUDED_ENV_VARS;
for (const [key, value] of Object.entries(settings.env)) {
if (RELOAD_EXCLUDED_KEYS.has(key)) continue;
if (PROJECT_ENV_HARDCODED_EXCLUSIONS.includes(key)) continue;
if (excludedVars.includes(key)) continue;
if (typeof value !== 'string') continue;
setRuntimeEnvIfUnset(effectiveEnv, key, value);
}
}
const overlayKeys = Object.keys(effectiveEnv)
.filter((key) => effectiveEnv[key] !== baseEnv[key])
.sort();
return {
effectiveEnv: Object.freeze({ ...effectiveEnv }),
overlayKeys: Object.freeze(overlayKeys),
envFilePaths: Object.freeze([...envFilePaths]),
envFileReadFailed: parsedEnvFiles.readFailed,
envFileReadFailures: Object.freeze([...parsedEnvFiles.readFailures]),
};
}
/**
@ -338,59 +475,37 @@ export function loadEnvironment(
): void {
const userLevelPaths = getUserLevelEnvPaths();
const envFilePaths = findEnvFiles(settings, startDir, userLevelPaths);
const parsedEnvFiles = parseEnvFiles(envFilePaths, userLevelPaths);
// Cloud Shell environment variable handling
if (process.env['CLOUD_SHELL'] === 'true') {
setUpCloudShellEnvironmentFromFiles(envFilePaths);
setUpCloudShellEnvironmentInEnv(process.env, parsedEnvFiles.files);
}
// Step 1: Load from .env files (higher priority than settings.env)
// Only set if not already present in process.env (no-override mode)
for (const envFilePath of envFilePaths) {
try {
const envFileContent = fs.readFileSync(envFilePath, 'utf-8');
const parsedEnv = dotenv.parse(envFileContent);
for (const envFile of parsedEnvFiles.files) {
const excludedVars =
settings?.advanced?.excludedEnvVars || DEFAULT_EXCLUDED_ENV_VARS;
// homeScoped: `.env` lives under the user's home Qwen dir or `~/.env` —
// only these may set QWEN_HOME / QWEN_RUNTIME_DIR.
// qwenScoped: any `.env` whose immediate parent is `.qwen` (including
// `<repo>/.qwen/.env`) — exempt from the user `excludedEnvVars` list.
for (const key in envFile.parsedEnv) {
if (!canApplyParsedEnvKey(envFile, key, excludedVars)) continue;
const excludedVars =
settings?.advanced?.excludedEnvVars || DEFAULT_EXCLUDED_ENV_VARS;
const normalizedEnvFilePath = path.normalize(envFilePath);
// homeScoped: `.env` lives under the user's home Qwen dir or `~/.env` —
// only these may set QWEN_HOME / QWEN_RUNTIME_DIR.
// qwenScoped: any `.env` whose immediate parent is `.qwen` (including
// `<repo>/.qwen/.env`) — exempt from the user `excludedEnvVars` list.
const isHomeScopedEnvFile = userLevelPaths.has(normalizedEnvFilePath);
const isQwenScopedEnvFile =
isHomeScopedEnvFile ||
path.basename(path.dirname(normalizedEnvFilePath)) === QWEN_DIR;
for (const key in parsedEnv) {
if (Object.hasOwn(parsedEnv, key)) {
if (
!isHomeScopedEnvFile &&
PROJECT_ENV_HARDCODED_EXCLUSIONS.includes(key)
) {
continue;
}
if (!isQwenScopedEnvFile && excludedVars.includes(key)) {
continue;
}
const existingValue = process.env[key];
const isEffectivelyUnset =
!Object.hasOwn(process.env, key) || existingValue === '';
if (isEffectivelyUnset) {
process.env[key] = parsedEnv[key];
dotEnvSourcedKeys.add(key);
}
// Seed snapshot with ALL parsed keys (not just written ones)
// so child processes can detect deletions on first reload.
if (!lastReloadSnapshotSeeded && !lastReloadSnapshot.has(key)) {
lastReloadSnapshot.set(key, parsedEnv[key]!);
}
}
const existingValue = process.env[key];
const isEffectivelyUnset =
!Object.hasOwn(process.env, key) || existingValue === '';
if (isEffectivelyUnset) {
process.env[key] = envFile.parsedEnv[key];
dotEnvSourcedKeys.add(key);
}
// Seed snapshot with ALL parsed keys (not just written ones)
// so child processes can detect deletions on first reload.
if (!lastReloadSnapshotSeeded && !lastReloadSnapshot.has(key)) {
lastReloadSnapshot.set(key, envFile.parsedEnv[key]!);
}
} catch (_e) {
// Errors are ignored to match the behavior of `dotenv.config({ quiet: true })`.
}
}
@ -399,6 +514,9 @@ export function loadEnvironment(
// settings.json could otherwise redirect global state after path bootstrap.
if (settings.env) {
for (const [key, value] of Object.entries(settings.env)) {
if (RELOAD_EXCLUDED_KEYS.has(key)) {
continue;
}
if (PROJECT_ENV_HARDCODED_EXCLUSIONS.includes(key)) {
continue;
}
@ -441,44 +559,27 @@ export function reloadEnvironment(
): EnvReloadResult {
const userLevelPaths = getUserLevelEnvPaths();
const envFilePaths = findEnvFiles(settings, workspaceCwd, userLevelPaths);
const parsedEnvFiles = parseEnvFiles(envFilePaths, userLevelPaths);
if (process.env['CLOUD_SHELL'] === 'true') {
setUpCloudShellEnvironmentFromFiles(envFilePaths);
setUpCloudShellEnvironmentInEnv(process.env, parsedEnvFiles.files);
}
// Build the set of new keys from .env (higher priority) + settings.env
let dotEnvReadFailed = false;
const dotEnvReadFailed = parsedEnvFiles.readFailed;
const newDotEnvKeys = new Map<string, string>();
const newSettingsEnvKeys = new Map<string, string>();
for (const envFilePath of envFilePaths) {
try {
const envFileContent = fs.readFileSync(envFilePath, 'utf-8');
const parsedEnv = dotenv.parse(envFileContent);
const excludedVars =
settings?.advanced?.excludedEnvVars || DEFAULT_EXCLUDED_ENV_VARS;
const normalizedEnvFilePath = path.normalize(envFilePath);
const isHomeScopedEnvFile = userLevelPaths.has(normalizedEnvFilePath);
const isQwenScopedEnvFile =
isHomeScopedEnvFile ||
path.basename(path.dirname(normalizedEnvFilePath)) === QWEN_DIR;
for (const key in parsedEnv) {
if (!Object.hasOwn(parsedEnv, key)) continue;
if (RELOAD_EXCLUDED_KEYS.has(key)) continue;
if (
!isHomeScopedEnvFile &&
PROJECT_ENV_HARDCODED_EXCLUSIONS.includes(key)
) {
continue;
}
if (!isQwenScopedEnvFile && excludedVars.includes(key)) continue;
if (!newDotEnvKeys.has(key)) {
newDotEnvKeys.set(key, parsedEnv[key]!);
}
for (const envFile of parsedEnvFiles.files) {
const excludedVars =
settings?.advanced?.excludedEnvVars || DEFAULT_EXCLUDED_ENV_VARS;
for (const key in envFile.parsedEnv) {
if (!canApplyParsedEnvKey(envFile, key, excludedVars, { reload: true })) {
continue;
}
if (!newDotEnvKeys.has(key)) {
newDotEnvKeys.set(key, envFile.parsedEnv[key]!);
}
} catch {
dotEnvReadFailed = true;
}
}

View file

@ -578,7 +578,27 @@ function toRpcError(err: unknown): {
case 'InvalidClientIdError':
return { code: RPC.INVALID_PARAMS, message: errMsg(err) };
case 'SessionLimitExceededError':
return { code: RPC.INTERNAL_ERROR, message: errMsg(err) };
return {
code: RPC.INTERNAL_ERROR,
message: errMsg(err),
data: {
errorKind: 'session_limit_exceeded',
limit: (err as { limit?: unknown }).limit,
scope: 'workspace',
retryable: true,
},
};
case 'TotalSessionLimitExceededError':
return {
code: RPC.INTERNAL_ERROR,
message: errMsg(err),
data: {
errorKind: 'session_limit_exceeded',
limit: (err as { limit?: unknown }).limit,
scope: (err as { scope?: unknown }).scope,
retryable: true,
},
};
case 'PromptQueueFullError': {
const promptErr = err as {
sessionId?: unknown;

View file

@ -28,8 +28,10 @@ import {
PermissionForbiddenError,
PermissionPolicyNotImplementedError,
PromptQueueFullError,
SessionLimitExceededError,
SessionShellClientRequiredError,
SessionShellDisabledError,
TotalSessionLimitExceededError,
} from '@qwen-code/acp-bridge/bridgeErrors';
import { SessionService, Storage } from '@qwen-code/qwen-code-core';
import {
@ -1166,6 +1168,80 @@ describe('ACP Streamable HTTP transport (over the wire)', () => {
expect(frame.result.sessionId).toBe('sess-1');
});
it('maps workspace session admission failures to retryable RPC error data', async () => {
bridge.spawnOrAttach = async () => {
throw new SessionLimitExceededError(20);
};
const connId = await initialize();
const connStream = await openStream(connId);
const got = takeFrames(connStream, 1);
await new Promise((r) => setTimeout(r, 50));
const ack = await post(connId, {
jsonrpc: '2.0',
id: 3,
method: 'session/new',
params: { cwd: '/ws' },
});
expect(ack.status).toBe(202);
const [frame] = (await got) as Array<{
id: number;
error: {
code: number;
data: {
errorKind: string;
limit: number;
scope: string;
retryable: boolean;
};
};
}>;
expect(frame.id).toBe(3);
expect(frame.error.code).toBe(-32603);
expect(frame.error.data).toMatchObject({
errorKind: 'session_limit_exceeded',
limit: 20,
scope: 'workspace',
retryable: true,
});
});
it('maps total session admission failures to retryable RPC error data', async () => {
bridge.spawnOrAttach = async () => {
throw new TotalSessionLimitExceededError(10);
};
const connId = await initialize();
const connStream = await openStream(connId);
const got = takeFrames(connStream, 1);
await new Promise((r) => setTimeout(r, 50));
const ack = await post(connId, {
jsonrpc: '2.0',
id: 3,
method: 'session/new',
params: { cwd: '/ws' },
});
expect(ack.status).toBe(202);
const [frame] = (await got) as Array<{
id: number;
error: {
code: number;
data: {
errorKind: string;
limit: number;
scope: string;
retryable: boolean;
};
};
}>;
expect(frame.id).toBe(3);
expect(frame.error.code).toBe(-32603);
expect(frame.error.data).toMatchObject({
errorKind: 'session_limit_exceeded',
limit: 10,
scope: 'total',
retryable: true,
});
});
it('prompt streams session/update then the final result', async () => {
bridge.promptBehavior = async (_s, q) => {
q.push({

View file

@ -55,6 +55,9 @@ export type {
} from '@qwen-code/acp-bridge';
export type {
BridgeFreshSessionAdmission,
BridgeFreshSessionAdmissionContext,
BridgeFreshSessionReservation,
BridgeOptions,
DaemonStatusProvider,
} from '@qwen-code/acp-bridge/bridgeOptions';
@ -110,6 +113,7 @@ export {
McpServerRestartFailedError,
SessionBusyError,
InvalidRewindTargetError,
TotalSessionLimitExceededError,
NOT_CURRENTLY_GENERATING_CANCEL_MESSAGE,
// Multi-client permission coordination errors.
CancelSentinelCollisionError,

View file

@ -27,7 +27,7 @@ import {
type ServeWorkspaceEnvStatus,
} from '@qwen-code/acp-bridge';
import { getGitVersion, getNpmVersion } from '../utils/systemInfo.js';
import { buildEnvStatusFromProcess } from './env-snapshot.js';
import { buildEnvStatusFromEnv, snapshotProcessEnv } from './env-snapshot.js';
const REQUIRED_NODE_MAJOR = 22;
@ -38,7 +38,12 @@ const REQUIRED_NODE_MAJOR = 22;
* called only from the route handlers, so per-request allocation is
* fine).
*/
export function createDaemonStatusProvider(): DaemonStatusProvider {
export function createDaemonStatusProvider(
options: {
env?: Readonly<Record<string, string | undefined>>;
} = {},
): DaemonStatusProvider {
const readEnv = () => options.env ?? snapshotProcessEnv();
return {
async getEnvStatus(
boundWorkspace: string,
@ -48,13 +53,13 @@ export function createDaemonStatusProvider(): DaemonStatusProvider {
// in a resolved Promise to match the async `DaemonStatusProvider`
// contract. Future async-needing implementations (e.g. reading
// a config file) get the seam without changing the bridge.
return buildEnvStatusFromProcess(boundWorkspace, acpChannelLive);
return buildEnvStatusFromEnv(boundWorkspace, acpChannelLive, readEnv());
},
async getDaemonPreflightCells(
boundWorkspace: string,
): Promise<ServePreflightCell[]> {
return buildDaemonPreflightCells(boundWorkspace);
return buildDaemonPreflightCells(boundWorkspace, readEnv());
},
};
}
@ -75,6 +80,7 @@ export function createDaemonStatusProvider(): DaemonStatusProvider {
*/
async function buildDaemonPreflightCells(
boundWorkspace: string,
env: Readonly<Record<string, string | undefined>>,
): Promise<ServePreflightCell[]> {
// Each builder returns (or eventually returns) one cell. We run them via
// `Promise.allSettled` after wrapping every call in `Promise.resolve().then`
@ -125,7 +131,7 @@ async function buildDaemonPreflightCells(
// Mirrors `defaultSpawnChannelFactory`'s lookup so the preflight cell
// reflects the path the child would actually be spawned from.
const cliEntryCell = (): ServePreflightCell => {
const cliEntry = process.env['QWEN_CLI_ENTRY'] || process.argv[1] || '';
const cliEntry = env['QWEN_CLI_ENTRY'] || process.argv[1] || '';
if (cliEntry) {
return {
kind: 'cli_entry',
@ -133,9 +139,7 @@ async function buildDaemonPreflightCells(
locality: 'daemon',
detail: {
path: cliEntry,
source: process.env['QWEN_CLI_ENTRY']
? 'QWEN_CLI_ENTRY'
: 'process.argv[1]',
source: env['QWEN_CLI_ENTRY'] ? 'QWEN_CLI_ENTRY' : 'process.argv[1]',
},
};
}

View file

@ -43,6 +43,66 @@ afterEach(() => {
});
describe('buildDaemonStatusResponse', () => {
it('includes maxTotalSessions in daemon status limits', async () => {
const options = makeOptions();
options.opts.maxTotalSessions = 50;
const response = await buildDaemonStatusResponse('summary', options);
expect(response.limits.maxTotalSessions).toBe(50);
});
it('warns when total session capacity is high and reports in-flight admission', async () => {
const options = makeOptions({
totalAdmissionInFlight: 1,
bridgeSnapshot: {
...BASE_BRIDGE_SNAPSHOT,
sessionCount: 7,
},
});
options.opts.maxTotalSessions = 10;
const response = await buildDaemonStatusResponse('summary', options);
expect(response.runtime.sessions).toMatchObject({
active: 7,
admissionInFlight: 1,
});
expect(response).toMatchObject({
status: 'warning',
issues: expect.arrayContaining([
expect.objectContaining({ code: 'total_session_capacity_high' }),
]),
});
});
it('uses total admission live count for total session capacity warnings', async () => {
const options = makeOptions({
totalAdmissionLiveCount: 8,
totalAdmissionInFlight: 1,
bridgeSnapshot: {
...BASE_BRIDGE_SNAPSHOT,
sessionCount: 1,
},
});
options.opts.maxTotalSessions = 10;
const response = await buildDaemonStatusResponse('summary', options);
expect(response.runtime.sessions).toMatchObject({
active: 1,
admissionInFlight: 1,
});
expect(response).toMatchObject({
status: 'warning',
issues: expect.arrayContaining([
expect.objectContaining({
code: 'total_session_capacity_high',
message: 'Total active and in-flight sessions are at 9/10.',
}),
]),
});
});
it('reports every runtime issue code from daemon counters', async () => {
const response = await buildDaemonStatusResponse(
'summary',
@ -615,6 +675,8 @@ interface MakeOptionsInput {
activePromptCount?: number;
pendingPromptTotal?: number;
lastActivityAt?: number | null;
totalAdmissionLiveCount?: number;
totalAdmissionInFlight?: number;
}
function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions {
@ -678,6 +740,16 @@ function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions {
...(input.perfSnapshot
? { getPerfSnapshot: () => input.perfSnapshot! }
: {}),
...(input.totalAdmissionInFlight === undefined
? {}
: {
getTotalSessionAdmissionSnapshot: () => ({
liveCount:
input.totalAdmissionLiveCount ??
(input.bridgeSnapshot ?? BASE_BRIDGE_SNAPSHOT).sessionCount,
inFlight: input.totalAdmissionInFlight!,
}),
}),
};
}

View file

@ -21,6 +21,7 @@ import type {
DaemonWorkspaceService,
WorkspaceRequestContext,
} from './workspace-service/index.js';
import type { TotalSessionAdmissionSnapshot } from './total-session-admission.js';
// Re-export so downstream consumers (server.ts, routes, the SDK type mirror)
// import the bucket shape from the status module alongside the rest of the
@ -61,6 +62,7 @@ export interface DaemonStartupSnapshot {
export interface DaemonStatusIssue {
code:
| 'session_capacity_high'
| 'total_session_capacity_high'
| 'connection_capacity_high'
| 'pending_permissions'
| 'acp_channel_down'
@ -102,6 +104,7 @@ export interface BuildDaemonStatusOptions {
getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot;
getPerfSnapshot?: () => DaemonPerfSnapshot;
getMetricsSeries?: () => DaemonMetricsBucket[];
getTotalSessionAdmissionSnapshot?: () => TotalSessionAdmissionSnapshot;
}
interface DaemonStatusSection<T> {
@ -140,6 +143,7 @@ interface DaemonStatusSecurity {
interface DaemonStatusLimits {
maxSessions: number | null;
maxTotalSessions: number | null;
maxPendingPromptsPerSession: number | null;
listenerMaxConnections: number | null;
eventRingSize: number;
@ -153,7 +157,7 @@ interface DaemonStatusLimits {
interface DaemonStatusRuntime {
loading?: boolean;
error?: string;
sessions: { active: number };
sessions: { active: number; admissionInFlight?: number };
permissions: {
pending: number;
policy: string;
@ -284,6 +288,7 @@ export async function buildDaemonStatusResponse(
state: 'disabled',
channels: [],
};
const totalAdmissionSnapshot = input.getTotalSessionAdmissionSnapshot?.();
const issues: DaemonStatusIssue[] = [];
let full: FullDaemonStatus | undefined;
@ -294,6 +299,7 @@ export async function buildDaemonStatusResponse(
rateLimitHits,
input,
channelWorker,
totalAdmissionSnapshot,
);
if (detail === 'full') {
@ -335,6 +341,7 @@ export async function buildDaemonStatusResponse(
},
limits: {
maxSessions: bridgeSnapshot.limits.maxSessions,
maxTotalSessions: positiveFiniteOrNull(input.opts.maxTotalSessions),
maxPendingPromptsPerSession:
bridgeSnapshot.limits.maxPendingPromptsPerSession,
listenerMaxConnections: listenerMaxConnections(input.opts.maxConnections),
@ -350,7 +357,12 @@ export async function buildDaemonStatusResponse(
features: [...input.features],
},
runtime: {
sessions: { active: bridgeSnapshot.sessionCount },
sessions: {
active: bridgeSnapshot.sessionCount,
...(totalAdmissionSnapshot
? { admissionInFlight: totalAdmissionSnapshot.inFlight }
: {}),
},
permissions: {
pending: bridgeSnapshot.pendingPermissionCount,
policy: bridgeSnapshot.permissionPolicy,
@ -523,6 +535,7 @@ function pushRuntimeIssues(
rateLimitHits: Record<RateLimitTier, number>,
input: BuildDaemonStatusOptions,
channelWorker: ChannelWorkerSnapshot,
totalAdmissionSnapshot: TotalSessionAdmissionSnapshot | undefined,
): void {
if (
bridgeSnapshot.limits.maxSessions !== null &&
@ -537,6 +550,20 @@ function pushRuntimeIssues(
});
}
const maxTotalSessions = positiveFiniteOrNull(input.opts.maxTotalSessions);
if (maxTotalSessions !== null) {
const totalActive =
(totalAdmissionSnapshot?.liveCount ?? bridgeSnapshot.sessionCount) +
(totalAdmissionSnapshot?.inFlight ?? 0);
if (totalActive / maxTotalSessions >= CAPACITY_WARNING_RATIO) {
issues.push({
code: 'total_session_capacity_high',
severity: 'warning',
message: `Total active and in-flight sessions are at ${totalActive}/${maxTotalSessions}.`,
});
}
}
if (
acpSnapshot !== undefined &&
acpSnapshot.connectionCap !== null &&

View file

@ -10,8 +10,10 @@ import {
ENV_NONSECRET_VARS,
ENV_PROXY_VARS,
ENV_SECRET_VARS,
buildEnvStatusFromEnv,
buildEnvStatusFromProcess,
readProxyVar,
snapshotProcessEnv,
} from './env-snapshot.js';
const TRACKED_ENV = [
@ -44,6 +46,44 @@ afterEach(() => {
});
describe('buildEnvStatusFromProcess', () => {
it('snapshots process.env into an independent copy', () => {
const key = 'QWEN_TEST_ENV_SNAPSHOT_COPY';
const previous = process.env[key];
delete process.env[key];
try {
const snapshot = snapshotProcessEnv();
snapshot[key] = 'mutated';
expect(process.env[key]).toBeUndefined();
process.env[key] = 'from-process';
expect(snapshot[key]).toBe('mutated');
} finally {
if (previous === undefined) {
delete process.env[key];
} else {
process.env[key] = previous;
}
}
});
it('can build a snapshot from an injected runtime env without reading process.env', () => {
const status = buildEnvStatusFromEnv('/ws', false, {
DASHSCOPE_API_KEY: 'sk-runtime',
HTTPS_PROXY: 'http://user:pass@runtime-proxy.local:8080',
});
const key = status.cells.find(
(c) => c.kind === 'env_var' && c.name === 'DASHSCOPE_API_KEY',
);
expect(key).toMatchObject({ present: true, status: 'ok' });
const proxy = status.cells.find(
(c) => c.kind === 'proxy' && c.name === 'HTTPS_PROXY',
);
expect(proxy!.value).toBe('runtime-proxy.local:8080');
expect(process.env['DASHSCOPE_API_KEY']).toBeUndefined();
});
it('emits a runtime cell whose value matches the actual runtime version', () => {
const status = buildEnvStatusFromProcess('/ws', false);
const runtime = status.cells.find((c) => c.kind === 'runtime');

View file

@ -67,6 +67,10 @@ const PROXY_VARS = [
'ALL_PROXY',
] as const;
export function snapshotProcessEnv(): Record<string, string | undefined> {
return { ...process.env };
}
/**
* Resolve a proxy env var, preferring the uppercase canonical form and
* falling back to the lowercase variant only when the uppercase is
@ -134,6 +138,18 @@ function safeProxyValue(name: string, raw: string): string {
export function buildEnvStatusFromProcess(
workspaceCwd: string,
acpChannelLive: boolean,
): ServeWorkspaceEnvStatus {
return buildEnvStatusFromEnv(
workspaceCwd,
acpChannelLive,
snapshotProcessEnv(),
);
}
export function buildEnvStatusFromEnv(
workspaceCwd: string,
acpChannelLive: boolean,
sourceEnv: Readonly<NodeJS.ProcessEnv>,
): ServeWorkspaceEnvStatus {
// `process.env` is shared mutable state — any concurrent code path
// (auth flow, settings reload, child boot) can mutate it mid-snapshot.
@ -141,7 +157,7 @@ export function buildEnvStatusFromProcess(
// env, and a client polling `/workspace/env` can never see a torn
// half-pre-init / half-post-init snapshot. Copy is cheap (a few hundred
// string refs) and atomic from JS' single-threaded execution model.
const env = { ...process.env };
const env = { ...sourceEnv };
const cells: ServeEnvCell[] = [];
// Under Bun, `process.versions.node` is the pinned node-compat shim

View file

@ -613,6 +613,7 @@ describe('serve fast path argument parsing', () => {
['hostname', ['--hostname', '127.0.0.1']],
['token', ['--token', 'token']],
['max-sessions', ['--max-sessions', '10']],
['max-total-sessions', ['--max-total-sessions', '20']],
[
'max-pending-prompts-per-session',
['--max-pending-prompts-per-session', '5'],
@ -678,6 +679,7 @@ describe('serve fast path argument parsing', () => {
},
});
expect(fastPathParsed).not.toHaveProperty('options.maxSessions');
expect(fastPathParsed).not.toHaveProperty('options.maxTotalSessions');
expect(fastPathParsed).not.toHaveProperty('options.maxConnections');
expect(fastPathParsed).not.toHaveProperty('options.eventRingSize');
expect(fastPathParsed).not.toHaveProperty(

View file

@ -35,6 +35,7 @@ const NUMBER_OPTIONS = new Map<
>([
['port', 'port'],
['maxSessions', 'max-sessions'],
['maxTotalSessions', 'max-total-sessions'],
['maxPendingPromptsPerSession', 'max-pending-prompts-per-session'],
['maxConnections', 'max-connections'],
['eventRingSize', 'event-ring-size'],

View file

@ -0,0 +1,80 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import * as fs from 'node:fs';
import * as path from 'node:path';
import { fileURLToPath } from 'node:url';
import { describe, expect, it } from 'vitest';
const repoRoot = path.resolve(
path.dirname(fileURLToPath(import.meta.url)),
'..',
'..',
'..',
'..',
);
const scannedRoots = [
path.join(repoRoot, 'packages', 'cli', 'src', 'serve'),
path.join(repoRoot, 'packages', 'acp-bridge', 'src'),
];
const allowedProcessEnvFiles = new Set(
[
'packages/acp-bridge/src/bridge.ts',
'packages/acp-bridge/src/bridgeOptions.ts',
'packages/acp-bridge/src/spawnChannel.ts',
'packages/cli/src/serve/acp-http/index.ts',
'packages/cli/src/serve/channel-worker-supervisor.ts',
'packages/cli/src/serve/daemon-logger.ts',
'packages/cli/src/serve/debug-mode.ts',
'packages/cli/src/serve/env-snapshot.ts',
'packages/cli/src/serve/fast-path-settings.ts',
'packages/cli/src/serve/fast-path.ts',
'packages/cli/src/serve/fs/audit.ts',
'packages/cli/src/serve/routes/workspace-setup-github.ts',
'packages/cli/src/serve/run-qwen-serve.ts',
'packages/cli/src/serve/server/fs-factory.ts',
'packages/cli/src/serve/server/serve-features.ts',
].map((file) => path.normalize(file)),
);
function listTypeScriptFiles(dir: string): string[] {
const out: string[] = [];
for (const entry of fs.readdirSync(dir, { withFileTypes: true })) {
const full = path.join(dir, entry.name);
if (entry.isDirectory()) {
listTypeScriptFiles(full).forEach((file) => out.push(file));
} else if (entry.isFile() && full.endsWith('.ts')) {
out.push(full);
}
}
return out;
}
function stripComments(source: string): string {
return source.replace(/\/\*[\s\S]*?\*\//g, '').replace(/^\s*\/\/.*$/gm, '');
}
describe('serve process.env guard', () => {
it('keeps workspace-scoped serve and acp-bridge code off direct process.env reads', () => {
const offenders: string[] = [];
for (const root of scannedRoots) {
for (const file of listTypeScriptFiles(root)) {
if (file.endsWith('.test.ts')) continue;
const relative = path.normalize(path.relative(repoRoot, file));
if (allowedProcessEnvFiles.has(relative)) continue;
if (
stripComments(fs.readFileSync(file, 'utf8')).includes('process.env')
) {
offenders.push(relative);
}
}
}
expect(offenders).toEqual([]);
});
});

View file

@ -60,6 +60,7 @@ vi.mock('@modelcontextprotocol/sdk/client/index.js', () => ({
Client: sdkMocks.MockClient,
}));
vi.mock('@modelcontextprotocol/sdk/client/stdio.js', () => ({
DEFAULT_INHERITED_ENV_VARS: ['HOME', 'PATH'],
StdioClientTransport: sdkMocks.MockStdioClientTransport,
}));
vi.mock('@modelcontextprotocol/sdk/client/streamableHttp.js', () => ({
@ -411,14 +412,30 @@ describe('helpers', () => {
expect(sdkMocks.state.stdioTransports).toHaveLength(0);
});
it('buildTransport merges stdio env only when provided', () => {
buildTransport({
command: 'node',
args: ['server.mjs'],
env: { A2UI_TOKEN: 'secret' },
cwd: '/workspace',
});
buildTransport({ command: 'node' });
it('buildTransport uses the scrubbed runtime env for stdio transports', () => {
const runtimeEnv = {
HOME: '/runtime/home',
PATH: '/runtime/bin',
RUNTIME_ONLY: 'yes',
A2UI_TOKEN: 'base',
OPENAI_API_KEY: 'runtime-key',
QWEN_SERVER_TOKEN: 'daemon-secret',
BASH_FUNC_bad: '() { ignored; }',
SHELL_FUNC: '() { ignored; }',
};
buildTransport(
{
command: 'node',
args: ['server.mjs'],
env: {
A2UI_TOKEN: 'secret',
QWEN_SERVER_TOKEN: 'explicit-secret',
},
cwd: '/workspace',
},
runtimeEnv,
);
buildTransport({ command: 'node' }, runtimeEnv);
const withEnv = sdkMocks.state.stdioTransports[0] as {
options: {
@ -436,9 +453,19 @@ describe('helpers', () => {
args: ['server.mjs'],
cwd: '/workspace',
});
expect(withEnv.options.env?.['A2UI_TOKEN']).toBe('secret');
expect(withEnv.options.env?.['PATH']).toBe(process.env['PATH']);
expect(withoutEnv.options).not.toHaveProperty('env');
expect(withEnv.options.env).toMatchObject({
HOME: '/runtime/home',
PATH: '/runtime/bin',
RUNTIME_ONLY: 'yes',
A2UI_TOKEN: 'secret',
OPENAI_API_KEY: 'runtime-key',
});
expect(withEnv.options.env?.['BASH_FUNC_bad']).toBeUndefined();
expect(withEnv.options.env?.['SHELL_FUNC']).toBeUndefined();
expect(withEnv.options.env?.['QWEN_SERVER_TOKEN']).toBeUndefined();
expect(withoutEnv.options.env?.['PATH']).toBe('/runtime/bin');
expect(withoutEnv.options.env?.['A2UI_TOKEN']).toBe('base');
expect(withoutEnv.options.env?.['QWEN_SERVER_TOKEN']).toBeUndefined();
});
it('callA2uiAction connects, calls the action tool, and closes resources', async () => {

View file

@ -34,12 +34,17 @@ import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js'
import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js';
import { writeStderrLine } from '../../utils/stdioHelpers.js';
import { QWEN_SERVER_TOKEN_ENV } from '../channel-worker-env.js';
import { snapshotProcessEnv } from '../env-snapshot.js';
const A2UI_MIME = 'application/a2ui+json';
// Standard action-tool name from the official A2UI-over-MCP guide
// (a2ui.org/guides/a2ui_over_mcp).
const ACTION_TOOL = 'action';
const CALL_TIMEOUT_MS = 15_000;
const SCRUBBED_STDIO_ENV_KEYS: ReadonlySet<string> = new Set([
QWEN_SERVER_TOKEN_ENV,
]);
export interface McpServerConfigLike {
command?: string;
@ -86,6 +91,7 @@ interface RegisterA2uiActionRoutesOptions {
cfg: McpServerConfigLike,
args: A2uiActionArgs,
) => Promise<A2uiActionResult>;
env?: Readonly<Record<string, string | undefined>>;
}
/** Exported for unit testing. */
@ -129,24 +135,45 @@ export async function findFromSettingsFile(
}
/** Build a one-shot transport from the config shape: stdio (command) or streamable HTTP (httpUrl). */
export function buildTransport(cfg: McpServerConfigLike): Transport {
export function buildTransport(
cfg: McpServerConfigLike,
baseEnv: Readonly<Record<string, string | undefined>> = snapshotProcessEnv(),
): Transport {
if (typeof cfg.httpUrl === 'string') {
return new StreamableHTTPClientTransport(new URL(cfg.httpUrl));
}
return new StdioClientTransport({
command: cfg.command!,
args: cfg.args ?? [],
// spawn() treats `env` as a complete replacement, not a merge — a partial
// env (e.g. {API_KEY}) would strip PATH/HOME and break the child. Merge
// over process.env like packages/core/src/tools/mcp-client.ts does; when
// unset, let the SDK apply its safe default environment.
...(cfg.env
? { env: { ...process.env, ...cfg.env } as Record<string, string> }
: {}),
// Passing `env` prevents the SDK from reading live process.env while
// keeping daemon MCP stdio behavior aligned with the core CLI client.
env: buildStdioServerEnv(baseEnv, cfg.env),
cwd: cfg.cwd,
});
}
function buildStdioServerEnv(
baseEnv: Readonly<Record<string, string | undefined>>,
serverEnv: Record<string, string> | undefined,
): Record<string, string> {
const env: Record<string, string> = {};
for (const [key, value] of Object.entries(baseEnv)) {
if (
value !== undefined &&
!key.startsWith('BASH_FUNC_') &&
!value.startsWith('()') &&
!SCRUBBED_STDIO_ENV_KEYS.has(key)
) {
env[key] = value;
}
}
const merged = { ...env, ...(serverEnv ?? {}) };
for (const key of SCRUBBED_STDIO_ENV_KEYS) {
delete merged[key];
}
return merged;
}
/** Exported for unit testing the MCP content normalization rules. */
export function extractA2uiActionResult(
result: A2uiToolResult,
@ -192,8 +219,9 @@ export function extractA2uiActionResult(
export async function callA2uiAction(
cfg: McpServerConfigLike,
args: A2uiActionArgs,
env?: Readonly<Record<string, string | undefined>>,
): Promise<A2uiActionResult> {
const transport = buildTransport(cfg);
const transport = buildTransport(cfg, env);
const client = new Client({ name: 'qwen-serve-a2ui', version: '0.0.1' });
try {
await client.connect(transport, { timeout: CALL_TIMEOUT_MS });
@ -216,7 +244,10 @@ export function registerA2uiActionRoutes(
opts: RegisterA2uiActionRoutesOptions,
): void {
const { boundWorkspace, mutate, safeBody, getMcpServers } = opts;
const callAction = opts.callAction ?? callA2uiAction;
const callAction =
opts.callAction ??
((cfg: McpServerConfigLike, args: A2uiActionArgs) =>
callA2uiAction(cfg, args, opts.env));
app.post(
'/session/:id/a2ui-action',

View file

@ -25,6 +25,7 @@ import type { ServeOptions } from '../types.js';
import type { ChannelWorkerSnapshot } from '../channel-worker-supervisor.js';
import type { DaemonWorkspaceService } from '../workspace-service/index.js';
import { getServeProtocolVersions } from '../capabilities.js';
import type { TotalSessionAdmissionSnapshot } from '../total-session-admission.js';
interface RegisterDaemonStatusRoutesDeps {
opts: ServeOptions;
@ -46,6 +47,7 @@ interface RegisterDaemonStatusRoutesDeps {
getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot;
getPerfSnapshot?: () => DaemonPerfSnapshot;
getMetricsSeries?: () => DaemonMetricsBucket[];
getTotalSessionAdmissionSnapshot?: () => TotalSessionAdmissionSnapshot;
}
export function registerDaemonStatusRoutes(
@ -82,6 +84,8 @@ export function registerDaemonStatusRoutes(
getChannelWorkerSnapshot: deps.getChannelWorkerSnapshot,
getPerfSnapshot: deps.getPerfSnapshot,
getMetricsSeries: deps.getMetricsSeries,
getTotalSessionAdmissionSnapshot:
deps.getTotalSessionAdmissionSnapshot,
}),
);
} catch (err) {

View file

@ -34,6 +34,7 @@ import type {
import * as qwenCore from '@qwen-code/qwen-code-core';
import * as serverModule from './server.js';
import * as settingsRuntime from '../config/settings.js';
import * as environmentRuntime from '../config/environment.js';
import * as trustedFoldersRuntime from '../config/trustedFolders.js';
import type {
ChannelWorkerSnapshot,
@ -965,6 +966,9 @@ describe('runQwenServe pre-listen bridge option validation', () => {
it.each([
['maxSessions', Number.NaN, /maxSessions/],
['maxSessions', -1, /maxSessions/],
['maxTotalSessions', Number.NaN, /maxTotalSessions/],
['maxTotalSessions', -1, /maxTotalSessions/],
['maxTotalSessions', 1.5, /maxTotalSessions/],
['eventRingSize', 0, /eventRingSize/],
['eventRingSize', 1.5, /eventRingSize/],
['eventRingSize', Number.POSITIVE_INFINITY, /eventRingSize/],
@ -1301,6 +1305,257 @@ describe('runQwenServe runtime startup failures', () => {
}
});
it('rebuilds runtime env from the immutable daemon base after workspace reload', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-runtime-env-reload-')),
);
const originalBase = process.env['QWEN_TEST_BOOT_BASE'];
const originalLeak = process.env['QWEN_TEST_RELOAD_LEAK'];
const originalRemoved = process.env['QWEN_TEST_REMOVED_FROM_DOTENV'];
process.env['QWEN_TEST_BOOT_BASE'] = 'base';
process.env['QWEN_TEST_REMOVED_FROM_DOTENV'] = 'stale';
delete process.env['QWEN_TEST_RELOAD_LEAK'];
vi.spyOn(qwenCore, 'resolveTelemetrySettings').mockResolvedValue({
enabled: false,
sensitiveSpanAttributeMaxLength: 1024 * 1024,
});
vi.spyOn(settingsRuntime, 'loadSettings').mockImplementation(
(...args: Parameters<typeof settingsRuntime.loadSettings>) => {
const options = args[1];
const isReload =
typeof options === 'object' && options?.skipLoadEnvironment === true;
return {
merged: {
env: {
QWEN_TEST_RUNTIME_VALUE: isReload ? 'reloaded' : 'boot',
},
},
} as unknown as ReturnType<typeof settingsRuntime.loadSettings>;
},
);
vi.spyOn(settingsRuntime, 'reloadEnvironment').mockImplementation(() => {
process.env['QWEN_TEST_RELOAD_LEAK'] = 'workspace-a';
delete process.env['QWEN_TEST_REMOVED_FROM_DOTENV'];
return {
updatedKeys: ['QWEN_TEST_RELOAD_LEAK'],
removedKeys: ['QWEN_TEST_REMOVED_FROM_DOTENV'],
};
});
vi.spyOn(trustedFoldersRuntime, 'getWorkspaceTrustStatus').mockReturnValue({
effective: { state: 'trusted' },
} as ReturnType<typeof trustedFoldersRuntime.getWorkspaceTrustStatus>);
const buildRuntimeEnvironment = vi.spyOn(
environmentRuntime,
'buildRuntimeEnvironment',
);
let workspace:
| {
reload(ctx: {
route: string;
workspaceCwd: string;
}): Promise<unknown>;
}
| undefined;
let primaryRuntimeEnv:
| {
effectiveEnv?: NodeJS.ProcessEnv;
}
| undefined;
vi.spyOn(serverModule, 'createServeApp').mockImplementation(
(_opts, _getPort, deps) => {
workspace = deps?.workspace as typeof workspace;
primaryRuntimeEnv = deps?.primaryRuntimeEnv as typeof primaryRuntimeEnv;
return express();
},
);
const handle = await runQwenServe(
{
port: 0,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
maxSessions: 1,
serveWebShell: false,
},
{
bridge: makeRuntimeBridge(),
bootSettings: {},
daemonLogBaseDir: path.join(tmpDir, 'debug'),
resolveOnListen: true,
},
);
try {
await handle.runtimeReady;
expect(workspace).toBeDefined();
expect(primaryRuntimeEnv?.effectiveEnv).toBeDefined();
const capturedRuntimeEnv = primaryRuntimeEnv!.effectiveEnv!;
expect(capturedRuntimeEnv['QWEN_TEST_RUNTIME_VALUE']).toBe('boot');
await workspace!.reload({
route: 'POST /workspace/reload',
workspaceCwd: tmpDir,
});
const reloadBaseEnv = buildRuntimeEnvironment.mock.calls.at(-1)?.[2];
expect(reloadBaseEnv?.['QWEN_TEST_BOOT_BASE']).toBe('base');
expect(reloadBaseEnv?.['QWEN_TEST_REMOVED_FROM_DOTENV']).toBe('stale');
expect(reloadBaseEnv?.['QWEN_TEST_RELOAD_LEAK']).toBeUndefined();
expect(primaryRuntimeEnv!.effectiveEnv).toBe(capturedRuntimeEnv);
expect(capturedRuntimeEnv['QWEN_TEST_RUNTIME_VALUE']).toBe('reloaded');
expect(capturedRuntimeEnv['QWEN_TEST_REMOVED_FROM_DOTENV']).toBe('stale');
expect(capturedRuntimeEnv['QWEN_TEST_RELOAD_LEAK']).toBeUndefined();
} finally {
if (originalBase === undefined) {
delete process.env['QWEN_TEST_BOOT_BASE'];
} else {
process.env['QWEN_TEST_BOOT_BASE'] = originalBase;
}
if (originalLeak === undefined) {
delete process.env['QWEN_TEST_RELOAD_LEAK'];
} else {
process.env['QWEN_TEST_RELOAD_LEAK'] = originalLeak;
}
if (originalRemoved === undefined) {
delete process.env['QWEN_TEST_REMOVED_FROM_DOTENV'];
} else {
process.env['QWEN_TEST_REMOVED_FROM_DOTENV'] = originalRemoved;
}
await handle.close();
}
});
it('preserves previous runtime env and marks fallback when reload env rebuild fails', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-runtime-env-fallback-')),
);
vi.spyOn(qwenCore, 'resolveTelemetrySettings').mockResolvedValue({
enabled: false,
sensitiveSpanAttributeMaxLength: 1024 * 1024,
});
vi.spyOn(settingsRuntime, 'loadSettings').mockImplementation(
(...args: Parameters<typeof settingsRuntime.loadSettings>) => {
const options = args[1];
const isReload =
typeof options === 'object' && options?.skipLoadEnvironment === true;
return {
merged: {
env: {
QWEN_TEST_RUNTIME_VALUE: isReload ? 'reloaded' : 'boot',
},
},
} as unknown as ReturnType<typeof settingsRuntime.loadSettings>;
},
);
vi.spyOn(settingsRuntime, 'reloadEnvironment').mockReturnValue({
updatedKeys: [],
removedKeys: [],
});
vi.spyOn(trustedFoldersRuntime, 'getWorkspaceTrustStatus').mockReturnValue({
effective: { state: 'trusted' },
} as ReturnType<typeof trustedFoldersRuntime.getWorkspaceTrustStatus>);
const buildRuntimeEnvironmentActual =
environmentRuntime.buildRuntimeEnvironment;
let failReloadBuild = false;
vi.spyOn(environmentRuntime, 'buildRuntimeEnvironment').mockImplementation(
(
...args: Parameters<typeof environmentRuntime.buildRuntimeEnvironment>
) => {
if (failReloadBuild) {
throw new Error('runtime env rebuild failed');
}
return buildRuntimeEnvironmentActual(...args);
},
);
let workspace:
| {
reload(ctx: {
route: string;
workspaceCwd: string;
}): Promise<unknown>;
}
| undefined;
let primaryRuntimeEnv:
| {
effectiveEnv?: NodeJS.ProcessEnv;
fallbackReason?: string;
}
| undefined;
vi.spyOn(serverModule, 'createServeApp').mockImplementation(
(_opts, _getPort, deps) => {
workspace = deps?.workspace as typeof workspace;
primaryRuntimeEnv = deps?.primaryRuntimeEnv as typeof primaryRuntimeEnv;
return express();
},
);
const logBaseDir = path.join(tmpDir, 'debug');
const handle = await runQwenServe(
{
port: 0,
hostname: '127.0.0.1',
mode: 'http-bridge',
workspace: tmpDir,
maxSessions: 1,
serveWebShell: false,
},
{
bridge: makeRuntimeBridge(),
bootSettings: {},
daemonLogBaseDir: logBaseDir,
resolveOnListen: true,
},
);
let closed = false;
try {
await handle.runtimeReady;
expect(workspace).toBeDefined();
expect(primaryRuntimeEnv?.effectiveEnv).toBeDefined();
const capturedRuntimeEnv = primaryRuntimeEnv!.effectiveEnv!;
expect(capturedRuntimeEnv['QWEN_TEST_RUNTIME_VALUE']).toBe('boot');
failReloadBuild = true;
await workspace!.reload({
route: 'POST /workspace/reload',
workspaceCwd: tmpDir,
});
expect(primaryRuntimeEnv!.effectiveEnv).toBe(capturedRuntimeEnv);
expect(capturedRuntimeEnv['QWEN_TEST_RUNTIME_VALUE']).toBe('boot');
expect(primaryRuntimeEnv!.fallbackReason).toBe(
'runtime env rebuild failed',
);
failReloadBuild = false;
await workspace!.reload({
route: 'POST /workspace/reload',
workspaceCwd: tmpDir,
});
expect(primaryRuntimeEnv!.effectiveEnv).toBe(capturedRuntimeEnv);
expect(capturedRuntimeEnv['QWEN_TEST_RUNTIME_VALUE']).toBe('reloaded');
expect(primaryRuntimeEnv!.fallbackReason).toBeUndefined();
await handle.close();
closed = true;
const logPath = path.join(
logBaseDir,
'daemon',
`serve-${process.pid}.log`,
);
const log = fs.readFileSync(logPath, 'utf8');
expect(log).toContain(
'failed to rebuild runtime env snapshot after daemon env reload; preserving previous runtime env',
);
} finally {
if (!closed) {
await handle.close();
}
}
});
it('filters secondary workspace roots before constructing the bridge filesystem', async () => {
tmpDir = fs.realpathSync(
fs.mkdtempSync(path.join(os.tmpdir(), 'qws-runtime-roots-')),

View file

@ -487,6 +487,7 @@ export interface RunHandle {
type CoreRuntime = typeof import('@qwen-code/qwen-code-core');
type ProviderConfig = NonNullable<ReturnType<CoreRuntime['findProviderById']>>;
type SettingsRuntime = typeof import('../config/settings.js');
type EnvironmentRuntime = typeof import('../config/environment.js');
type LoadedSettingsAdapterRuntime =
typeof import('../config/loadedSettingsAdapter.js');
type TrustedFoldersRuntime = typeof import('../config/trustedFolders.js');
@ -707,21 +708,25 @@ async function resolveDaemonLogBaseDirForRun(input: {
let settingsRuntimePromise:
| Promise<{
settings: SettingsRuntime;
environment: EnvironmentRuntime;
loadedSettingsAdapter: LoadedSettingsAdapterRuntime;
trustedFolders: TrustedFoldersRuntime;
}>
| undefined;
function loadSettingsRuntimeModules(): Promise<{
settings: SettingsRuntime;
environment: EnvironmentRuntime;
loadedSettingsAdapter: LoadedSettingsAdapterRuntime;
trustedFolders: TrustedFoldersRuntime;
}> {
settingsRuntimePromise ??= Promise.all([
import('../config/settings.js'),
import('../config/environment.js'),
import('../config/loadedSettingsAdapter.js'),
import('../config/trustedFolders.js'),
]).then(([settings, loadedSettingsAdapter, trustedFolders]) => ({
]).then(([settings, environment, loadedSettingsAdapter, trustedFolders]) => ({
settings,
environment,
loadedSettingsAdapter,
trustedFolders,
}));
@ -738,6 +743,7 @@ async function loadServeRuntimeModules() {
daemonStatusProviderModule,
workspaceProvidersStatusModule,
workspaceSkillsStatusModule,
totalSessionAdmissionModule,
] = await Promise.all([
import('./server.js'),
import('@qwen-code/acp-bridge/bridge'),
@ -747,6 +753,7 @@ async function loadServeRuntimeModules() {
import('./daemon-status-provider.js'),
import('./workspace-providers-status.js'),
import('./workspace-skills-status.js'),
import('./total-session-admission.js'),
]);
return {
createServeApp: serverModule.createServeApp,
@ -765,6 +772,8 @@ async function loadServeRuntimeModules() {
workspaceProvidersStatusModule.createWorkspaceProvidersStatusProvider,
createWorkspaceSkillsStatusProvider:
workspaceSkillsStatusModule.createWorkspaceSkillsStatusProvider,
createTotalSessionAdmissionController:
totalSessionAdmissionModule.createTotalSessionAdmissionController,
};
}
@ -1138,6 +1147,7 @@ function createBootstrapServeApp(input: {
},
limits: {
maxSessions: advertisedMaxSessions(opts.maxSessions),
maxTotalSessions: positiveFiniteOrNull(opts.maxTotalSessions),
maxPendingPromptsPerSession: advertisedMaxPendingPromptsPerSession(
opts.maxPendingPromptsPerSession,
),
@ -1345,6 +1355,9 @@ export async function runQwenServe(
},
};
preResolveServeFastPathHomeEnvOverrides();
const daemonRuntimeBaseEnv: Readonly<NodeJS.ProcessEnv> = Object.freeze({
...process.env,
});
// Trim both sources. Common gotcha: `export QWEN_SERVER_TOKEN=$(cat
// token.txt)` keeps the file's trailing `\n` in the env value, so the
@ -1770,6 +1783,14 @@ export async function runQwenServe(
);
}
}
if (opts.maxTotalSessions !== undefined) {
if (!isNonNegativeIntegerOrInfinity(opts.maxTotalSessions)) {
throw new TypeError(
`Invalid maxTotalSessions: ${opts.maxTotalSessions}. Must be a non-negative integer ` +
`(0 / Infinity = unlimited).`,
);
}
}
if (opts.maxPendingPromptsPerSession !== undefined) {
if (!isNonNegativeIntegerOrInfinity(opts.maxPendingPromptsPerSession)) {
throw new TypeError(
@ -2014,6 +2035,64 @@ export async function runQwenServe(
{ workspace: boundWorkspace },
);
}
const runtimeEnvSnapshot = runtimeBootSettings
? settingsRuntime.environment.buildRuntimeEnvironment(
runtimeBootSettings.merged,
boundWorkspace,
daemonRuntimeBaseEnv,
)
: {
effectiveEnv: { ...daemonRuntimeBaseEnv },
overlayKeys: Object.freeze([] as string[]),
envFilePaths: Object.freeze([] as string[]),
envFileReadFailed: false,
envFileReadFailures: Object.freeze([]),
};
const logRuntimeEnvFileReadFailures = (
workspace: string,
snapshot: {
readonly envFileReadFailed: boolean;
readonly envFileReadFailures?: ReadonlyArray<{
readonly path: string;
readonly error: string;
}>;
},
): void => {
if (!snapshot.envFileReadFailed) return;
const failedFiles = snapshot.envFileReadFailures ?? [];
daemonLog.warn('one or more runtime env files could not be read', {
workspace,
...(failedFiles.length > 0 ? { failedFiles } : {}),
});
};
logRuntimeEnvFileReadFailures(boundWorkspace, runtimeEnvSnapshot);
const runtimeEffectiveEnv: NodeJS.ProcessEnv = {
...runtimeEnvSnapshot.effectiveEnv,
};
const replaceRuntimeEffectiveEnv = (
nextEnv: Readonly<NodeJS.ProcessEnv>,
): void => {
for (const key of Object.keys(runtimeEffectiveEnv)) {
delete runtimeEffectiveEnv[key];
}
Object.assign(runtimeEffectiveEnv, nextEnv);
};
const primaryRuntimeEnv: {
mode: 'runtime-overlay';
overlayKeys: string[];
envFilePaths: string[];
effectiveEnv: NodeJS.ProcessEnv;
envFileReadFailed: boolean;
envFileReadFailures: Array<{ path: string; error: string }>;
fallbackReason?: string;
} = {
mode: 'runtime-overlay' as const,
overlayKeys: [...runtimeEnvSnapshot.overlayKeys],
effectiveEnv: runtimeEffectiveEnv,
envFilePaths: [...runtimeEnvSnapshot.envFilePaths],
envFileReadFailed: runtimeEnvSnapshot.envFileReadFailed,
envFileReadFailures: [...runtimeEnvSnapshot.envFileReadFailures],
};
const daemonWorkspaceHash = core.hashDaemonWorkspace(boundWorkspace);
let daemonTelemetrySettings: TelemetrySettings;
try {
@ -2203,6 +2282,7 @@ export async function runQwenServe(
...(customIgnoreFiles !== undefined ? { customIgnoreFiles } : {}),
});
const channelFactory = runtime.createSpawnChannelFactory({
sourceEnv: runtimeEffectiveEnv,
onDiagnosticLine: diagnosticSink,
pipeHooks: {
onMessageSent: (bytes) => recordPipeMessage('outbound', bytes),
@ -2218,9 +2298,13 @@ export async function runQwenServe(
? { extraArgs: ['--experimental-lsp'] }
: {}),
});
const statusProvider = runtime.createDaemonStatusProvider();
const statusProvider = runtime.createDaemonStatusProvider({
env: runtimeEffectiveEnv,
});
const workspaceProvidersStatusProvider =
runtime.createWorkspaceProvidersStatusProvider();
runtime.createWorkspaceProvidersStatusProvider({
env: runtimeEffectiveEnv,
});
const workspaceSkillsStatusProvider =
runtime.createWorkspaceSkillsStatusProvider();
// Reverse tool channel (issue #5626, Phase 2). ONE sender registry shared
@ -2229,6 +2313,12 @@ export async function runQwenServe(
// (which registers a per-connection `ClientMcpRegistrar`'s sender on
// `mcp_register`). Inert unless `opts.clientMcpOverWs` is on.
const clientMcpSenderRegistry = new ClientMcpSenderRegistry();
const totalSessionAdmission = runtime.createTotalSessionAdmissionController(
{
maxTotalSessions: opts.maxTotalSessions,
getBridges: () => (bridgeRef ? [bridgeRef] : []),
},
);
const persistDisabledToolsFn = (
workspace: string,
toolName: string,
@ -2308,6 +2398,7 @@ export async function runQwenServe(
// connection that hosts a named client MCP server (#5626).
clientMcpSender: clientMcpSenderRegistry.lookup,
maxSessions: opts.maxSessions,
freshSessionAdmission: totalSessionAdmission.admit,
...(opts.maxPendingPromptsPerSession !== undefined
? { maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession }
: {}),
@ -2363,10 +2454,64 @@ export async function runQwenServe(
const fresh = settingsRuntime.settings.loadSettings(workspace, {
skipLoadEnvironment: true,
});
return settingsRuntime.settings.reloadEnvironment(
const result = settingsRuntime.settings.reloadEnvironment(
fresh.merged,
workspace,
);
let refreshedRuntimeEnv: ReturnType<
EnvironmentRuntime['buildRuntimeEnvironment']
>;
let fallbackReason: string | undefined;
try {
refreshedRuntimeEnv =
settingsRuntime.environment.buildRuntimeEnvironment(
fresh.merged,
workspace,
daemonRuntimeBaseEnv,
);
} catch (err) {
fallbackReason = err instanceof Error ? err.message : String(err);
daemonLog.warn(
'failed to rebuild runtime env snapshot after daemon env reload; preserving previous runtime env',
{
error: fallbackReason,
},
);
refreshedRuntimeEnv = {
effectiveEnv: { ...runtimeEffectiveEnv },
overlayKeys: [...primaryRuntimeEnv.overlayKeys],
envFilePaths: [...primaryRuntimeEnv.envFilePaths],
envFileReadFailed: primaryRuntimeEnv.envFileReadFailed ?? false,
envFileReadFailures: [
...(primaryRuntimeEnv.envFileReadFailures ?? []),
],
};
}
logRuntimeEnvFileReadFailures(workspace, refreshedRuntimeEnv);
replaceRuntimeEffectiveEnv(refreshedRuntimeEnv.effectiveEnv);
if (fallbackReason) {
primaryRuntimeEnv.fallbackReason = fallbackReason;
} else {
delete primaryRuntimeEnv.fallbackReason;
}
primaryRuntimeEnv.envFileReadFailed =
refreshedRuntimeEnv.envFileReadFailed;
primaryRuntimeEnv.envFileReadFailures.splice(
0,
primaryRuntimeEnv.envFileReadFailures.length,
...refreshedRuntimeEnv.envFileReadFailures,
);
primaryRuntimeEnv.overlayKeys.splice(
0,
primaryRuntimeEnv.overlayKeys.length,
...refreshedRuntimeEnv.overlayKeys,
);
primaryRuntimeEnv.envFilePaths.splice(
0,
primaryRuntimeEnv.envFilePaths.length,
...refreshedRuntimeEnv.envFilePaths,
);
return result;
}),
queryWorkspaceStatus: (method, idle) =>
bridge.queryWorkspaceStatus(method, idle),
@ -2537,6 +2682,7 @@ export async function runQwenServe(
...(customIgnoreFiles !== undefined ? { customIgnoreFiles } : {}),
}),
primaryWorkspaceTrusted: trustedWorkspace,
primaryRuntimeEnv,
daemonLog,
getChannelWorkerSnapshot,
getPerfSnapshot: () => ({
@ -2556,6 +2702,7 @@ export async function runQwenServe(
},
}),
getMetricsSeries: () => metricsRing.snapshot(),
getTotalSessionAdmissionSnapshot: totalSessionAdmission.snapshot,
recordDaemonRequest: (durationMs, statusCode) =>
metricsRing.recordRequest(durationMs, statusCode),
workspace: workspaceService,

View file

@ -13,8 +13,8 @@ import { describe, it, expect, afterEach, beforeEach, vi } from 'vitest';
import request from 'supertest';
import { trace, type Span } from '@opentelemetry/api';
import {
computeKeepaliveIntervalMs,
createServeApp,
computeKeepaliveIntervalMs,
detectFromLoopback,
listWorkspaceSessionsForResponse,
PromptDeadlineExceededError,
@ -72,6 +72,7 @@ import {
SessionBusyError,
SessionLimitExceededError,
SessionNotFoundError,
TotalSessionLimitExceededError,
WorkspaceMismatchError,
type BridgeHeartbeatResult,
type BridgeHeartbeatState,
@ -122,6 +123,14 @@ import {
ClientMcpSenderRegistry,
createClientMcpServerProvider,
} from './acp-http/client-mcp-sender-registry.js';
import {
DeviceFlowRegistry,
TooManyActiveDeviceFlowsError,
UpstreamDeviceFlowError,
brandSecret,
type DeviceFlowProvider,
type DeviceFlowRegistry as DeviceFlowRegistryType,
} from './auth/device-flow.js';
import { resetHomeEnvBootstrapForTesting } from '../config/settings.js';
import {
resetTrustedFoldersForTesting,
@ -1842,29 +1851,6 @@ function abortableBridgePromptImpl(): FakeBridgeOpts['promptImpl'] {
});
}
describe('computeKeepaliveIntervalMs', () => {
it('keeps the heartbeat strictly under the reaper window (beats before reap)', () => {
// A small custom idle timeout must not get an interval ≥ the window, or the
// session is reaped before the first heartbeat. Half the window is the cap.
for (const idle of [1_000, 2_000, 10_000, 40_000, 60_000, 120_000]) {
const interval = computeKeepaliveIntervalMs(idle);
expect(interval).toBeLessThanOrEqual(Math.floor(idle / 2));
expect(interval).toBeGreaterThanOrEqual(1);
}
});
it('uses ~a third of a large window (default 30m → 10m cap)', () => {
expect(computeKeepaliveIntervalMs(30 * 60_000)).toBe(10 * 60_000); // MAX cap
expect(computeKeepaliveIntervalMs(15 * 60_000)).toBe(5 * 60_000); // /3
});
it('returns the relaxed max cadence when the reaper is disabled (≤ 0)', () => {
// No reaping → no heartbeat pressure, but the revive loop still runs.
expect(computeKeepaliveIntervalMs(0)).toBe(10 * 60_000);
expect(computeKeepaliveIntervalMs(-5)).toBe(10 * 60_000);
});
});
describe('createServeApp', () => {
it('rejects client-MCP over WS with an injected bridge but no matching sender registry', () => {
expect(() =>
@ -12081,8 +12067,55 @@ describe('createServeApp', () => {
expect(res.body).toMatchObject({
code: 'session_limit_exceeded',
limit: 20,
scope: 'workspace',
});
});
it('503 + Retry-After + total scope and daemon log when bridge throws TotalSessionLimitExceededError', async () => {
const bridge = fakeBridge({
spawnImpl: async () => {
throw new TotalSessionLimitExceededError(10);
},
});
const daemonLog = fakeDaemonLog();
const app = createServeApp(baseOpts, undefined, {
bridge,
daemonLog,
});
const res = await request(app)
.post('/session')
.set('Host', `127.0.0.1:${baseOpts.port}`)
.send({ cwd: '/work/a' });
expect(res.status).toBe(503);
expect(res.headers['retry-after']).toBe('5');
expect(res.body).toMatchObject({
code: 'session_limit_exceeded',
limit: 10,
scope: 'total',
});
expect(daemonLog.warn).toHaveBeenCalledWith(
'total session admission rejected',
expect.objectContaining({
route: 'POST /session',
limit: 10,
scope: 'total',
}),
);
});
});
});
describe('computeKeepaliveIntervalMs', () => {
it('keeps the first heartbeat inside small custom idle windows', () => {
expect(computeKeepaliveIntervalMs(10_000)).toBe(5_000);
});
it('caps large idle windows at the relaxed maximum cadence', () => {
expect(computeKeepaliveIntervalMs(60 * 60_000)).toBe(10 * 60_000);
});
it('uses the relaxed cadence when idle reaping is disabled', () => {
expect(computeKeepaliveIntervalMs(0)).toBe(10 * 60_000);
});
});
@ -12091,12 +12124,11 @@ describe('runQwenServe', () => {
let runtimeDir: string | undefined;
beforeEach(async () => {
// These tests spawn a real daemon bound to the repo cwd. Redirect the
// per-project runtime dir (where scheduled_tasks.json lives) into a temp
// dir so the daemon's scheduled-task rehydration reads an empty schedule
// instead of the developer's real ~/.qwen — otherwise it would try to
// reload a real bound session and hang these startup/shutdown tests.
runtimeDir = await fsp.mkdtemp(path.join(os.tmpdir(), 'run-qwen-serve-'));
runtimeDir = await fsp.mkdtemp(
path.join(os.tmpdir(), 'qwen-serve-runtime-'),
);
// Keep real scheduled task state out of startup/shutdown tests; otherwise a
// developer's ~/.qwen/scheduled_tasks.json could rehydrate sessions here.
Storage.setRuntimeBaseDir(runtimeDir);
});
@ -14185,6 +14217,94 @@ describe('createServeApp ServeAppDeps.fsFactory wiring (#4175 PR 18)', () => {
expect(locals.workspaceRegistry!.primary.trusted).toBe(true);
});
it('threads primary runtime env metadata into the default registry runtime', async () => {
const { createServeApp } = await import('./server.js');
const primaryRuntimeEnv = {
mode: 'runtime-overlay',
overlayKeys: ['OPENAI_API_KEY'],
envFilePaths: [],
effectiveEnv: { OPENAI_API_KEY: 'runtime-key' },
} as const;
const app = createServeApp(
{
port: 0,
hostname: '127.0.0.1',
workspace: '/work/bound',
} as Parameters<typeof createServeApp>[0],
() => 0,
{ primaryRuntimeEnv } as Parameters<typeof createServeApp>[2],
);
const locals = app.locals as { workspaceRegistry?: WorkspaceRegistry };
expect(locals.workspaceRegistry!.primary.env).toBe(primaryRuntimeEnv);
});
it('wires total admission into the internally-created bridge', async () => {
type AdmissionContext = {
operation: 'spawn';
workspaceCwd: string;
};
let freshSessionAdmission:
| ((context: AdmissionContext) => { release(): void })
| undefined;
vi.resetModules();
vi.doMock('./acp-session-bridge.js', async () => {
const actual = await vi.importActual<
typeof import('./acp-session-bridge.js')
>('./acp-session-bridge.js');
return {
...actual,
createAcpSessionBridge: vi.fn((opts: unknown) => {
freshSessionAdmission = (
opts as {
freshSessionAdmission?: typeof freshSessionAdmission;
}
).freshSessionAdmission;
return fakeBridge();
}),
};
});
try {
const { createServeApp } = await import('./server.js');
createServeApp(
{
port: 0,
hostname: '127.0.0.1',
workspace: WS_BOUND,
maxTotalSessions: 1,
} as Parameters<typeof createServeApp>[0],
() => 0,
);
expect(freshSessionAdmission).toBeDefined();
const first = freshSessionAdmission!({
operation: 'spawn',
workspaceCwd: WS_BOUND,
});
let rejection: unknown;
try {
freshSessionAdmission!({
operation: 'spawn',
workspaceCwd: WS_BOUND,
});
} catch (err) {
rejection = err;
}
expect(rejection).toMatchObject({
name: 'TotalSessionLimitExceededError',
limit: 1,
scope: 'total',
operation: 'spawn',
workspaceCwd: WS_BOUND,
});
first.release();
} finally {
vi.doUnmock('./acp-session-bridge.js');
vi.resetModules();
}
});
it('uses an injected workspace registry as the primary runtime source', async () => {
const { createServeApp } = await import('./server.js');
const runtime = makeInjectedWorkspaceRuntime();
@ -14472,7 +14592,7 @@ describe('auth device-flow routes', () => {
// whose `poll` is scripted per-test. Lives at the top of the suite so
// every `it()` can compose it with the registry.
function makeFakeProvider(): {
provider: import('./auth/device-flow.js').DeviceFlowProvider;
provider: DeviceFlowProvider;
startCount: () => number;
} {
let starts = 0;
@ -14485,12 +14605,8 @@ describe('auth device-flow routes', () => {
deviceCode:
// Use the brandSecret helper so the secret follows the same
// redaction shape the production provider produces.
(await import('./auth/device-flow.js')).brandSecret(
`device-${starts}`,
),
pkceVerifier: (await import('./auth/device-flow.js')).brandSecret(
`pkce-${starts}`,
),
brandSecret(`device-${starts}`),
pkceVerifier: brandSecret(`pkce-${starts}`),
userCode: `USER-${starts}`,
verificationUri: 'https://idp.example/verify',
verificationUriComplete: 'https://idp.example/verify?u=AB12',
@ -14971,17 +15087,15 @@ describe('auth device-flow routes', () => {
// must surface as 502 with code:'upstream_error' instead of falling
// through `sendBridgeError`'s generic 500 path. Build a fake
// provider whose start always throws.
const { UpstreamDeviceFlowError } = await import('./auth/device-flow.js');
const failingProvider: import('./auth/device-flow.js').DeviceFlowProvider =
{
providerId: 'qwen-oauth',
async start() {
throw new UpstreamDeviceFlowError('mocked upstream outage');
},
async poll() {
return { kind: 'pending' as const };
},
};
const failingProvider: DeviceFlowProvider = {
providerId: 'qwen-oauth',
async start() {
throw new UpstreamDeviceFlowError('mocked upstream outage');
},
async poll() {
return { kind: 'pending' as const };
},
};
const bridge = fakeBridge();
const app = createServeApp({ ...baseOpts, token: 'tkn' }, undefined, {
bridge,
@ -15000,10 +15114,7 @@ describe('auth device-flow routes', () => {
it('sweeper-driven auto-expiry transitions a stale entry to status:error and surfaces over GET', async () => {
// PR 21 fold-in 0 P1-13: cover the time-based expiry path via an
// injected registry with a controlled clock + manual sweeper trigger.
const { DeviceFlowRegistry, brandSecret } = await import(
'./auth/device-flow.js'
);
const fakeProvider: import('./auth/device-flow.js').DeviceFlowProvider = {
const fakeProvider: DeviceFlowProvider = {
providerId: 'qwen-oauth',
async start() {
return {
@ -15103,9 +15214,6 @@ describe('auth device-flow routes', () => {
it('POST returns 409 too_many_active_flows when registry cap is reached', async () => {
// Inject a fake registry whose `start` always throws the cap error.
const { TooManyActiveDeviceFlowsError } = await import(
'./auth/device-flow.js'
);
const fakeRegistry = {
start: async () => {
throw new TooManyActiveDeviceFlowsError();
@ -15114,7 +15222,7 @@ describe('auth device-flow routes', () => {
cancel: () => undefined,
listPending: () => [],
dispose: () => {},
} as unknown as import('./auth/device-flow.js').DeviceFlowRegistry;
} as unknown as DeviceFlowRegistryType;
const bridge = fakeBridge();
const app = createServeApp({ ...baseOpts, token: 'tkn' }, undefined, {

View file

@ -96,6 +96,10 @@ import {
} from './routes/workspace-voice.js';
import { registerA2uiActionRoutes } from './routes/a2ui-action.js';
import { setRateLimiter } from './rate-limit.js';
import {
createTotalSessionAdmissionController,
type TotalSessionAdmissionSnapshot,
} from './total-session-admission.js';
import {
sendBridgeError as sendBridgeErrorResponse,
sendPermissionVoteError as sendPermissionVoteErrorResponse,
@ -122,6 +126,7 @@ import { installSelfOriginStripMiddleware } from './server/self-origin.js';
import {
createSingleWorkspaceRegistry,
type WorkspaceRegistry,
type WorkspaceRuntimeEnvMetadata,
} from './workspace-registry.js';
import { registerWorkspaceLifecycleRoutes } from './routes/workspace-lifecycle.js';
import { registerWorkspaceMcpControlRoutes } from './routes/workspace-mcp-control.js';
@ -163,6 +168,12 @@ function describeRegistryPrimaryForConflict(
);
}
function getRuntimeEffectiveEnv(
metadata: WorkspaceRuntimeEnvMetadata | undefined,
): Readonly<Record<string, string | undefined>> | undefined {
return metadata?.effectiveEnv;
}
export interface ServeAppDeps {
/** Bridge instance; tests inject a fake. Defaults to a fresh real one. */
bridge?: AcpSessionBridge;
@ -247,6 +258,7 @@ export interface ServeAppDeps {
getPerfSnapshot?: () => DaemonPerfSnapshot;
/** Rolling metrics series for the Daemon Status charts (oldest→newest). */
getMetricsSeries?: () => DaemonMetricsBucket[];
getTotalSessionAdmissionSnapshot?: () => TotalSessionAdmissionSnapshot;
/**
* Sink fed one (durationMs, statusCode) per matched daemon HTTP request, so
* the metrics ring can bucket request rate and latency for the charts.
@ -287,6 +299,7 @@ export interface ServeAppDeps {
clientMcpSenderRegistry?: ClientMcpSenderRegistry;
workspaceRegistry?: WorkspaceRegistry;
primaryWorkspaceTrusted?: boolean;
primaryRuntimeEnv?: WorkspaceRuntimeEnvMetadata;
voiceTranscriber?: WorkspaceVoiceRouteDeps['transcribe'];
}
@ -477,6 +490,9 @@ export function createServeApp(
injectedWorkspaceRegistry?.primary.clientMcpSenderRegistry ??
deps.clientMcpSenderRegistry ??
new ClientMcpSenderRegistry();
const primaryRuntimeEnvMetadata =
injectedWorkspaceRegistry?.primary.env ?? deps.primaryRuntimeEnv;
const primaryEffectiveEnv = getRuntimeEffectiveEnv(primaryRuntimeEnvMetadata);
const { languageCodes, currentServeFeatures, invalidateServeFeaturesCache } =
createServeFeatures({
opts,
@ -488,12 +504,28 @@ export function createServeApp(
deps.workspace !== undefined || injectedWorkspaceRegistry !== undefined,
sessionShellCommandEnabled,
});
const statusProvider = deps.statusProvider ?? createDaemonStatusProvider();
const statusProvider =
deps.statusProvider ??
createDaemonStatusProvider(
primaryEffectiveEnv ? { env: primaryEffectiveEnv } : {},
);
let defaultBridgeForAdmission: AcpSessionBridge | undefined;
const totalSessionAdmission =
!deps.bridge && !injectedWorkspaceRegistry
? createTotalSessionAdmissionController({
maxTotalSessions: opts.maxTotalSessions,
getBridges: () =>
defaultBridgeForAdmission ? [defaultBridgeForAdmission] : [],
})
: undefined;
const bridge =
injectedWorkspaceRegistry?.primary.bridge ??
deps.bridge ??
createAcpSessionBridge({
maxSessions: opts.maxSessions,
...(totalSessionAdmission
? { freshSessionAdmission: totalSessionAdmission.admit }
: {}),
maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession,
eventRingSize: opts.eventRingSize,
permissionResponseTimeoutMs: opts.permissionResponseTimeoutMs,
@ -509,6 +541,9 @@ export function createServeApp(
// ext-method by reaching the WS connection that hosts the named server.
clientMcpSender: clientMcpSenderRegistry.lookup,
});
if (!injectedWorkspaceRegistry && !deps.bridge) {
defaultBridgeForAdmission = bridge;
}
const archiveCoordinator = new SessionArchiveCoordinator();
installSelfOriginStripMiddleware(app, getPort);
@ -547,8 +582,9 @@ export function createServeApp(
boundWorkspace,
contextFilename: deps.contextFilename ?? 'QWEN.md',
statusProvider,
workspaceProvidersStatusProvider:
createWorkspaceProvidersStatusProvider(),
workspaceProvidersStatusProvider: createWorkspaceProvidersStatusProvider(
primaryEffectiveEnv ? { env: primaryEffectiveEnv } : {},
),
workspaceSkillsStatusProvider: createWorkspaceSkillsStatusProvider(),
isChannelLive: () => bridge.isChannelLive(),
persistDisabledTools:
@ -585,7 +621,10 @@ export function createServeApp(
workspaceCwd: boundWorkspace,
primary: true,
trusted: deps.primaryWorkspaceTrusted ?? false,
env: { mode: 'parent-process', overlayKeys: [] },
env: primaryRuntimeEnvMetadata ?? {
mode: 'parent-process',
overlayKeys: [],
},
bridge,
workspaceService: workspace,
routeFileSystemFactory: fsFactory,
@ -728,6 +767,8 @@ export function createServeApp(
getChannelWorkerSnapshot: deps.getChannelWorkerSnapshot,
getPerfSnapshot: deps.getPerfSnapshot,
getMetricsSeries: deps.getMetricsSeries,
getTotalSessionAdmissionSnapshot:
deps.getTotalSessionAdmissionSnapshot ?? totalSessionAdmission?.snapshot,
});
registerCapabilitiesRoutes(app, {
@ -870,6 +911,7 @@ export function createServeApp(
boundWorkspace: primaryBoundWorkspace,
mutate,
safeBody,
env: getRuntimeEffectiveEnv(primaryRuntime.env),
// UI-server discovery uses the daemon's workspace MCP status, which
// includes servers registered at runtime.
getMcpServers: async () => {
@ -1074,7 +1116,9 @@ export function createServeApp(
extraWsRoutes: [
{
path: '/voice/stream',
onConnection: createVoiceWsConnectionHandler(primaryBoundWorkspace),
onConnection: createVoiceWsConnectionHandler(primaryBoundWorkspace, {
env: getRuntimeEffectiveEnv(primaryRuntime.env),
}),
},
],
});

View file

@ -40,6 +40,7 @@ import {
WorkspaceInitRaceError,
WorkspaceInitSymlinkError,
WorkspaceMismatchError,
TotalSessionLimitExceededError,
} from '../acp-session-bridge.js';
import type { DaemonLogger } from '../daemon-logger.js';
@ -368,6 +369,37 @@ export function sendBridgeError(
error: err.message,
code: 'session_limit_exceeded',
limit: err.limit,
scope: 'workspace',
});
return;
}
if (err instanceof TotalSessionLimitExceededError) {
const totalSessionError = err as TotalSessionLimitExceededError & {
operation?: string;
workspaceCwd?: string;
sourceSessionId?: string;
};
daemonLog?.warn('total session admission rejected', {
...(ctx?.route ? { route: ctx.route } : {}),
...(ctx?.sessionId ? { sessionId: ctx.sessionId } : {}),
limit: totalSessionError.limit,
scope: totalSessionError.scope,
...(totalSessionError.operation
? { operation: totalSessionError.operation }
: {}),
...(totalSessionError.workspaceCwd
? { workspaceCwd: totalSessionError.workspaceCwd }
: {}),
...(totalSessionError.sourceSessionId
? { sourceSessionId: totalSessionError.sourceSessionId }
: {}),
});
res.set('Retry-After', '5');
res.status(503).json({
error: err.message,
code: 'session_limit_exceeded',
limit: err.limit,
scope: err.scope,
});
return;
}

View file

@ -0,0 +1,89 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { describe, expect, it } from 'vitest';
import { TotalSessionLimitExceededError } from './acp-session-bridge.js';
import { createTotalSessionAdmissionController } from './total-session-admission.js';
describe('createTotalSessionAdmissionController', () => {
it('counts live bridge sessions plus in-flight reservations across runtimes', () => {
const bridges = [{ sessionCount: 1 }, { sessionCount: 0 }];
const admission = createTotalSessionAdmissionController({
maxTotalSessions: 2,
getBridges: () => bridges,
});
const reservation = admission.admit({
operation: 'spawn',
workspaceCwd: '/work/a',
});
expect(admission.snapshot()).toEqual({ liveCount: 1, inFlight: 1 });
expect(() =>
admission.admit({ operation: 'spawn', workspaceCwd: '/work/b' }),
).toThrow(TotalSessionLimitExceededError);
try {
admission.admit({ operation: 'spawn', workspaceCwd: '/work/b' });
} catch (err) {
expect(err).toMatchObject({
operation: 'spawn',
workspaceCwd: '/work/b',
});
}
if (!reservation) throw new Error('expected reservation');
reservation.release();
expect(admission.snapshot()).toEqual({ liveCount: 1, inFlight: 0 });
bridges[1]!.sessionCount = 1;
expect(admission.snapshot()).toEqual({ liveCount: 2, inFlight: 0 });
expect(() =>
admission.admit({ operation: 'spawn', workspaceCwd: '/work/c' }),
).toThrow(TotalSessionLimitExceededError);
});
it('treats undefined, zero, and Infinity as unlimited', () => {
for (const maxTotalSessions of [undefined, 0, Infinity]) {
const admission = createTotalSessionAdmissionController({
maxTotalSessions,
getBridges: () => [{ sessionCount: 99 }],
});
const reservation = admission.admit({
operation: 'spawn',
workspaceCwd: '/work/a',
});
if (!reservation) throw new Error('expected reservation');
reservation.release();
}
});
it('ignores duplicate release calls', () => {
const admission = createTotalSessionAdmissionController({
maxTotalSessions: 1,
getBridges: () => [{ sessionCount: 0 }],
});
const reservation = admission.admit({
operation: 'spawn',
workspaceCwd: '/work/a',
});
if (!reservation) throw new Error('expected reservation');
expect(() =>
admission.admit({ operation: 'spawn', workspaceCwd: '/work/b' }),
).toThrow(TotalSessionLimitExceededError);
reservation.release();
reservation.release();
const nextReservation = admission.admit({
operation: 'spawn',
workspaceCwd: '/work/c',
});
if (!nextReservation) throw new Error('expected reservation');
nextReservation.release();
});
});

View file

@ -0,0 +1,80 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import {
TotalSessionLimitExceededError,
type BridgeFreshSessionAdmission,
type BridgeFreshSessionAdmissionContext,
type BridgeFreshSessionReservation,
} from './acp-session-bridge.js';
interface SessionCountSource {
readonly sessionCount: number;
}
export interface TotalSessionAdmissionOptions {
readonly maxTotalSessions?: number;
readonly getBridges: () => readonly SessionCountSource[];
}
export interface TotalSessionAdmissionSnapshot {
readonly liveCount: number;
readonly inFlight: number;
}
export interface TotalSessionAdmissionController {
readonly admit: BridgeFreshSessionAdmission;
readonly snapshot: () => TotalSessionAdmissionSnapshot;
}
export function createTotalSessionAdmissionController({
maxTotalSessions,
getBridges,
}: TotalSessionAdmissionOptions): TotalSessionAdmissionController {
let inFlight = 0;
const limit =
maxTotalSessions === undefined ||
maxTotalSessions === 0 ||
maxTotalSessions === Number.POSITIVE_INFINITY
? Number.POSITIVE_INFINITY
: maxTotalSessions;
return {
admit(
context: BridgeFreshSessionAdmissionContext,
): BridgeFreshSessionReservation {
if (limit !== Number.POSITIVE_INFINITY) {
if (getLiveCount(getBridges()) + inFlight >= limit) {
throw Object.assign(new TotalSessionLimitExceededError(limit), {
operation: context.operation,
workspaceCwd: context.workspaceCwd,
...(context.sessionId ? { sessionId: context.sessionId } : {}),
...(context.sourceSessionId
? { sourceSessionId: context.sourceSessionId }
: {}),
});
}
}
inFlight++;
let released = false;
return {
release() {
if (released) return;
released = true;
inFlight--;
},
};
},
snapshot() {
return { liveCount: getLiveCount(getBridges()), inFlight };
},
};
}
function getLiveCount(bridges: readonly SessionCountSource[]): number {
return bridges.reduce((sum, bridge) => sum + bridge.sessionCount, 0);
}

View file

@ -57,6 +57,12 @@ export interface ServeOptions {
* `0` or `Infinity` to disable.
*/
maxSessions?: number;
/**
* Non-negative integer cap on concurrent live sessions across all workspace
* runtimes. Defaults to unlimited until multi-workspace sessions are ungated.
* `0` or `Infinity` disables the cap.
*/
maxTotalSessions?: number;
/**
* Per-session cap on accepted prompts that have not settled yet.
* Defaults to 5. `0` or `Infinity` disables the cap.

View file

@ -0,0 +1,91 @@
/**
* @license
* Copyright 2026 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { AuthType } from '@qwen-code/qwen-code-core';
import { afterEach, describe, expect, it, vi } from 'vitest';
const mocks = vi.hoisted(() => ({
loadSettings: vi.fn(),
getAuthTypeFromEnv: vi.fn(),
resolveCliGenerationConfig: vi.fn(),
resolveVoiceTranscriptionConfig: vi.fn(),
isStreamingVoiceModel: vi.fn(),
}));
vi.mock('../../config/settings.js', () => ({
loadSettings: mocks.loadSettings,
}));
vi.mock('../../utils/modelConfigUtils.js', () => ({
getAuthTypeFromEnv: mocks.getAuthTypeFromEnv,
resolveCliGenerationConfig: mocks.resolveCliGenerationConfig,
}));
vi.mock('../../services/voice-transcriber.js', () => ({
resolveVoiceTranscriptionConfig: mocks.resolveVoiceTranscriptionConfig,
isStreamingVoiceModel: mocks.isStreamingVoiceModel,
}));
const originalDashscopeApiKey = process.env['DASHSCOPE_API_KEY'];
describe('loadDaemonVoiceContext', () => {
afterEach(() => {
if (originalDashscopeApiKey === undefined) {
delete process.env['DASHSCOPE_API_KEY'];
} else {
process.env['DASHSCOPE_API_KEY'] = originalDashscopeApiKey;
}
vi.resetModules();
vi.resetAllMocks();
});
it('uses the injected runtime env for voice auth and model config resolution', async () => {
const injectedEnv = {
DASHSCOPE_API_KEY: 'runtime-key',
OPENAI_API_KEY: 'runtime-openai-key',
};
process.env['DASHSCOPE_API_KEY'] = 'process-key';
mocks.loadSettings.mockReturnValue({
merged: {
voiceModel: 'qwen3-asr-flash',
modelProviders: {},
},
});
mocks.getAuthTypeFromEnv.mockReturnValue(AuthType.USE_OPENAI);
mocks.resolveCliGenerationConfig.mockReturnValue({
generationConfig: {},
sources: {},
});
mocks.isStreamingVoiceModel.mockReturnValue(false);
const { loadDaemonVoiceContext } = await import(
'./resolve-voice-config.js'
);
const context = loadDaemonVoiceContext('/work/voice', { env: injectedEnv });
expect(context.voiceModel).toBe('qwen3-asr-flash');
expect(context.env).toBe(injectedEnv);
expect(mocks.getAuthTypeFromEnv).toHaveBeenCalledWith(injectedEnv);
expect(mocks.resolveCliGenerationConfig).toHaveBeenCalledWith(
expect.objectContaining({
selectedAuthType: AuthType.USE_OPENAI,
env: injectedEnv,
}),
);
expect(mocks.resolveVoiceTranscriptionConfig).toHaveBeenCalledWith(
expect.objectContaining({
env: injectedEnv,
settings: expect.objectContaining({
merged: expect.objectContaining({ voiceModel: 'qwen3-asr-flash' }),
}),
voiceModel: 'qwen3-asr-flash',
}),
);
expect(mocks.loadSettings).toHaveBeenCalledWith('/work/voice', {
skipLoadEnvironment: true,
});
});
});

View file

@ -11,6 +11,7 @@ import {
getAuthTypeFromEnv,
resolveCliGenerationConfig,
} from '../../utils/modelConfigUtils.js';
import { snapshotProcessEnv } from '../env-snapshot.js';
import {
isStreamingVoiceModel,
resolveVoiceTranscriptionConfig,
@ -27,6 +28,7 @@ export interface DaemonVoiceContext {
settings: LoadedSettings;
/** A `ModelsConfig` — satisfies the resolver's structural `getAllConfiguredModels`. */
models: VoiceModelLookup;
env?: Readonly<Record<string, string | undefined>>;
voiceModel: string;
/** True for realtime models (open an upstream WS); false → batch on stop. */
streaming: boolean;
@ -44,15 +46,18 @@ function readVoiceModel(settings: LoadedSettings): string | undefined {
* `workspace-providers-status.ts` so the daemon resolves the same configured
* models the CLI would without constructing a full CLI `Config`.
*/
function buildModelsConfig(settings: LoadedSettings): ModelsConfig {
function buildModelsConfig(
settings: LoadedSettings,
env: Readonly<Record<string, string | undefined>>,
): ModelsConfig {
const merged = settings.merged;
const selectedAuthType =
merged.security?.auth?.selectedType ?? getAuthTypeFromEnv();
merged.security?.auth?.selectedType ?? getAuthTypeFromEnv(env);
const resolvedCliConfig = resolveCliGenerationConfig({
argv: {},
settings: merged,
selectedAuthType,
env: process.env as Record<string, string | undefined>,
env,
});
return new ModelsConfig({
initialAuthType: selectedAuthType,
@ -69,18 +74,31 @@ function buildModelsConfig(settings: LoadedSettings): ModelsConfig {
*/
export function loadDaemonVoiceContext(
workspaceCwd: string,
options: { env?: Readonly<Record<string, string | undefined>> } = {},
): DaemonVoiceContext {
const settings = loadSettings(workspaceCwd);
const settings = loadSettings(
workspaceCwd,
options.env ? { skipLoadEnvironment: true } : true,
);
const voiceModel = readVoiceModel(settings);
if (!voiceModel) {
throw new Error('No voice model is configured for this workspace.');
}
const models = buildModelsConfig(settings);
const models = buildModelsConfig(
settings,
options.env ?? snapshotProcessEnv(),
);
// Validates transcribable + baseUrl + apiKey presence (throws otherwise).
resolveVoiceTranscriptionConfig({ config: models, settings, voiceModel });
resolveVoiceTranscriptionConfig({
config: models,
settings,
voiceModel,
env: options.env,
});
return {
settings,
models,
...(options.env ? { env: options.env } : {}),
voiceModel,
streaming: isStreamingVoiceModel(voiceModel),
};

View file

@ -68,6 +68,7 @@ function encodeWav(pcm: Uint8Array): Uint8Array {
/** Injection seams for unit tests; production uses the reused CLI pipeline. */
export interface VoiceWsDeps {
loadContext?: (workspaceCwd: string) => DaemonVoiceContext;
env?: Readonly<Record<string, string | undefined>>;
openStream?: (
ctx: DaemonVoiceContext,
callbacks: VoiceStreamCallbacks,
@ -84,6 +85,7 @@ async function defaultOpenStream(
config: ctx.models,
settings: ctx.settings,
voiceModel: ctx.voiceModel,
env: ctx.env,
});
await assertVoiceBaseUrlNetworkAllowed(cfg);
return await openVoiceStreamWithRetry(() =>
@ -103,7 +105,12 @@ function defaultTranscribe(
): Promise<string> {
return transcribeVoiceAudio(
{ data: encodeWav(pcm), mimeType: 'audio/wav' },
{ config: ctx.models, settings: ctx.settings, voiceModel: ctx.voiceModel },
{
config: ctx.models,
settings: ctx.settings,
voiceModel: ctx.voiceModel,
env: ctx.env,
},
).catch((error: unknown) => {
debugLogger.debug(
`[voice-ws] batch transcription error: ${errMessage(error)}`,
@ -171,7 +178,10 @@ export function createVoiceWsConnectionHandler(
boundWorkspace: string,
deps: VoiceWsDeps = {},
): (ws: WebSocket, req: IncomingMessage) => void {
const loadContext = deps.loadContext ?? loadDaemonVoiceContext;
const loadContext =
deps.loadContext ??
((workspaceCwd: string) =>
loadDaemonVoiceContext(workspaceCwd, { env: deps.env }));
const openStream = deps.openStream ?? defaultOpenStream;
const transcribe = deps.transcribe ?? defaultTranscribe;
// Shared across all connections from this daemon (factory closure).

View file

@ -353,6 +353,30 @@ describe('createWorkspaceProvidersStatusProvider', () => {
).toBe(true);
});
it('does not load workspace env files into process.env when env is injected', async () => {
const originalOpenaiApiKey = process.env['OPENAI_API_KEY'];
delete process.env['OPENAI_API_KEY'];
await fs.writeFile(path.join(workspace, '.env'), 'OPENAI_API_KEY=leak');
await writeUserSettings({
security: { auth: { selectedType: 'openai' } },
modelProviders: {
openai: [{ id: 'env-model', name: 'Env Model' }],
},
});
const provider = createWorkspaceProvidersStatusProvider({
env: { OPENAI_MODEL: 'env-model', OPENAI_API_KEY: 'runtime-key' },
});
try {
const result = await provider(workspace, false);
expect(result.current?.modelId).toBe('env-model(openai)');
expect(process.env['OPENAI_API_KEY']).toBeUndefined();
} finally {
restoreEnv('OPENAI_API_KEY', originalOpenaiApiKey);
}
});
it('includes only non-empty fast model settings in current selection', async () => {
const provider = createWorkspaceProvidersStatusProvider({ env: {} });
await writeUserSettings({

View file

@ -32,6 +32,7 @@ import {
parseAcpBaseModelId,
sanitizeProviderBaseUrl,
} from '../utils/acpModelUtils.js';
import { snapshotProcessEnv } from './env-snapshot.js';
const debugLogger = createDebugLogger('WORKSPACE_PROVIDERS_STATUS');
@ -58,12 +59,14 @@ function buildWorkspaceProvidersStatus(
options: WorkspaceProvidersStatusProviderOptions,
): ServeWorkspaceProvidersStatus {
try {
const loaded = loadSettings(workspaceCwd);
const loaded = loadSettings(
workspaceCwd,
options.env ? { skipLoadEnvironment: true } : true,
);
const settings = loaded.merged;
const env =
options.env ?? (process.env as Record<string, string | undefined>);
const env = options.env ?? snapshotProcessEnv();
const selectedAuthType =
settings.security?.auth?.selectedType ?? getAuthTypeFromEnv();
settings.security?.auth?.selectedType ?? getAuthTypeFromEnv(env);
const argv: CliGenerationConfigInputs['argv'] = {
model: options.argv?.model,
openaiApiKey: options.argv?.openaiApiKey,

View file

@ -13,8 +13,16 @@ import type { WorkspaceFileSystemFactory } from './fs/index.js';
import type { DaemonWorkspaceService } from './workspace-service/types.js';
export interface WorkspaceRuntimeEnvMetadata {
readonly mode: 'parent-process';
readonly mode: 'parent-process' | 'runtime-overlay';
readonly overlayKeys: readonly string[];
readonly effectiveEnv?: Readonly<NodeJS.ProcessEnv>;
readonly envFilePaths?: readonly string[];
readonly envFileReadFailed?: boolean;
readonly envFileReadFailures?: ReadonlyArray<{
readonly path: string;
readonly error: string;
}>;
readonly fallbackReason?: string;
}
export interface WorkspaceRuntime {

View file

@ -65,6 +65,7 @@ interface ResolveVoiceTranscriptionConfigArgs {
config: VoiceModelSource;
settings: LoadedSettings;
voiceModel: string;
env?: Readonly<Record<string, string | undefined>>;
}
interface TranscribeVoiceAudioArgs extends ResolveVoiceTranscriptionConfigArgs {
@ -241,12 +242,13 @@ function readApiKey(
settings: LoadedSettings,
model: AvailableModel,
baseUrl: string,
env: Readonly<Record<string, string | undefined>> | undefined,
): string | undefined {
if (!model.envKey && !isQwenBaseUrl(baseUrl)) {
return undefined;
}
const envKey = model.envKey ?? DEFAULT_OPENAI_API_KEY;
const envValue = process.env[envKey];
const envValue = (env ?? process.env)[envKey];
if (envValue && envValue.trim().length > 0) {
return envValue.trim();
}
@ -267,6 +269,7 @@ export function resolveVoiceTranscriptionConfig({
config,
settings,
voiceModel,
env,
}: ResolveVoiceTranscriptionConfigArgs): VoiceTranscriptionConfig {
const matches = config
.getAllConfiguredModels()
@ -305,7 +308,7 @@ export function resolveVoiceTranscriptionConfig({
);
}
const apiKey = readApiKey(settings, model, normalizedBaseUrl);
const apiKey = readApiKey(settings, model, normalizedBaseUrl, env);
if (model.envKey && !apiKey) {
throw new Error(`Voice model '${voiceModel}' requires ${model.envKey}.`);
}

View file

@ -72,6 +72,32 @@ describe('voice-transcriber', () => {
});
});
it('uses the supplied env snapshot for model API keys', () => {
vi.stubEnv('DASHSCOPE_API_KEY', 'process-key');
const config = createConfig([
{
id: 'qwen3-asr-flash',
label: 'Qwen ASR',
authType: AuthType.USE_OPENAI,
baseUrl: 'https://dashscope.example/v1',
envKey: 'DASHSCOPE_API_KEY',
},
]);
expect(
resolveVoiceTranscriptionConfig({
config,
settings: createSettings(),
voiceModel: 'qwen3-asr-flash',
env: { DASHSCOPE_API_KEY: 'runtime-key' },
}),
).toEqual({
model: 'qwen3-asr-flash',
baseUrl: 'https://dashscope.example/v1',
apiKey: 'runtime-key',
});
});
it('routes known voice models by model id instead of user protocol', () => {
expect(resolveVoiceTransport('qwen3-asr-flash')).toBe('qwen-asr-chat');
expect(resolveVoiceTransport('qwen3-asr-flash-2026-02-10')).toBe(

View file

@ -55,6 +55,19 @@ describe('modelConfigUtils', () => {
expect(getAuthTypeFromEnv()).toBe(AuthType.USE_OPENAI);
});
it('uses an injected env instead of process.env when provided', () => {
process.env['OPENAI_API_KEY'] = 'process-key';
process.env['OPENAI_MODEL'] = 'process-model';
process.env['OPENAI_BASE_URL'] = 'https://process.example';
expect(
getAuthTypeFromEnv({
GEMINI_API_KEY: 'runtime-key',
GEMINI_MODEL: 'runtime-model',
}),
).toBe(AuthType.USE_GEMINI);
});
it('should return USE_OPENAI when the model is given via QWEN_MODEL', () => {
// QWEN_MODEL is a valid USE_OPENAI model var (see AUTH_ENV_MODEL_VARS),
// so a config that sets it instead of OPENAI_MODEL must still resolve.

View file

@ -197,31 +197,33 @@ export interface ResolvedCliGenerationConfig {
warnings: string[];
}
export function getAuthTypeFromEnv(): AuthType | undefined {
if (process.env['QWEN_OAUTH']) {
export function getAuthTypeFromEnv(
env: Record<string, string | undefined> = process.env,
): AuthType | undefined {
if (env['QWEN_OAUTH']) {
return AuthType.QWEN_OAUTH;
}
if (
process.env['OPENAI_API_KEY'] &&
(process.env['OPENAI_MODEL'] || process.env['QWEN_MODEL']) &&
process.env['OPENAI_BASE_URL']
env['OPENAI_API_KEY'] &&
(env['OPENAI_MODEL'] || env['QWEN_MODEL']) &&
env['OPENAI_BASE_URL']
) {
return AuthType.USE_OPENAI;
}
if (process.env['GEMINI_API_KEY'] && process.env['GEMINI_MODEL']) {
if (env['GEMINI_API_KEY'] && env['GEMINI_MODEL']) {
return AuthType.USE_GEMINI;
}
if (process.env['GOOGLE_API_KEY'] && process.env['GOOGLE_MODEL']) {
if (env['GOOGLE_API_KEY'] && env['GOOGLE_MODEL']) {
return AuthType.USE_VERTEX_AI;
}
if (
process.env['ANTHROPIC_API_KEY'] &&
process.env['ANTHROPIC_MODEL'] &&
process.env['ANTHROPIC_BASE_URL']
env['ANTHROPIC_API_KEY'] &&
env['ANTHROPIC_MODEL'] &&
env['ANTHROPIC_BASE_URL']
) {
return AuthType.USE_ANTHROPIC;
}