bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix(core): use registry for daemon orphan recovery

This commit is contained in:
i-trytoohard 2026-05-18 02:51:47 +05:30
parent 37d3a86d6d
commit caadafab3f
5 changed files with 93 additions and 322 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

111
packages/cli/__tests__/commands/start.test.ts
View File

@ -32,7 +32,6 @@ const {
mockSessionManager,
mockWaitForPortAndOpen,
mockSpawn,
mockFindPidByPort,
mockKillProcessTree,
mockSweepDaemonChildren,
mockScanAoOrphans,
@ -57,7 +56,6 @@ const {
},
mockWaitForPortAndOpen: vi.fn().mockResolvedValue(undefined),
mockSpawn: vi.fn(),
mockFindPidByPort: vi.fn(),
mockKillProcessTree: vi.fn(),
mockSweepDaemonChildren: vi.fn(),
mockScanAoOrphans: vi.fn(),
@ -148,7 +146,6 @@ vi.mock("@aoagents/ao-core", async (importOriginal) => {
if (path) return actual.loadConfig(path);
return mockConfigRef.current;
},
findPidByPort: mockFindPidByPort,
killProcessTree: mockKillProcessTree,
sweepDaemonChildren: mockSweepDaemonChildren,
scanAoOrphans: mockScanAoOrphans,
@ -408,8 +405,6 @@ beforeEach(async () => {
});
mockWaitForPortAndOpen.mockReset();
mockWaitForPortAndOpen.mockResolvedValue(undefined);
mockFindPidByPort.mockReset();
mockFindPidByPort.mockResolvedValue(null);
mockKillProcessTree.mockReset();
mockKillProcessTree.mockResolvedValue(undefined);
mockSweepDaemonChildren.mockReset();
@ -1766,20 +1761,7 @@ describe("start command — orchestrator session strategy display", () => {
// ---------------------------------------------------------------------------
describe("stop command", () => {
/** Helper: mock exec to simulate a dashboard process on a given port. */
function mockDashboardOnPort(dashboardPort: number, pid = "12345"): void {
mockExec.mockImplementation(async (cmd: string, args: string[] = []) => {
if (cmd === "kill") return { stdout: "", stderr: "" };
if (cmd === "ps") return { stdout: "node /fake/web/dist-server/start-all.js", stderr: "" };
if (cmd === "lsof") {
const portArg = args.find((a) => a.startsWith(":"));
if (portArg === `:${dashboardPort}`) return { stdout: pid, stderr: "" };
}
throw new Error("no process");
});
}
it("stops the actual numbered orchestrator session and dashboard", async () => {
it("stops the actual numbered orchestrator session", async () => {
mockConfigRef.current = makeConfig({ "my-app": makeProject() });
// Issue #1048: ao stop must look up the real numbered orchestrator id
// (e.g. app-orchestrator-3) via sm.list — never the phantom `${prefix}-orchestrator`.
@ -1795,8 +1777,6 @@ describe("stop command", () => {
},
]);
mockSessionManager.kill.mockResolvedValue({ cleaned: true, alreadyTerminated: false });
mockDashboardOnPort(3000);
await program.parseAsync(["node", "test", "stop"]);
expect(mockSessionManager.kill).toHaveBeenCalledWith("app-orchestrator-3", {
@ -1911,101 +1891,12 @@ describe("stop command", () => {
},
]);
mockSessionManager.kill.mockResolvedValue({ cleaned: true, alreadyTerminated: false });
mockDashboardOnPort(3000);
await program.parseAsync(["node", "test", "stop", "--purge-session"]);
expect(mockSessionManager.kill).toHaveBeenCalledWith("app-orchestrator", {
purgeOpenCode: true,
});
});
it("calls killProcessTree with numeric PID when findPidByPort returns a PID", async () => {
mockConfigRef.current = makeConfig({ "my-app": makeProject() });
mockSessionManager.list.mockResolvedValue([]);
mockFindPidByPort.mockResolvedValue("1234");
// killDashboardOnPort verifies the PID is an AO dashboard via `ps` on Unix
// before killing. Stub it to return a matching cmdline so we reach the kill.
mockExec.mockImplementation(async (cmd: string) => {
if (cmd === "ps") return { stdout: "node /fake/web/dist-server/start-all.js", stderr: "" };
throw new Error("no process");
});
await program.parseAsync(["node", "test", "stop"]);
expect(mockFindPidByPort).toHaveBeenCalledWith(3000);
expect(mockKillProcessTree).toHaveBeenCalledWith(1234);
});
it("does not call killProcessTree when findPidByPort returns null", async () => {
mockConfigRef.current = makeConfig({ "my-app": makeProject() });
mockSessionManager.list.mockResolvedValue([]);
mockFindPidByPort.mockResolvedValue(null);
await program.parseAsync(["node", "test", "stop"]);
expect(mockFindPidByPort).toHaveBeenCalledWith(3000);
expect(mockKillProcessTree).not.toHaveBeenCalled();
});
// Recovers from issue #645: when the configured port was busy at start, the
// dashboard auto-reassigned to port+N and `ao stop` couldn't find it. The
// port-scan fallback in stopDashboard walks port+1..port+MAX_PORT_SCAN.
// Skip on Windows: killDashboardOnPort skips the `ps` cmdline verification
// there (uses netstat trust), so the assertions on `ps` output don't apply.
it.skipIf(process.platform === "win32")(
"finds orphaned dashboard on a reassigned port via port scan",
async () => {
mockConfigRef.current = makeConfig({ "my-app": makeProject() });
mockSessionManager.list.mockResolvedValue([]);
// Port 3000 has nothing; port 3001 has the orphaned dashboard
mockFindPidByPort.mockImplementation(async (port: number) =>
port === 3001 ? "99999" : null,
);
// ps cmdline check inside killDashboardOnPort must pass for the kill to fire
mockExec.mockImplementation(async (cmd: string) => {
if (cmd === "ps") return { stdout: "node /fake/web/dist-server/start-all.js", stderr: "" };
throw new Error("no process");
});
await program.parseAsync(["node", "test", "stop"]);
expect(mockKillProcessTree).toHaveBeenCalledWith(99999);
const output = vi
.mocked(console.log)
.mock.calls.map((c) => c.join(" "))
.join("\n");
expect(output).toContain("was on port 3001");
},
);
// Windows parallel: the port-scan fallback must still find the orphaned
// dashboard, but killDashboardOnPort intentionally skips the `ps` cmdline
// check (no `ps` on Windows; we trust netstat output via findPidByPort).
// Ensures a developer who breaks the Windows port-scan path is caught.
it.runIf(process.platform === "win32")(
"finds orphaned dashboard on a reassigned port via port scan (Windows)",
async () => {
mockConfigRef.current = makeConfig({ "my-app": makeProject() });
mockSessionManager.list.mockResolvedValue([]);
mockFindPidByPort.mockImplementation(async (port: number) =>
port === 3001 ? "99999" : null,
);
await program.parseAsync(["node", "test", "stop"]);
expect(mockKillProcessTree).toHaveBeenCalledWith(99999);
// `ps` must NOT be invoked on Windows — the cmdline verification is
// skipped by design in killDashboardOnPort.
const psCalls = mockExec.mock.calls.filter((c) => c[0] === "ps");
expect(psCalls).toHaveLength(0);
const output = vi
.mocked(console.log)
.mock.calls.map((c) => c.join(" "))
.join("\n");
expect(output).toContain("was on port 3001");
},
);
});
// ---------------------------------------------------------------------------

67
packages/cli/src/commands/start.ts
View File

@ -29,7 +29,6 @@ import {
isWindows,
isMac,
isLinux,
findPidByPort,
killProcessTree,
loadLocalProjectConfigDetailed,
recordActivityEvent,
@ -45,10 +44,10 @@ import {
scanAoOrphans,
reapAoOrphans,
type DaemonChildSweepResult,
type AoOrphanProcess,
type DaemonChildOrphan,
} from "@aoagents/ao-core";
import { parse as yamlParse, stringify as yamlStringify } from "yaml";
import { exec, execSilent, git } from "../lib/shell.js";
import { 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";
@ -1253,64 +1252,6 @@ async function runStartup(
return port;
}
/**
* Stop dashboard server.
* Uses platform adapter to find the process listening on the port, then kills it.
* Best effort — if it fails, just warn the user.
*/
/** Pattern matching AO dashboard processes (production and dev mode). */
const DASHBOARD_CMD_PATTERN = /next-server|start-all\.js|next dev|ao-web/;
/**
* Check whether a process listening on the given port is an AO dashboard
* (next-server, start-all.js, or next dev). Only kills matching PIDs,
* leaving unrelated co-listeners (sidecars, SO_REUSEPORT) untouched.
*/
async function killDashboardOnPort(port: number): Promise<boolean> {
try {
const pid = await findPidByPort(port);
if (!pid) return false;
// On Unix, verify the process is actually a dashboard before killing so
// unrelated co-listeners (sidecars, SO_REUSEPORT) are left untouched.
// findPidByPort on Windows uses netstat; we trust the port match there.
if (!isWindows()) {
try {
const { stdout: cmdline } = await exec("ps", ["-p", String(pid), "-o", "args="]);
if (!DASHBOARD_CMD_PATTERN.test(cmdline)) return false;
} catch {
return false;
}
}
await killProcessTree(Number(pid));
return true;
} catch {
return false;
}
}
async function stopDashboard(port: number): Promise<void> {
// 1. Try the expected port — verify it's a dashboard before killing
if (await killDashboardOnPort(port)) {
console.log(chalk.green("Dashboard stopped"));
return;
}
// 2. Fallback: scan nearby ports to find an orphaned dashboard
// that was auto-reassigned when the original port was busy.
// Uses killDashboardOnPort to verify the process is actually an
// AO dashboard before killing, avoiding collateral damage.
for (let p = port + 1; p <= port + MAX_PORT_SCAN; p++) {
if (await killDashboardOnPort(p)) {
console.log(chalk.green(`Dashboard stopped (was on port ${p})`));
return;
}
}
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` : ""
@ -1328,7 +1269,7 @@ async function sweepRegisteredDaemonChildren(ownerPid?: number): Promise<void> {
}
}
function describeAoOrphans(orphans: AoOrphanProcess[]): string {
function describeAoOrphans(orphans: DaemonChildOrphan[]): string {
return orphans
.map((orphan) => `${orphan.pid} (${orphan.role})`)
.slice(0, 8)
@ -1928,7 +1869,6 @@ export function registerStop(program: Command): void {
projectIds[0];
project = config.projects[_projectId];
}
const port = config.port ?? DEFAULT_PORT;
if (projectArg) {
console.log(chalk.bold(`\nStopping orchestrator for ${chalk.cyan(project.name)}\n`));
@ -2128,7 +2068,6 @@ export function registerStop(program: Command): void {
} else {
await sweepRegisteredDaemonChildren();
}
await stopDashboard(running?.port ?? port);
}
// Targeted stop deliberately does NOT edit `running.json` from this
// child CLI process. The long-lived parent supervises lifecycle

94
packages/core/src/__tests__/daemon-children.test.ts
View File

@ -6,9 +6,10 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest";
import {
__getDaemonChildrenRegistryFile,
clearDaemonChildrenRegistry,
detectAoOrphansFromPsOutput,
getDaemonChildren,
registerDaemonChild,
reapAoOrphans,
scanAoOrphans,
spawnManagedDaemonChild,
sweepDaemonChildren,
unregisterDaemonChild,
@ -144,28 +145,75 @@ describe("daemon child registry", () => {
});
});
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");
describe("AO orphan recovery", () => {
let tmpHome: string;
let originalHome: string | undefined;
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" }),
]);
beforeEach(() => {
tmpHome = mkdtempSync(join(tmpdir(), "ao-daemon-orphans-"));
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("finds and reaps registered children whose owning daemon is gone", async () => {
const liveOwnerChild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 30_000)"], {
stdio: "ignore",
});
const orphanChild = spawn(process.execPath, ["-e", "setTimeout(() => {}, 30_000)"], {
stdio: "ignore",
});
try {
expect(liveOwnerChild.pid).toBeTypeOf("number");
expect(orphanChild.pid).toBeTypeOf("number");
const liveOwnerPid = liveOwnerChild.pid as number;
const orphanPid = orphanChild.pid as number;
const missingOwnerPid = 999_999_999;
registerDaemonChild({
pid: liveOwnerPid,
parentPid: process.pid,
role: "live-owner-child",
command: "node live-owner.js",
});
registerDaemonChild({
pid: orphanPid,
parentPid: missingOwnerPid,
role: "orphaned-child",
command: "node orphan.js",
});
const orphans = await scanAoOrphans();
expect(orphans).toEqual([
expect.objectContaining({
pid: orphanPid,
parentPid: missingOwnerPid,
role: "orphaned-child",
}),
]);
const result = await reapAoOrphans(orphans, 1_000);
expect(result.attempted).toBe(1);
expect(isProcessAliveForTest(orphanPid)).toBe(false);
expect(isProcessAliveForTest(liveOwnerPid)).toBe(true);
expect(getDaemonChildren()).toContainEqual(
expect.objectContaining({ pid: liveOwnerPid, role: "live-owner-child" }),
);
expect(getDaemonChildren()).not.toContainEqual(expect.objectContaining({ pid: orphanPid }));
} finally {
liveOwnerChild.kill("SIGKILL");
orphanChild.kill("SIGKILL");
await waitForChildExit(liveOwnerChild);
await waitForChildExit(orphanChild);
}
});
});

139
packages/core/src/daemon-children.ts
View File

@ -1,9 +1,4 @@
import {
type ChildProcess,
type SpawnOptions,
execFile as execFileCb,
spawn,
} from "node:child_process";
import { type ChildProcess, type SpawnOptions, spawn } from "node:child_process";
import {
existsSync,
mkdirSync,
@ -16,11 +11,9 @@ import {
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;
@ -39,11 +32,12 @@ export interface DaemonChildSweepResult {
failed: number;
}
export interface AoOrphanProcess {
export interface DaemonChildOrphan {
pid: number;
ppid: number;
command: string;
parentPid: number;
role: string;
startedAt: string;
command?: string;
}
export interface DaemonChildSweepOptions {
@ -381,120 +375,20 @@ export async function sweepDaemonChildren(
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 scanAoOrphans(): Promise<DaemonChildOrphan[]> {
return getDaemonChildren()
.filter((entry) => !isProcessAlive(entry.parentPid))
.map((entry) => ({
pid: entry.pid,
parentPid: entry.parentPid,
role: entry.role,
startedAt: entry.startedAt,
command: entry.command,
}));
}
export async function reapAoOrphans(
orphans: AoOrphanProcess[],
orphans: DaemonChildOrphan[],
graceMs: number = DEFAULT_GRACE_MS,
): Promise<DaemonChildSweepResult> {
const result: DaemonChildSweepResult = {
@ -520,6 +414,7 @@ export async function reapAoOrphans(
result.forceKilled = stillAliveAfterTerm.size - stillAliveAfterKill.size;
result.failed = stillAliveAfterKill.size;
pruneSweptDaemonChildren(new Set(pids));
return result;
}

4
packages/core/src/index.ts
View File

@ -528,14 +528,12 @@ export {
registerChildReaper,
spawnManagedDaemonChild,
sweepDaemonChildren,
classifyAoOrphanCommand,
detectAoOrphansFromPsOutput,
scanAoOrphans,
reapAoOrphans,
type DaemonChildEntry,
type DaemonChildSweepOptions,
type DaemonChildSweepResult,
type AoOrphanProcess,
type DaemonChildOrphan,
} from "./daemon-children.js";
// Activity event logging — structured diagnostic event trail