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