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

fix: clean up dashboard shutdown diagnostics

This commit is contained in:
i-trytoohard 2026-05-15 06:11:12 +05:30
parent a66a087ef6
commit 5ed9d23aff
7 changed files with 538 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
packages/integration-tests/src/daemon-children.integration.test.ts
View File

@ -114,12 +114,20 @@ describe.skipIf(!canRun)("daemon child reaping (integration)", () => {
AO_CALLER_TYPE: "agent",
AO_CONFIG_PATH: configPath,
AO_GLOBAL_CONFIG: configPath,
AO_SHUTDOWN_GRACE_MS: "15000",
PORT: String(port),
};
const start = spawn(tsxBin, [cliEntry, "start", "--no-orchestrator", "--reap-orphans"], {
cwd: repoPath,
env,
stdio: "ignore",
stdio: ["ignore", "pipe", "pipe"],
});
let startOutput = "";
start.stdout?.on("data", (chunk: Buffer) => {
startOutput += chunk.toString();
});
start.stderr?.on("data", (chunk: Buffer) => {
startOutput += chunk.toString();
});
startPid = start.pid;
expect(startPid).toBeTypeOf("number");
@ -139,11 +147,16 @@ describe.skipIf(!canRun)("daemon child reaping (integration)", () => {
const childPids = await readChildPids(runningPid!);
expect(childPids.length).toBeGreaterThan(0);
await execFileAsync(tsxBin, [cliEntry, "stop", "--all"], { cwd: repoPath, env, timeout: 20_000 });
await execFileAsync(tsxBin, [cliEntry, "stop", "--all"], {
cwd: repoPath,
env,
timeout: 20_000,
});
await sleep(5_000);
const stillAlive = childPids.filter(isAlive);
expect(stillAlive).toEqual([]);
expect(isAlive(runningPid!)).toBe(false);
expect(startOutput).not.toMatch(/exited with code 1/);
}, 60_000);
});

33
packages/web/server/__tests__/direct-terminal-ws.integration.test.ts
View File

@ -7,12 +7,16 @@
* response, mirroring what the browser's MuxProvider does.
*/
import { describe, it, expect, beforeAll, afterAll } from "vitest";
import { describe, it, expect, beforeAll, afterAll, vi } from "vitest";
import { execFileSync } from "node:child_process";
import { request, type IncomingMessage } from "node:http";
import { WebSocket } from "ws";
import { findTmux } from "../tmux-utils.js";
import { createDirectTerminalServer, type DirectTerminalServer } from "../direct-terminal-ws.js";
import {
createDirectTerminalServer,
createDirectTerminalShutdownHandler,
type DirectTerminalServer,
} from "../direct-terminal-ws.js";
const TMUX = findTmux();
const TEST_SESSION = `ao-test-integration-${process.pid}`;
@ -25,6 +29,31 @@ const describeWithTmux = TMUX ? describe : describe.skip;
let terminal: DirectTerminalServer;
let port: number;
describe("direct terminal shutdown handler", () => {
it("runs shutdown only once when duplicate SIGTERM/SIGINT signals arrive", () => {
const shutdown = vi.fn();
const log = vi.fn();
const warn = vi.fn();
const exit = vi.fn();
const handleShutdown = createDirectTerminalShutdownHandler(shutdown, {
log,
warn,
exit,
forceTimeoutMs: 0,
});
handleShutdown("SIGTERM");
handleShutdown("SIGTERM");
handleShutdown("SIGINT");
expect(shutdown).toHaveBeenCalledTimes(1);
expect(log).toHaveBeenCalledTimes(1);
expect(log).toHaveBeenCalledWith("[DirectTerminal] Received SIGTERM, shutting down...");
expect(warn).not.toHaveBeenCalled();
expect(exit).not.toHaveBeenCalled();
});
});
// =============================================================================
// Helpers
// =============================================================================

94
packages/web/server/__tests__/mux-websocket.test.ts
View File

@ -307,6 +307,30 @@ describe("SessionBroadcaster", () => {
expect(callback).not.toHaveBeenCalled();
});
it("suppresses abort warnings when shutdown aborts an in-flight fetch", async () => {
const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {});
let rejectFetch: ((err: unknown) => void) | undefined;
mockFetch.mockImplementationOnce(
() =>
new Promise((_resolve, reject) => {
rejectFetch = reject;
}),
);
const callback = vi.fn();
broadcaster.subscribe(callback);
broadcaster.shutdown();
rejectFetch?.("shutdown");
await vi.advanceTimersByTimeAsync(0);
expect(callback).not.toHaveBeenCalled();
expect(warnSpy).not.toHaveBeenCalledWith(
"[SessionBroadcaster] fetchSnapshot error:",
"shutdown",
);
warnSpy.mockRestore();
});
});
describe("disconnect", () => {
@ -1076,3 +1100,73 @@ describe("TerminalManager.open — re-attach skipped when tmux session is gone (
expect(exitCb).not.toHaveBeenCalled();
});
});
describe("TerminalManager shutdown", () => {
let capturedOnExit: ((evt: { exitCode: number }) => Promise<void> | void) | undefined;
let pty: {
onData: ReturnType<typeof vi.fn>;
onExit: ReturnType<typeof vi.fn>;
write: ReturnType<typeof vi.fn>;
resize: ReturnType<typeof vi.fn>;
kill: ReturnType<typeof vi.fn>;
};
beforeEach(() => {
vi.useFakeTimers();
mockSpawn.mockReset();
mockPtySpawn.mockReset();
mockTmuxHasSession.mockReset();
capturedOnExit = undefined;
mockSpawn.mockImplementation(() => new EventEmitter());
pty = {
onData: vi.fn(),
onExit: vi.fn((cb: (evt: { exitCode: number }) => Promise<void> | void) => {
capturedOnExit = cb;
}),
write: vi.fn(),
resize: vi.fn(),
kill: vi.fn(),
};
mockPtySpawn.mockReturnValue(pty);
});
afterEach(() => {
vi.clearAllTimers();
vi.useRealTimers();
});
it("detaches PTYs, lets them drain, and reports normal exit code 0 without re-attaching", async () => {
const mgr = new TerminalManager("/usr/bin/tmux");
const exitCb = vi.fn();
mgr.subscribe("ao-177", undefined, vi.fn(), exitCb);
const shutdownPromise = mgr.shutdownGracefully(1000);
expect(pty.write).toHaveBeenCalledWith("\x02d");
await capturedOnExit!({ exitCode: 0 });
await vi.advanceTimersByTimeAsync(1000);
await shutdownPromise;
expect(pty.kill).not.toHaveBeenCalled();
expect(mockTmuxHasSession).not.toHaveBeenCalled();
expect(mockPtySpawn).toHaveBeenCalledTimes(1);
expect(exitCb).toHaveBeenCalledWith(0);
});
it("does not hard-kill PTYs when WebSocket subscribers close during the drain window", async () => {
const mgr = new TerminalManager("/usr/bin/tmux");
const unsubscribe = mgr.subscribe("ao-177", undefined, vi.fn(), vi.fn());
const shutdownPromise = mgr.shutdownGracefully(1000);
unsubscribe();
expect(pty.write).toHaveBeenCalledWith("\x02d");
expect(pty.kill).not.toHaveBeenCalled();
await capturedOnExit!({ exitCode: 0 });
await vi.advanceTimersByTimeAsync(1000);
await shutdownPromise;
expect(pty.kill).not.toHaveBeenCalled();
});
});

25
packages/web/server/__tests__/start-all.test.ts Normal file
View File

@ -0,0 +1,25 @@
import { describe, expect, it } from "vitest";
import {
DEFAULT_SHUTDOWN_GRACE_MS,
formatCleanShutdownMessage,
formatForceKillFallbackMessage,
getShutdownGraceMs,
} from "../start-all.js";
describe("start-all shutdown diagnostics", () => {
it("uses a 15s default shutdown grace window with env override", () => {
expect(DEFAULT_SHUTDOWN_GRACE_MS).toBe(15_000);
expect(getShutdownGraceMs({})).toBe(15_000);
expect(getShutdownGraceMs({ AO_SHUTDOWN_GRACE_MS: "30000" })).toBe(30_000);
expect(getShutdownGraceMs({ AO_SHUTDOWN_GRACE_MS: "nope" })).toBe(15_000);
});
it("keeps shutdown log wording explicit about clean vs safety-fallback exits", () => {
expect(`info: ${formatCleanShutdownMessage(1234)}`).toMatchInlineSnapshot(
`"info: all children shut down cleanly in 1234ms"`,
);
expect(`warn: ${formatForceKillFallbackMessage("next", 15000)}`).toMatchInlineSnapshot(
`"warn: next did not exit within 15000ms — sent SIGKILL (safety fallback, no orphans leaked)"`,
);
});
});

97
packages/web/server/direct-terminal-ws.ts
View File

@ -4,13 +4,51 @@
*/
import { createServer, type Server } from "node:http";
import type { WebSocketServer } from "ws";
import { isWindows } from "@aoagents/ao-core";
import { findTmux } from "./tmux-utils.js";
import { createMuxWebSocket } from "./mux-websocket.js";
import {
createMuxWebSocket,
PTY_SHUTDOWN_DRAIN_MS,
type MuxWebSocketServer,
} from "./mux-websocket.js";
export interface DirectTerminalServer {
server: Server;
shutdown: () => void;
shutdown: (opts?: { drainMs?: number }) => void;
}
export function createDirectTerminalShutdownHandler(
shutdown: () => void | Promise<void>,
opts: {
log?: (message: string) => void;
warn?: (message: string) => void;
exit?: (code: number) => never | void;
forceTimeoutMs?: number;
} = {},
): (signal: string) => void {
const log = opts.log ?? console.log;
const warn = opts.warn ?? console.warn;
const exit = opts.exit ?? process.exit;
const forceTimeoutMs = opts.forceTimeoutMs ?? 5_000;
let shuttingDown = false;
return (signal: string) => {
if (shuttingDown) return;
shuttingDown = true;
log(`[DirectTerminal] Received ${signal}, shutting down...`);
void Promise.resolve(shutdown());
if (forceTimeoutMs > 0) {
const forceExitTimer = setTimeout(() => {
warn(
`[DirectTerminal] warn: forced shutdown after ${forceTimeoutMs}ms (safety fallback, no orphans leaked)`,
);
exit(0);
}, forceTimeoutMs);
forceExitTimer.unref();
}
};
}
/**
@ -20,7 +58,7 @@ export interface DirectTerminalServer {
export function createDirectTerminalServer(tmuxPath?: string | null): DirectTerminalServer {
const TMUX = tmuxPath ?? findTmux();
let muxWss: WebSocketServer | null = null;
let muxWss: MuxWebSocketServer | null = null;
const metrics = {
totalConnections: 0,
@ -69,25 +107,37 @@ export function createDirectTerminalServer(tmuxPath?: string | null): DirectTerm
server.on("upgrade", (request, socket, head) => {
const pathname = new URL(request.url ?? "/", "ws://localhost").pathname;
if ((pathname === "/mux" || pathname === "/ao-terminal-mux") && muxWss) {
muxWss.handleUpgrade(request, socket, head, (ws) => {
muxWss!.emit("connection", ws, request);
const mux = muxWss;
if ((pathname === "/mux" || pathname === "/ao-terminal-mux") && mux) {
mux.handleUpgrade(request, socket, head, (ws) => {
mux.emit("connection", ws, request);
});
} else {
socket.destroy();
}
});
function shutdown() {
// Terminate all connected mux clients — this triggers their 'close' events
// which unsubscribe terminal callbacks and kill PTY processes.
if (muxWss) {
for (const client of muxWss.clients) {
client.terminate();
function shutdown(opts: { drainMs?: number } = {}) {
const drainMs = opts.drainMs ?? 0;
void (async () => {
await muxWss?.shutdownGracefully?.(drainMs);
if (muxWss) {
// Send a normal close frame first so browsers receive any final PTY
// exit messages. After the PTY drain window, terminate stragglers.
for (const client of muxWss.clients) {
client.close(1001, "server shutting down");
}
const terminateTimer = setTimeout(() => {
if (!muxWss) return;
for (const client of muxWss.clients) {
client.terminate();
}
}, 200);
terminateTimer.unref();
muxWss.close();
}
muxWss.close();
}
server.close();
server.close();
})();
}
return { server, shutdown };
@ -107,7 +157,7 @@ if (isMainModule) {
const TMUX = findTmux();
if (TMUX) {
console.log(`[DirectTerminal] Using tmux: ${TMUX}`);
} else if (process.platform === "win32") {
} else if (isWindows()) {
console.log(`[DirectTerminal] Windows mode — using named pipe relay to PTY hosts`);
} else {
console.log(`[DirectTerminal] No tmux available — terminal relay may be limited`);
@ -119,15 +169,10 @@ if (isMainModule) {
console.log(`[DirectTerminal] WebSocket server listening on port ${PORT}`);
});
function handleShutdown(signal: string) {
console.log(`[DirectTerminal] Received ${signal}, shutting down...`);
shutdown();
const forceExitTimer = setTimeout(() => {
console.error("[DirectTerminal] Forced shutdown after timeout");
process.exit(1);
}, 5000);
forceExitTimer.unref();
}
const handleShutdown = createDirectTerminalShutdownHandler(
() => shutdown({ drainMs: PTY_SHUTDOWN_DRAIN_MS }),
{ forceTimeoutMs: 15_000 },
);
process.on("SIGINT", () => handleShutdown("SIGINT"));
process.on("SIGTERM", () => handleShutdown("SIGTERM"));

146
packages/web/server/mux-websocket.ts
View File

@ -83,7 +83,9 @@ export class SessionBroadcaster {
private subscribers = new Set<(sessions: SessionPatch[]) => void>();
private errorSubscribers = new Set<(error: string) => void>();
private intervalId: ReturnType<typeof setInterval> | null = null;
private controllers = new Set<AbortController>();
private polling = false;
private shuttingDown = false;
// Tracks the last fetch outcome so we only emit ui.session_broadcast_failed on
// the healthy → failing transition (not every 3s during an outage).
private lastFetchOk = true;
@ -173,6 +175,7 @@ export class SessionBroadcaster {
error: string | null;
}> {
const controller = new AbortController();
this.controllers.add(controller);
const timeoutId = setTimeout(() => controller.abort(), 4000);
try {
const res = await fetch(`${this.baseUrl}/api/sessions/patches`, {
@ -190,13 +193,26 @@ export class SessionBroadcaster {
return { sessions: data.sessions ?? null, error: null };
} catch (err) {
clearTimeout(timeoutId);
if (this.shuttingDown && (controller.signal.aborted || isAbortError(err))) {
return { sessions: null, error: null };
}
const msg = err instanceof Error ? err.message : String(err);
console.warn("[SessionBroadcaster] fetchSnapshot error:", msg);
this.recordFetchFailure(msg);
return { sessions: null, error: msg };
} finally {
this.controllers.delete(controller);
}
}
shutdown(): void {
this.shuttingDown = true;
for (const controller of this.controllers) {
controller.abort();
}
this.disconnect();
}
/**
* Emit ui.session_broadcast_failed once per healthy→failing transition.
* The broadcaster polls every 3s; emitting on every failure during a long
@ -381,6 +397,14 @@ export class NotificationBroadcaster {
}
}
function isAbortError(err: unknown): boolean {
if (err instanceof DOMException) return err.name === "AbortError";
if (err instanceof Error) {
return err.name === "AbortError" || /aborted|abort/i.test(err.message);
}
return false;
}
// node-pty is an optionalDependency — load dynamically
/* eslint-disable @typescript-eslint/consistent-type-imports -- node-pty is optional; static import would crash if missing */
type IPty = import("node-pty").IPty;
@ -446,6 +470,7 @@ interface ManagedTerminal {
const RING_BUFFER_MAX = 50 * 1024; // 50KB max per terminal
const WS_BUFFER_HIGH_WATERMARK = 64 * 1024; // 64KB
const MAX_REATTACH_ATTEMPTS = 3;
export const PTY_SHUTDOWN_DRAIN_MS = 1_000;
/**
* Grace period a freshly-attached PTY must survive before its successful
* attach is allowed to reset the re-attach counter. Prevents tight crash
@ -467,6 +492,7 @@ export class TerminalManager {
private terminals = new Map<string, ManagedTerminal>();
private TMUX: string;
private spawnHelperRepairAttempted = false;
private shuttingDown = false;
constructor(tmuxPath?: string) {
const resolved = tmuxPath ?? findTmux();
@ -653,6 +679,17 @@ export class TerminalManager {
terminal.pty = null;
let reattachError: string | undefined;
if (this.shuttingDown) {
if (terminal.resetTimer) {
clearTimeout(terminal.resetTimer);
terminal.resetTimer = undefined;
}
for (const cb of terminal.exitCallbacks) {
cb(exitCode);
}
return;
}
// Skip the re-attach loop entirely when the underlying tmux session is
// gone (e.g. user pressed Ctrl-C in the pane and the launch command
// exited, taking the only window with it). Without this guard we
@ -830,11 +867,11 @@ export class TerminalManager {
clearTimeout(terminal.resetTimer);
terminal.resetTimer = undefined;
}
if (terminal.pty) {
if (!this.shuttingDown && terminal.pty) {
terminal.pty.kill();
terminal.pty = null;
}
this.terminals.delete(key);
if (!this.shuttingDown) this.terminals.delete(key);
}
};
}
@ -847,6 +884,103 @@ export class TerminalManager {
if (!terminal) return "";
return terminal.buffer.join("");
}
private async findTmuxClientName(tmuxSessionId: string, ptyPid: number): Promise<string | null> {
return new Promise((resolve) => {
const proc = spawn(this.TMUX, [
"list-clients",
"-t",
`=${tmuxSessionId}`,
"-F",
"#{client_pid}\t#{client_name}",
]);
let stdout = "";
let settled = false;
const finish = (clientName: string | null) => {
if (settled) return;
settled = true;
clearTimeout(timer);
resolve(clientName);
};
const timer = setTimeout(() => {
proc.kill("SIGTERM");
finish(null);
}, 500);
timer.unref();
if (!proc.stdout) {
finish(null);
return;
}
proc.stdout.on("data", (chunk: Buffer) => {
stdout += chunk.toString("utf8");
});
proc.on("error", () => finish(null));
proc.on("close", () => {
for (const line of stdout.split("\n")) {
const [pidText, clientName] = line.split("\t");
if (Number(pidText) === ptyPid && clientName) {
finish(clientName);
return;
}
}
finish(null);
});
});
}
private async detachManagedPtyClient(terminal: ManagedTerminal): Promise<void> {
const ptyPid = terminal.pty?.pid;
if (ptyPid) {
const clientName = await this.findTmuxClientName(terminal.tmuxSessionId, ptyPid);
if (clientName) {
try {
spawn(this.TMUX, ["detach-client", "-t", clientName]).on("error", () => {
// Best-effort; fallback below covers spawn failures.
});
return;
} catch {
// Fall through to the PTY-local detach key.
}
}
}
try {
// Fallback to the default tmux detach key (Ctrl-B, d). This targets only
// the managed PTY client instead of detaching every client on the session.
terminal.pty?.write("\x02d");
} catch {
// Best-effort; the kill fallback below handles stubborn PTYs.
}
}
async shutdownGracefully(drainMs = PTY_SHUTDOWN_DRAIN_MS): Promise<void> {
if (this.shuttingDown) return;
this.shuttingDown = true;
const attached = [...this.terminals.values()].filter((terminal) => terminal.pty);
for (const terminal of attached) {
if (terminal.resetTimer) {
clearTimeout(terminal.resetTimer);
terminal.resetTimer = undefined;
}
await this.detachManagedPtyClient(terminal);
}
if (drainMs > 0 && attached.some((terminal) => terminal.pty)) {
await new Promise<void>((resolve) => setTimeout(resolve, drainMs));
}
for (const terminal of attached) {
if (terminal.pty) {
try {
terminal.pty.kill();
} catch {
// Process is already gone or cannot be signalled; shutdown continues.
}
}
}
}
}
// ── Windows Pipe Relay (extracted for testability) ──
@ -865,6 +999,10 @@ export interface PipeRelayDeps {
resolvePipePath: (id: string, projectId?: string) => string | null;
}
export interface MuxWebSocketServer extends WebSocketServer {
shutdownGracefully?: (drainMs?: number) => Promise<void>;
}
/**
* Handle a Windows terminal message by relaying through named pipes.
* Extracted from the WebSocket connection handler for testability.
@ -1418,5 +1556,9 @@ export function createMuxWebSocket(tmuxPath?: string | null): WebSocketServer |
});
console.log("[MuxServer] Mux WebSocket server created (noServer mode)");
(wss as MuxWebSocketServer).shutdownGracefully = async (drainMs = PTY_SHUTDOWN_DRAIN_MS) => {
broadcaster.shutdown();
await terminalManager?.shutdownGracefully(drainMs);
};
return wss;
}

214
packages/web/server/start-all.ts
View File

@ -22,13 +22,45 @@ const __dirname = dirname(__filename);
// Resolve paths relative to the package root (one level up from dist-server/)
const pkgRoot = resolve(__dirname, "..");
const children: ChildProcess[] = [];
markDaemonShutdownHandlerInstalled();
export const DEFAULT_SHUTDOWN_GRACE_MS = 15_000;
const POST_SIGKILL_VERIFY_MS = 2_000;
interface ManagedChild {
label: string;
child: ChildProcess;
exited: boolean;
}
const children: ManagedChild[] = [];
let shuttingDown = false;
function log(label: string, msg: string): void {
process.stdout.write(`[${label}] ${msg}\n`);
}
function logLevel(label: string, level: "info" | "warn" | "error", msg: string): void {
log(label, `${level}: ${msg}`);
}
export function getShutdownGraceMs(
env: Record<string, string | undefined> = process.env as Record<string, string | undefined>,
): number {
const raw = env["AO_SHUTDOWN_GRACE_MS"];
if (!raw) return DEFAULT_SHUTDOWN_GRACE_MS;
const parsed = Math.floor(Number(raw));
if (!Number.isFinite(parsed) || parsed <= 0) return DEFAULT_SHUTDOWN_GRACE_MS;
return parsed;
}
export function formatCleanShutdownMessage(elapsedMs: number): string {
return `all children shut down cleanly in ${elapsedMs}ms`;
}
export function formatForceKillFallbackMessage(label: string, graceMs: number): string {
return `${label} did not exit within ${graceMs}ms — sent SIGKILL (safety fallback, no orphans leaked)`;
}
function spawnProcess(
label: string,
command: string,
@ -59,21 +91,23 @@ function spawnProcess(
}
});
child.on("exit", (code) => {
child.on("exit", (code: number | null) => {
const current = children[slotIndex];
if (current?.child === child) current.exited = true;
log(label, `exited with code ${code}`);
if (!shuttingDown && opts?.restart && code !== 0 && restarts < maxRestarts) {
restarts++;
log(label, `restarting (attempt ${restarts}/${maxRestarts})`);
const replacement = launch();
// Replace in-place — slot was assigned on first push
children[slotIndex] = replacement;
children[slotIndex] = { label, child: replacement, exited: false };
}
});
// Only push on first launch; restarts replace the existing slot
if (slotIndex === -1) {
slotIndex = children.length;
children.push(child);
children.push({ label, child, exited: false });
}
return child;
@ -105,68 +139,100 @@ function resolveNextBin(): string {
}
}
// Start Next.js production server
const port = process.env["PORT"] || "3000";
const pathBasedMux = process.env["AO_PATH_BASED_MUX"] === "1";
// When AO_PATH_BASED_MUX=1, single-port-server.js owns PORT and Next.js is
// shifted to PORT + 1000 (overridable via NEXT_INTERNAL_PORT). The proxy
// forwards HTTP to Next.js and tunnels `/ao-terminal-mux` WS upgrades to
// direct-terminal-ws. Default off — Next.js stays on PORT directly.
const NEXT_INTERNAL_OFFSET = 1000;
const nextPort = pathBasedMux
? (process.env["NEXT_INTERNAL_PORT"] ?? String(parseInt(port, 10) + NEXT_INTERNAL_OFFSET))
: port;
const nextBin = resolveNextBin();
if (isWindows() && nextBin !== "next") {
// On Windows, run the JS entry point via the current node binary.
// spawn() can't execute .js files directly on Windows.
spawnProcess("next", process.execPath, [nextBin, "start", "-p", nextPort]);
} else {
spawnProcess("next", nextBin, ["start", "-p", nextPort]);
}
if (pathBasedMux) {
// Surface the internal port to the child so it doesn't have to re-derive
// the offset; pin it explicitly.
process.env["NEXT_INTERNAL_PORT"] = nextPort;
spawnProcess("single-port", process.execPath, [resolve(__dirname, "single-port-server.js")]);
}
// Start direct terminal WebSocket server (auto-restart on crash)
spawnProcess("direct-terminal", "node", [resolve(__dirname, "direct-terminal-ws.js")], {
restart: true,
});
// Graceful shutdown — send SIGTERM to children and wait for them to exit
let shuttingDown = false;
function cleanup(): void {
if (shuttingDown) return;
shuttingDown = true;
let alive = children.length;
if (alive === 0) {
const graceMs = getShutdownGraceMs();
const shutdownStartedAt = Date.now();
const alive = new Set(children.filter(({ exited }) => !exited));
let forceFallbackStarted = false;
const finishIfDone = (): void => {
if (alive.size > 0) return;
if (!forceFallbackStarted) {
logLevel("start-all", "info", formatCleanShutdownMessage(Date.now() - shutdownStartedAt));
}
process.exit(0);
};
if (alive.size === 0) {
finishIfDone();
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);
forceFallbackStarted = true;
void forceKillRemainingChildren(alive, graceMs);
}, graceMs);
forceTimer.unref();
for (const child of children) {
child.on("exit", () => {
alive--;
if (alive <= 0) {
clearTimeout(forceTimer);
async function forceKillRemainingChildren(stuckChildren: Set<ManagedChild>, grace: number) {
const stuck = [...stuckChildren].filter(({ exited }) => !exited);
if (stuck.length === 0) {
finishIfDone();
return;
}
const results = await Promise.all(
stuck.map(async ({ label, child }) => {
logLevel("start-all", "warn", formatForceKillFallbackMessage(label, grace));
const pid = child.pid;
if (!pid) {
try {
return child.kill("SIGKILL");
} catch {
return false;
}
}
try {
await killProcessTree(pid, "SIGKILL");
return true;
} catch {
try {
return child.kill("SIGKILL");
} catch {
return false;
}
}
}),
);
if (results.some((sent) => !sent)) {
logLevel(
"start-all",
"error",
"SIGKILL fallback failed for one or more children; manual cleanup may be required",
);
process.exit(1);
return;
}
const verifyTimer = setTimeout(() => {
const stillAlive = [...stuckChildren].filter(({ exited }) => !exited);
if (stillAlive.length === 0) {
process.exit(0);
return;
}
logLevel(
"start-all",
"error",
`${stillAlive.map(({ label }) => label).join(", ")} remained alive after SIGKILL; manual cleanup may be required`,
);
process.exit(1);
}, POST_SIGKILL_VERIFY_MS);
verifyTimer.unref();
}
for (const info of alive) {
const { child } = info;
child.on("exit", () => {
info.exited = true;
alive.delete(info);
if (alive.size <= 0) clearTimeout(forceTimer);
finishIfDone();
});
const pid = child.pid;
if (pid) {
@ -179,5 +245,41 @@ function cleanup(): void {
}
}
process.on("SIGINT", cleanup);
process.on("SIGTERM", cleanup);
export function runStartAll(): void {
markDaemonShutdownHandlerInstalled();
// Start Next.js production server
const port = process.env["PORT"] || "3000";
const pathBasedMux = process.env["AO_PATH_BASED_MUX"] === "1";
const NEXT_INTERNAL_OFFSET = 1000;
const nextPort = pathBasedMux
? (process.env["NEXT_INTERNAL_PORT"] ?? String(parseInt(port, 10) + NEXT_INTERNAL_OFFSET))
: port;
const nextBin = resolveNextBin();
if (isWindows() && nextBin !== "next") {
// On Windows, run the JS entry point via the current node binary.
// spawn() can't execute .js files directly on Windows.
spawnProcess("next", process.execPath, [nextBin, "start", "-p", nextPort]);
} else {
spawnProcess("next", nextBin, ["start", "-p", nextPort]);
}
if (pathBasedMux) {
// Surface the internal port to the child so it doesn't have to re-derive
// the offset; pin it explicitly.
process.env["NEXT_INTERNAL_PORT"] = nextPort;
spawnProcess("single-port", process.execPath, [resolve(__dirname, "single-port-server.js")]);
}
// Start direct terminal WebSocket server (auto-restart on crash)
spawnProcess("direct-terminal", "node", [resolve(__dirname, "direct-terminal-ws.js")], {
restart: true,
});
process.on("SIGINT", cleanup);
process.on("SIGTERM", cleanup);
}
const isMainModule = process.argv[1] ? resolve(process.argv[1]) === __filename : false;
if (isMainModule) runStartAll();