From 35f5c265d3d0af7eea3889dbfde03e77e23b8f11 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Thu, 6 Aug 2026 11:44:41 +0200 Subject: [PATCH] feat: add telemetry reference adapter and span composition --- README.md | 2 +- packages/agent/CHANGELOG.md | 2 +- packages/agent/src/harness/agent-harness.ts | 21 +- packages/agent/src/harness/telemetry.ts | 4 + packages/agent/src/index.ts | 11 +- packages/agent/test/harness/telemetry.test.ts | 33 +- packages/coding-agent/CHANGELOG.md | 2 +- packages/telemetry/CHANGELOG.md | 2 +- packages/telemetry/README.md | 461 +++++++++++++++++- packages/telemetry/package.json | 4 + packages/telemetry/src/index.ts | 123 ++++- packages/telemetry/src/memory.ts | 219 +++++++++ packages/telemetry/src/noop.ts | 20 + packages/telemetry/src/testing/conformance.ts | 315 ++++++++++++ packages/telemetry/src/testing/index.ts | 6 + packages/telemetry/src/testing/types.ts | 18 + packages/telemetry/test/conformance.test.ts | 46 ++ packages/telemetry/test/telemetry.test.ts | 83 ++++ tsconfig.json | 1 + vitest.base.ts | 2 + 20 files changed, 1331 insertions(+), 44 deletions(-) create mode 100644 packages/telemetry/src/memory.ts create mode 100644 packages/telemetry/src/noop.ts create mode 100644 packages/telemetry/src/testing/conformance.ts create mode 100644 packages/telemetry/src/testing/index.ts create mode 100644 packages/telemetry/src/testing/types.ts create mode 100644 packages/telemetry/test/conformance.test.ts diff --git a/README.md b/README.md index c90ba28d0..e68f9385d 100644 --- a/README.md +++ b/README.md @@ -27,7 +27,7 @@ To learn more about Pi: | Package | Description | |---------|-------------| -| **[@earendil-works/pi-telemetry](packages/telemetry)** | Vendor-neutral telemetry contracts and typed schema utilities | +| **[@earendil-works/pi-telemetry](packages/telemetry)** | Vendor-neutral telemetry contracts, reference adapter, conformance tests, and typed schemas | | **[@earendil-works/pi-ai](packages/ai)** | Unified multi-provider LLM API (OpenAI, Anthropic, Google, etc.) | | **[@earendil-works/pi-agent-core](packages/agent)** | Agent runtime with tool calling and state management | | **[@earendil-works/pi-coding-agent](packages/coding-agent)** | Interactive coding agent CLI | diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 7241660f4..318148970 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -10,7 +10,7 @@ ### Added -- Added typed AI-request and harness telemetry schemas, callback helpers, and a generated schema reference. +- Added typed AI-request and harness telemetry schemas, their combined schema tuple, callback helpers, and a generated schema reference. - Added bounded `Session.findEntriesOnBranch()` and `findEntryOnBranch()` queries with explicit traversal, filtering, ordering, and limit options. - Added a compile-complete `AgentHarness` v2 scaffold; unfinished operation paths reject with `HarnessNotImplemented` while durable execution is implemented. - Added `JsonlSessionRepo`, a v4 append-only JSONL session repository with metadata validation and shared storage semantics ([#7611](https://github.com/earendil-works/pi/pull/7611) by [@davidbrai](https://github.com/davidbrai)). diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 550d19ce6..3802900db 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -22,6 +22,7 @@ import type { Session, SessionTree, } from "./session/index.ts"; +import type { TelemetryContext } from "./telemetry.ts"; import type { AgentHarnessResources, PromptTemplate, Skill } from "./types.ts"; export class LaneBusy extends TaggedError("LaneBusy")<{ @@ -233,24 +234,6 @@ class UnavailableRegistry implements Hooks, Events { } } -export interface ExecutionSpan extends ExecutionContext { - addEvent(name: string, attributes?: SpanAttributes): void; - setAttributes(attributes: SpanAttributes): void; - end(result: SpanEnd): void; -} - -export interface ExecutionContext { - startSpan(name: string, attributes?: SpanAttributes): ExecutionSpan; -} -export interface SpanAttributes { - [name: string]: string | number | boolean | undefined; -} -export interface SpanEnd { - status: "ok" | "error"; - error?: { name: string; message: string }; - attributes?: SpanAttributes; -} - export type HarnessTool = AgentTool & { replay?: "never" | "safe" }; export type Resources = AgentHarnessResources; export type StreamOptions = SimpleStreamOptions; @@ -276,7 +259,7 @@ export interface AgentHarnessOptions { drive?: "automatic" | "manual"; toProviderMessages?: (messages: AgentMessage[]) => Message[] | Promise; entryProjectors?: Record; - context?: ExecutionContext; + context?: TelemetryContext; } export interface WatchHandle { diff --git a/packages/agent/src/harness/telemetry.ts b/packages/agent/src/harness/telemetry.ts index 159108279..c7ba25c57 100644 --- a/packages/agent/src/harness/telemetry.ts +++ b/packages/agent/src/harness/telemetry.ts @@ -36,6 +36,7 @@ export type { TelemetrySpan, TelemetrySpanDefinition, TelemetryStartAttributeDefinition, + TypedSpanStarter, } from "@earendil-works/pi-telemetry"; export const AI_TELEMETRY_SCHEMA = { @@ -570,6 +571,9 @@ export const HARNESS_TELEMETRY_SCHEMA = { }, } as const satisfies TelemetrySchemaDefinition; +/** Combined typed span vocabulary for agent-owned AI-request and harness telemetry. */ +export const AGENT_TELEMETRY_SCHEMAS = [AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA] as const; + export type HarnessSpanName = TelemetrySchemaSpanName; export type HarnessSpanStartAttributes = TelemetrySchemaSpanStartAttributes< typeof HARNESS_TELEMETRY_SCHEMA, diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 998c6ed54..b1315b68b 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -8,6 +8,8 @@ export type { InferOptionalAttributes, InferRequiredAndOptionalAttributes, InferStartAttributes, + RecordedTelemetryEvent, + RecordedTelemetrySpan, SchemaTelemetrySpan, SpanAttributes, SpanAttributes as TelemetrySpanAttributes, @@ -30,8 +32,14 @@ export type { TelemetrySpan, TelemetrySpanDefinition, TelemetryStartAttributeDefinition, + TypedSpanStarter, +} from "@earendil-works/pi-telemetry"; +export { + createTypedSpanStarter, + defineTelemetrySchema, + InMemoryTelemetryContext, + NOOP_TELEMETRY_CONTEXT, } from "@earendil-works/pi-telemetry"; -export { defineTelemetrySchema, NOOP_TELEMETRY_CONTEXT } from "@earendil-works/pi-telemetry"; export * from "./agent.ts"; // Loop functions export * from "./agent-loop.ts"; @@ -92,6 +100,7 @@ export type { HarnessTelemetrySpan, } from "./harness/telemetry.ts"; export { + AGENT_TELEMETRY_SCHEMAS, AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA, startAiSpan, diff --git a/packages/agent/test/harness/telemetry.test.ts b/packages/agent/test/harness/telemetry.test.ts index e344a4350..cb2b34023 100644 --- a/packages/agent/test/harness/telemetry.test.ts +++ b/packages/agent/test/harness/telemetry.test.ts @@ -1,9 +1,10 @@ import { readFileSync } from "node:fs"; import { resolve } from "node:path"; -import { NOOP_TELEMETRY_CONTEXT, type TelemetryContext } from "@earendil-works/pi-telemetry"; +import { createTypedSpanStarter, NOOP_TELEMETRY_CONTEXT, type TelemetryContext } from "@earendil-works/pi-telemetry"; import { describe, expect, expectTypeOf, it } from "vitest"; import { renderAgentTelemetrySchemaMarkdown } from "../../scripts/generate-telemetry-docs.ts"; import { + AGENT_TELEMETRY_SCHEMAS, AI_TELEMETRY_SCHEMA, type AiSpanEndAttributes, type AiSpanStartAttributes, @@ -18,6 +19,7 @@ describe("agent telemetry schemas", () => { it("serializes both schemas and generates the checked-in reference", () => { expect(() => JSON.stringify(AI_TELEMETRY_SCHEMA)).not.toThrow(); expect(() => JSON.stringify(HARNESS_TELEMETRY_SCHEMA)).not.toThrow(); + expect(AGENT_TELEMETRY_SCHEMAS).toEqual([AI_TELEMETRY_SCHEMA, HARNESS_TELEMETRY_SCHEMA]); expect(Object.keys(HARNESS_TELEMETRY_SCHEMA.spans)).toEqual([ "pi.harness.run", "pi.harness.compaction", @@ -35,6 +37,35 @@ describe("agent telemetry schemas", () => { expect(actual).toBe(renderAgentTelemetrySchemaMarkdown()); }); + it("starts AI-request and harness spans through one composed typed starter", async () => { + const startSpan = createTypedSpanStarter(NOOP_TELEMETRY_CONTEXT, AGENT_TELEMETRY_SCHEMAS); + await startSpan( + "pi.harness.step", + { + "pi.lane.name": "main", + "pi.operation.id": "operation", + "pi.step.kind": "assistant", + "pi.step.attempt": 1, + }, + async (stepSpan, startChildSpan) => { + stepSpan.setAttributes({ "pi.step.outcome": "succeeded" }); + await startChildSpan( + "pi.ai.request", + { + "pi.ai.operation": "stream", + "pi.ai.provider": "provider", + "pi.ai.model": "model", + "pi.ai.api": "api", + "pi.ai.streaming": true, + }, + (requestSpan) => { + requestSpan.setAttributes({ "pi.ai.response.stop_reason": "stop" }); + }, + ); + }, + ); + }); + it("infers exact AI start and optional end attributes", async () => { type Start = AiSpanStartAttributes<"pi.ai.request">; type End = AiSpanEndAttributes<"pi.ai.request">; diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index f9ae0a607..73675edba 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -105,7 +105,7 @@ - Added inherited opt-in vLLM `thinking_token_budget` support for OpenAI-compatible models, reserving output tokens for the final answer ([#7638](https://github.com/earendil-works/pi/pull/7638) by [@bnsd55](https://github.com/bnsd55)). - Added inherited support for OpenAI-compatible streams that omit `finish_reason`, using `compat.supportsFinishReason` to infer normal and tool-use stops when the stream ends. See [OpenAI Compatibility](docs/models.md#openai-compatibility). - Added inherited deferred provider request contracts, durable response handles, authenticated fetch/cancel dispatch, and faux-provider support for pending, ready, failed, and cancelled responses ([#7339](https://github.com/earendil-works/pi/pull/7339) by [@davidbrai](https://github.com/davidbrai)). -- Added inherited typed telemetry contracts, callback helpers, and schema references for pi-ai and the agent harness. See the [pi-ai telemetry schema](../ai/docs/telemetry-schema.md) and [agent harness telemetry schema](../agent/docs/telemetry-schema.md). +- Added inherited vendor-neutral telemetry contracts plus agent-owned typed AI-request and harness schemas, composed span starters, and callback helpers. See the [agent telemetry schema reference](../agent/docs/telemetry-schema.md). - Added inherited structured Amazon Bedrock failure diagnostics with HTTP status, modeled error code, and AWS request id when available ([#7286](https://github.com/earendil-works/pi/pull/7286) by [@brianstanley](https://github.com/brianstanley)). - Added inherited `AgentOptions.shouldStopAfterTurn` for gracefully stopping after a completed turn before queued messages or another model call are processed. See [Agent Options](../agent/README.md#agent-options) ([#7367](https://github.com/earendil-works/pi/pull/7367) by [@acmerfight](https://github.com/acmerfight)). - Added inherited v4 `JsonlSessionRepo` support for append-only JSONL harness sessions ([#7611](https://github.com/earendil-works/pi/pull/7611) by [@davidbrai](https://github.com/davidbrai)). diff --git a/packages/telemetry/CHANGELOG.md b/packages/telemetry/CHANGELOG.md index dc2f991ce..cb41256a1 100644 --- a/packages/telemetry/CHANGELOG.md +++ b/packages/telemetry/CHANGELOG.md @@ -4,4 +4,4 @@ ### Added -- Added the callback-based telemetry context contract, shared no-op context, and typed serializable schema utilities. +- Added the callback-based telemetry context contract, shared no-op context, deterministic in-memory reference adapter, reusable adapter conformance suite, typed serializable schema utilities, and multi-schema typed span starters with explicit child propagation. diff --git a/packages/telemetry/README.md b/packages/telemetry/README.md index 02805a7a5..0c00dfd71 100644 --- a/packages/telemetry/README.md +++ b/packages/telemetry/README.md @@ -2,4 +2,463 @@ Vendor-neutral telemetry contracts and typed schema utilities for pi packages. -This package contains no exporter and depends on no telemetry backend. Applications provide a `TelemetryContext` adapter; pi packages pass contexts explicitly and define their own domain schemas. +This package provides: + +- an explicit, callback-based `TelemetryContext` / `TelemetrySpan` contract; +- a shared `NOOP_TELEMETRY_CONTEXT`; +- a reference `InMemoryTelemetryContext` implementation; +- serializable schema definitions with inferred TypeScript types; +- no exporter, global current-span state, or dependency on a telemetry backend. + +Applications can use the in-memory reference or provide an adapter for OpenTelemetry, Sentry, logs, or another backend. Pi packages pass telemetry contexts explicitly and define their domain schemas separately. + +## Table of Contents + +- [Installation](#installation) +- [Telemetry Concepts](#telemetry-concepts) +- [Core Context API](#core-context-api) +- [Adapter Contract](#adapter-contract) +- [No-op Context](#no-op-context) +- [In-Memory Reference Adapter](#in-memory-reference-adapter) +- [Adapter Conformance](#adapter-conformance) +- [Typed Schemas](#typed-schemas) + - [Start and Completion Attributes](#start-and-completion-attributes) +- [Schema Metadata](#schema-metadata) +- [Pi Package Integration](#pi-package-integration) +- [Security and Portability](#security-and-portability) +- [API Reference](#api-reference) +- [Development](#development) +- [License](#license) + +## Installation + +```bash +npm install @earendil-works/pi-telemetry +``` + +## Telemetry Concepts + +Telemetry describes what a program did while it was running. This package models that work using spans, attributes, events, statuses, and explicit context: + +| Concept | Plain-language meaning | +|---|---| +| **Span** | A timed record of one operation, such as loading an account or making an AI request. It begins before the work and ends when the work finishes. | +| **Parent and child spans** | Operations can contain smaller operations. A request span might contain a cache lookup and a database query. Together they form a tree showing where time was spent. | +| **Attribute** | A named fact attached to a span, such as `provider: "openai"`, `cache.hit: true`, or `item_count: 12`. Attributes describe the operation and its result. | +| **Event** | A named occurrence at a point during a span, such as `retry.scheduled` or `cache.lookup`. Events have no duration and may carry their own attributes. | +| **Status** | The operation's outcome: `ok` or `error`. An error status may include an error name and message. | +| **Context** | A handle identifying where new work belongs in the span tree. Starting a span from a context makes it a child of that context. | + +For example, loading an account could produce this telemetry: + +```text +example.account.load span +├─ attributes: account.id=123, found=true facts about the span +├─ event: example.cache.lookup occurrence during the span +│ └─ attribute: cache.hit=false fact about the event +└─ status: ok final outcome +``` + +A span is diagnostic data, not business state. Recording it must not change whether the account load runs, succeeds, fails, or is persisted. An adapter translates these generic concepts into the corresponding concepts used by OpenTelemetry, Sentry, logs, or another backend. + +## Core Context API + +A `TelemetryContext` starts a span around a callback. The callback receives a `TelemetrySpan`, which is also the explicit parent context for child spans. + +```typescript +import { + NOOP_TELEMETRY_CONTEXT, + type TelemetryContext, +} from '@earendil-works/pi-telemetry'; + +async function loadAccount( + accountId: string, + telemetryContext: TelemetryContext = NOOP_TELEMETRY_CONTEXT, +) { + return telemetryContext.startSpan( + { + name: 'example.account.load', + attributes: { 'example.account.id': accountId }, + }, + async (span) => { + const account = await readAccount(accountId); + span.setAttributes({ 'example.account.found': account !== undefined }); + return account; + }, + ); +} +``` + +Pass the callback span to lower-level work to create explicit nesting: + +```typescript +return telemetryContext.startSpan({ name: 'example.parent' }, async (parentSpan) => { + return parentSpan.startSpan({ name: 'example.child' }, async (childSpan) => { + childSpan.addEvent('example.cache.lookup', { 'example.cache.hit': true }); + return performWork(); + }); +}); +``` + +There is no public `end()` method. `startSpan()` owns settlement and keeps the span open until the callback's value or promise settles. For an expected failure represented by a normal return value, set the status explicitly: + +```typescript +return telemetryContext.startSpan({ name: 'example.save' }, async (span) => { + const result = await save(); + if (!result.ok) { + span.setStatus({ + status: 'error', + error: { name: 'SaveError', message: result.reason }, + }); + } + return result; +}); +``` + +## Adapter Contract + +An adapter implements `TelemetryContext` and bridges the generic API to its backend. It must: + +- create a child span and invoke the callback synchronously, exactly once; +- preserve the callback's returned value and rejection value, returning a promise rejected with the same value after a synchronous throw; +- keep the native span open until a returned promise settles; +- treat normal completion as `ok` and throws/rejections as errors unless an explicit status was set; +- make repeated `setStatus()` calls last-write-wins; +- merge `setAttributes()` calls, with later defined values replacing earlier values and `undefined` ignored; +- make recording methods synchronous, passive, and non-throwing; +- ignore calls made after settlement; +- ignore a failed recording call atomically, suppress backend failures, and still execute the business callback exactly once. + +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. + +## No-op Context + +Use `NOOP_TELEMETRY_CONTEXT` when telemetry is optional: + +```typescript +import { NOOP_TELEMETRY_CONTEXT } from '@earendil-works/pi-telemetry'; + +const result = await NOOP_TELEMETRY_CONTEXT.startSpan( + { name: 'example.operation' }, + () => runOperation(), +); +``` + +The no-op context: + +- invokes callbacks synchronously; +- preserves returned values and asynchronous rejections, and converts a synchronous throw to a promise rejected with the same value; +- uses one shared frozen inert span, including for nested spans; +- does not inspect or retain names, attributes, events, or statuses. + +## In-Memory Reference Adapter + +`InMemoryTelemetryContext` is the backend-neutral reference implementation. It is useful for tests, local diagnostics, and applications that intentionally want process-local capture without an exporter: + +```typescript +import { InMemoryTelemetryContext } from '@earendil-works/pi-telemetry'; + +const telemetry = new InMemoryTelemetryContext(); + +await telemetry.startSpan( + { name: 'example.operation', attributes: { input: 'demo' } }, + async (span) => { + span.addEvent('example.started'); + span.setAttributes({ output_count: 3 }); + }, +); + +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. + +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. + +## Adapter Conformance + +`@earendil-works/pi-telemetry/testing` exports a runner-independent conformance suite modeled as grouped cases. A fixture supplies a fresh context and converts its backend's finished spans into normalized `RecordedTelemetrySpan` snapshots: + +```typescript +import { + createTelemetryAdapterConformance, + type TelemetryAdapterFixture, +} from '@earendil-works/pi-telemetry/testing'; +import { describe, it } from 'vitest'; + +const conformance = createTelemetryAdapterConformance(async () => { + const adapter = createMyTelemetryAdapter(); + return { + context: adapter.context, + getSpans: async () => adapter.normalizedSpans(), + async [Symbol.asyncDispose]() { + await adapter.close(); + }, + } satisfies TelemetryAdapterFixture; +}); + +for (const group of new Set(conformance.map((testCase) => testCase.group))) { + describe(group, () => { + for (const testCase of conformance.filter((candidate) => candidate.group === group)) { + it(testCase.name, () => testCase.run()); + } + }); +} +``` + +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. + +## Typed Schemas + +The low-level span API intentionally accepts open names and attribute bags so adapters remain generic. Domain packages can define closed, serializable schemas and infer exact TypeScript types from them. + +```typescript +import { + createTypedSpanStarter, + defineTelemetrySchema, +} from '@earendil-works/pi-telemetry'; + +export const EXAMPLE_TELEMETRY_SCHEMA = defineTelemetrySchema({ + version: 1, + spans: { + 'example.read': { + description: 'Read one resource', + parents: { kind: 'any' }, + startAttributes: { + 'example.resource': { + type: 'string', + required: true, + values: ['account', 'project'], + description: 'Resource kind', + }, + }, + endAttributes: { + 'example.item_count': { + type: 'number', + description: 'Number of returned items', + }, + }, + events: { + 'example.cache': { + description: 'Cache lookup result', + attributes: { + 'example.cache.hit': { + type: 'boolean', + required: true, + description: 'Whether the cache contained the resource', + }, + }, + }, + }, + status: { + default: 'ok', + errorWhen: 'The read throws or returns an error result', + }, + }, + }, +} as const); + +const startSpan = createTypedSpanStarter( + telemetryContext, + [EXAMPLE_TELEMETRY_SCHEMA], +); +``` + +The starter exposes one overload per span and checks names and attributes at compile time. Union-valued names must be narrowed before a call, preserving the relationship between each runtime name and its attribute schema. Its callback receives a child starter over the same schemas, already bound to the callback span: + +```typescript +await startSpan( + 'example.read', + { 'example.resource': 'account' }, + async (span, startChildSpan) => { + span.addEvent('example.cache', { 'example.cache.hit': true }); + const accounts = await readAccounts(); + span.setAttributes({ 'example.item_count': accounts.length }); + + await startChildSpan( + 'example.read', + { 'example.resource': 'project' }, + async (childSpan) => { + const projects = await readProjects(); + childSpan.setAttributes({ 'example.item_count': projects.length }); + }, + ); + + return accounts; + }, +); +``` + +### Start and Completion Attributes + +`startAttributes` and `endAttributes` describe when an attribute is normally known, not separate runtime storage: + +| Schema field | How values are recorded | Requiredness | +|---|---|---| +| `startAttributes` | Passed in the typed starter's `attributes` argument when the span is created | Each definition explicitly sets `required: true` or `false` | +| `endAttributes` | Added later through the schema-scoped span's `setAttributes()` method | Always optional | + +Both sets become ordinary attributes on the same backend span. There is no separate end-attribute payload or end callback. In the preceding example, `example.resource` is known when `example.read` starts, while `example.item_count` is known only after `readAccounts()` returns: + +```typescript +await startSpan( + 'example.read', + { 'example.resource': 'account' }, // required start attribute + async (span) => { + const accounts = await readAccounts(); + span.setAttributes({ + 'example.item_count': accounts.length, // optional completion attribute + }); + return accounts; + }, +); // resolving the callback settles the span +``` + +“End” means completion enrichment: an end attribute may be set at any point while the callback is active, and it may be omitted when unavailable. Calling `setAttributes()` zero times is valid. This matters for early failures, cancellation, and provider-specific data that may not exist on every path. + +Repeated `setAttributes()` calls merge into the same attribute bag. A later defined value replaces an earlier value for the same key, while `undefined` is ignored. The schema-scoped method accepts only the current span's declared end attributes. + +Attributes do not end the span. Returning, resolving, throwing, or rejecting from the callback controls settlement; `startSpan()` performs the actual end operation. Adapter calls made after settlement are inert. + +A starter can compose multiple independently versioned schemas: + +```typescript +import { AGENT_TELEMETRY_SCHEMAS } from '@earendil-works/pi-agent-core'; + +const startAgentSpan = createTypedSpanStarter( + telemetryContext, + AGENT_TELEMETRY_SCHEMAS, +); +``` + +Inline schema arrays retain their tuple types automatically. Separately declared arrays should use `as const`. Literal duplicate span names across the array are rejected at compile time; schemas are not merged, inspected, or retained at runtime. + +Schema-derived types reject missing required attributes, unknown keys, invalid closed-set values, undeclared events, and attributes on empty schemas. End attributes are always optional enrichment; the type system does not require `setAttributes()` to be called. + +`defineTelemetrySchema()` is a typed identity function. It returns ordinary JSON-serializable data and performs no runtime validation or parent-rule enforcement. + +## Schema Metadata + +Supported attribute types are: + +- `string`, `number`, and `boolean`; +- `string[]`, `number[]`, and `boolean[]`. + +Attribute definitions support: + +- `values`: a closed set for scalar values; +- `elementValues`: a closed set for array elements; +- `examples`: documentation examples; +- `sensitive`: marks data requiring special handling; +- `cardinality`: records expected `low` or `high` cardinality. + +Start and event attributes declare `required`. End attributes do not; see [Start and Completion Attributes](#start-and-completion-attributes). + +Parent metadata is descriptive schema data: + +- `{ kind: 'any' }`: root or any caller span; +- `{ kind: 'root_or_external' }`: root or a caller-owned span outside the schema; +- `{ kind: 'spans', spans: [...] }`: only the listed schema spans. + +Adapters do not need to understand schema objects. Instrumentation helpers and tests use them to keep emitted names and attributes consistent. + +## Pi Package Integration + +Package ownership is intentionally split: + +- `@earendil-works/pi-telemetry` owns the vendor-neutral contract, no-op and in-memory reference contexts, schema utilities, and adapter conformance suite; +- `@earendil-works/pi-ai` accepts and propagates `telemetryContext` in provider request options but owns no telemetry schema; +- `@earendil-works/pi-agent-core` owns and exports the pi AI-request and harness schemas, their combined readonly schema tuple, and typed span helpers. + +```typescript +import { + AGENT_TELEMETRY_SCHEMAS, + AI_TELEMETRY_SCHEMA, + HARNESS_TELEMETRY_SCHEMA, + startAiSpan, + startHarnessSpan, +} from '@earendil-works/pi-agent-core'; +``` + +The pi schemas use pi-owned `pi.ai.*`, `pi.harness.*`, and `pi.session.*` names. Adapters may translate them to backend conventions without changing the emitted pi vocabulary. + +## 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. + +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. + +The package does not use `AsyncLocalStorage` or another runtime-specific ambient context API. It is suitable for Node.js, Bun, browsers, and workers; backend adapters remain responsible for their own runtime compatibility. + +## API Reference + +### Core types and values + +| Export | Purpose | +|---|---| +| `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 | +| `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 | + +### Schema definitions and inference + +| Export | Purpose | +|---|---| +| `defineTelemetrySchema()` | Typed identity helper for serializable schema data | +| `createTypedSpanStarter()` | Binds a parent context to one or more schema vocabularies | +| `TypedSpanStarter` | Exact starter type with recursively child-bound callbacks | +| `TelemetrySchemaDefinition` | Top-level schema shape | +| `TelemetrySpanDefinition` | Span metadata, parents, attributes, events, and status rule | +| `TelemetryAttributeType` | Supported scalar and array type names | +| `TelemetryAttributeMetadata` | Description, sensitivity, and cardinality metadata | +| `TelemetryAttributeDefinition` | Attribute type, allowed values, examples, and metadata | +| `TelemetryStartAttributeDefinition` | Start attribute definition with requiredness | +| `TelemetryEventAttributeDefinition` | Event attribute definition with requiredness | +| `TelemetryEventDefinition` | Event description and attribute definitions | +| `TelemetryParentDefinition` | Open, external-root, or finite schema-parent rule | +| `TelemetrySchemaSpanName` | Union of declared span names | +| `TelemetrySchemaSpanStartAttributes` | Exact inferred start attributes for one span | +| `TelemetrySchemaSpanEndAttributes` | Optional inferred end attributes for one span | +| `TelemetrySchemaSpanEventName` | Union of events declared by one span | +| `TelemetrySchemaSpanEventAttributes` | Exact inferred attributes for one event | +| `SchemaTelemetrySpan` | Span view restricted to one schema span | +| `TelemetrySchemaSpanUnion` | Discriminated union of all spans in a schema | +| `InferStartAttributes` | Required and optional values inferred from start definitions | +| `InferOptionalAttributes` | Optional values inferred from end definitions | +| `InferEventAttributes` | Required and optional values inferred from event definitions | +| `InferRequiredAndOptionalAttributes` | Shared inference utility for definitions with requiredness | +| `ExactTelemetryAttributes` | Rejects keys outside an expected attribute set | + +### Testing subpath + +| Export | Purpose | +|---|---| +| `createTelemetryAdapterConformance()` | Creates runner-independent adapter conformance cases | +| `TelemetryAdapterFixture` | Fresh context and normalized snapshot reader for one case | +| `TelemetryAdapterFixtureFactory` | Creates isolated fixtures | +| `TelemetryAdapterConformanceCase` | Grouped case that test runners execute | + +## Development + +From this package directory: + +```bash +npm test +npm run build +``` + +Repository-wide type checking, formatting, linting, and smoke checks run with: + +```bash +npm run check +``` + +## License + +MIT diff --git a/packages/telemetry/package.json b/packages/telemetry/package.json index 104acd027..332cee58f 100644 --- a/packages/telemetry/package.json +++ b/packages/telemetry/package.json @@ -9,6 +9,10 @@ ".": { "types": "./dist/index.d.ts", "import": "./dist/index.js" + }, + "./testing": { + "types": "./dist/testing/index.d.ts", + "import": "./dist/testing/index.js" } }, "files": [ diff --git a/packages/telemetry/src/index.ts b/packages/telemetry/src/index.ts index 094812526..3d3909791 100644 --- a/packages/telemetry/src/index.ts +++ b/packages/telemetry/src/index.ts @@ -21,24 +21,7 @@ export interface TelemetrySpan extends TelemetryContext { setStatus(status: SpanStatus): void; } -function startNoopSpan(_options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { - try { - return Promise.resolve(callback(noopTelemetrySpan)); - } catch (error) { - return Promise.reject(error); - } -} - -const noopTelemetrySpan: TelemetrySpan = { - startSpan: startNoopSpan, - addEvent: () => {}, - setAttributes: () => {}, - setStatus: () => {}, -}; -Object.freeze(noopTelemetrySpan); - -/** Shared telemetry context used when an application does not provide one. */ -export const NOOP_TELEMETRY_CONTEXT: TelemetryContext = noopTelemetrySpan; +export { NOOP_TELEMETRY_CONTEXT } from "./noop.ts"; export type TelemetryAttributeType = "string" | "number" | "boolean" | "string[]" | "number[]" | "boolean[]"; @@ -268,3 +251,107 @@ export type TelemetrySchemaSpanUnion = }; }; }[TelemetrySchemaSpanName]; + +type TelemetrySchemaTuple = readonly [TelemetrySchemaDefinition, ...TelemetrySchemaDefinition[]]; + +type SpanNameInSchema = Schema extends TelemetrySchemaDefinition + ? TelemetrySchemaSpanName + : never; + +type SpanNameInSchemas = SpanNameInSchema; + +type SpanStartAttributesInSchema< + Schema extends TelemetrySchemaDefinition, + Name extends string, +> = Schema extends TelemetrySchemaDefinition + ? Name extends TelemetrySchemaSpanName + ? TelemetrySchemaSpanStartAttributes + : never + : never; + +type SpanInSchema< + Schema extends TelemetrySchemaDefinition, + Name extends string, +> = Schema extends TelemetrySchemaDefinition + ? Name extends TelemetrySchemaSpanName + ? SchemaTelemetrySpan + : never + : never; + +type DuplicateTelemetrySpanNames< + Schemas extends readonly TelemetrySchemaDefinition[], + Seen extends string = never, +> = Schemas extends readonly [ + infer Schema extends TelemetrySchemaDefinition, + ...infer Rest extends readonly TelemetrySchemaDefinition[], +] + ? + | Extract, Seen> + | DuplicateTelemetrySpanNames> + : never; + +type UniqueTelemetrySchemas = [DuplicateTelemetrySpanNames] extends [ + never, +] + ? unknown + : { readonly "duplicate telemetry span names": DuplicateTelemetrySpanNames }; + +type UnionToIntersection = (Union extends unknown ? (value: Union) => void : never) extends ( + value: infer Intersection, +) => void + ? Intersection + : never; + +type TypedSpanStarterForName> = < + const Attributes extends SpanStartAttributesInSchema, + Result, +>( + name: Name, + attributes: ExactTelemetryAttributes, Attributes>, + callback: ( + span: SpanInSchema, + startChildSpan: TypedSpanStarter, + ) => Result | Promise, +) => Promise; + +/** A per-span overload set bound to one explicit parent context and one or more schemas. */ +export type TypedSpanStarter = UnionToIntersection< + { + [Name in SpanNameInSchemas]: TypedSpanStarterForName; + }[SpanNameInSchemas] +>; + +function bindTypedSpanStarter( + telemetryContext: TelemetryContext, +): TypedSpanStarter { + const startSpan = ( + name: SpanNameInSchemas, + attributes: SpanAttributes, + callback: ( + span: SpanInSchema>, + startChildSpan: TypedSpanStarter, + ) => unknown, + ): Promise => + telemetryContext.startSpan({ name, attributes }, (span) => + callback( + span as SpanInSchema>, + bindTypedSpanStarter(span), + ), + ); + + return startSpan as TypedSpanStarter; +} + +/** + * Bind an explicit parent context to the combined span vocabulary of one or more schemas. + * Schema values are used only for type inference; no runtime schema validation is performed. + */ +export function createTypedSpanStarter( + telemetryContext: TelemetryContext, + _schemas: Schemas & UniqueTelemetrySchemas, +): TypedSpanStarter { + return bindTypedSpanStarter(telemetryContext); +} + +export type { RecordedTelemetryEvent, RecordedTelemetrySpan } from "./memory.ts"; +export { InMemoryTelemetryContext } from "./memory.ts"; diff --git a/packages/telemetry/src/memory.ts b/packages/telemetry/src/memory.ts new file mode 100644 index 000000000..fd5985a6d --- /dev/null +++ b/packages/telemetry/src/memory.ts @@ -0,0 +1,219 @@ +import type { + AttributeValue, + SpanAttributes, + SpanOptions, + SpanStatus, + TelemetryContext, + TelemetrySpan, +} from "./index.ts"; +import { NOOP_TELEMETRY_CONTEXT } from "./noop.ts"; + +export interface RecordedTelemetryEvent { + readonly name: string; + readonly attributes: Readonly; +} + +export interface RecordedTelemetrySpan { + readonly id: number; + readonly parentId: number | null; + readonly name: string; + readonly attributes: Readonly; + readonly events: readonly RecordedTelemetryEvent[]; + readonly status: SpanStatus; + readonly settled: boolean; + readonly endSequence?: number; +} + +interface MutableRecordedTelemetryEvent { + name: string; + attributes: SpanAttributes; +} + +interface MutableRecordedTelemetrySpan { + id: number; + parentId: number | null; + name: string; + attributes: SpanAttributes; + events: MutableRecordedTelemetryEvent[]; + status: SpanStatus; + explicitStatus: boolean; + settled: boolean; + endSequence?: number; +} + +interface InMemoryTelemetryState { + spans: MutableRecordedTelemetrySpan[]; + nextSpanId: number; + nextEndSequence: number; +} + +function copyAttributeValue(value: AttributeValue): AttributeValue { + return Array.isArray(value) ? ([...value] as AttributeValue) : value; +} + +function copyAttributes(attributes?: SpanAttributes): SpanAttributes { + const copy: SpanAttributes = {}; + if (!attributes) return copy; + for (const [name, value] of Object.entries(attributes)) { + if (value !== undefined) copy[name] = copyAttributeValue(value); + } + return copy; +} + +function mergeAttributes(current: SpanAttributes, attributes: SpanAttributes): SpanAttributes { + const merged = copyAttributes(current); + for (const [name, value] of Object.entries(attributes)) { + if (value !== undefined) merged[name] = copyAttributeValue(value); + } + return merged; +} + +function copyStatus(status: SpanStatus): SpanStatus { + if (status.status === "ok") return { status: "ok" }; + return status.error + ? { status: "error", error: { name: status.error.name, message: status.error.message } } + : { status: "error" }; +} + +function automaticErrorStatus(error: unknown): SpanStatus { + try { + if (error instanceof Error) { + return { status: "error", error: { name: error.name, message: error.message } }; + } + } catch { + // Error inspection is passive. Fall through to an error status without details. + } + return { status: "error" }; +} + +function settleSpan( + state: InMemoryTelemetryState, + span: MutableRecordedTelemetrySpan, + failed: boolean, + error?: unknown, +): void { + if (span.settled) return; + if (failed && !span.explicitStatus) span.status = automaticErrorStatus(error); + span.settled = true; + span.endSequence = state.nextEndSequence++; +} + +function createSpan( + state: InMemoryTelemetryState, + parent: MutableRecordedTelemetrySpan | undefined, + options: SpanOptions, +): MutableRecordedTelemetrySpan { + const name = options.name; + const attributes = copyAttributes(options.attributes); + return { + id: state.nextSpanId++, + parentId: parent?.id ?? null, + name, + attributes, + events: [], + status: { status: "ok" }, + explicitStatus: false, + settled: false, + }; +} + +function startInMemorySpan( + state: InMemoryTelemetryState, + parent: MutableRecordedTelemetrySpan | undefined, + options: SpanOptions, + callback: (span: TelemetrySpan) => T | Promise, +): Promise { + if (parent?.settled) return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); + + let recordedSpan: MutableRecordedTelemetrySpan; + try { + recordedSpan = createSpan(state, parent, options); + state.spans.push(recordedSpan); + } catch { + return NOOP_TELEMETRY_CONTEXT.startSpan(options, callback); + } + + const span: TelemetrySpan = { + startSpan: ( + childOptions: SpanOptions, + childCallback: (child: TelemetrySpan) => Result | Promise, + ) => startInMemorySpan(state, recordedSpan, childOptions, childCallback), + addEvent(name, attributes) { + if (recordedSpan.settled) return; + try { + recordedSpan.events.push({ name, attributes: copyAttributes(attributes) }); + } catch { + // Recording is passive. Ignore malformed or unreadable telemetry payloads. + } + }, + setAttributes(attributes) { + if (recordedSpan.settled) return; + try { + recordedSpan.attributes = mergeAttributes(recordedSpan.attributes, attributes); + } catch { + // Recording is passive. Ignore malformed or unreadable telemetry payloads. + } + }, + setStatus(status) { + if (recordedSpan.settled) return; + try { + recordedSpan.status = copyStatus(status); + recordedSpan.explicitStatus = true; + } catch { + // Recording is passive. Ignore malformed or unreadable telemetry payloads. + } + }, + }; + + let result: T | Promise; + try { + result = callback(span); + } catch (error) { + settleSpan(state, recordedSpan, true, error); + return Promise.reject(error); + } + + return Promise.resolve(result).then( + (value) => { + settleSpan(state, recordedSpan, false); + return value; + }, + (error: unknown) => { + settleSpan(state, recordedSpan, true, error); + throw error; + }, + ); +} + +/** + * 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 { + private readonly state: InMemoryTelemetryState = { + spans: [], + nextSpanId: 1, + nextEndSequence: 1, + }; + + startSpan(options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { + return startInMemorySpan(this.state, undefined, options, callback); + } + + /** Returns detached snapshots in span-start order. */ + getSpans(): readonly RecordedTelemetrySpan[] { + return this.state.spans.map((span) => ({ + id: span.id, + parentId: span.parentId, + name: span.name, + attributes: copyAttributes(span.attributes), + events: span.events.map((event) => ({ + name: event.name, + attributes: copyAttributes(event.attributes), + })), + status: copyStatus(span.status), + settled: span.settled, + ...(span.endSequence === undefined ? {} : { endSequence: span.endSequence }), + })); + } +} diff --git a/packages/telemetry/src/noop.ts b/packages/telemetry/src/noop.ts new file mode 100644 index 000000000..48244d3a3 --- /dev/null +++ b/packages/telemetry/src/noop.ts @@ -0,0 +1,20 @@ +import type { SpanOptions, TelemetryContext, TelemetrySpan } from "./index.ts"; + +function startNoopSpan(_options: SpanOptions, callback: (span: TelemetrySpan) => T | Promise): Promise { + try { + return Promise.resolve(callback(noopTelemetrySpan)); + } catch (error) { + return Promise.reject(error); + } +} + +const noopTelemetrySpan: TelemetrySpan = { + startSpan: startNoopSpan, + addEvent: () => {}, + setAttributes: () => {}, + setStatus: () => {}, +}; +Object.freeze(noopTelemetrySpan); + +/** Shared telemetry context used when an application does not provide one. */ +export const NOOP_TELEMETRY_CONTEXT: TelemetryContext = noopTelemetrySpan; diff --git a/packages/telemetry/src/testing/conformance.ts b/packages/telemetry/src/testing/conformance.ts new file mode 100644 index 000000000..f90ba588b --- /dev/null +++ b/packages/telemetry/src/testing/conformance.ts @@ -0,0 +1,315 @@ +import { deepStrictEqual, doesNotThrow, fail, ok, strictEqual } from "node:assert/strict"; +import type { SpanAttributes, SpanOptions, SpanStatus, TelemetrySpan } from "../index.ts"; +import type { RecordedTelemetrySpan } from "../memory.ts"; +import type { + TelemetryAdapterConformanceCase, + TelemetryAdapterFixture, + TelemetryAdapterFixtureFactory, +} from "./types.ts"; + +type ConformanceTest = (fixture: TelemetryAdapterFixture) => Promise; + +function createCase( + factory: TelemetryAdapterFixtureFactory, + group: string, + name: string, + test: ConformanceTest, +): TelemetryAdapterConformanceCase { + return { + group, + name, + async run() { + await using fixture = await factory(); + await test(fixture); + }, + }; +} + +function findSpan(spans: readonly RecordedTelemetrySpan[], name: string): RecordedTelemetrySpan { + const span = spans.find((candidate) => candidate.name === name); + ok(span, `Expected recorded span ${name}`); + return span; +} + +async function rejectsWithSameValue(operation: Promise, expected: unknown): Promise { + try { + await operation; + fail("Expected operation to reject"); + } catch (error) { + strictEqual(error, expected); + } +} + +function unreadable(value: T): T { + return new Proxy(value, { + get: () => { + throw new Error("read"); + }, + getOwnPropertyDescriptor: () => { + throw new Error("inspect"); + }, + getPrototypeOf: () => { + throw new Error("prototype"); + }, + ownKeys: () => { + throw new Error("enumerate"); + }, + }); +} + +/** Creates runner-independent cases for the callback telemetry adapter contract. */ +export function createTelemetryAdapterConformance( + factory: TelemetryAdapterFixtureFactory, +): readonly TelemetryAdapterConformanceCase[] { + return [ + createCase( + factory, + "callback lifecycle", + "admits once synchronously and preserves the result", + async (fixture) => { + let admitted = false; + let calls = 0; + const expected = { value: 42 }; + const result = fixture.context.startSpan({ name: "success" }, () => { + admitted = true; + calls++; + return expected; + }); + + strictEqual(admitted, true); + strictEqual(calls, 1); + strictEqual(await result, expected); + deepStrictEqual(findSpan(await fixture.getSpans(), "success").status, { status: "ok" }); + strictEqual(findSpan(await fixture.getSpans(), "success").settled, true); + }, + ), + + createCase( + factory, + "callback lifecycle", + "preserves synchronous and asynchronous rejection values", + async (fixture) => { + const syncError = new Error("sync"); + await rejectsWithSameValue( + fixture.context.startSpan({ name: "sync-error" }, () => { + throw syncError; + }), + syncError, + ); + + const asyncError = { kind: "async" }; + await rejectsWithSameValue( + fixture.context.startSpan({ name: "async-error" }, async () => { + throw asyncError; + }), + asyncError, + ); + + await rejectsWithSameValue( + fixture.context.startSpan({ name: "undefined-error" }, () => Promise.reject(undefined)), + undefined, + ); + + const unreadableError = unreadable({ kind: "unreadable" }); + await rejectsWithSameValue( + fixture.context.startSpan({ name: "unreadable-error" }, () => { + throw unreadableError; + }), + unreadableError, + ); + + const asyncUnreadableError = unreadable({ kind: "async-unreadable" }); + await rejectsWithSameValue( + fixture.context.startSpan({ name: "async-unreadable-error" }, () => + Promise.reject(asyncUnreadableError), + ), + asyncUnreadableError, + ); + + const spans = await fixture.getSpans(); + for (const name of [ + "sync-error", + "async-error", + "undefined-error", + "unreadable-error", + "async-unreadable-error", + ]) { + strictEqual(findSpan(spans, name).status.status, "error"); + } + }, + ), + + createCase(factory, "status", "uses last explicit status without automatic overwrite", async (fixture) => { + await fixture.context.startSpan({ name: "last-status" }, (span) => { + span.setStatus({ status: "error", error: { name: "Expected", message: "first" } }); + span.setStatus({ status: "ok" }); + }); + + const thrown = new Error("after explicit status"); + await rejectsWithSameValue( + fixture.context.startSpan({ name: "explicit-before-throw" }, (span) => { + span.setStatus({ status: "ok" }); + throw thrown; + }), + thrown, + ); + + const rejected = new Error("after async explicit status"); + await rejectsWithSameValue( + fixture.context.startSpan({ name: "explicit-before-rejection" }, (span) => { + span.setStatus({ status: "error", error: { name: "Expected", message: "async failure" } }); + return Promise.reject(rejected); + }), + rejected, + ); + + await fixture.context.startSpan({ name: "expected-failure" }, (span) => { + span.setStatus({ status: "error", error: { name: "Expected", message: "returned failure" } }); + return { ok: false }; + }); + + const spans = await fixture.getSpans(); + deepStrictEqual(findSpan(spans, "last-status").status, { status: "ok" }); + deepStrictEqual(findSpan(spans, "explicit-before-throw").status, { status: "ok" }); + deepStrictEqual(findSpan(spans, "explicit-before-rejection").status, { + status: "error", + error: { name: "Expected", message: "async failure" }, + }); + deepStrictEqual(findSpan(spans, "expected-failure").status, { + status: "error", + error: { name: "Expected", message: "returned failure" }, + }); + }), + + createCase(factory, "recording", "merges attributes and records ordered events", async (fixture) => { + await fixture.context.startSpan( + { + name: "recording", + attributes: { start: "value", overwrite: "start", ignored: undefined }, + }, + (span) => { + span.setAttributes({ count: 1, overwrite: "middle" }); + span.setAttributes({ count: undefined, overwrite: "end" }); + span.addEvent("first", { index: 1, ignored: undefined }); + span.addEvent("second", { index: 2 }); + }, + ); + + const span = findSpan(await fixture.getSpans(), "recording"); + deepStrictEqual(span.attributes, { start: "value", overwrite: "end", count: 1 }); + deepStrictEqual(span.events, [ + { name: "first", attributes: { index: 1 } }, + { name: "second", attributes: { index: 2 } }, + ]); + }), + + createCase(factory, "recording", "ignores failed attribute calls atomically", async (fixture) => { + await fixture.context.startSpan({ name: "atomic-attributes", attributes: { retained: "value" } }, (span) => { + const attributes: SpanAttributes = { + partial: "must not survive", + unreadable: unreadable(["value"]), + }; + doesNotThrow(() => span.setAttributes(attributes)); + }); + + deepStrictEqual(findSpan(await fixture.getSpans(), "atomic-attributes").attributes, { + retained: "value", + }); + }), + + createCase(factory, "recording", "makes calls after settlement inert", async (fixture) => { + let settledSpan: TelemetrySpan | undefined; + await fixture.context.startSpan({ name: "settled", attributes: { value: "initial" } }, (span) => { + settledSpan = span; + }); + const capturedSpan = settledSpan; + if (!capturedSpan) throw new Error("Expected callback span"); + + doesNotThrow(() => capturedSpan.setAttributes({ value: "late" })); + doesNotThrow(() => capturedSpan.addEvent("late", { value: true })); + doesNotThrow(() => capturedSpan.setStatus({ status: "error" })); + let childAdmitted = false; + const childResult = capturedSpan.startSpan({ name: "late-child" }, () => { + childAdmitted = true; + return 7; + }); + strictEqual(childAdmitted, true); + strictEqual(await childResult, 7); + + const spans = await fixture.getSpans(); + strictEqual(spans.length, 1); + deepStrictEqual(spans[0]?.attributes, { value: "initial" }); + deepStrictEqual(spans[0]?.events, []); + deepStrictEqual(spans[0]?.status, { status: "ok" }); + }), + + createCase(factory, "parentage", "records nested and concurrent child relationships", async (fixture) => { + let releaseFirst: (() => void) | undefined; + const firstGate = new Promise((resolve) => { + releaseFirst = resolve; + }); + + await fixture.context.startSpan({ name: "parent" }, async (parent) => { + const first = parent.startSpan({ name: "first-child" }, async () => { + await firstGate; + }); + const second = parent.startSpan({ name: "second-child" }, () => "done"); + strictEqual(await second, "done"); + releaseFirst?.(); + await first; + }); + + const spans = await fixture.getSpans(); + const parent = findSpan(spans, "parent"); + const first = findSpan(spans, "first-child"); + const second = findSpan(spans, "second-child"); + strictEqual(parent.parentId, null); + strictEqual(first.parentId, parent.id); + strictEqual(second.parentId, parent.id); + ok(second.endSequence !== undefined && first.endSequence !== undefined && parent.endSequence !== undefined); + ok(second.endSequence < first.endSequence); + ok(first.endSequence < parent.endSequence); + }), + + createCase(factory, "passivity", "suppresses unreadable telemetry payload failures", async (fixture) => { + let calls = 0; + const options = unreadable({ name: "unreadable-options", attributes: { secret: "value" } }); + const result = fixture.context.startSpan(options, () => { + calls++; + return 9; + }); + + strictEqual(calls, 1); + strictEqual(await result, 9); + deepStrictEqual(await fixture.getSpans(), []); + + await fixture.context.startSpan({ name: "unreadable-recording" }, (span) => { + const attributes = unreadable({ secret: "value" }); + const status = unreadable({ status: "ok" }); + doesNotThrow(() => span.setAttributes(attributes)); + doesNotThrow(() => span.addEvent("unreadable-event", attributes)); + doesNotThrow(() => span.setStatus(status)); + }); + + const recorded = await fixture.getSpans(); + strictEqual(recorded.length, 1); + deepStrictEqual(recorded[0]?.attributes, {}); + deepStrictEqual(recorded[0]?.events, []); + deepStrictEqual(recorded[0]?.status, { status: "ok" }); + }), + + createCase(factory, "passivity", "ignores failed status calls atomically", async (fixture) => { + const rejection = new Error("rejected after unreadable status"); + await rejectsWithSameValue( + fixture.context.startSpan({ name: "unreadable-status" }, (span) => { + const status = unreadable({ status: "ok" }); + doesNotThrow(() => span.setStatus(status)); + return Promise.reject(rejection); + }), + rejection, + ); + + strictEqual(findSpan(await fixture.getSpans(), "unreadable-status").status.status, "error"); + }), + ]; +} diff --git a/packages/telemetry/src/testing/index.ts b/packages/telemetry/src/testing/index.ts new file mode 100644 index 000000000..9ecad668c --- /dev/null +++ b/packages/telemetry/src/testing/index.ts @@ -0,0 +1,6 @@ +export { createTelemetryAdapterConformance } from "./conformance.ts"; +export type { + TelemetryAdapterConformanceCase, + TelemetryAdapterFixture, + TelemetryAdapterFixtureFactory, +} from "./types.ts"; diff --git a/packages/telemetry/src/testing/types.ts b/packages/telemetry/src/testing/types.ts new file mode 100644 index 000000000..1267083f3 --- /dev/null +++ b/packages/telemetry/src/testing/types.ts @@ -0,0 +1,18 @@ +import type { TelemetryContext } from "../index.ts"; +import type { 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; + getSpans(): Promise; +} + +/** Creates an isolated adapter fixture for one conformance case. */ +export type TelemetryAdapterFixtureFactory = () => Promise; + +/** A runner-independent conformance case that can be registered with any test framework. */ +export interface TelemetryAdapterConformanceCase { + readonly group: string; + readonly name: string; + run(): Promise; +} diff --git a/packages/telemetry/test/conformance.test.ts b/packages/telemetry/test/conformance.test.ts new file mode 100644 index 000000000..ba98cdec5 --- /dev/null +++ b/packages/telemetry/test/conformance.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; +import { InMemoryTelemetryContext, type SpanAttributes } from "../src/index.ts"; +import { createTelemetryAdapterConformance, type TelemetryAdapterFixture } from "../src/testing/index.ts"; + +const conformance = createTelemetryAdapterConformance(() => { + const context = new InMemoryTelemetryContext(); + return Promise.resolve({ + context, + getSpans: () => Promise.resolve(context.getSpans()), + [Symbol.asyncDispose]: () => Promise.resolve(), + }); +}); + +describe("InMemoryTelemetryContext conformance", () => { + for (const group of new Set(conformance.map((testCase) => testCase.group))) { + describe(group, () => { + for (const testCase of conformance.filter((candidate) => candidate.group === group)) { + it(testCase.name, () => testCase.run()); + } + }); + } + + it("returns detached snapshots without exposing mutable recording state", async () => { + const context = new InMemoryTelemetryContext(); + let openSettled: boolean | undefined; + let openEndSequence: number | undefined; + await context.startSpan({ name: "snapshot", attributes: { tags: ["initial"] } }, (span) => { + span.addEvent("event", { value: 1 }); + const [open] = context.getSpans(); + openSettled = open?.settled; + openEndSequence = open?.endSequence; + }); + + expect(openSettled).toBe(false); + expect(openEndSequence).toBeUndefined(); + const [first] = context.getSpans(); + expect(first?.settled).toBe(true); + expect(first?.endSequence).toBe(1); + (first?.attributes as SpanAttributes).tags = ["mutated"]; + (first?.events[0]?.attributes as SpanAttributes).value = 2; + + const [second] = context.getSpans(); + expect(second?.attributes).toEqual({ tags: ["initial"] }); + expect(second?.events).toEqual([{ name: "event", attributes: { value: 1 } }]); + }); +}); diff --git a/packages/telemetry/test/telemetry.test.ts b/packages/telemetry/test/telemetry.test.ts index c09784ee0..fbbbb1040 100644 --- a/packages/telemetry/test/telemetry.test.ts +++ b/packages/telemetry/test/telemetry.test.ts @@ -1,6 +1,8 @@ import { describe, expect, expectTypeOf, it } from "vitest"; import { + createTypedSpanStarter, defineTelemetrySchema, + InMemoryTelemetryContext, NOOP_TELEMETRY_CONTEXT, type SchemaTelemetrySpan, type SpanAttributes, @@ -68,6 +70,87 @@ describe("telemetry schemas", () => { }; expectTypeOf(compileTimeFailures).toBeFunction(); }); + + it("combines schema vocabularies and binds child starters to their parent spans", async () => { + const operationSchema = defineTelemetrySchema({ + version: 1, + spans: { + operation: { + description: "Operation", + parents: { kind: "root_or_external" }, + startAttributes: { + kind: { type: "string", required: true, values: ["read", "write"], description: "Kind" }, + }, + endAttributes: {}, + status: { default: "ok", errorWhen: "The operation fails" }, + }, + }, + } as const); + const requestSchema = defineTelemetrySchema({ + version: 3, + spans: { + request: { + description: "Request", + parents: { kind: "spans", spans: ["operation"] }, + startAttributes: { + provider: { type: "string", required: true, description: "Provider" }, + }, + endAttributes: { + response: { type: "string", description: "Response kind" }, + }, + status: { default: "ok", errorWhen: "The request fails" }, + }, + }, + } as const); + const telemetryContext = new InMemoryTelemetryContext(); + const startSpan = createTypedSpanStarter(telemetryContext, [operationSchema, requestSchema]); + + const result = await startSpan("operation", { kind: "read" }, (_operationSpan, startChildSpan) => + startChildSpan("request", { provider: "example" }, (requestSpan) => { + requestSpan.setAttributes({ response: "cached" }); + return 42; + }), + ); + + expect(result).toBe(42); + expectTypeOf(result).toEqualTypeOf(); + const spans = telemetryContext.getSpans(); + const operationSpan = spans.find((span) => span.name === "operation"); + const requestSpan = spans.find((span) => span.name === "request"); + expect(operationSpan?.parentId).toBeNull(); + expect(requestSpan?.parentId).toBe(operationSpan?.id); + expect(() => + createTypedSpanStarter(telemetryContext, unreadable([operationSchema, requestSchema] as const)), + ).not.toThrow(); + + const syncError = { kind: "sync" }; + const syncResult = startSpan("operation", { kind: "write" }, () => { + throw syncError; + }); + await expect(syncResult).rejects.toBe(syncError); + + const asyncError = { kind: "async" }; + const asyncResult = startSpan("request", { provider: "example" }, async () => { + throw asyncError; + }); + await expect(asyncResult).rejects.toBe(asyncError); + + const compileTimeFailures = () => { + const spanName: "operation" | "request" = Math.random() > 0.5 ? "operation" : "request"; + // @ts-expect-error union-valued names must be narrowed to preserve name and attribute correlation + void startSpan(spanName, { kind: "read" }, () => {}); + const extraRequestAttributes = { provider: "example", unknown: true } as const; + // @ts-expect-error variables with unknown attributes are rejected + void startSpan("request", extraRequestAttributes, () => {}); + // @ts-expect-error unknown span names are rejected across the combined vocabulary + void startSpan("unknown", {}, () => {}); + // @ts-expect-error attributes are selected from the schema that owns the span + void startSpan("request", { kind: "read" }, () => {}); + // @ts-expect-error duplicate span names across schemas are rejected + void createTypedSpanStarter(telemetryContext, [operationSchema, operationSchema]); + }; + expectTypeOf(compileTimeFailures).toBeFunction(); + }); }); describe("NOOP_TELEMETRY_CONTEXT", () => { diff --git a/tsconfig.json b/tsconfig.json index 85cd390c3..a62e6ac5b 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -7,6 +7,7 @@ "paths": { "*": ["./*"], "@earendil-works/pi-telemetry": ["./packages/telemetry/src/index.ts"], + "@earendil-works/pi-telemetry/testing": ["./packages/telemetry/src/testing/index.ts"], "@earendil-works/pi-ai": ["./packages/ai/src/index.ts"], "@earendil-works/pi-ai/oauth": ["./packages/ai/src/oauth.ts"], "@earendil-works/pi-ai/*": ["./packages/ai/src/*.ts", "./packages/ai/src/providers/*.ts"], diff --git a/vitest.base.ts b/vitest.base.ts index 8796ba7e9..a07223a0f 100644 --- a/vitest.base.ts +++ b/vitest.base.ts @@ -3,6 +3,7 @@ import { defineConfig } from "vitest/config"; export const workspaceSourcePaths = { telemetryIndex: fileURLToPath(new URL("./packages/telemetry/src/index.ts", import.meta.url)), + telemetryTesting: fileURLToPath(new URL("./packages/telemetry/src/testing/index.ts", import.meta.url)), aiIndex: fileURLToPath(new URL("./packages/ai/src/index.ts", import.meta.url)), aiCompat: fileURLToPath(new URL("./packages/ai/src/compat.ts", import.meta.url)), aiOAuth: fileURLToPath(new URL("./packages/ai/src/oauth.ts", import.meta.url)), @@ -16,6 +17,7 @@ export default defineConfig({ resolve: { alias: [ { find: /^@earendil-works\/pi-telemetry$/, replacement: workspaceSourcePaths.telemetryIndex }, + { find: /^@earendil-works\/pi-telemetry\/testing$/, replacement: workspaceSourcePaths.telemetryTesting }, { find: /^@earendil-works\/pi-ai$/, replacement: workspaceSourcePaths.aiIndex }, { find: /^@earendil-works\/pi-ai\/compat$/, replacement: workspaceSourcePaths.aiCompat }, { find: /^@earendil-works\/pi-ai\/oauth$/, replacement: workspaceSourcePaths.aiOAuth },