Fix review issues: busy-wait loops, lock race, import hygiene, typed errors
- Replace CPU-burning spin loops in running-state.ts with async setTimeout and jittered backoff; make all exports async - Fix TOCTOU race in acquireLock by extracting tryAcquire helper with retry loop instead of force-remove-and-retry-once - Hoist dynamic imports in addProjectToConfig/choice-2 to static imports - Add try/finally around readline in detectAgentRuntime - Validate session prefix uniqueness in "Start new orchestrator" menu - Add ConfigNotFoundError class to @composio/ao-core, replace fragile string matching in ao start with instanceof check - Fix setup.sh to recommend `ao start` instead of deprecated `ao init` - Fix start-all.ts: resolve next binary with fallback, wait for children on SIGTERM instead of immediate process.exit Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
suraj-markup
parent
511d2b7fb9
commit
596913aa34
|
|
@ -10,7 +10,7 @@
|
|||
*/
|
||||
|
||||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { existsSync, writeFileSync } from "node:fs";
|
||||
import { existsSync, readFileSync, writeFileSync } from "node:fs";
|
||||
import { resolve, basename } from "node:path";
|
||||
import { cwd } from "node:process";
|
||||
import chalk from "chalk";
|
||||
|
|
@ -28,12 +28,13 @@ import {
|
|||
generateConfigFromUrl,
|
||||
configToYaml,
|
||||
normalizeOrchestratorSessionStrategy,
|
||||
ConfigNotFoundError,
|
||||
type OrchestratorConfig,
|
||||
type ProjectConfig,
|
||||
type ParsedRepoUrl,
|
||||
} from "@composio/ao-core";
|
||||
import { stringify as yamlStringify } from "yaml";
|
||||
import { exec, execSilent } from "../lib/shell.js";
|
||||
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 {
|
||||
|
|
@ -46,10 +47,11 @@ import {
|
|||
} from "../lib/web-dir.js";
|
||||
import { cleanNextCache } from "../lib/dashboard-rebuild.js";
|
||||
import { preflight } from "../lib/preflight.js";
|
||||
import { register, unregister, isAlreadyRunning, getRunning } from "../lib/running-state.js";
|
||||
import { register, unregister, isAlreadyRunning, getRunning, waitForExit } from "../lib/running-state.js";
|
||||
import { isHumanCaller } from "../lib/caller-context.js";
|
||||
import { detectEnvironment } from "../lib/detect-env.js";
|
||||
import { detectAgentRuntime } from "../lib/detect-agent.js";
|
||||
import { detectDefaultBranch } from "../lib/git-utils.js";
|
||||
import {
|
||||
detectProjectType,
|
||||
generateRulesFromTemplates,
|
||||
|
|
@ -338,11 +340,6 @@ async function addProjectToConfig(
|
|||
config: OrchestratorConfig,
|
||||
projectPath: string,
|
||||
): Promise<string> {
|
||||
const { readFileSync } = await import("node:fs");
|
||||
const { parse: yamlParse } = await import("yaml");
|
||||
const { git } = await import("../lib/shell.js");
|
||||
const { detectDefaultBranch } = await import("../lib/git-utils.js");
|
||||
|
||||
const resolvedPath = resolve(projectPath.replace(/^~/, process.env["HOME"] || ""));
|
||||
const projectId = basename(resolvedPath);
|
||||
|
||||
|
|
@ -743,7 +740,7 @@ export function registerStart(program: Command): void {
|
|||
try {
|
||||
loadedConfig = loadConfig();
|
||||
} catch (err) {
|
||||
if (err instanceof Error && err.message.includes("No agent-orchestrator.yaml found")) {
|
||||
if (err instanceof ConfigNotFoundError) {
|
||||
// First run — auto-create config
|
||||
loadedConfig = await autoCreateConfig(cwd());
|
||||
} else {
|
||||
|
|
@ -755,7 +752,7 @@ export function registerStart(program: Command): void {
|
|||
}
|
||||
|
||||
// ── Already-running detection (Step 9) ──
|
||||
const running = isAlreadyRunning();
|
||||
const running = await isAlreadyRunning();
|
||||
if (running) {
|
||||
if (isHumanCaller()) {
|
||||
console.log(chalk.cyan(`\nℹ AO is already running.`));
|
||||
|
|
@ -783,14 +780,24 @@ export function registerStart(program: Command): void {
|
|||
process.exit(0);
|
||||
} else if (choice.trim() === "2") {
|
||||
// Generate unique orchestrator: same project, new session
|
||||
const suffix = Math.random().toString(36).slice(2, 6);
|
||||
const newId = `${projectId}-${suffix}`;
|
||||
const newPrefix = generateSessionPrefix(newId);
|
||||
|
||||
const { readFileSync } = await import("node:fs");
|
||||
const { parse: yamlParse } = await import("yaml");
|
||||
const rawYaml = readFileSync(config.configPath, "utf-8");
|
||||
const rawConfig = yamlParse(rawYaml);
|
||||
|
||||
// Collect existing prefixes to avoid collisions
|
||||
const existingPrefixes = new Set(
|
||||
Object.values(rawConfig.projects as Record<string, Record<string, unknown>>).map(
|
||||
(p) => p.sessionPrefix as string,
|
||||
).filter(Boolean),
|
||||
);
|
||||
|
||||
let newId: string;
|
||||
let newPrefix: string;
|
||||
do {
|
||||
const suffix = Math.random().toString(36).slice(2, 6);
|
||||
newId = `${projectId}-${suffix}`;
|
||||
newPrefix = generateSessionPrefix(newId);
|
||||
} while (rawConfig.projects[newId] || existingPrefixes.has(newPrefix));
|
||||
|
||||
rawConfig.projects[newId] = {
|
||||
...rawConfig.projects[projectId],
|
||||
sessionPrefix: newPrefix,
|
||||
|
|
@ -803,12 +810,11 @@ export function registerStart(program: Command): void {
|
|||
// Continue to startup below
|
||||
} else if (choice.trim() === "3") {
|
||||
try { process.kill(running.pid, "SIGTERM"); } catch { /* already dead */ }
|
||||
const { waitForExit } = await import("../lib/running-state.js");
|
||||
if (!waitForExit(running.pid, 5000)) {
|
||||
if (!(await waitForExit(running.pid, 5000))) {
|
||||
console.log(chalk.yellow(" Process didn't exit cleanly, sending SIGKILL..."));
|
||||
try { process.kill(running.pid, "SIGKILL"); } catch { /* already dead */ }
|
||||
}
|
||||
unregister();
|
||||
await unregister();
|
||||
console.log(chalk.yellow("\n Stopped existing instance. Restarting...\n"));
|
||||
// Continue to startup below
|
||||
} else {
|
||||
|
|
@ -828,7 +834,7 @@ export function registerStart(program: Command): void {
|
|||
const actualPort = await runStartup(config, projectId, project, opts);
|
||||
|
||||
// ── Register in running.json (Step 10) ──
|
||||
register({
|
||||
await register({
|
||||
pid: process.pid,
|
||||
configPath: config.configPath,
|
||||
port: actualPort,
|
||||
|
|
@ -869,7 +875,7 @@ export function registerStop(program: Command): void {
|
|||
) => {
|
||||
try {
|
||||
// Check running.json first
|
||||
const running = getRunning();
|
||||
const running = await getRunning();
|
||||
|
||||
if (opts.all) {
|
||||
// --all: kill via running.json if available, then fallback to config
|
||||
|
|
@ -879,7 +885,7 @@ export function registerStop(program: Command): void {
|
|||
} catch {
|
||||
// Already dead
|
||||
}
|
||||
unregister();
|
||||
await unregister();
|
||||
console.log(
|
||||
chalk.green(`\n✓ Stopped AO on port ${running.port}`),
|
||||
);
|
||||
|
|
@ -925,7 +931,7 @@ export function registerStop(program: Command): void {
|
|||
} catch {
|
||||
// Already dead
|
||||
}
|
||||
unregister();
|
||||
await unregister();
|
||||
}
|
||||
await stopDashboard(running?.port ?? port);
|
||||
|
||||
|
|
|
|||
|
|
@ -75,20 +75,23 @@ export async function detectAgentRuntime(): Promise<string> {
|
|||
const { createInterface } = await import("node:readline/promises");
|
||||
const rl = createInterface({ input: process.stdin, output: process.stdout });
|
||||
|
||||
console.log("\n Multiple agent runtimes detected:\n");
|
||||
available.forEach((a, i) => {
|
||||
console.log(` ${i + 1}. ${a.displayName} (${a.name})`);
|
||||
});
|
||||
console.log();
|
||||
try {
|
||||
console.log("\n Multiple agent runtimes detected:\n");
|
||||
available.forEach((a, i) => {
|
||||
console.log(` ${i + 1}. ${a.displayName} (${a.name})`);
|
||||
});
|
||||
console.log();
|
||||
|
||||
const answer = await rl.question(` Choose default agent [1-${available.length}]: `);
|
||||
rl.close();
|
||||
const answer = await rl.question(` Choose default agent [1-${available.length}]: `);
|
||||
|
||||
const idx = parseInt(answer.trim(), 10) - 1;
|
||||
if (idx >= 0 && idx < available.length) {
|
||||
return available[idx].name;
|
||||
const idx = parseInt(answer.trim(), 10) - 1;
|
||||
if (idx >= 0 && idx < available.length) {
|
||||
return available[idx].name;
|
||||
}
|
||||
|
||||
// Invalid input — default to first
|
||||
return available[0].name;
|
||||
} finally {
|
||||
rl.close();
|
||||
}
|
||||
|
||||
// Invalid input — default to first
|
||||
return available[0].name;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
import { readFileSync, writeFileSync, mkdirSync, unlinkSync, openSync, closeSync, constants } from "node:fs";
|
||||
import { join } from "node:path";
|
||||
import { homedir } from "node:os";
|
||||
import { setTimeout as sleep } from "node:timers/promises";
|
||||
|
||||
export interface RunningState {
|
||||
pid: number;
|
||||
|
|
@ -27,38 +28,47 @@ function isProcessAlive(pid: number): boolean {
|
|||
}
|
||||
}
|
||||
|
||||
/** Try to create the lockfile atomically. Returns a release function on success, null on failure. */
|
||||
function tryAcquire(): (() => void) | null {
|
||||
try {
|
||||
const fd = openSync(LOCK_FILE, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY);
|
||||
closeSync(fd);
|
||||
return () => {
|
||||
try { unlinkSync(LOCK_FILE); } catch { /* best effort */ }
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Simple advisory lockfile. Uses O_EXCL to atomically create the lock.
|
||||
* Returns a release function. If lock cannot be acquired within timeout, throws.
|
||||
* Advisory lockfile using O_EXCL for atomic creation.
|
||||
* Retries with jittered backoff. After timeout, assumes the lock is stale
|
||||
* (holder crashed) and force-removes it before one final atomic attempt.
|
||||
*/
|
||||
function acquireLock(timeoutMs = 5000): () => void {
|
||||
async function acquireLock(timeoutMs = 5000): Promise<() => void> {
|
||||
ensureDir();
|
||||
|
||||
const start = Date.now();
|
||||
let attempt = 0;
|
||||
|
||||
while (true) {
|
||||
try {
|
||||
const fd = openSync(LOCK_FILE, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY);
|
||||
closeSync(fd);
|
||||
return () => {
|
||||
try { unlinkSync(LOCK_FILE); } catch { /* best effort */ }
|
||||
};
|
||||
} catch {
|
||||
if (Date.now() - start > timeoutMs) {
|
||||
// Stale lock — force remove and retry once
|
||||
try { unlinkSync(LOCK_FILE); } catch { /* ignore */ }
|
||||
try {
|
||||
const fd = openSync(LOCK_FILE, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY);
|
||||
closeSync(fd);
|
||||
return () => {
|
||||
try { unlinkSync(LOCK_FILE); } catch { /* best effort */ }
|
||||
};
|
||||
} catch {
|
||||
throw new Error("Could not acquire running.json lock");
|
||||
}
|
||||
}
|
||||
// Spin wait 50ms
|
||||
const end = Date.now() + 50;
|
||||
while (Date.now() < end) { /* busy wait */ }
|
||||
const release = tryAcquire();
|
||||
if (release) return release;
|
||||
|
||||
if (Date.now() - start > timeoutMs) {
|
||||
// Likely stale — remove and make one final atomic attempt.
|
||||
try { unlinkSync(LOCK_FILE); } catch { /* ignore */ }
|
||||
const finalRelease = tryAcquire();
|
||||
if (finalRelease) return finalRelease;
|
||||
throw new Error("Could not acquire running.json lock");
|
||||
}
|
||||
|
||||
// Jittered backoff: 30-70ms base, growing with attempts (capped at 200ms)
|
||||
const baseMs = Math.min(50 + attempt * 20, 200);
|
||||
const jitter = Math.floor(Math.random() * 40) - 20;
|
||||
await sleep(baseMs + jitter);
|
||||
attempt++;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -86,8 +96,8 @@ function writeState(state: RunningState | null): void {
|
|||
* Register the current AO instance as running.
|
||||
* Uses a lockfile to prevent concurrent registration.
|
||||
*/
|
||||
export function register(entry: RunningState): void {
|
||||
const release = acquireLock();
|
||||
export async function register(entry: RunningState): Promise<void> {
|
||||
const release = await acquireLock();
|
||||
try {
|
||||
writeState(entry);
|
||||
} finally {
|
||||
|
|
@ -98,8 +108,8 @@ export function register(entry: RunningState): void {
|
|||
/**
|
||||
* Unregister the running AO instance.
|
||||
*/
|
||||
export function unregister(): void {
|
||||
const release = acquireLock();
|
||||
export async function unregister(): Promise<void> {
|
||||
const release = await acquireLock();
|
||||
try {
|
||||
writeState(null);
|
||||
} finally {
|
||||
|
|
@ -111,8 +121,8 @@ export function unregister(): void {
|
|||
* Get the currently running AO instance, if any.
|
||||
* Auto-prunes stale entries (dead PIDs).
|
||||
*/
|
||||
export function getRunning(): RunningState | null {
|
||||
const release = acquireLock();
|
||||
export async function getRunning(): Promise<RunningState | null> {
|
||||
const release = await acquireLock();
|
||||
try {
|
||||
const state = readState();
|
||||
if (!state) return null;
|
||||
|
|
@ -133,7 +143,7 @@ export function getRunning(): RunningState | null {
|
|||
* Check if AO is already running.
|
||||
* Returns the running state if alive, null otherwise.
|
||||
*/
|
||||
export function isAlreadyRunning(): RunningState | null {
|
||||
export async function isAlreadyRunning(): Promise<RunningState | null> {
|
||||
return getRunning();
|
||||
}
|
||||
|
||||
|
|
@ -141,13 +151,11 @@ export function isAlreadyRunning(): RunningState | null {
|
|||
* Wait for a process to exit, polling isProcessAlive.
|
||||
* Returns true if the process exited, false if timeout reached.
|
||||
*/
|
||||
export function waitForExit(pid: number, timeoutMs = 5000): boolean {
|
||||
export async function waitForExit(pid: number, timeoutMs = 5000): Promise<boolean> {
|
||||
const start = Date.now();
|
||||
while (Date.now() - start < timeoutMs) {
|
||||
if (!isProcessAlive(pid)) return true;
|
||||
// Spin wait 100ms
|
||||
const end = Date.now() + 100;
|
||||
while (Date.now() < end) { /* busy wait */ }
|
||||
await sleep(100);
|
||||
}
|
||||
return !isProcessAlive(pid);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import { mkdirSync, writeFileSync, rmSync, realpathSync } from "node:fs";
|
|||
import { join } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { loadConfig, findConfigFile } from "../src/config.js";
|
||||
import { ConfigNotFoundError } from "../src/types.js";
|
||||
|
||||
describe("Config Loading", () => {
|
||||
let testDir: string;
|
||||
|
|
@ -110,7 +111,7 @@ projects:
|
|||
});
|
||||
|
||||
it("should throw error if config not found", () => {
|
||||
expect(() => loadConfig()).toThrow("No agent-orchestrator.yaml found");
|
||||
expect(() => loadConfig()).toThrow(ConfigNotFoundError);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ import { resolve, join, basename } from "node:path";
|
|||
import { homedir } from "node:os";
|
||||
import { parse as parseYaml } from "yaml";
|
||||
import { z } from "zod";
|
||||
import type { OrchestratorConfig } from "./types.js";
|
||||
import { ConfigNotFoundError, type OrchestratorConfig } from "./types.js";
|
||||
import { generateSessionPrefix } from "./paths.js";
|
||||
|
||||
function inferScmPlugin(project: {
|
||||
|
|
@ -467,7 +467,7 @@ export function loadConfig(configPath?: string): OrchestratorConfig {
|
|||
const path = configPath ?? findConfigFile();
|
||||
|
||||
if (!path) {
|
||||
throw new Error("No agent-orchestrator.yaml found. Run `ao init` to create one.");
|
||||
throw new ConfigNotFoundError();
|
||||
}
|
||||
|
||||
const raw = readFileSync(path, "utf-8");
|
||||
|
|
@ -488,7 +488,7 @@ export function loadConfigWithPath(configPath?: string): {
|
|||
const path = configPath ?? findConfigFile();
|
||||
|
||||
if (!path) {
|
||||
throw new Error("No agent-orchestrator.yaml found. Run `ao init` to create one.");
|
||||
throw new ConfigNotFoundError();
|
||||
}
|
||||
|
||||
const raw = readFileSync(path, "utf-8");
|
||||
|
|
|
|||
|
|
@ -1310,3 +1310,11 @@ export class SessionNotFoundError extends Error {
|
|||
this.name = "SessionNotFoundError";
|
||||
}
|
||||
}
|
||||
|
||||
/** Thrown when no agent-orchestrator.yaml config file can be found. */
|
||||
export class ConfigNotFoundError extends Error {
|
||||
constructor(message?: string) {
|
||||
super(message ?? "No agent-orchestrator.yaml found. Run `ao start` to create one.");
|
||||
this.name = "ConfigNotFoundError";
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,9 @@
|
|||
|
||||
import { spawn, 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";
|
||||
|
||||
const __filename = fileURLToPath(import.meta.url);
|
||||
const __dirname = dirname(__filename);
|
||||
|
|
@ -47,10 +49,28 @@ function spawnProcess(label: string, command: string, args: string[]): ChildProc
|
|||
return child;
|
||||
}
|
||||
|
||||
// Start Next.js production server (use local binary, not npx, to avoid slow global lookup)
|
||||
/**
|
||||
* Resolve the `next` CLI binary path.
|
||||
* Tries the local .bin shim first (fast), then falls back to require.resolve (hoisted deps).
|
||||
*/
|
||||
function resolveNextBin(): string {
|
||||
const localBin = resolve(pkgRoot, "node_modules", ".bin", "next");
|
||||
if (existsSync(localBin)) return localBin;
|
||||
|
||||
// Hoisted node_modules — resolve the actual next CLI entry
|
||||
const require = createRequire(resolve(pkgRoot, "package.json"));
|
||||
try {
|
||||
const nextPkg = require.resolve("next/package.json");
|
||||
return resolve(dirname(nextPkg), "dist", "bin", "next");
|
||||
} catch {
|
||||
// Last resort — rely on PATH
|
||||
return "next";
|
||||
}
|
||||
}
|
||||
|
||||
// Start Next.js production server
|
||||
const port = process.env["PORT"] || "3000";
|
||||
const nextBin = resolve(pkgRoot, "node_modules", ".bin", "next");
|
||||
spawnProcess("next", nextBin, ["start", "-p", port]);
|
||||
spawnProcess("next", resolveNextBin(), ["start", "-p", port]);
|
||||
|
||||
// Start terminal WebSocket server
|
||||
spawnProcess("terminal", "node", [resolve(__dirname, "terminal-websocket.js")]);
|
||||
|
|
@ -58,12 +78,36 @@ spawnProcess("terminal", "node", [resolve(__dirname, "terminal-websocket.js")]);
|
|||
// Start direct terminal WebSocket server
|
||||
spawnProcess("direct-terminal", "node", [resolve(__dirname, "direct-terminal-ws.js")]);
|
||||
|
||||
// Graceful shutdown
|
||||
// Graceful shutdown — send SIGTERM to children and wait for them to exit
|
||||
let shuttingDown = false;
|
||||
|
||||
function cleanup(): void {
|
||||
for (const child of children) {
|
||||
child.kill();
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
|
||||
let alive = children.length;
|
||||
if (alive === 0) {
|
||||
process.exit(0);
|
||||
return;
|
||||
}
|
||||
|
||||
// Force exit after 5s if children don't exit cleanly
|
||||
const forceTimer = setTimeout(() => {
|
||||
log("start-all", "Children did not exit in time, forcing shutdown");
|
||||
process.exit(1);
|
||||
}, 5000);
|
||||
forceTimer.unref();
|
||||
|
||||
for (const child of children) {
|
||||
child.on("exit", () => {
|
||||
alive--;
|
||||
if (alive <= 0) {
|
||||
clearTimeout(forceTimer);
|
||||
process.exit(0);
|
||||
}
|
||||
});
|
||||
child.kill("SIGTERM");
|
||||
}
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
process.on("SIGINT", cleanup);
|
||||
|
|
|
|||
|
|
@ -170,16 +170,12 @@ echo "Setup complete!"
|
|||
echo ""
|
||||
echo "What's next:"
|
||||
echo ""
|
||||
echo " Navigate to your project directory and initialize:"
|
||||
echo " Navigate to your project directory and start:"
|
||||
echo ""
|
||||
echo " cd ~/your-project"
|
||||
echo " ao init # auto-detects everything, zero prompts"
|
||||
echo ""
|
||||
echo " Then start the orchestrator + dashboard:"
|
||||
echo ""
|
||||
echo " ao start # run this from your project directory"
|
||||
echo " ao start # auto-detects, creates config, launches dashboard"
|
||||
echo ""
|
||||
echo " Want to add more projects later?"
|
||||
echo ""
|
||||
echo " ao add-project ~/path/to/another-repo"
|
||||
echo " ao start ~/path/to/another-repo"
|
||||
echo ""
|
||||
|
|
|
|||
Loading…
Reference in New Issue