mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-10 01:29:17 +00:00
feat(cli): Add channel worker settings reload for serve --channel (#6598)
The daemon-managed channel worker reads each channel's settings (tokens, proxy, per-channel model) once when it starts, so applying settings.json changes previously required restarting the whole daemon. This adds an explicit reload that stops and relaunches the worker so it re-reads settings.json, without bouncing the daemon or its live sessions. The reload is exposed as a strict-gated POST /workspace/channel/reload route, an SDK reloadChannelWorker() method, and a qwen channel reload CLI command, advertised through a channel_reload capability only when the daemon was started with --channel. The worker supervisor gains a restart() that coalesces concurrent reloads onto a single relaunch, resets the crash-restart budget so a failed worker recovers, and latches a disposed flag on hard shutdown so a racing reload cannot relaunch a worker into a tearing-down daemon. Refs #5976
This commit is contained in:
jinye
GitHub
parent
53243de0c0
commit
fd613eae56
19 changed files with 868 additions and 24 deletions
|
|
@ -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 <names>` 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`
|
||||
|
|
|
|||
|
|
@ -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 <name>` 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 <name>` 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 <names>` 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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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`,
|
||||
|
|
|
|||
|
|
@ -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.')
|
||||
|
|
|
|||
115
packages/cli/src/commands/channel/reload.test.ts
Normal file
115
packages/cli/src/commands/channel/reload.test.ts
Normal file
|
|
@ -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<typeof reloadCommand.handler>;
|
||||
|
||||
async function runHandler(argv: Record<string, unknown>): Promise<void> {
|
||||
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);
|
||||
});
|
||||
});
|
||||
123
packages/cli/src/commands/channel/reload.ts
Normal file
123
packages/cli/src/commands/channel/reload.ts
Normal file
|
|
@ -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<ChannelReloadResultLike>;
|
||||
}
|
||||
|
||||
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<unknown, ReloadArgs> = {
|
||||
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);
|
||||
}
|
||||
},
|
||||
};
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -68,6 +68,12 @@ export interface ChannelWorkerSnapshot {
|
|||
export interface ChannelWorkerSupervisor {
|
||||
start(): Promise<void>;
|
||||
stop(): Promise<void>;
|
||||
/**
|
||||
* 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<ChannelWorkerSnapshot>;
|
||||
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<ChannelWorkerSnapshot> | 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;
|
||||
}
|
||||
|
|
|
|||
58
packages/cli/src/serve/routes/workspace-channel-control.ts
Normal file
58
packages/cli/src/serve/routes/workspace-channel-control.ts
Normal file
|
|
@ -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<ChannelWorkerSnapshot>;
|
||||
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 <name>`.',
|
||||
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',
|
||||
});
|
||||
}
|
||||
},
|
||||
);
|
||||
}
|
||||
|
|
@ -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),
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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: {
|
||||
|
|
|
|||
|
|
@ -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 =>
|
||||
|
|
|
|||
|
|
@ -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<ChannelWorkerSnapshot>;
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<DaemonChannelReloadResult> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -328,6 +328,8 @@ export type {
|
|||
DaemonMcpManageResult,
|
||||
DaemonMcpRestartResult,
|
||||
DaemonReloadResponse,
|
||||
DaemonChannelReloadResult,
|
||||
DaemonChannelWorkerSnapshot,
|
||||
DaemonRewindResult,
|
||||
DaemonRewindSnapshotInfo,
|
||||
DaemonSessionBtwResult,
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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(() =>
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue