返回 CodeWhale
ingest.ts
根目录 / pet / src / core / ingest.ts
1 import { fromCodewhaleRuntime, fromCodewhaleSession, isCodewhaleRuntimeDocument, isCodewhaleSession } from './codewhale.js';
2 import { validateBundle, evidenceToTrace, validateObservation, bundleFromTrace } from './evidence.js';
3 import { CATEGORIES, errorOnsetOf, type Category, type PrivacyMode, type Status, type Trace, type WhaleEvent } from './model.js';
4
5 export interface ImportOptions {
6 privacy?: PrivacyMode; maxEvents?: number; maxBytes?: number; maxTraces?: number;
7 /** Optional local-only redaction hook. Never invoked by a network service. */
8 redactor?: (value: unknown, path: string) => unknown;
9 }
10 type Obj = Record<string, any>;
11 const obj = (v: unknown): Obj => v !== null && typeof v === 'object' && !Array.isArray(v) ? v as Obj : {};
12 const list = (v: unknown): any[] => Array.isArray(v) ? v : [];
13 const str = (v: unknown): string | undefined => typeof v === 'string' && v.length ? v : undefined;
14 const SECRET_KEY = /(?:^|[._-])(?:api[._-]?key|authorization|password|passwd|secret|access[._-]?token|refresh[._-]?token|cookie|private[._-]?key)(?:$|[._-])/i;
15 const SAFE_META = /^(?:gen_ai\.(?:usage\.(?:input_tokens(?:\.cached)?|output_tokens|cache_read\.input_tokens|cost)|request\.(?:model|max_tokens)|response\.model|provider\.name|operation\.name)|whalesong\.(?:category|agent_id)|agent\.(?:id|parent_id|state)|(?:http|rpc)\.(?:response\.status_code|method)|retry\.(?:count|attempt)|context\.(?:tokens|limit)|phase|spawnedAgentId|parentAgentId|targetType|iteration|benchmark|blocked)$/;
16
17 /** Best effort, not a guarantee of de-identification. Preserves structure, not secrets. */
18 export function redact(value: unknown, path = '', hook?: ImportOptions['redactor'], depth = 0, seen = new WeakSet<object>()): unknown {
19 if (depth > 48) return '[REDACTED: nesting limit]';
20 const key = path.split('/').at(-1) ?? '';
21 if (SECRET_KEY.test(key) || /^(apiKey|accessToken|refreshToken|privateKey)$/i.test(key)) return '[REDACTED]';
22 // OTLP attributes encode a sensitive name in `key`, not in the JSON path.
23 // Redact the AnyValue itself while preserving a valid OTLP value envelope.
24 const record = obj(value);
25 if (typeof record.key === 'string' && Object.hasOwn(record, 'value') &&
26 (SECRET_KEY.test(record.key) || /^(apiKey|accessToken|refreshToken|privateKey)$/i.test(record.key))) {
27 const safe = { ...record, value: { stringValue: '[REDACTED]' } };
28 // Only the known KeyValue fields are kept; arbitrary siblings cannot bypass redaction.
29 const cleaned = { key: record.key, value: safe.value };
30 return hook ? hook(cleaned, path) : cleaned;
31 }
32 if (value && typeof value === 'object') { if (seen.has(value)) return '[REDACTED: cycle]'; seen.add(value); }
33 let out: unknown = value;
34 if (typeof value === 'string') {
35 out = value
36 .replace(/-----BEGIN [^-]*PRIVATE KEY-----[\s\S]*?-----END [^-]*PRIVATE KEY-----/g, '[REDACTED: private key]')
37 .replace(/\b(?:sk-[A-Za-z0-9_-]{16,}|AKIA[0-9A-Z]{16})\b/g, '[REDACTED: key]')
38 .replace(/\b(?:gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|xox[baprs]-[A-Za-z0-9-]{16,})\b/g, '[REDACTED: token]')
39 .replace(/(https?:\/\/)[^\s/@:]+:[^\s/@]+@/gi, '$1[REDACTED]@')
40 .replace(/\bBearer\s+[A-Za-z0-9._~+\/-]+=*/gi, 'Bearer [REDACTED]')
41 .replace(/\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b/g, '[REDACTED: JWT]')
42 .replace(/((?:api[_-]?key|password|secret)\s*[=:]\s*)[^\s,;"'}]+/gi, '$1[REDACTED]')
43 .replace(/\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi, '[REDACTED: email]');
44 } else if (Array.isArray(value)) {
45 out = value.map((v, i) => redact(v, `${path}/${i}`, hook, depth + 1, seen));
46 } else if (value && typeof value === 'object') {
47 out = Object.fromEntries(Object.entries(value).map(([k, v]) => [k, redact(v, `${path}/${k}`, hook, depth + 1, seen)]));
48 }
49 if (value && typeof value === 'object') seen.delete(value);
50 return hook ? hook(out, path) : out;
51 }
52 export function privacyEvent(e: WhaleEvent, mode: PrivacyMode): WhaleEvent {
53 if (mode === 'metadata') {
54 const { payload: _payload, raw: _raw, ...rest } = e;
55 return { ...rest, links: e.links?.map(link=>({traceId:link.traceId,spanId:link.spanId})), attributes: Object.fromEntries(Object.entries(e.attributes).filter(([k, v]) =>
56 SAFE_META.test(k) && (typeof v !== 'object' || v === null)
57 || ['whalesong.container', 'codewhale.container', 'whalesong.waiting'].includes(k) && typeof v === 'boolean'
58 // Relative timestamps can be negative before the importer rebases them.
59 || k === 'whalesong.error_onset_ms' && typeof v === 'number' && Number.isFinite(v))) };
60 }
61 return e;
62 }
63 function number(v: unknown, field: string, optional = true): number | undefined {
64 if (v === undefined || v === null) {
65 if (optional) return undefined;
66 throw new Error(`Missing ${field}. Times must be numeric milliseconds.`);
67 }
68 if (typeof v !== 'number' || !Number.isFinite(v) || Math.abs(v) > Number.MAX_SAFE_INTEGER) throw new Error(`Invalid ${field}: expected a finite, safely representable number.`);
69 return v;
70 }
71 function nonnegative(v: unknown, field: string): number | undefined {
72 const n = number(v, field); if (n !== undefined && n < 0) throw new Error(`${field} must be nonnegative.`); return n;
73 }
74 function numericAttr(v: unknown): number | undefined {
75 if (v === undefined || v === null || v === '') return undefined;
76 const n = Number(v); return Number.isFinite(n) && n >= 0 ? n : undefined;
77 }
78 export function categoryFor(name: string, a: Obj): Category {
79 const explicit = a['whalesong.category'] ?? a.category;
80 if (CATEGORIES.includes(explicit)) return explicit;
81 const n = name.toLowerCase(), op = String(a['gen_ai.operation.name'] ?? '').toLowerCase();
82 if (/exception|^error\b/.test(n)) return 'error';
83 if (/spawn|fork|subagent/.test(n) || op === 'invoke_agent') return 'agent';
84 if (/message\.send|handoff|agent\.message/.test(n)) return 'communication';
85 if (/retrieve|retrieval|context|embedding|vector|memory|rag/.test(n)) return 'memory';
86 if (/browser|navigate|screenshot|click|playwright/.test(n)) return 'browser';
87 if (/read_file|write_file|list_dir|file\.|filesystem|fs\.|patch/.test(n)) return 'filesystem';
88 if (/exec|shell|run_test|cargo|pytest|compile/.test(n)) return 'code';
89 if (a['gen_ai.request.model'] || a['llm.model_name'] || /reason|completion|generate|chat|llm/.test(n) || ['chat', 'generate_content', 'text_completion'].includes(op)) return 'reasoning';
90 if (a['gen_ai.tool.name'] || a['tool.name'] || /tool|search|function/.test(n)) return 'tool';
91 if (a['http.request.method'] || a['http.method'] || a['rpc.system'] || /http|request|api|fetch|network/.test(n)) return 'network';
92 if (/user|human|approval/.test(n)) return 'human';
93 if (/orchestrat|workflow|phase|join|session|root/.test(n)) return 'orchestration';
94 return 'other';
95 }
96 export function decodeAnyValue(v: unknown): unknown {
97 const a = obj(v);
98 if ('stringValue' in a) return a.stringValue;
99 if ('boolValue' in a) return a.boolValue;
100 if ('doubleValue' in a) return a.doubleValue;
101 if ('intValue' in a) {
102 const n = Number(a.intValue);
103 return Number.isSafeInteger(n) ? n : String(a.intValue);
104 }
105 if ('bytesValue' in a) return a.bytesValue;
106 if ('arrayValue' in a) return list(obj(a.arrayValue).values).map(decodeAnyValue);
107 if ('kvlistValue' in a) return attributes(obj(a.kvlistValue).values);
108 return v;
109 }
110 export function attributes(value: unknown): Obj {
111 if (!Array.isArray(value)) return obj(value);
112 return Object.fromEntries(value.filter(x => typeof x?.key === 'string').map(x => [x.key, decodeAnyValue(x.value)]));
113 }
114 function ns(v: unknown, field: string): bigint {
115 if (typeof v === 'number' && !Number.isSafeInteger(v)) throw new Error(`${field} lost precision: encode OTLP nanoseconds as a decimal string.`);
116 const s = String(v ?? '');
117 if (!/^\d{1,20}$/.test(s) || BigInt(s) > 18446744073709551615n) throw new Error(`Invalid ${field}: expected an OTLP nanosecond integer string.`);
118 return BigInt(s);
119 }
120 function otelStatus(v: unknown): Status {
121 const c = obj(v).code;
122 return c === 2 || c === 'STATUS_CODE_ERROR' ? 'error' : c === 1 || c === 'STATUS_CODE_OK' ? 'success' : 'unknown';
123 }
124 function normalizedEvent(v: unknown, index: number): WhaleEvent {
125 const a = obj(v), id = str(a.id), traceId = str(a.traceId), name = str(a.name);
126 if (!id || !traceId || !name) throw new Error(`Record ${index + 1} requires nonempty id, traceId, and name.`);
127 if (a.schemaVersion !== undefined && a.schemaVersion !== 1) throw new Error(`Record ${index + 1}: unsupported schemaVersion ${a.schemaVersion}.`);
128 const startTime = number(a.startTime, 'startTime', false)!;
129 const endTime = number(a.endTime, 'endTime') ?? startTime;
130 if (endTime < startTime) throw new Error(`Record ${index + 1}: endTime precedes startTime.`);
131 if (a.category !== undefined && !CATEGORIES.includes(a.category)) throw new Error(`Unknown category "${a.category}". Use "other" plus subtype for extensions.`);
132 const allowed: Status[] = ['pending', 'running', 'success', 'error', 'unknown'];
133 if (a.status !== undefined && !allowed.includes(a.status)) throw new Error(`Record ${index + 1}: invalid status.`);
134 const at = obj(a.attributes);
135 return {
136 schemaVersion: 1, id, traceId, name, parentId: str(a.parentId), startTime, endTime,
137 openEnded: a.endTime === undefined || a.openEnded === true,
138 agentId: str(a.agentId) ?? 'unattributed', agentType: str(a.agentType), parentAgentId: str(a.parentAgentId),
139 category: a.category ?? categoryFor(name, at), subtype: str(a.subtype),
140 model: str(a.model), provider: str(a.provider), tool: str(a.tool),
141 inputTokens: nonnegative(a.inputTokens, 'inputTokens'), outputTokens: nonnegative(a.outputTokens, 'outputTokens'),
142 cachedTokens: nonnegative(a.cachedTokens, 'cachedTokens'), cost: nonnegative(a.cost, 'cost'),
143 costCurrency: str(a.costCurrency), latency: nonnegative(a.latency, 'latency'),
144 contextTokens: nonnegative(a.contextTokens, 'contextTokens'), contextLimit: nonnegative(a.contextLimit, 'contextLimit'),
145 retry: nonnegative(a.retry, 'retry'), status: a.status ?? 'unknown',
146 sourceId: str(a.sourceId), targetId: str(a.targetId), targetType: str(a.targetType),
147 links: list(a.links).filter(x => typeof x?.traceId === 'string' && typeof x?.spanId === 'string'),
148 attributes: at, payload: a.payload, raw: a.raw ?? v,
149 observation: a.observation === undefined ? undefined : validateObservation(a.observation),
150 };
151 }
152 interface OtlpRecord { span: Obj; resource: Obj; scope: Obj; resourceSchema?: string; scopeSchema?: string }
153 function otlpRecords(doc: Obj): OtlpRecord[] {
154 const out: OtlpRecord[] = [];
155 for (const r of list(doc.resourceSpans)) {
156 for (const s of list(r.scopeSpans ?? r.instrumentationLibrarySpans)) {
157 for (const span of list(s.spans)) out.push({ span: obj(span), resource: obj(r.resource), scope: obj(s.scope ?? s.instrumentationLibrary), resourceSchema: r.schemaUrl, scopeSchema: s.schemaUrl });
158 }
159 }
160 return out;
161 }
162 function fromOTLP(doc: Obj, maxEvents: number): { events: WhaleEvent[]; origins: Map<string, string>; warnings: string[] } {
163 const records = otlpRecords(doc), bases = new Map<string, bigint>(), warnings: string[] = [];
164 if (!records.length) throw new Error('No spans found in resourceSpans[].scopeSpans[].spans[].');
165 if (records.length > maxEvents) throw new Error(`Import exceeds the ${maxEvents.toLocaleString()} event limit.`);
166 for (const { span: s } of records) {
167 if (!str(s.traceId) || !str(s.spanId)) throw new Error('Every OTLP span requires traceId and spanId.');
168 const start = ns(s.startTimeUnixNano, 'startTimeUnixNano');
169 const end = s.endTimeUnixNano === undefined ? start : ns(s.endTimeUnixNano, 'endTimeUnixNano');
170 if (end < start) throw new Error(`OTLP span ${s.spanId}: end precedes start.`);
171 let earliest = start;
172 for (const e of list(s.events)) { const t = ns(e.timeUnixNano, 'event.timeUnixNano'); if (t < earliest) earliest = t; }
173 if (!bases.has(s.traceId) || earliest < bases.get(s.traceId)!) bases.set(s.traceId, earliest);
174 if ((s.droppedEventsCount ?? 0) > 0) warnings.push(`Span ${s.spanId} reports ${s.droppedEventsCount} dropped events; coverage is incomplete.`);
175 }
176 const events: WhaleEvent[] = [];
177 for (const rec of records) {
178 const s = rec.span, a = { ...attributes(rec.resource.attributes), ...attributes(s.attributes) };
179 const origin = bases.get(s.traceId)!, start = ns(s.startTimeUnixNano, 'startTimeUnixNano');
180 const end = s.endTimeUnixNano === undefined ? start : ns(s.endTimeUnixNano, 'endTimeUnixNano');
181 const name = String(s.name ?? 'unnamed span');
182 const e: WhaleEvent = {
183 schemaVersion: 1, id: s.spanId, traceId: s.traceId, parentId: str(s.parentSpanId),
184 name, startTime: Number(start - origin) / 1e6, endTime: Number(end - origin) / 1e6,
185 openEnded: s.endTimeUnixNano === undefined,
186 agentId: String(a['whalesong.agent_id'] ?? a['gen_ai.agent.id'] ?? a['agent.id'] ?? a['service.name'] ?? 'unattributed'),
187 parentAgentId: str(a['agent.parent_id']), agentType: str(a['gen_ai.agent.name']),
188 category: categoryFor(name, a), model: str(a['gen_ai.request.model'] ?? a['gen_ai.response.model'] ?? a['llm.model_name']),
189 provider: str(a['gen_ai.provider.name'] ?? a['gen_ai.system']), tool: str(a['gen_ai.tool.name'] ?? a['tool.name']),
190 inputTokens: numericAttr(a['gen_ai.usage.input_tokens'] ?? a['llm.token_count.prompt']),
191 outputTokens: numericAttr(a['gen_ai.usage.output_tokens'] ?? a['llm.token_count.completion']),
192 cachedTokens: numericAttr(a['gen_ai.usage.cache_read.input_tokens'] ?? a['gen_ai.usage.input_tokens.cached']),
193 cost: numericAttr(a['whalesong.cost'] ?? a['gen_ai.usage.cost']), costCurrency: str(a['whalesong.cost_currency']),
194 contextTokens: numericAttr(a['context.tokens']), contextLimit: numericAttr(a['context.limit']),
195 retry: numericAttr(a['retry.count'] ?? a['retry.attempt']), status: otelStatus(s.status),
196 latency: Number(end - start) / 1e6, attributes: a,
197 sourceId: str(a['whalesong.source_id']), targetId: str(a['whalesong.target_id']), targetType: str(a['whalesong.target_type']),
198 links: list(s.links).map(l => ({ traceId: l.traceId, spanId: l.spanId, attributes: attributes(l.attributes) })),
199 payload: a['gen_ai.input.messages'] !== undefined || a['gen_ai.output.messages'] !== undefined ? {
200 request: a['gen_ai.input.messages'], response: a['gen_ai.output.messages'],
201 } : undefined,
202 raw: rec,
203 };
204 events.push(e);
205 for (const [i, record] of list(s.events).entries()) {
206 const ea = attributes(record.attributes), time = Number(ns(record.timeUnixNano, 'event.timeUnixNano') - origin) / 1e6;
207 const ename = String(record.name ?? 'span event');
208 events.push({ schemaVersion: 1, id: `${s.spanId}/event/${i}`, traceId: s.traceId, parentId: s.spanId,
209 startTime: time, endTime: time, agentId: e.agentId, name: ename, category: categoryFor(ename, ea),
210 status: ename === 'exception' ? 'error' : 'unknown', attributes: ea, raw: { event: record, spanId: s.spanId, resource: rec.resource, scope: rec.scope } });
211 }
212 if (events.length > maxEvents) throw new Error(`Import exceeds the ${maxEvents.toLocaleString()} event limit, including span events.`);
213 }
214 return { events, origins: new Map([...bases].map(([k, v]) => [k, v.toString()])), warnings };
215 }
216
217 /** Parse strictly: malformed lines or duplicate identities never disappear silently. */
218 export function importTrace(text: string, filename = 'Imported trace', options: ImportOptions = {}): Trace[] {
219 const mode = options.privacy ?? 'redact', maxEvents = options.maxEvents ?? 250_000;
220 if (!['redact', 'metadata', 'retain'].includes(mode)) throw new Error('Unknown privacy mode.');
221 const maxTraces = options.maxTraces ?? 8;
222 if (!Number.isInteger(maxEvents) || maxEvents < 1 || maxEvents > 250_000) throw new Error('maxEvents must be in [1, 250000].');
223 if (!Number.isInteger(maxTraces) || maxTraces < 1 || maxTraces > 64) throw new Error('maxTraces must be in [1, 64].');
224 if (new TextEncoder().encode(text).length > (options.maxBytes ?? 64 * 1024 * 1024)) throw new Error('File exceeds the 64 MiB MVP import limit. Split the export by trace.');
225 const trimmed = text.replace(/^\uFEFF/, '').trim();
226 if (!trimmed) throw new Error('The trace file is empty.');
227 let document: unknown;
228 try { document = JSON.parse(trimmed); }
229 catch {
230 document = trimmed.split(/\r?\n/).filter(l => l.trim()).map((line, i) => {
231 try { return JSON.parse(line); } catch { throw new Error(`Invalid JSON on nonempty line ${i + 1}. Import cancelled; no rows were skipped.`); }
232 });
233 }
234 // Transform before both normalization and raw retention, so raw cannot bypass redaction.
235 const safe = mode === 'retain' ? (options.redactor ? options.redactor(document, '') : document) : redact(document, '', options.redactor);
236 const root = obj(safe);
237 if(root.format === 'whalesong.evidence/v1') return [evidenceToTrace(validateBundle(root, Math.min(maxEvents, 100_000)))];
238 if (isCodewhaleSession(safe)) {
239 const trace = fromCodewhaleSession(safe, filename, maxEvents);
240 trace.privacy = mode;
241 trace.events = trace.events.map(e => privacyEvent(e, mode));
242 return [trace];
243 }
244 const records = Array.isArray(safe) ? safe : Array.isArray(root.events) ? root.events : null;
245 if (isCodewhaleRuntimeDocument(records ?? [safe])) {
246 const trace = fromCodewhaleRuntime(records ?? [safe], filename, maxEvents);
247 trace.privacy = mode;
248 trace.events = trace.events.map(e => privacyEvent(e, mode));
249 return [trace];
250 }
251 const isOTLP = Array.isArray(root.resourceSpans);
252 let all: WhaleEvent[], origins = new Map<string, string>(), warnings: string[] = [];
253 if (isOTLP) { const r = fromOTLP(root, maxEvents); all = r.events; origins = r.origins; warnings = r.warnings; }
254 else {
255 const incoming = records ?? [safe];
256 if (incoming.length > maxEvents) throw new Error(`Import exceeds the ${maxEvents.toLocaleString()} event limit.`);
257 all = incoming.map(normalizedEvent);
258 }
259 if (!all.length) throw new Error('The file contains no events.');
260 const groups = new Map<string, WhaleEvent[]>(), ids = new Set<string>();
261 for (const e of all) {
262 const key = `${e.traceId}\0${e.id}`;
263 if (ids.has(key)) throw new Error(`Duplicate event identity (${e.traceId}, ${e.id}). Import cancelled.`);
264 ids.add(key);
265 const group = groups.get(e.traceId) ?? []; group.push(privacyEvent(e, mode)); groups.set(e.traceId, group);
266 if (groups.size > maxTraces) throw new Error(`This import contains more than ${maxTraces} traces. Split it by trace ID.`);
267 }
268 return [...groups].map(([id, events]) => {
269 events.sort((a, b) => a.startTime - b.startTime || a.id.localeCompare(b.id));
270 const base = isOTLP ? 0 : events[0].startTime;
271 if (!isOTLP && events.some(e => Math.abs(e.endTime - base) > Number.MAX_SAFE_INTEGER)) throw new Error('Trace duration exceeds safely representable milliseconds.');
272 for (const e of events) {
273 if (e.attributes['whalesong.error_onset_ms'] !== undefined) e.attributes['whalesong.error_onset_ms'] = errorOnsetOf(e) - base;
274 e.startTime -= base; e.endTime -= base;
275 }
276 const localIds = new Set(events.map(e => e.id)), missingParents = events.filter(e => e.parentId && !localIds.has(e.parentId)).length;
277 const traceWarnings = [...warnings];
278 if (missingParents) traceWarnings.push(`${missingParents} parent spans are absent from this trace; no parent relationship was invented.`);
279 if (events.some(e => e.openEnded)) traceWarnings.push('Open spans have unknown duration and are displayed as onset-only, not extended into invented activity.');
280 const currencies = new Set(events.filter(e => e.cost !== undefined).map(e => e.costCurrency ?? 'unspecified'));
281 if (currencies.size > 1) traceWarnings.push('Mixed cost currencies: aggregate cost comparison is disabled.');
282 return { id, name: groups.size > 1 ? `${filename} · ${id.slice(0, 8)}` : String(root.name ?? filename),
283 events, duration: Math.max(1, events.reduce((m, e) => Math.max(m, e.endTime, e.startTime, e.status === 'error' ? errorOnsetOf(e) : 0), 0)),
284 originTime: origins.get(id) ?? (root.originTime !== undefined && base === 0 ? str(root.originTime) : `${base} ms`),
285 source: isOTLP ? 'otlp' as const : 'jsonl' as const, privacy: mode, warnings: [...new Set(traceWarnings)],
286 metadata: { ...(mode === 'metadata' ? {} : obj(root.metadata)), timeUnit: 'ms', originUnit: isOTLP ? 'unix-nanoseconds' : 'milliseconds', sourceFilename: filename },
287 };
288 });
289 }
290
291 export function exportJSONL(trace: Trace): string {
292 const bundle=bundleFromTrace(trace);
293 if(bundle)return JSON.stringify(bundle);
294 return trace.events.map(e => JSON.stringify(e)).join('\n') + '\n';
295 }
296
296 lines TYPESCRIPT