feat(agent): 引擎进程管理——空闲回收、并发上限、崩溃退避重启、日志轮转(A3.10)

This commit is contained in:
zh 2026-09-29 10:46:13 +08:00
parent b20dbd17d0
commit 70a78a695d
7 changed files with 506 additions and 26 deletions

View File

@ -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(',')}`);
});

View File

@ -281,6 +281,10 @@ export function createPiRuntime(options: PiRuntimeOptions = {}): AgentRuntime {
abort(): void { abort(): void {
void client.abort(); void client.abort();
}, },
/** 日志轮转用:把引擎 stderr 交给进程管理层(它只保留尾部) */
stderrTail(): string {
return client.getStderr();
},
async dispose(): Promise<void> { async dispose(): Promise<void> {
await client.stop(); await client.stop();
}, },

View File

@ -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<SupervisorOptions, 'onLog' | 'now'> = {
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<string, Entry>();
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<SupervisorOptions> = {}) {
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<RuntimeSession>): Promise<RuntimeSession> {
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<void> {
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<void> {
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<void> {
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<void> {
if (!entry.runtimeSession) return;
try {
this.drainStderr('', entry);
await entry.runtimeSession.dispose();
} catch {
/* 回收失败不阻塞:进程可能已经死了 */
}
}
private now(): number {
return this.options.now?.() ?? Date.now();
}
}

View File

@ -87,6 +87,11 @@ export interface RuntimeSession {
abort(): void; abort(): void;
/** 释放运行时资源(子进程、连接) */ /** 释放运行时资源(子进程、连接) */
dispose(): Promise<void>; dispose(): Promise<void>;
/**
* 可选的引擎 stderr 快照:进程管理层用它做日志轮转(只保留尾部)。
* 不是所有引擎都有子进程 stderr,缺省实现可以不给。
*/
stderrTail?(): string;
} }
export interface AgentRuntime { export interface AgentRuntime {

View File

@ -12,6 +12,7 @@ import { createFakeRuntime } from '../agent/fake-runtime.js';
import type { AgentRuntime, RuntimeSession } from '../agent/runtime.js'; import type { AgentRuntime, RuntimeSession } from '../agent/runtime.js';
import { createRulePlanner, plannerToFakeTurn } from '../agent/rule-planner.js'; import { createRulePlanner, plannerToFakeTurn } from '../agent/rule-planner.js';
import { toToolDescriptors } from '../agent/tools.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 type { SessionRepository } from '../core/storage/sessions.js';
import { import {
capabilityFingerprint, capabilityFingerprint,
@ -40,6 +41,10 @@ export interface ApplicationServiceOptions {
contextRepository?: ContextRepository; contextRepository?: ContextRepository;
/** 上下文预算覆盖(测试用) */ /** 上下文预算覆盖(测试用) */
contextBudget?: Partial<ContextBudget>; contextBudget?: Partial<ContextBudget>;
/** 引擎进程管理(A3.10):不传则用默认值自建 */
supervisor?: RuntimeSupervisor;
/** 进程管理参数覆盖(测试用) */
supervisorOptions?: Partial<SupervisorOptions>;
} }
export class ApplicationService { export class ApplicationService {
@ -56,9 +61,8 @@ export class ApplicationService {
/** 已经注入过上下文的运行时会话:同一运行时会话只注入一次(它自己有原生上下文) */ /** 已经注入过上下文的运行时会话:同一运行时会话只注入一次(它自己有原生上下文) */
private readonly injectedRuntimeSessions = new Set<string>(); private readonly injectedRuntimeSessions = new Set<string>();
/** 每个平台会话一个运行时会话(ADR-0006:首版一进程一会话) */ /** 每个平台会话一个运行时会话(ADR-0006:首版一进程一会话) */
private readonly runtimeSessions = new Map<string, RuntimeSession>(); /** 引擎进程管理:空闲回收、并发上限、崩溃退避重启(A3.10) */
/** 会话 → 工具集指纹:身份或可见能力变了就重建运行时会话 */ readonly supervisor: RuntimeSupervisor;
private readonly runtimeFingerprints = new Map<string, string>();
constructor(options: ApplicationServiceOptions = {}) { constructor(options: ApplicationServiceOptions = {}) {
this.version = options.version ?? '0.1.0'; this.version = options.version ?? '0.1.0';
@ -66,6 +70,7 @@ export class ApplicationService {
this.visibility = options.capabilityVisibility ?? { defaultAllow: true, denyWrite: false, grants: [] }; this.visibility = options.capabilityVisibility ?? { defaultAllow: true, denyWrite: false, grants: [] };
this.contextRepository = options.contextRepository; this.contextRepository = options.contextRepository;
this.contextBudget = options.contextBudget; this.contextBudget = options.contextBudget;
this.supervisor = options.supervisor ?? new RuntimeSupervisor(options.supervisorOptions ?? {});
this.core = createCore({ registry: this.registry }); this.core = createCore({ registry: this.registry });
this.runtime = options.runtime ?? createFakeRuntime({ planner: plannerToFakeTurn(createRulePlanner()) }); this.runtime = options.runtime ?? createFakeRuntime({ planner: plannerToFakeTurn(createRulePlanner()) });
} }
@ -215,7 +220,8 @@ export class ApplicationService {
type: 'run.state', type: 'run.state',
data: { state: 'cancelling' }, data: { state: 'cancelling' },
}); });
this.runtimeSessions.get(sessionId)?.abort(); // 取消:让引擎停在这一轮,回合自然结束(资源等空闲回收或失败回收)
this.supervisor.get(sessionId)?.abort();
for (const run of record.runs.values()) run.abort.abort(); for (const run of record.runs.values()) run.abort.abort();
return { state: 'cancelling' }; return { state: 'cancelling' };
} }
@ -258,6 +264,13 @@ export class ApplicationService {
if (result.text !== '') { if (result.text !== '') {
this.sessions.appendMessage(sessionId, { messageId: assistantMessageId, role: 'assistant', ts: Date.now(), text: 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) // 检查点:把这次构建出的摘要与来源落库,摘要与原始消息都保留(实施设计 1.3)
this.contextRepository?.save({ this.contextRepository?.save({
sessionId, sessionId,
@ -269,6 +282,7 @@ export class ApplicationService {
estimatedChars: context.estimatedChars, estimatedChars: context.estimatedChars,
}); });
} catch (error) { } catch (error) {
await this.supervisor.evict(sessionId, 'failed', '引擎调用抛错').catch(() => undefined);
record.bus.publish({ record.bus.publish({
sessionId, sessionId,
runId, runId,
@ -302,24 +316,22 @@ export class ApplicationService {
private async runtimeSessionFor(record: SessionRecord): Promise<RuntimeSession> { private async runtimeSessionFor(record: SessionRecord): Promise<RuntimeSession> {
const sessionId = record.summary.sessionId; const sessionId = record.summary.sessionId;
const capabilities = this.visibleCapabilitiesFor(record.owner); const capabilities = this.visibleCapabilitiesFor(record.owner);
const fingerprint = `${record.owner.userId}|${capabilityFingerprint(capabilities)}`;
const existing = this.runtimeSessions.get(sessionId);
// 身份或工具集变了就重建:工具集不能跨身份复用(架构设计 5.1「按身份给工具」) // 身份或工具集变了就重建:工具集不能跨身份复用(架构设计 5.1「按身份给工具」)
if (existing && this.runtimeFingerprints.get(sessionId) === fingerprint) return existing; const fingerprint = `${record.owner.userId}|${capabilityFingerprint(capabilities)}`;
if (existing) { return this.supervisor.acquire(sessionId, fingerprint, () =>
await existing.dispose().catch(() => undefined); this.runtime.createSession({
this.runtimeSessions.delete(sessionId);
}
const created = await this.runtime.createSession({
sessionId, sessionId,
subject: { userId: record.owner.userId, tenantId: record.owner.tenantId }, subject: { userId: record.owner.userId, tenantId: record.owner.tenantId },
projectId: record.summary.projectId, projectId: record.summary.projectId,
system: PLATFORM_SYSTEM_PROMPT, system: PLATFORM_SYSTEM_PROMPT,
tools: toToolDescriptors(capabilities), tools: toToolDescriptors(capabilities),
history: record.messages, history: record.messages,
}); }),
this.runtimeSessions.set(sessionId, created); );
this.runtimeFingerprints.set(sessionId, fingerprint); }
return created;
/** 进程管理统计(观测与用例用) */
processStats() {
return this.supervisor.stats();
} }
} }

View File

@ -13,6 +13,7 @@ import type { CapabilityVisibility } from './core/registry/visibility.js';
import { openDatabase, type Database } from './core/storage/db.js'; import { openDatabase, type Database } from './core/storage/db.js';
import { SessionRepository } from './core/storage/sessions.js'; import { SessionRepository } from './core/storage/sessions.js';
import { ContextRepository } from './core/storage/context.js'; import { ContextRepository } from './core/storage/context.js';
import { RuntimeSupervisor } from './agent/process-manager.js';
import { ApplicationService } from './app/service.js'; import { ApplicationService } from './app/service.js';
import { createHttpServer } from './entry/http.js'; import { createHttpServer } from './entry/http.js';
@ -54,12 +55,19 @@ export async function startServer(options: StartOptions = {}): Promise<RunningSe
const bridgeRef: { info?: { baseUrl: string; internalToken: string } } = {}; const bridgeRef: { info?: { baseUrl: string; internalToken: string } } = {};
const runtime = options.runtime ?? (await selectRuntime(config, options.engine, () => bridgeRef.info)); const runtime = options.runtime ?? (await selectRuntime(config, options.engine, () => bridgeRef.info));
const database: Database = openDatabase(options.databasePath ?? config.database.path); 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({ const service = new ApplicationService({
...(options.version === undefined ? {} : { version: options.version }), ...(options.version === undefined ? {} : { version: options.version }),
runtime, runtime,
sessionRepository: new SessionRepository(database), sessionRepository: new SessionRepository(database),
capabilityVisibility: options.capabilityVisibility ?? config.capabilities, capabilityVisibility: options.capabilityVisibility ?? config.capabilities,
contextRepository: new ContextRepository(database), contextRepository: new ContextRepository(database),
supervisor,
}); });
// 开发期可选:APS_DEV_PROVIDER=1 时注册假 Provider,用于本地联调与演示(默认不注册,能力清单为空) // 开发期可选:APS_DEV_PROVIDER=1 时注册假 Provider,用于本地联调与演示(默认不注册,能力清单为空)
if (process.env.APS_DEV_PROVIDER === '1') { if (process.env.APS_DEV_PROVIDER === '1') {
@ -67,6 +75,7 @@ export async function startServer(options: StartOptions = {}): Promise<RunningSe
await service.registerProvider(createFakeProvider()); await service.registerProvider(createFakeProvider());
} }
supervisor.startSweeper();
const server: Server = createHttpServer(service, internalToken); const server: Server = createHttpServer(service, internalToken);
await new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
@ -85,14 +94,17 @@ export async function startServer(options: StartOptions = {}): Promise<RunningSe
service, service,
runtime, runtime,
internalToken, internalToken,
close: () => close: async () => {
new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve())); server.close((error) => (error ? reject(error) : resolve()));
// 事件流是长连接,关停时必须主动断开,否则 close 会一直等它们 // 事件流是长连接,关停时必须主动断开,否则 close 会一直等它们
// 客户端断线后按 Last-Event-ID 重连,不会丢事件(协议文档第 5 节) // 客户端断线后按 Last-Event-ID 重连,不会丢事件(协议文档第 5 节)
server.closeAllConnections(); server.closeAllConnections();
});
// 等引擎子进程回收完成再返回:关停后不该有残留进程(A3.10)
await supervisor.disposeAll();
database.close(); database.close();
}), },
}; };
} }
/** 引擎选择:显式注入 > 配置里的 engine.kind;选 pi 但没密钥时明确回退到假引擎 */ /** 引擎选择:显式注入 > 配置里的 engine.kind;选 pi 但没密钥时明确回退到假引擎 */

View File

@ -30,6 +30,10 @@
"types": "./dist/agent/fake-runtime.d.ts", "types": "./dist/agent/fake-runtime.d.ts",
"default": "./dist/agent/fake-runtime.js" "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": { "./agent/pi/runtime": {
"types": "./dist/agent/pi/runtime.d.ts", "types": "./dist/agent/pi/runtime.d.ts",
"default": "./dist/agent/pi/runtime.js" "default": "./dist/agent/pi/runtime.js"