@elsium-ai/observe

Observability, tracing, cost intelligence, metrics, audit trails, provenance tracking, and OpenTelemetry compatibility for ElsiumAI.

Install

npm install @elsium-ai/observe @elsium-ai/core

What’s Inside

Spans

Low-level building blocks for distributed tracing. Each span records a named operation with timing, metadata, events, and parent-child relationships.

SpanKind

type SpanKind = 'llm' | 'tool' | 'agent' | 'workflow' | 'custom'

SpanStatus

type SpanStatus = 'running' | 'ok' | 'error'

SpanEvent

interface SpanEvent {
  name: string
  timestamp: number
  data?: Record<string, unknown>
}

SpanData

The serializable representation of a span, returned by span.toJSON().

interface SpanData {
  id: string
  traceId: string
  parentId?: string
  name: string
  kind: SpanKind
  status: SpanStatus
  startTime: number
  endTime?: number
  durationMs?: number
  metadata: Record<string, unknown>
  events: SpanEvent[]
}

Span

The live span interface used during operation execution.

interface Span {
  readonly id: string
  readonly traceId: string
  readonly name: string
  readonly kind: SpanKind

  addEvent(name: string, data?: Record<string, unknown>): void
  setMetadata(key: string, value: unknown): void
  end(result?: { status?: SpanStatus; metadata?: Record<string, unknown> }): void
  child(name: string, kind?: SpanKind): Span
  toJSON(): SpanData
}

SpanHandler

Callback invoked when a span ends.

type SpanHandler = (span: SpanData) => void

createSpan()

Creates a new span for tracing an operation.

function createSpan(
  name: string,
  options?: {
    traceId?: string
    parentId?: string
    kind?: SpanKind
    onEnd?: SpanHandler
  },
): Span

Returns: Span

import { createSpan } from '@elsium-ai/observe'

const span = createSpan('generate-summary', { kind: 'llm' })
span.setMetadata('model', 'gpt-4o')
span.addEvent('prompt-sent', { tokens: 120 })

// Create a child span for a sub-operation
const child = span.child('embed-context', 'tool')
child.end()

span.end({ status: 'ok', metadata: { outputTokens: 350 } })

console.log(span.toJSON())

Cost Engine

Budget enforcement and cost intelligence for LLM usage. Tracks spend across models, agents, users, and features. Detects runaway loops and suggests cheaper model alternatives.

BudgetConfig

interface BudgetConfig {
  totalBudget?: number
  dailyBudget?: number
  perUser?: number
  perFeature?: number
  perAgent?: number
}

LoopDetectionConfig

interface LoopDetectionConfig {
  maxCallsPerMinute?: number
  maxCostPerMinute?: number
}

CostAlert

interface CostAlert {
  type: 'threshold' | 'loop_detected' | 'budget_exceeded' | 'projection_warning'
  dimension: string
  currentValue: number
  limit: number
  message: string
  timestamp: number
}

CostDimension

Aggregated cost data for a single dimension (model, agent, user, or feature).

interface CostDimension {
  totalCost: number
  totalTokens: number
  callCount: number
  firstCallAt: number
  lastCallAt: number
}

CostIntelligenceReport

Full cost intelligence report with projections and recommendations.

interface CostIntelligenceReport {
  totalSpend: number
  totalTokens: number
  totalCalls: number
  projectedDailySpend: number
  projectedMonthlySpend: number
  byModel: Record<string, CostDimension>
  byAgent: Record<string, CostDimension>
  byUser: Record<string, CostDimension>
  byFeature: Record<string, CostDimension>
  recommendations: string[]
  alerts: CostAlert[]
}

ModelSuggestion

A recommendation to switch to a cheaper model.

interface ModelSuggestion {
  currentModel: string
  suggestedModel: string
  estimatedSavings: number
  reason: string
}

ModelTierEntry

Defines the pricing tier for a model.

interface ModelTierEntry {
  tier: 'low' | 'mid' | 'high'
  costPerMToken: number
}

CostEngineConfig

interface CostEngineConfig {
  totalBudget?: number
  dailyBudget?: number
  perUser?: number
  perFeature?: number
  perAgent?: number
  loopDetection?: LoopDetectionConfig
  onAlert?: (alert: CostAlert) => void
  alertThresholds?: number[]
}

CostEngine

interface CostEngine {
  middleware(): Middleware
  getReport(): CostIntelligenceReport
  suggestModel(currentModel: string, inputTokens: number): ModelSuggestion | null
  trackCall(
    response: LLMResponse,
    dimensions?: { agent?: string; user?: string; feature?: string },
  ): void
  reset(): void
}

createCostEngine()

Creates a cost engine with budget enforcement, alerting, and cost intelligence.

function createCostEngine(config?: CostEngineConfig): CostEngine

Returns: CostEngine

import { createCostEngine } from '@elsium-ai/observe'

const engine = createCostEngine({
  totalBudget: 50.0,
  dailyBudget: 5.0,
  perAgent: 10.0,
  loopDetection: { maxCallsPerMinute: 60, maxCostPerMinute: 1.0 },
  alertThresholds: [0.5, 0.8, 0.95],
  onAlert: (alert) => console.warn(alert.message),
})

// Use as middleware in an ElsiumAI gateway
// gateway.use(engine.middleware())

// Or track calls manually
// engine.trackCall(response, { agent: 'summarizer', user: 'user-123' })

// Get a full intelligence report
const report = engine.getReport()
console.log('Projected monthly spend:', report.projectedMonthlySpend)
console.log('Recommendations:', report.recommendations)

// Get model optimization suggestions
const suggestion = engine.suggestModel('claude-opus-4-6', 200)
if (suggestion) {
  console.log(`Switch to ${suggestion.suggestedModel} for ${suggestion.estimatedSavings.toFixed(0)}% savings`)
}

registerModelTier()

Registers a custom model with its pricing tier so the cost engine can track it.

function registerModelTier(model: string, entry: ModelTierEntry): void

import { registerModelTier } from '@elsium-ai/observe'

registerModelTier('my-custom-model', { tier: 'mid', costPerMToken: 1.5 })

The cost engine ships with built-in tiers for common models including GPT-4o, GPT-4.1, GPT-5, Claude Sonnet 4.6, Claude Opus 4.6, Claude Haiku 4.5, Gemini 2.0 Flash, Gemini 2.5 Pro, o1, o3, o4-mini, and more.

createLocalCostStore()

In-memory reference adapter for the CostStore port — the async-first contract for cost attribution across processes and instances. This package ships only the in-memory adapter. For multi-instance deployments with shared budget, implement CostStore against your own backend (SQLite, Postgres, Redis, DynamoDB, …); no DB drivers are bundled.

reserve / commit / release let callers pre-reserve a budget before the LLM call commits, so concurrent requests racing against the same tenant cap don’t double-spend.

createLocalCostStore(options?: LocalCostStoreOptions): CostStore

Options:

CostStore methods:

CostDimensionKey is one of 'model' | 'agent' | 'user' | 'feature' | 'tenant' | 'workflow' | 'workflowStep' | 'traceId'.

import { createLocalCostStore } from '@elsium-ai/observe'

const store = createLocalCostStore({ reservationTtlMs: 30_000 })

// Reserve before the call, commit the real cost after.
const token = await store.reserve({ model: 'claude-sonnet-4-20250514', tenant: 'acme' }, 0.02)
const response = await gw.complete(request)
await store.commit(token, response.cost)

// Aggregate spend per tenant over the last hour.
const buckets = await store.aggregate('tenant', undefined, {
  fromMs: Date.now() - 3_600_000,
  toMs: Date.now(),
})
// buckets: [{ key: 'acme', cost, tokens, calls, firstAt, lastAt }, ...]

detectDrift()

Compares output distributions between two snapshots of the same canonical input set — typically “yesterday’s model version” vs “today’s”. Reports exact-match rate, length-distribution shift, tool-call divergence, and (optionally, via a pluggable provider) semantic-similarity drift. Designed to run in production against sampled real traffic, not only in CI.

Semantic similarity is bring-your-own via the SimilarityProvider port (cosine over your embeddings, an LLM-as-judge, anything). No embedding library is bundled; if no provider is supplied, semantic metrics are omitted and the composite score is re-normalized.

detectDrift(config: DriftDetectionConfig): Promise<DriftReport>

Config (DriftDetectionConfig):

A DriftSample is { input: string; output: string; tokens?: number; toolCalls?: readonly string[] }. Samples are paired by input (usually a hash of the prompt); unpaired inputs are reported in mismatchedInputs.

Result (DriftReport):

import { detectDrift } from '@elsium-ai/observe'

const report = await detectDrift({
  baseline: [{ input: 'q1', output: 'Paris', toolCalls: ['search'] }],
  current: [{ input: 'q1', output: 'Paris, France', toolCalls: ['search'] }],
  // Optional: bring your own semantic similarity.
  similarity: { similarity: async (a, b) => cosine(await embed(a), await embed(b)) },
})

if (report.overallDrift > 0.3) {
  console.warn('Model drift detected', report.mismatchedInputs)
}

Tracer

High-level tracing API that wraps spans with sampling, console output, cost tracking, and pluggable exporters.

TracerOutput

type TracerOutput = 'console' | 'json-file' | TracerExporter

TracerExporter

Interface for custom span exporters.

interface TracerExporter {
  name: string
  export(spans: SpanData[]): void | Promise<void>
}

TracerConfig

interface TracerConfig {
  output?: TracerOutput[]
  costTracking?: boolean
  samplingRate?: number
  maxSpans?: number
}

CostReport

interface CostReport {
  totalCost: number
  totalTokens: number
  totalInputTokens: number
  totalOutputTokens: number
  callCount: number
  byModel: Record<
    string,
    {
      cost: number
      tokens: number
      calls: number
    }
  >
}

Tracer

interface Tracer {
  startSpan(name: string, kind?: SpanKind): Span
  getSpans(): SpanData[]
  getCostReport(): CostReport
  trackLLMCall(data: {
    model: string
    inputTokens: number
    outputTokens: number
    cost: number
    latencyMs: number
  }): void
  reset(): void
  flush(): Promise<void>
}

observe()

Creates a tracer instance for recording spans and LLM call costs.

function observe(config?: TracerConfig): Tracer

Returns: Tracer

import { observe } from '@elsium-ai/observe'

const tracer = observe({
  output: ['console'],
  samplingRate: 1.0,
  costTracking: true,
})

// Start a span
const span = tracer.startSpan('chat-completion', 'llm')
span.setMetadata('model', 'gpt-4o')
span.end()

// Track an LLM call for cost reporting
tracer.trackLLMCall({
  model: 'gpt-4o',
  inputTokens: 500,
  outputTokens: 200,
  cost: 0.0035,
  latencyMs: 1200,
})

// Get cost report
const report = tracer.getCostReport()
console.log('Total cost:', report.totalCost)

// Flush spans to all exporters
await tracer.flush()

You can provide a custom exporter:

import { observe } from '@elsium-ai/observe'
import type { TracerExporter } from '@elsium-ai/observe'

const myExporter: TracerExporter = {
  name: 'my-backend',
  async export(spans) {
    await fetch('https://my-telemetry.example.com/spans', {
      method: 'POST',
      body: JSON.stringify(spans),
    })
  },
}

const tracer = observe({ output: [myExporter] })

Metrics

General-purpose metrics collection with counters, gauges, and histograms.

MetricEntry

interface MetricEntry {
  name: string
  type: 'counter' | 'gauge' | 'histogram'
  value: number
  tags: Record<string, string>
  timestamp: number
}

MetricsCollector

interface MetricsCollector {
  increment(name: string, value?: number, tags?: Record<string, string>): void
  gauge(name: string, value: number, tags?: Record<string, string>): void
  histogram(name: string, value: number, tags?: Record<string, string>): void
  getMetrics(): MetricEntry[]
  reset(): void
}

createMetrics()

Creates a metrics collector for counters, gauges, and histograms.

function createMetrics(options?: { maxEntries?: number }): MetricsCollector

Returns: MetricsCollector

import { createMetrics } from '@elsium-ai/observe'

const metrics = createMetrics()

// Increment a counter (default increment is 1)
metrics.increment('llm.calls', 1, { model: 'gpt-4o' })

// Set a gauge to a current value
metrics.gauge('queue.depth', 42, { queue: 'embeddings' })

// Record a histogram observation
metrics.histogram('llm.latency_ms', 1200, { model: 'gpt-4o' })

// Retrieve all recorded entries
const entries = metrics.getMetrics()
console.log(`Recorded ${entries.length} metric entries`)

// Reset all state
metrics.reset()

Audit Trail

Tamper-evident audit logging with SHA-256 hash-chaining. Every event is linked to the previous one via its hash, enabling integrity verification of the full event history.

AuditEventType

type AuditEventType =
  | 'llm_call'
  | 'tool_execution'
  | 'security_violation'
  | 'budget_alert'
  | 'policy_violation'
  | 'auth_event'
  | 'approval_request'
  | 'approval_decision'
  | 'config_change'

AuditEvent

interface AuditEvent {
  id: string
  sequenceId: number
  type: AuditEventType
  timestamp: number
  actor?: string
  traceId?: string
  data: Record<string, unknown>
  hash: string
  previousHash: string
}

AuditStorageAdapter

Interface for custom audit storage backends (database, file system, etc.).

interface AuditStorageAdapter {
  append(event: AuditEvent): void | Promise<void>
  query(filter: AuditQueryFilter): AuditEvent[] | Promise<AuditEvent[]>
  count(): number | Promise<number>
  verifyIntegrity(): AuditIntegrityResult | Promise<AuditIntegrityResult>
  getLastHash?(): string | Promise<string>
}

AuditQueryFilter

interface AuditQueryFilter {
  type?: AuditEventType | AuditEventType[]
  actor?: string
  traceId?: string
  fromTimestamp?: number
  toTimestamp?: number
  limit?: number
  offset?: number
}

AuditIntegrityResult

interface AuditIntegrityResult {
  valid: boolean
  totalEvents: number
  brokenAt?: number
  chainComplete?: boolean
}

AuditTrailConfig

interface AuditTrailConfig {
  storage?: AuditStorageAdapter | 'memory'
  hashChain?: boolean
  maxEvents?: number
  onError?: (error: unknown) => void
}

AuditTrail

interface AuditTrail {
  log(
    type: AuditEventType,
    data: Record<string, unknown>,
    options?: { actor?: string; traceId?: string },
  ): void
  query(filter: AuditQueryFilter): Promise<AuditEvent[]>
  verifyIntegrity(): Promise<AuditIntegrityResult>
  readonly count: number
}

createAuditTrail()

Creates a hash-chained audit trail for tamper-evident event logging.

function createAuditTrail(config?: AuditTrailConfig): AuditTrail

Returns: AuditTrail

import { createAuditTrail } from '@elsium-ai/observe'

const audit = createAuditTrail({ hashChain: true })

// Log events
audit.log('llm_call', {
  model: 'gpt-4o',
  inputTokens: 500,
  outputTokens: 200,
  cost: 0.0035,
}, { actor: 'user-123', traceId: 'trc_abc' })

audit.log('tool_execution', {
  tool: 'web-search',
  query: 'latest news',
  resultCount: 10,
})

// Query events
const llmEvents = await audit.query({ type: 'llm_call', limit: 50 })
console.log(`Found ${llmEvents.length} LLM call events`)

// Verify the hash chain has not been tampered with
const integrity = await audit.verifyIntegrity()
console.log('Audit chain valid:', integrity.valid)
console.log('Total events:', integrity.totalEvents)

auditMiddleware()

Creates an ElsiumAI middleware that automatically logs every LLM call (success or failure) to the given audit trail.

function auditMiddleware(auditTrail: AuditTrail): Middleware

Returns: Middleware (from @elsium-ai/core)

import { createAuditTrail, auditMiddleware } from '@elsium-ai/observe'

const audit = createAuditTrail({ hashChain: true })
const middleware = auditMiddleware(audit)

// Use with an ElsiumAI gateway
// gateway.use(middleware)

The middleware automatically records llm_call events containing provider, model, inputTokens, outputTokens, totalTokens, cost, latencyMs, and stopReason. On errors, it records the error message and success: false.

Audit Sinks

Sinks forward audit events to external destinations. A SinkManager batches events in memory and fans them out to one or more sinks with retry, backpressure, and optional dead-lettering. Every sink implements the AuditSink interface:

interface AuditSink {
  name: string
  filter?: (event: AuditEvent) => boolean
  send(events: AuditEvent[]): Promise<void>
  shutdown?(): Promise<void>
}

createSinkManager()

Buffers events, dispatches them in batches, and delivers each batch to every configured sink with exponential-backoff retry and jitter. Failed deliveries (after retries) are routed to an optional deadLetterSink. The internal flush timer is unref’d so it won’t keep the process alive.

createSinkManager(config: SinkManagerConfig): SinkManager

Config (SinkManagerConfig):

SinkManager methods: dispatch(event) (enqueue, non-blocking), flush(): Promise<void> (drain buffer and await in-flight), shutdown(): Promise<void> (stop timer, flush, then call each sink’s shutdown).

import {
  createSinkManager,
  createWebhookSink,
  createDatadogSink,
  createJsonlSink,
} from '@elsium-ai/observe'

const manager = createSinkManager({
  sinks: [createWebhookSink({ url: 'https://siem.example.com/ingest' }), createDatadogSink({ apiKey: env('DD_API_KEY') })],
  deadLetterSink: createJsonlSink({ path: './audit-dlq.jsonl' }),
  batch: { size: 100, intervalMs: 2000 },
  onError: (sink, err) => console.error('[audit-sink]', sink, err),
})

manager.dispatch(event)
await manager.shutdown() // on process exit

createWebhookSink()

POSTs (or PUTs) batches as { events } JSON to an HTTP endpoint, with an abortable timeout.

createWebhookSink(config: WebhookSinkConfig): AuditSink

createSplunkSink()

Sends events to Splunk HTTP Event Collector (HEC), one newline-delimited JSON event per record.

createSplunkSink(config: SplunkSinkConfig): AuditSink

createDatadogSink()

Ships events to the Datadog Log Intake API. Security-violation events are tagged status: 'error'; others info. The endpoint is derived from site.

createDatadogSink(config: DatadogSinkConfig): AuditSink

createJsonlSink()

Appends events as newline-delimited JSON to a local file, creating parent directories as needed. Writes are serialized through a write lock; fsync is called after each batch by default for durability. Implements shutdown to flush and close the handle.

createJsonlSink(config: JsonlSinkConfig): AuditSink

Provenance

Tracks the full lineage of every AI-generated output by hashing the prompt, model, config, input, and output. Enables reproducibility audits and output-to-source tracing.

ProvenanceRecord

interface ProvenanceRecord {
  id: string
  outputHash: string
  promptVersion: string
  modelVersion: string
  configHash: string
  inputHash: string
  timestamp: number
  traceId?: string
  metadata?: Record<string, unknown>
}

All hash fields (outputHash, promptVersion, modelVersion, configHash, inputHash) are SHA-256 hex digests of their respective inputs.

ProvenanceTracker

interface ProvenanceTracker {
  record(data: {
    prompt: string
    model: string
    config: Record<string, unknown>
    input: string
    output: string
    traceId?: string
    metadata?: Record<string, unknown>
  }): ProvenanceRecord
  query(filter: {
    outputHash?: string
    promptVersion?: string
    modelVersion?: string
    traceId?: string
  }): ProvenanceRecord[]
  getLineage(outputHash: string): ProvenanceRecord[]
  readonly count: number
  clear(): void
}

createProvenanceTracker()

Creates a provenance tracker for recording and querying output lineage.

function createProvenanceTracker(options?: {
  maxRecords?: number
}): ProvenanceTracker

Returns: ProvenanceTracker

import { createProvenanceTracker } from '@elsium-ai/observe'

const provenance = createProvenanceTracker()

// Record a generation
const record = provenance.record({
  prompt: 'Summarize the following article...',
  model: 'gpt-4o',
  config: { temperature: 0.7, maxTokens: 500 },
  input: 'The article text here...',
  output: 'A concise summary of the article.',
  traceId: 'trc_abc',
})

console.log('Output hash:', record.outputHash)

// Query records by output hash
const matches = provenance.query({ outputHash: record.outputHash })

// Get the full lineage for a given output (all records sharing the same traceId)
const lineage = provenance.getLineage(record.outputHash)
console.log(`Lineage has ${lineage.length} steps`)

// Check count and clear
console.log('Total records:', provenance.count)
provenance.clear()

Experiments

createExperiment(config)

createExperiment(config: ExperimentConfig): Experiment

Creates an A/B testing and evaluation framework for comparing models, prompts, or configurations.

createFileExperimentStore(dir)

createFileExperimentStore(dir: string): ExperimentStore

Creates a file-based experiment store for persisting experiment results.

import { createExperiment, createFileExperimentStore } from '@elsium-ai/observe'

const store = createFileExperimentStore('./experiments')

const experiment = createExperiment({
  name: 'prompt-comparison',
  variants: [
    { name: 'concise', weight: 1, config: { systemPrompt: 'Be concise.' } },
    { name: 'detailed', weight: 1, config: { systemPrompt: 'Be detailed and thorough.' } },
  ],
  store,
})

const variant = experiment.assign(userId)
const result = await runWithVariant(variant)
experiment.record(variant.name, result)

const analysis = experiment.results()

Instrumentation

instrumentComplete(complete, tracer)

instrumentComplete(complete: CompleteFn, tracer: Tracer): CompleteFn

Wraps an LLM complete function with automatic span creation, recording model, tokens, cost, and errors.

instrumentAgent(agent, tracer)

instrumentAgent(agent: Agent, tracer: Tracer): Agent

Wraps an agent with automatic tracing, creating spans for each run and tool invocation.

import { observe, instrumentComplete, instrumentAgent } from '@elsium-ai/observe'
import { defineAgent } from '@elsium-ai/agents'

const tracer = observe()

// Instrument gateway completions
const tracedComplete = instrumentComplete(gw.complete.bind(gw), tracer)
const response = await tracedComplete(request) // automatically traced

// Instrument an agent
const agent = defineAgent({ name: 'assistant', /* ... */ })
const tracedAgent = instrumentAgent(agent, tracer)
const result = await tracedAgent.run('Hello') // automatically traced with per-tool spans

OpenTelemetry

Compatibility layer for converting ElsiumAI spans to the OpenTelemetry format, propagating W3C Trace Context headers, and exporting via OTLP JSON to any OTel-compatible backend (Jaeger, Grafana Tempo, Datadog, Honeycomb, etc.).

OTelSpanKind

type OTelSpanKind = 0 | 1 | 2 | 3 | 4 | 5
// 0 = UNSPECIFIED, 1 = INTERNAL, 2 = SERVER, 3 = CLIENT, 4 = PRODUCER, 5 = CONSUMER

OTelStatusCode

type OTelStatusCode = 0 | 1 | 2
// 0 = UNSET, 1 = OK, 2 = ERROR

OTelAttributeValue

interface OTelAttributeValue {
  stringValue?: string
  intValue?: number
  doubleValue?: number
  boolValue?: boolean
  arrayValue?: { values: OTelAttributeValue[] }
}

OTelAttribute

interface OTelAttribute {
  key: string
  value: OTelAttributeValue
}

OTelEvent

interface OTelEvent {
  name: string
  timeUnixNano: string
  attributes: OTelAttribute[]
}

OTelSpan

The OpenTelemetry-compatible span representation.

interface OTelSpan {
  traceId: string
  spanId: string
  parentSpanId?: string
  name: string
  kind: OTelSpanKind
  startTimeUnixNano: string
  endTimeUnixNano: string
  attributes: OTelAttribute[]
  events: OTelEvent[]
  status: {
    code: OTelStatusCode
    message?: string
  }
}

OTelResource

interface OTelResource {
  attributes: OTelAttribute[]
}

OTelExportRequest

The full OTLP JSON export payload structure.

interface OTelExportRequest {
  resourceSpans: Array<{
    resource: OTelResource
    scopeSpans: Array<{
      scope: {
        name: string
        version: string
      }
      spans: OTelSpan[]
    }>
  }>
}

TraceContext

Parsed W3C Trace Context.

interface TraceContext {
  traceId: string
  spanId: string
  traceFlags: number
  traceState?: string
}

OTLPExporterConfig

interface OTLPExporterConfig {
  /** OTLP endpoint URL (e.g. http://localhost:4318/v1/traces) */
  endpoint: string
  /** Optional headers (e.g. for auth) */
  headers?: Record<string, string>
  /** Service name for resource attributes */
  serviceName?: string
  /** Service version */
  serviceVersion?: string
  /** Batch size before sending */
  batchSize?: number
  /** Flush interval in ms */
  flushIntervalMs?: number
}

toOTelSpan()

Converts an ElsiumAI SpanData to an OpenTelemetry-compatible span.

function toOTelSpan(span: SpanData): OTelSpan

Returns: OTelSpan

import { createSpan, toOTelSpan } from '@elsium-ai/observe'

const span = createSpan('my-operation', { kind: 'llm' })
span.end()

const otelSpan = toOTelSpan(span.toJSON())
console.log(otelSpan.traceId, otelSpan.spanId)

ElsiumAI span kinds are mapped as follows: llm to CLIENT (3), tool/agent/workflow to INTERNAL (1), custom to UNSPECIFIED (0).

toOTelExportRequest()

Builds a full OTLP JSON export request from a batch of spans.

function toOTelExportRequest(
  spans: SpanData[],
  options?: {
    serviceName?: string
    serviceVersion?: string
  },
): OTelExportRequest

Returns: OTelExportRequest

import { observe, toOTelExportRequest } from '@elsium-ai/observe'

const tracer = observe({ output: [] })
const span = tracer.startSpan('process-request', 'agent')
span.end()

const payload = toOTelExportRequest(tracer.getSpans(), {
  serviceName: 'my-ai-service',
  serviceVersion: '1.0.0',
})

// Send to any OTLP-compatible endpoint
await fetch('http://localhost:4318/v1/traces', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify(payload),
})

toTraceparent()

Creates a W3C traceparent header value from a span.

function toTraceparent(span: SpanData): string

Returns: string — Format: 00-{traceId}-{spanId}-01

import { createSpan, toTraceparent } from '@elsium-ai/observe'

const span = createSpan('outgoing-call', { kind: 'llm' })
span.end()

const header = toTraceparent(span.toJSON())
// e.g. "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"

parseTraceparent()

Parses a W3C traceparent header string into a TraceContext.

function parseTraceparent(header: string): TraceContext | null

Returns: TraceContext | null — Returns null if the header is malformed or uses an unsupported version.

import { parseTraceparent } from '@elsium-ai/observe'

const ctx = parseTraceparent('00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01')
if (ctx) {
  console.log('Trace ID:', ctx.traceId)
  console.log('Span ID:', ctx.spanId)
  console.log('Sampled:', ctx.traceFlags === 1)
}

injectTraceContext()

Injects the traceparent header into an outgoing HTTP headers object for distributed trace propagation.

function injectTraceContext(
  span: SpanData,
  headers?: Record<string, string>,
): Record<string, string>

Returns: Record<string, string> — A new headers object with traceparent set.

import { createSpan, injectTraceContext } from '@elsium-ai/observe'

const span = createSpan('api-call', { kind: 'llm' })
const headers = injectTraceContext(span.toJSON(), {
  'Content-Type': 'application/json',
  Authorization: 'Bearer token',
})
// headers now includes { traceparent: '00-...-...-01', ... }

await fetch('https://api.example.com/generate', { headers })
span.end()

extractTraceContext()

Extracts a TraceContext from incoming HTTP headers. Checks both traceparent and Traceparent keys.

function extractTraceContext(
  headers: Record<string, string | undefined>,
): TraceContext | null

Returns: TraceContext | null

import { extractTraceContext, createSpan } from '@elsium-ai/observe'

// In an HTTP handler
function handleRequest(req: { headers: Record<string, string> }) {
  const parentCtx = extractTraceContext(req.headers)

  const span = createSpan('handle-request', {
    traceId: parentCtx?.traceId,
    parentId: parentCtx?.spanId,
    kind: 'agent',
  })

  // ... process request ...
  span.end()
}

createOTLPExporter()

Creates an OTLP JSON exporter that sends spans to any OTel-compatible backend. Supports batching and automatic periodic flushing.

function createOTLPExporter(config: OTLPExporterConfig): TracerExporter

Returns: TracerExporter

import { observe, createOTLPExporter } from '@elsium-ai/observe'

const exporter = createOTLPExporter({
  endpoint: 'http://localhost:4318/v1/traces',
  serviceName: 'my-ai-service',
  serviceVersion: '1.0.0',
  headers: { Authorization: 'Bearer my-token' },
  batchSize: 50,
  flushIntervalMs: 10000,
})

const tracer = observe({ output: [exporter] })

const span = tracer.startSpan('generate', 'llm')
span.end()

// Spans are batched and sent automatically, or flush manually:
await tracer.flush()

Compliance Reporting

Generate compliance reports against regulatory frameworks from audit trail data.

generateComplianceReport

generateComplianceReport(
  auditTrail: AuditTrail,
  config: ComplianceReportConfig,
): Promise<ComplianceReport>

Supported Frameworks:

Config:

import { createAuditTrail, generateComplianceReport, formatComplianceReport } from '@elsium-ai/observe'

const audit = createAuditTrail({ hashChain: true })

const report = await generateComplianceReport(audit, {
  framework: 'eu-ai-act',
  systemName: 'medical-triage-ai',
  systemVersion: '2.1.0',
  reportPeriod: {
    from: Date.now() - 30 * 24 * 60 * 60 * 1000,
    to: Date.now(),
  },
  riskLevel: 'high',
})

formatComplianceReport

formatComplianceReport(report: ComplianceReport): string

Signed Proofs (Verifiable Agent Execution)

Per-run, cryptographically signed execution proofs. Each agent run is recorded as a SHA-256 hash chain of events (LLM calls, tool calls, RAG retrievals, policy decisions), and the chain head is signed with Ed25519. Anyone holding the public key can later verify — offline — that the run happened exactly as recorded and was produced by the holder of the signing key. Two proofs can be diffed to check replay reproducibility.

Builds on @elsium-ai/core crypto (createEd25519Signer, createKeyRegistry, WriteOnceStore). Proof event data stores hashes of inputs/outputs (requestHash, responseHash, inputHash, …), not raw payloads, so proofs are safe to share without leaking prompt content.

createProofRecorder()

Creates a recorder bound to a signer. Open a session per agent run, record events, then finalize() to produce a signed ExecutionProof.

createProofRecorder(config: ProofRecorderConfig): ProofRecorder

Config (ProofRecorderConfig):

ProofRecorder methods:

StartSessionOptions:

ProofSession methods: recordLLMCall(summary), recordToolCall(summary), recordRagRetrieve(summary), recordPolicyDecision(summary), recordCustom(data) — each appends a chained ProofEvent and returns it — plus finalize(options?).

finalize(options?: FinalizeOptions) appends an agent.output event if finalOutput is given, signs the chain head, and returns the ExecutionProof. If a WriteOnceStore is passed via options.store, the proof is persisted under options.storeKey?.(proofId) (default <proofId>.json). Recording after finalize throws.

import { createProofRecorder } from '@elsium-ai/observe'
import {
  createEd25519Signer,
  createKeyRegistry,
  createInMemoryWriteOnceStore,
  generateEd25519KeyPair,
} from '@elsium-ai/core'

const keyPair = generateEd25519KeyPair()
const signer = createEd25519Signer({ privateKey: keyPair.privateKey, keyId: 'agent-key-1' })
const store = createInMemoryWriteOnceStore()

const recorder = createProofRecorder({ signer })

const session = recorder.startSession({
  agentId: 'support-agent',
  agentVersion: '1.4.0',
  inputs: { messages: [{ role: 'user', content: 'reset my password' }] },
  reproducibility: { modelVersions: { primary: 'claude-sonnet-4-20250514' } },
})

session.recordToolCall({ tool: 'lookup_account', inputHash: 'a1b2…', outputHash: 'c3d4…' })
session.recordPolicyDecision({ rule: 'pii-redaction', result: 'allow' })

const proof = await session.finalize({ finalOutput: 'Password reset link sent.', store })
// proof.chainHead is signed; proof.signature carries keyId + Ed25519 value

verifyProof()

Verifies a proof offline in two stages: (1) recompute the hash chain (each hashPrev/hashSelf and the chainHead), then (2) verify the Ed25519 signature over the chain head against the trusted keys in registry. Returns early with chainValid: false if the chain is broken.

verifyProof(proof: ExecutionProof, registry: KeyRegistry): VerifyProofResult

Result (VerifyProofResult):

import { verifyProof } from '@elsium-ai/observe'
import { createKeyRegistry } from '@elsium-ai/core'

const registry = createKeyRegistry()
registry.add('agent-key-1', keyPair.publicKey)

const result = verifyProof(proof, registry)
if (!result.valid) {
  console.error('Proof rejected:', result.reason, 'broken at', result.chainBrokenAt)
}

compareProofs()

Diffs two proofs event-by-event to assess replay reproducibility. Two strategies:

• 'structural' (default): ignores non-deterministic event types (llm.call data beyond model/provider, agent.input, agent.output, custom) and compares deterministic events (tool calls, policy decisions, RAG) by data. matches is true when no deltas are found.
• 'bit-exact': compares event type and hashSelf; additionally requires chainHeadMatch for matches.

compareProofs(proofA: ExecutionProof, proofB: ExecutionProof, options?: { strategy?: ReplayStrategy }): ReplayDiff

Result (ReplayDiff):

import { compareProofs } from '@elsium-ai/observe'

const diff = compareProofs(recordedProof, replayedProof, { strategy: 'structural' })
if (!diff.matches) {
  console.warn('Replay diverged', diff.deltas)
}

ExecutionProof shape

interface ExecutionProof {
  version: 'elsium-proof/v1'
  proofId: string
  agentId: string
  agentVersion?: string
  startedAt: string  // ISO 8601
  endedAt: string    // ISO 8601
  events: ProofEvent[]
  chainHead: string  // hashSelf of the final event
  signature: Signature  // Ed25519, carries keyId
  reproducibility?: ReproducibilityHints
}

Each ProofEvent is { sequence, type, timestamp, data, hashPrev, hashSelf } where type is one of agent.input, agent.output, llm.call, tool.call, rag.retrieve, policy.evaluated, custom. The constants PROOF_VERSION ('elsium-proof/v1') and PROOF_SESSION_METADATA_KEY ('proofSessionId', the ctx.metadata key the middleware reads) are also exported.

AI-BOM (agent composition manifest)

A proof records one run. An AI-BOM records the composition every run inherits: which models, prompts, tools, MCP servers, eval datasets, policies and thresholds the agent is built from — signed, so composition cannot change without detection.

generateAiBom()

generateAiBom(input: AiBomInput, options: GenerateAiBomOptions): Promise<AiBom>

import { createEd25519Signer, generateEd25519KeyPair } from '@elsium-ai/core'
import { generateAiBom } from '@elsium-ai/observe'

const pair = generateEd25519KeyPair()
const signer = createEd25519Signer({ privateKey: pair.privateKey, keyId: 'release-key' })

const bom = await generateAiBom(
  {
    agentId: 'loan-underwriter',
    agentVersion: '2.3.1',
    environment: 'production',
    models: [
      { provider: 'anthropic', model: 'claude-sonnet-4-6', role: 'primary', region: 'eu-west-1' },
    ],
    prompts: [{ name: 'system', version: '7', content: SYSTEM_PROMPT }],
    tools: [creditCheck],                 // a real Tool is accepted by shape
    datasets: [goldenSetManifest],        // a real DatasetManifest is too
    policies: [{ name: 'lending', version: '3', mode: 'enforce', document: bundle }],
    thresholds: { confidenceFloor: 0.8, maxCostUsd: 5 },
    runtime: { framework: 'elsium-ai', frameworkVersion: '0.20.0' },
  },
  { signer },
)

AiBomInput fields are all optional except agentId. Components may be supplied either pre-hashed ({ name, sha256 }) or as raw content ({ name, content } / { name, document }) for the framework to hash.

Component sources are matched structurally, not by import: @elsium-ai/observe depends on core alone, so passing a Tool from @elsium-ai/tools or a DatasetManifest from @elsium-ai/testing works without coupling the packages.

Generation is deterministic and order-independent — components are sorted by identity before hashing, so moving a defineTool call does not read as drift. Pass bomId and clock for byte-identical fixtures.

verifyAiBom()

verifyAiBom(bom: AiBom, registry: KeyRegistry): Promise<VerifyAiBomResult>

Verifies three layers inner-to-outer: components still hashes to componentsHash, the header still hashes to digest, and digest carries a signature from a trusted key.

const result = await verifyAiBom(bom, registry)
// {
//   valid: true, signatureValid: true, componentsHashValid: true, digestValid: true,
//   checked: { componentsHash: true, digest: true, signature: true },
// }

Verification stops at the innermost failure, so a false verdict may mean “failed” or “never evaluated”. Read checked to tell them apart rather than reporting an unevaluated signature as an invalid one.

diffAiBom() / passesGate()

diffAiBom(approved: AiBom, current: AiBom): AiBomDiff
passesGate(diff: AiBomDiff, failOn?: DriftSeverity): boolean   // default 'critical'

const diff = diffAiBom(approvedBom, shippedBom)
// diff.drifts:    ComponentDrift[]  -- kind, componentKind, id, field, severity, reason
// diff.counts:    { critical, major, minor }
// diff.identical: true when the shipped agent matches what was approved

if (!passesGate(diff)) throw new Error('composition drift')

Severity answers one question: did the blast radius grow, or a control weaken?

A removal is rarely critical — losing a capability is a correctness problem. Losing a control is a security one, which is why a removed policy ranks critical while a removed tool ranks major.

diffAiBom is purely structural and does not verify signatures. Verify both BOMs first: a diff against an unverified baseline proves nothing.

AiBom shape

interface AiBom {
  version: 'elsium-aibom/v1'
  bomId: string
  agentId: string
  agentVersion?: string
  generatedAt: string      // ISO 8601
  environment?: string
  components: AiBomComponents
  componentsHash: string   // SHA-256 over the canonical components
  digest: string           // SHA-256 over the header + componentsHash -- what is signed
  signature: Signature     // Ed25519, carries keyId
  metadata?: Record<string, unknown>
}

AiBomComponents holds models, prompts, tools, mcpServers, datasets, policies, thresholds and runtime. serializeAiBom(bom) emits canonical JSON — stable bytes regardless of key insertion order — for the file you commit as the approved BOM. AIBOM_VERSION is exported as a constant.

CI usage

elsium bom verify ./aibom.json --public-key ./release.pub
elsium bom diff ./approved-bom.json ./aibom.json --fail-on critical

See the elsium bom command reference for every flag.

Policy Simulation

Available from @elsium-ai/observe@0.20.0.

Governance controls fail on adoption, not on design. Nobody enables a rule in production without knowing what it breaks, so policies sit in monitor-only mode indefinitely and protect nothing. Recorded ExecutionProofs already hold what happened — replaying a candidate policy over them turns “what will this block?” into an answer with traceIds attached.

Same shape as terraform plan or opa eval: decide against history, then commit.

simulatePolicy()

simulatePolicy(traces: readonly ExecutionProof[], probe: PolicyProbe): PolicySimulation

Read-only. Nothing is executed and no side effect is replayed — a tool call denied in simulation still happened in the recording.

import { simulatePolicy, flowPolicyProbe } from '@elsium-ai/observe'
import { createFlowPolicy, createLabel, lethalTrifectaRule } from '@elsium-ai/core'

const result = simulatePolicy(
  traces,
  flowPolicyProbe({
    policy: createFlowPolicy([lethalTrifectaRule()]),
    initial: createLabel({ classes: ['secret'], origin: 'trusted', source: 'vault' }),
  }),
)

// {
//   traces: 100, evaluated: 200, allowed: 197, denied: 3,
//   denials: [{ traceId: 'proof_...', sequence: 2, subject: 'tool:send_email', rule: 'lethal-trifecta' }],
//   byRule: { 'lethal-trifecta': 3 },
//   affectedTraces: ['proof_...'], affectedTraceRatio: 0.03,
// }

affectedTraceRatio is the number that decides whether a rule ships: it counts runs touched, not decision points, because one run breaking is what a user experiences.

comparePolicies()

comparePolicies(traces, { baseline: PolicyProbe; candidate: PolicyProbe }): PolicyComparison

const plan = comparePolicies(traces, { baseline: current, candidate: proposed })
// { newlyDenied: [...], newlyAllowed: [...], unchanged: 197, baseline: {...}, candidate: {...} }

newlyDenied is the blast radius everyone asks about. newlyAllowed is the one people forget — a relaxed policy silently permitting what used to be blocked is how controls erode.

flowPolicyProbe()

flowPolicyProbe(options: FlowProbeOptions): PolicyProbe

Replays information flow over a recorded run: walks events in order, accumulates provenance as the live tracker would, and checks each sink at the point it was reached. A tool is checked before its own output joins the context — otherwise every tool would appear to taint itself.

Defaults mirror the live enforcement points: rag.retrieve and tool.call results are untrusted, llm.call output is model.

PolicyProbe

type PolicyProbe = (trace: ExecutionProof) => SimulatedDecision[]

A port, not a closed set. Policy engines differ in what they decide over — flows to a sink, authorization requests, budgets — and simulation should not need to know. flowPolicyProbe ships with the package; other engines plug in the same way.

Formatting

formatSimulation(simulation: PolicySimulation): string
formatComparison(comparison: PolicyComparison): string

Plan-style output for a terminal or a CI comment. See Policy Simulation in the Fundamentals guide for the worked example where the default rule blocks 100% of traffic and the narrowed one blocks 3%.

Studio Exporter

createStudioExporter()

A TracerExporter that writes traces, X-Ray history, and cost reports to a local directory for consumption by Elsium Studio tooling. Spans are written per-trace as <dir>/traces/<traceId>.json; X-Ray entries accumulate (newest first, capped at 500) in <dir>/xray-history.json; cost reports are written to <dir>/cost-report.json. All writes are best-effort and never throw (failures are logged).

createStudioExporter(config?: StudioExporterConfig): StudioExporter

Config (StudioExporterConfig):

In addition to the TracerExporter.export(spans) method, StudioExporter adds writeXRayEntry(entry: Record<string, unknown>) and writeCostReport(report: CostReport).

import { createStudioExporter, observe } from '@elsium-ai/observe'

const studio = createStudioExporter({ dir: '.elsium' })

const tracer = observe({ output: [studio] })
// spans are persisted under .elsium/traces/<traceId>.json

studio.writeCostReport(tracer.getCostReport())

Part of ElsiumAI

This package is the observability layer of the ElsiumAI framework. See the full documentation for guides and examples.

License

MIT