import { EventEmitter } from 'node:events'; import { existsSync, statSync } from 'node:fs'; import { randomUUID } from 'node:crypto'; import { homedir } from 'node:os'; import { basename, join } from 'node:path'; import pty from '@homebridge/node-pty-prebuilt-multiarch'; import { FLOW, REPLAY_TAIL_BYTES, type SessionActivity, type SessionSummary } from '@arboretum/shared'; import { RingBuffer } from './ring-buffer.js'; import { buildSpawnSpec } from './claude-launcher.js'; import { readClaudeBinPath } from './claude-settings.js'; import { findByPid } from './session-registry.js'; import { SessionActivityTracker } from './claude-adapter.js'; import type { PushService } from './push-service.js'; import type { Db } from '../db/index.js'; // 4 Mo : conserve assez d'historique pour que le replay (REPLAY_TAIL_BYTES = 1 Mo) reste largement // dans le ring et qu'on puisse remonter une conversation Claude après ré-attache. const RING_CAPACITY = 4 * 1024 * 1024; const KILL_GRACE_MS = 5000; /** Délai avant envoi d'une notif push sur passage en `waiting` : annulé si la session repart (faux positif). */ const NOTIFY_DEBOUNCE_MS = 1500; /** Capture du claudeSessionId après spawn : poll du registre par pid (waitReady validé S1). */ const CLAUDE_ID_POLL_MS = 400; const CLAUDE_ID_TIMEOUT_MS = 60_000; /** * Ligne `sessions` telle que lue pour construire un SessionSummary historique (session terminée). * `type` (et non `interface`) pour que TS infère l'index signature implicite → cast direct depuis * le `Record` de node:sqlite. */ type HistoricalRow = { id: string; cwd: string; command: string; title: string | null; created_at: string; ended_at: string | null; exit_code: number | null; claude_session_id: string | null; added_dirs: string | null; group_id: string | null; archived_at: string | null; }; /** Parse la colonne `added_dirs` (JSON array de chemins) de façon défensive ; [] si NULL/invalide. */ function parseAddedDirs(raw: string | null): string[] { if (!raw) return []; try { const v = JSON.parse(raw); return Array.isArray(v) ? v.filter((x): x is string => typeof x === 'string') : []; } catch { return []; } } /** Lien entre un client WS attaché et une session. La gateway fournit les callbacks d'envoi. */ export interface ClientBinding { channel: number; mode: 'interactive' | 'observer'; controlling: boolean; sentBytes: number; ackedBytes: number; lagging: boolean; sendOutput: (payload: Buffer) => void; sendResync: (payload: Buffer) => void; onDetached: (reason: 'session_exit' | 'replaced') => void; onControlChanged: (controlling: boolean) => void; } interface ManagedSession { id: string; cwd: string; command: 'claude' | 'bash'; title: string | null; createdAt: string; proc: pty.IPty; ring: RingBuffer; clients: Set; paused: boolean; exited: { exitCode: number | null; signal: number | null } | null; killTimer: NodeJS.Timeout | null; /** ID interne du CLI claude, résolu via le registre après spawn (null pour bash / pas encore prêt). */ claudeSessionId: string | null; /** répertoires supplémentaires reliés dans la session (--add-dir) ; [] pour une session mono-repo (P6). */ addedDirs: string[]; /** groupe propriétaire d'une session de groupe multi-repo ; null sinon (P6). */ groupId: string | null; /** détection d'état fin (busy/waiting/idle + dialogue) ; null pour bash (P3-B). */ tracker: SessionActivityTracker | null; /** dernière activité notifiée (détection du front montant vers `waiting` pour le push P4-B). */ prevActivity: SessionActivity | null; /** timer de notif push débouncée (annulé si la session quitte `waiting` avant l'échéance). */ notifyTimer: NodeJS.Timeout | null; } export interface PtyManagerEvents { session_update: [SessionSummary]; session_exit: [{ sessionId: string; exitCode: number | null; signal: number | null }]; } export class PtyManager extends EventEmitter { private readonly live = new Map(); constructor( private readonly db: Db, private readonly sessionsDir: string = join(homedir(), '.claude', 'sessions'), private readonly push: PushService | null = null, ) { super(); } spawn(opts: { cwd: string; command?: 'claude' | 'bash'; resume?: { claudeSessionId: string; fork?: boolean }; /** répertoires supplémentaires à relier (session de groupe multi-repo, P6). */ addDirs?: string[]; /** groupe propriétaire (session de groupe, P6). */ groupId?: string; }): SessionSummary { const cwd = opts.cwd; if (!existsSync(cwd) || !statSync(cwd).isDirectory()) { throw Object.assign(new Error(`Not a directory: ${cwd}`), { statusCode: 400 }); } // Dédoublonne et écarte le cwd primaire ; valide chaque répertoire supplémentaire (comme le cwd). const addedDirs = [...new Set(opts.addDirs ?? [])].filter((d) => d !== cwd); for (const dir of addedDirs) { if (!existsSync(dir) || !statSync(dir).isDirectory()) { throw Object.assign(new Error(`Not a directory: ${dir}`), { statusCode: 400 }); } } // Un resume/fork est toujours une session claude (le cwd d'origine est garanti par l'appelant — S1). const command = opts.resume ? 'claude' : (opts.command ?? 'claude'); // Override de chemin du binaire claude (réglage UI) lu à chaque spawn → effet sans redémarrage. const claudeBinPath = command === 'claude' ? readClaudeBinPath(this.db) : null; const spec = buildSpawnSpec({ command, ...(claudeBinPath ? { claudeBinPath } : {}), ...(opts.resume ? { resume: opts.resume } : {}), ...(addedDirs.length ? { addDirs: addedDirs } : {}), }); const id = randomUUID(); const proc = pty.spawn(spec.file, spec.args, { name: 'xterm-256color', cols: 120, rows: 32, cwd, env: spec.env as { [key: string]: string }, }); const session: ManagedSession = { id, cwd, command, title: null, createdAt: new Date().toISOString(), proc, ring: new RingBuffer(RING_CAPACITY), clients: new Set(), paused: false, exited: null, killTimer: null, claudeSessionId: null, addedDirs, groupId: opts.groupId ?? null, tracker: null, prevActivity: null, notifyTimer: null, }; // Détection d'état fin (P3-B) : uniquement pour claude (bash n'a pas de registre). if (command === 'claude') { session.tracker = new SessionActivityTracker(proc.pid, this.sessionsDir, () => { if (session.exited) return; const summary = this.summarize(session); this.maybeNotify(session, summary.activity ?? null); this.emit('session_update', summary); }); } this.live.set(id, session); this.db .prepare('INSERT INTO sessions (id, cwd, command, created_at, resumed_from, added_dirs, group_id) VALUES (?, ?, ?, ?, ?, ?, ?)') .run( id, cwd, command, session.createdAt, opts.resume?.claudeSessionId ?? null, addedDirs.length ? JSON.stringify(addedDirs) : null, session.groupId, ); proc.onData((data) => this.handleOutput(session, Buffer.from(data, 'utf8'))); proc.onExit(({ exitCode, signal }) => this.handleExit(session, exitCode, signal ?? null)); if (command === 'claude') this.captureClaudeSessionId(session); const summary = this.summarize(session); this.emit('session_update', summary); return summary; } /** Résout le claudeSessionId du CLI en pollant le registre par pid, puis le persiste (waitReady — S1). */ private captureClaudeSessionId(s: ManagedSession): void { const deadline = Date.now() + CLAUDE_ID_TIMEOUT_MS; const tick = (): void => { if (s.exited || s.claudeSessionId) return; const entry = findByPid(this.sessionsDir, s.proc.pid); if (entry?.claudeSessionId) { s.claudeSessionId = entry.claudeSessionId; this.db.prepare('UPDATE sessions SET claude_session_id = ? WHERE id = ?').run(entry.claudeSessionId, s.id); this.emit('session_update', this.summarize(s)); return; } if (Date.now() < deadline) setTimeout(tick, CLAUDE_ID_POLL_MS).unref(); }; setTimeout(tick, CLAUDE_ID_POLL_MS).unref(); } /** * Contexte de session de groupe (P6) à réinjecter au resume : derniers `added_dirs`/`group_id` * persistés pour ce claudeSessionId. Permet à `--resume` de re-relier les mêmes répertoires. */ groupSessionContext(claudeSessionId: string): { addedDirs: string[]; groupId: string | null } | null { const row = this.db .prepare('SELECT added_dirs, group_id FROM sessions WHERE claude_session_id = ? AND added_dirs IS NOT NULL ORDER BY created_at DESC LIMIT 1') .get(claudeSessionId) as { added_dirs: string | null; group_id: string | null } | undefined; if (!row) return null; return { addedDirs: parseAddedDirs(row.added_dirs), groupId: row.group_id }; } /** * Cible de reprise d'une session managée MORTE (P2/P6), résolue par UUID Arboretum : son cwd * d'origine, son claudeSessionId et son contexte de groupe, lus en DB. null si l'id ne correspond * pas à une session managée morte, reprenable (claude + claudeSessionId connu). * Complète `DiscoveryService.getDiscovered`, qui ne couvre QUE les sessions claude EXTERNES * (une managée connue est justement exclue de la découverte). */ resumeTargetById(id: string): { cwd: string; claudeSessionId: string; addedDirs: string[]; groupId: string | null } | null { if (this.live.has(id)) return null; // vivante : pas de resume direct (fork via le même chemin) const row = this.db .prepare( "SELECT cwd, claude_session_id, added_dirs, group_id FROM sessions WHERE id = ? AND ended_at IS NOT NULL AND claude_session_id IS NOT NULL AND command = 'claude'", ) .get(id) as { cwd: string; claude_session_id: string; added_dirs: string | null; group_id: string | null } | undefined; if (!row) return null; return { cwd: row.cwd, claudeSessionId: row.claude_session_id, addedDirs: parseAddedDirs(row.added_dirs), groupId: row.group_id }; } /** Session managée VIVANTE portant ce claudeSessionId (garde-fou anti-resume d'une session vivante). */ findLiveByClaudeSessionId(claudeSessionId: string): SessionSummary | null { for (const s of this.live.values()) { if (s.claudeSessionId === claudeSessionId) return this.summarize(s); } return null; } /** * Ensemble des claudeSessionId déjà connus d'Arboretum (sessions managées vivantes + historique DB). * Sert au DiscoveryService à exclure les doublons (une managée ne doit pas réapparaître en découverte). */ knownClaudeSessionIds(): Set { const rows = this.db .prepare('SELECT claude_session_id FROM sessions WHERE claude_session_id IS NOT NULL') .all() as Array<{ claude_session_id: string }>; const set = new Set(rows.map((r) => r.claude_session_id)); for (const s of this.live.values()) if (s.claudeSessionId) set.add(s.claudeSessionId); return set; } list(opts?: { includeArchived?: boolean }): SessionSummary[] { const liveSummaries = [...this.live.values()].map((s) => this.summarize(s)); const liveIds = new Set(this.live.keys()); const rows = this.db .prepare('SELECT id, cwd, command, title, created_at, ended_at, exit_code, claude_session_id, added_dirs, group_id, archived_at FROM sessions ORDER BY created_at DESC LIMIT 100') .all() as HistoricalRow[]; const historical: SessionSummary[] = rows .filter((r) => !liveIds.has(r.id)) // Sessions auto-archivées exclues par défaut (filtre indépendant et cumulable avec hidden). .filter((r) => (opts?.includeArchived ? true : r.archived_at == null)) .map((r) => this.historicalSummary(r)); return [...liveSummaries, ...historical]; } /** Construit le SessionSummary d'une ligne historique (session managée terminée). */ private historicalSummary(r: HistoricalRow): SessionSummary { const addedDirs = parseAddedDirs(r.added_dirs); return { id: r.id, cwd: r.cwd, command: r.command, title: r.title, status: 'exited' as const, live: false, createdAt: r.created_at, endedAt: r.ended_at, exitCode: r.exit_code, clients: 0, source: 'managed' as const, claudeSessionId: r.claude_session_id, pid: null, // une session claude morte avec un claudeSessionId connu est reprenable (--resume direct). resumable: r.command === 'claude' && r.claude_session_id != null, attachable: false, registryStatus: null, ...(addedDirs.length ? { addedDirs } : {}), groupId: r.group_id, archived: r.archived_at != null, }; } /** * Ré-émet un `session_update` pour une session historique (P10) — utilisé au dés-archivage pour * que tous les clients rafraîchissent le row (le champ `archived` repasse à false). No-op si la * session est vivante (déjà couverte par le flux live) ou inconnue. */ emitHistoricalUpdate(id: string): void { if (this.live.has(id)) return; const r = this.db .prepare('SELECT id, cwd, command, title, created_at, ended_at, exit_code, claude_session_id, added_dirs, group_id, archived_at FROM sessions WHERE id = ?') .get(id) as HistoricalRow | undefined; if (!r) return; this.emit('session_update', this.historicalSummary(r)); } get(id: string): SessionSummary | null { const s = this.live.get(id); return s ? this.summarize(s) : null; } kill(id: string): boolean { const s = this.live.get(id); if (!s || s.exited) return false; try { process.kill(s.proc.pid, 'SIGTERM'); } catch { return false; } s.killTimer ??= setTimeout(() => { if (!s.exited) { try { process.kill(s.proc.pid, 'SIGKILL'); } catch { /* déjà mort */ } } }, KILL_GRACE_MS); return true; } /** Arrêt du daemon : SIGTERM à toutes les sessions (le CLI nettoie son registre sur SIGTERM — spike S1). */ shutdown(): void { for (const id of this.live.keys()) this.kill(id); } // ---- attach / detach / io ---- attach(sessionId: string, binding: ClientBinding, cols: number, rows: number): { ok: true; controlling: boolean } | { ok: false; code: 'NOT_FOUND' | 'SESSION_EXITED' } { const s = this.live.get(sessionId); if (!s) return { ok: false, code: 'NOT_FOUND' }; if (s.exited) return { ok: false, code: 'SESSION_EXITED' }; const hasController = [...s.clients].some((c) => c.controlling); binding.controlling = binding.mode === 'interactive' && !hasController; s.clients.add(binding); if (binding.controlling) { s.proc.resize(cols, rows); s.tracker?.resize(cols, rows); } // Replay : reset terminal + queue du ring (l'écran TUI courant se reconstitue) binding.sendResync(s.ring.tail(REPLAY_TAIL_BYTES)); binding.sentBytes = 0; binding.ackedBytes = 0; return { ok: true, controlling: binding.controlling }; } detach(sessionId: string, binding: ClientBinding): void { const s = this.live.get(sessionId); if (!s) return; s.clients.delete(binding); if (binding.controlling) { const next = [...s.clients].find((c) => c.mode === 'interactive'); if (next) { next.controlling = true; next.onControlChanged(true); } } this.updateFlowControl(s); } write(sessionId: string, binding: ClientBinding, data: string): 'ok' | 'not_controlling' | 'gone' { const s = this.live.get(sessionId); if (!s || s.exited) return 'gone'; // Outil mono-utilisateur : tous les interactifs peuvent écrire ; les observers jamais. if (binding.mode !== 'interactive') return 'not_controlling'; s.proc.write(data); return 'ok'; } /** * Répond à un dialogue Claude sans clavier (P4-A) : traduit une intention de haut * niveau en keystrokes PTY, validée contre l'état fin du tracker (P3-B). * - 'select' N : positionne le curseur sur l'option N puis confirme (`"N\r"`) — protocole acté spike S3. * - 'confirm' : valide l'option pré-sélectionnée (`"\r"`). * - 'deny' : refus universel (Esc). * Réutilise le chemin write (mono-utilisateur : tout interactif peut répondre, observers non). */ answer( sessionId: string, binding: ClientBinding, action: 'select' | 'confirm' | 'deny', optionN?: number, ): 'ok' | 'not_controlling' | 'gone' | 'invalid' { const s = this.live.get(sessionId); if (!s || s.exited) return 'gone'; if (binding.mode !== 'interactive') return 'not_controlling'; const act = s.tracker?.snapshot(); if (action === 'select') { // L'option doit exister dans le dialogue courant (anti-frappe fantôme mobile). if (!act?.dialog?.options.some((o) => o.n === optionN)) return 'invalid'; s.proc.write(`${optionN}\r`); return 'ok'; } // confirm/deny n'exigent qu'un état d'attente (le dialogue Trust précède le registre — S1). if (act?.activity !== 'waiting') return 'invalid'; s.proc.write(action === 'deny' ? '\x1b' : '\r'); return 'ok'; } resize(sessionId: string, binding: ClientBinding, cols: number, rows: number): void { const s = this.live.get(sessionId); if (!s || s.exited || !binding.controlling) return; s.proc.resize(cols, rows); s.tracker?.resize(cols, rows); } ack(sessionId: string, binding: ClientBinding, bytes: number): void { const s = this.live.get(sessionId); if (!s) return; binding.ackedBytes = Math.max(binding.ackedBytes, bytes); if (binding.lagging && binding.sentBytes - binding.ackedBytes < FLOW.LOW_WATERMARK) { binding.lagging = false; binding.sendResync(s.ring.tail(REPLAY_TAIL_BYTES)); binding.sentBytes = 0; binding.ackedBytes = 0; } this.updateFlowControl(s); } // ---- interne ---- /** * Push P4-B : notifie sur le FRONT MONTANT vers un état qui requiert l'attention — `waiting` * (un dialogue bloque) ou `idle` atteint depuis `busy` (Claude a terminé sa réponse, la session * redevient disponible). Le tracker réémet souvent le même état → on ne réagit qu'au changement. * Debounce annulable : un état ultra-bref (Claude repart tout seul) ne déclenche pas de notif — * à l'échéance on revérifie l'état réel. Cible tous les abonnements (un seul utilisateur). */ private maybeNotify(s: ManagedSession, next: SessionActivity | null): void { const prev = s.prevActivity; s.prevActivity = next; if (!this.push) return; const notifiable = (next === 'waiting' && prev !== 'waiting') || (next === 'idle' && prev === 'busy'); if (notifiable) { const target = next; // 'waiting' | 'idle' if (s.notifyTimer) clearTimeout(s.notifyTimer); s.notifyTimer = setTimeout(() => { s.notifyTimer = null; const act = s.tracker?.snapshot(); if (s.exited || act?.activity !== target) return; // faux positif : l'état a déjà changé const base = { sessionId: s.id, title: basename(s.cwd) || s.cwd, url: `/sessions/${s.id}` }; void this.push?.notify( target === 'idle' ? { ...base, body: 'available for new instructions', kind: null } : { ...base, body: act.dialog?.waitingFor ?? act.waitingFor ?? 'waiting for your input', kind: act.dialog?.kind ?? null }, ); }, NOTIFY_DEBOUNCE_MS); s.notifyTimer.unref(); } else if (s.notifyTimer && next !== 'waiting' && next !== 'idle') { // On quitte un état notifiable avant l'échéance (ex. Claude repart en busy) → annule. clearTimeout(s.notifyTimer); s.notifyTimer = null; } } private handleOutput(s: ManagedSession, chunk: Buffer): void { s.ring.write(chunk); s.tracker?.feed(chunk); for (const c of s.clients) { if (c.lagging) continue; c.sendOutput(chunk); c.sentBytes += chunk.length; if (c.sentBytes - c.ackedBytes > FLOW.LAGGING_BYTES) c.lagging = true; } this.updateFlowControl(s); } /** * pause() seulement quand TOUS les clients interactifs non-lagging dépassent HIGH ; * resume() quand le min repasse sous LOW. Les observers ne freinent jamais le PTY. */ private updateFlowControl(s: ManagedSession): void { if (s.exited) return; const interactive = [...s.clients].filter((c) => c.mode === 'interactive' && !c.lagging); if (interactive.length === 0) { if (s.paused) { s.proc.resume(); s.paused = false; } return; } const outstandings = interactive.map((c) => c.sentBytes - c.ackedBytes); const min = Math.min(...outstandings); if (!s.paused && outstandings.every((o) => o > FLOW.HIGH_WATERMARK)) { s.proc.pause(); s.paused = true; } else if (s.paused && min < FLOW.LOW_WATERMARK) { s.proc.resume(); s.paused = false; } } private handleExit(s: ManagedSession, exitCode: number | null, signal: number | null): void { s.exited = { exitCode, signal }; s.tracker?.dispose(); s.tracker = null; if (s.killTimer) clearTimeout(s.killTimer); if (s.notifyTimer) clearTimeout(s.notifyTimer); const endedAt = new Date().toISOString(); this.db.prepare('UPDATE sessions SET ended_at = ?, exit_code = ? WHERE id = ?').run(endedAt, exitCode, s.id); for (const c of s.clients) c.onDetached('session_exit'); s.clients.clear(); this.live.delete(s.id); this.emit('session_exit', { sessionId: s.id, exitCode, signal }); this.emit('session_update', { ...this.summarize(s), live: false, status: 'exited', endedAt }); } private summarize(s: ManagedSession): SessionSummary { const act = s.tracker?.snapshot(); return { id: s.id, cwd: s.cwd, command: s.command, title: s.title, status: s.exited ? 'exited' : 'running', live: !s.exited, createdAt: s.createdAt, endedAt: null, exitCode: s.exited?.exitCode ?? null, clients: s.clients.size, source: 'managed', claudeSessionId: s.claudeSessionId, pid: s.exited ? null : s.proc.pid, // une managée vivante ne se resume pas (corruption) ; une managée claude morte oui. resumable: !!s.exited && s.command === 'claude' && s.claudeSessionId != null, attachable: !s.exited, // statut brut du registre dérivé de l'activité fine (P3-B) ; null pour bash. registryStatus: act?.activity ?? null, activity: act?.activity ?? null, waitingFor: act?.waitingFor ?? null, dialog: act?.dialog ?? null, ...(s.addedDirs.length ? { addedDirs: s.addedDirs } : {}), groupId: s.groupId, }; } }