fix: preserve WeChat attachment task context

control-plane/README.md

@@ -20,7 +20,8 @@
 - 插件回执必须携带服务端 `execution_id` 和领取连接；完成、阻断及待回查状态不可被后续 `running` 回执覆盖。执行租约过期会进入待回查，不会重新入队。
 - Chrome 插件最低兼容版本由正式操作台与根目录发布清单共同门禁。插件包含分段前门禁、ERP 只读唯一解析、严格写前门禁、当前窄生命周期适配、写入前 `write_started` 持久化和写后回查；扩展后台重启后也不会重跑同一任务。本次双轨解析不修改插件执行契约或扩展版本。
 - 渠道 Adapter 由 AgentBus 负责；控制平面只作为 AgentBus Bot 连接到文档中的 WebSocket，不实现微信、个人微信或其他渠道协议。
-- 微信桥接器把正文放在严格的 `New WeChat message` / `Conversation:` / `Text:` 三行传输信封中；AgentBus listener 会在任务快照前只解开这一已知信封，并把 `Text:` 同行值及后续行作为业务正文。近似、缺失字段或空正文的包装保持原文，不能通过忽略任意未知标签来绕过 Program parser 的失败关闭。
+- 微信桥接器把正文放在严格的 `New WeChat message` / `Conversation:` / `Text:` 三行传输信封中；AgentBus listener 会在任务快照前只解开这一已知信封，把 `Text:` 同行值及后续行作为业务正文，并在帧没有显式 `conversation_id` 时使用信封中的 `Conversation:` 值。显式字段仍优先；近似、缺失字段或空正文的包装保持原文，不能通过忽略任意未知标签来绕过 Program parser 的失败关闭。
+- 微信侧的 `[WeChat attachment: 文件名]` 只是一段传输占位文字，不代表控制面已经收到文件。若同一帧没有符合契约的 `payload.attachments[]`，listener 会在进入任务服务前失败关闭、保留原名单任务的等待状态，并返回“附件内容未传到平台”；不会把占位文字创建成新业务任务。附件元数据、HTTPS URL、DNS、大小或摘要校验失败时返回对应的安全摘要，仍不回显 URL、文件字节或名单内容。
 - AgentBus 入站消息会复用 `TaskService` 的任务/会话/解析队列，解析完成后通过同一 WebSocket 返回一次 `task.result`。组织级“全自动化”关闭时，手工与 AgentBus 新任务都要求管理员确认；开启后，两种来源的合法解析结果都自动进入 ERP 队列，不再按来源或创建、名单、安排、修改、取消/恢复、导出等业务类型保留人工例外。操作台在 EventSource 建连/重连、30 秒后台刷新以及页面重新可见或聚焦时重新读取数据库权威开关，避免后台变更后按钮仍显示旧值。缺资料、解析失败、歧义、插件校验失败或 ERP 回查不确定时仍会停止，不会绕过校验或重试不确定写入。
 - 一个组织可以维护多个“用户渠道”。渠道代表外部 AgentBus 用户身份，不等同于平台管理员账号；管理员在独立根路径 `/channels` 的“AgentBus 渠道”目录中创建、停用、启用、轮换或删除渠道。删除会停止对应 listener、移除服务端保存的 key 和该渠道尚存的持久化回执；历史任务本体保留，其 `channel_id` 按数据库契约置空。每个渠道独立保存加密后的 AgentBus key，并建立独立 WebSocket listener；列表和日志都不会回显 key。`AGENTBUS_WS_URL`、重连策略和客户端类型仍是全局连接配置，`AGENTBUS_BOT_ADDRESS` 可作为渠道 bot address 的默认值。仍由完整旧环境变量托管的兼容渠道会自动重建，必须先移除环境配置并重启服务，才允许删除其数据库记录。
 - `/history` 支持逐条彻底删除，以及勾选当前页后批量删除。单条使用 `DELETE /api/tasks/:taskId`，批量使用 `POST /api/tasks/bulk-delete`（一次 1–100 个且不能重复）；两者都要求管理员 mutation 会话、same-origin 与 CSRF 门禁。批量删除会在同一数据库事务中按组织锁定并核对全部目标，任一任务不存在或不属于当前组织时整批回滚；成功后任务、生命周期、尝试、会话、附件元数据和 AgentBus 回执按外键级联删除，task-scoped audit/outbox 行显式删除。OSS 附件对象在事务提交后使用已冻结的 storage key 逐一清理，清理异常写入服务日志。对已确认或正在插件流程中的任务，页面会明确警告：停止插件只是尽力而为，已经发生的 ERP 写入及已投递到外部渠道的副本不会因删除平台历史而撤回。

control-plane/src/agentbus.ts

@@ -14,6 +14,7 @@ import type {
   AgentBusAttachmentContent
 } from './task-service.js';
 import {
+  InputAttachmentError,
   downloadAgentBusInputAttachment,
   parseAgentBusInputAttachment
 } from './input-attachment.js';
@@ -175,6 +176,31 @@ function text(value: unknown): string {
 const WECHAT_MESSAGE_HEADER = 'New WeChat message';
 const WECHAT_CONVERSATION_LINE = /^Conversation\s*[:：]\s*(.+)$/i;
 const WECHAT_TEXT_LINE = /^Text\s*[:：]\s*(.*)$/i;
+const WECHAT_ATTACHMENT_PLACEHOLDER = /^\[WeChat attachment:\s*[^\]\r\n]+\]$/iu;
+
+interface WeChatTransportEnvelope {
+  businessText: string;
+  conversationId: string;
+  attachmentPlaceholder: boolean;
+}
+
+function parseWeChatTransportEnvelope(value: unknown): WeChatTransportEnvelope | null {
+  const original = text(value);
+  if (!original) return null;
+  const lines = original.split(/\r?\n/);
+  if (lines.length < 3 || lines[0].trim() !== WECHAT_MESSAGE_HEADER) return null;
+  const conversation = WECHAT_CONVERSATION_LINE.exec(lines[1]?.trim() || '');
+  const firstTextLine = WECHAT_TEXT_LINE.exec(lines[2]?.trim() || '');
+  const conversationId = text(conversation?.[1]);
+  if (!conversationId || !firstTextLine) return null;
+  const businessText = [firstTextLine[1], ...lines.slice(3)].join('\n').trim();
+  if (!businessText) return null;
+  return {
+    businessText,
+    conversationId: conversationId.slice(0, 200),
+    attachmentPlaceholder: WECHAT_ATTACHMENT_PLACEHOLDER.test(businessText)
+  };
+}
 
 /**
  * The WeChat-to-AgentBus bridge places the actual business instruction inside
@@ -183,15 +209,7 @@ const WECHAT_TEXT_LINE = /^Text\s*[:：]\s*(.*)$/i;
  * bodies stay unchanged so transport adaptation cannot hide business input.
  */
 export function extractAgentBusBusinessText(value: unknown): string {
-  const original = text(value);
-  if (!original) return original;
-  const lines = original.split(/\r?\n/);
-  if (lines.length < 3 || lines[0].trim() !== WECHAT_MESSAGE_HEADER) return original;
-  const conversation = WECHAT_CONVERSATION_LINE.exec(lines[1]?.trim() || '');
-  const firstTextLine = WECHAT_TEXT_LINE.exec(lines[2]?.trim() || '');
-  if (!conversation || !text(conversation[1]) || !firstTextLine) return original;
-  const businessText = [firstTextLine[1], ...lines.slice(3)].join('\n').trim();
-  return businessText || original;
+  return parseWeChatTransportEnvelope(value)?.businessText || text(value);
 }
 
 function jsonObject(value: unknown): Record<string, unknown> {
@@ -352,6 +370,9 @@ export function parseAgentBusFrame(raw: RawData | string): AgentBusFrame | null
     if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
     const frame = value as Record<string, unknown>;
     const payload = jsonObject(frame.payload);
+    const envelope = parseWeChatTransportEnvelope(payload.text);
+    const conversationId = text(frame.conversation_id) || text(payload.conversation_id)
+      || envelope?.conversationId;
     return {
       ...frame,
       id: text(frame.id) || undefined,
@@ -360,7 +381,7 @@ export function parseAgentBusFrame(raw: RawData | string): AgentBusFrame | null
       to: text(frame.to) || undefined,
       session_id: text(frame.session_id) || undefined,
       epoch: Number.isFinite(Number(frame.epoch)) ? Number(frame.epoch) : undefined,
-      conversation_id: text(frame.conversation_id) || undefined,
+      conversation_id: conversationId || undefined,
       reply_to: text(frame.reply_to) || undefined,
       payload: {
         ...payload,
@@ -876,12 +897,18 @@ export class AgentBusListener {
     this.logFrame('inbound_task_accepted', frame, 'AgentBus inbound task accepted');
     const processing = this.processInboundTask(frame)
       .catch((error) => {
+        const attachmentError = error instanceof InputAttachmentError ? error : null;
         this.logger.error({
           agentbus_event: 'task_processing_failed',
           ...frameLogData(frame, this.config.AGENTBUS_LOG_PAYLOADS),
+          error_code: attachmentError?.code || null,
           error: error instanceof Error ? error.message : String(error)
         }, 'AgentBus task processing failed');
-        this.queueFinalReply(frame, 'failed', '任务处理失败，请稍后重试。');
+        this.queueFinalReply(
+          frame,
+          'failed',
+          attachmentError?.message || '任务处理失败，请稍后重试。'
+        );
       })
       .finally(() => {
         this.inFlightTaskIds.delete(taskId);
@@ -893,12 +920,19 @@ export class AgentBusListener {
   private async processInboundTask(frame: AgentBusFrame): Promise<void> {
     const taskId = text(frame.id);
     const rawMessage = text(frame.payload?.text);
-    const message = extractAgentBusBusinessText(rawMessage);
+    const transportEnvelope = parseWeChatTransportEnvelope(rawMessage);
+    const message = transportEnvelope?.businessText || extractAgentBusBusinessText(rawMessage);
     const conversationId = frameConversationId(frame);
     const rawAttachments = Array.isArray(frame.payload?.attachments) ? frame.payload!.attachments! : [];
     if (rawAttachments.length > 1) {
       throw new Error('AgentBus 名单业务每次只能发送一个附件。');
     }
+    if (!rawAttachments.length && transportEnvelope?.attachmentPlaceholder) {
+      throw new InputAttachmentError(
+        'roster_attachment_metadata_missing',
+        '附件内容未传到平台，原任务仍在等待附件。请检查微信桥接器的文件转发后重新发送。'
+      );
+    }
     const attachments = rawAttachments.length
       ? [await downloadAgentBusInputAttachment(
           parseAgentBusInputAttachment(rawAttachments[0], this.config.ARTIFACT_MAX_BYTES),

control-plane/test/agentbus.test.ts

@@ -108,6 +108,13 @@ class FakeSocket {
   }
 }
 
+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',
@@ -188,6 +195,21 @@ test('WeChat transport envelope exposes only its business text to the global par
   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}`;
@@ -510,7 +532,7 @@ test('AgentBus attachment references omit archived visitor XLS and keep only XLS
   }]);
 });
 
-test('AgentBus listener connects with the documented Authorization header and returns one final result', async () => {
+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 = '';
@@ -608,6 +630,7 @@ test('AgentBus listener connects with the documented Authorization header and re
       }
     }
   });
+  t.after(() => listener.stop());
 
   listener.start();
   assert.match(capturedUrl, /[?&]ready=1/);
@@ -645,11 +668,11 @@ test('AgentBus listener connects with the documented Authorization header and re
     payload: { text: wrappedBusinessText, reply_policy: { progress: true } }
   }));
 
-  await new Promise((resolve) => setTimeout(resolve, 50));
+  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, 'agentbus:channel:wechat:user-1');
+  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');
@@ -675,7 +698,80 @@ test('AgentBus listener connects with the documented Authorization header and re
   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/);
-  listener.stop();
+});
+
+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'));
+  assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'outbound_result_sent'));
 });
 
 test('listener reload stop does not overwrite the channel status as disabled', () => {