vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-07-09 17:19:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

refactor(cli): split serve server assembly (#5937)

Browse source 
Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye 2026-06-27 20:18:19 +08:00 • committed by GitHub
parent f33dd61f8a
commit 94ea0884bb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 972 additions and 650 deletions
Download patch file Download diff file Expand all files Collapse all files

64
.qwen/design/serve-server-final-split.md Normal file
View file

@ -0,0 +1,64 @@
# serve server.ts final split
## Goal
Continue the staged `packages/cli/src/serve/server.ts` split without changing daemon behavior. This pass moves the remaining inline REST handlers, small middleware helpers, capability construction, device-flow registry setup, and rate-limiter setup into focused internal modules. `createServeApp()` remains the composition point for daemon state, middleware order, route registration, ACP transport mount, Web Shell fallback, and final error handling.
## Middleware And Route Order
The assembly order is part of the daemon contract and must stay visually auditable in `createServeApp()`:
1. same-origin `Origin` stripping
2. CORS and host allowlist
3. pre-auth `/health` and `/demo` on allowed loopback setups
4. access logging
5. Web Shell static assets
6. bearer auth
7. rate limit
8. JSON body parser and JSON parser error mapper
9. post-auth `/health` and `/demo` when required
10. daemon telemetry
11. REST route groups
12. ACP HTTP and WebSocket routes
13. Web Shell fallback
14. final error handler
## Extracted Boundaries
`server/self-origin.ts`, `server/access-log.ts`, `server/rate-limiter-setup.ts`, and `server/error-handlers.ts` own small middleware/setup blocks that previously lived inline in `createServeApp()`. They are intentionally thin and keep the same registration order in `server.ts`.
`server/serve-features.ts` owns the language-code list, voice transcription capability cache, and advertised feature envelope input construction. Its cache invalidation function is still called by workspace settings reload/change paths.
`server/device-flow-registry.ts` owns default Qwen OAuth provider registration, event sink wiring, audit stderr breadcrumbs, and `app.locals` registry installation.
`routes/capabilities.ts` owns `GET /capabilities`.
`routes/workspace-mcp-control.ts` owns MCP restart/manage/runtime add/remove mutations.
`routes/workspace-lifecycle.ts` owns `/workspace/init` and `/workspace/reload`.
`routes/workspace-tools.ts` owns `/workspace/tools/:name/enable`.
Each route module receives only the dependencies it needs. None of the new modules import `server.ts`, which keeps dependency direction one-way and avoids cycles.
## Remaining In `server.ts`
`server.ts` still owns app creation, bound-workspace canonicalization, bridge/filesystem/workspace construction, mutation gate creation, route ordering, ACP HTTP/WebSocket mount, Web Shell static/fallback placement, and the compatibility exports consumed by existing callers.
The file is not required to drop below 200 lines in this PR. The acceptance criterion is that it has no inline REST endpoint handlers and reads as an assembly file whose behavioral ordering can be reviewed in one place.
## Non-goals
This pass does not change response bodies, status codes, headers, SSE frames, ACP behavior, auth gates, rate-limit tiers, device-flow semantics, or error taxonomy. It does not remove `status.ts`, `event-bus.ts`, or `in-memory-channel.ts` compatibility shims. It does not rename historical docs or introduce a Router framework or a single god context for routes.
## Audit Notes
Round 1 checked architecture boundaries and kept the existing `registerXRoutes(app, deps)` pattern instead of adding a Router abstraction.
Round 2 checked dependency direction and moved device-flow/runtime setup behind helpers without letting any route module import `server.ts`.
Round 3 checked failure paths and kept bridge error mapping, JSON body parser errors, strict mutation gates, and client-id validation call sites behavior-preserving.
Round 4 checked compatibility and retained public exports from `server.ts` for `run-qwen-serve.ts`, ACP HTTP callers, and tests.
Round 5 checked testing strategy and uses focused `server.test.ts`, route tests, ACP HTTP tests, typecheck, build, lint, inline endpoint grep, and `git diff --check`.

59
packages/cli/src/serve/routes/capabilities.ts Normal file
View file

@ -0,0 +1,59 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application } from 'express';
import type { AcpSessionBridge } from '../acp-session-bridge.js';
import { getServeProtocolVersions } from '../capabilities.js';
import type { getAdvertisedServeFeatures } from '../capabilities.js';
import { advertisedMaxPendingPromptsPerSession } from '../server/serve-features.js';
import {
CAPABILITIES_SCHEMA_VERSION,
type CapabilitiesEnvelope,
type ServeOptions,
} from '../types.js';
interface RegisterCapabilitiesRoutesDeps {
qwenCodeVersion?: string;
mode: ServeOptions['mode'];
currentServeFeatures: () => ReturnType<typeof getAdvertisedServeFeatures>;
boundWorkspace: string;
permissionPolicy: AcpSessionBridge['permissionPolicy'];
maxPendingPromptsPerSession: ServeOptions['maxPendingPromptsPerSession'];
languageCodes: string[];
}
export function registerCapabilitiesRoutes(
app: Application,
deps: RegisterCapabilitiesRoutesDeps,
): void {
app.get('/capabilities', (_req, res) => {
const envelope: CapabilitiesEnvelope = {
v: CAPABILITIES_SCHEMA_VERSION,
protocolVersions: getServeProtocolVersions(),
...(deps.qwenCodeVersion
? { qwenCodeVersion: deps.qwenCodeVersion }
: {}),
mode: deps.mode,
features: deps.currentServeFeatures(),
modelServices: [],
// Surface the bound workspace so clients can detect mismatch pre-flight
// and omit `cwd` on `POST /session`.
workspaceCwd: deps.boundWorkspace,
// Advertise supported transport families so SDK clients can
// auto-negotiate the best available transport via negotiateTransport().
transports: ['rest'],
// Active mediation policy under the `policy` namespace.
policy: { permission: deps.permissionPolicy },
limits: {
maxPendingPromptsPerSession: advertisedMaxPendingPromptsPerSession(
deps.maxPendingPromptsPerSession,
),
},
supportedLanguages: deps.languageCodes,
};
res.status(200).json(envelope);
});
}

75
packages/cli/src/serve/routes/workspace-lifecycle.ts Normal file
View file

@ -0,0 +1,75 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application, Request, RequestHandler, Response } from 'express';
import type { SendBridgeError } from '../server/error-response.js';
import { createBuildWorkspaceCtx } from '../server/request-helpers.js';
import type { DaemonWorkspaceService } from '../workspace-service/index.js';
interface RegisterWorkspaceLifecycleRoutesDeps {
boundWorkspace: string;
workspace: DaemonWorkspaceService;
mutate: (opts?: { strict?: boolean }) => RequestHandler;
safeBody: (req: Request) => Record<string, unknown>;
sendBridgeError: SendBridgeError;
invalidateServeFeaturesCache: () => void;
parseAndValidateClientId: (
req: Request,
res: Response,
) => string | undefined | null;
}
export function registerWorkspaceLifecycleRoutes(
app: Application,
deps: RegisterWorkspaceLifecycleRoutesDeps,
): void {
const {
boundWorkspace,
workspace,
mutate,
safeBody,
sendBridgeError,
invalidateServeFeaturesCache,
parseAndValidateClientId,
} = deps;
const buildWorkspaceCtx = createBuildWorkspaceCtx(boundWorkspace);
app.post('/workspace/init', mutate({ strict: true }), async (req, res) => {
const body = safeBody(req);
const force = body['force'];
if (force !== undefined && typeof force !== 'boolean') {
res.status(400).json({
error: '`force` must be a boolean when provided',
code: 'invalid_force_flag',
});
return;
}
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
try {
const ctx = buildWorkspaceCtx('POST /workspace/init', clientId);
const result = await workspace.initWorkspace(ctx, {
force: force === true,
});
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, { route: 'POST /workspace/init' });
}
});
app.post('/workspace/reload', mutate({ strict: true }), async (req, res) => {
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
try {
const ctx = buildWorkspaceCtx('POST /workspace/reload', clientId);
const result = await workspace.reload(ctx);
invalidateServeFeaturesCache();
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, { route: 'POST /workspace/reload' });
}
});
}

221
packages/cli/src/serve/routes/workspace-mcp-control.ts Normal file
View file

@ -0,0 +1,221 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application, Request, RequestHandler, Response } from 'express';
import type { AcpSessionBridge } from '../acp-session-bridge.js';
import type { SendBridgeError } from '../server/error-response.js';
import {
createBuildWorkspaceCtx,
MAX_SERVER_NAME_LENGTH,
validateMcpRuntimeServerName,
} from '../server/request-helpers.js';
import type { DaemonWorkspaceService } from '../workspace-service/index.js';
interface RegisterWorkspaceMcpControlRoutesDeps {
boundWorkspace: string;
bridge: AcpSessionBridge;
workspace: DaemonWorkspaceService;
mutate: (opts?: { strict?: boolean }) => RequestHandler;
safeBody: (req: Request) => Record<string, unknown>;
sendBridgeError: SendBridgeError;
parseAndValidateClientId: (
req: Request,
res: Response,
) => string | undefined | null;
}
export function registerWorkspaceMcpControlRoutes(
app: Application,
deps: RegisterWorkspaceMcpControlRoutesDeps,
): void {
const {
boundWorkspace,
bridge,
workspace,
mutate,
safeBody,
sendBridgeError,
parseAndValidateClientId,
} = deps;
const buildWorkspaceCtx = createBuildWorkspaceCtx(boundWorkspace);
app.post(
'/workspace/mcp/:server/restart',
mutate({ strict: true }),
async (req, res) => {
const serverName = req.params['server'];
if (!serverName || typeof serverName !== 'string') {
res.status(400).json({
error: 'Server name path parameter is required',
code: 'invalid_server_name',
});
return;
}
if (serverName.length > MAX_SERVER_NAME_LENGTH) {
res.status(400).json({
error: `Server name exceeds ${MAX_SERVER_NAME_LENGTH}-character limit`,
code: 'invalid_server_name',
});
return;
}
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
let entryIndex: number | undefined;
const rawEntryIndex = req.query['entryIndex'];
if (rawEntryIndex !== undefined && rawEntryIndex !== '*') {
const candidate =
typeof rawEntryIndex === 'string' ? rawEntryIndex : undefined;
const parsed =
candidate !== undefined ? Number.parseInt(candidate, 10) : NaN;
if (
!Number.isInteger(parsed) ||
parsed < 0 ||
String(parsed) !== candidate
) {
res.status(400).json({
error:
'`entryIndex` query parameter must be a non-negative integer or "*"',
code: 'invalid_entry_index',
});
return;
}
entryIndex = parsed;
}
try {
const ctx = buildWorkspaceCtx(
'POST /workspace/mcp/:server/restart',
clientId,
);
const result = await workspace.restartMcpServer(
ctx,
serverName,
entryIndex !== undefined ? { entryIndex } : undefined,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'POST /workspace/mcp/:server/restart',
});
}
},
);
for (const [routeAction, bridgeAction] of [
['enable', 'enable'],
['disable', 'disable'],
['authenticate', 'authenticate'],
['clear-auth', 'clear-auth'],
] as const) {
app.post(
`/workspace/mcp/:server/${routeAction}`,
mutate({ strict: true }),
async (req, res) => {
const serverName = req.params['server'];
if (!serverName || typeof serverName !== 'string') {
res.status(400).json({
error: 'Server name path parameter is required',
code: 'invalid_server_name',
});
return;
}
if (serverName.length > MAX_SERVER_NAME_LENGTH) {
res.status(400).json({
error: `Server name exceeds ${MAX_SERVER_NAME_LENGTH}-character limit`,
code: 'invalid_server_name',
});
return;
}
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
try {
const result = await bridge.manageMcpServer(
serverName,
bridgeAction,
clientId,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: `POST /workspace/mcp/:server/${routeAction}`,
});
}
},
);
}
app.post(
'/workspace/mcp/servers',
mutate({ strict: true }),
async (req, res) => {
const body = safeBody(req);
const name = body['name'];
if (!validateMcpRuntimeServerName(name, res)) return;
const config = body['config'];
if (
typeof config !== 'object' ||
config === null ||
Array.isArray(config)
) {
res.status(400).json({
error: '`config` must be a non-null object',
code: 'missing_required_field',
field: 'config',
});
return;
}
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
if (!clientId) {
res.status(400).json({
error:
'`X-Qwen-Client-Id` header is required for runtime MCP mutation',
code: 'missing_client_id',
});
return;
}
try {
const result = await bridge.addRuntimeMcpServer(
name,
config as Record<string, unknown>,
clientId,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'POST /workspace/mcp/servers',
});
}
},
);
app.delete(
'/workspace/mcp/servers/:name',
mutate({ strict: true }),
async (req, res) => {
const name = req.params['name'] ?? '';
if (!validateMcpRuntimeServerName(name, res)) return;
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
if (!clientId) {
res.status(400).json({
error:
'`X-Qwen-Client-Id` header is required for runtime MCP mutation',
code: 'missing_client_id',
});
return;
}
try {
const result = await bridge.removeRuntimeMcpServer(name, clientId);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'DELETE /workspace/mcp/servers/:name',
});
}
},
);
}

97
packages/cli/src/serve/routes/workspace-tools.ts Normal file
View file

@ -0,0 +1,97 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application, Request, RequestHandler, Response } from 'express';
import type { SendBridgeError } from '../server/error-response.js';
import {
createBuildWorkspaceCtx,
MAX_TOOL_NAME_LENGTH,
} from '../server/request-helpers.js';
import type { DaemonWorkspaceService } from '../workspace-service/index.js';
interface RegisterWorkspaceToolsRoutesDeps {
boundWorkspace: string;
workspace: DaemonWorkspaceService;
mutate: (opts?: { strict?: boolean }) => RequestHandler;
safeBody: (req: Request) => Record<string, unknown>;
sendBridgeError: SendBridgeError;
parseAndValidateClientId: (
req: Request,
res: Response,
) => string | undefined | null;
}
export function registerWorkspaceToolsRoutes(
app: Application,
deps: RegisterWorkspaceToolsRoutesDeps,
): void {
const {
boundWorkspace,
workspace,
mutate,
safeBody,
sendBridgeError,
parseAndValidateClientId,
} = deps;
const buildWorkspaceCtx = createBuildWorkspaceCtx(boundWorkspace);
app.post(
'/workspace/tools/:name/enable',
mutate({ strict: true }),
async (req, res) => {
const rawToolName = req.params['name'];
if (!rawToolName || typeof rawToolName !== 'string') {
res.status(400).json({
error: 'Tool name path parameter is required',
code: 'invalid_tool_name',
});
return;
}
const toolName = rawToolName.trim();
if (toolName.length === 0) {
res.status(400).json({
error: 'Tool name path parameter is required',
code: 'invalid_tool_name',
});
return;
}
if (toolName.length > MAX_TOOL_NAME_LENGTH) {
res.status(400).json({
error: `Tool name exceeds ${MAX_TOOL_NAME_LENGTH}-character limit`,
code: 'invalid_tool_name',
});
return;
}
const body = safeBody(req);
const enabled = body['enabled'];
if (typeof enabled !== 'boolean') {
res.status(400).json({
error: '`enabled` is required and must be a boolean',
code: 'invalid_enabled_flag',
});
return;
}
const clientId = parseAndValidateClientId(req, res);
if (clientId === null) return;
try {
const ctx = buildWorkspaceCtx(
'POST /workspace/tools/:name/enable',
clientId,
);
const result = await workspace.setWorkspaceToolEnabled(
ctx,
toolName,
enabled,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'POST /workspace/tools/:name/enable',
});
}
},
);
}

728
packages/cli/src/serve/server.ts
View file

@ -5,8 +5,7 @@
*/
import express from 'express';
import type { Application, Request, Response } from 'express';
import { writeStderrLine } from '../utils/stdioHelpers.js';
import type { Application } from 'express';
import type { DaemonLogger } from './daemon-logger.js';
import type { DaemonStartupSnapshot } from './daemon-status.js';
import {
@ -17,20 +16,14 @@ import {
hostAllowlist,
parseAllowOriginPatterns,
} from './auth.js';
import {
import type {
DeviceFlowProvider,
DeviceFlowRegistry,
setDeviceFlowRegistry,
type DeviceFlowEventSink,
type DeviceFlowProvider,
type DeviceFlowProviderId,
} from './auth/device-flow.js';
import type { DaemonStatusProvider } from '@qwen-code/acp-bridge';
import { QwenOAuthDeviceFlowProvider } from './auth/qwen-device-flow-provider.js';
import { createBridgeFileSystemAdapter } from './bridge-file-system-adapter.js';
import { createDaemonStatusProvider } from './daemon-status-provider.js';
import { createWorkspaceProvidersStatusProvider } from './workspace-providers-status.js';
import { SUPPORTED_LANGUAGES } from '../i18n/index.js';
import { loadSettings } from '../config/settings.js';
import { mountAcpHttp, type AcpHttpHandle } from './acp-http/index.js';
import { createVoiceWsConnectionHandler } from './voice/voice-ws.js';
import {
@ -39,12 +32,6 @@ import {
type AcpSessionBridge,
} from './acp-session-bridge.js';
import {
getAdvertisedServeFeatures,
getServeProtocolVersions,
} from './capabilities.js';
import {
CAPABILITIES_SCHEMA_VERSION,
type CapabilitiesEnvelope,
type ServeAuthProviderInstallRequest,
type ServeAuthProviderInstallResult,
type ServeOptions,
@ -74,6 +61,7 @@ import {
createDaemonWorkspaceService,
type DaemonWorkspaceService,
} from './workspace-service/index.js';
import { registerCapabilitiesRoutes } from './routes/capabilities.js';
import { registerWorkspacePermissionsRoutes } from './routes/workspace-permissions.js';
import { registerWorkspaceSettingsRoutes } from './routes/workspace-settings.js';
import {
@ -84,13 +72,8 @@ import {
registerWorkspaceVoiceRoutes,
type WorkspaceVoiceRouteDeps,
} from './routes/workspace-voice.js';
import { hasConfiguredBatchVoiceTranscriptionModel } from '../services/voice-service.js';
import { registerA2uiActionRoutes } from './routes/a2ui-action.js';
import {
createRateLimiter,
setRateLimiter,
type RateLimiterInstance,
} from './rate-limit.js';
import { setRateLimiter } from './rate-limit.js';
import {
sendBridgeError as sendBridgeErrorResponse,
sendPermissionVoteError as sendPermissionVoteErrorResponse,
@ -99,15 +82,23 @@ import {
import { resolveBridgeFsFactory } from './server/fs-factory.js';
import {
createBuildWorkspaceCtx,
MAX_SERVER_NAME_LENGTH,
MAX_TOOL_NAME_LENGTH,
parseAndValidateWorkspaceClientId,
parseClientIdHeader,
safeBody,
sendJsonBodyParserError,
validateMcpRuntimeServerName,
} from './server/request-helpers.js';
import { daemonTelemetryMiddleware } from './server/telemetry.js';
import { installAccessLogMiddleware } from './server/access-log.js';
import { setupDeviceFlowRegistry } from './server/device-flow-registry.js';
import {
installFinalErrorHandler,
installJsonBodyParser,
} from './server/error-handlers.js';
import { installRateLimiter } from './server/rate-limiter-setup.js';
import { createServeFeatures } from './server/serve-features.js';
import { installSelfOriginStripMiddleware } from './server/self-origin.js';
import { registerWorkspaceLifecycleRoutes } from './routes/workspace-lifecycle.js';
import { registerWorkspaceMcpControlRoutes } from './routes/workspace-mcp-control.js';
import { registerWorkspaceToolsRoutes } from './routes/workspace-tools.js';
export {
createDefaultFsAuditEmit,
@ -128,23 +119,6 @@ export type {
} from './server/session-list.js';
export { getActiveSseCount } from './routes/sse-events.js';
function isWorkspaceVoiceTranscriptionAvailable(
boundWorkspace: string,
): boolean {
try {
return hasConfiguredBatchVoiceTranscriptionModel(
loadSettings(boundWorkspace),
);
} catch (err) {
writeStderrLine(
`qwen serve: workspace voice transcription capability check failed: ${
err instanceof Error ? err.message : String(err)
}`,
);
return false;
}
}
/**
* Module-scoped once-per-process guard for the `createServeApp`
* default-trust stderr warning. Without this, tests calling
@ -247,17 +221,6 @@ export interface ServeAppDeps {
voiceTranscriber?: WorkspaceVoiceRouteDeps['transcribe'];
}
// Keep in sync with acp-bridge bridge.ts and SDK DaemonClient.ts.
const DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION = 5;
function advertisedMaxPendingPromptsPerSession(
value: number | undefined,
): number | null {
if (value === undefined) return DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION;
if (value === 0 || value === Number.POSITIVE_INFINITY) return null;
return value;
}
/**
* Build the Express app for `qwen serve`. Pure function — no side effects on
* the network or process; `runQwenServe` does the listen/signal handling.
@ -267,31 +230,9 @@ function advertisedMaxPendingPromptsPerSession(
* resolves. Defaults to `opts.port` for callers (e.g. tests) that pin a port
* up front.
*
* Supported routes:
* - `GET /health`
* - `GET /daemon/status`
* - `GET /capabilities`
* - `GET /workspace/mcp`
* - `GET /workspace/skills`
* - `GET /workspace/providers`
* - `GET /workspace/env`
* - `GET /workspace/preflight`
* - `POST /session`
* - `POST /session/:id/load`
* - `POST /session/:id/resume`
* - `GET /workspace/:id/sessions`
* - `GET /session/:id/status`
* - `GET /session/:id/context`
* - `GET /session/:id/supported-commands`
* - `GET /session/:id/tasks`
* - `GET /session/:id/lsp`
* - `POST /session/:id/prompt`
* - `POST /session/:id/cancel`
* - `POST /session/:id/heartbeat`
* - `POST /session/:id/model`
* - `GET /session/:id/events` (SSE)
* - `POST /session/:id/permission/:requestId`
* - `POST /permission/:requestId`
* Route modules are registered below in middleware order. Keep this file as
* the assembly point so auth/rate-limit/body-parser/REST/ACP/Web Shell order
* stays reviewable in one place.
*
* **Workspace validation contract.** `createServeApp` itself does NOT
* verify that `opts.workspace` exists or is a directory — it
@ -347,19 +288,18 @@ export function createServeApp(
injected: deps.fsFactory,
trusted: false,
});
let cachedVoiceTranscriptionAvailable: boolean | undefined;
const invalidateServeFeaturesCache = () => {
cachedVoiceTranscriptionAvailable = undefined;
};
const getCachedVoiceTranscriptionAvailable = () => {
cachedVoiceTranscriptionAvailable ??=
isWorkspaceVoiceTranscriptionAvailable(boundWorkspace);
return cachedVoiceTranscriptionAvailable;
};
const tokenConfigured =
typeof opts.token === 'string' && opts.token.length > 0;
const sessionShellCommandEnabled =
opts.enableSessionShell === true && tokenConfigured;
const { languageCodes, currentServeFeatures, invalidateServeFeaturesCache } =
createServeFeatures({
opts,
boundWorkspace,
persistSettingAvailable: deps.persistSetting !== undefined,
reloadAvailable: deps.workspace !== undefined,
sessionShellCommandEnabled,
});
const statusProvider = deps.statusProvider ?? createDaemonStatusProvider();
const bridge =
deps.bridge ??
@ -378,34 +318,7 @@ export function createServeApp(
fileSystem: createBridgeFileSystemAdapter(fsFactory),
});
// Allow same-origin requests from the demo page. Browsers send an
// `Origin` header on same-origin POST/fetch calls; `denyBrowserOriginCors`
// below would reject them. This middleware strips `Origin` when it
// matches the daemon's own address so the demo page's API calls pass
// through. Only loopback origins are matched — non-loopback deployments
// require the operator to front the daemon with a reverse proxy for
// browser access anyway (per the threat-model docs).
let cachedStripPort = -1;
let cachedSelfOrigins: Set<string> = new Set();
app.use((req: import('express').Request, _res, next) => {
const origin = req.headers.origin;
if (origin) {
const port = getPort();
if (port !== cachedStripPort) {
cachedStripPort = port;
cachedSelfOrigins = new Set([
`http://127.0.0.1:${port}`,
`http://localhost:${port}`,
`http://[::1]:${port}`,
`http://host.docker.internal:${port}`,
]);
}
if (cachedSelfOrigins.has(origin)) {
delete req.headers.origin;
}
}
next();
});
installSelfOriginStripMiddleware(app, getPort);
// Park the factory on `app.locals` so route handlers can pick it up
// via `req.app.locals.fsFactory` without re-threading the value
@ -416,69 +329,13 @@ export function createServeApp(
// compute workspace-relative response paths without re-resolving.
(app.locals as { boundWorkspace?: string }).boundWorkspace = boundWorkspace;
// Wire the device-flow registry. Default builds a single Qwen
// provider; tests inject `deps.deviceFlowRegistry` or
// `deps.deviceFlowProviders` to stub the OAuth client only.
const deviceFlowProviderMap = new Map<
DeviceFlowProviderId,
DeviceFlowProvider
>();
for (const provider of deps.deviceFlowProviders ?? []) {
deviceFlowProviderMap.set(provider.providerId, provider);
}
if (!deviceFlowProviderMap.has('qwen-oauth')) {
deviceFlowProviderMap.set('qwen-oauth', new QwenOAuthDeviceFlowProvider());
}
const deviceFlowEventSink: DeviceFlowEventSink = {
publish(emission, originatorClientId) {
bridge.publishWorkspaceEvent({
type: `auth_device_flow_${emission.type}`,
data: emission.data,
...(originatorClientId ? { originatorClientId } : {}),
});
},
};
const deviceFlowRegistry =
deps.deviceFlowRegistry ??
new DeviceFlowRegistry({
events: deviceFlowEventSink,
audit: {
record(line) {
// Structured stderr breadcrumb; deviceFlowId truncated to first
// 8 chars so log
// skimmers can follow a flow without retaining full uuids.
const id = line.deviceFlowId.slice(0, 8);
const parts = [
`[serve] auth.device-flow:`,
`provider=${line.providerId}`,
`deviceFlowId=${id}...`,
line.clientId ? `clientId=${line.clientId}` : 'clientId=-',
`status=${line.status}`,
];
if (line.errorKind) parts.push(`errorKind=${line.errorKind}`);
if (line.expiresInMs !== undefined) {
parts.push(`expiresInMs=${Math.max(0, line.expiresInMs)}`);
}
// Include `line.hint` for operator-only breadcrumbs that
// aren't surfaced over SSE. Bound at 1 KiB.
if (line.hint) {
const STDERR_HINT_MAX = 1_024;
const hint =
line.hint.length > STDERR_HINT_MAX
? `${line.hint.slice(0, STDERR_HINT_MAX)}…[+${line.hint.length - STDERR_HINT_MAX} bytes truncated]`
: line.hint;
// Quote the hint so multi-word values stay parseable.
parts.push(`hint=${JSON.stringify(hint)}`);
}
writeStderrLine(parts.join(' '));
},
},
resolveProvider: (providerId) => deviceFlowProviderMap.get(providerId),
const { deviceFlowRegistry, getSupportedDeviceFlowProviders } =
setupDeviceFlowRegistry({
app,
bridge,
registry: deps.deviceFlowRegistry,
providers: deps.deviceFlowProviders,
});
// Park the registry on `app.locals` so request handlers can reach it.
// Typed accessor prevents a string-key typo from silently detaching
// `runQwenServe`'s shutdown dispose call.
setDeviceFlowRegistry(app, deviceFlowRegistry);
const { daemonLog } = deps;
@ -526,8 +383,6 @@ export function createServeApp(
bridge.publishWorkspaceEvent(event);
},
});
let rateLimiter: RateLimiterInstance | undefined;
// Order matters: rejection guards (CORS / Host allowlist / bearer auth)
// run BEFORE the JSON body parser. Otherwise an unauthenticated POST
// gets a full 10MB `JSON.parse` before the 401 fires — a trivially
@ -568,58 +423,7 @@ export function createServeApp(
healthDemoRoutes.register(app);
}
// Access-log middleware. Registered BEFORE bearerAuth and JSON parser
// so auth rejections (401) and malformed-body errors (400) are also
// captured in the daemon log. Excluded:
// - GET /health (high-frequency probe, would drown signal)
// - Successful SSE streams (GET .../events with 200) — logged inline
// at open/close; failed SSE handshakes (4xx) are still recorded.
if (daemonLog) {
const SESSION_ID_RE = /\/session\/([^/]+)/;
app.use((req, res, next) => {
const { method, path: reqPath } = req;
if (
(method === 'GET' && reqPath === '/health') ||
(method === 'POST' && reqPath.endsWith('/heartbeat'))
) {
return next();
}
const startMs = Date.now();
res.on('finish', () => {
try {
const status = res.statusCode;
if (
method === 'GET' &&
reqPath.endsWith('/events') &&
status === 200
) {
return;
}
const durationMs = Date.now() - startMs;
const sessionMatch = reqPath.match(SESSION_ID_RE);
const sessionId = sessionMatch?.[1];
const clientId = req.headers['x-qwen-client-id'] as
| string
| undefined;
const ctx = {
route: `${method} ${reqPath}`,
...(sessionId ? { sessionId } : {}),
...(clientId ? { clientId } : {}),
status,
durationMs,
};
if (status >= 400) {
daemonLog.warn('request completed', ctx);
} else {
daemonLog.info('request completed', ctx);
}
} catch {
// Logging failure must not affect the request.
}
});
next();
});
}
installAccessLogMiddleware(app, daemonLog);
// Serve the Web Shell static assets (/ and /assets) BEFORE bearerAuth. The
// static shell carries no secrets and a browser cannot attach an
@ -641,47 +445,8 @@ export function createServeApp(
// Rate limiter: after auth (only count authenticated requests),
// before body parser (reject early without burning JSON.parse CPU).
if (opts.rateLimit) {
const windowMs = opts.rateLimitWindowMs ?? 60_000;
rateLimiter = createRateLimiter({
tiers: {
prompt: { windowMs, max: opts.rateLimitPrompt ?? 10 },
mutation: { windowMs, max: opts.rateLimitMutation ?? 30 },
read: { windowMs, max: opts.rateLimitRead ?? 120 },
},
hostname: opts.hostname,
onLimitReached: daemonLog
? (tier, key, suppressed) => {
daemonLog.warn(
`rate limit hit${suppressed > 0 ? ` (${suppressed} suppressed)` : ''}`,
{ tier, key: key.slice(0, 64) },
);
}
: undefined,
onError: daemonLog
? (err, path) => {
daemonLog.warn(
`rate limiter error (fail-open): ${err instanceof Error ? err.message : String(err)}`,
{ path },
);
}
: undefined,
});
app.use(rateLimiter.middleware);
}
app.use(express.json({ limit: '10mb' }));
app.use(
(
err: unknown,
_req: import('express').Request,
res: import('express').Response,
next: import('express').NextFunction,
) => {
if (sendJsonBodyParserError(res, err)) return;
next(err);
},
);
const rateLimiter = installRateLimiter(app, opts, daemonLog);
installJsonBodyParser(app);
if (!healthDemoRoutes.exposeHealthPreAuth) {
// Non-loopback OR loopback with `--require-auth`: register
@ -703,30 +468,6 @@ export function createServeApp(
const buildWorkspaceCtx = createBuildWorkspaceCtx(boundWorkspace);
const LANGUAGE_CODES = [...SUPPORTED_LANGUAGES.map((l) => l.code), 'auto'];
const currentServeFeatures = () =>
getAdvertisedServeFeatures(undefined, {
requireAuth: opts.requireAuth === true,
mcpPoolActive: opts.mcpPoolActive !== false,
allowOriginActive:
opts.allowOrigins !== undefined && opts.allowOrigins.length > 0,
...(opts.promptDeadlineMs !== undefined
? { promptDeadlineMs: opts.promptDeadlineMs }
: {}),
...(opts.writerIdleTimeoutMs !== undefined
? { writerIdleTimeoutMs: opts.writerIdleTimeoutMs }
: {}),
persistSettingAvailable: deps.persistSetting !== undefined,
sessionShellCommandEnabled,
rateLimit: opts.rateLimit === true,
reloadAvailable: deps.workspace !== undefined,
voiceTranscriptionAvailable: getCachedVoiceTranscriptionAvailable(),
// Advertised whenever the `/voice/stream` WS endpoint exists (ACP HTTP
// on). A configured token no longer suppresses it — the browser carries
// the bearer token via the WS subprotocol, which the upgrade listener
// verifies (acp-http/index.ts).
voiceWsAvailable: process.env['QWEN_SERVE_ACP_HTTP'] !== '0',
});
const acpHandleRef: { current?: AcpHttpHandle } = {};
registerDaemonStatusRoutes(app, {
@ -741,41 +482,19 @@ export function createServeApp(
getRateLimiter: () => rateLimiter,
getRestSseActive: getActiveSseCount,
currentServeFeatures,
getSupportedDeviceFlowProviders: () =>
Array.from(deviceFlowProviderMap.keys()),
getSupportedDeviceFlowProviders,
deviceFlowRegistry,
sessionShellCommandEnabled,
});
app.get('/capabilities', (_req, res) => {
const envelope: CapabilitiesEnvelope = {
v: CAPABILITIES_SCHEMA_VERSION,
protocolVersions: getServeProtocolVersions(),
...(deps.qwenCodeVersion
? { qwenCodeVersion: deps.qwenCodeVersion }
: {}),
mode: opts.mode,
features: currentServeFeatures(),
modelServices: [],
// Surface the bound workspace so clients can detect mismatch
// pre-flight and omit `cwd` on `POST /session`.
workspaceCwd: boundWorkspace,
// Advertise supported transport families so SDK clients can
// auto-negotiate the best available transport via
// `negotiateTransport()`. REST is always available; future PRs
// will add 'acp-http' / 'acp-ws' entries when the corresponding
// routes are wired.
transports: ['rest'],
// Active mediation policy under the `policy` namespace.
policy: { permission: bridge.permissionPolicy },
limits: {
maxPendingPromptsPerSession: advertisedMaxPendingPromptsPerSession(
opts.maxPendingPromptsPerSession,
),
},
supportedLanguages: LANGUAGE_CODES,
};
res.status(200).json(envelope);
registerCapabilitiesRoutes(app, {
qwenCodeVersion: deps.qwenCodeVersion,
mode: opts.mode,
currentServeFeatures,
boundWorkspace,
permissionPolicy: bridge.permissionPolicy,
maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession,
languageCodes,
});
registerWorkspaceStatusRoutes(app, {
@ -914,8 +633,7 @@ export function createServeApp(
registerWorkspaceAuthRoutes(app, {
mutate,
deviceFlowRegistry,
getSupportedDeviceFlowProviders: () =>
Array.from(deviceFlowProviderMap.keys()),
getSupportedDeviceFlowProviders,
sendBridgeError,
boundWorkspace,
allowPrivateAuthBaseUrl: opts.allowPrivateAuthBaseUrl === true,
@ -930,306 +648,38 @@ export function createServeApp(
daemonLog,
promptDeadlineMs: opts.promptDeadlineMs,
sessionShellCommandEnabled,
languageCodes: LANGUAGE_CODES,
languageCodes,
});
app.post(
'/workspace/mcp/:server/restart',
mutate({ strict: true }),
async (req, res) => {
// Single-server MCP restart with budget pre-check. Soft refusals
// are 200 OK with `{restarted:false, skipped:true, reason}`.
const serverName = req.params['server'];
if (!serverName || typeof serverName !== 'string') {
res.status(400).json({
error: 'Server name path parameter is required',
code: 'invalid_server_name',
});
return;
}
// Cap server name length to prevent unbounded path-parameter input.
if (serverName.length > MAX_SERVER_NAME_LENGTH) {
res.status(400).json({
error: `Server name exceeds ${MAX_SERVER_NAME_LENGTH}-character limit`,
code: 'invalid_server_name',
});
return;
}
// Validate `X-Qwen-Client-Id` against known client ids.
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
// Parse `?entryIndex=` for pool-mode targeted restarts. Accepts
// a non-negative integer or `*` / omitted (restart all).
let entryIndex: number | undefined;
const rawEntryIndex = req.query['entryIndex'];
if (rawEntryIndex !== undefined && rawEntryIndex !== '*') {
const candidate =
typeof rawEntryIndex === 'string' ? rawEntryIndex : undefined;
const parsed =
candidate !== undefined ? Number.parseInt(candidate, 10) : NaN;
if (
!Number.isInteger(parsed) ||
parsed < 0 ||
String(parsed) !== candidate
) {
res.status(400).json({
error:
'`entryIndex` query parameter must be a non-negative integer or "*"',
code: 'invalid_entry_index',
});
return;
}
entryIndex = parsed;
}
try {
const ctx = buildWorkspaceCtx(
'POST /workspace/mcp/:server/restart',
clientId,
);
const result = await workspace.restartMcpServer(
ctx,
serverName,
entryIndex !== undefined ? { entryIndex } : undefined,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'POST /workspace/mcp/:server/restart',
});
}
},
);
for (const [routeAction, bridgeAction] of [
['enable', 'enable'],
['disable', 'disable'],
['authenticate', 'authenticate'],
['clear-auth', 'clear-auth'],
] as const) {
app.post(
`/workspace/mcp/:server/${routeAction}`,
mutate({ strict: true }),
async (req, res) => {
const serverName = req.params['server'];
if (!serverName || typeof serverName !== 'string') {
res.status(400).json({
error: 'Server name path parameter is required',
code: 'invalid_server_name',
});
return;
}
if (serverName.length > MAX_SERVER_NAME_LENGTH) {
res.status(400).json({
error: `Server name exceeds ${MAX_SERVER_NAME_LENGTH}-character limit`,
code: 'invalid_server_name',
});
return;
}
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
try {
const result = await bridge.manageMcpServer(
serverName,
bridgeAction,
clientId,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: `POST /workspace/mcp/:server/${routeAction}`,
});
}
},
);
}
// Add a runtime MCP server.
app.post(
'/workspace/mcp/servers',
mutate({ strict: true }),
async (req, res) => {
const body = safeBody(req);
const name = body['name'];
if (!validateMcpRuntimeServerName(name, res)) return;
// Validate config: must be a non-null object
const config = body['config'];
if (
typeof config !== 'object' ||
config === null ||
Array.isArray(config)
) {
res.status(400).json({
error: '`config` must be a non-null object',
code: 'missing_required_field',
field: 'config',
});
return;
}
// Validate client identity (required for runtime MCP mutation)
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
if (!clientId) {
res.status(400).json({
error:
'`X-Qwen-Client-Id` header is required for runtime MCP mutation',
code: 'missing_client_id',
});
return;
}
try {
const result = await bridge.addRuntimeMcpServer(
name,
config as Record<string, unknown>,
clientId,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'POST /workspace/mcp/servers',
});
}
},
);
// Remove a runtime MCP server. Idempotent.
app.delete(
'/workspace/mcp/servers/:name',
mutate({ strict: true }),
async (req, res) => {
const name = req.params['name'] ?? '';
if (!validateMcpRuntimeServerName(name, res)) return;
// Validate client identity (required for runtime MCP mutation)
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
if (!clientId) {
res.status(400).json({
error:
'`X-Qwen-Client-Id` header is required for runtime MCP mutation',
code: 'missing_client_id',
});
return;
}
try {
const result = await bridge.removeRuntimeMcpServer(name, clientId);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'DELETE /workspace/mcp/servers/:name',
});
}
},
);
app.post('/workspace/init', mutate({ strict: true }), async (req, res) => {
// #4175 Wave 4 PR 17. Scaffold-only init: the workspace service
// writes an empty QWEN.md without invoking the LLM. Default refuses
// overwrite (409); body `{force: true}` overrides.
const body = safeBody(req);
const force = body['force'];
if (force !== undefined && typeof force !== 'boolean') {
res.status(400).json({
error: '`force` must be a boolean when provided',
code: 'invalid_force_flag',
});
return;
}
// Validate against known client ids.
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
try {
const ctx = buildWorkspaceCtx('POST /workspace/init', clientId);
const result = await workspace.initWorkspace(ctx, {
force: force === true,
});
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, { route: 'POST /workspace/init' });
}
registerWorkspaceMcpControlRoutes(app, {
boundWorkspace,
bridge,
workspace,
mutate,
safeBody,
sendBridgeError,
parseAndValidateClientId: (req, res) =>
parseAndValidateWorkspaceClientId(req, res, bridge),
});
registerWorkspaceLifecycleRoutes(app, {
boundWorkspace,
workspace,
mutate,
safeBody,
sendBridgeError,
invalidateServeFeaturesCache,
parseAndValidateClientId: (req, res) =>
parseAndValidateWorkspaceClientId(req, res, bridge),
});
registerWorkspaceToolsRoutes(app, {
boundWorkspace,
workspace,
mutate,
safeBody,
sendBridgeError,
parseAndValidateClientId: (req, res) =>
parseAndValidateWorkspaceClientId(req, res, bridge),
});
app.post(
'/workspace/reload',
mutate({ strict: true }),
async (req: Request, res: Response) => {
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
try {
const ctx = buildWorkspaceCtx('POST /workspace/reload', clientId);
const result = await workspace.reload(ctx);
invalidateServeFeaturesCache();
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, { route: 'POST /workspace/reload' });
}
},
);
app.post(
'/workspace/tools/:name/enable',
mutate({ strict: true }),
async (req, res) => {
// Toggles a tool name in the workspace `tools.disabled` settings
// list. Strict-gated alongside other
// mutation routes; bridge writes the file directly (no
// ACP roundtrip) and fan-outs `tool_toggled` to every live
// session SSE bus. Already-registered tools in live sessions
// are NOT retroactively unregistered — toggling takes effect on
// the next ACP child spawn or session refresh.
const rawToolName = req.params['name'];
if (!rawToolName || typeof rawToolName !== 'string') {
res.status(400).json({
error: 'Tool name path parameter is required',
code: 'invalid_tool_name',
});
return;
}
// Trim before persistence so the write path matches the read path.
const toolName = rawToolName.trim();
if (toolName.length === 0) {
res.status(400).json({
error: 'Tool name path parameter is required',
code: 'invalid_tool_name',
});
return;
}
// Cap tool name length to prevent settings file bloat.
if (toolName.length > MAX_TOOL_NAME_LENGTH) {
res.status(400).json({
error: `Tool name exceeds ${MAX_TOOL_NAME_LENGTH}-character limit`,
code: 'invalid_tool_name',
});
return;
}
const body = safeBody(req);
const enabled = body['enabled'];
if (typeof enabled !== 'boolean') {
res.status(400).json({
error: '`enabled` is required and must be a boolean',
code: 'invalid_enabled_flag',
});
return;
}
// Validate against known client ids.
const clientId = parseAndValidateWorkspaceClientId(req, res, bridge);
if (clientId === null) return;
try {
const ctx = buildWorkspaceCtx(
'POST /workspace/tools/:name/enable',
clientId,
);
const result = await workspace.setWorkspaceToolEnabled(
ctx,
toolName,
enabled,
);
res.status(200).json(result);
} catch (err) {
sendBridgeError(res, err, {
route: 'POST /workspace/tools/:name/enable',
});
}
},
);
registerPermissionRoutes(app, {
bridge,
@ -1282,29 +732,7 @@ export function createServeApp(
mountWebShellSpaFallback(app, webShellDir);
}
// Final error handler. `express.json()` throws `SyntaxError` (with
// `status: 400`) on malformed body — without this 4-arg middleware
// Express renders an HTML error page, which trips SDK clients that
// expect a JSON body on every response. Anything else bubbling out
// is a programmer error; log it and return a JSON 500 (matches the
// route-level `sendBridgeError` shape so clients have one error
// contract to parse).
app.use(
(
err: unknown,
_req: import('express').Request,
res: import('express').Response,
_next: import('express').NextFunction,
) => {
if (sendJsonBodyParserError(res, err)) return;
writeStderrLine(
`qwen serve: unhandled error: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`,
);
if (!res.headersSent) {
res.status(500).json({ error: 'Internal server error' });
}
},
);
installFinalErrorHandler(app);
if (rateLimiter) {
setRateLimiter(app, rateLimiter);

55
packages/cli/src/serve/server/access-log.ts Normal file
View file

@ -0,0 +1,55 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application } from 'express';
import type { DaemonLogger } from '../daemon-logger.js';
const SESSION_ID_RE = /\/session\/([^/]+)/;
export function installAccessLogMiddleware(
app: Application,
daemonLog: DaemonLogger | undefined,
): void {
if (!daemonLog) return;
app.use((req, res, next) => {
const { method, path: reqPath } = req;
if (
(method === 'GET' && reqPath === '/health') ||
(method === 'POST' && reqPath.endsWith('/heartbeat'))
) {
return next();
}
const startMs = Date.now();
res.on('finish', () => {
try {
const status = res.statusCode;
if (method === 'GET' && reqPath.endsWith('/events') && status === 200) {
return;
}
const durationMs = Date.now() - startMs;
const sessionMatch = reqPath.match(SESSION_ID_RE);
const sessionId = sessionMatch?.[1];
const clientId = req.headers['x-qwen-client-id'] as string | undefined;
const ctx = {
route: `${method} ${reqPath}`,
...(sessionId ? { sessionId } : {}),
...(clientId ? { clientId } : {}),
status,
durationMs,
};
if (status >= 400) {
daemonLog.warn('request completed', ctx);
} else {
daemonLog.info('request completed', ctx);
}
} catch {
// Logging failure must not affect the request.
}
});
next();
});
}

99
packages/cli/src/serve/server/device-flow-registry.ts Normal file
View file

@ -0,0 +1,99 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application } from 'express';
import { writeStderrLine } from '../../utils/stdioHelpers.js';
import type { AcpSessionBridge } from '../acp-session-bridge.js';
import {
DeviceFlowRegistry,
setDeviceFlowRegistry,
type DeviceFlowEventSink,
type DeviceFlowProvider,
type DeviceFlowProviderId,
} from '../auth/device-flow.js';
import { QwenOAuthDeviceFlowProvider } from '../auth/qwen-device-flow-provider.js';
interface SetupDeviceFlowRegistryDeps {
app: Application;
bridge: AcpSessionBridge;
registry?: DeviceFlowRegistry;
providers?: DeviceFlowProvider[];
}
export interface ServeDeviceFlowRuntime {
deviceFlowRegistry: DeviceFlowRegistry;
getSupportedDeviceFlowProviders: () => DeviceFlowProviderId[];
}
export function setupDeviceFlowRegistry(
deps: SetupDeviceFlowRegistryDeps,
): ServeDeviceFlowRuntime {
const deviceFlowProviderMap = new Map<
DeviceFlowProviderId,
DeviceFlowProvider
>();
for (const provider of deps.providers ?? []) {
deviceFlowProviderMap.set(provider.providerId, provider);
}
if (!deviceFlowProviderMap.has('qwen-oauth')) {
deviceFlowProviderMap.set('qwen-oauth', new QwenOAuthDeviceFlowProvider());
}
const deviceFlowEventSink: DeviceFlowEventSink = {
publish(emission, originatorClientId) {
deps.bridge.publishWorkspaceEvent({
type: `auth_device_flow_${emission.type}`,
data: emission.data,
...(originatorClientId ? { originatorClientId } : {}),
});
},
};
const deviceFlowRegistry =
deps.registry ??
new DeviceFlowRegistry({
events: deviceFlowEventSink,
audit: {
record(line) {
// Structured stderr breadcrumb; deviceFlowId truncated to first
// 8 chars so log skimmers can follow a flow without retaining
// full uuids.
const id = line.deviceFlowId.slice(0, 8);
const parts = [
`[serve] auth.device-flow:`,
`provider=${line.providerId}`,
`deviceFlowId=${id}...`,
line.clientId ? `clientId=${line.clientId}` : 'clientId=-',
`status=${line.status}`,
];
if (line.errorKind) parts.push(`errorKind=${line.errorKind}`);
if (line.expiresInMs !== undefined) {
parts.push(`expiresInMs=${Math.max(0, line.expiresInMs)}`);
}
// Include `line.hint` for operator-only breadcrumbs that aren't
// surfaced over SSE. Bound at 1 KiB.
if (line.hint) {
const STDERR_HINT_MAX = 1_024;
const hint =
line.hint.length > STDERR_HINT_MAX
? `${line.hint.slice(0, STDERR_HINT_MAX)}…[+${line.hint.length - STDERR_HINT_MAX} bytes truncated]`
: line.hint;
// Quote the hint so multi-word values stay parseable.
parts.push(`hint=${JSON.stringify(hint)}`);
}
writeStderrLine(parts.join(' '));
},
},
resolveProvider: (providerId) => deviceFlowProviderMap.get(providerId),
});
setDeviceFlowRegistry(deps.app, deviceFlowRegistry);
return {
deviceFlowRegistry,
getSupportedDeviceFlowProviders: () =>
Array.from(deviceFlowProviderMap.keys()),
};
}

30
packages/cli/src/serve/server/error-handlers.ts Normal file
View file

@ -0,0 +1,30 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import express from 'express';
import type { Application, NextFunction, Request, Response } from 'express';
import { writeStderrLine } from '../../utils/stdioHelpers.js';
import { sendJsonBodyParserError } from './request-helpers.js';
export function installJsonBodyParser(app: Application): void {
app.use(express.json({ limit: '10mb' }));
app.use((err: unknown, _req: Request, res: Response, next: NextFunction) => {
if (sendJsonBodyParserError(res, err)) return;
next(err);
});
}
export function installFinalErrorHandler(app: Application): void {
app.use((err: unknown, _req: Request, res: Response, _next: NextFunction) => {
if (sendJsonBodyParserError(res, err)) return;
writeStderrLine(
`qwen serve: unhandled error: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`,
);
if (!res.headersSent) {
res.status(500).json({ error: 'Internal server error' });
}
});
}

46
packages/cli/src/serve/server/rate-limiter-setup.ts Normal file
View file

@ -0,0 +1,46 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application } from 'express';
import type { DaemonLogger } from '../daemon-logger.js';
import { createRateLimiter, type RateLimiterInstance } from '../rate-limit.js';
import type { ServeOptions } from '../types.js';
export function installRateLimiter(
app: Application,
opts: ServeOptions,
daemonLog: DaemonLogger | undefined,
): RateLimiterInstance | undefined {
if (!opts.rateLimit) return undefined;
const windowMs = opts.rateLimitWindowMs ?? 60_000;
const rateLimiter = createRateLimiter({
tiers: {
prompt: { windowMs, max: opts.rateLimitPrompt ?? 10 },
mutation: { windowMs, max: opts.rateLimitMutation ?? 30 },
read: { windowMs, max: opts.rateLimitRead ?? 120 },
},
hostname: opts.hostname,
onLimitReached: daemonLog
? (tier, key, suppressed) => {
daemonLog.warn(
`rate limit hit${suppressed > 0 ? ` (${suppressed} suppressed)` : ''}`,
{ tier, key: key.slice(0, 64) },
);
}
: undefined,
onError: daemonLog
? (err, path) => {
daemonLog.warn(
`rate limiter error (fail-open): ${err instanceof Error ? err.message : String(err)}`,
{ path },
);
}
: undefined,
});
app.use(rateLimiter.middleware);
return rateLimiter;
}

40
packages/cli/src/serve/server/self-origin.ts Normal file
View file

@ -0,0 +1,40 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import type { Application, Request } from 'express';
/**
* Allow same-origin requests from the demo page. Browsers send an `Origin`
* header on same-origin POST/fetch calls; the browser-origin wall would reject
* them. Only loopback origins are matched.
*/
export function installSelfOriginStripMiddleware(
app: Application,
getPort: () => number,
): void {
let cachedStripPort = -1;
let cachedSelfOrigins: Set<string> = new Set();
app.use((req: Request, _res, next) => {
const origin = req.headers.origin;
if (origin) {
const port = getPort();
if (port !== cachedStripPort) {
cachedStripPort = port;
cachedSelfOrigins = new Set([
`http://127.0.0.1:${port}`,
`http://localhost:${port}`,
`http://[::1]:${port}`,
`http://host.docker.internal:${port}`,
]);
}
if (cachedSelfOrigins.has(origin)) {
delete req.headers.origin;
}
}
next();
});
}

108
packages/cli/src/serve/server/serve-features.ts Normal file
View file

@ -0,0 +1,108 @@
/**
* @license
* Copyright 2025 Qwen Team
* SPDX-License-Identifier: Apache-2.0
*/
import { loadSettings } from '../../config/settings.js';
import { SUPPORTED_LANGUAGES } from '../../i18n/index.js';
import { hasConfiguredBatchVoiceTranscriptionModel } from '../../services/voice-service.js';
import { writeStderrLine } from '../../utils/stdioHelpers.js';
import { getAdvertisedServeFeatures } from '../capabilities.js';
import type { ServeOptions } from '../types.js';
// Keep in sync with acp-bridge bridge.ts and SDK DaemonClient.ts.
const DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION = 5;
export const SERVE_LANGUAGE_CODES = [
...SUPPORTED_LANGUAGES.map((language) => language.code),
'auto',
];
export function advertisedMaxPendingPromptsPerSession(
value: number | undefined,
): number | null {
if (value === undefined) return DEFAULT_MAX_PENDING_PROMPTS_PER_SESSION;
if (value === 0 || value === Number.POSITIVE_INFINITY) return null;
return value;
}
interface CreateServeFeaturesDeps {
opts: ServeOptions;
boundWorkspace: string;
persistSettingAvailable: boolean;
reloadAvailable: boolean;
sessionShellCommandEnabled: boolean;
}
export interface ServeFeaturesRuntime {
languageCodes: string[];
currentServeFeatures: () => ReturnType<typeof getAdvertisedServeFeatures>;
invalidateServeFeaturesCache: () => void;
}
export function createServeFeatures(
deps: CreateServeFeaturesDeps,
): ServeFeaturesRuntime {
const {
opts,
boundWorkspace,
persistSettingAvailable,
reloadAvailable,
sessionShellCommandEnabled,
} = deps;
let cachedVoiceTranscriptionAvailable: boolean | undefined;
const invalidateServeFeaturesCache = () => {
cachedVoiceTranscriptionAvailable = undefined;
};
const getCachedVoiceTranscriptionAvailable = () => {
cachedVoiceTranscriptionAvailable ??=
isWorkspaceVoiceTranscriptionAvailable(boundWorkspace);
return cachedVoiceTranscriptionAvailable;
};
return {
languageCodes: SERVE_LANGUAGE_CODES,
invalidateServeFeaturesCache,
currentServeFeatures: () =>
getAdvertisedServeFeatures(undefined, {
requireAuth: opts.requireAuth === true,
mcpPoolActive: opts.mcpPoolActive !== false,
allowOriginActive:
opts.allowOrigins !== undefined && opts.allowOrigins.length > 0,
...(opts.promptDeadlineMs !== undefined
? { promptDeadlineMs: opts.promptDeadlineMs }
: {}),
...(opts.writerIdleTimeoutMs !== undefined
? { writerIdleTimeoutMs: opts.writerIdleTimeoutMs }
: {}),
persistSettingAvailable,
sessionShellCommandEnabled,
rateLimit: opts.rateLimit === true,
reloadAvailable,
voiceTranscriptionAvailable: getCachedVoiceTranscriptionAvailable(),
// Advertised whenever the `/voice/stream` WS endpoint exists (ACP HTTP
// on). A configured token no longer suppresses it — the browser carries
// the bearer token via the WS subprotocol, which the upgrade listener
// verifies (acp-http/index.ts).
voiceWsAvailable: process.env['QWEN_SERVE_ACP_HTTP'] !== '0',
}),
};
}
function isWorkspaceVoiceTranscriptionAvailable(
boundWorkspace: string,
): boolean {
try {
return hasConfiguredBatchVoiceTranscriptionModel(
loadSettings(boundWorkspace),
);
} catch (err) {
writeStderrLine(
`qwen serve: workspace voice transcription capability check failed: ${
err instanceof Error ? err.message : String(err)
}`,
);
return false;
}
}