From a6c9dbd5996df6f36ddedb45e2b5e984cb64927f Mon Sep 17 00:00:00 2001 From: Jayesh Sharma Date: Fri, 6 Mar 2026 21:35:18 +0530 Subject: [PATCH] feat: add lifecycle worker automation --- .../__tests__/commands/review-check.test.ts | 13 +- packages/cli/__tests__/commands/spawn.test.ts | 20 +- packages/cli/__tests__/commands/start.test.ts | 32 +++ packages/cli/src/commands/lifecycle-worker.ts | 63 +++++ packages/cli/src/commands/review-check.ts | 18 +- packages/cli/src/commands/spawn.ts | 3 + packages/cli/src/commands/start.ts | 39 +++ packages/cli/src/index.ts | 2 + .../cli/src/lib/create-session-manager.ts | 15 +- packages/cli/src/lib/lifecycle-service.ts | 246 ++++++++++++++++++ .../src/__tests__/lifecycle-manager.test.ts | 223 ++++++++++++++++ .../src/__tests__/session-manager.test.ts | 60 ++++- packages/core/src/lifecycle-manager.ts | 210 +++++++++++++-- packages/core/src/session-manager.ts | 214 ++++++++++++--- 14 files changed, 1083 insertions(+), 75 deletions(-) create mode 100644 packages/cli/src/commands/lifecycle-worker.ts create mode 100644 packages/cli/src/lib/lifecycle-service.ts diff --git a/packages/cli/__tests__/commands/review-check.test.ts b/packages/cli/__tests__/commands/review-check.test.ts index d63c9f4b0..cb0a490ac 100644 --- a/packages/cli/__tests__/commands/review-check.test.ts +++ b/packages/cli/__tests__/commands/review-check.test.ts @@ -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 () => { diff --git a/packages/cli/__tests__/commands/spawn.test.ts b/packages/cli/__tests__/commands/spawn.test.ts index b30e0bcb1..943fd711e 100644 --- a/packages/cli/__tests__/commands/spawn.test.ts +++ b/packages/cli/__tests__/commands/spawn.test.ts @@ -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 | 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 => 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", diff --git a/packages/cli/__tests__/commands/start.test.ts b/packages/cli/__tests__/commands/start.test.ts index d5619ba02..a7a28ccb6 100644 --- a/packages/cli/__tests__/commands/start.test.ts +++ b/packages/cli/__tests__/commands/start.test.ts @@ -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 => 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(" ")) diff --git a/packages/cli/src/commands/lifecycle-worker.ts b/packages/cli/src/commands/lifecycle-worker.ts new file mode 100644 index 000000000..80538c09d --- /dev/null +++ b/packages/cli/src/commands/lifecycle-worker.ts @@ -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 ID from config") + .option("--interval-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); + }); +} diff --git a/packages/cli/src/commands/review-check.ts b/packages/cli/src/commands/review-check.ts index a8263b018..faf5dae33 100644 --- a/packages/cli/src/commands/review-check.ts +++ b/packages/cli/src/commands/review-check.ts @@ -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}`)); diff --git a/packages/cli/src/commands/spawn.ts b/packages/cli/src/commands/spawn.ts index dee43d157..1b2d10a56 100644 --- a/packages/cli/src/commands/spawn.ts +++ b/packages/cli/src/commands/spawn.ts @@ -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); diff --git a/packages/cli/src/commands/start.ts b/packages/cli/src/commands/start.ts index 25cae0c20..1f3e5ae51 100644 --- a/packages/cli/src/commands/start.ts +++ b/packages/cli/src/commands/start.ts @@ -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 { const sessionId = `${project.sessionPrefix}-orchestrator`; + const shouldStartLifecycle = opts?.dashboard !== false || opts?.orchestrator !== false; + let lifecycleStatus: Awaited> | 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); diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index b8b0b5a75..f7cf1bbb6 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -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(); diff --git a/packages/cli/src/lib/create-session-manager.ts b/packages/cli/src/lib/create-session-manager.ts index 5299611c3..954806fe7 100644 --- a/packages/cli/src/lib/create-session-manager.ts +++ b/packages/cli/src/lib/create-session-manager.ts @@ -10,9 +10,11 @@ import { createPluginRegistry, createSessionManager, + createLifecycleManager, type OrchestratorConfig, type SessionManager, type PluginRegistry, + type LifecycleManager, } from "@composio/ao-core"; let registryPromise: Promise | null = null; @@ -22,7 +24,7 @@ let registryPromise: Promise | 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 { +export async function getRegistry(config: OrchestratorConfig): Promise { if (!registryPromise) { registryPromise = (async () => { const registry = createPluginRegistry(); @@ -45,3 +47,14 @@ export async function getSessionManager(config: OrchestratorConfig): Promise { + const registry = await getRegistry(config); + const sessionManager = createSessionManager({ config, registry }); + return createLifecycleManager({ config, registry, sessionManager }); +} diff --git a/packages/cli/src/lib/lifecycle-service.ts b/packages/cli/src/lib/lifecycle-service.ts new file mode 100644 index 000000000..b6af2ac00 --- /dev/null +++ b/packages/cli/src/lib/lifecycle-service.ts @@ -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 { + 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 { + 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 { + 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 { + 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; +} diff --git a/packages/core/src/__tests__/lifecycle-manager.test.ts b/packages/core/src/__tests__/lifecycle-manager.test.ts index faa903e9c..4f2289d88 100644 --- a/packages/core/src/__tests__/lifecycle-manager.test.ts +++ b/packages/core/src/__tests__/lifecycle-manager.test.ts @@ -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", diff --git a/packages/core/src/__tests__/session-manager.test.ts b/packages/core/src/__tests__/session-manager.test.ts index 5935cd6ff..a9e32e3cc 100644 --- a/packages/core/src/__tests__/session-manager.test.ts +++ b/packages/core/src/__tests__/session-manager.test.ts @@ -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", diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index 3ac87a8e4..bb6186e0c 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -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>, + ): 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 { + const project = config.projects[session.projectId]; + if (!project || !session.pr) return; + + const scm = project.scm ? registry.get("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 { 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. */ diff --git a/packages/core/src/session-manager.ts b/packages/core/src/session-manager.ts index 66374c974..755627879 100644 --- a/packages/core/src/session-manager.ts +++ b/packages/core/src/session-manager.ts @@ -117,6 +117,16 @@ const PR_TRACKING_STATUSES: ReadonlySet = 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 { + 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 { - // Find the session in any project's sessions directory - let raw: Record | 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(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(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", - handle.runtimeName ?? - (project ? (project.runtime ?? config.defaults.runtime) : config.defaults.runtime), - ); + const runtimePlugin = registry.get("runtime", runtimeName); if (!runtimePlugin) { throw new Error(`No runtime plugin for session ${sessionId}`); } - await runtimePlugin.sendMessage(handle, message); + const agentPlugin = registry.get("agent", agentName); + if (!agentPlugin) { + throw new Error(`No agent plugin for session ${sessionId}`); + } + + const captureOutput = async (handle: RuntimeHandle): Promise => { + 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 => { + 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 => { + 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 => { + 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 => { + 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(