Merge pull request #389 from pikasTech/fix/2792-child-reaping

修复 manager 外部命令子进程回收

src/mgr/abortable-child.ts

@@ -18,12 +18,15 @@ export function waitForChildProcess(child: ChildProcess, input: { signal?: Abort
   return new Promise<ChildProcessExit>((resolve, reject) => {
     let settled = false;
     let aborted = false;
+    let spawnError: Error | null = null;
+    let exit: ChildProcessExit | null = null;
     let killTimer: NodeJS.Timeout | null = null;
 
     const cleanup = (): void => {
       if (killTimer) clearTimeout(killTimer);
       input.signal?.removeEventListener("abort", onAbort);
       child.removeListener("error", onError);
+      child.removeListener("exit", onExit);
       child.removeListener("close", onClose);
     };
     const finish = (callback: () => void): void => {
@@ -42,13 +45,21 @@ export function waitForChildProcess(child: ChildProcess, input: { signal?: Abort
       killTimer.unref?.();
     };
     const onError = (error: Error): void => {
-      finish(() => reject(aborted ? abortedError(input.signal, input.label) : new AgentRunError("infra-failed", `failed to start ${input.label}: ${redactText(error.message)}`, { httpStatus: 503 })));
+      spawnError = error;
+    };
+    const onExit = (code: number | null, signal: NodeJS.Signals | null): void => {
+      exit = { code, signal };
     };
     const onClose = (code: number | null, signal: NodeJS.Signals | null): void => {
-      finish(() => aborted ? reject(abortedError(input.signal, input.label)) : resolve({ code, signal }));
+      finish(() => {
+        if (aborted) reject(abortedError(input.signal, input.label));
+        else if (spawnError) reject(new AgentRunError("infra-failed", `failed to start ${input.label}: ${redactText(spawnError.message)}`, { httpStatus: 503 }));
+        else resolve(exit ?? { code, signal });
+      });
     };
 
     child.once("error", onError);
+    child.once("exit", onExit);
     child.once("close", onClose);
     input.signal?.addEventListener("abort", onAbort, { once: true });
     if (input.signal?.aborted) onAbort();

src/mgr/provider-profiles.ts

@@ -7,6 +7,7 @@ import type { AgentRunStore } from "./store.js";
 import type { BackendProfile, ExecutionPolicy, JsonRecord, JsonValue } from "../common/types.js";
 import { asRecord, validateBackendProfile } from "../common/validation.js";
 import { redactJson, redactText } from "../common/redaction.js";
+import { waitForChildProcess } from "./abortable-child.js";
 import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js";
 import { buildRunResult } from "./result.js";
 import { runnerJobStatusSummary } from "./runner-job-status.js";
@@ -809,13 +810,9 @@ async function runKubectl(kubectlCommand: string, args: string[], stdin?: string
   child.stderr.setEncoding("utf8");
   child.stdout.on("data", (chunk) => { stdout += String(chunk); });
   child.stderr.on("data", (chunk) => { stderr += String(chunk); });
+  const resultPromise = waitForChildProcess(child, { label: "kubectl" });
   child.stdin.end(stdin ?? "");
-  const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
-    child.on("error", reject);
-    child.on("close", (code, signal) => resolve({ code, signal }));
-  }).catch((error: unknown) => {
-    throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
-  });
+  const result = await resultPromise;
   return { ...result, stdout, stderr };
 }
 

src/mgr/runner-reconciler.ts

@@ -8,6 +8,7 @@ import { agentRunBusinessTraceId } from "../common/otel-trace.js";
 import { nowIso, stableHash } from "../common/validation.js";
 import type { AgentRunStore } from "./store.js";
 import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
+import { waitForChildProcess } from "./abortable-child.js";
 import { reconcileStalePendingRunners, type RunnerRetentionOptions } from "./runner-retention.js";
 
 export interface RunnerReconcilerOptions {
@@ -1020,12 +1021,7 @@ async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ cod
   child.stderr.setEncoding("utf8");
   child.stdout.on("data", (chunk) => { stdout += String(chunk); });
   child.stderr.on("data", (chunk) => { stderr += String(chunk); });
-  const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
-    child.on("error", reject);
-    child.on("close", (code, signal) => resolve({ code, signal }));
-  }).catch((error: unknown) => {
-    throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
-  });
+  const result = await waitForChildProcess(child, { label: "kubectl" });
   return { ...result, stdout, stderr };
 }
 

src/mgr/tool-credentials.ts

@@ -4,6 +4,7 @@ import { AgentRunError } from "../common/errors.js";
 import type { JsonRecord } from "../common/types.js";
 import { asRecord } from "../common/validation.js";
 import { redactJson, redactText } from "../common/redaction.js";
+import { waitForChildProcess } from "./abortable-child.js";
 
 const defaultNamespace = "agentrun-v01";
 const annotationPrefix = "agentrun.pikastech.local/tool-credential";
@@ -251,13 +252,9 @@ async function runKubectl(kubectlCommand: string, args: string[], stdin?: string
   child.stderr.setEncoding("utf8");
   child.stdout.on("data", (chunk) => { stdout += String(chunk); });
   child.stderr.on("data", (chunk) => { stderr += String(chunk); });
+  const resultPromise = waitForChildProcess(child, { label: "kubectl" });
   child.stdin.end(stdin ?? "");
-  const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
-    child.on("error", reject);
-    child.on("close", (code, signal) => resolve({ code, signal }));
-  }).catch((error: unknown) => {
-    throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
-  });
+  const result = await resultPromise;
   return { ...result, stdout, stderr };
 }
 

src/selftest/cases/24-runner-startup-retry.ts

@@ -1,5 +1,5 @@
 import assert from "node:assert/strict";
-import { chmod, writeFile } from "node:fs/promises";
+import { chmod, readdir, readFile, writeFile } from "node:fs/promises";
 import path from "node:path";
 import type { JsonRecord } from "../../common/types.js";
 import { reconcileRunnerJobsOnce, type RunnerStartupRecoveryOptions } from "../../mgr/runner-reconciler.js";
@@ -28,6 +28,7 @@ const selfTest: SelfTestCase = async (context) => {
     await assertDeterministicFailureDoesNotRetry(context, fixturePath, kubectlCommand);
     await assertFailedJobWithoutTerminalOutboxFails(context, fixturePath, kubectlCommand);
     await assertSucceededJobWithoutTerminalOutboxFails(context, fixturePath, kubectlCommand);
+    await assertRepeatedReconcileReapsKubectl(fixturePath, kubectlCommand);
     await assertRunnerBootObservationLifecycle();
   } finally {
     if (previousFixture === undefined) delete process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH;
@@ -41,12 +42,39 @@ const selfTest: SelfTestCase = async (context) => {
       "image-not-found-fails-without-retry",
       "failed-job-without-terminal-outbox-fails",
       "succeeded-job-without-terminal-outbox-fails",
+      "repeated-manager-kubectl-commands-are-reaped",
       "semantic-failure-events-carry-workbench-fields",
       "runner-git-mirror-boot-is-visible-before-claim",
     ],
   };
 };
 
+async function assertRepeatedReconcileReapsKubectl(fixturePath: string, kubectlCommand: string): Promise<void> {
+  if (process.platform !== "linux") return;
+  const fact = createRunnerFact("child-reaping");
+  await writeFixture(fixturePath, "", "", "running");
+  for (let index = 0; index < 32; index++) {
+    await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: policy });
+  }
+  assert.deepEqual(await ownedZombieChildren(), []);
+}
+
+async function ownedZombieChildren(): Promise<Array<{ pid: number; command: string }>> {
+  const taskIds = await readdir(`/proc/${process.pid}/task`);
+  const childIds = new Set<string>();
+  for (const taskId of taskIds) {
+    const children = await readFile(`/proc/${process.pid}/task/${taskId}/children`, "utf8").catch(() => "");
+    for (const pid of children.trim().split(/\s+/u).filter(Boolean)) childIds.add(pid);
+  }
+  const zombies: Array<{ pid: number; command: string }> = [];
+  for (const pid of childIds) {
+    const stat = await readFile(`/proc/${pid}/stat`, "utf8").catch(() => "");
+    const match = stat.match(/^\d+ \((.*)\) ([A-Z])/u);
+    if (match?.[2] === "Z") zombies.push({ pid: Number(pid), command: match[1] ?? "" });
+  }
+  return zombies.sort((left, right) => left.pid - right.pid);
+}
+
 async function assertRunnerBootObservationLifecycle(): Promise<void> {
   const fact = createRunnerFact("boot-observation");
   const identity = { runId: fact.runId, commandId: fact.commandId, attemptId: "attempt-boot-observation", runnerId: "runner-boot-observation" };