/** * 引擎进程管理用例(任务 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(',')}`); });