QIWEI-skill/scripts/agent-runtime-reliability-smoke-test.js

'use strict';

const assert = require('assert/strict');
const fs = require('fs');
const os = require('os');
const path = require('path');

const { AgentWorkbenchDb } = require('../mcp/src/core/agent-workbench-db');
const { AgentWorkbenchService } = require('../mcp/src/core/agent-workbench-service');
const { QiweiAgentPoller } = require('../mcp/src/dashboard/agent-service').__testing;
const { GroupAgentService } = require('../mcp/src/dashboard/group-agent-service');
const { extractExplicitCustomerIntelligence } = require('../mcp/src/core/agent-runtime');

const results = [];

function fixtureDir(prefix) {
  return fs.mkdtempSync(path.join(os.tmpdir(), `${prefix}-`));
}

function closeFixture({ service, db, dir }) {
  service?.stopBackgroundWorkers?.();
  db?.close?.();
  if (dir) fs.rmSync(dir, { recursive: true, force: true });
}

function qualifiedOutput(content = '已收到，我先按当前信息整理下一步。') {
  return {
    content,
    confidence: 0.96,
    intent: '信息确认',
    reason: '可靠性测试固定输出。',
    requiresHuman: false,
    profileUpdates: {},
    tasks: [],
    alerts: [],
    citations: [],
    toolTrace: [],
  };
}

async function check(name, fn) {
  await fn();
  results.push({ name, status: 'passed' });
}

async function testBatchIsolationAndCursorCommit() {
  const dir = fixtureDir('qiwei-poller-isolation');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), { defaultMode: 'review' });
  const processed = [];
  let syncCalls = 0;
  const poller = new QiweiAgentPoller({
    config: { messageRetryAttempts: 1, messageRetryBaseMs: 0, intervalMs: 0 },
    db,
    qiwei: {
      async syncMessages() {
        syncCalls += 1;
        return syncCalls === 1
          ? { syncMsgList: [{ seq: 1 }, { seq: 2 }, { seq: 3 }], travelSyncKey: 3 }
          : { syncMsgList: [], travelSyncKey: 3 };
      },
    },
    service: {},
  });
  poller.process = async message => {
    processed.push(Number(message.seq));
    if (Number(message.seq) === 2) throw new Error('fixture message failure');
    return { status: 'processed' };
  };
  poller.running = true;
  poller.waitInterval = async () => { poller.running = false; };
  try {
    await poller.loop(0);
    assert.deepEqual(processed, [1, 2, 3], '同一批次后续消息必须继续处理');
    assert.equal(Number(db.getPollState('sync_key', '0')), 3, '批次完成后游标必须推进');
    assert.equal(poller.metrics.failed, 1);
    assert.equal(poller.metrics.lastBatchFailed, 1);
    assert.equal(db.listAudit(100).some(item => item.action === 'poller_message_error'), true);
  } finally {
    closeFixture({ db, dir });
  }
}

async function testStalledCursorDoesNotSpin() {
  const dir = fixtureDir('qiwei-poller-stall');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), { defaultMode: 'review' });
  let syncCalls = 0;
  let waits = 0;
  const poller = new QiweiAgentPoller({
    config: { messageRetryAttempts: 1, messageRetryBaseMs: 0, intervalMs: 0 },
    db,
    qiwei: {
      async syncMessages() {
        syncCalls += 1;
        return { syncMsgList: [{ seq: 4 }], travelSyncKey: 4 };
      },
    },
    service: {},
  });
  poller.process = async () => ({ status: 'ignored_not_allowlisted' });
  poller.running = true;
  poller.waitInterval = async () => {
    waits += 1;
    poller.running = false;
  };
  try {
    await poller.loop(4);
    assert.equal(syncCalls, 1, '停滞游标必须经过等待，不得在同一轮忙循环');
    assert.equal(waits, 1);
    assert.equal(poller.metrics.cursorStalls, 1);
    assert.equal(db.listAudit(100).some(item => item.action === 'poller_cursor_stalled'), true);
    assert.equal(Number(db.getPollState('sync_key', '0')), 4, '停滞批次不得伪造游标');
  } finally {
    closeFixture({ db, dir });
  }
}

async function testPollerGenerationRetry() {
  const dir = fixtureDir('qiwei-poller-retry');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), { defaultMode: 'review' });
  let processCalls = 0;
  let draftCalls = 0;
  const poller = new QiweiAgentPoller({
    config: { messageRetryAttempts: 2, messageRetryBaseMs: 0, intervalMs: 0 },
    db,
    qiwei: {},
    service: {
      async generateLatestDraft() {
        draftCalls += 1;
        return { status: 'pending_review' };
      },
    },
  });
  poller.process = async () => {
    processCalls += 1;
    return {
      status: 'agent_failed',
      errorCode: 'network_error',
      error: 'network timeout',
      conversation: { id: 'conversation-retry' },
      message: { id: 'message-retry' },
    };
  };
  try {
    const result = await poller.processWithRetry({ seq: 11, msgServerId: 'external-retry' });
    assert.equal(result.status, 'pending_review');
    assert.equal(processCalls, 1, '已入库消息重试不得重新回放 wire message');
    assert.equal(draftCalls, 1);
    assert.equal(db.listAudit(100).some(item => item.action === 'poller_message_retry_scheduled'), true);
  } finally {
    closeFixture({ db, dir });
  }
}

async function testPollerAutopilotSendRetry() {
  const dir = fixtureDir('qiwei-poller-send-retry');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), { defaultMode: 'autopilot' });
  let processCalls = 0;
  let draftCalls = 0;
  const poller = new QiweiAgentPoller({
    config: { messageRetryAttempts: 2, messageRetryBaseMs: 0, intervalMs: 0 },
    db,
    qiwei: {},
    service: {
      async generateLatestDraft(conversationId) {
        draftCalls += 1;
        return {
          status: 'autopilot_sent',
          conversation: { id: conversationId },
          message: { id: 'message-send-retry' },
          reply: { id: 'outbound-send-retry' },
        };
      },
    },
  });
  poller.process = async () => {
    processCalls += 1;
    return {
      status: 'autopilot_send_failed',
      errorCode: 'upstream_send_not_acknowledged',
      error: '企微上游暂时未确认发送结果',
      conversation: { id: 'conversation-send-retry' },
      message: { id: 'message-send-retry' },
    };
  };
  try {
    const result = await poller.processWithRetry({ seq: 12, msgServerId: 'external-send-retry' });
    assert.equal(result.status, 'autopilot_sent');
    assert.equal(processCalls, 1, '发送失败重试不得重新回放原始企微消息');
    assert.equal(draftCalls, 1, '发送失败应只重新生成最新草稿一次');
    assert.equal(db.listAudit(100).some(item => item.action === 'poller_message_retry_scheduled'), true);
  } finally {
    closeFixture({ db, dir });
  }
}

async function waitFor(predicate, message) {
  for (let attempt = 0; attempt < 80; attempt += 1) {
    if (predicate()) return;
    await new Promise(resolve => setTimeout(resolve, 5));
  }
  throw new Error(message);
}

function groupFixture(dir, options = {}) {
  const messages = [];
  const roomId = 'reliability-group';
  const groups = { [roomId]: { roomName: '可靠性测试群' } };
  const statePath = path.join(dir, 'group-agent-replies.json');
  const runtime = {
    db: { globalState: () => ({ paused: false }) },
    config: { qiwei: { groupSendRetryAttempts: options.sendAttempts ?? 2, groupSendRetryBaseMs: 0 } },
    agent: options.agent,
    qiwei: options.qiwei,
  };
  const service = new GroupAgentService({
    projectRoot: dir,
    statePath,
    getRuntime: () => runtime,
    loadGroups: () => groups,
    loadMessages: () => messages,
    appendMessage: (_id, message) => messages.push(message),
  });
  const inbound = (id, content) => ({
    fromRoomId: roomId,
    msgUniqueIdentifier: id,
    msgType: 1,
    seq: messages.length + 1,
    senderId: 'group-customer',
    senderName: '群客户',
    content,
    timestamp: Date.now(),
  });
  return { service, roomId, inbound };
}

async function testGroupDeferredGenerationQueuesEveryInbound() {
  const dir = fixtureDir('qiwei-group-deferred');
  let agentCalls = 0;
  let releaseFirst;
  const firstGate = new Promise(resolve => { releaseFirst = resolve; });
  const { service, inbound } = groupFixture(dir, {
    agent: {
      async run() {
        agentCalls += 1;
        if (agentCalls === 1) await firstGate;
        return qualifiedOutput(`群聊草稿 ${agentCalls}`);
      },
    },
    qiwei: { async sendText() { return { isSendSuccess: true }; } },
  });
  try {
    const first = await service.ingestPolledMessage(inbound('group-deferred-1', '第一条群消息'), {}, { deferGeneration: true });
    const second = await service.ingestPolledMessage(inbound('group-deferred-2', '第二条群消息'), {}, { deferGeneration: true });
    assert.equal(first.status, 'group_generation_queued');
    assert.equal(second.status, 'group_generation_queued');
    await waitFor(() => agentCalls === 1, '首条群消息必须开始后台生成');
    releaseFirst();
    await waitFor(() => agentCalls === 2, '生成期间到达的第二条群消息必须顺序进入生成队列');
    await waitFor(() => Boolean(service.list()[0].pendingReply), '群聊生成完成后必须留下待审核草稿');
    const pending = service.list()[0].pendingReply;
    assert.equal(pending.sourceMessageId, 'group-deferred-2', '最新已处理群消息必须获得独立草稿');
  } finally {
    fs.rmSync(dir, { recursive: true, force: true });
  }
}

async function testGroupAutoSendRetriesWithoutFalseSent() {
  const dir = fixtureDir('qiwei-group-send-retry');
  let sendCalls = 0;
  const { service, roomId, inbound } = groupFixture(dir, {
    sendAttempts: 2,
    agent: { async run() { return qualifiedOutput('群聊自动回复'); } },
    qiwei: {
      async sendText() {
        sendCalls += 1;
        return sendCalls === 1 ? { isSendSuccess: false } : { isSendSuccess: true };
      },
    },
  });
  try {
    service.setMode(roomId, 'auto');
    const result = await service.ingestPolledMessage(inbound('group-send-retry-1', '请介绍一下服务'));
    assert.equal(result.status, 'auto_sent');
    assert.equal(sendCalls, 2, '上游首次未确认时必须有限重试');
    const group = service.list()[0];
    assert.equal(group.pendingReply, null, '确认发送后不得保留 pending 草稿');
    assert.equal(group.audit.some(item => item.action === 'group_send_retry_scheduled'), true);
    assert.equal(group.audit.some(item => item.action === 'group_message_auto_sent'), true);
  } finally {
    fs.rmSync(dir, { recursive: true, force: true });
  }
}

async function testGroupAutoSendFailureKeepsPendingDraft() {
  const dir = fixtureDir('qiwei-group-send-failed');
  let sendCalls = 0;
  const { service, roomId, inbound } = groupFixture(dir, {
    sendAttempts: 1,
    agent: { async run() { return qualifiedOutput('仍需人工确认的群聊回复'); } },
    qiwei: { async sendText() { sendCalls += 1; return { isSendSuccess: false }; } },
  });
  try {
    service.setMode(roomId, 'auto');
    const result = await service.ingestPolledMessage(inbound('group-send-failed-1', '我需要帮助'));
    assert.equal(result.status, 'pending_review');
    assert.equal(sendCalls, 1);
    const group = service.list()[0];
    assert.equal(group.pendingReply?.status, 'pending', '未确认投递的群消息不能伪造为 sent');
    assert.equal(group.audit.some(item => item.action === 'group_auto_send_failed'), true);
    assert.equal(group.audit.some(item => item.action === 'group_message_auto_sent'), false);
  } finally {
    fs.rmSync(dir, { recursive: true, force: true });
  }
}

async function testGenerationRetryAndRealtimeIntelligence() {
  const dir = fixtureDir('qiwei-runtime-reliability');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), {
    defaultMode: 'review',
    autoSendConfidence: 0.88,
  });
  let agentCalls = 0;
  let releaseAgent;
  const gate = new Promise(resolve => { releaseAgent = resolve; });
  const service = new AgentWorkbenchService({
    db,
    agent: {
      modelClient: { isConfigured: () => true },
      async run() {
        agentCalls += 1;
        if (agentCalls === 1) return { content: '' };
        await gate;
        return qualifiedOutput();
      },
    },
    qiwei: {
      isConfigured: () => true,
      async sendText() { return { isSendSuccess: true }; },
    },
    config: {
      accountKey: 'reliability-fixture',
      agent: {
        provider: 'fixture',
        model: 'fixture',
        apiKey: 'fixture',
        generationRetryAttempts: 2,
        generationRetryBaseMs: 0,
        qualityPassScore: 60,
      },
      qiwei: { allowedSenders: ['contact-realtime'] },
      memory: { enabled: false, extractionIntervalMs: 60000 },
    },
  });
  try {
    const inbound = service.ingestInbound({
      externalId: 'realtime-intelligence-1',
      contactId: 'contact-realtime',
      contactName: '实时画像测试',
      content: '预算 300 万，明年置换',
    });
    // The model is deliberately held. Rule-based intelligence must already
    // be queryable while generation is waiting.
    await new Promise(resolve => setImmediate(resolve));
    const conversation = db.getConversationByContactId('contact-realtime');
    const profile = db.getProfile(conversation.id).profile;
    assert.equal(profile.budgetWan, 300);
    assert.equal(profile.purpose, '置换');
    assert.equal(profile.timeline, '明年');
    assert.equal(db.listCustomerAlerts(conversation.id).some(item => item.business_key === 'high_intent:core_demand_ready'), true);
    assert.equal(db.listAudit(100, conversation.id).some(item => item.action === 'customer_intelligence_realtime_updated'), true);

    releaseAgent();
    const result = await inbound;
    assert.equal(result.status, 'pending_review');
    assert.equal(agentCalls, 2, '空回复应触发一次模型重试');
    assert.equal(db.listAudit(100, conversation.id).some(item => item.action === 'agent_generation_retry_scheduled'), true);
  } finally {
    closeFixture({ service, db, dir });
  }
}

async function testGenerationFallbackAutoReply() {
  const dir = fixtureDir('qiwei-generation-fallback');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), {
    defaultMode: 'autopilot',
    autoSendConfidence: 0.88,
  });
  let agentCalls = 0;
  let sendCalls = 0;
  const service = new AgentWorkbenchService({
    db,
    agent: {
      modelClient: { isConfigured: () => true },
      async run() {
        agentCalls += 1;
        throw new Error('Claude Code 处理超时');
      },
    },
    qiwei: {
      isConfigured: () => true,
      async sendText() {
        sendCalls += 1;
        return { isSendSuccess: true };
      },
    },
    config: {
      accountKey: 'fallback-fixture',
      agent: {
        provider: 'fixture',
        model: 'fixture',
        apiKey: 'fixture',
        generationRetryAttempts: 1,
        generationRetryBaseMs: 0,
        qualityPassScore: 60,
      },
      qiwei: { allowedSenders: ['contact-fallback'] },
      memory: { enabled: false, extractionIntervalMs: 60000 },
    },
  });
  try {
    const greeting = await service.ingestInbound({
      externalId: 'fallback-greeting-1',
      contactId: 'contact-fallback',
      contactName: '自动回复兜底测试',
      content: '你好',
    });
    assert.equal(greeting.status, 'autopilot_sent');
    assert.equal(agentCalls, 0, '简单问候应绕过易超时的模型调用');
    assert.equal(sendCalls, 1, '简单问候必须实际发送一次');
    assert.equal(db.listAudit(100).some(item => item.action === 'agent_generation_shortcut'), true);
    assert.equal(db.listAudit(100).some(item => item.action === 'autopilot_message_sent'), true);

    const general = await service.ingestInbound({
      externalId: 'fallback-general-1',
      contactId: 'contact-fallback',
      contactName: '自动回复兜底测试',
      content: '我想了解一下服务',
    });
    assert.equal(general.status, 'autopilot_sent');
    assert.equal(agentCalls, 1, '普通消息仍先尝试模型');
    assert.equal(sendCalls, 2, '模型超时后应发送即时确认');
    assert.equal(db.listAudit(100).some(item => item.action === 'agent_generation_fallback_created'), true);
  } finally {
    closeFixture({ service, db, dir });
  }
}

async function testRestartRecoversAcknowledgedGeneration() {
  const dir = fixtureDir('qiwei-generation-recovery');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), {
    defaultMode: 'autopilot',
    autoSendConfidence: 0.88,
  });
  const config = {
    accountKey: 'generation-recovery-fixture',
    agent: {
      provider: 'fixture',
      model: 'fixture',
      apiKey: 'fixture',
      generationRetryAttempts: 1,
      generationRetryBaseMs: 0,
      generationRecovery: { pendingAgeMs: 5000, retryCooldownMs: 5000, maxAttempts: 2 },
      qualityPassScore: 60,
    },
    qiwei: { allowedSenders: ['contact-recovery'] },
    memory: { enabled: false },
  };
  const initial = new AgentWorkbenchService({
    db,
    agent: { modelClient: { isConfigured: () => true }, async run() { return new Promise(() => {}); } },
    qiwei: { isConfigured: () => true, async sendText() { return { isSendSuccess: true }; } },
    config,
  });
  let recovered;
  try {
    const queued = await initial.ingestInbound({
      externalId: 'generation-recovery-1',
      contactId: 'contact-recovery',
      contactName: '恢复测试客户',
      content: '我想了解一下服务',
    }, { awaitGeneration: false });
    assert.equal(queued.status, 'generation_queued');
    const conversation = db.getConversationByContactId('contact-recovery');
    assert.equal(db.getMessage(queued.message.id).status, 'generation_pending');
    db.db.prepare("UPDATE messages SET created_at=? WHERE id=?").run(new Date(Date.now() - 6000).toISOString(), queued.message.id);

    let sendCalls = 0;
    recovered = new AgentWorkbenchService({
      db,
      agent: { modelClient: { isConfigured: () => true }, async run() { throw new Error('network timeout'); } },
      qiwei: {
        isConfigured: () => true,
        async sendText() { sendCalls += 1; return { isSendSuccess: true }; },
      },
      config,
    });
    const result = await recovered.recoverPendingGenerations('fixture:restart');
    assert.equal(result.recovered, 1, '已 ACK 但未生成的入站消息必须在重启后恢复');
    assert.equal(sendCalls, 1, `恢复任务在全自动模式下必须完成实际投递：${JSON.stringify(result)}`);
    assert.equal(db.getMessage(queued.message.id).status, 'reply_sent');
    assert.equal(db.listAudit(100, conversation.id).some(item => item.action === 'agent_generation_recovery_completed'), true);
  } finally {
    initial.stopBackgroundWorkers();
    recovered?.stopBackgroundWorkers();
    db.close();
    fs.rmSync(dir, { recursive: true, force: true });
  }
}

async function testRiskFallbackStaysHumanReview() {
  const dir = fixtureDir('qiwei-generation-fallback-risk');
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), { defaultMode: 'autopilot' });
  let sendCalls = 0;
  const service = new AgentWorkbenchService({
    db,
    agent: {
      modelClient: { isConfigured: () => true },
      async run() { throw new Error('network timeout'); },
    },
    qiwei: {
      isConfigured: () => true,
      async sendText() { sendCalls += 1; return { isSendSuccess: true }; },
    },
    config: {
      accountKey: 'fallback-risk-fixture',
      agent: { provider: 'fixture', model: 'fixture', apiKey: 'fixture', generationRetryAttempts: 1, generationRetryBaseMs: 0, qualityPassScore: 60 },
      qiwei: { allowedSenders: ['contact-risk'] },
      memory: { enabled: false, extractionIntervalMs: 60000 },
    },
  });
  try {
    const result = await service.ingestInbound({
      externalId: 'fallback-risk-1',
      contactId: 'contact-risk',
      contactName: '风险兜底测试',
      content: '我要投诉这个问题',
    });
    assert.equal(result.status, 'pending_review');
    assert.equal(sendCalls, 0, '风险消息的兜底只能进入人工审核');
    assert.equal(Boolean(result.draft.requires_human), true);
    assert.equal(db.listAudit(100).some(item => item.action === 'agent_generation_fallback_created'), true);
  } finally {
    closeFixture({ service, db, dir });
  }
}

async function main() {
  await check('批次内单条失败不阻断后续消息并提交游标', testBatchIsolationAndCursorCommit);
  await check('停滞游标经过等待而不是忙循环', testStalledCursorDoesNotSpin);
  await check('瞬时 Agent 失败走最新草稿有限重试', testPollerGenerationRetry);
  await check('自动回复发送未确认时走最新草稿有限重试', testPollerAutopilotSendRetry);
  await check('群聊回调/轮询后台生成时不丢失同群后续消息', testGroupDeferredGenerationQueuesEveryInbound);
  await check('群聊自动回复在上游未确认时有限重试后再标记已发送', testGroupAutoSendRetriesWithoutFalseSent);
  await check('群聊自动回复失败时保留待审核草稿且不伪造已发送', testGroupAutoSendFailureKeepsPendingDraft);
  await check('模型重试与实时画像/需求/预警在生成期间可见', testGenerationRetryAndRealtimeIntelligence);
  await check('模型超时后问候与普通消息均有自动回复兜底', testGenerationFallbackAutoReply);
  await check('回调已确认但生成中断的消息会在重启后恢复投递', testRestartRecoversAcknowledgedGeneration);
  await check('风险消息的模型超时兜底保持人工审核', testRiskFallbackStaysHumanReview);
  process.stdout.write(`${JSON.stringify({ status: 'passed', results }, null, 2)}\n`);
}

main().catch(error => {
  process.stderr.write(`${JSON.stringify({ status: 'failed', message: error.message, stack: error.stack }, null, 2)}\n`);
  process.exitCode = 1;
});