merge: integrate attachment correlation and server diagnostics

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,85 @@ 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'
+    && /^[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('listener reload stop does not overwrite the channel status as disabled', () => {

control-plane/test/diagnostics.test.ts

@@ -0,0 +1,180 @@
+import assert from 'node:assert/strict';
+import { Writable } from 'node:stream';
+import test from 'node:test';
+import { loadConfig } from '../src/config.js';
+import {
+  diagnosticDurationMs,
+  diagnosticError,
+  diagnosticRequestPath,
+  normalizeRequestId
+} from '../src/diagnostics.js';
+import { buildServer, createControlPlaneLogger } from '../src/server.js';
+import { TaskService, type TaskEvent } from '../src/task-service.js';
+
+function testConfig(overrides: NodeJS.ProcessEnv = {}) {
+  return loadConfig({
+    NODE_ENV: 'test',
+    FIELD_ENCRYPTION_KEY: Buffer.alloc(32, 41).toString('base64'),
+    DATABASE_URL: 'postgresql://invalid:invalid@127.0.0.1:1/invalid',
+    ...overrides
+  });
+}
+
+function logCollector(): { destination: Writable; records: Array<Record<string, unknown>> } {
+  const chunks: string[] = [];
+  const records: Array<Record<string, unknown>> = [];
+  const destination = new Writable({
+    write(chunk, _encoding, callback) {
+      chunks.push(String(chunk));
+      const lines = chunks.join('').split('\n');
+      chunks.length = 0;
+      const remainder = lines.pop() || '';
+      if (remainder) chunks.push(remainder);
+      for (const line of lines) {
+        if (line.trim()) records.push(JSON.parse(line) as Record<string, unknown>);
+      }
+      callback();
+    }
+  });
+  return { destination, records };
+}
+
+test('diagnostic errors preserve code and stack location without leaking the error message', () => {
+  const secret = 'customer-secret-value';
+  const error = new Error(`download https://user:password@example.test/list.xlsx?token=${secret}`) as Error & {
+    code: string;
+    errno: number;
+    syscall: string;
+  };
+  error.code = 'ECONNRESET';
+  error.errno = -54;
+  error.syscall = 'read';
+  const metadata = diagnosticError(error, 'download_failed');
+  const serialized = JSON.stringify(metadata);
+  assert.equal(metadata.error_code, 'ECONNRESET');
+  assert.equal(metadata.error_name, 'Error');
+  assert.match(String(metadata.error_fingerprint), /^[a-f0-9]{24}$/u);
+  assert.equal(metadata.error_errno, -54);
+  assert.equal(metadata.error_syscall, 'read');
+  assert.doesNotMatch(serialized, /customer-secret-value|password|example\.test|list\.xlsx/u);
+});
+
+test('diagnostic request identifiers and paths are stable and query-safe', () => {
+  assert.equal(normalizeRequestId('request:wechat:12345678'), 'request:wechat:12345678');
+  const generated = normalizeRequestId('token=must-not-be-used');
+  assert.match(generated, /^[a-f0-9-]{36}$/u);
+  assert.doesNotMatch(generated, /token/u);
+  assert.equal(diagnosticRequestPath('/api/tasks/TASK-1?token=secret#fragment'), '/api/tasks/TASK-1');
+  const startedAt = process.hrtime.bigint() - 2_000_000n;
+  assert.ok(diagnosticDurationMs(startedAt) >= 1);
+});
+
+test('control-plane logger redacts credentials and includes deployment identity', async () => {
+  const { destination, records } = logCollector();
+  const logger = createControlPlaneLogger(testConfig({ DEPLOYMENT_REVISION: 'commit-2360506' }), destination);
+  logger.info({
+    diagnostic_event: 'test.redaction',
+    password: 'root-secret',
+    nested: { token: 'nested-secret' },
+    safe_value: 'visible'
+  }, 'diagnostic test');
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.equal(records.length, 1);
+  assert.equal(records[0].service, 'ltjt-control-plane');
+  assert.equal(records[0].deployment_revision, 'commit-2360506');
+  assert.equal(records[0].password, '[REDACTED]');
+  assert.deepEqual(records[0].nested, { token: '[REDACTED]' });
+  assert.equal(records[0].safe_value, 'visible');
+});
+
+test('HTTP diagnostics reuse one request ID and never log query values', async () => {
+  const { destination, records } = logCollector();
+  const { app } = await buildServer({
+    config: testConfig(),
+    startParserLoop: false,
+    loggerDestination: destination,
+    parser: {
+      async parse() {
+        return { blockers: ['test parser'] };
+      },
+      async checkConnection() {
+        return { ok: false, configured: false };
+      }
+    }
+  });
+  const response = await app.inject({
+    method: 'GET',
+    url: '/health/live?token=query-secret',
+    headers: { 'x-request-id': 'request:test:12345678' }
+  });
+  assert.equal(response.statusCode, 200);
+  assert.equal(response.headers['x-request-id'], 'request:test:12345678');
+  await app.close();
+  await new Promise((resolve) => setImmediate(resolve));
+
+  const started = records.find((record) => record.diagnostic_event === 'http.request.started');
+  const completed = records.find((record) => record.diagnostic_event === 'http.request.completed');
+  assert.equal(started?.request_id, 'request:test:12345678');
+  assert.equal(completed?.request_id, 'request:test:12345678');
+  assert.equal(started?.path, '/health/live');
+  assert.equal(completed?.path, '/health/live');
+  assert.equal(completed?.status_code, 200);
+  assert.doesNotMatch(JSON.stringify(records), /query-secret/u);
+});
+
+test('task state and audit diagnostics record keys but never business values', async () => {
+  const logs: Array<Record<string, unknown>> = [];
+  const service = new TaskService(
+    testConfig(),
+    {} as never,
+    {
+      info(metadata) { logs.push(metadata); },
+      warn(metadata) { logs.push(metadata); },
+      error(metadata) { logs.push(metadata); }
+    }
+  ) as unknown as {
+    notify(event: TaskEvent): void;
+    audit(
+      client: { query: (...args: unknown[]) => Promise<{ rowCount: number }> },
+      context: { organizationId: string; userId: string; requestId: string },
+      eventType: string,
+      entityId: string,
+      metadata: Record<string, unknown>
+    ): Promise<void>;
+  };
+  service.notify({
+    id: 8,
+    organization_id: 'org-secret',
+    task_id: 'TASK-DIAGNOSTIC-1',
+    status: 'awaiting_attachment',
+    stage: 'input',
+    message: 'customer-secret-message',
+    payload: { customer_name: 'customer-secret-value', row_count: 12 },
+    created_at: new Date().toISOString()
+  });
+  await service.audit(
+    { async query() { return { rowCount: 1 }; } },
+    { organizationId: 'org-secret', userId: 'user-secret', requestId: 'request:audit:12345678' },
+    'task.passenger_roster_attachment_rejected',
+    'TASK-DIAGNOSTIC-1',
+    { customer_name: 'customer-secret-value', error_code: 'roster_file_invalid' }
+  );
+
+  assert.ok(logs.some((log) => log.diagnostic_event === 'task.state.emitted'
+    && Array.isArray(log.payload_keys)
+    && (log.payload_keys as string[]).includes('customer_name')));
+  assert.ok(logs.some((log) => log.diagnostic_event === 'audit.event.staged'
+    && log.request_id === 'request:audit:12345678'));
+  assert.doesNotMatch(JSON.stringify(logs), /customer-secret-value|customer-secret-message|org-secret|user-secret/u);
+});
+
+test('production configuration rejects raw AgentBus payload logging', () => {
+  assert.throws(
+    () => testConfig({ NODE_ENV: 'production', AGENTBUS_LOG_PAYLOADS: 'true' }),
+    /AGENTBUS_LOG_PAYLOADS must remain false in production/u
+  );
+  assert.throws(
+    () => testConfig({ LOG_LEVEL: 'verbose' }),
+    /Invalid enum value/u
+  );
+});

control-plane/test/input-attachment.test.ts

@@ -5,6 +5,7 @@ import { sha256Bytes } from '../src/crypto.js';
 import {
   InputAttachmentError,
   decodeInlineInputAttachment,
+  downloadAgentBusInputAttachment,
   isPrivateOrReservedIp,
   parseAgentBusInputAttachment,
   validateAgentBusAttachmentUrl
@@ -80,6 +81,23 @@ test('AgentBus attachment metadata is normalized without exposing URL credential
   assert.equal(reference.sha256, 'a'.repeat(64));
 });
 
+test('AgentBus attachment diagnostics expose stages and codes without URL data', async () => {
+  const events: Array<{ event: string; metadata: Record<string, unknown> }> = [];
+  await assert.rejects(
+    () => downloadAgentBusInputAttachment({
+      name: 'synthetic.xls',
+      contentType: 'application/vnd.ms-excel',
+      size: 128,
+      url: 'https://127.0.0.1/private-roster.xls?token=secret'
+    }, 1_000, (event, metadata) => events.push({ event, metadata })),
+    (error: unknown) => error instanceof InputAttachmentError
+      && error.code === 'roster_attachment_url_unsafe'
+  );
+  assert.deepEqual(events.map((event) => event.event), ['download_started', 'download_failed']);
+  assert.equal(events[1].metadata.error_code, 'roster_attachment_url_unsafe');
+  assert.doesNotMatch(JSON.stringify(events), /127\.0\.0\.1|private-roster|token|secret/u);
+});
+
 test('input attachment migration stores only encrypted normalized data and waiting-task index', async () => {
   const sql = await readFile(new URL('../migrations/014_task_input_attachments.sql', import.meta.url), 'utf8');
   assert.match(sql, /CREATE TABLE IF NOT EXISTS task_input_attachments/);