fix(daemon): 修复 PM2 God 的 systemd cgroup 归属

src/cli/pm2-existing-client.ts

@@ -0,0 +1,116 @@
+#!/usr/bin/env node
+/** Execute a bounded PM2 mutation only when its RPC daemon already exists. */
+import { createRequire } from 'node:module';
+import {
+  inspectLinuxPm2GodOwnership,
+  revalidateLinuxPm2GodProcess,
+  type LinuxPm2GodProcess,
+} from '../core/pm2-lifecycle-owner.js';
+
+const require = createRequire(import.meta.url);
+const pm2 = require('pm2') as any;
+const ABSENT_EXIT_CODE = 4;
+const CONNECT_TIMEOUT_MS = 5_000;
+const MUTATION_TIMEOUT_MS = 25_000;
+
+type Mutation =
+  | { operation: 'start'; target: string; only?: string; updateEnv?: boolean }
+  | { operation: 'restart'; target: string; updateEnv?: boolean }
+  | { operation: 'stop' | 'delete'; target: string };
+type Request = Mutation & { expectedGod: LinuxPm2GodProcess };
+
+function fail(message: string, code = 1): never {
+  console.error(message);
+  process.exit(code);
+}
+
+let request: Request;
+try {
+  request = JSON.parse(process.argv[2] ?? '') as Request;
+} catch (error) {
+  fail(`invalid PM2 existing-daemon request: ${error instanceof Error ? error.message : String(error)}`);
+}
+
+const timer = setTimeout(() => fail('PM2 existing-daemon RPC connection timed out'), CONNECT_TIMEOUT_MS);
+timer.unref();
+let mutationTimer: NodeJS.Timeout | undefined;
+let mutationSettled = false;
+
+function complete(error?: Error | null): void {
+  mutationSettled = true;
+  if (mutationTimer) clearTimeout(mutationTimer);
+  if (error) fail(`PM2 existing-daemon ${request.operation} failed: ${error.message}`);
+  pm2.disconnect((disconnectError?: Error | null) => {
+    if (disconnectError) fail(`PM2 existing-daemon disconnect failed: ${disconnectError.message}`);
+    process.exit(0);
+  });
+}
+
+pm2.Client.pingDaemon((alive: boolean) => {
+  if (!alive) {
+    clearTimeout(timer);
+    fail('PM2 God disappeared before mutation; refusing to daemonize from this caller', ABSENT_EXIT_CODE);
+  }
+  pm2.Client.launchRPC((connectError: Error | null | undefined) => {
+    if (connectError) fail(`PM2 existing-daemon RPC connection failed: ${connectError.message}`);
+    clearTimeout(timer);
+    const rpcSocket = pm2.Client.client?.sock;
+    if (!rpcSocket?.set) fail('PM2 existing-daemon RPC socket cannot disable reconnect');
+    // pm2-axon otherwise queues an in-flight request and reconnects to a new
+    // rpc.sock owner after 100ms. A generation check cannot bind the eventual
+    // recipient unless reconnection is disabled before the check and send.
+    rpcSocket.set('retry timeout', 0);
+    rpcSocket.set('retry max timeout', 0);
+    rpcSocket.retry = 0;
+    rpcSocket.once('close', () => {
+      if (!mutationSettled) fail(`PM2 God disconnected before ${request.operation} completed`);
+    });
+    // launchRPC pins this client to one socket generation. Revalidate the exact
+    // PID/birth/cgroup only after that connection exists: a God replaced after
+    // the parent's ownership check can no longer receive a mutation first.
+    const home = process.env.PM2_HOME ?? '';
+    const ownership = inspectLinuxPm2GodOwnership(home);
+    const current = ownership.kind === 'absent' || ownership.processes.length !== 1
+      ? undefined
+      : ownership.processes[0];
+    if (!request.expectedGod?.startIdentity
+        || !current
+        || current.pid !== request.expectedGod.pid
+        || current.startIdentity !== request.expectedGod.startIdentity
+        || current.cgroup !== request.expectedGod.cgroup
+        || !revalidateLinuxPm2GodProcess(request.expectedGod, home)) {
+      mutationSettled = true;
+      pm2.disconnect(() => fail(
+        `PM2 God generation changed before mutation (expected pid ${request.expectedGod?.pid ?? 'unknown'})`,
+        ABSENT_EXIT_CODE,
+      ));
+      return;
+    }
+    // Keep the helper alive until PM2 acknowledges the mutation. With Axon
+    // reconnect disabled, a closed socket must never let an unacknowledged
+    // queued request fall through Node's beforeExit path with status 0.
+    mutationTimer = setTimeout(
+      () => fail(`PM2 existing-daemon ${request.operation} response timed out`),
+      MUTATION_TIMEOUT_MS,
+    );
+    switch (request.operation) {
+      case 'start':
+        pm2.start(request.target, {
+          ...(request.only ? { only: request.only } : {}),
+          ...(request.updateEnv ? { updateEnv: true } : {}),
+        }, complete);
+        return;
+      case 'restart':
+        pm2.restart(request.target, { updateEnv: request.updateEnv === true }, complete);
+        return;
+      case 'stop':
+        pm2.stop(request.target, complete);
+        return;
+      case 'delete':
+        pm2.delete(request.target, complete);
+        return;
+      default:
+        fail('unsupported PM2 existing-daemon mutation');
+    }
+  });
+});

src/cli/pm2-existing.ts

@@ -0,0 +1,85 @@
+import { spawnSync } from 'node:child_process';
+import { existsSync } from 'node:fs';
+import { join } from 'node:path';
+import type { LinuxPm2GodProcess } from '../core/pm2-lifecycle-owner.js';
+
+const ABSENT_EXIT_CODE = 4;
+
+type ExistingPm2Mutation =
+  | { operation: 'start'; target: string; only?: string; updateEnv?: boolean }
+  | { operation: 'restart'; target: string; updateEnv?: boolean }
+  | { operation: 'stop' | 'delete'; target: string };
+
+type ExistingPm2Request = ExistingPm2Mutation & { expectedGod: LinuxPm2GodProcess };
+
+function mutationFromArgs(args: string[]): ExistingPm2Mutation {
+  const [operation, target] = args;
+  if (!target) throw new Error(`unsupported PM2 mutation without target: ${args.join(' ')}`);
+  if (operation === 'stop' || operation === 'delete') {
+    if (args.length !== 2) throw new Error(`unsupported PM2 mutation: ${args.join(' ')}`);
+    return { operation, target };
+  }
+  if (operation !== 'start') throw new Error(`unsupported PM2 mutation: ${args.join(' ')}`);
+  const onlyIndex = args.indexOf('--only');
+  const only = onlyIndex >= 0 ? args[onlyIndex + 1] : undefined;
+  const updateEnv = args.includes('--update-env');
+  const expectedLength = 2 + (only ? 2 : 0) + (updateEnv ? 1 : 0);
+  if (args.length !== expectedLength || (onlyIndex >= 0 && !only)) {
+    throw new Error(`unsupported PM2 start mutation: ${args.join(' ')}`);
+  }
+  if (updateEnv && !only) return { operation: 'restart', target, updateEnv: true };
+  return {
+    operation: 'start',
+    target,
+    ...(only ? { only } : {}),
+    ...(updateEnv ? { updateEnv: true } : {}),
+  };
+}
+
+function helperArgs(pkgRoot: string, request: ExistingPm2Request): string[] {
+  const built = join(pkgRoot, 'dist', 'cli', 'pm2-existing-client.js');
+  const payload = JSON.stringify(request);
+  // Source/test execution must not silently select a stale dist helper with an
+  // older request contract. Installed production code runs from dist and uses
+  // the sibling built helper.
+  if (import.meta.url.includes('/dist/cli/pm2-existing.js') && existsSync(built)) {
+    return [built, payload];
+  }
+  return ['--import', 'tsx', join(pkgRoot, 'src', 'cli', 'pm2-existing-client.ts'), payload];
+}
+
+/** Mutate an attested existing God without PM2's public connect/daemonize path. */
+export function runExistingPm2Command(input: {
+  pkgRoot: string;
+  home: string;
+  args: string[];
+  inherit?: boolean;
+  timeoutMs?: number;
+  env?: NodeJS.ProcessEnv;
+  nodePath?: string;
+  expectedGod: LinuxPm2GodProcess;
+}): void {
+  if (!input.expectedGod.startIdentity) {
+    throw new Error(`PM2 God pid ${input.expectedGod.pid} has no process-birth identity`);
+  }
+  const request = { ...mutationFromArgs(input.args), expectedGod: input.expectedGod };
+  const result = spawnSync(
+    input.nodePath ?? process.execPath,
+    helperArgs(input.pkgRoot, request),
+    {
+      stdio: input.inherit === false ? 'pipe' : 'inherit',
+      env: {
+        ...(input.env ?? process.env),
+        PM2_HOME: input.home,
+      },
+      timeout: input.timeoutMs ?? 30_000,
+    },
+  );
+  if (result.status === 0) return;
+  const stderr = String(result.stderr ?? '').trim();
+  const detail = result.error?.message ?? (stderr || `status ${result.status}`);
+  if (result.status === ABSENT_EXIT_CODE) {
+    throw new Error(`PM2 God disappeared before mutation; no replacement daemon was created: ${detail}`);
+  }
+  throw new Error(`PM2 existing-daemon ${input.args.join(' ')} failed: ${detail}`);
+}

src/cli/pm2-readonly-client.ts

@@ -0,0 +1,111 @@
+#!/usr/bin/env node
+/**
+ * Side-effect-free PM2 observer.
+ *
+ * PM2's public `connect`/CLI path daemonizes when the RPC socket disappears.
+ * This helper deliberately calls pingDaemon + launchRPC directly and never
+ * calls Client.start/connect, so a status/logs race cannot create a new God in
+ * the observer's cgroup.
+ */
+import { createRequire } from 'node:module';
+import {
+  inspectLinuxPm2GodOwnership,
+  revalidateLinuxPm2GodProcess,
+  type LinuxPm2GodProcess,
+} from '../core/pm2-lifecycle-owner.js';
+
+const require = createRequire(import.meta.url);
+const pm2 = require('pm2') as any;
+const mode = process.argv[2];
+const target = process.argv[3] || 'all';
+const lines = Number.parseInt(process.argv[4] || '50', 10);
+const CONNECT_TIMEOUT_MS = 5_000;
+const ABSENT_EXIT_CODE = 3;
+let expectedGod: LinuxPm2GodProcess | undefined;
+try {
+  expectedGod = process.env.BOTMUX_PM2_EXPECTED_GOD
+    ? JSON.parse(process.env.BOTMUX_PM2_EXPECTED_GOD) as LinuxPm2GodProcess
+    : undefined;
+} catch (error) {
+  fail(`invalid PM2 read-only generation binding: ${error instanceof Error ? error.message : String(error)}`);
+}
+
+function fail(message: string, code = 1): never {
+  console.error(message);
+  process.exit(code);
+}
+
+function disableAxonReconnect(socket: any, label: string): void {
+  if (!socket?.set) fail(`PM2 read-only ${label} socket cannot disable reconnect`);
+  socket.set('retry timeout', 0);
+  socket.set('retry max timeout', 0);
+  // pm2-axon caches the current backoff separately after connect.
+  socket.retry = 0;
+}
+
+function revalidateExpectedGod(phase: string): void {
+  if (!expectedGod) return;
+  const home = process.env.PM2_HOME ?? '';
+  const ownership = inspectLinuxPm2GodOwnership(home);
+  const current = ownership.kind === 'absent' || ownership.processes.length !== 1
+    ? undefined
+    : ownership.processes[0];
+  if (!expectedGod.startIdentity
+      || !current
+      || current.pid !== expectedGod.pid
+      || current.startIdentity !== expectedGod.startIdentity
+      || current.cgroup !== expectedGod.cgroup
+      || !revalidateLinuxPm2GodProcess(expectedGod, home)) {
+    fail(`PM2 God generation changed before read-only ${phase}`);
+  }
+}
+
+const timer = setTimeout(() => {
+  fail('PM2 read-only RPC connection timed out');
+}, CONNECT_TIMEOUT_MS);
+timer.unref();
+
+pm2.Client.pingDaemon((alive: boolean) => {
+  if (!alive) {
+    clearTimeout(timer);
+    if (mode === 'jlist') process.exit(ABSENT_EXIT_CODE);
+    console.log(mode === 'logs' ? 'daemon 未在运行，暂无 PM2 日志。' : 'daemon 未在运行。');
+    process.exit(0);
+  }
+  pm2.Client.launchRPC((connectError: Error | null | undefined) => {
+    if (connectError) fail(`PM2 read-only RPC connection failed: ${connectError.message}`);
+    clearTimeout(timer);
+    const rpcSocket = pm2.Client.client?.sock;
+    disableAxonReconnect(rpcSocket, 'RPC');
+    revalidateExpectedGod(mode);
+    if (mode === 'jlist') {
+      pm2.list((error: Error | null | undefined, list: unknown[]) => {
+        if (error) fail(`PM2 read-only jlist failed: ${error.message}`);
+        process.stdout.write(JSON.stringify(Array.isArray(list) ? list : []));
+        pm2.disconnect(() => process.exit(0));
+      });
+      return;
+    }
+    if (mode === 'status') {
+      pm2.speedList(null);
+      return;
+    }
+    if (mode === 'logs') {
+      const launchBus = pm2.Client.launchBus.bind(pm2.Client);
+      pm2.Client.launchBus = (callback: (...args: any[]) => void) => {
+        launchBus((error: Error | null | undefined, bus: unknown, busSocket: any) => {
+          if (error) fail(`PM2 read-only bus connection failed: ${error.message}`);
+          disableAxonReconnect(busSocket, 'bus');
+          revalidateExpectedGod('logs bus connection');
+          // Closing this generation's publisher must end the observer. Never
+          // reconnect to a replacement God that later owns the same pathname.
+          busSocket.once('close', () => process.exit(0));
+          callback(error, bus, busSocket);
+        });
+      };
+      pm2.streamLogs(target, Number.isFinite(lines) ? lines : 50, false, undefined, false);
+      return;
+    }
+    fail(`unknown PM2 read-only mode: ${mode}`);
+  });
+});