fmode-app/scripts/loadtest-agent.mjs

#!/usr/bin/env node
/**
 * 容器并发压力测试（多会话）
 *
 * 用法：
 *   node scripts/loadtest-agent.mjs --agent agent-node-kangshifu --n 10
 *   node scripts/loadtest-agent.mjs --agent agent-node-kangshifu --n 10 --timeout 600
 *
 * ## 为什么需要它
 *
 * 容器自己造不出 N 个独立会话 —— 它只有自己的 agent 身份，没有用户侧身份
 * （2026-11-14 kangshifu 实测：4 条路全不通）。
 *
 * 本脚本从**外部**用 `RP5Ht55NeX` 给用户签身份、建 N 个独立房间，
 * 再**同时**发消息，制造真实并发。
 *
 * ## 测什么
 *
 *   1. 完整性：N 条里几条真的回了
 *   2. 串台：每个房间的回复里有没有混进别的房间的 token
 *   3. 端到端耗时：发出 → 收到 ✅ 用了多久
 *   4. 文件产出：回复里出现的 s3 链接
 *
 * 🔴 **每个房间的指令带唯一 token**，回复里必须出现**自己那条**的 token。
 *    出现别人的 token = 串台，这是「一房间一会话」的核心验证。
 */
import { readFileSync, writeFileSync, mkdirSync } from 'node:fs';
import { homedir } from 'node:os';
import { join, dirname } from 'node:path';

const FUNCTIONS = 'https://server.fmode.cn/api/functions';
const FN_PROVISION = 'RP5Ht55NeX';

const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

function arg(name, fallback) {
  const i = process.argv.indexOf(`--${name}`);
  return i >= 0 && process.argv[i + 1] ? process.argv[i + 1] : fallback;
}

const AGENT = arg('agent', 'agent-node-kangshifu');
const N = Number(arg('n', '10'));
/** 单条等待上限（秒）。容器并发时可能排队，给足 */
const TIMEOUT_S = Number(arg('timeout', '600'));
/** 每轮轮询间隔（毫秒） */
const POLL_MS = Number(arg('poll', '3000'));
const OUT = arg('out', `docs/03-interface/loadtest-${AGENT}-n${N}.md`);

function loadToken() {
  /* 允许用 --token 或 LT_TOKEN 覆盖，方便用测试账号（如 ksftest01）跑压测 */
  const override = arg('token', process.env.LT_TOKEN);
  if (override) {
    if (!override.startsWith('r:')) throw new Error('--token 必须以 r: 开头');
    return { token: override, src: '--token/环境变量' };
  }
  const p = join(homedir(), '.fmode', 'config', 'user.json');
  const t = JSON.parse(readFileSync(p, 'utf8'))?.sessionToken;
  if (typeof t !== 'string' || !t.startsWith('r:')) throw new Error(`读不到会话凭据：${p}`);
  return { token: t, src: p };
}
const { token: TOKEN, src: TOKEN_SRC } = loadToken();

async function provision(agentId) {
  const r = await fetch(FUNCTIONS, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'X-Parse-Application-Id': 'ncloudmaster' },
    body: JSON.stringify({ id: FN_PROVISION, token: TOKEN, params: { agentId } }),
  });
  const j = await r.json();
  const d = j.result ?? j;
  if (!d?.accessToken) throw new Error(`provision 失败：${JSON.stringify(j).slice(0, 300)}`);
  return d;
}

/** 生成一个可辨识的 token（8 位） */
function mkToken(i) {
  return `T${i}-${Math.random().toString(36).slice(2, 8).toUpperCase()}`;
}

async function main() {
  console.log(`\n=== 容器并发压测 ===`);
  console.log(`容器  : ${AGENT}`);
  console.log(`并发数: ${N}`);
  console.log(`凭据  : ${TOKEN_SRC}`);
  console.log(`超时  : ${TIMEOUT_S}s\n`);

  const prov = await provision(AGENT);
  const agentInfo = await provision(AGENT);
  console.log(`用户: ${prov.userMxId} | 容器: ${agentInfo.agentMxId}\n`);

  const mx = await import('matrix-js-sdk');
  const client = mx.createClient({
    baseUrl: prov.homeserver,
    accessToken: prov.accessToken,
    userId: prov.userMxId,
  });
  const hs = prov.homeserver.replace(/\/$/, '');
  const auth = { Authorization: `Bearer ${prov.accessToken}` };

  // ── 1. 建 N 个独立房间 ──────────────────────────────────────────────
  console.log('=== [1] 建房间 ===');
  const peers = [];
  for (let i = 1; i <= N; i++) {
    const room = (
      await client.createRoom({
        invite: [agentInfo.agentMxId],
        name: `loadtest-n${N}-${i}-${Date.now()}`,
        preset: mx.Preset.TrustedPrivateChat,
      })
    ).room_id;
    peers.push({ i, room, token: mkToken(i), sentId: null, ackedAt: null, events: [], joined: false });
    console.log(`  [${i}] ${room}`);
  }

  // ── 2. 等容器进房（并发等待，最多 120 秒）───────────────────────────
  console.log('\n=== [2] 等容器进房 ===');
  const joinDeadline = Date.now() + 120_000;
  while (Date.now() < joinDeadline) {
    const pend = peers.filter((p) => !p.joined);
    if (!pend.length) break;
    await Promise.all(
      pend.map(async (p) => {
        const r = await fetch(
          `${hs}/_matrix/client/v3/rooms/${encodeURIComponent(p.room)}/joined_members`,
          { headers: auth },
        ).catch(() => null);
        const j = r ? await r.json().catch(() => ({})) : {};
        if ((j.joined ?? {})[agentInfo.agentMxId]) p.joined = true;
      }),
    );
    await sleep(2000);
  }
  const joinedCount = peers.filter((p) => p.joined).length;
  console.log(`  已进房: ${joinedCount}/${N}`);
  if (joinedCount === 0) throw new Error('容器一个房间都没进，压测无法进行');

  // ── 3. 同时发消息 ────────────────────────────────────────────────────
  console.log('\n=== [3] 并发发送 ===');
  const t0 = Date.now();
  /**
   * 🔴 指令必须逼它走**真实交付链路**（2026-11-14 修正）。
   *
   * 上一版的坑：只说「写文件并上传」，容器用 `uploader.mjs put` 直传，
   * 链接是 `https://s3.fmode.cn/user/<uid>/report/...` —— **不是**
   * `/deliverables/<agentId>/...`，App 侧 `isAgentOutputUrl()` 不认，
   * 等于测了个假链路。
   *
   * 现在明确要求：写 8KB 内容（足够撑满一次真实上传）+ 回复时用 `MEDIA:` 行，
   * 走 `plugin-deliverable-inject` → 出站媒体 → `/deliverables/` 前缀。
   */
  const payloadFor = (i, token) => {
    /* 约 8KB 的可辨识内容：既压上传，又能验证文件内容没错乱 */
    const lines = [];
    for (let k = 0; k < 120; k++) {
      lines.push(`L${String(k).padStart(3, '0')} n=${i} token=${token} payload=${'x'.repeat(40)}`);
    }
    return { head: `LOADTEST n=${i} token=${token}`, body: lines.join('\n') };
  };

  await Promise.all(
    peers.map(async (p) => {
      if (!p.joined) return;
      const { head, body } = payloadFor(p.i, p.token);
      p.expectBytes = body.length;
      const body400 = [
        `压测任务 ${p.i}。严格按下面做，不要做别的事，不要解释：`,
        ``,
        `1. 用 shell 创建文件 /opt/data/work/loadtest-n${N}/msg-${p.i}.md，内容严格为下面 ${body.split('\n').length} 行（每行都要有）：`,
        ``,
        '```',
        body,
        '```',
        ``,
        `2. 确认第一行是：${head}`,
        `3. 在回复正文里**单独一行**写：MEDIA:/opt/data/work/loadtest-n${N}/msg-${p.i}.md`,
        `4. 回复里必须原样出现 token：${p.token}`,
      ].join('\n');
      p.promptBytes = body400.length;
      const sent = await client.sendEvent(p.room, 'm.room.message', {
        msgtype: 'm.text',
        body: body400,
        'm.mentions': { user_ids: [agentInfo.agentMxId] },
      });
      p.sentId = sent.event_id;
      p.sentAt = Date.now();
    }),
  );
  console.log(`  ${peers.filter((p) => p.sentId).length} 条已发出（每条含约 8KB 写入负载），等待回复…\n`);

  // ── 4. 轮询收集 ──────────────────────────────────────────────────────
  const deadline = t0 + TIMEOUT_S * 1000;
  let lastReport = 0;
  while (Date.now() < deadline) {
    await sleep(POLL_MS);

    await Promise.all(
      peers.map(async (p) => {
        if (!p.sentId || p.ackedAt) return;
        const r = await fetch(
          `${hs}/_matrix/client/v3/rooms/${encodeURIComponent(p.room)}/messages?dir=b&limit=60`,
          { headers: auth },
        ).catch(() => null);
        if (!r) return;
        const j = await r.json().catch(() => ({}));
        const seen = new Set(p.events.map((e) => e.id));
        for (const ev of (j.chunk ?? []).reverse()) {
          if (seen.has(ev.event_id)) continue;
          if (ev.type === 'm.reaction') {
            const rel = ev.content?.['m.relates_to'];
            if (rel?.event_id === p.sentId && String(rel.key ?? '').includes('✅')) {
              p.ackedAt = Date.now();
            }
            p.events.push({ id: ev.event_id, type: 'reaction', body: '' });
            continue;
          }
          if (ev.type !== 'm.room.message' || ev.sender === prov.userMxId) continue;
          p.events.push({
            id: ev.event_id,
            type: ev.content?.msgtype || 'm.text',
            body: String(ev.content?.body ?? ''),
            file: ev.content?.file,
          });
        }
      }),
    );

    const done = peers.filter((p) => p.ackedAt).length;
    if (Date.now() - lastReport > 15000) {
      lastReport = Date.now();
      const elapsed = Math.round((Date.now() - t0) / 1000);
      console.log(`  [${elapsed}s] 完成 ${done}/${peers.filter((p) => p.joined).length}`);
    }
    if (done === peers.filter((p) => p.joined).length) break;
  }

  // ── 5. 结果分析 ──────────────────────────────────────────────────────
  const bound = peers.filter((p) => p.joined);
  const rows = bound.map((p) => {
    const bodies = p.events.filter((e) => e.body).map((e) => e.body).join('\n');
    /* 串台检测：这个房间的回复里出现了**别的房间**的 token */
    const others = bound.filter((q) => q !== p).map((q) => q.token);
    const leakedIn = others.filter((t) => bodies.includes(t));
    const hasOwn = bodies.includes(p.token);
    const urls = [...bodies.matchAll(/https:\/\/s3\.fmode\.cn\/[^\s)`]+/g)].map((m) => m[0]);
    const files = p.events.filter((e) => e.file).map((e) => e.file?.url).filter(Boolean);
    const allUrls = [...new Set([...urls, ...files])];
    /*
     * 🔴 链接前缀判定（2026-11-14 新增）。
     *
     * App 侧 `s3-link.ts:290` 要求 `/deliverables/` 前缀，否则不认成果。
     * 上一版压测容器走 `uploader.mjs put`，给出 `/user/<uid>/report/...`，
     * 测出来"成功"但 App 根本看不到 —— 这是假测试。
     */
    const deliverableUrls = allUrls.filter((u) => u.includes('/deliverables/'));
    const wrongPrefixUrls = allUrls.filter((u) => !u.includes('/deliverables/'));
    return {
      i: p.i,
      room: p.room,
      token: p.token,
      ok: !!p.ackedAt,
      ms: p.ackedAt ? p.ackedAt - p.sentAt : null,
      hasOwn,
      leakedIn,
      urls: allUrls,
      deliverableUrls,
      wrongPrefixUrls,
      replyLen: bodies.length,
      expectBytes: p.expectBytes || 0,
    };
  });

  // ── 5b. S3 三重验证（并发 HEAD，最多 30 秒）─────────────────────────
  console.log('\n=== [4b] 验证 s3 链接 ──');
  await Promise.all(
    rows.map(async (r) => {
      r.urlChecks = await Promise.all(
        r.urls.map(async (u) => {
          try {
            const ac = new AbortController();
            const to = setTimeout(() => ac.abort(), 15000);
            const resp = await fetch(u, { method: 'GET', signal: ac.signal });
            clearTimeout(to);
            const text = resp.ok ? await resp.text() : '';
            return {
              url: u,
              status: resp.status,
              ok: resp.ok,
              bytes: text.length,
              /* 内容里有没有本房的 token —— 验证「文件内容没错乱/没串台」 */
              hasOwnToken: text.includes(r.token),
              hasOtherToken: rows.some((q) => q.i !== r.i && text.includes(q.token)),
            };
          } catch (e) {
            return { url: u, status: 0, ok: false, error: String(e?.message || e) };
          }
        }),
      );
      r.contentOk = r.urlChecks.some((c) => c.ok && c.hasOwnToken && !c.hasOtherToken);
      r.sizeExpectOk = r.urlChecks.some((c) => c.ok && c.bytes >= r.expectBytes * 0.9);
    }),
  );

  const doneN = rows.filter((r) => r.ok).length;
  const crossTalk = rows.filter((r) => r.leakedIn.length > 0);
  const noOwn = rows.filter((r) => r.ok && !r.hasOwn);
  const withDeliverable = rows.filter((r) => r.deliverableUrls.length > 0);
  const wrongPrefix = rows.filter((r) => r.urls.length > 0 && r.deliverableUrls.length === 0);
  const contentVerified = rows.filter((r) => r.contentOk);

  console.log('\n=== [5] 结果 ===');
  console.log(`完整性     : ${doneN}/${bound.length}`);
  console.log(`串台（消息）: ${crossTalk.length} 个房间收到别人的 token`);
  console.log(`有回复但无本房 token : ${noOwn.length}`);
  console.log(`🔗 /deliverables/ 前缀 : ${withDeliverable.length}/${rows.filter((r) => r.urls.length).length}`);
  console.log(`⚠️ 链接前缀错误 : ${wrongPrefix.length}`);
  console.log(`📄 内容验证通过 : ${contentVerified.length}/${rows.filter((r) => r.urls.length).length}`);
  const times = rows.filter((r) => r.ms).map((r) => r.ms);
  if (times.length) {
    times.sort((a, b) => a - b);
    const p = (q) => times[Math.min(times.length - 1, Math.floor(times.length * q))];
    console.log(`耗时(秒)   : min=${(times[0] / 1000).toFixed(1)} p50=${(p(0.5) / 1000).toFixed(1)} p95=${(p(0.95) / 1000).toFixed(1)} max=${(times[times.length - 1] / 1000).toFixed(1)}`);
  }

  for (const r of rows) {
    const chk = r.urlChecks?.[0];
    const flags = [
      r.ok ? `✅ ${(r.ms / 1000).toFixed(1)}s` : '❌ 超时无 ✅',
      r.hasOwn ? '' : '⚠️回复无本房token',
      r.leakedIn.length ? `🔴串台:${r.leakedIn.join(',')}` : '',
      r.deliverableUrls.length ? `🔗deliverables` : r.urls.length ? '⚠️前缀非deliverables' : '❌无链接',
      chk ? (chk.ok ? `S3 ${chk.status}/${chk.bytes}B` : `S3 ${chk.status}`) : 'S3 未测',
      r.contentOk ? '内容✓' : r.urls.length ? '内容✗' : '',
    ]
      .filter(Boolean)
      .join(' ');
    console.log(`  [${String(r.i).padStart(2)}] ${flags}`);
  }

  // ── 6. 落盘 ──────────────────────────────────────────────────────────
  const lines = [
    `# 压测结果 — ${AGENT} — 并发 ${N}`,
    ``,
    `- 时间：${new Date().toISOString()}`,
    `- 用户：${prov.userMxId}`,
    `- 容器：${agentInfo.agentMxId}`,
    `- 超时：${TIMEOUT_S}s`,
    ``,
    `## 汇总`,
    ``,
    `| 指标 | 值 |`,
    `|---|---|`,
    `| 建房间 | ${N} |`,
    `| 容器进房 | ${bound.length} |`,
    `| 发出 | ${peers.filter((p) => p.sentId).length} |`,
    `| **完成** | **${doneN}** |`,
    `| 串台房间数 | ${crossTalk.length} |`,
    `| 有回复但无本房 token | ${noOwn.length} |`,
    `| **/deliverables/ 前缀正确** | **${withDeliverable.length}** |`,
    `| 链接前缀错误（App 不认） | ${wrongPrefix.length} |`,
    `| **S3 内容验证通过** | **${contentVerified.length}** |`,
    times.length ? `| 耗时 p50 | ${(times[Math.floor(times.length * 0.5)] / 1000).toFixed(1)}s |` : '',
    times.length ? `| 耗时 max | ${(times[times.length - 1] / 1000).toFixed(1)}s |` : '',
    ``,
    `## 逐房间`,
    ``,
    `| # | 房间 | token | 完成 | 耗时 | 本房token | 串台 | deliverables | S3 | 内容 |`,
    `|---|---|---|---|---|---|---|---|---|---|`,
    ...rows.map(
      (r) =>
        `| ${r.i} | \`${r.room}\` | ${r.token} | ${r.ok ? '✅' : '❌'} | ${r.ms ? (r.ms / 1000).toFixed(1) + 's' : '—'} | ${r.hasOwn ? '✅' : '❌'} | ${r.leakedIn.length ? '🔴 ' + r.leakedIn.join(',') : '—'} | ${r.deliverableUrls.length ? '✅' : r.urls.length ? '⚠️' : '❌'} | ${r.urlChecks?.[0]?.ok ? `${r.urlChecks[0].status}/${r.urlChecks[0].bytes}B` : '—'} | ${r.contentOk ? '✅' : '❌'} |`,
    ),
    ``,
    `## 串台详情`,
    ``,
    crossTalk.length
      ? crossTalk.map((r) => `- 房间 ${r.i} 收到了：${r.leakedIn.join(', ')}`).join('\n')
      : '无串台。',
    ``,
    `## 链接前缀问题`,
    ``,
    wrongPrefix.length
      ? wrongPrefix
          .map(
            (r) => `- 房间 ${r.i}：\n${r.urls.map((u) => `  - \`${u}\``).join('\n')}\n  （不是 \`/deliverables/\`，App 侧 \`isAgentOutputUrl()\` 不认）`,
          )
          .join('\n')
      : '全部链接都是 `/deliverables/` 前缀。',
    ``,
    `## 失败明细`,
    ``,
    rows.filter((r) => !r.ok).length
      ? rows.filter((r) => !r.ok).map((r) => `- 房间 ${r.i}（${r.room}）：${r.replyLen} 字符回复，未收到 ✅`).join('\n')
      : '无失败。',
    ``,
  ];
  mkdirSync(dirname(OUT), { recursive: true });
  writeFileSync(OUT, lines.join('\n'));
  console.log(`\n结果已写入 ${OUT}`);

  process.exit(doneN === bound.length && crossTalk.length === 0 ? 0 : 1);
}

main().catch((e) => {
  console.error('\n❌ 压测失败：', e instanceof Error ? e.message : e);
  process.exit(2);
});