refactor(cli): remove lifecycle-worker subprocess, poll in-process (#1186)
* refactor(cli): remove lifecycle-worker subprocess model, run polling in-process Replaces the per-project `ao lifecycle-worker` subprocess with an in-process polling loop managed inside `lifecycle-service`. All registered projects are now polled from the single long-lived `ao start` process. - Drops the `lifecycle-worker` command and its registration. - Rewrites `lifecycle-service` around a `Map<projectId, ActiveLoop>`, with SIGINT/SIGTERM/beforeExit graceful shutdown. - Removes PID-file coordination (PID file, log file, status, etc.) — these only existed to track subprocess state. - Per-project error isolation is preserved: the core lifecycle manager's `pollAll()` already catches per-cycle errors, and stop failures in one project can't prevent others from stopping. - Updates `start`/`spawn` callers and tests to drop the PID/logFile shape. - Adds `lifecycle-service.test.ts` covering idempotency, unknown projects, error isolation across projects, and graceful stop-all. Closes #1185 * fix(cli): address bugbot review on lifecycle-service in-process refactor - `ao spawn` / `ao batch-spawn` no longer call `ensureLifecycleWorker`. That call used to start `setInterval` polling in the one-shot spawn process, which (a) kept the CLI alive forever after the session spawned and (b) duplicated polling already running in `ao start`. Replaced with a `warnIfAONotRunning()` helper that checks `running.json` and prints a hint if the orchestrator isn't up. - `ao stop` no longer calls `stopLifecycleWorker`. That call always ran against a fresh in-memory map (stop is a separate process from start), so it always returned false and printed "Lifecycle worker not running" misleadingly. SIGTERM to the `ao start` PID already triggers the shared shutdown handler in `lifecycle-service`, which stops every loop. - Drop duplicated shutdown closure inside `registerSignalsOnce` — signal handlers now reference `stopAllLifecycleWorkers` directly. - Update tests accordingly: spawn.test.ts mocks `running-state.getRunning`, start.test.ts drops `stopLifecycleWorker` expectations. * fix(cli): drop SIGINT/SIGTERM listeners from lifecycle-service Installing listeners for those signals removes Node.js's default "exit on signal" behavior (per Node docs: "its default behavior will be removed — Node.js will no longer exit"). Since the registered listener doesn't call process.exit(), the `ao start` process would hang on SIGTERM with the setInterval timer keeping the event loop alive forever — effectively breaking `ao stop`. Default signal handling terminates the process cleanly; the OS reclaims the interval timer and dashboard child. `stopAllLifecycleWorkers` stays exported for callers that want explicit cleanup before exit. * fix(cli): project-scoped spawn warning + flush lifecycle health on exit Addresses reviewer feedback on PR #1186: - `warnIfAONotRunning` now takes a projectId and warns not only when no AO is running, but also when the running instance isn't polling the target project (e.g. `ao start A` then `ao spawn` in B left users silent about the fact that B wasn't being polled). - `running.json` now records only the project this `ao start` actually polls, not every project in config. Previously this list was a lie — `ensureLifecycleWorker` is called for the selected project only. - `ao start` installs SIGINT/SIGTERM handlers that call `stopAllLifecycleWorkers()` (flushing per-project "stopped" health state) and then `process.exit()`. Installing the handler safely requires an explicit exit because SIGINT/SIGTERM listeners remove Node's default exit behavior. * fix(cli): remove dead stopLifecycleWorker, add missing test mock Addresses bugbot comments on PR #1186: - Delete `stopLifecycleWorker` from `lifecycle-service.ts`. It was only called from `ao stop` in the old subprocess model; in the in-process model, SIGTERM to the `ao start` pid + the shutdown handler in `start.ts` covers cleanup. No production caller remains. - Add `stopAllLifecycleWorkers: vi.fn()` to `start.test.ts`'s `lifecycle-service.js` mock. Without it, `vi.mock` replaced the module and the named import resolved to undefined; the shutdown handler's try/catch would silently swallow the resulting TypeError, hiding any regression in how shutdown is wired. - Update `lifecycle-service.test.ts` to drop references to the removed export (one test removed, one repurposed as a no-op smoke test for `stopAllLifecycleWorkers` against an empty active map).
This commit is contained in:
Harsh Batheja
GitHub
parent
3540e512b6
commit
95fbecc379
|
|
@ -4,7 +4,7 @@ import { join } from "node:path";
|
|||
import { tmpdir } from "node:os";
|
||||
import { type Session, type SessionManager, getProjectBaseDir } from "@aoagents/ao-core";
|
||||
|
||||
const { mockExec, mockConfigRef, mockSessionManager, mockEnsureLifecycleWorker } = vi.hoisted(
|
||||
const { mockExec, mockConfigRef, mockSessionManager, mockGetRunning } = vi.hoisted(
|
||||
() => ({
|
||||
mockExec: vi.fn(),
|
||||
mockConfigRef: { current: null as Record<string, unknown> | null },
|
||||
|
|
@ -18,7 +18,7 @@ const { mockExec, mockConfigRef, mockSessionManager, mockEnsureLifecycleWorker }
|
|||
send: vi.fn(),
|
||||
claimPR: vi.fn(),
|
||||
},
|
||||
mockEnsureLifecycleWorker: vi.fn(),
|
||||
mockGetRunning: vi.fn(),
|
||||
}),
|
||||
);
|
||||
|
||||
|
|
@ -56,8 +56,8 @@ vi.mock("../../src/lib/create-session-manager.js", () => ({
|
|||
getSessionManager: async (): Promise<SessionManager> => mockSessionManager as SessionManager,
|
||||
}));
|
||||
|
||||
vi.mock("../../src/lib/lifecycle-service.js", () => ({
|
||||
ensureLifecycleWorker: (...args: unknown[]) => mockEnsureLifecycleWorker(...args),
|
||||
vi.mock("../../src/lib/running-state.js", () => ({
|
||||
getRunning: () => mockGetRunning(),
|
||||
}));
|
||||
|
||||
vi.mock("../../src/lib/metadata.js", () => ({
|
||||
|
|
@ -121,14 +121,8 @@ beforeEach(() => {
|
|||
mockSessionManager.spawn.mockReset();
|
||||
mockSessionManager.claimPR.mockReset();
|
||||
mockExec.mockReset();
|
||||
mockEnsureLifecycleWorker.mockReset();
|
||||
mockEnsureLifecycleWorker.mockResolvedValue({
|
||||
running: true,
|
||||
started: true,
|
||||
pid: 12345,
|
||||
pidFile: "/tmp/lifecycle-worker.pid",
|
||||
logFile: "/tmp/lifecycle-worker.log",
|
||||
});
|
||||
mockGetRunning.mockReset();
|
||||
mockGetRunning.mockResolvedValue({ pid: 1234, port: 3000, startedAt: "", projects: ["my-app"] });
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
|
|
@ -165,11 +159,6 @@ describe("spawn command", () => {
|
|||
// Single arg = issue; project is auto-detected (only one project in config)
|
||||
await program.parseAsync(["node", "test", "spawn", "INT-100"]);
|
||||
|
||||
expect(mockEnsureLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"my-app",
|
||||
);
|
||||
|
||||
expect(mockSessionManager.spawn).toHaveBeenCalledWith({
|
||||
projectId: "my-app",
|
||||
issueId: "INT-100",
|
||||
|
|
@ -248,10 +237,6 @@ describe("spawn command", () => {
|
|||
|
||||
await program.parseAsync(["node", "test", "spawn", "INT-42"]);
|
||||
|
||||
expect(mockEnsureLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"backend",
|
||||
);
|
||||
expect(mockSessionManager.spawn).toHaveBeenCalledWith({
|
||||
projectId: "backend",
|
||||
issueId: "INT-42",
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ const {
|
|||
mockWaitForPortAndOpen,
|
||||
mockSpawn,
|
||||
mockEnsureLifecycleWorker,
|
||||
mockStopLifecycleWorker,
|
||||
} = vi.hoisted(() => ({
|
||||
mockExec: vi.fn(),
|
||||
mockExecSilent: vi.fn(),
|
||||
|
|
@ -42,7 +41,6 @@ const {
|
|||
mockWaitForPortAndOpen: vi.fn().mockResolvedValue(undefined),
|
||||
mockSpawn: vi.fn(),
|
||||
mockEnsureLifecycleWorker: vi.fn(),
|
||||
mockStopLifecycleWorker: vi.fn(),
|
||||
}));
|
||||
|
||||
const { mockDetectOpenClawInstallation } = vi.hoisted(() => ({
|
||||
|
|
@ -101,7 +99,7 @@ vi.mock("../../src/lib/create-session-manager.js", () => ({
|
|||
|
||||
vi.mock("../../src/lib/lifecycle-service.js", () => ({
|
||||
ensureLifecycleWorker: (...args: unknown[]) => mockEnsureLifecycleWorker(...args),
|
||||
stopLifecycleWorker: (...args: unknown[]) => mockStopLifecycleWorker(...args),
|
||||
stopAllLifecycleWorkers: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../../src/lib/web-dir.js", () => ({
|
||||
|
|
@ -238,12 +236,7 @@ beforeEach(() => {
|
|||
mockEnsureLifecycleWorker.mockResolvedValue({
|
||||
running: true,
|
||||
started: true,
|
||||
pid: 12345,
|
||||
pidFile: "/tmp/lifecycle-worker.pid",
|
||||
logFile: "/tmp/lifecycle-worker.log",
|
||||
});
|
||||
mockStopLifecycleWorker.mockReset();
|
||||
mockStopLifecycleWorker.mockResolvedValue(true);
|
||||
mockDetectOpenClawInstallation.mockReset();
|
||||
mockDetectOpenClawInstallation.mockResolvedValue({
|
||||
state: "missing",
|
||||
|
|
@ -1039,10 +1032,6 @@ describe("stop command", () => {
|
|||
expect(mockSessionManager.kill).toHaveBeenCalledWith("app-orchestrator", {
|
||||
purgeOpenCode: false,
|
||||
});
|
||||
expect(mockStopLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"my-app",
|
||||
);
|
||||
const output = vi
|
||||
.mocked(console.log)
|
||||
.mock.calls.map((c) => c.join(" "))
|
||||
|
|
@ -1058,10 +1047,6 @@ describe("stop command", () => {
|
|||
await program.parseAsync(["node", "test", "stop"]);
|
||||
|
||||
expect(mockSessionManager.kill).not.toHaveBeenCalled();
|
||||
expect(mockStopLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"my-app",
|
||||
);
|
||||
const output = vi
|
||||
.mocked(console.log)
|
||||
.mock.calls.map((c) => c.join(" "))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,147 @@
|
|||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { LifecycleManager, OrchestratorConfig } from "@aoagents/ao-core";
|
||||
|
||||
const mockGetLifecycleManager = vi.fn();
|
||||
|
||||
vi.mock("../../src/lib/create-session-manager.js", () => ({
|
||||
getLifecycleManager: (...args: unknown[]) => mockGetLifecycleManager(...args),
|
||||
}));
|
||||
|
||||
// Import after mocks
|
||||
import {
|
||||
ensureLifecycleWorker,
|
||||
stopAllLifecycleWorkers,
|
||||
isLifecycleWorkerRunning,
|
||||
listLifecycleWorkers,
|
||||
} from "../../src/lib/lifecycle-service.js";
|
||||
|
||||
function makeConfig(projectIds: string[]): OrchestratorConfig {
|
||||
return {
|
||||
configPath: "/tmp/agent-orchestrator.yaml",
|
||||
port: 3000,
|
||||
readyThresholdMs: 300_000,
|
||||
defaults: { runtime: "tmux", agent: "claude-code", workspace: "worktree", notifiers: [] },
|
||||
projects: Object.fromEntries(
|
||||
projectIds.map((id) => [id, { name: id, repo: "", path: "/tmp", defaultBranch: "main" }]),
|
||||
),
|
||||
notifiers: {},
|
||||
notificationRouting: {},
|
||||
reactions: {},
|
||||
} as OrchestratorConfig;
|
||||
}
|
||||
|
||||
function makeFakeLifecycle(overrides?: Partial<LifecycleManager>): LifecycleManager & {
|
||||
start: ReturnType<typeof vi.fn>;
|
||||
stop: ReturnType<typeof vi.fn>;
|
||||
} {
|
||||
const start = vi.fn();
|
||||
const stop = vi.fn();
|
||||
return { start, stop, ...overrides } as unknown as LifecycleManager & {
|
||||
start: typeof start;
|
||||
stop: typeof stop;
|
||||
};
|
||||
}
|
||||
|
||||
describe("lifecycle-service", () => {
|
||||
beforeEach(() => {
|
||||
stopAllLifecycleWorkers();
|
||||
mockGetLifecycleManager.mockReset();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
stopAllLifecycleWorkers();
|
||||
});
|
||||
|
||||
it("starts polling in-process for a known project", async () => {
|
||||
const lifecycle = makeFakeLifecycle();
|
||||
mockGetLifecycleManager.mockResolvedValue(lifecycle);
|
||||
|
||||
const result = await ensureLifecycleWorker(makeConfig(["app"]), "app", 1000);
|
||||
|
||||
expect(result).toEqual({ running: true, started: true });
|
||||
expect(lifecycle.start).toHaveBeenCalledWith(1000);
|
||||
expect(isLifecycleWorkerRunning("app")).toBe(true);
|
||||
});
|
||||
|
||||
it("is idempotent — second ensure is a no-op", async () => {
|
||||
const lifecycle = makeFakeLifecycle();
|
||||
mockGetLifecycleManager.mockResolvedValue(lifecycle);
|
||||
|
||||
const first = await ensureLifecycleWorker(makeConfig(["app"]), "app");
|
||||
const second = await ensureLifecycleWorker(makeConfig(["app"]), "app");
|
||||
|
||||
expect(first.started).toBe(true);
|
||||
expect(second.started).toBe(false);
|
||||
expect(lifecycle.start).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("throws on unknown projects", async () => {
|
||||
await expect(
|
||||
ensureLifecycleWorker(makeConfig(["app"]), "missing"),
|
||||
).rejects.toThrow(/Unknown project/);
|
||||
});
|
||||
|
||||
it("isolates errors: one project failing to start does not affect another", async () => {
|
||||
const healthy = makeFakeLifecycle();
|
||||
mockGetLifecycleManager.mockImplementation(async (_config, projectId: string) => {
|
||||
if (projectId === "broken") {
|
||||
throw new Error("boom — broken project plugin");
|
||||
}
|
||||
return healthy;
|
||||
});
|
||||
|
||||
const config = makeConfig(["healthy", "broken"]);
|
||||
|
||||
await expect(ensureLifecycleWorker(config, "broken")).rejects.toThrow(/boom/);
|
||||
expect(isLifecycleWorkerRunning("broken")).toBe(false);
|
||||
|
||||
const result = await ensureLifecycleWorker(config, "healthy");
|
||||
expect(result.started).toBe(true);
|
||||
expect(healthy.start).toHaveBeenCalledTimes(1);
|
||||
expect(isLifecycleWorkerRunning("healthy")).toBe(true);
|
||||
});
|
||||
|
||||
it("stopAllLifecycleWorkers is a no-op when nothing is active", () => {
|
||||
expect(() => stopAllLifecycleWorkers()).not.toThrow();
|
||||
expect(listLifecycleWorkers()).toEqual([]);
|
||||
});
|
||||
|
||||
it("stopAllLifecycleWorkers stops every registered project", async () => {
|
||||
const a = makeFakeLifecycle();
|
||||
const b = makeFakeLifecycle();
|
||||
mockGetLifecycleManager.mockImplementation(async (_cfg, projectId: string) =>
|
||||
projectId === "a" ? a : b,
|
||||
);
|
||||
|
||||
const config = makeConfig(["a", "b"]);
|
||||
await ensureLifecycleWorker(config, "a");
|
||||
await ensureLifecycleWorker(config, "b");
|
||||
|
||||
expect(listLifecycleWorkers().sort()).toEqual(["a", "b"]);
|
||||
|
||||
stopAllLifecycleWorkers();
|
||||
|
||||
expect(a.stop).toHaveBeenCalledTimes(1);
|
||||
expect(b.stop).toHaveBeenCalledTimes(1);
|
||||
expect(listLifecycleWorkers()).toEqual([]);
|
||||
});
|
||||
|
||||
it("a throwing stop on one project does not prevent others from stopping", async () => {
|
||||
const broken = makeFakeLifecycle();
|
||||
(broken.stop as ReturnType<typeof vi.fn>).mockImplementation(() => {
|
||||
throw new Error("stop failed");
|
||||
});
|
||||
const healthy = makeFakeLifecycle();
|
||||
mockGetLifecycleManager.mockImplementation(async (_cfg, projectId: string) =>
|
||||
projectId === "broken" ? broken : healthy,
|
||||
);
|
||||
|
||||
const config = makeConfig(["broken", "healthy"]);
|
||||
await ensureLifecycleWorker(config, "broken");
|
||||
await ensureLifecycleWorker(config, "healthy");
|
||||
|
||||
expect(() => stopAllLifecycleWorkers()).not.toThrow();
|
||||
expect(healthy.stop).toHaveBeenCalledTimes(1);
|
||||
expect(listLifecycleWorkers()).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,139 +0,0 @@
|
|||
import type { Command } from "commander";
|
||||
import chalk from "chalk";
|
||||
import { createCorrelationId, createProjectObserver, loadConfig } from "@aoagents/ao-core";
|
||||
import { getLifecycleManager } from "../lib/create-session-manager.js";
|
||||
import {
|
||||
clearLifecycleWorkerPid,
|
||||
getLifecycleWorkerStatus,
|
||||
writeLifecycleWorkerPid,
|
||||
} from "../lib/lifecycle-service.js";
|
||||
|
||||
function parseInterval(value: string): number {
|
||||
const parsed = Number.parseInt(value, 10);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : 30_000;
|
||||
}
|
||||
|
||||
export function registerLifecycleWorker(program: Command): void {
|
||||
program
|
||||
.command("lifecycle-worker")
|
||||
.description("Internal lifecycle polling worker")
|
||||
.argument("<project>", "Project ID from config")
|
||||
.option("--interval-ms <ms>", "Polling interval in milliseconds", "30000")
|
||||
.action(async (projectId: string, opts: { intervalMs?: string }) => {
|
||||
const config = loadConfig();
|
||||
const observer = createProjectObserver(config, "lifecycle-worker");
|
||||
if (!config.projects[projectId]) {
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: "error",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
reason: `Unknown project: ${projectId}`,
|
||||
details: { projectId },
|
||||
});
|
||||
console.error(chalk.red(`Unknown project: ${projectId}`));
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const existing = getLifecycleWorkerStatus(config, projectId);
|
||||
if (existing.running && existing.pid !== process.pid) {
|
||||
// Another lifecycle worker is already running for this project — exit
|
||||
// silently to avoid duplicate polling loops.
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: "warn",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
reason: `Worker already running with pid ${existing.pid}`,
|
||||
details: { projectId, pid: existing.pid },
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const lifecycle = await getLifecycleManager(config, projectId);
|
||||
const intervalMs = parseInterval(opts.intervalMs ?? "30000");
|
||||
let shuttingDown = false;
|
||||
let heartbeat: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
const shutdown = (code: number): void => {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
if (heartbeat) clearInterval(heartbeat);
|
||||
lifecycle.stop();
|
||||
clearLifecycleWorkerPid(config, projectId, process.pid);
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: code === 0 ? "warn" : "error",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
reason: code === 0 ? "Worker stopped" : "Worker exited unexpectedly",
|
||||
details: { projectId, pid: process.pid, exitCode: code },
|
||||
});
|
||||
// Flush stdout/stderr before exiting so crash messages reach the log file
|
||||
const done = (): void => process.exit(code);
|
||||
if (process.stdout.writableFinished && process.stderr.writableFinished) {
|
||||
done();
|
||||
} else {
|
||||
let flushed = 0;
|
||||
const tryExit = (): void => {
|
||||
flushed++;
|
||||
if (flushed >= 2) done();
|
||||
};
|
||||
process.stdout.write("", tryExit);
|
||||
process.stderr.write("", tryExit);
|
||||
// Hard exit if flush hangs
|
||||
setTimeout(done, 1_000).unref();
|
||||
}
|
||||
};
|
||||
|
||||
process.on("SIGINT", () => shutdown(0));
|
||||
process.on("SIGTERM", () => shutdown(0));
|
||||
process.on("uncaughtException", (err) => {
|
||||
observer.recordOperation({
|
||||
metric: "lifecycle_poll",
|
||||
operation: "lifecycle.worker_crash",
|
||||
outcome: "failure",
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
projectId,
|
||||
reason: err instanceof Error ? err.message : String(err),
|
||||
level: "error",
|
||||
});
|
||||
shutdown(1);
|
||||
});
|
||||
process.on("unhandledRejection", (reason) => {
|
||||
observer.recordOperation({
|
||||
metric: "lifecycle_poll",
|
||||
operation: "lifecycle.worker_rejection",
|
||||
outcome: "failure",
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
projectId,
|
||||
reason: reason instanceof Error ? reason.message : String(reason),
|
||||
level: "error",
|
||||
});
|
||||
shutdown(1);
|
||||
});
|
||||
|
||||
writeLifecycleWorkerPid(config, projectId, process.pid);
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: "ok",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
details: { projectId, pid: process.pid, intervalMs },
|
||||
});
|
||||
|
||||
// Periodic heartbeat so we can verify the worker is alive from the log
|
||||
heartbeat = setInterval(() => {
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: "ok",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-worker"),
|
||||
details: { projectId, pid: process.pid, intervalMs, heartbeat: true },
|
||||
});
|
||||
}, 5 * 60_000); // every 5 minutes
|
||||
heartbeat.unref();
|
||||
|
||||
lifecycle.start(intervalMs);
|
||||
});
|
||||
}
|
||||
|
|
@ -11,9 +11,9 @@ import { DEFAULT_PORT } from "../lib/constants.js";
|
|||
import { exec } from "../lib/shell.js";
|
||||
import { banner } from "../lib/format.js";
|
||||
import { getSessionManager } from "../lib/create-session-manager.js";
|
||||
import { ensureLifecycleWorker } from "../lib/lifecycle-service.js";
|
||||
import { preflight } from "../lib/preflight.js";
|
||||
import { findProjectForDirectory } from "../lib/project-resolution.js";
|
||||
import { getRunning } from "../lib/running-state.js";
|
||||
|
||||
/**
|
||||
* Auto-detect the project ID from the config.
|
||||
|
|
@ -54,6 +54,32 @@ interface SpawnClaimOptions {
|
|||
assignOnGithub?: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Lifecycle polling runs in-process inside the long-lived `ao start` process.
|
||||
* `ao spawn` is a one-shot CLI — it can't start polling in its own process
|
||||
* (the interval would keep the CLI alive forever and duplicate work). Warn
|
||||
* when no `ao start` is running, or when the running instance isn't covering
|
||||
* this project (e.g. `ao start A` then `ao spawn` in B).
|
||||
*/
|
||||
async function warnIfAONotRunning(projectId: string): Promise<void> {
|
||||
const running = await getRunning();
|
||||
if (!running) {
|
||||
console.log(
|
||||
chalk.yellow(
|
||||
"⚠ AO is not running — lifecycle polling is inactive. Run `ao start` so the new session is tracked.",
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (!running.projects.includes(projectId)) {
|
||||
console.log(
|
||||
chalk.yellow(
|
||||
`⚠ The running AO instance (pid ${running.pid}) is not polling project "${projectId}". Run \`ao start ${projectId}\` so the new session is tracked.`,
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run pre-flight checks for a project once, before any sessions are spawned.
|
||||
* Validates runtime and tracker prerequisites so failures surface immediately
|
||||
|
|
@ -230,7 +256,7 @@ export function registerSpawn(program: Command): void {
|
|||
|
||||
try {
|
||||
await runSpawnPreflight(config, projectId, claimOptions);
|
||||
await ensureLifecycleWorker(config, projectId);
|
||||
await warnIfAONotRunning(projectId);
|
||||
|
||||
await spawnSession(config, projectId, issueId, opts.open, opts.agent, claimOptions, opts.prompt);
|
||||
} catch (err) {
|
||||
|
|
@ -276,7 +302,7 @@ export function registerBatchSpawn(program: Command): void {
|
|||
// Pre-flight once before the loop so a missing prerequisite fails fast
|
||||
try {
|
||||
await runSpawnPreflight(config, projectId);
|
||||
await ensureLifecycleWorker(config, projectId);
|
||||
await warnIfAONotRunning(projectId);
|
||||
} catch (err) {
|
||||
console.error(chalk.red(`✗ ${err instanceof Error ? err.message : String(err)}`));
|
||||
process.exit(1);
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ import {
|
|||
import { parse as yamlParse, stringify as yamlStringify } from "yaml";
|
||||
import { exec, execSilent, git } from "../lib/shell.js";
|
||||
import { getSessionManager } from "../lib/create-session-manager.js";
|
||||
import { ensureLifecycleWorker, stopLifecycleWorker } from "../lib/lifecycle-service.js";
|
||||
import { ensureLifecycleWorker, stopAllLifecycleWorkers } from "../lib/lifecycle-service.js";
|
||||
import {
|
||||
findWebDir,
|
||||
buildDashboardEnv,
|
||||
|
|
@ -993,8 +993,8 @@ async function runStartup(
|
|||
lifecycleStatus = await ensureLifecycleWorker(config, projectId);
|
||||
spinner.succeed(
|
||||
lifecycleStatus.started
|
||||
? `Lifecycle worker started${lifecycleStatus.pid ? ` (PID ${lifecycleStatus.pid})` : ""}`
|
||||
: `Lifecycle worker already running${lifecycleStatus.pid ? ` (PID ${lifecycleStatus.pid})` : ""}`,
|
||||
? "Lifecycle polling started"
|
||||
: "Lifecycle polling already running",
|
||||
);
|
||||
} catch (err) {
|
||||
spinner.fail("Lifecycle worker failed to start");
|
||||
|
|
@ -1090,10 +1090,7 @@ async function runStartup(
|
|||
|
||||
if (shouldStartLifecycle && lifecycleStatus) {
|
||||
const lifecycleLabel = lifecycleStatus.started ? "started" : "already running";
|
||||
const lifecycleTarget = lifecycleStatus.pid
|
||||
? `${lifecycleLabel} (PID ${lifecycleStatus.pid})`
|
||||
: lifecycleLabel;
|
||||
console.log(chalk.cyan("Lifecycle:"), lifecycleTarget);
|
||||
console.log(chalk.cyan("Lifecycle:"), lifecycleLabel);
|
||||
}
|
||||
|
||||
if (hasExistingOrchestrators) {
|
||||
|
|
@ -1370,13 +1367,35 @@ export function registerStart(program: Command): void {
|
|||
const actualPort = await runStartup(config, projectId, project, opts);
|
||||
|
||||
// ── Register in running.json (Step 11) ──
|
||||
// Only record the project this invocation actually polls. Other
|
||||
// configured projects are not covered by this lifecycle loop, and
|
||||
// `ao spawn` relies on this list to decide whether to warn users.
|
||||
await register({
|
||||
pid: process.pid,
|
||||
configPath: config.configPath,
|
||||
port: actualPort,
|
||||
startedAt: new Date().toISOString(),
|
||||
projects: Object.keys(config.projects),
|
||||
projects: [projectId],
|
||||
});
|
||||
|
||||
// Install shutdown handlers so `ao stop` (which sends SIGTERM to
|
||||
// this pid) flushes lifecycle health state before exit. Handlers
|
||||
// MUST call process.exit() — installing a SIGINT/SIGTERM listener
|
||||
// removes Node's default exit behavior, so without an explicit
|
||||
// exit the interval timer would keep the event loop alive.
|
||||
let shuttingDown = false;
|
||||
const shutdown = (signal: NodeJS.Signals): void => {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
try {
|
||||
stopAllLifecycleWorkers();
|
||||
} catch {
|
||||
// Best-effort cleanup — never block shutdown on observability.
|
||||
}
|
||||
process.exit(signal === "SIGINT" ? 130 : 0);
|
||||
};
|
||||
process.once("SIGINT", shutdown);
|
||||
process.once("SIGTERM", shutdown);
|
||||
} catch (err) {
|
||||
if (err instanceof Error) {
|
||||
console.error(chalk.red("\nError:"), err.message);
|
||||
|
|
@ -1451,12 +1470,11 @@ export function registerStop(program: Command): void {
|
|||
console.log(chalk.yellow(`Orchestrator session "${sessionId}" is not running`));
|
||||
}
|
||||
|
||||
const lifecycleStopped = await stopLifecycleWorker(config, _projectId);
|
||||
if (lifecycleStopped) {
|
||||
console.log(chalk.green("Lifecycle worker stopped"));
|
||||
} else {
|
||||
console.log(chalk.yellow("Lifecycle worker not running"));
|
||||
}
|
||||
// Lifecycle polling runs in-process inside the `ao start` process
|
||||
// (registered via `running.json`). Sending SIGTERM to that PID below
|
||||
// triggers the shared shutdown handler in `lifecycle-service`, which
|
||||
// stops every per-project loop. No explicit stop call needed here —
|
||||
// this CLI invocation is a separate process with an empty active map.
|
||||
|
||||
// Stop dashboard — kill parent PID from running.json, then also stop
|
||||
// any dashboard child process via lsof (parent SIGTERM may not propagate)
|
||||
|
|
|
|||
|
|
@ -1,254 +1,97 @@
|
|||
import { spawn } from "node:child_process";
|
||||
import {
|
||||
closeSync,
|
||||
existsSync,
|
||||
mkdirSync,
|
||||
openSync,
|
||||
readFileSync,
|
||||
unlinkSync,
|
||||
writeFileSync,
|
||||
} from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { getProjectBaseDir, type OrchestratorConfig } from "@aoagents/ao-core";
|
||||
createCorrelationId,
|
||||
createProjectObserver,
|
||||
type LifecycleManager,
|
||||
type OrchestratorConfig,
|
||||
} from "@aoagents/ao-core";
|
||||
import { getLifecycleManager } from "./create-session-manager.js";
|
||||
|
||||
const LIFECYCLE_PID_FILE = "lifecycle-worker.pid";
|
||||
const LIFECYCLE_LOG_FILE = "lifecycle-worker.log";
|
||||
const DEFAULT_START_TIMEOUT_MS = 5_000;
|
||||
const STOP_TIMEOUT_MS = 5_000;
|
||||
const DEFAULT_INTERVAL_MS = 30_000;
|
||||
|
||||
interface ActiveLoop {
|
||||
lifecycle: LifecycleManager;
|
||||
stop: () => void;
|
||||
}
|
||||
|
||||
const active = new Map<string, ActiveLoop>();
|
||||
|
||||
// Note: no SIGINT/SIGTERM listeners are installed here. Adding a listener for
|
||||
// those signals removes Node.js's default "exit on signal" behavior, which
|
||||
// would leave `ao start` hanging when `ao stop` sends SIGTERM (the setInterval
|
||||
// keeps the event loop alive forever). Default signal handling terminates the
|
||||
// process cleanly; the OS reclaims the interval timer. Callers that need to
|
||||
// flush state explicitly before exit can call `stopAllLifecycleWorkers()`.
|
||||
|
||||
export interface LifecycleWorkerStatus {
|
||||
running: boolean;
|
||||
pid: number | null;
|
||||
pidFile: string;
|
||||
logFile: string;
|
||||
}
|
||||
|
||||
function getProjectBase(config: OrchestratorConfig, projectId: string): string {
|
||||
const project = config.projects[projectId];
|
||||
if (!project) {
|
||||
throw new Error(`Unknown project: ${projectId}`);
|
||||
}
|
||||
return getProjectBaseDir(config.configPath, project.path);
|
||||
}
|
||||
|
||||
export function getLifecyclePidFile(config: OrchestratorConfig, projectId: string): string {
|
||||
return join(getProjectBase(config, projectId), LIFECYCLE_PID_FILE);
|
||||
}
|
||||
|
||||
export function getLifecycleLogFile(config: OrchestratorConfig, projectId: string): string {
|
||||
return join(getProjectBase(config, projectId), LIFECYCLE_LOG_FILE);
|
||||
}
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
function isProcessRunning(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch (err) {
|
||||
if (
|
||||
err &&
|
||||
typeof err === "object" &&
|
||||
"code" in err &&
|
||||
(err as NodeJS.ErrnoException).code === "EPERM"
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function readPid(pidFile: string): number | null {
|
||||
if (!existsSync(pidFile)) return null;
|
||||
|
||||
try {
|
||||
const raw = readFileSync(pidFile, "utf-8").trim();
|
||||
const pid = Number.parseInt(raw, 10);
|
||||
return Number.isFinite(pid) && pid > 0 ? pid : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export function writeLifecycleWorkerPid(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
pid: number,
|
||||
): void {
|
||||
const pidFile = getLifecyclePidFile(config, projectId);
|
||||
mkdirSync(getProjectBase(config, projectId), { recursive: true });
|
||||
writeFileSync(pidFile, `${pid}\n`, "utf-8");
|
||||
}
|
||||
|
||||
export function clearLifecycleWorkerPid(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
pid?: number,
|
||||
): void {
|
||||
const pidFile = getLifecyclePidFile(config, projectId);
|
||||
if (!existsSync(pidFile)) return;
|
||||
|
||||
if (pid !== undefined) {
|
||||
const currentPid = readPid(pidFile);
|
||||
if (currentPid !== null && currentPid !== pid) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
unlinkSync(pidFile);
|
||||
} catch {
|
||||
// Best effort cleanup
|
||||
}
|
||||
}
|
||||
|
||||
export function getLifecycleWorkerStatus(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
): LifecycleWorkerStatus {
|
||||
const pidFile = getLifecyclePidFile(config, projectId);
|
||||
const logFile = getLifecycleLogFile(config, projectId);
|
||||
const pid = readPid(pidFile);
|
||||
|
||||
if (pid !== null && isProcessRunning(pid)) {
|
||||
return { running: true, pid, pidFile, logFile };
|
||||
}
|
||||
|
||||
if (pid !== null) {
|
||||
clearLifecycleWorkerPid(config, projectId, pid);
|
||||
}
|
||||
|
||||
return { running: false, pid: null, pidFile, logFile };
|
||||
}
|
||||
|
||||
function resolveLifecycleWorkerLaunch(projectId: string): { command: string; args: string[] } {
|
||||
const entry = process.argv[1];
|
||||
const workerArgs = ["lifecycle-worker", projectId];
|
||||
|
||||
if (entry && /\.(?:c|m)?js$/i.test(entry)) {
|
||||
return {
|
||||
command: process.execPath,
|
||||
args: [entry, ...workerArgs],
|
||||
};
|
||||
}
|
||||
|
||||
if (entry && /\.ts$/i.test(entry)) {
|
||||
return {
|
||||
command: "npx",
|
||||
args: ["tsx", entry, ...workerArgs],
|
||||
};
|
||||
}
|
||||
|
||||
return {
|
||||
command: "ao",
|
||||
args: workerArgs,
|
||||
};
|
||||
}
|
||||
|
||||
async function waitForLifecycleWorker(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
timeoutMs = DEFAULT_START_TIMEOUT_MS,
|
||||
): Promise<LifecycleWorkerStatus> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
|
||||
while (Date.now() < deadline) {
|
||||
const status = getLifecycleWorkerStatus(config, projectId);
|
||||
if (status.running) {
|
||||
return status;
|
||||
}
|
||||
await sleep(100);
|
||||
}
|
||||
|
||||
return getLifecycleWorkerStatus(config, projectId);
|
||||
started: boolean;
|
||||
}
|
||||
|
||||
export async function ensureLifecycleWorker(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
): Promise<LifecycleWorkerStatus & { started: boolean }> {
|
||||
const current = getLifecycleWorkerStatus(config, projectId);
|
||||
if (current.running) {
|
||||
return { ...current, started: false };
|
||||
intervalMs: number = DEFAULT_INTERVAL_MS,
|
||||
): Promise<LifecycleWorkerStatus> {
|
||||
if (!config.projects[projectId]) {
|
||||
throw new Error(`Unknown project: ${projectId}`);
|
||||
}
|
||||
|
||||
const baseDir = getProjectBase(config, projectId);
|
||||
const logFile = getLifecycleLogFile(config, projectId);
|
||||
mkdirSync(baseDir, { recursive: true });
|
||||
|
||||
const stdoutFd = openSync(logFile, "a");
|
||||
const stderrFd = openSync(logFile, "a");
|
||||
|
||||
try {
|
||||
const launch = resolveLifecycleWorkerLaunch(projectId);
|
||||
const child = spawn(launch.command, launch.args, {
|
||||
cwd: process.cwd(),
|
||||
detached: true,
|
||||
stdio: ["ignore", stdoutFd, stderrFd],
|
||||
env: {
|
||||
...process.env,
|
||||
AO_LIFECYCLE_PROJECT: projectId,
|
||||
AO_CONFIG_PATH: config.configPath,
|
||||
},
|
||||
});
|
||||
|
||||
child.unref();
|
||||
|
||||
// Write PID from the parent immediately after spawn to close the TOCTOU
|
||||
// window: without this, a second concurrent `ensureLifecycleWorker` call
|
||||
// could pass the "not running" check before the child writes its own PID.
|
||||
if (child.pid) {
|
||||
writeLifecycleWorkerPid(config, projectId, child.pid);
|
||||
}
|
||||
} finally {
|
||||
closeSync(stdoutFd);
|
||||
closeSync(stderrFd);
|
||||
if (active.has(projectId)) {
|
||||
return { running: true, started: false };
|
||||
}
|
||||
|
||||
const status = await waitForLifecycleWorker(config, projectId);
|
||||
if (!status.running) {
|
||||
throw new Error(
|
||||
`Lifecycle worker failed to start for project ${projectId}. See ${status.logFile}`,
|
||||
);
|
||||
}
|
||||
const observer = createProjectObserver(config, "lifecycle-service");
|
||||
const lifecycle = await getLifecycleManager(config, projectId);
|
||||
|
||||
return { ...status, started: true };
|
||||
lifecycle.start(intervalMs);
|
||||
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: "ok",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-service"),
|
||||
details: { projectId, intervalMs, inProcess: true },
|
||||
});
|
||||
|
||||
active.set(projectId, {
|
||||
lifecycle,
|
||||
stop: () => {
|
||||
try {
|
||||
lifecycle.stop();
|
||||
} finally {
|
||||
observer.setHealth({
|
||||
surface: "lifecycle.worker",
|
||||
status: "warn",
|
||||
projectId,
|
||||
correlationId: createCorrelationId("lifecycle-service"),
|
||||
reason: "Lifecycle polling stopped",
|
||||
details: { projectId },
|
||||
});
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
return { running: true, started: true };
|
||||
}
|
||||
|
||||
export async function stopLifecycleWorker(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
): Promise<boolean> {
|
||||
const status = getLifecycleWorkerStatus(config, projectId);
|
||||
if (!status.running || status.pid === null) {
|
||||
clearLifecycleWorkerPid(config, projectId);
|
||||
return false;
|
||||
}
|
||||
|
||||
try {
|
||||
process.kill(status.pid, "SIGTERM");
|
||||
} catch {
|
||||
clearLifecycleWorkerPid(config, projectId, status.pid);
|
||||
return false;
|
||||
}
|
||||
|
||||
const deadline = Date.now() + STOP_TIMEOUT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
if (!isProcessRunning(status.pid)) {
|
||||
clearLifecycleWorkerPid(config, projectId, status.pid);
|
||||
return true;
|
||||
export function stopAllLifecycleWorkers(): void {
|
||||
for (const projectId of Array.from(active.keys())) {
|
||||
const entry = active.get(projectId);
|
||||
if (entry) {
|
||||
try {
|
||||
entry.stop();
|
||||
} catch {
|
||||
// Best-effort
|
||||
}
|
||||
}
|
||||
await sleep(100);
|
||||
active.delete(projectId);
|
||||
}
|
||||
|
||||
try {
|
||||
process.kill(status.pid, "SIGKILL");
|
||||
} catch {
|
||||
// Best effort hard stop
|
||||
}
|
||||
|
||||
clearLifecycleWorkerPid(config, projectId, status.pid);
|
||||
return true;
|
||||
}
|
||||
|
||||
export function isLifecycleWorkerRunning(projectId: string): boolean {
|
||||
return active.has(projectId);
|
||||
}
|
||||
|
||||
export function listLifecycleWorkers(): string[] {
|
||||
return Array.from(active.keys());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,7 +8,6 @@ import { registerReviewCheck } from "./commands/review-check.js";
|
|||
import { registerDashboard } from "./commands/dashboard.js";
|
||||
import { registerOpen } from "./commands/open.js";
|
||||
import { registerStart, registerStop } from "./commands/start.js";
|
||||
import { registerLifecycleWorker } from "./commands/lifecycle-worker.js";
|
||||
import { registerVerify } from "./commands/verify.js";
|
||||
import { registerDoctor } from "./commands/doctor.js";
|
||||
import { registerUpdate } from "./commands/update.js";
|
||||
|
|
@ -36,7 +35,6 @@ export function createProgram(): Command {
|
|||
registerReviewCheck(program);
|
||||
registerDashboard(program);
|
||||
registerOpen(program);
|
||||
registerLifecycleWorker(program);
|
||||
registerVerify(program);
|
||||
registerDoctor(program);
|
||||
registerUpdate(program);
|
||||
|
|
|
|||
Loading…
Reference in New Issue