import type { AcpConfig } from "../config.js"; import { chatKeyFor, type DurableSessionStore, type SessionBinding } from "../core/durable-session-store.js"; import type { RoleRegistry } from "../roles/role-registry.js"; import type { AcpBackendRegistry } from "./backend-registry.js"; import type { ConversationRequest, ConversationResponse, ConversationRuntime, RuntimeStats } from "./types.js"; import { AcpWorker } from "./worker.js"; export class AcpSessionManager implements ConversationRuntime { private readonly workers = new Map(); private readonly inFlight = new Map(); private readonly sweeper: NodeJS.Timeout; private crashes = 0; private shuttingDown = false; constructor( private readonly config: AcpConfig, private readonly backends: AcpBackendRegistry, private readonly roles: RoleRegistry, private readonly store: DurableSessionStore ) { this.sweeper = setInterval(() => void this.sweep(), config.sweepIntervalMs); this.sweeper.unref(); } async prompt(request: ConversationRequest): Promise { if (this.shuttingDown) throw new Error("ACP runtime is shutting down"); const chatKey = chatKeyFor(request.platform, request.chatId); const role = this.roles.get(this.store.getSelectedRole(chatKey, this.roles.defaultId())); let binding = this.store.getBinding(chatKey, role.id); if (binding && (binding.roleFingerprint !== role.fingerprint || binding.backendId !== role.backend || binding.workspace !== role.workspace)) { await this.dropBinding(binding); binding = undefined; } const worker = await this.acquireWorker(role.id, binding); this.inFlight.set(chatKey, worker); try { if (!binding) { const now = Date.now(); await worker.prompt(role.bootstrap); binding = { chatKey, roleId: role.id, backendId: role.backend, nativeSessionId: worker.nativeSessionId!, workspace: role.workspace, roleFingerprint: role.fingerprint, createdAt: now, updatedAt: now }; await this.store.setBinding(binding); } const text = await worker.prompt(request.text); await this.store.touchBinding(chatKey, role.id); return { text: text || "(ACP agent returned no text)", roleId: role.id, backendId: role.backend }; } catch (error) { if (worker.nativeSessionId) this.workers.delete(workerKey(worker.backend.id, worker.nativeSessionId)); await worker.terminate(); throw error; } finally { if (this.inFlight.get(chatKey) === worker) this.inFlight.delete(chatKey); } } async cancel(platform: string, chatId: string): Promise { const worker = this.inFlight.get(chatKeyFor(platform, chatId)); return worker ? worker.cancel() : false; } async reset(platform: string, chatId: string): Promise { const chatKey = chatKeyFor(platform, chatId); await this.cancel(platform, chatId); const roleId = this.store.getSelectedRole(chatKey, this.roles.defaultId()); const binding = await this.store.deleteBinding(chatKey, roleId); if (binding) await this.stopWorker(binding.backendId, binding.nativeSessionId); } async selectRole(platform: string, chatId: string, roleId: string): Promise { this.roles.get(roleId); await this.store.setSelectedRole(chatKeyFor(platform, chatId), roleId); } selectedRole(platform: string, chatId: string): string { return this.store.getSelectedRole(chatKeyFor(platform, chatId), this.roles.defaultId()); } status(platform: string, chatId: string): Record { const chatKey = chatKeyFor(platform, chatId); const roleId = this.store.getSelectedRole(chatKey, this.roles.defaultId()); const binding = this.store.getBinding(chatKey, roleId); return { role: roleId, backend: this.roles.get(roleId).backend, persisted: Boolean(binding), running: this.inFlight.has(chatKey) }; } stats(): RuntimeStats { return { activeWorkers: this.workers.size, inFlight: this.inFlight.size, crashes: this.crashes, persistedBindings: this.store.stats().bindings }; } async shutdown(): Promise { if (this.shuttingDown) return; this.shuttingDown = true; clearInterval(this.sweeper); await Promise.all([...this.inFlight.values()].map((worker) => worker.cancel().catch(() => false))); await Promise.all([...this.workers.values()].map((worker) => worker.terminate())); this.workers.clear(); this.inFlight.clear(); } private async acquireWorker(roleId: string, binding?: SessionBinding): Promise { const role = this.roles.get(roleId); if (binding) { const existing = this.workers.get(workerKey(binding.backendId, binding.nativeSessionId)); if (existing) return existing; } await this.ensureCapacity(); const worker = new AcpWorker(this.backends.get(role.backend), role, this.config, (crashed, error) => { this.crashes++; if (crashed.nativeSessionId) this.workers.delete(workerKey(crashed.backend.id, crashed.nativeSessionId)); console.error(`ACP worker crash: ${error.message}`); }); const nativeSessionId = await worker.start(binding?.nativeSessionId); this.workers.set(workerKey(role.backend, nativeSessionId), worker); return worker; } private async ensureCapacity(): Promise { if (this.workers.size < this.config.maxProcesses) return; const candidate = [...this.workers.entries()].filter(([, worker]) => !worker.inFlight).sort((a, b) => a[1].lastUsedAt - b[1].lastUsedAt)[0]; if (!candidate) throw new Error(`ACP worker limit reached (${this.config.maxProcesses})`); this.workers.delete(candidate[0]); await candidate[1].terminate(); } private async sweep(): Promise { const cutoff = Date.now() - this.config.idleTimeoutMs; const expired = [...this.workers.entries()].filter(([, worker]) => !worker.inFlight && worker.lastUsedAt < cutoff); for (const [key, worker] of expired) { this.workers.delete(key); await worker.terminate(); } } private async dropBinding(binding: SessionBinding): Promise { await this.store.deleteBinding(binding.chatKey, binding.roleId); await this.stopWorker(binding.backendId, binding.nativeSessionId); } private async stopWorker(backendId: string, nativeSessionId: string): Promise { const key = workerKey(backendId, nativeSessionId); const worker = this.workers.get(key); if (!worker) return; this.workers.delete(key); await worker.terminate(); } } function workerKey(backendId: string, nativeSessionId: string): string { return `${backendId}:${nativeSessionId}`; }