fix: return AgentBus attachment rejections

.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.

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,

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<string, unknown>).event, 'task.progress');
+  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('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<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: { 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<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'
+    },
+    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[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'));
+});