QIWEI-skill/scripts/agent-generation-concurrency-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 results = [];

function deferred() {
  let resolve;
  let reject;
  const promise = new Promise((resolvePromise, rejectPromise) => {
    resolve = resolvePromise;
    reject = rejectPromise;
  });
  return { promise, resolve, reject };
}

function qualifiedOutput(content, profileUpdates = {}) {
  return {
    content,
    confidence: 0.96,
    intent: 'next_step',
    reason: 'Deterministic concurrency fixture.',
    requiresHuman: false,
    profileUpdates,
    tasks: [],
    alerts: [],
    citations: [],
    toolTrace: [],
  };
}

function setup({ mode = 'review', agentRun, journeyDefinition = null } = {}) {
  const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'qiwei-generation-concurrency-'));
  const db = new AgentWorkbenchDb(path.join(dir, 'workbench.db'), {
    defaultMode: mode,
    autoSendConfidence: 0.8,
  });
  const sent = [];
  const service = new AgentWorkbenchService({
    db,
    agent: {
      modelClient: { isConfigured: () => true },
      run: agentRun,
    },
    qiwei: {
      isConfigured: () => true,
      async sendText(toId, content) {
        sent.push({ toId, content });
        return { isSendSuccess: true };
      },
    },
    config: {
      accountKey: 'generation-concurrency-smoke',
      agent: {
        apiKey: 'fixture',
        model: 'fixture',
        provider: 'fixture',
        qualityPassScore: 82,
        journeyDefinition,
      },
      qiwei: { allowedSenders: ['contact-1', 'contact-2'] },
      memory: { enabled: false },
    },
  });
  return {
    dir,
    db,
    sent,
    service,
    close() {
      service.stopBackgroundWorkers();
      db.close();
      fs.rmSync(dir, { recursive: true, force: true });
    },
  };
}

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

async function runStaleMode(mode) {
  const firstStarted = deferred();
  const firstReply = deferred();
  let calls = 0;
  let active = 0;
  let maxActive = 0;
  const ctx = setup({
    mode,
    agentRun: async input => {
      calls += 1;
      active += 1;
      maxActive = Math.max(maxActive, active);
      try {
        if (calls === 1) {
          firstStarted.resolve();
          await firstReply.promise;
          return qualifiedOutput('第一条旧回复不应落库；建议忽略该结果。', { staleField: 'must-not-persist' });
        }
        assert.match(input.inboundContent, /第二条/);
        return qualifiedOutput('第二条信息已作为当前依据；建议按最新内容继续推进。', { latestField: 'persisted' });
      } finally {
        active -= 1;
      }
    },
  });
  try {
    const first = ctx.service.ingestInbound({
      externalId: `${mode}-first`,
      contactId: 'contact-1',
      content: '第一条问题，请给建议。',
    });
    await firstStarted.promise;
    const second = ctx.service.ingestInbound({
      externalId: `${mode}-second`,
      contactId: 'contact-1',
      content: '第二条补充，请按最新内容处理。',
    });
    assert.equal(ctx.sent.length, 0);
    firstReply.resolve();

    const firstResult = await first;
    const secondResult = await second;
    assert.equal(firstResult.status, 'stale_discarded');
    assert.equal(secondResult.status, mode === 'review' ? 'pending_review' : mode === 'auto' ? 'sent' : 'autopilot_sent');
    assert.equal(calls, 2);
    assert.equal(maxActive, 1);
    assert.equal(ctx.sent.length, mode === 'review' ? 0 : 1);

    const conversation = ctx.db.getConversationByContactId('contact-1');
    const profile = ctx.db.getProfile(conversation.id).profile;
    assert.equal(profile.staleField, undefined);
    assert.equal(profile.latestField, 'persisted');
    const drafts = ctx.db.listDrafts({ conversationId: conversation.id });
    assert.equal(drafts.some(draft => draft.inbound_message_id === firstResult.message.id), false);
    assert.equal(ctx.db.listConversationAppliedActions(conversation.id).some(action => (
      action.payload?.inboundMessageId === firstResult.message.id
    )), false);
    assert.equal(ctx.db.listAudit(100, conversation.id).filter(item => item.action === 'agent_response_stale_discarded').length, 1);
  } finally {
    ctx.close();
  }
}

async function main() {
  for (const mode of ['review', 'auto', 'autopilot']) {
    await check(`${mode} discards a pending stale generation and serializes the next turn`, () => runStaleMode(mode));
  }

  await check('direct autopilot and automatic approval reject superseded inbound messages', async () => {
    const ctx = setup({ mode: 'autopilot', agentRun: async () => qualifiedOutput('当前回复可执行；建议继续推进。') });
    try {
      const conversation = ctx.db.ensureConversation('contact-1', 'Fixture');
      const oldInbound = ctx.db.insertMessage({
        conversationId: conversation.id,
        direction: 'inbound',
        senderType: 'customer',
        content: '旧问题',
      }).message;
      const staleDraft = ctx.db.createDraft({
        conversationId: conversation.id,
        inboundMessageId: oldInbound.id,
        content: '旧草稿',
        confidence: 0.99,
        requiresHuman: false,
        citations: [],
        toolTrace: [],
      });
      ctx.db.insertMessage({
        conversationId: conversation.id,
        direction: 'inbound',
        senderType: 'customer',
        content: '新问题',
      });
      await assert.rejects(() => ctx.service.sendAutopilotReply(conversation, oldInbound, {
        content: '旧自动回复',
        qualityPassed: true,
        requiresHuman: false,
      }), error => error.code === 'AGENT_RESPONSE_STALE');
      const autoApproval = await ctx.service.approveDraft(staleDraft.id, { actor: 'agent:auto' });
      assert.equal(autoApproval.status, 'stale_discarded');
      assert.equal(autoApproval.draft.status, 'rejected');
      assert.equal(ctx.sent.length, 0);
    } finally {
      ctx.close();
    }
  });

  await check('human approval remains available for an intentionally reviewed older draft', async () => {
    const ctx = setup({ mode: 'review', agentRun: async () => qualifiedOutput('当前回复可执行；建议继续推进。') });
    try {
      const conversation = ctx.db.ensureConversation('contact-1', 'Fixture');
      const oldInbound = ctx.db.insertMessage({ conversationId: conversation.id, direction: 'inbound', senderType: 'customer', content: '旧问题' }).message;
      const draft = ctx.db.createDraft({
        conversationId: conversation.id,
        inboundMessageId: oldInbound.id,
        content: '人工确认后的旧问题答复',
        confidence: 0.1,
        requiresHuman: true,
        citations: [],
        toolTrace: [],
      });
      ctx.db.insertMessage({ conversationId: conversation.id, direction: 'inbound', senderType: 'customer', content: '新问题' });
      const result = await ctx.service.approveDraft(draft.id, { actor: 'human', content: '人工明确选择发送的内容' });
      assert.equal(result.status, 'sent');
      assert.equal(ctx.sent.length, 1);
    } finally {
      ctx.close();
    }
  });

  await check('journey remains planned for a draft and advances only after a real send', async () => {
    const journeyDefinition = {
      id: 'generic-service-flow',
      initialStageId: 'understand',
      stages: [
        { id: 'understand', requiredFacts: ['goal'], nextStageId: 'qualify' },
        { id: 'qualify', requiredFacts: ['budget'], nextStageId: 'act' },
        { id: 'act', requiredFacts: [], nextStageId: null },
      ],
    };
    const ctx = setup({
      mode: 'review',
      journeyDefinition,
      agentRun: async () => qualifiedOutput(
        '可以围绕当前目标继续推进；建议先整理一个可执行方案。',
        { goal: 'improve service', budget: 'confirmed range' },
      ),
    });
    try {
      const generated = await ctx.service.ingestInbound({
        externalId: 'journey-message',
        contactId: 'contact-1',
        content: '目标和投入范围都明确了，请给下一步建议。',
      });
      assert.equal(generated.status, 'pending_review');
      let journey = ctx.db.getConversationJourney(generated.conversation.id);
      assert.equal(journey.currentStageId, 'understand');
      assert.equal(journey.status, 'active');
      assert.equal(journey.appliedActions.length, 1);
      assert.equal(journey.appliedActions[0].status, 'planned');

      await ctx.service.approveDraft(generated.draft.id, { actor: 'human' });
      journey = ctx.db.getConversationJourney(generated.conversation.id);
      assert.equal(journey.currentStageId, 'act');
      assert.equal(journey.status, 'completed');
      assert.equal(journey.appliedActions[0].status, 'confirmed');
      assert.equal(journey.appliedActions[0].confirmationEvidence.type, 'send');
      assert.deepEqual(journey.state.stageHistory.filter(item => item.kind === 'skip').map(item => item.stageId), ['qualify']);
    } finally {
      ctx.close();
    }
  });

  process.stdout.write(`${JSON.stringify({ status: 'ok', checks: results.length, results }, null, 2)}\n`);
}

main().catch(error => {
  console.error(error);
  process.exitCode = 1;
});