diff --git a/package-lock.json b/package-lock.json index 406d54f9d..f98a1303b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -6184,6 +6184,7 @@ "version": "1.3.0", "license": "MIT", "dependencies": { + "@earendil-works/pi-telemetry": "1.3.0", "typebox": "1.3.27" }, "devDependencies": { @@ -6343,6 +6344,7 @@ "@earendil-works/chord": "1.3.0", "@earendil-works/pi-ai": "1.3.0", "@earendil-works/pi-codemode": "1.3.0", + "@earendil-works/pi-telemetry": "1.3.0", "@silvia-odwyer/photon": "0.3.3", "diff": "8.0.4", "typebox": "1.3.27" diff --git a/package.json b/package.json index f377231fb..04462a1d5 100644 --- a/package.json +++ b/package.json @@ -13,8 +13,8 @@ ], "scripts": { "clean": "npm run clean --workspaces", - "build": "cd packages/chord && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build && cd ../durable && npm run build && cd ../env && npm run build && cd ../agent && npm run build && cd ../coding-agent && npm run build", - "build:offline": "cd packages/chord && npm run build && cd ../tui && npm run build && cd ../telemetry && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build:offline && cd ../durable && npm run build && cd ../env && npm run build && cd ../agent && npm run build && cd ../coding-agent && npm run build", + "build": "cd packages/telemetry && npm run build && cd ../chord && npm run build && cd ../tui && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build && cd ../durable && npm run build && cd ../env && npm run build && cd ../agent && npm run build && cd ../coding-agent && npm run build", + "build:offline": "cd packages/telemetry && npm run build && cd ../chord && npm run build && cd ../tui && npm run build && cd ../codemode && npm run build && cd ../mcp && npm run build && cd ../ai && npm run build:offline && cd ../durable && npm run build && cd ../env && npm run build && cd ../agent && npm run build && cd ../coding-agent && npm run build", "build:native:darwin": "npm --prefix packages/tui run build:native:darwin", "build:native:linux": "npm --prefix packages/tui run build:native:linux", "build:native:win32": "npm --prefix packages/tui run build:native:win32", diff --git a/packages/chord/CHANGELOG.md b/packages/chord/CHANGELOG.md index e9d1c4165..0ad555288 100644 --- a/packages/chord/CHANGELOG.md +++ b/packages/chord/CHANGELOG.md @@ -2,8 +2,13 @@ ## [Unreleased] +### Breaking Changes + +- Custom `Context` implementations must provide `telemetry: TelemetryContext`; use `NOOP_TELEMETRY_CONTEXT` when disabled. + ### Added +- `Context.telemetry` is an explicit, non-optional telemetry parent (no-op by default). `withTelemetry(telemetry, context)` derives it without changing values or cancellation; `traceIn(context, options, fn)` starts a child span and passes its derived context to the callback. - `@earendil-works/chord/sqlite`: portable SQLite interfaces, atomic migration runner, and schema-prefix validation, with Node (`/sqlite/node`) and Cloudflare Durable Object (`/sqlite/cloudflare`) drivers. Moved from Durable, which keeps its storage implementations and re-exports the moved public names. - `@earendil-works/chord/output`: `Progress` takes an optional `ProgressClock` (`now`, `setTimer`) to pace commits on, by default `Date.now` and `setTimeout`. diff --git a/packages/chord/README.md b/packages/chord/README.md index 157b4b51d..d05dc79f6 100644 --- a/packages/chord/README.md +++ b/packages/chord/README.md @@ -16,7 +16,7 @@ Contracts are runtime values described with [TypeBox](https://github.com/sinclai |--------|----------|---------| | `@earendil-works/chord` | Contracts (`local`, `remote`, `method`, `opaque`), declarations (through the factory's `self`: `state`, `derive`, `resource`, `on`, `task`), hosts (`createHost`, `defineFacet`, `defineRegistry`), links (`memoryLinkPair`, `ipcLink`, `webSocketLink`, `acceptWebSockets`), errors (`ChordError`, `isChordError`, `forbidden`, `invalid`). Also the shared base: `Context`, `JsonValue`, `JsonRepresentation`, `Draft`, `copyJson`, `isJsonValue`, and `replicatedState` with its types. | Node and browsers | | `@earendil-works/chord/node` | Child host processes: `createHostManager`, `forkDriver`, `childControl`, `processLock`; the facet module loader `nodeLoader`. | Node | -| `@earendil-works/chord/context` | Explicit cancellation and invocation values: `BACKGROUND_CONTEXT`, `withCancel`, `withAbortSignal`, `withoutAbortSignal`, `withContextValue`, `awaitWithContext`. | Node and browsers | +| `@earendil-works/chord/context` | Explicit cancellation, telemetry, and invocation values: `BACKGROUND_CONTEXT`, `withCancel`, `withAbortSignal`, `withoutAbortSignal`, `withTelemetry`, `traceIn`, `withContextValue`, `awaitWithContext`. | Node and browsers | | `@earendil-works/chord/delta` | The transactional JSON tracker, structural diff, immutable apply, and the wire encoder/decoder for operation batches. | Node and browsers | | `@earendil-works/chord/output` | Bounded output streams and paced progress: `OutputBuffer` (head or tail window of a stream, `OutputLimits`, `OutputSkip`), `boundOutput`, `sanitizeOutput`, `utf8ByteLength`, and `Progress` (adaptive commits of a changing value). | Node and browsers | | `@earendil-works/chord/sqlite` | Portable `SqliteDatabase`, `SqliteExecutor`, and `SqliteValue`, atomic `applySqliteMigrations(database, migrations, { table })`, and `validateSqliteSchemaPrefix`. The metadata table must be a valid, non-reserved SQLite identifier; migration versions start at 1 and are contiguous. | Any SQLite adapter | @@ -69,6 +69,26 @@ await host.root(async (ctx) => { - [Specification](docs/spec.md): compatibility, lifetime, and reload semantics, plus known limitations and implementation decisions. - [Delta guide](src/delta/README.md): ownership, lifecycle, operation, and replica contracts of `@earendil-works/chord/delta`. +## Explicit Context telemetry + +`Context` has a fixed `telemetry: TelemetryContext` member, alongside `abortSignal`. It is never undefined: `BACKGROUND_CONTEXT` and `TODO_CONTEXT` use `NOOP_TELEMETRY_CONTEXT` from `@earendil-works/pi-telemetry`. + +```ts +import { BACKGROUND_CONTEXT, traceIn, withTelemetry } from '@earendil-works/chord/context'; +import { InMemoryTelemetryContext } from '@earendil-works/pi-telemetry'; + +const telemetry = new InMemoryTelemetryContext(); +const context = withTelemetry(telemetry, BACKGROUND_CONTEXT); +await traceIn(context, { name: 'app.load' }, async (child) => { + child.telemetry.addEvent('cache.miss'); + return traceIn(child, { name: 'app.read' }, (nested) => readAccount(nested)); +}); +``` + +`withTelemetry(telemetry, context)` derives a new context; it never changes its parent, cancellation, or typed values. `traceIn(context, options, fn)` creates a callback-managed child span, and `fn` receives the derived context with `telemetry` narrowed to `TelemetrySpan`, including `addLink()`. Values, cancellation, callback results, and rejection identities are preserved. Concurrent calls keep independent parents; there is no ambient current span. `value()`, `ContextKey`, and `withContextValue()` remain supported. + +These helpers operate on the standalone `Context` API. They do not yet instrument Chord hosts, methods, or network frames. + ## Tracking JSON deltas Import the standalone transactional tracker from `@earendil-works/chord/delta`: diff --git a/packages/chord/package.json b/packages/chord/package.json index 973063fb5..f88036b93 100644 --- a/packages/chord/package.json +++ b/packages/chord/package.json @@ -80,6 +80,7 @@ "node": ">=22.19.0" }, "dependencies": { + "@earendil-works/pi-telemetry": "1.3.0", "typebox": "1.3.27" }, "devDependencies": { diff --git a/packages/chord/src/context/index.ts b/packages/chord/src/context/index.ts index 7426083f6..9df2bc23b 100644 --- a/packages/chord/src/context/index.ts +++ b/packages/chord/src/context/index.ts @@ -1,3 +1,10 @@ +import { + NOOP_TELEMETRY_CONTEXT, + type SpanOptions, + type TelemetryContext, + type TelemetrySpan, +} from "@earendil-works/pi-telemetry"; + /** Typed identity for one value carried by a {@link Context}. */ export interface ContextKey { readonly token: symbol; @@ -8,6 +15,7 @@ export interface ContextKey { /** Immutable invocation-scoped values passed explicitly through operations. */ export interface Context { readonly abortSignal: AbortSignal | undefined; + readonly telemetry: TelemetryContext; value(key: ContextKey): T | undefined; toString(): string; } @@ -16,6 +24,10 @@ const ABORT_SIGNAL_CONTEXT_KEY: ContextKey = Object.fre token: Symbol("chord.abortSignal"), }); +const TELEMETRY_CONTEXT_KEY: ContextKey = Object.freeze({ + token: Symbol("chord.telemetry"), +}); + abstract class BaseContext implements Context { abstract value(key: ContextKey): T | undefined; abstract toString(): string; @@ -23,6 +35,10 @@ abstract class BaseContext implements Context { get abortSignal(): AbortSignal | undefined { return this.value(ABORT_SIGNAL_CONTEXT_KEY); } + + get telemetry(): TelemetryContext { + return this.value(TELEMETRY_CONTEXT_KEY) ?? NOOP_TELEMETRY_CONTEXT; + } } class EmptyContext extends BaseContext { @@ -76,6 +92,23 @@ export function withContextValue(key: ContextKey, value: T, parent: Contex return new ContextValue(parent, key, value); } +/** Derive an explicit telemetry parent without changing cancellation or application values. */ +export function withTelemetry( + telemetry: T, + context: Context, +): Context & { readonly telemetry: T } { + return withContextValue(TELEMETRY_CONTEXT_KEY, telemetry, context) as Context & { readonly telemetry: T }; +} + +/** Run a child span with a derived context whose telemetry member is that span. */ +export function traceIn( + context: Context, + options: SpanOptions, + fn: (context: Context & { readonly telemetry: TelemetrySpan }) => T | Promise, +): Promise { + return context.telemetry.startSpan(options, (span) => fn(withTelemetry(span, context))); +} + /** * Derive a context cancelled by either the parent signal or the supplied signal. * The parent context remains unchanged. diff --git a/packages/chord/test/context.test.ts b/packages/chord/test/context.test.ts index 3384d3891..d45b88d89 100644 --- a/packages/chord/test/context.test.ts +++ b/packages/chord/test/context.test.ts @@ -1,13 +1,16 @@ +import { InMemoryTelemetryContext, NOOP_TELEMETRY_CONTEXT } from "@earendil-works/pi-telemetry"; import { describe, expect, it, vi } from "vitest"; import { awaitWithContext, BACKGROUND_CONTEXT, createContextKey, TODO_CONTEXT, + traceIn, withAbortSignal, withCancel, withContextValue, withoutAbortSignal, + withTelemetry, } from "../src/context/index.ts"; describe("Context", () => { @@ -16,6 +19,8 @@ describe("Context", () => { expect(TODO_CONTEXT).not.toBe(BACKGROUND_CONTEXT); expect(TODO_CONTEXT.abortSignal).toBeUndefined(); + expect(TODO_CONTEXT.telemetry).toBe(NOOP_TELEMETRY_CONTEXT); + expect(BACKGROUND_CONTEXT.telemetry).toBe(NOOP_TELEMETRY_CONTEXT); expect(TODO_CONTEXT.value(key)).toBeUndefined(); expect(String(BACKGROUND_CONTEXT)).toBe("[Context BACKGROUND_CONTEXT]"); expect(String(TODO_CONTEXT)).toBe("[Context TODO_CONTEXT]"); @@ -70,6 +75,75 @@ describe("Context", () => { expect(cleanup.value(key)).toBe("preserved"); }); + it("derives telemetry independently from values and cancellation", async () => { + const telemetry = new InMemoryTelemetryContext(); + const other = new InMemoryTelemetryContext(); + const key = createContextKey("principal"); + const controller = new AbortController(); + const base = withContextValue(key, "owner", withAbortSignal(controller.signal, BACKGROUND_CONTEXT)); + const context = withTelemetry(telemetry, base); + const replaced = withTelemetry(other, context); + expect(base.telemetry).toBe(NOOP_TELEMETRY_CONTEXT); + expect(context.telemetry).toBe(telemetry); + expect(replaced.telemetry).toBe(other); + expect(replaced.value(key)).toBe("owner"); + expect(replaced.abortSignal).toBe(base.abortSignal); + expect(withoutAbortSignal(context).telemetry).toBe(telemetry); + await traceIn(context, { name: "outer" }, async (child) => { + expect(child.value(key)).toBe("owner"); + expect(child.abortSignal).toBe(context.abortSignal); + child.telemetry.addEvent("started"); + await traceIn(child, { name: "inner" }, (nested) => { + nested.telemetry.addLink({ context: child.telemetry }); + return 42; + }); + }); + const [outer, inner] = telemetry.getSpans(); + expect(outer?.parentId).toBeNull(); + expect(inner?.parentId).toBe(outer?.id); + expect(inner?.links[0]?.context).toEqual(outer?.context); + expect(context.telemetry).toBe(telemetry); + controller.abort(); + expect(context.abortSignal?.aborted).toBe(true); + }); + + it("isolates concurrent traceIn branches and preserves callback results and errors", async () => { + const telemetry = new InMemoryTelemetryContext(); + const context = withTelemetry(telemetry, BACKGROUND_CONTEXT); + let release: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const first = traceIn(context, { name: "first" }, async (child) => { + await gate; + return traceIn(child, { name: "first.child" }, () => "first"); + }); + const second = traceIn(context, { name: "second" }, async (child) => { + const result = await traceIn(child, { name: "second.child" }, () => "second"); + release?.(); + return result; + }); + expect(await Promise.all([first, second])).toEqual(["first", "second"]); + const spans = telemetry.getSpans(); + for (const name of ["first", "second"]) { + const parent = spans.find((span) => span.name === name); + expect(spans.find((span) => span.name === `${name}.child`)?.parentId).toBe(parent?.id); + } + const failure = { reason: "same value" }; + await expect( + traceIn(context, { name: "throw" }, () => { + throw failure; + }), + ).rejects.toBe(failure); + await expect(traceIn(context, { name: "reject" }, () => Promise.reject(failure))).rejects.toBe(failure); + expect( + await traceIn(BACKGROUND_CONTEXT, { name: "noop" }, (child) => { + child.telemetry.addLink({ context: telemetry }); + return 42; + }), + ).toBe(42); + }); + it("stops waiting when the invocation is cancelled", async () => { const controller = new AbortController(); const context = withAbortSignal(controller.signal, BACKGROUND_CONTEXT); diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 6645eb0dd..419fc31f9 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -12,6 +12,7 @@ ### Added +- `HarnessOptions.telemetry` accepts a host-owned telemetry context with optional propagation and metrics capabilities. This change adds the configuration type only; Durable instrumentation and persisted carriers are not yet implemented. - `settings.compaction.modelOverrides` sets compaction thresholds per model, keyed by exact `"provider/modelId"`, so one harness can serve small- and large-window models; the threshold check, overflow recovery, and the compaction task use the entry of the model they act for ([#10942](https://github.com/earendil-works/pi/issues/10942)). - Durable sleeps: a sleeping task holds no invocation, survives restarts, and resumes at its checkpoint once the Harness clock reaches `until`. - `HarnessOptions.setTimer` arms the timers of the Session, the scheduler, and its tasks (progress throttles, idle-memory eviction, sleeps), also for `createSession()`; with `now`, they run on a host or fake clock. A foreground timer the host throws on or declines falls back to `setTimeout`. `ToolExecutionApi.now()` reads the Harness clock. diff --git a/packages/durable/README.md b/packages/durable/README.md index fea9629de..c3f3d3521 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -75,7 +75,9 @@ What happened: - `submit()` durably admits your input and returns a `Submission`. A built-in generation task calls the model and appends the answer. - `wait()` resolves once the input is answered (`done`) or has failed (`unanswered`, with a reason). -Every async call takes a Chord `Context`, which carries cancellation. `BACKGROUND_CONTEXT` never cancels. Cancelling a wait only cancels that wait, never the work. +Every async call takes a Chord `Context`, which carries cancellation and explicit telemetry. `BACKGROUND_CONTEXT` never cancels and uses no-op telemetry. Cancelling a wait only cancels that wait, never the work. + +`HarnessOptions.telemetry` accepts a host-owned `TelemetryContext` adapter, optionally implementing `TelemetryPropagation` and `TelemetryMetrics`, from `@earendil-works/pi-telemetry`. This reserves the adapter configuration boundary; Durable currently emits no spans or metrics and does not persist carriers. Its instrumentation and schema are separate work. ## Concepts diff --git a/packages/durable/package.json b/packages/durable/package.json index bb43f2fd8..5823f4240 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -147,6 +147,7 @@ "@earendil-works/chord": "1.3.0", "@earendil-works/pi-ai": "1.3.0", "@earendil-works/pi-codemode": "1.3.0", + "@earendil-works/pi-telemetry": "1.3.0", "@silvia-odwyer/photon": "0.3.3", "diff": "8.0.4", "typebox": "1.3.27" diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index cca3ef601..690caf2e4 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -14,6 +14,7 @@ import type { Usage, UserMessage, } from "@earendil-works/pi-ai"; +import type { TelemetryContext, TelemetryMetrics, TelemetryPropagation } from "@earendil-works/pi-telemetry"; import type { ExecutionEnv, ShellOutputSkip, ShellOutputWindow } from "../env/index.ts"; import type { ConversationId, @@ -605,6 +606,8 @@ export type EnvTarget = { }; export type HarnessOptions = { + /** Host-owned adapter; propagation and metrics are optional. Durable instrumentation is not yet emitted. */ + readonly telemetry?: TelemetryContext & Partial; /** * pi-ai model access used by generation, compaction, hooks and tools: a `Models`, or a function returning the current * one, called at each use, so a host can replace its model configuration without reopening the Harness. diff --git a/packages/durable/test/harness-telemetry-options.test.ts b/packages/durable/test/harness-telemetry-options.test.ts new file mode 100644 index 000000000..a63532e20 --- /dev/null +++ b/packages/durable/test/harness-telemetry-options.test.ts @@ -0,0 +1,24 @@ +import { InMemoryTelemetryContext, NOOP_TELEMETRY_CONTEXT } from "@earendil-works/pi-telemetry"; +import { describe, expect, expectTypeOf, it } from "vitest"; +import type { HarnessOptions } from "../src/harness/types.ts"; + +describe("HarnessOptions.telemetry", () => { + it("accepts one adapter with optional propagation and metrics capabilities", () => { + const adapter = new InMemoryTelemetryContext(); + const full: Pick = { telemetry: adapter }; + const contextOnly: Pick = { telemetry: NOOP_TELEMETRY_CONTEXT }; + const omitted: Pick = {}; + expect(full.telemetry).toBe(adapter); + expect(contextOnly.telemetry).toBe(NOOP_TELEMETRY_CONTEXT); + expect(omitted.telemetry).toBeUndefined(); + expectTypeOf(full.telemetry?.inject).toExtend< + ((context: NonNullable) => unknown) | undefined + >(); + const compileTimeFailures = () => { + // @ts-expect-error telemetry requires the span-starting context capability + const invalid: Pick = { telemetry: { counter: adapter.counter.bind(adapter) } }; + return invalid; + }; + expectTypeOf(compileTimeFailures).toBeFunction(); + }); +}); diff --git a/packages/telemetry/CHANGELOG.md b/packages/telemetry/CHANGELOG.md index 11ad5c5e8..c094cb21f 100644 --- a/packages/telemetry/CHANGELOG.md +++ b/packages/telemetry/CHANGELOG.md @@ -2,6 +2,18 @@ ## [Unreleased] +### Breaking Changes + +- Adapters must implement `TelemetrySpan.addLink()`. Normalized span snapshots include carrier-shaped context and parent identities and ordered links. + +### Added + +- Opaque string-map carriers and optional `TelemetryPropagation` injection/extraction, with passive root fallback for absent or invalid carriers and shared `NOOP_TELEMETRY_PROPAGATION`. +- Initial `SpanOptions.links` and late `span.addLink()` with context targets and optional attributes, including settled target spans and extracted remote parents. +- Optional `TelemetryMetrics` counters, up-down counters, and histograms with boundary hints and exemplar contexts, plus shared `NOOP_TELEMETRY_METRICS` instruments. +- Metric schema definitions and `createTypedMetrics()` with exact instrument names, kinds, and required/optional low-cardinality attributes. +- In-memory propagation across processes, link and measurement snapshots, and adapter conformance cases for propagation, links, and optional metrics. + ## [1.3.0] - 2026-10-15 ## [1.2.0] - 2026-10-12 diff --git a/packages/telemetry/README.md b/packages/telemetry/README.md index 0c00dfd71..8bfc5d480 100644 --- a/packages/telemetry/README.md +++ b/packages/telemetry/README.md @@ -6,6 +6,8 @@ This package provides: - an explicit, callback-based `TelemetryContext` / `TelemetrySpan` contract; - a shared `NOOP_TELEMETRY_CONTEXT`; +- opaque cross-process propagation and span links; +- counters, up-down counters, histograms, and typed metric schemas; - a reference `InMemoryTelemetryContext` implementation; - serializable schema definitions with inferred TypeScript types; - no exporter, global current-span state, or dependency on a telemetry backend. @@ -18,6 +20,8 @@ Applications can use the in-memory reference or provide an adapter for OpenTelem - [Telemetry Concepts](#telemetry-concepts) - [Core Context API](#core-context-api) - [Adapter Contract](#adapter-contract) +- [Propagation and Span Links](#propagation-and-span-links) +- [Metrics](#metrics) - [No-op Context](#no-op-context) - [In-Memory Reference Adapter](#in-memory-reference-adapter) - [Adapter Conformance](#adapter-conformance) @@ -131,6 +135,105 @@ An adapter implements `TelemetryContext` and bridges the generic API to its back Adapters may activate backend-native ambient context internally for automatic instrumentation, but pi code always propagates the parent through `TelemetryContext` arguments. Exporter buffering, flushing, sampling, backend IDs, and backend-specific context objects belong to the adapter. Use the [adapter conformance suite](#adapter-conformance) to check these observable semantics. +## Propagation and Span Links + +An adapter may also implement `TelemetryPropagation`. Keep that capability on the one adapter configured by the application, not on every context: + +```typescript +interface TelemetryPropagation { + inject(context: TelemetryContext): TelemetryCarrier | undefined; + extract(carrier: TelemetryCarrier | undefined): TelemetryContext; +} +type TelemetryCarrier = Record; +``` + +A carrier is an opaque JSON-safe string map. Adapters own its format; W3C `traceparent` and `tracestate` fit without requiring them. Transport code sends only the carrier, never a context or span: + +```typescript +const caller = new InMemoryTelemetryContext(); +const provider = new InMemoryTelemetryContext(); +await caller.startSpan({ name: 'rpc.client' }, async (span) => { + const wire = JSON.stringify(caller.inject(span)); + await provider.extract(JSON.parse(wire)).startSpan( + { name: 'rpc.server' }, + (serverSpan) => handleRequest(serverSpan), + ); +}); +``` + +Injection returns a detached carrier or `undefined` for roots, no-op contexts, and unsupported contexts. Extraction returns a receiver-owned parent. Absent, invalid, unknown, or unreadable carriers fall back to the adapter's root, so incoming work is still recorded. Both methods are synchronous, passive, and non-throwing. Trace metadata is untrusted diagnostics, never authorization. Transport boundaries own carrier size limits. + +A link relates work without changing its parent. Use `SpanOptions.links` when the target is known at creation, or `span.addLink()` after the span has started: + +```typescript +await telemetry.startSpan( + { name: 'activation', links: [{ context: firstGetter, attributes: { reason: 'trigger' } }] }, + async (activation) => { + laterGetter.addLink({ context: activation }); + await activate(); + }, +); +``` + +Link targets are `TelemetryContext` handles, including spans and extracted remote parents. For a remote link, first call `adapter.extract(carrier)`. A settled target remains linkable and injectable; a settled source ignores `addLink()`. Links append in order, copy attributes, and ignore root, no-op, unsupported, or malformed targets. Failed link calls are atomic and never throw. The typed span starter's callback also supports `addLink()`; initial links use the low-level `startSpan()` API. + +`NOOP_TELEMETRY_PROPAGATION` never inspects its arguments: injection returns `undefined`, extraction returns `NOOP_TELEMETRY_CONTEXT`. + +## Metrics + +An adapter may implement `TelemetryMetrics` alongside `TelemetryContext` and `TelemetryPropagation`. Applications configure one adapter, not separate backends for each capability: + +```typescript +const bytes = telemetry.counter('network.bytes', { + description: 'Encoded WebSocket bytes', unit: 'By', +}); +bytes.add(encodedSize, { direction: 'out' }); + +const active = telemetry.upDownCounter('calls.active'); +active.add(1); +active.add(-1); + +const duration = telemetry.histogram('call.duration', { + unit: 'ms', boundaries: [1, 10, 100, 1000], +}); +duration.record(elapsedMs, { method: 'read' }, currentSpan); +``` + +Factories take an instrument name and optional `InstrumentOptions` (`description`, `unit`). Histograms additionally accept `HistogramOptions.boundaries`: finite, strictly increasing bucket hints. Backends own aggregation and may translate these hints. Counters accept non-negative finite increments; up-down counters and histograms accept finite signed values. Histogram `record(value, attributes?, context?)` optionally correlates a measurement with a span, called an exemplar. Counter `add(value, attributes?)` has no context argument. + +Factories and recording methods are synchronous, passive, and non-throwing. Invalid measurements, unreadable metadata/attributes, or invalid boundaries are ignored atomically. `NOOP_TELEMETRY_METRICS` returns shared frozen instruments without inspecting inputs. Record only values already available to business code; do not serialize or measure extra data for telemetry. + +Use only low-cardinality attributes: bounded values such as direction or method, not request IDs, user IDs, prompts, or arbitrary paths. The low-level attribute bag remains open; instrumentation owns this policy, not backend-specific heuristics. + +### Typed metric schemas + +`TelemetrySchemaDefinition.metrics` optionally declares instrument names, kinds, metadata, and required/optional attributes. `createTypedMetrics()` combines one or more schemas and rejects duplicate metric names, incorrect instrument kinds, unknown attributes, and missing required attributes: + +```typescript +const schema = defineTelemetrySchema({ + version: 1, + spans: {}, + metrics: { + 'network.bytes': { + kind: 'counter', description: 'Encoded bytes', unit: 'By', + attributes: { + direction: { + type: 'string', required: true, values: ['in', 'out'], + description: 'Direction', cardinality: 'low', + }, + }, + }, + }, +} as const); + +const metrics = createTypedMetrics(telemetry, [schema]); +metrics.counter('network.bytes').add(encodedSize, { direction: 'out' }); +``` + +Metric definitions use `kind: 'counter' | 'upDownCounter' | 'histogram'`; only histograms may declare `boundaries`. Metric attributes use the same scalar/array definitions as spans, but reject `cardinality: 'high'`. Omitting the cardinality annotation does not waive the low-cardinality policy. Unlike the typed span starter, the metric helper reads and retains schema instrument metadata at runtime. It performs no runtime attribute validation. + +`InMemoryTelemetryContext.getMetrics()` returns ordered, detached `RecordedTelemetryMeasurement` snapshots: `kind`, `name`, optional `description`/`unit`/`boundaries`, `value`, copied `attributes`, and a carrier-shaped exemplar `context` or `null`. This is a measurement log, not aggregated counters or buckets; storage is unbounded and records no timestamps. + ## No-op Context Use `NOOP_TELEMETRY_CONTEXT` when telemetry is optional: @@ -171,7 +274,9 @@ await telemetry.startSpan( console.log(telemetry.getSpans()); ``` -`getSpans()` returns detached snapshots in span-start order. Each `RecordedTelemetrySpan` contains a deterministic numeric ID, parent ID, merged attributes, ordered events, final status, settlement state, and deterministic end sequence. It records no timestamps. +`getSpans()` returns detached snapshots in span-start order. Each `RecordedTelemetrySpan` contains a deterministic numeric ID, local parent ID, carrier-shaped `context` and `parentContext` snapshots, ordered `links`, merged attributes, ordered events, final status, settlement state, and deterministic end sequence. It records no timestamps. A remote parent has `parentId: null` but a non-null `parentContext`; a root has both null. A `RecordedTelemetryLink` contains a target carrier and copied attributes. + +The adapter implements `TelemetryPropagation` with its own small carrier codec: a unique recorder scope and a deterministic numeric span ID. The carrier survives JSON round trips and works across independent processes without a shared registry. Its keys are an implementation detail, not a public wire format for other adapters. The adapter is safe to use as an ordinary `TelemetryContext`, but storage is unbounded and process-local. Create a fresh instance to isolate tests or recording scopes, and do not capture sensitive attributes unless the caller's data policy allows them. @@ -206,7 +311,7 @@ for (const group of new Set(conformance.map((testCase) => testCase.group))) { } ``` -The suite checks synchronous single admission, result and rejection identity, automatic and explicit status, attribute merging, event ordering, inert post-settlement calls, nested and concurrent parentage, and suppression of unreadable telemetry payload failures. `getSpans()` may flush an asynchronous exporter before returning. The testing subpath uses Node's assertion APIs; the root telemetry package remains runtime-neutral. +The suite checks synchronous single admission, result and rejection identity, automatic and explicit status, attribute merging, event ordering, inert post-settlement calls, nested and concurrent parentage, links, and suppression of unreadable telemetry payload failures. Fixtures may supply `propagation` to additionally check JSON round trips, multiple adapter hops, and safe fallback for malformed carriers. Fixtures may also supply `metrics` and `getMetrics()` to check ordered measurements, histogram exemplars, input copying, and atomic suppression of invalid metadata or measurements. `getSpans()` may flush an asynchronous exporter before returning. The testing subpath uses Node's assertion APIs; the root telemetry package remains runtime-neutral. ## Typed Schemas @@ -384,7 +489,7 @@ The pi schemas use pi-owned `pi.ai.*`, `pi.harness.*`, and `pi.session.*` names. ## Security and Portability -Telemetry is process-local diagnostics, not durable application state. Do not persist a `TelemetryContext`, `TelemetrySpan`, or backend-native trace object in records, messages, snapshots, or deferred handles. +Telemetry is diagnostics, not durable application state. Do not persist a `TelemetryContext`, `TelemetrySpan`, or backend-native trace object in records, messages, snapshots, or deferred handles. An injected `TelemetryCarrier` may cross a transport or be stored as optional diagnostic correlation data. It is best effort: an adapter change may lose correlation, but must not prevent work. Attribute values are intentionally limited to primitive scalars and arrays. Domain instrumentation should avoid prompts, completions, tool arguments or output, file contents, provider payloads, headers, credentials, and free-form error details unless its schema and data policy explicitly allow them. @@ -398,13 +503,24 @@ The package does not use `AsyncLocalStorage` or another runtime-specific ambient |---|---| | `TelemetryContext` | Starts callback-managed child spans | | `TelemetrySpan` | Records attributes, events, and status; also acts as a child context | -| `SpanOptions` | Span name and optional start attributes | +| `SpanOptions` | Span name, optional start attributes and links | +| `SpanLink` | Context target and optional link attributes | +| `TelemetryCarrier` | Opaque JSON-safe string map | +| `TelemetryPropagation` | Optional adapter injection and extraction capability | +| `NOOP_TELEMETRY_PROPAGATION` | Shared passive propagation for disabled telemetry | +| `TelemetryMetrics` | Optional adapter instrument factories | +| `TelemetryCounter` / `TelemetryUpDownCounter` | Instruments with `add()` | +| `TelemetryHistogram` | Instrument with `record()` and optional exemplar context | +| `InstrumentOptions` / `HistogramOptions` | Description, unit, and histogram boundary hints | +| `NOOP_TELEMETRY_METRICS` | Shared passive metric instruments | +| `RecordedTelemetryMeasurement` | Captured measurement and optional exemplar | | `SpanAttributes` / `AttributeValue` | Open adapter-level attribute bag and supported values | | `SpanStatus` | Explicit `ok` or `error` status | | `NOOP_TELEMETRY_CONTEXT` | Shared passive context for disabled telemetry | | `InMemoryTelemetryContext` | Reference adapter with deterministic process-local recording | | `RecordedTelemetrySpan` | Normalized captured span snapshot | | `RecordedTelemetryEvent` | Normalized captured event snapshot | +| `RecordedTelemetryLink` | Captured target carrier and link attributes | ### Schema definitions and inference @@ -412,6 +528,9 @@ The package does not use `AsyncLocalStorage` or another runtime-specific ambient |---|---| | `defineTelemetrySchema()` | Typed identity helper for serializable schema data | | `createTypedSpanStarter()` | Binds a parent context to one or more schema vocabularies | +| `createTypedMetrics()` / `TypedTelemetryMetrics` | Binds metric names, kinds, metadata, and attributes to schemas | +| `TelemetryMetricDefinition` / `TelemetryMetricAttributeDefinition` | Serializable instrument metadata and low-cardinality attribute definitions | +| `SchemaTelemetryCounter` / `SchemaTelemetryHistogram` | Schema-constrained recording methods | | `TypedSpanStarter` | Exact starter type with recursively child-bound callbacks | | `TelemetrySchemaDefinition` | Top-level schema shape | | `TelemetrySpanDefinition` | Span metadata, parents, attributes, events, and status rule | diff --git a/packages/telemetry/src/index.ts b/packages/telemetry/src/index.ts index 3d3909791..4d76e73a0 100644 --- a/packages/telemetry/src/index.ts +++ b/packages/telemetry/src/index.ts @@ -1,12 +1,29 @@ +import type { TelemetryMetricDefinition } from "./metrics.ts"; + export type AttributeValue = string | number | boolean | readonly string[] | readonly number[] | readonly boolean[]; export interface SpanAttributes { [name: string]: AttributeValue | undefined; } +/** Opaque, JSON-serializable adapter-owned propagation data. */ +export type TelemetryCarrier = Record; + +/** Optional transport capability implemented by the application's telemetry adapter. */ +export interface TelemetryPropagation { + inject(context: TelemetryContext): TelemetryCarrier | undefined; + extract(carrier: TelemetryCarrier | undefined): TelemetryContext; +} + +export interface SpanLink { + context: TelemetryContext; + attributes?: SpanAttributes; +} + export interface SpanOptions { name: string; attributes?: SpanAttributes; + links?: readonly SpanLink[]; } export type SpanStatus = { status: "ok" } | { status: "error"; error?: { name: string; message: string } }; @@ -16,12 +33,27 @@ export interface TelemetryContext { } export interface TelemetrySpan extends TelemetryContext { + addLink(link: SpanLink): void; addEvent(name: string, attributes?: SpanAttributes): void; setAttributes(attributes: SpanAttributes): void; setStatus(status: SpanStatus): void; } -export { NOOP_TELEMETRY_CONTEXT } from "./noop.ts"; +export type { + HistogramOptions, + InstrumentOptions, + SchemaTelemetryCounter, + SchemaTelemetryHistogram, + TelemetryCounter, + TelemetryHistogram, + TelemetryMetricAttributeDefinition, + TelemetryMetricDefinition, + TelemetryMetrics, + TelemetryUpDownCounter, + TypedTelemetryMetrics, +} from "./metrics.ts"; +export { createTypedMetrics } from "./metrics.ts"; +export { NOOP_TELEMETRY_CONTEXT, NOOP_TELEMETRY_METRICS, NOOP_TELEMETRY_PROPAGATION } from "./noop.ts"; export type TelemetryAttributeType = "string" | "number" | "boolean" | "string[]" | "number[]" | "boolean[]"; @@ -66,6 +98,7 @@ export interface TelemetrySpanDefinition { export interface TelemetrySchemaDefinition { version: number; spans: Record; + metrics?: Record; } /** Typed identity helper for serializable telemetry schema data. */ @@ -353,5 +386,10 @@ export function createTypedSpanStarter(telemetryContext); } -export type { RecordedTelemetryEvent, RecordedTelemetrySpan } from "./memory.ts"; +export type { + RecordedTelemetryEvent, + RecordedTelemetryLink, + RecordedTelemetryMeasurement, + RecordedTelemetrySpan, +} from "./memory.ts"; export { InMemoryTelemetryContext } from "./memory.ts"; diff --git a/packages/telemetry/src/memory.ts b/packages/telemetry/src/memory.ts index fd5985a6d..2db67fe4a 100644 --- a/packages/telemetry/src/memory.ts +++ b/packages/telemetry/src/memory.ts @@ -1,11 +1,22 @@ import type { AttributeValue, SpanAttributes, + SpanLink, SpanOptions, SpanStatus, + TelemetryCarrier, TelemetryContext, + TelemetryPropagation, TelemetrySpan, } from "./index.ts"; +import type { + HistogramOptions, + InstrumentOptions, + TelemetryCounter, + TelemetryHistogram, + TelemetryMetrics, + TelemetryUpDownCounter, +} from "./metrics.ts"; import { NOOP_TELEMETRY_CONTEXT } from "./noop.ts"; export interface RecordedTelemetryEvent { @@ -13,9 +24,18 @@ export interface RecordedTelemetryEvent { readonly attributes: Readonly; } +export interface RecordedTelemetryLink { + readonly context: Readonly; + readonly attributes: Readonly; +} + export interface RecordedTelemetrySpan { readonly id: number; + /** Local parent ID, or null for roots and parents owned by another recorder. */ readonly parentId: number | null; + readonly context: Readonly; + readonly parentContext: Readonly | null; + readonly links: readonly RecordedTelemetryLink[]; readonly name: string; readonly attributes: Readonly; readonly events: readonly RecordedTelemetryEvent[]; @@ -24,6 +44,17 @@ export interface RecordedTelemetrySpan { readonly endSequence?: number; } +export interface RecordedTelemetryMeasurement { + readonly kind: "counter" | "upDownCounter" | "histogram"; + readonly name: string; + readonly description?: string; + readonly unit?: string; + readonly boundaries?: readonly number[]; + readonly value: number; + readonly attributes: Readonly; + readonly context: Readonly | null; +} + interface MutableRecordedTelemetryEvent { name: string; attributes: SpanAttributes; @@ -32,6 +63,9 @@ interface MutableRecordedTelemetryEvent { interface MutableRecordedTelemetrySpan { id: number; parentId: number | null; + context: TelemetryCarrier; + parentContext: TelemetryCarrier | null; + links: RecordedTelemetryLink[]; name: string; attributes: SpanAttributes; events: MutableRecordedTelemetryEvent[]; @@ -41,8 +75,19 @@ interface MutableRecordedTelemetrySpan { endSequence?: number; } +interface InMemoryParent { + context: TelemetryCarrier; + localId: number | null; + span?: MutableRecordedTelemetrySpan; +} + +// Identity survives settlement without exposing backend IDs on the generic context contract. +const contextCarriers = new WeakMap(); + interface InMemoryTelemetryState { + scope: string; spans: MutableRecordedTelemetrySpan[]; + measurements: RecordedTelemetryMeasurement[]; nextSpanId: number; nextEndSequence: number; } @@ -68,6 +113,16 @@ function mergeAttributes(current: SpanAttributes, attributes: SpanAttributes): S return merged; } +function copyLink(link: SpanLink): RecordedTelemetryLink | undefined { + try { + const context = contextCarriers.get(link.context); + if (!context) return undefined; + return { context: { ...context }, attributes: copyAttributes(link.attributes) }; + } catch { + return undefined; + } +} + function copyStatus(status: SpanStatus): SpanStatus { if (status.status === "ok") return { status: "ok" }; return status.error @@ -100,14 +155,27 @@ function settleSpan( function createSpan( state: InMemoryTelemetryState, - parent: MutableRecordedTelemetrySpan | undefined, + parent: InMemoryParent | undefined, options: SpanOptions, ): MutableRecordedTelemetrySpan { const name = options.name; const attributes = copyAttributes(options.attributes); + const links: RecordedTelemetryLink[] = []; + try { + for (const link of options.links ?? []) { + const copy = copyLink(link); + if (copy) links.push(copy); + } + } catch { + // An unreadable link list must not suppress the operation's span. + } + const id = state.nextSpanId++; return { - id: state.nextSpanId++, - parentId: parent?.id ?? null, + id, + parentId: parent?.localId ?? null, + context: { "pi-memory-scope": state.scope, "pi-memory-span": String(id) }, + parentContext: parent ? { ...parent.context } : null, + links, name, attributes, events: [], @@ -119,11 +187,11 @@ function createSpan( function startInMemorySpan( state: InMemoryTelemetryState, - parent: MutableRecordedTelemetrySpan | undefined, + parent: InMemoryParent | undefined, options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise, ): Promise { - if (parent?.settled) return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); + if (parent?.span?.settled) return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); let recordedSpan: MutableRecordedTelemetrySpan; try { @@ -137,7 +205,18 @@ function startInMemorySpan( startSpan: ( childOptions: SpanOptions, childCallback: (child: TelemetrySpan) => Result | Promise, - ) => startInMemorySpan(state, recordedSpan, childOptions, childCallback), + ) => + startInMemorySpan( + state, + { context: recordedSpan.context, localId: recordedSpan.id, span: recordedSpan }, + childOptions, + childCallback, + ), + addLink(link) { + if (recordedSpan.settled) return; + const copy = copyLink(link); + if (copy) recordedSpan.links.push(copy); + }, addEvent(name, attributes) { if (recordedSpan.settled) return; try { @@ -165,6 +244,8 @@ function startInMemorySpan( }, }; + contextCarriers.set(span, recordedSpan.context); + let result: T | Promise; try { result = callback(span); @@ -185,13 +266,60 @@ function startInMemorySpan( ); } +function createMeasurementRecorder( + state: InMemoryTelemetryState, + kind: RecordedTelemetryMeasurement["kind"], + name: string, + options?: HistogramOptions, +): (value: number, attributes?: SpanAttributes, context?: TelemetryContext) => void { + try { + const description = options?.description; + const unit = options?.unit; + const boundaries = + kind === "histogram" && options?.boundaries !== undefined ? [...options.boundaries] : undefined; + if ( + typeof name !== "string" || + (description !== undefined && typeof description !== "string") || + (unit !== undefined && typeof unit !== "string") + ) + return () => {}; + if (boundaries?.some((value, index) => !Number.isFinite(value) || (index > 0 && value <= boundaries[index - 1]!))) + return () => {}; + const metadata = { + kind, + name, + ...(description === undefined ? {} : { description }), + ...(unit === undefined ? {} : { unit }), + ...(boundaries === undefined ? {} : { boundaries }), + }; + return (value, attributes, context) => { + if (!Number.isFinite(value) || (kind === "counter" && value < 0)) return; + try { + const carrier = context === undefined ? undefined : contextCarriers.get(context); + state.measurements.push({ + ...metadata, + value, + attributes: copyAttributes(attributes), + context: carrier ? { ...carrier } : null, + }); + } catch { + // Measurements are passive; discard the whole failed call. + } + }; + } catch { + return () => {}; + } +} + /** * Backend-neutral reference implementation that records spans in process memory. * Create a fresh instance to isolate tests or independent recording scopes. */ -export class InMemoryTelemetryContext implements TelemetryContext { +export class InMemoryTelemetryContext implements TelemetryContext, TelemetryPropagation, TelemetryMetrics { private readonly state: InMemoryTelemetryState = { + scope: crypto.randomUUID(), spans: [], + measurements: [], nextSpanId: 1, nextEndSequence: 1, }; @@ -200,11 +328,73 @@ export class InMemoryTelemetryContext implements TelemetryContext { return startInMemorySpan(this.state, undefined, options, callback); } + /** Returns a detached carrier for a known span or extracted parent, including settled spans. */ + inject(context: TelemetryContext): TelemetryCarrier | undefined { + const carrier = contextCarriers.get(context); + return carrier ? { ...carrier } : undefined; + } + + /** Unrecognized or unreadable carriers fall back to this recorder's root context. */ + extract(carrier: TelemetryCarrier | undefined): TelemetryContext { + try { + if (!carrier || typeof carrier !== "object" || Array.isArray(carrier)) return this; + if (!Object.hasOwn(carrier, "pi-memory-scope") || !Object.hasOwn(carrier, "pi-memory-span")) return this; + if (Object.values(carrier).some((value) => typeof value !== "string")) return this; + const scope = carrier["pi-memory-scope"]; + const span = carrier["pi-memory-span"]; + if (typeof scope !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/.test(scope)) + return this; + if (typeof span !== "string" || !/^[1-9][0-9]*$/.test(span) || !Number.isSafeInteger(Number(span))) + return this; + const context = { "pi-memory-scope": scope, "pi-memory-span": span }; + const localId = + scope === this.state.scope && this.state.spans.some((candidate) => candidate.id === Number(span)) + ? Number(span) + : null; + const parent: TelemetryContext = { + startSpan: (options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise) => + startInMemorySpan(this.state, { context, localId }, options, callback), + }; + contextCarriers.set(parent, context); + return parent; + } catch { + return this; + } + } + + counter(name: string, options?: InstrumentOptions): TelemetryCounter { + return { add: createMeasurementRecorder(this.state, "counter", name, options) }; + } + + upDownCounter(name: string, options?: InstrumentOptions): TelemetryUpDownCounter { + return { add: createMeasurementRecorder(this.state, "upDownCounter", name, options) }; + } + + histogram(name: string, options?: HistogramOptions): TelemetryHistogram { + return { record: createMeasurementRecorder(this.state, "histogram", name, options) }; + } + + /** Returns detached measurements in recording order, without aggregation. */ + getMetrics(): readonly RecordedTelemetryMeasurement[] { + return this.state.measurements.map((measurement) => ({ + ...measurement, + ...(measurement.boundaries === undefined ? {} : { boundaries: [...measurement.boundaries] }), + attributes: copyAttributes(measurement.attributes), + context: measurement.context ? { ...measurement.context } : null, + })); + } + /** Returns detached snapshots in span-start order. */ getSpans(): readonly RecordedTelemetrySpan[] { return this.state.spans.map((span) => ({ id: span.id, parentId: span.parentId, + context: { ...span.context }, + parentContext: span.parentContext ? { ...span.parentContext } : null, + links: span.links.map((link) => ({ + context: { ...link.context }, + attributes: copyAttributes(link.attributes), + })), name: span.name, attributes: copyAttributes(span.attributes), events: span.events.map((event) => ({ diff --git a/packages/telemetry/src/metrics.ts b/packages/telemetry/src/metrics.ts new file mode 100644 index 000000000..428b8a6bd --- /dev/null +++ b/packages/telemetry/src/metrics.ts @@ -0,0 +1,140 @@ +import type { + ExactTelemetryAttributes, + InferRequiredAndOptionalAttributes, + SpanAttributes, + TelemetryContext, + TelemetrySchemaDefinition, + TelemetryStartAttributeDefinition, +} from "./index.ts"; + +export interface InstrumentOptions { + description?: string; + unit?: string; +} + +export interface HistogramOptions extends InstrumentOptions { + boundaries?: readonly number[]; +} + +export interface TelemetryCounter { + add(value: number, attributes?: SpanAttributes): void; +} + +export interface TelemetryUpDownCounter { + add(value: number, attributes?: SpanAttributes): void; +} + +export interface TelemetryHistogram { + record(value: number, attributes?: SpanAttributes, context?: TelemetryContext): void; +} + +/** Optional adapter capability. Factories and recording methods are synchronous and non-throwing. */ +export interface TelemetryMetrics { + counter(name: string, options?: InstrumentOptions): TelemetryCounter; + upDownCounter(name: string, options?: InstrumentOptions): TelemetryUpDownCounter; + histogram(name: string, options?: HistogramOptions): TelemetryHistogram; +} + +export type TelemetryMetricAttributeDefinition = TelemetryStartAttributeDefinition & { cardinality?: "low" }; + +export type TelemetryMetricDefinition = { + description: string; + unit?: string; + attributes: Record; +} & ({ kind: "counter" | "upDownCounter"; boundaries?: never } | { kind: "histogram"; boundaries?: readonly number[] }); + +type MetricKind = TelemetryMetricDefinition["kind"]; +type MetricSchemas = readonly [TelemetrySchemaDefinition, ...TelemetrySchemaDefinition[]]; + +type MetricsOf = Schema extends { + metrics: infer Metrics extends Record; +} + ? Metrics + : Record; + +type MetricNames< + Schema extends TelemetrySchemaDefinition, + Kind extends MetricKind, +> = Schema extends TelemetrySchemaDefinition + ? { + [Name in keyof MetricsOf]: MetricsOf[Name] extends { kind: Kind } ? Name : never; + }[keyof MetricsOf] & + string + : never; + +type MetricByName< + Schema extends TelemetrySchemaDefinition, + Name extends string, +> = Schema extends TelemetrySchemaDefinition + ? Name extends keyof MetricsOf + ? MetricsOf[Name] + : never + : never; + +type DuplicateMetricNames< + Schemas extends readonly TelemetrySchemaDefinition[], + Seen extends string = never, +> = Schemas extends readonly [ + infer Schema extends TelemetrySchemaDefinition, + ...infer Rest extends readonly TelemetrySchemaDefinition[], +] + ? Extract, Seen> | DuplicateMetricNames & string)> + : never; + +type MetricAttributes = InferRequiredAndOptionalAttributes< + Definition["attributes"] +>; + +type MeasurementArguments< + Definition extends TelemetryMetricDefinition, + Attributes extends MetricAttributes, +> = { + [Name in keyof Definition["attributes"]]: Definition["attributes"][Name]["required"] extends true ? Name : never; +}[keyof Definition["attributes"]] extends never + ? [attributes?: ExactTelemetryAttributes, Attributes>] + : [attributes: ExactTelemetryAttributes, Attributes>]; + +export type SchemaTelemetryCounter = { + add = MetricAttributes>( + value: number, + ...args: MeasurementArguments + ): void; +}; + +export type SchemaTelemetryHistogram = { + record = MetricAttributes>( + value: number, + ...args: [...MeasurementArguments, context?: TelemetryContext] + ): void; +}; + +export interface TypedTelemetryMetrics { + counter>( + name: Name, + ): SchemaTelemetryCounter>; + upDownCounter>( + name: Name, + ): SchemaTelemetryCounter>; + histogram>( + name: Name, + ): SchemaTelemetryHistogram>; +} + +/** Bind instrument names, kinds, metadata, and exact measurement attributes to one or more schemas. */ +export function createTypedMetrics( + metrics: TelemetryMetrics, + schemas: Schemas & + ([DuplicateMetricNames] extends [never] + ? unknown + : { readonly "duplicate telemetry metric names": DuplicateMetricNames }), +): TypedTelemetryMetrics { + const definitions = new Map(); + for (const schema of schemas) { + for (const [name, definition] of Object.entries(schema.metrics ?? {})) definitions.set(name, definition); + } + return { + counter: (name: string) => metrics.counter(name, definitions.get(name)), + upDownCounter: (name: string) => metrics.upDownCounter(name, definitions.get(name)), + histogram: (name: string) => metrics.histogram(name, definitions.get(name)), + } as unknown as TypedTelemetryMetrics; +} diff --git a/packages/telemetry/src/noop.ts b/packages/telemetry/src/noop.ts index 48244d3a3..5527b47da 100644 --- a/packages/telemetry/src/noop.ts +++ b/packages/telemetry/src/noop.ts @@ -1,4 +1,5 @@ -import type { SpanOptions, TelemetryContext, TelemetrySpan } from "./index.ts"; +import type { SpanOptions, TelemetryContext, TelemetryPropagation, TelemetrySpan } from "./index.ts"; +import type { TelemetryCounter, TelemetryHistogram, TelemetryMetrics, TelemetryUpDownCounter } from "./metrics.ts"; function startNoopSpan(_options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { try { @@ -10,6 +11,7 @@ function startNoopSpan(_options: SpanOptions, callback: (span: TelemetrySpan) const noopTelemetrySpan: TelemetrySpan = { startSpan: startNoopSpan, + addLink: () => {}, addEvent: () => {}, setAttributes: () => {}, setStatus: () => {}, @@ -18,3 +20,20 @@ Object.freeze(noopTelemetrySpan); /** Shared telemetry context used when an application does not provide one. */ export const NOOP_TELEMETRY_CONTEXT: TelemetryContext = noopTelemetrySpan; + +/** Shared propagation capability that neither inspects nor retains its inputs. */ +export const NOOP_TELEMETRY_PROPAGATION: TelemetryPropagation = Object.freeze({ + inject: () => undefined, + extract: () => NOOP_TELEMETRY_CONTEXT, +}); + +const noopCounter: TelemetryCounter = Object.freeze({ add: () => {} }); +const noopUpDownCounter: TelemetryUpDownCounter = Object.freeze({ add: () => {} }); +const noopHistogram: TelemetryHistogram = Object.freeze({ record: () => {} }); + +/** Shared instruments that never inspect or retain metadata, measurements, or contexts. */ +export const NOOP_TELEMETRY_METRICS: TelemetryMetrics = Object.freeze({ + counter: () => noopCounter, + upDownCounter: () => noopUpDownCounter, + histogram: () => noopHistogram, +}); diff --git a/packages/telemetry/src/testing/conformance.ts b/packages/telemetry/src/testing/conformance.ts index f90ba588b..df147246e 100644 --- a/packages/telemetry/src/testing/conformance.ts +++ b/packages/telemetry/src/testing/conformance.ts @@ -1,5 +1,13 @@ import { deepStrictEqual, doesNotThrow, fail, ok, strictEqual } from "node:assert/strict"; -import type { SpanAttributes, SpanOptions, SpanStatus, TelemetrySpan } from "../index.ts"; +import { + NOOP_TELEMETRY_CONTEXT, + type SpanAttributes, + type SpanLink, + type SpanOptions, + type SpanStatus, + type TelemetryCarrier, + type TelemetrySpan, +} from "../index.ts"; import type { RecordedTelemetrySpan } from "../memory.ts"; import type { TelemetryAdapterConformanceCase, @@ -62,6 +70,173 @@ export function createTelemetryAdapterConformance( factory: TelemetryAdapterFixtureFactory, ): readonly TelemetryAdapterConformanceCase[] { return [ + createCase( + factory, + "metrics", + "records ordered measurements and optional histogram exemplars", + async (fixture) => { + if (!fixture.metrics) return; + ok(fixture.getMetrics, "Metrics fixtures require a normalized measurement reader"); + fixture.metrics + .counter("bytes", { description: "Encoded bytes", unit: "By" }) + .add(12, { direction: "out" }); + const active = fixture.metrics.upDownCounter("active"); + active.add(1); + active.add(-1); + const histogram = fixture.metrics.histogram("duration", { unit: "ms", boundaries: [1, 10] }); + await fixture.context.startSpan({ name: "measured" }, (span) => + histogram.record(5, { method: "read" }, span), + ); + const measurements = await fixture.getMetrics(); + deepStrictEqual( + measurements.map((measurement) => [measurement.kind, measurement.name, measurement.value]), + [ + ["counter", "bytes", 12], + ["upDownCounter", "active", 1], + ["upDownCounter", "active", -1], + ["histogram", "duration", 5], + ], + ); + deepStrictEqual(measurements[0]?.attributes, { direction: "out" }); + strictEqual(measurements[0]?.description, "Encoded bytes"); + strictEqual(measurements[0]?.unit, "By"); + deepStrictEqual(measurements[3]?.boundaries, [1, 10]); + deepStrictEqual(measurements[3]?.context, findSpan(await fixture.getSpans(), "measured").context); + }, + ), + createCase( + factory, + "metrics", + "copies inputs and suppresses invalid measurements atomically", + async (fixture) => { + if (!fixture.metrics) return; + ok(fixture.getMetrics, "Metrics fixtures require a normalized measurement reader"); + const boundaries = [1, 10]; + const options = { description: "Original", boundaries }; + const histogram = fixture.metrics.histogram("snapshot", options); + boundaries.push(100); + options.description = "Changed"; + const attributes = { tags: ["original"] }; + histogram.record(-1, attributes); + attributes.tags.push("changed"); + const counter = fixture.metrics.counter("invalid"); + for (const value of [NaN, Infinity, -Infinity, -1]) doesNotThrow(() => counter.add(value)); + doesNotThrow(() => counter.add(1, { partial: true, broken: unreadable(["value"]) })); + doesNotThrow(() => fixture.metrics?.histogram("bad-options", unreadable({ unit: "ms" })).record(1)); + doesNotThrow(() => fixture.metrics?.histogram("bad-boundaries", { boundaries: [2, 1] }).record(1)); + const measurements = await fixture.getMetrics(); + strictEqual(measurements.length, 1); + deepStrictEqual(measurements[0]?.attributes, { tags: ["original"] }); + deepStrictEqual(measurements[0]?.boundaries, [1, 10]); + strictEqual(measurements[0]?.description, "Original"); + strictEqual(measurements[0]?.context, null); + }, + ), + createCase(factory, "links", "records initial and late links without changing parents", async (fixture) => { + let target: TelemetrySpan | undefined; + await fixture.context.startSpan({ name: "target" }, (span) => { + target = span; + }); + if (!target) throw new Error("Expected target span"); + const targetSpan = target; + const attributes = { reason: ["initial"] }; + await fixture.context.startSpan({ name: "linked", links: [{ context: target, attributes }] }, (span) => { + attributes.reason.push("mutated"); + span.addLink({ context: targetSpan, attributes: { reason: "late" } }); + span.addLink({ context: NOOP_TELEMETRY_CONTEXT }); + }); + const spans = await fixture.getSpans(); + const linked = findSpan(spans, "linked"); + strictEqual(linked.parentId, null); + strictEqual(linked.parentContext, null); + deepStrictEqual(linked.links, [ + { context: findSpan(spans, "target").context, attributes: { reason: ["initial"] } }, + { context: findSpan(spans, "target").context, attributes: { reason: "late" } }, + ]); + }), + createCase( + factory, + "links", + "suppresses malformed links atomically and ignores settled sources", + async (fixture) => { + let source: TelemetrySpan | undefined; + await fixture.context.startSpan({ name: "link-source" }, (span) => { + source = span; + doesNotThrow(() => span.addLink(unreadable({ context: span }))); + doesNotThrow(() => + span.addLink({ context: span, attributes: { partial: true, broken: unreadable(["value"]) } }), + ); + }); + if (!source) throw new Error("Expected source span"); + const sourceSpan = source; + doesNotThrow(() => sourceSpan.addLink({ context: sourceSpan })); + deepStrictEqual(findSpan(await fixture.getSpans(), "link-source").links, []); + }, + ), + createCase( + factory, + "propagation", + "roundtrips parent identity through JSON without retaining the carrier", + async (fixture) => { + if (!fixture.propagation) return; + const propagation = fixture.propagation; + await fixture.context.startSpan({ name: "sender" }, async (span) => { + const carrier = propagation.inject(span); + ok(carrier); + const wire = JSON.parse(JSON.stringify(carrier)) as TelemetryCarrier; + const parent = propagation.extract(wire); + for (const key of Object.keys(wire)) wire[key] = "mutated"; + await parent.startSpan({ name: "receiver" }, () => 42); + deepStrictEqual(propagation.inject(span), carrier); + }); + const spans = await fixture.getSpans(); + deepStrictEqual(findSpan(spans, "receiver").parentContext, findSpan(spans, "sender").context); + }, + ), + createCase(factory, "propagation", "unknown and unreadable carriers produce recorded roots", async (fixture) => { + if (!fixture.propagation) return; + strictEqual(fixture.propagation.inject(NOOP_TELEMETRY_CONTEXT), undefined); + for (const [index, carrier] of [ + undefined, + {}, + { unknown: "format" }, + unreadable({ bad: "value" }), + ].entries()) { + let result = 0; + doesNotThrow(() => { + const context = fixture.propagation?.extract(carrier); + if (!context) throw new Error("Expected root context"); + void context.startSpan({ name: `fallback-${index}` }, () => { + result = 42; + }); + }); + strictEqual(result, 42); + } + for (const span of await fixture.getSpans()) strictEqual(span.parentContext, null); + }), + createCase(factory, "propagation", "propagates across fresh adapters and multiple hops", async (fixture) => { + if (!fixture.propagation) return; + await using receiver = await factory(); + await using worker = await factory(); + if (!receiver.propagation || !worker.propagation) throw new Error("Expected propagation on each fixture"); + let senderCarrier: TelemetryCarrier | undefined; + await fixture.context.startSpan({ name: "sender" }, (span) => { + senderCarrier = fixture.propagation?.inject(span); + }); + ok(senderCarrier); + let receiverCarrier: TelemetryCarrier | undefined; + await receiver.propagation + .extract(JSON.parse(JSON.stringify(senderCarrier)) as TelemetryCarrier) + .startSpan({ name: "receiver" }, (span) => { + receiverCarrier = receiver.propagation?.inject(span); + }); + ok(receiverCarrier); + await worker.propagation + .extract(JSON.parse(JSON.stringify(receiverCarrier)) as TelemetryCarrier) + .startSpan({ name: "worker" }, () => {}); + deepStrictEqual(findSpan(await receiver.getSpans(), "receiver").parentContext, senderCarrier); + deepStrictEqual(findSpan(await worker.getSpans(), "worker").parentContext, receiverCarrier); + }), createCase( factory, "callback lifecycle", diff --git a/packages/telemetry/src/testing/types.ts b/packages/telemetry/src/testing/types.ts index 1267083f3..d0940790a 100644 --- a/packages/telemetry/src/testing/types.ts +++ b/packages/telemetry/src/testing/types.ts @@ -1,9 +1,12 @@ -import type { TelemetryContext } from "../index.ts"; -import type { RecordedTelemetrySpan } from "../memory.ts"; +import type { TelemetryContext, TelemetryMetrics, TelemetryPropagation } from "../index.ts"; +import type { RecordedTelemetryMeasurement, RecordedTelemetrySpan } from "../memory.ts"; /** A fresh adapter instance and normalized snapshot reader owned by one conformance case. */ export interface TelemetryAdapterFixture extends AsyncDisposable { readonly context: TelemetryContext; + readonly propagation?: TelemetryPropagation; + readonly metrics?: TelemetryMetrics; + getMetrics?(): Promise; getSpans(): Promise; } diff --git a/packages/telemetry/test/conformance.test.ts b/packages/telemetry/test/conformance.test.ts index ba98cdec5..7203b1247 100644 --- a/packages/telemetry/test/conformance.test.ts +++ b/packages/telemetry/test/conformance.test.ts @@ -6,6 +6,9 @@ const conformance = createTelemetryAdapterConformance(() => { const context = new InMemoryTelemetryContext(); return Promise.resolve({ context, + propagation: context, + metrics: context, + getMetrics: () => Promise.resolve(context.getMetrics()), getSpans: () => Promise.resolve(context.getSpans()), [Symbol.asyncDispose]: () => Promise.resolve(), }); diff --git a/packages/telemetry/test/metrics.test.ts b/packages/telemetry/test/metrics.test.ts new file mode 100644 index 000000000..11e691f17 --- /dev/null +++ b/packages/telemetry/test/metrics.test.ts @@ -0,0 +1,196 @@ +import { describe, expect, expectTypeOf, it } from "vitest"; +import { + createTypedMetrics, + defineTelemetrySchema, + type HistogramOptions, + InMemoryTelemetryContext, + type InstrumentOptions, + NOOP_TELEMETRY_CONTEXT, + NOOP_TELEMETRY_METRICS, + type SpanAttributes, + type TelemetryCarrier, + type TelemetryMetricDefinition, +} from "../src/index.ts"; + +function unreadable(value: T): T { + return new Proxy(value, { + get() { + throw new Error("read"); + }, + ownKeys() { + throw new Error("enumerate"); + }, + }); +} + +const traffic = defineTelemetrySchema({ + version: 1, + spans: {}, + metrics: { + "network.bytes": { + kind: "counter", + description: "Encoded bytes", + unit: "By", + attributes: { + direction: { + type: "string", + values: ["in", "out"], + required: true, + description: "Direction", + cardinality: "low", + }, + }, + }, + "network.active": { kind: "upDownCounter", description: "Open links", attributes: {} }, + }, +} as const); +const latency = defineTelemetrySchema({ + version: 1, + spans: {}, + metrics: { + "call.duration": { + kind: "histogram", + description: "Call duration", + unit: "ms", + boundaries: [1, 10, 100], + attributes: { method: { type: "string", required: false, description: "Method" } }, + }, + }, +} as const); + +describe("metrics", () => { + it("binds schema names, kinds, metadata, and exact measurement attributes", async () => { + const adapter = new InMemoryTelemetryContext(); + const metrics = createTypedMetrics(adapter, [traffic, latency]); + metrics.counter("network.bytes").add(128, { direction: "out" }); + metrics.upDownCounter("network.active").add(1); + metrics.upDownCounter("network.active").add(-1); + await adapter.startSpan({ name: "request" }, (span) => { + metrics.histogram("call.duration").record(12, { method: "read" }, span); + metrics.histogram("call.duration").record(13); + metrics.histogram("call.duration").record(14, undefined, span); + }); + const measurements = adapter.getMetrics(); + expect(measurements[0]).toMatchObject({ + kind: "counter", + name: "network.bytes", + description: "Encoded bytes", + unit: "By", + value: 128, + attributes: { direction: "out" }, + }); + expect(measurements[3]).toMatchObject({ + kind: "histogram", + boundaries: [1, 10, 100], + context: adapter.getSpans()[0]?.context, + }); + expect(measurements[4]?.context).toBeNull(); + expect(measurements[5]?.context).toEqual(adapter.getSpans()[0]?.context); + const compileTimeFailures = () => { + // @ts-expect-error counter names are restricted to counters + metrics.counter("call.duration"); + // @ts-expect-error histogram names are restricted to histograms + metrics.histogram("network.bytes"); + // @ts-expect-error unknown names are rejected + metrics.upDownCounter("unknown"); + // @ts-expect-error required metric attributes cannot be omitted + metrics.counter("network.bytes").add(1); + // @ts-expect-error closed-set metric values are exact + metrics.counter("network.bytes").add(1, { direction: "sideways" }); + const extra = { direction: "in", unknown: true } as const; + // @ts-expect-error variables containing unknown attributes are rejected + metrics.counter("network.bytes").add(1, extra); + // @ts-expect-error an empty attribute schema rejects every attribute + metrics.upDownCounter("network.active").add(1, { unknown: true }); + // @ts-expect-error duplicate metric names are rejected + createTypedMetrics(adapter, [traffic, traffic]); + const invalid: TelemetryMetricDefinition = { + kind: "counter", + description: "Invalid", + attributes: { + // @ts-expect-error high-cardinality attributes do not belong on metrics + request: { type: "string", required: false, description: "Request", cardinality: "high" }, + }, + }; + return invalid; + }; + expectTypeOf(compileTimeFailures).toBeFunction(); + expect(() => JSON.stringify(traffic)).not.toThrow(); + }); + + it("requires histogram attributes when the schema declares them", () => { + const schema = defineTelemetrySchema({ + version: 1, + spans: {}, + metrics: { + required: { + kind: "histogram", + description: "Required tag", + attributes: { + cache: { type: "boolean", required: true, description: "Cache hit" }, + }, + }, + }, + } as const); + const metrics = createTypedMetrics(new InMemoryTelemetryContext(), [schema]); + metrics.histogram("required").record(1, { cache: true }); + const compileTimeFailures = () => { + // @ts-expect-error histogram attributes are required + metrics.histogram("required").record(1); + // @ts-expect-error histogram values match their attribute schema + metrics.histogram("required").record(1, { cache: "true" }); + // @ts-expect-error optional exemplar context does not replace required attributes + metrics.histogram("required").record(1, NOOP_TELEMETRY_CONTEXT); + }; + expectTypeOf(compileTimeFailures).toBeFunction(); + }); + + it("returns detached measurement snapshots including exemplars and boundaries", async () => { + const telemetry = new InMemoryTelemetryContext(); + const histogram = telemetry.histogram("duration", { boundaries: [1, 2] }); + await telemetry.startSpan({ name: "request" }, (span) => histogram.record(1, { tags: ["original"] }, span)); + const measurement = telemetry.getMetrics()[0]; + if (!measurement?.context) throw new Error("Expected measurement context"); + (measurement.boundaries as number[]).push(3); + (measurement.attributes as SpanAttributes).tags = ["mutated"]; + (measurement.context as TelemetryCarrier)["pi-memory-span"] = "999"; + expect(telemetry.getMetrics()[0]).toMatchObject({ + boundaries: [1, 2], + attributes: { tags: ["original"] }, + context: telemetry.getSpans()[0]?.context, + }); + }); + + it("ignores malformed metadata, nonfinite values, and unreadable attributes", () => { + const telemetry = new InMemoryTelemetryContext(); + telemetry.counter("bad", unreadable({ unit: "By" })).add(1); + telemetry.histogram("bad", unreadable({ boundaries: [1] })).record(1); + for (const boundaries of [[1, 1], [2, 1], [NaN], [Infinity]]) + telemetry.histogram("bad", { boundaries }).record(1); + for (const value of [NaN, Infinity, -Infinity]) { + telemetry.counter("bad").add(value); + telemetry.upDownCounter("bad").add(value); + telemetry.histogram("bad").record(value); + } + telemetry.counter("bad").add(-1); + telemetry.histogram("bad").record(1, unreadable({ tag: "value" })); + expect(telemetry.getMetrics()).toEqual([]); + telemetry.histogram("valid").record(-1, {}, NOOP_TELEMETRY_CONTEXT); + expect(telemetry.getMetrics()[0]?.context).toBeNull(); + }); + + it("shares frozen no-op instruments without inspecting arguments", () => { + const options = unreadable({ unit: "By" }); + const attributes = unreadable({ secret: "value" }); + const counter = NOOP_TELEMETRY_METRICS.counter("bytes", options); + const active = NOOP_TELEMETRY_METRICS.upDownCounter("active", options); + const histogram = NOOP_TELEMETRY_METRICS.histogram("duration", options); + expect(counter).toBe(NOOP_TELEMETRY_METRICS.counter("other")); + expect(Object.isFrozen(counter)).toBe(true); + expect(Object.isFrozen(active)).toBe(true); + expect(Object.isFrozen(histogram)).toBe(true); + expect(() => counter.add(1, attributes)).not.toThrow(); + expect(() => active.add(-1, attributes)).not.toThrow(); + expect(() => histogram.record(1, attributes, unreadable(NOOP_TELEMETRY_CONTEXT))).not.toThrow(); + }); +}); diff --git a/packages/telemetry/test/propagation-worker.ts b/packages/telemetry/test/propagation-worker.ts new file mode 100644 index 000000000..50f8d47ac --- /dev/null +++ b/packages/telemetry/test/propagation-worker.ts @@ -0,0 +1,6 @@ +import { InMemoryTelemetryContext, type TelemetryCarrier } from "../src/index.ts"; + +const telemetry = new InMemoryTelemetryContext(); +const carrier = JSON.parse(process.argv[2] ?? "{}") as TelemetryCarrier; +await telemetry.extract(carrier).startSpan({ name: "remote" }, () => {}); +process.stdout.write(JSON.stringify(telemetry.getSpans())); diff --git a/packages/telemetry/test/propagation.test.ts b/packages/telemetry/test/propagation.test.ts new file mode 100644 index 000000000..4707a4a2f --- /dev/null +++ b/packages/telemetry/test/propagation.test.ts @@ -0,0 +1,135 @@ +import { execFileSync } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { describe, expect, it } from "vitest"; +import { + InMemoryTelemetryContext, + NOOP_TELEMETRY_CONTEXT, + NOOP_TELEMETRY_PROPAGATION, + type RecordedTelemetrySpan, + type SpanAttributes, + type TelemetryCarrier, + type TelemetrySpan, +} from "../src/index.ts"; + +function unreadable(value: T): T { + return new Proxy(value, { + get() { + throw new Error("unreadable"); + }, + }); +} + +describe("propagation and links", () => { + it("continues a parent in another Node process without shared recorder state", async () => { + const telemetry = new InMemoryTelemetryContext(); + let carrier: TelemetryCarrier | undefined; + await telemetry.startSpan({ name: "caller" }, (span) => { + carrier = telemetry.inject(span); + }); + const output = execFileSync( + process.execPath, + [fileURLToPath(new URL("./propagation-worker.ts", import.meta.url)), JSON.stringify(carrier)], + { encoding: "utf8" }, + ); + const recorded = JSON.parse(output) as RecordedTelemetrySpan[]; + expect(recorded).toHaveLength(1); + expect(recorded[0]?.parentContext).toEqual(carrier); + expect(recorded[0]?.parentId).toBeNull(); + expect(recorded[0]?.context).not.toEqual(carrier); + }); + + it("isolates crossed concurrent remote parents and links shared roots", async () => { + const caller = new InMemoryTelemetryContext(); + const provider = new InMemoryTelemetryContext(); + let activation: TelemetrySpan | undefined; + let finishActivation: (() => void) | undefined; + const gate = new Promise((resolve) => { + finishActivation = resolve; + }); + let shared: Promise | undefined; + await Promise.all( + ["first", "later"].map((name) => + caller.startSpan({ name }, async (trigger) => { + const parent = provider.extract(caller.inject(trigger)); + await parent.startSpan({ name: `${name}.method` }, async (getter) => { + if (!shared) + shared = provider.startSpan({ name: "activation", links: [{ context: getter }] }, async (span) => { + activation = span; + await gate; + }); + if (activation) getter.addLink({ context: activation }); + if (name === "later") finishActivation?.(); + await shared; + }); + }), + ), + ); + const spans = provider.getSpans(); + const root = spans.find((span) => span.name === "activation"); + expect(root?.parentContext).toBeNull(); + const first = spans.find((span) => span.name === "first.method"); + const later = spans.find((span) => span.name === "later.method"); + expect(first?.parentContext).toEqual(caller.getSpans()[0]?.context); + expect(later?.parentContext).toEqual(caller.getSpans()[1]?.context); + expect(root?.links[0]?.context).toEqual(first?.context); + expect(later?.links[0]?.context).toEqual(root?.context); + }); + + it("detaches carrier and link snapshots and retains settled target identities", async () => { + const telemetry = new InMemoryTelemetryContext(); + let target: TelemetrySpan | undefined; + await telemetry.startSpan({ name: "target" }, (span) => { + target = span; + }); + if (!target) throw new Error("Expected target"); + const original = telemetry.inject(target); + const injected = telemetry.inject(target); + if (!injected) throw new Error("Expected carrier"); + injected["pi-memory-span"] = "999"; + expect(telemetry.inject(target)).toEqual(original); + await telemetry.startSpan( + { name: "linked", links: [{ context: target, attributes: { tags: ["original"] } }] }, + () => {}, + ); + const [first, linked] = telemetry.getSpans(); + if (!first || !linked?.links[0]) throw new Error("Expected spans and link"); + (first.context as TelemetryCarrier)["pi-memory-span"] = "123"; + (linked.links[0].context as TelemetryCarrier)["pi-memory-span"] = "456"; + (linked.links[0].attributes as SpanAttributes).tags = ["mutated"]; + expect(telemetry.getSpans()[1]?.links).toEqual([{ context: original, attributes: { tags: ["original"] } }]); + }); + + it("links a new root to an extracted remote target", async () => { + const sender = new InMemoryTelemetryContext(); + const receiver = new InMemoryTelemetryContext(); + let carrier: TelemetryCarrier | undefined; + await sender.startSpan({ name: "trigger" }, (span) => { + carrier = sender.inject(span); + }); + const target = receiver.extract(JSON.parse(JSON.stringify(carrier)) as TelemetryCarrier); + await receiver.startSpan( + { name: "background", links: [{ context: target, attributes: { reason: "restart" } }] }, + (span) => { + span.addLink({ context: target }); + }, + ); + const [recorded] = receiver.getSpans(); + expect(recorded?.parentContext).toBeNull(); + expect(recorded?.links).toEqual([ + { context: carrier, attributes: { reason: "restart" } }, + { context: carrier, attributes: {} }, + ]); + }); + + it("rejects malformed memory carriers and never inspects no-op payloads", async () => { + const telemetry = new InMemoryTelemetryContext(); + for (const carrier of [{ "pi-memory-scope": "bad", "pi-memory-span": "1" }, unreadable({ bad: "value" })]) { + expect(telemetry.extract(carrier)).toBe(telemetry); + } + expect(NOOP_TELEMETRY_PROPAGATION.inject(unreadable(NOOP_TELEMETRY_CONTEXT))).toBeUndefined(); + expect(NOOP_TELEMETRY_PROPAGATION.extract(unreadable({ bad: "value" }))).toBe(NOOP_TELEMETRY_CONTEXT); + await NOOP_TELEMETRY_CONTEXT.startSpan({ name: "noop" }, (span) => { + expect(() => span.addLink(unreadable({ context: span }))).not.toThrow(); + }); + }); +}); diff --git a/scripts/check-entry-graphs.mjs b/scripts/check-entry-graphs.mjs index da4cce2ee..7dde4e553 100644 --- a/scripts/check-entry-graphs.mjs +++ b/scripts/check-entry-graphs.mjs @@ -59,8 +59,9 @@ const BUDGETS = { ".": { // Tool argument validation reaches TypeBox; provider-session creation reaches pi-ai's lean UUID utility; the // memory and SQLite storages share scan order and cursor handling. Values from Chord's root (copyJson, - // replicatedState) reach its host runtime through the root barrel. - maxFiles: 66, + // replicatedState) reach its host runtime through the root barrel. Chord's context reaches pi-telemetry's root + // (index, noop, metrics, memory) for its NOOP telemetry default. + maxFiles: 70, forbid: [ "packages/ai/src/index.ts", "packages/ai/src/utils/typebox-helpers.ts",