QIWEI-skill/scripts/session-reply-delivery-regression-smoke-test.js

'use strict';

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

const RUN_ROOT = fs.mkdtempSync(path.join(os.tmpdir(), 'qiwei-session-delivery-'));
process.env.QIWEI_MESSAGES_DIR = path.join(RUN_ROOT, 'messages');
process.env.QIWEI_MESSAGE_ARCHIVE_ENABLED = '1';

const { AgentWorkbenchDb } = require('../mcp/src/core/agent-workbench-db');
const { AgentWorkbenchService } = require('../mcp/src/core/agent-workbench-service');
const { appendChatRecord, drainQueues } = require('../mcp/src/core/message-archive');

const results = [];

function check(name, fn) {
  return Promise.resolve().then(fn).then(() => {
    results.push({ name, status: 'passed' });
  });
}

function findFiles(root) {
  if (!fs.existsSync(root)) return [];
  const files = [];
  for (const entry of fs.readdirSync(root, { withFileTypes: true })) {
    const target = path.join(root, entry.name);
    if (entry.isDirectory()) files.push(...findFiles(target));
    else files.push(target);
  }
  return files;
}

function qualifiedOutput(content) {
  return {
    content,
    confidence: 0.96,
    intent: 'session_delivery_regression',
    reason: 'Deterministic delivery fixture.',
    requiresHuman: false,
    qualityPassed: true,
    qualityScore: 96,
    qualityChecks: [],
    citations: [],
    toolTrace: [],
  };
}

async function testLocalOutboundMergesRemoteEcho() {
  const dbPath = path.join(RUN_ROOT, 'merge', 'workbench.db');
  fs.mkdirSync(path.dirname(dbPath), { recursive: true });
  const db = new AgentWorkbenchDb(dbPath, { globalPaused: false, defaultMode: 'autopilot' });
  try {
    const conversation = db.ensureConversation('contact-merge', 'Fixture');
    const createdAt = '2026-08-18T10:00:00.000Z';
    const local = db.insertMessage({
      conversationId: conversation.id,
      direction: 'outbound',
      senderType: 'agent',
      content: '这条方案我先给您记下，您看周六上午方便吗？',
      status: 'sent',
      createdAt,
      raw: { source: 'autopilot', inboundMessageId: 'inbound-1' },
    }).message;
    const match = db.findRecentOutboundMatch(conversation.id, local.content, '2026-08-18T10:00:12.000Z', 300);
    assert.equal(match.id, local.id);

    const merged = db.attachExternalId(local.id, 'remote-outbound-1', { source: 'manual_sync' });
    assert.equal(merged.external_id, 'remote-outbound-1');
    const replay = db.insertMessage({
      conversationId: conversation.id,
      externalId: 'remote-outbound-1',
      direction: 'outbound',
      senderType: 'human',
      content: local.content,
      status: 'sent',
      createdAt: '2026-08-18T10:00:12.000Z',
      raw: { source: 'manual_sync' },
    });
    assert.equal(replay.created, false);
    assert.equal(db.listMessages(conversation.id, 20).filter(item => item.direction === 'outbound').length, 1);
  } finally {
    db.close();
  }
}

async function testConcurrentAutopilotSendsOnce() {
  const dbPath = path.join(RUN_ROOT, 'concurrency', 'workbench.db');
  fs.mkdirSync(path.dirname(dbPath), { recursive: true });
  const db = new AgentWorkbenchDb(dbPath, { globalPaused: false, defaultMode: 'autopilot' });
  const sent = [];
  const service = new AgentWorkbenchService({
    db,
    agent: { modelClient: { isConfigured: () => true } },
    qiwei: {
      isConfigured: () => true,
      async sendText(toId, content) {
        sent.push({ toId, content });
        return { isSendSuccess: true, msgServerId: 'remote-concurrent-1' };
      },
    },
    config: {
      accountKey: 'session-delivery-regression',
      agent: { qualityPassScore: 82 },
      qiwei: { allowedSenders: ['contact-concurrent'] },
      memory: { enabled: false },
    },
  });
  try {
    const conversation = db.ensureConversation('contact-concurrent', 'Fixture');
    const inbound = db.insertMessage({
      conversationId: conversation.id,
      externalId: 'remote-inbound-1',
      direction: 'inbound',
      senderType: 'customer',
      content: '请把当前最合适的方案发我',
    }).message;
    const output = qualifiedOutput('按当前条件，这个方案先给您参考；您看周六上午方便吗？');
    const [first, second] = await Promise.all([
      service.sendAutopilotReply(conversation, inbound, output),
      service.sendAutopilotReply(conversation, inbound, output),
    ]);
    assert.equal(sent.length, 1);
    assert.equal(first.status, 'autopilot_sent');
    assert.equal(second.status, 'autopilot_already_sent');
    assert.equal(db.listMessages(conversation.id, 20).filter(item => item.direction === 'outbound').length, 1);
    assert.equal(db.listAudit(50, conversation.id).filter(item => item.action === 'autopilot_duplicate_send_suppressed').length, 1);
  } finally {
    if (typeof service.stopBackgroundWorkers === 'function') service.stopBackgroundWorkers();
    db.close();
  }
}

async function testCrossProcessDeliveryClaimSendsOnce() {
  const dbPath = path.join(RUN_ROOT, 'cross-process', 'workbench.db');
  fs.mkdirSync(path.dirname(dbPath), { recursive: true });
  const dbA = new AgentWorkbenchDb(dbPath, { globalPaused: false, defaultMode: 'autopilot' });
  const dbB = new AgentWorkbenchDb(dbPath, { globalPaused: false, defaultMode: 'autopilot' });
  const sent = [];
  const makeService = db => new AgentWorkbenchService({
    db,
    agent: { modelClient: { isConfigured: () => true } },
    qiwei: {
      isConfigured: () => true,
      async sendText(toId, content) {
        sent.push({ toId, content });
        await new Promise(resolve => setTimeout(resolve, 30));
        return { isSendSuccess: true, msgServerId: 'remote-cross-process-1' };
      },
    },
    config: {
      accountKey: 'session-delivery-cross-process',
      agent: { qualityPassScore: 82 },
      qiwei: { allowedSenders: ['contact-cross-process'] },
      memory: { enabled: false },
    },
  });
  const serviceA = makeService(dbA);
  const serviceB = makeService(dbB);
  try {
    const conversation = dbA.ensureConversation('contact-cross-process', 'Fixture');
    const inbound = dbA.insertMessage({
      conversationId: conversation.id,
      externalId: 'remote-inbound-cross-process-1',
      direction: 'inbound',
      senderType: 'customer',
      content: '请把当前最合适的方案发我',
    }).message;
    const output = qualifiedOutput('按当前条件，这个方案先给您参考；您看周六上午方便吗？');
    const [first, second] = await Promise.all([
      serviceA.sendAutopilotReply(conversation, inbound, output),
      serviceB.sendAutopilotReply(conversation, inbound, output),
    ]);
    assert.equal(sent.length, 1);
    assert.deepEqual([first.status, second.status].sort(), ['autopilot_already_sent', 'autopilot_sent']);
    assert.equal(dbA.listMessages(conversation.id, 20).filter(item => item.direction === 'outbound').length, 1);
    assert.equal(dbA.listAudit(50, conversation.id).filter(item => item.action === 'autopilot_delivery_in_flight_suppressed').length, 1);
  } finally {
    if (typeof serviceA.stopBackgroundWorkers === 'function') serviceA.stopBackgroundWorkers();
    if (typeof serviceB.stopBackgroundWorkers === 'function') serviceB.stopBackgroundWorkers();
    dbA.close();
    dbB.close();
  }
}

async function testGlobalPauseSuppressesQueuedAutopilotDelivery() {
  const dbPath = path.join(RUN_ROOT, 'global-pause', 'workbench.db');
  fs.mkdirSync(path.dirname(dbPath), { recursive: true });
  const db = new AgentWorkbenchDb(dbPath, { globalPaused: false, defaultMode: 'autopilot' });
  const sent = [];
  const service = new AgentWorkbenchService({
    db,
    agent: { modelClient: { isConfigured: () => true } },
    qiwei: {
      isConfigured: () => true,
      async sendText(toId, content) {
        sent.push({ toId, content });
        return { isSendSuccess: true, msgServerId: 'remote-global-pause-1' };
      },
    },
    config: {
      accountKey: 'session-delivery-global-pause',
      agent: { qualityPassScore: 82 },
      qiwei: { allowedSenders: ['contact-global-pause'] },
      memory: { enabled: false },
    },
  });
  try {
    const conversation = db.ensureConversation('contact-global-pause', 'Fixture');
    db.setConversationMode(conversation.id, 'autopilot');
    const inbound = db.insertMessage({
      conversationId: conversation.id,
      externalId: 'remote-inbound-global-pause-1',
      direction: 'inbound',
      senderType: 'customer',
      content: '暂停后不要自动发出',
    }).message;
    service.setGlobal({ paused: true }, 'regression');
    const result = await service.sendAutopilotReply(conversation, inbound, qualifiedOutput('这条不应出站'));
    assert.equal(result.status, 'paused');
    assert.equal(sent.length, 0);
    assert.equal(db.listMessages(conversation.id, 20).filter(item => item.direction === 'outbound').length, 0);
    assert.equal(db.listAudit(50, conversation.id).filter(item => item.action === 'autopilot_paused_delivery_suppressed').length, 1);
  } finally {
    if (typeof service.stopBackgroundWorkers === 'function') service.stopBackgroundWorkers();
    db.close();
  }
}

async function testArchiveDeduplicatesSameVisibleMessage() {
  const createdAt = '2026-08-18T10:00:00.000Z';
  const payload = {
    wxid: 'contact-archive',
    messageId: 'local-outbound-archive',
    externalId: 'remote-outbound-archive',
    dir: 'out',
    senderType: 'agent',
    content: '同一条可见消息只应归档一次。',
    createdAt,
  };
  await appendChatRecord(payload);
  await appendChatRecord({ ...payload, messageId: 'remote-outbound-archive', source: 'manual_sync' });
  await drainQueues();
  const files = findFiles(path.join(RUN_ROOT, 'messages'));
  const rows = files.flatMap(file => fs.readFileSync(file, 'utf8').split(/\r?\n/).filter(Boolean).map(line => JSON.parse(line)));
  assert.equal(rows.filter(row => row.content === payload.content && row.dir === 'out').length, 1);
}

async function main() {
  await check('local outbound is merged with the remote echo', testLocalOutboundMergesRemoteEcho);
  await check('concurrent autopilot delivery emits one visible outbound', testConcurrentAutopilotSendsOnce);
  await check('cross-process delivery claim emits one visible outbound', testCrossProcessDeliveryClaimSendsOnce);
  await check('global pause suppresses queued autopilot delivery', testGlobalPauseSuppressesQueuedAutopilotDelivery);
  await check('chat archive suppresses duplicate visible rows', testArchiveDeduplicatesSameVisibleMessage);
  process.stdout.write(`${JSON.stringify({ status: 'ok', checks: results.length, results }, null, 2)}\n`);
}

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