aps-agent/scripts/process-manager.test.mjs

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