From 70a78a695dce5209d11f56cc8354d72fe48ee283 Mon Sep 17 00:00:00 2001 From: zh Date: Tue, 29 Sep 2026 10:46:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E5=BC=95=E6=93=8E=E8=BF=9B?= =?UTF-8?q?=E7=A8=8B=E7=AE=A1=E7=90=86=E2=80=94=E2=80=94=E7=A9=BA=E9=97=B2?= =?UTF-8?q?=E5=9B=9E=E6=94=B6=E3=80=81=E5=B9=B6=E5=8F=91=E4=B8=8A=E9=99=90?= =?UTF-8?q?=E3=80=81=E5=B4=A9=E6=BA=83=E9=80=80=E9=81=BF=E9=87=8D=E5=90=AF?= =?UTF-8?q?=E3=80=81=E6=97=A5=E5=BF=97=E8=BD=AE=E8=BD=AC=EF=BC=88A3.10?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/process-manager.test.mjs | 184 ++++++++++++++++++++++ server/agent/pi/runtime.ts | 4 + server/agent/process-manager.ts | 259 +++++++++++++++++++++++++++++++ server/agent/runtime.ts | 5 + server/app/service.ts | 56 ++++--- server/index.ts | 20 ++- server/package.json | 4 + 7 files changed, 506 insertions(+), 26 deletions(-) create mode 100644 scripts/process-manager.test.mjs create mode 100644 server/agent/process-manager.ts diff --git a/scripts/process-manager.test.mjs b/scripts/process-manager.test.mjs new file mode 100644 index 0000000..84b89b2 --- /dev/null +++ b/scripts/process-manager.test.mjs @@ -0,0 +1,184 @@ +/** + * 引擎进程管理用例（任务 A3.10）。 + * + * 五条验收：可启停、超时中断、崩溃按退避重启且上限可配、日志轮转、平台退出时子进程回收。 + * 超时中断与真进程回收在 pi-runtime / 关停用例里覆盖，这里覆盖调度、重建与退避逻辑。 + */ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +import { RuntimeSupervisor } from '@aps/server/agent/process-manager'; + +const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + +/** 造一个可观测的假运行时会话 */ +function fakeSession(label, log) { + return { + id: label, + toolExecution: 'platform', + async *send() { + yield { type: 'turn_end', reason: 'stop' }; + }, + resolveToolResult() {}, + abort() { + log.push(`abort:${label}`); + }, + async dispose() { + log.push(`dispose:${label}`); + }, + stderrTail: () => log.filter((item) => item.startsWith('stderr')).join(''), + }; +} + +function countingFactory(log) { + let created = 0; + return async () => { + created += 1; + const label = `s${created}`; + log.push(`create:${label}`); + return fakeSession(label, log); + }; +} + +test('空闲回收：到点回收，下一次请求重建', async () => { + const log = []; + const supervisor = new RuntimeSupervisor({ idleTimeoutMs: 30, sweepIntervalMs: 10, maxConcurrent: 4 }); + const create = countingFactory(log); + + await supervisor.acquire('k1', 'fp', create); + assert.equal(supervisor.size, 1); + await sleep(80); + await supervisor.sweepIdle(); + assert.equal(supervisor.size, 0, '空闲会话应被回收'); + assert.equal(supervisor.stats().evictedIdle, 1); + assert.ok(log.includes('dispose:s1'), '回收要真的释放引擎资源'); + + await supervisor.acquire('k1', 'fp', create); + assert.equal(supervisor.stats().active, 1); + assert.ok(log.includes('create:s2'), '下一次请求应重建'); +}); + +test('并发上限：满了明确拒绝（rate_limited 429），回收后可再用', async () => { + const log = []; + const supervisor = new RuntimeSupervisor({ maxConcurrent: 1, idleTimeoutMs: 60_000 }); + const create = countingFactory(log); + + await supervisor.acquire('a', 'fp', create); + await assert.rejects( + () => supervisor.acquire('b', 'fp', create), + (error) => { + assert.equal(error.code, 'rate_limited'); + assert.equal(error.status, 429); + assert.equal(error.retryable, true); + return true; + }, + ); + assert.equal(supervisor.stats().rejectedBusy, 1); + + await supervisor.evict('a', 'idle'); + await supervisor.acquire('b', 'fp', create); + assert.equal(supervisor.stats().active, 1); +}); + +test('崩溃退避重启：按指数退避等待，超过上限停止重试，正常回合清零', async () => { + const log = []; + const supervisor = new RuntimeSupervisor({ maxConcurrent: 4, idleTimeoutMs: 60_000, maxRestarts: 2, baseBackoffMs: 20 }); + const create = countingFactory(log); + + await supervisor.acquire('k', 'fp', create); + await supervisor.evict('k', 'failed', '进程退出'); + // 第一次退避：立即请求被拒 + await assert.rejects(() => supervisor.acquire('k', 'fp', create), /退避重启/); + await sleep(40); + await supervisor.acquire('k', 'fp', create); + assert.equal(supervisor.stats().active, 1, '退避到点后应重建'); + + // 正常回合清零连续失败计数 + supervisor.markHealthy('k'); + await supervisor.evict('k', 'failed'); + await sleep(40); + await supervisor.acquire('k', 'fp', create); + assert.equal(supervisor.stats().active, 1); + + // 连续失败到上限：不再重试 + supervisor.touch('k'); + await supervisor.evict('k', 'failed'); + await sleep(90); + await supervisor.acquire('k', 'fp', create); + await supervisor.evict('k', 'failed'); + await assert.rejects( + () => supervisor.acquire('k', 'fp', create), + (error) => { + assert.equal(error.code, 'unavailable'); + assert.equal(error.retryable, false); + return true; + }, + '超过重启上限后应停止重试', + ); + assert.ok(supervisor.stats().rejectedBackoff >= 1); +}); + +test('工具集/身份变化：旧会话作废并立即重建，不进退避也不计崩溃', async () => { + const log = []; + const supervisor = new RuntimeSupervisor({ maxConcurrent: 4, idleTimeoutMs: 60_000, maxRestarts: 1, baseBackoffMs: 5_000 }); + const create = countingFactory(log); + + await supervisor.acquire('k', 'fp-a', create); + // 指纹变化（例如新注册了 Provider）：立刻换新，不能像崩溃那样把用户的请求挡在退避里 + await supervisor.acquire('k', 'fp-b', create); + assert.ok(log.includes('create:s2'), '指纹变化应立刻重建运行时会话'); + assert.equal(supervisor.stats().active, 1); + assert.equal(supervisor.stats().evictedStale, 1); + assert.equal(supervisor.stats().evictedFailed, 0, '更换工具集不是故障，不该计入崩溃回收'); + assert.ok(log.includes('dispose:s1'), '旧会话要真的释放'); +}); +test('平台关停：全部回收并记录引擎日志尾部', async () => { + const log = []; + const supervisor = new RuntimeSupervisor({ maxConcurrent: 4, idleTimeoutMs: 60_000, onLog: (message) => log.push(message) }); + const create = countingFactory(log); + await supervisor.acquire('a', 'fp', create); + await supervisor.acquire('b', 'fp', create); + + await supervisor.disposeAll(); + assert.equal(supervisor.stats().active, 0); + assert.equal(supervisor.stats().evictedClosed, 2); + assert.ok(log.includes('dispose:s1') && log.includes('dispose:s2'), '关停要把所有引擎资源释放'); +}); + +test('服务层：关停平台时引擎会话被释放（无残留进程）', async () => { + const log = []; + let created = 0; + const runtime = { + name: 'counting', + toolExecution: 'platform', + async createSession() { + created += 1; + return fakeSession(`svc${created}`, log); + }, + }; + + process.env.APS_DEV_PROVIDER = '1'; + const { startServer } = await import('../server/dist/index.js'); + const server = await startServer({ host: '127.0.0.1', port: 0, version: 'test', databasePath: ':memory:', engine: 'fake', runtime }); + try { + const created1 = await ( + await fetch(`${server.url}/v1/sessions`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ scope: 'project', projectId: 'p-default', title: '关停用例' }), + }) + ).json(); + const sessionId = created1.session.sessionId; + await fetch(`${server.url}/v1/sessions/${sessionId}/messages`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ text: '你好', idempotencyKey: 'idem-proc-0001' }), + }); + await sleep(200); + assert.equal(server.service.processStats().active, 1, '发过消息后应有一个活跃引擎会话'); + } finally { + await server.close(); + delete process.env.APS_DEV_PROVIDER; + } + assert.ok(log.includes('dispose:svc1'), `关停平台应释放引擎会话，实际日志：${log.join(',')}`); +}); \ No newline at end of file diff --git a/server/agent/pi/runtime.ts b/server/agent/pi/runtime.ts index 3b9d56f..46c24e5 100644 --- a/server/agent/pi/runtime.ts +++ b/server/agent/pi/runtime.ts @@ -281,6 +281,10 @@ export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime { abort(): void { void client.abort(); }, + /** 日志轮转用：把引擎 stderr 交给进程管理层（它只保留尾部） */ + stderrTail(): string { + return client.getStderr(); + }, async dispose(): Promise { await client.stop(); }, diff --git a/server/agent/process-manager.ts b/server/agent/process-manager.ts new file mode 100644 index 0000000..a1eb9fa --- /dev/null +++ b/server/agent/process-manager.ts @@ -0,0 +1,259 @@ +/** + * 引擎进程管理（任务 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(); + } +} \ No newline at end of file diff --git a/server/agent/runtime.ts b/server/agent/runtime.ts index f94754f..8f95bf4 100644 --- a/server/agent/runtime.ts +++ b/server/agent/runtime.ts @@ -87,6 +87,11 @@ export interface RuntimeSession { abort(): void; /** 释放运行时资源（子进程、连接） */ dispose(): Promise; + /** + * 可选的引擎 stderr 快照：进程管理层用它做日志轮转（只保留尾部）。 + * 不是所有引擎都有子进程 stderr，缺省实现可以不给。 + */ + stderrTail?(): string; } export interface AgentRuntime { diff --git a/server/app/service.ts b/server/app/service.ts index eed3b95..11cf86b 100644 --- a/server/app/service.ts +++ b/server/app/service.ts @@ -12,6 +12,7 @@ import { createFakeRuntime } from '../agent/fake-runtime.js'; import type { AgentRuntime, RuntimeSession } from '../agent/runtime.js'; import { createRulePlanner, plannerToFakeTurn } from '../agent/rule-planner.js'; import { toToolDescriptors } from '../agent/tools.js'; +import { RuntimeSupervisor, type SupervisorOptions } from '../agent/process-manager.js'; import type { SessionRepository } from '../core/storage/sessions.js'; import { capabilityFingerprint, @@ -40,6 +41,10 @@ export interface ApplicationServiceOptions { contextRepository?: ContextRepository; /** 上下文预算覆盖（测试用） */ contextBudget?: Partial; + /** 引擎进程管理（A3.10）：不传则用默认值自建 */ + supervisor?: RuntimeSupervisor; + /** 进程管理参数覆盖（测试用） */ + supervisorOptions?: Partial; } export class ApplicationService { @@ -56,9 +61,8 @@ export class ApplicationService { /** 已经注入过上下文的运行时会话：同一运行时会话只注入一次（它自己有原生上下文） */ private readonly injectedRuntimeSessions = new Set(); /** 每个平台会话一个运行时会话（ADR-0006：首版一进程一会话） */ - private readonly runtimeSessions = new Map(); - /** 会话 → 工具集指纹：身份或可见能力变了就重建运行时会话 */ - private readonly runtimeFingerprints = new Map(); + /** 引擎进程管理：空闲回收、并发上限、崩溃退避重启（A3.10） */ + readonly supervisor: RuntimeSupervisor; constructor(options: ApplicationServiceOptions = {}) { this.version = options.version ?? '0.1.0'; @@ -66,6 +70,7 @@ export class ApplicationService { this.visibility = options.capabilityVisibility ?? { defaultAllow: true, denyWrite: false, grants: [] }; this.contextRepository = options.contextRepository; this.contextBudget = options.contextBudget; + this.supervisor = options.supervisor ?? new RuntimeSupervisor(options.supervisorOptions ?? {}); this.core = createCore({ registry: this.registry }); this.runtime = options.runtime ?? createFakeRuntime({ planner: plannerToFakeTurn(createRulePlanner()) }); } @@ -215,7 +220,8 @@ export class ApplicationService { type: 'run.state', data: { state: 'cancelling' }, }); - this.runtimeSessions.get(sessionId)?.abort(); + // 取消：让引擎停在这一轮，回合自然结束（资源等空闲回收或失败回收） + this.supervisor.get(sessionId)?.abort(); for (const run of record.runs.values()) run.abort.abort(); return { state: 'cancelling' }; } @@ -258,6 +264,13 @@ export class ApplicationService { if (result.text !== '') { this.sessions.appendMessage(sessionId, { messageId: assistantMessageId, role: 'assistant', ts: Date.now(), text: result.text }); } + // 正常回合：清零连续失败计数；错误回合：回收后按退避重建（A3.10） + if (result.finishReason === 'error') { + await this.supervisor.evict(sessionId, 'failed', '回合以错误收尾'); + } else { + this.supervisor.markHealthy(sessionId); + this.supervisor.touch(sessionId); + } // 检查点：把这次构建出的摘要与来源落库，摘要与原始消息都保留（实施设计 1.3） this.contextRepository?.save({ sessionId, @@ -269,6 +282,7 @@ export class ApplicationService { estimatedChars: context.estimatedChars, }); } catch (error) { + await this.supervisor.evict(sessionId, 'failed', '引擎调用抛错').catch(() => undefined); record.bus.publish({ sessionId, runId, @@ -302,24 +316,22 @@ export class ApplicationService { private async runtimeSessionFor(record: SessionRecord): Promise { const sessionId = record.summary.sessionId; const capabilities = this.visibleCapabilitiesFor(record.owner); - const fingerprint = `${record.owner.userId}|${capabilityFingerprint(capabilities)}`; - const existing = this.runtimeSessions.get(sessionId); // 身份或工具集变了就重建：工具集不能跨身份复用（架构设计 5.1「按身份给工具」） - if (existing && this.runtimeFingerprints.get(sessionId) === fingerprint) return existing; - if (existing) { - await existing.dispose().catch(() => undefined); - this.runtimeSessions.delete(sessionId); - } - const created = await this.runtime.createSession({ - sessionId, - subject: { userId: record.owner.userId, tenantId: record.owner.tenantId }, - projectId: record.summary.projectId, - system: PLATFORM_SYSTEM_PROMPT, - tools: toToolDescriptors(capabilities), - history: record.messages, - }); - this.runtimeSessions.set(sessionId, created); - this.runtimeFingerprints.set(sessionId, fingerprint); - return created; + const fingerprint = `${record.owner.userId}|${capabilityFingerprint(capabilities)}`; + return this.supervisor.acquire(sessionId, fingerprint, () => + this.runtime.createSession({ + sessionId, + subject: { userId: record.owner.userId, tenantId: record.owner.tenantId }, + projectId: record.summary.projectId, + system: PLATFORM_SYSTEM_PROMPT, + tools: toToolDescriptors(capabilities), + history: record.messages, + }), + ); + } + + /** 进程管理统计（观测与用例用） */ + processStats() { + return this.supervisor.stats(); } } \ No newline at end of file diff --git a/server/index.ts b/server/index.ts index 8469601..acaa2ad 100644 --- a/server/index.ts +++ b/server/index.ts @@ -13,6 +13,7 @@ import type { CapabilityVisibility } from './core/registry/visibility.js'; import { openDatabase, type Database } from './core/storage/db.js'; import { SessionRepository } from './core/storage/sessions.js'; import { ContextRepository } from './core/storage/context.js'; +import { RuntimeSupervisor } from './agent/process-manager.js'; import { ApplicationService } from './app/service.js'; import { createHttpServer } from './entry/http.js'; @@ -54,12 +55,19 @@ export async function startServer(options: StartOptions = {}): Promise bridgeRef.info)); const database: Database = openDatabase(options.databasePath ?? config.database.path); + // 引擎进程管理（A3.10）：空闲回收与并发上限可按环境变量调；日志走 stderr + const supervisor = new RuntimeSupervisor({ + maxConcurrent: Number.parseInt(process.env.APS_MAX_CONCURRENT_ENGINES ?? '4', 10), + idleTimeoutMs: Number.parseInt(process.env.APS_ENGINE_IDLE_MS ?? String(15 * 60_000), 10), + onLog: (message) => console.error(message), + }); const service = new ApplicationService({ ...(options.version === undefined ? {} : { version: options.version }), runtime, sessionRepository: new SessionRepository(database), capabilityVisibility: options.capabilityVisibility ?? config.capabilities, contextRepository: new ContextRepository(database), + supervisor, }); // 开发期可选：APS_DEV_PROVIDER=1 时注册假 Provider，用于本地联调与演示（默认不注册，能力清单为空） if (process.env.APS_DEV_PROVIDER === '1') { @@ -67,6 +75,7 @@ export async function startServer(options: StartOptions = {}): Promise((resolve, reject) => { @@ -85,14 +94,17 @@ export async function startServer(options: StartOptions = {}): Promise - new Promise((resolve, reject) => { + close: async () => { + await new Promise((resolve, reject) => { server.close((error) => (error ? reject(error) : resolve())); // 事件流是长连接，关停时必须主动断开，否则 close 会一直等它们 // 客户端断线后按 Last-Event-ID 重连，不会丢事件（协议文档第 5 节） server.closeAllConnections(); - database.close(); - }), + }); + // 等引擎子进程回收完成再返回：关停后不该有残留进程（A3.10） + await supervisor.disposeAll(); + database.close(); + }, }; } /** 引擎选择：显式注入 > 配置里的 engine.kind；选 pi 但没密钥时明确回退到假引擎 */ diff --git a/server/package.json b/server/package.json index 41f0448..e98f05a 100644 --- a/server/package.json +++ b/server/package.json @@ -30,6 +30,10 @@ "types": "./dist/agent/fake-runtime.d.ts", "default": "./dist/agent/fake-runtime.js" }, + "./agent/process-manager": { + "types": "./dist/agent/process-manager.d.ts", + "default": "./dist/agent/process-manager.js" + }, "./agent/pi/runtime": { "types": "./dist/agent/pi/runtime.d.ts", "default": "./dist/agent/pi/runtime.js"