From de86054b9d08ed30d4ea89fea17ded4619b3224d Mon Sep 17 00:00:00 2001 From: Prateek Date: Fri, 13 Feb 2026 17:37:43 +0530 Subject: [PATCH] feat: implement core services (metadata, event-bus, tmux, session-manager, lifecycle-manager) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implement the 5 core service modules in packages/core/src/: - metadata.ts: Flat-file key=value session metadata read/write matching existing bash script format. Supports read, write, update, delete (with archive), and list operations. - event-bus.ts: In-process pub/sub with JSONL append-only persistence. Supports typed and wildcard listeners, priority inference from event type, filtered history queries, and a createEvent() helper. - tmux.ts: Async wrappers around tmux commands using child_process.execFile (no shell injection). Handles long/multiline messages via load-buffer/ paste-buffer. Covers list, create, send-keys, capture-pane, kill, and pane TTY discovery. - session-manager.ts: Session CRUD that orchestrates Runtime + Agent + Workspace plugins. Full spawn pipeline (workspace → runtime → agent), list with live runtime checks, kill with cleanup, batch cleanup checking PR state + issue state + runtime liveness, and message delivery. - lifecycle-manager.ts: State machine per session with configurable polling interval. Detects status transitions, emits events, triggers configured reactions (send-to-agent, notify, auto-merge) with retry counting and time-based escalation. Routes notifications by priority to configured notifier plugins. All services implement interfaces defined in types.ts. Updated index.ts to export all new modules. Co-Authored-By: Claude Opus 4.6 --- packages/core/src/event-bus.ts | 193 ++++++++++ packages/core/src/index.ts | 33 ++ packages/core/src/lifecycle-manager.ts | 466 +++++++++++++++++++++++++ packages/core/src/metadata.ts | 179 ++++++++++ packages/core/src/session-manager.ts | 428 +++++++++++++++++++++++ packages/core/src/tmux.ts | 194 ++++++++++ 6 files changed, 1493 insertions(+) create mode 100644 packages/core/src/event-bus.ts create mode 100644 packages/core/src/lifecycle-manager.ts create mode 100644 packages/core/src/metadata.ts create mode 100644 packages/core/src/session-manager.ts create mode 100644 packages/core/src/tmux.ts diff --git a/packages/core/src/event-bus.ts b/packages/core/src/event-bus.ts new file mode 100644 index 000000000..92c21eadf --- /dev/null +++ b/packages/core/src/event-bus.ts @@ -0,0 +1,193 @@ +/** + * Event Bus — in-process pub/sub with JSONL persistence. + * + * Events are OrchestratorEvent objects. Supports: + * - Typed event listeners (by EventType) + * - Wildcard listeners ("*") + * - JSONL file persistence (append-only) + * - Filtered history queries + */ + +import { appendFileSync, readFileSync, existsSync, mkdirSync } from "node:fs"; +import { join, dirname } from "node:path"; +import { randomUUID } from "node:crypto"; +import type { + EventBus, + EventFilter, + EventType, + EventPriority, + OrchestratorEvent, + SessionId, +} from "./types.js"; + +type EventHandler = (event: OrchestratorEvent) => void; + +/** + * Create an OrchestratorEvent with defaults filled in. + */ +export function createEvent( + type: EventType, + opts: { + sessionId: SessionId; + projectId: string; + message: string; + priority?: EventPriority; + data?: Record; + } +): OrchestratorEvent { + return { + id: randomUUID(), + type, + priority: opts.priority ?? inferPriority(type), + sessionId: opts.sessionId, + projectId: opts.projectId, + timestamp: new Date(), + message: opts.message, + data: opts.data ?? {}, + }; +} + +/** Infer a reasonable priority from event type. */ +function inferPriority(type: EventType): EventPriority { + if (type.includes("stuck") || type.includes("needs_input") || type.includes("errored")) { + return "urgent"; + } + if (type.includes("approved") || type.includes("ready") || type.includes("merged")) { + return "action"; + } + if (type.includes("fail") || type.includes("changes_requested") || type.includes("conflicts")) { + return "warning"; + } + return "info"; +} + +/** Serialize an event for JSONL storage. */ +function serializeEvent(event: OrchestratorEvent): string { + return JSON.stringify({ + ...event, + timestamp: event.timestamp.toISOString(), + }); +} + +/** Deserialize an event from JSONL. */ +function deserializeEvent(line: string): OrchestratorEvent | null { + try { + const raw = JSON.parse(line) as Record; + return { + ...raw, + timestamp: new Date(raw["timestamp"] as string), + } as OrchestratorEvent; + } catch { + return null; + } +} + +/** + * Create an EventBus implementation. + * + * @param logPath - Path to the JSONL event log file. If null, persistence is disabled. + */ +export function createEventBus(logPath: string | null): EventBus { + const handlers = new Map>(); + const history: OrchestratorEvent[] = []; + + // Load existing history from JSONL file + if (logPath && existsSync(logPath)) { + const content = readFileSync(logPath, "utf-8"); + for (const line of content.split("\n")) { + if (!line.trim()) continue; + const event = deserializeEvent(line); + if (event) history.push(event); + } + } + + // Ensure log directory exists + if (logPath) { + mkdirSync(dirname(logPath), { recursive: true }); + } + + function getOrCreateSet(key: string): Set { + let set = handlers.get(key); + if (!set) { + set = new Set(); + handlers.set(key, set); + } + return set; + } + + return { + emit(event: OrchestratorEvent): void { + // Persist to JSONL + if (logPath) { + appendFileSync(logPath, serializeEvent(event) + "\n", "utf-8"); + } + + // Store in memory + history.push(event); + + // Notify type-specific handlers + const typeHandlers = handlers.get(event.type); + if (typeHandlers) { + for (const handler of typeHandlers) { + try { + handler(event); + } catch { + // Don't let handler errors break the bus + } + } + } + + // Notify wildcard handlers + const wildcardHandlers = handlers.get("*"); + if (wildcardHandlers) { + for (const handler of wildcardHandlers) { + try { + handler(event); + } catch { + // Don't let handler errors break the bus + } + } + } + }, + + on(event: EventType | "*", handler: EventHandler): void { + getOrCreateSet(event).add(handler); + }, + + off(event: EventType | "*", handler: EventHandler): void { + const set = handlers.get(event); + if (set) { + set.delete(handler); + if (set.size === 0) handlers.delete(event); + } + }, + + getHistory(filter?: EventFilter): OrchestratorEvent[] { + let result = history; + + if (filter) { + if (filter.sessionId) { + result = result.filter((e) => e.sessionId === filter.sessionId); + } + if (filter.projectId) { + result = result.filter((e) => e.projectId === filter.projectId); + } + if (filter.type) { + result = result.filter((e) => e.type === filter.type); + } + if (filter.priority) { + result = result.filter((e) => e.priority === filter.priority); + } + if (filter.since) { + const since = filter.since; + result = result.filter((e) => e.timestamp >= since); + } + if (filter.limit) { + result = result.slice(-filter.limit); + } + } + + return result; + }, + }; +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 96af10407..c0354d4fc 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -13,3 +13,36 @@ export { loadConfig, validateConfig, getDefaultConfig } from "./config.js"; // Plugin registry export { createPluginRegistry } from "./plugin-registry.js"; + +// Metadata — flat-file session metadata read/write +export { + readMetadata, + readMetadataRaw, + writeMetadata, + updateMetadata, + deleteMetadata, + listMetadata, +} from "./metadata.js"; + +// Event bus — pub/sub with JSONL persistence +export { createEventBus, createEvent } from "./event-bus.js"; + +// tmux — command wrappers +export { + isTmuxAvailable, + listSessions as listTmuxSessions, + hasSession as hasTmuxSession, + newSession as newTmuxSession, + sendKeys as tmuxSendKeys, + capturePane as tmuxCapturePane, + killSession as killTmuxSession, + getPaneTTY as getTmuxPaneTTY, +} from "./tmux.js"; + +// Session manager — session CRUD +export { createSessionManager } from "./session-manager.js"; +export type { SessionManagerDeps } from "./session-manager.js"; + +// Lifecycle manager — state machine + reaction engine +export { createLifecycleManager } from "./lifecycle-manager.js"; +export type { LifecycleManagerDeps } from "./lifecycle-manager.js"; diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts new file mode 100644 index 000000000..9f6282fce --- /dev/null +++ b/packages/core/src/lifecycle-manager.ts @@ -0,0 +1,466 @@ +/** + * Lifecycle Manager — state machine + polling loop + reaction engine. + * + * Periodically polls all sessions and: + * 1. Detects state transitions (spawning → working → pr_open → etc.) + * 2. Emits events on transitions + * 3. Triggers reactions (auto-handle CI failures, review comments, etc.) + * 4. Escalates to human notification when auto-handling fails + * + * Reference: scripts/claude-session-status, scripts/claude-review-check + */ + +import type { + LifecycleManager, + SessionManager, + SessionId, + SessionStatus, + EventType, + EventBus, + OrchestratorEvent, + OrchestratorConfig, + ProjectConfig, + ReactionConfig, + ReactionResult, + PluginRegistry, + Agent, + SCM, + Notifier, + Session, + EventPriority, +} from "./types.js"; +import { createEvent } from "./event-bus.js"; +import { updateMetadata } from "./metadata.js"; + +type EventHandler = (event: OrchestratorEvent) => void; + +/** Parse a duration string like "10m", "30s", "1h" to milliseconds. */ +function parseDuration(str: string): number { + const match = str.match(/^(\d+)(s|m|h)$/); + if (!match) return 0; + const value = parseInt(match[1], 10); + switch (match[2]) { + case "s": return value * 1000; + case "m": return value * 60_000; + case "h": return value * 3_600_000; + default: return 0; + } +} + +/** Map from event type prefix to session status. */ +const STATUS_FROM_EVENT: Partial> = { + "session.spawned": "spawning", + "session.working": "working", + "session.stuck": "stuck", + "session.needs_input": "needs_input", + "session.errored": "errored", + "session.killed": "killed", + "pr.created": "pr_open", + "ci.failing": "ci_failed", + "review.changes_requested": "changes_requested", + "review.approved": "approved", + "merge.ready": "mergeable", + "merge.completed": "merged", +}; + +/** Determine which event type corresponds to a status transition. */ +function statusToEventType( + from: SessionStatus | undefined, + to: SessionStatus +): EventType | null { + switch (to) { + case "working": return "session.working"; + case "pr_open": return "pr.created"; + case "ci_failed": return "ci.failing"; + case "review_pending": return "review.pending"; + case "changes_requested": return "review.changes_requested"; + case "approved": return "review.approved"; + case "mergeable": return "merge.ready"; + case "merged": return "merge.completed"; + case "needs_input": return "session.needs_input"; + case "stuck": return "session.stuck"; + case "errored": return "session.errored"; + case "killed": return "session.killed"; + default: return null; + } +} + +/** Map event type to reaction config key. */ +function eventToReactionKey(eventType: EventType): string | null { + switch (eventType) { + case "ci.failing": return "ci-failed"; + case "review.changes_requested": return "changes-requested"; + case "automated_review.found": return "bugbot-comments"; + case "merge.conflicts": return "merge-conflicts"; + case "merge.ready": return "approved-and-green"; + case "session.stuck": return "agent-stuck"; + case "session.needs_input": return "agent-needs-input"; + case "session.exited": return "agent-exited"; + case "summary.all_complete": return "all-complete"; + default: return null; + } +} + +export interface LifecycleManagerDeps { + config: OrchestratorConfig; + registry: PluginRegistry; + sessionManager: SessionManager; + eventBus: EventBus; +} + +/** Track retry counts for reactions per session. */ +interface ReactionTracker { + retries: number; + firstTriggered: Date; +} + +/** Create a LifecycleManager instance. */ +export function createLifecycleManager( + deps: LifecycleManagerDeps +): LifecycleManager { + const { config, registry, sessionManager, eventBus } = deps; + + const states = new Map(); + const handlers = new Map>(); + const reactionTrackers = new Map(); // "sessionId:reactionKey" + let pollTimer: ReturnType | null = null; + + function getOrCreateSet(key: string): Set { + let set = handlers.get(key); + if (!set) { + set = new Set(); + handlers.set(key, set); + } + return set; + } + + function emitToHandlers(event: OrchestratorEvent): void { + const typeHandlers = handlers.get(event.type); + if (typeHandlers) { + for (const handler of typeHandlers) { + try { handler(event); } catch { /* ignore */ } + } + } + const wildcardHandlers = handlers.get("*"); + if (wildcardHandlers) { + for (const handler of wildcardHandlers) { + try { handler(event); } catch { /* ignore */ } + } + } + } + + /** Determine current status for a session by polling plugins. */ + async function determineStatus(session: Session): Promise { + const project = config.projects[session.projectId]; + if (!project) return session.status; + + const agent = registry.get( + "agent", + project.agent ?? config.defaults.agent + ); + const scm = project.scm + ? registry.get("scm", project.scm.plugin) + : null; + + // 1. Check if runtime is alive + if (session.runtimeHandle) { + const runtime = registry.get( + "runtime", + project.runtime ?? config.defaults.runtime + ); + if (runtime) { + const alive = await runtime.isAlive(session.runtimeHandle).catch(() => true); + if (!alive) return "killed"; + } + } + + // 2. Check agent activity + if (agent) { + const activity = await agent.detectActivity(session).catch(() => session.activity); + if (activity === "exited") return "killed"; + if (activity === "blocked") return "stuck"; + if (activity === "waiting_input") return "needs_input"; + } + + // 3. Check PR state if PR exists + if (session.pr && scm) { + try { + const prState = await scm.getPRState(session.pr); + if (prState === "merged") return "merged"; + if (prState === "closed") return "killed"; + + // Check CI + const ciStatus = await scm.getCISummary(session.pr); + if (ciStatus === "failing") return "ci_failed"; + + // Check reviews + const reviewDecision = await scm.getReviewDecision(session.pr); + if (reviewDecision === "changes_requested") return "changes_requested"; + if (reviewDecision === "approved") { + // Check merge readiness + const mergeReady = await scm.getMergeability(session.pr); + if (mergeReady.mergeable) return "mergeable"; + return "approved"; + } + if (reviewDecision === "pending") return "review_pending"; + + return "pr_open"; + } catch { + // SCM check failed — keep current status + } + } + + // 4. Default: if agent is active, it's working + return session.status === "spawning" ? "working" : session.status; + } + + /** Execute a reaction for an event. */ + async function executeReaction( + event: OrchestratorEvent, + reactionKey: string, + reactionConfig: ReactionConfig + ): Promise { + const trackerKey = `${event.sessionId}:${reactionKey}`; + let tracker = reactionTrackers.get(trackerKey); + + if (!tracker) { + tracker = { retries: 0, firstTriggered: new Date() }; + reactionTrackers.set(trackerKey, tracker); + } + + // Check if we should escalate + const maxRetries = reactionConfig.retries ?? Infinity; + const escalateAfter = reactionConfig.escalateAfter; + let shouldEscalate = false; + + if (tracker.retries >= maxRetries) { + shouldEscalate = true; + } + + if (typeof escalateAfter === "string") { + const durationMs = parseDuration(escalateAfter); + if (durationMs > 0 && Date.now() - tracker.firstTriggered.getTime() > durationMs) { + shouldEscalate = true; + } + } + + if (typeof escalateAfter === "number" && tracker.retries >= escalateAfter) { + shouldEscalate = true; + } + + if (shouldEscalate) { + // Escalate to human + await notifyHuman(event, reactionConfig.priority ?? "urgent"); + return { + reactionType: reactionKey, + success: true, + action: "escalated", + escalated: true, + }; + } + + // Execute the reaction action + switch (reactionConfig.action) { + case "send-to-agent": { + if (reactionConfig.message) { + try { + await sessionManager.send(event.sessionId, reactionConfig.message); + tracker.retries++; + + eventBus.emit( + createEvent("reaction.triggered", { + sessionId: event.sessionId, + projectId: event.projectId, + message: `Reaction '${reactionKey}' sent message to agent`, + data: { reactionKey, retries: tracker.retries }, + }) + ); + + return { + reactionType: reactionKey, + success: true, + action: "send-to-agent", + message: reactionConfig.message, + escalated: false, + }; + } catch { + // Send failed — escalate + await notifyHuman(event, reactionConfig.priority ?? "warning"); + return { + reactionType: reactionKey, + success: false, + action: "send-to-agent", + escalated: true, + }; + } + } + break; + } + + case "notify": { + await notifyHuman(event, reactionConfig.priority ?? "info"); + return { + reactionType: reactionKey, + success: true, + action: "notify", + escalated: false, + }; + } + + case "auto-merge": { + // Auto-merge is handled by the SCM plugin + // For now, just notify + await notifyHuman(event, "action"); + return { + reactionType: reactionKey, + success: true, + action: "auto-merge", + escalated: false, + }; + } + } + + return { + reactionType: reactionKey, + success: false, + action: reactionConfig.action, + escalated: false, + }; + } + + /** Send a notification to all configured notifiers. */ + async function notifyHuman( + event: OrchestratorEvent, + priority: EventPriority + ): Promise { + const eventWithPriority = { ...event, priority }; + const notifierNames = config.notificationRouting[priority] ?? config.defaults.notifiers; + + for (const name of notifierNames) { + const notifier = registry.get("notifier", name); + if (notifier) { + try { + await notifier.notify(eventWithPriority); + } catch { + // Notifier failed — not much we can do + } + } + } + } + + /** Poll a single session and handle state transitions. */ + async function checkSession(session: Session): Promise { + const oldStatus = states.get(session.id) ?? session.status; + const newStatus = await determineStatus(session); + + if (newStatus !== oldStatus) { + // State transition detected + states.set(session.id, newStatus); + + // Update metadata + updateMetadata(config.dataDir, session.id, { status: newStatus }); + + // Emit transition event + const eventType = statusToEventType(oldStatus, newStatus); + if (eventType) { + const event = createEvent(eventType, { + sessionId: session.id, + projectId: session.projectId, + message: `${session.id}: ${oldStatus} → ${newStatus}`, + data: { oldStatus, newStatus }, + }); + + eventBus.emit(event); + emitToHandlers(event); + + // Check if there's a reaction for this event + const reactionKey = eventToReactionKey(eventType); + if (reactionKey) { + // Check project-specific overrides first, then global + const project = config.projects[session.projectId]; + const reactionConfig = + project?.reactions?.[reactionKey] ?? + config.reactions[reactionKey]; + + if (reactionConfig && (reactionConfig as ReactionConfig).auto !== false) { + await executeReaction( + event, + reactionKey, + reactionConfig as ReactionConfig + ); + } + } + } + } else { + // No transition but track current state + states.set(session.id, newStatus); + } + } + + /** Run one polling cycle across all sessions. */ + async function pollAll(): Promise { + try { + const sessions = await sessionManager.list(); + + for (const session of sessions) { + if (session.status === "merged" || session.status === "killed") continue; + await checkSession(session); + } + + // Check if all sessions are complete + const activeSessions = sessions.filter( + (s) => s.status !== "merged" && s.status !== "killed" + ); + if (sessions.length > 0 && activeSessions.length === 0) { + const event = createEvent("summary.all_complete", { + sessionId: "system", + projectId: "all", + message: `All ${sessions.length} sessions complete`, + data: { totalSessions: sessions.length }, + }); + eventBus.emit(event); + emitToHandlers(event); + } + } catch { + // Poll cycle failed — will retry next interval + } + } + + return { + start(intervalMs = 30_000): void { + if (pollTimer) return; // Already running + pollTimer = setInterval(() => void pollAll(), intervalMs); + // Run immediately on start + void pollAll(); + }, + + stop(): void { + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = null; + } + }, + + getStates(): Map { + return new Map(states); + }, + + async check(sessionId: SessionId): Promise { + const session = await sessionManager.get(sessionId); + if (!session) throw new Error(`Session ${sessionId} not found`); + await checkSession(session); + }, + + on(event: EventType | "*", handler: EventHandler): void { + getOrCreateSet(event).add(handler); + }, + + off(event: EventType | "*", handler: EventHandler): void { + const set = handlers.get(event); + if (set) { + set.delete(handler); + if (set.size === 0) handlers.delete(event); + } + }, + }; +} diff --git a/packages/core/src/metadata.ts b/packages/core/src/metadata.ts new file mode 100644 index 000000000..dd98211ec --- /dev/null +++ b/packages/core/src/metadata.ts @@ -0,0 +1,179 @@ +/** + * Flat-file metadata read/write. + * + * Each session has a metadata file in the data directory, stored as + * key=value pairs (one per line), matching the existing bash script format. + * + * Example file contents: + * worktree=/Users/foo/.worktrees/ao/ao-3 + * branch=feat/INT-1234 + * status=working + * pr=https://github.com/org/repo/pull/42 + * issue=https://linear.app/team/issue/INT-1234 + */ + +import { readFileSync, writeFileSync, existsSync, mkdirSync, unlinkSync, readdirSync } from "node:fs"; +import { join, dirname } from "node:path"; +import type { SessionId, SessionMetadata } from "./types.js"; + +/** + * Parse a key=value metadata file into a record. + * Lines starting with # are comments. Empty lines are skipped. + * Only the first `=` is used as the delimiter (values can contain `=`). + */ +function parseMetadataFile(content: string): Record { + const result: Record = {}; + for (const line of content.split("\n")) { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith("#")) continue; + const eqIndex = trimmed.indexOf("="); + if (eqIndex === -1) continue; + const key = trimmed.slice(0, eqIndex).trim(); + const value = trimmed.slice(eqIndex + 1).trim(); + if (key) result[key] = value; + } + return result; +} + +/** Serialize a record back to key=value format. */ +function serializeMetadata(data: Record): string { + return Object.entries(data) + .filter(([, v]) => v !== undefined && v !== "") + .map(([k, v]) => `${k}=${v}`) + .join("\n") + "\n"; +} + +/** Get the metadata file path for a session. */ +function metadataPath(dataDir: string, sessionId: SessionId): string { + return join(dataDir, "sessions", sessionId); +} + +/** + * Read metadata for a session. Returns null if the file doesn't exist. + */ +export function readMetadata( + dataDir: string, + sessionId: SessionId +): SessionMetadata | null { + const path = metadataPath(dataDir, sessionId); + if (!existsSync(path)) return null; + + const content = readFileSync(path, "utf-8"); + const raw = parseMetadataFile(content); + + return { + worktree: raw["worktree"] ?? "", + branch: raw["branch"] ?? "", + status: raw["status"] ?? "unknown", + issue: raw["issue"], + pr: raw["pr"], + summary: raw["summary"], + project: raw["project"], + createdAt: raw["createdAt"], + runtimeHandle: raw["runtimeHandle"], + }; +} + +/** + * Read raw metadata as a string record (for arbitrary keys). + */ +export function readMetadataRaw( + dataDir: string, + sessionId: SessionId +): Record | null { + const path = metadataPath(dataDir, sessionId); + if (!existsSync(path)) return null; + return parseMetadataFile(readFileSync(path, "utf-8")); +} + +/** + * Write full metadata for a session (overwrites existing file). + */ +export function writeMetadata( + dataDir: string, + sessionId: SessionId, + metadata: SessionMetadata +): void { + const path = metadataPath(dataDir, sessionId); + mkdirSync(dirname(path), { recursive: true }); + + const data: Record = { + worktree: metadata.worktree, + branch: metadata.branch, + status: metadata.status, + }; + + if (metadata.issue) data["issue"] = metadata.issue; + if (metadata.pr) data["pr"] = metadata.pr; + if (metadata.summary) data["summary"] = metadata.summary; + if (metadata.project) data["project"] = metadata.project; + if (metadata.createdAt) data["createdAt"] = metadata.createdAt; + if (metadata.runtimeHandle) data["runtimeHandle"] = metadata.runtimeHandle; + + writeFileSync(path, serializeMetadata(data), "utf-8"); +} + +/** + * Update specific fields in a session's metadata. + * Reads existing file, merges updates, writes back. + */ +export function updateMetadata( + dataDir: string, + sessionId: SessionId, + updates: Partial> +): void { + const path = metadataPath(dataDir, sessionId); + let existing: Record = {}; + + if (existsSync(path)) { + existing = parseMetadataFile(readFileSync(path, "utf-8")); + } + + // Merge updates — delete keys set to empty string + for (const [key, value] of Object.entries(updates)) { + if (value === undefined) continue; + if (value === "") { + delete existing[key]; + } else { + existing[key] = value; + } + } + + mkdirSync(dirname(path), { recursive: true }); + writeFileSync(path, serializeMetadata(existing), "utf-8"); +} + +/** + * Delete a session's metadata file. + * Optionally archive it to an `archive/` subdirectory. + */ +export function deleteMetadata( + dataDir: string, + sessionId: SessionId, + archive = true +): void { + const path = metadataPath(dataDir, sessionId); + if (!existsSync(path)) return; + + if (archive) { + const archiveDir = join(dataDir, "sessions", "archive"); + mkdirSync(archiveDir, { recursive: true }); + const timestamp = new Date().toISOString().replace(/[:.]/g, "-"); + const archivePath = join(archiveDir, `${sessionId}_${timestamp}`); + writeFileSync(archivePath, readFileSync(path, "utf-8")); + } + + unlinkSync(path); +} + +/** + * List all session IDs that have metadata files. + */ +export function listMetadata(dataDir: string): SessionId[] { + const dir = join(dataDir, "sessions"); + if (!existsSync(dir)) return []; + + return readdirSync(dir).filter( + (name) => name !== "archive" && !name.startsWith(".") + ); +} diff --git a/packages/core/src/session-manager.ts b/packages/core/src/session-manager.ts new file mode 100644 index 000000000..dffd06726 --- /dev/null +++ b/packages/core/src/session-manager.ts @@ -0,0 +1,428 @@ +/** + * Session Manager — CRUD for agent sessions. + * + * Orchestrates Runtime, Agent, and Workspace plugins to: + * - Spawn new sessions (create workspace → create runtime → launch agent) + * - List sessions (from metadata + live runtime checks) + * - Kill sessions (agent → runtime → workspace cleanup) + * - Cleanup completed sessions (PR merged / issue closed) + * - Send messages to running sessions + * + * Reference: scripts/claude-ao-session, scripts/send-to-session + */ + +import { execFile } from "node:child_process"; +import type { + SessionManager, + Session, + SessionId, + SessionSpawnConfig, + SessionStatus, + CleanupResult, + OrchestratorConfig, + ProjectConfig, + Runtime, + Agent, + Workspace, + Tracker, + SCM, + EventBus, + PluginRegistry, +} from "./types.js"; +import { + readMetadata, + writeMetadata, + updateMetadata, + deleteMetadata, + listMetadata, +} from "./metadata.js"; +import { createEvent } from "./event-bus.js"; + +/** Get the current git branch in a directory. */ +function getGitBranch(dir: string): Promise { + return new Promise((resolve) => { + execFile("git", ["-C", dir, "branch", "--show-current"], (error, stdout) => { + resolve(error ? "unknown" : stdout.trim()); + }); + }); +} + +/** Get the next session number for a project. */ +function getNextSessionNumber( + existingSessions: string[], + prefix: string +): number { + let max = 0; + const pattern = new RegExp(`^${prefix}-(\\d+)$`); + for (const name of existingSessions) { + const match = name.match(pattern); + if (match) { + const num = parseInt(match[1], 10); + if (num > max) max = num; + } + } + return max + 1; +} + +/** Reconstruct a Session object from metadata + live data. */ +function metadataToSession( + sessionId: SessionId, + meta: Record +): Session { + return { + id: sessionId, + projectId: meta["project"] ?? "", + status: (meta["status"] ?? "unknown") as SessionStatus, + activity: "idle", + branch: meta["branch"] || null, + issueId: meta["issue"] || null, + pr: meta["pr"] + ? { + number: parseInt(meta["pr"].match(/\/(\d+)$/)?.[1] ?? "0", 10), + url: meta["pr"], + title: "", + owner: "", + repo: "", + branch: meta["branch"] ?? "", + baseBranch: "", + isDraft: false, + } + : null, + workspacePath: meta["worktree"] || null, + runtimeHandle: meta["runtimeHandle"] + ? JSON.parse(meta["runtimeHandle"]) + : null, + agentInfo: meta["summary"] + ? { summary: meta["summary"], agentSessionId: null } + : null, + createdAt: meta["createdAt"] ? new Date(meta["createdAt"]) : new Date(), + lastActivityAt: new Date(), + metadata: meta, + }; +} + +export interface SessionManagerDeps { + config: OrchestratorConfig; + registry: PluginRegistry; + eventBus: EventBus; +} + +/** Create a SessionManager instance. */ +export function createSessionManager(deps: SessionManagerDeps): SessionManager { + const { config, registry, eventBus } = deps; + + /** Resolve which plugins to use for a project. */ + function resolvePlugins(project: ProjectConfig) { + const runtime = registry.get( + "runtime", + project.runtime ?? config.defaults.runtime + ); + const agent = registry.get( + "agent", + project.agent ?? config.defaults.agent + ); + const workspace = registry.get( + "workspace", + project.workspace ?? config.defaults.workspace + ); + const tracker = project.tracker + ? registry.get("tracker", project.tracker.plugin) + : null; + const scm = project.scm + ? registry.get("scm", project.scm.plugin) + : null; + + return { runtime, agent, workspace, tracker, scm }; + } + + return { + async spawn(spawnConfig: SessionSpawnConfig): Promise { + const project = config.projects[spawnConfig.projectId]; + if (!project) { + throw new Error(`Unknown project: ${spawnConfig.projectId}`); + } + + const plugins = resolvePlugins(project); + if (!plugins.runtime) { + throw new Error(`Runtime plugin '${project.runtime ?? config.defaults.runtime}' not found`); + } + if (!plugins.agent) { + throw new Error(`Agent plugin '${project.agent ?? config.defaults.agent}' not found`); + } + + // Determine session ID + const existingSessions = listMetadata(config.dataDir); + const num = getNextSessionNumber(existingSessions, project.sessionPrefix); + const sessionId = `${project.sessionPrefix}-${num}`; + + // Determine branch name + let branch = spawnConfig.branch ?? project.defaultBranch; + if (spawnConfig.issueId && plugins.tracker) { + branch = plugins.tracker.branchName(spawnConfig.issueId, project); + } else if (spawnConfig.issueId) { + branch = `feat/${spawnConfig.issueId}`; + } + + // Create workspace (if workspace plugin is available) + let workspacePath = project.path; + if (plugins.workspace) { + const wsInfo = await plugins.workspace.create({ + projectId: spawnConfig.projectId, + project, + sessionId, + branch, + }); + workspacePath = wsInfo.path; + + // Run post-create hooks + if (plugins.workspace.postCreate) { + await plugins.workspace.postCreate(wsInfo, project); + } + } + + // Get agent launch config + const agentLaunchConfig = { + sessionId, + projectConfig: project, + issueId: spawnConfig.issueId, + prompt: spawnConfig.prompt, + permissions: project.agentConfig?.permissions, + model: project.agentConfig?.model, + }; + + const launchCommand = plugins.agent.getLaunchCommand(agentLaunchConfig); + const environment = plugins.agent.getEnvironment(agentLaunchConfig); + + // Create runtime session + const handle = await plugins.runtime.create({ + sessionId, + workspacePath, + launchCommand, + environment: { + ...environment, + AO_SESSION: sessionId, + }, + }); + + // Write metadata + writeMetadata(config.dataDir, sessionId, { + worktree: workspacePath, + branch, + status: "spawning", + issue: spawnConfig.issueId, + project: spawnConfig.projectId, + createdAt: new Date().toISOString(), + runtimeHandle: JSON.stringify(handle), + }); + + // Post-launch setup (if agent supports it) + const session: Session = { + id: sessionId, + projectId: spawnConfig.projectId, + status: "spawning", + activity: "active", + branch, + issueId: spawnConfig.issueId ?? null, + pr: null, + workspacePath, + runtimeHandle: handle, + agentInfo: null, + createdAt: new Date(), + lastActivityAt: new Date(), + metadata: {}, + }; + + if (plugins.agent.postLaunchSetup) { + await plugins.agent.postLaunchSetup(session); + } + + // Emit event + eventBus.emit( + createEvent("session.spawned", { + sessionId, + projectId: spawnConfig.projectId, + message: `Session ${sessionId} spawned${spawnConfig.issueId ? ` for ${spawnConfig.issueId}` : ""}`, + data: { branch, workspacePath, issueId: spawnConfig.issueId }, + }) + ); + + return session; + }, + + async list(projectId?: string): Promise { + const sessionIds = listMetadata(config.dataDir); + const sessions: Session[] = []; + + for (const sid of sessionIds) { + const raw = readMetadata(config.dataDir, sid); + if (!raw) continue; + + // Filter by project if specified + if (projectId && raw.project !== projectId) continue; + + const session = metadataToSession(sid, raw as unknown as Record); + + // Check if runtime is still alive + if (session.runtimeHandle) { + const project = config.projects[session.projectId]; + if (project) { + const plugins = resolvePlugins(project); + if (plugins.runtime) { + try { + const alive = await plugins.runtime.isAlive(session.runtimeHandle); + if (!alive) { + session.status = "killed"; + session.activity = "exited"; + } + } catch { + // Can't check — assume still alive + } + } + } + } + + sessions.push(session); + } + + return sessions; + }, + + async get(sessionId: SessionId): Promise { + const raw = readMetadata(config.dataDir, sessionId); + if (!raw) return null; + return metadataToSession(sessionId, raw as unknown as Record); + }, + + async kill(sessionId: SessionId): Promise { + const raw = readMetadata(config.dataDir, sessionId); + if (!raw) throw new Error(`Session ${sessionId} not found`); + + const projectId = raw.project ?? ""; + const project = config.projects[projectId]; + + // Destroy runtime + if (raw.runtimeHandle && project) { + const plugins = resolvePlugins(project); + if (plugins.runtime) { + try { + const handle = JSON.parse(raw.runtimeHandle); + await plugins.runtime.destroy(handle); + } catch { + // Runtime might already be gone + } + } + } + + // Destroy workspace + if (raw.worktree && project) { + const plugins = resolvePlugins(project); + if (plugins.workspace) { + try { + await plugins.workspace.destroy(raw.worktree); + } catch { + // Workspace might already be gone + } + } + } + + // Archive metadata + deleteMetadata(config.dataDir, sessionId, true); + + // Emit event + eventBus.emit( + createEvent("session.killed", { + sessionId, + projectId, + message: `Session ${sessionId} killed`, + }) + ); + }, + + async cleanup(projectId?: string): Promise { + const result: CleanupResult = { killed: [], skipped: [], errors: [] }; + const sessions = await this.list(projectId); + + for (const session of sessions) { + try { + const project = config.projects[session.projectId]; + if (!project) { + result.skipped.push(session.id); + continue; + } + + const plugins = resolvePlugins(project); + let shouldKill = false; + + // Check if PR is merged + if (session.pr && plugins.scm) { + try { + const prState = await plugins.scm.getPRState(session.pr); + if (prState === "merged" || prState === "closed") { + shouldKill = true; + } + } catch { + // Can't check PR — skip + } + } + + // Check if issue is completed + if (!shouldKill && session.issueId && plugins.tracker) { + try { + const completed = await plugins.tracker.isCompleted( + session.issueId, + project + ); + if (completed) shouldKill = true; + } catch { + // Can't check issue — skip + } + } + + // Check if runtime is dead + if (!shouldKill && session.runtimeHandle && plugins.runtime) { + try { + const alive = await plugins.runtime.isAlive(session.runtimeHandle); + if (!alive) shouldKill = true; + } catch { + // Can't check — skip + } + } + + if (shouldKill) { + await this.kill(session.id); + result.killed.push(session.id); + } else { + result.skipped.push(session.id); + } + } catch (err) { + result.errors.push({ + sessionId: session.id, + error: err instanceof Error ? err.message : String(err), + }); + } + } + + return result; + }, + + async send(sessionId: SessionId, message: string): Promise { + const raw = readMetadata(config.dataDir, sessionId); + if (!raw) throw new Error(`Session ${sessionId} not found`); + + const project = config.projects[raw.project ?? ""]; + if (!project) throw new Error(`Project not found for session ${sessionId}`); + + const plugins = resolvePlugins(project); + if (!plugins.runtime) { + throw new Error(`No runtime plugin for session ${sessionId}`); + } + + if (!raw.runtimeHandle) { + throw new Error(`No runtime handle for session ${sessionId}`); + } + + const handle = JSON.parse(raw.runtimeHandle); + await plugins.runtime.sendMessage(handle, message); + }, + }; +} diff --git a/packages/core/src/tmux.ts b/packages/core/src/tmux.ts new file mode 100644 index 000000000..767bdecf7 --- /dev/null +++ b/packages/core/src/tmux.ts @@ -0,0 +1,194 @@ +/** + * tmux command wrappers — async helpers for tmux operations. + * + * Uses child_process.execFile for safe command execution (no shell injection). + */ + +import { execFile } from "node:child_process"; + +/** Run a tmux command and return stdout. */ +function tmux(...args: string[]): Promise { + return new Promise((resolve, reject) => { + execFile("tmux", args, { timeout: 10_000 }, (error, stdout, stderr) => { + if (error) { + // tmux exits non-zero for many benign cases (no sessions, etc.) + reject(new Error(`tmux ${args[0]} failed: ${stderr || error.message}`)); + return; + } + resolve(stdout); + }); + }); +} + +/** Check if tmux server is running. */ +export async function isTmuxAvailable(): Promise { + try { + await tmux("list-sessions", "-F", "#{session_name}"); + return true; + } catch { + return false; + } +} + +export interface TmuxSessionInfo { + name: string; + created: string; + attached: boolean; + windows: number; +} + +/** List all tmux sessions. */ +export async function listSessions(): Promise { + try { + const output = await tmux( + "list-sessions", + "-F", + "#{session_name}\t#{session_created_string}\t#{session_attached}\t#{session_windows}" + ); + + return output + .trim() + .split("\n") + .filter(Boolean) + .map((line) => { + const [name = "", created = "", attached = "0", windows = "1"] = line.split("\t"); + return { + name, + created, + attached: attached !== "0", + windows: parseInt(windows, 10) || 1, + }; + }); + } catch { + // No tmux server or no sessions + return []; + } +} + +/** Check if a specific tmux session exists. */ +export async function hasSession(sessionName: string): Promise { + try { + await tmux("has-session", "-t", sessionName); + return true; + } catch { + return false; + } +} + +export interface NewSessionOptions { + /** Session name */ + name: string; + /** Working directory */ + cwd: string; + /** Initial command to run */ + command?: string; + /** Environment variables to set */ + environment?: Record; + /** Window width/height */ + width?: number; + height?: number; +} + +/** Create a new tmux session (detached). */ +export async function newSession(opts: NewSessionOptions): Promise { + const args = ["new-session", "-d", "-s", opts.name, "-c", opts.cwd]; + + // Add environment variables + if (opts.environment) { + for (const [key, value] of Object.entries(opts.environment)) { + args.push("-e", `${key}=${value}`); + } + } + + // Window size + if (opts.width) { + args.push("-x", String(opts.width)); + } + if (opts.height) { + args.push("-y", String(opts.height)); + } + + await tmux(...args); + + // Send the initial command if provided + if (opts.command) { + await sendKeys(opts.name, opts.command); + } +} + +/** + * Send keys (text + Enter) to a tmux session. + * For long/multiline messages, uses load-buffer + paste-buffer. + */ +export async function sendKeys( + sessionName: string, + text: string, + pressEnter = true +): Promise { + if (text.includes("\n") || text.length > 200) { + // Use load-buffer for long/multiline text (avoids tmux escaping issues) + const { writeFileSync, unlinkSync } = await import("node:fs"); + const { tmpdir } = await import("node:os"); + const { join } = await import("node:path"); + const { randomUUID } = await import("node:crypto"); + + const tmpFile = join(tmpdir(), `ao-tmux-${randomUUID()}.txt`); + writeFileSync(tmpFile, text, "utf-8"); + + try { + await tmux("load-buffer", tmpFile); + await tmux("paste-buffer", "-t", sessionName); + } finally { + try { unlinkSync(tmpFile); } catch { /* ignore cleanup errors */ } + } + } else { + await tmux("send-keys", "-t", sessionName, text); + } + + if (pressEnter) { + // Small delay for paste to complete before sending Enter + if (text.includes("\n") || text.length > 200) { + await new Promise((resolve) => setTimeout(resolve, 300)); + } + await tmux("send-keys", "-t", sessionName, "Enter"); + } +} + +/** + * Capture recent output from a tmux pane. + * + * @param sessionName - tmux session name + * @param lines - Number of scrollback lines to capture (default 30) + */ +export async function capturePane( + sessionName: string, + lines = 30 +): Promise { + return tmux( + "capture-pane", + "-t", + sessionName, + "-p", + "-S", + `-${lines}` + ); +} + +/** Kill a tmux session. */ +export async function killSession(sessionName: string): Promise { + await tmux("kill-session", "-t", sessionName); +} + +/** + * Get the TTY device for a tmux session's first pane. + * Useful for finding processes running in the session. + */ +export async function getPaneTTY(sessionName: string): Promise { + try { + const output = await tmux("list-panes", "-t", sessionName, "-F", "#{pane_tty}"); + const tty = output.trim().split("\n")[0]; + return tty || null; + } catch { + return null; + } +}