Telemetry — Overview

Overview

The @tagna/udiot telemetry subsystem is a unified observability stack built into the framework. It provides four interlocking pillars: metrics, distributed tracing, structured logging, and v0.16 run-event streams for long-running AI and automation work. They are wired together through a shared context propagation system.

The built-in exporters send JSON over HTTP using resource/scope envelopes. Their current payloads include framework-specific metric fields, object-shaped attributes and string span statuses. The implementation and unit fixtures do not establish standard OTLP ingestion compatibility with arbitrary collectors or hosted backends. Validate the actual receiver or provide an application-owned exporter. Logging emits structured records; configure delivery separately.

Key Concepts

• MetricsRegistry — a global singleton that accumulates counters, gauges, and histograms in memory, then flushes them to one or more MetricsExporter implementations on a configurable interval.
• startSpan — the tracing entry point. Every span automatically inherits the ambient trace context from the current async scope and exports the completed span to registered TraceExporter instances.
• TelemetryLogger — a structured logger that attaches the active traceId/spanId to every log line, correlating logs with traces automatically.
• RunEventStream — a typed publish/subscribe stream keyed by runId. It records agent steps, retries, token usage, cost, and terminal outcomes for replay and live UI tails.
• TelemetryContext — a lightweight context object (traceId, spanId, requestId, userId, etc.) that flows through the call stack. Telemetry primitives read from and write to this context.
• SDK init / shutdown — initializeTelemetry() wires exporters and starts the flush loop; shutdownTelemetry() drains the buffer and stops the loop before process exit.

Architecture

Application Code
       │
       ├─► getMetricsRegistry()         (counters / gauges / histograms)
       │         │
       ├─► startSpan(name, opts, fn)    (distributed traces)
       │         │
       ├─► getDefaultRunEventStream()   (agent run history / live tail)
       │         │
       └─► getTelemetryLogger()         (structured logs)
                 │
    ┌────────────┴─────────────────────────────────────┐
    │               TelemetryContext                   │
    │  { traceId, spanId, requestId, userId, tags… }   │
    └────────────┬─────────────────────────────────────┘
                 │ propagated automatically
                 ▼
    ┌────────────────────────────────────┐
    │            Exporters               │
    │  OTLPMetricsExporter               │
    │  OTLPTraceExporter                 │
    │  ConsoleMetricsExporter            │
    └───────────┬────────────────────────┘
                │
                ▼
    Backend (Jaeger / Grafana / Datadog / New Relic / …)

SDK Initialization

Import everything from the top-level telemetry barrel:

import {
  initializeTelemetry,
  shutdownTelemetry,
  OTLPMetricsExporter,
  OTLPTraceExporter,
} from '@tagna/udiot/telemetry';

initializeTelemetry(options?: TelemetryInitOptions): void

Call once at process startup. Safe to call multiple times — subsequent calls stack additional exporters.

shutdownTelemetry(): Promise<void>

Stops the periodic flush loop and flushes all remaining metrics. Call during graceful shutdown (SIGTERM, beforeExit, etc.).

TelemetryInitOptions

Quick Start

import {
  initializeTelemetry,
  shutdownTelemetry,
  getMetricsRegistry,
  startSpan,
  getTelemetryLogger,
  OTLPMetricsExporter,
  OTLPTraceExporter,
  registerTraceExporter,
} from '@tagna/udiot/telemetry';

// 1. Initialize the SDK
initializeTelemetry({
  metrics: {
    exporters: [new OTLPMetricsExporter({ url: 'http://otel-collector:4318/v1/metrics' })],
    flushIntervalMs: 60_000,
  },
});

registerTraceExporter(
  new OTLPTraceExporter({ url: 'http://otel-collector:4318/v1/traces' })
);

// 2. Record a metric
const registry = getMetricsRegistry();
registry.increment('http.requests.total', 1, { method: 'GET', route: '/health' });

// 3. Create a span
const result = await startSpan('processOrder', { kind: 'server' }, async (span) => {
  // Business logic here. The span captures start/end time and status automatically.
  return { orderId: '123' };
});

// 4. Log a structured message (traceId attached automatically)
const logger = getTelemetryLogger();
logger.info('order.processed', { orderId: result.orderId });

// 5. Graceful shutdown
process.on('SIGTERM', async () => {
  await shutdownTelemetry();
  process.exit(0);
});

Common Patterns

Framework-managed initialization — if you use the UDIOT framework config system, call ensureTelemetryFromConfig() instead of initializeTelemetry(). It reads from the telemetry key of your framework config and is idempotent.

Test isolation — call resetMetricsRegistry() and resetTraceExporters() in afterEach to prevent metric state from leaking between tests.

Agent progress streams — pass a runId into LLMClient calls, AgentRunner, or AgentLoop contexts. The runtime publishes RunEvent records that can be replayed or tailed:

import { getDefaultRunEventStream } from '@tagna/udiot/telemetry';

const events = getDefaultRunEventStream();

for await (const event of events.subscribe('run_123', { untilTerminal: true })) {
  console.log(event.agent, event.step, event.outcome);
}

See run-events.md for memory, Redis, and Postgres-backed run-event streams.

Integration Notes

• All telemetry exports are fire-and-forget: a failed HTTP export logs a warning and does not throw, so your application code is never disrupted by an unreachable collector.
• The flush interval timer is unreffed (via timer.unref?.()) so it does not prevent the Node.js process from exiting naturally.
• Context propagation uses a simple module-level variable (activeContext). It is not AsyncLocalStorage-based; if you need per-async-task isolation in heavily concurrent code, set context explicitly with withTelemetryContext() inside each task.