/** * 引擎进程管理(任务 A3.10,ADR-0006)。 * * 两个引擎共用一套:会话级资源登记、空闲回收、并发上限、崩溃退避重启、日志(stderr)环形缓冲。 * 平台不关心引擎是子进程还是别的形态——只管「谁在用、用了多久、坏了怎么办」。 * * 默认值(实施设计 9.5):一进程一会话、空闲 15 分钟回收、并发上限 4、连续重启上限 5。 * 重启上限是「连续」的:一次正常回合会把计数清零(markHealthy)。 */ import { AppError } from '../app/errors.js'; import type { RuntimeSession } from './runtime.js'; export interface SupervisorOptions { maxConcurrent: number; idleTimeoutMs: number; sweepIntervalMs: number; /** 单个会话允许的连续重启次数 */ maxRestarts: number; /** 退避基数:第 n 次重启等待 base * 2^(n-1) */ baseBackoffMs: number; /** 引擎日志(stderr 尾部)输出口 */ onLog?: (message: string) => void; /** 便于测试注入时钟 */ now?: () => number; } const DEFAULTS: Omit = { maxConcurrent: 4, idleTimeoutMs: 15 * 60_000, sweepIntervalMs: 60_000, maxRestarts: 5, baseBackoffMs: 1_000, }; const STDERR_TAIL_CHARS = 2000; interface Entry { /** 死了的条目保留为「墓碑」:记着重试次数与退避截止时间 */ dead: boolean; runtimeSession: RuntimeSession | undefined; fingerprint: string; lastUsedAt: number; restarts: number; /** 到点之前不接受新请求;Infinity 表示已达重启上限 */ retryAfter: number; stderrTail: string; stderrOffset: number; } export interface SupervisorStats { active: number; evictedIdle: number; evictedFailed: number; evictedClosed: number; evictedStale: number; rejectedBusy: number; rejectedBackoff: number; } export class RuntimeSupervisor { private readonly entries = new Map(); private readonly options: SupervisorOptions; private sweeper: NodeJS.Timeout | undefined; private readonly counters = { evictedIdle: 0, evictedFailed: 0, evictedClosed: 0, evictedStale: 0, rejectedBusy: 0, rejectedBackoff: 0 }; constructor(options: Partial = {}) { this.options = { ...DEFAULTS, ...options }; } /** 活跃(还活着的)引擎会话数,墓碑不算 */ get size(): number { return [...this.entries.values()].filter((entry) => !entry.dead).length; } stats(): SupervisorStats { return { active: this.size, ...this.counters }; } get(key: string): RuntimeSession | undefined { const entry = this.entries.get(key); return entry?.dead === true ? undefined : entry?.runtimeSession; } /** * 取会话对应的引擎资源:命中且指纹一致就复用;墓碑到点后重建;并发满或退避中则明确拒绝。 */ async acquire(key: string, fingerprint: string, create: () => Promise): Promise { const now = this.now(); let entry = this.entries.get(key); if (entry && entry.fingerprint !== fingerprint) { await this.evict(key, 'stale', '工具集或身份变化'); entry = this.entries.get(key); } if (entry?.dead === true) { if (entry.retryAfter === Number.POSITIVE_INFINITY) { this.counters.rejectedBackoff += 1; throw new AppError('unavailable', `引擎连续启动失败已达上限(${this.options.maxRestarts} 次),停止重试`, { status: 503, retryable: false, }); } if (entry.retryAfter > now) { this.counters.rejectedBackoff += 1; const waitSeconds = Math.max(0.1, Math.round(((entry.retryAfter - now) / 1000) * 10) / 10); throw new AppError('rate_limited', `引擎正在退避重启(第 ${entry.restarts} 次),请 ${waitSeconds} 秒后重试`, { status: 429, retryable: true, }); } const runtimeSession = await create(); this.entries.set(key, { dead: false, runtimeSession, fingerprint, lastUsedAt: now, restarts: entry.restarts, retryAfter: 0, stderrTail: entry.stderrTail, stderrOffset: 0, }); return runtimeSession; } if (entry) { entry.lastUsedAt = now; this.drainStderr(key, entry); return entry.runtimeSession as RuntimeSession; } if (this.size >= this.options.maxConcurrent) { this.counters.rejectedBusy += 1; throw new AppError('rate_limited', `同时活跃的引擎会话已达上限(${this.options.maxConcurrent}),请稍后重试`, { status: 429, retryable: true, }); } const runtimeSession = await create(); this.entries.set(key, { dead: false, runtimeSession, fingerprint, lastUsedAt: now, restarts: 0, retryAfter: 0, stderrTail: '', stderrOffset: 0, }); return runtimeSession; } touch(key: string): void { const entry = this.entries.get(key); if (entry) entry.lastUsedAt = this.now(); } /** 一次正常回合之后调用:连续重启计数清零(上限管的是「连续」失败) */ markHealthy(key: string): void { const entry = this.entries.get(key); if (entry && !entry.dead) entry.restarts = 0; } /** * 回收:idle(空闲)、failed(崩溃或回合失败)、closed(平台关停)、 * stale(工具集或身份变化,旧会话作废——这不是故障,立即重建且不进退避) */ async evict(key: string, reason: 'idle' | 'failed' | 'closed' | 'stale', detail?: string): Promise { const entry = this.entries.get(key); if (!entry) return; if (reason !== 'failed') { this.entries.delete(key); if (reason === 'idle') this.counters.evictedIdle += 1; else if (reason === 'stale') { this.counters.evictedStale += 1; this.logTail(key, entry, detail ?? '配置变化'); } else { this.counters.evictedClosed += 1; this.logTail(key, entry, detail ?? '平台关停'); } await this.closeEntry(entry); return; } this.counters.evictedFailed += 1; const restarts = entry.restarts + 1; await this.closeEntry(entry); if (restarts > this.options.maxRestarts) { this.options.onLog?.(`[process] 会话 ${key} 连续启动失败已达上限(${this.options.maxRestarts}),停止重试:${detail ?? ''}`); this.entries.set(key, { ...entry, dead: true, runtimeSession: undefined, restarts, retryAfter: Number.POSITIVE_INFINITY }); return; } const backoff = this.options.baseBackoffMs * 2 ** (restarts - 1); this.options.onLog?.(`[process] 会话 ${key} 异常回收(${detail ?? '失败'}),第 ${restarts} 次重启将在 ${backoff}ms 后允许`); this.entries.set(key, { ...entry, dead: true, runtimeSession: undefined, restarts, retryAfter: this.now() + backoff }); } /** 平台关停:全部回收,避免子进程残留 */ async disposeAll(): Promise { for (const key of [...this.entries.keys()]) await this.evict(key, 'closed'); this.stopSweeper(); } startSweeper(): void { if (this.sweeper) return; this.sweeper = setInterval(() => { void this.sweepIdle(); }, this.options.sweepIntervalMs); this.sweeper.unref(); } stopSweeper(): void { if (this.sweeper) clearInterval(this.sweeper); this.sweeper = undefined; } async sweepIdle(): Promise { const now = this.now(); for (const [key, entry] of [...this.entries]) { if (entry.dead) continue; // 墓碑由 acquire 处理 if (now - entry.lastUsedAt >= this.options.idleTimeoutMs) { await this.evict(key, 'idle'); } } } /** 引擎 stderr 环形缓冲:只留最近一段,回收时打日志 */ private drainStderr(key: string, entry: Entry): void { if (!entry.runtimeSession) return; const snapshot = entry.runtimeSession.stderrTail?.() ?? ''; if (snapshot.length > entry.stderrOffset) { const delta = snapshot.slice(entry.stderrOffset); entry.stderrOffset = snapshot.length; entry.stderrTail = `${entry.stderrTail}${delta}`.slice(-STDERR_TAIL_CHARS); } } private logTail(key: string, entry: Entry, detail: string): void { if (entry.stderrTail.trim() === '') return; this.options.onLog?.(`[process] 会话 ${key} ${detail},引擎 stderr 尾部:${entry.stderrTail.slice(-400)}`); } private async closeEntry(entry: Entry): Promise { if (!entry.runtimeSession) return; try { this.drainStderr('', entry); await entry.runtimeSession.dispose(); } catch { /* 回收失败不阻塞:进程可能已经死了 */ } } private now(): number { return this.options.now?.() ?? Date.now(); } }