From a3963ad0f629818edfa921261acfeb28dde9c9dd Mon Sep 17 00:00:00 2001 From: inman Date: Sun, 30 Aug 2026 16:01:42 +0800 Subject: [PATCH] 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')));