diff --git a/.project-docs/30-worklog/tasks/20260831-fix-agentbus-attachment-reply-6f2c9a41.md b/.project-docs/30-worklog/tasks/20260831-fix-agentbus-attachment-reply-6f2c9a41.md new file mode 100644 index 0000000..1bfd356 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260831-fix-agentbus-attachment-reply-6f2c9a41.md @@ -0,0 +1,55 @@ +# Task: Fix AgentBus attachment rejection reply + +## Identity + +- Task ID: 20260831-fix-agentbus-attachment-reply-6f2c9a41 +- Mode: Feature +- Branch: main +- Worktree: /Users/inmanx/Documents/lwltAPI +- Base commit: d7a821da1627548186e8834350e09a66c44140d7 +- Owner: codex +- Status: Planning + +## Scope + +- Diagnose why a structured AgentBus roster attachment that returns `input_attachment.status = rejected` leaves the original task waiting but sends no rejection feedback to the channel user. +- Add an attempt-level failed result for rejected roster attachments in both legacy in-memory and database-backed durable AgentBus delivery modes. +- Preserve the original task's `awaiting_attachment` state, avoid scheduling parsing for rejected bytes, and add focused regression coverage. +- Run the full repository gates, commit the bounded fix on `main`, and push it non-forced to `origin/main`. + +## Intent And Constraints + +- Keep the existing fail-closed roster workflow: a rejected workbook must not enter Program parsing or ERP execution, and the same waiting task must remain available for a corrected attachment. +- Return the existing bounded `input_attachment.message`; do not expose attachment URL, hostname, IP, file name, workbook bytes, cell values, raw converter output, or unsafe error details. +- Preserve the durable accepted/result outbox, inbound frame reply routing, conversation correlation, and idempotent delivery keys. +- Do not deploy, restart services, mutate Kubernetes, access ERP, read secrets, retry the live task, or send an external message. + +## Outcome + +- Confirmed the feedback gap was not an AgentBus transport exception. Workbook rejection is represented as a successful `TaskMessageResult` while the business task intentionally remains `awaiting_attachment`; therefore the generic exception reply did not run and the terminal-task finalizer correctly skipped the still-waiting task. +- Added an explicit attachment-attempt rejection branch immediately after task ingestion. It emits a failed `task.result` through the database outbox for durable channels or the existing pending-reply path for legacy listeners, then returns without requesting the parse queue. +- Kept the task state and reusable waiting workflow unchanged. The channel receives the existing safe workbook rejection text and can submit a corrected `.xls/.xlsx` attachment in a later message. +- Added direct-listener and durable-channel regressions covering reply routing, failure status/text, retained waiting state, and zero parse-queue requests. + +## Verification + +- Focused AgentBus tests: 16/16 passed. +- `node --run check:repo`: 10/10 passed. +- `node --run check`: passed. +- `node --run test:control-plane`: 139/139 passed. +- `node --run test:legacy`: 256/256 passed. +- `node --run build`: passed. +- `check_project_docs.py`: passed. +- `check_doc_drift.py --task-id 20260831-fix-agentbus-attachment-reply-6f2c9a41`: passed. +- `git diff --check`: passed. + +## Follow-ups + +- Deploy/restart and verify the corrected reply with a real WeChat attachment only under separate authorization. + +## Promotion Candidates + +- Target canonical AgentBus/data-flow memory during a later Integration task. +- Proposal: roster workbook rejection is an attachment-attempt failure that must create an immediate failed AgentBus result even though the reusable business task remains nonterminal in `awaiting_attachment`. +- Evidence: `control-plane/src/agentbus.ts`, `control-plane/test/agentbus.test.ts`, and the focused 16/16 AgentBus regression. +- Human confirmation required: no for the bounded reply behavior; deployment and live retry remain separately authorized. diff --git a/control-plane/src/agentbus.ts b/control-plane/src/agentbus.ts index 80393ea..1fb4015 100644 --- a/control-plane/src/agentbus.ts +++ b/control-plane/src/agentbus.ts @@ -1019,6 +1019,30 @@ export class AgentBusListener { attached: result.attached, created: result.created }, 'AgentBus message ingested into task service'); + const rejectedAttachment = result.input_attachment?.status === 'rejected' + ? result.input_attachment + : null; + if (rejectedAttachment) { + this.logger.warn({ + agentbus_event: 'attachment_rejection_reply_requested', + inbound_frame_id: taskId, + task_id: result.task.task_id, + task_status: result.task.status, + error_code: rejectedAttachment.error_code || 'roster_workbook_processing_failed' + }, 'AgentBus roster attachment rejection reply requested'); + if (this.durableDeliveryEnabled() && this.channel && this.tasks.enqueueAgentBusResult) { + await this.tasks.enqueueAgentBusResult.call(this.tasks, this.channel.id, taskId, { + event: 'task.result', + status: 'failed', + task_id: result.task.task_id, + text: rejectedAttachment.message.slice(0, 20_000) + }); + await this.flushDurableDeliveries(); + } else { + this.queueFinalReply(frame, 'failed', rejectedAttachment.message); + } + return; + } this.logger.info({ agentbus_event: 'parse_queue_requested', inbound_frame_id: taskId, diff --git a/control-plane/test/agentbus.test.ts b/control-plane/test/agentbus.test.ts index 8e61a9e..17dea91 100644 --- a/control-plane/test/agentbus.test.ts +++ b/control-plane/test/agentbus.test.ts @@ -779,6 +779,73 @@ test('WeChat attachment placeholder without file metadata fails closed before ta 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).event, 'task.progress'); + assert.equal((socket.sent[1].payload as Record).event, 'task.result'); + assert.equal((socket.sent[1].payload as Record).status, 'failed'); + assert.equal((socket.sent[1].payload as Record).text, rejectionMessage); + assert.equal(socket.sent[1].reply_to, 'wechat-attachment-rejected'); + assert.equal(socket.sent[1].conversation_id, 'conversation-attachment-rejected'); +}); + test('listener reload stop does not overwrite the channel status as disabled', () => { const statuses: string[] = []; const listener = new AgentBusListener({ @@ -1049,3 +1116,148 @@ test('durable channel listener persists route and resends accepted/result delive assert.deepEqual(deliveryState.failures, []); listener.stop(); }); + +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 = {}; + 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: { event: 'task.progress', status: 'accepted', text: '已受理,正在处理。' }, + attempt_count: 1 + }; + const result: AgentBusDelivery = { + ...accepted, + id: 'delivery-attachment-rejected-result', + delivery_kind: 'result', + payload: resultPayload + }; + const logs: Array<{ level: string; metadata: Record; 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' + }, + 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).event, 'task.progress'); + assert.equal((socket.sent[1].payload as Record).event, 'task.result'); + assert.equal((socket.sent[1].payload as Record).status, 'failed'); + assert.equal((socket.sent[1].payload as Record).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')); +});