diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index 7046888c6..a3670d9d3 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -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 { - 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>).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); diff --git a/packages/cli/src/lib/detect-agent.ts b/packages/cli/src/lib/detect-agent.ts index 09f3dc63d..3195826b1 100644 --- a/packages/cli/src/lib/detect-agent.ts +++ b/packages/cli/src/lib/detect-agent.ts @@ -75,20 +75,23 @@ export async function detectAgentRuntime(): Promise { 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; } diff --git a/packages/cli/src/lib/running-state.ts b/packages/cli/src/lib/running-state.ts index e4ded21ab..666a64780 100644 --- a/packages/cli/src/lib/running-state.ts +++ b/packages/cli/src/lib/running-state.ts @@ -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 { + 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 { + 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 { + 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 { 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 { 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); } diff --git a/packages/core/__tests__/config.test.ts b/packages/core/__tests__/config.test.ts index b06da050c..208557b6b 100644 --- a/packages/core/__tests__/config.test.ts +++ b/packages/core/__tests__/config.test.ts @@ -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); }); }); diff --git a/packages/core/src/config.ts b/packages/core/src/config.ts index d0b704633..5ddc731bf 100644 --- a/packages/core/src/config.ts +++ b/packages/core/src/config.ts @@ -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"); diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 39a279ac6..3e1880196 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -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"; + } +} diff --git a/packages/web/server/start-all.ts b/packages/web/server/start-all.ts index 170aaa82e..50c0bd8b7 100644 --- a/packages/web/server/start-all.ts +++ b/packages/web/server/start-all.ts @@ -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); diff --git a/scripts/setup.sh b/scripts/setup.sh index 4b65dfc6c..a41f6d7cc 100755 --- a/scripts/setup.sh +++ b/scripts/setup.sh @@ -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 ""