diff --git a/benchmarks/lib/langfuse-drafts.ts b/benchmarks/lib/langfuse-drafts.ts new file mode 100644 index 0000000..16c68b2 --- /dev/null +++ b/benchmarks/lib/langfuse-drafts.ts @@ -0,0 +1,487 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { privateJson } from './contracts.js'; + +const OBSERVATION_FIELDS = 'core,metadata'; +const MAX_OBSERVATION_PAGES = 20; +const MAX_SCORE_PAGES = 100; +const MAX_ROWS = 20_000; +const TERMINAL_REASONS = new Set([ + 'completed', + 'released', + 'runner_exit', + 'stopped', +]); +const ERROR_CATEGORIES = new Set([ + 'provider_failure', + 'runner_error', + 'runtime_error', + 'tool_error', + 'runner_exit', + 'stopped', +]); +const OPAQUE_ID = /^[A-Za-z0-9_.:-]{1,160}$/; +const TOOL_NAME = /^[A-Za-z0-9_.:-]{1,120}$/; + +export interface LangfuseObservation { + id?: unknown; + traceId?: unknown; + startTime?: unknown; + name?: unknown; + metadata?: unknown; + input?: unknown; + output?: unknown; + statusMessage?: unknown; + [key: string]: unknown; +} + +export interface LangfuseScore { + dataType?: unknown; + value?: unknown; + subject?: unknown; + comment?: unknown; + metadata?: unknown; + [key: string]: unknown; +} + +export interface LangfuseDraftOptions { + endpoint: string; + publicKey: string; + secretKey: string; + from: Date; + to: Date; + minScore: number; + fetcher?: typeof fetch; +} + +export interface LangfuseDraftBundle { + schemaVersion: 1; + generatedAt: string; + selection: { + failedTraceCount: number; + lowScoreTraceCount: number; + draftCount: number; + skippedMissingRunIdentity: number; + bodyFieldsRequested: false; + holdoutRead: false; + casesModified: false; + }; + drafts: Array>; +} + +interface SelectedDraft { + runId: string | null; + turnId: string | null; + terminalReason: string | null; + errorCategory: string | null; + lowScoreCount: number; + failed: boolean; + toolNames: string[]; +} + +export function isSafeDraftEndpoint(value: string): boolean { + try { + const url = new URL(value); + return ( + (url.protocol === 'https:' || + (url.protocol === 'http:' && + ['localhost', '127.0.0.1'].includes(url.hostname.toLowerCase()))) && + !url.username && + !url.password && + !url.search && + !url.hash + ); + } catch { + return false; + } +} + +function assertMetadataOnly(row: Record): void { + for (const field of [ + 'input', + 'output', + 'statusMessage', + 'comment', + 'metadataText', + 'prompt', + 'messages', + 'model', + 'userId', + 'sessionId', + ]) { + if (Object.hasOwn(row, field) && containsContent(row[field])) + throw new Error('langfuse_content_field_rejected'); + } +} + +function containsContent(value: unknown): boolean { + if (value === null || value === undefined) return false; + if (typeof value === 'string') return value.trim().length > 0; + if (Array.isArray(value)) return value.some(containsContent); + if (typeof value === 'object') + return Object.values(value).some(containsContent); + return true; +} + +function safeOpaqueId(value: unknown): string | null { + return typeof value === 'string' && OPAQUE_ID.test(value) ? value : null; +} + +function safeToolName(value: unknown): string | null { + return typeof value === 'string' && TOOL_NAME.test(value) ? value : null; +} + +function safeEnum(value: unknown, allowed: Set): string | null { + return typeof value === 'string' && allowed.has(value) ? value : null; +} + +function metadataOf(row: LangfuseObservation): Record { + assertMetadataOnly(row); + return row.metadata && typeof row.metadata === 'object' + ? (row.metadata as Record) + : {}; +} + +function observationTraceId(row: LangfuseObservation): string | null { + assertMetadataOnly(row); + return typeof row.traceId === 'string' && row.traceId.length <= 160 + ? row.traceId + : null; +} + +function observationId(row: LangfuseObservation): string | null { + assertMetadataOnly(row); + return typeof row.id === 'string' && row.id.length <= 160 ? row.id : null; +} + +function scoreTraceId( + score: LangfuseScore, + traceByObservationId: Map, +): string | null { + assertMetadataOnly(score); + const subject = score.subject; + if (!subject || typeof subject !== 'object') return null; + const value = subject as Record; + if (value.kind === 'trace' && typeof value.id === 'string') return value.id; + if (value.kind === 'observation' && typeof value.id === 'string') + return traceByObservationId.get(value.id) ?? null; + // Session and experiment scores do not identify a single run and are not + // expanded into individual drafts. + return null; +} + +export function buildLangfuseDraftBundle(input: { + roots: LangfuseObservation[]; + tools: LangfuseObservation[]; + scores: LangfuseScore[]; + minScore?: number; + generatedAt?: Date; +}): LangfuseDraftBundle { + const minScore = input.minScore ?? 4; + if (!Number.isFinite(minScore) || minScore <= 1 || minScore > 5) + throw new Error('langfuse_min_score_invalid'); + + const traceByObservationId = new Map(); + for (const row of [...input.roots, ...input.tools]) { + const id = observationId(row); + const traceId = observationTraceId(row); + if (id && traceId) traceByObservationId.set(id, traceId); + } + + const lowScoreCounts = new Map(); + for (const score of input.scores) { + assertMetadataOnly(score); + if (score.dataType !== 'NUMERIC' || typeof score.value !== 'number') + continue; + if (score.value >= minScore) continue; + const traceId = scoreTraceId(score, traceByObservationId); + if (traceId) + lowScoreCounts.set(traceId, (lowScoreCounts.get(traceId) ?? 0) + 1); + } + + const toolsByTrace = new Map>(); + for (const row of input.tools) { + const traceId = observationTraceId(row); + if (!traceId) continue; + const name = safeToolName(metadataOf(row)['attributes.gen_ai.tool.name']); + if (!name) continue; + const names = toolsByTrace.get(traceId) ?? new Set(); + names.add(name); + toolsByTrace.set(traceId, names); + } + + const selected: SelectedDraft[] = []; + let skippedMissingRunIdentity = 0; + for (const row of input.roots) { + const traceId = observationTraceId(row); + if (!traceId) continue; + const metadata = metadataOf(row); + const terminalReason = safeEnum( + metadata['attributes.commerceagent.run_finished.reason'], + TERMINAL_REASONS, + ); + const errorCategory = safeEnum( + metadata['attributes.error.type'], + ERROR_CATEGORIES, + ); + const lowScoreCount = lowScoreCounts.get(traceId) ?? 0; + const failed = + errorCategory !== null || + (terminalReason !== null && terminalReason !== 'completed'); + if (!failed && lowScoreCount === 0) continue; + const runId = safeOpaqueId(metadata.run_id); + const turnId = safeOpaqueId(metadata.turn_id); + if (!runId && !turnId) { + skippedMissingRunIdentity++; + continue; + } + selected.push({ + runId, + turnId, + terminalReason, + errorCategory, + lowScoreCount, + failed, + toolNames: [...(toolsByTrace.get(traceId) ?? [])].sort(), + }); + } + + selected.sort((a, b) => + `${a.turnId ?? ''}|${a.runId ?? ''}`.localeCompare( + `${b.turnId ?? ''}|${b.runId ?? ''}`, + ), + ); + const drafts = selected.map((item, index) => ({ + schemaVersion: 1, + draftId: `draft-${String(index + 1).padStart(4, '0')}`, + source: { + runId: item.runId, + turnId: item.turnId, + signal: item.failed + ? item.lowScoreCount > 0 + ? 'failure_and_low_score' + : 'failure' + : 'low_score', + terminalReason: item.terminalReason, + errorCategory: item.errorCategory, + toolNames: item.toolNames, + lowScoreCount: item.lowScoreCount, + }, + caseDraft: { + id: '', + category: '', + riskLevel: '', + split: '', + prompt: '', + fixture_variant: '', + expected: { + status: '', + facts: {}, + evidence_required: [], + evidence_allowed: [], + forbidden_conclusions: [], + }, + tools: { required: [], forbidden: [], max_calls: null }, + instructions: {}, + rubric: '', + }, + promotionRequirements: { + humanReviewRequired: true, + referenceAnswerSelfTestRequired: true, + mutationTestsRequired: true, + automaticPromotion: false, + }, + })); + + return { + schemaVersion: 1, + generatedAt: (input.generatedAt ?? new Date()).toISOString(), + selection: { + failedTraceCount: selected.filter((item) => item.failed).length, + lowScoreTraceCount: selected.filter((item) => item.lowScoreCount > 0) + .length, + draftCount: drafts.length, + skippedMissingRunIdentity, + bodyFieldsRequested: false, + holdoutRead: false, + casesModified: false, + }, + drafts, + }; +} + +function iso(value: Date): string { + if (!Number.isFinite(value.getTime())) + throw new Error('langfuse_time_range_invalid'); + return value.toISOString(); +} + +function assertRange(from: Date, to: Date): void { + if (!Number.isFinite(from.getTime()) || !Number.isFinite(to.getTime())) + throw new Error('langfuse_time_range_invalid'); + if (from.getTime() >= to.getTime()) + throw new Error('langfuse_time_range_invalid'); + if (to.getTime() - from.getTime() > 31 * 24 * 60 * 60 * 1_000) + throw new Error('langfuse_time_range_too_wide'); +} + +async function fetchPages>(input: { + endpoint: string; + auth: string; + path: string; + from: Date; + to: Date; + params: Record; + maxPages: number; + maxRows: number; + pageSize?: number; + timeParameters?: { from: string; to: string }; + fetcher: typeof fetch; +}): Promise { + const rows: T[] = []; + let cursor: string | null = null; + const visitedCursors = new Set(); + for (let page = 0; page < input.maxPages; page++) { + const url = new URL(`${input.endpoint.replace(/\/+$/, '')}${input.path}`); + const timeParameters = input.timeParameters ?? { + from: 'fromStartTime', + to: 'toStartTime', + }; + url.searchParams.set(timeParameters.from, iso(input.from)); + url.searchParams.set(timeParameters.to, iso(input.to)); + url.searchParams.set('limit', String(input.pageSize ?? 1000)); + for (const [key, value] of Object.entries(input.params)) + url.searchParams.set(key, value); + if (cursor) url.searchParams.set('cursor', cursor); + + let response: Response; + try { + response = await input.fetcher(url, { + headers: { + Authorization: `Basic ${input.auth}`, + Accept: 'application/json', + }, + redirect: 'error', + signal: AbortSignal.timeout(10_000), + }); + } catch { + throw new Error('langfuse_request_failed'); + } + if (!response.ok) { + if (response.status === 401 || response.status === 403) + throw new Error('langfuse_auth_failed'); + if (response.status === 429) throw new Error('langfuse_rate_limited'); + throw new Error(`langfuse_http_${response.status}`); + } + let payload: { data?: unknown; meta?: { cursor?: unknown } }; + try { + payload = (await response.json()) as typeof payload; + } catch { + throw new Error('langfuse_response_invalid'); + } + if (!Array.isArray(payload.data)) + throw new Error('langfuse_response_invalid'); + for (const row of payload.data) { + if (!row || typeof row !== 'object' || Array.isArray(row)) + throw new Error('langfuse_response_invalid'); + assertMetadataOnly(row as Record); + rows.push(row as T); + if (rows.length > input.maxRows) + throw new Error('langfuse_result_limit_exceeded'); + } + const next = payload.meta?.cursor; + if (next === undefined || next === null || next === '') return rows; + if (typeof next !== 'string' || next.length > 2048) + throw new Error('langfuse_cursor_invalid'); + if (visitedCursors.has(next)) throw new Error('langfuse_cursor_repeated'); + visitedCursors.add(next); + cursor = next; + } + throw new Error('langfuse_page_limit_exceeded'); +} + +export async function collectLangfuseDraftBundle( + input: LangfuseDraftOptions, +): Promise { + assertRange(input.from, input.to); + if (!isSafeDraftEndpoint(input.endpoint)) + throw new Error('langfuse_endpoint_invalid'); + if (!input.publicKey.trim() || !input.secretKey.trim()) + throw new Error('langfuse_credentials_missing'); + if ( + !Number.isFinite(input.minScore) || + input.minScore <= 1 || + input.minScore > 5 + ) + throw new Error('langfuse_min_score_invalid'); + const fetcher = input.fetcher ?? fetch; + const auth = Buffer.from(`${input.publicKey}:${input.secretKey}`).toString( + 'base64', + ); + const fetchObservations = (name: string) => + fetchPages({ + endpoint: input.endpoint, + auth, + path: '/api/public/v2/observations', + from: input.from, + to: input.to, + params: { fields: OBSERVATION_FIELDS, name }, + maxPages: MAX_OBSERVATION_PAGES, + maxRows: MAX_ROWS, + fetcher, + }); + const [roots, tools] = await Promise.all([ + fetchObservations('commerceagent.turn'), + fetchObservations('commerceagent.tool'), + ]); + const lowScores = await fetchPages({ + endpoint: input.endpoint, + auth, + path: '/api/public/v3/scores', + from: input.from, + to: input.to, + params: { + fields: 'subject', + dataType: 'NUMERIC', + valueMax: String(Math.max(0, input.minScore - 0.000001)), + }, + maxPages: MAX_SCORE_PAGES, + maxRows: MAX_ROWS, + pageSize: 100, + timeParameters: { from: 'fromTimestamp', to: 'toTimestamp' }, + fetcher, + }); + return buildLangfuseDraftBundle({ + roots, + tools, + scores: lowScores, + minScore: input.minScore, + }); +} + +export function writeLangfuseDraftBundle( + benchmarkDirectory: string, + bundle: LangfuseDraftBundle, + generatedAt = new Date(), +): string { + const outputRoot = path.resolve( + benchmarkDirectory, + 'results', + 'langfuse-drafts', + ); + const timestamp = generatedAt + .toISOString() + .replace(/[:.]/g, '-') + .replace(/Z$/, 'Z'); + const outputDirectory = path.join(outputRoot, `run-${timestamp}`); + if (!outputDirectory.startsWith(`${outputRoot}${path.sep}`)) + throw new Error('langfuse_output_path_invalid'); + fs.mkdirSync(outputRoot, { recursive: true, mode: 0o700 }); + fs.chmodSync(outputRoot, 0o700); + if (fs.existsSync(outputDirectory)) + throw new Error('langfuse_output_already_exists'); + fs.mkdirSync(outputDirectory, { mode: 0o700 }); + fs.chmodSync(outputDirectory, 0o700); + privateJson(path.join(outputDirectory, 'drafts.json'), bundle); + return outputDirectory; +} diff --git a/benchmarks/package.json b/benchmarks/package.json index a4e06a6..7c12e9d 100644 --- a/benchmarks/package.json +++ b/benchmarks/package.json @@ -17,7 +17,8 @@ "regrade": "tsx scripts/regrade-business.ts", "judge": "tsx scripts/judge.ts", "judge:batch": "node --import tsx scripts/judge-calibration-batch.ts", - "calibrate": "tsx scripts/calibrate.ts" + "calibrate": "tsx scripts/calibrate.ts", + "langfuse-drafts": "tsx scripts/langfuse-drafts.ts" }, "devDependencies": { "promptfoo": "0.124.1", diff --git a/benchmarks/scripts/langfuse-drafts.ts b/benchmarks/scripts/langfuse-drafts.ts new file mode 100644 index 0000000..3aff6d5 --- /dev/null +++ b/benchmarks/scripts/langfuse-drafts.ts @@ -0,0 +1,89 @@ +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { + collectLangfuseDraftBundle, + isSafeDraftEndpoint, + writeLangfuseDraftBundle, +} from '../lib/langfuse-drafts.js'; + +const repository = fileURLToPath(new URL('../../', import.meta.url)); +const benchmarkDirectory = path.join(repository, 'benchmarks'); + +function option(args: string[], name: string): string | undefined { + const index = args.indexOf(name); + return index < 0 ? undefined : args[index + 1]; +} + +function parseDate(value: string | undefined, fallback: Date): Date { + if (value === undefined) return fallback; + const date = new Date(value); + if (!Number.isFinite(date.getTime())) + throw new Error('langfuse_time_range_invalid'); + return date; +} + +async function readCredentials() { + // runtime-config derives the main data directory from cwd, so select the + // repository before loading it. It reads encrypted local settings and then + // the existing COMMERCEAGENT_OTEL_* environment fallbacks. + process.chdir(repository); + const runtime = await import('../../src/runtime-config.js'); + const settings = runtime.getSystemSettings(); + const endpoint = settings.observabilityEndpoint; + const publicKey = settings.observabilityPublicKey; + const secretKey = runtime.getObservabilitySecret(); + if (!endpoint || !publicKey || !secretKey) + throw new Error('langfuse_credentials_missing'); + if (!isSafeDraftEndpoint(endpoint)) + throw new Error('langfuse_endpoint_invalid'); + return { endpoint, publicKey, secretKey }; +} + +export async function runLangfuseDraftExport(args = process.argv.slice(2)) { + if (args.includes('--help') || args.includes('-h')) { + process.stdout.write( + 'Usage: npm run bench:langfuse-drafts -- [--from ISO] [--to ISO] [--min-score 4]\n' + + 'Reads metadata-only failures and low numeric scores; writes skeleton drafts under benchmarks/results/langfuse-drafts/.\n', + ); + return null; + } + const to = parseDate(option(args, '--to'), new Date()); + const from = parseDate( + option(args, '--from'), + new Date(to.getTime() - 7 * 24 * 60 * 60 * 1_000), + ); + const minScore = Number(option(args, '--min-score') ?? '4'); + if (!Number.isFinite(minScore) || minScore <= 1 || minScore > 5) + throw new Error('langfuse_min_score_invalid'); + + const credentials = await readCredentials(); + const bundle = await collectLangfuseDraftBundle({ + ...credentials, + from, + to, + minScore, + }); + const outputDirectory = writeLangfuseDraftBundle(benchmarkDirectory, bundle); + const relativeOutput = path.relative(repository, outputDirectory); + process.stdout.write( + JSON.stringify({ + ...bundle.selection, + outputDirectory: relativeOutput, + automaticPromotion: false, + }) + '\n', + ); + return { bundle, outputDirectory }; +} + +if ( + process.argv[1] && + path.resolve(process.argv[1]) === fileURLToPath(import.meta.url) +) + try { + await runLangfuseDraftExport(); + } catch (error) { + process.stderr.write( + `${error instanceof Error && /^[a-z0-9_:-]+$/i.test(error.message) ? error.message : 'langfuse_draft_export_failed'}\n`, + ); + process.exitCode = 1; + } diff --git a/docs/ACL-MATRIX.md b/docs/ACL-MATRIX.md index 0359647..7ad4152 100644 --- a/docs/ACL-MATRIX.md +++ b/docs/ACL-MATRIX.md @@ -203,6 +203,8 @@ read-only 投影按 host-issued turn ID 精确匹配当前或已接纳的 queued | 路由族/操作 | 权限 | | ------------------------------------------------- | -------------------------------------------------- | | Provider、系统容量、Host 集成、注册策略、系统外观 | `manage_system_config` | +| 查看 OpenTelemetry 配置 | `manage_system_config` | +| 修改 OpenTelemetry 配置和密钥 | admin + `manage_system_config` | | Plugin Catalog 手动扫描 | admin / `manage_system_config` 路径中的 admin 检查 | | 系统 MCP 写入 | admin | | 用户创建、禁用、恢复、角色与权限 | `manage_users` | @@ -216,6 +218,13 @@ read-only 投影按 host-issued turn ID 精确匹配当前或已接纳的 queued 系统 MCP 默认仅 admin 可用；只有显式设置为 shared 后，普通成员的 Agent 才能进入 有效能力清单。API 不回传 Secret 明文。 +CommerceAgent 不提供 Langfuse Trace 查看接口。`manage_system_config` 可配置导出目标； +只有同时具备 `manage_system_config` 的管理员可以修改 OpenTelemetry 开关、地址和密钥； +具备该权限的用户可读取非 Secret 设置。Trace 查看由 Langfuse 项目自身的成员权限决定。 +`user_id_hash` 和 `workspace_id_hash` +用于 Trace 搜索与过滤，不构成访问控制。部署者应限制 Langfuse 项目成员，避免未获授权的 +用户查看其他用户的 Trace。 + ## 7. WebSocket 连接建立时把认证用户 ID、角色和 Permission 固定到 Session；每个操作仍重新检查 diff --git a/docs/API.md b/docs/API.md index 68200fb..5dd264b 100644 --- a/docs/API.md +++ b/docs/API.md @@ -73,6 +73,42 @@ Public： `data/extra/` 和任务运行历史 - `POST /api/groups/:jid/reset-owner`，admin break-glass - `GET /api/groups/:jid/messages` + +## 系统设置与 OpenTelemetry + +`GET|PUT /api/config/system` 受 `manage_system_config` 权限保护。OpenTelemetry 字段的 +修改还要求调用者角色为 admin；其他系统配置仍按原权限策略处理。系统设置响应包含 +`observabilityEnabled`、`observabilityEndpoint`、`observabilityPublicKey`、 +`observabilitySampleRate`、`observabilityIncludeContent` 和只读的 +`observabilityHasSecretKey`。写入请求可带 `observabilitySecretKey`；Secret Key +只写入本机 AES-256-GCM 加密文件，读取响应永远不会返回它。该设置遵循 +Web 持久配置优先于 `COMMERCEAGENT_OTEL_*` 环境变量，环境变量优先于代码默认值。 + +默认关闭。开启后通过 OTLP/HTTP 向配置的 Langfuse 根地址发送 trace，路径为 +`/api/public/otel/v1/traces`，使用 Basic Auth（用户名为 Public Key，密码为 +Secret Key）。公网地址必须使用 HTTPS；HTTP 仅允许 `localhost` 和 `127.0.0.1` +用于本机调试。不要求特定部署形态，可使用 Langfuse 云版或自托管实例。 + +trace 记录运行 ID、轮次 ID、工作区与用户 ID 的 SHA-256 哈希、Profile 版本、模型名、 +工具名、Token、耗时、错误类别和 `run_finished.reason`。每次**轮次**按采样率独立 +抽样，同一会话的不同轮次可能有不同的采样结果。默认不导出正文、工具参数或工具返回。 +只有显式开启 `observabilityIncludeContent` 后，才会导出经过脱敏的用户输入、助手输出、 +工具输入和工具返回；API Key、Endpoint、常见第三方令牌、手机号、邮箱、身份证号、银行卡号、 +IP 地址、地址、收件人/客户字段、Cookie 和令牌会在本机移除或替换后再进入 span。如果 +脱敏后仍命中高风险模式，正文会降级为只发送原文长度和 SHA-256。正文脱敏是尽力而为， +不保证识别所有自由文本中的个人信息（例如未标注的人名）；处理真实客户数据的生产环境 +建议保持元数据模式，不开启正文上报。导出异步执行；导出失败不会影响 Agent 回复或 +usage/计费记录，只增加进程内失败计数并输出一次不含端点、凭据和 payload 的分类日志。 + +CommerceAgent 不提供 Langfuse Trace 浏览路由。`manage_system_config` 用户可配置目标 +地址和密钥；Trace 查看权限由目标 Langfuse 项目的成员权限控制。由于同一项目内的哈希 +标识只用于筛选，不是授权边界，部署者应将项目查看权限限制给获准的运维人员。 + +Langfuse 可使用云版或自托管实例；CommerceAgent 只需要对应项目的 API 地址、Public Key +和 Secret Key。benchmark 的 `bench:langfuse-drafts` 命令是只读回流工具，只查询元数据与 +评分字段，不读取 trace 正文、工具参数/返回或评分评论；输出仅为被 Git 忽略的空白结构草稿， +不会写入业务用例或 holdout。需要正文时，必须由人工在受控环境中另行补全并完成脱敏审核。 + - `DELETE /api/groups/:jid/messages/:messageId` - `GET|PUT /api/groups/:jid/env` - `GET|PUT /api/groups/:jid/mcp`，仅兼容旧客户端 diff --git a/docs/BENCHMARK-REVISION.md b/docs/BENCHMARK-REVISION.md index 9cc8d78..18597ee 100644 --- a/docs/BENCHMARK-REVISION.md +++ b/docs/BENCHMARK-REVISION.md @@ -425,3 +425,11 @@ snake_case、camelCase 和省略输入字段测试全部失败。 - 验证：Fake Messages 服务 6 种响应脚本、预算计算和选型门槛测试 8 项通过；68 条 reference/oracle 自检通过、932 个变异被拒绝；当前源码 Docker 镜像完整 E2E selftest 通过（1 次 attempt、2 次 fixture MCP 调用、测量有效、parity/reassessment/report/leak check 通过）。无真实模型调用、无 holdout 运行、无选型结论。 - 后续最终代码路径把 `bench:selftest` 改为驱动真实 Promptfoo Provider 与运行器，而非手工拼接 Promptfoo 结果；CI/本地 `make bench-selftest` 必须从当前 checkout 构建镜像，避免旧 latest 镜像的 entrypoint 漂移。完整 selftest 最终复跑通过，synthetic candidate 保持 `gateway_pending`，因此业务通过只验证链路，不计入模型质量。 - 最终验收：全量 `make test` 为 354 个测试文件、3162 项通过；benchmark typecheck、项目 typecheck、`make build`、`docs:check`、`format:check`、`git diff --check` 均通过。预算 CLI 已用历史 50-attempt run 验证输出 150 次筛选 attempts、CNY 14.13 估算和冷启动时间代理值。离线 full selftest 报告 1 次业务 attempt 测量有效、2 次 fixture MCP 成功调用、独立重评/报告/泄漏探针通过、真实模型调用 0、holdout 0。 + +### 阶段 3 Langfuse 元数据回流草稿（2026-10-24） + +- 主 Agent `runAgent` 与 Conversation Agent 都通过 `withObservabilityCleanup` 在所有返回/异常路径收尾；每条已注册 run 在 `containerTimeout + 60s` 后还有超时关闭兜底，并清除 runId、chatJid、turnId 映射和定时器。新轮次遇到同 chat 的旧 trace 会先关闭旧 run，避免静默漏 trace。 +- 新增只读 `bench:langfuse-drafts`：从系统设置或 `COMMERCEAGENT_OTEL_*` 环境回退读取 Langfuse 地址/Key；查询限时窗的 Observations API v2 `core,metadata` 与 Scores API v3 数值分数，不请求 input/output、状态文本、工具参数/返回、评分评论、模型名、Endpoint、用户或工作区标识。Trace ID 只在内存中关联，不写入草稿。 +- 输出仅为 `benchmarks/results/langfuse-drafts/` 下 0700/0600 的空白结构草稿；Prompt、业务分类、风险等级、split、期望状态/事实/证据、工具计划和 rubric 均留空。脚本不读取或写入 cases/dev、cases/holdout，不触碰 holdout ledger，不提供自动提升或提交功能。草稿必须由人补全、通过 reference/oracle 自检和变异测试并确认后，才能手动进入回归用例。 +- 本机 Langfuse Compose 的 Web 端口改映射到 `127.0.0.1:3100`，CommerceAgent 保持默认 3000；主系统设置地址同步到 3100，正文上报关闭。所有 Compose 宿主端口保持 loopback 绑定，历史联调 trace 仍为 0。 +- 验证：Langfuse 草稿与 OTel 生命周期专项测试 16 项通过；benchmark typecheck、`make typecheck`、`make build`、`docs:check`、格式检查通过。全量测试沿用 4 workers 和临时权限镜像，358 个测试文件、3197 项通过。先以真实本机 Langfuse 空项目验证 CLI 为 0 observations/0 drafts；随后创建 1 条无正文合成失败 trace 和 1 条数值低分，CLI 产出 1 份骨架草稿，run/turn、终态、低分计数均匹配，traceId 和正文均未落盘。检查到兼容 API 在 `statusMessage`/`comment` 上可返回 null 占位；脚本允许 null/空值并仍拒绝非空内容。合成 trace、score 与输出目录已删除，查询确认剩余合成 observations/scores 均为 0。没有读取真实会话、写入 cases、读取 holdout 或调用外部模型。阶段 4 及之后未开始。 diff --git a/docs/BENCHMARK.md b/docs/BENCHMARK.md index 4006756..42fbeac 100644 --- a/docs/BENCHMARK.md +++ b/docs/BENCHMARK.md @@ -240,7 +240,37 @@ Provider 导入无论成功或失败均立即删除 `provider-snapshot`。每个 Container 验收或模型质量结论。 阶段 4 的业务用例集已加入。阶段 5 已按真实 fixture 工具返回修正 Judge 输入，并完成相同样本诊断和 20 条新人工盲标。新样本在关键维度几乎全为高分，且每个关键维度都没有 3 条人工低分，样本信息不足；本轮 κ 不作报告，也不能据此判断 Judge 好坏或进入排名。阶段 6 只按确定性断言排名；Profile 对比和正式多候选运行尚未开始。最新结果见 [验证记录](BENCHMARK-VALIDATION.md)。 -第一版不引入 Langfuse；线上观测仍需先定义脱敏和保留期限。 + +## Langfuse 失败回流草稿 + +CommerceAgent 的 Langfuse OTel 导出为独立 opt-in 设置；benchmark 默认使用隔离 `DATA_DIR`， +不读取或继承生产 Langfuse 配置，因此 `bench:run` 与 `bench:selftest` 不会向 Langfuse 写入 trace。 +回流命令只读 Langfuse API，地址和密钥取自 CommerceAgent 系统设置，或由已有的 +`COMMERCEAGENT_OTEL_ENDPOINT`、`COMMERCEAGENT_OTEL_PUBLIC_KEY`、 +`COMMERCEAGENT_OTEL_SECRET_KEY` 环境回退配置提供；不要把密钥放进命令行参数。 + +```bash +npm run bench:langfuse-drafts -- --from 2026-10-17T00:00:00Z --to 2026-10-24T00:00:00Z --min-score 4 +``` + +省略时间时查询最近 7 天；单次查询窗口最多 31 天。目标需支持 Langfuse Observations API v2 +和 Scores API v3（Langfuse v4+）。脚本使用 Observations API 的 `core,metadata` 字段读取 +`commerceagent.turn` 与 `commerceagent.tool`，并用 Scores API v3 +识别低于阈值的数值评分。默认阈值为 4（1–5 评分中低于 4）；只保留 run/turn ID、终态、 +错误类别、工具名和低分条数用于选择与引用。查询不会请求 input/output、工具参数或返回、 +状态文本、评分评论、模型名、用户/工作区标识或 Endpoint。trace ID 只在内存里关联观察和评分， +不会写入草稿。若兼容服务仍返回被禁止字段且其值非空，导出会中止；null 或空字符串占位会被 +视为空值并忽略，绝不会写入草稿。 + +草稿写入被 `.gitignore` 排除的 `benchmarks/results/langfuse-drafts/`，文件权限为本机私有模式。 +它只生成结构骨架：Prompt、业务分类、风险等级、split、期望状态、事实、证据、工具计划和 +rubric 都留空，不推断预期答案，也不从回复正文补写内容。脚本不读取或写入 +`benchmarks/cases/dev/`、`benchmarks/cases/holdout/`，不修改 holdout ledger、不自动提交， +也没有把草稿提升为正式用例的命令。 + +人工审核者必须在受控环境中补全合成或已批准的任务内容及风险等级、期望状态；真实会话正文 +不能由此脚本导出。进入正式回归集前，每条草稿都必须通过固定参考答案自检、变异测试并由人 +确认。holdout 只允许在最终正式运行中使用一次，回流和调试阶段不得读取或触碰 holdout。 ## 开发验证 diff --git a/package-lock.json b/package-lock.json index 9d3bd86..af3bbd5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -12,6 +12,9 @@ "@anthropic-ai/claude-agent-sdk": "0.3.205", "@hono/node-server": "^2.0.12", "@larksuiteoapi/node-sdk": "^1.58.0", + "@opentelemetry/api": "1.9.0", + "@opentelemetry/exporter-trace-otlp-http": "0.200.0", + "@opentelemetry/sdk-trace-base": "2.0.1", "adm-zip": "^0.6.0", "baileys": "^7.0.0-rc13", "bcryptjs": "^3.0.3", @@ -1535,6 +1538,241 @@ "@emnapi/runtime": "^1.7.1" } }, + "node_modules/@opentelemetry/api": { + "version": "1.9.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.0.tgz", + "integrity": "sha512-3giAOQvZiH5F9bMlMiv8+GSPMeqg0dbaeo58/0SlA9sxSqZhnUtxzX9/2FzyhS9sWQf5S0GJE0AKBrFqjpeYcg==", + "license": "Apache-2.0", + "engines": { + "node": ">=8.0.0" + } + }, + "node_modules/@opentelemetry/api-logs": { + "version": "0.200.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/api-logs/-/api-logs-0.200.0.tgz", + "integrity": "sha512-IKJBQxh91qJ+3ssRly5hYEJ8NDHu9oY/B1PXVSCWf7zytmYO9RNLB0Ox9XQ/fJ8m6gY6Q6NtBWlmXfaXt5Uc4Q==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/api": "^1.3.0" + }, + "engines": { + "node": ">=8.0.0" + } + }, + "node_modules/@opentelemetry/core": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-2.0.0.tgz", + "integrity": "sha512-SLX36allrcnVaPYG3R78F/UZZsBsvbc7lMCLx37LyH5MJ1KAAZ2E3mW9OAD3zGz0G8q/BtoS5VUrjzDydhD6LQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.0.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/exporter-trace-otlp-http": { + "version": "0.200.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/exporter-trace-otlp-http/-/exporter-trace-otlp-http-0.200.0.tgz", + "integrity": "sha512-Goi//m/7ZHeUedxTGVmEzH19NgqJY+Bzr6zXo1Rni1+hwqaksEyJ44gdlEMREu6dzX1DlAaH/qSykSVzdrdafA==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.0", + "@opentelemetry/otlp-exporter-base": "0.200.0", + "@opentelemetry/otlp-transformer": "0.200.0", + "@opentelemetry/resources": "2.0.0", + "@opentelemetry/sdk-trace-base": "2.0.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/exporter-trace-otlp-http/node_modules/@opentelemetry/sdk-trace-base": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-2.0.0.tgz", + "integrity": "sha512-qQnYdX+ZCkonM7tA5iU4fSRsVxbFGml8jbxOgipRGMFHKaXKHQ30js03rTobYjKjIfnOsZSbHKWF0/0v0OQGfw==", + "deprecated": "Use @opentelemetry/sdk-trace package", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.0", + "@opentelemetry/resources": "2.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.3.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/otlp-exporter-base": { + "version": "0.200.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-exporter-base/-/otlp-exporter-base-0.200.0.tgz", + "integrity": "sha512-IxJgA3FD7q4V6gGq4bnmQM5nTIyMDkoGFGrBrrDjB6onEiq1pafma55V+bHvGYLWvcqbBbRfezr1GED88lacEQ==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.0", + "@opentelemetry/otlp-transformer": "0.200.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/otlp-transformer": { + "version": "0.200.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/otlp-transformer/-/otlp-transformer-0.200.0.tgz", + "integrity": "sha512-+9YDZbYybOnv7sWzebWOeK6gKyt2XE7iarSyBFkwwnP559pEevKOUD8NyDHhRjCSp13ybh9iVXlMfcj/DwF/yw==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/api-logs": "0.200.0", + "@opentelemetry/core": "2.0.0", + "@opentelemetry/resources": "2.0.0", + "@opentelemetry/sdk-logs": "0.200.0", + "@opentelemetry/sdk-metrics": "2.0.0", + "@opentelemetry/sdk-trace-base": "2.0.0", + "protobufjs": "^7.3.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": "^1.3.0" + } + }, + "node_modules/@opentelemetry/otlp-transformer/node_modules/@opentelemetry/sdk-trace-base": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-2.0.0.tgz", + "integrity": "sha512-qQnYdX+ZCkonM7tA5iU4fSRsVxbFGml8jbxOgipRGMFHKaXKHQ30js03rTobYjKjIfnOsZSbHKWF0/0v0OQGfw==", + "deprecated": "Use @opentelemetry/sdk-trace package", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.0", + "@opentelemetry/resources": "2.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.3.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/resources": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-2.0.0.tgz", + "integrity": "sha512-rnZr6dML2z4IARI4zPGQV4arDikF/9OXZQzrC01dLmn0CZxU5U5OLd/m1T7YkGRj5UitjeoCtg/zorlgMQcdTg==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.0", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.3.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/sdk-logs": { + "version": "0.200.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-logs/-/sdk-logs-0.200.0.tgz", + "integrity": "sha512-VZG870063NLfObmQQNtCVcdXXLzI3vOjjrRENmU37HYiPFa0ZXpXVDsTD02Nh3AT3xYJzQaWKl2X2lQ2l7TWJA==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/api-logs": "0.200.0", + "@opentelemetry/core": "2.0.0", + "@opentelemetry/resources": "2.0.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.4.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/sdk-metrics": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-2.0.0.tgz", + "integrity": "sha512-Bvy8QDjO05umd0+j+gDeWcTaVa1/R2lDj/eOvjzpm8VQj1K1vVZJuyjThpV5/lSHyYW2JaHF2IQ7Z8twJFAhjA==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.0", + "@opentelemetry/resources": "2.0.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.9.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/sdk-trace-base": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-2.0.1.tgz", + "integrity": "sha512-xYLlvk/xdScGx1aEqvxLwf6sXQLXCjk3/1SQT9X9AoN5rXRhkdvIFShuNNmtTEPRBqcsMbS4p/gJLNI2wXaDuQ==", + "deprecated": "Use @opentelemetry/sdk-trace package", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.1", + "@opentelemetry/resources": "2.0.1", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.3.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/sdk-trace-base/node_modules/@opentelemetry/core": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/core/-/core-2.0.1.tgz", + "integrity": "sha512-MaZk9SJIDgo1peKevlbhP6+IwIiNPNmswNL4AF0WaQJLbHXjr9SrZMgS12+iqr9ToV4ZVosCcc0f8Rg67LXjxw==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.0.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/sdk-trace-base/node_modules/@opentelemetry/resources": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/resources/-/resources-2.0.1.tgz", + "integrity": "sha512-dZOB3R6zvBwDKnHDTB4X1xtMArB/d324VsbiPkX/Yu0Q8T2xceRthoIVFhJdvgVM2QhGVUyX9tzwiNxGtoBJUw==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/core": "2.0.1", + "@opentelemetry/semantic-conventions": "^1.29.0" + }, + "engines": { + "node": "^18.19.0 || >=20.6.0" + }, + "peerDependencies": { + "@opentelemetry/api": ">=1.3.0 <1.10.0" + } + }, + "node_modules/@opentelemetry/semantic-conventions": { + "version": "1.43.0", + "resolved": "https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.43.0.tgz", + "integrity": "sha512-eSYWTm620tTk45EKSedaUL8MFYI8hW164hIXsgIHyxu3VobUB3fFCu5t0hQby6OoWRPsG1KkKUG2M5UadiLiVg==", + "license": "Apache-2.0", + "engines": { + "node": ">=14" + } + }, "node_modules/@oxc-project/types": { "version": "0.133.0", "resolved": "https://registry.npmjs.org/@oxc-project/types/-/types-0.133.0.tgz", diff --git a/package.json b/package.json index e06c2f4..bfb2eef 100644 --- a/package.json +++ b/package.json @@ -36,6 +36,7 @@ "bench:judge": "npm --prefix benchmarks run judge --", "bench:judge:batch": "npm --prefix benchmarks run judge:batch --", "bench:calibrate": "npm --prefix benchmarks run calibrate --", + "bench:langfuse-drafts": "npm --prefix benchmarks run langfuse-drafts --", "bench:selftest": "npm --prefix benchmarks run selftest --", "bench:review": "npm --prefix benchmarks run review --" }, @@ -43,6 +44,9 @@ "@anthropic-ai/claude-agent-sdk": "0.3.205", "@hono/node-server": "^2.0.12", "@larksuiteoapi/node-sdk": "^1.58.0", + "@opentelemetry/api": "1.9.0", + "@opentelemetry/exporter-trace-otlp-http": "0.200.0", + "@opentelemetry/sdk-trace-base": "2.0.1", "adm-zip": "^0.6.0", "baileys": "^7.0.0-rc13", "bcryptjs": "^3.0.3", diff --git a/src/index.ts b/src/index.ts index 9cc6a0a..584fa71 100644 --- a/src/index.ts +++ b/src/index.ts @@ -395,6 +395,12 @@ import { resolveChannelAccountFallbackWorkspace, } from './channel-account-routing.js'; import { testChannelAccountCredentials } from './channel-account-connectivity.js'; +import { + recordObservabilityEvent, + recordObservabilityFailure, + startObservabilityRun, + withObservabilityCleanup, +} from './otel-observability.js'; import { buildAgentProfilePrompt, hasAgentProfilePrompts, @@ -9373,6 +9379,23 @@ function ensureTerminalContainerStarted(chatJid: string): boolean { } async function runAgent( + ...args: Parameters +): ReturnType { + const [group, , chatJid] = args; + return withObservabilityCleanup( + chatJid, + () => queue.getActiveQueryId(chatJid) ?? undefined, + () => runAgentTurn(...args), + (result) => { + if (queue.getRecentStopDisposition(group.folder) === 'discard') + return 'stopped'; + if (result.status === 'success') return 'completed'; + return result.status === 'closed' ? 'released' : 'runner_exit'; + }, + ); +} + +async function runAgentTurn( group: RegisteredGroup, prompt: string, chatJid: string, @@ -9390,6 +9413,20 @@ async function runAgent( const resolvedAgentProfile = resolveEffectiveAgentProfile( agentProfile ?? getAgentProfileForWorkspace(group.folder, group.created_by), ); + const observabilityRunId = queue.getActiveQueryId(chatJid) ?? undefined; + if (observabilityRunId || turnId) { + startObservabilityRun( + { + runId: observabilityRunId, + chatJid, + turnId, + workspace: group.folder, + profileVersion: resolvedAgentProfile?.version, + userId: group.created_by ?? undefined, + }, + prompt, + ); + } const interactionMode = resolveRuntimeInteractionMode( getWorkspaceInteractionMode(group.folder), { agentKind: 'main' }, @@ -9452,6 +9489,16 @@ async function runAgent( // Wrap onOutput to track session ID from streamed results const wrappedOnOutput = async (output: ContainerOutput) => { + if (output.providerFailure) { + recordObservabilityFailure( + chatJid, + observabilityRunId, + 'provider_failure', + ); + } else if (output.status === 'error') { + recordObservabilityFailure(chatJid, observabilityRunId, 'runner_error'); + } + recordObservabilityEvent(observabilityRunId, output.streamEvent); queue.markRunnerActivity(chatJid); if ( output.ipcReceipts?.length && @@ -13970,6 +14017,19 @@ async function handleDiscordIpcRequest( async function processAgentConversation( chatJid: string, agentId: string, +): Promise { + const virtualChatJid = `${chatJid}#agent:${agentId}`; + return withObservabilityCleanup( + virtualChatJid, + () => queue.getActiveQueryId(virtualChatJid) ?? undefined, + () => processAgentConversationTurn(chatJid, agentId), + (result) => (result === false ? 'runner_exit' : 'completed'), + ); +} + +async function processAgentConversationTurn( + chatJid: string, + agentId: string, ): Promise { const agent = getAgent(agentId); if (!agent || (agent.kind !== 'conversation' && agent.kind !== 'spawn')) { @@ -14222,6 +14282,20 @@ async function processAgentConversation( formatMessages(missedMessages, { knownMessageIds: knownReferencedMessageIds, }); + const observabilityRunId = queue.getActiveQueryId(virtualJid) ?? undefined; + if (observabilityRunId || lastProcessed.id) { + startObservabilityRun( + { + runId: observabilityRunId, + chatJid: virtualJid, + turnId: lastProcessed.id, + workspace: effectiveGroup.folder, + profileVersion: agentProfile?.version, + userId: effectiveGroup.created_by ?? undefined, + }, + prompt, + ); + } const images = collectMessageImages(virtualChatJid, missedMessages, { knownMessageIds: activeSessionReferencedMessageIds, }); @@ -15189,6 +15263,20 @@ async function processAgentConversation( }; const wrappedOnOutput = async (output: ContainerOutput) => { + if (output.providerFailure) { + recordObservabilityFailure( + virtualJid, + observabilityRunId, + 'provider_failure', + ); + } else if (output.status === 'error') { + recordObservabilityFailure( + virtualJid, + observabilityRunId, + 'runner_error', + ); + } + recordObservabilityEvent(observabilityRunId, output.streamEvent); // #547: warm-lifecycle bookkeeping — mark activity, and flag query-idle on // a substantive result / interruption so the runner can be kept warm. queue.markRunnerActivity(virtualJid); diff --git a/src/otel-observability.ts b/src/otel-observability.ts new file mode 100644 index 0000000..c17dc3d --- /dev/null +++ b/src/otel-observability.ts @@ -0,0 +1,541 @@ +import { createHash, randomUUID } from 'node:crypto'; + +import { ROOT_CONTEXT, trace, type Span } from '@opentelemetry/api'; +import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http'; +import { + BasicTracerProvider, + BatchSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; + +import type { StreamEvent } from './types.js'; +import { serializeRedactedContent } from './otel-redaction.js'; +import { + getObservabilitySecret, + getSystemSettings, + isSafeOtelEndpoint, +} from './runtime-config.js'; +import { logger } from './logger.js'; + +const SERVICE_NAME = 'commerceagent'; +const OTEL_TIMEOUT_MS = 2_000; +const FLUSH_TIMEOUT_MS = 2_500; +const STALE_RUN_TIMEOUT_BUFFER_MS = 60_000; +const activeRuns = new Map(); +const activeRunsByChatJid = new Map(); +const activeRunsByTurnId = new Map(); +const activeRunTimeouts = new Map(); +let exportFailureCount = 0; +let loggedExportFailure = false; + +export interface ObservabilityRunIdentity { + runId?: string; + chatJid?: string; + turnId?: string; + workspace: string; + profileVersion?: number; + userId?: string; +} + +export interface ObservabilityExportConfig { + enabled: boolean; + endpoint: string; + publicKey: string; + secretKey: string; + sampleRate: number; + includeContent: boolean; +} + +function hashIdentity(value: string | undefined): string { + if (!value) return 'unknown'; + return createHash('sha256').update(value).digest('hex'); +} + +function reportExportFailure(): void { + exportFailureCount += 1; + if (loggedExportFailure) return; + loggedExportFailure = true; + logger.warn( + { category: 'otel_export_failed' }, + 'OpenTelemetry trace export failed; export failures will be counted without logging payloads', + ); +} + +export function getObservabilityExportFailureCount(): number { + return exportFailureCount; +} + +function readExportConfig(): ObservabilityExportConfig { + const settings = getSystemSettings(); + return { + enabled: settings.observabilityEnabled, + endpoint: settings.observabilityEndpoint, + publicKey: settings.observabilityPublicKey, + secretKey: getObservabilitySecret(), + sampleRate: settings.observabilitySampleRate, + includeContent: settings.observabilityIncludeContent, + }; +} + +function makeExporter(config: ObservabilityExportConfig): { + provider: BasicTracerProvider; + tracer: ReturnType; +} | null { + if (!config.enabled || Math.random() >= config.sampleRate) { + return null; + } + const endpoint = config.endpoint.trim().replace(/\/+$/, ''); + const publicKey = config.publicKey.trim(); + const secret = config.secretKey; + if (!endpoint || !isSafeOtelEndpoint(endpoint) || !publicKey || !secret) { + return null; + } + + const auth = Buffer.from(`${publicKey}:${secret}`).toString('base64'); + const exporter = new OTLPTraceExporter({ + url: `${endpoint}/api/public/otel/v1/traces`, + headers: { + Authorization: `Basic ${auth}`, + 'x-langfuse-ingestion-version': '4', + }, + timeoutMillis: OTEL_TIMEOUT_MS, + }); + const originalExport = exporter.export.bind(exporter); + exporter.export = (spans, callback) => { + originalExport(spans, (result) => { + if (result.code !== 0) reportExportFailure(); + callback(result); + }); + }; + const provider = new BasicTracerProvider({ + forceFlushTimeoutMillis: FLUSH_TIMEOUT_MS, + spanProcessors: [ + new BatchSpanProcessor(exporter, { + maxQueueSize: 256, + maxExportBatchSize: 64, + scheduledDelayMillis: 1_000, + exportTimeoutMillis: OTEL_TIMEOUT_MS, + }), + ], + }); + return { provider, tracer: provider.getTracer(SERVICE_NAME, '1') }; +} + +export class ObservabilityRun { + private readonly provider: BasicTracerProvider; + private readonly tracer: ReturnType; + private readonly root: Span; + private readonly startedAt = Date.now(); + private readonly toolSpans = new Map(); + private readonly modelSpans = new Map(); + private readonly includeContent: boolean; + private readonly assistantOutput: string[] = []; + private assistantOutputLength = 0; + private runId: string; + private finished = false; + readonly chatJid?: string; + readonly turnId?: string; + + constructor( + identity: ObservabilityRunIdentity, + config: ObservabilityExportConfig, + initialInput?: string, + ) { + const initialized = makeExporter(config); + if (!initialized) throw new Error('observability exporter not configured'); + this.provider = initialized.provider; + this.tracer = initialized.tracer; + this.includeContent = config.includeContent; + this.chatJid = identity.chatJid; + this.turnId = identity.turnId; + this.runId = + identity.runId ?? + (identity.turnId ? `turn-${identity.turnId}` : randomUUID()); + const userHash = hashIdentity(identity.userId); + const workspaceHash = hashIdentity(identity.workspace); + this.root = initialized.tracer.startSpan('commerceagent.turn', { + attributes: { + 'langfuse.trace.name': 'commerceagent.turn', + 'langfuse.user.id': userHash, + 'langfuse.session.id': workspaceHash, + 'langfuse.trace.metadata.run_id': this.runId, + 'langfuse.trace.metadata.turn_id': identity.turnId ?? 'unknown', + 'langfuse.trace.metadata.workspace_id_hash': workspaceHash, + 'langfuse.trace.metadata.agent_profile_version': + identity.profileVersion ?? -1, + 'langfuse.trace.metadata.user_id_hash': userHash, + 'service.name': SERVICE_NAME, + }, + }); + if (this.includeContent && initialInput) { + this.root.setAttribute( + 'langfuse.observation.input', + serializeRedactedContent(initialInput), + ); + } + } + + private parentContext() { + return trace.setSpanContext(ROOT_CONTEXT, this.root.spanContext()); + } + + bindRunId(runId: string): void { + if (!runId || runId === this.runId) return; + this.runId = runId; + this.root.setAttribute('langfuse.trace.metadata.run_id', runId); + } + + record(event: StreamEvent): void { + if (this.finished) return; + const parent = this.parentContext(); + if (event.eventType === 'tool_use_start' && event.toolUseId) { + const span = this.tracer.startSpan( + 'commerceagent.tool', + { + attributes: { + 'langfuse.observation.type': 'tool', + 'gen_ai.operation.name': 'execute_tool', + 'gen_ai.tool.name': event.toolName ?? 'unknown', + 'commerceagent.agent_scope': event.agentScope ?? 'unknown', + ...(this.includeContent + ? { + 'langfuse.observation.input': serializeRedactedContent( + event.toolInput ?? event.toolInputSummary ?? '', + ), + } + : {}), + }, + }, + parent, + ); + this.toolSpans.set(event.toolUseId, span); + return; + } + if (event.eventType === 'tool_use_end' && event.toolUseId) { + this.toolSpans + .get(event.toolUseId) + ?.setAttribute('commerceagent.tool.call_completed', true); + return; + } + if (event.eventType === 'tool_result' && event.toolUseId) { + const span = this.toolSpans.get(event.toolUseId); + // Only inspect the local error marker; never attach the result text. + if (span && event.toolResult?.startsWith('⚠️ ')) { + span.setAttribute('error.type', 'tool_error'); + span.setStatus({ code: 2 }); + } + if (span && this.includeContent && event.toolResult) { + span.setAttribute( + 'langfuse.observation.output', + serializeRedactedContent(event.toolResult), + ); + } + span?.end(); + this.toolSpans.delete(event.toolUseId); + return; + } + if (event.eventType === 'raw_sdk_event') { + if (event.apiRequestId) { + this.root.setAttribute( + 'commerceagent.last_api_request_id', + event.apiRequestId, + ); + } + const messageId = event.messageUuid; + if (event.rawType === 'stream_event/message_start' && messageId) { + const span = this.tracer.startSpan( + 'gen_ai.request', + { + attributes: { + 'langfuse.observation.type': 'generation', + 'gen_ai.operation.name': 'chat', + ...(event.apiRequestId + ? { 'gen_ai.request.id': event.apiRequestId } + : {}), + }, + }, + parent, + ); + this.modelSpans.set(messageId, span); + } else if (event.rawType === 'stream_event/message_stop' && messageId) { + this.modelSpans.get(messageId)?.end(); + this.modelSpans.delete(messageId); + } + return; + } + if (event.eventType === 'usage' && event.usage) { + const usage = event.usage; + if (usage.eventId) { + this.root.setAttribute('commerceagent.usage.event_id', usage.eventId); + } + this.root.setAttributes({ + 'gen_ai.usage.input_tokens': usage.inputTokens, + 'gen_ai.usage.output_tokens': usage.outputTokens, + 'gen_ai.usage.cache_read_input_tokens': usage.cacheReadInputTokens, + 'gen_ai.usage.cache_creation_input_tokens': + usage.cacheCreationInputTokens, + 'gen_ai.usage.reasoning_tokens': usage.reasoningTokens, + 'commerceagent.usage.cost_usd': usage.costUSD, + 'commerceagent.usage.duration_ms': usage.durationMs, + 'commerceagent.usage.num_turns': usage.numTurns, + }); + const models = Object.entries(usage.modelUsage ?? {}); + this.root.setAttribute('gen_ai.request.count', models.length); + for (const [model, modelUsage] of models) { + const span = this.tracer.startSpan( + 'gen_ai.model_usage', + { + attributes: { + 'langfuse.observation.type': 'generation', + 'gen_ai.request.model': model, + 'gen_ai.usage.input_tokens': modelUsage.inputTokens, + 'gen_ai.usage.output_tokens': modelUsage.outputTokens, + 'gen_ai.usage.cache_read_input_tokens': + modelUsage.cacheReadInputTokens, + 'gen_ai.usage.cache_creation_input_tokens': + modelUsage.cacheCreationInputTokens, + 'gen_ai.usage.reasoning_tokens': modelUsage.reasoningTokens, + 'commerceagent.usage.cost_usd': modelUsage.costUSD, + }, + }, + parent, + ); + span.end(); + } + return; + } + if ( + this.includeContent && + event.eventType === 'text_delta' && + event.text && + this.assistantOutputLength < 16_000 + ) { + const remaining = 16_000 - this.assistantOutputLength; + const next = event.text.slice(0, remaining); + this.assistantOutput.push(next); + this.assistantOutputLength += next.length; + return; + } + if (event.eventType === 'status' && event.statusText) { + const normalized = event.statusText.toLowerCase(); + if (normalized.includes('error') || normalized.includes('failed')) { + this.root.setAttribute('error.type', 'runtime_error'); + } + } + } + + recordFailure( + category: 'provider_failure' | 'runner_error' | 'runtime_error', + ): void { + this.root.setAttribute('error.type', category); + this.root.setStatus({ code: 2 }); + } + + finish(reason: string): void { + if (this.finished) return; + this.finished = true; + const duration = Math.max(0, Date.now() - this.startedAt); + for (const span of this.toolSpans.values()) span.end(); + for (const span of this.modelSpans.values()) span.end(); + this.toolSpans.clear(); + this.modelSpans.clear(); + this.root.setAttributes({ + 'commerceagent.run_finished.reason': reason, + 'commerceagent.run.duration_ms': duration, + }); + if (this.includeContent && this.assistantOutput.length > 0) { + this.root.setAttribute( + 'langfuse.observation.output', + serializeRedactedContent(this.assistantOutput.join('')), + ); + } + if (reason === 'runner_exit' || reason === 'stopped') { + this.root.setAttribute('error.type', reason); + this.root.setStatus({ code: 2 }); + } + this.root.end(); + // Export is intentionally detached from queue/session completion. + const flush = this.provider + .forceFlush() + .catch(() => reportExportFailure()) + .finally(() => { + void this.provider.shutdown().catch(() => reportExportFailure()); + }); + let flushTimeout: NodeJS.Timeout | undefined; + const boundedWait = new Promise((resolve) => { + flushTimeout = setTimeout(resolve, FLUSH_TIMEOUT_MS); + }); + void Promise.race([flush, boundedWait]).finally(() => { + if (flushTimeout) clearTimeout(flushTimeout); + }); + } +} + +function removeActiveRun(run: ObservabilityRun): void { + const timeout = activeRunTimeouts.get(run); + if (timeout) clearTimeout(timeout); + activeRunTimeouts.delete(run); + for (const [alias, candidate] of activeRuns) { + if (candidate === run) activeRuns.delete(alias); + } + for (const [jid, candidate] of activeRunsByChatJid) { + if (candidate === run) activeRunsByChatJid.delete(jid); + } + for (const [turnId, candidate] of activeRunsByTurnId) { + if (candidate === run) activeRunsByTurnId.delete(turnId); + } +} + +function finishAndRemoveActiveRun(run: ObservabilityRun, reason: string): void { + removeActiveRun(run); + run.finish(reason); +} + +function configuredStaleRunTimeoutMs(): number { + try { + const timeout = getSystemSettings().containerTimeout; + return Math.max(1_000, timeout) + STALE_RUN_TIMEOUT_BUFFER_MS; + } catch { + return 1_800_000 + STALE_RUN_TIMEOUT_BUFFER_MS; + } +} + +export function getActiveObservabilityRunCount(): number { + return new Set(activeRuns.values()).size; +} + +export function startObservabilityRunWithConfig( + identity: ObservabilityRunIdentity, + config: ObservabilityExportConfig, + initialInput?: string, + staleTimeoutMs = configuredStaleRunTimeoutMs(), +): void { + const runId = + identity.runId ?? (identity.turnId ? `turn-${identity.turnId}` : undefined); + if (!runId) return; + + const activeChatRun = identity.chatJid + ? activeRunsByChatJid.get(identity.chatJid) + : undefined; + if (activeChatRun) { + // Duplicate start notifications for the same turn are harmless. A new + // turn on the same chat means the old run missed its terminal callback; + // close it rather than silently suppressing this turn's trace. + if (identity.turnId && activeChatRun.turnId === identity.turnId) return; + finishAndRemoveActiveRun(activeChatRun, 'runner_exit'); + } + + const activeRun = activeRuns.get(runId); + if (activeRun) { + if ( + activeRun.chatJid === identity.chatJid && + activeRun.turnId === identity.turnId + ) { + return; + } + finishAndRemoveActiveRun(activeRun, 'runner_exit'); + } + + try { + const run = new ObservabilityRun( + { ...identity, runId }, + config, + initialInput, + ); + activeRuns.set(runId, run); + if (identity.chatJid) activeRunsByChatJid.set(identity.chatJid, run); + if (identity.turnId) activeRunsByTurnId.set(identity.turnId, run); + const timeout = setTimeout( + () => finishAndRemoveActiveRun(run, 'runner_exit'), + staleTimeoutMs, + ); + timeout.unref?.(); + activeRunTimeouts.set(run, timeout); + } catch { + // Misconfiguration must not affect the agent turn; no secret or endpoint + // detail is logged here. + } +} + +export function startObservabilityRun( + identity: ObservabilityRunIdentity, + initialInput?: string, +): void { + try { + startObservabilityRunWithConfig(identity, readExportConfig(), initialInput); + } catch { + // Observability configuration reads must never interrupt the agent turn. + } +} + +export async function withObservabilityCleanup( + chatJid: string, + getRunId: () => string | undefined, + operation: () => Promise, + getFinishReason: (result: T) => string, +): Promise { + let reason = 'runner_exit'; + try { + const result = await operation(); + reason = getFinishReason(result); + return result; + } finally { + let runId = ''; + try { + runId = getRunId() ?? ''; + } catch { + // The chat mapping remains the cleanup fallback if queue state is gone. + } + finishObservabilityRun(chatJid, runId, reason); + } +} + +export function createObservabilityRun( + identity: ObservabilityRunIdentity, + config: ObservabilityExportConfig, + initialInput?: string, +): ObservabilityRun | null { + try { + return new ObservabilityRun(identity, config, initialInput); + } catch { + return null; + } +} + +export function recordObservabilityEvent( + runId: string | undefined, + event: StreamEvent | undefined, +): void { + if (!event) return; + const run = + (runId ? activeRuns.get(runId) : undefined) ?? + (event.queryRunId ? activeRuns.get(event.queryRunId) : undefined) ?? + (event.turnId ? activeRunsByTurnId.get(event.turnId) : undefined); + if (!run) return; + if (event.queryRunId) { + run.bindRunId(event.queryRunId); + activeRuns.set(event.queryRunId, run); + } + run.record(event); +} + +export function recordObservabilityFailure( + chatJid: string, + runId: string | undefined, + category: 'provider_failure' | 'runner_error' | 'runtime_error', +): void { + const run = + (runId ? activeRuns.get(runId) : undefined) ?? + activeRunsByChatJid.get(chatJid); + run?.recordFailure(category); +} + +export function finishObservabilityRun( + chatJid: string, + runId: string, + reason: string, +): void { + const run = activeRuns.get(runId) ?? activeRunsByChatJid.get(chatJid); + if (!run) return; + run.bindRunId(runId); + finishAndRemoveActiveRun(run, reason); +} diff --git a/src/otel-redaction.ts b/src/otel-redaction.ts new file mode 100644 index 0000000..ddfada5 --- /dev/null +++ b/src/otel-redaction.ts @@ -0,0 +1,279 @@ +import { createHash } from 'node:crypto'; +import { isIP } from 'node:net'; + +const REDACTED = '[REDACTED]'; +const MAX_CONTENT_LENGTH = 16_000; + +const SENSITIVE_FIELD = + /(?:api[_ -]?key|secret|token|auth(?:orization)?|password|cookie|credential|endpoint|base[_ -]?url|url|email|e-mail|phone|mobile|telephone|contact|address|street|recipient|consignee|receiver|customer(?:[_ -]?(?:name|info|profile))?|ship[_ -]?to|shipping[_ -]?contact|delivery[_ -]?contact|full[_ -]?name|收件|收货|联系人|地址|电话|手机|邮箱|客户信息|姓名|令牌|密钥|凭据)/i; + +const SENSITIVE_ASSIGNMENT = + /((?:api[_ -]?key|secret(?:[_ -]?key)?|access[_ -]?token|refresh[_ -]?token|token|auth(?:orization)?|password|endpoint|base[_ -]?url)\s*[:=]\s*)["']?[^,\s;"'}]+/gi; + +const SECRET_SCHEME = + /((?:proxy-)?authorization\s*[:=]\s*)(?:basic|bearer|token)\s+[^\s,;"']+/gi; +const SCHEME_CREDENTIAL = /\b(?:basic|bearer|token)\s+[A-Za-z0-9+/=_\-.~]+/gi; +const JWT = + /\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b/g; +const GITHUB_TOKEN = + /\b(?:gh[pousr]_[A-Za-z0-9]{16,}|github_pat_[A-Za-z0-9_]{16,})\b/gi; +const AWS_ACCESS_KEY = /\b(?:AKIA|ASIA)[A-Z0-9]{16}\b/g; +const SLACK_TOKEN = /\bxox[baprs]-[A-Za-z0-9-]{10,}\b/gi; +const GITLAB_TOKEN = /\bglpat-[A-Za-z0-9_-]{16,}\b/gi; +const API_STYLE_TOKEN = /\b(?:sk|rk|pk)-[A-Za-z0-9_-]{12,}\b/gi; +const LONG_HEX = /\b[A-Fa-f0-9]{32,}\b/g; +const HIGH_ENTROPY_TOKEN = /\b[A-Za-z0-9_+/=-]{24,}\b/g; +const EMAIL = /[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}/gi; +const NATIONAL_ID = /(?(); + for (const char of value) counts.set(char, (counts.get(char) ?? 0) + 1); + let entropy = 0; + for (const count of counts.values()) { + const probability = count / value.length; + entropy -= probability * Math.log2(probability); + } + return entropy; +} + +function isHighEntropyToken(value: string): boolean { + return ( + value.length >= 24 && + /[A-Za-z]/.test(value) && + /\d/.test(value) && + shannonEntropy(value) >= 3.5 + ); +} + +function isLuhnValid(value: string): boolean { + const digits = value.replace(/\D/g, ''); + if (digits.length < 13 || digits.length > 19) return false; + let sum = 0; + let double = false; + for (let index = digits.length - 1; index >= 0; index -= 1) { + let digit = Number(digits[index]); + if (double) { + digit *= 2; + if (digit > 9) digit -= 9; + } + sum += digit; + double = !double; + } + return sum % 10 === 0; +} + +function redactCardNumbers(value: string): string { + return value.replace(/(? + isLuhnValid(match) ? REDACTED : match, + ); +} + +function redactIpAddresses(value: string): string { + const bracketed = value.replace( + /\[([0-9A-Fa-f:.]+)\](?::\d{1,5})?/g, + (match, host: string) => (isIP(host) === 6 ? REDACTED : match), + ); + const ipv4 = bracketed.replace( + /(? { + const [host, rawPort] = match.split(':'); + const port = rawPort === undefined ? undefined : Number(rawPort); + return isIP(host) === 4 && (port === undefined || port <= 65535) + ? REDACTED + : match; + }, + ); + return ipv4.replace( + /(? (isIP(match) === 6 ? REDACTED : match), + ); +} + +function redactChineseAddresses(value: string): string { + return ( + value + // Directly redact addresses in the four municipalities, which have no + // province component (for example 北京市朝阳区建国路88号3楼). + .replace( + /(?:北京市|天津市|上海市|重庆市)[\u4e00-\u9fff]{2,}(?:区|县)[^,，。;；\r\n]{2,}?(?:路|街|巷|弄|号)[^,，。;；\r\n]*/g, + REDACTED, + ) + .replace( + /(?:中国)?[\u4e00-\u9fff]{2,}(?:省|自治区)[\u4e00-\u9fff]{2,}(?:市|州|盟)[\u4e00-\u9fff]{2,}(?:区|县|镇|乡)[\u4e00-\u9fff0-9-]{2,}(?:路|街|巷|弄|号|楼|室)[\u4e00-\u9fff0-9-]*/g, + REDACTED, + ) + ); +} + +function redactText(value: string): string { + const preservedIds: string[] = []; + let result = value.replace(EVIDENCE_ID, (id) => { + const index = preservedIds.push(id) - 1; + return `__CA_EVIDENCE_ID_${index}__`; + }); + result = result + // Run credential/token patterns before phone/card patterns so digits in a + // token can never be partially replaced while leaving the token prefix. + .replace(SECRET_SCHEME, `$1${REDACTED}`) + .replace(SCHEME_CREDENTIAL, REDACTED) + .replace(SENSITIVE_ASSIGNMENT, `$1${REDACTED}`) + .replace(GITHUB_TOKEN, REDACTED) + .replace(AWS_ACCESS_KEY, REDACTED) + .replace(SLACK_TOKEN, REDACTED) + .replace(GITLAB_TOKEN, REDACTED) + .replace(API_STYLE_TOKEN, REDACTED) + .replace(JWT, REDACTED) + .replace(LONG_HEX, REDACTED) + .replace(HIGH_ENTROPY_TOKEN, (match) => + isHighEntropyToken(match) ? REDACTED : match, + ) + .replace(/https?:\/\/[^\s<>"'`]+/gi, REDACTED) + .replace(/\b(?:Cookie|Set-Cookie)\s*:\s*[^\r\n]+/gi, REDACTED) + .replace(EMAIL, REDACTED) + .replace(NATIONAL_ID, REDACTED); + + result = redactCardNumbers(result); + result = result.replace(PHONE, REDACTED); + result = redactIpAddresses(result); + result = redactChineseAddresses(result); + result = result.replace( + /((?:recipient(?:[_ -]?(?:name|address))?|consignee|receiver(?:[_ -]?(?:name|address))?|customer(?:[_ -]?(?:name|info|profile))?|shipping\s*address|delivery\s*address|address|收件人|收货人|联系人|收件地址|收货地址|配送地址|地址)\s*(?:是|[:：=])\s*)[^,，。;；\r\n]+/gi, + `$1${REDACTED}`, + ); + result = result.replace( + /\b\d{1,6}\s+[A-Za-z0-9.'-]+(?:\s+[A-Za-z0-9.'-]+){0,5}\s+(?:Street|St|Road|Rd|Avenue|Ave|Boulevard|Blvd|Lane|Ln|Drive|Dr|Apartment|Apt|Unit)\.?\s*(?:#\s*)?[A-Za-z0-9-]*\b/gi, + REDACTED, + ); + result = result.replace( + /__CA_EVIDENCE_ID_(\d+)__/g, + (placeholder, rawIndex) => preservedIds[Number(rawIndex)] ?? placeholder, + ); + return result.slice(0, MAX_CONTENT_LENGTH); +} + +/** Recursively remove sensitive fields and redact sensitive values in free text. */ +export function redactObservabilityValue( + value: unknown, + fieldName = '', +): unknown { + if (SENSITIVE_FIELD.test(fieldName)) return REDACTED; + if (typeof value === 'string') { + const trimmed = value.trim(); + if (trimmed.startsWith('{') || trimmed.startsWith('[')) { + try { + return JSON.stringify( + redactObservabilityValue(JSON.parse(trimmed)), + ).slice(0, MAX_CONTENT_LENGTH); + } catch { + // Treat malformed JSON as ordinary text and still apply pattern rules. + } + } + return redactText(value); + } + if (Array.isArray(value)) { + return value.map((item) => redactObservabilityValue(item)); + } + if (value && typeof value === 'object') { + return Object.fromEntries( + Object.entries(value).map(([key, child]) => [ + key, + redactObservabilityValue(child, key), + ]), + ); + } + return value; +} + +export function getResidualSensitiveCategories(value: string): string[] { + const scanValue = value.replace(EVIDENCE_ID, '[SAFE_EVIDENCE_ID]'); + const has = (pattern: RegExp) => { + pattern.lastIndex = 0; + return pattern.test(scanValue); + }; + const categories: string[] = []; + if (has(SECRET_SCHEME)) categories.push('auth_scheme'); + if (has(SCHEME_CREDENTIAL)) categories.push('credential_scheme'); + const assignmentPattern = new RegExp( + SENSITIVE_ASSIGNMENT.source, + SENSITIVE_ASSIGNMENT.flags, + ); + const hasUnredactedAssignment = Array.from( + scanValue.matchAll(assignmentPattern), + ).some((match) => { + const complete = match[0] ?? ''; + const prefix = match[1] ?? ''; + const assignedValue = complete.slice(prefix.length).replace(/^['"]/, ''); + return !assignedValue.startsWith(REDACTED); + }); + if (hasUnredactedAssignment) categories.push('credential_assignment'); + if (has(GITHUB_TOKEN)) categories.push('github_token'); + if (has(AWS_ACCESS_KEY)) categories.push('aws_access_key'); + if (has(SLACK_TOKEN)) categories.push('slack_token'); + if (has(GITLAB_TOKEN)) categories.push('gitlab_token'); + if (has(API_STYLE_TOKEN)) categories.push('api_style_token'); + if (has(JWT)) categories.push('jwt'); + if (has(LONG_HEX)) categories.push('long_hex'); + if (new RegExp(NATIONAL_ID.source, NATIONAL_ID.flags).test(scanValue)) { + categories.push('national_id'); + } + if (has(EMAIL)) categories.push('email'); + if (has(PHONE)) categories.push('phone'); + if (redactCardNumbers(scanValue) !== scanValue) + categories.push('payment_card'); + if (redactIpAddresses(scanValue) !== scanValue) categories.push('ip_address'); + if (redactChineseAddresses(scanValue) !== scanValue) + categories.push('street_address'); + const highEntropyPattern = new RegExp( + HIGH_ENTROPY_TOKEN.source, + HIGH_ENTROPY_TOKEN.flags, + ); + if ( + Array.from(scanValue.matchAll(highEntropyPattern)).some(([match]) => + isHighEntropyToken(match), + ) + ) { + categories.push('high_entropy_token'); + } + return categories; +} + +function serialize(value: unknown): string { + if (typeof value === 'string') return value; + try { + return JSON.stringify(value) ?? String(value); + } catch { + return String(value); + } +} + +/** + * If sanitization leaves a high-risk pattern, fail closed with metadata only. + * Exported so the fallback can be directly tested with simulated residuals. + */ +export function failClosedOnResidualSensitiveContent( + originalContent: string, + redactedContent: string, +): string { + if (getResidualSensitiveCategories(redactedContent).length === 0) { + return redactedContent; + } + return JSON.stringify({ + content_redacted: true, + original_length: originalContent.length, + original_sha256: createHash('sha256').update(originalContent).digest('hex'), + }); +} + +export function serializeRedactedContent(value: unknown): string { + const original = serialize(value); + const redacted = serialize(redactObservabilityValue(value)).slice( + 0, + MAX_CONTENT_LENGTH, + ); + return failClosedOnResidualSensitiveContent(original, redacted); +} diff --git a/src/routes/config.ts b/src/routes/config.ts index f38fe70..c90f168 100644 --- a/src/routes/config.ts +++ b/src/routes/config.ts @@ -108,6 +108,9 @@ import { getAppearanceConfig, saveAppearanceConfig, getSystemSettings, + hasObservabilitySecret, + isSafeOtelEndpoint, + saveObservabilitySecret, getEffectiveExternalDir, getContainerEnvConfig, saveSystemSettings, @@ -2220,6 +2223,12 @@ function toSystemSettingsResponse( maxLoginAttempts: settings.maxLoginAttempts, loginLockoutMinutes: settings.loginLockoutMinutes, fallbackModel: settings.fallbackModel, + observabilityEnabled: settings.observabilityEnabled, + observabilityEndpoint: settings.observabilityEndpoint, + observabilityPublicKey: settings.observabilityPublicKey, + observabilitySampleRate: settings.observabilitySampleRate, + observabilityIncludeContent: settings.observabilityIncludeContent, + observabilityHasSecretKey: hasObservabilitySecret(), }; } @@ -2256,11 +2265,36 @@ configRoutes.put( ); } + const actor = c.get('user') as AuthUser; + const writesObservabilitySettings = Object.keys(validation.data).some( + (key) => key.startsWith('observability'), + ); + if (writesObservabilitySettings && actor.role !== 'admin') { + return c.json( + { error: 'Only administrators can change OpenTelemetry settings' }, + 403, + ); + } + try { const before = toSystemSettingsResponse(getSystemSettings()); // Deprecated compatibility inputs. Accept them so stale Web clients do // not fail the whole request, but never persist or apply them. const effectiveSettings = { ...validation.data }; + const secretKey = effectiveSettings.observabilitySecretKey; + delete effectiveSettings.observabilitySecretKey; + if ( + effectiveSettings.observabilityEndpoint && + !isSafeOtelEndpoint(effectiveSettings.observabilityEndpoint) + ) { + return c.json( + { + error: + 'OpenTelemetry 地址必须使用 HTTPS；HTTP 仅允许 localhost 或 127.0.0.1。', + }, + 400, + ); + } for (const key of [ 'maxConcurrentHostProcesses', 'maxConcurrentScripts', @@ -2271,6 +2305,25 @@ configRoutes.put( ] as const) { delete effectiveSettings[key]; } + const candidateSettings = { + ...getSystemSettings(), + ...effectiveSettings, + }; + if ( + candidateSettings.observabilityEnabled && + (!candidateSettings.observabilityEndpoint || + !candidateSettings.observabilityPublicKey || + !(secretKey || hasObservabilitySecret())) + ) { + return c.json( + { + error: + '启用 OpenTelemetry 前，请填写服务地址、Public Key 和 Secret Key。', + }, + 400, + ); + } + if (secretKey?.trim()) saveObservabilitySecret(secretKey); const saved = saveSystemSettings(effectiveSettings); const response = toSystemSettingsResponse(saved); const changedFields = changedSettingFields( @@ -2278,8 +2331,8 @@ configRoutes.put( response, effectiveSettings, ); + if (secretKey) changedFields.push('observabilitySecretKey'); if (changedFields.length > 0) { - const actor = c.get('user') as AuthUser; logAuthEvent({ event_type: 'system_settings_updated', username: actor.username, diff --git a/src/runtime-config.ts b/src/runtime-config.ts index 3b62b34..cfff0de 100644 --- a/src/runtime-config.ts +++ b/src/runtime-config.ts @@ -3921,6 +3921,7 @@ const SYSTEM_SETTINGS_FILE = path.join( CLAUDE_CONFIG_DIR, 'system-settings.json', ); +const OTEL_SECRET_FILE = path.join(CLAUDE_CONFIG_DIR, 'otel-secret.json'); export interface SystemSettings { containerTimeout: number; @@ -3955,6 +3956,11 @@ export interface SystemSettings { // false = 关闭定时扫描，admin 仍可手点 POST /api/plugins/catalog/scan。 // 适用于不希望本机私有 plugin 自动入共享 catalog 的环境。 pluginAutoScan: boolean; + observabilityEnabled: boolean; + observabilityEndpoint: string; + observabilityPublicKey: string; + observabilitySampleRate: number; + observabilityIncludeContent: boolean; } // Upper bound for the login lockout window. auth.ts reclaims login-attempt @@ -3984,6 +3990,11 @@ const DEFAULT_SYSTEM_SETTINGS: SystemSettings = { mainAgentAutoCompactPercentage: 0, fallbackModel: '', pluginAutoScan: true, + observabilityEnabled: false, + observabilityEndpoint: '', + observabilityPublicKey: '', + observabilitySampleRate: 1, + observabilityIncludeContent: false, }; type SystemSettingsSource = 'file' | 'env' | 'api'; @@ -4077,6 +4088,35 @@ function normalizeSystemSettings( } } + let observabilityEndpoint = DEFAULT_SYSTEM_SETTINGS.observabilityEndpoint; + if (raw.observabilityEndpoint !== undefined) { + if (typeof raw.observabilityEndpoint === 'string') { + const candidate = raw.observabilityEndpoint.trim().replace(/\/+$/, ''); + if ( + candidate.length <= 512 && + (candidate === '' || isSafeOtelEndpoint(candidate)) + ) { + observabilityEndpoint = candidate; + } else { + invalidFields.add('observabilityEndpoint'); + } + } else { + invalidFields.add('observabilityEndpoint'); + } + } + + let observabilityPublicKey = DEFAULT_SYSTEM_SETTINGS.observabilityPublicKey; + if (raw.observabilityPublicKey !== undefined) { + if ( + typeof raw.observabilityPublicKey === 'string' && + raw.observabilityPublicKey.length <= 256 + ) { + observabilityPublicKey = raw.observabilityPublicKey.trim(); + } else { + invalidFields.add('observabilityPublicKey'); + } + } + let billingCurrency = DEFAULT_SYSTEM_SETTINGS.billingCurrency; if (raw.billingCurrency !== undefined) { if ( @@ -4197,6 +4237,25 @@ function normalizeSystemSettings( 'pluginAutoScan', DEFAULT_SYSTEM_SETTINGS.pluginAutoScan, ), + observabilityEnabled: booleanField( + 'observabilityEnabled', + DEFAULT_SYSTEM_SETTINGS.observabilityEnabled, + ), + observabilityEndpoint, + observabilityPublicKey, + observabilitySampleRate: numberField( + 'observabilitySampleRate', + DEFAULT_SYSTEM_SETTINGS.observabilitySampleRate, + 0, + 1, + false, + ), + // Stage 1 stores the explicit opt-in, but intentionally exports metadata + // only until Stage 2 adds content redaction. + observabilityIncludeContent: booleanField( + 'observabilityIncludeContent', + DEFAULT_SYSTEM_SETTINGS.observabilityIncludeContent, + ), }; if (invalidFields.size > 0) { @@ -4217,7 +4276,96 @@ function readSystemSettingsFromFile(): SystemSettings | null { const raw = JSON.parse( fs.readFileSync(SYSTEM_SETTINGS_FILE, 'utf-8'), ) as Record; - return normalizeSystemSettings(raw, 'file'); + return normalizeSystemSettings( + { + observabilityEnabled: + raw.observabilityEnabled ?? + parseEnvBoolean(process.env.COMMERCEAGENT_OTEL_ENABLED), + observabilityEndpoint: + raw.observabilityEndpoint ?? process.env.COMMERCEAGENT_OTEL_ENDPOINT, + observabilityPublicKey: + raw.observabilityPublicKey ?? process.env.COMMERCEAGENT_OTEL_PUBLIC_KEY, + observabilitySampleRate: + raw.observabilitySampleRate ?? + parseEnvNumber(process.env.COMMERCEAGENT_OTEL_SAMPLE_RATE), + observabilityIncludeContent: + raw.observabilityIncludeContent ?? + parseEnvBoolean(process.env.COMMERCEAGENT_OTEL_INCLUDE_CONTENT), + ...raw, + }, + 'file', + ); +} + +function parseEnvBoolean(value: string | undefined): boolean | undefined { + if (value === 'true') return true; + if (value === 'false') return false; + return undefined; +} + +function parseEnvNumber(value: string | undefined): number | undefined { + if (!value?.trim()) return undefined; + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : undefined; +} + +export function isSafeOtelEndpoint(value: string): boolean { + try { + const url = new URL(value); + return ( + (url.protocol === 'https:' || + (url.protocol === 'http:' && + ['localhost', '127.0.0.1'].includes(url.hostname.toLowerCase()))) && + !url.username && + !url.password && + !url.search && + !url.hash + ); + } catch { + return false; + } +} + +interface StoredOtelSecret { + iv: string; + tag: string; + data: string; +} + +export function saveObservabilitySecret(secret: string): void { + const normalized = secret.trim(); + if (normalized.length > 2048) + throw new Error('Observability secret is too long'); + if (!normalized) { + try { + fs.unlinkSync(OTEL_SECRET_FILE); + } catch (err) { + if ((err as NodeJS.ErrnoException).code !== 'ENOENT') throw err; + } + return; + } + fs.mkdirSync(CLAUDE_CONFIG_DIR, { recursive: true }); + const encrypted = encryptChannelSecret({ secret: normalized }); + writeSecretFile(OTEL_SECRET_FILE, JSON.stringify(encrypted) + '\n'); +} + +export function hasObservabilitySecret(): boolean { + return Boolean(getObservabilitySecret()); +} + +export function getObservabilitySecret(): string { + try { + if (fs.existsSync(OTEL_SECRET_FILE)) { + const encrypted = JSON.parse( + fs.readFileSync(OTEL_SECRET_FILE, 'utf8'), + ) as StoredOtelSecret; + const payload = decryptChannelSecret<{ secret?: unknown }>(encrypted); + if (typeof payload.secret === 'string') return payload.secret; + } + } catch { + logger.warn('Failed to read observability secret'); + } + return process.env.COMMERCEAGENT_OTEL_SECRET_KEY || ''; } function buildEnvFallbackSettings(): SystemSettings { @@ -4244,6 +4392,12 @@ function buildEnvFallbackSettings(): SystemSettings { process.env.AUTO_COMPACT_PERCENTAGE, fallbackModel: process.env.FALLBACK_MODEL, pluginAutoScan: process.env.PLUGIN_AUTO_SCAN, + observabilityEnabled: process.env.COMMERCEAGENT_OTEL_ENABLED, + observabilityEndpoint: process.env.COMMERCEAGENT_OTEL_ENDPOINT, + observabilityPublicKey: process.env.COMMERCEAGENT_OTEL_PUBLIC_KEY, + observabilitySampleRate: process.env.COMMERCEAGENT_OTEL_SAMPLE_RATE, + observabilityIncludeContent: + process.env.COMMERCEAGENT_OTEL_INCLUDE_CONTENT, }, 'env', ); diff --git a/src/schemas.ts b/src/schemas.ts index 4b0007e..1a1c2ee 100644 --- a/src/schemas.ts +++ b/src/schemas.ts @@ -721,6 +721,12 @@ export const SystemSettingsSchema = z maxLoginAttempts: z.number().int().min(1).max(100).optional(), loginLockoutMinutes: z.number().int().min(1).max(1440).optional(), fallbackModel: z.string().max(64).optional(), + observabilityEnabled: z.boolean().optional(), + observabilityEndpoint: z.string().max(512).optional(), + observabilityPublicKey: z.string().max(256).optional(), + observabilitySampleRate: z.number().min(0).max(1).optional(), + observabilityIncludeContent: z.boolean().optional(), + observabilitySecretKey: z.string().max(2048).optional(), }) .strict(); diff --git a/src/web.ts b/src/web.ts index 98e6c28..f1156a1 100644 --- a/src/web.ts +++ b/src/web.ts @@ -126,6 +126,7 @@ import { PLUGIN_EXPANSION_ATTACHMENT_TYPE } from './plugin-expander-sentinel.js' import { persistPluginExpansion } from './plugin-expander-store.js'; import { logger } from './logger.js'; import { recordRunContextSnapshot } from './run-context-snapshot.js'; +import { finishObservabilityRun } from './otel-observability.js'; import { RunStreamFence } from './run-stream-fence.js'; import { executeSessionReset, @@ -3191,6 +3192,7 @@ export function broadcastRunFinished( reason: RunFinishReason, finishedAt: number, ): void { + finishObservabilityRun(chatJid, runId, reason); const baseJid = stripRuntimeJidSuffix(chatJid); const jid = normalizeRuntimeJid(chatJid); const allowedUserIds = getGroupAllowedUserIds(baseJid); diff --git a/tests/benchmark-langfuse-drafts.test.ts b/tests/benchmark-langfuse-drafts.test.ts new file mode 100644 index 0000000..87cc481 --- /dev/null +++ b/tests/benchmark-langfuse-drafts.test.ts @@ -0,0 +1,311 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; + +import { + buildLangfuseDraftBundle, + collectLangfuseDraftBundle, + writeLangfuseDraftBundle, + type LangfuseObservation, + type LangfuseScore, +} from '../benchmarks/lib/langfuse-drafts.js'; + +const roots: LangfuseObservation[] = [ + { + id: 'observation-root-failure', + traceId: 'internal-trace-failure', + startTime: '2026-10-23T00:00:00.000Z', + name: 'commerceagent.turn', + metadata: { + run_id: 'run-synthetic-failure', + turn_id: 'turn-synthetic-failure', + 'attributes.commerceagent.run_finished.reason': 'runner_exit', + 'attributes.error.type': 'runner_error', + 'attributes.gen_ai.request.model': 'secret-model-name', + 'attributes.langfuse.user.id': 'private-user-hash', + 'attributes.langfuse.session.id': 'private-workspace-hash', + }, + }, + { + id: 'observation-root-low-score', + traceId: 'internal-trace-low-score', + startTime: '2026-10-23T00:01:00.000Z', + name: 'commerceagent.turn', + metadata: { + run_id: 'run-synthetic-low-score', + turn_id: 'turn-synthetic-low-score', + 'attributes.commerceagent.run_finished.reason': 'completed', + }, + }, + { + id: 'observation-root-no-identity', + traceId: 'internal-trace-no-identity', + startTime: '2026-10-23T00:02:00.000Z', + name: 'commerceagent.turn', + metadata: { + 'attributes.commerceagent.run_finished.reason': 'stopped', + }, + }, +]; + +const tools: LangfuseObservation[] = [ + { + id: 'observation-tool-1', + traceId: 'internal-trace-failure', + name: 'commerceagent.tool', + metadata: { + 'attributes.gen_ai.tool.name': 'mcp__commerce_fixture__get_order', + 'attributes.customer_email': 'customer@example.test', + }, + }, + { + id: 'observation-tool-2', + traceId: 'internal-trace-low-score', + name: 'commerceagent.tool', + metadata: { + 'attributes.gen_ai.tool.name': 'mcp__commerce_fixture__search_policy', + }, + }, +]; + +const scores: LangfuseScore[] = [ + { + dataType: 'NUMERIC', + value: 2, + subject: { kind: 'trace', id: 'internal-trace-low-score' }, + }, + { + dataType: 'NUMERIC', + value: 5, + subject: { kind: 'trace', id: 'internal-trace-failure' }, + }, +]; + +const tempDirectories: string[] = []; +afterEach(() => { + for (const directory of tempDirectories.splice(0)) + fs.rmSync(directory, { recursive: true, force: true }); +}); + +describe('Langfuse metadata-only regression drafts', () => { + it('builds empty structured skeletons without copying trace content or identity', () => { + const bundle = buildLangfuseDraftBundle({ + roots, + tools, + scores, + generatedAt: new Date('2026-10-24T00:00:00.000Z'), + }); + const serialized = JSON.stringify(bundle); + + expect(bundle.selection).toMatchObject({ + failedTraceCount: 1, + lowScoreTraceCount: 1, + draftCount: 2, + skippedMissingRunIdentity: 1, + bodyFieldsRequested: false, + holdoutRead: false, + casesModified: false, + }); + expect(bundle.drafts[0]).toMatchObject({ + source: { + runId: 'run-synthetic-failure', + turnId: 'turn-synthetic-failure', + signal: 'failure', + terminalReason: 'runner_exit', + errorCategory: 'runner_error', + toolNames: ['mcp__commerce_fixture__get_order'], + }, + caseDraft: { + id: '', + category: '', + riskLevel: '', + split: '', + prompt: '', + expected: { status: '', facts: {} }, + }, + promotionRequirements: { + humanReviewRequired: true, + referenceAnswerSelfTestRequired: true, + mutationTestsRequired: true, + automaticPromotion: false, + }, + }); + expect(bundle.drafts[1].source).toMatchObject({ + signal: 'low_score', + lowScoreCount: 1, + }); + for (const forbidden of [ + 'internal-trace-', + 'secret-model-name', + 'private-user-hash', + 'private-workspace-hash', + 'customer@example.test', + 'input', + 'output', + ]) + expect(serialized).not.toContain(forbidden); + }); + + it('writes only under the ignored results draft directory with private permissions', () => { + const benchmarkDirectory = fs.mkdtempSync( + path.join(os.tmpdir(), 'commerce-langfuse-drafts-'), + ); + tempDirectories.push(benchmarkDirectory); + const bundle = buildLangfuseDraftBundle({ roots, tools, scores }); + const output = writeLangfuseDraftBundle( + benchmarkDirectory, + bundle, + new Date('2026-10-24T00:00:00.000Z'), + ); + + expect(output).toBe( + path.join( + benchmarkDirectory, + 'results', + 'langfuse-drafts', + 'run-2026-10-24T00-00-00-000Z', + ), + ); + expect(fs.existsSync(path.join(output, 'drafts.json'))).toBe(true); + expect(fs.statSync(output).mode & 0o777).toBe(0o700); + expect(fs.statSync(path.join(output, 'drafts.json')).mode & 0o777).toBe( + 0o600, + ); + expect(fs.existsSync(path.join(benchmarkDirectory, 'cases'))).toBe(false); + }); + + it('requests bounded metadata fields and numeric scores without body fields', async () => { + const requests: Array<{ url: URL; authorization: string | null }> = []; + const fetcher: typeof fetch = async (input, init) => { + const url = new URL(String(input)); + requests.push({ + url, + authorization: new Headers(init?.headers).get('authorization'), + }); + let data: unknown[] = []; + if (url.pathname.endsWith('/observations')) { + data = + url.searchParams.get('name') === 'commerceagent.turn' + ? [roots[0]] + : [tools[0]]; + } else if (url.pathname.endsWith('/scores')) { + data = [scores[0]]; + } + return new Response(JSON.stringify({ data, meta: { cursor: null } }), { + status: 200, + headers: { 'content-type': 'application/json' }, + }); + }; + + const bundle = await collectLangfuseDraftBundle({ + endpoint: 'http://127.0.0.1:3100', + publicKey: 'pk-synthetic', + secretKey: 'sk-synthetic', + from: new Date('2026-10-20T00:00:00.000Z'), + to: new Date('2026-10-24T00:00:00.000Z'), + minScore: 4, + fetcher, + }); + + expect(bundle.selection.draftCount).toBe(1); + expect(requests).toHaveLength(3); + for (const request of requests) { + expect( + request.url.searchParams.get('fromStartTime') ?? + request.url.searchParams.get('fromTimestamp'), + ).toBe('2026-10-20T00:00:00.000Z'); + expect( + request.url.searchParams.get('toStartTime') ?? + request.url.searchParams.get('toTimestamp'), + ).toBe('2026-10-24T00:00:00.000Z'); + expect(request.url.searchParams.get('fields')).not.toContain('io'); + expect(request.authorization).toMatch(/^Basic /); + expect(request.url.toString()).not.toContain('sk-synthetic'); + } + const scoreRequest = requests.find((request) => + request.url.pathname.endsWith('/scores'), + ); + expect(scoreRequest?.url.searchParams.get('dataType')).toBe('NUMERIC'); + expect(scoreRequest?.url.searchParams.get('valueMax')).toBe('3.999999'); + }); + + it('rejects unexpected input/output fields from Langfuse responses', async () => { + const fetcher: typeof fetch = async () => + new Response( + JSON.stringify({ + data: [ + { + ...roots[0], + input: 'PRIVATE_SYNTHETIC_MESSAGE', + }, + ], + meta: { cursor: null }, + }), + { status: 200 }, + ); + + await expect( + collectLangfuseDraftBundle({ + endpoint: 'http://127.0.0.1:3100', + publicKey: 'pk-synthetic', + secretKey: 'sk-synthetic', + from: new Date('2026-10-20T00:00:00.000Z'), + to: new Date('2026-10-24T00:00:00.000Z'), + minScore: 4, + fetcher, + }), + ).rejects.toThrow('langfuse_content_field_rejected'); + }); + + it('rejects score comments instead of exporting free-form annotations', async () => { + const fetcher: typeof fetch = async (input) => { + const url = new URL(String(input)); + const data = url.pathname.endsWith('/scores') + ? [{ ...scores[0], comment: 'PRIVATE_SYNTHETIC_REVIEW_TEXT' }] + : []; + return new Response(JSON.stringify({ data, meta: { cursor: null } }), { + status: 200, + }); + }; + + await expect( + collectLangfuseDraftBundle({ + endpoint: 'http://127.0.0.1:3100', + publicKey: 'pk-synthetic', + secretKey: 'sk-synthetic', + from: new Date('2026-10-20T00:00:00.000Z'), + to: new Date('2026-10-24T00:00:00.000Z'), + minScore: 4, + fetcher, + }), + ).rejects.toThrow('langfuse_content_field_rejected'); + }); + + it('accepts null or empty optional body fields without copying them', async () => { + const fetcher: typeof fetch = async (input) => { + const url = new URL(String(input)); + const data = url.pathname.endsWith('/scores') + ? [{ ...scores[0], comment: null, metadata: { note: '' } }] + : url.searchParams.get('name') === 'commerceagent.turn' + ? [{ ...roots[0], input: null, output: '', statusMessage: null }] + : tools; + return new Response(JSON.stringify({ data, meta: { cursor: null } }), { + status: 200, + }); + }; + + const bundle = await collectLangfuseDraftBundle({ + endpoint: 'http://127.0.0.1:3100', + publicKey: 'pk-synthetic', + secretKey: 'sk-synthetic', + from: new Date('2026-10-20T00:00:00.000Z'), + to: new Date('2026-10-24T00:00:00.000Z'), + minScore: 4, + fetcher, + }); + expect(bundle.selection.draftCount).toBe(1); + expect(JSON.stringify(bundle)).not.toContain('statusMessage'); + expect(JSON.stringify(bundle)).not.toContain('comment'); + }); +}); diff --git a/tests/otel-observability.test.ts b/tests/otel-observability.test.ts new file mode 100644 index 0000000..8d18abe --- /dev/null +++ b/tests/otel-observability.test.ts @@ -0,0 +1,511 @@ +import { createServer, type Server } from 'node:http'; +import { once } from 'node:events'; + +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import type { StreamEvent } from '../shared/stream-event.js'; +import { createHash } from 'node:crypto'; +import { + createObservabilityRun, + finishObservabilityRun, + getActiveObservabilityRunCount, + getObservabilityExportFailureCount, + startObservabilityRunWithConfig, + type ObservabilityExportConfig, + withObservabilityCleanup, +} from '../src/otel-observability.js'; + +const identity = { + runId: 'run-synthetic-123', + turnId: 'turn-synthetic-456', + workspace: '/synthetic/workspace', + profileVersion: 7, + userId: 'user-synthetic-789', +}; + +const config: ObservabilityExportConfig = { + enabled: true, + endpoint: 'http://127.0.0.1:4318', + publicKey: 'pk-test', + secretKey: 'sk-test', + sampleRate: 1, + includeContent: false, +}; + +async function listen(server: Server): Promise { + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const address = server.address(); + if (!address || typeof address === 'string') throw new Error('No test port'); + return address.port; +} + +async function close(server: Server): Promise { + if (!server.listening) return; + server.close(); + await once(server, 'close'); +} + +function createCollector() { + const requestBodies: string[] = []; + const waiters: Array<{ count: number; resolve: () => void }> = []; + const server = createServer((req, res) => { + let body = ''; + req.setEncoding('utf8'); + req.on('data', (chunk: string) => (body += chunk)); + req.on('end', () => { + requestBodies.push(body); + res.writeHead(200).end(); + for (const waiter of waiters.splice(0)) { + if (requestBodies.length >= waiter.count) waiter.resolve(); + else waiters.push(waiter); + } + }); + }); + return { + server, + requestBodies, + waitForRequests(count: number) { + if (requestBodies.length >= count) return Promise.resolve(); + return new Promise((resolve) => waiters.push({ count, resolve })); + }, + }; +} + +describe('OpenTelemetry runtime export', () => { + const servers: Server[] = []; + + afterEach(async () => { + await Promise.all(servers.splice(0).map(close)); + }); + + it('does not create an exporter or make a request while disabled', async () => { + const server = createServer((_req, res) => { + res.writeHead(200).end(); + }); + servers.push(server); + const port = await listen(server); + const run = createObservabilityRun(identity, { + ...config, + endpoint: `http://127.0.0.1:${port}`, + enabled: false, + }); + + expect(run).toBeNull(); + }); + + it('exports metadata without any message or tool content', async () => { + let requestBody = ''; + let resolveRequest!: () => void; + const requestReceived = new Promise((resolve) => { + resolveRequest = resolve; + }); + const server = createServer((req, res) => { + req.setEncoding('utf8'); + req.on('data', (chunk: string) => (requestBody += chunk)); + req.on('end', () => { + res.writeHead(200).end(); + resolveRequest(); + }); + }); + servers.push(server); + const port = await listen(server); + const run = createObservabilityRun(identity, { + ...config, + endpoint: `http://127.0.0.1:${port}`, + }); + expect(run).not.toBeNull(); + + const promptMarker = 'PRIVATE_PROMPT_SENTINEL_4f32'; + const toolInputMarker = 'PRIVATE_TOOL_INPUT_SENTINEL_92bd'; + const toolResultMarker = 'PRIVATE_TOOL_RESULT_SENTINEL_e130'; + const assistantMarker = 'PRIVATE_ASSISTANT_SENTINEL_581e'; + run!.record({ + eventType: 'raw_sdk_event', + rawType: 'stream_event/message_start', + messageUuid: 'api-message-id-1', + apiRequestId: 'request-id-1', + text: promptMarker, + rawEvent: { content: promptMarker }, + } as StreamEvent); + run!.record({ + eventType: 'tool_use_start', + toolName: 'get_order', + toolUseId: 'tool-call-1', + toolInputSummary: toolInputMarker, + toolInput: { order: toolInputMarker }, + } as StreamEvent); + run!.record({ + eventType: 'tool_result', + toolUseId: 'tool-call-1', + toolResult: toolResultMarker, + detail: toolResultMarker, + } as StreamEvent); + run!.record({ + eventType: 'text_delta', + text: assistantMarker, + } as StreamEvent); + run!.record({ + eventType: 'usage', + usage: { + eventId: 'usage-event-id-1', + inputTokens: 10, + outputTokens: 5, + cacheReadInputTokens: 2, + cacheCreationInputTokens: 0, + reasoningTokens: 0, + costUSD: 0.001, + durationMs: 20, + numTurns: 1, + modelUsage: { + 'model-synthetic': { + inputTokens: 10, + outputTokens: 5, + cacheReadInputTokens: 2, + cacheCreationInputTokens: 0, + reasoningTokens: 0, + costUSD: 0.001, + }, + }, + }, + } as StreamEvent); + run!.finish('completed'); + await requestReceived; + + for (const marker of [ + promptMarker, + toolInputMarker, + toolResultMarker, + assistantMarker, + ]) { + expect(requestBody).not.toContain(marker); + } + expect(requestBody).toContain('run-synthetic-123'); + expect(requestBody).toContain('turn-synthetic-456'); + expect(requestBody).toContain('model-synthetic'); + expect(requestBody).toContain('get_order'); + expect(requestBody).toContain('usage-event-id-1'); + expect(requestBody).toContain('completed'); + }); + + it('exports only locally redacted content when explicitly enabled', async () => { + let requestBody = ''; + let resolveRequest!: () => void; + const requestReceived = new Promise((resolve) => { + resolveRequest = resolve; + }); + const server = createServer((req, res) => { + req.setEncoding('utf8'); + req.on('data', (chunk: string) => (requestBody += chunk)); + req.on('end', () => { + res.writeHead(200).end(); + resolveRequest(); + }); + }); + servers.push(server); + const port = await listen(server); + const run = createObservabilityRun( + identity, + { + ...config, + endpoint: `http://127.0.0.1:${port}`, + includeContent: true, + }, + 'Please deliver the parcel and confirm the order status. Email private@example.test, call +1 415-555-2671, use https://api.example.test/v1 with api_key=sk-live-otelsynthetic1234567890 Bearer synthetic-bearer-value. Address: 123 Main Street.', + ); + expect(run).not.toBeNull(); + run!.record({ + eventType: 'tool_use_start', + toolName: 'lookup_order', + toolUseId: 'private-tool-call', + toolInput: { + order_id: 'synthetic-order-123', + recipientName: 'Synthetic Recipient Name', + shipping_address: '88 Example Road, Example City', + phoneNumber: '13800138000', + }, + } as StreamEvent); + run!.record({ + eventType: 'tool_result', + toolUseId: 'private-tool-call', + toolResult: + 'Order synthetic-order-123 is delayed. recipient: Synthetic Recipient Name, address: 88 Example Road, Example City; Cookie: session=synthetic-cookie; Authorization: Bearer synthetic-result-token', + detail: 'detail-must-not-be-exported', + } as StreamEvent); + run!.record({ + eventType: 'text_delta', + text: 'Answer: the parcel is delayed and the order is still active.', + } as StreamEvent); + run!.finish('completed'); + await requestReceived; + + for (const sensitive of [ + 'private@example.test', + '+1 415-555-2671', + 'https://api.example.test/v1', + 'sk-live-otelsynthetic1234567890', + 'synthetic-bearer-value', + '123 Main Street', + 'Synthetic Recipient Name', + '88 Example Road, Example City', + '13800138000', + 'synthetic-cookie', + 'synthetic-result-token', + 'detail-must-not-be-exported', + ]) { + expect(requestBody).not.toContain(sensitive); + } + expect(requestBody).toContain('Please deliver the parcel'); + expect(requestBody).toContain('Answer: the parcel is delayed'); + expect(requestBody).toContain('[REDACTED]'); + }); + + it('keeps user identity hashes scoped to each trace', async () => { + const requestBodies: string[] = []; + let resolveRequests!: () => void; + const requestsReceived = new Promise((resolve) => { + resolveRequests = resolve; + }); + const server = createServer((req, res) => { + let body = ''; + req.setEncoding('utf8'); + req.on('data', (chunk: string) => (body += chunk)); + req.on('end', () => { + requestBodies.push(body); + res.writeHead(200).end(); + if (requestBodies.length === 2) resolveRequests(); + }); + }); + servers.push(server); + const port = await listen(server); + const userA = createObservabilityRun( + { ...identity, runId: 'run-user-a', userId: 'private-user-a' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + ); + const userB = createObservabilityRun( + { ...identity, runId: 'run-user-b', userId: 'private-user-b' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + ); + expect(userA).not.toBeNull(); + expect(userB).not.toBeNull(); + userA!.finish('completed'); + userB!.finish('completed'); + await requestsReceived; + + const hashA = createHash('sha256').update('private-user-a').digest('hex'); + const hashB = createHash('sha256').update('private-user-b').digest('hex'); + expect(requestBodies[0]).toContain(hashA); + expect(requestBodies[0]).not.toContain(hashB); + expect(requestBodies[0]).not.toContain('private-user-b'); + expect(requestBodies[1]).toContain(hashB); + expect(requestBodies[1]).not.toContain(hashA); + expect(requestBodies[1]).not.toContain('private-user-a'); + }); + + it('rebinds a fallback turn trace to the exact queue run ID before export', async () => { + let requestBody = ''; + let resolveRequest!: () => void; + const requestReceived = new Promise((resolve) => { + resolveRequest = resolve; + }); + const server = createServer((req, res) => { + req.setEncoding('utf8'); + req.on('data', (chunk: string) => (requestBody += chunk)); + req.on('end', () => { + res.writeHead(200).end(); + resolveRequest(); + }); + }); + servers.push(server); + const port = await listen(server); + const run = createObservabilityRun( + { + turnId: 'turn-fallback-1', + chatJid: 'synthetic-chat', + workspace: '/synthetic/workspace', + profileVersion: 7, + userId: 'user-synthetic-789', + }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + ); + expect(run).not.toBeNull(); + run!.bindRunId('queue-run-exact-1'); + run!.finish('completed'); + await requestReceived; + + expect(requestBody).toContain('queue-run-exact-1'); + expect(requestBody).not.toContain('turn-turn-fallback-1'); + expect(requestBody).toContain('turn-fallback-1'); + }); + + it('cleans up an early return and allows the next turn on the same chat', async () => { + const collector = createCollector(); + servers.push(collector.server); + const port = await listen(collector.server); + const chatJid = 'synthetic-early-return-chat'; + startObservabilityRunWithConfig( + { ...identity, runId: undefined, chatJid, turnId: 'early-turn-1' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 10_000, + ); + + const result = await withObservabilityCleanup( + chatJid, + () => undefined, + async () => false, + (value) => (value ? 'completed' : 'runner_exit'), + ); + expect(result).toBe(false); + expect(getActiveObservabilityRunCount()).toBe(0); + await collector.waitForRequests(1); + + startObservabilityRunWithConfig( + { ...identity, runId: undefined, chatJid, turnId: 'early-turn-2' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 10_000, + ); + expect(getActiveObservabilityRunCount()).toBe(1); + finishObservabilityRun(chatJid, '', 'completed'); + await collector.waitForRequests(2); + expect(getActiveObservabilityRunCount()).toBe(0); + expect(collector.requestBodies[0]).toContain('early-turn-1'); + expect(collector.requestBodies[1]).toContain('early-turn-2'); + }); + + it('cleans up and rethrows when a conversation turn throws', async () => { + const collector = createCollector(); + servers.push(collector.server); + const port = await listen(collector.server); + const chatJid = 'synthetic-thrown-chat'; + startObservabilityRunWithConfig( + { ...identity, runId: undefined, chatJid, turnId: 'throw-turn-1' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 10_000, + ); + + await expect( + withObservabilityCleanup( + chatJid, + () => undefined, + async () => { + throw new Error('synthetic turn failure'); + }, + () => 'completed', + ), + ).rejects.toThrow('synthetic turn failure'); + expect(getActiveObservabilityRunCount()).toBe(0); + await collector.waitForRequests(1); + expect(collector.requestBodies[0]).toContain('runner_exit'); + }); + + it('expires a run without run_finished and starts the following turn', async () => { + const collector = createCollector(); + servers.push(collector.server); + const port = await listen(collector.server); + const chatJid = 'synthetic-missing-terminal-chat'; + startObservabilityRunWithConfig( + { ...identity, runId: undefined, chatJid, turnId: 'missing-finish-1' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 25, + ); + expect(getActiveObservabilityRunCount()).toBe(1); + + await collector.waitForRequests(1); + await vi.waitFor(() => { + expect(getActiveObservabilityRunCount()).toBe(0); + }); + expect(collector.requestBodies[0]).toContain('runner_exit'); + + startObservabilityRunWithConfig( + { ...identity, runId: undefined, chatJid, turnId: 'missing-finish-2' }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 10_000, + ); + finishObservabilityRun(chatJid, '', 'completed'); + await collector.waitForRequests(2); + expect(getActiveObservabilityRunCount()).toBe(0); + expect(collector.requestBodies[1]).toContain('missing-finish-2'); + }); + + it('keeps simultaneous chat runs isolated when one finishes', async () => { + const collector = createCollector(); + servers.push(collector.server); + const port = await listen(collector.server); + startObservabilityRunWithConfig( + { + ...identity, + runId: undefined, + chatJid: 'synthetic-parallel-chat-a', + turnId: 'parallel-turn-a', + }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 10_000, + ); + startObservabilityRunWithConfig( + { + ...identity, + runId: undefined, + chatJid: 'synthetic-parallel-chat-b', + turnId: 'parallel-turn-b', + }, + { ...config, endpoint: `http://127.0.0.1:${port}` }, + undefined, + 10_000, + ); + expect(getActiveObservabilityRunCount()).toBe(2); + + finishObservabilityRun('synthetic-parallel-chat-a', '', 'completed'); + await collector.waitForRequests(1); + expect(getActiveObservabilityRunCount()).toBe(1); + finishObservabilityRun('synthetic-parallel-chat-b', '', 'completed'); + await collector.waitForRequests(2); + expect(getActiveObservabilityRunCount()).toBe(0); + expect(collector.requestBodies[0]).toContain('parallel-turn-a'); + expect(collector.requestBodies[1]).toContain('parallel-turn-b'); + }); + + it('keeps turn completion and usage independent of an unreachable exporter', async () => { + const run = createObservabilityRun(identity, { + ...config, + endpoint: 'http://127.0.0.1:1', + }); + expect(run).not.toBeNull(); + const usage = { + inputTokens: 30, + outputTokens: 7, + costUSD: 0.002, + }; + const before = { ...usage }; + const startedAt = Date.now(); + run!.record({ + eventType: 'usage', + usage: { + eventId: 'usage-survives-export-failure', + inputTokens: usage.inputTokens, + outputTokens: usage.outputTokens, + cacheReadInputTokens: 0, + cacheCreationInputTokens: 0, + reasoningTokens: 0, + costUSD: usage.costUSD, + durationMs: 15, + numTurns: 1, + }, + } as StreamEvent); + expect(() => run!.finish('completed')).not.toThrow(); + expect(Date.now() - startedAt).toBeLessThan(100); + expect(usage).toEqual(before); + + await vi.waitFor( + () => { + expect(getObservabilityExportFailureCount()).toBeGreaterThan(0); + }, + { timeout: 5_000 }, + ); + }); +}); diff --git a/tests/otel-redaction.test.ts b/tests/otel-redaction.test.ts new file mode 100644 index 0000000..99d8cdd --- /dev/null +++ b/tests/otel-redaction.test.ts @@ -0,0 +1,145 @@ +import { describe, expect, test } from 'vitest'; + +import { + failClosedOnResidualSensitiveContent, + redactObservabilityValue, + serializeRedactedContent, +} from '../src/otel-redaction.js'; + +describe('OpenTelemetry content redaction', () => { + test('redacts sensitive fields recursively while preserving unrelated facts', () => { + const result = serializeRedactedContent({ + order_id: 'fixture-order-17', + status: 'delayed', + recipientName: 'Synthetic Recipient', + shipping_address: '8 Example Road, Example City', + customer: { + name: 'Nested Customer Name', + phoneNumber: '13800138000', + email: 'recipient@example.test', + }, + apiKey: 'sk-live-synthetic-key-123456789', + endpoint: 'https://api.example.test/anthropic', + }); + + for (const sensitive of [ + 'Synthetic Recipient', + 'Nested Customer Name', + '8 Example Road', + '13800138000', + 'recipient@example.test', + 'sk-live-synthetic-key-123456789', + 'https://api.example.test/anthropic', + ]) { + expect(result).not.toContain(sensitive); + } + expect(result).toContain('fixture-order-17'); + expect(result).toContain('delayed'); + expect(result).toContain('[REDACTED]'); + }); + + test('redacts credentials, endpoint URLs, contact information and addresses in free text', () => { + const text = + 'Email private@example.test; phone +1 415-555-2671; address 123 Main Street. ' + + 'Visit https://service.example.test/api?secret=x; api_key=sk-live-text-key-123456789; ' + + 'Authorization: Bearer synthetic-bearer; Cookie: session=synthetic-cookie; ' + + 'Endpoint: api.example.test/v1; token=synthetic-free-token; ' + + 'JWT eyJsyntheticheader.eyJsyntheticpayload.synthetic-signature-value; ' + + '地址：北京市朝阳区建国路88号。'; + const result = serializeRedactedContent(text); + + for (const sensitive of [ + 'private@example.test', + '+1 415-555-2671', + '123 Main Street', + 'https://service.example.test/api?secret=x', + 'sk-live-text-key-123456789', + 'synthetic-bearer', + 'synthetic-cookie', + 'api.example.test/v1', + 'synthetic-free-token', + 'eyJsyntheticheader.eyJsyntheticpayload.synthetic-signature-value', + '北京市朝阳区建国路88号', + ]) { + expect(result).not.toContain(sensitive); + } + expect(result).toContain('[REDACTED]'); + }); + + test('bounds serialized content and can redact a named sensitive scalar', () => { + expect(redactObservabilityValue('synthetic-value', 'access_token')).toBe( + '[REDACTED]', + ); + expect(serializeRedactedContent('x'.repeat(20_000))).toHaveLength(16_000); + }); + + test('redacts credentials, provider tokens, personal identifiers, cards, IPs and naked municipality addresses', () => { + const sensitiveSamples = [ + 'Authorization: Basic dXNlcjpwYXNz', + 'ghp_1234567890abcdefghijklmnopqrstuv', + 'gho_1234567890abcdefghijklmnopqrstuv', + 'github_pat_1234567890_abcdefghijklmnopqrstuvwxyz', + 'AKIAIOSFODNN7EXAMPLE', + 'ASIAIOSFODNN7EXAMPLE', + 'xoxb-SYNTHETIC-ONLY-DO-NOT-USE', + 'glpat-1234567890abcdefghijklmnop', + '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef', + 'o9R5q0nZ7W2cK4mY8sA1vP6d', + '11010519491231002X', + '4111111111111111', + '10.0.3.12:8080', + '[2001:db8::1]:443', + '北京市朝阳区建国路88号3楼', + ]; + const raw = `Sensitive values: ${sensitiveSamples.join(' | ')}`; + const result = serializeRedactedContent({ + text: raw, + order_id: 'ORD-2026-0912', + amount: 129.5, + date: '2026-09-12', + }); + + for (const sensitive of sensitiveSamples) { + expect(result).not.toContain(sensitive); + } + expect(result).toContain('ORD-2026-0912'); + expect(result).toContain('129.5'); + expect(result).toContain('2026-09-12'); + }); + + test('fails closed to original length and hash if a high-risk value survives redaction', () => { + const original = 'This body contains a synthetic credential.'; + const fallback = failClosedOnResidualSensitiveContent( + original, + 'Authorization: Basic dXNlcjpwYXNz', + ); + const parsed = JSON.parse(fallback); + expect(parsed).toEqual({ + content_redacted: true, + original_length: original.length, + original_sha256: expect.any(String), + }); + expect(fallback).not.toContain('dXNlcjpwYXNz'); + expect(fallback).not.toContain(original); + }); + + test('preserves uppercase fixture evidence IDs without allowing long phone or card values', () => { + const evidenceIds = [ + 'AD-CAMP-100-20260901', + 'ORDER-100', + 'POLICY-US-RETURN-1', + 'CH-220', + 'REFUND-001', + 'HANDOFF-001', + ]; + const raw = `${evidenceIds.join(' ')} ORDER-13800138000 ORDER-4111111111111111`; + const result = serializeRedactedContent(raw); + + for (const evidenceId of evidenceIds) { + expect(result).toContain(evidenceId); + } + expect(result).not.toContain('13800138000'); + expect(result).not.toContain('4111111111111111'); + expect(result).toContain('ORDER-[REDACTED]'); + }); +}); diff --git a/tests/routes-system-settings-security.test.ts b/tests/routes-system-settings-security.test.ts index 14fa355..952ebc3 100644 --- a/tests/routes-system-settings-security.test.ts +++ b/tests/routes-system-settings-security.test.ts @@ -34,7 +34,9 @@ vi.mock('../src/middleware/auth.ts', async (importOriginal) => { display_name: 'Settings Security User', role: process.env.COMMERCEAGENT_TEST_ROLE ?? 'member', status: 'active', - permissions: JSON.parse(process.env.COMMERCEAGENT_TEST_PERMISSIONS ?? '[]'), + permissions: JSON.parse( + process.env.COMMERCEAGENT_TEST_PERMISSIONS ?? '[]', + ), must_change_password: false, }); return next(); @@ -112,6 +114,51 @@ afterAll(() => { }); describe('system settings capability boundaries', () => { + test('users without manage_system_config cannot read or change OTel settings', async () => { + asUser('member'); + expect((await app.request('/api/config/system')).status).toBe(403); + const response = await app.request('/api/config/system', { + method: 'PUT', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + observabilityEnabled: true, + observabilitySecretKey: 'synthetic-denied-secret', + }), + }); + expect(response.status).toBe(403); + expect(fs.existsSync(path.join(tmpDir, 'config', 'otel-secret.json'))).toBe( + false, + ); + }); + + test('non-admin with manage_system_config still cannot change OTel settings', async () => { + asUser('member', ['manage_system_config']); + const response = await app.request('/api/config/system', { + method: 'PUT', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ observabilityEnabled: true }), + }); + expect(response.status).toBe(403); + }); + + test('rejects remote HTTP OTel endpoints before storing credentials', async () => { + asUser('admin', ['manage_system_config']); + const response = await app.request('/api/config/system', { + method: 'PUT', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ + observabilityEnabled: true, + observabilityEndpoint: 'http://otel.example.test', + observabilityPublicKey: 'pk-synthetic', + observabilitySecretKey: 'synthetic-endpoint-secret', + }), + }); + expect(response.status).toBe(400); + expect(fs.existsSync(path.join(tmpDir, 'config', 'otel-secret.json'))).toBe( + false, + ); + }); + test('system config response excludes host and billing fields', async () => { asUser('member', ['manage_system_config']); const response = await app.request('/api/config/system'); @@ -137,6 +184,29 @@ describe('system settings capability boundaries', () => { expect(response.status).toBe(400); }); + test('stores OTel Secret Key encrypted and never returns it from system config', async () => { + asUser('admin', ['manage_system_config']); + const secret = 'synthetic-otel-secret-never-return-this'; + const saved = await app.request('/api/config/system', { + method: 'PUT', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ observabilitySecretKey: secret }), + }); + expect(saved.status).toBe(200); + const responseBody = await saved.text(); + expect(responseBody).not.toContain(secret); + expect(JSON.parse(responseBody).observabilityHasSecretKey).toBe(true); + + const stored = fs.readFileSync( + path.join(tmpDir, 'config', 'otel-secret.json'), + 'utf8', + ); + expect(stored).not.toContain(secret); + expect( + fs.statSync(path.join(tmpDir, 'config', 'otel-secret.json')).mode & 0o777, + ).toBe(0o600); + }); + test('stale clients may submit retired automation fields, which are ignored', async () => { asUser('member', ['manage_system_config']); const response = await app.request('/api/config/system', { diff --git a/tests/runtime-system-settings-normalization.test.ts b/tests/runtime-system-settings-normalization.test.ts index a8a12e4..bc04f10 100644 --- a/tests/runtime-system-settings-normalization.test.ts +++ b/tests/runtime-system-settings-normalization.test.ts @@ -10,6 +10,10 @@ const tmpDir = fs.mkdtempSync( process.env.CONTAINER_TIMEOUT = '-5'; process.env.MAX_CONCURRENT_CONTAINERS = '999'; process.env.ADMIN_HOST_ONLY_MODE = 'true'; +process.env.COMMERCEAGENT_OTEL_ENABLED = 'true'; +process.env.COMMERCEAGENT_OTEL_ENDPOINT = 'https://otel-env.example'; +process.env.COMMERCEAGENT_OTEL_PUBLIC_KEY = 'pk-env'; +process.env.COMMERCEAGENT_OTEL_SAMPLE_RATE = '0.25'; vi.mock('../src/config.js', () => ({ ASSISTANT_NAME: 'CommerceAgent', @@ -27,21 +31,56 @@ afterAll(() => { delete process.env.CONTAINER_TIMEOUT; delete process.env.MAX_CONCURRENT_CONTAINERS; delete process.env.ADMIN_HOST_ONLY_MODE; + delete process.env.COMMERCEAGENT_OTEL_ENABLED; + delete process.env.COMMERCEAGENT_OTEL_ENDPOINT; + delete process.env.COMMERCEAGENT_OTEL_PUBLIC_KEY; + delete process.env.COMMERCEAGENT_OTEL_SAMPLE_RATE; fs.rmSync(tmpDir, { recursive: true, force: true }); }); describe('system settings normalization', () => { + test('OTel endpoints require HTTPS except the two local HTTP targets', () => { + expect(runtime.isSafeOtelEndpoint('https://otel.example')).toBe(true); + expect(runtime.isSafeOtelEndpoint('http://localhost:3000')).toBe(true); + expect(runtime.isSafeOtelEndpoint('http://127.0.0.1:3000')).toBe(true); + expect(runtime.isSafeOtelEndpoint('http://127.0.0.2:3000')).toBe(false); + expect(runtime.isSafeOtelEndpoint('http://otel.example')).toBe(false); + expect(runtime.isSafeOtelEndpoint('http://[::1]:3000')).toBe(false); + expect(runtime.isSafeOtelEndpoint('https://user:pass@otel.example')).toBe( + false, + ); + expect( + runtime.isSafeOtelEndpoint('https://otel.example/path?secret=x'), + ).toBe(false); + }); + test('env, API save, and file reads share the same bounds', () => { const fromEnv = runtime.getSystemSettings(); expect(fromEnv.containerTimeout).toBe(60_000); expect(fromEnv.maxConcurrentContainers).toBe(100); expect(fromEnv.adminHostOnlyMode).toBe(true); + expect(fromEnv.observabilityEnabled).toBe(true); + expect(fromEnv.observabilityEndpoint).toBe('https://otel-env.example'); + expect(fromEnv.observabilityPublicKey).toBe('pk-env'); + expect(fromEnv.observabilitySampleRate).toBe(0.25); + expect(fromEnv.observabilityIncludeContent).toBe(false); const saved = runtime.saveSystemSettings({ mainAgentAutoCompactPercentage: 95, } as any); expect(saved.mainAgentAutoCompactPercentage).toBe(90); + const persisted = runtime.saveSystemSettings({ + observabilityEnabled: false, + observabilityEndpoint: 'https://otel-persisted.example', + observabilitySampleRate: 0.75, + }); + expect(persisted.observabilityEnabled).toBe(false); + expect(persisted.observabilityEndpoint).toBe( + 'https://otel-persisted.example', + ); + expect(persisted.observabilitySampleRate).toBe(0.75); + const settingsFile = path.join(tmpDir, 'config', 'system-settings.json'); fs.writeFileSync( settingsFile, diff --git a/tests/system-settings-projection.test.ts b/tests/system-settings-projection.test.ts index 380fb28..4788db8 100644 --- a/tests/system-settings-projection.test.ts +++ b/tests/system-settings-projection.test.ts @@ -24,6 +24,12 @@ describe('system settings API projection', () => { const configRoute = read('src/routes/config.ts'); for (const key of uiKeys) { + if (key === 'observabilityHasSecretKey') { + expect(configRoute).toContain( + 'observabilityHasSecretKey: hasObservabilitySecret(),', + ); + continue; + } expect( configRoute.includes(`${key}: settings.${key},`), `system settings projection is missing "${key}"`, diff --git a/web/src/components/settings/SystemSettingsSection.tsx b/web/src/components/settings/SystemSettingsSection.tsx index 89763d6..cfbcb84 100644 --- a/web/src/components/settings/SystemSettingsSection.tsx +++ b/web/src/components/settings/SystemSettingsSection.tsx @@ -276,6 +276,15 @@ export function SystemSettingsSection({ // fallbackModel 是自由字符串，独立于数值 field 抽象；仅在 runtime scope 展示。 const [fallbackModel, setFallbackModel] = useState(''); const [initialFallbackModel, setInitialFallbackModel] = useState(''); + const [observabilityEnabled, setObservabilityEnabled] = useState(false); + const [observabilityEndpoint, setObservabilityEndpoint] = useState(''); + const [observabilityPublicKey, setObservabilityPublicKey] = useState(''); + const [observabilitySecretKey, setObservabilitySecretKey] = useState(''); + const [observabilitySampleRate, setObservabilitySampleRate] = useState('1'); + const [observabilityIncludeContent, setObservabilityIncludeContent] = + useState(false); + const [observabilityHasSecretKey, setObservabilityHasSecretKey] = + useState(false); const loadSettings = useCallback(async () => { if (!canManage) return; @@ -289,6 +298,13 @@ export function SystemSettingsSection({ setInitialValues(nextValues); setFallbackModel(data.fallbackModel ?? ''); setInitialFallbackModel(data.fallbackModel ?? ''); + setObservabilityEnabled(data.observabilityEnabled ?? false); + setObservabilityEndpoint(data.observabilityEndpoint ?? ''); + setObservabilityPublicKey(data.observabilityPublicKey ?? ''); + setObservabilitySecretKey(''); + setObservabilitySampleRate(String(data.observabilitySampleRate ?? 1)); + setObservabilityIncludeContent(data.observabilityIncludeContent ?? false); + setObservabilityHasSecretKey(data.observabilityHasSecretKey ?? false); setTouched({}); setSubmitted(false); } catch (error) { @@ -324,6 +340,17 @@ export function SystemSettingsSection({ const fallbackModelDirty = scope === 'runtime' && fallbackModel.trim() !== initialFallbackModel.trim(); + const observabilityDirty = + scope === 'runtime' && + (!!observabilityEnabled !== !!settings?.observabilityEnabled || + observabilityEndpoint.trim() !== + (settings?.observabilityEndpoint ?? '') || + observabilityPublicKey.trim() !== + (settings?.observabilityPublicKey ?? '') || + observabilitySampleRate !== + String(settings?.observabilitySampleRate ?? 1) || + observabilityIncludeContent !== !!settings?.observabilityIncludeContent || + observabilitySecretKey.length > 0); const dirty = useMemo( () => @@ -332,21 +359,46 @@ export function SystemSettingsSection({ activeFields.some( (field) => values[field.key] !== initialValues[field.key], )) || + fallbackModelDirty || + observabilityDirty, + [ + activeFields, + initialValues, + values, fallbackModelDirty, - [activeFields, initialValues, values, fallbackModelDirty], + observabilityDirty, + ], ); const handleSave = async () => { if (!settings || !values) return; setSubmitted(true); if (Object.keys(errors).length > 0) return; + const sampleRate = Number(observabilitySampleRate); + if ( + scope === 'runtime' && + (!Number.isFinite(sampleRate) || sampleRate < 0 || sampleRate > 1) + ) { + toast.error('采样率必须是 0 到 1 之间的数字。'); + return; + } - const payload: Partial = {}; + const payload: Partial & { + observabilitySecretKey?: string; + } = {}; for (const field of activeFields) { payload[field.key] = field.toStored(Number(values[field.key])); } if (scope === 'runtime') { payload.fallbackModel = fallbackModel.trim(); + payload.observabilityEnabled = observabilityEnabled; + payload.observabilityEndpoint = observabilityEndpoint.trim(); + payload.observabilityPublicKey = observabilityPublicKey.trim(); + payload.observabilitySampleRate = sampleRate; + payload.observabilityIncludeContent = observabilityIncludeContent; + if (observabilitySecretKey) { + payload.observabilitySecretKey = observabilitySecretKey; + } } setSaving(true); @@ -358,6 +410,13 @@ export function SystemSettingsSection({ setInitialValues(nextValues); setFallbackModel(data.fallbackModel ?? ''); setInitialFallbackModel(data.fallbackModel ?? ''); + setObservabilityEnabled(data.observabilityEnabled ?? false); + setObservabilityEndpoint(data.observabilityEndpoint ?? ''); + setObservabilityPublicKey(data.observabilityPublicKey ?? ''); + setObservabilitySecretKey(''); + setObservabilitySampleRate(String(data.observabilitySampleRate ?? 1)); + setObservabilityIncludeContent(data.observabilityIncludeContent ?? false); + setObservabilityHasSecretKey(data.observabilityHasSecretKey ?? false); setTouched({}); setSubmitted(false); toast.success('系统参数已保存，将应用于后续启动的任务'); @@ -466,6 +525,111 @@ export function SystemSettingsSection({ )} + {scope === 'runtime' && ( +
+
+

+ OpenTelemetry 导出 +

+

+ 可连接 Langfuse + 云版或自托管服务。默认关闭；默认仅发送运行元数据。显式开启正文上报后，用户输入、助手输出和工具数据会先脱敏再导出。 +

+
+
+
+ + +
+
+ + + setObservabilityEndpoint(event.target.value) + } + placeholder="https://cloud.langfuse.com" + autoComplete="url" + /> +

+ 填写服务根地址，导出路径由 CommerceAgent 自动补全；HTTP 仅支持 + localhost 与 127.0.0.1。 +

+
+
+ + + setObservabilityPublicKey(event.target.value) + } + autoComplete="off" + /> +
+
+ + + setObservabilitySecretKey(event.target.value) + } + placeholder={ + observabilityHasSecretKey + ? '已配置；留空保持不变' + : '仅保存时提交' + } + autoComplete="new-password" + /> +

+ 保存后使用 AES-256-GCM 加密并以本机权限保护；API 不会返回密钥。 +

+
+
+ + + setObservabilitySampleRate(event.target.value) + } + /> +
+
+
+ +

+ 每个轮次按采样率独立抽样。API + Key、Endpoint、联系方式、地址、收件信息、Cookie + 和令牌会在本机脱敏；脱敏尽力而为，不保证识别所有个人信息。生产环境处理真实客户数据时建议保持关闭正文上报。 +

+
+ +
+
+
+ )} +

{dirty ? '有尚未保存的系统参数' : '系统参数已保存'} @@ -643,8 +807,8 @@ export function HostIntegrationSettingsSection({ className="mt-1 text-xs leading-5 text-muted-foreground" > 开启后自动继承宿主机提示词、Rules、全部 Skills 与 MCP， - 无需再逐项选择；CommerceAgent 管理的能力继续附加。普通用户的 默认 - CommerceAgent 始终使用托管配置。 + 无需再逐项选择；CommerceAgent 管理的能力继续附加。普通用户的 + 默认 CommerceAgent 始终使用托管配置。