From f67868bea5f2c9311e1d915c7d67374616a0945e Mon Sep 17 00:00:00 2001 From: jedarden Date: Tue, 21 Apr 2026 12:34:30 -0400 Subject: [PATCH] =?UTF-8?q?feat(bd-h40):=20extract=20Normalizer=20componen?= =?UTF-8?q?t=20=E2=80=94=20decouple=20parse=20from=20ingest?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Create src/normalizer.ts with normalize(raw, source): NeedleEvent supporting all 5 sources: jsonl, otlp-log, otlp-span-start, otlp-span-end, otlp-metric - Move JSONL parsing logic (canonical, legacy NEEDLE, flat legacy) from parser.ts into normalizer.ts - Refactor parser.ts to thin facade delegating to normalizer - Tailer now imports directly from normalizer (pure source, no parse coupling) - Add 63 unit tests covering all source paths, edge cases, and cross-source parity between JSONL and OTLP-log Co-Authored-By: Claude Opus 4.7 --- src/normalizer.test.ts | 859 +++++++++++++++++++++++++++++++++++++++++ src/normalizer.ts | 541 ++++++++++++++++++++++++++ src/parser.ts | 392 ++----------------- src/tailer.ts | 4 +- 4 files changed, 1424 insertions(+), 372 deletions(-) create mode 100644 src/normalizer.test.ts create mode 100644 src/normalizer.ts diff --git a/src/normalizer.test.ts b/src/normalizer.test.ts new file mode 100644 index 0000000..ddcbed0 --- /dev/null +++ b/src/normalizer.test.ts @@ -0,0 +1,859 @@ +/** + * Tests for FABRIC Normalizer + * + * Covers all 5 source paths: + * - jsonl (canonical NeedleEvent, legacy NEEDLE, flat legacy) + * - otlp-log + * - otlp-span-start + * - otlp-span-end + * - otlp-metric + */ + +import { describe, it, expect } from 'vitest'; +import { + normalize, + normalizeToLogEvent, + needleEventToLogEvent, + NormalizerSource, +} from './normalizer.js'; +import { NeedleEvent, NEEDLE_EVENT_SCHEMA_VERSION, LogEvent } from './types.js'; + +// ── Helpers ────────────────────────────────────────────────────── + +function canonicalNeedleEvent(overrides: Partial = {}): NeedleEvent { + return { + timestamp: '2026-03-04T16:17:34.008Z', + event_type: 'worker.started', + worker_id: 'claude-anthropic-sonnet-test', + session_id: 'test-session', + sequence: 1, + data: {}, + ...overrides, + }; +} + +// ── normalize() ────────────────────────────────────────────────── + +describe('normalize', () => { + it('returns null for unknown source', () => { + // Exhaustive switch means this can't happen at compile time, + // but we verify the default branch returns null + const result = normalize('anything', 'jsonl' as NormalizerSource); + // Should return a value or null depending on content + expect(result === null || typeof result === 'object').toBe(true); + }); +}); + +// ── JSONL source ───────────────────────────────────────────────── + +describe('normalize – jsonl source', () => { + describe('canonical NeedleEvent format', () => { + it('parses a canonical NeedleEvent JSON string', () => { + const raw = JSON.stringify(canonicalNeedleEvent()); + const result = normalize(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('worker.started'); + expect(result!.worker_id).toBe('claude-anthropic-sonnet-test'); + expect(result!.session_id).toBe('test-session'); + expect(result!.sequence).toBe(1); + }); + + it('parses a canonical NeedleEvent object passed as string', () => { + // normalizeJsonl expects a string — objects must be JSON-stringified first + const obj = canonicalNeedleEvent(); + const result = normalize(JSON.stringify(obj), 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('worker.started'); + }); + + it('preserves bead_id', () => { + const raw = JSON.stringify(canonicalNeedleEvent({ bead_id: 'bd-abc' })); + const result = normalize(raw, 'jsonl'); + expect(result!.bead_id).toBe('bd-abc'); + }); + + it('preserves data payload', () => { + const raw = JSON.stringify(canonicalNeedleEvent({ + data: { pid: 1234, workspace: '/home/test' }, + })); + const result = normalize(raw, 'jsonl'); + expect(result!.data.pid).toBe(1234); + expect(result!.data.workspace).toBe('/home/test'); + }); + + it('throws on schema version mismatch', () => { + const raw = JSON.stringify({ + ...canonicalNeedleEvent(), + schema_version: 999, + }); + expect(() => normalize(raw, 'jsonl')).toThrow(/schema mismatch/); + }); + + it('accepts matching schema version', () => { + const raw = JSON.stringify({ + ...canonicalNeedleEvent(), + schema_version: NEEDLE_EVENT_SCHEMA_VERSION, + }); + const result = normalize(raw, 'jsonl'); + expect(result).not.toBeNull(); + }); + + it('returns null if required fields are missing', () => { + // Missing worker_id + const raw = JSON.stringify({ + timestamp: '2026-03-04T16:17:34.008Z', + event_type: 'test', + session_id: 's1', + sequence: 1, + }); + expect(normalize(raw, 'jsonl')).toBeNull(); + }); + }); + + describe('legacy NEEDLE format', () => { + it('parses NEEDLE format with worker object', () => { + const raw = JSON.stringify({ + ts: '2026-03-04T16:17:34.008Z', + event: 'worker.started', + session: 'test-session', + worker: { + runner: 'claude', + provider: 'anthropic', + model: 'sonnet', + identifier: 'test12', + }, + data: { pid: 1929276 }, + }); + const result = normalize(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.worker_id).toBe('claude-anthropic-sonnet-test12'); + expect(result!.event_type).toBe('worker.started'); + expect(result!.session_id).toBe('test-session'); + // sequence is -1 for legacy format (not available) + expect(result!.sequence).toBe(-1); + }); + + it('parses NEEDLE format with worker string', () => { + const raw = JSON.stringify({ + ts: '2026-03-04T16:17:34.008Z', + event: 'bead.claimed', + session: 'test-session', + worker: 'claude-anthropic-opus-prod', + data: { bead_id: 'bd-2ok0' }, + }); + const result = normalize(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.worker_id).toBe('claude-anthropic-opus-prod'); + expect(result!.bead_id).toBe('bd-2ok0'); + }); + + it('preserves level in data', () => { + const raw = JSON.stringify({ + ts: '2026-03-04T16:17:34.008Z', + event: 'bead.claim_retry', + level: 'warn', + session: 'test-session', + worker: 'worker-1', + data: {}, + }); + const result = normalize(raw, 'jsonl'); + expect(result!.data.level).toBe('warn'); + }); + + it('preserves provider/model from worker object in data', () => { + const raw = JSON.stringify({ + ts: '2026-03-04T16:17:34.008Z', + event: 'worker.started', + session: 'test-session', + worker: { + runner: 'claude', + provider: 'anthropic', + model: 'opus', + identifier: 'prod', + }, + data: {}, + }); + const result = normalize(raw, 'jsonl'); + expect(result!.data.provider).toBe('anthropic'); + expect(result!.data.model).toBe('opus'); + }); + + it('promotes bead_id from data to top level', () => { + const raw = JSON.stringify({ + ts: '2026-03-04T16:17:34.008Z', + event: 'bead.claimed', + session: 'test-session', + worker: 'worker-1', + data: { bead_id: 'bd-xyz', title: 'Test task' }, + }); + const result = normalize(raw, 'jsonl'); + expect(result!.bead_id).toBe('bd-xyz'); + // bead_id should be removed from data + expect(result!.data.bead_id).toBeUndefined(); + // other fields preserved + expect(result!.data.title).toBe('Test task'); + }); + }); + + describe('flat legacy format', () => { + it('parses flat legacy format', () => { + const raw = JSON.stringify({ + ts: 1709337600000, + worker: 'w-abc123', + level: 'info', + msg: 'Test message', + }); + const result = normalize(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.worker_id).toBe('w-abc123'); + expect(result!.event_type).toBe('Test message'); + expect(result!.data.level).toBe('info'); + }); + + it('preserves optional fields in data', () => { + const raw = JSON.stringify({ + ts: 1709337600000, + worker: 'w-test', + level: 'debug', + msg: 'Tool call', + tool: 'Read', + path: '/src/main.ts', + bead: 'bd-xyz', + duration_ms: 5000, + error: 'some error', + }); + const result = normalize(raw, 'jsonl'); + expect(result!.data.tool).toBe('Read'); + expect(result!.data.path).toBe('/src/main.ts'); + expect(result!.bead_id).toBe('bd-xyz'); + expect(result!.data.duration_ms).toBe(5000); + expect(result!.data.error).toBe('some error'); + }); + + it('preserves extra fields in data', () => { + const raw = JSON.stringify({ + ts: 1709337600000, + worker: 'w-test', + level: 'info', + msg: 'Test', + customField: 'custom value', + tokens: 150, + }); + const result = normalize(raw, 'jsonl'); + expect(result!.data.customField).toBe('custom value'); + expect(result!.data.tokens).toBe(150); + }); + + it('returns null for invalid flat format', () => { + // Missing msg + const raw = JSON.stringify({ + ts: 1709337600000, + worker: 'w-test', + level: 'info', + }); + expect(normalize(raw, 'jsonl')).toBeNull(); + }); + + it('returns null for invalid level', () => { + const raw = JSON.stringify({ + ts: 1709337600000, + worker: 'w-test', + level: 'invalid', + msg: 'Test', + }); + expect(normalize(raw, 'jsonl')).toBeNull(); + }); + }); + + describe('edge cases', () => { + it('returns null for empty string', () => { + expect(normalize('', 'jsonl')).toBeNull(); + }); + + it('returns null for whitespace-only string', () => { + expect(normalize(' \n\t ', 'jsonl')).toBeNull(); + }); + + it('returns null for non-JSON string', () => { + expect(normalize('not valid json', 'jsonl')).toBeNull(); + }); + + it('returns null for null input', () => { + expect(normalize(null as unknown as string, 'jsonl')).toBeNull(); + }); + + it('returns null for number input', () => { + // normalizeJsonl expects a string; non-string input via jsonl source + // is a caller bug — verify it doesn't crash + expect(() => normalize(42 as unknown as string, 'jsonl')).not.toThrow(); + }); + }); +}); + +// ── OTLP-log source ────────────────────────────────────────────── + +describe('normalize – otlp-log source', () => { + it('parses OTLP LogRecord with attributes array', () => { + const record = { + timeUnixNano: '1772641054008000000', + attributes: [ + { key: 'event_type', value: { stringValue: 'worker.started' } }, + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + { key: 'session_id', value: { stringValue: 'sess-1' } }, + { key: 'sequence', value: { intValue: '5' } }, + { key: 'bead_id', value: { stringValue: 'bd-123' } }, + { key: 'custom', value: { stringValue: 'data' } }, + ], + }; + const result = normalize(record, 'otlp-log'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('worker.started'); + expect(result!.worker_id).toBe('tcb-alpha'); + expect(result!.session_id).toBe('sess-1'); + expect(result!.sequence).toBe(5); + expect(result!.bead_id).toBe('bd-123'); + expect(result!.data.custom).toBe('data'); + // Standard fields should not be in data + expect(result!.data.event_type).toBeUndefined(); + expect(result!.data.worker_id).toBeUndefined(); + }); + + it('parses OTLP LogRecord with plain object attributes', () => { + const record = { + timeUnixNano: '1772641054008000000', + attributes: { + event_type: 'bead.claimed', + worker_id: 'tcb-bravo', + session_id: 'sess-2', + }, + }; + const result = normalize(record, 'otlp-log'); + expect(result).not.toBeNull(); + expect(result!.worker_id).toBe('tcb-bravo'); + }); + + it('uses current time when timeUnixNano is missing', () => { + const record = { + attributes: [ + { key: 'event_type', value: { stringValue: 'test' } }, + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + const result = normalize(record, 'otlp-log'); + expect(result).not.toBeNull(); + // Should have a valid ISO timestamp + expect(typeof result!.timestamp).toBe('string'); + expect(Date.parse(result!.timestamp)).not.toBeNaN(); + }); + + it('returns null when worker_id is missing', () => { + const record = { + attributes: [ + { key: 'event_type', value: { stringValue: 'test' } }, + ], + }; + expect(normalize(record, 'otlp-log')).toBeNull(); + }); + + it('returns null when event_type is missing', () => { + const record = { + attributes: [ + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + expect(normalize(record, 'otlp-log')).toBeNull(); + }); + + it('returns null for null input', () => { + expect(normalize(null, 'otlp-log')).toBeNull(); + }); + + it('parses int and double values from attributes', () => { + const record = { + timeUnixNano: '1772641054008000000', + attributes: [ + { key: 'event_type', value: { stringValue: 'test' } }, + { key: 'worker_id', value: { stringValue: 'w-1' } }, + { key: 'count', value: { intValue: '42' } }, + { key: 'rate', value: { doubleValue: 3.14 } }, + { key: 'active', value: { boolValue: true } }, + ], + }; + const result = normalize(record, 'otlp-log'); + expect(result!.data.count).toBe(42); + expect(result!.data.rate).toBe(3.14); + expect(result!.data.active).toBe(true); + }); +}); + +// ── OTLP-span-start source ─────────────────────────────────────── + +describe('normalize – otlp-span-start source', () => { + it('parses OTLP Span start', () => { + const span = { + startTimeUnixNano: '1772641054008000000', + attributes: [ + { key: 'event_type', value: { stringValue: 'bead.claimed' } }, + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + { key: 'session_id', value: { stringValue: 'sess-1' } }, + { key: 'sequence', value: { intValue: '10' } }, + { key: 'bead_id', value: { stringValue: 'bd-span' } }, + ], + }; + const result = normalize(span, 'otlp-span-start'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('bead.claimed'); + expect(result!.worker_id).toBe('tcb-alpha'); + expect(result!.session_id).toBe('sess-1'); + expect(result!.sequence).toBe(10); + expect(result!.bead_id).toBe('bd-span'); + }); + + it('defaults to bead.claimed when event_type is missing', () => { + const span = { + startTimeUnixNano: '1772641054008000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(span, 'otlp-span-start'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('bead.claimed'); + }); + + it('uses startTimeUnixNano for timestamp', () => { + const span = { + startTimeUnixNano: '1772641054008000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + const result = normalize(span, 'otlp-span-start'); + // 1772641054008000000 ns → 1772641054008 ms → 2026-03-04T16:17:34.008Z + expect(result!.timestamp).toBe('2026-03-04T16:17:34.008Z'); + }); + + it('accepts snake_case start_time_unix_nano', () => { + const span = { + start_time_unix_nano: '1772641054008000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + const result = normalize(span, 'otlp-span-start'); + expect(result!.timestamp).toBe('2026-03-04T16:17:34.008Z'); + }); + + it('returns null when worker_id is missing', () => { + const span = { + attributes: [], + }; + expect(normalize(span, 'otlp-span-start')).toBeNull(); + }); + + it('returns null for null input', () => { + expect(normalize(null, 'otlp-span-start')).toBeNull(); + }); +}); + +// ── OTLP-span-end source ───────────────────────────────────────── + +describe('normalize – otlp-span-end source', () => { + it('parses OTLP Span end with OK status', () => { + const span = { + startTimeUnixNano: '1772641054008000000', + endTimeUnixNano: '1772641058000000000', // ~3.9s later + status: { code: 'OK' }, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + { key: 'session_id', value: { stringValue: 'sess-1' } }, + { key: 'bead_id', value: { stringValue: 'bd-end' } }, + ], + }; + const result = normalize(span, 'otlp-span-end'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('bead.completed'); + expect(result!.worker_id).toBe('tcb-alpha'); + expect(result!.bead_id).toBe('bd-end'); + // Duration computed from start/end times + expect(result!.data.duration_ms).toBe(3992); + }); + + it('maps ERROR status to bead.failed', () => { + const span = { + endTimeUnixNano: '1772641058000000000', + status: { code: 'ERROR', message: 'Timeout exceeded' }, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(span, 'otlp-span-end'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('bead.failed'); + expect(result!.data.error).toBe('Timeout exceeded'); + }); + + it('maps numeric error code (2) to bead.failed', () => { + const span = { + endTimeUnixNano: '1772641058000000000', + status: { code: 2 }, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(span, 'otlp-span-end'); + expect(result!.event_type).toBe('bead.failed'); + }); + + it('uses explicit event_type from attributes when present', () => { + const span = { + endTimeUnixNano: '1772641058000000000', + status: { code: 'OK' }, + attributes: [ + { key: 'event_type', value: { stringValue: 'custom.event' } }, + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(span, 'otlp-span-end'); + expect(result!.event_type).toBe('custom.event'); + }); + + it('uses endTime for timestamp', () => { + const span = { + endTimeUnixNano: '1772641058000000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + const result = normalize(span, 'otlp-span-end'); + expect(result!.timestamp).toBe('2026-03-04T16:17:38.000Z'); + }); + + it('accepts snake_case end_time_unix_nano', () => { + const span = { + end_time_unix_nano: '1772641058000000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + const result = normalize(span, 'otlp-span-end'); + expect(result!.timestamp).toBe('2026-03-04T16:17:38.000Z'); + }); + + it('returns null when worker_id is missing', () => { + const span = { + endTimeUnixNano: '1772641058000000000', + attributes: [], + }; + expect(normalize(span, 'otlp-span-end')).toBeNull(); + }); + + it('returns null for null input', () => { + expect(normalize(null, 'otlp-span-end')).toBeNull(); + }); +}); + +// ── OTLP-metric source ─────────────────────────────────────────── + +describe('normalize – otlp-metric source', () => { + it('parses OTLP Metric with asDouble value', () => { + const metric = { + name: 'tokens.used', + timeUnixNano: '1772641054008000000', + asDouble: 42.5, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + { key: 'session_id', value: { stringValue: 'sess-1' } }, + { key: 'bead_id', value: { stringValue: 'bd-metric' } }, + { key: 'model', value: { stringValue: 'sonnet' } }, + ], + }; + const result = normalize(metric, 'otlp-metric'); + expect(result).not.toBeNull(); + expect(result!.event_type).toBe('metric.tokens.used'); + expect(result!.worker_id).toBe('tcb-alpha'); + expect(result!.session_id).toBe('sess-1'); + expect(result!.bead_id).toBe('bd-metric'); + expect(result!.data.value).toBe(42.5); + expect(result!.data.metric_name).toBe('tokens.used'); + expect(result!.data.model).toBe('sonnet'); + }); + + it('parses OTLP Metric with asInt value', () => { + const metric = { + name: 'beads.completed', + timeUnixNano: '1772641054008000000', + asInt: 10, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(metric, 'otlp-metric'); + expect(result!.data.value).toBe(10); + }); + + it('parses OTLP Metric with plain value field', () => { + const metric = { + name: 'cost.usd', + timeUnixNano: '1772641054008000000', + value: 0.05, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(metric, 'otlp-metric'); + expect(result!.data.value).toBe(0.05); + }); + + it('defaults event_type to metric.unknown when name is missing', () => { + const metric = { + timeUnixNano: '1772641054008000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + ], + }; + const result = normalize(metric, 'otlp-metric'); + expect(result!.event_type).toBe('metric.unknown'); + expect(result!.data.metric_name).toBeUndefined(); + }); + + it('returns null when worker_id is missing', () => { + const metric = { + name: 'test', + timeUnixNano: '1772641054008000000', + attributes: [], + }; + expect(normalize(metric, 'otlp-metric')).toBeNull(); + }); + + it('returns null for null input', () => { + expect(normalize(null, 'otlp-metric')).toBeNull(); + }); + + it('uses snake_case time_unix_nano', () => { + const metric = { + name: 'test', + time_unix_nano: '1772641054008000000', + attributes: [ + { key: 'worker_id', value: { stringValue: 'w-1' } }, + ], + }; + const result = normalize(metric, 'otlp-metric'); + expect(result!.timestamp).toBe('2026-03-04T16:17:34.008Z'); + }); +}); + +// ── normalizeToLogEvent ────────────────────────────────────────── + +describe('normalizeToLogEvent', () => { + it('returns null for invalid input', () => { + expect(normalizeToLogEvent('not json', 'jsonl')).toBeNull(); + expect(normalizeToLogEvent(null, 'otlp-log')).toBeNull(); + }); + + it('returns a valid LogEvent from JSONL canonical format', () => { + const raw = JSON.stringify(canonicalNeedleEvent({ + bead_id: 'bd-test', + data: { duration_ms: 5000, tool: 'Read', path: '/src/main.ts' }, + })); + const result = normalizeToLogEvent(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.ts).toBe(new Date('2026-03-04T16:17:34.008Z').getTime()); + expect(result!.worker).toBe('claude-anthropic-sonnet-test'); + expect(result!.msg).toBe('worker.started'); + expect(result!.session).toBe('test-session'); + expect(result!.bead).toBe('bd-test'); + expect(result!.duration_ms).toBe(5000); + expect(result!.tool).toBe('Read'); + expect(result!.path).toBe('/src/main.ts'); + }); + + it('returns a valid LogEvent from JSONL NEEDLE format', () => { + const raw = JSON.stringify({ + ts: '2026-03-04T16:17:34.008Z', + event: 'worker.started', + level: 'info', + session: 'test-session', + worker: { + runner: 'claude', + provider: 'anthropic', + model: 'sonnet', + identifier: 'test12', + }, + data: { pid: 1929276 }, + }); + const result = normalizeToLogEvent(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.worker).toBe('claude-anthropic-sonnet-test12'); + expect(result!.msg).toBe('worker.started'); + expect(result!.level).toBe('info'); + expect(result!.session).toBe('test-session'); + expect(result!.provider).toBe('anthropic'); + expect(result!.model).toBe('sonnet'); + }); + + it('returns a valid LogEvent from JSONL flat legacy format', () => { + const raw = JSON.stringify({ + ts: 1709337600000, + worker: 'w-test', + level: 'info', + msg: 'Test message', + }); + const result = normalizeToLogEvent(raw, 'jsonl'); + expect(result).not.toBeNull(); + expect(result!.ts).toBe(1709337600000); + expect(result!.worker).toBe('w-test'); + expect(result!.level).toBe('info'); + expect(result!.msg).toBe('Test message'); + }); + + it('returns a valid LogEvent from OTLP-log', () => { + const record = { + timeUnixNano: '1772641054008000000', + attributes: [ + { key: 'event_type', value: { stringValue: 'bead.claimed' } }, + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + { key: 'session_id', value: { stringValue: 'sess-1' } }, + ], + }; + const result = normalizeToLogEvent(record, 'otlp-log'); + expect(result).not.toBeNull(); + expect(result!.worker).toBe('tcb-alpha'); + expect(result!.msg).toBe('bead.claimed'); + expect(result!.session).toBe('sess-1'); + }); + + it('returns a valid LogEvent from OTLP-metric', () => { + const metric = { + name: 'tokens.used', + timeUnixNano: '1772641054008000000', + asDouble: 100, + attributes: [ + { key: 'worker_id', value: { stringValue: 'tcb-alpha' } }, + { key: 'session_id', value: { stringValue: 'sess-1' } }, + ], + }; + const result = normalizeToLogEvent(metric, 'otlp-metric'); + expect(result).not.toBeNull(); + expect(result!.worker).toBe('tcb-alpha'); + expect(result!.msg).toBe('metric.tokens.used'); + }); +}); + +// ── needleEventToLogEvent ──────────────────────────────────────── + +describe('needleEventToLogEvent', () => { + it('converts NeedleEvent to LogEvent', () => { + const ne = canonicalNeedleEvent({ + bead_id: 'bd-abc', + data: { + duration_ms: 3000, + error: 'test error', + tool: 'Bash', + path: '/tmp/test', + provider: 'anthropic', + model: 'sonnet', + }, + }); + const result = needleEventToLogEvent(ne); + expect(result.ts).toBe(new Date('2026-03-04T16:17:34.008Z').getTime()); + expect(result.worker).toBe('claude-anthropic-sonnet-test'); + expect(result.msg).toBe('worker.started'); + expect(result.session).toBe('test-session'); + expect(result.bead).toBe('bd-abc'); + expect(result.duration_ms).toBe(3000); + expect(result.error).toBe('test error'); + expect(result.tool).toBe('Bash'); + expect(result.path).toBe('/tmp/test'); + expect(result.provider).toBe('anthropic'); + expect(result.model).toBe('sonnet'); + }); + + it('infers level from event name', () => { + // error.* → error + expect(needleEventToLogEvent(canonicalNeedleEvent({ event_type: 'error.agent_crash' })).level).toBe('error'); + // *.failed → warn + expect(needleEventToLogEvent(canonicalNeedleEvent({ event_type: 'bead.failed' })).level).toBe('warn'); + // *.retry → warn + expect(needleEventToLogEvent(canonicalNeedleEvent({ event_type: 'claim.retry' })).level).toBe('warn'); + // debug.* → debug + expect(needleEventToLogEvent(canonicalNeedleEvent({ event_type: 'debug.probe' })).level).toBe('debug'); + // default → info + expect(needleEventToLogEvent(canonicalNeedleEvent({ event_type: 'worker.started' })).level).toBe('info'); + }); + + it('uses explicit level from data when present', () => { + const ne = canonicalNeedleEvent({ + event_type: 'bead.claim_retry', + data: { level: 'warn' }, + }); + const result = needleEventToLogEvent(ne); + expect(result.level).toBe('warn'); + }); + + it('copies extra data fields to LogEvent', () => { + const ne = canonicalNeedleEvent({ + data: { + pid: 1234, + workspace: '/home/test', + custom: 'value', + }, + }); + const result = needleEventToLogEvent(ne); + expect(result.pid).toBe(1234); + expect(result.workspace).toBe('/home/test'); + expect(result.custom).toBe('value'); + }); + + it('does not overwrite standard LogEvent fields with data fields', () => { + const ne = canonicalNeedleEvent({ + data: { + duration_ms: 5000, + }, + }); + const result = needleEventToLogEvent(ne); + expect(result.duration_ms).toBe(5000); + // 'ts' is set from timestamp, not from data + expect(result.ts).toBe(new Date('2026-03-04T16:17:34.008Z').getTime()); + }); +}); + +// ── Cross-source parity ────────────────────────────────────────── + +describe('cross-source parity', () => { + it('JSONL and OTLP-log produce equivalent LogEvents', () => { + const canonicalNe = canonicalNeedleEvent({ + bead_id: 'bd-parity', + data: { duration_ms: 1000 }, + }); + const jsonlLine = JSON.stringify(canonicalNe); + + const otlpRecord = { + timeUnixNano: String(new Date('2026-03-04T16:17:34.008Z').getTime() * 1_000_000), + attributes: [ + { key: 'event_type', value: { stringValue: canonicalNe.event_type } }, + { key: 'worker_id', value: { stringValue: canonicalNe.worker_id } }, + { key: 'session_id', value: { stringValue: canonicalNe.session_id } }, + { key: 'sequence', value: { intValue: String(canonicalNe.sequence) } }, + { key: 'bead_id', value: { stringValue: canonicalNe.bead_id! } }, + { key: 'duration_ms', value: { intValue: '1000' } }, + ], + }; + + const jsonlResult = normalizeToLogEvent(jsonlLine, 'jsonl'); + const otlpResult = normalizeToLogEvent(otlpRecord, 'otlp-log'); + + // Both should produce equivalent LogEvents + expect(jsonlResult).not.toBeNull(); + expect(otlpResult).not.toBeNull(); + + expect(jsonlResult!.worker).toBe(otlpResult!.worker); + expect(jsonlResult!.msg).toBe(otlpResult!.msg); + expect(jsonlResult!.session).toBe(otlpResult!.session); + expect(jsonlResult!.bead).toBe(otlpResult!.bead); + expect(jsonlResult!.level).toBe(otlpResult!.level); + // duration_ms in data is promoted to top-level + expect(jsonlResult!.duration_ms).toBe(1000); + expect(otlpResult!.duration_ms).toBe(1000); + }); +}); diff --git a/src/normalizer.ts b/src/normalizer.ts new file mode 100644 index 0000000..71553df --- /dev/null +++ b/src/normalizer.ts @@ -0,0 +1,541 @@ +/** + * FABRIC Normalizer + * + * Normalizes raw events from multiple ingestion sources into the canonical + * NeedleEvent shape. Each source (JSONL, OTLP) has its own parsing path, + * but all produce the same output type so the downstream event bus sees + * a uniform stream. + */ + +import { + LogEvent, + LogLevel, + NeedleEvent, + NEEDLE_EVENT_SCHEMA_VERSION, +} from './types.js'; + +// ── Source types ────────────────────────────────────────────── + +export type NormalizerSource = + | 'jsonl' + | 'otlp-log' + | 'otlp-span-start' + | 'otlp-span-end' + | 'otlp-metric'; + +// ── Internal interfaces for legacy formats ──────────────────── + +interface NeedleWorkerObject { + runner: string; + provider: string; + model: string; + identifier: string; +} + +interface NeedleLogEntry { + ts: string; + event: string; + level?: string; + session: string; + worker: string | NeedleWorkerObject; + data: Record; +} + +// ── Main entry point ────────────────────────────────────────── + +/** + * Normalize a raw input into the canonical NeedleEvent shape. + * + * @param raw Raw payload (string for JSONL, object for OTLP) + * @param source Which ingestion source produced this input + * @returns Normalized NeedleEvent, or null if the input is invalid/unrecognized + */ +export function normalize( + raw: string | unknown, + source: NormalizerSource, +): NeedleEvent | null { + switch (source) { + case 'jsonl': + return normalizeJsonl(raw as string); + case 'otlp-log': + return normalizeOtlpLog(raw); + case 'otlp-span-start': + return normalizeOtlpSpanStart(raw); + case 'otlp-span-end': + return normalizeOtlpSpanEnd(raw); + case 'otlp-metric': + return normalizeOtlpMetric(raw); + default: + return null; + } +} + +/** + * Convenience: normalize and then convert to the legacy LogEvent shape + * used by the existing store / TUI / web consumers. + */ +export function normalizeToLogEvent( + raw: string | unknown, + source: NormalizerSource, +): LogEvent | null { + const ne = normalize(raw, source); + return ne ? needleEventToLogEvent(ne) : null; +} + +// ── JSONL source ────────────────────────────────────────────── + +function normalizeJsonl(raw: string | unknown): NeedleEvent | null { + // Accept pre-parsed objects directly + let parsed: unknown; + if (typeof raw === 'string') { + if (!raw || !raw.trim()) return null; + try { + parsed = JSON.parse(raw); + } catch { + return null; + } + } else { + parsed = raw; + } + + // 1) Canonical NeedleEvent wire format (has timestamp, event_type, worker_id, …) + const canonical = parseCanonicalNeedleEvent(parsed); + if (canonical) return canonical; + + // 2) Legacy NEEDLE JSONL format (ts: string ISO, event: string, worker: obj|string) + if (isNeedleLogEntry(parsed)) { + return normalizeNeedleLogEntry(parsed); + } + + // 3) Legacy flat format (ts: number, worker: string, level, msg) + return normalizeLegacyLogEntry(parsed); +} + +/** + * Parse a raw object already matching the canonical NeedleEvent schema. + * Returns null (no throw) on structural mismatch; throws on version mismatch. + */ +function parseCanonicalNeedleEvent(raw: unknown): NeedleEvent | null { + if (typeof raw !== 'object' || raw === null) return null; + const obj = raw as Record; + + if (typeof obj.timestamp !== 'string') return null; + if (typeof obj.event_type !== 'string') return null; + if (typeof obj.worker_id !== 'string') return null; + if (typeof obj.session_id !== 'string') return null; + if (typeof obj.sequence !== 'number') return null; + + if (obj.schema_version !== undefined && obj.schema_version !== NEEDLE_EVENT_SCHEMA_VERSION) { + throw new Error( + `NeedleEvent schema mismatch: got ${obj.schema_version}, expected ${NEEDLE_EVENT_SCHEMA_VERSION}`, + ); + } + + const event: NeedleEvent = { + timestamp: obj.timestamp, + event_type: obj.event_type, + worker_id: obj.worker_id, + session_id: obj.session_id, + sequence: obj.sequence, + data: typeof obj.data === 'object' && obj.data !== null + ? obj.data as Record + : {}, + }; + + if (typeof obj.bead_id === 'string') { + event.bead_id = obj.bead_id; + } + + return event; +} + +function isNeedleLogEntry(parsed: unknown): parsed is NeedleLogEntry { + if (typeof parsed !== 'object' || parsed === null) return false; + const obj = parsed as Record; + return ( + typeof obj.ts === 'string' && + typeof obj.event === 'string' && + (typeof obj.worker === 'object' || typeof obj.worker === 'string') + ); +} + +function normalizeNeedleLogEntry(entry: NeedleLogEntry): NeedleEvent { + const workerId = typeof entry.worker === 'string' + ? entry.worker + : `${entry.worker.runner}-${entry.worker.provider}-${entry.worker.model}-${entry.worker.identifier}`; + + const data = entry.data || {}; + + const ne: NeedleEvent = { + timestamp: entry.ts, + event_type: entry.event, + worker_id: workerId, + session_id: entry.session, + sequence: -1, // not available in legacy NEEDLE JSONL + data: { ...data }, + }; + + // Preserve level in data for downstream conversion + if (entry.level) ne.data.level = entry.level; + + // Flatten worker object fields into data when available + if (typeof entry.worker === 'object') { + ne.data.provider = entry.worker.provider; + ne.data.model = entry.worker.model; + } + + // Promote bead_id if present in data + if (typeof data.bead_id === 'string') { + ne.bead_id = data.bead_id; + delete ne.data.bead_id; + } + + return ne; +} + +function normalizeLegacyLogEntry(parsed: unknown): NeedleEvent | null { + if (typeof parsed !== 'object' || parsed === null) return null; + const obj = parsed as Record; + + if (typeof obj.ts !== 'number') return null; + if (typeof obj.worker !== 'string') return null; + if (!isValidLogLevel(obj.level)) return null; + if (typeof obj.msg !== 'string') return null; + + // Collect extra fields into data + const standardFields = new Set([ + 'ts', 'worker', 'level', 'msg', 'tool', 'path', 'bead', 'duration_ms', 'error', + 'session', 'provider', 'model', + ]); + const data: Record = {}; + for (const key of Object.keys(obj)) { + if (!standardFields.has(key)) { + data[key] = obj[key]; + } + } + + const ne: NeedleEvent = { + timestamp: new Date(obj.ts).toISOString(), + event_type: obj.msg, + worker_id: obj.worker, + session_id: '', + sequence: -1, + data: { + ...data, + level: obj.level, + }, + }; + + if (typeof obj.tool === 'string') ne.data.tool = obj.tool; + if (typeof obj.path === 'string') ne.data.path = obj.path; + if (typeof obj.bead === 'string') ne.bead_id = obj.bead; + if (typeof obj.duration_ms === 'number') ne.data.duration_ms = obj.duration_ms; + if (typeof obj.error === 'string') ne.data.error = obj.error; + + return ne; +} + +// ── OTLP source stubs ───────────────────────────────────────── + +/** + * Normalize an OTLP LogRecord into NeedleEvent. + * + * Expected input shape (already JSON-parsed from the OTLP/HTTP JSON export + * or decoded from OTLP/gRPC protobuf): + * + * { + * timeUnixNano: string, + * body: { stringValue: string }, + * attributes: [ + * { key: "event_type", value: { stringValue: "worker.started" } }, + * { key: "worker_id", value: { stringValue: "tcb-alpha" } }, + * … + * ] + * } + */ +function normalizeOtlpLog(raw: unknown): NeedleEvent | null { + if (typeof raw !== 'object' || raw === null) return null; + const record = raw as Record; + + const attrs = otlpAttrs(record); + + const event_type = attrs.get('event_type'); + const worker_id = attrs.get('worker_id'); + const session_id = attrs.get('session_id'); + const sequence = attrs.get('sequence'); + + if (!event_type || !worker_id) return null; + + const timestamp = record.timeUnixNano + ? otlpNanoToISO(record.timeUnixNano as string | number) + : new Date().toISOString(); + + const data: Record = {}; + for (const [k, v] of attrs) { + if (k !== 'event_type' && k !== 'worker_id' && k !== 'session_id' && k !== 'sequence' && k !== 'bead_id') { + data[k] = v; + } + } + + const ne: NeedleEvent = { + timestamp, + event_type: String(event_type), + worker_id: String(worker_id), + session_id: session_id ? String(session_id) : '', + sequence: typeof sequence === 'number' ? sequence : -1, + data, + }; + + const bead_id = attrs.get('bead_id'); + if (bead_id) ne.bead_id = String(bead_id); + + return ne; +} + +/** + * Normalize an OTLP Span start event into NeedleEvent. + * + * A span start maps to a bead.claimed or worker.started event depending on + * the span's attributes. + */ +function normalizeOtlpSpanStart(raw: unknown): NeedleEvent | null { + if (typeof raw !== 'object' || raw === null) return null; + const span = raw as Record; + + const attrs = otlpAttrs(span); + const event_type = attrs.get('event_type') || 'bead.claimed'; + const worker_id = attrs.get('worker_id'); + if (!worker_id) return null; + + const startTime = span.startTimeUnixNano || span.start_time_unix_nano; + const timestamp = startTime + ? otlpNanoToISO(startTime as string | number) + : new Date().toISOString(); + + const data: Record = {}; + for (const [k, v] of attrs) { + if (k !== 'event_type' && k !== 'worker_id' && k !== 'session_id' && k !== 'sequence' && k !== 'bead_id') { + data[k] = v; + } + } + + const ne: NeedleEvent = { + timestamp, + event_type: String(event_type), + worker_id: String(worker_id), + session_id: attrs.get('session_id') ? String(attrs.get('session_id')) : '', + sequence: typeof attrs.get('sequence') === 'number' ? attrs.get('sequence') as number : -1, + data, + }; + + const bead_id = attrs.get('bead_id'); + if (bead_id) ne.bead_id = String(bead_id); + + return ne; +} + +/** + * Normalize an OTLP Span end event into NeedleEvent. + * + * A span end maps to a bead.completed or bead.failed event depending on + * the span's status. + */ +function normalizeOtlpSpanEnd(raw: unknown): NeedleEvent | null { + if (typeof raw !== 'object' || raw === null) return null; + const span = raw as Record; + + const attrs = otlpAttrs(span); + + // Determine event_type from span status + const status = span.status as Record | undefined; + const code = status?.code; + const spanEventType = attrs.get('event_type'); + const event_type = spanEventType + ? String(spanEventType) + : (code === 'ERROR' || code === 2) ? 'bead.failed' : 'bead.completed'; + + const worker_id = attrs.get('worker_id'); + if (!worker_id) return null; + + const endTime = span.endTimeUnixNano || span.end_time_unix_nano; + const timestamp = endTime + ? otlpNanoToISO(endTime as string | number) + : new Date().toISOString(); + + const data: Record = {}; + for (const [k, v] of attrs) { + if (k !== 'event_type' && k !== 'worker_id' && k !== 'session_id' && k !== 'sequence' && k !== 'bead_id') { + data[k] = v; + } + } + + // Include duration if both start and end times are available + const startTime = span.startTimeUnixNano || span.start_time_unix_nano; + if (startTime && endTime) { + const startNs = Number(startTime); + const endNs = Number(endTime); + data.duration_ms = Math.round((endNs - startNs) / 1_000_000); + } + + // Include error message from status + if (status?.message) { + data.error = status.message; + } + + const ne: NeedleEvent = { + timestamp, + event_type, + worker_id: String(worker_id), + session_id: attrs.get('session_id') ? String(attrs.get('session_id')) : '', + sequence: typeof attrs.get('sequence') === 'number' ? attrs.get('sequence') as number : -1, + data, + }; + + const bead_id = attrs.get('bead_id'); + if (bead_id) ne.bead_id = String(bead_id); + + return ne; +} + +/** + * Normalize an OTLP Metric data point into NeedleEvent. + * + * Maps OTLP gauge/sum/histogram data points with NEEDLE attributes + * into the canonical event stream. + */ +function normalizeOtlpMetric(raw: unknown): NeedleEvent | null { + if (typeof raw !== 'object' || raw === null) return null; + const metricPoint = raw as Record; + + const attrs = otlpAttrs(metricPoint); + + const metricName = metricPoint.name as string | undefined; + const worker_id = attrs.get('worker_id'); + if (!worker_id) return null; + + const timeUnixNano = metricPoint.timeUnixNano || metricPoint.time_unix_nano; + const timestamp = timeUnixNano + ? otlpNanoToISO(timeUnixNano as string | number) + : new Date().toISOString(); + + const data: Record = {}; + for (const [k, v] of attrs) { + if (k !== 'worker_id' && k !== 'session_id' && k !== 'bead_id') { + data[k] = v; + } + } + if (metricName) data.metric_name = metricName; + + // Extract value depending on metric type + if (metricPoint.asDouble !== undefined) data.value = metricPoint.asDouble; + else if (metricPoint.asInt !== undefined) data.value = metricPoint.asInt; + else if (typeof metricPoint.value === 'number') data.value = metricPoint.value; + + const ne: NeedleEvent = { + timestamp, + event_type: `metric.${metricName || 'unknown'}`, + worker_id: String(worker_id), + session_id: attrs.get('session_id') ? String(attrs.get('session_id')) : '', + sequence: -1, + data, + }; + + const bead_id = attrs.get('bead_id'); + if (bead_id) ne.bead_id = String(bead_id); + + return ne; +} + +// ── NeedleEvent → LogEvent conversion ───────────────────────── + +/** + * Convert a canonical NeedleEvent into the legacy LogEvent shape. + * + * This adapter preserves backward compatibility with existing UI consumers + * that depend on LogEvent's flat structure. + */ +export function needleEventToLogEvent(ne: NeedleEvent): LogEvent { + const ts = new Date(ne.timestamp).getTime(); + const level = inferLogLevel(ne.event_type, ne.data.level as string | undefined); + const event: LogEvent = { + ts, + worker: ne.worker_id, + level, + msg: ne.event_type, + }; + + if (ne.session_id) event.session = ne.session_id; + if (ne.bead_id !== undefined) event.bead = ne.bead_id; + + const data = ne.data; + if (typeof data.duration_ms === 'number') event.duration_ms = data.duration_ms; + if (typeof data.error === 'string') event.error = data.error; + if (typeof data.tool === 'string') event.tool = data.tool; + if (typeof data.path === 'string') event.path = data.path; + if (typeof data.provider === 'string') event.provider = data.provider; + if (typeof data.model === 'string') event.model = data.model; + + const reserved = new Set([ + 'duration_ms', 'error', 'tool', 'path', 'provider', 'model', 'level', + 'bead_id', + ]); + for (const key of Object.keys(data)) { + if (!reserved.has(key) && !(key in event)) { + event[key] = data[key]; + } + } + + return event; +} + +// ── Shared helpers ──────────────────────────────────────────── + +function isValidLogLevel(level: unknown): level is LogLevel { + return level === 'debug' || level === 'info' || level === 'warn' || level === 'error'; +} + +function inferLogLevel(eventName: string, explicitLevel?: string): LogLevel { + if (isValidLogLevel(explicitLevel)) return explicitLevel; + if (eventName.startsWith('error.')) return 'error'; + if (eventName.endsWith('.failed') || eventName.endsWith('.retry')) return 'warn'; + if (eventName.startsWith('debug.')) return 'debug'; + return 'info'; +} + +/** + * Extract OTLP attributes array into a plain Map. + * + * Accepts either: + * - attributes: [{ key, value: { stringValue | intValue | doubleValue } }] + * - attributes as a plain object (some JSON exporters use this form) + */ +function otlpAttrs(record: Record): Map { + const map = new Map(); + const attrs = record.attributes; + + if (Array.isArray(attrs)) { + for (const attr of attrs) { + if (typeof attr !== 'object' || attr === null) continue; + const a = attr as Record; + const key = a.key as string | undefined; + if (!key) continue; + const val = a.value as Record | undefined; + if (!val) continue; + if ('stringValue' in val) map.set(key, val.stringValue); + else if ('intValue' in val) map.set(key, Number(val.intValue)); + else if ('doubleValue' in val) map.set(key, val.doubleValue); + else if ('boolValue' in val) map.set(key, val.boolValue); + } + } else if (typeof attrs === 'object' && attrs !== null) { + for (const [key, val] of Object.entries(attrs as Record)) { + map.set(key, val); + } + } + + return map; +} + +function otlpNanoToISO(nano: string | number): string { + const ms = Math.round(Number(nano) / 1_000_000); + return new Date(ms).toISOString(); +} diff --git a/src/parser.ts b/src/parser.ts index e08a16b..33119e2 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -9,7 +9,6 @@ import { LogEvent, LogLevel, NeedleEvent, - NEEDLE_EVENT_SCHEMA_VERSION, ConversationEvent, PromptEvent, ResponseEvent, @@ -18,371 +17,42 @@ import { ToolResultEvent, ConversationParseOptions, } from './types.js'; +import { normalizeToLogEvent, normalize, needleEventToLogEvent } from './normalizer.js'; + +// Re-export from normalizer for backward compatibility +export { normalize, needleEventToLogEvent } from './normalizer.js'; /** - * Parse a raw JSON object into a canonical NeedleEvent. + * Parse a JSON log line into a canonical NeedleEvent. * - * Validates the wire-format fields and asserts schema version compatibility. - * Returns null if the object does not match the NeedleEvent shape. - * - * @throws Error if schema_version is present but doesn't match NEEDLE_EVENT_SCHEMA_VERSION - */ -export function parseNeedleEvent(raw: unknown): NeedleEvent | null { - if (typeof raw !== 'object' || raw === null) return null; - const obj = raw as Record; - - // Required fields - if (typeof obj.timestamp !== 'string') return null; - if (typeof obj.event_type !== 'string') return null; - if (typeof obj.worker_id !== 'string') return null; - if (typeof obj.session_id !== 'string') return null; - if (typeof obj.sequence !== 'number') return null; - - // Schema version assertion — present in newer NEEDLE output - if (obj.schema_version !== undefined && obj.schema_version !== NEEDLE_EVENT_SCHEMA_VERSION) { - throw new Error( - `NeedleEvent schema mismatch: got ${obj.schema_version}, expected ${NEEDLE_EVENT_SCHEMA_VERSION}` - ); - } - - const event: NeedleEvent = { - timestamp: obj.timestamp, - event_type: obj.event_type, - worker_id: obj.worker_id, - session_id: obj.session_id, - sequence: obj.sequence, - data: typeof obj.data === 'object' && obj.data !== null - ? obj.data as Record - : {}, - }; - - if (typeof obj.bead_id === 'string') { - event.bead_id = obj.bead_id; - } - - return event; -} - -/** - * Convert a canonical NeedleEvent into the legacy LogEvent shape. - * - * This adapter preserves backward compatibility with existing UI consumers - * that depend on LogEvent's flat structure. - */ -export function needleEventToLogEvent(ne: NeedleEvent): LogEvent { - const ts = new Date(ne.timestamp).getTime(); - const level = inferLogLevel(ne.event_type); - const event: LogEvent = { - ts, - worker: ne.worker_id, - level, - msg: ne.event_type, - session: ne.session_id, - }; - - if (ne.bead_id) event.bead = ne.bead_id; - - const data = ne.data; - if (typeof data.duration_ms === 'number') event.duration_ms = data.duration_ms; - if (typeof data.error === 'string') event.error = data.error; - if (typeof data.tool === 'string') event.tool = data.tool; - if (typeof data.path === 'string') event.path = data.path; - - for (const key of Object.keys(data)) { - if (!isStandardField(key) && !(key in event)) { - event[key] = data[key]; - } - } - - return event; -} - -/** - * Check if a parsed object matches the canonical NeedleEvent wire format. - * Canonical format uses `timestamp`, `event_type`, `worker_id`, `session_id`, `sequence`. - */ -function isCanonicalNeedleFormat(obj: Record): boolean { - return ( - typeof obj.timestamp === 'string' && - typeof obj.event_type === 'string' && - typeof obj.worker_id === 'string' && - typeof obj.session_id === 'string' && - typeof obj.sequence === 'number' - ); -} - -/** - * Parse a single log line - * - * Supports three formats: + * Handles all JSONL sub-formats: * 1. Canonical NeedleEvent: {timestamp, event_type, worker_id, session_id, sequence, data} * 2. Legacy NEEDLE format: {ts: ISO string, event: string, worker: {...}, session: string, data: {...}} * 3. Flat legacy format: {ts: Unix ms, worker: string, level: string, msg: string} * - * @param line - Raw log line (JSON string) - * @returns Parsed LogEvent or null if invalid + * @returns NeedleEvent with all fields preserved, or null if the line is unparseable + */ +export function parseNeedleEvent(line: string): NeedleEvent | null { + return normalize(line, 'jsonl'); +} + +/** + * Parse a single log line into a legacy LogEvent. + * + * Thin adapter: calls parseNeedleEvent to get the canonical shape, + * then projects to the flat LogEvent for backward-compatible UI consumers. */ export function parseLogLine(line: string): LogEvent | null { - // Skip empty lines - if (!line || !line.trim()) { - return null; - } - - try { - const parsed = JSON.parse(line); - - // Canonical NeedleEvent wire format (top priority) - if (typeof parsed === 'object' && parsed !== null && isCanonicalNeedleFormat(parsed as Record)) { - const ne = parseNeedleEvent(parsed); - if (ne) return needleEventToLogEvent(ne); - } - - // Legacy NEEDLE format - if (isNeedleFormat(parsed)) { - return parseNeedleFormat(parsed); - } - - // Legacy format validation - if (typeof parsed.ts !== 'number') { - return null; - } - if (typeof parsed.worker !== 'string') { - return null; - } - if (!isValidLogLevel(parsed.level)) { - return null; - } - if (typeof parsed.msg !== 'string') { - return null; - } - - // Construct LogEvent with validated fields - const event: LogEvent = { - ts: parsed.ts, - worker: parsed.worker, - level: parsed.level, - msg: parsed.msg, - }; - - // Copy optional fields if present - if (typeof parsed.tool === 'string') event.tool = parsed.tool; - if (typeof parsed.path === 'string') event.path = parsed.path; - if (typeof parsed.bead === 'string') event.bead = parsed.bead; - if (typeof parsed.duration_ms === 'number') event.duration_ms = parsed.duration_ms; - if (typeof parsed.error === 'string') event.error = parsed.error; - - // Copy any additional fields - for (const key of Object.keys(parsed)) { - if (!isStandardField(key) && !(key in event)) { - event[key] = parsed[key]; - } - } - - return event; - } catch { - // Not valid JSON - return null; - } + const ne = parseNeedleEvent(line); + return ne ? needleEventToLogEvent(ne) : null; } /** - * NEEDLE worker object — legacy format only, present in some tests. - * Production NEEDLE emits worker as a flat string: runner-provider-model-identifier. - */ -interface NeedleWorkerObject { - runner: string; // e.g., "claude" - provider: string; // e.g., "anthropic", "openai" - model: string; // e.g., "sonnet", "gpt-4o" - identifier: string; // e.g., "alpha", "bravo" -} - -/** - * NEEDLE log format interface. - * worker is a flat string in current NEEDLE output (runner-provider-model-identifier). - * The object form is retained for backward compat with legacy test fixtures. - */ -interface NeedleLogEntry { - ts: string; // ISO 8601 timestamp - event: string; // Event type (e.g., "worker.started", "bead.claimed") - level?: string; // Log level — always present in current NEEDLE output - session: string; // Session identifier - worker: string | NeedleWorkerObject; // Flat string in production; object in legacy fixtures - data: Record; // Event-specific payload -} - -/** - * Check if parsed object matches NEEDLE format - */ -function isNeedleFormat(parsed: unknown): parsed is NeedleLogEntry { - if (typeof parsed !== 'object' || parsed === null) return false; - const obj = parsed as Record; - - // NEEDLE format has: ts (string), event (string), worker (object or string) - // Worker can be either: - // - Object: {runner, provider, model, identifier} (legacy) - // - String: "runner-provider-model-identifier" (aligned format) - return ( - typeof obj.ts === 'string' && - typeof obj.event === 'string' && - (typeof obj.worker === 'object' || typeof obj.worker === 'string') - ); -} - -/** - * Parse NEEDLE format log entry - */ -function parseNeedleFormat(entry: NeedleLogEntry): LogEvent { - // Convert ISO timestamp to Unix milliseconds - const ts = new Date(entry.ts).getTime(); - - // Handle worker as string (current NEEDLE format) or object (legacy test fixtures) - const worker = typeof entry.worker === 'string' - ? entry.worker - : `${entry.worker.runner}-${entry.worker.provider}-${entry.worker.model}-${entry.worker.identifier}`; - - // Use event type as message - const msg = entry.event; - - // Use the level NEEDLE provides; fall back to inference only for legacy entries without it - const level = isValidLogLevel(entry.level) ? entry.level as LogLevel : inferLogLevel(entry.event); - - // Build LogEvent - const event: LogEvent = { - ts, - worker, - level, - msg, - }; - - // Extract optional fields from data payload - const data = entry.data || {}; - - // Extract bead_id (map to 'bead' field) - if (typeof data.bead_id === 'string') { - event.bead = data.bead_id; - } - - // Extract duration_ms - if (typeof data.duration_ms === 'number') { - event.duration_ms = data.duration_ms; - } - - // Extract error if present - if (typeof data.error === 'string') { - event.error = data.error; - } - - // Extract tool if present - if (typeof data.tool === 'string') { - event.tool = data.tool; - } - - // Extract path if present - if (typeof data.path === 'string') { - event.path = data.path; - } - - // Copy session and, when available from the object form, provider/model - event.session = entry.session; - if (typeof entry.worker !== 'string') { - event.provider = entry.worker.provider; - event.model = entry.worker.model; - } - - // Copy remaining data fields (excluding already extracted ones) - const extractedFields = ['bead_id', 'duration_ms', 'error', 'tool', 'path']; - for (const key of Object.keys(data)) { - if (!extractedFields.includes(key) && !(key in event)) { - event[key] = data[key]; - } - } - - return event; -} - -/** - * Infer log level from event name. - * - * Mirrors NEEDLE's _needle_telemetry_infer_level rules exactly: - * error.* → error - * *.failed → warn - * *.retry → warn - * debug.* → debug - * everything else → info - * - * This is only used as a fallback when the event's level field is absent - * (legacy log entries). Current NEEDLE always includes level explicitly. - */ -function inferLogLevel(eventName: string): LogLevel { - if (eventName.startsWith('error.')) return 'error'; - if (eventName.endsWith('.failed') || eventName.endsWith('.retry')) return 'warn'; - if (eventName.startsWith('debug.')) return 'debug'; - return 'info'; -} - -/** - * Parse a JSON object directly into a LogEvent - * - * Used for HTTP-ingested events that are already parsed as JSON objects. - * - * @param obj - Parsed JSON object - * @returns Parsed LogEvent or null if invalid + * Parse a JSON object directly into a LogEvent. + * Delegates to the normalizer for all format detection. */ export function parseEventObject(obj: unknown): LogEvent | null { - if (typeof obj !== 'object' || obj === null) { - return null; - } - - // Canonical NeedleEvent wire format (top priority) - const rec = obj as Record; - if (isCanonicalNeedleFormat(rec)) { - const ne = parseNeedleEvent(obj); - if (ne) return needleEventToLogEvent(ne); - } - - // Legacy NEEDLE format - if (isNeedleFormat(obj)) { - return parseNeedleFormat(obj); - } - - // Try as flat legacy format - validate required fields - const parsed = obj as Record; - if (typeof parsed.ts !== 'number') { - return null; - } - if (typeof parsed.worker !== 'string') { - return null; - } - if (!isValidLogLevel(parsed.level)) { - return null; - } - if (typeof parsed.msg !== 'string') { - return null; - } - - // Construct LogEvent with validated fields - const event: LogEvent = { - ts: parsed.ts, - worker: parsed.worker, - level: parsed.level, - msg: parsed.msg, - }; - - // Copy optional fields if present - if (typeof parsed.tool === 'string') event.tool = parsed.tool; - if (typeof parsed.path === 'string') event.path = parsed.path; - if (typeof parsed.bead === 'string') event.bead = parsed.bead; - if (typeof parsed.duration_ms === 'number') event.duration_ms = parsed.duration_ms; - if (typeof parsed.error === 'string') event.error = parsed.error; - - // Copy any additional fields - for (const key of Object.keys(parsed)) { - if (!isStandardField(key) && !(key in event)) { - event[key] = parsed[key]; - } - } - - return event; + return normalizeToLogEvent(obj, 'jsonl'); } /** @@ -449,24 +119,6 @@ export interface FormatOptions { colorize?: boolean; } -/** - * Check if level is valid - */ -function isValidLogLevel(level: unknown): level is LogLevel { - return level === 'debug' || level === 'info' || level === 'warn' || level === 'error'; -} - -/** - * Check if field is a standard LogEvent field - */ -function isStandardField(key: string): boolean { - return [ - 'ts', 'worker', 'level', 'msg', 'tool', 'path', 'bead', 'duration_ms', 'error', - // NEEDLE-specific fields - 'session', 'provider', 'model' - ].includes(key); -} - /** * Format timestamp for display */ diff --git a/src/tailer.ts b/src/tailer.ts index 28a3bb9..472e3a1 100644 --- a/src/tailer.ts +++ b/src/tailer.ts @@ -8,7 +8,7 @@ import * as fs from 'fs'; import * as path from 'path'; import { EventEmitter } from 'events'; import { LogEvent } from './types.js'; -import { parseLogLine } from './parser.js'; +import { normalizeToLogEvent } from './normalizer.js'; export interface TailerOptions { /** Path to log file or directory */ @@ -180,7 +180,7 @@ export class LogTailer extends EventEmitter { this.emit('line', line); if (this.parseJson) { - const event = parseLogLine(line); + const event = normalizeToLogEvent(line, 'jsonl'); if (event) { this.emit('event', event); }