feat(eve): build the OTel pipeline from a declared value

.changeset/otel-pipeline-primitive.md

@@ -0,0 +1,5 @@
+---
+"eve": patch
+---
+
+eve now builds its OpenTelemetry pipeline from a declared `otel()` value with the local trace spool as an ordinary span processor, instead of registering the provider and the spool as one unit. Internal groundwork for authored instrumentation providers; the spans and attributes eve records are unchanged.

packages/eve/scripts/vendor-compiled/declarations/@opentelemetry/api.d.ts

@@ -55,8 +55,13 @@ export interface TextMapGetter<Carrier = unknown> {
   keys(carrier: Carrier): string[];
 }
 
+export interface TextMapSetter<Carrier = unknown> {
+  set(carrier: Carrier, key: string, value: string): void;
+}
+
 export declare const propagation: {
   extract<Carrier>(context: Context, carrier: Carrier, getter: TextMapGetter<Carrier>): Context;
+  inject<Carrier>(context: Context, carrier: Carrier, setter: TextMapSetter<Carrier>): void;
 };
 
 export declare const trace: {

packages/eve/scripts/vendor-compiled/declarations/@vercel/otel.d.ts

@@ -10,13 +10,38 @@ export interface IdGenerator {
   generateTraceId(): string;
 }
 
+/**
+ * A `TextMapPropagator`, or one of the names `@vercel/otel` resolves for you.
+ * Structural rather than imported: the instance comes from whichever
+ * `@opentelemetry/*` build the app installed, not from eve's.
+ */
+export type PropagatorOrName =
+  | { inject(...args: never[]): void; extract(...args: never[]): unknown; fields(): string[] }
+  | "auto"
+  | "none"
+  | "tracecontext"
+  | "baggage";
+
+/** A `Sampler`, or one of the names `@vercel/otel` resolves for you. */
+export type SamplerOrName =
+  | { shouldSample(...args: never[]): unknown; toString(): string }
+  | "auto"
+  | "always_off"
+  | "always_on"
+  | "parentbased_always_off"
+  | "parentbased_always_on"
+  | "parentbased_traceidratio"
+  | "traceidratio";
+
 export interface Configuration {
+  readonly attributes?: Readonly<Record<string, unknown>>;
   readonly autoDetectResources?: boolean;
   readonly idGenerator?: IdGenerator;
   readonly instrumentations?: readonly unknown[];
-  readonly propagators?: readonly ["none"];
+  readonly propagators?: readonly PropagatorOrName[];
   readonly serviceName?: string;
   readonly spanProcessors?: readonly SpanProcessor[];
+  readonly traceSampler?: SamplerOrName;
 }
 
 export declare function registerOTel(configuration?: Configuration | string): void;

packages/eve/src/tracing/agent-trace-span-processor.ts

@@ -11,8 +11,6 @@ export class AgentTraceSpanProcessor implements SpanProcessor {
   readonly #children: readonly SpanProcessor[];
   readonly #ownedTraceIds = new Set<string>();
   readonly #sessionTraceIds = new Map<string, Set<string>>();
-  #attached = false;
-
   constructor(children: readonly SpanProcessor[]) {
     this.#children = children;
   }
@@ -21,12 +19,7 @@ export class AgentTraceSpanProcessor implements SpanProcessor {
     await Promise.all(this.#children.map((child) => child.forceFlush()));
   }
 
-  isAttached(): boolean {
-    return this.#attached;
-  }
-
   onStart(span: unknown, parentContext: unknown): void {
-    this.#attached = true;
     if (!isSpanLike(span)) return;
     const sessionId = span.attributes["agent.session.id"];
     if (typeof sessionId === "string") {

packages/eve/src/tracing/local-instrumentation-runtime-conflict.scenario.test.ts

@@ -25,6 +25,6 @@ describe("local instrumentation runtime ownership", () => {
         frameworkVersion: "test",
         serviceName: "test-agent",
       }),
-    ).toThrow(/another runtime already exists/u);
+    ).toThrow(/another runtime already owns the global tracer provider/u);
   });
 });

packages/eve/src/tracing/local-instrumentation-runtime.ts

@@ -1,10 +1,7 @@
-import { context, trace } from "#compiled/@opentelemetry/api/index.js";
-import { registerOTel } from "#compiled/@vercel/otel/index.js";
+import { trace } from "#compiled/@opentelemetry/api/index.js";
 
 import { ContextAgentTraceStateStore } from "#tracing/agent-trace-context-store.js";
 import { createAgentOtelInstrumentation } from "#tracing/agent-otel-provider.js";
-import { AgentSpanIdGenerator } from "#tracing/agent-span-id-generator.js";
-import { AgentTraceSpanProcessor } from "#tracing/agent-trace-span-processor.js";
 import {
   createInstrumentationHooks,
   type InstrumentationProviderDefinition,
@@ -14,11 +11,9 @@ import {
   registerInstrumentationRuntime,
   type InstrumentationRuntime,
 } from "#harness/instrumentation-runtime.js";
-import {
-  requestLocalTraceStorePrune,
-  resolveLocalTraceRetentionSettings,
-} from "#tracing/local-trace-retention.js";
-import { LocalTraceSpanProcessor } from "#tracing/local-trace-span-processor.js";
+import { localTraces } from "#tracing/local-traces.js";
+import { mergeOtelDeclarations, otel } from "#tracing/otel-declaration.js";
+import { registerOtelPipeline } from "#tracing/otel-registration.js";
 
 /** Installs the zero-config local OTel runtime once in an `eve dev` worker. */
 export function installLocalInstrumentationRuntime(input: {
@@ -29,28 +24,15 @@ export function installLocalInstrumentationRuntime(input: {
   const existing = getInstrumentationRuntime();
   if (existing !== undefined) return existing;
 
-  const retention = resolveLocalTraceRetentionSettings();
-  // `EVE_TRACES=off` removes the writer but keeps the runtime: agent context
-  // still has to propagate so AI SDK and user spans nest correctly.
-  const processor = new AgentTraceSpanProcessor(
-    retention.enabled ? [new LocalTraceSpanProcessor(input.appRoot)] : [],
-  );
-  const idGenerator = new AgentSpanIdGenerator();
-  registerOTel({
-    autoDetectResources: false,
-    idGenerator,
-    instrumentations: [],
-    propagators: ["none"],
+  // The zero-config default expressed with the same primitive an authored
+  // `instrumentation.ts` would use, so this path exercises it.
+  const spool = localTraces({ appRoot: input.appRoot });
+  const merged = mergeOtelDeclarations([otel({ spanProcessors: [spool] })]);
+  const idGenerator = registerOtelPipeline({
+    options: merged ?? {},
     serviceName: input.serviceName,
-    spanProcessors: [processor],
   });
-  const probe = trace.getTracer("eve.registration").startSpan("eve.otel.registration");
-  const activeContext = trace.setSpan(context.active(), probe);
-  const contextAttached = context.with(activeContext, () => trace.getActiveSpan() === probe);
-  probe.end();
-  if (!processor.isAttached() || !contextAttached) {
-    throw new Error("eve could not register OpenTelemetry because another runtime already exists.");
-  }
+
   const agentOtel = createAgentOtelInstrumentation({
     captureContent: process.env.EVE_TRACES_CONTENT !== "off",
     frameworkVersion: input.frameworkVersion,
@@ -64,33 +46,14 @@ export function installLocalInstrumentationRuntime(input: {
       "session.failed": releaseSessionTrace,
     },
   };
-  // Startup sweep: a store left oversized by a killed dev server is bounded
-  // before this worker adds to it.
-  requestPrune();
 
   return registerInstrumentationRuntime({
-    forceFlush: () => processor.forceFlush(),
+    forceFlush: () => spool.forceFlush(),
     hooks: createInstrumentationHooks([agentOtel.hook, releaseTrace]),
     runInContext: agentOtel.runInContext,
   });
 
   async function releaseSessionTrace(event: { readonly sessionId: string }): Promise<void> {
-    // Settle pending segment writes before dropping liveness: a sweep already
-    // running reads the same live set, so releasing first would expose the
-    // trace to eviction while it is still being written.
-    await processor.forceFlush();
-    if (!processor.releaseSession(event.sessionId)) return;
-    requestPrune();
-  }
-
-  function requestPrune(): void {
-    if (!retention.enabled) return;
-    requestLocalTraceStorePrune({
-      activeTraceIds: processor.activeTraceIds(),
-      appRoot: input.appRoot,
-      maxAgeMs: retention.maxAgeMs,
-      maxTotalBytes: retention.maxTotalBytes,
-      retainCount: retention.retainCount,
-    });
+    await spool.releaseSession(event.sessionId);
   }
 }

packages/eve/src/tracing/local-traces.test.ts

@@ -0,0 +1,50 @@
+import { describe, expect, it, vi } from "vitest";
+
+import { localTraces } from "#tracing/local-traces.js";
+
+vi.mock("#tracing/local-trace-span-processor.js", () => ({
+  LocalTraceSpanProcessor: class {
+    async forceFlush(): Promise<void> {}
+    onEnd(): void {}
+    onStart(): void {}
+    async shutdown(): Promise<void> {}
+  },
+}));
+
+vi.mock("#tracing/local-trace-retention.js", () => ({
+  requestLocalTraceStorePrune: vi.fn(),
+  resolveLocalTraceRetentionSettings: () => ({
+    enabled: true,
+    maxAgeMs: 1,
+    maxTotalBytes: 1,
+    retainCount: 1,
+  }),
+}));
+
+function agentSpan(sessionId: string, traceId: string): unknown {
+  return {
+    attributes: { "agent.session.id": sessionId },
+    spanContext: () => ({ traceId }),
+  };
+}
+
+describe("localTraces", () => {
+  it("reports whether the released session owned any traces", async () => {
+    const spool = localTraces({ appRoot: "/tmp/eve-local-traces-test" });
+    spool.onStart(agentSpan("session-one", "a".repeat(32)), undefined);
+
+    // A subagent child owns none, so releasing it leaves the trace pinned.
+    await expect(spool.releaseSession("child-one")).resolves.toBe(false);
+    await expect(spool.releaseSession("session-one")).resolves.toBe(true);
+    // Releasing twice is not an error, it just owns nothing the second time.
+    await expect(spool.releaseSession("session-one")).resolves.toBe(false);
+  });
+
+  it("is a span processor, so it composes wherever one goes", () => {
+    const spool = localTraces({ appRoot: "/tmp/eve-local-traces-test" });
+    expect(typeof spool.onStart).toBe("function");
+    expect(typeof spool.onEnd).toBe("function");
+    expect(typeof spool.forceFlush).toBe("function");
+    expect(typeof spool.shutdown).toBe("function");
+  });
+});

packages/eve/src/tracing/local-traces.ts

@@ -0,0 +1,77 @@
+import type { SpanProcessor } from "#compiled/@vercel/otel/index.js";
+
+import { AgentTraceSpanProcessor } from "#tracing/agent-trace-span-processor.js";
+import { LocalTraceSpanProcessor } from "#tracing/local-trace-span-processor.js";
+import {
+  requestLocalTraceStorePrune,
+  resolveLocalTraceRetentionSettings,
+} from "#tracing/local-trace-retention.js";
+
+/**
+ * The local spool, as a span processor.
+ *
+ * Session liveness and retention live in here rather than in the runtime that
+ * installs it, so nothing an author puts in the same `spanProcessors` list can
+ * see them — and eve's accept filter never reaches an author's exporters.
+ */
+export interface LocalTracesProcessor extends SpanProcessor {
+  /**
+   * Settles pending writes and drops one root session's liveness, then bounds
+   * the store. A subagent child owns no traces, so releasing one is a no-op
+   * and leaves the shared trace pinned until its root finishes.
+   */
+  releaseSession(sessionId: string): Promise<boolean>;
+}
+
+/**
+ * Writes the OTLP/JSON spool under `.eve/traces/v1`.
+ *
+ * `EVE_TRACES=off` removes the writer but keeps the processor: eve still has
+ * to observe spans to track which session owns which trace.
+ */
+export function localTraces(input: { readonly appRoot?: string } = {}): LocalTracesProcessor {
+  const appRoot = input.appRoot ?? resolveAppRoot();
+  const retention = resolveLocalTraceRetentionSettings();
+  const processor = new AgentTraceSpanProcessor(
+    retention.enabled ? [new LocalTraceSpanProcessor(appRoot)] : [],
+  );
+
+  const requestPrune = (): void => {
+    if (!retention.enabled) return;
+    requestLocalTraceStorePrune({
+      activeTraceIds: processor.activeTraceIds(),
+      appRoot,
+      maxAgeMs: retention.maxAgeMs,
+      maxTotalBytes: retention.maxTotalBytes,
+      retainCount: retention.retainCount,
+    });
+  };
+
+  // Startup sweep: a store left oversized by a killed dev server is bounded
+  // before this worker adds to it.
+  requestPrune();
+
+  return {
+    forceFlush: () => processor.forceFlush(),
+    onEnd: (span) => processor.onEnd(span),
+    onStart: (span, parentContext) => processor.onStart(span, parentContext),
+    async releaseSession(sessionId) {
+      // Settle pending segment writes before dropping liveness: a sweep already
+      // running reads the same live set, so releasing first would expose the
+      // trace to eviction while it is still being written.
+      await processor.forceFlush();
+      if (!processor.releaseSession(sessionId)) return false;
+      requestPrune();
+      return true;
+    },
+    shutdown: () => processor.shutdown(),
+  };
+}
+
+function resolveAppRoot(): string {
+  const appRoot = process.env["EVE_DEV_WORKER_APP_ROOT"];
+  if (appRoot === undefined) {
+    throw new Error("EVE_DEV_WORKER_APP_ROOT is required for local tracing.");
+  }
+  return appRoot;
+}

packages/eve/src/tracing/otel-declaration.test.ts

@@ -0,0 +1,79 @@
+import type { SpanProcessor } from "#compiled/@vercel/otel/index.js";
+import { describe, expect, it } from "vitest";
+
+import { isOtelDeclaration, mergeOtelDeclarations, otel } from "#tracing/otel-declaration.js";
+
+/** The merge only ever moves processors, so a fresh no-op is identity enough. */
+function processor(): SpanProcessor {
+  return {
+    forceFlush: async () => undefined,
+    onEnd: () => undefined,
+    onStart: () => undefined,
+    shutdown: async () => undefined,
+  };
+}
+
+describe("otel", () => {
+  it("declares a pipeline without registering anything", () => {
+    const declaration = otel({ spanProcessors: [processor()] });
+    expect(isOtelDeclaration(declaration)).toBe(true);
+    expect(isOtelDeclaration({ options: {} })).toBe(false);
+  });
+});
+
+describe("mergeOtelDeclarations", () => {
+  it("is absent when nothing declared a pipeline", () => {
+    expect(mergeOtelDeclarations([])).toBeUndefined();
+  });
+
+  it("concatenates span processors in declaration order", () => {
+    const [first, second, third] = [processor(), processor(), processor()];
+    const merged = mergeOtelDeclarations([
+      otel({ spanProcessors: [first, second] }),
+      otel({ spanProcessors: [third] }),
+      otel(),
+    ]);
+
+    expect(merged?.spanProcessors).toStrictEqual([first, second, third]);
+  });
+
+  it("carries a singleton declared exactly once", () => {
+    const merged = mergeOtelDeclarations([
+      otel({ sampler: "always_on" }),
+      otel({ propagators: ["tracecontext"] }),
+      otel({ resource: { "service.version": "abc" } }),
+    ]);
+
+    expect(merged).toMatchObject({
+      propagators: ["tracecontext"],
+      resource: { "service.version": "abc" },
+      sampler: "always_on",
+    });
+  });
+
+  // A process has one tracer provider, so letting the first declaration win
+  // would silently discard the second — the failure this throw exists to stop.
+  it.each([
+    { key: "resource", one: otel({ resource: { a: "1" } }), two: otel({ resource: { b: "2" } }) },
+    { key: "sampler", one: otel({ sampler: "always_on" }), two: otel({ sampler: "always_off" }) },
+    {
+      key: "propagators",
+      one: otel({ propagators: ["tracecontext"] }),
+      two: otel({ propagators: ["baggage"] }),
+    },
+  ])("refuses a second $key rather than picking one", ({ key, one, two }) => {
+    expect(() => mergeOtelDeclarations([one, two])).toThrow(
+      new RegExp(`declares \`${key}\` in more than one`, "u"),
+    );
+  });
+
+  it("names both declarations in the collision, so the author can find them", () => {
+    expect(() =>
+      mergeOtelDeclarations([
+        otel(),
+        otel({ sampler: "always_on" }),
+        otel({ sampler: "always_off" }),
+      ]),
+    ).toThrow(/\(1 and 2\)/u);
+  });
+});