From 7d324b537d1acea491532ba33545ba58984f4178 Mon Sep 17 00:00:00 2001 From: i-trytoohard Date: Fri, 15 May 2026 03:38:09 +0530 Subject: [PATCH] fix(cli): reap daemon children on stop+SIGINT, sweep orphans on start (closes #1848) (#1849) * feat(core): add managed daemon child registry * fix(cli): reap daemon children on stop and shutdown * test(cli): cover daemon child reaping * chore: version packages for 0.9.0 * fix(core): avoid regex in orphan process scan * test(web): expect managed child spawn helper * fix(core): let daemon shutdown own signal exit * fix(web): mark shutdown ownership before spawning children * fix(core): wait for managed children before fallback exit * docs: document daemon process management architecture * fix(core): scope daemon child sweeps by owner pid * docs: link process architecture to PR changes * docs: clarify legacy messaging watcher status * chore(core): remove unused messaging watcher orphan pattern * chore: remove version bump from orphan reaping PR * docs: remove process management design draft --------- Co-authored-by: i-trytoohard <193449657+i-trytoohard@users.noreply.github.com> --- eslint.config.js | 25 + packages/cli/__tests__/commands/start.test.ts | 81 ++- packages/cli/__tests__/lib/daemon.test.ts | 21 +- packages/cli/src/commands/start.ts | 120 ++-- packages/cli/src/lib/daemon.ts | 3 +- packages/cli/src/lib/shutdown.ts | 13 +- .../src/__tests__/daemon-children.test.ts | 171 ++++++ packages/core/src/daemon-children.ts | 528 ++++++++++++++++++ packages/core/src/index.ts | 25 +- packages/core/src/platform.ts | 4 + .../src/daemon-children.integration.test.ts | 149 +++++ .../__tests__/server-compatibility.test.ts | 4 +- packages/web/server/start-all.ts | 22 +- 13 files changed, 1102 insertions(+), 64 deletions(-) create mode 100644 packages/core/src/__tests__/daemon-children.test.ts create mode 100644 packages/core/src/daemon-children.ts create mode 100644 packages/integration-tests/src/daemon-children.integration.test.ts diff --git a/eslint.config.js b/eslint.config.js index 86d1cd57f..c6dcc531e 100644 --- a/eslint.config.js +++ b/eslint.config.js @@ -99,6 +99,31 @@ export default tseslint.config( }, }, + // Long-running daemon entrypoints must use the managed-child API so any + // subprocess they own is registered and reaped on stop/SIGINT. + { + files: ["packages/cli/src/commands/start.ts", "packages/web/server/start-all.ts"], + rules: { + "no-restricted-imports": [ + "error", + { + paths: [ + { + name: "node:child_process", + importNames: ["spawn"], + message: "Use spawnManagedDaemonChild() for daemon-owned subprocesses.", + }, + { + name: "child_process", + importNames: ["spawn"], + message: "Use spawnManagedDaemonChild() for daemon-owned subprocesses.", + }, + ], + }, + ], + }, + }, + // Scripts directory - Node.js environment { files: ["scripts/**/*.js", "scripts/**/*.mjs", "packages/*/scripts/**/*.js"], diff --git a/packages/cli/__tests__/commands/start.test.ts b/packages/cli/__tests__/commands/start.test.ts index 841830617..60a247c8e 100644 --- a/packages/cli/__tests__/commands/start.test.ts +++ b/packages/cli/__tests__/commands/start.test.ts @@ -34,6 +34,9 @@ const { mockSpawn, mockFindPidByPort, mockKillProcessTree, + mockSweepDaemonChildren, + mockScanAoOrphans, + mockReapAoOrphans, mockStartProjectSupervisor, } = vi.hoisted(() => ({ mockExec: vi.fn(), @@ -56,6 +59,9 @@ const { mockSpawn: vi.fn(), mockFindPidByPort: vi.fn(), mockKillProcessTree: vi.fn(), + mockSweepDaemonChildren: vi.fn(), + mockScanAoOrphans: vi.fn(), + mockReapAoOrphans: vi.fn(), mockStartProjectSupervisor: vi.fn(), })); @@ -144,6 +150,9 @@ vi.mock("@aoagents/ao-core", async (importOriginal) => { }, findPidByPort: mockFindPidByPort, killProcessTree: mockKillProcessTree, + sweepDaemonChildren: mockSweepDaemonChildren, + scanAoOrphans: mockScanAoOrphans, + reapAoOrphans: mockReapAoOrphans, }; }); @@ -282,6 +291,7 @@ import { registerStart, registerStop, autoCreateConfig } from "../../src/command let tmpDir: string; let program: Command; let cwdSpy: ReturnType; +let originalAoGlobalConfig: string | undefined; function createSpawnChild(options?: { /** Emit `error` instead of `close`. */ @@ -319,6 +329,8 @@ function createSpawnChild(options?: { beforeEach(async () => { tmpDir = mkdtempSync(join(tmpdir(), "ao-start-test-")); + originalAoGlobalConfig = process.env["AO_GLOBAL_CONFIG"]; + process.env["AO_GLOBAL_CONFIG"] = join(tmpDir, "global-agent-orchestrator.yaml"); program = new Command(); program.exitOverride(); @@ -398,6 +410,22 @@ beforeEach(async () => { mockFindPidByPort.mockResolvedValue(null); mockKillProcessTree.mockReset(); mockKillProcessTree.mockResolvedValue(undefined); + mockSweepDaemonChildren.mockReset(); + mockSweepDaemonChildren.mockResolvedValue({ + attempted: 0, + terminated: 0, + forceKilled: 0, + failed: 0, + }); + mockScanAoOrphans.mockReset(); + mockScanAoOrphans.mockResolvedValue([]); + mockReapAoOrphans.mockReset(); + mockReapAoOrphans.mockResolvedValue({ + attempted: 0, + terminated: 0, + forceKilled: 0, + failed: 0, + }); mockStartProjectSupervisor.mockReset(); mockStartProjectSupervisor.mockResolvedValue({ stop: vi.fn(), reconcileNow: vi.fn() }); mockDetectOpenClawInstallation.mockReset(); @@ -436,6 +464,8 @@ beforeEach(async () => { afterEach(() => { if (cwdSpy) cwdSpy.mockRestore(); + if (originalAoGlobalConfig === undefined) delete process.env["AO_GLOBAL_CONFIG"]; + else process.env["AO_GLOBAL_CONFIG"] = originalAoGlobalConfig; rmSync(tmpDir, { recursive: true, force: true }); vi.restoreAllMocks(); }); @@ -1986,6 +2016,7 @@ describe("start command — platform-aware runtime fallback", () => { .join("\n"); expect(output).toContain("Stopped sessions for"); expect(output).not.toContain("Dashboard stopped"); + expect(mockSweepDaemonChildren).not.toHaveBeenCalled(); }); it("targeted stop does NOT unregister running.json", async () => { @@ -2110,6 +2141,7 @@ describe("start command — platform-aware runtime fallback", () => { // not a direct process.kill — that's how it gets `taskkill /T /F` on // Windows and process-group kill on Unix. Assert on the mock. expect(mockKillProcessTree).toHaveBeenCalledWith(99999, "SIGTERM"); + expect(mockSweepDaemonChildren).toHaveBeenCalledWith({ ownerPid: 99999 }); expect(mockUnregister).toHaveBeenCalled(); expect(mockRemoveProjectFromRunning).not.toHaveBeenCalled(); }); @@ -2390,7 +2422,12 @@ describe("start command — already-running detection", () => { globalConfigPath, yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { "my-app": { name: "My App", @@ -2582,7 +2619,12 @@ describe("start command — already-running detection", () => { configPath, yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { "my-app": { name: "My App", @@ -2636,7 +2678,12 @@ describe("start command — already-running detection", () => { const { stringify: yamlStringify } = await import("yaml"); const originalYaml = yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { "my-app": { name: "My App", @@ -2689,7 +2736,12 @@ describe("start command — path-based deduplication in addProjectToConfig", () configPath, yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { "my-app": { name: "My App", @@ -2742,7 +2794,12 @@ describe("start command — path-based deduplication in addProjectToConfig", () configPath, yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { "old-name": { name: "Old Name", @@ -2802,7 +2859,12 @@ describe("start command — global registry mutations", () => { globalConfigPath, yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { current: { projectId: "current", @@ -2903,7 +2965,12 @@ describe("start command — global registry mutations", () => { globalConfigPath, yamlStringify( { - defaults: { runtime: "process", agent: "claude-code", workspace: "worktree", notifiers: [] }, + defaults: { + runtime: "process", + agent: "claude-code", + workspace: "worktree", + notifiers: [], + }, projects: { current: { projectId: "current", diff --git a/packages/cli/__tests__/lib/daemon.test.ts b/packages/cli/__tests__/lib/daemon.test.ts index c50e1edd1..990fa5b80 100644 --- a/packages/cli/__tests__/lib/daemon.test.ts +++ b/packages/cli/__tests__/lib/daemon.test.ts @@ -1,11 +1,13 @@ import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; import type * as AoCore from "@aoagents/ao-core"; -const { mockUnregister, mockWaitForExit, mockKillProcessTree } = vi.hoisted(() => ({ - mockUnregister: vi.fn(), - mockWaitForExit: vi.fn(), - mockKillProcessTree: vi.fn(), -})); +const { mockUnregister, mockWaitForExit, mockKillProcessTree, mockSweepDaemonChildren } = + vi.hoisted(() => ({ + mockUnregister: vi.fn(), + mockWaitForExit: vi.fn(), + mockKillProcessTree: vi.fn(), + mockSweepDaemonChildren: vi.fn(), + })); vi.mock("../../src/lib/running-state.js", () => ({ unregister: mockUnregister, @@ -17,6 +19,7 @@ vi.mock("@aoagents/ao-core", async () => { return { ...actual, killProcessTree: mockKillProcessTree, + sweepDaemonChildren: mockSweepDaemonChildren, }; }); @@ -37,6 +40,13 @@ beforeEach(() => { mockWaitForExit.mockReset(); mockKillProcessTree.mockReset(); mockKillProcessTree.mockResolvedValue(undefined); + mockSweepDaemonChildren.mockReset(); + mockSweepDaemonChildren.mockResolvedValue({ + attempted: 0, + terminated: 0, + forceKilled: 0, + failed: 0, + }); }); afterEach(() => { @@ -93,6 +103,7 @@ describe("killExistingDaemon", () => { it("uses killProcessTree(SIGTERM), awaits exit, and unregisters on the happy path", async () => { mockWaitForExit.mockResolvedValueOnce(true); await killExistingDaemon(fakeRunning); + expect(mockSweepDaemonChildren).toHaveBeenCalledWith({ ownerPid: 12345 }); expect(mockKillProcessTree).toHaveBeenCalledWith(12345, "SIGTERM"); expect(mockKillProcessTree).toHaveBeenCalledTimes(1); expect(mockWaitForExit).toHaveBeenCalledWith(12345, 5000); diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index f29be5559..1b5841a30 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -9,7 +9,7 @@ * (or equivalent flag) at launch time — no file writing required. */ -import { spawn, type ChildProcess } from "node:child_process"; +import { type ChildProcess } from "node:child_process"; import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { resolve, basename, dirname } from "node:path"; import { cwd } from "node:process"; @@ -27,6 +27,8 @@ import { isTerminalSession, getDefaultRuntime, isWindows, + isMac, + isLinux, findPidByPort, killProcessTree, loadLocalProjectConfigDetailed, @@ -37,9 +39,15 @@ import { type ProjectConfig, type ParsedRepoUrl, writeLocalProjectConfig, + spawnManagedDaemonChild, + sweepDaemonChildren, + scanAoOrphans, + reapAoOrphans, + type DaemonChildSweepResult, + type AoOrphanProcess, } from "@aoagents/ao-core"; import { parse as yamlParse, stringify as yamlStringify } from "yaml"; -import { exec, execSilent, forwardSignalsToChild, git } from "../lib/shell.js"; +import { exec, execSilent, git } from "../lib/shell.js"; import { getSessionManager } from "../lib/create-session-manager.js"; import { listLifecycleWorkers } from "../lib/lifecycle-service.js"; import { startBunTmpJanitor } from "../lib/bun-tmp-janitor.js"; @@ -96,7 +104,7 @@ import { tryInstallWithAttempts, } from "../lib/install-helpers.js"; import { ensureGit, runtimePreflight } from "../lib/startup-preflight.js"; -import { installShutdownHandlers } from "../lib/shutdown.js"; +import { installShutdownHandlers, isShutdownInProgress } from "../lib/shutdown.js"; import { resolveOrCreateProject } from "../lib/resolve-project.js"; import { pathsEqual } from "../lib/path-equality.js"; import { maybePromptForUpdateChannel } from "../lib/update-channel-onboarding.js"; @@ -308,10 +316,10 @@ async function promptAgentSelection(): Promise<{ } function ghInstallAttempts(): InstallAttempt[] { - if (process.platform === "darwin") { + if (isMac()) { return [{ cmd: "brew", args: ["install", "gh"], label: "brew install gh" }]; } - if (process.platform === "linux") { + if (isLinux()) { return [ { cmd: "sudo", @@ -801,7 +809,7 @@ async function startDashboard( if (useDevServer) { // Monorepo with --dev: use pnpm run dev (tsx watch, HMR, etc.) console.log(chalk.dim(" Mode: development (HMR enabled)")); - child = spawn("pnpm", ["run", "dev"], { + child = spawnManagedDaemonChild("dashboard", "pnpm", ["run", "dev"], { cwd: webDir, stdio: "inherit", detached: !isWindows(), @@ -814,7 +822,7 @@ async function startDashboard( console.log(chalk.dim(" Tip: use --dev for hot reload when editing dashboard UI\n")); } const startScript = resolve(webDir, "dist-server", "start-all.js"); - child = spawn("node", [startScript], { + child = spawnManagedDaemonChild("dashboard", "node", [startScript], { cwd: webDir, stdio: "inherit", detached: !isWindows(), @@ -857,6 +865,11 @@ async function runStartup( // presence of `updateChannel` in the global config). await maybePromptForUpdateChannel(); + // Install the parent shutdown path before spawning any managed children. + // This guarantees a SIGINT/SIGTERM in the middle of startup still performs + // the full AO cleanup instead of relying on Node's default signal exit. + installShutdownHandlers({ configPath: config.configPath, projectId }); + const shouldStartLifecycle = opts?.dashboard !== false || opts?.orchestrator !== false; let port = config.port ?? DEFAULT_PORT; console.log(chalk.bold(`\nStarting orchestrator for ${chalk.cyan(project.name)}\n`)); @@ -1122,35 +1135,9 @@ async function runStartup( // Keep dashboard process alive if it was started if (dashboardProcess) { - const pid = dashboardProcess.pid; - - // On Unix the dashboard is spawned with detached:true (own process group) - // so Ctrl+C only reaches AO's process group, not the dashboard's. Forward - // SIGINT/SIGTERM so the dashboard group is also cleaned up on exit. - // On Windows, detached:false keeps child in the same console — - // Ctrl+C reaches both processes. No signal forwarding needed. - if (!isWindows() && pid) { - forwardSignalsToChild(pid, dashboardProcess); - } - - // Also kill the dashboard child when the parent exits for any reason - // (normal exit path after lifecycle flush). The `exit` event is - // synchronous and fires regardless of platform, so it covers the cases - // where forwardSignalsToChild doesn't (Windows, or non-signal exits). - /* c8 ignore start -- exit handler only fires on process termination */ - const killDashboardChild = (): void => { - try { - dashboardProcess?.kill("SIGTERM"); - } catch { - // already dead - } - }; - /* c8 ignore stop */ - process.on("exit", killDashboardChild); - dashboardProcess.on("exit", (code) => { - process.removeListener("exit", killDashboardChild); if (openAbort) openAbort.abort(); + if (isShutdownInProgress()) return; if (code !== 0 && code !== null) { console.error(chalk.red(`Dashboard exited with code ${code}`)); } @@ -1219,6 +1206,60 @@ async function stopDashboard(port: number): Promise { console.log(chalk.yellow("Could not stop dashboard (may not be running)")); } +function formatSweepSummary(result: DaemonChildSweepResult): string { + return `${result.terminated} graceful, ${result.forceKilled} force-killed${ + result.failed > 0 ? `, ${result.failed} failed` : "" + }`; +} + +async function sweepRegisteredDaemonChildren(ownerPid?: number): Promise { + const result = await sweepDaemonChildren({ ownerPid }); + if (result.attempted > 0) { + console.log( + chalk.dim( + ` Swept ${result.attempted} registered daemon child(ren): ${formatSweepSummary(result)}`, + ), + ); + } +} + +function describeAoOrphans(orphans: AoOrphanProcess[]): string { + return orphans + .map((orphan) => `${orphan.pid} (${orphan.role})`) + .slice(0, 8) + .join(", "); +} + +async function maybeSweepAoOrphansOnStart(reapOrphans: boolean | undefined): Promise { + const orphans = await scanAoOrphans(); + if (orphans.length === 0) return; + + if (!reapOrphans && isHumanCaller()) { + console.log( + chalk.yellow( + `\n Found ${orphans.length} orphaned AO child process(es): ${describeAoOrphans(orphans)}`, + ), + ); + reapOrphans = await promptConfirm("Kill orphaned AO child processes before starting?", true); + } + + if (!reapOrphans) { + console.log( + chalk.yellow( + ` Found ${orphans.length} orphaned AO child process(es). Run \`ao start --reap-orphans\` to clean them up.`, + ), + ); + return; + } + + const result = await reapAoOrphans(orphans); + console.log( + chalk.green( + ` Reaped ${result.attempted} orphaned AO child process(es): ${formatSweepSummary(result)}`, + ), + ); +} + /** * Spawn an orchestrator session against an already-running daemon, invalidate * the dashboard's project cache, and surface enough context for the user to @@ -1305,6 +1346,7 @@ export function registerStart(program: Command): void { .option("--rebuild", "Clean and rebuild dashboard before starting") .option("--dev", "Use Next.js dev server with hot reload (for dashboard UI development)") .option("--interactive", "Prompt to configure config settings") + .option("--reap-orphans", "Kill orphaned AO child processes before starting") .action( async ( projectArg?: string, @@ -1314,6 +1356,7 @@ export function registerStart(program: Command): void { rebuild?: boolean; dev?: boolean; interactive?: boolean; + reapOrphans?: boolean; }, ) => { let releaseStartupLock: (() => void) | undefined; @@ -1326,6 +1369,7 @@ export function registerStart(program: Command): void { try { releaseStartupLock = await acquireStartupLock(); + await maybeSweepAoOrphansOnStart(opts?.reapOrphans); let config: OrchestratorConfig; let projectId: string; let project: ProjectConfig; @@ -1603,9 +1647,7 @@ export function registerStart(program: Command): void { }); // Ctrl+C and `ao stop` (which sends SIGTERM) perform a full - // graceful shutdown: kill sessions, record last-stop state for - // restore, unregister, then exit. See lib/shutdown.ts. - installShutdownHandlers({ configPath: config.configPath, projectId }); + // graceful shutdown via the handler installed inside runStartup(). } catch (err) { if (err instanceof Error) { console.error(chalk.red("\nError:"), err.message); @@ -1691,6 +1733,7 @@ export function registerStop(program: Command): void { // protocol so node-pty disposes ConPTY gracefully (avoids WER // 0x800700e8). No-op on non-Windows. await sweepWindowsPtyHostsBeforeParentKill(); + await sweepRegisteredDaemonChildren(running.pid); // killProcessTree handles process trees on Windows (taskkill /T /F) // and process groups on Unix; it swallows "already dead" internally. await killProcessTree(running.pid, "SIGTERM"); @@ -1828,8 +1871,11 @@ export function registerStop(program: Command): void { // protocol so node-pty disposes ConPTY gracefully (avoids WER // 0x800700e8). No-op on non-Windows. await sweepWindowsPtyHostsBeforeParentKill(); + await sweepRegisteredDaemonChildren(running.pid); await killProcessTree(running.pid, "SIGTERM"); await unregister(); + } else { + await sweepRegisteredDaemonChildren(); } await stopDashboard(running?.port ?? port); } diff --git a/packages/cli/src/lib/daemon.ts b/packages/cli/src/lib/daemon.ts index 9a5c60e6a..5ccccd924 100644 --- a/packages/cli/src/lib/daemon.ts +++ b/packages/cli/src/lib/daemon.ts @@ -20,7 +20,7 @@ */ import chalk from "chalk"; -import { killProcessTree } from "@aoagents/ao-core"; +import { killProcessTree, sweepDaemonChildren } from "@aoagents/ao-core"; import { unregister, waitForExit, type RunningState } from "./running-state.js"; /** @@ -90,6 +90,7 @@ export function attachToDaemon(running: RunningState): AttachedDaemon { * internally. */ export async function killExistingDaemon(running: RunningState): Promise { + await sweepDaemonChildren({ ownerPid: running.pid }); await killProcessTree(running.pid, "SIGTERM"); if (!(await waitForExit(running.pid, 5000))) { console.log(chalk.yellow(" Process didn't exit cleanly, sending SIGKILL...")); diff --git a/packages/cli/src/lib/shutdown.ts b/packages/cli/src/lib/shutdown.ts index 397efe795..228f5f88d 100644 --- a/packages/cli/src/lib/shutdown.ts +++ b/packages/cli/src/lib/shutdown.ts @@ -12,7 +12,12 @@ * see ao-118 plan PR B). */ -import { isTerminalSession, loadConfig } from "@aoagents/ao-core"; +import { + isTerminalSession, + loadConfig, + markDaemonShutdownHandlerInstalled, + sweepDaemonChildren, +} from "@aoagents/ao-core"; import { stopBunTmpJanitor } from "./bun-tmp-janitor.js"; import { getSessionManager } from "./create-session-manager.js"; import { stopAllLifecycleWorkers } from "./lifecycle-service.js"; @@ -36,6 +41,10 @@ export interface ShutdownContext { let handlersInstalled = false; let shuttingDown = false; +export function isShutdownInProgress(): boolean { + return shuttingDown; +} + /** * Install SIGINT/SIGTERM handlers. Process-wide idempotent — calling * this more than once is a no-op. Only the first signal triggers @@ -45,6 +54,7 @@ let shuttingDown = false; export function installShutdownHandlers(ctx: ShutdownContext): void { if (handlersInstalled) return; handlersInstalled = true; + markDaemonShutdownHandlerInstalled(); const shutdown = (signal: NodeJS.Signals): void => { if (shuttingDown) return; @@ -105,6 +115,7 @@ export function installShutdownHandlers(ctx: ShutdownContext): void { }); } + await sweepDaemonChildren({ ownerPid: process.pid }); await unregister(); } catch { // Best-effort — always exit even if cleanup fails diff --git a/packages/core/src/__tests__/daemon-children.test.ts b/packages/core/src/__tests__/daemon-children.test.ts new file mode 100644 index 000000000..2f4ec8f4b --- /dev/null +++ b/packages/core/src/__tests__/daemon-children.test.ts @@ -0,0 +1,171 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + __getDaemonChildrenRegistryFile, + clearDaemonChildrenRegistry, + detectAoOrphansFromPsOutput, + getDaemonChildren, + registerDaemonChild, + spawnManagedDaemonChild, + sweepDaemonChildren, + unregisterDaemonChild, +} from "../daemon-children.js"; + +function isProcessAliveForTest(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (err: unknown) { + return (err as { code?: string }).code === "EPERM"; + } +} + +async function waitForChildExit(child: ChildProcess): Promise { + if (child.exitCode !== null || child.signalCode !== null) return; + await Promise.race([ + new Promise((resolve) => child.once("exit", () => resolve())), + new Promise((resolve) => setTimeout(resolve, 2_000)), + ]); +} + +describe("daemon child registry", () => { + let tmpHome: string; + let originalHome: string | undefined; + + beforeEach(() => { + tmpHome = mkdtempSync(join(tmpdir(), "ao-daemon-children-")); + originalHome = process.env["HOME"]; + process.env["HOME"] = tmpHome; + clearDaemonChildrenRegistry(); + }); + + afterEach(() => { + clearDaemonChildrenRegistry(); + if (originalHome === undefined) delete process.env["HOME"]; + else process.env["HOME"] = originalHome; + rmSync(tmpHome, { recursive: true, force: true }); + }); + + it("writes, reads, and unregisters daemon child pids", () => { + registerDaemonChild({ + pid: process.pid, + parentPid: process.pid, + role: "dashboard", + command: "node dist-server/start-all.js", + }); + + expect(getDaemonChildren()).toEqual([ + expect.objectContaining({ + pid: process.pid, + parentPid: process.pid, + role: "dashboard", + command: "node dist-server/start-all.js", + }), + ]); + + const raw = JSON.parse(readFileSync(__getDaemonChildrenRegistryFile(), "utf-8")) as unknown[]; + expect(raw).toHaveLength(1); + + unregisterDaemonChild(process.pid); + expect(getDaemonChildren()).toEqual([]); + }); + + it("sweeps only children owned by the requested daemon pid", async () => { + const targetChild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 30_000)"], { + stdio: "ignore", + }); + const otherChild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 30_000)"], { + stdio: "ignore", + }); + + try { + expect(targetChild.pid).toBeTypeOf("number"); + expect(otherChild.pid).toBeTypeOf("number"); + const targetPid = targetChild.pid as number; + const otherPid = otherChild.pid as number; + + registerDaemonChild({ + pid: targetPid, + parentPid: 111, + role: "owned-by-target", + command: "node target.js", + }); + registerDaemonChild({ + pid: otherPid, + parentPid: 222, + role: "owned-by-other-daemon", + command: "node other.js", + }); + + const result = await sweepDaemonChildren({ ownerPid: 111, graceMs: 1_000 }); + + expect(result.attempted).toBe(1); + expect(isProcessAliveForTest(otherPid)).toBe(true); + expect(getDaemonChildren()).toContainEqual( + expect.objectContaining({ + pid: otherPid, + parentPid: 222, + role: "owned-by-other-daemon", + }), + ); + } finally { + targetChild.kill("SIGKILL"); + otherChild.kill("SIGKILL"); + await waitForChildExit(targetChild); + await waitForChildExit(otherChild); + } + }); + + it("spawnManagedDaemonChild makes registry tracking the default for daemon spawns", async () => { + const child = spawnManagedDaemonChild( + "test-child", + process.execPath, + ["-e", "setTimeout(() => {}, 30_000)"], + { stdio: "ignore" }, + ); + + expect(child.pid).toBeTypeOf("number"); + expect(getDaemonChildren()).toContainEqual( + expect.objectContaining({ + pid: child.pid, + role: "test-child", + parentPid: process.pid, + command: `${process.execPath} -e setTimeout(() => {}, 30_000)`, + }), + ); + + child.kill("SIGTERM"); + await new Promise((resolve) => child.once("exit", () => resolve())); + + expect(getDaemonChildren()).not.toContainEqual(expect.objectContaining({ pid: child.pid })); + }); +}); + +describe("AO orphan detection", () => { + it("detects PPID=1 AO dashboard, websocket, and lifecycle processes", () => { + const output = [ + " 90350 1 node next-server (v15.5.15)", + " 90351 1 node /opt/homebrew/lib/node_modules/@aoagents/ao-web/dist-server/start-all.js", + " 47457 1 node /opt/homebrew/lib/node_modules/@aoagents/ao-web@0.2.4/dist-server/start-all.js", + " 47458 1 node @aoagents/ao-web@0.2.4 dist-server/start-all.js", + " 47575 1 node /opt/homebrew/lib/node_modules/@aoagents/ao-web@0.2.4/dist-server/terminal-websocket.js", + " 47580 1 node /opt/homebrew/lib/node_modules/@aoagents/ao-web@0.2.4/dist-server/direct-terminal-ws.js", + " 9914 1 node /opt/homebrew/bin/ao lifecycle-worker codex-startup-factory", + " 22222 3333 node /opt/homebrew/bin/ao lifecycle-worker not-an-orphan", + " 44444 1 node unrelated-server.js", + ].join("\n"); + + expect(detectAoOrphansFromPsOutput(output)).toEqual([ + expect.objectContaining({ pid: 90350, role: "next-server" }), + expect.objectContaining({ pid: 90351, role: "ao-web" }), + expect.objectContaining({ pid: 47457, role: "ao-web" }), + expect.objectContaining({ pid: 47458, role: "ao-web" }), + expect.objectContaining({ pid: 47575, role: "ao-web" }), + expect.objectContaining({ pid: 47580, role: "ao-web" }), + expect.objectContaining({ pid: 9914, role: "lifecycle-worker" }), + ]); + }); +}); diff --git a/packages/core/src/daemon-children.ts b/packages/core/src/daemon-children.ts new file mode 100644 index 000000000..1e3d55008 --- /dev/null +++ b/packages/core/src/daemon-children.ts @@ -0,0 +1,528 @@ +import { + type ChildProcess, + type SpawnOptions, + execFile as execFileCb, + spawn, +} from "node:child_process"; +import { + existsSync, + mkdirSync, + readFileSync, + rmSync, + rmdirSync, + statSync, + unlinkSync, +} from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { promisify } from "node:util"; +import { atomicWriteFileSync } from "./atomic-write.js"; +import { isWindows, killProcessTree } from "./platform.js"; + +const execFileAsync = promisify(execFileCb); +const DEFAULT_GRACE_MS = 5_000; +const LOCK_STALE_MS = 10_000; + +export interface DaemonChildEntry { + pid: number; + role: string; + parentPid: number; + startedAt: string; + command?: string; +} + +export interface DaemonChildSweepResult { + attempted: number; + terminated: number; + forceKilled: number; + failed: number; +} + +export interface AoOrphanProcess { + pid: number; + ppid: number; + command: string; + role: string; +} + +export interface DaemonChildSweepOptions { + ownerPid?: number; + graceMs?: number; +} + +function getRegistryFile(): string { + return join(homedir(), ".agent-orchestrator", "daemon-children.json"); +} + +function getLockDir(): string { + return join(homedir(), ".agent-orchestrator", "daemon-children.lock"); +} + +function ensureStateDir(): void { + mkdirSync(join(homedir(), ".agent-orchestrator"), { recursive: true }); +} + +function sleepSync(ms: number): void { + const buffer = new SharedArrayBuffer(4); + const view = new Int32Array(buffer); + Atomics.wait(view, 0, 0, ms); +} + +function acquireRegistryLock(): () => void { + ensureStateDir(); + const lockDir = getLockDir(); + const deadline = Date.now() + 5_000; + while (true) { + try { + mkdirSync(lockDir); + return () => { + try { + rmdirSync(lockDir); + } catch { + // Best effort. + } + }; + } catch (err: unknown) { + const code = (err as NodeJS.ErrnoException).code; + if (code !== "EEXIST") throw err; + try { + const ageMs = Date.now() - statSync(lockDir).mtimeMs; + if (ageMs > LOCK_STALE_MS) { + rmSync(lockDir, { recursive: true, force: true }); + continue; + } + } catch { + // Retry lock acquisition if the lock disappeared between calls. + } + if (Date.now() > deadline) { + throw new Error(`Could not acquire daemon child registry lock (${lockDir})`, { + cause: err, + }); + } + sleepSync(25); + } + } +} + +function isDaemonChildEntry(value: unknown): value is DaemonChildEntry { + if (typeof value !== "object" || value === null) return false; + const entry = value as Partial; + return ( + typeof entry.pid === "number" && + entry.pid > 0 && + typeof entry.role === "string" && + typeof entry.parentPid === "number" && + typeof entry.startedAt === "string" && + (entry.command === undefined || typeof entry.command === "string") + ); +} + +function readRawDaemonChildren(): DaemonChildEntry[] { + const file = getRegistryFile(); + if (!existsSync(file)) return []; + try { + const parsed = JSON.parse(readFileSync(file, "utf-8")) as unknown; + if (!Array.isArray(parsed)) return []; + return parsed.filter(isDaemonChildEntry); + } catch { + return []; + } +} + +function writeRawDaemonChildren(entries: DaemonChildEntry[]): void { + const file = getRegistryFile(); + if (entries.length === 0) { + try { + unlinkSync(file); + } catch { + // File may not exist. + } + return; + } + ensureStateDir(); + atomicWriteFileSync(file, JSON.stringify(entries, null, 2)); +} + +function isProcessAlive(pid: number): boolean { + if (pid <= 0) return false; + try { + process.kill(pid, 0); + return true; + } catch (err: unknown) { + return (err as { code?: string }).code === "EPERM"; + } +} + +async function waitForProcessesExit(pids: number[], timeoutMs: number): Promise> { + const alive = new Set(pids.filter(isProcessAlive)); + const deadline = Date.now() + timeoutMs; + + while (alive.size > 0 && Date.now() < deadline) { + for (const pid of alive) { + if (!isProcessAlive(pid)) alive.delete(pid); + } + if (alive.size > 0) { + await sleep(Math.min(50, Math.max(0, deadline - Date.now()))); + } + } + + for (const pid of alive) { + if (!isProcessAlive(pid)) alive.delete(pid); + } + + return alive; +} + +export function registerDaemonChild(entry: Omit): void { + if (entry.pid <= 0) return; + const release = acquireRegistryLock(); + try { + const next = readRawDaemonChildren().filter((existing) => existing.pid !== entry.pid); + next.push({ ...entry, startedAt: new Date().toISOString() }); + writeRawDaemonChildren(next); + } finally { + release(); + } +} + +export function unregisterDaemonChild(pid: number): void { + const release = acquireRegistryLock(); + try { + const before = readRawDaemonChildren(); + const next = before.filter((entry) => entry.pid !== pid); + if (next.length === before.length) return; + writeRawDaemonChildren(next); + } finally { + release(); + } +} + +export function getDaemonChildren(): DaemonChildEntry[] { + const release = acquireRegistryLock(); + try { + const all = readRawDaemonChildren(); + const live = all.filter((entry) => isProcessAlive(entry.pid)); + if (live.length !== all.length) writeRawDaemonChildren(live); + return live; + } finally { + release(); + } +} + +export function clearDaemonChildrenRegistry(): void { + const release = acquireRegistryLock(); + try { + writeRawDaemonChildren([]); + } finally { + release(); + } +} + +function pruneSweptDaemonChildren(sweptPids: Set): void { + const release = acquireRegistryLock(); + try { + const next = readRawDaemonChildren().filter( + (entry) => !sweptPids.has(entry.pid) || isProcessAlive(entry.pid), + ); + writeRawDaemonChildren(next); + } finally { + release(); + } +} + +const reapedChildren = new WeakSet(); +const managedChildren = new Map(); +let managedSignalHandlersInstalled = false; +let daemonShutdownHandlerInstalled = false; +let fallbackShutdownStarted = false; + +/** + * Tell the managed child reaper that this process owns an application-level + * SIGINT/SIGTERM shutdown path. The reaper will still forward signals to + * children, but it will not install its default 50ms fallback exit; the owning + * shutdown handler is responsible for exiting after its async cleanup finishes. + */ +export function markDaemonShutdownHandlerInstalled(): void { + daemonShutdownHandlerInstalled = true; +} + +function getManagedChildPids(): number[] { + return [...managedChildren.keys()]; +} + +function getSignalExitCode(signal: NodeJS.Signals): number { + if (signal === "SIGINT") return 130; + if (signal === "SIGTERM") return 143; + return 1; +} + +function terminateManagedChildren(signal: "SIGTERM" | "SIGKILL" = "SIGTERM"): void { + for (const [pid, child] of managedChildren) { + void killProcessTree(pid, signal); + try { + child.kill(signal); + } catch { + // Already gone. + } + } +} + +async function exitAfterManagedChildren(signal: NodeJS.Signals): Promise { + const exitCode = getSignalExitCode(signal); + const pids = getManagedChildPids(); + + terminateManagedChildren("SIGTERM"); + + const stillAlive = await waitForProcessesExit(pids, DEFAULT_GRACE_MS); + if (stillAlive.size > 0) { + terminateManagedChildren("SIGKILL"); + await waitForProcessesExit([...stillAlive], 1_000); + } + + process.exit(exitCode); +} + +function installManagedSignalHandlers(): void { + if (managedSignalHandlersInstalled || isWindows()) return; + managedSignalHandlersInstalled = true; + + const forward = (signal: NodeJS.Signals): void => { + terminateManagedChildren(); + + // Installing a signal listener disables Node's default "exit on signal" + // behaviour. If no application-level shutdown handler is present, preserve + // that default after giving managed children the same graceful + // SIGTERM→wait→SIGKILL lifecycle used by `ao stop`. + if (!daemonShutdownHandlerInstalled && !fallbackShutdownStarted) { + fallbackShutdownStarted = true; + void exitAfterManagedChildren(signal); + } + }; + + process.on("SIGINT", forward); + process.on("SIGTERM", forward); + process.on("exit", () => terminateManagedChildren()); +} + +/** + * Track a long-running daemon child in the pid registry and forward parent + * shutdown to it. If the owning process has its own shutdown handler, that + * handler remains responsible for exiting; otherwise the managed signal + * handler preserves Node's default signal-exit behaviour after forwarding. + */ +export function registerChildReaper(child: ChildProcess, role: string, command?: string): void { + if (reapedChildren.has(child)) return; + reapedChildren.add(child); + + const pid = child.pid; + if (!pid) return; + + registerDaemonChild({ pid, role, parentPid: process.pid, command }); + managedChildren.set(pid, child); + installManagedSignalHandlers(); + + const cleanup = (): void => { + managedChildren.delete(pid); + unregisterDaemonChild(pid); + }; + + child.once("exit", cleanup); + child.once("error", cleanup); +} + +/** + * The required interface for long-running subprocesses owned by the AO daemon. + * Callers get normal child_process.spawn behaviour, plus pid registry, + * signal forwarding, process-group cleanup, and registry unregister on exit. + */ +export function spawnManagedDaemonChild( + role: string, + command: string, + args: readonly string[], + options: SpawnOptions = {}, +): ChildProcess { + const child = spawn(command, [...args], options); + registerChildReaper(child, role, [command, ...args].join(" ")); + return child; +} + +export async function sweepDaemonChildren( + options: DaemonChildSweepOptions = {}, +): Promise { + const { ownerPid, graceMs = DEFAULT_GRACE_MS } = options; + const entries = getDaemonChildren().filter( + (entry) => ownerPid === undefined || entry.parentPid === ownerPid, + ); + const result: DaemonChildSweepResult = { + attempted: entries.length, + terminated: 0, + forceKilled: 0, + failed: 0, + }; + + for (const entry of entries) { + await killProcessTree(entry.pid, "SIGTERM"); + } + + const pids = entries.map((entry) => entry.pid); + const stillAliveAfterTerm = await waitForProcessesExit(pids, graceMs); + result.terminated = pids.length - stillAliveAfterTerm.size; + + for (const entry of entries.filter((entry) => stillAliveAfterTerm.has(entry.pid))) { + await killProcessTree(entry.pid, "SIGKILL"); + } + + const stillAliveAfterKill = await waitForProcessesExit([...stillAliveAfterTerm], 1_000); + result.forceKilled = stillAliveAfterTerm.size - stillAliveAfterKill.size; + result.failed = stillAliveAfterKill.size; + + pruneSweptDaemonChildren(new Set(entries.map((entry) => entry.pid))); + return result; +} + +function isAsciiWhitespace(char: string): boolean { + return char === " " || char === "\t" || char === "\r" || char === "\n"; +} + +function normalizeCommand(command: string): string { + return command.replaceAll("\\", "/").toLowerCase(); +} + +function firstCommandWord(command: string): string { + let start = 0; + while (start < command.length && isAsciiWhitespace(command[start] ?? "")) start++; + + let end = start; + while (end < command.length && !isAsciiWhitespace(command[end] ?? "")) end++; + + return command.slice(start, end); +} + +function commandLooksLikeNode(command: string): boolean { + const executable = firstCommandWord(normalizeCommand(command)); + const basename = executable.slice(executable.lastIndexOf("/") + 1); + return basename === "node" || basename === "node.exe"; +} + +export function classifyAoOrphanCommand(command: string): string | null { + if (!commandLooksLikeNode(command)) return null; + + const normalized = normalizeCommand(command); + + if ( + normalized.includes("@aoagents/ao-web") && + (normalized.includes("/dist-server/") || normalized.includes(" dist-server/")) + ) { + return "ao-web"; + } + if ( + normalized.includes("/ao lifecycle-worker ") || + normalized.includes(" ao lifecycle-worker ") + ) { + return "lifecycle-worker"; + } + if (normalized.includes("next-server")) { + return "next-server"; + } + return null; +} + +function parseLeadingUnsignedInt( + value: string, + start: number, +): { value: number; next: number } | null { + let index = start; + while (index < value.length && isAsciiWhitespace(value[index] ?? "")) index++; + + const firstDigit = index; + while (index < value.length) { + const code = value.charCodeAt(index); + if (code < 48 || code > 57) break; + index++; + } + + if (index === firstDigit) return null; + return { value: Number(value.slice(firstDigit, index)), next: index }; +} + +function parsePsLine(line: string): AoOrphanProcess | null { + const pid = parseLeadingUnsignedInt(line, 0); + if (!pid) return null; + + const ppid = parseLeadingUnsignedInt(line, pid.next); + if (!ppid) return null; + + let commandStart = ppid.next; + while (commandStart < line.length && isAsciiWhitespace(line[commandStart] ?? "")) { + commandStart++; + } + + const command = line.slice(commandStart); + if (!Number.isFinite(pid.value) || !Number.isFinite(ppid.value) || ppid.value !== 1) { + return null; + } + + const role = classifyAoOrphanCommand(command); + if (!role) return null; + return { pid: pid.value, ppid: ppid.value, command, role }; +} + +export function detectAoOrphansFromPsOutput(psOutput: string): AoOrphanProcess[] { + const orphans: AoOrphanProcess[] = []; + for (const rawLine of psOutput.replaceAll("\r\n", "\n").split("\n")) { + const line = rawLine.trimStart(); + if (!line) continue; + + const orphan = parsePsLine(line); + if (orphan) orphans.push(orphan); + } + return orphans; +} + +export async function scanAoOrphans(): Promise { + if (isWindows()) return []; + try { + const { stdout } = await execFileAsync("ps", ["-axo", "pid,ppid,command"], { + windowsHide: true, + maxBuffer: 10 * 1024 * 1024, + }); + return detectAoOrphansFromPsOutput(stdout); + } catch { + return []; + } +} + +export async function reapAoOrphans( + orphans: AoOrphanProcess[], + graceMs: number = DEFAULT_GRACE_MS, +): Promise { + const result: DaemonChildSweepResult = { + attempted: orphans.length, + terminated: 0, + forceKilled: 0, + failed: 0, + }; + + for (const orphan of orphans) { + await killProcessTree(orphan.pid, "SIGTERM"); + } + + const pids = orphans.map((orphan) => orphan.pid); + const stillAliveAfterTerm = await waitForProcessesExit(pids, graceMs); + result.terminated = pids.length - stillAliveAfterTerm.size; + + for (const orphan of orphans.filter((orphan) => stillAliveAfterTerm.has(orphan.pid))) { + await killProcessTree(orphan.pid, "SIGKILL"); + } + + const stillAliveAfterKill = await waitForProcessesExit([...stillAliveAfterTerm], 1_000); + result.forceKilled = stillAliveAfterTerm.size - stillAliveAfterKill.size; + result.failed = stillAliveAfterKill.size; + + return result; +} + +export function __getDaemonChildrenRegistryFile(): string { + return getRegistryFile(); +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 6adcdaae4..6cd385068 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -164,10 +164,7 @@ export { resetOpenCodeSessionListCache, } from "./opencode-shared.js"; export type { OpenCodeSessionListEntry } from "./opencode-shared.js"; -export { - getWorkspaceAgentsMdPath, - writeWorkspaceOpenCodeAgentsMd, -} from "./opencode-agents-md.js"; +export { getWorkspaceAgentsMdPath, writeWorkspaceOpenCodeAgentsMd } from "./opencode-agents-md.js"; export { writeOpenCodeConfig } from "./opencode-config.js"; export { getOrchestratorSessionId, @@ -276,6 +273,7 @@ export { export { isWindows, isMac, + isLinux, getDefaultRuntime, getShell, killProcessTree, @@ -412,6 +410,25 @@ export { type WindowsPtyHostEntry, } from "./windows-pty-registry.js"; +export { + registerDaemonChild, + unregisterDaemonChild, + getDaemonChildren, + clearDaemonChildrenRegistry, + markDaemonShutdownHandlerInstalled, + registerChildReaper, + spawnManagedDaemonChild, + sweepDaemonChildren, + classifyAoOrphanCommand, + detectAoOrphansFromPsOutput, + scanAoOrphans, + reapAoOrphans, + type DaemonChildEntry, + type DaemonChildSweepOptions, + type DaemonChildSweepResult, + type AoOrphanProcess, +} from "./daemon-children.js"; + // Activity event logging — structured diagnostic event trail export { recordActivityEvent, droppedEventCount } from "./activity-events.js"; export { isActivityEventsFtsEnabled, closeDb } from "./events-db.js"; diff --git a/packages/core/src/platform.ts b/packages/core/src/platform.ts index 763a953f0..5cde1af87 100644 --- a/packages/core/src/platform.ts +++ b/packages/core/src/platform.ts @@ -20,6 +20,10 @@ export function isMac(): boolean { return process.platform === "darwin"; } +export function isLinux(): boolean { + return process.platform === "linux"; +} + export function getDefaultRuntime(): "tmux" | "process" { return isWindows() ? "process" : "tmux"; } diff --git a/packages/integration-tests/src/daemon-children.integration.test.ts b/packages/integration-tests/src/daemon-children.integration.test.ts new file mode 100644 index 000000000..fd234ea0c --- /dev/null +++ b/packages/integration-tests/src/daemon-children.integration.test.ts @@ -0,0 +1,149 @@ +import { spawn, execFile } from "node:child_process"; +import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { mkdtemp, rm, realpath } from "node:fs/promises"; +import { createServer } from "node:net"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { promisify } from "node:util"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { isWindows, killProcessTree } from "@aoagents/ao-core"; +import { sleep } from "./helpers/polling.js"; + +const execFileAsync = promisify(execFile); +const __dirname = dirname(fileURLToPath(import.meta.url)); +const repoRoot = resolve(__dirname, "../../.."); +const cliEntry = join(repoRoot, "packages/cli/src/index.ts"); +const tsxBin = join(repoRoot, "packages/cli/node_modules/.bin/tsx"); +const dashboardEntry = join(repoRoot, "packages/web/dist-server/start-all.js"); + +const canRun = !isWindows() && existsSync(tsxBin) && existsSync(dashboardEntry); + +async function getFreePort(): Promise { + return new Promise((resolve, reject) => { + const server = createServer(); + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + server.close(() => { + if (address && typeof address === "object") resolve(address.port); + else reject(new Error("Could not allocate a free port")); + }); + }); + }); +} + +async function readChildPids(pid: number): Promise { + try { + const { stdout } = await execFileAsync("pgrep", ["-P", String(pid)]); + return stdout + .split(/\s+/) + .map((value) => Number(value)) + .filter((value) => Number.isFinite(value) && value > 0); + } catch { + return []; + } +} + +function isAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (err: unknown) { + return (err as { code?: string }).code === "EPERM"; + } +} + +describe.skipIf(!canRun)("daemon child reaping (integration)", () => { + let tmpHome: string; + let repoPath: string; + let configPath: string; + let startPid: number | undefined; + let port: number; + + beforeEach(async () => { + tmpHome = await realpath(await mkdtemp(join(tmpdir(), "ao-daemon-int-home-"))); + port = await getFreePort(); + repoPath = join(tmpHome, "repo"); + mkdirSync(repoPath, { recursive: true }); + await execFileAsync("git", ["init"], { cwd: repoPath }); + await execFileAsync("git", ["config", "user.email", "test@example.com"], { cwd: repoPath }); + await execFileAsync("git", ["config", "user.name", "Test User"], { cwd: repoPath }); + writeFileSync(join(repoPath, "README.md"), "# daemon child reaping\n"); + await execFileAsync("git", ["add", "."], { cwd: repoPath }); + await execFileAsync("git", ["commit", "-m", "Initial commit"], { cwd: repoPath }); + + configPath = join(repoPath, "agent-orchestrator.yaml"); + writeFileSync( + configPath, + ["runtime: process", "agent: claude-code", "workspace: worktree"].join("\n"), + ); + + const globalConfigPath = join(tmpHome, "global-agent-orchestrator.yaml"); + writeFileSync( + globalConfigPath, + [ + `port: ${port}`, + "defaults:", + " runtime: process", + " agent: claude-code", + " workspace: worktree", + " notifiers: []", + "projects:", + " daemon-int:", + " displayName: Daemon Integration", + ` path: ${JSON.stringify(repoPath)}`, + " defaultBranch: main", + " sessionPrefix: daemon-int", + ].join("\n"), + ); + configPath = globalConfigPath; + }, 30_000); + + afterEach(async () => { + if (startPid && isAlive(startPid)) { + await killProcessTree(startPid, "SIGKILL"); + } + await rm(tmpHome, { recursive: true, force: true }).catch(() => {}); + }, 30_000); + + it("ao stop terminates children spawned by ao start", async () => { + const env = { + ...process.env, + HOME: tmpHome, + AO_CALLER_TYPE: "agent", + AO_CONFIG_PATH: configPath, + AO_GLOBAL_CONFIG: configPath, + PORT: String(port), + }; + const start = spawn(tsxBin, [cliEntry, "start", "--no-orchestrator", "--reap-orphans"], { + cwd: repoPath, + env, + stdio: "ignore", + }); + startPid = start.pid; + expect(startPid).toBeTypeOf("number"); + + const runningPath = join(tmpHome, ".agent-orchestrator/running.json"); + let runningPid: number | undefined; + for (let i = 0; i < 100; i++) { + if (existsSync(runningPath)) { + const running = JSON.parse(readFileSync(runningPath, "utf-8")) as { pid?: number }; + runningPid = running.pid; + break; + } + await sleep(100); + } + expect(runningPid).toBeTypeOf("number"); + + const childPids = await readChildPids(runningPid!); + expect(childPids.length).toBeGreaterThan(0); + + await execFileAsync(tsxBin, [cliEntry, "stop", "--all"], { cwd: repoPath, env, timeout: 20_000 }); + await sleep(5_000); + + const stillAlive = childPids.filter(isAlive); + expect(stillAlive).toEqual([]); + expect(isAlive(runningPid!)).toBe(false); + }, 60_000); +}); diff --git a/packages/web/server/__tests__/server-compatibility.test.ts b/packages/web/server/__tests__/server-compatibility.test.ts index 25ffcf20d..efcfc01db 100644 --- a/packages/web/server/__tests__/server-compatibility.test.ts +++ b/packages/web/server/__tests__/server-compatibility.test.ts @@ -54,11 +54,11 @@ describe("start-all.ts", () => { it("kills child process trees during shutdown", () => { expect(source).toMatch(/killProcessTree/); - expect(source).toMatch(/detached:\s*process\.platform\s*!==\s*["']win32["']/); + expect(source).toMatch(/spawnManagedDaemonChild/); + expect(source).toMatch(/detached:\s*!\s*isWindows\(\)/); }); }); - describe("OrchestratorConfig compatibility", () => { it("OrchestratorConfig does not have dataDir property", () => { const typesSource = readFileSync( diff --git a/packages/web/server/start-all.ts b/packages/web/server/start-all.ts index dd3c3dbaf..d54b11c33 100644 --- a/packages/web/server/start-all.ts +++ b/packages/web/server/start-all.ts @@ -4,12 +4,17 @@ * Replaces the dev-only `concurrently` setup. */ -import { spawn, type ChildProcess } from "node:child_process"; +import { type ChildProcess } from "node:child_process"; import { resolve, dirname } from "node:path"; import { existsSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { createRequire } from "node:module"; -import { killProcessTree } from "@aoagents/ao-core"; +import { + isWindows, + killProcessTree, + markDaemonShutdownHandlerInstalled, + spawnManagedDaemonChild, +} from "@aoagents/ao-core"; const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); @@ -18,6 +23,7 @@ const __dirname = dirname(__filename); const pkgRoot = resolve(__dirname, ".."); const children: ChildProcess[] = []; +markDaemonShutdownHandlerInstalled(); function log(label: string, msg: string): void { process.stdout.write(`[${label}] ${msg}\n`); @@ -34,11 +40,11 @@ function spawnProcess( let slotIndex = -1; function launch(): ChildProcess { - const child = spawn(command, args, { + const child = spawnManagedDaemonChild(`dashboard:${label}`, command, args, { cwd: pkgRoot, stdio: ["ignore", "pipe", "pipe"], env: process.env, - detached: process.platform !== "win32", + detached: !isWindows(), }); child.stdout?.on("data", (data: Buffer) => { @@ -83,7 +89,7 @@ function spawnProcess( function resolveNextBin(): string { // On Windows, .bin/next is a POSIX shell shim that spawn() cannot execute. // Skip it and go straight to the JS entry point. - if (process.platform !== "win32") { + if (!isWindows()) { const localBin = resolve(pkgRoot, "node_modules", ".bin", "next"); if (existsSync(localBin)) return localBin; } @@ -103,7 +109,7 @@ function resolveNextBin(): string { const port = process.env["PORT"] || "3000"; const nextBin = resolveNextBin(); -if (process.platform === "win32" && nextBin !== "next") { +if (isWindows() && nextBin !== "next") { // On Windows, run the JS entry point via the current node binary. // spawn() can't execute .js files directly on Windows. spawnProcess("next", process.execPath, [nextBin, "start", "-p", port]); @@ -112,7 +118,9 @@ if (process.platform === "win32" && nextBin !== "next") { } // Start direct terminal WebSocket server (auto-restart on crash) -spawnProcess("direct-terminal", "node", [resolve(__dirname, "direct-terminal-ws.js")], { restart: true }); +spawnProcess("direct-terminal", "node", [resolve(__dirname, "direct-terminal-ws.js")], { + restart: true, +}); // Graceful shutdown — send SIGTERM to children and wait for them to exit let shuttingDown = false;