diff --git a/CHANGELOG.md b/CHANGELOG.md index 42f68b8d..0a7fd4d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `@chatbot/*` package namespace and infrastructure/database identifiers are unchanged. ### Added +- Graph-agent playground executions now persist a per-node trace in + `AgentExecution.output.trace` (no schema change) and show graph nodes in the console + trace tab. Persisted and streamed trace data is redacted with irreversible placeholders + (`[CPF]`, `[CNPJ]`, `[CARTAO]`, `[EMAIL]`, `[TELEFONE]`, `[IP]`, `[TOKEN]`) and size-capped. + Only structured identifiers are detected; names, addresses and free text are not. - `SUPPORT.md` describing where to get help - `.github/CODEOWNERS` for automatic review routing - README badges (build status, license) diff --git a/apps/web-ui/app/api/agents/[id]/playground/resume/route.ts b/apps/web-ui/app/api/agents/[id]/playground/resume/route.ts index 33e735c3..ee00a2b3 100644 --- a/apps/web-ui/app/api/agents/[id]/playground/resume/route.ts +++ b/apps/web-ui/app/api/agents/[id]/playground/resume/route.ts @@ -9,6 +9,8 @@ import { PausedExecutionService, LlmProviderService, TenantConfigService, + NodeTraceCollector, + sanitizeExecutionEvent, } from '@chatbot/shared'; import { createLLMProvider, type TenantLLMConfig } from '@chatbot/ai'; import { GraphExecutor, createNodeExecutors } from '@chatbot/agent-studio/server'; @@ -125,6 +127,8 @@ export async function POST( let fullText = ''; let pausedAgain = false; + // Continue the per-node trace persisted before the pause (already redacted). + const traceCollector = NodeTraceCollector.fromOutput(agentExecution.output); const stream = new ReadableStream({ async start(controller) { @@ -140,13 +144,15 @@ export async function POST( metadata, { onEvent: (event) => { - sendEvent(event.type, event); + traceCollector.handleEvent(event); + sendEvent(event.type, sanitizeExecutionEvent(event)); if (event.type === 'text_delta') { fullText += (event as any).delta; } }, onPause: async (pauseInfo) => { pausedAgain = true; + traceCollector.recordPause(); await pausedExecService.create({ tenantId, agentId, @@ -159,7 +165,7 @@ export async function POST( }); await db.agentExecution.update({ where: { id: paused.executionId }, - data: { status: 'paused' }, + data: { status: 'paused', output: { ...traceCollector.snapshot() } }, }); logger.info({ executionId: paused.executionId, resumeToken: pauseInfo.resumeToken }, 'Playground resumed execution paused again'); }, @@ -171,7 +177,7 @@ export async function POST( where: { id: paused.executionId }, data: { status: 'completed', - output: { text: fullText }, + output: { text: fullText, ...traceCollector.snapshot() }, completedAt: new Date(), }, }); @@ -181,7 +187,7 @@ export async function POST( sendEvent('execution_error', { error: errorMessage }); await db.agentExecution.update({ where: { id: paused.executionId }, - data: { status: 'failed', output: { error: errorMessage }, completedAt: new Date() }, + data: { status: 'failed', output: { error: errorMessage, ...traceCollector.snapshot() }, completedAt: new Date() }, }); logger.error({ executionId: paused.executionId, error: errorMessage }, 'Graph resume execution failed'); } finally { diff --git a/apps/web-ui/app/api/agents/[id]/playground/route.ts b/apps/web-ui/app/api/agents/[id]/playground/route.ts index 060612b3..73c776c3 100644 --- a/apps/web-ui/app/api/agents/[id]/playground/route.ts +++ b/apps/web-ui/app/api/agents/[id]/playground/route.ts @@ -9,6 +9,8 @@ import { LlmProviderService, PausedExecutionService, playgroundRequestSchema, + NodeTraceCollector, + sanitizeExecutionEvent, } from '@chatbot/shared'; import { S3Service } from '@chatbot/shared/server'; import { streamChat, createLLMProvider, type TenantLLMConfig, ContentResolver, type MessageAttachment, buildBuiltInTools } from '@chatbot/ai'; @@ -352,6 +354,8 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id: let graphTtftMs: number | undefined; let paused = false; const pausedExecService = new PausedExecutionService(db); + // Per-node trace, redacted and size-capped before it is persisted or streamed. + const traceCollector = new NodeTraceCollector(); const stream = new ReadableStream({ async start(controller) { @@ -379,13 +383,15 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id: if (event.type === 'text_delta' && graphTtftMs === undefined) { graphTtftMs = Date.now() - graphStartTime; } - sendEvent(event.type, event); + traceCollector.handleEvent(event); + sendEvent(event.type, sanitizeExecutionEvent(event)); if (event.type === 'text_delta') { fullText += (event as any).delta; } }, onPause: async (pauseInfo) => { paused = true; + traceCollector.recordPause(); await pausedExecService.create({ tenantId, agentId: id, @@ -398,7 +404,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id: }); await db.agentExecution.update({ where: { id: execution.id }, - data: { status: 'paused' }, + data: { status: 'paused', output: { ...traceCollector.snapshot() } }, }); sendEvent('execution_paused', { resumeToken: pauseInfo.resumeToken, @@ -422,7 +428,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id: where: { id: execution.id }, data: { status: 'completed', - output: { text: fullText }, + output: { text: fullText, ...traceCollector.snapshot() }, completedAt: new Date(), }, }); @@ -440,7 +446,7 @@ export async function POST(req: NextRequest, { params }: { params: Promise<{ id: }); await db.agentExecution.update({ where: { id: execution.id }, - data: { status: 'failed', output: { error: errorMessage }, completedAt: new Date() }, + data: { status: 'failed', output: { error: errorMessage, ...traceCollector.snapshot() }, completedAt: new Date() }, }); logger.error({ executionId: execution.id, error: errorMessage }, 'Graph execution failed'); } finally { diff --git a/apps/web-ui/components/agents/playground/console-trace.tsx b/apps/web-ui/components/agents/playground/console-trace.tsx index e17ce927..4ce46311 100644 --- a/apps/web-ui/components/agents/playground/console-trace.tsx +++ b/apps/web-ui/components/agents/playground/console-trace.tsx @@ -21,9 +21,35 @@ interface TraceItem { output?: unknown; } +/** Subset of a graph `NodeTraceEntry` as streamed (already redacted by the server). */ +interface NodeTracePayload { + nodeId?: string; + nodeType?: string; + nodeLabel?: string; + status?: string; + durationMs?: number; + input?: unknown; + output?: unknown; + error?: string; +} + +function applyNodeTrace(item: TraceItem, trace: NodeTracePayload): void { + item.name = trace.nodeLabel ?? item.name; + item.type = trace.nodeType ?? item.type; + item.status = trace.status === 'failed' ? 'failed' : 'completed'; + item.durationMs = trace.durationMs; + item.input = trace.input; + item.output = + trace.error !== undefined + ? { ...((trace.output as Record | undefined) ?? {}), error: trace.error } + : trace.output; +} + function buildTraceItems(events: ConsoleEvent[]): TraceItem[] { const items: TraceItem[] = []; const toolCalls = new Map(); + // Latest running item per graph node id (a node may run more than once in loops). + const runningNodes = new Map(); for (const event of events) { if (event.type === 'tool_call') { @@ -66,6 +92,36 @@ function buildTraceItems(events: ConsoleEvent[]): TraceItem[] { execItem.status = event.data.error ? 'failed' : 'completed'; execItem.durationMs = event.data.durationMs as number; } + } else if (event.type === 'node_start') { + const nodeId = String(event.data.nodeId ?? 'node'); + const item: TraceItem = { + id: event.id, + type: String(event.data.nodeType ?? 'node'), + name: nodeId, + status: 'running', + }; + runningNodes.set(nodeId, item); + items.push(item); + } else if (event.type === 'node_complete') { + const trace = (event.data.trace ?? {}) as NodeTracePayload; + const nodeId = String(event.data.nodeId ?? trace.nodeId ?? 'node'); + let item = runningNodes.get(nodeId); + if (!item) { + item = { id: event.id, type: trace.nodeType ?? 'node', name: nodeId, status: 'running' }; + items.push(item); + } + applyNodeTrace(item, trace); + runningNodes.delete(nodeId); + } else if (event.type === 'node_error') { + const nodeId = String(event.data.nodeId ?? 'node'); + let item = runningNodes.get(nodeId); + if (!item) { + item = { id: event.id, type: 'node', name: nodeId, status: 'running' }; + items.push(item); + } + item.status = 'failed'; + item.output = { error: String(event.data.error ?? 'Node failed') }; + runningNodes.delete(nodeId); } else if (event.type === 'error') { items.push({ id: event.id, diff --git a/libs/shared/src/index.ts b/libs/shared/src/index.ts index b2e49a33..47bb3722 100644 --- a/libs/shared/src/index.ts +++ b/libs/shared/src/index.ts @@ -7,6 +7,30 @@ export { createLogger } from './logging/logger'; // Utils export { withTimeout, TimeoutError } from './utils/with-timeout'; +// PII redaction (irreversible placeholders, for persisted/displayed observability data) +export { + redactText, + redactDeep, + DEFAULT_PII_RULES, + SAUDE_RULES, + REDACTED_PLACEHOLDER, + TRUNCATED_PLACEHOLDER, + NodeTraceCollector, + sanitizeExecutionEvent, + sanitizeTraceEntry, + truncateValue, +} from './pii'; +export type { + PiiRule, + PiiCategory, + CustomPiiPattern, + RedactOptions, + NodeTraceLike, + PersistedNodeTrace, + PersistedTraceOutput, + TraceEvent, +} from './pii'; + // Database export { getPrismaClient, disconnectPrisma } from './db/prisma-client'; export { getTenantClient, TENANT_SCOPED_MODELS } from './db/tenant-middleware'; diff --git a/libs/shared/src/pii/index.ts b/libs/shared/src/pii/index.ts new file mode 100644 index 00000000..023599ee --- /dev/null +++ b/libs/shared/src/pii/index.ts @@ -0,0 +1,27 @@ +export { + redactText, + redactDeep, + isValidCpf, + isValidCnpj, + isValidCardNumber, + isValidIpv4, + isValidBrPhone, + DEFAULT_PII_RULES, + CREDENCIAIS_RULES, + IDENTIFICADORES_RULES, + FINANCEIRO_RULES, + REDE_E_CONTATO_RULES, + SAUDE_RULES, + REDACTED_PLACEHOLDER, + TRUNCATED_PLACEHOLDER, +} from './redact'; +export type { PiiRule, PiiCategory, CustomPiiPattern, RedactOptions } from './redact'; +export { + NodeTraceCollector, + sanitizeExecutionEvent, + sanitizeTraceEntry, + truncateValue, + MAX_FIELD_CHARS, + MAX_TRACE_ENTRIES, +} from './node-trace'; +export type { NodeTraceLike, PersistedNodeTrace, PersistedTraceOutput, TraceEvent } from './node-trace'; diff --git a/libs/shared/src/pii/node-trace.test.ts b/libs/shared/src/pii/node-trace.test.ts new file mode 100644 index 00000000..312b16d4 --- /dev/null +++ b/libs/shared/src/pii/node-trace.test.ts @@ -0,0 +1,181 @@ +import { describe, it, expect } from 'vitest'; +import { + NodeTraceCollector, + sanitizeExecutionEvent, + sanitizeTraceEntry, + truncateValue, + MAX_FIELD_CHARS, + type NodeTraceLike, +} from './node-trace'; + +const CPF = '529.982.247-25'; + +function entry(overrides: Partial = {}): NodeTraceLike { + return { + nodeId: 'n1', + nodeType: 'tool', + status: 'completed', + startedAt: '2026-10-18T00:00:00.000Z', + ...overrides, + }; +} + +describe('truncateValue', () => { + it('keeps small values and cuts long strings', () => { + expect(truncateValue('abc', 10)).toBe('abc'); + expect(truncateValue('a'.repeat(20), 10)).toBe(`${'a'.repeat(10)}[TRUNCADO]`); + }); + + it('replaces large objects by a marker with preview', () => { + const out = truncateValue({ big: 'x'.repeat(100) }, 50) as { truncated: boolean; originalLength: number; preview: string }; + expect(out.truncated).toBe(true); + expect(out.originalLength).toBeGreaterThan(50); + expect(out.preview.endsWith('[TRUNCADO]')).toBe(true); + }); + + it('passes through primitives', () => { + expect(truncateValue(5)).toBe(5); + expect(truncateValue(null)).toBeNull(); + }); +}); + +describe('sanitizeTraceEntry', () => { + it('redacts input, output and error', () => { + const out = sanitizeTraceEntry( + entry({ + input: { toolName: 'lookup', parameters: { cpf: CPF } }, + output: { result: `email ana@exemplo.com` }, + error: `falhou para ${CPF}`, + }), + ); + expect(JSON.stringify(out)).not.toContain('529.982'); + expect(JSON.stringify(out)).not.toContain('ana@exemplo.com'); + expect(out.input).toEqual({ toolName: 'lookup', parameters: { cpf: '[CPF]' } }); + expect(out.error).toBe('falhou para [CPF]'); + }); + + it('redacts before truncating so a cut never leaves part of an identifier', () => { + const padding = 'x'.repeat(MAX_FIELD_CHARS - 5); + const out = sanitizeTraceEntry(entry({ output: { body: `${padding}${CPF}` } })); + const body = (out.output as { body: string }).body; + expect(body).not.toMatch(/\d{3}\.\d{3}/); + }); + + it('does not mutate the original entry', () => { + const original = entry({ input: { v: CPF } }); + sanitizeTraceEntry(original); + expect(original.input).toEqual({ v: CPF }); + }); +}); + +describe('NodeTraceCollector', () => { + it('collects completed nodes redacted', () => { + const c = new NodeTraceCollector(); + c.handleEvent({ type: 'node_start', nodeId: 'n1', nodeType: 'tool' }); + c.handleEvent({ type: 'node_complete', nodeId: 'n1', trace: entry({ input: { v: CPF } }) }); + const { trace } = c.snapshot(); + expect(trace).toHaveLength(1); + expect(trace[0]?.input).toEqual({ v: '[CPF]' }); + }); + + it('reconstructs a failed node from node_start + node_error', () => { + const c = new NodeTraceCollector(); + c.handleEvent({ type: 'node_start', nodeId: 'n2', nodeType: 'http' }); + c.handleEvent({ type: 'node_error', nodeId: 'n2', error: `bad request for ${CPF}` }); + const { trace } = c.snapshot(); + expect(trace[0]).toMatchObject({ nodeId: 'n2', nodeType: 'http', status: 'failed', error: 'bad request for [CPF]' }); + expect(typeof trace[0]?.durationMs).toBe('number'); + }); + + it('records node_error without a prior node_start as unknown type', () => { + const c = new NodeTraceCollector(); + c.handleEvent({ type: 'node_error', nodeId: 'n9', error: 'no executor' }); + expect(c.snapshot().trace[0]).toMatchObject({ nodeId: 'n9', nodeType: 'unknown', status: 'failed' }); + }); + + it('records the pausing node', () => { + const c = new NodeTraceCollector(); + c.handleEvent({ type: 'node_start', nodeId: 'h1', nodeType: 'human' }); + c.recordPause(); + expect(c.snapshot().trace[0]).toMatchObject({ nodeId: 'h1', nodeType: 'human', status: 'paused' }); + }); + + it('recordPause without a current node is a no-op', () => { + const c = new NodeTraceCollector(); + c.recordPause(); + expect(c.snapshot().trace).toHaveLength(0); + }); + + it('caps the number of entries and reports dropped ones', () => { + const c = new NodeTraceCollector([], 0, 2); + for (let i = 0; i < 5; i++) { + c.handleEvent({ type: 'node_complete', nodeId: `n${i}`, trace: entry({ nodeId: `n${i}` }) }); + } + expect(c.snapshot()).toMatchObject({ traceDropped: 3 }); + expect(c.snapshot().trace).toHaveLength(2); + }); + + it('ignores unrelated events', () => { + const c = new NodeTraceCollector(); + c.handleEvent({ type: 'text_delta', nodeId: 'n1', delta: 'oi' }); + expect(c.snapshot().trace).toHaveLength(0); + }); + + it('continues from a previously persisted output (resume)', () => { + const first = new NodeTraceCollector(); + first.handleEvent({ type: 'node_complete', nodeId: 'a', trace: entry({ nodeId: 'a' }) }); + const persisted = JSON.parse(JSON.stringify({ text: 'x', ...first.snapshot() })); + + const resumed = NodeTraceCollector.fromOutput(persisted); + resumed.handleEvent({ type: 'node_complete', nodeId: 'b', trace: entry({ nodeId: 'b' }) }); + expect(resumed.snapshot().trace.map((t) => t.nodeId)).toEqual(['a', 'b']); + }); + + it('fromOutput tolerates null and malformed values', () => { + expect(NodeTraceCollector.fromOutput(null).snapshot().trace).toEqual([]); + expect(NodeTraceCollector.fromOutput({ trace: 'nope' }).snapshot().trace).toEqual([]); + expect(NodeTraceCollector.fromOutput({ error: 'x' }).snapshot().trace).toEqual([]); + }); +}); + +describe('sanitizeExecutionEvent', () => { + it('redacts node_complete, node_error and state_update', () => { + const complete = sanitizeExecutionEvent({ + type: 'node_complete', + nodeId: 'n1', + trace: entry({ output: { r: CPF } }), + }) as { trace: NodeTraceLike }; + expect(complete.trace.output).toEqual({ r: '[CPF]' }); + + const err = sanitizeExecutionEvent({ type: 'node_error', nodeId: 'n1', error: CPF }) as { error: string }; + expect(err.error).toBe('[CPF]'); + + const state = sanitizeExecutionEvent({ type: 'state_update', channels: { who: CPF } }) as { + channels: Record; + }; + expect(state.channels).toEqual({ who: '[CPF]' }); + }); + + it('redacts execution_complete trace and final state, keeping other fields', () => { + const out = sanitizeExecutionEvent({ + type: 'execution_complete', + trace: [entry({ input: { v: CPF } })], + finalState: { + channels: { who: CPF }, + messages: [{ role: 'user', content: CPF }], + currentNodeId: null, + metadata: { executionId: 'e1' }, + }, + }) as { trace: NodeTraceLike[]; finalState: { channels: unknown; messages: unknown; currentNodeId: unknown; metadata: unknown } }; + expect(JSON.stringify(out)).not.toContain('529.982'); + expect(out.finalState.currentNodeId).toBeNull(); + expect(out.finalState.metadata).toEqual({ executionId: 'e1' }); + }); + + it('leaves text_delta and other events unchanged', () => { + const delta = { type: 'text_delta', nodeId: 'n1', delta: `seu CPF ${CPF}` }; + expect(sanitizeExecutionEvent(delta)).toBe(delta); + const start = { type: 'node_start', nodeId: 'n1', nodeType: 'llm' }; + expect(sanitizeExecutionEvent(start)).toBe(start); + }); +}); diff --git a/libs/shared/src/pii/node-trace.ts b/libs/shared/src/pii/node-trace.ts new file mode 100644 index 00000000..c6a473a8 --- /dev/null +++ b/libs/shared/src/pii/node-trace.ts @@ -0,0 +1,237 @@ +/** + * PII-safe handling of per-node execution traces of graph agents. + * + * - Every persisted/streamed trace entry is redacted (see ./redact) and then + * truncated, in that order, so a cut can never leave half of an identifier. + * - Types are structural on purpose: this module must not depend on + * `@chatbot/agent-studio` (which depends on `@chatbot/shared`). + */ + +import { redactDeep, redactText, TRUNCATED_PLACEHOLDER } from './redact'; + +export interface NodeTraceLike { + nodeId: string; + nodeType: string; + nodeLabel?: string; + status: 'completed' | 'failed' | 'skipped' | 'paused'; + startedAt: string; + completedAt?: string; + input?: Record; + output?: Record; + error?: string; + durationMs?: number; +} + +export type JsonValue = string | number | boolean | null | JsonValue[] | { [key: string]: JsonValue }; + +/** + * A redacted, size-capped trace entry. Declared as a JSON-safe type alias (not an + * interface) so it can be stored in a Prisma `Json` column without casts. + */ +export type PersistedNodeTrace = { + nodeId: string; + nodeType: string; + nodeLabel?: string; + status: NodeTraceLike['status']; + startedAt: string; + completedAt?: string; + input?: { [key: string]: JsonValue }; + output?: { [key: string]: JsonValue }; + error?: string; + durationMs?: number; +}; + +export type PersistedTraceOutput = { + trace: PersistedNodeTrace[]; + /** Number of entries dropped after MAX_TRACE_ENTRIES was reached. */ + traceDropped?: number; +}; + +/** Event shape accepted by the collector/sanitizer (structurally compatible with agent-studio's ExecutionEvent). */ +export type TraceEvent = { type: string } & Record; + +export const MAX_FIELD_CHARS = 4000; +export const MAX_TRACE_ENTRIES = 200; + +/** + * Caps the size of a (already redacted) value. Strings are cut; objects/arrays + * larger than `maxChars` once serialised are replaced by a marker object with + * a preview. + */ +export function truncateValue(value: unknown, maxChars: number = MAX_FIELD_CHARS): unknown { + if (typeof value === 'string') { + return value.length > maxChars ? `${value.slice(0, maxChars)}${TRUNCATED_PLACEHOLDER}` : value; + } + if (value === null || typeof value !== 'object') return value; + let json: string; + try { + json = JSON.stringify(value); + } catch { + return { truncated: true, preview: TRUNCATED_PLACEHOLDER }; + } + if (json.length <= maxChars) return value; + return { truncated: true, originalLength: json.length, preview: `${json.slice(0, maxChars)}${TRUNCATED_PLACEHOLDER}` }; +} + +function redactAndTruncate(value: unknown): unknown { + return truncateValue(redactDeep(value)); +} + +function redactAndTruncateRecord(record: Record | undefined): Record | undefined { + if (!record) return record; + const redacted = redactDeep(record); + const out: Record = {}; + for (const [key, value] of Object.entries(redacted)) out[key] = truncateValue(value); + return out; +} + +/** Returns a redacted, size-capped copy of a trace entry. */ +export function sanitizeTraceEntry(entry: NodeTraceLike): PersistedNodeTrace { + const out: PersistedNodeTrace = { + nodeId: entry.nodeId, + nodeType: entry.nodeType, + status: entry.status, + startedAt: entry.startedAt, + }; + if (entry.nodeLabel !== undefined) out.nodeLabel = redactText(entry.nodeLabel); + if (entry.completedAt !== undefined) out.completedAt = entry.completedAt; + if (entry.durationMs !== undefined) out.durationMs = entry.durationMs; + // redactDeep/truncateValue only transform strings/containers of already-serialisable + // executor data, so the result is JSON-safe. + if (entry.input !== undefined) out.input = redactAndTruncateRecord(entry.input) as { [key: string]: JsonValue }; + if (entry.output !== undefined) out.output = redactAndTruncateRecord(entry.output) as { [key: string]: JsonValue }; + if (entry.error !== undefined) out.error = redactAndTruncate(entry.error) as string; + return out; +} + +/** + * Accumulates redacted per-node trace entries from executor events. + * + * The graph executor emits `node_complete` only for successful nodes, so failed + * and paused nodes are reconstructed from `node_start` + `node_error` / + * `recordPause()`. + */ +export class NodeTraceCollector { + private entries: PersistedNodeTrace[]; + private dropped: number; + private current?: { nodeId: string; nodeType: string; startedAtMs: number }; + + constructor( + initial: PersistedNodeTrace[] = [], + initialDropped = 0, + private readonly maxEntries: number = MAX_TRACE_ENTRIES, + ) { + this.entries = initial.slice(0, maxEntries); + this.dropped = initialDropped + Math.max(0, initial.length - maxEntries); + } + + /** Rebuilds a collector from a previously persisted `AgentExecution.output` (e.g. on resume). */ + static fromOutput(output: unknown, maxEntries: number = MAX_TRACE_ENTRIES): NodeTraceCollector { + if (output && typeof output === 'object') { + const record = output as { trace?: unknown; traceDropped?: unknown }; + const trace = Array.isArray(record.trace) ? (record.trace as PersistedNodeTrace[]) : []; + const dropped = typeof record.traceDropped === 'number' ? record.traceDropped : 0; + return new NodeTraceCollector(trace, dropped, maxEntries); + } + return new NodeTraceCollector([], 0, maxEntries); + } + + handleEvent(event: TraceEvent): void { + switch (event.type) { + case 'node_start': + this.current = { + nodeId: String(event.nodeId), + nodeType: String(event.nodeType), + startedAtMs: Date.now(), + }; + break; + case 'node_complete': + if (event.trace && typeof event.trace === 'object') { + this.add(sanitizeTraceEntry(event.trace as NodeTraceLike)); + } + this.current = undefined; + break; + case 'node_error': { + const nowMs = Date.now(); + const known = this.current && this.current.nodeId === String(event.nodeId) ? this.current : undefined; + this.add({ + nodeId: String(event.nodeId), + nodeType: known?.nodeType ?? 'unknown', + status: 'failed', + startedAt: new Date(known?.startedAtMs ?? nowMs).toISOString(), + completedAt: new Date(nowMs).toISOString(), + durationMs: known ? nowMs - known.startedAtMs : undefined, + error: redactAndTruncate(String(event.error ?? '')) as string, + }); + this.current = undefined; + break; + } + default: + break; + } + } + + /** Records the node that is pausing the execution (e.g. a human node). */ + recordPause(): void { + if (!this.current) return; + const nowMs = Date.now(); + this.add({ + nodeId: this.current.nodeId, + nodeType: this.current.nodeType, + status: 'paused', + startedAt: new Date(this.current.startedAtMs).toISOString(), + completedAt: new Date(nowMs).toISOString(), + durationMs: nowMs - this.current.startedAtMs, + }); + this.current = undefined; + } + + snapshot(): PersistedTraceOutput { + return this.dropped > 0 + ? { trace: [...this.entries], traceDropped: this.dropped } + : { trace: [...this.entries] }; + } + + private add(entry: PersistedNodeTrace): void { + if (this.entries.length >= this.maxEntries) { + this.dropped++; + return; + } + this.entries.push(entry); + } +} + +/** + * Returns the version of an executor event that is safe to stream to the + * browser console. `text_delta` (the answer shown in the chat) and events + * without payload are returned unchanged. + */ +export function sanitizeExecutionEvent(event: TraceEvent): TraceEvent { + switch (event.type) { + case 'node_complete': + return event.trace && typeof event.trace === 'object' + ? { ...event, trace: sanitizeTraceEntry(event.trace as NodeTraceLike) } + : event; + case 'node_error': + return { ...event, error: redactAndTruncate(String(event.error ?? '')) }; + case 'state_update': + return { ...event, channels: redactAndTruncateRecord(event.channels as Record | undefined) }; + case 'execution_complete': { + const trace = Array.isArray(event.trace) ? (event.trace as NodeTraceLike[]).map(sanitizeTraceEntry) : event.trace; + const finalState = event.finalState as Record | undefined; + return { + ...event, + trace, + finalState: finalState + ? { + ...finalState, + channels: redactAndTruncateRecord(finalState.channels as Record | undefined), + messages: redactAndTruncate(finalState.messages), + } + : finalState, + }; + } + default: + return event; + } +} diff --git a/libs/shared/src/pii/redact.test.ts b/libs/shared/src/pii/redact.test.ts new file mode 100644 index 00000000..43293c42 --- /dev/null +++ b/libs/shared/src/pii/redact.test.ts @@ -0,0 +1,145 @@ +import { describe, it, expect } from 'vitest'; +import { + redactText, + redactDeep, + isValidCpf, + isValidCnpj, + isValidCardNumber, + isValidIpv4, + isValidBrPhone, + SAUDE_RULES, + DEFAULT_PII_RULES, +} from './redact'; + +describe('validators', () => { + it('validates CPF check digits and rejects repeated sequences', () => { + expect(isValidCpf('529.982.247-25')).toBe(true); + expect(isValidCpf('52998224725')).toBe(true); + expect(isValidCpf('529.982.247-26')).toBe(false); + expect(isValidCpf('111.111.111-11')).toBe(false); + }); + + it('validates CNPJ check digits and rejects repeated sequences', () => { + expect(isValidCnpj('11.222.333/0001-81')).toBe(true); + expect(isValidCnpj('11222333000181')).toBe(true); + expect(isValidCnpj('11.222.333/0001-82')).toBe(false); + expect(isValidCnpj('00000000000000')).toBe(false); + }); + + it('validates card numbers with Luhn', () => { + expect(isValidCardNumber('4111 1111 1111 1111')).toBe(true); + expect(isValidCardNumber('5555-5555-5555-4444')).toBe(true); + expect(isValidCardNumber('4111 1111 1111 1112')).toBe(false); + expect(isValidCardNumber('1729200000000')).toBe(false); + }); + + it('validates IPv4 octet range', () => { + expect(isValidIpv4('192.168.0.1')).toBe(true); + expect(isValidIpv4('256.1.1.1')).toBe(false); + }); + + it('validates BR phones by DDD and shape', () => { + expect(isValidBrPhone('(11) 98765-4321')).toBe(true); + expect(isValidBrPhone('+55 11 98765-4321')).toBe(true); + expect(isValidBrPhone('11987654321')).toBe(true); + expect(isValidBrPhone('(11) 3333-4444')).toBe(true); + expect(isValidBrPhone('(10) 98765-4321')).toBe(false); + expect(isValidBrPhone('1133334444')).toBe(false); + expect(isValidBrPhone('1729200000')).toBe(false); + }); +}); + +describe('redactText', () => { + it('replaces CPF and discards the original', () => { + const out = redactText('Meu CPF é 529.982.247-25, obrigado'); + expect(out).toBe('Meu CPF é [CPF], obrigado'); + expect(out).not.toContain('529'); + }); + + it('replaces unformatted CPF and CNPJ', () => { + expect(redactText('cpf 52998224725')).toBe('cpf [CPF]'); + expect(redactText('cnpj 11.222.333/0001-81')).toBe('cnpj [CNPJ]'); + expect(redactText('cnpj 11222333000181')).toBe('cnpj [CNPJ]'); + }); + + it('keeps digit runs that fail validation', () => { + expect(redactText('pedido 529.982.247-26')).toBe('pedido 529.982.247-26'); + expect(redactText('timestamp 1729200000000')).toBe('timestamp 1729200000000'); + }); + + it('replaces e-mail, card, IP, phone and tokens', () => { + expect(redactText('mail ana.souza+x@exemplo.com.br')).toBe('mail [EMAIL]'); + expect(redactText('cartao 4111 1111 1111 1111')).toBe('cartao [CARTAO]'); + expect(redactText('ip 10.0.0.255.')).toBe('ip [IP].'); + expect(redactText('tel (11) 98765-4321')).toBe('tel [TELEFONE]'); + expect(redactText('key sk-abcdefghijklmnop1234')).toBe('key [TOKEN]'); + expect(redactText('id AKIAABCDEFGHIJKLMNOP')).toBe('id [TOKEN]'); + expect(redactText(`pat ghp_${'a'.repeat(36)}`)).toBe('pat [TOKEN]'); + expect(redactText('Authorization: Bearer abcdefghijklmnopqrstuv')).toBe('Authorization: Bearer [TOKEN]'); + }); + + it('does not touch invalid IPs', () => { + expect(redactText('v 300.1.1.1')).toBe('v 300.1.1.1'); + }); + + it('applies CPF/CNPJ/card before phone (no partial phone matches inside them)', () => { + expect(redactText('11987654321 and 52998224725')).toBe('[TELEFONE] and [CPF]'); + expect(redactText('4111111111111111')).toBe('[CARTAO]'); + }); + + it('is idempotent', () => { + const once = redactText('cpf 529.982.247-25 mail a@b.com ip 1.2.3.4'); + expect(redactText(once)).toBe(once); + }); + + it('returns empty input untouched', () => { + expect(redactText('')).toBe(''); + }); + + it('applies custom patterns as [REDACTED] by default', () => { + const out = redactText('processo PRC-123456', { extraPatterns: [{ pattern: /PRC-\d+/ }] }); + expect(out).toBe('processo [REDACTED]'); + const custom = redactText('processo PRC-1', { extraPatterns: [{ pattern: /PRC-\d+/g, placeholder: '[PROC]' }] }); + expect(custom).toBe('processo [PROC]'); + }); +}); + +describe('redactDeep', () => { + it('redacts strings in nested objects and arrays without mutating the input', () => { + const input = { + a: 'cpf 529.982.247-25', + n: 42, + ok: true, + nested: { list: ['x@y.com', { deep: 'ip 8.8.8.8' }], none: null }, + }; + const snapshot = JSON.stringify(input); + const out = redactDeep(input); + expect(out).toEqual({ + a: 'cpf [CPF]', + n: 42, + ok: true, + nested: { list: ['[EMAIL]', { deep: 'ip [IP]' }], none: null }, + }); + expect(JSON.stringify(input)).toBe(snapshot); + }); + + it('handles circular references without throwing', () => { + const a: Record = { v: 'a@b.com' }; + a.self = a; + const out = redactDeep(a) as Record; + expect(out.v).toBe('[EMAIL]'); + expect(out.self).toBe('[TRUNCADO]'); + }); + + it('leaves non-plain objects such as Date untouched', () => { + const d = new Date(0); + expect((redactDeep({ d }) as { d: Date }).d).toBe(d); + }); +}); + +describe('rule set', () => { + it('reserves an empty health category', () => { + expect(SAUDE_RULES).toHaveLength(0); + expect(DEFAULT_PII_RULES.some((r) => r.category === 'saude')).toBe(false); + }); +}); diff --git a/libs/shared/src/pii/redact.ts b/libs/shared/src/pii/redact.ts new file mode 100644 index 00000000..4846c3f4 --- /dev/null +++ b/libs/shared/src/pii/redact.ts @@ -0,0 +1,267 @@ +/** + * Irreversible PII redaction for data that is persisted or displayed for + * observability purposes (e.g. per-node execution traces). + * + * Design rules: + * - A match is replaced by a short placeholder (`[CPF]`, `[EMAIL]`, ...). The + * original value is discarded: there is no vault and no mapping back. + * - Only structured identifiers are detected. Names, addresses, free text and + * health information are NOT detected (see SAUDE_RULES, reserved for later). + * - Where a cheap validation exists (check digits, Luhn, octet range, DDD) it is + * applied to reduce false positives. + * - Redaction applies to the stored/displayed copy only; callers must keep + * passing the original data to the agent. + */ + +export type PiiCategory = 'identificadores' | 'financeiro' | 'credenciais' | 'saude'; + +export interface PiiRule { + id: string; + category: PiiCategory; + placeholder: string; + /** Must carry the global flag. */ + pattern: RegExp; + /** Optional extra check on the matched text; return false to keep the text. */ + validate?: (match: string) => boolean; +} + +export interface CustomPiiPattern { + pattern: RegExp; + /** Defaults to `[REDACTED]`. */ + placeholder?: string; +} + +export interface RedactOptions { + /** Extra patterns (e.g. per-tenant). Always replaced by their placeholder. */ + extraPatterns?: CustomPiiPattern[]; +} + +export const REDACTED_PLACEHOLDER = '[REDACTED]'; +export const TRUNCATED_PLACEHOLDER = '[TRUNCADO]'; + +const MAX_DEPTH = 32; + +// --------------------------------------------------------------------------- +// Validators +// --------------------------------------------------------------------------- + +function onlyDigits(value: string): string { + return value.replace(/\D/g, ''); +} + +function allSameDigit(digits: string): boolean { + return /^(\d)\1+$/.test(digits); +} + +export function isValidCpf(value: string): boolean { + const digits = onlyDigits(value); + if (digits.length !== 11 || allSameDigit(digits)) return false; + const d = digits.split('').map(Number); + const check = (len: number): number => { + let sum = 0; + for (let i = 0; i < len; i++) sum += (d[i] as number) * (len + 1 - i); + return ((sum * 10) % 11) % 10; + }; + return check(9) === d[9] && check(10) === d[10]; +} + +export function isValidCnpj(value: string): boolean { + const digits = onlyDigits(value); + if (digits.length !== 14 || allSameDigit(digits)) return false; + const d = digits.split('').map(Number); + const check = (len: number): number => { + const weights = len === 12 ? [5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2] : [6, 5, 4, 3, 2, 9, 8, 7, 6, 5, 4, 3, 2]; + let sum = 0; + for (let i = 0; i < len; i++) sum += (d[i] as number) * (weights[i] as number); + const rest = sum % 11; + return rest < 2 ? 0 : 11 - rest; + }; + return check(12) === d[12] && check(13) === d[13]; +} + +export function isValidCardNumber(value: string): boolean { + const digits = onlyDigits(value); + if (digits.length < 13 || digits.length > 19) return false; + // Major card networks start with 2-6; this also avoids epoch-millisecond strings. + if (!/^[2-6]/.test(digits) || allSameDigit(digits)) return false; + let sum = 0; + let double = false; + for (let i = digits.length - 1; i >= 0; i--) { + let n = Number(digits[i]); + if (double) { + n *= 2; + if (n > 9) n -= 9; + } + sum += n; + double = !double; + } + return sum % 10 === 0; +} + +export function isValidIpv4(value: string): boolean { + const parts = value.split('.'); + return parts.length === 4 && parts.every((p) => /^\d{1,3}$/.test(p) && Number(p) <= 255); +} + +const BR_DDD = new Set([ + 11, 12, 13, 14, 15, 16, 17, 18, 19, 21, 22, 24, 27, 28, 31, 32, 33, 34, 35, 37, 38, 41, 42, 43, 44, 45, 46, 47, 48, + 49, 51, 53, 54, 55, 61, 62, 63, 64, 65, 66, 67, 68, 69, 71, 73, 74, 75, 77, 79, 81, 82, 83, 84, 85, 86, 87, 88, 89, + 91, 92, 93, 94, 95, 96, 97, 98, 99, +]); + +export function isValidBrPhone(value: string): boolean { + let digits = onlyDigits(value); + const hasSeparators = /\D/.test(value); + if ((digits.length === 12 || digits.length === 13) && digits.startsWith('55')) digits = digits.slice(2); + if (digits.length !== 10 && digits.length !== 11) return false; + if (!BR_DDD.has(Number(digits.slice(0, 2)))) return false; + if (digits.length === 11) return digits[2] === '9'; + // 10 digits (landline): require visible formatting so bare digit runs + // (e.g. epoch seconds) are not mistaken for phone numbers. + return hasSeparators && /^[2-5]/.test(digits.slice(2)); +} + +// --------------------------------------------------------------------------- +// Rules, organised by category. Order matters: more specific / longer digit +// patterns come before looser ones (CNPJ, CPF and card before phone). +// --------------------------------------------------------------------------- + +export const CREDENCIAIS_RULES: PiiRule[] = [ + { + id: 'token', + category: 'credenciais', + placeholder: '[TOKEN]', + pattern: + /\b(?:sk-[A-Za-z0-9_-]{16,}|AKIA[0-9A-Z]{16}|gh[pousr]_[A-Za-z0-9]{36,}|xox[baprs]-[A-Za-z0-9-]{10,}|eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,})/g, + }, + { + id: 'bearer-token', + category: 'credenciais', + placeholder: '[TOKEN]', + pattern: /(?<=\bBearer\s+)[A-Za-z0-9._~+/=-]{16,}/g, + }, +]; + +export const IDENTIFICADORES_RULES: PiiRule[] = [ + { + id: 'email', + category: 'identificadores', + placeholder: '[EMAIL]', + pattern: /[A-Za-z0-9._%+-]+@[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)*\.[A-Za-z]{2,}/g, + }, + { + id: 'cnpj', + category: 'identificadores', + placeholder: '[CNPJ]', + pattern: /(? (rule.validate && !rule.validate(match) ? match : rule.placeholder)); +} + +/** Replaces PII found in a string by placeholders. Original values are discarded. */ +export function redactText(input: string, options: RedactOptions = {}): string { + if (!input) return input; + let out = input; + for (const rule of DEFAULT_PII_RULES) out = applyRule(out, rule); + for (const custom of options.extraPatterns ?? []) { + out = out.replace(toGlobal(custom.pattern), custom.placeholder ?? REDACTED_PLACEHOLDER); + } + return out; +} + +/** + * Returns a copy of `value` with every string redacted. Arrays and plain + * objects are walked recursively; other values (numbers, booleans, Dates, ...) + * are returned as-is. Circular references and excessive depth are replaced + * by a marker instead of throwing. + */ +export function redactDeep(value: T, options: RedactOptions = {}): T { + const ancestors = new WeakSet(); + + const walk = (node: unknown, depth: number): unknown => { + if (typeof node === 'string') return redactText(node, options); + if (node === null || typeof node !== 'object') return node; + if (depth >= MAX_DEPTH || ancestors.has(node)) return TRUNCATED_PLACEHOLDER; + + if (Array.isArray(node)) { + ancestors.add(node); + const result = node.map((item) => walk(item, depth + 1)); + ancestors.delete(node); + return result; + } + + const proto = Object.getPrototypeOf(node); + if (proto !== Object.prototype && proto !== null) return node; + + ancestors.add(node); + const result: Record = {}; + for (const [key, item] of Object.entries(node)) result[key] = walk(item, depth + 1); + ancestors.delete(node); + return result; + }; + + return walk(value, 0) as T; +}