diff --git a/docs/developers/daemon/15-channel-adapters.md b/docs/developers/daemon/15-channel-adapters.md index 298078f703..64935e313b 100644 --- a/docs/developers/daemon/15-channel-adapters.md +++ b/docs/developers/daemon/15-channel-adapters.md @@ -162,6 +162,15 @@ sequenceDiagram - `shutdown()` closes every active session and the underlying transport (the channel's WebSocket / long-poll). - DingTalk's WebSocket stream supports server-push; WeChat's long-poll requires a backoff strategy on idle responses; Telegram's long-poll has a built-in `timeout` parameter. +### Settings reload (`POST /workspace/channel/reload`) + +The daemon reads channel settings from `settings.json` once, when the channel worker starts (`packages/cli/src/commands/channel/daemon-worker.ts` → `loadSettings` → `loadChannelsConfig`). To apply changes without a full daemon restart, the daemon exposes `POST /workspace/channel/reload` (strict mutation gate; SDK `DaemonClient.reloadChannelWorker()`; CLI `qwen channel reload`): + +- The route calls `ChannelWorkerSupervisor.restart()` (`packages/cli/src/serve/channel-worker-supervisor.ts`), which stops the current worker child and relaunches it. The relaunched worker re-reads `settings.json`, so channel tokens, `proxy`, and per-channel `model` all take effect. +- Concurrent reloads coalesce onto a single stop+relaunch. `restart()` also resets the crash-restart budget, so a worker parked in `failed` recovers on an explicit reload. +- If the relaunch fails (for example, settings were edited into an invalid state), the channels stay down, the route returns 5xx with the latest snapshot, and `GET /daemon/status` reports `failed`. +- The `channel_reload` capability and the route are advertised only when the daemon was started with `--channel`. Adding a brand-new channel name to a `--channel ` selection still requires a daemon restart; `--channel all` picks up newly-configured channels on reload. + ## Dependencies - `packages/channels/base/` — `ChannelBase`, `DaemonChannelBridge`, `types.ts` (`ChannelConfig`, `Envelope`, `SessionScope`, `ChannelPlugin`). @@ -196,6 +205,8 @@ Channel-specific keys layer on top (DingTalk: `streamCredentials`; WeChat: `ilin - `packages/channels/base/src/DaemonChannelBridge.ts` - `packages/channels/base/src/ChannelBase.ts` - `packages/channels/base/src/types.ts` +- `packages/cli/src/serve/channel-worker-supervisor.ts` (worker supervision + `restart()`) +- `packages/cli/src/serve/routes/workspace-channel-control.ts` (`POST /workspace/channel/reload`) - `packages/channels/dingtalk/src/DingtalkAdapter.ts` - `packages/channels/weixin/src/WeixinAdapter.ts` - `packages/channels/telegram/src/TelegramAdapter.ts` diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index 5773f30137..2ff9c8fd0f 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -15,7 +15,7 @@ Run Qwen Code as a local HTTP daemon so multiple clients (IDE plugins, web UIs, - **Reconnect-safe streaming** — SSE with `Last-Event-ID` reconnect lets a client drop and pick up exactly where it left off (within the ring's replay window). - **First-responder permissions** — when the agent asks for permission to run a tool, every connected client sees the request; whichever client answers first wins. - **One daemon, one workspace** — each `qwen serve` process binds to exactly one workspace at boot (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02). Multi-workspace deployments run one daemon per workspace on separate ports (or behind an orchestrator). -- **Experimental daemon-managed channels** — `qwen serve --channel ` starts a channel worker owned by the daemon lifecycle. The worker is a separate process, connects back to the daemon through the SDK, and reports its state in `GET /daemon/status`. +- **Experimental daemon-managed channels** — `qwen serve --channel ` starts a channel worker owned by the daemon lifecycle. The worker is a separate process, connects back to the daemon through the SDK, and reports its state in `GET /daemon/status`. After editing channel settings, reload it in place with `POST /workspace/channel/reload` (or `qwen channel reload`) — no full daemon restart needed. - **Remote runtime control** ([#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 17) — change a session's approval mode (`POST /session/:id/approval-mode`), toggle a tool per workspace (`POST /workspace/tools/:name/enable`), scaffold an empty `QWEN.md` (`POST /workspace/init`, mechanical only — does NOT call the model; for AI-fill, follow up with `POST /session/:id/prompt`), restart a single MCP server with a budget pre-check (`POST /workspace/mcp/:server/restart`), or add/remove MCP servers at runtime without a daemon restart (`POST /workspace/mcp/servers`, `DELETE /workspace/mcp/servers/:name`). All strict-gated — configure `--token` first. - **Session recap** ([#4175](https://github.com/QwenLM/qwen-code/issues/4175) follow-up) — fetch a one-sentence "where did I leave off" summary of an active session (`POST /session/:id/recap`). Wraps core's `generateSessionRecap` as a side-query against the fast model; pollutes neither the main chat history nor the SSE stream. Non-strict gate (same posture as `/prompt`); SDK helper `client.recapSession(sessionId)`. - **Known limit — token-cost amplification:** the route is a pure-cost endpoint (each call is an LLM side-query, no state benefit) and the daemon has no per-route rate limit in v1. On a no-token loopback default a buggy or malicious local client can spam it to burn tokens. Configure `--token` (and optionally `--require-auth`) on shared dev hosts before exposing the daemon. @@ -96,7 +96,9 @@ qwen serve --channel telegram --channel feishu qwen serve --channel all ``` -This mode is experimental and daemon-managed. It does not replace the standalone `qwen channel start` command: standalone channels still use the ACP-backed `AcpBridge` service. With `qwen serve --channel`, the daemon launches one channel worker process after the HTTP runtime is ready. If the worker exits after startup, the daemon keeps running and `GET /daemon/status` reports a `channel_worker_exited` warning. Automatic worker restart is deferred. +This mode is experimental and daemon-managed. It does not replace the standalone `qwen channel start` command: standalone channels still use the ACP-backed `AcpBridge` service. With `qwen serve --channel`, the daemon launches one channel worker process after the HTTP runtime is ready. If the worker crashes after startup, the daemon keeps running, relaunches it under a bounded restart policy, and reports its state (including `channel_worker_exited` warnings) in `GET /daemon/status`. + +The daemon reads each channel's settings (tokens, `proxy`, per-channel `model`) from `settings.json` once, when the worker starts. To apply changes without restarting the whole daemon, call `POST /workspace/channel/reload` (strict-gated; SDK `client.reloadChannelWorker()`, or `qwen channel reload`). The daemon stops and relaunches the channel worker, which re-reads `settings.json`; every selected channel briefly disconnects and reconnects, and persisted threads are restored from disk. The route is advertised as the `channel_reload` capability only when the daemon was started with `--channel`. Adding a brand-new channel name to a `--channel ` selection still requires a daemon restart (or use `--channel all`, which picks up newly-configured channels on reload). The daemon is bound to one workspace, so every selected channel's `cwd` must resolve to the daemon workspace. `--channel all` cannot be combined with named channels. diff --git a/integration-tests/cli/qwen-serve-routes.test.ts b/integration-tests/cli/qwen-serve-routes.test.ts index 81acb10c79..074dc0b6e9 100644 --- a/integration-tests/cli/qwen-serve-routes.test.ts +++ b/integration-tests/cli/qwen-serve-routes.test.ts @@ -225,9 +225,10 @@ describe('qwen serve — capabilities envelope', () => { // // Conditional tags absent under this suite's spawn flags (no // `--require-auth` / `--allow-origin` / deadline env vars / - // rate-limit opt-in, no configured batch ASR model): `require_auth`, - // `allow_origin`, `cdp_tunnel_over_ws`, `prompt_absolute_deadline`, - // `writer_idle_timeout`, `workspace_voice_transcription`, `rate_limit`. + // rate-limit opt-in, no `--channel`, no configured batch ASR model): + // `require_auth`, `allow_origin`, `cdp_tunnel_over_ws`, + // `prompt_absolute_deadline`, `writer_idle_timeout`, + // `workspace_voice_transcription`, `rate_limit`, `channel_reload`. // Pool tags (`mcp_workspace_pool`, `mcp_pool_restart`) ARE present // because the workspace MCP pool is on by default, as are // `workspace_settings`, `workspace_permissions`, `workspace_voice`, diff --git a/packages/cli/src/commands/channel.ts b/packages/cli/src/commands/channel.ts index a8a22830ec..efa423fc1e 100644 --- a/packages/cli/src/commands/channel.ts +++ b/packages/cli/src/commands/channel.ts @@ -2,6 +2,7 @@ import type { CommandModule, Argv } from 'yargs'; import { startCommand } from './channel/start.js'; import { stopCommand } from './channel/stop.js'; import { statusCommand } from './channel/status.js'; +import { reloadCommand } from './channel/reload.js'; import { daemonWorkerCommand } from './channel/daemon-worker.js'; import { pairingListCommand, @@ -30,6 +31,7 @@ export const channelCommand: CommandModule = { .command(daemonWorkerCommand) .command(stopCommand) .command(statusCommand) + .command(reloadCommand) .command(pairingCommand) .command(configureWeixinCommand) .demandCommand(1, 'You need at least one command before continuing.') diff --git a/packages/cli/src/commands/channel/reload.test.ts b/packages/cli/src/commands/channel/reload.test.ts new file mode 100644 index 0000000000..fd8f63b187 --- /dev/null +++ b/packages/cli/src/commands/channel/reload.test.ts @@ -0,0 +1,115 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const mockReloadChannelWorker = vi.hoisted(() => vi.fn()); +const mockDaemonClient = vi.hoisted(() => + vi.fn(() => ({ reloadChannelWorker: mockReloadChannelWorker })), +); +const mockWriteStdoutLine = vi.hoisted(() => vi.fn()); +const mockWriteStderrLine = vi.hoisted(() => vi.fn()); + +vi.mock('@qwen-code/sdk/daemon', () => ({ + DaemonClient: mockDaemonClient, +})); +vi.mock('../../utils/stdioHelpers.js', () => ({ + writeStdoutLine: mockWriteStdoutLine, + writeStderrLine: mockWriteStderrLine, +})); + +import { reloadCommand } from './reload.js'; + +type ReloadHandler = NonNullable; + +async function runHandler(argv: Record): Promise { + await (reloadCommand.handler as ReloadHandler)({ + _: [], + $0: 'qwen', + ...argv, + } as never); +} + +describe('channel reload command', () => { + beforeEach(() => { + vi.spyOn(process, 'exit').mockImplementation((() => undefined) as never); + vi.unstubAllEnvs(); + vi.stubEnv('QWEN_DAEMON_URL', undefined); + vi.stubEnv('QWEN_SERVER_TOKEN', undefined); + vi.stubEnv('QWEN_DAEMON_TOKEN', undefined); + }); + + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + mockReloadChannelWorker.mockReset(); + mockDaemonClient.mockClear(); + mockWriteStdoutLine.mockClear(); + mockWriteStderrLine.mockClear(); + }); + + it('reloads via the resolved daemon URL/token and prints the snapshot', async () => { + mockReloadChannelWorker.mockResolvedValue({ + reloaded: true, + worker: { + enabled: true, + state: 'running', + channels: ['telegram'], + pid: 4321, + restartCount: 2, + }, + }); + + await runHandler({ 'daemon-url': 'http://daemon:9', token: 'secret' }); + + expect(mockDaemonClient).toHaveBeenCalledWith({ + baseUrl: 'http://daemon:9', + token: 'secret', + }); + expect(mockReloadChannelWorker).toHaveBeenCalledTimes(1); + const line = mockWriteStdoutLine.mock.calls[0]?.[0] as string; + expect(line).toContain('state=running'); + expect(line).toContain('channels=telegram'); + expect(line).toContain('pid=4321'); + expect(line).toContain('restarts=2'); + expect(process.exit).toHaveBeenCalledWith(0); + expect(mockWriteStderrLine).not.toHaveBeenCalled(); + }); + + it('defaults the daemon URL and omits the token when neither flag nor env is set', async () => { + mockReloadChannelWorker.mockResolvedValue({ + reloaded: true, + worker: { enabled: true, state: 'running', channels: [] }, + }); + + await runHandler({}); + + expect(mockDaemonClient).toHaveBeenCalledWith({ + baseUrl: 'http://127.0.0.1:4170', + }); + }); + + it('falls back to QWEN_DAEMON_URL and QWEN_SERVER_TOKEN from the environment', async () => { + vi.stubEnv('QWEN_DAEMON_URL', 'http://env-daemon:5'); + vi.stubEnv('QWEN_SERVER_TOKEN', 'env-token'); + mockReloadChannelWorker.mockResolvedValue({ + reloaded: true, + worker: { enabled: true, state: 'running', channels: [] }, + }); + + await runHandler({}); + + expect(mockDaemonClient).toHaveBeenCalledWith({ + baseUrl: 'http://env-daemon:5', + token: 'env-token', + }); + }); + + it('reports failures on stderr and exits non-zero', async () => { + mockReloadChannelWorker.mockRejectedValue(new Error('no channel worker')); + + await runHandler({ 'daemon-url': 'http://daemon:9' }); + + const line = mockWriteStderrLine.mock.calls[0]?.[0] as string; + expect(line).toContain('Reload failed'); + expect(line).toContain('no channel worker'); + expect(process.exit).toHaveBeenCalledWith(1); + }); +}); diff --git a/packages/cli/src/commands/channel/reload.ts b/packages/cli/src/commands/channel/reload.ts new file mode 100644 index 0000000000..14cb30c546 --- /dev/null +++ b/packages/cli/src/commands/channel/reload.ts @@ -0,0 +1,123 @@ +import type { CommandModule } from 'yargs'; +import { writeStderrLine, writeStdoutLine } from '../../utils/stdioHelpers.js'; +import { + QWEN_DAEMON_TOKEN_ENV, + QWEN_DAEMON_URL_ENV, + QWEN_SERVER_TOKEN_ENV, +} from '../../serve/channel-worker-env.js'; + +const DEFAULT_DAEMON_URL = 'http://127.0.0.1:4170'; + +// Structural subset of the SDK's DaemonChannelReloadResult. Kept local (like +// daemon-worker.ts's DaemonSdkLike) so this command doesn't take a static +// type dependency on the SDK subpath. +interface ChannelReloadResultLike { + reloaded: boolean; + worker: { + state: string; + channels: string[]; + pid?: number; + restartCount?: number; + error?: string; + }; +} + +interface DaemonClientLike { + reloadChannelWorker(opts?: { + clientId?: string; + timeoutMs?: number; + }): Promise; +} + +interface DaemonSdkLike { + DaemonClient: new (opts: { + baseUrl: string; + token?: string; + }) => DaemonClientLike; +} + +interface ReloadArgs { + 'daemon-url'?: string; + token?: string; + timeout?: number; +} + +function resolveDaemonUrl(flag: string | undefined): string { + // `||` (not `??`) so an empty flag or empty env var falls through to the + // default rather than producing an unusable empty base URL. + return flag || process.env[QWEN_DAEMON_URL_ENV] || DEFAULT_DAEMON_URL; +} + +function resolveToken(flag: string | undefined): string | undefined { + return ( + flag ?? + process.env[QWEN_SERVER_TOKEN_ENV] ?? + process.env[QWEN_DAEMON_TOKEN_ENV] + ); +} + +export const reloadCommand: CommandModule = { + command: 'reload', + describe: + 'Reload the daemon-managed channel worker so it re-reads settings.json', + builder: (yargs) => + yargs + .option('daemon-url', { + type: 'string', + description: `Daemon base URL (default: $${QWEN_DAEMON_URL_ENV} or ${DEFAULT_DAEMON_URL})`, + }) + .option('token', { + type: 'string', + description: `Bearer token (default: $${QWEN_SERVER_TOKEN_ENV})`, + }) + .option('timeout', { + type: 'number', + description: 'Request timeout in milliseconds', + }), + handler: async (argv) => { + const baseUrl = resolveDaemonUrl(argv['daemon-url']); + const token = resolveToken(argv.token); + + let sdk: DaemonSdkLike; + try { + sdk = (await import('@qwen-code/sdk/daemon')) as unknown as DaemonSdkLike; + } catch (err) { + writeStderrLine( + `[Channel] Failed to load daemon SDK: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + process.exit(1); + } + + const client = new sdk.DaemonClient({ + baseUrl, + ...(token ? { token } : {}), + }); + + try { + const result = await client.reloadChannelWorker( + argv.timeout !== undefined ? { timeoutMs: argv.timeout } : undefined, + ); + const worker = result.worker; + const parts = [ + `state=${worker.state}`, + `channels=${worker.channels.join(', ') || 'none'}`, + ...(worker.pid !== undefined ? [`pid=${worker.pid}`] : []), + ...(worker.restartCount !== undefined + ? [`restarts=${worker.restartCount}`] + : []), + ...(worker.error ? [`error=${worker.error}`] : []), + ]; + writeStdoutLine(`[Channel] Reloaded (${parts.join(', ')}).`); + process.exit(0); + } catch (err) { + writeStderrLine( + `[Channel] Reload failed (${baseUrl}): ${ + err instanceof Error ? err.message : String(err) + }`, + ); + process.exit(1); + } + }, +}; diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index b35f1ad461..3720317448 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -259,6 +259,13 @@ export const SERVE_CAPABILITY_REGISTRY = { session_branch: { since: 'v1' }, rate_limit: { since: 'v1' }, workspace_reload: { since: 'v1' }, + // Daemon supports reloading its daemon-managed channel worker via + // `POST /workspace/channel/reload`. The worker is stopped and relaunched; + // on relaunch it re-reads settings.json (channels / proxy / per-channel + // model), so channel settings changes apply without a full daemon restart. + // Advertised CONDITIONALLY — only when the daemon was started with + // `--channel` (i.e. a channel worker exists to reload). + channel_reload: { since: 'v1' }, // Multi-workspace sessions closed loop (issue #6378 Phase 2a). Advertised // only when one daemon hosts more than one registered workspace runtime. multi_workspace_sessions: { since: 'v1' }, @@ -315,6 +322,12 @@ export interface AdvertiseFeatureToggles { sessionArtifactsPersistenceAvailable?: boolean; rateLimit?: boolean; reloadAvailable?: boolean; + /** + * Whether the daemon exposes the channel worker reload route + * (`channel_reload`). Set only when the daemon was started with + * `--channel`, so a channel worker exists to reload. + */ + channelReloadAvailable?: boolean; /** * Whether the daemon will accept client-hosted MCP servers over the WS * (`client_mcp_over_ws`, issue #5626). @@ -402,6 +415,7 @@ export const CONDITIONAL_SERVE_FEATURES: ReadonlyMap< ], ['rate_limit', (toggles) => toggles.rateLimit === true], ['workspace_reload', (toggles) => toggles.reloadAvailable === true], + ['channel_reload', (toggles) => toggles.channelReloadAvailable === true], [ 'multi_workspace_sessions', (toggles) => toggles.multiWorkspaceSessionsEnabled === true, diff --git a/packages/cli/src/serve/channel-worker-supervisor.test.ts b/packages/cli/src/serve/channel-worker-supervisor.test.ts index 5b0bc880e2..8b751d6cb2 100644 --- a/packages/cli/src/serve/channel-worker-supervisor.test.ts +++ b/packages/cli/src/serve/channel-worker-supervisor.test.ts @@ -1986,4 +1986,195 @@ describe('createChannelWorkerSupervisor', () => { requestedChannels: ['telegram'], }); }); + + describe('restart()', () => { + const waitForCalls = async (getCount: () => number, count: number) => { + for (let i = 0; i < 100 && getCount() < count; i++) { + await new Promise((resolve) => setTimeout(resolve, 0)); + } + }; + + it('stops the running worker and relaunches it so settings are re-read', async () => { + const child1 = new FakeChild(); + const child2 = new FakeChild(); + const queue = [child1, child2]; + const spawnWorker = vi.fn(() => queue.shift()!); + const supervisor = createChannelWorkerSupervisor({ + cliEntryPath: '/repo/dist/index.js', + daemonUrl: 'http://127.0.0.1:4170', + workspace: '/workspace', + selection: { mode: 'names', names: ['telegram'] }, + spawnWorker, + }); + + const started = supervisor.start(); + child1.emit('message', { + type: 'ready', + pid: 111, + channels: ['telegram'], + }); + await started; + expect(supervisor.snapshot()).toMatchObject({ + state: 'running', + pid: 111, + }); + + const restarted = supervisor.restart(); + await waitForCalls(() => spawnWorker.mock.calls.length, 2); + child2.emit('message', { + type: 'ready', + pid: 222, + channels: ['telegram'], + }); + const snapshot = await restarted; + + expect(child1.kill).toHaveBeenCalled(); + expect(spawnWorker).toHaveBeenCalledTimes(2); + expect(snapshot).toMatchObject({ state: 'running', pid: 222 }); + expect(supervisor.snapshot()).toMatchObject({ + state: 'running', + pid: 222, + }); + }); + + it('coalesces concurrent restarts onto a single relaunch', async () => { + const child1 = new FakeChild(); + const child2 = new FakeChild(); + const queue = [child1, child2]; + const spawnWorker = vi.fn(() => queue.shift()!); + const supervisor = createChannelWorkerSupervisor({ + cliEntryPath: '/repo/dist/index.js', + daemonUrl: 'http://127.0.0.1:4170', + workspace: '/workspace', + selection: { mode: 'names', names: ['telegram'] }, + spawnWorker, + }); + + const started = supervisor.start(); + child1.emit('message', { + type: 'ready', + pid: 111, + channels: ['telegram'], + }); + await started; + + const first = supervisor.restart(); + const second = supervisor.restart(); + await waitForCalls(() => spawnWorker.mock.calls.length, 2); + child2.emit('message', { + type: 'ready', + pid: 222, + channels: ['telegram'], + }); + const [a, b] = await Promise.all([first, second]); + + // Initial start + exactly one relaunch: concurrent restarts share the + // same in-flight promise and never fork a second worker. + expect(spawnWorker).toHaveBeenCalledTimes(2); + expect(a).toMatchObject({ state: 'running', pid: 222 }); + expect(b).toMatchObject({ state: 'running', pid: 222 }); + }); + + it('rejects on relaunch failure, then recovers on a later restart', async () => { + const child1 = new FakeChild(); + const child2 = new FakeChild(); + let call = 0; + const spawnWorker = vi.fn(() => { + call += 1; + if (call === 2) { + throw new Error('fork failed'); + } + return call === 1 ? child1 : child2; + }); + const supervisor = createChannelWorkerSupervisor({ + cliEntryPath: '/repo/dist/index.js', + daemonUrl: 'http://127.0.0.1:4170', + workspace: '/workspace', + selection: { mode: 'names', names: ['telegram'] }, + spawnWorker, + }); + + const started = supervisor.start(); + child1.emit('message', { + type: 'ready', + pid: 111, + channels: ['telegram'], + }); + await started; + + // Second spawn throws — the relaunch fails and the worker parks in + // `failed` (channels down), and restart() surfaces the error. + await expect(supervisor.restart()).rejects.toThrow(/fork failed/); + expect(supervisor.snapshot()).toMatchObject({ state: 'failed' }); + + // A later restart resets the budget and recovers once spawning works. + const recovered = supervisor.restart(); + await waitForCalls(() => spawnWorker.mock.calls.length, 3); + child2.emit('message', { + type: 'ready', + pid: 333, + channels: ['telegram'], + }); + expect(await recovered).toMatchObject({ state: 'running', pid: 333 }); + }); + + it('does not relaunch after killAllSync latches disposed', async () => { + const child1 = new FakeChild(); + const spawnWorker = vi.fn(() => child1); + const supervisor = createChannelWorkerSupervisor({ + cliEntryPath: '/repo/dist/index.js', + daemonUrl: 'http://127.0.0.1:4170', + workspace: '/workspace', + selection: { mode: 'names', names: ['telegram'] }, + spawnWorker, + }); + + const started = supervisor.start(); + child1.emit('message', { + type: 'ready', + pid: 111, + channels: ['telegram'], + }); + await started; + + supervisor.killAllSync(); + expect(spawnWorker).toHaveBeenCalledTimes(1); + + // A reload after a hard shutdown must be a no-op, not a relaunch. + const snapshot = await supervisor.restart(); + expect(spawnWorker).toHaveBeenCalledTimes(1); + expect(snapshot.state).toBe('stopped'); + }); + + it('does not fork a new worker when killAllSync races an in-flight restart', async () => { + const child1 = new FakeChild(); + const child2 = new FakeChild(); + const queue = [child1, child2]; + const spawnWorker = vi.fn(() => queue.shift()!); + const supervisor = createChannelWorkerSupervisor({ + cliEntryPath: '/repo/dist/index.js', + daemonUrl: 'http://127.0.0.1:4170', + workspace: '/workspace', + selection: { mode: 'names', names: ['telegram'] }, + spawnWorker, + }); + + const started = supervisor.start(); + child1.emit('message', { + type: 'ready', + pid: 111, + channels: ['telegram'], + }); + await started; + + // Begin a reload, then simulate a hard daemon shutdown before the + // relaunch's start() runs. The disposed latch must prevent a new fork + // (which would otherwise orphan a worker spawned during shutdown). + const restarted = supervisor.restart(); + supervisor.killAllSync(); + await restarted; + + expect(spawnWorker).toHaveBeenCalledTimes(1); + }); + }); }); diff --git a/packages/cli/src/serve/channel-worker-supervisor.ts b/packages/cli/src/serve/channel-worker-supervisor.ts index 5de3acfc50..b3875cc856 100644 --- a/packages/cli/src/serve/channel-worker-supervisor.ts +++ b/packages/cli/src/serve/channel-worker-supervisor.ts @@ -68,6 +68,12 @@ export interface ChannelWorkerSnapshot { export interface ChannelWorkerSupervisor { start(): Promise; stop(): Promise; + /** + * Stop the current worker (if any) and relaunch it. The relaunched worker + * re-reads settings.json, so this is how settings changes are applied + * without restarting the whole daemon. Rejects if the relaunch fails. + */ + restart(): Promise; killAllSync(): void; snapshot(): ChannelWorkerSnapshot; } @@ -448,6 +454,8 @@ export function createChannelWorkerSupervisor( let restartTimer: NodeJS.Timeout | undefined; let staleHeartbeatTimer: NodeJS.Timeout | undefined; let restartAttemptTimes: number[] = []; + let restarting: Promise | undefined; + let disposed = false; const snapshotCopy = (): ChannelWorkerSnapshot => ({ ...snapshot, @@ -815,9 +823,12 @@ export function createChannelWorkerSupervisor( }); }; - return { + const supervisor: ChannelWorkerSupervisor = { async start() { - if (child) return; + // `disposed` is latched only by killAllSync() (hard shutdown), so the + // supported stop()/start() reuse lifecycle is preserved; this guard just + // prevents a relaunch into a daemon that is being force-torn-down. + if (disposed || child) return; stopping = false; clearRestartTimer(); restartAttemptTimes = []; @@ -858,7 +869,29 @@ export function createChannelWorkerSupervisor( stopping = false; snapshot = { ...snapshot, state: 'stopped' }; }, + async restart() { + // A hard shutdown (killAllSync) latches `disposed`; a reload racing that + // must not relaunch a worker into a tearing-down daemon. + if (disposed) return snapshotCopy(); + // Coalesce concurrent reloads onto one stop+relaunch so a burst of + // reload requests cannot fork multiple workers. + restarting ??= (async () => { + try { + await supervisor.stop(); + // start() bails if a child is still attached (stop cleared it) or if + // killAllSync latched `disposed` mid-reload — avoiding an orphaned + // fork. It also resets the restart budget, so a worker previously + // parked in `failed` recovers on an explicit reload. + await supervisor.start(); + return snapshotCopy(); + } finally { + restarting = undefined; + } + })(); + return restarting; + }, killAllSync() { + disposed = true; if ( !child || snapshot.state === 'exited' || @@ -888,4 +921,5 @@ export function createChannelWorkerSupervisor( return snapshotCopy(); }, }; + return supervisor; } diff --git a/packages/cli/src/serve/routes/workspace-channel-control.ts b/packages/cli/src/serve/routes/workspace-channel-control.ts new file mode 100644 index 0000000000..f55a708b86 --- /dev/null +++ b/packages/cli/src/serve/routes/workspace-channel-control.ts @@ -0,0 +1,58 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { Application, Request, RequestHandler, Response } from 'express'; +import type { ChannelWorkerSnapshot } from '../channel-worker-supervisor.js'; +import type { SendBridgeError } from '../server/error-response.js'; + +interface RegisterWorkspaceChannelControlRoutesDeps { + getChannelWorkerSnapshot: () => ChannelWorkerSnapshot; + reloadChannelWorker: () => Promise; + mutate: (opts?: { strict?: boolean }) => RequestHandler; + sendBridgeError: SendBridgeError; + parseAndValidateClientId: ( + req: Request, + res: Response, + ) => string | undefined | null; +} + +export function registerWorkspaceChannelControlRoutes( + app: Application, + deps: RegisterWorkspaceChannelControlRoutesDeps, +): void { + const { + getChannelWorkerSnapshot, + reloadChannelWorker, + mutate, + sendBridgeError, + parseAndValidateClientId, + } = deps; + + app.post( + '/workspace/channel/reload', + mutate({ strict: true }), + async (req, res) => { + const clientId = parseAndValidateClientId(req, res); + if (clientId === null) return; + if (!getChannelWorkerSnapshot().enabled) { + res.status(409).json({ + error: + 'This daemon has no channel worker to reload. Start it with `qwen serve --channel `.', + code: 'channel_worker_not_enabled', + }); + return; + } + try { + const worker = await reloadChannelWorker(); + res.status(200).json({ reloaded: true, worker }); + } catch (err) { + sendBridgeError(res, err, { + route: 'POST /workspace/channel/reload', + }); + } + }, + ); +} diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index d060ccc693..7226f00320 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -3843,6 +3843,7 @@ describe('runQwenServe channel worker supervisor', () => { return { start: vi.fn().mockResolvedValue(undefined), stop: vi.fn().mockResolvedValue(undefined), + restart: vi.fn().mockResolvedValue(snapshot), killAllSync: vi.fn(), snapshot: vi.fn(() => snapshot), }; diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 8249b1002b..715ea0acb9 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -570,6 +570,9 @@ function createDisabledChannelWorkerSupervisor(): ChannelWorkerSupervisor { return { async start() {}, async stop() {}, + async restart() { + return { ...snapshot, channels: [] }; + }, killAllSync() {}, snapshot: () => ({ ...snapshot, channels: [] }), }; @@ -861,6 +864,7 @@ function currentServeFeaturesForRunQwenServe( sessionArtifactsPersistenceAvailable, rateLimit: opts.rateLimit === true, reloadAvailable: true, + channelReloadAvailable: opts.channelSelection !== undefined, // Advertise the same WS feature flags as the runtime path (serve-features.ts) // so the bootstrap `/capabilities` window doesn't briefly under-report them. clientMcpOverWsEnabled: opts.clientMcpOverWs === true, @@ -2096,6 +2100,7 @@ export async function runQwenServe( let channelWorker: ChannelWorkerSupervisor = createDisabledChannelWorkerSupervisor(); const getChannelWorkerSnapshot = () => channelWorker.snapshot(); + const reloadChannelWorker = () => channelWorker.restart(); const writeChannelWorkerPidfile = ( snapshot = channelWorker.snapshot(), options: { clearWorkerPid?: boolean } = {}, @@ -3174,6 +3179,10 @@ export async function runQwenServe( primaryRuntimeEnv, daemonLog, getChannelWorkerSnapshot, + // Gate both the `channel_reload` capability and the reload route on the + // presence of this dep, so it is advertised only when a channel worker + // exists to reload. + ...(opts.channelSelection ? { reloadChannelWorker } : {}), getPerfSnapshot: () => ({ eventLoop: currentDaemonEventLoopMonitor.snapshot(), promptQueueWait: { diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 4adfc71d16..cbb9beeb18 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -20,6 +20,7 @@ import { PromptDeadlineExceededError, resolvePromptDeadlineMs, } from './server.js'; +import type { ChannelWorkerSnapshot } from './channel-worker-supervisor.js'; import { runQwenServe, type RunHandle } from './run-qwen-serve.js'; import { resolveWebShellDir, @@ -379,6 +380,7 @@ const EXPECTED_REGISTERED_FEATURES = [ 'session_branch', 'rate_limit', 'workspace_reload', + 'channel_reload', 'multi_workspace_sessions', 'client_mcp_over_ws', 'cdp_tunnel_over_ws', @@ -2132,6 +2134,20 @@ describe('createServeApp', () => { ); continue; } + if (feature === 'channel_reload') { + expect(predicate({ channelReloadAvailable: true })).toBe(true); + expect(predicate({ channelReloadAvailable: false })).toBe(false); + expect(predicate({})).toBe(false); + expect( + getAdvertisedServeFeatures(undefined, { + channelReloadAvailable: true, + }), + ).toContain(feature); + expect(getAdvertisedServeFeatures(undefined, {})).not.toContain( + feature, + ); + continue; + } if (feature === 'multi_workspace_sessions') { expect(predicate({ multiWorkspaceSessionsEnabled: true })).toBe(true); expect(predicate({ multiWorkspaceSessionsEnabled: false })).toBe( @@ -9651,6 +9667,149 @@ describe('createServeApp', () => { }); }); + describe('POST /workspace/channel/reload', () => { + const tokenOpts: ServeOptions = { ...baseOpts, token: 'secret' }; + const auth = (req: request.Test): request.Test => + req + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + const runningSnapshot: ChannelWorkerSnapshot = { + enabled: true, + state: 'running', + channels: ['telegram'], + pid: 4321, + }; + const disabledSnapshot: ChannelWorkerSnapshot = { + enabled: false, + state: 'disabled', + channels: [], + }; + + it('requires strict mutation auth before reloading', async () => { + const reloadChannelWorker = vi.fn(async () => runningSnapshot); + const app = createServeApp(baseOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + getChannelWorkerSnapshot: () => runningSnapshot, + reloadChannelWorker, + }); + + const res = await request(app) + .post('/workspace/channel/reload') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({}); + + expect(res.status).toBe(401); + expect(res.body.code).toBe('token_required'); + expect(reloadChannelWorker).not.toHaveBeenCalled(); + }); + + it('reloads the channel worker and returns the post-reload snapshot', async () => { + const reloadedSnapshot: ChannelWorkerSnapshot = { + ...runningSnapshot, + pid: 9999, + restartCount: 1, + }; + const reloadChannelWorker = vi.fn(async () => reloadedSnapshot); + const app = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + getChannelWorkerSnapshot: () => runningSnapshot, + reloadChannelWorker, + }); + + const res = await auth( + request(app).post('/workspace/channel/reload'), + ).send({}); + + expect(res.status).toBe(200); + expect(res.body).toEqual({ reloaded: true, worker: reloadedSnapshot }); + expect(reloadChannelWorker).toHaveBeenCalledTimes(1); + }); + + it('returns 409 when no channel worker is enabled', async () => { + const reloadChannelWorker = vi.fn(async () => disabledSnapshot); + const app = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + getChannelWorkerSnapshot: () => disabledSnapshot, + reloadChannelWorker, + }); + + const res = await auth( + request(app).post('/workspace/channel/reload'), + ).send({}); + + expect(res.status).toBe(409); + expect(res.body.code).toBe('channel_worker_not_enabled'); + expect(reloadChannelWorker).not.toHaveBeenCalled(); + }); + + it('maps relaunch failures through sendBridgeError', async () => { + const reloadChannelWorker = vi.fn(async () => { + throw new Error('relaunch failed'); + }); + const app = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + getChannelWorkerSnapshot: () => runningSnapshot, + reloadChannelWorker, + }); + + const res = await auth( + request(app).post('/workspace/channel/reload'), + ).send({}); + + expect(res.status).toBe(500); + expect(res.body).toEqual({ error: 'relaunch failed' }); + expect(reloadChannelWorker).toHaveBeenCalledTimes(1); + }); + + it('advertises channel_reload only when the reload dep is wired', async () => { + const withWorker = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + getChannelWorkerSnapshot: () => runningSnapshot, + reloadChannelWorker: vi.fn(async () => runningSnapshot), + }); + const withCaps = await auth(request(withWorker).get('/capabilities')); + expect(withCaps.body.features).toContain('channel_reload'); + + const withoutWorker = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + }); + const withoutCaps = await auth( + request(withoutWorker).get('/capabilities'), + ); + expect(withoutCaps.body.features).not.toContain('channel_reload'); + + // Asymmetric deps must NOT advertise: the route needs both + // getChannelWorkerSnapshot and reloadChannelWorker, so the capability + // must not appear when only one is wired (else it would advertise while + // the route 404s). + const onlyReload = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + reloadChannelWorker: vi.fn(async () => runningSnapshot), + }); + const onlyReloadCaps = await auth( + request(onlyReload).get('/capabilities'), + ); + expect(onlyReloadCaps.body.features).not.toContain('channel_reload'); + + const onlySnapshot = createServeApp(tokenOpts, undefined, { + bridge: fakeBridge(), + boundWorkspace: WS_BOUND, + getChannelWorkerSnapshot: () => runningSnapshot, + }); + const onlySnapshotCaps = await auth( + request(onlySnapshot).get('/capabilities'), + ); + expect(onlySnapshotCaps.body.features).not.toContain('channel_reload'); + }); + }); + describe('workspace trust routes', () => { const tokenOpts: ServeOptions = { ...baseOpts, token: 'secret' }; const auth = (req: request.Test): request.Test => diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index a4b9a87819..40000e846c 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -131,6 +131,7 @@ import { } from './workspace-registry.js'; import { registerWorkspaceLifecycleRoutes } from './routes/workspace-lifecycle.js'; import { registerWorkspaceMcpControlRoutes } from './routes/workspace-mcp-control.js'; +import { registerWorkspaceChannelControlRoutes } from './routes/workspace-channel-control.js'; import { registerWorkspaceToolsRoutes } from './routes/workspace-tools.js'; export { @@ -256,6 +257,13 @@ export interface ServeAppDeps { daemonLog?: DaemonLogger; startup?: DaemonStartupSnapshot; getChannelWorkerSnapshot?: () => ChannelWorkerSnapshot; + /** + * Stop and relaunch the daemon-managed channel worker so it re-reads + * settings.json. Wired only when the daemon owns a channel worker; its + * presence gates the `channel_reload` capability and the + * `POST /workspace/channel/reload` route. + */ + reloadChannelWorker?: () => Promise; getPerfSnapshot?: () => DaemonPerfSnapshot; /** Rolling metrics series for the Daemon Status charts (oldest→newest). */ getMetricsSeries?: () => DaemonMetricsBucket[]; @@ -506,6 +514,12 @@ export function createServeApp( // runtime, so it has the same reload surface as legacy deps.workspace. reloadAvailable: deps.workspace !== undefined || injectedWorkspaceRegistry !== undefined, + // Advertise `channel_reload` only when BOTH deps the route needs are + // wired — the same condition that gates route registration below — so an + // embedder can never see the capability advertised while the route 404s. + channelReloadAvailable: + deps.getChannelWorkerSnapshot !== undefined && + deps.reloadChannelWorker !== undefined, sessionShellCommandEnabled, multiWorkspaceSessionsEnabled: (injectedWorkspaceRegistry?.list().length ?? 1) > 1, @@ -986,6 +1000,18 @@ export function createServeApp( parseAndValidateClientId: (req, res) => parseAndValidateWorkspaceClientId(req, res, primaryBridge), }); + if (deps.getChannelWorkerSnapshot && deps.reloadChannelWorker) { + const getChannelWorkerSnapshot = deps.getChannelWorkerSnapshot; + const reloadChannelWorker = deps.reloadChannelWorker; + registerWorkspaceChannelControlRoutes(app, { + getChannelWorkerSnapshot, + reloadChannelWorker, + mutate, + sendBridgeError, + parseAndValidateClientId: (req, res) => + parseAndValidateWorkspaceClientId(req, res, primaryBridge), + }); + } registerWorkspaceLifecycleRoutes(app, { boundWorkspace: primaryBoundWorkspace, workspace: primaryWorkspace, diff --git a/packages/cli/src/serve/server/serve-features.ts b/packages/cli/src/serve/server/serve-features.ts index 1a1c87ae96..4faff440ad 100644 --- a/packages/cli/src/serve/server/serve-features.ts +++ b/packages/cli/src/serve/server/serve-features.ts @@ -43,6 +43,7 @@ interface CreateServeFeaturesDeps { persistSettingAvailable: boolean; sessionArtifactsPersistenceAvailable: boolean; reloadAvailable: boolean; + channelReloadAvailable: boolean; sessionShellCommandEnabled: boolean; multiWorkspaceSessionsEnabled: boolean; } @@ -62,6 +63,7 @@ export function createServeFeatures( persistSettingAvailable, sessionArtifactsPersistenceAvailable, reloadAvailable, + channelReloadAvailable, sessionShellCommandEnabled, multiWorkspaceSessionsEnabled, } = deps; @@ -95,6 +97,7 @@ export function createServeFeatures( sessionArtifactsPersistenceAvailable, rateLimit: opts.rateLimit === true, reloadAvailable, + channelReloadAvailable, multiWorkspaceSessionsEnabled, clientMcpOverWsEnabled: opts.clientMcpOverWs === true, cdpTunnelOverWsEnabled: opts.cdpTunnelOverWs === true, diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 2e2ff0b7e4..b63e650d34 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -96,6 +96,7 @@ import type { DaemonInitWorkspaceResult, DaemonMcpRestartResult, DaemonReloadResponse, + DaemonChannelReloadResult, DaemonMcpManageAction, DaemonMcpManageResult, DaemonSessionBtwResult, @@ -2444,6 +2445,36 @@ export class DaemonClient { ); } + /** + * Reload the daemon-managed channel worker: the daemon stops and relaunches + * it so it re-reads settings.json (channels / proxy / per-channel model). + * Requires the daemon to have been started with `--channel`; otherwise the + * route responds 409. Pre-flight the `channel_reload` capability. + */ + async reloadChannelWorker(opts?: { + clientId?: string; + timeoutMs?: number; + }): Promise { + return await this.fetchWithTimeout( + `${this.baseUrl}/workspace/channel/reload`, + { + method: 'POST', + headers: this.headers( + { 'Content-Type': 'application/json' }, + opts?.clientId, + ), + body: '{}', + }, + async (res) => { + if (!res.ok) { + throw await this.failOnError(res, 'POST /workspace/channel/reload'); + } + return (await res.json()) as DaemonChannelReloadResult; + }, + opts?.timeoutMs, + ); + } + async manageMcpServer( serverName: string, action: DaemonMcpManageAction, diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index e83b880de9..36284dcee4 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -328,6 +328,8 @@ export type { DaemonMcpManageResult, DaemonMcpRestartResult, DaemonReloadResponse, + DaemonChannelReloadResult, + DaemonChannelWorkerSnapshot, DaemonRewindResult, DaemonRewindSnapshotInfo, DaemonSessionBtwResult, diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 17bdd88e3d..6f729f7d1b 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -327,23 +327,7 @@ export interface DaemonStatusReport { channel: { live: boolean }; // Mirrors the daemon's ChannelWorkerSnapshot. `state` and `signal` are // widened to string to avoid coupling the wire type to the daemon's unions. - channelWorker: { - enabled: boolean; - state: string; - channels: string[]; - requestedChannels?: string[]; - pid?: number; - startedAt?: string; - exitCode?: number | null; - signal?: string | null; - error?: string; - restartCount?: number; - lastExitAt?: string; - lastRestartAt?: string; - nextRestartAt?: string; - lastHeartbeatAt?: string; - staleHeartbeatAt?: string; - }; + channelWorker: DaemonChannelWorkerSnapshot; transport: { restSseActive: number; acp: { @@ -2150,6 +2134,38 @@ export interface DaemonReloadResponse { childError?: string; } +/** + * Mirrors the daemon's ChannelWorkerSnapshot. `state` and `signal` are + * widened to string to avoid coupling the wire type to the daemon's unions. + */ +export interface DaemonChannelWorkerSnapshot { + enabled: boolean; + state: string; + channels: string[]; + requestedChannels?: string[]; + pid?: number; + startedAt?: string; + exitCode?: number | null; + signal?: string | null; + error?: string; + restartCount?: number; + lastExitAt?: string; + lastRestartAt?: string; + nextRestartAt?: string; + lastHeartbeatAt?: string; + staleHeartbeatAt?: string; +} + +/** + * Result of `POST /workspace/channel/reload`: the daemon stopped and + * relaunched its channel worker (which re-reads settings.json). `worker` is + * the post-reload snapshot. + */ +export interface DaemonChannelReloadResult { + reloaded: boolean; + worker: DaemonChannelWorkerSnapshot; +} + export type DaemonMcpRestartResult = | { serverName: string; diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index 008ff06ed8..3d9ff3a819 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -3017,6 +3017,52 @@ describe('DaemonClient', () => { }); }); + describe('reloadChannelWorker', () => { + it('POSTs /workspace/channel/reload with an empty body and returns the typed result', async () => { + const worker = { + enabled: true, + state: 'running', + channels: ['telegram'], + pid: 4321, + restartCount: 1, + }; + const { fetch, calls } = recordingFetch(() => + jsonResponse(200, { reloaded: true, worker }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const result = await client.reloadChannelWorker(); + expect(result).toEqual({ reloaded: true, worker }); + expect(calls[0]?.url).toBe('http://daemon/workspace/channel/reload'); + expect(calls[0]?.method).toBe('POST'); + expect(calls[0]?.body).toBe('{}'); + }); + + it('forwards the client id header', async () => { + const { fetch, calls } = recordingFetch(() => + jsonResponse(200, { + reloaded: true, + worker: { enabled: true, state: 'running', channels: [] }, + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + await client.reloadChannelWorker({ clientId: 'client-9' }); + expect(calls[0]?.headers['x-qwen-client-id']).toBe('client-9'); + }); + + it('rejects on a non-2xx response (channels not enabled)', async () => { + const { fetch } = recordingFetch(() => + jsonResponse(409, { + error: 'no channel worker', + code: 'channel_worker_not_enabled', + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + await expect(client.reloadChannelWorker()).rejects.toMatchObject({ + status: 409, + }); + }); + }); + describe('restartMcpServer (#4175 Wave 4 PR 17)', () => { it('POSTs an empty body and returns the typed result on success', async () => { const { fetch, calls } = recordingFetch(() =>