From dffe4eedc3b62a416b2bf456d36c1316f679b3e8 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 14:46:07 -0400 Subject: [PATCH 1/4] feat(dev): OTLP trace storage for local dev MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pure OTLP wire handling (per-trace batch partitioning, id normalization, frontend shaping) and append-only per-trace JSONL storage. A batch routinely carries spans from several traces, so persistence partitions by trace id — writing whole batches under the first id corrupts trace identity. Consumed by the OTLP collector in #1980, which stacks on this. --- src/core/dev/otel/store.test.ts | 135 ++++++++++++ src/core/dev/otel/store.ts | 129 +++++++++++ src/core/dev/otel/transforms.test.ts | 243 +++++++++++++++++++++ src/core/dev/otel/transforms.ts | 311 +++++++++++++++++++++++++++ src/core/dev/otel/types.ts | 65 ++++++ 5 files changed, 883 insertions(+) create mode 100644 src/core/dev/otel/store.test.ts create mode 100644 src/core/dev/otel/store.ts create mode 100644 src/core/dev/otel/transforms.test.ts create mode 100644 src/core/dev/otel/transforms.ts create mode 100644 src/core/dev/otel/types.ts diff --git a/src/core/dev/otel/store.test.ts b/src/core/dev/otel/store.test.ts new file mode 100644 index 000000000..f4bb450c7 --- /dev/null +++ b/src/core/dev/otel/store.test.ts @@ -0,0 +1,135 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { TraceStore } from "./store"; +import type { OtlpPayload } from "./types"; + +const TRACE_A = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +const TRACE_B = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; + +function payload( + traceId: string, + options: { serviceName?: string; startNano?: string; name?: string } = {}, +): OtlpPayload { + return { + resourceSpans: [ + { + resource: { + attributes: [ + { key: "service.name", value: { stringValue: options.serviceName ?? "agent-1" } }, + ], + }, + scopeSpans: [ + { + scope: { name: "test" }, + spans: [ + { + traceId, + spanId: "0123456789abcdef", + name: options.name ?? "invoke_agent strands", + kind: 1, + startTimeUnixNano: options.startNano ?? `${BigInt(Date.now()) * 1_000_000n}`, + endTimeUnixNano: options.startNano ?? `${BigInt(Date.now()) * 1_000_000n}`, + }, + ], + }, + ], + }, + ], + }; +} + +let directory: string; +let store: TraceStore; + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), "trace-store-")); + store = new TraceStore(directory); +}); + +afterEach(async () => { + await rm(directory, { recursive: true, force: true }); +}); + +describe("TraceStore", () => { + test("append then list returns the trace with metadata", async () => { + await store.append(payload(TRACE_A)); + const traces = await store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.traceId).toBe(TRACE_A); + expect(traces[0]!.spanCount).toBe("1"); + expect(traces[0]!.resourceSpans).toBeDefined(); + }); + + test("appends to the same trace accumulate spans", async () => { + await store.append(payload(TRACE_A)); + await store.append(payload(TRACE_A, { name: "tool_use" })); + const traces = await store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.spanCount).toBe("2"); + }); + + test("payloads without a trace id are dropped", async () => { + await store.append({ resourceSpans: [] }); + expect(await store.list()).toEqual([]); + }); + + test("a batch carrying several traces lands in each trace's own file", async () => { + const batch = payload(TRACE_A); + batch.resourceSpans![0]!.scopeSpans![0]!.spans!.push({ + ...batch.resourceSpans![0]!.scopeSpans![0]!.spans![0]!, + traceId: TRACE_B, + name: "tool_use", + }); + await store.append(batch); + + const traces = await store.list(); + expect(traces.map((trace) => trace.traceId).sort()).toEqual([TRACE_A, TRACE_B]); + expect(traces.every((trace) => trace.spanCount === "1")).toBe(true); + expect(await store.get(TRACE_B)).toBeDefined(); + }); + + test("list filters by service name", async () => { + await store.append(payload(TRACE_A, { serviceName: "agent-1" })); + await store.append(payload(TRACE_B, { serviceName: "agent-2" })); + const traces = await store.list({ serviceName: "agent-2" }); + expect(traces.map((trace) => trace.traceId)).toEqual([TRACE_B]); + }); + + test("list filters by time window and sorts newest first", async () => { + const oldNano = `${BigInt(Date.now() - 24 * 60 * 60 * 1000) * 1_000_000n}`; + await store.append(payload(TRACE_A, { startNano: oldNano })); + await store.append(payload(TRACE_B)); + + expect((await store.list()).map((trace) => trace.traceId)).toEqual([TRACE_B]); + + const all = await store.list({ startTime: 0 }); + expect(all.map((trace) => trace.traceId)).toEqual([TRACE_B, TRACE_A]); + }); + + test("get returns the trace detail or undefined for unknown ids", async () => { + await store.append(payload(TRACE_A)); + const detail = await store.get(TRACE_A); + expect(detail?.resourceSpans).toBeDefined(); + expect(await store.get(TRACE_B)).toBeUndefined(); + }); + + test("skips malformed lines and files without failing", async () => { + await store.append(payload(TRACE_A)); + await writeFile(join(directory, `${TRACE_A}.otlp.jsonl`), "{not json}\n", { + flag: "a", + }); + await writeFile(join(directory, "garbage.otlp.jsonl"), "also not json\n"); + + const traces = await store.list(); + expect(traces).toHaveLength(1); + expect(traces[0]!.spanCount).toBe("1"); + }); + + test("list on a directory that does not exist returns empty", async () => { + const empty = new TraceStore(join(directory, "missing")); + expect(await empty.list()).toEqual([]); + expect(await empty.get(TRACE_A)).toBeUndefined(); + }); +}); diff --git a/src/core/dev/otel/store.ts b/src/core/dev/otel/store.ts new file mode 100644 index 000000000..57e7e4593 --- /dev/null +++ b/src/core/dev/otel/store.ts @@ -0,0 +1,129 @@ +import { appendFile, mkdir, readFile, readdir } from "node:fs/promises"; +import { join } from "node:path"; +import { buildTraceDetail, extractTraceMeta, partitionByTraceId } from "./transforms"; +import type { OtlpPayload, OtlpResourceLog, OtlpResourceSpan } from "./types"; + +const OTLP_EXT = ".otlp.jsonl"; +const DEFAULT_LIST_WINDOW_MS = 12 * 60 * 60 * 1000; + +export interface TraceSummary { + traceId: string; + timestamp: string; + sessionId?: string; + spanCount: string; + resourceSpans?: unknown[]; + resourceLogs?: unknown[]; +} + +export interface TraceDetail { + resourceSpans?: unknown[]; + resourceLogs?: unknown[]; +} + +export interface ListTracesOptions { + serviceName?: string; + startTime?: number; + endTime?: number; +} + +/** + * Append-only local trace storage: one JSON Lines file per trace (named by its + * trace id), each line a per-trace slice of an OTLP export payload. No in-memory + * state — reads go to disk on demand, which is fine because the inspector only + * fetches traces on user actions. Malformed files and lines are skipped, never fatal. + */ +export class TraceStore { + constructor(private readonly directory: string) {} + + /** + * Persist one OTLP export payload, partitioned by trace id so a batch that + * carries several traces lands in each trace's own file. Spans and log + * records without a trace id are dropped. + */ + public async append(payload: OtlpPayload): Promise { + const partitions = partitionByTraceId(payload); + if (partitions.size === 0) return; + + await mkdir(this.directory, { recursive: true }); + await Promise.all( + [...partitions].map(([traceId, partition]) => + appendFile( + join(this.directory, `${sanitize(traceId)}${OTLP_EXT}`), + JSON.stringify(partition) + "\n", + ), + ), + ); + } + + /** List traces newest-first, filtered by service name and time range (default: last 12 hours). */ + public async list(options: ListTracesOptions = {}): Promise { + const now = Date.now(); + const start = options.startTime ?? now - DEFAULT_LIST_WINDOW_MS; + const end = options.endTime ?? now; + + const summaries: TraceSummary[] = []; + for (const file of await this.traceFiles()) { + const trace = await this.readTraceFile(file); + if (!trace) continue; + + const meta = extractTraceMeta(trace.resourceSpans, trace.resourceLogs); + if (!meta.traceId) continue; + if (meta.lastSeen < start || meta.firstSeen > end) continue; + if (options.serviceName && meta.serviceName !== options.serviceName) continue; + + summaries.push({ + traceId: meta.traceId, + timestamp: new Date(meta.lastSeen).toISOString(), + sessionId: meta.sessionId, + spanCount: String(meta.spanCount), + ...buildTraceDetail(trace.resourceSpans, trace.resourceLogs), + }); + } + + return summaries.sort((a, b) => b.timestamp.localeCompare(a.timestamp)); + } + + /** All spans and logs for one trace, or undefined when the trace is unknown. */ + public async get(traceId: string): Promise { + const trace = await this.readTraceFile(`${sanitize(traceId)}${OTLP_EXT}`); + if (!trace) return undefined; + return buildTraceDetail(trace.resourceSpans, trace.resourceLogs); + } + + private async traceFiles(): Promise { + try { + return (await readdir(this.directory)).filter((file) => file.endsWith(OTLP_EXT)); + } catch { + return []; + } + } + + private async readTraceFile( + fileName: string, + ): Promise<{ resourceSpans: OtlpResourceSpan[]; resourceLogs: OtlpResourceLog[] } | undefined> { + let content: string; + try { + content = await readFile(join(this.directory, fileName), "utf8"); + } catch { + return undefined; + } + + const resourceSpans: OtlpResourceSpan[] = []; + const resourceLogs: OtlpResourceLog[] = []; + for (const line of content.split("\n")) { + if (!line.trim()) continue; + try { + const payload = JSON.parse(line) as OtlpPayload; + if (payload.resourceSpans) resourceSpans.push(...payload.resourceSpans); + if (payload.resourceLogs) resourceLogs.push(...payload.resourceLogs); + } catch { + // Skip malformed lines — a partially written line must not break reads. + } + } + return { resourceSpans, resourceLogs }; + } +} + +function sanitize(value: string): string { + return value.replace(/[^a-zA-Z0-9_-]/g, "_"); +} diff --git a/src/core/dev/otel/transforms.test.ts b/src/core/dev/otel/transforms.test.ts new file mode 100644 index 000000000..0b3bc2eef --- /dev/null +++ b/src/core/dev/otel/transforms.test.ts @@ -0,0 +1,243 @@ +import { describe, expect, test } from "bun:test"; +import { + buildTraceDetail, + extractAnyValue, + extractTraceMeta, + flattenAttributes, + hexFromB64OrString, + nanoToMs, + partitionByTraceId, +} from "./transforms"; +import type { OtlpResourceLog, OtlpResourceSpan } from "./types"; + +const TRACE_ID_HEX = "0123456789abcdef0123456789abcdef"; +const TRACE_ID_B64 = Buffer.from(TRACE_ID_HEX, "hex").toString("base64"); +const SPAN_ID_HEX = "0123456789abcdef"; + +function resourceSpan(overrides: { serviceName?: string; spans: object[] }): OtlpResourceSpan { + return { + resource: overrides.serviceName + ? { attributes: [{ key: "service.name", value: { stringValue: overrides.serviceName } }] } + : undefined, + scopeSpans: [{ scope: { name: "test-scope" }, spans: overrides.spans }], + }; +} + +const agentSpan = { + traceId: TRACE_ID_B64, + spanId: SPAN_ID_HEX, + name: "invoke_agent strands", + kind: 1, + startTimeUnixNano: "1700000000000000000", + endTimeUnixNano: "1700000001500000000", + attributes: [ + { key: "gen_ai.prompt", value: { stringValue: "hello" } }, + { key: "session.id", value: { stringValue: "session-1" } }, + ], +}; + +describe("extractTraceMeta", () => { + test("collects trace id, time bounds, session, service, and span count", () => { + const meta = extractTraceMeta( + [resourceSpan({ serviceName: "my-agent", spans: [agentSpan] })], + [], + ); + expect(meta).toEqual({ + traceId: TRACE_ID_HEX, + firstSeen: 1700000000000, + lastSeen: 1700000001500, + sessionId: "session-1", + serviceName: "my-agent", + spanCount: 1, + }); + }); + + test("counts log records and falls back to observed time", () => { + const logs: OtlpResourceLog[] = [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "log-agent" } }] }, + scopeLogs: [ + { + scope: {}, + logRecords: [{ traceId: TRACE_ID_HEX, observedTimeUnixNano: "1700000002000000000" }], + }, + ], + }, + ]; + const meta = extractTraceMeta([], logs); + expect(meta.traceId).toBe(TRACE_ID_HEX); + expect(meta.serviceName).toBe("log-agent"); + expect(meta.spanCount).toBe(1); + expect(meta.firstSeen).toBe(1700000002000); + expect(meta.lastSeen).toBe(1700000002000); + }); + + test("defaults time bounds to now when no timestamps exist", () => { + const before = Date.now(); + const meta = extractTraceMeta([], []); + expect(meta.firstSeen).toBeGreaterThanOrEqual(before); + expect(meta.lastSeen).toBeGreaterThanOrEqual(before); + expect(meta.traceId).toBeUndefined(); + }); +}); + +describe("partitionByTraceId", () => { + const OTHER_TRACE_HEX = "ffffffffffffffffffffffffffffffff"; + + test("splits a batch carrying several traces into per-trace payloads", () => { + const otherSpan = { ...agentSpan, traceId: OTHER_TRACE_HEX, name: "tool_use" }; + const partitions = partitionByTraceId({ + resourceSpans: [resourceSpan({ serviceName: "svc", spans: [agentSpan, otherSpan] })], + }); + + expect([...partitions.keys()].sort()).toEqual([TRACE_ID_HEX, OTHER_TRACE_HEX]); + const first = partitions.get(TRACE_ID_HEX)!.resourceSpans![0] as OtlpResourceSpan; + expect(first.scopeSpans![0]!.spans).toEqual([agentSpan]); + expect(first.resource).toBeDefined(); + const second = partitions.get(OTHER_TRACE_HEX)!.resourceSpans![0] as OtlpResourceSpan; + expect(second.scopeSpans![0]!.spans).toEqual([otherSpan]); + }); + + test("partitions log records by trace id and keys base64 ids as hex", () => { + const partitions = partitionByTraceId({ + resourceLogs: [ + { + scopeLogs: [ + { + scope: {}, + logRecords: [{ traceId: TRACE_ID_B64 }, { traceId: OTHER_TRACE_HEX }], + }, + ], + }, + ], + }); + + expect([...partitions.keys()].sort()).toEqual([TRACE_ID_HEX, OTHER_TRACE_HEX]); + }); + + test("drops spans without a trace id and returns empty for empty payloads", () => { + expect(partitionByTraceId({}).size).toBe(0); + const partitions = partitionByTraceId({ + resourceSpans: [resourceSpan({ spans: [{ name: "orphan" }] })], + }); + expect(partitions.size).toBe(0); + }); +}); + +describe("buildTraceDetail", () => { + test("hexes ids, flattens attributes, and unwraps log bodies", () => { + const detail = buildTraceDetail( + [resourceSpan({ serviceName: "svc", spans: [agentSpan] })], + [ + { + resource: { attributes: [{ key: "service.name", value: { stringValue: "svc" } }] }, + scopeLogs: [ + { + scope: {}, + logRecords: [ + { traceId: TRACE_ID_B64, spanId: SPAN_ID_HEX, body: { stringValue: "log line" } }, + ], + }, + ], + }, + ], + ); + + const spans = detail.resourceSpans as { + resource: { attributes: Record }; + scopeSpans: { spans: { traceId: string; attributes: Record }[] }[]; + }[]; + expect(spans[0]!.resource.attributes).toEqual({ "service.name": "svc" }); + expect(spans[0]!.scopeSpans[0]!.spans[0]!.traceId).toBe(TRACE_ID_HEX); + expect(spans[0]!.scopeSpans[0]!.spans[0]!.attributes).toEqual({ + "gen_ai.prompt": "hello", + "session.id": "session-1", + }); + + const logs = detail.resourceLogs as { + scopeLogs: { logRecords: { traceId: string; body: unknown }[] }[]; + }[]; + expect(logs[0]!.scopeLogs[0]!.logRecords[0]!.traceId).toBe(TRACE_ID_HEX); + expect(logs[0]!.scopeLogs[0]!.logRecords[0]!.body).toBe("log line"); + }); + + test("filters transport noise but keeps meaningful spans", () => { + const noiseSpans = [ + { name: "GET / http send", attributes: [] }, + { + name: "http.request", + attributes: [{ key: "asgi.event.type", value: { stringValue: "http.request" } }], + }, + { name: "POST", kind: 3, attributes: [] }, + { + name: "POST /invocations", + kind: 2, + attributes: [{ key: "http.method", value: { stringValue: "POST" } }], + }, + ]; + const detail = buildTraceDetail([resourceSpan({ spans: [...noiseSpans, agentSpan] })], []); + const spans = detail.resourceSpans as { scopeSpans: { spans: { name: string }[] }[] }[]; + expect(spans[0]!.scopeSpans[0]!.spans.map((span) => span.name)).toEqual([ + "invoke_agent strands", + ]); + }); + + test("string span kinds from JSON ingest are normalized before filtering", () => { + const detail = buildTraceDetail( + [resourceSpan({ spans: [{ name: "POST", kind: "SPAN_KIND_CLIENT", attributes: [] }] })], + [], + ); + expect(detail.resourceSpans).toBeUndefined(); + }); + + test("returns undefined sections when everything is filtered or empty", () => { + expect(buildTraceDetail([], [])).toEqual({ resourceSpans: undefined, resourceLogs: undefined }); + }); +}); + +describe("helpers", () => { + test("nanoToMs converts and handles absence", () => { + expect(nanoToMs("1700000000123456789")).toBe(1700000000123); + expect(nanoToMs(undefined)).toBe(0); + }); + + test("hexFromB64OrString accepts hex, base64, and empty", () => { + expect(hexFromB64OrString(TRACE_ID_HEX.toUpperCase())).toBe(TRACE_ID_HEX); + expect(hexFromB64OrString(TRACE_ID_B64)).toBe(TRACE_ID_HEX); + expect(hexFromB64OrString(undefined)).toBe(""); + }); + + test("flattenAttributes handles typed values, arrays, and flat passthrough", () => { + expect( + flattenAttributes([ + { key: "s", value: { stringValue: "x" } }, + { key: "i", value: { intValue: "42" } }, + { key: "d", value: { doubleValue: 1.5 } }, + { key: "b", value: { boolValue: true } }, + { key: "a", value: { arrayValue: { values: [{ stringValue: "y" }, { intValue: "7" }] } } }, + { key: "skipped" }, + ]), + ).toEqual({ s: "x", i: 42, d: 1.5, b: true, a: ["y", "7"] }); + expect(flattenAttributes({ already: "flat" })).toEqual({ already: "flat" }); + expect(flattenAttributes([])).toBeUndefined(); + expect(flattenAttributes(undefined)).toBeUndefined(); + }); + + test("extractAnyValue unwraps nested kvlist and array values", () => { + expect( + extractAnyValue({ + kvlistValue: { + values: [ + { + key: "nested", + value: { arrayValue: { values: [{ intValue: "1" }, { boolValue: false }] } }, + }, + { key: "plain", value: { stringValue: "v" } }, + ], + }, + }), + ).toEqual({ nested: [1, false], plain: "v" }); + expect(extractAnyValue("passthrough")).toBe("passthrough"); + expect(extractAnyValue(null)).toBeNull(); + }); +}); diff --git a/src/core/dev/otel/transforms.ts b/src/core/dev/otel/transforms.ts new file mode 100644 index 000000000..1c161d5e0 --- /dev/null +++ b/src/core/dev/otel/transforms.ts @@ -0,0 +1,311 @@ +import type { + OtlpAttributes, + OtlpAttributeValue, + OtlpPayload, + OtlpResource, + OtlpResourceLog, + OtlpResourceSpan, +} from "./types"; + +export interface TraceMeta { + traceId?: string; + firstSeen: number; + lastSeen: number; + sessionId?: string; + serviceName?: string; + spanCount: number; +} + +/** Extract listing metadata (trace id, time bounds, session, service, count) from raw OTLP arrays. */ +export function extractTraceMeta( + resourceSpans: OtlpResourceSpan[], + resourceLogs: OtlpResourceLog[], +): TraceMeta { + const meta: TraceMeta = { firstSeen: Infinity, lastSeen: 0, spanCount: 0 }; + + for (const resourceSpan of resourceSpans) { + meta.serviceName ??= getResourceAttribute(resourceSpan.resource, "service.name"); + for (const scopeSpan of resourceSpan.scopeSpans ?? []) { + for (const span of scopeSpan.spans ?? []) { + meta.spanCount++; + meta.traceId ??= hexFromB64OrString(span.traceId) || undefined; + widenTimeBounds(meta, nanoToMs(span.startTimeUnixNano)); + widenTimeBounds(meta, nanoToMs(span.endTimeUnixNano)); + meta.sessionId ??= + getAttributeValue(span.attributes, "session.id") ?? + getAttributeValue(span.attributes, "attributes.session.id"); + } + } + } + + for (const resourceLog of resourceLogs) { + meta.serviceName ??= getResourceAttribute(resourceLog.resource, "service.name"); + for (const scopeLog of resourceLog.scopeLogs ?? []) { + for (const record of scopeLog.logRecords ?? []) { + meta.spanCount++; + meta.traceId ??= hexFromB64OrString(record.traceId) || undefined; + widenTimeBounds( + meta, + nanoToMs(record.timeUnixNano) || nanoToMs(record.observedTimeUnixNano), + ); + } + } + } + + const now = Date.now(); + if (meta.firstSeen === Infinity) meta.firstSeen = now; + if (meta.lastSeen === 0) meta.lastSeen = now; + return meta; +} + +/** + * Split one OTLP export payload into per-trace payloads, keyed by hex trace id. + * A single export batch routinely carries spans from several traces (SDKs batch + * by time, not by trace), so persistence must not attribute a whole batch to + * the first trace id it sees. Spans and log records without a trace id are dropped. + * Resource and scope structure is preserved within each partition. + */ +export function partitionByTraceId(payload: OtlpPayload): Map { + const partitions = new Map(); + const partition = (traceId: string): OtlpPayload => { + let entry = partitions.get(traceId); + if (!entry) { + entry = {}; + partitions.set(traceId, entry); + } + return entry; + }; + + for (const resourceSpan of payload.resourceSpans ?? []) { + for (const scopeSpan of resourceSpan.scopeSpans ?? []) { + const byTrace = groupBy(scopeSpan.spans ?? [], (span) => hexFromB64OrString(span.traceId)); + for (const [traceId, spans] of byTrace) { + (partition(traceId).resourceSpans ??= []).push({ + resource: resourceSpan.resource, + scopeSpans: [{ scope: scopeSpan.scope, spans }], + }); + } + } + } + + for (const resourceLog of payload.resourceLogs ?? []) { + for (const scopeLog of resourceLog.scopeLogs ?? []) { + const byTrace = groupBy(scopeLog.logRecords ?? [], (record) => + hexFromB64OrString(record.traceId), + ); + for (const [traceId, logRecords] of byTrace) { + (partition(traceId).resourceLogs ??= []).push({ + resource: resourceLog.resource, + scopeLogs: [{ scope: scopeLog.scope, logRecords }], + }); + } + } + } + + return partitions; +} + +function groupBy(items: T[], key: (item: T) => string): Map { + const groups = new Map(); + for (const item of items) { + const groupKey = key(item); + if (!groupKey) continue; + const group = groups.get(groupKey); + if (group) group.push(item); + else groups.set(groupKey, [item]); + } + return groups; +} + +/** + * Build frontend-ready trace detail from raw OTLP arrays: ids to hex, attributes + * flattened to plain records, transport-noise spans dropped, log bodies unwrapped. + */ +export function buildTraceDetail( + resourceSpans: OtlpResourceSpan[], + resourceLogs: OtlpResourceLog[], +): { resourceSpans?: unknown[]; resourceLogs?: unknown[] } { + const spans = resourceSpans + .map((resourceSpan) => ({ + resource: resourceSpan.resource + ? { attributes: flattenAttributes(resourceSpan.resource.attributes) } + : undefined, + scopeSpans: resourceSpan.scopeSpans + ?.map((scopeSpan) => ({ + scope: scopeSpan.scope, + spans: scopeSpan.spans + ?.map((span) => ({ + ...span, + traceId: hexFromB64OrString(span.traceId), + spanId: hexFromB64OrString(span.spanId), + parentSpanId: hexFromB64OrString(span.parentSpanId), + attributes: flattenAttributes(span.attributes), + })) + .filter((span) => isMeaningfulSpan(span)), + })) + .filter((scopeSpan) => scopeSpan.spans && scopeSpan.spans.length > 0), + })) + .filter((resourceSpan) => resourceSpan.scopeSpans && resourceSpan.scopeSpans.length > 0); + + const logs = resourceLogs + .map((resourceLog) => ({ + resource: resourceLog.resource + ? { attributes: flattenAttributes(resourceLog.resource.attributes) } + : undefined, + scopeLogs: resourceLog.scopeLogs?.map((scopeLog) => ({ + scope: scopeLog.scope, + logRecords: scopeLog.logRecords?.map((record) => ({ + ...record, + traceId: hexFromB64OrString(record.traceId), + spanId: hexFromB64OrString(record.spanId), + body: record.body === undefined ? undefined : extractAnyValue(record.body), + attributes: flattenAttributes(record.attributes), + })), + })), + })) + .filter((resourceLog) => resourceLog.scopeLogs && resourceLog.scopeLogs.length > 0); + + return { + resourceSpans: spans.length > 0 ? spans : undefined, + resourceLogs: logs.length > 0 ? logs : undefined, + }; +} + +/** + * Whether a span carries application-level signal. Filters ASGI transport events, + * bare HTTP client/server noise, and other framework spans that add nothing in the UI. + */ +function isMeaningfulSpan(span: { + name?: string; + kind?: number | string; + attributes?: Record; +}): boolean { + const name = span.name ?? ""; + const attributes = span.attributes ?? {}; + const kind = normalizeSpanKind(span.kind); + + if (name.endsWith(" http send") || name.endsWith(" http receive")) return false; + if (attributes["asgi.event.type"]) return false; + if (Object.keys(attributes).some((key) => key.startsWith("gen_ai."))) return true; + if (attributes["rpc.system"] || attributes["rpc.method"]) return true; + + const scopeHints = ["strands", "bedrock", "langchain", "crewai", "autogen", "google_adk"]; + if (scopeHints.some((hint) => name.toLowerCase().includes(hint))) return true; + if (name === "tool_use" || name === "tool_call" || attributes["tool.name"]) return true; + + if (kind === SPAN_KIND.CLIENT && (name === "POST" || name === "GET" || name.startsWith("HTTP "))) + return false; + if (kind === SPAN_KIND.SERVER && name.startsWith("POST /") && attributes["http.method"]) + return false; + + return true; +} + +const SPAN_KIND = { INTERNAL: 1, SERVER: 2, CLIENT: 3, PRODUCER: 4, CONSUMER: 5 } as const; + +/** Normalize a span kind from its protobuf enum name or number to the numeric value. */ +function normalizeSpanKind(kind: number | string | undefined): number { + if (typeof kind === "number") return kind; + if (typeof kind === "string") { + const name = kind.replace(/^SPAN_KIND_/, "") as keyof typeof SPAN_KIND; + return SPAN_KIND[name] ?? 0; + } + return 0; +} + +/** Convert a nanosecond timestamp string to milliseconds (0 when absent). */ +export function nanoToMs(nano: string | undefined): number { + if (!nano) return 0; + return Math.floor(Number(nano) / 1_000_000); +} + +/** + * Normalize a trace/span id that may be base64 (protobuf JSON conversion) or + * already hex (JSON ingest) into lowercase hex. + */ +export function hexFromB64OrString(value: string | undefined): string { + if (!value) return ""; + if (/^[0-9a-f]+$/i.test(value) && (value.length === 32 || value.length === 16)) + return value.toLowerCase(); + try { + return Buffer.from(value, "base64").toString("hex"); + } catch { + return value; + } +} + +/** Flatten OTLP attributes into a plain record; passes already-flat records through. */ +export function flattenAttributes( + attributes: OtlpAttributes | undefined, +): Record | undefined { + if (!attributes) return undefined; + if (!Array.isArray(attributes)) return attributes; + if (attributes.length === 0) return undefined; + + const flat: Record = {}; + for (const attribute of attributes) { + if (!attribute.value) continue; + const value = attribute.value; + if (value.stringValue !== undefined) flat[attribute.key] = value.stringValue; + else if (value.intValue !== undefined) flat[attribute.key] = Number(value.intValue); + else if (value.doubleValue !== undefined) flat[attribute.key] = value.doubleValue; + else if (value.boolValue !== undefined) flat[attribute.key] = value.boolValue; + else if (value.arrayValue?.values) { + flat[attribute.key] = value.arrayValue.values.map( + (item: OtlpAttributeValue) => + item.stringValue ?? item.intValue ?? item.doubleValue ?? item.boolValue ?? null, + ); + } + } + return flat; +} + +/** Unwrap an OTLP AnyValue (string/int/double/bool/array/kvlist) into a plain value. */ +export function extractAnyValue(value: unknown): unknown { + if (!value || typeof value !== "object") return value; + const anyValue = value as Record; + if (anyValue.stringValue !== undefined) return anyValue.stringValue; + if (anyValue.intValue !== undefined) return Number(anyValue.intValue); + if (anyValue.doubleValue !== undefined) return anyValue.doubleValue; + if (anyValue.boolValue !== undefined) return anyValue.boolValue; + if (anyValue.arrayValue && typeof anyValue.arrayValue === "object") { + const { values } = anyValue.arrayValue as { values?: unknown[] }; + return (values ?? []).map(extractAnyValue); + } + if (anyValue.kvlistValue && typeof anyValue.kvlistValue === "object") { + const { values } = anyValue.kvlistValue as { values?: { key: string; value?: unknown }[] }; + const record: Record = {}; + for (const entry of values ?? []) { + record[entry.key] = entry.value === undefined ? undefined : extractAnyValue(entry.value); + } + return record; + } + return value; +} + +function getResourceAttribute(resource: OtlpResource | undefined, key: string): string | undefined { + return getAttributeValue(resource?.attributes, key); +} + +function getAttributeValue( + attributes: OtlpAttributes | undefined, + key: string, +): string | undefined { + if (!attributes) return undefined; + if (Array.isArray(attributes)) { + const attribute = attributes.find((entry) => entry.key === key); + if (!attribute?.value) return undefined; + return ( + attribute.value.stringValue ?? + (attribute.value.intValue != null ? String(attribute.value.intValue) : undefined) + ); + } + const value = attributes[key]; + return typeof value === "string" ? value : undefined; +} + +function widenTimeBounds(meta: TraceMeta, timeMs: number): void { + if (!timeMs) return; + if (timeMs < meta.firstSeen) meta.firstSeen = timeMs; + if (timeMs > meta.lastSeen) meta.lastSeen = timeMs; +} diff --git a/src/core/dev/otel/types.ts b/src/core/dev/otel/types.ts new file mode 100644 index 000000000..a458de9d4 --- /dev/null +++ b/src/core/dev/otel/types.ts @@ -0,0 +1,65 @@ +/** + * Wire shapes for OTLP/HTTP payloads after protobuf JSON conversion or JSON ingest. + * Attributes appear either as OTLP key/value arrays (from the SDK exporters) or as + * already-flat records (after our own flattening) — helpers accept both. + */ + +export interface OtlpAttributeValue { + stringValue?: string; + intValue?: string; + doubleValue?: number; + boolValue?: boolean; + arrayValue?: { values?: OtlpAttributeValue[] }; + kvlistValue?: { values?: OtlpAttribute[] }; +} + +export interface OtlpAttribute { + key: string; + value?: OtlpAttributeValue; +} + +export type OtlpAttributes = OtlpAttribute[] | Record; + +export interface OtlpResource { + attributes?: OtlpAttributes; +} + +export interface OtlpSpan { + traceId?: string; + spanId?: string; + parentSpanId?: string; + name?: string; + kind?: number | string; + startTimeUnixNano?: string; + endTimeUnixNano?: string; + attributes?: OtlpAttributes; + status?: { code?: number; message?: string }; + events?: unknown[]; +} + +export interface OtlpResourceSpan { + resource?: OtlpResource; + scopeSpans?: { scope?: { name?: string; version?: string }; spans?: OtlpSpan[] }[]; +} + +export interface OtlpLogRecord { + timeUnixNano?: string; + observedTimeUnixNano?: string; + severityNumber?: number; + severityText?: string; + body?: unknown; + attributes?: OtlpAttributes; + traceId?: string; + spanId?: string; +} + +export interface OtlpResourceLog { + resource?: OtlpResource; + scopeLogs?: { scope?: { name?: string; version?: string }; logRecords?: OtlpLogRecord[] }[]; +} + +/** One OTLP export payload: what a single POST /v1/traces or /v1/logs carries. */ +export interface OtlpPayload { + resourceSpans?: OtlpResourceSpan[]; + resourceLogs?: OtlpResourceLog[]; +} From fad9c3bd33c5864739341448190a9b66192ab7ed Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Wed, 19 Aug 2026 15:27:46 -0400 Subject: [PATCH 2/4] fix(dev): match trace filters against every participating service A distributed trace spans several local agents whose exports append to the same trace file; filtering by any participant must find it, not only the first service seen. --- src/core/dev/otel/store.test.ts | 13 ++++++++++--- src/core/dev/otel/store.ts | 2 +- src/core/dev/otel/transforms.test.ts | 4 ++-- src/core/dev/otel/transforms.ts | 13 +++++++++---- 4 files changed, 22 insertions(+), 10 deletions(-) diff --git a/src/core/dev/otel/store.test.ts b/src/core/dev/otel/store.test.ts index f4bb450c7..78b77abe9 100644 --- a/src/core/dev/otel/store.test.ts +++ b/src/core/dev/otel/store.test.ts @@ -90,11 +90,18 @@ describe("TraceStore", () => { expect(await store.get(TRACE_B)).toBeDefined(); }); - test("list filters by service name", async () => { + test("list filters by service name, matching every participant of a distributed trace", async () => { await store.append(payload(TRACE_A, { serviceName: "agent-1" })); + // agent-2 contributes spans to the SAME trace (distributed) and owns its own trace. + await store.append(payload(TRACE_A, { serviceName: "agent-2", name: "tool_use" })); await store.append(payload(TRACE_B, { serviceName: "agent-2" })); - const traces = await store.list({ serviceName: "agent-2" }); - expect(traces.map((trace) => trace.traceId)).toEqual([TRACE_B]); + + expect((await store.list({ serviceName: "agent-2" })).map((t) => t.traceId).sort()).toEqual([ + TRACE_A, + TRACE_B, + ]); + expect((await store.list({ serviceName: "agent-1" })).map((t) => t.traceId)).toEqual([TRACE_A]); + expect(await store.list({ serviceName: "agent-3" })).toEqual([]); }); test("list filters by time window and sorts newest first", async () => { diff --git a/src/core/dev/otel/store.ts b/src/core/dev/otel/store.ts index 57e7e4593..67bbcef69 100644 --- a/src/core/dev/otel/store.ts +++ b/src/core/dev/otel/store.ts @@ -69,7 +69,7 @@ export class TraceStore { const meta = extractTraceMeta(trace.resourceSpans, trace.resourceLogs); if (!meta.traceId) continue; if (meta.lastSeen < start || meta.firstSeen > end) continue; - if (options.serviceName && meta.serviceName !== options.serviceName) continue; + if (options.serviceName && !meta.serviceNames.includes(options.serviceName)) continue; summaries.push({ traceId: meta.traceId, diff --git a/src/core/dev/otel/transforms.test.ts b/src/core/dev/otel/transforms.test.ts index 0b3bc2eef..efdfc88ac 100644 --- a/src/core/dev/otel/transforms.test.ts +++ b/src/core/dev/otel/transforms.test.ts @@ -47,7 +47,7 @@ describe("extractTraceMeta", () => { firstSeen: 1700000000000, lastSeen: 1700000001500, sessionId: "session-1", - serviceName: "my-agent", + serviceNames: ["my-agent"], spanCount: 1, }); }); @@ -66,7 +66,7 @@ describe("extractTraceMeta", () => { ]; const meta = extractTraceMeta([], logs); expect(meta.traceId).toBe(TRACE_ID_HEX); - expect(meta.serviceName).toBe("log-agent"); + expect(meta.serviceNames).toEqual(["log-agent"]); expect(meta.spanCount).toBe(1); expect(meta.firstSeen).toBe(1700000002000); expect(meta.lastSeen).toBe(1700000002000); diff --git a/src/core/dev/otel/transforms.ts b/src/core/dev/otel/transforms.ts index 1c161d5e0..df2f87637 100644 --- a/src/core/dev/otel/transforms.ts +++ b/src/core/dev/otel/transforms.ts @@ -12,7 +12,8 @@ export interface TraceMeta { firstSeen: number; lastSeen: number; sessionId?: string; - serviceName?: string; + /** Every service participating in the trace — a distributed trace spans several local agents. */ + serviceNames: string[]; spanCount: number; } @@ -21,10 +22,12 @@ export function extractTraceMeta( resourceSpans: OtlpResourceSpan[], resourceLogs: OtlpResourceLog[], ): TraceMeta { - const meta: TraceMeta = { firstSeen: Infinity, lastSeen: 0, spanCount: 0 }; + const meta: TraceMeta = { firstSeen: Infinity, lastSeen: 0, spanCount: 0, serviceNames: [] }; + const services = new Set(); for (const resourceSpan of resourceSpans) { - meta.serviceName ??= getResourceAttribute(resourceSpan.resource, "service.name"); + const service = getResourceAttribute(resourceSpan.resource, "service.name"); + if (service) services.add(service); for (const scopeSpan of resourceSpan.scopeSpans ?? []) { for (const span of scopeSpan.spans ?? []) { meta.spanCount++; @@ -39,7 +42,8 @@ export function extractTraceMeta( } for (const resourceLog of resourceLogs) { - meta.serviceName ??= getResourceAttribute(resourceLog.resource, "service.name"); + const service = getResourceAttribute(resourceLog.resource, "service.name"); + if (service) services.add(service); for (const scopeLog of resourceLog.scopeLogs ?? []) { for (const record of scopeLog.logRecords ?? []) { meta.spanCount++; @@ -55,6 +59,7 @@ export function extractTraceMeta( const now = Date.now(); if (meta.firstSeen === Infinity) meta.firstSeen = now; if (meta.lastSeen === 0) meta.lastSeen = now; + meta.serviceNames = [...services]; return meta; } From 4035b7bc6b3229d9a4ad4ebabe69ac6181f9bfea Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 11:20:17 -0400 Subject: [PATCH 3/4] fix(dev): address review on OTLP trace storage - narrow OtlpAttributes to the key/value wire form (the flat-record variant had no producer); drop the dead passthrough branches - flatten array attributes through extractAnyValue so ints stay numeric and nested kvlists survive (was stringifying and dropping them) - count rendered (post-filter) spans for the list summary instead of raw records, so the count matches the waterfall the inspector shows - surface non-ENOENT fs errors from reads instead of masking them as empty - add newest-N limit to list() for the inspector's per-invocation poll --- src/core/dev/otel/store.test.ts | 30 ++++++++++++++++++--- src/core/dev/otel/store.ts | 38 ++++++++++++++++++++++----- src/core/dev/otel/transforms.test.ts | 11 +++----- src/core/dev/otel/transforms.ts | 39 ++++++++++------------------ src/core/dev/otel/types.ts | 6 ++--- 5 files changed, 77 insertions(+), 47 deletions(-) diff --git a/src/core/dev/otel/store.test.ts b/src/core/dev/otel/store.test.ts index 78b77abe9..53d04cc2a 100644 --- a/src/core/dev/otel/store.test.ts +++ b/src/core/dev/otel/store.test.ts @@ -1,5 +1,5 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; -import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { TraceStore } from "./store"; @@ -104,7 +104,7 @@ describe("TraceStore", () => { expect(await store.list({ serviceName: "agent-3" })).toEqual([]); }); - test("list filters by time window and sorts newest first", async () => { + test("list filters by time window, sorts newest first, and caps to limit", async () => { const oldNano = `${BigInt(Date.now() - 24 * 60 * 60 * 1000) * 1_000_000n}`; await store.append(payload(TRACE_A, { startNano: oldNano })); await store.append(payload(TRACE_B)); @@ -113,12 +113,23 @@ describe("TraceStore", () => { const all = await store.list({ startTime: 0 }); expect(all.map((trace) => trace.traceId)).toEqual([TRACE_B, TRACE_A]); + + // limit keeps the newest N after sorting. + expect((await store.list({ startTime: 0, limit: 1 })).map((trace) => trace.traceId)).toEqual([ + TRACE_B, + ]); }); - test("get returns the trace detail or undefined for unknown ids", async () => { + test("get merges spans across appends and is undefined for unknown ids", async () => { await store.append(payload(TRACE_A)); + await store.append(payload(TRACE_A, { name: "tool_use" })); + const detail = await store.get(TRACE_A); - expect(detail?.resourceSpans).toBeDefined(); + const spans = (detail!.resourceSpans as { scopeSpans: { spans: { name: string }[] }[] }[]) + .flatMap((resourceSpan) => resourceSpan.scopeSpans) + .flatMap((scopeSpan) => scopeSpan.spans); + expect(spans.map((span) => span.name).sort()).toEqual(["invoke_agent strands", "tool_use"]); + expect(await store.get(TRACE_B)).toBeUndefined(); }); @@ -139,4 +150,15 @@ describe("TraceStore", () => { expect(await empty.list()).toEqual([]); expect(await empty.get(TRACE_A)).toBeUndefined(); }); + + test("non-ENOENT fs errors bubble up rather than reading as empty", async () => { + // readdir on a path that is a file, not a directory -> ENOTDIR must throw. + const asFile = join(directory, "file"); + await writeFile(asFile, "x"); + expect(new TraceStore(asFile).list()).rejects.toThrow(); + + // readFile on a trace path that is a directory -> EISDIR must throw. + await mkdir(join(directory, `${TRACE_A}.otlp.jsonl`)); + expect(store.list()).rejects.toThrow(); + }); }); diff --git a/src/core/dev/otel/store.ts b/src/core/dev/otel/store.ts index 67bbcef69..2d6c0ea00 100644 --- a/src/core/dev/otel/store.ts +++ b/src/core/dev/otel/store.ts @@ -24,6 +24,8 @@ export interface ListTracesOptions { serviceName?: string; startTime?: number; endTime?: number; + /** Keep only the newest N traces — the inspector re-polls this on every invocation. */ + limit?: number; } /** @@ -67,20 +69,24 @@ export class TraceStore { if (!trace) continue; const meta = extractTraceMeta(trace.resourceSpans, trace.resourceLogs); + // No id means every line failed to parse (empty/corrupt file), not a real trace. if (!meta.traceId) continue; if (meta.lastSeen < start || meta.firstSeen > end) continue; if (options.serviceName && !meta.serviceNames.includes(options.serviceName)) continue; + const detail = buildTraceDetail(trace.resourceSpans, trace.resourceLogs); summaries.push({ traceId: meta.traceId, timestamp: new Date(meta.lastSeen).toISOString(), sessionId: meta.sessionId, - spanCount: String(meta.spanCount), - ...buildTraceDetail(trace.resourceSpans, trace.resourceLogs), + // Count the spans the UI actually renders (post noise-filter), not raw records. + spanCount: String(countRenderedSpans(detail.resourceSpans)), + ...detail, }); } - return summaries.sort((a, b) => b.timestamp.localeCompare(a.timestamp)); + summaries.sort((a, b) => b.timestamp.localeCompare(a.timestamp)); + return options.limit === undefined ? summaries : summaries.slice(0, options.limit); } /** All spans and logs for one trace, or undefined when the trace is unknown. */ @@ -93,8 +99,9 @@ export class TraceStore { private async traceFiles(): Promise { try { return (await readdir(this.directory)).filter((file) => file.endsWith(OTLP_EXT)); - } catch { - return []; + } catch (error) { + if (isNotFound(error)) return []; // No traces persisted yet — the dir is created on first append. + throw error; } } @@ -104,8 +111,11 @@ export class TraceStore { let content: string; try { content = await readFile(join(this.directory, fileName), "utf8"); - } catch { - return undefined; + } catch (error) { + // Unknown trace (get) or a file removed between listing and read; any other + // fault (permissions, bad path) is real and must not read as "no trace". + if (isNotFound(error)) return undefined; + throw error; } const resourceSpans: OtlpResourceSpan[] = []; @@ -127,3 +137,17 @@ export class TraceStore { function sanitize(value: string): string { return value.replace(/[^a-zA-Z0-9_-]/g, "_"); } + +/** Number of spans in a built trace detail — what the inspector's waterfall shows. */ +function countRenderedSpans(resourceSpans: TraceDetail["resourceSpans"]): number { + let count = 0; + for (const resourceSpan of (resourceSpans ?? []) as OtlpResourceSpan[]) { + for (const scopeSpan of resourceSpan.scopeSpans ?? []) count += scopeSpan.spans?.length ?? 0; + } + return count; +} + +/** A missing directory or file — the only fs error reads should treat as "empty". */ +function isNotFound(error: unknown): boolean { + return (error as NodeJS.ErrnoException | null)?.code === "ENOENT"; +} diff --git a/src/core/dev/otel/transforms.test.ts b/src/core/dev/otel/transforms.test.ts index efdfc88ac..b506c84be 100644 --- a/src/core/dev/otel/transforms.test.ts +++ b/src/core/dev/otel/transforms.test.ts @@ -37,7 +37,7 @@ const agentSpan = { }; describe("extractTraceMeta", () => { - test("collects trace id, time bounds, session, service, and span count", () => { + test("collects trace id, time bounds, session, and service", () => { const meta = extractTraceMeta( [resourceSpan({ serviceName: "my-agent", spans: [agentSpan] })], [], @@ -48,11 +48,10 @@ describe("extractTraceMeta", () => { lastSeen: 1700000001500, sessionId: "session-1", serviceNames: ["my-agent"], - spanCount: 1, }); }); - test("counts log records and falls back to observed time", () => { + test("reads trace id, service, and observed time from logs alone", () => { const logs: OtlpResourceLog[] = [ { resource: { attributes: [{ key: "service.name", value: { stringValue: "log-agent" } }] }, @@ -67,7 +66,6 @@ describe("extractTraceMeta", () => { const meta = extractTraceMeta([], logs); expect(meta.traceId).toBe(TRACE_ID_HEX); expect(meta.serviceNames).toEqual(["log-agent"]); - expect(meta.spanCount).toBe(1); expect(meta.firstSeen).toBe(1700000002000); expect(meta.lastSeen).toBe(1700000002000); }); @@ -207,7 +205,7 @@ describe("helpers", () => { expect(hexFromB64OrString(undefined)).toBe(""); }); - test("flattenAttributes handles typed values, arrays, and flat passthrough", () => { + test("flattenAttributes handles typed values and arrays, empty for none", () => { expect( flattenAttributes([ { key: "s", value: { stringValue: "x" } }, @@ -217,8 +215,7 @@ describe("helpers", () => { { key: "a", value: { arrayValue: { values: [{ stringValue: "y" }, { intValue: "7" }] } } }, { key: "skipped" }, ]), - ).toEqual({ s: "x", i: 42, d: 1.5, b: true, a: ["y", "7"] }); - expect(flattenAttributes({ already: "flat" })).toEqual({ already: "flat" }); + ).toEqual({ s: "x", i: 42, d: 1.5, b: true, a: ["y", 7] }); expect(flattenAttributes([])).toBeUndefined(); expect(flattenAttributes(undefined)).toBeUndefined(); }); diff --git a/src/core/dev/otel/transforms.ts b/src/core/dev/otel/transforms.ts index df2f87637..8bedd76ba 100644 --- a/src/core/dev/otel/transforms.ts +++ b/src/core/dev/otel/transforms.ts @@ -1,6 +1,5 @@ import type { OtlpAttributes, - OtlpAttributeValue, OtlpPayload, OtlpResource, OtlpResourceLog, @@ -14,15 +13,14 @@ export interface TraceMeta { sessionId?: string; /** Every service participating in the trace — a distributed trace spans several local agents. */ serviceNames: string[]; - spanCount: number; } -/** Extract listing metadata (trace id, time bounds, session, service, count) from raw OTLP arrays. */ +/** Extract listing metadata (trace id, time bounds, session, service) from raw OTLP arrays. */ export function extractTraceMeta( resourceSpans: OtlpResourceSpan[], resourceLogs: OtlpResourceLog[], ): TraceMeta { - const meta: TraceMeta = { firstSeen: Infinity, lastSeen: 0, spanCount: 0, serviceNames: [] }; + const meta: TraceMeta = { firstSeen: Infinity, lastSeen: 0, serviceNames: [] }; const services = new Set(); for (const resourceSpan of resourceSpans) { @@ -30,7 +28,6 @@ export function extractTraceMeta( if (service) services.add(service); for (const scopeSpan of resourceSpan.scopeSpans ?? []) { for (const span of scopeSpan.spans ?? []) { - meta.spanCount++; meta.traceId ??= hexFromB64OrString(span.traceId) || undefined; widenTimeBounds(meta, nanoToMs(span.startTimeUnixNano)); widenTimeBounds(meta, nanoToMs(span.endTimeUnixNano)); @@ -46,7 +43,6 @@ export function extractTraceMeta( if (service) services.add(service); for (const scopeLog of resourceLog.scopeLogs ?? []) { for (const record of scopeLog.logRecords ?? []) { - meta.spanCount++; meta.traceId ??= hexFromB64OrString(record.traceId) || undefined; widenTimeBounds( meta, @@ -239,13 +235,11 @@ export function hexFromB64OrString(value: string | undefined): string { } } -/** Flatten OTLP attributes into a plain record; passes already-flat records through. */ +/** Flatten an OTLP key/value attribute array into a plain record. */ export function flattenAttributes( attributes: OtlpAttributes | undefined, ): Record | undefined { - if (!attributes) return undefined; - if (!Array.isArray(attributes)) return attributes; - if (attributes.length === 0) return undefined; + if (!attributes || attributes.length === 0) return undefined; const flat: Record = {}; for (const attribute of attributes) { @@ -255,12 +249,9 @@ export function flattenAttributes( else if (value.intValue !== undefined) flat[attribute.key] = Number(value.intValue); else if (value.doubleValue !== undefined) flat[attribute.key] = value.doubleValue; else if (value.boolValue !== undefined) flat[attribute.key] = value.boolValue; - else if (value.arrayValue?.values) { - flat[attribute.key] = value.arrayValue.values.map( - (item: OtlpAttributeValue) => - item.stringValue ?? item.intValue ?? item.doubleValue ?? item.boolValue ?? null, - ); - } + // Arrays (and any nested kvlist within them) share the AnyValue unwrapping below. + else if (value.arrayValue?.values) + flat[attribute.key] = value.arrayValue.values.map(extractAnyValue); } return flat; } @@ -297,16 +288,12 @@ function getAttributeValue( key: string, ): string | undefined { if (!attributes) return undefined; - if (Array.isArray(attributes)) { - const attribute = attributes.find((entry) => entry.key === key); - if (!attribute?.value) return undefined; - return ( - attribute.value.stringValue ?? - (attribute.value.intValue != null ? String(attribute.value.intValue) : undefined) - ); - } - const value = attributes[key]; - return typeof value === "string" ? value : undefined; + const attribute = attributes.find((entry) => entry.key === key); + if (!attribute?.value) return undefined; + return ( + attribute.value.stringValue ?? + (attribute.value.intValue != null ? String(attribute.value.intValue) : undefined) + ); } function widenTimeBounds(meta: TraceMeta, timeMs: number): void { diff --git a/src/core/dev/otel/types.ts b/src/core/dev/otel/types.ts index a458de9d4..2b0375445 100644 --- a/src/core/dev/otel/types.ts +++ b/src/core/dev/otel/types.ts @@ -1,7 +1,7 @@ /** * Wire shapes for OTLP/HTTP payloads after protobuf JSON conversion or JSON ingest. - * Attributes appear either as OTLP key/value arrays (from the SDK exporters) or as - * already-flat records (after our own flattening) — helpers accept both. + * Attributes always arrive as OTLP key/value arrays; we flatten them to plain + * records only for display output, which is never read back through these types. */ export interface OtlpAttributeValue { @@ -18,7 +18,7 @@ export interface OtlpAttribute { value?: OtlpAttributeValue; } -export type OtlpAttributes = OtlpAttribute[] | Record; +export type OtlpAttributes = OtlpAttribute[]; export interface OtlpResource { attributes?: OtlpAttributes; From 50743541c8e6d56b6d348c29e19c32adc9828e75 Mon Sep 17 00:00:00 2001 From: Tejas Kashinath Date: Fri, 21 Aug 2026 11:39:53 -0400 Subject: [PATCH 4/4] fix(dev): flatten kvlist-valued OTLP attributes instead of dropping them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit flattenAttributes hand-rolled a value branch per AnyValue kind and had no kvlistValue case, so an attribute whose value is a kvlist (or anything the chain didn't enumerate) silently vanished. Route every attribute value through extractAnyValue, which already unwraps all variants including kvlist — smaller and complete. Addresses Gitika's review on transforms.ts (reuse extractAnyValue; kvlist must not disappear). --- src/core/dev/otel/transforms.test.ts | 8 ++++++-- src/core/dev/otel/transforms.ts | 11 +++-------- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/src/core/dev/otel/transforms.test.ts b/src/core/dev/otel/transforms.test.ts index b506c84be..180391284 100644 --- a/src/core/dev/otel/transforms.test.ts +++ b/src/core/dev/otel/transforms.test.ts @@ -205,7 +205,7 @@ describe("helpers", () => { expect(hexFromB64OrString(undefined)).toBe(""); }); - test("flattenAttributes handles typed values and arrays, empty for none", () => { + test("flattenAttributes handles typed values, arrays, and kvlist, empty for none", () => { expect( flattenAttributes([ { key: "s", value: { stringValue: "x" } }, @@ -213,9 +213,13 @@ describe("helpers", () => { { key: "d", value: { doubleValue: 1.5 } }, { key: "b", value: { boolValue: true } }, { key: "a", value: { arrayValue: { values: [{ stringValue: "y" }, { intValue: "7" }] } } }, + { + key: "kv", + value: { kvlistValue: { values: [{ key: "inner", value: { intValue: "3" } }] } }, + }, { key: "skipped" }, ]), - ).toEqual({ s: "x", i: 42, d: 1.5, b: true, a: ["y", 7] }); + ).toEqual({ s: "x", i: 42, d: 1.5, b: true, a: ["y", 7], kv: { inner: 3 } }); expect(flattenAttributes([])).toBeUndefined(); expect(flattenAttributes(undefined)).toBeUndefined(); }); diff --git a/src/core/dev/otel/transforms.ts b/src/core/dev/otel/transforms.ts index 8bedd76ba..cef0f5ca1 100644 --- a/src/core/dev/otel/transforms.ts +++ b/src/core/dev/otel/transforms.ts @@ -244,14 +244,9 @@ export function flattenAttributes( const flat: Record = {}; for (const attribute of attributes) { if (!attribute.value) continue; - const value = attribute.value; - if (value.stringValue !== undefined) flat[attribute.key] = value.stringValue; - else if (value.intValue !== undefined) flat[attribute.key] = Number(value.intValue); - else if (value.doubleValue !== undefined) flat[attribute.key] = value.doubleValue; - else if (value.boolValue !== undefined) flat[attribute.key] = value.boolValue; - // Arrays (and any nested kvlist within them) share the AnyValue unwrapping below. - else if (value.arrayValue?.values) - flat[attribute.key] = value.arrayValue.values.map(extractAnyValue); + // One unwrap for every AnyValue variant — string/int/double/bool/array/kvlist — + // so nested and kvlist-valued attributes flatten instead of silently dropping. + flat[attribute.key] = extractAnyValue(attribute.value); } return flat; }