Observability

Standing back from the hedge to see what's actually growing where. This page is the read-side observability surface. For the write-side audit ledger that these decorators wrap, see Audit ledger and AuditableCall row.

The substrate emits two structured write streams ready for an observability stack: every StreamEvent fires on the runtime's event emitter, and every LLM call writes one AuditableCall row to the ledger. Both are typed; both are the seams to wire into your telemetry pipeline.

This page documents the patterns that hold up in production — without breaking determinism, without blocking the turn, and without inventing a second write site for data the substrate is already writing.

import type { ProviderDecisionLedger, AuditableCall } from "@pleach/core/audit";
import type { StreamEvent } from "@pleach/core";

Observability is a thematic island. Four distinct surfaces (lineage, observability, otel-observability, runtime-inspector) — not a three-concept cluster. Readers usually arrive already knowing which one they need. See What lives outside the cluster pattern.

The two write streams

Wire both. The lifecycle stream covers turn-level shape (when did this turn start, how many tool calls fired, did it interrupt); the audit ledger covers per-call detail (which model, which family, how many tokens, how long, what was the decision).

OpenTelemetry pattern

The canonical OTel integration wraps the audit ledger in a span tracer and subscribes turn-level events to a span builder.

// lib/observability/tracedLedger.ts
import { trace, SpanKind, SpanStatusCode } from "@opentelemetry/api";
import type { ProviderDecisionLedger, AuditableCall } from "@pleach/core/audit";

const tracer = trace.getTracer("@pleach/core");

export class TracedLedger implements ProviderDecisionLedger {
  constructor(private primary: ProviderDecisionLedger) {}

  async recordCall(call: AuditableCall): Promise<void> {
    const span = tracer.startSpan(`llm.${call.call.callClass}`, {
      kind: SpanKind.CLIENT,
      attributes: {
        "llm.model":         call.call.model,
        "llm.family":        call.call.provider,
        "llm.transport":     call.call.transport,
        "llm.call_class":    call.call.callClass,
        "llm.session_id":    call.sessionId,
        "llm.turn_id":       call.turnId,
        "llm.stage_id":      call.stageId,
        "llm.tokens.input":  call.tokenCost?.inputTokens,
        "llm.tokens.output": call.tokenCost?.outputTokens,
        "llm.latency_ms":    call.outcome.latencyMs,
        "llm.fallback":      call.decision.selectedReason,
        "llm.cache_hit":     (call.cacheBreakpoint?.cacheReadTokens ?? 0) > 0,
      },
    });

    if (call.outcome.status === "failed") {
      span.setStatus({ code: SpanStatusCode.ERROR, message: call.outcome.finishReason ?? undefined });
    }

    span.end();
    return this.primary.recordCall(call);
  }
}

Wire it as the active ledger:

import { setProviderDecisionLedgerFactory } from "@pleach/core/runtime";

setProviderDecisionLedgerFactory({
  fromSupabase: (client) =>
    new TracedLedger(new SupabaseProviderDecisionLedger(client)),
});

Per-call spans are the load-bearing telemetry. Don't roll your own span at the seam call site — the ledger is the seam the substrate already calls. A span emitted from a stream observer hook fires per chunk (potentially hundreds per turn) and breaks the sync-only observer contract; a span emitted from a custom provider wrapper double-counts against the ledger's own span and makes the trace tree confusing. One span per recordCall is the right cardinality — one per LLM call, attributed by turnId and stageId.

Turn-level spans

Wrap each turn in an outer span via the runtime event stream:

const activeSpans = new Map<string, ReturnType<typeof tracer.startSpan>>();

runtime.on("step.start", (event) => {
  if (event.step === "anchor-plan") {
    const span = tracer.startSpan("agent.turn", {
      attributes: { "session.id": event.sessionId },
    });
    activeSpans.set(event.sessionId, span);
  }
});

runtime.on("step.end", (event) => {
  if (event.step === "post-turn") {
    activeSpans.get(event.sessionId)?.end();
    activeSpans.delete(event.sessionId);
  }
});

Turn-level spans + per-call child spans give you a complete trace tree per turn — what stages fired, what models were called, where the latency went.

OTEL spans (in-process)

The substrate now emits four span types directly: session.turn, llm.invocation, graph.stage, and tool.execution. Parent-threading nests them naturally per turn — the turn span contains stage spans, which contain LLM and tool spans.

A runtime.spans facet exposes the exporter lifecycle plus introspection reads. Its snapshot() returns { inFlightCount, isShutdown } — exporter state, not the span records. To inspect the trace tree itself during debugging, wire the reference CapturingOtelExporter and read its captured array (no OTLP collector needed). See /docs/otel-observability.

The ledger-decorator pattern above still works for per-call span emission. The four built-in spans complement it; they don't replace it. Pick the decorator when you want per-recordCall control; use the built-in spans when you want the full turn tree for free.

For the auto-flush controls (otelFlushIntervalTurns), the OTLP wiring recipe, and the pleach.tenant_id attribute, see /docs/otel-observability.

Datadog pattern

Datadog's dd-trace library auto-instruments many libraries. For the substrate, the integration shape is the same as OTel — a ledger decorator emits per-call metrics:

// lib/observability/datadogLedger.ts
import StatsD from "hot-shots";
import type { ProviderDecisionLedger, AuditableCall } from "@pleach/core/audit";

const dogstatsd = new StatsD({ host: process.env.DD_AGENT_HOST });

export class DatadogLedger implements ProviderDecisionLedger {
  constructor(private primary: ProviderDecisionLedger) {}

  async recordCall(call: AuditableCall): Promise<void> {
    const tags = [
      `model:${call.call.model}`,
      `family:${call.call.provider}`,
      `call_class:${call.call.callClass}`,
      `outcome:${call.outcome.status}`,
    ];

    dogstatsd.increment("llm.calls", 1, tags);
    dogstatsd.histogram("llm.latency_ms", call.outcome.latencyMs, tags);
    dogstatsd.histogram("llm.tokens.input",  call.tokenCost?.inputTokens  ?? 0, tags);
    dogstatsd.histogram("llm.tokens.output", call.tokenCost?.outputTokens ?? 0, tags);

    if (call.outcome.status === "failed") {
      dogstatsd.increment("llm.errors", 1, [...tags, `reason:${call.outcome.finishReason}`]);
    }

    return this.primary.recordCall(call);
  }
}

For Datadog APM (distributed tracing), use the @opentelemetry/exporter-trace-otlp-http exporter pointed at Datadog's OTLP intake — the OTel pattern above lands in Datadog APM unchanged.

Honeycomb pattern

Honeycomb consumes OTLP directly. Configure the SDK to point at Honeycomb's OTLP endpoint:

import { NodeSDK } from "@opentelemetry/sdk-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";

const sdk = new NodeSDK({
  traceExporter: new OTLPTraceExporter({
    url: "https://api.honeycomb.io/v1/traces",
    headers: { "x-honeycomb-team": process.env.HONEYCOMB_API_KEY! },
  }),
  serviceName: "my-agent",
});

sdk.start();

Configuring the SDK is necessary but not sufficient: @pleach/core never reads the global tracer provider, so the SDK alone captures no pleach spans. Pass a bridge implementing OtelExporter to the runtime via config.otelExporter to forward the four substrate spans into this pipeline — see the OTLP bridge recipe.

Honeycomb's strength is high-cardinality field aggregation — the per-call ledger row's model_id, turn_id, tenant_id make it a natural fit. Don't pre-aggregate; let Honeycomb do it.

Prometheus pattern

For self-hosted Prometheus, expose a metrics endpoint that reads from an in-memory counter the ledger increments:

// lib/observability/promLedger.ts
import { Counter, Histogram, register } from "prom-client";

const llmCalls = new Counter({
  name: "pleach_llm_calls_total",
  help: "Total LLM calls",
  labelNames: ["model", "family", "call_class", "outcome"],
});

const llmLatency = new Histogram({
  name: "pleach_llm_latency_ms",
  help: "LLM call latency in ms",
  labelNames: ["model", "family", "call_class"],
  buckets: [50, 100, 250, 500, 1000, 2500, 5000, 10000, 30000],
});

export class PromLedger implements ProviderDecisionLedger {
  constructor(private primary: ProviderDecisionLedger) {}

  async recordCall(call: AuditableCall): Promise<void> {
    const labels = {
      model:      call.call.model,
      family:     call.call.provider,
      call_class: call.call.callClass,
    };
    llmCalls.inc({ ...labels, outcome: call.outcome.status });
    llmLatency.observe(labels, call.outcome.latencyMs);
    return this.primary.recordCall(call);
  }
}

Mount the /metrics endpoint:

app.get("/metrics", async (req, res) => {
  res.set("Content-Type", register.contentType);
  res.end(await register.metrics());
});

For high-throughput deployments, the Counter/Histogram update cost is amortized into the ledger write — both are microsecond-scale.

Structured logging

The substrate's default loggers write event types and ids, not payloads. To stream the full event log into a structured logger:

import pino from "pino";
import type { StreamEvent } from "@pleach/core";

const log = pino({ level: process.env.LOG_LEVEL ?? "info" });

const interestingEvents: StreamEvent["type"][] = [
  "error",
  "tool.failed",
  "sync.conflict",
  "stream.truncated",
  "interrupt.requested",
];

for (const type of interestingEvents) {
  runtime.on(type, (event) => {
    log[type === "error" ? "error" : "info"]({
      event_type: type,
      session_id: event.sessionId,
      ...event,
    });
  });
}

Don't log every event variant — message.delta fires per chunk and floods logs. The audit ledger is the per-call surface; structured logs are for the lifecycle shape.

Forwarding tool.completed to a log sink

Subscribe to the lifecycle stream and push completion events to Loki, Honeycomb, or any sink that takes JSON:

runtime.on("tool.completed", async (event) => {
  await fetch(process.env.LOKI_PUSH_URL!, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      streams: [{
        stream: { app: "knowledge-base", tool: event.toolName },
        values: [[String(Date.now() * 1e6), JSON.stringify({
          sessionId:  event.sessionId,
          turnId:     event.turnId,
          toolName:   event.toolName,
          durationMs: event.durationMs,
          status:     event.status,
        })]],
      }],
    }),
  }).catch(() => {});
});

The .catch(() => {}) matters: telemetry failures must not break the turn.

Per-tenant dashboards

Tenant attribution is in the ledger row payload. The query layer groups it:

import { getAggregateUsage } from "@pleach/core/query";

const usage = await getAggregateUsage(client, store, {
  userId:    "tenant-service-account",
  groupBy:   "model",
  dateRange: { from: "2026-06-01", to: "2026-07-01" },
});

Feed this into your dashboarding tool (Grafana, Looker, Metabase). The ledger schema is stable; queries written against it survive substrate upgrades.

Joining the ledger to billing

The load-bearing join key is payload->>'tenantId' on harness_auditable_calls against your billing table's tenant column (typically tenants.id or organizations.id). A second join on payload->>'turnId' lets a per-conversation invoice line-item exist — the same turnId that runtime.executeMessage stamps onto the ledger row is the one the React layer surfaces in DevTools, so support requests like "why was this conversation expensive" map back to a single WHERE payload->>'turnId' = $1 query.

SELECT
  b.tenant_name,
  SUM((c.payload->'tokenUsage'->>'in')::int)  AS input_tokens,
  SUM((c.payload->'tokenUsage'->>'out')::int) AS output_tokens
FROM harness_auditable_calls c
JOIN billing_tenants b ON b.id = c.payload->>'tenantId'
WHERE c.created_at >= date_trunc('month', now())
GROUP BY 1;

Per-day fallback rate

The audit row's payload.fallbackReason is the seam. Rolled up by day, it answers "how often is the primary model failing":

SELECT
  date_trunc('day', created_at)                                AS day,
  COUNT(*)                                                     AS total_calls,
  COUNT(*) FILTER (WHERE payload->>'fallbackReason' IS NOT NULL
                     AND payload->>'fallbackReason' <> 'none') AS fallback_calls,
  ROUND(
    100.0 * COUNT(*) FILTER (WHERE payload->>'fallbackReason' IS NOT NULL
                               AND payload->>'fallbackReason' <> 'none')
    / NULLIF(COUNT(*), 0),
    2
  ) AS fallback_pct
FROM harness_auditable_calls
WHERE created_at >= now() - interval '30 days'
GROUP BY 1
ORDER BY 1;

A rising fallback_pct is the early signal for a primary-provider incident; pair with the alert thresholds below.

What to alert on

Five signals from the substrate that correlate to user-facing issues:

Don't alert on message.delta rate or token counts — those are business metrics, not reliability signals. A rising input-token count on a tenant is usually "they wrote a longer prompt this week," not "the substrate broke." Wire those into the billing dashboard, not the on-call paging path.

Extending observability — plugin hooks

The patterns above show the read-side: subscribe to the runtime's existing streams, decorate the ledger, fold the event log. Sometimes the right move is to inject observability into the stream — capture a custom probe per chunk, sanitize the rendered output, attach a span at a specific node. Two plugin hooks are the supported path.

contributeStreamObservers — per-chunk probes

Stream observers fire per stream delta on a specific seam invocation. They're the hook for per-chunk telemetry — a redaction gate, a latency probe, a coherence-score sampler. The registration declares a when filter (which callClass, stageId, optional nodeId triggers the observer) and a factory that returns the StreamIterationObserver whose onChunk runs per delta.

import type { HarnessPlugin } from "@pleach/core"
import type { StreamObserverRegistration } from "@pleach/core/plugins/stream"

const latencyProbe: StreamObserverRegistration = {
  when: { callClass: "synthesize", stageId: "synthesize" },
  factory: (ctx) => {
    let firstChunkAt: number | undefined
    return {
      onChunk: (chunk) => {
        if (firstChunkAt === undefined) {
          firstChunkAt = performance.now()
          recordTtft(ctx.nodeId, firstChunkAt)
        }
        return { kind: "continue" }
      },
    }
  },
}

export const myPlugin: HarnessPlugin = {
  name: "ttft-probe",
  version: "0.1.0",
  contributeStreamObservers: () => [latencyProbe],
}

Three rules hold for every stream observer:

• onChunk is synchronous. Returning a Promise is rejected for replay-determinism. The observer can hand work off to a background queue, but it can't await.
• The when filter is AND-composed. Every set field must match. callClass: "*" matches any class; omitting stageId matches any stage.
• The factory must not throw at construction. PluginManager pre-screens factories via smokeValidateFactories at register time; a throwing factory surfaces as PluginValidationError before any session uses it.

contributeFinalizationPasses — output sanitization

A SanitizerPass runs on the rendered synthesizer output before it lands in the audit row. The hook for redaction passes, citation post-processing, output-shape normalization. Each pass declares an id and an order (lower runs first; ties resolve by registration order); the apply(content, ctx) method returns the rewritten string.

import type { HarnessPlugin } from "@pleach/core"
import type {
  SanitizerPass,
  FinalizationPassContext,
} from "@pleach/core/plugins"

const redactInternalIds: SanitizerPass = {
  id: "redact-internal-ids",
  order: 100,
  apply(content: string, ctx: FinalizationPassContext): string {
    return content.replace(/INTERNAL-\d{6}/g, "[REDACTED]")
  },
}

export const redactionPlugin: HarnessPlugin = {
  name: "internal-id-redactor",
  version: "0.1.0",
  contributeFinalizationPasses: () => [redactInternalIds],
}

Finalization passes run after the FabricationDetector hooks, so a detector that flagged a suspect span has already routed before sanitization runs. See Fabrication detection for the pre-finalization guard surface and Plugin contract for the full hook set.

What not to do

A few patterns that fight the substrate:

• Don't log raw AuditableCall payloads without wiring PIIRedactor. The payload carries user input; raw logs are a PII risk.
• Don't block the turn on telemetry writes. The ledger contract is fire-and-forget; a telemetry decorator that throws fails the call, which fails the turn.
• Don't add wall-clock timestamps to spans inside the chain. The substrate's IDs (ULIDs) carry timing; spans get their own timing from the tracer. Two clocks disagree about subtle things.
• Don't roll your own per-call instrumentation in plugin observer hooks. Observers are sync-only by contract; making them telemetry sinks risks breaking the sync property.

Where to go next