feat: add lifecycle worker automation
This commit is contained in:
Jayesh Sharma
parent
533ad2498d
commit
a6c9dbd599
|
|
@ -269,18 +269,11 @@ describe("review-check command", () => {
|
|||
|
||||
const output = consoleSpy.mock.calls.map((c) => String(c[0])).join("\n");
|
||||
expect(output).toContain("Fix prompt sent");
|
||||
|
||||
// Should have sent C-c, C-u, message, Enter
|
||||
expect(mockExec).toHaveBeenCalledWith("tmux", ["send-keys", "-t", "app-1", "C-c"]);
|
||||
expect(mockExec).toHaveBeenCalledWith("tmux", ["send-keys", "-t", "app-1", "C-u"]);
|
||||
expect(mockExec).toHaveBeenCalledWith("tmux", [
|
||||
"send-keys",
|
||||
"-t",
|
||||
expect(mockSessionManager.send).toHaveBeenCalledWith(
|
||||
"app-1",
|
||||
"-l",
|
||||
expect.stringContaining("review comments"),
|
||||
]);
|
||||
expect(mockExec).toHaveBeenCalledWith("tmux", ["send-keys", "-t", "app-1", "Enter"]);
|
||||
);
|
||||
expect(mockExec).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("handles gh returning null (API failure)", async () => {
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import { join } from "node:path";
|
|||
import { tmpdir } from "node:os";
|
||||
import { type Session, type SessionManager, getProjectBaseDir } from "@composio/ao-core";
|
||||
|
||||
const { mockExec, mockConfigRef, mockSessionManager } = vi.hoisted(() => ({
|
||||
const { mockExec, mockConfigRef, mockSessionManager, mockEnsureLifecycleWorker } = vi.hoisted(() => ({
|
||||
mockExec: vi.fn(),
|
||||
mockConfigRef: { current: null as Record<string, unknown> | null },
|
||||
mockSessionManager: {
|
||||
|
|
@ -17,6 +17,7 @@ const { mockExec, mockConfigRef, mockSessionManager } = vi.hoisted(() => ({
|
|||
send: vi.fn(),
|
||||
claimPR: vi.fn(),
|
||||
},
|
||||
mockEnsureLifecycleWorker: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../../src/lib/shell.js", () => ({
|
||||
|
|
@ -52,6 +53,10 @@ 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/metadata.js", () => ({
|
||||
findSessionForIssue: vi.fn().mockResolvedValue(null),
|
||||
writeMetadata: vi.fn(),
|
||||
|
|
@ -108,6 +113,14 @@ 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',
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
|
|
@ -144,6 +157,11 @@ describe("spawn command", () => {
|
|||
|
||||
await program.parseAsync(["node", "test", "spawn", "my-app", "INT-100"]);
|
||||
|
||||
expect(mockEnsureLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"my-app",
|
||||
);
|
||||
|
||||
// Must delegate to session manager
|
||||
expect(mockSessionManager.spawn).toHaveBeenCalledWith({
|
||||
projectId: "my-app",
|
||||
|
|
|
|||
|
|
@ -22,6 +22,8 @@ const {
|
|||
mockSessionManager,
|
||||
mockWaitForPortAndOpen,
|
||||
mockSpawn,
|
||||
mockEnsureLifecycleWorker,
|
||||
mockStopLifecycleWorker,
|
||||
} = vi.hoisted(() => ({
|
||||
mockExec: vi.fn(),
|
||||
mockExecSilent: vi.fn(),
|
||||
|
|
@ -38,6 +40,8 @@ const {
|
|||
},
|
||||
mockWaitForPortAndOpen: vi.fn().mockResolvedValue(undefined),
|
||||
mockSpawn: vi.fn(),
|
||||
mockEnsureLifecycleWorker: vi.fn(),
|
||||
mockStopLifecycleWorker: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../../src/lib/shell.js", () => ({
|
||||
|
|
@ -76,6 +80,11 @@ 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),
|
||||
stopLifecycleWorker: (...args: unknown[]) => mockStopLifecycleWorker(...args),
|
||||
}));
|
||||
|
||||
vi.mock("../../src/lib/web-dir.js", () => ({
|
||||
findWebDir: vi.fn().mockReturnValue("/fake/web"),
|
||||
buildDashboardEnv: vi.fn().mockResolvedValue({}),
|
||||
|
|
@ -146,6 +155,16 @@ beforeEach(() => {
|
|||
mockExecSilent.mockResolvedValue(null);
|
||||
mockWaitForPortAndOpen.mockReset();
|
||||
mockWaitForPortAndOpen.mockResolvedValue(undefined);
|
||||
mockEnsureLifecycleWorker.mockReset();
|
||||
mockEnsureLifecycleWorker.mockResolvedValue({
|
||||
running: true,
|
||||
started: true,
|
||||
pid: 12345,
|
||||
pidFile: "/tmp/lifecycle-worker.pid",
|
||||
logFile: "/tmp/lifecycle-worker.log",
|
||||
});
|
||||
mockStopLifecycleWorker.mockReset();
|
||||
mockStopLifecycleWorker.mockResolvedValue(true);
|
||||
mockSpawn.mockClear();
|
||||
});
|
||||
|
||||
|
|
@ -560,6 +579,10 @@ describe("start command — browser open waits for port", () => {
|
|||
expect(port).toBe(3000);
|
||||
expect(url).toContain("/sessions/app-orchestrator");
|
||||
expect(signal).toBeInstanceOf(AbortSignal);
|
||||
expect(mockEnsureLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"my-app",
|
||||
);
|
||||
});
|
||||
|
||||
it("skips browser open with --no-dashboard", async () => {
|
||||
|
|
@ -568,6 +591,7 @@ describe("start command — browser open waits for port", () => {
|
|||
await program.parseAsync(["node", "test", "start", "--no-dashboard", "--no-orchestrator"]);
|
||||
|
||||
expect(mockWaitForPortAndOpen).not.toHaveBeenCalled();
|
||||
expect(mockEnsureLifecycleWorker).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -585,6 +609,10 @@ describe("stop command", () => {
|
|||
await program.parseAsync(["node", "test", "stop"]);
|
||||
|
||||
expect(mockSessionManager.kill).toHaveBeenCalledWith("app-orchestrator");
|
||||
expect(mockStopLifecycleWorker).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ configPath: expect.any(String) }),
|
||||
"my-app",
|
||||
);
|
||||
const output = vi
|
||||
.mocked(console.log)
|
||||
.mock.calls.map((c) => c.join(" "))
|
||||
|
|
@ -600,6 +628,10 @@ 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,63 @@
|
|||
import type { Command } from "commander";
|
||||
import chalk from "chalk";
|
||||
import { loadConfig } from "@composio/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();
|
||||
if (!config.projects[projectId]) {
|
||||
console.error(chalk.red(`Unknown project: ${projectId}`));
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
const existing = getLifecycleWorkerStatus(config, projectId);
|
||||
if (existing.running && existing.pid !== process.pid) {
|
||||
return;
|
||||
}
|
||||
|
||||
const lifecycle = await getLifecycleManager(config);
|
||||
const intervalMs = parseInterval(opts.intervalMs ?? "30000");
|
||||
let shuttingDown = false;
|
||||
|
||||
const shutdown = (code: number): void => {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
lifecycle.stop();
|
||||
clearLifecycleWorkerPid(config, projectId, process.pid);
|
||||
process.exit(code);
|
||||
};
|
||||
|
||||
process.on("SIGINT", () => shutdown(0));
|
||||
process.on("SIGTERM", () => shutdown(0));
|
||||
process.on("uncaughtException", (err) => {
|
||||
console.error(`[ao lifecycle] Worker crashed for ${projectId}:`, err);
|
||||
shutdown(1);
|
||||
});
|
||||
process.on("unhandledRejection", (reason) => {
|
||||
console.error(`[ao lifecycle] Worker crashed for ${projectId}:`, reason);
|
||||
shutdown(1);
|
||||
});
|
||||
|
||||
writeLifecycleWorkerPid(config, projectId, process.pid);
|
||||
console.log(
|
||||
`[ao lifecycle] Started for ${projectId} (pid=${process.pid}, interval=${intervalMs}ms)`,
|
||||
);
|
||||
lifecycle.start(intervalMs);
|
||||
});
|
||||
}
|
||||
|
|
@ -2,17 +2,19 @@ import chalk from "chalk";
|
|||
import ora from "ora";
|
||||
import type { Command } from "commander";
|
||||
import { loadConfig } from "@composio/ao-core";
|
||||
import { exec, gh } from "../lib/shell.js";
|
||||
import { gh } from "../lib/shell.js";
|
||||
import { getSessionManager } from "../lib/create-session-manager.js";
|
||||
|
||||
interface ReviewInfo {
|
||||
sessionId: string;
|
||||
tmuxTarget: string;
|
||||
prNumber: string;
|
||||
pendingComments: number;
|
||||
reviewDecision: string | null;
|
||||
}
|
||||
|
||||
const REVIEW_FIX_PROMPT =
|
||||
"There are review comments on your PR. Check with `gh pr view --comments` and `gh api` for inline comments. Address each one, push fixes, and reply.";
|
||||
|
||||
async function checkPRReviews(
|
||||
repo: string,
|
||||
prNumber: string,
|
||||
|
|
@ -93,7 +95,6 @@ export function registerReviewCheck(program: Command): void {
|
|||
if (pendingComments > 0 || reviewDecision === "CHANGES_REQUESTED") {
|
||||
results.push({
|
||||
sessionId: session.id,
|
||||
tmuxTarget: session.runtimeHandle?.id ?? session.id,
|
||||
prNumber: prNum,
|
||||
pendingComments,
|
||||
reviewDecision,
|
||||
|
|
@ -128,16 +129,7 @@ export function registerReviewCheck(program: Command): void {
|
|||
|
||||
if (!opts.dryRun) {
|
||||
try {
|
||||
// Interrupt busy agent and clear partial input before sending
|
||||
await exec("tmux", ["send-keys", "-t", result.tmuxTarget, "C-c"]);
|
||||
await new Promise((resolve) => setTimeout(resolve, 500));
|
||||
await exec("tmux", ["send-keys", "-t", result.tmuxTarget, "C-u"]);
|
||||
await new Promise((resolve) => setTimeout(resolve, 200));
|
||||
const message =
|
||||
"There are review comments on your PR. Check with `gh pr view --comments` and `gh api` for inline comments. Address each one, push fixes, and reply.";
|
||||
await exec("tmux", ["send-keys", "-t", result.tmuxTarget, "-l", message]);
|
||||
await new Promise((resolve) => setTimeout(resolve, 200));
|
||||
await exec("tmux", ["send-keys", "-t", result.tmuxTarget, "Enter"]);
|
||||
await sm.send(result.sessionId, REVIEW_FIX_PROMPT);
|
||||
console.log(chalk.green(` -> Fix prompt sent`));
|
||||
} catch (err) {
|
||||
console.error(chalk.red(` -> Failed to send: ${err}`));
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import { loadConfig, type OrchestratorConfig } from "@composio/ao-core";
|
|||
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";
|
||||
|
||||
interface SpawnClaimOptions {
|
||||
|
|
@ -151,6 +152,7 @@ export function registerSpawn(program: Command): void {
|
|||
|
||||
try {
|
||||
await runSpawnPreflight(config, projectId, { claimPr: opts.claimPr });
|
||||
await ensureLifecycleWorker(config, projectId);
|
||||
await spawnSession(config, projectId, issueId, opts.open, opts.agent, {
|
||||
claimPr: opts.claimPr,
|
||||
assignOnGithub: opts.assignOnGithub,
|
||||
|
|
@ -191,6 +193,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);
|
||||
} catch (err) {
|
||||
console.error(chalk.red(`✗ ${err instanceof Error ? err.message : String(err)}`));
|
||||
process.exit(1);
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ import {
|
|||
} from "@composio/ao-core";
|
||||
import { exec, execSilent } from "../lib/shell.js";
|
||||
import { getSessionManager } from "../lib/create-session-manager.js";
|
||||
import { ensureLifecycleWorker, stopLifecycleWorker } from "../lib/lifecycle-service.js";
|
||||
import { findWebDir, buildDashboardEnv, waitForPortAndOpen, isPortAvailable, findFreePort, MAX_PORT_SCAN } from "../lib/web-dir.js";
|
||||
import { cleanNextCache } from "../lib/dashboard-rebuild.js";
|
||||
import { preflight } from "../lib/preflight.js";
|
||||
|
|
@ -249,6 +250,8 @@ async function runStartup(
|
|||
opts?: { dashboard?: boolean; orchestrator?: boolean; rebuild?: boolean; autoPort?: boolean },
|
||||
): Promise<void> {
|
||||
const sessionId = `${project.sessionPrefix}-orchestrator`;
|
||||
const shouldStartLifecycle = opts?.dashboard !== false || opts?.orchestrator !== false;
|
||||
let lifecycleStatus: Awaited<ReturnType<typeof ensureLifecycleWorker>> | null = null;
|
||||
let port = config.port ?? DEFAULT_PORT;
|
||||
|
||||
console.log(chalk.bold(`\nStarting orchestrator for ${chalk.cyan(project.name)}\n`));
|
||||
|
|
@ -296,6 +299,27 @@ async function runStartup(
|
|||
console.log(chalk.dim(" (Dashboard will be ready in a few seconds)\n"));
|
||||
}
|
||||
|
||||
if (shouldStartLifecycle) {
|
||||
try {
|
||||
spinner.start("Starting lifecycle worker");
|
||||
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})` : ""}`,
|
||||
);
|
||||
} catch (err) {
|
||||
spinner.fail("Lifecycle worker failed to start");
|
||||
if (dashboardProcess) {
|
||||
dashboardProcess.kill();
|
||||
}
|
||||
throw new Error(
|
||||
`Failed to start lifecycle worker: ${err instanceof Error ? err.message : String(err)}`,
|
||||
{ cause: err },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Create orchestrator session (unless --no-orchestrator or already exists)
|
||||
let tmuxTarget = sessionId;
|
||||
if (opts?.orchestrator !== false) {
|
||||
|
|
@ -341,6 +365,14 @@ async function runStartup(
|
|||
console.log(chalk.cyan("Dashboard:"), `http://localhost:${port}`);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
if (opts?.orchestrator !== false && !exists) {
|
||||
console.log(chalk.cyan("Orchestrator:"), `tmux attach -t ${tmuxTarget}`);
|
||||
} else if (exists) {
|
||||
|
|
@ -481,6 +513,13 @@ 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"));
|
||||
}
|
||||
|
||||
// Stop dashboard
|
||||
await stopDashboard(port);
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ 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";
|
||||
|
||||
const program = new Command();
|
||||
|
||||
|
|
@ -29,5 +30,6 @@ registerSend(program);
|
|||
registerReviewCheck(program);
|
||||
registerDashboard(program);
|
||||
registerOpen(program);
|
||||
registerLifecycleWorker(program);
|
||||
|
||||
program.parse();
|
||||
|
|
|
|||
|
|
@ -10,9 +10,11 @@
|
|||
import {
|
||||
createPluginRegistry,
|
||||
createSessionManager,
|
||||
createLifecycleManager,
|
||||
type OrchestratorConfig,
|
||||
type SessionManager,
|
||||
type PluginRegistry,
|
||||
type LifecycleManager,
|
||||
} from "@composio/ao-core";
|
||||
|
||||
let registryPromise: Promise<PluginRegistry> | null = null;
|
||||
|
|
@ -22,7 +24,7 @@ let registryPromise: Promise<PluginRegistry> | null = null;
|
|||
* Caches the Promise (not the resolved value) so concurrent callers
|
||||
* await the same initialization rather than racing.
|
||||
*/
|
||||
async function getRegistry(config: OrchestratorConfig): Promise<PluginRegistry> {
|
||||
export async function getRegistry(config: OrchestratorConfig): Promise<PluginRegistry> {
|
||||
if (!registryPromise) {
|
||||
registryPromise = (async () => {
|
||||
const registry = createPluginRegistry();
|
||||
|
|
@ -45,3 +47,14 @@ export async function getSessionManager(config: OrchestratorConfig): Promise<Ses
|
|||
return createSessionManager({ config, registry });
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a LifecycleManager backed by core's implementation.
|
||||
* Shares the same plugin registry initialization path as SessionManager.
|
||||
*/
|
||||
export async function getLifecycleManager(
|
||||
config: OrchestratorConfig,
|
||||
): Promise<LifecycleManager> {
|
||||
const registry = await getRegistry(config);
|
||||
const sessionManager = createSessionManager({ config, registry });
|
||||
return createLifecycleManager({ config, registry, sessionManager });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,246 @@
|
|||
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 "@composio/ao-core";
|
||||
|
||||
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;
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
export async function ensureLifecycleWorker(
|
||||
config: OrchestratorConfig,
|
||||
projectId: string,
|
||||
): Promise<LifecycleWorkerStatus & { started: boolean }> {
|
||||
const current = getLifecycleWorkerStatus(config, projectId);
|
||||
if (current.running) {
|
||||
return { ...current, started: false };
|
||||
}
|
||||
|
||||
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,
|
||||
},
|
||||
});
|
||||
|
||||
child.unref();
|
||||
} finally {
|
||||
closeSync(stdoutFd);
|
||||
closeSync(stderrFd);
|
||||
}
|
||||
|
||||
const status = await waitForLifecycleWorker(config, projectId);
|
||||
if (!status.running) {
|
||||
throw new Error(
|
||||
`Lifecycle worker failed to start for project ${projectId}. See ${status.logFile}`,
|
||||
);
|
||||
}
|
||||
|
||||
return { ...status, 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;
|
||||
}
|
||||
await sleep(100);
|
||||
}
|
||||
|
||||
try {
|
||||
process.kill(status.pid, "SIGKILL");
|
||||
} catch {
|
||||
// Best effort hard stop
|
||||
}
|
||||
|
||||
clearLifecycleWorkerPid(config, projectId, status.pid);
|
||||
return true;
|
||||
}
|
||||
|
|
@ -830,6 +830,229 @@ describe("reactions", () => {
|
|||
expect(mockNotifier.notify).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("dispatches unresolved review comments even when reviewDecision stays unchanged", async () => {
|
||||
config.reactions = {
|
||||
"changes-requested": {
|
||||
auto: true,
|
||||
action: "send-to-agent",
|
||||
message: "Handle review comments.",
|
||||
},
|
||||
};
|
||||
|
||||
const mockSCM: SCM = {
|
||||
name: "mock-scm",
|
||||
detectPR: vi.fn(),
|
||||
getPRState: vi.fn().mockResolvedValue("open"),
|
||||
mergePR: vi.fn(),
|
||||
closePR: vi.fn(),
|
||||
getCIChecks: vi.fn(),
|
||||
getCISummary: vi.fn().mockResolvedValue("passing"),
|
||||
getReviews: vi.fn(),
|
||||
getReviewDecision: vi.fn().mockResolvedValue("none"),
|
||||
getPendingComments: vi.fn().mockResolvedValue([
|
||||
{
|
||||
id: "c1",
|
||||
author: "reviewer",
|
||||
body: "Please rename this helper",
|
||||
path: "src/app.ts",
|
||||
line: 12,
|
||||
isResolved: false,
|
||||
createdAt: new Date(),
|
||||
url: "https://example.com/comment/1",
|
||||
},
|
||||
]),
|
||||
getAutomatedComments: vi.fn().mockResolvedValue([]),
|
||||
getMergeability: vi.fn(),
|
||||
};
|
||||
|
||||
const registryWithSCM: PluginRegistry = {
|
||||
...mockRegistry,
|
||||
get: vi.fn().mockImplementation((slot: string) => {
|
||||
if (slot === "runtime") return mockRuntime;
|
||||
if (slot === "agent") return mockAgent;
|
||||
if (slot === "scm") return mockSCM;
|
||||
return null;
|
||||
}),
|
||||
};
|
||||
|
||||
const session = makeSession({ status: "pr_open", pr: makePR() });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
vi.mocked(mockSessionManager.send).mockResolvedValue(undefined);
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "pr_open",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
|
||||
expect(mockSessionManager.send).toHaveBeenCalledWith("app-1", "Handle review comments.");
|
||||
|
||||
vi.mocked(mockSessionManager.send).mockClear();
|
||||
await lm.check("app-1");
|
||||
expect(mockSessionManager.send).not.toHaveBeenCalled();
|
||||
|
||||
const metadata = readMetadataRaw(sessionsDir, "app-1");
|
||||
expect(metadata?.["lastPendingReviewDispatchHash"]).toBe("c1");
|
||||
});
|
||||
|
||||
it("does not double-send when changes_requested transition already triggered the reaction", async () => {
|
||||
config.reactions = {
|
||||
"changes-requested": {
|
||||
auto: true,
|
||||
action: "send-to-agent",
|
||||
message: "Handle requested changes.",
|
||||
},
|
||||
};
|
||||
|
||||
const mockSCM: SCM = {
|
||||
name: "mock-scm",
|
||||
detectPR: vi.fn(),
|
||||
getPRState: vi.fn().mockResolvedValue("open"),
|
||||
mergePR: vi.fn(),
|
||||
closePR: vi.fn(),
|
||||
getCIChecks: vi.fn(),
|
||||
getCISummary: vi.fn().mockResolvedValue("passing"),
|
||||
getReviews: vi.fn(),
|
||||
getReviewDecision: vi.fn().mockResolvedValue("changes_requested"),
|
||||
getPendingComments: vi.fn().mockResolvedValue([
|
||||
{
|
||||
id: "c1",
|
||||
author: "reviewer",
|
||||
body: "Please add validation",
|
||||
path: "src/route.ts",
|
||||
line: 44,
|
||||
isResolved: false,
|
||||
createdAt: new Date(),
|
||||
url: "https://example.com/comment/2",
|
||||
},
|
||||
]),
|
||||
getAutomatedComments: vi.fn().mockResolvedValue([]),
|
||||
getMergeability: vi.fn(),
|
||||
};
|
||||
|
||||
const registryWithSCM: PluginRegistry = {
|
||||
...mockRegistry,
|
||||
get: vi.fn().mockImplementation((slot: string) => {
|
||||
if (slot === "runtime") return mockRuntime;
|
||||
if (slot === "agent") return mockAgent;
|
||||
if (slot === "scm") return mockSCM;
|
||||
return null;
|
||||
}),
|
||||
};
|
||||
|
||||
const session = makeSession({ status: "pr_open", pr: makePR() });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
vi.mocked(mockSessionManager.send).mockResolvedValue(undefined);
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "pr_open",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
await lm.check("app-1");
|
||||
|
||||
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
|
||||
expect(mockSessionManager.send).toHaveBeenCalledWith(
|
||||
"app-1",
|
||||
"Handle requested changes.",
|
||||
);
|
||||
});
|
||||
|
||||
it("dispatches automated review comments only once for an unchanged backlog", async () => {
|
||||
config.reactions = {
|
||||
"bugbot-comments": {
|
||||
auto: true,
|
||||
action: "send-to-agent",
|
||||
message: "Handle automated review findings.",
|
||||
},
|
||||
};
|
||||
|
||||
const mockSCM: SCM = {
|
||||
name: "mock-scm",
|
||||
detectPR: vi.fn(),
|
||||
getPRState: vi.fn().mockResolvedValue("open"),
|
||||
mergePR: vi.fn(),
|
||||
closePR: vi.fn(),
|
||||
getCIChecks: vi.fn(),
|
||||
getCISummary: vi.fn().mockResolvedValue("passing"),
|
||||
getReviews: vi.fn(),
|
||||
getReviewDecision: vi.fn().mockResolvedValue("none"),
|
||||
getPendingComments: vi.fn().mockResolvedValue([]),
|
||||
getAutomatedComments: vi.fn().mockResolvedValue([
|
||||
{
|
||||
id: "bot-1",
|
||||
botName: "cursor[bot]",
|
||||
body: "Potential issue detected",
|
||||
path: "src/worker.ts",
|
||||
line: 9,
|
||||
severity: "warning",
|
||||
createdAt: new Date(),
|
||||
url: "https://example.com/comment/3",
|
||||
},
|
||||
]),
|
||||
getMergeability: vi.fn(),
|
||||
};
|
||||
|
||||
const registryWithSCM: PluginRegistry = {
|
||||
...mockRegistry,
|
||||
get: vi.fn().mockImplementation((slot: string) => {
|
||||
if (slot === "runtime") return mockRuntime;
|
||||
if (slot === "agent") return mockAgent;
|
||||
if (slot === "scm") return mockSCM;
|
||||
return null;
|
||||
}),
|
||||
};
|
||||
|
||||
const session = makeSession({ status: "pr_open", pr: makePR() });
|
||||
vi.mocked(mockSessionManager.get).mockResolvedValue(session);
|
||||
vi.mocked(mockSessionManager.send).mockResolvedValue(undefined);
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "pr_open",
|
||||
project: "my-app",
|
||||
});
|
||||
|
||||
const lm = createLifecycleManager({
|
||||
config,
|
||||
registry: registryWithSCM,
|
||||
sessionManager: mockSessionManager,
|
||||
});
|
||||
|
||||
await lm.check("app-1");
|
||||
expect(mockSessionManager.send).toHaveBeenCalledTimes(1);
|
||||
expect(mockSessionManager.send).toHaveBeenCalledWith(
|
||||
"app-1",
|
||||
"Handle automated review findings.",
|
||||
);
|
||||
|
||||
vi.mocked(mockSessionManager.send).mockClear();
|
||||
await lm.check("app-1");
|
||||
expect(mockSessionManager.send).not.toHaveBeenCalled();
|
||||
|
||||
const metadata = readMetadataRaw(sessionsDir, "app-1");
|
||||
expect(metadata?.["lastAutomatedReviewDispatchHash"]).toBe("bot-1");
|
||||
});
|
||||
|
||||
it("notifies humans on significant transitions without reaction config", async () => {
|
||||
const mockNotifier: Notifier = {
|
||||
name: "mock-notifier",
|
||||
|
|
|
|||
|
|
@ -1228,7 +1228,7 @@ describe("cleanup", () => {
|
|||
});
|
||||
|
||||
describe("send", () => {
|
||||
it("sends message via runtime.sendMessage", async () => {
|
||||
it("sends message via runtime.sendMessage and confirms delivery", async () => {
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
|
|
@ -1236,6 +1236,9 @@ describe("send", () => {
|
|||
project: "my-app",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
vi.mocked(mockRuntime.getOutput)
|
||||
.mockResolvedValueOnce("before")
|
||||
.mockResolvedValueOnce("after");
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.send("app-1", "Fix the CI failures");
|
||||
|
|
@ -1243,6 +1246,56 @@ describe("send", () => {
|
|||
expect(mockRuntime.sendMessage).toHaveBeenCalledWith(makeHandle("rt-1"), "Fix the CI failures");
|
||||
});
|
||||
|
||||
it("restores a dead session before sending the message", async () => {
|
||||
const wsPath = join(tmpDir, "ws-app-1");
|
||||
mkdirSync(wsPath, { recursive: true });
|
||||
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: wsPath,
|
||||
branch: "feat/TEST-1",
|
||||
status: "working",
|
||||
project: "my-app",
|
||||
issue: "TEST-1",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-old")),
|
||||
});
|
||||
|
||||
vi.mocked(mockRuntime.isAlive).mockImplementation(async (handle) => handle.id !== "rt-old");
|
||||
vi.mocked(mockAgent.isProcessRunning).mockImplementation(
|
||||
async (handle) => handle.id !== "rt-old",
|
||||
);
|
||||
vi.mocked(mockRuntime.create).mockResolvedValue(makeHandle("rt-restored"));
|
||||
vi.mocked(mockRuntime.getOutput)
|
||||
.mockResolvedValueOnce("restored prompt")
|
||||
.mockResolvedValueOnce("before send")
|
||||
.mockResolvedValueOnce("after send");
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.send("app-1", "Please fix the review comments");
|
||||
|
||||
expect(mockRuntime.create).toHaveBeenCalled();
|
||||
expect(mockRuntime.sendMessage).toHaveBeenCalledWith(
|
||||
makeHandle("rt-restored"),
|
||||
"Please fix the review comments",
|
||||
);
|
||||
});
|
||||
|
||||
it("throws when delivery cannot be confirmed", async () => {
|
||||
writeMetadata(sessionsDir, "app-1", {
|
||||
worktree: "/tmp",
|
||||
branch: "main",
|
||||
status: "working",
|
||||
project: "my-app",
|
||||
runtimeHandle: JSON.stringify(makeHandle("rt-1")),
|
||||
});
|
||||
vi.mocked(mockRuntime.getOutput).mockResolvedValue("steady output");
|
||||
vi.mocked(mockAgent.detectActivity).mockReturnValue("idle");
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await expect(sm.send("app-1", "Fix the CI failures")).rejects.toThrow(
|
||||
"could not be confirmed",
|
||||
);
|
||||
});
|
||||
|
||||
it("throws for nonexistent session", async () => {
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await expect(sm.send("nope", "hello")).rejects.toThrow("not found");
|
||||
|
|
@ -1255,10 +1308,13 @@ describe("send", () => {
|
|||
status: "working",
|
||||
project: "my-app",
|
||||
});
|
||||
vi.mocked(mockRuntime.getOutput)
|
||||
.mockResolvedValueOnce("before")
|
||||
.mockResolvedValueOnce("after");
|
||||
|
||||
const sm = createSessionManager({ config, registry: mockRegistry });
|
||||
await sm.send("app-1", "hello");
|
||||
// Should use session ID "app-1" as the handle id with default runtime
|
||||
|
||||
expect(mockRuntime.sendMessage).toHaveBeenCalledWith(
|
||||
{ id: "app-1", runtimeName: "mock", data: {} },
|
||||
"hello",
|
||||
|
|
|
|||
|
|
@ -417,6 +417,190 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
};
|
||||
}
|
||||
|
||||
function clearReactionTracker(sessionId: SessionId, reactionKey: string): void {
|
||||
reactionTrackers.delete(`${sessionId}:${reactionKey}`);
|
||||
}
|
||||
|
||||
function getReactionConfigForSession(
|
||||
session: Session,
|
||||
reactionKey: string,
|
||||
): ReactionConfig | null {
|
||||
const project = config.projects[session.projectId];
|
||||
const globalReaction = config.reactions[reactionKey];
|
||||
const projectReaction = project?.reactions?.[reactionKey];
|
||||
const reactionConfig = projectReaction
|
||||
? { ...globalReaction, ...projectReaction }
|
||||
: globalReaction;
|
||||
return reactionConfig ? (reactionConfig as ReactionConfig) : null;
|
||||
}
|
||||
|
||||
function updateSessionMetadata(
|
||||
session: Session,
|
||||
updates: Partial<Record<string, string>>,
|
||||
): void {
|
||||
const project = config.projects[session.projectId];
|
||||
if (!project) return;
|
||||
|
||||
const sessionsDir = getSessionsDir(config.configPath, project.path);
|
||||
updateMetadata(sessionsDir, session.id, updates);
|
||||
|
||||
for (const [key, value] of Object.entries(updates)) {
|
||||
if (value === undefined) continue;
|
||||
if (value === "") {
|
||||
delete session.metadata[key];
|
||||
} else {
|
||||
session.metadata[key] = value;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function makeFingerprint(ids: string[]): string {
|
||||
return [...ids].sort().join(",");
|
||||
}
|
||||
|
||||
async function maybeDispatchReviewBacklog(
|
||||
session: Session,
|
||||
oldStatus: SessionStatus,
|
||||
newStatus: SessionStatus,
|
||||
transitionReaction?: { key: string; result: ReactionResult | null },
|
||||
): Promise<void> {
|
||||
const project = config.projects[session.projectId];
|
||||
if (!project || !session.pr) return;
|
||||
|
||||
const scm = project.scm ? registry.get<SCM>("scm", project.scm.plugin) : null;
|
||||
if (!scm) return;
|
||||
|
||||
const humanReactionKey = "changes-requested";
|
||||
const automatedReactionKey = "bugbot-comments";
|
||||
|
||||
if (newStatus === "merged" || newStatus === "killed") {
|
||||
clearReactionTracker(session.id, humanReactionKey);
|
||||
clearReactionTracker(session.id, automatedReactionKey);
|
||||
updateSessionMetadata(session, {
|
||||
lastPendingReviewFingerprint: "",
|
||||
lastPendingReviewDispatchHash: "",
|
||||
lastPendingReviewDispatchAt: "",
|
||||
lastAutomatedReviewFingerprint: "",
|
||||
lastAutomatedReviewDispatchHash: "",
|
||||
lastAutomatedReviewDispatchAt: "",
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const [pendingResult, automatedResult] = await Promise.allSettled([
|
||||
scm.getPendingComments(session.pr),
|
||||
scm.getAutomatedComments(session.pr),
|
||||
]);
|
||||
|
||||
const pendingComments =
|
||||
pendingResult.status === "fulfilled" && Array.isArray(pendingResult.value)
|
||||
? pendingResult.value
|
||||
: [];
|
||||
const automatedComments =
|
||||
automatedResult.status === "fulfilled" && Array.isArray(automatedResult.value)
|
||||
? automatedResult.value
|
||||
: [];
|
||||
|
||||
const pendingFingerprint = makeFingerprint(pendingComments.map((comment) => comment.id));
|
||||
const lastPendingFingerprint = session.metadata["lastPendingReviewFingerprint"] ?? "";
|
||||
const lastPendingDispatchHash = session.metadata["lastPendingReviewDispatchHash"] ?? "";
|
||||
|
||||
if (
|
||||
pendingFingerprint !== lastPendingFingerprint &&
|
||||
transitionReaction?.key !== humanReactionKey
|
||||
) {
|
||||
clearReactionTracker(session.id, humanReactionKey);
|
||||
}
|
||||
if (pendingFingerprint !== lastPendingFingerprint) {
|
||||
updateSessionMetadata(session, {
|
||||
lastPendingReviewFingerprint: pendingFingerprint,
|
||||
});
|
||||
}
|
||||
|
||||
if (!pendingFingerprint) {
|
||||
clearReactionTracker(session.id, humanReactionKey);
|
||||
updateSessionMetadata(session, {
|
||||
lastPendingReviewFingerprint: "",
|
||||
lastPendingReviewDispatchHash: "",
|
||||
lastPendingReviewDispatchAt: "",
|
||||
});
|
||||
} else if (
|
||||
transitionReaction?.key === humanReactionKey &&
|
||||
transitionReaction.result?.success
|
||||
) {
|
||||
if (lastPendingDispatchHash !== pendingFingerprint) {
|
||||
updateSessionMetadata(session, {
|
||||
lastPendingReviewDispatchHash: pendingFingerprint,
|
||||
lastPendingReviewDispatchAt: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
} else if (
|
||||
!(oldStatus !== newStatus && newStatus === "changes_requested") &&
|
||||
pendingFingerprint !== lastPendingDispatchHash
|
||||
) {
|
||||
const reactionConfig = getReactionConfigForSession(session, humanReactionKey);
|
||||
if (
|
||||
reactionConfig &&
|
||||
reactionConfig.action &&
|
||||
(reactionConfig.auto !== false || reactionConfig.action === "notify")
|
||||
) {
|
||||
const result = await executeReaction(
|
||||
session.id,
|
||||
session.projectId,
|
||||
humanReactionKey,
|
||||
reactionConfig,
|
||||
);
|
||||
if (result.success) {
|
||||
updateSessionMetadata(session, {
|
||||
lastPendingReviewDispatchHash: pendingFingerprint,
|
||||
lastPendingReviewDispatchAt: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const automatedFingerprint = makeFingerprint(automatedComments.map((comment) => comment.id));
|
||||
const lastAutomatedFingerprint = session.metadata["lastAutomatedReviewFingerprint"] ?? "";
|
||||
const lastAutomatedDispatchHash =
|
||||
session.metadata["lastAutomatedReviewDispatchHash"] ?? "";
|
||||
|
||||
if (automatedFingerprint !== lastAutomatedFingerprint) {
|
||||
clearReactionTracker(session.id, automatedReactionKey);
|
||||
updateSessionMetadata(session, {
|
||||
lastAutomatedReviewFingerprint: automatedFingerprint,
|
||||
});
|
||||
}
|
||||
|
||||
if (!automatedFingerprint) {
|
||||
clearReactionTracker(session.id, automatedReactionKey);
|
||||
updateSessionMetadata(session, {
|
||||
lastAutomatedReviewFingerprint: "",
|
||||
lastAutomatedReviewDispatchHash: "",
|
||||
lastAutomatedReviewDispatchAt: "",
|
||||
});
|
||||
} else if (automatedFingerprint !== lastAutomatedDispatchHash) {
|
||||
const reactionConfig = getReactionConfigForSession(session, automatedReactionKey);
|
||||
if (
|
||||
reactionConfig &&
|
||||
reactionConfig.action &&
|
||||
(reactionConfig.auto !== false || reactionConfig.action === "notify")
|
||||
) {
|
||||
const result = await executeReaction(
|
||||
session.id,
|
||||
session.projectId,
|
||||
automatedReactionKey,
|
||||
reactionConfig,
|
||||
);
|
||||
if (result.success) {
|
||||
updateSessionMetadata(session, {
|
||||
lastAutomatedReviewDispatchHash: automatedFingerprint,
|
||||
lastAutomatedReviewDispatchAt: new Date().toISOString(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Send a notification to all configured notifiers. */
|
||||
async function notifyHuman(event: OrchestratorEvent, priority: EventPriority): Promise<void> {
|
||||
const eventWithPriority = { ...event, priority };
|
||||
|
|
@ -443,17 +627,12 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
const oldStatus =
|
||||
tracked ?? ((session.metadata?.["status"] as SessionStatus | undefined) || session.status);
|
||||
const newStatus = await determineStatus(session);
|
||||
let transitionReaction: { key: string; result: ReactionResult | null } | undefined;
|
||||
|
||||
if (newStatus !== oldStatus) {
|
||||
// State transition detected
|
||||
states.set(session.id, newStatus);
|
||||
|
||||
// Update metadata — session.projectId is the config key (e.g., "my-app")
|
||||
const project = config.projects[session.projectId];
|
||||
if (project) {
|
||||
const sessionsDir = getSessionsDir(config.configPath, project.path);
|
||||
updateMetadata(sessionsDir, session.id, { status: newStatus });
|
||||
}
|
||||
updateSessionMetadata(session, { status: newStatus });
|
||||
|
||||
// Reset allCompleteEmitted when any session becomes active again
|
||||
if (newStatus !== "merged" && newStatus !== "killed") {
|
||||
|
|
@ -465,7 +644,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
if (oldEventType) {
|
||||
const oldReactionKey = eventToReactionKey(oldEventType);
|
||||
if (oldReactionKey) {
|
||||
reactionTrackers.delete(`${session.id}:${oldReactionKey}`);
|
||||
clearReactionTracker(session.id, oldReactionKey);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -476,23 +655,18 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
const reactionKey = eventToReactionKey(eventType);
|
||||
|
||||
if (reactionKey) {
|
||||
// Merge project-specific overrides with global defaults
|
||||
const project = config.projects[session.projectId];
|
||||
const globalReaction = config.reactions[reactionKey];
|
||||
const projectReaction = project?.reactions?.[reactionKey];
|
||||
const reactionConfig = projectReaction
|
||||
? { ...globalReaction, ...projectReaction }
|
||||
: globalReaction;
|
||||
const reactionConfig = getReactionConfigForSession(session, reactionKey);
|
||||
|
||||
if (reactionConfig && reactionConfig.action) {
|
||||
// auto: false skips automated agent actions but still allows notifications
|
||||
if (reactionConfig.auto !== false || reactionConfig.action === "notify") {
|
||||
await executeReaction(
|
||||
const reactionResult = await executeReaction(
|
||||
session.id,
|
||||
session.projectId,
|
||||
reactionKey,
|
||||
reactionConfig as ReactionConfig,
|
||||
reactionConfig,
|
||||
);
|
||||
transitionReaction = { key: reactionKey, result: reactionResult };
|
||||
// Reaction is handling this event — suppress immediate human notification.
|
||||
// "send-to-agent" retries + escalates on its own; "notify"/"auto-merge"
|
||||
// already call notifyHuman internally. Notifying here would bypass the
|
||||
|
|
@ -520,6 +694,8 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan
|
|||
// No transition but track current state
|
||||
states.set(session.id, newStatus);
|
||||
}
|
||||
|
||||
await maybeDispatchReviewBacklog(session, oldStatus, newStatus, transitionReaction);
|
||||
}
|
||||
|
||||
/** Run one polling cycle across all sessions. */
|
||||
|
|
|
|||
|
|
@ -117,6 +117,16 @@ const PR_TRACKING_STATUSES: ReadonlySet<string> = new Set([
|
|||
"mergeable",
|
||||
]);
|
||||
|
||||
const SEND_RESTORE_READY_TIMEOUT_MS = 5_000;
|
||||
const SEND_RESTORE_READY_POLL_MS = 500;
|
||||
const SEND_CONFIRMATION_ATTEMPTS = 3;
|
||||
const SEND_CONFIRMATION_POLL_MS = 500;
|
||||
const SEND_CONFIRMATION_OUTPUT_LINES = 20;
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
/** Validate and normalize a status string. */
|
||||
function validateStatus(raw: string | undefined): SessionStatus {
|
||||
// Bash scripts write "starting" — treat as "working"
|
||||
|
|
@ -994,44 +1004,186 @@ export function createSessionManager(deps: SessionManagerDeps): SessionManager {
|
|||
}
|
||||
|
||||
async function send(sessionId: SessionId, message: string): Promise<void> {
|
||||
// Find the session in any project's sessions directory
|
||||
let raw: Record<string, string> | null = null;
|
||||
for (const project of Object.values(config.projects)) {
|
||||
const sessionsDir = getProjectSessionsDir(project);
|
||||
const metadata = readMetadataRaw(sessionsDir, sessionId);
|
||||
if (metadata) {
|
||||
raw = metadata;
|
||||
break;
|
||||
}
|
||||
const located = findSessionRecord(sessionId);
|
||||
if (!located) {
|
||||
throw new Error(`Session ${sessionId} not found`);
|
||||
}
|
||||
|
||||
if (!raw) throw new Error(`Session ${sessionId} not found`);
|
||||
const { raw, project } = located;
|
||||
const parsedHandle = raw["runtimeHandle"]
|
||||
? safeJsonParse<RuntimeHandle>(raw["runtimeHandle"])
|
||||
: null;
|
||||
const runtimeName = parsedHandle?.runtimeName ?? project.runtime ?? config.defaults.runtime;
|
||||
const agentName = raw["agent"] ?? project.agent ?? config.defaults.agent;
|
||||
|
||||
// Build handle: use stored runtimeHandle, or fall back to session ID as tmux session name
|
||||
let handle: RuntimeHandle;
|
||||
if (raw["runtimeHandle"]) {
|
||||
const parsed = safeJsonParse<RuntimeHandle>(raw["runtimeHandle"]);
|
||||
if (!parsed) {
|
||||
throw new Error(`Corrupted runtime handle for session ${sessionId}`);
|
||||
}
|
||||
handle = parsed;
|
||||
} else {
|
||||
// Sessions created by bash scripts don't have runtimeHandle — use session ID as tmux handle
|
||||
handle = { id: sessionId, runtimeName: config.defaults.runtime, data: {} };
|
||||
}
|
||||
|
||||
// Prefer handle.runtimeName to find the correct plugin
|
||||
const project = config.projects[raw["project"] ?? ""];
|
||||
const runtimePlugin = registry.get<Runtime>(
|
||||
"runtime",
|
||||
handle.runtimeName ??
|
||||
(project ? (project.runtime ?? config.defaults.runtime) : config.defaults.runtime),
|
||||
);
|
||||
const runtimePlugin = registry.get<Runtime>("runtime", runtimeName);
|
||||
if (!runtimePlugin) {
|
||||
throw new Error(`No runtime plugin for session ${sessionId}`);
|
||||
}
|
||||
|
||||
await runtimePlugin.sendMessage(handle, message);
|
||||
const agentPlugin = registry.get<Agent>("agent", agentName);
|
||||
if (!agentPlugin) {
|
||||
throw new Error(`No agent plugin for session ${sessionId}`);
|
||||
}
|
||||
|
||||
const captureOutput = async (handle: RuntimeHandle): Promise<string> => {
|
||||
try {
|
||||
return (await runtimePlugin.getOutput(handle, SEND_CONFIRMATION_OUTPUT_LINES)) ?? "";
|
||||
} catch {
|
||||
return "";
|
||||
}
|
||||
};
|
||||
|
||||
const detectActivityFromOutput = (output: string) => {
|
||||
if (!output) return null;
|
||||
try {
|
||||
return agentPlugin.detectActivity(output);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
};
|
||||
|
||||
const hasQueuedMessage = (output: string): boolean => {
|
||||
return output.includes("Press up to edit queued messages");
|
||||
};
|
||||
|
||||
const waitForRestoredSession = async (restoredSession: Session): Promise<void> => {
|
||||
const handle = restoredSession.runtimeHandle;
|
||||
if (!handle) {
|
||||
return;
|
||||
}
|
||||
|
||||
const deadline = Date.now() + SEND_RESTORE_READY_TIMEOUT_MS;
|
||||
while (true) {
|
||||
const [runtimeAlive, processRunning, output] = await Promise.all([
|
||||
runtimePlugin.isAlive(handle).catch(() => true),
|
||||
agentPlugin.isProcessRunning(handle).catch(() => true),
|
||||
captureOutput(handle),
|
||||
]);
|
||||
|
||||
if (runtimeAlive && (processRunning || output.trim().length > 0)) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (Date.now() >= deadline) {
|
||||
return;
|
||||
}
|
||||
|
||||
await sleep(SEND_RESTORE_READY_POLL_MS);
|
||||
}
|
||||
};
|
||||
|
||||
const restoreForDelivery = async (reason: string, session: Session): Promise<Session> => {
|
||||
if (NON_RESTORABLE_STATUSES.has(session.status)) {
|
||||
throw new Error(`Cannot send to session ${sessionId}: ${reason}`);
|
||||
}
|
||||
|
||||
try {
|
||||
const restored = await restore(sessionId);
|
||||
await waitForRestoredSession(restored);
|
||||
return restored;
|
||||
} catch (err) {
|
||||
const detail = err instanceof Error ? err.message : String(err);
|
||||
throw new Error(`Cannot send to session ${sessionId}: ${reason} (${detail})`);
|
||||
}
|
||||
};
|
||||
|
||||
const prepareSession = async (forceRestore = false): Promise<Session> => {
|
||||
const current = await get(sessionId);
|
||||
if (!current) {
|
||||
throw new Error(`Session ${sessionId} not found`);
|
||||
}
|
||||
|
||||
const handle =
|
||||
current.runtimeHandle ??
|
||||
({
|
||||
id: sessionId,
|
||||
runtimeName,
|
||||
data: {},
|
||||
} satisfies RuntimeHandle);
|
||||
const normalized = current.runtimeHandle ? current : { ...current, runtimeHandle: handle };
|
||||
|
||||
if (forceRestore || isRestorable(normalized)) {
|
||||
return restoreForDelivery(
|
||||
forceRestore ? "session needed to be restarted before delivery" : "session is not running",
|
||||
normalized,
|
||||
);
|
||||
}
|
||||
|
||||
const [runtimeAlive, processRunning] = await Promise.all([
|
||||
runtimePlugin.isAlive(handle).catch(() => true),
|
||||
agentPlugin.isProcessRunning(handle).catch(() => true),
|
||||
]);
|
||||
|
||||
if (!runtimeAlive || !processRunning) {
|
||||
return restoreForDelivery(
|
||||
!runtimeAlive ? "runtime is not alive" : "agent process is not running",
|
||||
normalized,
|
||||
);
|
||||
}
|
||||
|
||||
return normalized;
|
||||
};
|
||||
|
||||
const sendWithConfirmation = async (session: Session): Promise<void> => {
|
||||
const handle = session.runtimeHandle;
|
||||
if (!handle) {
|
||||
throw new Error(`Session ${sessionId} has no runtime handle`);
|
||||
}
|
||||
|
||||
const baselineOutput = await captureOutput(handle);
|
||||
const baselineActivity = detectActivityFromOutput(baselineOutput) ?? session.activity;
|
||||
|
||||
await runtimePlugin.sendMessage(handle, message);
|
||||
|
||||
for (let attempt = 1; attempt <= SEND_CONFIRMATION_ATTEMPTS; attempt++) {
|
||||
const output = await captureOutput(handle);
|
||||
const activity = detectActivityFromOutput(output) ?? session.activity;
|
||||
const delivered =
|
||||
hasQueuedMessage(output) ||
|
||||
(output.length > 0 && output !== baselineOutput) ||
|
||||
(baselineActivity !== "active" && activity === "active") ||
|
||||
(baselineActivity !== "waiting_input" && activity === "waiting_input");
|
||||
|
||||
if (delivered) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (attempt < SEND_CONFIRMATION_ATTEMPTS) {
|
||||
await sleep(SEND_CONFIRMATION_POLL_MS);
|
||||
}
|
||||
}
|
||||
|
||||
throw new Error(`Message to session ${sessionId} could not be confirmed`);
|
||||
};
|
||||
|
||||
let prepared = await prepareSession();
|
||||
|
||||
try {
|
||||
await sendWithConfirmation(prepared);
|
||||
} catch (err) {
|
||||
const shouldRetryWithRestore =
|
||||
!(err instanceof Error && err.message.includes("could not be confirmed")) &&
|
||||
prepared.restoredAt === undefined &&
|
||||
!NON_RESTORABLE_STATUSES.has(prepared.status);
|
||||
|
||||
if (!shouldRetryWithRestore) {
|
||||
if (err instanceof Error) {
|
||||
throw err;
|
||||
}
|
||||
throw new Error(String(err));
|
||||
}
|
||||
|
||||
prepared = await prepareSession(true);
|
||||
try {
|
||||
await sendWithConfirmation(prepared);
|
||||
} catch (retryErr) {
|
||||
if (retryErr instanceof Error) {
|
||||
throw retryErr;
|
||||
}
|
||||
throw new Error(String(retryErr));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function claimPR(
|
||||
|
|
|
|||
Loading…
Reference in New Issue