feat: implement core services (metadata, event-bus, tmux, session-manager, lifecycle-manager)
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 <noreply@anthropic.com>
This commit is contained in:
Prateek
parent
0c535c98d7
commit
de86054b9d
|
|
@ -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<string, unknown>;
|
||||
}
|
||||
): 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<string, unknown>;
|
||||
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<string, Set<EventHandler>>();
|
||||
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<EventHandler> {
|
||||
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;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -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";
|
||||
|
|
|
|||
|
|
@ -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<Record<EventType, SessionStatus>> = {
|
||||
"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<SessionId, SessionStatus>();
|
||||
const handlers = new Map<string, Set<EventHandler>>();
|
||||
const reactionTrackers = new Map<string, ReactionTracker>(); // "sessionId:reactionKey"
|
||||
let pollTimer: ReturnType<typeof setInterval> | null = null;
|
||||
|
||||
function getOrCreateSet(key: string): Set<EventHandler> {
|
||||
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<SessionStatus> {
|
||||
const project = config.projects[session.projectId];
|
||||
if (!project) return session.status;
|
||||
|
||||
const agent = registry.get<Agent>(
|
||||
"agent",
|
||||
project.agent ?? config.defaults.agent
|
||||
);
|
||||
const scm = project.scm
|
||||
? registry.get<SCM>("scm", project.scm.plugin)
|
||||
: null;
|
||||
|
||||
// 1. Check if runtime is alive
|
||||
if (session.runtimeHandle) {
|
||||
const runtime = registry.get<import("./types.js").Runtime>(
|
||||
"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<ReactionResult> {
|
||||
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<void> {
|
||||
const eventWithPriority = { ...event, priority };
|
||||
const notifierNames = config.notificationRouting[priority] ?? config.defaults.notifiers;
|
||||
|
||||
for (const name of notifierNames) {
|
||||
const notifier = registry.get<Notifier>("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<void> {
|
||||
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<void> {
|
||||
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<SessionId, SessionStatus> {
|
||||
return new Map(states);
|
||||
},
|
||||
|
||||
async check(sessionId: SessionId): Promise<void> {
|
||||
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);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -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<string, string> {
|
||||
const result: Record<string, string> = {};
|
||||
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, string>): 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<string, string> | 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<string, string> = {
|
||||
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<Record<string, string>>
|
||||
): void {
|
||||
const path = metadataPath(dataDir, sessionId);
|
||||
let existing: Record<string, string> = {};
|
||||
|
||||
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(".")
|
||||
);
|
||||
}
|
||||
|
|
@ -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<string> {
|
||||
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<string, string>
|
||||
): 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>(
|
||||
"runtime",
|
||||
project.runtime ?? config.defaults.runtime
|
||||
);
|
||||
const agent = registry.get<Agent>(
|
||||
"agent",
|
||||
project.agent ?? config.defaults.agent
|
||||
);
|
||||
const workspace = registry.get<Workspace>(
|
||||
"workspace",
|
||||
project.workspace ?? config.defaults.workspace
|
||||
);
|
||||
const tracker = project.tracker
|
||||
? registry.get<Tracker>("tracker", project.tracker.plugin)
|
||||
: null;
|
||||
const scm = project.scm
|
||||
? registry.get<SCM>("scm", project.scm.plugin)
|
||||
: null;
|
||||
|
||||
return { runtime, agent, workspace, tracker, scm };
|
||||
}
|
||||
|
||||
return {
|
||||
async spawn(spawnConfig: SessionSpawnConfig): Promise<Session> {
|
||||
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<Session[]> {
|
||||
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<string, string>);
|
||||
|
||||
// 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<Session | null> {
|
||||
const raw = readMetadata(config.dataDir, sessionId);
|
||||
if (!raw) return null;
|
||||
return metadataToSession(sessionId, raw as unknown as Record<string, string>);
|
||||
},
|
||||
|
||||
async kill(sessionId: SessionId): Promise<void> {
|
||||
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<CleanupResult> {
|
||||
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<void> {
|
||||
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);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
|
@ -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<string> {
|
||||
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<boolean> {
|
||||
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<TmuxSessionInfo[]> {
|
||||
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<boolean> {
|
||||
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<string, string>;
|
||||
/** Window width/height */
|
||||
width?: number;
|
||||
height?: number;
|
||||
}
|
||||
|
||||
/** Create a new tmux session (detached). */
|
||||
export async function newSession(opts: NewSessionOptions): Promise<void> {
|
||||
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<void> {
|
||||
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<string> {
|
||||
return tmux(
|
||||
"capture-pane",
|
||||
"-t",
|
||||
sessionName,
|
||||
"-p",
|
||||
"-S",
|
||||
`-${lines}`
|
||||
);
|
||||
}
|
||||
|
||||
/** Kill a tmux session. */
|
||||
export async function killSession(sessionName: string): Promise<void> {
|
||||
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<string | null> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue