fix(ccm): reap Codex quota collector process trees (#215)

* fix(ccm): reap Codex quota collector process trees

* chore(ccm): record monitor reap changeset

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: PanQiWei

CHANGELOG.md

@@ -32,6 +32,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   count; at 46 modules it had roughly 11 tokens of headroom and any new module
   overflowed it.
 
+### Fixed
+
+- `ccm monitor` machine-wide Codex quota polling now owns the app-server process
+  group and waits for launcher close plus full tree disappearance before
+  publishing a result. Timeout and error paths escalate from `SIGTERM` to
+  `SIGKILL`, preventing one unreaped `MainThread` zombie from accumulating per
+  monitor tick.
+
 ## [0.22.0] — 2026-07-23
 
 > **Noncommercial license boundary release** — the first cc-master plugin

ccm/.changeset/reap-codex-quota-processes.md

@@ -0,0 +1,5 @@
+---
+'ccm': patch
+---
+
+修复 `ccm monitor` machine-wide Codex quota 轮询的进程回收竞态：collector 现在拥有独立的 app-server 进程组，超时或错误时按 `SIGTERM` → `SIGKILL` 升级清理，并等待 launcher close 与完整进程树消失后才发布结果，避免每轮监控累积一个未回收的 `MainThread` 僵尸进程。

ccm/apps/cli/src/codex-rate-limits.ts

@@ -8,6 +8,12 @@ import { MessageChannel, receiveMessageOnPort, Worker } from 'node:worker_thread
 import type { UsagePoolSignal, UsageSignal, WindowSignal } from '@ccm/engine';
 
 const DEFAULT_TIMEOUT_MS = 10_000;
+const WORKER_STARTUP_ALLOWANCE_MS = 1_000;
+const TERMINATION_GRACE_MS = 100;
+// Caller-side observation budget after escalation. A timed-out Worker remains unref'ed so it can
+// still own and reap the child instead of recreating the zombie leak this boundary prevents.
+const REAP_TIMEOUT_MS = 1_000;
+const REAP_POLL_MS = 5;
 
 interface RateLimitWindow {
   usedPercent?: unknown;
@@ -41,29 +47,45 @@ export function readCodexUsageSignal(
   const flag = new Int32Array(sab);
   const { port1, port2 } = new MessageChannel();
   let worker: Worker | null = null;
+  let workerChildReaped = false;
   try {
     worker = new Worker(WORKER_SOURCE, {
       eval: true,
       workerData: {
         codexBin,
         env,
         timeoutMs,
+        terminationGraceMs: TERMINATION_GRACE_MS,
+        reapPollMs: REAP_POLL_MS,
         sab,
         port: port2,
       },
       transferList: [port2],
     });
-    Atomics.wait(flag, 0, 0, timeoutMs + 1000);
+    const waitResult = Atomics.wait(
+      flag,
+      0,
+      0,
+      timeoutMs + WORKER_STARTUP_ALLOWANCE_MS + TERMINATION_GRACE_MS + REAP_TIMEOUT_MS,
+    );
+    if (waitResult === 'timed-out') return null;
     const msg = receiveMessageOnPort(port1)?.message as
-      | { ok?: boolean; result?: unknown }
+      | { ok?: boolean; reaped?: boolean; result?: unknown }
       | undefined;
+    if (msg?.reaped !== true) return null;
+    workerChildReaped = true;
     if (!msg?.ok) return null;
     return normalizeCodexRateLimits(msg.result);
   } catch {
     return null;
   } finally {
     try {
-      worker?.terminate();
+      if (workerChildReaped) {
+        void worker?.terminate();
+      } else {
+        // terminate() before child `close` discards libuv's wait/reap handle. Let cleanup finish.
+        worker?.unref();
+      }
     } catch {
       /* ignore */
     }
@@ -78,31 +100,135 @@ const { workerData } = await import('node:worker_threads');
 
 const flag = new Int32Array(workerData.sab);
 const port = workerData.port;
-let done = false;
+const ownsProcessGroup = process.platform !== 'win32';
+let child = null;
+let terminalPayload = null;
+let cleanupStarted = false;
+let launcherClosed = false;
+let treeGone = false;
+let published = false;
 let buffer = '';
+let responseTimer = null;
+let terminationTimer = null;
+let reapPollTimer = null;
 
-function finish(payload) {
-  if (done) return;
-  done = true;
-  try { port.postMessage(payload); } catch {}
+// Publishing the RPC payload and reaping the owned process tree are separate phases. The parent
+// receives no terminal message until both launcher close and complete tree disappearance hold.
+function clearTimer(timer) {
+  if (timer) clearTimeout(timer);
+}
+
+function processMissing(error) {
+  return error && typeof error === 'object' && error.code === 'ESRCH';
+}
+
+function observeTreeGone() {
+  if (treeGone) return true;
+  if (!child || !Number.isSafeInteger(child.pid) || child.pid <= 0) {
+    treeGone = launcherClosed;
+    return treeGone;
+  }
+  if (!ownsProcessGroup) {
+    treeGone = launcherClosed;
+    return treeGone;
+  }
+  try {
+    process.kill(-child.pid, 0);
+    return false;
+  } catch (error) {
+    if (processMissing(error)) treeGone = true;
+    return treeGone;
+  }
+}
+
+function signalOwnedTree(signal) {
+  if (!child || !Number.isSafeInteger(child.pid) || child.pid <= 0) {
+    treeGone = launcherClosed;
+    return false;
+  }
+  try {
+    if (ownsProcessGroup) {
+      process.kill(-child.pid, signal);
+      return true;
+    }
+    const signaled = child.kill(signal);
+    if (!signaled && launcherClosed) treeGone = true;
+    return signaled;
+  } catch (error) {
+    if (processMissing(error)) treeGone = true;
+    return false;
+  }
+}
+
+function publishReaped() {
+  if (published || !terminalPayload) return;
+  published = true;
+  clearTimer(responseTimer);
+  clearTimer(terminationTimer);
+  clearTimer(reapPollTimer);
+  try { port.postMessage({ ...terminalPayload, reaped: true }); } catch {}
   Atomics.store(flag, 0, 1);
   Atomics.notify(flag, 0);
-  try { child.stdin.end(); } catch {}
-  try { child.kill(); } catch {}
+  try { port.close(); } catch {}
+}
+
+function maybePublishReaped() {
+  if (!cleanupStarted || !launcherClosed || !observeTreeGone()) return false;
+  publishReaped();
+  return true;
+}
+
+function pollForReap() {
+  if (published || maybePublishReaped()) return;
+  clearTimer(reapPollTimer);
+  reapPollTimer = setTimeout(pollForReap, workerData.reapPollMs);
+}
+
+function requestFinish(payload) {
+  if (cleanupStarted) return;
+  cleanupStarted = true;
+  terminalPayload = payload;
+  clearTimer(responseTimer);
+  responseTimer = null;
+  try { child?.stdin.end(); } catch {}
+  signalOwnedTree('SIGTERM');
+  terminationTimer = setTimeout(() => {
+    if (published || maybePublishReaped()) return;
+    signalOwnedTree('SIGKILL');
+    maybePublishReaped();
+  }, workerData.terminationGraceMs);
+  pollForReap();
 }
 
 function write(msg) {
-  try { child.stdin.write(JSON.stringify(msg) + '\\n'); } catch { finish({ ok: false }); }
+  try { child.stdin.write(JSON.stringify(msg) + '\\n'); } catch { requestFinish({ ok: false }); }
 }
 
-const child = spawn(workerData.codexBin, ['app-server', '--stdio'], {
-  env: { ...process.env, ...workerData.env },
-  stdio: ['pipe', 'pipe', 'ignore'],
-});
+try {
+  child = spawn(workerData.codexBin, ['app-server', '--stdio'], {
+    detached: ownsProcessGroup,
+    env: { ...process.env, ...workerData.env },
+    stdio: ['pipe', 'pipe', 'ignore'],
+  });
+} catch {
+  launcherClosed = true;
+  treeGone = true;
+  requestFinish({ ok: false });
+}
 
-const timer = setTimeout(() => finish({ ok: false }), workerData.timeoutMs);
-child.once('error', () => finish({ ok: false }));
-child.once('exit', () => finish({ ok: false }));
+if (!child) {
+  maybePublishReaped();
+  return;
+}
+
+responseTimer = setTimeout(() => requestFinish({ ok: false }), workerData.timeoutMs);
+child.once('error', () => requestFinish({ ok: false }));
+child.once('exit', () => requestFinish({ ok: false }));
+child.once('close', () => {
+  launcherClosed = true;
+  if (!cleanupStarted) requestFinish({ ok: false });
+  maybePublishReaped();
+});
 child.stdout.on('data', (chunk) => {
   buffer += chunk.toString('utf8');
   let nl = buffer.indexOf('\\n');
@@ -123,8 +249,7 @@ function handleLine(line) {
     return;
   }
   if (msg.id === 6) {
-    clearTimeout(timer);
-    finish(msg.error ? { ok: false } : { ok: true, result: msg.result });
+    requestFinish(msg.error ? { ok: false } : { ok: true, result: msg.result });
   }
 }