From 23605066076f6c7c463564281379cc28bb27ec06 Mon Sep 17 00:00:00 2001 From: inman Date: Sun, 30 Aug 2026 15:41:55 +0800 Subject: [PATCH] fix: preserve WeChat attachment task context --- ...30-wechat-attachment-correlation-9f3a2c.md | 69 ++++++++++++ agent设计规范/agentbus-reply-contract.md | 2 + control-plane/README.md | 3 +- control-plane/src/agentbus.ts | 58 ++++++++-- control-plane/test/agentbus.test.ts | 104 +++++++++++++++++- 5 files changed, 219 insertions(+), 17 deletions(-) create mode 100644 .project-docs/30-worklog/tasks/20260830-wechat-attachment-correlation-9f3a2c.md diff --git a/.project-docs/30-worklog/tasks/20260830-wechat-attachment-correlation-9f3a2c.md b/.project-docs/30-worklog/tasks/20260830-wechat-attachment-correlation-9f3a2c.md new file mode 100644 index 0000000..36bb3c1 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260830-wechat-attachment-correlation-9f3a2c.md @@ -0,0 +1,69 @@ +# Task: Fix WeChat attachment task correlation + +## Identity + +- Task ID: 20260830-wechat-attachment-correlation-9f3a2c +- Mode: Feature +- Branch: codex/20260830-wechat-attachment-correlation-9f3a2c-wechat-attachment-correlation +- Worktree: /Users/inmanx/Documents/lwltAPI-worktrees/20260830-wechat-attachment-correlation-9f3a2c +- Base commit: e7aa58a203f9c05850a3d10e681b8800b856ee12 +- Owner: codex +- Status: Ready for Integration + +## Scope + +- Diagnose the deployed WeChat/AgentBus roster-attachment failure shown at 2026-08-30 15:19. +- Keep an attachment-only WeChat message from entering the ordinary new-task path when the bridge supplied only its text placeholder. +- Use the exact WeChat envelope `Conversation:` value for task correlation when no explicit AgentBus conversation field is present. +- Return safe, actionable attachment-ingress errors instead of collapsing metadata, URL, DNS, download, size, or digest failures into the generic task-processing message. +- Add focused regression coverage and update the active AgentBus transport documentation. + +## Intent And Constraints + +- Preserve the existing Program-only roster workflow: one `.xls/.xlsx` attachment must become `payload.attachments[]` with a bounded public HTTPS URL before it can resume an `awaiting_attachment` task. +- Do not weaken HTTPS, credential, redirect, DNS, private/reserved-network, size, or SHA-256 checks; absent file bytes must fail closed and leave the original task waiting. +- Do not change parser, operation, Schema, mapping, ERP, Chrome-extension, or release-artifact behavior. +- Do not read local secrets, deploy, restart services, mutate live tasks, access ERP, or send external messages. +- Treat the screenshots as evidence only. The raw production frame and server error code are unavailable, so distinguish the two supported failure branches in logs and user replies instead of asserting one without evidence. + +## Plan + +1. Add strict parsing for the observed WeChat transport envelope, including its conversation identifier and attachment-only placeholder. +2. Reject placeholder-only attachment messages before `TaskService.ingestMessage`, and surface the safe `InputAttachmentError` reason and code. +3. Add listener tests proving conversation correlation, no accidental task creation, actionable failure text, and unchanged strict-envelope behavior. +4. Update the AgentBus contract/readme, then run targeted and full repository verification. + +## Outcome + +- Confirmed the failure had two distinct ingress problems before task selection: the strict WeChat transport envelope exposed a `Conversation:` value that was not promoted to the AgentBus frame's conversation key, while an attachment card could arrive as placeholder text without the required `payload.attachments[]` file metadata. +- Added strict envelope parsing that preserves explicit AgentBus `conversation_id` precedence and otherwise uses the observed WeChat `Conversation:` value. A real structured attachment can therefore select the unique `awaiting_attachment` task in the same channel and conversation through the existing `TaskService` path. +- Added a fail-closed guard for `[WeChat attachment: ...]` placeholder-only frames. They now stop before `TaskService.ingestMessage`, do not create a second business task, leave the original roster task waiting, and return an actionable safe message that the file content never reached the platform. +- Preserved all existing attachment download controls. Pre-ingest `InputAttachmentError` failures now return their predefined safe summary and log a bounded error code instead of being collapsed into the generic task-processing failure; URLs, file bytes, and roster values remain absent from replies and logs. +- Updated the active AgentBus transport contract and control-plane README. No parser, Schema, mapping, ERP, Chrome-extension, release artifact, deployment, service process, live task, or external system was changed. + +## Verification + +- Focused AgentBus listener regression: 14/14 passed, including envelope conversation precedence, placeholder-only fail-closed behavior, zero task-ingestion calls, bounded error logging, and listener teardown. +- `node --run check:repo`: 9/9 passed. +- `node --run check`: passed. +- `node --run test:control-plane`: 128/128 passed. +- `node --run test:legacy`: 248/248 passed. +- `node --run build`: passed. +- `check_project_docs.py`: passed. +- `check_doc_drift.py --task-id 20260830-wechat-attachment-correlation-9f3a2c`: passed. +- `git diff --check`: passed. + +## Follow-ups + +- The external WeChat bridge must deliver each workbook as one real `payload.attachments[]` entry with the documented name, size/type/hash metadata and a control-plane-reachable public HTTPS URL. Placeholder text alone cannot supply file bytes and is intentionally not converted into an attachment. +- The raw 2026-08-30 production frame and its original server-side exception were not available. After this branch is integrated and deployed under separate authorization, observe the bounded `error_code` to distinguish missing metadata from URL, DNS, download, size, or digest failure without inspecting sensitive payloads. +- Integration, deployment, restart, and a live WeChat retry remain separate authorized actions; this Feature task performs none of them. + +## Promotion Candidates + +- Target: `.project-docs/20-architecture/data-flow.md` and `.project-docs/40-domain/business-rules.md`. + Proposal: record that a strict WeChat envelope may supply the AgentBus conversation fallback, but a roster attachment exists only when the frame carries a validated `payload.attachments[]` entry; placeholder text must never create a new task and must leave the prior roster task awaiting its file. + Evidence: `control-plane/src/agentbus.ts`, `control-plane/test/agentbus.test.ts`, `control-plane/README.md`, and `agent设计规范/agentbus-reply-contract.md` on this task branch; focused 14/14, control-plane 128/128, and legacy 248/248 tests passed. + Future impact: future bridge or channel adapters must preserve stable conversation identity and send file metadata/URL separately from user-visible attachment placeholder text; operational diagnosis should use bounded attachment error codes rather than payload logging. + Semantic conflicts: none found in active contracts; the proposal makes the existing structured-attachment requirement and task-correlation boundary explicit. + Human confirmation: not required for the rule itself because it preserves the existing fail-closed attachment contract, but Integration mode is required before writing canonical project memory. diff --git a/agent设计规范/agentbus-reply-contract.md b/agent设计规范/agentbus-reply-contract.md index 460f1f3..8ea66bf 100644 --- a/agent设计规范/agentbus-reply-contract.md +++ b/agent设计规范/agentbus-reply-contract.md @@ -17,6 +17,8 @@ 名单入站 `payload.attachments` 最多一个，使用 `name`、`content_type`、`size`、可选 `sha256` 和不含凭据的 HTTPS `url`。控制面拒绝 HTTP、URL 用户名密码、私网/保留地址、DNS 解析到私网、超过大小限制、大小或 SHA-256 不一致以及超过两次的重定向；下载时固定到已校验的公网 IP。附件 URL 和原始字节不写入任务 JSONB 或日志。 +微信传输信封的 `Conversation:` 在帧未显式提供 `conversation_id` 时作为会话键，显式字段优先。`[WeChat attachment: 文件名]` 只是桥接器的文字占位符：没有同时提供 `payload.attachments[]` 时，控制面不得把它作为新业务消息创建任务，也不得假装已收到文件；应保留原任务的 `awaiting_attachment` 状态并提示附件内容尚未传到平台。附件元数据或安全下载校验失败时只返回预定义的安全摘要，不回显 URL、文件内容或名单值。 + 附件通过程序模板校验后，任务从 `awaiting_attachment` 转为 `parse_queued`；校验失败则保留等待状态并只返回安全错误码和行列位置，不回显名单值。原始工作簿不持久化，只暂存加密 canonical TSV 供 Program Parser 使用。 ## 失败 diff --git a/control-plane/README.md b/control-plane/README.md index 68178e8..e5f30ad 100644 --- a/control-plane/README.md +++ b/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 写入及已投递到外部渠道的副本不会因删除平台历史而撤回。 diff --git a/control-plane/src/agentbus.ts b/control-plane/src/agentbus.ts index 040e952..f4fc556 100644 --- a/control-plane/src/agentbus.ts +++ b/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 { @@ -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; 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 { 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), diff --git a/control-plane/test/agentbus.test.ts b/control-plane/test/agentbus.test.ts index ef241f2..9ae54cb 100644 --- a/control-plane/test/agentbus.test.ts +++ b/control-plane/test/agentbus.test.ts @@ -108,6 +108,13 @@ class FakeSocket { } } +async function waitFor(predicate: () => boolean, timeoutMs = 1_000): Promise { + 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).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; 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).event, 'task.result'); + assert.equal((socket.sent[0].payload as Record).status, 'failed'); + assert.equal( + (socket.sent[0].payload as Record).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', () => {