feat(dev): OTLP trace storage for local dev

src/core/dev/otel/store.test.ts

@@ -0,0 +1,164 @@
+import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import { mkdir, 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, 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" }));
+
+    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, 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));
+
+    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]);
+
+    // limit keeps the newest N after sorting.
+    expect((await store.list({ startTime: 0, limit: 1 })).map((trace) => trace.traceId)).toEqual([
+      TRACE_B,
+    ]);
+  });
+
+  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);
+    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();
+  });
+
+  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();
+  });
+
+  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();
+  });
+});

src/core/dev/otel/store.ts

@@ -0,0 +1,153 @@
+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;
+  /** Keep only the newest N traces — the inspector re-polls this on every invocation. */
+  limit?: 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<void> {
+    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<TraceSummary[]> {
+    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);
+      // 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,
+        // Count the spans the UI actually renders (post noise-filter), not raw records.
+        spanCount: String(countRenderedSpans(detail.resourceSpans)),
+        ...detail,
+      });
+    }
+
+    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. */
+  public async get(traceId: string): Promise<TraceDetail | undefined> {
+    const trace = await this.readTraceFile(`${sanitize(traceId)}${OTLP_EXT}`);
+    if (!trace) return undefined;
+    return buildTraceDetail(trace.resourceSpans, trace.resourceLogs);
+  }
+
+  private async traceFiles(): Promise<string[]> {
+    try {
+      return (await readdir(this.directory)).filter((file) => file.endsWith(OTLP_EXT));
+    } catch (error) {
+      if (isNotFound(error)) return []; // No traces persisted yet — the dir is created on first append.
+      throw error;
+    }
+  }
+
+  private async readTraceFile(
+    fileName: string,
+  ): Promise<{ resourceSpans: OtlpResourceSpan[]; resourceLogs: OtlpResourceLog[] } | undefined> {
+    let content: string;
+    try {
+      content = await readFile(join(this.directory, fileName), "utf8");
+    } 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[] = [];
+    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, "_");
+}
+
+/** 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";
+}