* 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>
This commit is contained in:
i-trytoohard
GitHub
parent
89ad185195
commit
7d324b537d
|
|
@ -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"],
|
||||
|
|
|
|||
|
|
@ -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<typeof vi.spyOn>;
|
||||
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",
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
|||
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<void> {
|
||||
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<void> {
|
||||
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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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..."));
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
await Promise.race([
|
||||
new Promise<void>((resolve) => child.once("exit", () => resolve())),
|
||||
new Promise<void>((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<void>((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" }),
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
|
@ -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<DaemonChildEntry>;
|
||||
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<Set<number>> {
|
||||
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<DaemonChildEntry, "startedAt">): 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<number>): 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<ChildProcess>();
|
||||
const managedChildren = new Map<number, ChildProcess>();
|
||||
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<void> {
|
||||
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<DaemonChildSweepResult> {
|
||||
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<AoOrphanProcess[]> {
|
||||
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<DaemonChildSweepResult> {
|
||||
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();
|
||||
}
|
||||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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";
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<number> {
|
||||
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<number[]> {
|
||||
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);
|
||||
});
|
||||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
Loading…
Reference in New Issue