'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; });