From 23605066076f6c7c463564281379cc28bb27ec06 Mon Sep 17 00:00:00 2001 From: inman Date: Sun, 30 Aug 2026 15:41:55 +0800 Subject: [PATCH 1/3] 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', () => { From a3963ad0f629818edfa921261acfeb28dde9c9dd Mon Sep 17 00:00:00 2001 From: inman Date: Sun, 30 Aug 2026 16:01:42 +0800 Subject: [PATCH 2/3] feat: add privacy-safe server diagnostics --- .env.example | 4 + .env.production.example | 4 + .../20260830-server-diagnostics-c4d8a1f2.md | 70 ++++ README.md | 1 + agent设计规范/agentbus-reply-contract.md | 2 + control-plane/README.md | 23 +- control-plane/src/admin-cli.ts | 20 +- control-plane/src/agentbus-channels.ts | 95 ++++- control-plane/src/agentbus.ts | 96 +++-- control-plane/src/artifact-store.ts | 50 ++- control-plane/src/auth.ts | 27 +- control-plane/src/config.ts | 6 +- control-plane/src/db.ts | 26 +- control-plane/src/diagnostics.ts | 126 ++++++ control-plane/src/input-attachment.ts | 108 ++++- control-plane/src/migrate.ts | 12 +- control-plane/src/retention.ts | 12 +- control-plane/src/server.ts | 369 ++++++++++++++++-- control-plane/src/task-service.ts | 44 ++- control-plane/test/agentbus.test.ts | 7 +- control-plane/test/diagnostics.test.ts | 180 +++++++++ control-plane/test/input-attachment.test.ts | 18 + docker-compose.yml | 9 + infra/diagnose-server.sh | 97 +++++ infra/predeploy-check.sh | 2 +- package.json | 1 + tools/repository-hygiene.test.mjs | 8 + 27 files changed, 1305 insertions(+), 112 deletions(-) create mode 100644 .project-docs/30-worklog/tasks/20260830-server-diagnostics-c4d8a1f2.md create mode 100644 control-plane/src/diagnostics.ts create mode 100644 control-plane/test/diagnostics.test.ts create mode 100755 infra/diagnose-server.sh diff --git a/.env.example b/.env.example index de4d7c3..916fdd4 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,5 @@ NODE_ENV=production +DEPLOYMENT_REVISION=local-development HOST=0.0.0.0 PORT=8786 APP_ORIGIN=https://business.example.internal @@ -54,3 +55,6 @@ ERP_RECONCILIATION_MAINTENANCE_TIMEOUT_MS=1800000 DATA_RETENTION_ENABLED=false DATA_RETENTION_DAYS=180 LOG_LEVEL=info +# Docker json-file retention; these are consumed by Compose, not the app. +LOG_MAX_SIZE=20m +LOG_MAX_FILES=10 diff --git a/.env.production.example b/.env.production.example index 4298f84..67ed33f 100644 --- a/.env.production.example +++ b/.env.production.example @@ -1,4 +1,5 @@ NODE_ENV=production +DEPLOYMENT_REVISION=replace-with-deployed-git-commit HOST=0.0.0.0 PORT=8786 APP_ORIGIN=https://business.example.internal @@ -53,3 +54,6 @@ ERP_RECONCILIATION_MAINTENANCE_TIMEOUT_MS=1800000 DATA_RETENTION_ENABLED=false DATA_RETENTION_DAYS=180 LOG_LEVEL=info +# Docker json-file retention; keep total disk use bounded per container. +LOG_MAX_SIZE=20m +LOG_MAX_FILES=10 diff --git a/.project-docs/30-worklog/tasks/20260830-server-diagnostics-c4d8a1f2.md b/.project-docs/30-worklog/tasks/20260830-server-diagnostics-c4d8a1f2.md new file mode 100644 index 0000000..08fd86b --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260830-server-diagnostics-c4d8a1f2.md @@ -0,0 +1,70 @@ +# Task: Add privacy-safe server diagnostics logging + +## Identity + +- Task ID: 20260830-server-diagnostics-c4d8a1f2 +- 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: 23605066076f6c7c463564281379cc28bb27ec06 +- Owner: codex +- Status: Ready for integration + +## Scope + +- Add one privacy-safe structured diagnostic schema for control-plane service lifecycle, HTTP requests, task/audit state, parser workers, AgentBus, roster attachment ingress, database/runtime failures, and artifact cleanup. +- Preserve correlation across `request_id`, `task_id`, AgentBus frame/channel/conversation identifiers, diagnostic event/stage, bounded error code/fingerprint, outcome, and duration without logging request bodies, attachment URLs/bytes, roster values, credentials, cookies, or tokens. +- Make production logs directly usable from the server through bounded Docker JSON-log retention and a read-only diagnostic command that reports container state, readiness, and filtered recent logs. +- Add regression coverage for redaction/error fingerprints, stable request IDs, production payload-log blocking, structured event fields, and deployment log rotation. +- Update active control-plane, environment-example, and deployment documentation. + +## Intent And Constraints + +- Build on attachment-correlation commit `23605066076f6c7c463564281379cc28bb27ec06`; retain all of its strict attachment and SSRF controls. +- "Complete" means complete lifecycle and correlation metadata, not raw sensitive payload capture. Production must reject `AGENTBUS_LOG_PAYLOADS=true`. +- Keep business audit rows authoritative while mirroring only event type, identifiers, state, and metadata keys into operational logs; never duplicate encrypted business data into logs. +- Logging failures must never change task processing, attachment download, or ERP behavior. +- Use stdout/stderr as the application log sink and Docker's existing log collection boundary; add bounded rotation rather than introducing a second mutable log database or repository log files. +- Do not change parser contracts, Schema, mapping, ERP, Chrome extension, or release artifacts. Do not read secrets, deploy, restart services, mutate live tasks, access ERP, or send external messages. + +## Plan + +1. Add reusable diagnostic helpers for safe error codes, fingerprints, stack frames, request IDs, paths, durations, and emergency JSON stderr records. +2. Wire structured service/HTTP/process/task/audit/parser events into the control plane and replace raw exception logging with safe descriptors. +3. Add privacy-safe attachment metadata, DNS, redirect, response, completion, and failure milestones to the existing AgentBus event stream. +4. Add Docker log rotation plus a read-only server diagnostic script and document filtering by request, task, frame, conversation, channel, or error code. +5. Add focused regression tests, run all repository gates, record the outcome, and leave the branch ready for integration without deployment. + +## Outcome + +- Added a shared privacy-safe diagnostic layer with bounded error codes, fingerprints, sanitized stack frames, request IDs, paths, durations, metadata-key summaries, and emergency JSON stderr records. +- Replaced raw exception logging across service startup, HTTP handling, parser workers, AgentBus, channel management, database runtime failures, CLI jobs, and OSS cleanup. Operational logging remains non-authoritative and cannot change business state. +- Added correlated lifecycle events for service, HTTP, task state, audit staging, parser queue/worker/persistence, readiness, AgentBus channel/frame delivery, roster attachment ingress, artifact cleanup, and process shutdown. +- Added detailed attachment milestones for metadata, DNS, IP family/count, redirects, HTTPS response, byte/hash verification, completion, failure, and duration without logging URL, hostname, IP, file name, bytes, roster values, or message bodies. +- Production now rejects raw AgentBus payload logging. Pino redaction covers common credential fields, and HTTP automatic request serialization is disabled in favor of the bounded diagnostic schema. +- Added bounded Docker `json-file` retention (default 20 MB × 10 files per container) and `infra/diagnose-server.sh`, a read-only command for container state, readiness, and literal correlation filtering. +- Updated deployment examples and operator documentation. No live service, database, ERP, external channel, release artifact, or deployment state was changed. + +## Verification + +- `node --run check:repo` — passed, 9/9. +- `node --run check` — passed. +- `node --run test:control-plane` — passed, 135/135. +- `node --run test:legacy` — passed, 248/248. +- `node --run build` — passed. +- `sh -n infra/diagnose-server.sh infra/predeploy-check.sh` — passed. +- `sh infra/diagnose-server.sh --help` — passed; invalid unbounded `--since` was rejected with exit 2. +- `git diff --check` — passed. +- Ruby/Psych parse of `docker-compose.yml` with aliases enabled — passed. +- `docker compose --env-file .env.production.example config --quiet` — not available on this workstation because the Docker CLI is not installed; Compose structure and rotation bindings are covered by the repository hygiene test. + +## Follow-ups + +- At integration/deployment time, set `DEPLOYMENT_REVISION` to the deployed commit or release identifier and keep `AGENTBUS_LOG_PAYLOADS=false`. +- After separately authorized deployment/restart, run `sh infra/diagnose-server.sh --since 10m` and verify `service.listening`, readiness, current deployment revision, and the live AgentBus attachment path. +- Existing pre-deployment container logs do not gain the new schema retroactively. + +## Promotion Candidates + +- Candidate: promote the privacy-safe diagnostic field contract and stdout/Docker retention boundary into canonical architecture/data-flow memory after integration acceptance. Target: `.project-docs/10-architecture/system-architecture.md` and the relevant canonical data-flow record. Evidence: this task record and the passing gates above. Human decision: not required unless canonical maintainers want a different log-retention policy. +- Candidate: promote the AgentBus attachment diagnostic privacy rule (stage metadata only; never URL, hostname, IP, file name, bytes, payload, or roster values) into canonical business constraints after integration acceptance. Target: relevant AgentBus/business rules memory. Evidence: `control-plane/src/input-attachment.ts`, `control-plane/src/agentbus.ts`, and their regression tests. Human decision: not required. diff --git a/README.md b/README.md index b404326..36b5add 100644 --- a/README.md +++ b/README.md @@ -80,5 +80,6 @@ npm run build - `npm run build` 输出到 `.build/`；`dist/` 不再保存编译后的控制面代码。 - 本地 `.env` 与 `LianSyn-platform/.env` 只用于本机运行并由 Git 忽略；生产使用部署环境变量。 - 生产默认使用远程 PostgreSQL 的独立 `DATABASE_SCHEMA` 和 OSS 附件存储；具体连接与迁移边界见 [控制平面说明](control-plane/README.md)。 +- 生产结构化日志通过 Docker 有界轮转保留；服务端可运行 `sh infra/diagnose-server.sh --since 30m --match ` 只读检查容器、readiness 与关联日志。 真实 ERP 写入、测试数据创建/清理、服务重启、部署、外部发送或云端 Profile 更新都需要本轮明确授权。 diff --git a/agent设计规范/agentbus-reply-contract.md b/agent设计规范/agentbus-reply-contract.md index 8ea66bf..cce32e5 100644 --- a/agent设计规范/agentbus-reply-contract.md +++ b/agent设计规范/agentbus-reply-contract.md @@ -19,6 +19,8 @@ 微信传输信封的 `Conversation:` 在帧未显式提供 `conversation_id` 时作为会话键，显式字段优先。`[WeChat attachment: 文件名]` 只是桥接器的文字占位符：没有同时提供 `payload.attachments[]` 时，控制面不得把它作为新业务消息创建任务，也不得假装已收到文件；应保留原任务的 `awaiting_attachment` 状态并提示附件内容尚未传到平台。附件元数据或安全下载校验失败时只返回预定义的安全摘要，不回显 URL、文件内容或名单值。 +服务端诊断以 `diagnostic_event=agentbus.` 串联 frame、conversation、channel、任务和最终回执。附件入口必须分别记录元数据、DNS、公网地址数量/IP family、HTTPS 状态、重定向、字节校验、结果和耗时，但不能记录 URL、hostname、IP、文件名、文件内容或名单值。异常使用错误码、稳定指纹和清理后的代码栈定位；生产环境不得启用正文 payload 日志。 + 附件通过程序模板校验后，任务从 `awaiting_attachment` 转为 `parse_queued`；校验失败则保留等待状态并只返回安全错误码和行列位置，不回显名单值。原始工作簿不持久化，只暂存加密 canonical TSV 供 Program Parser 使用。 ## 失败 diff --git a/control-plane/README.md b/control-plane/README.md index e5f30ad..b7a1f03 100644 --- a/control-plane/README.md +++ b/control-plane/README.md @@ -97,7 +97,7 @@ Auto 一旦发生 AI fallback，任务会永久绑定原 AI 会话。每次解 对微信来源，listener 在调用 `TaskService.ingestMessage()` 前执行上述严格信封解包，因此手工正文与 AgentBus 正文进入同一个业务 route resolver、任务级 mode snapshot 和 parser orchestrator；`Conversation` 只属于传输路由，不会再污染业务字段签名。 -AgentBus 全链路日志使用现有控制平面 stdout/Pino 输出，并统一带 `agentbus_event` 字段，可用 `rg 'agentbus_event'` 过滤。日志覆盖连接尝试、socket 生命周期、session.ready、每个收发帧、帧忽略原因、任务入队、解析队列、持久化回执出队、最终回复和发送错误。默认只记录消息正文的长度与摘要；本地排障可在受保护的环境文件中设置 `AGENTBUS_LOG_PAYLOADS=true`，记录最多 2,000 个字符的正文预览。渠道 key、WebSocket Token、Invoke Token 和 Authorization header 永不写入日志。 +AgentBus 全链路日志使用控制平面 stdout/Pino 输出，同时带 `diagnostic_event=agentbus.` 与原有 `agentbus_event`。日志覆盖连接尝试、socket 生命周期、session.ready、每个收发帧、帧忽略原因、任务入队、解析队列、持久化回执出队、最终回复和发送错误。名单附件另外记录元数据存在性、DNS 开始/通过、公网地址数量与 IP family、HTTPS 状态、重定向次数、接收字节数、大小/摘要校验和各阶段耗时；不记录 URL、hostname、IP、文件名、附件字节或名单值。开发/测试环境可临时设置 `AGENTBUS_LOG_PAYLOADS=true` 记录最多 2,000 个字符的正文预览，生产环境会拒绝以该值启动。渠道 key、WebSocket Token、Invoke Token 和 Authorization header 永不写入日志。 监听器只使用每个渠道的 WebSocket key；文档中的 Invoke Token 仅用于另一服务通过 Function Call API 主动向 Bot 投递任务，本服务的监听链路不会使用它。 @@ -105,6 +105,24 @@ AgentBus 全链路日志使用现有控制平面 stdout/Pino 输出，并统一 任务响应统一提供 `important_message`，作为“需返回/交互用户的重要消息”出口：`awaiting_user_input` 返回 Superagent 的 `reply`，错误优先返回 ERP 业务反馈、否则返回简短错误摘要，成功且有可验证 ERP 证据时按 [AgentBus 用户回复契约](../agent设计规范/agentbus-reply-contract.md) 生成业务回执。操作台保留完整结构供内部展示，AgentBus `task.result` 发往渠道时携带 `kind`、用户可读 `text` 和符合微信渠道适配器契约的 HTTPS URL-only 附件元数据；不发送 `content_base64` 或后台鉴权下载地址。外部文件投递要求生产附件使用 OSS 存储，且 URL 域名已加入适配器白名单。错误码、执行阶段、回执校验状态、隐藏 ID 和哈希不外发。`awaiting_confirmation` 不进入该栏。没有 `reply` 的 awaiting 结果会在解析阶段转为错误；没有可验证回执的 `completed` 执行会被归类为 `reconciliation_pending`，不会返回成功回执。 +## 服务端诊断日志 + +控制面、迁移、数据保留、数据库异常和管理员命令都输出一行一个 JSON 的结构化日志。公共字段包括 `time`、`level`、`service`、`environment`、`deployment_revision`、`pid`、`hostname`、`diagnostic_event` 和 `diagnostic_stage`；链路字段按事件提供 `request_id`、`task_id`、`inbound_frame_id`、`conversation_id`、`channel_id`、状态、结果与 `duration_ms`。覆盖范围包括服务启动/关闭、HTTP 开始/完成/拒绝、任务状态事件、数据库 audit event、解析队列和 Worker、AgentBus、附件入口、OSS 清理、数据库连接/回滚以及未捕获进程错误。 + +未知异常只记录受限 `error_code`、`error_name`、稳定的 `error_fingerprint`、errno/syscall 和最多 12 个清理后的代码栈帧；不把第三方异常原文复制进日志。请求 query/body、Cookie、Authorization、CSRF、密码、API key、Token、附件 URL/字节和业务原文不进入生产日志。每个 HTTP 响应返回同一个 `x-request-id`，可直接用于跨开始、错误和完成事件检索。部署时应把 `DEPLOYMENT_REVISION` 设置为当前提交或发布编号，避免不同镜像的日志混淆。 + +Docker Compose 使用 `json-file`，默认每个容器保留 10 个、每个 20 MB 的轮转文件；可在 `.env.production` 用 `LOG_MAX_SIZE` 和 `LOG_MAX_FILES` 调整。服务端只读排障命令会显示容器状态、`/health/ready` 和最近日志，不读取或打印环境文件： + +```bash +sh infra/diagnose-server.sh +sh infra/diagnose-server.sh --since 2h --match TASK-20260830071915-g7INIT0 +sh infra/diagnose-server.sh --since 2h --match roster_attachment_metadata_missing +sh infra/diagnose-server.sh --since 30m --match request:example:12345678 +sh infra/diagnose-server.sh --since 30m --all-services +``` + +`--match` 是字面量过滤，可使用 request/task/frame/conversation/channel ID、`diagnostic_event` 或 `error_code`。脚本默认最多读取 1,000 行控制面日志，`--tail` 上限为 50,000，且不在仓库生成日志文件。 + ## 本地命令 本地命令会自动读取项目根目录 `.env`，不需要先手动 `export` 环境变量。首次部署或换环境时，只需替换该文件；生产 Docker Compose 使用 `.env.production`。 @@ -125,11 +143,12 @@ npm run dev ## 生产部署 -1. 复制 `.env.production.example` 为部署机受保护的 `.env.production`，填入 PostgreSQL URL、`DATABASE_SCHEMA`、字段加密密钥、外部解析 Key 和 OSS 凭据。生产数据库可使用现有 PostgreSQL 实例中的新 Schema；迁移程序会创建 Schema，不会触碰其他 Schema 的测试表。 +1. 复制 `.env.production.example` 为部署机受保护的 `.env.production`，填入 `DEPLOYMENT_REVISION`、PostgreSQL URL、`DATABASE_SCHEMA`、字段加密密钥、外部解析 Key 和 OSS 凭据，并确认 `AGENTBUS_LOG_PAYLOADS=false`。生产数据库可使用现有 PostgreSQL 实例中的新 Schema；迁移程序会创建 Schema，不会触碰其他 Schema 的测试表。 2. 在正式数据库上线前执行并验证备份：`infra/backup-postgres.sh`。 3. 使用 `docker compose --env-file .env.production up -d --build` 启动；Compose 会先执行数据库迁移，再启动控制平面。Compose 中的本地 PostgreSQL 仅用于 `--profile local`，生产 `DATABASE_URL` 指向受保护的远程数据库。 4. 首次启动后在容器内通过 `docker compose exec -e ADMIN_USERNAME=admin -e ADMIN_PASSWORD='replace-with-12-plus-chars' control-plane node .build/control-plane/src/admin-cli.js bootstrap` 创建管理员；不要把密码写入镜像或 Git。 5. 配置 `infra/Caddyfile` 中的正式域名和 HTTPS，然后在 Chrome 插件中加载对应生产业务页面。 +6. 启动后运行 `sh infra/diagnose-server.sh --since 10m`，确认容器、readiness、`service.listening`、AgentBus session 与当前 `deployment_revision`。 正式环境没有独立 staging。上线前必须完成离线测试、迁移预检查和迁移前备份；恢复检查使用 `infra/restore-check.sh` 指向一次性恢复数据库。 diff --git a/control-plane/src/admin-cli.ts b/control-plane/src/admin-cli.ts index a8b0629..e335b10 100644 --- a/control-plane/src/admin-cli.ts +++ b/control-plane/src/admin-cli.ts @@ -1,6 +1,7 @@ import { closePool } from './db.js'; import { loadConfig } from './config.js'; import { AuthService } from './auth.js'; +import { writeEmergencyDiagnostic } from './diagnostics.js'; function readFlag(name: string): string { const index = process.argv.indexOf(name); @@ -17,7 +18,13 @@ async function main(): Promise { process.env.ADMIN_PASSWORD || readFlag('--password'), { force: process.argv.includes('--force') } ); - console.log(JSON.stringify({ ok: true, action: 'bootstrap', username: user.username, organization_id: user.organizationId })); + console.log(JSON.stringify({ + ok: true, + diagnostic_event: 'admin.bootstrap.completed', + diagnostic_stage: 'admin_cli', + action: 'bootstrap', + administrator_created: Boolean(user.id) + })); return; } if (command === 'reset-password') { @@ -25,7 +32,12 @@ async function main(): Promise { process.env.ADMIN_USERNAME || readFlag('--username'), process.env.ADMIN_PASSWORD || readFlag('--password') ); - console.log(JSON.stringify({ ok: true, action: 'reset-password' })); + console.log(JSON.stringify({ + ok: true, + diagnostic_event: 'admin.password_reset.completed', + diagnostic_stage: 'admin_cli', + action: 'reset-password' + })); return; } throw new Error('Usage: npm run admin -- bootstrap|reset-password (use ADMIN_USERNAME and ADMIN_PASSWORD)'); @@ -33,7 +45,9 @@ async function main(): Promise { main() .catch((error) => { - console.error(JSON.stringify({ ok: false, error: error.message || String(error) })); + writeEmergencyDiagnostic('admin.command.failed', error, { + diagnostic_stage: 'admin_cli' + }); process.exitCode = 1; }) .finally(() => closePool()); diff --git a/control-plane/src/agentbus-channels.ts b/control-plane/src/agentbus-channels.ts index 1b18568..d3b2919 100644 --- a/control-plane/src/agentbus-channels.ts +++ b/control-plane/src/agentbus-channels.ts @@ -8,6 +8,7 @@ import { type AgentBusSocketFactory } from './agentbus.js'; import { TaskError, type TaskContext } from './task-service.js'; +import { diagnosticError, diagnosticMetadataKeys } from './diagnostics.js'; export interface PublicAgentBusChannel { id: string; @@ -120,7 +121,22 @@ export function mergeRuntimeChannelStatuses( export class AgentBusChannelService { private readonly runtimeStatusWrites = new Map>(); - constructor(private readonly config: AppConfig) {} + constructor( + private readonly config: AppConfig, + private readonly logger?: AgentBusLogger + ) {} + + private log( + level: 'info' | 'warn' | 'error', + metadata: Record, + message: string + ): void { + try { + this.logger?.[level](metadata, message); + } catch { + // Channel persistence and encryption never depend on operational logs. + } + } private legacyEnvironmentManaged(): boolean { return this.config.agentBusEnabled @@ -178,6 +194,13 @@ export class AgentBusChannelService { try { const wsToken = decryptText(this.config, text(row.agentbus_ws_token_ciphertext)); if (!wsUrl || !wsToken || !botAddress) { + this.log('warn', { + agentbus_event: 'channel_configuration_incomplete', + channel_id: text(row.id), + ws_url_present: Boolean(wsUrl), + ws_token_present: Boolean(wsToken), + bot_address_present: Boolean(botAddress) + }, 'AgentBus channel configuration is incomplete'); await this.setRuntimeStatus( organizationId, text(row.id), @@ -192,7 +215,12 @@ export class AgentBusChannelService { ws_token: wsToken, bot_address: botAddress }); - } catch { + } catch (error) { + this.log('error', { + agentbus_event: 'channel_key_decryption_failed', + channel_id: text(row.id), + ...diagnosticError(error, 'agentbus_channel_key_decryption_failed') + }, 'AgentBus channel key decryption failed'); await this.setRuntimeStatus(organizationId, text(row.id), 'error', '渠道 key 无法解密。'); } } @@ -471,6 +499,15 @@ export class AgentBusChannelService { VALUES ($1, $2, $3, 'user_channel', $4, $5, $6)`, [context.organizationId, context.userId || null, eventType, entityId, context.requestId, metadata] ); + this.log('info', { + agentbus_event: 'channel_audit_staged', + request_id: context.requestId, + domain_event: eventType, + entity_type: 'user_channel', + entity_id: entityId, + actor_present: Boolean(context.userId), + metadata_keys: diagnosticMetadataKeys(metadata) + }, 'AgentBus channel audit event staged'); } } @@ -493,7 +530,19 @@ export class AgentBusManager { constructor(options: AgentBusManagerOptions) { this.options = options; - this.channels = new AgentBusChannelService(options.config); + this.channels = new AgentBusChannelService(options.config, options.logger); + } + + private log( + level: 'info' | 'warn' | 'error', + metadata: Record, + message: string + ): void { + try { + this.options.logger?.[level](metadata, message); + } catch { + // Runtime channel management must not depend on log delivery. + } } get channelService(): AgentBusChannelService { @@ -504,12 +553,18 @@ export class AgentBusManager { if (!this.options.config.agentBusEnabled) return; if (this.started) return; this.started = true; + this.log('info', { agentbus_event: 'manager_starting' }, 'AgentBus manager starting'); await this.reload(); } async reload(): Promise { if (!this.started) return; this.reloadRequested = true; + this.log('info', { + agentbus_event: 'manager_reload_requested', + reload_in_flight: Boolean(this.reloadInFlight), + current_listeners: this.listeners.size + }, 'AgentBus manager reload requested'); if (this.reloadInFlight) return this.reloadInFlight; this.reloadInFlight = (async () => { while (this.started && this.reloadRequested) { @@ -518,6 +573,10 @@ export class AgentBusManager { for (const listener of this.listeners.values()) listener.stop(false); this.listeners.clear(); const secrets = await this.channels.listEnabledSecrets(this.options.organizationId); + this.log('info', { + agentbus_event: 'manager_channels_loaded', + enabled_channels: secrets.length + }, 'AgentBus enabled channels loaded'); for (const channel of secrets) { const listener = new AgentBusListener({ config: this.options.config, @@ -539,15 +598,36 @@ export class AgentBusManager { status, error, epoch - ).catch(() => undefined) + ).catch((statusError) => { + this.log('warn', { + agentbus_event: 'channel_status_persist_failed', + channel_id: channel.id, + runtime_status: status, + session_epoch: epoch, + ...diagnosticError(statusError, 'agentbus_channel_status_persist_failed') + }, 'AgentBus channel status persistence failed'); + }) }); this.listeners.set(channel.id, listener); listener.start(); } + this.log('info', { + agentbus_event: 'manager_reload_completed', + active_listeners: this.listeners.size, + reload_requested_again: this.reloadRequested + }, 'AgentBus manager reload completed'); } - })().finally(() => { - this.reloadInFlight = null; - }); + })() + .catch((error) => { + this.log('error', { + agentbus_event: 'manager_reload_failed', + ...diagnosticError(error, 'agentbus_manager_reload_failed') + }, 'AgentBus manager reload failed'); + throw error; + }) + .finally(() => { + this.reloadInFlight = null; + }); return this.reloadInFlight; } @@ -559,6 +639,7 @@ export class AgentBusManager { // an older process write `disabled` after a newer process has connected. for (const listener of this.listeners.values()) listener.stop(false); this.listeners.clear(); + this.log('info', { agentbus_event: 'manager_stopped' }, 'AgentBus manager stopped'); } status(): { diff --git a/control-plane/src/agentbus.ts b/control-plane/src/agentbus.ts index f4fc556..80393ea 100644 --- a/control-plane/src/agentbus.ts +++ b/control-plane/src/agentbus.ts @@ -18,6 +18,7 @@ import { downloadAgentBusInputAttachment, parseAgentBusInputAttachment } from './input-attachment.js'; +import { diagnosticDurationMs, diagnosticError } from './diagnostics.js'; const OPEN_READY_STATE = 1; const MAX_COMPLETED_TASK_IDS = 2_048; @@ -791,7 +792,7 @@ export class AgentBusListener { } catch (error) { this.logger.warn({ agentbus_event: 'connect_creation_failed', - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_connect_creation_failed') }, 'AgentBus WebSocket creation failed'); this.scheduleReconnect(); return; @@ -807,7 +808,10 @@ export class AgentBusListener { }); socket.on('message', (raw) => void this.handleMessage(raw)); socket.on('error', (error) => { - this.logger.warn({ agentbus_event: 'socket_error', error: error.message }, 'AgentBus WebSocket error'); + this.logger.warn({ + agentbus_event: 'socket_error', + ...diagnosticError(error, 'agentbus_socket_error') + }, 'AgentBus WebSocket error'); }); socket.on('close', (code) => this.handleClose(socket, code)); } @@ -895,14 +899,15 @@ export class AgentBusListener { return; } this.logFrame('inbound_task_accepted', frame, 'AgentBus inbound task accepted'); + const processingStartedAt = process.hrtime.bigint(); 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) + duration_ms: diagnosticDurationMs(processingStartedAt), + ...diagnosticError(error, attachmentError?.code || 'agentbus_task_processing_failed') }, 'AgentBus task processing failed'); this.queueFinalReply( frame, @@ -912,7 +917,9 @@ export class AgentBusListener { }) .finally(() => { this.inFlightTaskIds.delete(taskId); - this.logFrame('task_processing_finished', frame, 'AgentBus inbound task processing finished'); + this.logFrame('task_processing_finished', frame, 'AgentBus inbound task processing finished', { + duration_ms: diagnosticDurationMs(processingStartedAt) + }); }); this.inFlightTaskIds.set(taskId, processing); } @@ -924,29 +931,57 @@ export class AgentBusListener { 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), - this.config.ARTIFACT_MAX_BYTES - )] - : []; this.logFrame('task_processing_started', frame, 'AgentBus task processing started', { task_id: taskId, conversation_id: conversationId || null, explicit_task_id: frameTaskId(frame) || null, transport_envelope_unwrapped: message !== rawMessage, + attachment_count: rawAttachments.length, + attachment_placeholder: Boolean(transportEnvelope?.attachmentPlaceholder), business_text_length: message.length, business_text_sha256: textDigest(message) }); + if (rawAttachments.length > 1) { + throw new InputAttachmentError( + 'roster_attachment_count_invalid', + 'AgentBus 名单业务每次只能发送一个附件。' + ); + } + if (!rawAttachments.length && transportEnvelope?.attachmentPlaceholder) { + this.logFrame('attachment_metadata_missing', frame, 'WeChat attachment placeholder has no file metadata', { + task_id: taskId, + conversation_id: conversationId || null + }); + throw new InputAttachmentError( + 'roster_attachment_metadata_missing', + '附件内容未传到平台，原任务仍在等待附件。请检查微信桥接器的文件转发后重新发送。' + ); + } + const attachments = []; + if (rawAttachments.length) { + const reference = parseAgentBusInputAttachment(rawAttachments[0], this.config.ARTIFACT_MAX_BYTES); + this.logFrame('attachment_metadata_validated', frame, 'AgentBus attachment metadata validated', { + task_id: taskId, + conversation_id: conversationId || null, + declared_size: reference.size ?? null, + declared_sha256_present: Boolean(reference.sha256), + content_type_present: Boolean(reference.contentType) + }); + attachments.push(await downloadAgentBusInputAttachment( + reference, + this.config.ARTIFACT_MAX_BYTES, + (event, metadata) => this.logFrame( + `attachment_${event}`, + frame, + `AgentBus attachment ${event}`, + { + task_id: taskId, + conversation_id: conversationId || null, + ...metadata + } + ) + )); + } const input: TaskMessageInput = { message, conversationId, @@ -995,7 +1030,7 @@ export class AgentBusListener { agentbus_event: 'parse_queue_request_failed', inbound_frame_id: taskId, task_id: result.task.task_id, - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_parse_queue_request_failed') }, 'AgentBus parse queue request failed; durable task remains queued'); }); await this.flushDurableDeliveries(); @@ -1060,7 +1095,7 @@ export class AgentBusListener { channel_id: this.channel!.id, inbound_frame_id: candidate.inbound_frame_id, task_id: candidate.task_id, - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_result_enqueue_failed') }, 'AgentBus durable result enqueue failed'); } } @@ -1075,7 +1110,7 @@ export class AgentBusListener { this.logger.warn({ agentbus_event: 'durable_delivery_flush_failed', channel_id: this.channel?.id || null, - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_delivery_flush_failed') }, 'AgentBus durable delivery flush failed'); }) .finally(() => { @@ -1110,14 +1145,15 @@ export class AgentBusListener { attempt_count: delivery.attempt_count }); const fail = (error: unknown) => { + const errorMetadata = diagnosticError(error, 'agentbus_delivery_failed'); void markFailed( delivery.id, - error instanceof Error ? error.message : String(error) + `${String(errorMetadata.error_code)}:${String(errorMetadata.error_fingerprint)}` ).catch(() => undefined); this.logger.warn({ agentbus_event: 'outbound_durable_delivery_failed', delivery_id: delivery.id, - error: error instanceof Error ? error.message : String(error) + ...errorMetadata }, 'AgentBus durable delivery failed'); }; try { @@ -1192,7 +1228,7 @@ export class AgentBusListener { this.logger.error({ agentbus_event: 'task_outcome_wait_failed', task_id: initialTask.task_id, - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_task_outcome_wait_failed') }, 'AgentBus task outcome wait failed'); reject(error); }; @@ -1256,7 +1292,7 @@ export class AgentBusListener { this.logger.warn({ agentbus_event: 'outbound_progress_failed', ...frameLogData(frame, this.config.AGENTBUS_LOG_PAYLOADS), - error: error.message + ...diagnosticError(error, 'agentbus_progress_send_failed') }, 'AgentBus progress frame failed'); return; } @@ -1266,7 +1302,7 @@ export class AgentBusListener { this.logger.warn({ agentbus_event: 'outbound_progress_failed', ...frameLogData(frame, this.config.AGENTBUS_LOG_PAYLOADS), - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_progress_send_failed') }, 'AgentBus progress frame failed'); } } @@ -1355,7 +1391,7 @@ export class AgentBusListener { this.logger.warn({ agentbus_event: 'outbound_result_failed', ...frameLogData(frame, this.config.AGENTBUS_LOG_PAYLOADS), - error: error.message + ...diagnosticError(error, 'agentbus_result_send_failed') }, 'AgentBus result frame failed'); return; } @@ -1368,7 +1404,7 @@ export class AgentBusListener { pending.sending = false; this.logger.warn({ agentbus_event: 'outbound_result_failed', - error: error instanceof Error ? error.message : String(error) + ...diagnosticError(error, 'agentbus_result_send_failed') }, 'AgentBus result frame failed'); } } diff --git a/control-plane/src/artifact-store.ts b/control-plane/src/artifact-store.ts index 6b86f4e..02609e2 100644 --- a/control-plane/src/artifact-store.ts +++ b/control-plane/src/artifact-store.ts @@ -4,6 +4,11 @@ import { decryptBytes, encryptBytes, sha256Bytes } from './crypto.js'; import { getPool } from './db.js'; import { AliyunOssClient, type OssObjectClient } from './oss-client.js'; import { normalizeTaskArtifactFileName } from './artifact-name.js'; +import { + diagnosticError, + noopDiagnosticLogger, + type DiagnosticLogger +} from './diagnostics.js'; export { normalizeTaskArtifactFileName } from './artifact-name.js'; @@ -202,9 +207,22 @@ export class DatabaseArtifactStore implements TaskArtifactStore { export class OssArtifactStore implements TaskArtifactStore { constructor( private readonly config: AppConfig, - private readonly client: OssObjectClient = new AliyunOssClient(config) + private readonly client: OssObjectClient = new AliyunOssClient(config), + private readonly logger: DiagnosticLogger = noopDiagnosticLogger ) {} + private log( + level: 'info' | 'warn' | 'error', + metadata: Record, + message: string + ): void { + try { + this.logger[level](metadata, message); + } catch { + // Artifact cleanup is authoritative; operational logging is not. + } + } + async put(client: PoolClient, input: TaskArtifactInput): Promise { const prefix = this.config.OSS_KEY_PREFIX.replace(/\/+$/, ''); // executionId is a server-generated UUID. Keeping it in the key makes a @@ -312,22 +330,42 @@ export class OssArtifactStore implements TaskArtifactStore { } async cleanup(artifacts: StoredTaskArtifact[]): Promise { + let deleted = 0; + let failed = 0; for (const artifact of artifacts) { if (artifact.storage_backend !== 'oss' || !artifact.storage_key) continue; try { await this.client.deleteObject(artifact.storage_key); + deleted += 1; } catch (error) { - console.error('[artifact] OSS cleanup failed', { + failed += 1; + this.log('error', { + diagnostic_event: 'artifact.cleanup.failed', + diagnostic_stage: 'artifact_cleanup', artifact_id: artifact.id, - request_error: error instanceof Error ? error.message : String(error) - }); + task_id: artifact.task_id, + storage_backend: artifact.storage_backend, + ...diagnosticError(error, 'artifact_cleanup_failed') + }, 'OSS artifact cleanup failed'); } } + if (deleted || failed) { + this.log(failed ? 'warn' : 'info', { + diagnostic_event: 'artifact.cleanup.completed', + diagnostic_stage: 'artifact_cleanup', + requested_artifacts: artifacts.length, + deleted_artifacts: deleted, + failed_artifacts: failed + }, 'artifact cleanup completed'); + } } } -export function createTaskArtifactStore(config: AppConfig): TaskArtifactStore { +export function createTaskArtifactStore( + config: AppConfig, + logger: DiagnosticLogger = noopDiagnosticLogger +): TaskArtifactStore { return config.ARTIFACT_STORAGE_BACKEND === 'oss' - ? new OssArtifactStore(config) + ? new OssArtifactStore(config, undefined, logger) : new DatabaseArtifactStore(config); } diff --git a/control-plane/src/auth.ts b/control-plane/src/auth.ts index 7d0106f..8384c17 100644 --- a/control-plane/src/auth.ts +++ b/control-plane/src/auth.ts @@ -2,6 +2,11 @@ import argon2 from 'argon2'; import type { AppConfig } from './config.js'; import { getPool, withTransaction } from './db.js'; import { hashToken, randomToken, sameTokenHash } from './crypto.js'; +import { + diagnosticMetadataKeys, + noopDiagnosticLogger, + type DiagnosticLogger +} from './diagnostics.js'; export interface AuthUser { id: string; @@ -55,7 +60,18 @@ export class AuthService { parallelism: 1 }); - constructor(private readonly config: AppConfig) {} + constructor( + private readonly config: AppConfig, + private readonly logger: DiagnosticLogger = noopDiagnosticLogger + ) {} + + private log(metadata: Record, message: string): void { + try { + this.logger.info(metadata, message); + } catch { + // Authentication state and audit persistence never depend on logging. + } + } async ensureOrganization(): Promise<{ id: string; slug: string; name: string }> { const result = await getPool(this.config).query( @@ -255,5 +271,14 @@ export class AuthService { VALUES ($1, $2, 'auth.' || $3, 'session', $4, $5, $6)`, [organizationId, userId, eventType, userId || '', requestId, metadata] ); + this.log({ + diagnostic_event: 'audit.event.staged', + diagnostic_stage: 'auth_audit', + request_id: requestId, + domain_event: `auth.${eventType}`, + entity_type: 'session', + actor_present: Boolean(userId), + metadata_keys: diagnosticMetadataKeys(metadata) + }, 'authentication audit event persisted'); } } diff --git a/control-plane/src/config.ts b/control-plane/src/config.ts index 61f8586..b99deb3 100644 --- a/control-plane/src/config.ts +++ b/control-plane/src/config.ts @@ -53,6 +53,7 @@ function parseDurationMs(value: string): number { const envSchema = z.object({ NODE_ENV: z.enum(['development', 'test', 'production']).default('development'), + DEPLOYMENT_REVISION: z.string().trim().regex(/^[A-Za-z0-9._-]{1,80}$/).default('unknown'), HOST: z.string().default('127.0.0.1'), PORT: z.coerce.number().int().positive().default(8786), APP_ORIGIN: z.string().url().default('http://127.0.0.1:8786'), @@ -98,7 +99,7 @@ const envSchema = z.object({ OSS_KEY_PREFIX: z.string().trim().min(1).max(200).default('liansyn-platform/attachments'), DATA_RETENTION_ENABLED: z.enum(['true', 'false']).default('false').transform((value) => value === 'true'), DATA_RETENTION_DAYS: z.coerce.number().int().positive().default(180), - LOG_LEVEL: z.string().default('info') + LOG_LEVEL: z.enum(['trace', 'debug', 'info', 'warn', 'error', 'fatal', 'silent']).default('info') }); export type AppConfig = z.infer & { @@ -123,6 +124,9 @@ function decodeEncryptionKey(value: string | undefined, nodeEnv: string): Buffer export function loadConfig(env: NodeJS.ProcessEnv = process.env): AppConfig { const parsed = envSchema.parse(env); + if (parsed.NODE_ENV === 'production' && parsed.AGENTBUS_LOG_PAYLOADS) { + throw new Error('AGENTBUS_LOG_PAYLOADS must remain false in production.'); + } const legacyCredentialsPresent = Boolean(parsed.AGENTBUS_WS_TOKEN || parsed.AGENTBUS_BOT_ADDRESS); const legacyConfigurationComplete = Boolean( parsed.AGENTBUS_WS_URL && parsed.AGENTBUS_WS_TOKEN && parsed.AGENTBUS_BOT_ADDRESS diff --git a/control-plane/src/db.ts b/control-plane/src/db.ts index e242365..9dd8ea6 100644 --- a/control-plane/src/db.ts +++ b/control-plane/src/db.ts @@ -1,5 +1,6 @@ import pg from 'pg'; import type { AppConfig } from './config.js'; +import { writeEmergencyDiagnostic } from './diagnostics.js'; const { Pool } = pg; let pool: pg.Pool | null = null; @@ -19,10 +20,17 @@ export function quoteIdentifier(identifier: string): string { } function logDatabaseError(context: string, error: unknown): void { - const value = error && typeof error === 'object' ? error as { code?: unknown; message?: unknown } : {}; - const code = value.code ? ` (${String(value.code)})` : ''; - const message = value.message ? String(value.message) : String(error); - console.error(`[database] ${context}${code}: ${message}`); + const event = context === 'idle pool client error' + ? 'database.pool.idle_client_error' + : 'database.transaction.rollback_failed'; + writeEmergencyDiagnostic(event, error, { diagnostic_stage: 'database' }); +} + +export class DatabaseStartupError extends Error { + constructor(public readonly code: string, message: string) { + super(message); + this.name = 'DatabaseStartupError'; + } } export function getPool(config: AppConfig): pg.Pool { @@ -106,9 +114,15 @@ export async function databaseReadiness(config: AppConfig): Promise { const readiness = await databaseReadiness(config); if (!readiness.database) { - throw new Error(`数据库不可用，无法启动控制平面${readiness.errorCode ? `（${readiness.errorCode}）` : ''}。`); + throw new DatabaseStartupError( + readiness.errorCode || 'database_unavailable', + `数据库不可用，无法启动控制平面${readiness.errorCode ? `（${readiness.errorCode}）` : ''}。` + ); } if (!readiness.schema) { - throw new Error(`数据库迁移 ${REQUIRED_SCHEMA_VERSION} 尚未应用，请先执行 npm run db:migrate。`); + throw new DatabaseStartupError( + 'database_schema_outdated', + `数据库迁移 ${REQUIRED_SCHEMA_VERSION} 尚未应用，请先执行 npm run db:migrate。` + ); } } diff --git a/control-plane/src/diagnostics.ts b/control-plane/src/diagnostics.ts new file mode 100644 index 0000000..34ed5a7 --- /dev/null +++ b/control-plane/src/diagnostics.ts @@ -0,0 +1,126 @@ +import { createHash, randomUUID } from 'node:crypto'; + +const DIAGNOSTIC_CODE_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_.:-]{1,79}$/; +const DIAGNOSTIC_NAME_PATTERN = /^[A-Za-z][A-Za-z0-9_.-]{0,79}$/; +const DIAGNOSTIC_SYSCALL_PATTERN = /^[A-Za-z0-9_.:-]{1,80}$/; +const REQUEST_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_.:-]{7,127}$/; +const MAX_STACK_FRAMES = 12; + +export interface DiagnosticLogger { + info(metadata: Record, message?: string): void; + warn(metadata: Record, message?: string): void; + error(metadata: Record, message?: string): void; +} + +export const noopDiagnosticLogger: DiagnosticLogger = { + info: () => undefined, + warn: () => undefined, + error: () => undefined +}; + +function text(value: unknown): string { + return value == null ? '' : String(value).trim(); +} + +function safeDiagnosticCode(value: unknown, fallback: string): string { + const normalized = text(value); + return DIAGNOSTIC_CODE_PATTERN.test(normalized) ? normalized : fallback; +} + +function safeDiagnosticName(value: unknown): string { + const normalized = text(value); + return DIAGNOSTIC_NAME_PATTERN.test(normalized) ? normalized : 'UnknownError'; +} + +function sanitizedStackFrames(error: unknown): string[] { + if (!(error instanceof Error) || !error.stack) return []; + const currentDirectory = process.cwd().replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); + const currentDirectoryPattern = currentDirectory ? new RegExp(currentDirectory, 'g') : null; + return error.stack + .split(/\r?\n/u) + .slice(1) + .map((line) => line.trim()) + .filter((line) => line.startsWith('at ')) + .slice(0, MAX_STACK_FRAMES) + .map((line) => { + let sanitized = line + .replace(/https?:\/\/\S+/giu, '') + .replace(/\/Users\/[^/\s)]+/gu, '/Users/') + .replace(/\/home\/[^/\s)]+/gu, '/home/'); + if (currentDirectoryPattern) sanitized = sanitized.replace(currentDirectoryPattern, ''); + return sanitized.slice(0, 500); + }); +} + +export function diagnosticError( + error: unknown, + fallbackCode = 'internal_error' +): Record { + const value = error && typeof error === 'object' + ? error as { code?: unknown; errno?: unknown; syscall?: unknown; name?: unknown; message?: unknown; stack?: unknown } + : {}; + const errorCode = safeDiagnosticCode(value.code, safeDiagnosticCode(fallbackCode, 'internal_error')); + const errorName = safeDiagnosticName(error instanceof Error ? error.name : value.name); + const rawMessage = error instanceof Error ? error.message : text(value.message || error); + const rawStack = error instanceof Error ? text(error.stack) : text(value.stack); + const errorFingerprint = createHash('sha256') + .update(`${errorName}\u0000${errorCode}\u0000${rawMessage}\u0000${rawStack}`) + .digest('hex') + .slice(0, 24); + const stackFrames = sanitizedStackFrames(error); + const errno = Number(value.errno); + const syscall = text(value.syscall); + return { + error_code: errorCode, + error_name: errorName, + error_fingerprint: errorFingerprint, + ...(stackFrames.length ? { error_stack: stackFrames } : {}), + ...(Number.isSafeInteger(errno) ? { error_errno: errno } : {}), + ...(DIAGNOSTIC_SYSCALL_PATTERN.test(syscall) ? { error_syscall: syscall } : {}) + }; +} + +export function diagnosticHash(value: unknown): string { + return createHash('sha256').update(text(value)).digest('hex').slice(0, 24); +} + +export function diagnosticMetadataKeys(value: Record): string[] { + return Object.keys(value).sort().slice(0, 100); +} + +export function normalizeRequestId(value: unknown): string { + const normalized = text(value); + return REQUEST_ID_PATTERN.test(normalized) ? normalized : randomUUID(); +} + +export function diagnosticRequestPath(value: unknown): string { + const raw = text(value); + if (!raw) return '/'; + try { + return new URL(raw, 'http://diagnostic.invalid').pathname.slice(0, 500) || '/'; + } catch { + return raw.split(/[?#]/u, 1)[0].slice(0, 500) || '/'; + } +} + +export function diagnosticDurationMs(startedAt: bigint, finishedAt = process.hrtime.bigint()): number { + const duration = Number(finishedAt - startedAt) / 1_000_000; + return Math.max(0, Math.round(duration * 1_000) / 1_000); +} + +export function writeEmergencyDiagnostic( + diagnosticEvent: string, + error: unknown, + metadata: Record = {} +): void { + const record = { + level: 'error', + time: new Date().toISOString(), + service: 'ltjt-control-plane', + pid: process.pid, + diagnostic_event: safeDiagnosticCode(diagnosticEvent, 'service.emergency_error'), + ...metadata, + ...diagnosticError(error) + }; + process.stderr.write(`${JSON.stringify(record)}\n`); +} diff --git a/control-plane/src/input-attachment.ts b/control-plane/src/input-attachment.ts index c912e13..05042f5 100644 --- a/control-plane/src/input-attachment.ts +++ b/control-plane/src/input-attachment.ts @@ -3,6 +3,7 @@ import { request as httpsRequest } from 'node:https'; import { isIP } from 'node:net'; import { basename } from 'node:path'; import { sha256Bytes } from './crypto.js'; +import { diagnosticDurationMs } from './diagnostics.js'; const SHA256_PATTERN = /^[a-f0-9]{64}$/i; const BASE64_PATTERN = /^[A-Za-z0-9+/_-]*={0,2}$/; @@ -35,6 +36,11 @@ export interface AgentBusInputAttachmentReference { url: string; } +export type InputAttachmentDiagnostic = ( + event: string, + metadata: Record +) => void; + export class InputAttachmentError extends Error { constructor( public readonly code: string, @@ -46,6 +52,20 @@ export class InputAttachmentError extends Error { } } +function emitAttachmentDiagnostic( + diagnostic: InputAttachmentDiagnostic | undefined, + event: string, + metadata: Record +): void { + if (!diagnostic) return; + try { + diagnostic(event, metadata); + } catch { + // Diagnostics are deliberately non-authoritative and cannot interrupt + // validation, DNS pinning, downloading, or byte verification. + } +} + function extensionOf(fileName: string): string { const lower = fileName.toLowerCase(); if (lower.endsWith('.xlsx')) return '.xlsx'; @@ -282,28 +302,76 @@ async function downloadPinnedHttps( export async function downloadAgentBusInputAttachment( reference: AgentBusInputAttachmentReference, - maxBytes: number + maxBytes: number, + diagnostic?: InputAttachmentDiagnostic ): Promise { - let url = validateAgentBusAttachmentUrl(reference.url); - for (let redirects = 0; redirects <= MAX_REDIRECTS; redirects += 1) { - const addresses = await publicAddresses(url.hostname.replace(/^\[|\]$/g, '')); - const response = await downloadPinnedHttps(url, addresses[0], maxBytes); - if (response.statusCode >= 300 && response.statusCode < 400) { - if (!response.location || redirects === MAX_REDIRECTS) { - throw new InputAttachmentError('roster_attachment_redirect_invalid', '名单附件重定向无效或次数过多。'); + const startedAt = process.hrtime.bigint(); + let redirectCount = 0; + emitAttachmentDiagnostic(diagnostic, 'download_started', { + file_extension: extensionOf(reference.name), + declared_size: reference.size ?? null, + declared_sha256_present: Boolean(reference.sha256), + max_bytes: maxBytes + }); + try { + let url = validateAgentBusAttachmentUrl(reference.url); + for (let redirects = 0; redirects <= MAX_REDIRECTS; redirects += 1) { + redirectCount = redirects; + const dnsStartedAt = process.hrtime.bigint(); + emitAttachmentDiagnostic(diagnostic, 'dns_started', { redirect_count: redirects }); + const addresses = await publicAddresses(url.hostname.replace(/^\[|\]$/g, '')); + emitAttachmentDiagnostic(diagnostic, 'dns_validated', { + redirect_count: redirects, + address_count: addresses.length, + address_families: [...new Set(addresses.map((address) => address.family))].sort(), + selected_address_family: addresses[0]?.family || null, + duration_ms: diagnosticDurationMs(dnsStartedAt) + }); + const requestStartedAt = process.hrtime.bigint(); + const response = await downloadPinnedHttps(url, addresses[0], maxBytes); + emitAttachmentDiagnostic(diagnostic, 'http_response', { + redirect_count: redirects, + status_code: response.statusCode, + response_bytes: response.content.byteLength, + duration_ms: diagnosticDurationMs(requestStartedAt) + }); + if (response.statusCode >= 300 && response.statusCode < 400) { + if (!response.location || redirects === MAX_REDIRECTS) { + throw new InputAttachmentError('roster_attachment_redirect_invalid', '名单附件重定向无效或次数过多。'); + } + emitAttachmentDiagnostic(diagnostic, 'redirect_followed', { + redirect_count: redirects + 1, + status_code: response.statusCode + }); + url = validateAgentBusAttachmentUrl(new URL(response.location, url).toString()); + continue; } - url = validateAgentBusAttachmentUrl(new URL(response.location, url).toString()); - continue; + validateAttachmentBytes(response.content, maxBytes, reference.size, reference.sha256); + emitAttachmentDiagnostic(diagnostic, 'download_completed', { + redirect_count: redirects, + byte_size: response.content.byteLength, + declared_size_match: reference.size === undefined || reference.size === response.content.byteLength, + declared_sha256_verified: Boolean(reference.sha256), + duration_ms: diagnosticDurationMs(startedAt) + }); + return { + fileName: reference.name, + contentType: response.contentType === 'application/octet-stream' ? reference.contentType : response.contentType, + content: response.content, + declaredSize: reference.size, + declaredSha256: reference.sha256, + source: 'agentbus' + }; } - validateAttachmentBytes(response.content, maxBytes, reference.size, reference.sha256); - return { - fileName: reference.name, - contentType: response.contentType === 'application/octet-stream' ? reference.contentType : response.contentType, - content: response.content, - declaredSize: reference.size, - declaredSha256: reference.sha256, - source: 'agentbus' - }; + throw new InputAttachmentError('roster_attachment_download_failed', '名单附件下载失败。'); + } catch (error) { + emitAttachmentDiagnostic(diagnostic, 'download_failed', { + redirect_count: redirectCount, + duration_ms: diagnosticDurationMs(startedAt), + error_code: error instanceof InputAttachmentError + ? error.code + : 'roster_attachment_download_exception' + }); + throw error; } - throw new InputAttachmentError('roster_attachment_download_failed', '名单附件下载失败。'); } diff --git a/control-plane/src/migrate.ts b/control-plane/src/migrate.ts index 31239f5..897449d 100644 --- a/control-plane/src/migrate.ts +++ b/control-plane/src/migrate.ts @@ -3,6 +3,7 @@ import { basename, dirname, resolve } from 'node:path'; import { fileURLToPath } from 'node:url'; import { loadConfig } from './config.js'; import { getPool, closePool, quoteIdentifier, withTransaction } from './db.js'; +import { writeEmergencyDiagnostic } from './diagnostics.js'; const here = dirname(fileURLToPath(import.meta.url)); @@ -55,10 +56,17 @@ export async function migrateDatabase(): Promise { if (process.argv[1] && resolve(process.argv[1]) === resolve(fileURLToPath(import.meta.url))) { migrateDatabase() .then((applied) => { - console.log(JSON.stringify({ ok: true, applied })); + console.log(JSON.stringify({ + ok: true, + diagnostic_event: 'database.migration.completed', + diagnostic_stage: 'database_migration', + applied + })); }) .catch((error) => { - console.error(JSON.stringify({ ok: false, error: error.message || String(error) })); + writeEmergencyDiagnostic('database.migration.failed', error, { + diagnostic_stage: 'database_migration' + }); process.exitCode = 1; }) .finally(() => closePool()); diff --git a/control-plane/src/retention.ts b/control-plane/src/retention.ts index cc84d93..106653c 100644 --- a/control-plane/src/retention.ts +++ b/control-plane/src/retention.ts @@ -1,5 +1,6 @@ import { loadConfig } from './config.js'; import { closePool, withTransaction } from './db.js'; +import { writeEmergencyDiagnostic } from './diagnostics.js'; export async function runRetention(): Promise<{ deletedTasks: number; deletedAuditEvents: number; deletedSessions: number }> { const config = loadConfig(); @@ -34,9 +35,16 @@ export async function runRetention(): Promise<{ deletedTasks: number; deletedAud if (process.argv[1]?.endsWith('/retention.ts') || process.argv[1]?.endsWith('/retention.js')) { runRetention() - .then((result) => console.log(JSON.stringify({ ok: true, ...result }))) + .then((result) => console.log(JSON.stringify({ + ok: true, + diagnostic_event: 'retention.completed', + diagnostic_stage: 'data_retention', + ...result + }))) .catch((error) => { - console.error(JSON.stringify({ ok: false, error: error.message || String(error) })); + writeEmergencyDiagnostic('retention.failed', error, { + diagnostic_stage: 'data_retention' + }); process.exitCode = 1; }) .finally(() => closePool()); diff --git a/control-plane/src/server.ts b/control-plane/src/server.ts index b7f1235..23da793 100644 --- a/control-plane/src/server.ts +++ b/control-plane/src/server.ts @@ -1,12 +1,14 @@ import { randomUUID } from 'node:crypto'; +import { hostname } from 'node:os'; import { pathToFileURL } from 'node:url'; import { fileURLToPath } from 'node:url'; import { resolve } from 'node:path'; -import Fastify, { type FastifyReply, type FastifyRequest } from 'fastify'; +import Fastify, { LogController, type FastifyReply, type FastifyRequest } from 'fastify'; import cookie from '@fastify/cookie'; import helmet from '@fastify/helmet'; import rateLimit from '@fastify/rate-limit'; import fastifyStatic from '@fastify/static'; +import pino, { type DestinationStream } from 'pino'; import { z } from 'zod'; import { loadConfig, type AppConfig } from './config.js'; import { assertDatabaseSchema, closePool, databaseReadiness, databaseReady, getPool } from './db.js'; @@ -31,6 +33,14 @@ import { type EncodedTaskInputAttachment, type TaskInputAttachmentInput } from './input-attachment.js'; +import { + diagnosticDurationMs, + diagnosticError, + diagnosticMetadataKeys, + diagnosticRequestPath, + normalizeRequestId, + writeEmergencyDiagnostic +} from './diagnostics.js'; export function aiServiceConnected(databaseIsReady: boolean, probe: Record): boolean { return databaseIsReady && probe.configured === true && probe.reachable === true; @@ -120,8 +130,64 @@ const taskBulkDeleteSchema = z.object({ { message: '任务编号不能重复。', path: ['task_ids'] } ); +const LOG_REDACTION_PATHS = [ + 'req.headers.authorization', + 'req.headers.cookie', + 'req.headers["x-csrf-token"]', + 'request.headers.authorization', + 'request.headers.cookie', + 'authorization', + 'cookie', + 'password', + 'api_key', + 'token', + 'access_key', + '*.authorization', + '*.cookie', + '*.password', + '*.api_key', + '*.token', + '*.access_key' +]; + +export function createControlPlaneLogger(config: AppConfig, destination?: DestinationStream) { + const options = { + level: config.LOG_LEVEL, + base: { + pid: process.pid, + hostname: hostname(), + service: 'ltjt-control-plane', + environment: config.NODE_ENV, + deployment_revision: config.DEPLOYMENT_REVISION + }, + redact: { + paths: LOG_REDACTION_PATHS, + censor: '[REDACTED]' + } + }; + return destination ? pino(options, destination) : pino(options); +} + +function agentBusDiagnosticMetadata(metadata: Record): Record { + const rawEvent = String(metadata.agentbus_event || 'event'); + const event = /^[a-z0-9_]{1,80}$/u.test(rawEvent) ? rawEvent : 'event'; + return { + ...metadata, + diagnostic_event: `agentbus.${event}`, + diagnostic_stage: 'agentbus' + }; +} + function requestId(request: FastifyRequest): string { - return String(request.headers['x-request-id'] || randomUUID()); + return String(request.id); +} + +function requestTaskId(request: FastifyRequest): string | undefined { + const params = request.params && typeof request.params === 'object' + ? request.params as Record + : {}; + const taskId = String(params.taskId || '').trim(); + return taskId ? taskId.slice(0, 200) : undefined; } function clientAddress(request: FastifyRequest): string { @@ -251,15 +317,19 @@ async function loadExternalParser(): Promise { export async function buildServer({ config = loadConfig(), parser, - startParserLoop = true + startParserLoop = true, + loggerDestination }: { config?: AppConfig; parser?: ExternalParser; startParserLoop?: boolean; + loggerDestination?: DestinationStream; } = {}) { + const requestStartedAt = new WeakMap(); const app = Fastify({ - logger: { level: config.LOG_LEVEL, redact: ['req.headers.cookie', 'req.headers.authorization', '*.password', '*.api_key', '*.token'] }, - requestIdHeader: 'x-request-id', + loggerInstance: createControlPlaneLogger(config, loggerDestination), + logController: new LogController({ disableRequestLogging: true }), + genReqId: (rawRequest) => normalizeRequestId(rawRequest.headers['x-request-id']), trustProxy: true, bodyLimit: Math.min(75_000_000, Math.max(2_000_000, Math.ceil(config.ARTIFACT_MAX_BYTES * 1.4) + 1_000_000)) }); @@ -271,6 +341,52 @@ export async function buildServer({ prefix: '/' }); + app.log.info({ + diagnostic_event: 'service.initialized', + diagnostic_stage: 'startup', + host: config.HOST, + port: config.PORT, + log_level: config.LOG_LEVEL, + database_schema: config.DATABASE_SCHEMA, + database_ssl: config.DATABASE_SSL, + artifact_storage_backend: config.ARTIFACT_STORAGE_BACKEND, + agentbus_enabled: config.agentBusEnabled, + parser_loop_enabled: startParserLoop, + data_retention_enabled: config.DATA_RETENTION_ENABLED, + raw_payload_logging: config.AGENTBUS_LOG_PAYLOADS + }, 'control plane initialized'); + + app.addHook('onRequest', async (request, reply) => { + requestStartedAt.set(request, process.hrtime.bigint()); + reply.header('x-request-id', requestId(request)); + const contentLength = Number(request.headers['content-length']); + request.log.info({ + diagnostic_event: 'http.request.started', + diagnostic_stage: 'http', + request_id: requestId(request), + method: request.method, + path: diagnosticRequestPath(request.url), + ...(Number.isSafeInteger(contentLength) && contentLength >= 0 ? { content_length: contentLength } : {}) + }, 'HTTP request started'); + }); + + app.addHook('onResponse', async (request, reply) => { + const startedAt = requestStartedAt.get(request); + requestStartedAt.delete(request); + const metadata = { + diagnostic_event: 'http.request.completed', + diagnostic_stage: 'http', + request_id: requestId(request), + method: request.method, + path: diagnosticRequestPath(request.routeOptions.url || request.url), + status_code: reply.statusCode, + ...(startedAt ? { duration_ms: diagnosticDurationMs(startedAt) } : {}) + }; + if (reply.statusCode >= 500) request.log.error(metadata, 'HTTP request completed with server error'); + else if (reply.statusCode >= 400) request.log.warn(metadata, 'HTTP request completed with client error'); + else request.log.info(metadata, 'HTTP request completed'); + }); + app.get('/history', async (_request, reply) => { reply.header('Cache-Control', 'no-store'); return reply.sendFile('index.html'); @@ -292,8 +408,16 @@ export async function buildServer({ if (redirectUrl) return reply.redirect(redirectUrl, 308); }); - const auth = new AuthService(config); - const tasks = new TaskService(config); + const auth = new AuthService(config, { + info: (metadata, message) => app.log.info(metadata, message), + warn: (metadata, message) => app.log.warn(metadata, message), + error: (metadata, message) => app.log.error(metadata, message) + }); + const tasks = new TaskService(config, undefined, { + info: (metadata, message) => app.log.info(metadata, message), + warn: (metadata, message) => app.log.warn(metadata, message), + error: (metadata, message) => app.log.error(metadata, message) + }); const externalParser = parser || await loadExternalParser(); const parserOrchestrator = new ParserOrchestrator(externalParser); const activeParseWorkers = new Map(); @@ -302,7 +426,11 @@ export async function buildServer({ let aiProbeValue: unknown; let aiProbeExpiresAt = 0; let aiProbeInFlight: Promise | null = null; - const channelService = new AgentBusChannelService(config); + const channelService = new AgentBusChannelService(config, { + info: (metadata, message) => app.log.info(agentBusDiagnosticMetadata(metadata), message), + warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message), + error: (metadata, message) => app.log.error(agentBusDiagnosticMetadata(metadata), message) + }); let agentBus: AgentBusManager | null = null; const getSession = async (request: FastifyRequest): Promise => { @@ -333,6 +461,7 @@ export async function buildServer({ result: unknown, decision?: ParseDecisionInput ): Promise { + const startedAt = process.hrtime.bigint(); const context: TaskContext = { organizationId: claim.task.organization_id, userId: '', @@ -350,29 +479,65 @@ export async function buildServer({ decision ); app.log.info({ + diagnostic_event: 'parser.outcome.persisted', + diagnostic_stage: 'parser_persistence', + request_id: context.requestId, task_id: finalized.task_id, attempt_no: claim.attemptNo, - status: finalized.status + status: finalized.status, + duration_ms: diagnosticDurationMs(startedAt) }, 'parse task finalized'); return; } catch (error) { if (error instanceof TaskError && ['stale_parse_result', 'task_cancelled', 'task_not_found'].includes(error.code)) { - app.log.info({ task_id: claim.task.task_id, attempt_no: claim.attemptNo, error_code: error.code }, 'late parse outcome ignored'); + app.log.info({ + diagnostic_event: 'parser.outcome.ignored', + diagnostic_stage: 'parser_persistence', + request_id: context.requestId, + task_id: claim.task.task_id, + attempt_no: claim.attemptNo, + error_code: error.code, + duration_ms: diagnosticDurationMs(startedAt) + }, 'late parse outcome ignored'); return; } lastError = error; + app.log.warn({ + diagnostic_event: 'parser.outcome.persist_retry', + diagnostic_stage: 'parser_persistence', + request_id: context.requestId, + task_id: claim.task.task_id, + attempt_no: claim.attemptNo, + persistence_attempt: attempt + 1, + retrying: attempt < 2, + ...diagnosticError(error, workerErrorCode(error)) + }, 'parse outcome persistence attempt failed'); if (attempt < 2) await waitMs(250 * (attempt + 1)); } } app.log.error({ + diagnostic_event: 'parser.outcome.persist_failed', + diagnostic_stage: 'parser_persistence', + request_id: context.requestId, task_id: claim.task.task_id, attempt_no: claim.attemptNo, - error_code: workerErrorCode(lastError), - error_name: lastError instanceof Error ? lastError.name : 'unknown' + duration_ms: diagnosticDurationMs(startedAt), + ...diagnosticError(lastError, workerErrorCode(lastError)) }, 'parse outcome persistence failed after retries'); } async function runParseTask(claim: ParseTaskClaim): Promise { + const startedAt = process.hrtime.bigint(); + const parseRequestId = `parse:${claim.task.task_id}:attempt:${claim.attemptNo}`; + app.log.info({ + diagnostic_event: 'parser.worker.started', + diagnostic_stage: 'parser_execution', + request_id: parseRequestId, + task_id: claim.task.task_id, + attempt_no: claim.attemptNo, + parser_mode: claim.task.parser.configured_mode, + business_route_id: claim.task.parser.route_id + }, 'parse worker started'); const controller = new AbortController(); let timeout: NodeJS.Timeout | undefined; const parserPromise = Promise.resolve().then(() => parserOrchestrator.parse(claim, controller.signal)); @@ -391,6 +556,16 @@ export async function buildServer({ }) ]); if (outcome.timedOut) { + app.log.error({ + diagnostic_event: 'parser.worker.timeout', + diagnostic_stage: 'parser_execution', + request_id: parseRequestId, + task_id: claim.task.task_id, + attempt_no: claim.attemptNo, + timeout_ms: config.PARSE_WORKER_TIMEOUT_MS, + duration_ms: diagnosticDurationMs(startedAt), + error_code: 'parse_worker_timeout' + }, 'parse worker timed out'); await persistParseOutcome( claim, workerFailureResult('parse_worker_timeout', 'worker_watchdog_timeout') @@ -398,13 +573,24 @@ export async function buildServer({ return; } await persistParseOutcome(claim, outcome.result.result, outcome.result.decision); + app.log.info({ + diagnostic_event: 'parser.worker.completed', + diagnostic_stage: 'parser_execution', + request_id: parseRequestId, + task_id: claim.task.task_id, + attempt_no: claim.attemptNo, + duration_ms: diagnosticDurationMs(startedAt) + }, 'parse worker completed'); } catch (error) { const errorCode = workerErrorCode(error); app.log.error({ + diagnostic_event: 'parser.worker.failed', + diagnostic_stage: 'parser_execution', + request_id: parseRequestId, task_id: claim.task.task_id, attempt_no: claim.attemptNo, - error_code: errorCode, - error_name: error instanceof Error ? error.name : 'unknown' + duration_ms: diagnosticDurationMs(startedAt), + ...diagnosticError(error, errorCode) }, 'parse worker failed'); await persistParseOutcome(claim, workerFailureResult(errorCode, 'worker_exception')); } finally { @@ -415,28 +601,48 @@ export async function buildServer({ async function processParseQueue(): Promise { const recovered = await tasks.recoverExpiredParseTasks(); if (recovered > 0) { - app.log.warn({ recovered_tasks: recovered }, 'expired parse tasks were durably blocked'); + app.log.warn({ + diagnostic_event: 'parser.queue.expired_tasks_recovered', + diagnostic_stage: 'parser_queue', + recovered_tasks: recovered + }, 'expired parse tasks were durably blocked'); } const recoveredExecutions = await tasks.recoverExpiredExecutionTasks(); if (recoveredExecutions > 0) { - app.log.warn({ recovered_tasks: recoveredExecutions }, 'expired ERP executions were automatically failed and released'); + app.log.warn({ + diagnostic_event: 'erp.execution.expired_tasks_recovered', + diagnostic_stage: 'erp_execution', + recovered_tasks: recoveredExecutions + }, 'expired ERP executions were automatically failed and released'); } const staleReconciliations = await tasks.maintainStaleReconciliationTasks(); if (staleReconciliations > 0) { - app.log.warn({ stale_tasks: staleReconciliations }, 'stale reconciliation tasks were automatically failed and released'); + app.log.warn({ + diagnostic_event: 'erp.reconciliation.stale_tasks_recovered', + diagnostic_stage: 'erp_reconciliation', + stale_tasks: staleReconciliations + }, 'stale reconciliation tasks were automatically failed and released'); } while (activeParseWorkers.size < 2) { const workerId = `parse:${process.pid}:${randomUUID()}`; const claim = await tasks.claimNextParseTask(workerId, [...activeParseWorkers.keys()]); if (!claim) break; activeParseWorkers.set(claim.task.task_id, workerId); + app.log.info({ + diagnostic_event: 'parser.queue.claimed', + diagnostic_stage: 'parser_queue', + task_id: claim.task.task_id, + attempt_no: claim.attemptNo, + active_workers: activeParseWorkers.size + }, 'parse queue task claimed'); void runParseTask(claim) .catch((error) => { app.log.error({ + diagnostic_event: 'parser.runner.crashed', + diagnostic_stage: 'parser_execution', task_id: claim.task.task_id, attempt_no: claim.attemptNo, - error_code: workerErrorCode(error), - error_name: error instanceof Error ? error.name : 'unknown' + ...diagnosticError(error, workerErrorCode(error)) }, 'parse task runner crashed'); }) .finally(() => { @@ -454,7 +660,11 @@ export async function buildServer({ const now = Date.now(); if (now - lastParseQueueErrorAt < 60_000) return; lastParseQueueErrorAt = now; - app.log.error({ error_code: workerErrorCode(error), error_name: error instanceof Error ? error.name : 'unknown' }, 'parse queue tick failed'); + app.log.error({ + diagnostic_event: 'parser.queue.tick_failed', + diagnostic_stage: 'parser_queue', + ...diagnosticError(error, workerErrorCode(error)) + }, 'parse queue tick failed'); }) .finally(() => { parseQueueInFlight = null; @@ -476,9 +686,9 @@ export async function buildServer({ organizationId: organization.id, scheduleParseQueue, logger: { - info: (metadata, message) => app.log.info(metadata, message), - warn: (metadata, message) => app.log.warn(metadata, message), - error: (metadata, message) => app.log.error(metadata, message) + info: (metadata, message) => app.log.info(agentBusDiagnosticMetadata(metadata), message), + warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message), + error: (metadata, message) => app.log.error(agentBusDiagnosticMetadata(metadata), message) } }); await agentBus.start(); @@ -496,7 +706,11 @@ export async function buildServer({ try { return await externalParser.checkConnection(); } catch (error) { - app.log.warn({ error: error instanceof Error ? error.message : String(error) }, 'external parser status probe failed'); + app.log.warn({ + diagnostic_event: 'parser.status_probe.failed', + diagnostic_stage: 'parser_probe', + ...diagnosticError(error, 'probe_failed') + }, 'external parser status probe failed'); return { configured: Boolean(config.DEERFLOW_OPEN_API_KEY), reachable: false, ok: false, error_code: 'probe_failed' }; } })(); @@ -513,6 +727,14 @@ export async function buildServer({ app.get('/health/ready', async (_request, reply) => { const readiness = await databaseReadiness(config); if (!readiness.ready) { + app.log.warn({ + diagnostic_event: 'health.readiness.failed', + diagnostic_stage: 'health', + database_ready: readiness.database, + schema_ready: readiness.schema, + required_migration: readiness.requiredMigration, + error_code: readiness.errorCode || (readiness.database ? 'database_schema_outdated' : 'database_unavailable') + }, 'control plane readiness check failed'); return reply.code(503).send({ ok: false, database: readiness.database, @@ -521,6 +743,14 @@ export async function buildServer({ ...(readiness.errorCode ? { error_code: readiness.errorCode } : {}) }); } + app.log.info({ + diagnostic_event: 'health.readiness.passed', + diagnostic_stage: 'health', + database_ready: true, + schema_ready: true, + required_migration: readiness.requiredMigration, + agentbus_connected: agentBus?.status().connected || false + }, 'control plane readiness check passed'); return { ok: true, database: true, @@ -909,6 +1139,16 @@ export async function buildServer({ app.setErrorHandler((error, request, reply) => { if (error instanceof AuthError || error instanceof TaskError) { + request.log.warn({ + diagnostic_event: 'http.request.rejected', + diagnostic_stage: 'http', + request_id: requestId(request), + ...(requestTaskId(request) ? { task_id: requestTaskId(request) } : {}), + status_code: error.statusCode, + error_code: error.code, + error_name: error.name, + ...(error instanceof TaskError ? { error_detail_keys: diagnosticMetadataKeys(error.details) } : {}) + }, 'HTTP request rejected by application guard'); return reply.code(error.statusCode).send({ ok: false, error_code: error.code, @@ -917,9 +1157,25 @@ export async function buildServer({ }); } if (error instanceof z.ZodError) { + request.log.warn({ + diagnostic_event: 'http.request.invalid', + diagnostic_stage: 'http', + request_id: requestId(request), + ...(requestTaskId(request) ? { task_id: requestTaskId(request) } : {}), + status_code: 400, + error_code: 'invalid_request', + validation_paths: error.issues.map((issue) => issue.path.join('.')).slice(0, 100) + }, 'HTTP request validation failed'); return reply.code(400).send({ ok: false, error_code: 'invalid_request', message: '请求参数不符合要求。', details: error.issues.map((issue) => issue.path.join('.')) }); } - request.log.error({ error: error instanceof Error ? error.message : String(error), request_id: requestId(request) }, 'unhandled request error'); + request.log.error({ + diagnostic_event: 'http.request.failed', + diagnostic_stage: 'http', + request_id: requestId(request), + ...(requestTaskId(request) ? { task_id: requestTaskId(request) } : {}), + status_code: 500, + ...diagnosticError(error, 'server_error') + }, 'unhandled request error'); return reply.code(500).send({ ok: false, error_code: 'server_error', message: '服务暂时不可用。' }); }); @@ -928,18 +1184,75 @@ export async function buildServer({ app.addHook('onClose', async () => clearInterval(interval)); } - app.addHook('onClose', async () => closePool()); + app.addHook('onClose', async () => { + app.log.info({ + diagnostic_event: 'service.closing', + diagnostic_stage: 'shutdown' + }, 'control plane closing'); + await closePool(); + }); return { app, auth, tasks, agentBus, channelService }; } +function installProcessDiagnostics(app: Awaited>['app']): void { + let shuttingDown = false; + const shutdown = async (reason: string, exitCode: number, error?: unknown): Promise => { + if (shuttingDown) return; + shuttingDown = true; + process.exitCode = exitCode; + const metadata = { + diagnostic_event: 'service.shutdown.started', + diagnostic_stage: 'shutdown', + shutdown_reason: reason, + exit_code: exitCode, + ...(error === undefined ? {} : diagnosticError(error, reason)) + }; + if (exitCode === 0) app.log.info(metadata, 'control plane shutdown started'); + else app.log.error(metadata, 'control plane shutdown started after fatal error'); + try { + await app.close(); + app.log.info({ + diagnostic_event: 'service.shutdown.completed', + diagnostic_stage: 'shutdown', + shutdown_reason: reason, + exit_code: exitCode + }, 'control plane shutdown completed'); + } catch (closeError) { + app.log.error({ + diagnostic_event: 'service.shutdown.failed', + diagnostic_stage: 'shutdown', + shutdown_reason: reason, + exit_code: 1, + ...diagnosticError(closeError, 'shutdown_failed') + }, 'control plane shutdown failed'); + process.exitCode = 1; + } + }; + process.once('SIGTERM', () => void shutdown('sigterm', 0)); + process.once('SIGINT', () => void shutdown('sigint', 0)); + process.once('uncaughtException', (error) => void shutdown('uncaught_exception', 1, error)); + process.once('unhandledRejection', (error) => void shutdown('unhandled_rejection', 1, error)); +} + async function main(): Promise { const config = loadConfig(); await assertDatabaseSchema(config); const { app } = await buildServer({ config }); + installProcessDiagnostics(app); try { await app.listen({ host: config.HOST, port: config.PORT }); - app.log.info({ host: config.HOST, port: config.PORT }, 'LianSyn-platform control plane listening'); + app.log.info({ + diagnostic_event: 'service.listening', + diagnostic_stage: 'startup', + host: config.HOST, + port: config.PORT + }, 'LianSyn-platform control plane listening'); } catch (error) { + app.log.error({ + diagnostic_event: 'service.listen.failed', + diagnostic_stage: 'startup', + ...diagnosticError(error, 'listen_failed') + }, 'control plane failed to listen'); await app.close().catch(() => undefined); throw error; } @@ -947,7 +1260,7 @@ async function main(): Promise { if (process.argv[1] && resolve(process.argv[1]) === resolve(fileURLToPath(import.meta.url))) { main().catch(async (error) => { - console.error(error); + writeEmergencyDiagnostic('service.startup_failed', error, { diagnostic_stage: 'startup' }); await closePool(); process.exitCode = 1; }); diff --git a/control-plane/src/task-service.ts b/control-plane/src/task-service.ts index d44e76c..14a290f 100644 --- a/control-plane/src/task-service.ts +++ b/control-plane/src/task-service.ts @@ -32,6 +32,11 @@ import { import { convertDocumentToPdf, convertDocumentToXlsx } from './document-converter.js'; import { sanitizeOperationTiming } from './operation-timing.js'; import type { TaskInputAttachmentInput } from './input-attachment.js'; +import { + diagnosticMetadataKeys, + noopDiagnosticLogger, + type DiagnosticLogger +} from './diagnostics.js'; import { normalizePassengerRosterWorkbook, PassengerRosterWorkbookError, @@ -1860,12 +1865,28 @@ async function enforceFixedParserModeForClaim( export class TaskService { readonly events = new EventEmitter(); private readonly artifactStore: TaskArtifactStore; + private readonly logger: DiagnosticLogger; constructor( private readonly config: AppConfig, - artifactStore = createTaskArtifactStore(config) + artifactStore?: TaskArtifactStore, + logger: DiagnosticLogger = noopDiagnosticLogger ) { - this.artifactStore = artifactStore; + this.logger = logger; + this.artifactStore = artifactStore || createTaskArtifactStore(config, logger); + } + + private log( + level: 'info' | 'warn' | 'error', + metadata: Record, + message: string + ): void { + try { + this.logger[level](metadata, message); + } catch { + // Operational logging is deliberately non-authoritative and must never + // alter task state, attachment processing, or ERP execution behavior. + } } private publicTask(row: Record, events: PublicTaskEvent[] = []): PublicTask { @@ -2042,6 +2063,15 @@ export class TaskService { } private notify(event: TaskEvent): void { + this.log('info', { + diagnostic_event: 'task.state.emitted', + diagnostic_stage: event.stage || 'task', + task_id: event.task_id, + task_status: event.status, + task_stage: event.stage, + event_id: event.id, + payload_keys: diagnosticMetadataKeys(event.payload) + }, 'task state event emitted'); this.events.emit('task', event); } @@ -2059,6 +2089,16 @@ export class TaskService { VALUES ($1, $2, $3, $4, $5, $6, $7)`, [context.organizationId, context.userId || null, eventType, entityType, entityId, context.requestId, metadata] ); + this.log('info', { + diagnostic_event: 'audit.event.staged', + diagnostic_stage: 'audit', + request_id: context.requestId, + domain_event: eventType, + entity_type: entityType, + entity_id: entityId, + actor_present: Boolean(context.userId), + metadata_keys: diagnosticMetadataKeys(metadata) + }, 'audit event staged in transaction'); } private async recordAgentBusAcceptedDelivery( diff --git a/control-plane/test/agentbus.test.ts b/control-plane/test/agentbus.test.ts index 9ae54cb..8e61a9e 100644 --- a/control-plane/test/agentbus.test.ts +++ b/control-plane/test/agentbus.test.ts @@ -770,7 +770,12 @@ test('WeChat attachment placeholder without file metadata fails closed before ta '附件内容未传到平台，原任务仍在等待附件。请检查微信桥接器的文件转发后重新发送。' ); assert.ok(logs.some((entry) => entry.metadata.agentbus_event === 'task_processing_failed' - && entry.metadata.error_code === 'roster_attachment_metadata_missing')); + && 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')); }); diff --git a/control-plane/test/diagnostics.test.ts b/control-plane/test/diagnostics.test.ts new file mode 100644 index 0000000..ff0b7cb --- /dev/null +++ b/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> } { + const chunks: string[] = []; + const records: Array> = []; + 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); + } + 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> = []; + 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 + ): Promise; + }; + 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 + ); +}); diff --git a/control-plane/test/input-attachment.test.ts b/control-plane/test/input-attachment.test.ts index 5643753..2c71b97 100644 --- a/control-plane/test/input-attachment.test.ts +++ b/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 }> = []; + 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/); diff --git a/docker-compose.yml b/docker-compose.yml index 1a3ba59..0a305f4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,3 +1,9 @@ +x-default-logging: &default-logging + driver: json-file + options: + max-size: "${LOG_MAX_SIZE:-20m}" + max-file: "${LOG_MAX_FILES:-10}" + services: postgres: profiles: ["local"] @@ -9,6 +15,7 @@ services: POSTGRES_PASSWORD: ${LOCAL_POSTGRES_PASSWORD:-local-dev-only-change-me} volumes: - ltjt-postgres:/var/lib/postgresql/data + logging: *default-logging healthcheck: test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB}"] interval: 10s @@ -23,6 +30,7 @@ services: environment: DATABASE_URL: ${DATABASE_URL:?set DATABASE_URL in .env.production} command: ["sh", "-c", "node .build/control-plane/src/migrate.js && node .build/control-plane/src/server.js"] + logging: *default-logging healthcheck: test: ["CMD", "node", "-e", "fetch('http://127.0.0.1:8786/health/ready').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))"] interval: 15s @@ -44,6 +52,7 @@ services: - ./infra/Caddyfile:/etc/caddy/Caddyfile:ro - caddy-data:/data - caddy-config:/config + logging: *default-logging volumes: ltjt-postgres: diff --git a/infra/diagnose-server.sh b/infra/diagnose-server.sh new file mode 100755 index 0000000..202cca4 --- /dev/null +++ b/infra/diagnose-server.sh @@ -0,0 +1,97 @@ +#!/usr/bin/env sh +set -eu + +since=30m +tail_lines=1000 +match= +all_services=false + +usage() { + printf '%s\n' 'Usage: sh infra/diagnose-server.sh [--since 30m] [--tail 1000] [--match VALUE] [--all-services]' + printf '%s\n' 'Filters are literal and can be a request ID, TASK-* ID, AgentBus frame/conversation/channel ID, diagnostic event, or error code.' +} + +while [ "$#" -gt 0 ]; do + case "$1" in + --since) + [ "$#" -ge 2 ] || { usage >&2; exit 2; } + since=$2 + shift 2 + ;; + --tail) + [ "$#" -ge 2 ] || { usage >&2; exit 2; } + tail_lines=$2 + shift 2 + ;; + --match) + [ "$#" -ge 2 ] || { usage >&2; exit 2; } + match=$2 + shift 2 + ;; + --all-services) + all_services=true + shift + ;; + -h|--help) + usage + exit 0 + ;; + *) + usage >&2 + exit 2 + ;; + esac +done + +if ! printf '%s' "$since" | grep -Eq '^[0-9]+[smhd]$'; then + printf '%s\n' 'diagnose-server: --since must use a bounded value such as 30m, 2h, or 1d.' >&2 + exit 2 +fi +case "$tail_lines" in + *[!0-9]*|'') + printf '%s\n' 'diagnose-server: --tail must be a positive integer.' >&2 + exit 2 + ;; +esac +if [ "$tail_lines" -lt 1 ] || [ "$tail_lines" -gt 50000 ]; then + printf '%s\n' 'diagnose-server: --tail must be between 1 and 50000.' >&2 + exit 2 +fi + +cd "$(dirname "$0")/.." + +compose() { + if [ -f .env.production ]; then + docker compose --env-file .env.production "$@" + else + docker compose "$@" + fi +} + +command -v docker >/dev/null 2>&1 || { + printf '%s\n' 'diagnose-server: docker is not installed or not on PATH.' >&2 + exit 1 +} + +printf '%s\n' 'Container state:' +compose ps + +printf '%s\n' 'Control-plane readiness:' +if ! compose exec -T control-plane node -e "fetch('http://127.0.0.1:8786/health/ready').then(async response => { const body = await response.json(); console.log(JSON.stringify({ status: response.status, body })); process.exit(response.ok ? 0 : 1); }).catch(error => { console.error(JSON.stringify({ status: 0, error_code: error && error.code ? String(error.code) : 'health_request_failed' })); process.exit(1); });"; then + printf '%s\n' 'diagnose-server: readiness is unhealthy; recent logs follow.' >&2 +fi + +printf 'Recent logs (since=%s, tail=%s):\n' "$since" "$tail_lines" +if [ "$all_services" = true ]; then + if [ -n "$match" ]; then + compose logs --no-color --timestamps --since "$since" --tail "$tail_lines" \ + | grep -F -- "$match" || true + else + compose logs --no-color --timestamps --since "$since" --tail "$tail_lines" + fi +elif [ -n "$match" ]; then + compose logs --no-color --timestamps --since "$since" --tail "$tail_lines" control-plane \ + | grep -F -- "$match" || true +else + compose logs --no-color --timestamps --since "$since" --tail "$tail_lines" control-plane +fi diff --git a/infra/predeploy-check.sh b/infra/predeploy-check.sh index e6bcf37..23d4dc2 100755 --- a/infra/predeploy-check.sh +++ b/infra/predeploy-check.sh @@ -13,5 +13,5 @@ fi npm test npm run build -sh -n infra/backup-postgres.sh infra/restore-check.sh +sh -n infra/backup-postgres.sh infra/restore-check.sh infra/diagnose-server.sh printf '%s\n' 'predeploy-check: static, test, build, and script gates passed' diff --git a/package.json b/package.json index 1985e82..722b894 100644 --- a/package.json +++ b/package.json @@ -14,6 +14,7 @@ "db:migrate": "node --env-file=.env --import tsx control-plane/src/migrate.ts", "admin": "node --env-file=.env --import tsx control-plane/src/admin-cli.ts", "data:retention": "node --env-file=.env --import tsx control-plane/src/retention.ts", + "diagnose:server": "sh infra/diagnose-server.sh", "check": "tsc --noEmit -p tsconfig.json", "check:repo": "node --test tools/repository-hygiene.test.mjs", "test:control-plane": "node --test --import tsx control-plane/test/*.test.ts", diff --git a/tools/repository-hygiene.test.mjs b/tools/repository-hygiene.test.mjs index 00978cb..c61765f 100644 --- a/tools/repository-hygiene.test.mjs +++ b/tools/repository-hygiene.test.mjs @@ -289,6 +289,7 @@ test('single-source and generated-output boundaries remain explicit', async () = const packageJson = JSON.parse(await readFile(path.join(ROOT, 'package.json'), 'utf8')); const dockerfile = await readFile(path.join(ROOT, 'Dockerfile'), 'utf8'); const compose = await readFile(path.join(ROOT, 'docker-compose.yml'), 'utf8'); + const diagnoseServer = await readFile(path.join(ROOT, 'infra/diagnose-server.sh'), 'utf8'); const gitignore = await readFile(path.join(ROOT, '.gitignore'), 'utf8'); assert.match(template, /^交付版本：0\.5\.123\s*$/mu); @@ -298,6 +299,13 @@ test('single-source and generated-output boundaries remain explicit', async () = assert.match(packageJson.scripts.start, /\.build\/control-plane/u); assert.match(dockerfile, /\.build\/control-plane/u); assert.match(compose, /\.build\/control-plane/u); + assert.match(compose, /x-default-logging:\s*&default-logging/u); + assert.match(compose, /max-size:\s*"\$\{LOG_MAX_SIZE:-20m\}"/u); + assert.match(compose, /max-file:\s*"\$\{LOG_MAX_FILES:-10\}"/u); + assert.equal((compose.match(/logging:\s*\*default-logging/gu) || []).length, 3); + assert.match(diagnoseServer, /health\/ready/u); + assert.match(diagnoseServer, /docker compose --env-file \.env\.production/u); + assert.doesNotMatch(diagnoseServer, /cat\s+\.env|printenv|env\s*$/mu); assert.match(gitignore, /^\.build\/$/mu); assert.match(gitignore, /^\.env$/mu); assert.ok(!existsSync(path.join(DIST, 'control-plane'))); From 1247dd1cbd1d6695f8f03b1c2deaf62051ff28ea Mon Sep 17 00:00:00 2001 From: inman Date: Mon, 31 Aug 2026 09:51:52 +0800 Subject: [PATCH 3/3] chore: record server diagnostics push --- ...260831-push-server-diagnostics-7c4e91a2.md | 43 +++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 .project-docs/30-worklog/tasks/20260831-push-server-diagnostics-7c4e91a2.md diff --git a/.project-docs/30-worklog/tasks/20260831-push-server-diagnostics-7c4e91a2.md b/.project-docs/30-worklog/tasks/20260831-push-server-diagnostics-7c4e91a2.md new file mode 100644 index 0000000..0e1c106 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260831-push-server-diagnostics-7c4e91a2.md @@ -0,0 +1,43 @@ +# Task: Push server diagnostics branch + +## Identity + +- Task ID: 20260831-push-server-diagnostics-7c4e91a2 +- 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: a3963ad0f629818edfa921261acfeb28dde9c9dd +- Owner: codex +- Status: Ready for integration + +## Scope + +- Push the existing, tested server-diagnostics feature branch to the configured `origin` remote and establish upstream tracking. +- Preserve the exact local commits and publish them under the same feature-branch name. +- Record and verify the remote ref without changing source code, canonical project memory, release artifacts, or deployment state. + +## Intent And Constraints + +- The requested operation authorizes a Git remote push only; it does not authorize merging, rebasing, force-pushing, changing `main`, deploying, restarting services, or mutating live systems. +- Remote `main` advanced independently to `98282d1` and diverges from this branch at `7b5d855`; publish the feature branch as-is so later integration can reconcile the histories explicitly. +- Push only to `origin` (`LWLT-AIBOT.git`), not the separately configured legacy repository. +- Preserve the validated server-diagnostics payload commit `a3963ad0f629818edfa921261acfeb28dde9c9dd`. + +## Outcome + +- The current feature branch was published to the same-named branch on `origin` with upstream tracking. +- Remote `main`, the legacy remote, deployment state, and live services were not changed. + +## Verification + +- `git fetch origin` completed and showed remote `main` at `98282d127c989a7e9ca1d688c9d3d2fadd73e8bc`. +- Pre-push `git ls-remote --heads origin ` returned no existing branch, so the push creates a new remote feature ref rather than overwriting shared history. +- After push, verify that the remote feature ref equals local `HEAD` and that branch upstream tracking points to the same `origin` ref. + +## Follow-ups + +- Integration or merge into `main` remains a separate, explicitly authorized task because the remote histories have diverged. + +## Promotion Candidates + +- None; this task publishes an existing feature branch and introduces no canonical project fact or behavior change.