1634 lines
57 KiB
TypeScript
1634 lines
57 KiB
TypeScript
import assert from 'node:assert/strict';
|
||
import { EventEmitter } from 'node:events';
|
||
import test from 'node:test';
|
||
import { loadConfig } from '../src/config.js';
|
||
import {
|
||
AgentBusListener,
|
||
type AgentBusSocket,
|
||
type AgentBusTaskGateway,
|
||
createTaskResultFrame,
|
||
createTaskProgressFrame,
|
||
extractAgentBusBusinessText,
|
||
isInboundAgentBusTask,
|
||
parseAgentBusFrame,
|
||
sessionFromReadyFrame,
|
||
taskAttachmentReferences,
|
||
taskResultStatus,
|
||
taskResultText
|
||
} from '../src/agentbus.js';
|
||
import {
|
||
AgentBusManager,
|
||
mergeRuntimeChannelStatuses,
|
||
type PublicAgentBusChannel
|
||
} from '../src/agentbus-channels.js';
|
||
import {
|
||
AGENTBUS_FINAL_REPLY_ELIGIBLE_FIELD,
|
||
AGENTBUS_GENERIC_ACCEPTED_TEXT,
|
||
AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT,
|
||
AGENTBUS_ROSTER_WAITING_TEXT,
|
||
createAgentBusAcceptedDeliveryPayload,
|
||
publicAgentBusDeliveryPayload
|
||
} from '../src/agentbus-delivery.js';
|
||
import { resolveBusinessRoute } from '../src/business-routes.js';
|
||
import type { AgentBusDelivery, PublicTask } from '../src/task-service.js';
|
||
|
||
function makeTask(status: string, overrides: Partial<PublicTask> = {}): PublicTask {
|
||
return {
|
||
organization_id: 'org-1',
|
||
task_id: 'TASK-1',
|
||
source: 'manual',
|
||
channel_id: null,
|
||
channel_name: null,
|
||
raw_text: '测试指令',
|
||
conversation_id: 'conversation-1',
|
||
agent_session: {
|
||
status: 'active',
|
||
turn_no: 1,
|
||
recovery_count: 0,
|
||
session_id_present: false
|
||
},
|
||
parser: {
|
||
route_id: null,
|
||
input_contract_version: null,
|
||
configured_mode: 'ai',
|
||
config_revision: 0,
|
||
authoritative_engine: null,
|
||
engine_affinity: 'ai',
|
||
version: null,
|
||
fallback_reason: null,
|
||
comparison_status: null,
|
||
decision_id: null,
|
||
review_status: null,
|
||
diff_paths: [],
|
||
field_differences: [],
|
||
unreviewed_difference_count: 0,
|
||
shadow_differences: [],
|
||
reparse_count: 0
|
||
},
|
||
input_request: null,
|
||
important_message: null,
|
||
operation: null,
|
||
parse_response: null,
|
||
result: null,
|
||
summary: null,
|
||
status,
|
||
stage: 'parse',
|
||
message: '任务状态已更新。',
|
||
error: '',
|
||
failure: null,
|
||
success_receipt: null,
|
||
error_summary: null,
|
||
handoff_status: '',
|
||
confirmation_mode: 'manual',
|
||
confirmed_at: null,
|
||
created_at: new Date().toISOString(),
|
||
updated_at: new Date().toISOString(),
|
||
last_event_id: null,
|
||
events: [],
|
||
...overrides
|
||
};
|
||
}
|
||
|
||
class FakeSocket {
|
||
readyState = 0;
|
||
readonly sent: Record<string, unknown>[] = [];
|
||
private readonly handlers = new Map<string, Array<(...args: any[]) => void>>();
|
||
|
||
on(event: string, listener: (...args: any[]) => void): this {
|
||
const listeners = this.handlers.get(event) || [];
|
||
listeners.push(listener);
|
||
this.handlers.set(event, listeners);
|
||
return this;
|
||
}
|
||
|
||
emit(event: string, ...args: any[]): void {
|
||
for (const listener of this.handlers.get(event) || []) listener(...args);
|
||
}
|
||
|
||
send(data: string, callback?: (error?: Error) => void): void {
|
||
this.sent.push(JSON.parse(data) as Record<string, unknown>);
|
||
callback?.();
|
||
}
|
||
|
||
close(): void {
|
||
this.readyState = 3;
|
||
this.emit('close', 1000, Buffer.alloc(0));
|
||
}
|
||
}
|
||
|
||
async function waitFor(predicate: () => boolean, timeoutMs = 1_000): Promise<void> {
|
||
const deadline = Date.now() + timeoutMs;
|
||
while (!predicate() && Date.now() < deadline) {
|
||
await new Promise((resolve) => setTimeout(resolve, 5));
|
||
}
|
||
}
|
||
|
||
function testConfig() {
|
||
return loadConfig({
|
||
NODE_ENV: 'test',
|
||
FIELD_ENCRYPTION_KEY: Buffer.alloc(32, 4).toString('base64'),
|
||
AGENTBUS_ENABLED: 'true',
|
||
AGENTBUS_WS_URL: 'wss://mesh.nianxx.cn/ws',
|
||
AGENTBUS_WS_TOKEN: 'test-ws-token',
|
||
AGENTBUS_BOT_ADDRESS: 'bot:test:listener',
|
||
AGENTBUS_WS_RECONNECT_DELAY: '1ms',
|
||
AGENTBUS_TASK_TIMEOUT_MS: '1000',
|
||
AGENTBUS_LOG_PAYLOADS: 'true'
|
||
});
|
||
}
|
||
|
||
function publicChannelFixture(overrides: Partial<PublicAgentBusChannel> = {}): PublicAgentBusChannel {
|
||
return {
|
||
id: 'channel-1',
|
||
organization_id: 'org-1',
|
||
channel_type: 'agentbus',
|
||
display_name: '外部用户 A',
|
||
external_user_ref: 'external-user-a',
|
||
agentbus_bot_address: 'bot:test:listener',
|
||
owner_user_id: 'user-a',
|
||
owner_username: 'employee-a',
|
||
owner_role: 'user',
|
||
owner_erp_account: 'ERP-A',
|
||
routing_ready: true,
|
||
enabled: true,
|
||
status: 'disabled',
|
||
key_configured: true,
|
||
deletable: true,
|
||
last_connected_at: null,
|
||
last_error: null,
|
||
created_at: new Date().toISOString(),
|
||
updated_at: new Date().toISOString(),
|
||
...overrides
|
||
};
|
||
}
|
||
|
||
test('AgentBus configuration stays disabled until connection fields are supplied', () => {
|
||
const config = loadConfig({
|
||
NODE_ENV: 'test',
|
||
FIELD_ENCRYPTION_KEY: Buffer.alloc(32, 5).toString('base64')
|
||
});
|
||
assert.equal(config.agentBusEnabled, false);
|
||
assert.equal(config.AGENTBUS_WS_RECONNECT_DELAY, 5_000);
|
||
assert.equal(config.AGENTBUS_LOG_PAYLOADS, false);
|
||
});
|
||
|
||
test('AgentBus accepted delivery payloads keep final ownership metadata server-only', () => {
|
||
const waitingPayload = createAgentBusAcceptedDeliveryPayload({
|
||
text: AGENTBUS_ROSTER_WAITING_TEXT,
|
||
finalReplyEligible: false
|
||
});
|
||
assert.equal(waitingPayload.text, AGENTBUS_ROSTER_WAITING_TEXT);
|
||
assert.equal(waitingPayload[AGENTBUS_FINAL_REPLY_ELIGIBLE_FIELD], false);
|
||
assert.deepEqual(publicAgentBusDeliveryPayload(waitingPayload), {
|
||
event: 'task.progress',
|
||
status: 'accepted',
|
||
text: AGENTBUS_ROSTER_WAITING_TEXT
|
||
});
|
||
assert.deepEqual(createAgentBusAcceptedDeliveryPayload(), {
|
||
event: 'task.progress',
|
||
status: 'accepted',
|
||
text: AGENTBUS_GENERIC_ACCEPTED_TEXT,
|
||
[AGENTBUS_FINAL_REPLY_ELIGIBLE_FIELD]: true
|
||
});
|
||
});
|
||
|
||
test('AgentBus auto configuration rejects partial credentials', () => {
|
||
assert.throws(() => loadConfig({
|
||
NODE_ENV: 'test',
|
||
FIELD_ENCRYPTION_KEY: Buffer.alloc(32, 6).toString('base64'),
|
||
AGENTBUS_WS_TOKEN: 'partial-token'
|
||
}), /AGENTBUS_WS_URL/);
|
||
});
|
||
|
||
test('database-managed channels may use the global WebSocket URL without a global key', () => {
|
||
const config = loadConfig({
|
||
NODE_ENV: 'test',
|
||
FIELD_ENCRYPTION_KEY: Buffer.alloc(32, 7).toString('base64'),
|
||
AGENTBUS_ENABLED: 'true',
|
||
AGENTBUS_WS_URL: 'wss://mesh.nianxx.cn/ws'
|
||
});
|
||
assert.equal(config.agentBusEnabled, true);
|
||
assert.equal(config.AGENTBUS_WS_TOKEN, undefined);
|
||
});
|
||
|
||
test('WeChat transport envelope exposes only its business text to the global parser', () => {
|
||
const businessText = [
|
||
'独立团单个下单',
|
||
'发团日期：2026-11-02',
|
||
'预订客户：示例客户',
|
||
'产品搜索：示例产品',
|
||
'预估人数：15+1',
|
||
'用房数量：TWN 8，SGL 1'
|
||
].join('\n');
|
||
const wrapped = [
|
||
'New WeChat message',
|
||
'Conversation: conversation-example-1',
|
||
`Text: ${businessText}`
|
||
].join('\n');
|
||
const extracted = extractAgentBusBusinessText(wrapped);
|
||
assert.equal(extracted, businessText);
|
||
assert.equal(resolveBusinessRoute(extracted).routeId, 'team_order_create');
|
||
const envelopeConversation = parseAgentBusFrame(JSON.stringify({
|
||
id: 'wechat-envelope-conversation',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
payload: { text: wrapped }
|
||
}));
|
||
assert.equal(envelopeConversation?.conversation_id, 'conversation-example-1');
|
||
const explicitConversation = parseAgentBusFrame(JSON.stringify({
|
||
id: 'wechat-explicit-conversation',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-explicit-1',
|
||
payload: { text: wrapped }
|
||
}));
|
||
assert.equal(explicitConversation?.conversation_id, 'conversation-explicit-1');
|
||
|
||
const missingConversation = `New WeChat message\nConversation:\nText: ${businessText}`;
|
||
const wrongTextLabel = `New WeChat message\nConversation: conversation-example-1\nBody: ${businessText}`;
|
||
const emptyBody = 'New WeChat message\nConversation: conversation-example-1\nText: ';
|
||
assert.equal(extractAgentBusBusinessText(missingConversation), missingConversation);
|
||
assert.equal(extractAgentBusBusinessText(wrongTextLabel), wrongTextLabel);
|
||
assert.equal(extractAgentBusBusinessText(emptyBody), emptyBody.trim());
|
||
});
|
||
|
||
test('channel directory prefers the current listener over stale persisted disabled status', () => {
|
||
const channel = publicChannelFixture({ status: 'disabled', last_error: '旧进程状态' });
|
||
const connected = mergeRuntimeChannelStatuses([channel], [{
|
||
channel_id: channel.id,
|
||
enabled: true,
|
||
connected: true,
|
||
session_ready: true
|
||
}]);
|
||
assert.equal(connected[0].status, 'connected');
|
||
assert.equal(connected[0].last_error, null);
|
||
|
||
const connecting = mergeRuntimeChannelStatuses([channel], [{
|
||
channel_id: channel.id,
|
||
enabled: true,
|
||
connected: false,
|
||
session_ready: false
|
||
}]);
|
||
assert.equal(connecting[0].status, 'connecting');
|
||
|
||
const disabled = mergeRuntimeChannelStatuses(
|
||
[publicChannelFixture({ enabled: false, status: 'disabled' })],
|
||
[{ channel_id: channel.id, enabled: true, connected: true, session_ready: true }]
|
||
);
|
||
assert.equal(disabled[0].status, 'disabled');
|
||
});
|
||
|
||
test('AgentBus protocol helpers preserve reply routing fields', () => {
|
||
const inbound = parseAgentBusFrame(JSON.stringify({
|
||
id: 'channel-event-1',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-1',
|
||
payload: { text: '你好', reply_policy: { progress: true } }
|
||
}));
|
||
assert.ok(inbound);
|
||
assert.equal(isInboundAgentBusTask(inbound), true);
|
||
|
||
const session = { id: 'session-1', epoch: 3, address: 'bot:test:listener' };
|
||
const ready = sessionFromReadyFrame(parseAgentBusFrame(JSON.stringify({
|
||
id: 'ready-1',
|
||
type: 'event',
|
||
session_id: session.id,
|
||
epoch: session.epoch,
|
||
to: session.address,
|
||
payload: { event: 'session.ready' }
|
||
}))!);
|
||
assert.deepEqual(ready, session);
|
||
|
||
const progress = createTaskProgressFrame(inbound, session, '处理中', 'progress-1');
|
||
assert.equal(progress.reply_to, 'channel-event-1');
|
||
assert.equal(progress.conversation_id, 'conversation-1');
|
||
assert.equal((progress.payload as Record<string, unknown>).event, 'task.progress');
|
||
|
||
const result = createTaskResultFrame(inbound, session, 'completed', '已完成', 'result-1');
|
||
assert.equal(result.from, session.address);
|
||
assert.equal(result.to, inbound.from);
|
||
assert.equal(result.session_id, session.id);
|
||
assert.equal(result.epoch, session.epoch);
|
||
assert.equal(result.reply_to, inbound.id);
|
||
assert.equal((result.payload as Record<string, unknown>).event, 'task.result');
|
||
|
||
const important = {
|
||
kind: 'awaiting_user_input' as const,
|
||
text: '请补充房型分配。',
|
||
recorded_at: new Date().toISOString()
|
||
};
|
||
const resultWithImportantMessage = createTaskResultFrame(
|
||
inbound,
|
||
session,
|
||
'completed',
|
||
important.text,
|
||
'result-important-1',
|
||
important
|
||
);
|
||
assert.deepEqual((resultWithImportantMessage.payload as Record<string, unknown>).important_message, {
|
||
kind: 'awaiting_user_input',
|
||
text: '请补充房型分配。'
|
||
});
|
||
|
||
const detailedSuccess = {
|
||
kind: 'success' as const,
|
||
text: '创建完成。\n团号：LW-260907A-B',
|
||
recorded_at: new Date().toISOString(),
|
||
success_receipt: {
|
||
recorded_at: new Date().toISOString(),
|
||
receipt: {
|
||
group_number: 'LW-260907A-B',
|
||
verification_status: 'split_parent_completed'
|
||
}
|
||
}
|
||
};
|
||
const channelResult = createTaskResultFrame(
|
||
inbound,
|
||
session,
|
||
'completed',
|
||
'创建完成。\n团号：LW-260907A-B',
|
||
'result-channel-summary-1',
|
||
detailedSuccess
|
||
);
|
||
assert.deepEqual((channelResult.payload as Record<string, unknown>).important_message, {
|
||
kind: 'success',
|
||
text: '创建完成。\n团号：LW-260907A-B'
|
||
});
|
||
assert.doesNotMatch(JSON.stringify(channelResult.payload), /success_receipt|verification_status|split_parent_completed/);
|
||
|
||
const directFileResult = createTaskResultFrame(
|
||
inbound,
|
||
session,
|
||
'completed',
|
||
'团队文件已准备',
|
||
'result-file-1',
|
||
detailedSuccess,
|
||
[{
|
||
id: 'artifact-direct-1',
|
||
name: 'team.docx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
||
size: 3,
|
||
inline: false,
|
||
url: 'https://files.example.test/team.docx',
|
||
sha256: 'a'.repeat(64)
|
||
}]
|
||
);
|
||
assert.deepEqual((directFileResult.payload as Record<string, unknown>).attachments, [{
|
||
id: 'artifact-direct-1',
|
||
name: 'team.docx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
|
||
size: 3,
|
||
inline: false,
|
||
url: 'https://files.example.test/team.docx',
|
||
sha256: 'a'.repeat(64)
|
||
}]);
|
||
|
||
const insecureFileResult = createTaskResultFrame(
|
||
inbound,
|
||
session,
|
||
'completed',
|
||
'团队文件已准备',
|
||
'result-insecure-file-1',
|
||
{
|
||
kind: 'success',
|
||
text: '团队文件已准备',
|
||
recorded_at: new Date().toISOString(),
|
||
attachments: [{ name: 'team.docx', download_url: 'http://127.0.0.1:8786/file.docx' }]
|
||
}
|
||
);
|
||
assert.equal((insecureFileResult.payload as Record<string, unknown>).attachments, undefined);
|
||
|
||
const senderOnly = parseAgentBusFrame(JSON.stringify({
|
||
id: 'channel-event-without-conversation-1',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-2',
|
||
payload: { text: '补充信息' }
|
||
}));
|
||
assert.ok(senderOnly);
|
||
const fallbackProgress = createTaskProgressFrame(senderOnly, session, '处理中', 'progress-fallback-1');
|
||
assert.equal(fallbackProgress.conversation_id, 'agentbus:channel:wechat:user-2');
|
||
});
|
||
|
||
test('AgentBus result text uses the unified important message and preserves the confirmation gate', () => {
|
||
const needsInput = makeTask('awaiting_user_input', {
|
||
input_request: {
|
||
missing_fields: ['data.room_counts'],
|
||
questions: [{ field: 'data.room_counts', prompt: '请补充房型分配。' }],
|
||
captured_facts: {}
|
||
},
|
||
important_message: {
|
||
kind: 'awaiting_user_input',
|
||
text: '请补充房型分配，并说明是否一单一房。',
|
||
recorded_at: new Date().toISOString(),
|
||
input_request: {
|
||
missing_fields: ['data.room_counts'],
|
||
questions: [{ field: 'data.room_counts', prompt: '请补充房型分配。' }],
|
||
captured_facts: {}
|
||
}
|
||
}
|
||
});
|
||
assert.match(taskResultText(needsInput), /是否一单一房/);
|
||
assert.equal(taskResultStatus(needsInput), 'completed');
|
||
|
||
const completed = makeTask('completed', {
|
||
message: '散拼-创建母团计划已提交并完成回查。',
|
||
important_message: {
|
||
kind: 'success',
|
||
text: '散拼-创建母团计划已提交并完成回查。\n团号：LW-260907A-B',
|
||
recorded_at: new Date().toISOString(),
|
||
success_receipt: {
|
||
recorded_at: new Date().toISOString(),
|
||
receipt: {
|
||
group_numbers: ['LW-260907A-B', 'LW-260915A-A'],
|
||
room_counts: { SGL: 1, TWN: 8 },
|
||
verification_status: 'split_parent_completed'
|
||
}
|
||
}
|
||
}
|
||
});
|
||
assert.equal(
|
||
taskResultText(completed),
|
||
'散拼-创建母团计划已提交并完成回查。\n团号：LW-260907A-B'
|
||
);
|
||
assert.doesNotMatch(taskResultText(completed), /最终回执|verification_status|split_parent_completed/);
|
||
|
||
const awaitingConfirmation = makeTask('awaiting_confirmation');
|
||
assert.equal(taskResultText(awaitingConfirmation), '已受理，正在处理。');
|
||
assert.equal(taskResultStatus(awaitingConfirmation), 'completed');
|
||
|
||
const failed = makeTask('parse_failed', {
|
||
failure: {
|
||
error_code: 'blocked',
|
||
failure_stage: 'parse',
|
||
failure_source: 'agent',
|
||
failure_message: '资料不完整',
|
||
agent_returned: true,
|
||
plugin_dispatch_started: false,
|
||
erp_write_started: false,
|
||
no_plugin_dispatch: true,
|
||
no_erp_write: true,
|
||
validation_errors: []
|
||
}
|
||
});
|
||
assert.equal(taskResultStatus(failed), 'failed');
|
||
assert.match(taskResultText(failed), /资料不完整/);
|
||
|
||
const blocked = makeTask('blocked', {
|
||
important_message: {
|
||
kind: 'error',
|
||
text: '插件执行阻断：客户候选未唯一解析，已阻断 ERP 写入。；GetProduct ajax request was not observed; product_after_effect: expected exactly one existing LTJT option, found 0; Required fields still blank: customer, product.',
|
||
recorded_at: new Date().toISOString()
|
||
}
|
||
});
|
||
assert.equal(
|
||
taskResultText(blocked),
|
||
'插件执行阻断：客户候选未唯一解析，已阻断 ERP 写入。'
|
||
);
|
||
assert.doesNotMatch(taskResultText(blocked), /GetProduct|product_after_effect|Required fields/);
|
||
|
||
const erpFailure = makeTask('blocked', {
|
||
error_summary: {
|
||
recorded_at: new Date().toISOString(),
|
||
error_code: 'erp_business_rule_blocked',
|
||
stage: 'lifecycle_live_submit',
|
||
source: 'erp',
|
||
message: '此团还有【应收团款】账，不能取消！',
|
||
validation_errors: []
|
||
},
|
||
important_message: {
|
||
kind: 'error',
|
||
text: '此团还有【应收团款】账，不能取消！',
|
||
recorded_at: new Date().toISOString(),
|
||
error_summary: {
|
||
recorded_at: new Date().toISOString(),
|
||
error_code: 'erp_business_rule_blocked',
|
||
stage: 'lifecycle_live_submit',
|
||
source: 'erp',
|
||
message: '此团还有【应收团款】账，不能取消！',
|
||
validation_errors: []
|
||
}
|
||
}
|
||
});
|
||
assert.equal(taskResultStatus(erpFailure), 'failed');
|
||
assert.equal(taskResultText(erpFailure), '此团还有【应收团款】账，不能取消！');
|
||
|
||
const uncertainWrite = makeTask('reconciliation_pending', {
|
||
important_message: {
|
||
kind: 'error',
|
||
text: 'ERP 写入结果不确定：write_after_requery_mismatch',
|
||
recorded_at: new Date().toISOString()
|
||
}
|
||
});
|
||
assert.equal(taskResultStatus(uncertainWrite), 'failed');
|
||
assert.equal(
|
||
taskResultText(uncertainWrite),
|
||
'ERP 写入已发起，但系统尚未确认最终结果。请勿重复提交同一任务，等待任务所属账号只读核验。'
|
||
);
|
||
});
|
||
|
||
test('AgentBus attachment references omit archived visitor XLS and keep only XLSX delivery', () => {
|
||
const task = makeTask('completed', {
|
||
result: {
|
||
report: {
|
||
status: 'export_source_completed',
|
||
artifacts: [
|
||
{
|
||
artifact_id: 'artifact-visitor-xls',
|
||
file_name: 'Visitor.xls',
|
||
content_type: 'application/vnd.ms-excel',
|
||
bytes: 9377,
|
||
storage_backend: 'oss',
|
||
download_url: 'https://files.example.test/Visitor.xls',
|
||
agentbus_visible: false
|
||
},
|
||
{
|
||
artifact_id: 'artifact-visitor-xlsx',
|
||
file_name: 'Visitor.xlsx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
bytes: 5211,
|
||
storage_backend: 'oss',
|
||
download_url: 'https://files.example.test/Visitor.xlsx',
|
||
agentbus_visible: true
|
||
}
|
||
]
|
||
}
|
||
}
|
||
});
|
||
assert.deepEqual(taskAttachmentReferences(task), [{
|
||
artifact_id: 'artifact-visitor-xlsx',
|
||
name: 'Visitor.xlsx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
size: 5211,
|
||
url: 'https://files.example.test/Visitor.xlsx',
|
||
storage_backend: 'oss'
|
||
}]);
|
||
});
|
||
|
||
test('AgentBus listener connects with the documented Authorization header and returns one final result', async (t) => {
|
||
const config = testConfig();
|
||
const socket = new FakeSocket();
|
||
let capturedUrl = '';
|
||
let capturedHeaders: Record<string, string> = {};
|
||
let currentTask = makeTask('parse_queued');
|
||
const events = new EventEmitter();
|
||
const received: Array<Record<string, unknown>> = [];
|
||
let capturedContext: Record<string, unknown> = {};
|
||
const logs: Array<{ level: string; metadata: Record<string, unknown>; message?: string }> = [];
|
||
const tasks: AgentBusTaskGateway = {
|
||
events,
|
||
async ingestMessage(context, input) {
|
||
capturedContext = context as unknown as Record<string, unknown>;
|
||
received.push(input as unknown as Record<string, unknown>);
|
||
return { task: currentTask, attached: false, created: true };
|
||
},
|
||
async getTask() {
|
||
return currentTask;
|
||
}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config,
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {
|
||
currentTask = makeTask('awaiting_confirmation');
|
||
events.emit('task', {
|
||
id: 1,
|
||
organization_id: 'org-1',
|
||
task_id: currentTask.task_id,
|
||
status: currentTask.status,
|
||
stage: 'parse',
|
||
message: currentTask.message,
|
||
payload: {},
|
||
created_at: new Date().toISOString()
|
||
});
|
||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||
currentTask = makeTask('confirmed', {
|
||
confirmation_mode: 'automatic',
|
||
confirmed_at: new Date().toISOString(),
|
||
handoff_status: 'awaiting_handoff',
|
||
message: '指令解析完成，已进入全自动化 ERP 执行。'
|
||
});
|
||
events.emit('task', {
|
||
id: 2,
|
||
organization_id: 'org-1',
|
||
task_id: currentTask.task_id,
|
||
status: currentTask.status,
|
||
stage: 'automation',
|
||
message: currentTask.message,
|
||
payload: {},
|
||
created_at: new Date().toISOString()
|
||
});
|
||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||
currentTask = makeTask('completed', {
|
||
message: '散拼-创建母团计划已提交并完成回查。',
|
||
important_message: {
|
||
kind: 'success',
|
||
text: '散拼-创建母团计划已提交并完成回查。\n团号：LW-260907A-B',
|
||
recorded_at: new Date().toISOString(),
|
||
success_receipt: {
|
||
recorded_at: new Date().toISOString(),
|
||
receipt: {
|
||
group_number: 'LW-260907A-B',
|
||
verification_status: 'split_parent_completed'
|
||
}
|
||
}
|
||
}
|
||
});
|
||
events.emit('task', {
|
||
id: 3,
|
||
organization_id: 'org-1',
|
||
task_id: currentTask.task_id,
|
||
status: currentTask.status,
|
||
stage: 'browser_execution',
|
||
message: currentTask.message,
|
||
payload: {},
|
||
created_at: new Date().toISOString()
|
||
});
|
||
},
|
||
socketFactory: (url, options) => {
|
||
capturedUrl = url;
|
||
capturedHeaders = options.headers;
|
||
return socket as unknown as AgentBusSocket;
|
||
},
|
||
logger: {
|
||
info(metadata, message) {
|
||
logs.push({ level: 'info', metadata, message });
|
||
},
|
||
warn(metadata, message) {
|
||
logs.push({ level: 'warn', metadata, message });
|
||
},
|
||
error(metadata, message) {
|
||
logs.push({ level: 'error', metadata, message });
|
||
}
|
||
}
|
||
});
|
||
t.after(() => listener.stop());
|
||
|
||
listener.start();
|
||
assert.match(capturedUrl, /[?&]ready=1/);
|
||
assert.equal(capturedHeaders.Authorization, 'Bearer test-ws-token');
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-1',
|
||
type: 'event',
|
||
session_id: 'session-1',
|
||
epoch: 1,
|
||
to: 'bot:test:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ignored-frame-1',
|
||
type: 'message',
|
||
from: 'channel:wechat:user-1',
|
||
payload: { text: '格式不符合监听契约' }
|
||
}));
|
||
const wrappedBusinessText = [
|
||
'New WeChat message',
|
||
'Conversation: conversation-wechat-1',
|
||
'Text: 独立团单个下单',
|
||
'发团日期：2026-11-02',
|
||
'预订客户：示例客户',
|
||
'产品搜索：示例产品',
|
||
'预估人数：15+1',
|
||
'用房数量：TWN 8，SGL 1'
|
||
].join('\n');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'channel-event-1',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
payload: { text: wrappedBusinessText, reply_policy: { progress: true } }
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 2);
|
||
assert.equal(received.length, 1);
|
||
assert.equal(capturedContext.source, 'agentbus');
|
||
assert.equal(received[0].message, wrappedBusinessText.split('\n').slice(2).join('\n').replace(/^Text:\s*/, ''));
|
||
assert.equal(received[0].conversationId, 'conversation-wechat-1');
|
||
assert.equal(received[0].idempotencyKey, 'agentbus:channel-event-1');
|
||
assert.equal(socket.sent.length, 2);
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).event, 'task.progress');
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).status, 'accepted');
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).text, '已受理，正在处理。');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).event, 'task.result');
|
||
assert.equal(socket.sent[1].reply_to, 'channel-event-1');
|
||
assert.match(String((socket.sent[1].payload as Record<string, unknown>).text), /散拼-创建母团计划已提交并完成回查/);
|
||
assert.match(String((socket.sent[1].payload as Record<string, unknown>).text), /LW-260907A-B/);
|
||
assert.doesNotMatch(String((socket.sent[1].payload as Record<string, unknown>).text), /最终回执|verification_status|split_parent_completed/);
|
||
assert.deepEqual((socket.sent[1].payload as Record<string, unknown>).important_message, {
|
||
kind: 'success',
|
||
text: '散拼-创建母团计划已提交并完成回查。\n团号：LW-260907A-B'
|
||
});
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'listener_starting'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'session_ready'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'frame_ignored' && entry.metadata.ignore_reason === 'unexpected_type:message'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'frame_received' && entry.metadata.frame_id === 'channel-event-1'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'frame_received' && entry.metadata.text_preview === wrappedBusinessText));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'task_processing_started'
|
||
&& entry.metadata.transport_envelope_unwrapped === true));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'task_ingested' && entry.metadata.task_id === 'TASK-1'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'outbound_progress_sent'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'outbound_result_sent'));
|
||
assert.doesNotMatch(JSON.stringify(logs), /test-ws-token/);
|
||
});
|
||
|
||
test('WeChat attachment placeholder without file metadata fails closed before task ingestion', async (t) => {
|
||
const socket = new FakeSocket();
|
||
const events = new EventEmitter();
|
||
const logs: Array<{ level: string; metadata: Record<string, unknown>; message?: string }> = [];
|
||
let ingestCalls = 0;
|
||
const tasks: AgentBusTaskGateway = {
|
||
events,
|
||
async ingestMessage() {
|
||
ingestCalls += 1;
|
||
return { task: makeTask('parse_queued'), attached: false, created: true };
|
||
},
|
||
async getTask() {
|
||
return makeTask('failed');
|
||
}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config: testConfig(),
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {},
|
||
socketFactory: () => socket as unknown as AgentBusSocket,
|
||
logger: {
|
||
info(metadata, message) {
|
||
logs.push({ level: 'info', metadata, message });
|
||
},
|
||
warn(metadata, message) {
|
||
logs.push({ level: 'warn', metadata, message });
|
||
},
|
||
error(metadata, message) {
|
||
logs.push({ level: 'error', metadata, message });
|
||
}
|
||
}
|
||
});
|
||
t.after(() => listener.stop());
|
||
|
||
listener.start();
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-attachment-placeholder',
|
||
type: 'event',
|
||
session_id: 'session-attachment-placeholder',
|
||
epoch: 1,
|
||
to: 'bot:test:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'wechat-attachment-placeholder',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
payload: {
|
||
text: [
|
||
'New WeChat message',
|
||
'Conversation: thread:conversation-attachment-1',
|
||
'Text: [WeChat attachment: synthetic-roster.xlsx]'
|
||
].join('\n')
|
||
}
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 1);
|
||
assert.equal(ingestCalls, 0);
|
||
assert.equal(socket.sent.length, 1);
|
||
assert.equal(socket.sent[0].conversation_id, 'thread:conversation-attachment-1');
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).event, 'task.result');
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).status, 'failed');
|
||
assert.equal(
|
||
(socket.sent[0].payload as Record<string, unknown>).text,
|
||
'附件内容未传到平台，原任务仍在等待附件。请检查微信桥接器的文件转发后重新发送。'
|
||
);
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'task_processing_failed'
|
||
&& entry.metadata.error_code === 'roster_attachment_metadata_missing'
|
||
&& /^[a-f0-9]{24}$/u.test(String(entry.metadata.error_fingerprint))));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'task_processing_started'
|
||
&& entry.metadata.attachment_count === 0
|
||
&& entry.metadata.attachment_placeholder === true));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'attachment_metadata_missing'));
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'outbound_result_sent'));
|
||
});
|
||
|
||
test('attachment validation rejection returns an immediate failed result without scheduling parsing', async (t) => {
|
||
const socket = new FakeSocket();
|
||
const events = new EventEmitter();
|
||
const rejectionMessage = '名单附件校验未通过【roster_workbook_conversion_failed】，请按模板修正后重新发送。';
|
||
let parseQueueCalls = 0;
|
||
const tasks: AgentBusTaskGateway = {
|
||
events,
|
||
async ingestMessage() {
|
||
return {
|
||
task: makeTask('awaiting_attachment', {
|
||
task_id: 'TASK-ATTACHMENT-REJECTED',
|
||
source: 'agentbus',
|
||
message: rejectionMessage
|
||
}),
|
||
attached: true,
|
||
created: false,
|
||
input_attachment: {
|
||
status: 'rejected',
|
||
error_code: 'roster_workbook_conversion_failed',
|
||
message: rejectionMessage
|
||
}
|
||
};
|
||
},
|
||
async getTask() {
|
||
return makeTask('awaiting_attachment');
|
||
}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config: testConfig(),
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {
|
||
parseQueueCalls += 1;
|
||
},
|
||
socketFactory: () => socket as unknown as AgentBusSocket
|
||
});
|
||
t.after(() => listener.stop());
|
||
|
||
listener.start();
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-attachment-rejected',
|
||
type: 'event',
|
||
session_id: 'session-attachment-rejected',
|
||
epoch: 1,
|
||
to: 'bot:test:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'wechat-attachment-rejected',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-attachment-rejected',
|
||
payload: { text: '名单附件补充消息' }
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 2);
|
||
assert.equal(parseQueueCalls, 0);
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).event, 'task.progress');
|
||
assert.equal(
|
||
(socket.sent[0].payload as Record<string, unknown>).text,
|
||
AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT
|
||
);
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).event, 'task.result');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).status, 'failed');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).text, rejectionMessage);
|
||
assert.equal(socket.sent[1].reply_to, 'wechat-attachment-rejected');
|
||
assert.equal(socket.sent[1].conversation_id, 'conversation-attachment-rejected');
|
||
});
|
||
|
||
test('roster direct listener sends waiting, attachment-received, and one final reply across two frames', async (t) => {
|
||
const socket = new FakeSocket();
|
||
const events = new EventEmitter();
|
||
const taskId = 'TASK-ROSTER-REPLY-LIFECYCLE';
|
||
const finalText = '名单导入成功 子单号：D14541 本次导入：25人';
|
||
let parseQueueCalls = 0;
|
||
let ingestCalls = 0;
|
||
let currentTask = makeTask('awaiting_attachment', {
|
||
task_id: taskId,
|
||
source: 'agentbus',
|
||
message: AGENTBUS_ROSTER_WAITING_TEXT
|
||
});
|
||
const tasks: AgentBusTaskGateway = {
|
||
events,
|
||
async ingestMessage() {
|
||
ingestCalls += 1;
|
||
if (ingestCalls === 1) {
|
||
return { task: currentTask, attached: false, created: true };
|
||
}
|
||
currentTask = makeTask('parse_queued', {
|
||
task_id: taskId,
|
||
source: 'agentbus',
|
||
message: '名单附件已校验，共 25 行；任务已入队，等待程序解析。'
|
||
});
|
||
return {
|
||
task: currentTask,
|
||
attached: true,
|
||
created: false,
|
||
input_attachment: {
|
||
status: 'normalized',
|
||
row_count: 25,
|
||
message: '名单附件已校验，共 25 行，开始程序解析。'
|
||
}
|
||
};
|
||
},
|
||
async getTask() {
|
||
return currentTask;
|
||
}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config: testConfig(),
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {
|
||
parseQueueCalls += 1;
|
||
currentTask = makeTask('completed', {
|
||
task_id: taskId,
|
||
source: 'agentbus',
|
||
message: finalText,
|
||
important_message: {
|
||
kind: 'success',
|
||
text: finalText,
|
||
recorded_at: new Date().toISOString()
|
||
}
|
||
});
|
||
events.emit('task', {
|
||
id: 1,
|
||
organization_id: 'org-1',
|
||
task_id: taskId,
|
||
status: 'completed',
|
||
stage: 'browser_execution',
|
||
message: finalText,
|
||
payload: {},
|
||
created_at: new Date().toISOString()
|
||
});
|
||
},
|
||
socketFactory: () => socket as unknown as AgentBusSocket
|
||
});
|
||
t.after(() => listener.stop());
|
||
|
||
listener.start();
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-roster-lifecycle',
|
||
type: 'event',
|
||
session_id: 'session-roster-lifecycle',
|
||
epoch: 1,
|
||
to: 'bot:test:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'roster-directive-frame',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-roster-lifecycle',
|
||
payload: { text: '导入散拼子单名单' }
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 1);
|
||
assert.equal(parseQueueCalls, 0);
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).event, 'task.progress');
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).text, AGENTBUS_ROSTER_WAITING_TEXT);
|
||
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'roster-attachment-frame',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-roster-lifecycle',
|
||
payload: { text: '名单附件补充消息' }
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 3);
|
||
assert.equal(parseQueueCalls, 1);
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).event, 'task.progress');
|
||
assert.equal(
|
||
(socket.sent[1].payload as Record<string, unknown>).text,
|
||
AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT
|
||
);
|
||
const finalReplies = socket.sent.filter((frame) => {
|
||
const payload = frame.payload as Record<string, unknown>;
|
||
return payload.event === 'task.result' && payload.text === finalText;
|
||
});
|
||
assert.equal(finalReplies.length, 1);
|
||
assert.equal(finalReplies[0].reply_to, 'roster-attachment-frame');
|
||
});
|
||
|
||
test('listener reload stop does not overwrite the channel status as disabled', () => {
|
||
const statuses: string[] = [];
|
||
const listener = new AgentBusListener({
|
||
config: testConfig(),
|
||
tasks: {} as AgentBusTaskGateway,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {},
|
||
onStatusChange: (status) => statuses.push(status)
|
||
});
|
||
|
||
listener.stop(false);
|
||
assert.deepEqual(statuses, []);
|
||
listener.stop();
|
||
assert.deepEqual(statuses, ['disabled']);
|
||
});
|
||
|
||
test('channel manager replays a reload requested while another reload is in flight', async () => {
|
||
const manager = new AgentBusManager({
|
||
config: testConfig(),
|
||
tasks: {} as AgentBusTaskGateway,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {}
|
||
});
|
||
const service = manager.channelService as unknown as {
|
||
ensureLegacyChannel: (organizationId: string) => Promise<void>;
|
||
listEnabledSecrets: (organizationId: string) => Promise<[]>;
|
||
};
|
||
let listCalls = 0;
|
||
let releaseFirstList: (() => void) | undefined;
|
||
service.ensureLegacyChannel = async () => {};
|
||
service.listEnabledSecrets = async () => {
|
||
listCalls += 1;
|
||
if (listCalls === 1) {
|
||
await new Promise<void>((resolve) => {
|
||
releaseFirstList = resolve;
|
||
});
|
||
}
|
||
return [];
|
||
};
|
||
|
||
const initialReload = manager.start();
|
||
while (!releaseFirstList) await new Promise<void>((resolve) => setImmediate(resolve));
|
||
const mutationReload = manager.reload();
|
||
releaseFirstList();
|
||
await Promise.all([initialReload, mutationReload]);
|
||
|
||
assert.equal(listCalls, 2);
|
||
await manager.stop();
|
||
});
|
||
|
||
test('listener shutdown carries its session epoch for stale-writer protection', () => {
|
||
const socket = new FakeSocket();
|
||
const changes: Array<{ status: string; epoch: number | null }> = [];
|
||
const listener = new AgentBusListener({
|
||
config: testConfig(),
|
||
tasks: {} as AgentBusTaskGateway,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {},
|
||
socketFactory: () => socket as unknown as AgentBusSocket,
|
||
onStatusChange: (status, _error, epoch) => changes.push({ status, epoch })
|
||
});
|
||
|
||
listener.start();
|
||
socket.readyState = 1;
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-epoch-1',
|
||
type: 'event',
|
||
session_id: 'session-epoch-1',
|
||
epoch: 7,
|
||
to: 'bot:test:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
listener.stop();
|
||
|
||
assert.deepEqual(changes.at(-1), { status: 'disabled', epoch: 7 });
|
||
});
|
||
|
||
test('durable channel listener persists route and resends accepted/result deliveries', async () => {
|
||
const config = testConfig();
|
||
const socket = new FakeSocket();
|
||
const events = new EventEmitter();
|
||
let currentTask = makeTask('parse_queued', {
|
||
task_id: 'TASK-DURABLE',
|
||
channel_id: 'channel-1',
|
||
channel_name: '外部用户 A'
|
||
});
|
||
let acceptedCreated = false;
|
||
let resultEnqueued = false;
|
||
const deliveryState = {
|
||
acceptedDelivered: false,
|
||
resultDelivered: false,
|
||
failures: [] as string[]
|
||
};
|
||
let resultPayload: Record<string, unknown> = {};
|
||
const accepted: AgentBusDelivery = {
|
||
id: 'delivery-accepted',
|
||
channel_id: 'channel-1',
|
||
task_id: 'task-row-1',
|
||
inbound_frame_id: 'channel-durable-1',
|
||
inbound_from: 'channel:external:user-a',
|
||
conversation_id: 'conversation-durable-1',
|
||
delivery_kind: 'accepted',
|
||
payload: createAgentBusAcceptedDeliveryPayload(),
|
||
attempt_count: 1
|
||
};
|
||
const result: AgentBusDelivery = {
|
||
...accepted,
|
||
id: 'delivery-result',
|
||
delivery_kind: 'result',
|
||
payload: resultPayload
|
||
};
|
||
let capturedHeaders: Record<string, string> = {};
|
||
let receivedContext: Record<string, unknown> = {};
|
||
let receivedInput: Record<string, unknown> = {};
|
||
const tasks: AgentBusTaskGateway & { deliveryState: typeof deliveryState } = {
|
||
deliveryState,
|
||
events,
|
||
async ingestMessage(context, input) {
|
||
acceptedCreated = true;
|
||
receivedContext = context as unknown as Record<string, unknown>;
|
||
receivedInput = input as unknown as Record<string, unknown>;
|
||
return { task: currentTask, attached: false, created: true };
|
||
},
|
||
async getTask() {
|
||
return currentTask;
|
||
},
|
||
async listAgentBusFinalizationCandidates() {
|
||
if (acceptedCreated && deliveryState.acceptedDelivered && !resultEnqueued && currentTask.status === 'completed') {
|
||
return [{ channel_id: 'channel-1', task_id: currentTask.task_id, inbound_frame_id: accepted.inbound_frame_id }];
|
||
}
|
||
return [];
|
||
},
|
||
async enqueueAgentBusResult(_channelId, _inboundFrameId, payload) {
|
||
resultPayload = payload;
|
||
result.payload = resultPayload;
|
||
resultEnqueued = true;
|
||
},
|
||
async claimAgentBusDeliveries() {
|
||
if (acceptedCreated && !deliveryState.acceptedDelivered) return [accepted];
|
||
if (resultEnqueued && !deliveryState.resultDelivered) return [result];
|
||
return [];
|
||
},
|
||
async markAgentBusDeliveryDelivered(this: { deliveryState: typeof deliveryState }, deliveryId) {
|
||
if (deliveryId === accepted.id) this.deliveryState.acceptedDelivered = true;
|
||
if (deliveryId === result.id) this.deliveryState.resultDelivered = true;
|
||
},
|
||
async markAgentBusDeliveryFailed(this: { deliveryState: typeof deliveryState }, _deliveryId, errorMessage) {
|
||
this.deliveryState.failures.push(errorMessage);
|
||
},
|
||
async releaseAgentBusDeliveries() {}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config,
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {
|
||
currentTask = makeTask('completed', {
|
||
task_id: 'TASK-DURABLE',
|
||
channel_id: 'channel-1',
|
||
channel_name: '外部用户 A',
|
||
message: '任务已完成。',
|
||
result: {
|
||
report: {
|
||
status: 'export_source_completed',
|
||
artifacts: [
|
||
{
|
||
artifact_id: 'artifact-durable-source-xls',
|
||
file_name: 'Visitor.xls',
|
||
content_type: 'application/vnd.ms-excel',
|
||
bytes: 3,
|
||
sha256: 'b'.repeat(64),
|
||
storage_backend: 'oss',
|
||
download_url: 'https://files.example.test/Visitor.xls',
|
||
agentbus_visible: false
|
||
},
|
||
{
|
||
artifact_id: 'artifact-durable-1',
|
||
file_name: 'Visitor.xlsx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
bytes: 4,
|
||
sha256: 'a'.repeat(64),
|
||
storage_backend: 'oss',
|
||
download_url: 'https://files.example.test/Visitor.xlsx',
|
||
agentbus_visible: true
|
||
}
|
||
]
|
||
}
|
||
},
|
||
important_message: {
|
||
kind: 'success',
|
||
text: '任务已完成。',
|
||
recorded_at: new Date().toISOString()
|
||
}
|
||
});
|
||
},
|
||
socketFactory: (_url, options) => {
|
||
capturedHeaders = options.headers;
|
||
return socket as unknown as AgentBusSocket;
|
||
},
|
||
channel: {
|
||
id: 'channel-1',
|
||
displayName: '外部用户 A',
|
||
wsUrl: 'wss://mesh.nianxx.cn/ws',
|
||
wsToken: 'channel-ws-token',
|
||
botAddress: 'bot:channel-a:listener',
|
||
ownerUserId: 'user-a',
|
||
ownerRole: 'user'
|
||
}
|
||
});
|
||
|
||
listener.start();
|
||
assert.equal(capturedHeaders.Authorization, 'Bearer channel-ws-token');
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-durable-1',
|
||
type: 'event',
|
||
session_id: 'session-durable-1',
|
||
epoch: 4,
|
||
to: 'bot:channel-a:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'channel-durable-1',
|
||
type: 'event',
|
||
from: 'channel:external:user-a',
|
||
conversation_id: 'conversation-durable-1',
|
||
payload: { text: '创建一个新团' }
|
||
}));
|
||
|
||
const deadline = Date.now() + 3_000;
|
||
while (Date.now() < deadline && !deliveryState.resultDelivered) {
|
||
await new Promise((resolve) => setTimeout(resolve, 50));
|
||
}
|
||
assert.equal(receivedContext.channelId, 'channel-1');
|
||
assert.equal(receivedContext.userId, 'user-a');
|
||
assert.equal(receivedContext.role, 'user');
|
||
assert.equal(receivedInput.channelId, 'channel-1');
|
||
assert.deepEqual(receivedInput.agentBusRoute, {
|
||
inboundFrameId: 'channel-durable-1',
|
||
inboundFrom: 'channel:external:user-a',
|
||
conversationId: 'conversation-durable-1'
|
||
});
|
||
assert.deepEqual(resultPayload.attachments, [{
|
||
artifact_id: 'artifact-durable-1',
|
||
name: 'Visitor.xlsx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
size: 4,
|
||
url: 'https://files.example.test/Visitor.xlsx',
|
||
sha256: 'a'.repeat(64),
|
||
storage_backend: 'oss'
|
||
}]);
|
||
assert.doesNotMatch(JSON.stringify(resultPayload), /content_base64|download_url/);
|
||
assert.equal(socket.sent.length, 2);
|
||
assert.equal(socket.sent[0].id, 'bot-accepted-delivery-accepted');
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).event, 'task.progress');
|
||
assert.equal(socket.sent[1].id, 'bot-result-delivery-result');
|
||
assert.equal(socket.sent[1].reply_to, 'channel-durable-1');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).event, 'task.result');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).status, 'completed');
|
||
assert.deepEqual((socket.sent[1].payload as Record<string, unknown>).attachments, [{
|
||
id: 'artifact-durable-1',
|
||
name: 'Visitor.xlsx',
|
||
content_type: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
size: 4,
|
||
inline: false,
|
||
url: 'https://files.example.test/Visitor.xlsx',
|
||
sha256: 'a'.repeat(64)
|
||
}]);
|
||
assert.equal(deliveryState.acceptedDelivered, true);
|
||
assert.equal(deliveryState.resultDelivered, true);
|
||
assert.deepEqual(deliveryState.failures, []);
|
||
listener.stop();
|
||
});
|
||
|
||
test('durable roster lifecycle assigns the final result only to the attachment frame', async (t) => {
|
||
const socket = new FakeSocket();
|
||
const events = new EventEmitter();
|
||
const publicTaskId = 'TASK-DURABLE-ROSTER-LIFECYCLE';
|
||
const taskRowId = 'task-row-durable-roster-lifecycle';
|
||
const finalText = '名单导入成功 子单号：D14541 本次导入：25人';
|
||
const deliveries: AgentBusDelivery[] = [];
|
||
const deliveryStatuses = new Map<string, 'pending' | 'sending' | 'delivered'>();
|
||
let deliverySequence = 0;
|
||
let parseQueueCalls = 0;
|
||
let currentTask = makeTask('awaiting_attachment', {
|
||
task_id: publicTaskId,
|
||
source: 'agentbus',
|
||
channel_id: 'channel-1',
|
||
channel_name: '外部用户 A',
|
||
message: AGENTBUS_ROSTER_WAITING_TEXT
|
||
});
|
||
const addDelivery = (
|
||
kind: AgentBusDelivery['delivery_kind'],
|
||
inboundFrameId: string,
|
||
payload: Record<string, unknown>
|
||
) => {
|
||
deliverySequence += 1;
|
||
const delivery: AgentBusDelivery = {
|
||
id: `delivery-roster-${deliverySequence}`,
|
||
channel_id: 'channel-1',
|
||
task_id: taskRowId,
|
||
inbound_frame_id: inboundFrameId,
|
||
inbound_from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-durable-roster-lifecycle',
|
||
delivery_kind: kind,
|
||
payload,
|
||
attempt_count: 1
|
||
};
|
||
deliveries.push(delivery);
|
||
deliveryStatuses.set(delivery.id, 'pending');
|
||
return delivery;
|
||
};
|
||
let ingestCalls = 0;
|
||
const tasks: AgentBusTaskGateway = {
|
||
events,
|
||
async ingestMessage(_context, input) {
|
||
ingestCalls += 1;
|
||
const inboundFrameId = String(input.agentBusRoute?.inboundFrameId || '');
|
||
if (ingestCalls === 1) {
|
||
addDelivery('accepted', inboundFrameId, createAgentBusAcceptedDeliveryPayload({
|
||
text: AGENTBUS_ROSTER_WAITING_TEXT,
|
||
finalReplyEligible: false
|
||
}));
|
||
return { task: currentTask, attached: false, created: true };
|
||
}
|
||
addDelivery('accepted', inboundFrameId, createAgentBusAcceptedDeliveryPayload({
|
||
text: AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT,
|
||
finalReplyEligible: true
|
||
}));
|
||
currentTask = makeTask('parse_queued', {
|
||
task_id: publicTaskId,
|
||
source: 'agentbus',
|
||
channel_id: 'channel-1',
|
||
channel_name: '外部用户 A'
|
||
});
|
||
return {
|
||
task: currentTask,
|
||
attached: true,
|
||
created: false,
|
||
input_attachment: {
|
||
status: 'normalized',
|
||
row_count: 25,
|
||
message: '名单附件已校验，共 25 行，开始程序解析。'
|
||
}
|
||
};
|
||
},
|
||
async getTask() {
|
||
return currentTask;
|
||
},
|
||
async listAgentBusFinalizationCandidates() {
|
||
if (currentTask.status !== 'completed') return [];
|
||
const eligibleAccepted = deliveries.filter((delivery) => (
|
||
delivery.delivery_kind === 'accepted'
|
||
&& deliveryStatuses.get(delivery.id) === 'delivered'
|
||
&& delivery.payload[AGENTBUS_FINAL_REPLY_ELIGIBLE_FIELD] !== false
|
||
));
|
||
const latest = eligibleAccepted.at(-1);
|
||
if (!latest) return [];
|
||
const alreadyHasResult = deliveries.some((delivery) => (
|
||
delivery.delivery_kind === 'result'
|
||
&& delivery.inbound_frame_id === latest.inbound_frame_id
|
||
));
|
||
return alreadyHasResult ? [] : [{
|
||
channel_id: 'channel-1',
|
||
task_id: publicTaskId,
|
||
inbound_frame_id: latest.inbound_frame_id
|
||
}];
|
||
},
|
||
async enqueueAgentBusResult(_channelId, inboundFrameId, payload) {
|
||
addDelivery('result', inboundFrameId, payload);
|
||
},
|
||
async claimAgentBusDeliveries() {
|
||
const pending = deliveries.filter((delivery) => deliveryStatuses.get(delivery.id) === 'pending');
|
||
for (const delivery of pending) deliveryStatuses.set(delivery.id, 'sending');
|
||
return pending;
|
||
},
|
||
async markAgentBusDeliveryDelivered(deliveryId) {
|
||
deliveryStatuses.set(deliveryId, 'delivered');
|
||
},
|
||
async markAgentBusDeliveryFailed() {},
|
||
async releaseAgentBusDeliveries() {}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config: testConfig(),
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {
|
||
parseQueueCalls += 1;
|
||
currentTask = makeTask('completed', {
|
||
task_id: publicTaskId,
|
||
source: 'agentbus',
|
||
channel_id: 'channel-1',
|
||
channel_name: '外部用户 A',
|
||
message: finalText,
|
||
important_message: {
|
||
kind: 'success',
|
||
text: finalText,
|
||
recorded_at: new Date().toISOString()
|
||
}
|
||
});
|
||
},
|
||
socketFactory: () => socket as unknown as AgentBusSocket,
|
||
channel: {
|
||
id: 'channel-1',
|
||
displayName: '外部用户 A',
|
||
wsUrl: 'wss://mesh.nianxx.cn/ws',
|
||
wsToken: 'channel-ws-token',
|
||
botAddress: 'bot:channel-a:listener',
|
||
ownerUserId: 'user-a',
|
||
ownerRole: 'user'
|
||
}
|
||
});
|
||
t.after(() => listener.stop());
|
||
|
||
listener.start();
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-durable-roster-lifecycle',
|
||
type: 'event',
|
||
session_id: 'session-durable-roster-lifecycle',
|
||
epoch: 1,
|
||
to: 'bot:channel-a:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'durable-roster-directive-frame',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-durable-roster-lifecycle',
|
||
payload: { text: '导入散拼子单名单' }
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 1, 3_000);
|
||
assert.equal(parseQueueCalls, 0);
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).text, AGENTBUS_ROSTER_WAITING_TEXT);
|
||
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'durable-roster-attachment-frame',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-durable-roster-lifecycle',
|
||
payload: { text: '名单附件补充消息' }
|
||
}));
|
||
|
||
await waitFor(() => socket.sent.length === 3, 4_000);
|
||
assert.equal(parseQueueCalls, 1);
|
||
assert.deepEqual(socket.sent.map((frame) => {
|
||
const payload = frame.payload as Record<string, unknown>;
|
||
return { event: payload.event, text: payload.text, reply_to: frame.reply_to };
|
||
}), [
|
||
{
|
||
event: 'task.progress',
|
||
text: AGENTBUS_ROSTER_WAITING_TEXT,
|
||
reply_to: 'durable-roster-directive-frame'
|
||
},
|
||
{
|
||
event: 'task.progress',
|
||
text: AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT,
|
||
reply_to: 'durable-roster-attachment-frame'
|
||
},
|
||
{
|
||
event: 'task.result',
|
||
text: finalText,
|
||
reply_to: 'durable-roster-attachment-frame'
|
||
}
|
||
]);
|
||
assert.equal(
|
||
socket.sent.some((frame) => AGENTBUS_FINAL_REPLY_ELIGIBLE_FIELD in (frame.payload as Record<string, unknown>)),
|
||
false
|
||
);
|
||
});
|
||
|
||
test('durable channel persists and sends an attachment rejection result while the task keeps waiting', async (t) => {
|
||
const config = testConfig();
|
||
const socket = new FakeSocket();
|
||
const events = new EventEmitter();
|
||
const rejectionMessage = '名单附件校验未通过【roster_workbook_conversion_failed】，请按模板修正后重新发送。';
|
||
let acceptedCreated = false;
|
||
let resultEnqueued = false;
|
||
let acceptedDelivered = false;
|
||
let resultDelivered = false;
|
||
let parseQueueCalls = 0;
|
||
let resultPayload: Record<string, unknown> = {};
|
||
const accepted: AgentBusDelivery = {
|
||
id: 'delivery-attachment-rejected-accepted',
|
||
channel_id: 'channel-1',
|
||
task_id: 'task-row-attachment-rejected',
|
||
inbound_frame_id: 'channel-attachment-rejected',
|
||
inbound_from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-attachment-rejected',
|
||
delivery_kind: 'accepted',
|
||
payload: createAgentBusAcceptedDeliveryPayload({
|
||
text: AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT,
|
||
finalReplyEligible: true
|
||
}),
|
||
attempt_count: 1
|
||
};
|
||
const result: AgentBusDelivery = {
|
||
...accepted,
|
||
id: 'delivery-attachment-rejected-result',
|
||
delivery_kind: 'result',
|
||
payload: resultPayload
|
||
};
|
||
const logs: Array<{ level: string; metadata: Record<string, unknown>; message?: string }> = [];
|
||
const tasks: AgentBusTaskGateway = {
|
||
events,
|
||
async ingestMessage() {
|
||
acceptedCreated = true;
|
||
return {
|
||
task: makeTask('awaiting_attachment', {
|
||
task_id: 'TASK-ATTACHMENT-REJECTED',
|
||
source: 'agentbus',
|
||
channel_id: 'channel-1',
|
||
channel_name: '外部用户 A',
|
||
message: rejectionMessage
|
||
}),
|
||
attached: true,
|
||
created: false,
|
||
input_attachment: {
|
||
status: 'rejected',
|
||
error_code: 'roster_workbook_conversion_failed',
|
||
message: rejectionMessage
|
||
}
|
||
};
|
||
},
|
||
async getTask() {
|
||
return makeTask('awaiting_attachment');
|
||
},
|
||
async listAgentBusFinalizationCandidates() {
|
||
return [];
|
||
},
|
||
async enqueueAgentBusResult(channelId, inboundFrameId, payload) {
|
||
assert.equal(channelId, 'channel-1');
|
||
assert.equal(inboundFrameId, 'channel-attachment-rejected');
|
||
resultPayload = payload;
|
||
result.payload = resultPayload;
|
||
resultEnqueued = true;
|
||
},
|
||
async claimAgentBusDeliveries() {
|
||
const deliveries: AgentBusDelivery[] = [];
|
||
if (acceptedCreated && !acceptedDelivered) deliveries.push(accepted);
|
||
if (resultEnqueued && !resultDelivered) deliveries.push(result);
|
||
return deliveries;
|
||
},
|
||
async markAgentBusDeliveryDelivered(deliveryId) {
|
||
if (deliveryId === accepted.id) acceptedDelivered = true;
|
||
if (deliveryId === result.id) resultDelivered = true;
|
||
},
|
||
async markAgentBusDeliveryFailed() {},
|
||
async releaseAgentBusDeliveries() {}
|
||
};
|
||
const listener = new AgentBusListener({
|
||
config,
|
||
tasks,
|
||
organizationId: 'org-1',
|
||
scheduleParseQueue: async () => {
|
||
parseQueueCalls += 1;
|
||
},
|
||
socketFactory: () => socket as unknown as AgentBusSocket,
|
||
channel: {
|
||
id: 'channel-1',
|
||
displayName: '外部用户 A',
|
||
wsUrl: 'wss://mesh.nianxx.cn/ws',
|
||
wsToken: 'channel-ws-token',
|
||
botAddress: 'bot:channel-a:listener',
|
||
ownerUserId: 'user-a',
|
||
ownerRole: 'user'
|
||
},
|
||
logger: {
|
||
info(metadata, message) {
|
||
logs.push({ level: 'info', metadata, message });
|
||
},
|
||
warn(metadata, message) {
|
||
logs.push({ level: 'warn', metadata, message });
|
||
},
|
||
error(metadata, message) {
|
||
logs.push({ level: 'error', metadata, message });
|
||
}
|
||
}
|
||
});
|
||
t.after(() => listener.stop());
|
||
|
||
listener.start();
|
||
socket.readyState = 1;
|
||
socket.emit('open');
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'ready-durable-attachment-rejected',
|
||
type: 'event',
|
||
session_id: 'session-durable-attachment-rejected',
|
||
epoch: 4,
|
||
to: 'bot:channel-a:listener',
|
||
payload: { event: 'session.ready' }
|
||
}));
|
||
socket.emit('message', JSON.stringify({
|
||
id: 'channel-attachment-rejected',
|
||
type: 'event',
|
||
from: 'channel:wechat:user-1',
|
||
conversation_id: 'conversation-attachment-rejected',
|
||
payload: { text: '名单附件补充消息' }
|
||
}));
|
||
|
||
await waitFor(() => resultDelivered, 3_000);
|
||
assert.equal(acceptedDelivered, true);
|
||
assert.equal(resultDelivered, true);
|
||
assert.equal(parseQueueCalls, 0);
|
||
assert.deepEqual(resultPayload, {
|
||
event: 'task.result',
|
||
status: 'failed',
|
||
task_id: 'TASK-ATTACHMENT-REJECTED',
|
||
text: rejectionMessage
|
||
});
|
||
assert.equal(socket.sent.length, 2);
|
||
assert.equal((socket.sent[0].payload as Record<string, unknown>).event, 'task.progress');
|
||
assert.equal(
|
||
(socket.sent[0].payload as Record<string, unknown>).text,
|
||
AGENTBUS_ROSTER_ATTACHMENT_RECEIVED_TEXT
|
||
);
|
||
assert.equal(
|
||
(socket.sent[0].payload as Record<string, unknown>)[AGENTBUS_FINAL_REPLY_ELIGIBLE_FIELD],
|
||
undefined
|
||
);
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).event, 'task.result');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).status, 'failed');
|
||
assert.equal((socket.sent[1].payload as Record<string, unknown>).text, rejectionMessage);
|
||
assert.equal(socket.sent[1].reply_to, 'channel-attachment-rejected');
|
||
assert.equal(socket.sent[1].conversation_id, 'conversation-attachment-rejected');
|
||
assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'attachment_rejection_reply_requested'
|
||
&& entry.metadata.error_code === 'roster_workbook_conversion_failed'));
|
||
});
|