feat(cli): Harden daemon-managed channel worker

docs/developers/qwen-serve-protocol.md

@@ -303,10 +303,11 @@ warning severity, otherwise `ok`. Issue codes are stable and include
 `session_capacity_high`, `connection_capacity_high`, `pending_permissions`,
 `acp_channel_down`, `preflight_error`, `mcp_budget_warning`,
 `mcp_budget_exhausted`, `rate_limit_hits`, `channel_worker_exited`, and
-`workspace_status_unavailable`. During the short window after the listener is
-ready but before the full runtime is mounted, `/daemon/status` may report
-`daemon_runtime_starting`; if the async runtime mount fails, it reports
-`daemon_runtime_failed` while non-status runtime routes return `503`.
+`channel_worker_partial_connect`, and `workspace_status_unavailable`. During
+the short window after the listener is ready but before the full runtime is
+mounted, `/daemon/status` may report `daemon_runtime_starting`; if the async
+runtime mount fails, it reports `daemon_runtime_failed` while non-status
+runtime routes return `503`.
 
 `runtime.channel.live` reports the ACP bridge channel inside the daemon. It is
 not the channel-adapter worker. Daemon-managed channels use
@@ -315,6 +316,29 @@ not the channel-adapter worker. Daemon-managed channels use
 and then exits, `/daemon/status` keeps the daemon online and reports warning
 issue code `channel_worker_exited`.
 
+Daemon-managed channel worker startup remains fail-fast: if `qwen serve
+--channel ...` cannot start a worker that reaches ready, serve startup fails.
+After a worker has reached ready, unexpected exits are restarted by the serve
+supervisor within a bounded policy: up to 3 restart attempts in a 5 minute
+window, with 1s, 5s, then 15s backoff. The worker sends IPC heartbeats every
+15s; if no heartbeat is observed for 45s, the supervisor treats the worker as
+stale, kills it, records `staleHeartbeatAt`, and uses the same restart path.
+
+`runtime.channelWorker` may include additive operational fields:
+`requestedChannels`, `pid`, `startedAt`, `exitCode`, `signal`, `error`,
+`restartCount`, `lastExitAt`, `lastRestartAt`, `nextRestartAt`,
+`lastHeartbeatAt`, and `staleHeartbeatAt`. `restartCount` is the lifetime
+number of restart attempts made by this serve process; a running worker with
+`restartCount > 0` is healthy unless another issue applies. A running worker
+whose `requestedChannels` include names missing from `channels` reports
+`channel_worker_partial_connect`.
+
+`qwen channel status` continues to read pidfile metadata. During a restart
+window the serve-owned pidfile remains reserved, but `workerPid` is omitted so
+clients do not display a stale worker process. Worker stdout/stderr are
+forwarded into the daemon log with bearer tokens, sensitive worker environment
+values, and proxy URL credentials redacted.
+
 Security: the response never includes bearer tokens, client ids, full ACP
 connection ids, device-flow user codes, or verification URLs. `summary` omits
 the daemon log path; `full` may include it for authenticated operators.

packages/channels/base/src/sanitize.test.ts

@@ -19,8 +19,14 @@ const CSI = String.fromCharCode(0x009b); // CONTROL SEQUENCE INTRODUCER (another
 const ZWSP = String.fromCharCode(0x200b); // ZERO WIDTH SPACE
 const ZWNJ = String.fromCharCode(0x200c); // ZERO WIDTH NON-JOINER
 const ZWJ = String.fromCharCode(0x200d); // ZERO WIDTH JOINER
+const LRM = String.fromCharCode(0x200e); // LEFT-TO-RIGHT MARK
+const RLM = String.fromCharCode(0x200f); // RIGHT-TO-LEFT MARK
+const SHY = String.fromCharCode(0x00ad); // SOFT HYPHEN
+const ALM = String.fromCharCode(0x061c); // ARABIC LETTER MARK
+const INVISIBLE_PLUS = String.fromCharCode(0x2064); // INVISIBLE PLUS
 const WJ = String.fromCharCode(0x2060); // WORD JOINER
 const BOM = String.fromCharCode(0xfeff); // ZERO WIDTH NO-BREAK SPACE / BOM
+const VS16 = String.fromCharCode(0xfe0f); // VARIATION SELECTOR-16
 // U+1F389 PARTY POPPER as its UTF-16 surrogate pair, kept ASCII in source. A
 // length cap landing between the two units yields a lone surrogate (-> `replace`
 // char downstream); the sanitizers truncate on code-point boundaries to avoid it.
@@ -256,6 +262,19 @@ describe('sanitizeLogText', () => {
     expect(out).not.toContain(PDI);
   });
 
+  it('neutralizes format characters that can visually hide inside text', () => {
+    const out = sanitizeLogText(
+      `a${SHY}${ALM}${LRM}${RLM}${INVISIBLE_PLUS}${VS16}b`,
+      80,
+    );
+    expect(out).not.toContain(SHY);
+    expect(out).not.toContain(ALM);
+    expect(out).not.toContain(LRM);
+    expect(out).not.toContain(RLM);
+    expect(out).not.toContain(INVISIBLE_PLUS);
+    expect(out).not.toContain(VS16);
+  });
+
   it('caps to maxLen code points without splitting a surrogate pair', () => {
     // Cap by code point so an emoji at the boundary is never cut mid-pair.
     const out = sanitizeLogText(EMOJI.repeat(100), 5);

packages/channels/base/src/sanitize.ts

@@ -11,7 +11,7 @@
  * are stripped by each caller.
  */
 export const PROMPT_UNSAFE_INVISIBLES =
-  /[\u0080-\u009f\u200b-\u200d\u2028\u2029\u202a-\u202e\u2060\u2066-\u2069\ufeff]/g;
+  /[\u0080-\u009f\p{Cf}\u2028\u2029]|\p{Variation_Selector}/gu;
 
 /**
  * Truncate to at most `max` Unicode CODE POINTS (not UTF-16 code units). A cap

packages/cli/src/commands/channel/daemon-worker.test.ts

@@ -940,6 +940,152 @@ describe('daemonWorkerCommand', () => {
     }
   });
 
+  it('sends heartbeat messages while the daemon worker is live', async () => {
+    vi.useFakeTimers();
+    const exit = mockProcessExitNoThrow();
+    const send = vi.fn();
+    const restoreSend = stubProcessSend(send as NodeJS.Process['send']);
+    vi.stubEnv('QWEN_CHANNEL_DAEMON_WORKER', 'worker-token');
+    vi.stubEnv('QWEN_DAEMON_TOKEN', 'daemon-token');
+    vi.stubEnv('QWEN_DAEMON_URL', 'http://127.0.0.1:4170');
+    vi.stubEnv('QWEN_DAEMON_WORKSPACE', '/workspace');
+
+    try {
+      const handler = daemonWorkerCommand.handler({
+        channel: ['telegram'],
+        _: [],
+        $0: 'qwen',
+      });
+      await vi.waitFor(() => {
+        expect(send).toHaveBeenCalledWith(
+          expect.objectContaining({
+            type: 'ready',
+            channels: ['telegram'],
+            requestedChannels: ['telegram'],
+            pid: process.pid,
+          }),
+        );
+      });
+      send.mockClear();
+
+      await vi.advanceTimersByTimeAsync(15_000);
+
+      expect(send).toHaveBeenCalledWith(
+        expect.objectContaining({ type: 'heartbeat', pid: process.pid }),
+      );
+
+      process.emit('SIGTERM', 'SIGTERM');
+      await handler;
+      expect(exit).toHaveBeenCalledWith(0);
+
+      send.mockClear();
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(send).not.toHaveBeenCalledWith(
+        expect.objectContaining({ type: 'heartbeat' }),
+      );
+    } finally {
+      restoreSend();
+      vi.useRealTimers();
+    }
+  });
+
+  it('clears heartbeat messages when the IPC send channel is closed', async () => {
+    vi.useFakeTimers();
+    const exit = mockProcessExitNoThrow();
+    const send = vi.fn();
+    const restoreSend = stubProcessSend(send as NodeJS.Process['send']);
+    vi.stubEnv('QWEN_CHANNEL_DAEMON_WORKER', 'worker-token');
+    vi.stubEnv('QWEN_DAEMON_URL', 'http://127.0.0.1:4170');
+    vi.stubEnv('QWEN_DAEMON_WORKSPACE', '/workspace');
+
+    try {
+      const handler = daemonWorkerCommand.handler({
+        channel: ['telegram'],
+        _: [],
+        $0: 'qwen',
+      });
+      await vi.waitFor(() => {
+        expect(send).toHaveBeenCalledWith(
+          expect.objectContaining({
+            type: 'ready',
+            channels: ['telegram'],
+            requestedChannels: ['telegram'],
+            pid: process.pid,
+          }),
+        );
+      });
+      send.mockClear();
+      send.mockImplementation(() => {
+        throw Object.assign(new Error('Channel closed'), {
+          code: 'ERR_IPC_CHANNEL_CLOSED',
+        });
+      });
+
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(send).toHaveBeenCalledWith(
+        expect.objectContaining({ type: 'heartbeat' }),
+      );
+
+      send.mockClear();
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(send).not.toHaveBeenCalled();
+
+      process.emit('SIGTERM', 'SIGTERM');
+      await handler;
+      expect(exit).toHaveBeenCalledWith(0);
+    } finally {
+      restoreSend();
+      vi.useRealTimers();
+    }
+  });
+
+  it('clears heartbeat messages when parent IPC disconnects', async () => {
+    vi.useFakeTimers();
+    const exit = mockProcessExitNoThrow();
+    const send = vi.fn();
+    const restoreSend = stubProcessSend(send as NodeJS.Process['send']);
+    vi.stubEnv('QWEN_CHANNEL_DAEMON_WORKER', 'worker-token');
+    vi.stubEnv('QWEN_DAEMON_URL', 'http://127.0.0.1:4170');
+    vi.stubEnv('QWEN_DAEMON_WORKSPACE', '/workspace');
+
+    try {
+      const handler = daemonWorkerCommand.handler({
+        channel: ['telegram'],
+        _: [],
+        $0: 'qwen',
+      });
+      await vi.waitFor(() => {
+        expect(send).toHaveBeenCalledWith(
+          expect.objectContaining({
+            type: 'ready',
+            channels: ['telegram'],
+            requestedChannels: ['telegram'],
+            pid: process.pid,
+          }),
+        );
+      });
+
+      process.emit('disconnect');
+      send.mockClear();
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(send).not.toHaveBeenCalledWith(
+        expect.objectContaining({ type: 'heartbeat' }),
+      );
+
+      await handler;
+      expect(exit).toHaveBeenCalledWith(0);
+
+      send.mockClear();
+      await vi.advanceTimersByTimeAsync(15_000);
+      expect(send).not.toHaveBeenCalledWith(
+        expect.objectContaining({ type: 'heartbeat' }),
+      );
+    } finally {
+      restoreSend();
+      vi.useRealTimers();
+    }
+  });
+
   it('honors a shutdown signal received during async setup', async () => {
     const exit = mockProcessExitNoThrow();
     const restoreSend = stubProcessSend(vi.fn() as NodeJS.Process['send']);

packages/cli/src/commands/channel/daemon-worker.ts

@@ -17,6 +17,7 @@ import type { ServeChannelSelection } from '../../serve/types.js';
 import { normalizeServeChannelSelection } from '../../serve/channel-selection.js';
 import {
   CHANNEL_DAEMON_WORKER_SENTINEL,
+  CHANNEL_WORKER_HEARTBEAT_INTERVAL_MS,
   QWEN_DAEMON_TOKEN_ENV,
   QWEN_DAEMON_URL_ENV,
   QWEN_DAEMON_WORKSPACE_ENV,
@@ -502,6 +503,25 @@ export const daemonWorkerCommand: CommandModule<unknown, DaemonWorkerArgs> = {
       });
       removeEarlyShutdownHandlers();
 
+      let heartbeatTimer: NodeJS.Timeout | undefined;
+      const clearHeartbeat = () => {
+        if (!heartbeatTimer) return;
+        clearInterval(heartbeatTimer);
+        heartbeatTimer = undefined;
+      };
+      heartbeatTimer = setInterval(() => {
+        try {
+          process.send?.({
+            type: 'heartbeat',
+            pid: process.pid,
+            at: new Date().toISOString(),
+          });
+        } catch {
+          clearHeartbeat();
+        }
+      }, CHANNEL_WORKER_HEARTBEAT_INTERVAL_MS);
+      heartbeatTimer.unref();
+
       let shuttingDown = false;
       let exitCode = 0;
       let finish!: () => void;
@@ -513,6 +533,7 @@ export const daemonWorkerCommand: CommandModule<unknown, DaemonWorkerArgs> = {
           process.exit(1);
         } else {
           shuttingDown = true;
+          clearHeartbeat();
           try {
             await handle.close();
           } catch (err) {
@@ -526,6 +547,7 @@ export const daemonWorkerCommand: CommandModule<unknown, DaemonWorkerArgs> = {
               `[Channel] daemon worker failed to shut down after ${safeReason}: ${safeMessage}`,
             );
           } finally {
+            clearHeartbeat();
             finish();
           }
         }
@@ -540,6 +562,7 @@ export const daemonWorkerCommand: CommandModule<unknown, DaemonWorkerArgs> = {
         void shutdown(pendingShutdownReason);
       }
       await finished;
+      clearHeartbeat();
       process.removeListener('SIGINT', shutdown);
       process.removeListener('SIGTERM', shutdown);
       process.removeListener('disconnect', onDisconnect);

packages/cli/src/commands/channel/pidfile.test.ts

@@ -119,6 +119,7 @@ beforeEach(() => {
 });
 
 afterEach(() => {
+  vi.useRealTimers();
   process.kill = originalKill;
 });
 
@@ -201,6 +202,33 @@ describe('writeServiceInfo + readServiceInfo', () => {
     expect(fsFds.openedFlags).toContain(2 | 0x20000);
   });
 
+  it('preserves the serve reservation start time when worker metadata changes', () => {
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date('2026-07-01T01:00:00.000Z'));
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    process.kill = vi.fn(() => true) as any;
+
+    reserveServeServiceInfo({
+      channels: ['telegram'],
+      servePid: 4321,
+    });
+    vi.setSystemTime(new Date('2026-07-01T01:05:00.000Z'));
+    writeServeServiceInfo({
+      channels: ['telegram'],
+      servePid: 4321,
+      workerPid: 8765,
+    });
+
+    expect(readServiceInfo()).toMatchObject({
+      owner: 'serve',
+      pid: 4321,
+      servePid: 4321,
+      workerPid: 8765,
+      channels: ['telegram'],
+      startedAt: '2026-07-01T01:00:00.000Z',
+    });
+  });
+
   it('does not let serve metadata updates overwrite standalone pidfiles', () => {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     process.kill = vi.fn(() => true) as any;

packages/cli/src/commands/channel/pidfile.ts

@@ -156,22 +156,22 @@ export function writeServeServiceInfo({
   servePid?: number;
   workerPid?: number;
 }): void {
-  const info: ServiceInfo = {
+  const buildInfo = (startedAt: string): ServiceInfo => ({
     owner: 'serve',
     pid: servePid,
-    startedAt: new Date().toISOString(),
+    startedAt,
     channels,
     servePid,
     ...(workerPid !== undefined ? { workerPid } : {}),
-  };
+  });
 
   const filePath = pidFilePath();
   let fd: number;
   try {
     fd = openSync(filePath, constants.O_RDWR | constants.O_NOFOLLOW);
   } catch (err) {
     if ((err as NodeJS.ErrnoException).code === 'ENOENT') {
-      writeInfo(info, 'wx');
+      writeInfo(buildInfo(new Date().toISOString()), 'wx');
       return;
     }
     throw err;
@@ -195,6 +195,7 @@ export function writeServeServiceInfo({
         'Channel service pidfile is owned by another process.',
       );
     }
+    const info = buildInfo(existing.startedAt);
     ftruncateSync(fd, 0);
     writeSync(fd, JSON.stringify(info, null, 2), 0, 'utf-8');
   } finally {