Observability stack

withDevtools for dev + withTelemetry for OTel + a custom CruxPlugin for app-specific metrics.

This recipe layers Crux's three observability mechanisms into one stack: devtools for development, OpenTelemetry spans for production, and a custom CruxPlugin that consumes the canonical graph-record stream for app-specific metrics (cost per tenant, latency by region).

Primitives used

Local HTTP observability transport via config({ observability }) (or withDevtools())
withTelemetry() from @use-crux/otel
A custom CruxPlugin that installs via subscribeObservability() and cleans up in dispose
config({ plugins: [...] }) to compose them

When to reach for this pattern

You're going to production and need real observability, not just dev devtools
You already have an OTel-compatible APM (Datadog, Honeycomb, Grafana, New Relic)
You want app-specific metrics that aren't standard OTel attributes: tenant ID, feature flag bucket, cost per request

Full code

`lib/ai/plugins/cost-tracker.ts`

import type { CruxPlugin } from "@use-crux/core";
import { subscribeObservability } from "@use-crux/core/observability";

interface CostTrackerOptions {
  onCost: (event: {
    promptId: string | undefined;
    modelId: string | undefined;
    tenantId: string | undefined;
    inputTokens: number;
    outputTokens: number;
    estimatedCost: number;
    durationMs: number | undefined;
  }) => void | Promise<void>;
}

export function withCostTracker(options: CostTrackerOptions): CruxPlugin {
  return {
    name: "app:cost-tracker",
    install() {
      // Graph records are the only public instrumentation stream.
      // `span:start` carries primitive/identity; `span:end` carries metrics.
      // Join them by spanId rather than inventing per-event callback fields.
      const openGenerations = new Map<
        string,
        { promptId?: string; modelId?: string; tenantId?: string }
      >();

      const unsubscribe = subscribeObservability((record) => {
        if (record.type === "span:start") {
          if (
            record.primitive !== "generation.call" &&
            record.primitive !== "generation.stream"
          ) {
            return;
          }
          const attributes = record.attributes ?? {};
          openGenerations.set(record.spanId, {
            // Generation openSpan puts identity on attributes (and may also
            // promote promptId/model onto top-level span fields when present).
            promptId:
              typeof record.promptId === "string"
                ? record.promptId
                : typeof attributes.promptId === "string"
                  ? attributes.promptId
                  : undefined,
            modelId:
              typeof record.model === "string"
                ? record.model
                : typeof attributes.model === "string"
                  ? attributes.model
                  : undefined,
            // Correlator metadata is projected as meta.<key> attributes.
            tenantId:
              typeof attributes["meta.tenantId"] === "string"
                ? attributes["meta.tenantId"]
                : undefined,
          });
          return;
        }

        if (record.type !== "span:end") return;
        const start = openGenerations.get(record.spanId);
        if (!start) return;
        openGenerations.delete(record.spanId);

        const metrics = record.metrics ?? {};
        void options.onCost({
          promptId: start.promptId,
          modelId: start.modelId,
          tenantId: start.tenantId,
          inputTokens:
            typeof metrics.inputTokens === "number" ? metrics.inputTokens : 0,
          outputTokens:
            typeof metrics.outputTokens === "number" ? metrics.outputTokens : 0,
          estimatedCost:
            typeof metrics.costUsd === "number" ? metrics.costUsd : 0,
          durationMs:
            typeof metrics["gen.duration_ms"] === "number"
              ? metrics["gen.duration_ms"]
              : record.durationMs,
        });
      });

      return {
        dispose: unsubscribe,
      };
    },
  };
}

`crux.config.ts`

import { config } from "@use-crux/core";
import { withTelemetry } from "@use-crux/otel";

import { withCostTracker } from "./lib/ai/plugins/cost-tracker";

export default config({
  observability: {
    serverUrl: process.env.DEVTOOLS_URL,
    token: process.env.CRUX_DEVTOOLS_TOKEN,
  },
  plugins: [
    // OTel spans for production
    withTelemetry({
      serviceName: "my-app",
      attributes: {
        "deployment.environment": process.env.NODE_ENV ?? "development",
        ...(process.env.RELEASE_SHA
          ? { "service.version": process.env.RELEASE_SHA }
          : {}),
      },
    }),

    // Custom cost metric to your monitoring backend
    withCostTracker({
      onCost: async (event) => {
        await datadogClient.gauge("llm.cost", event.estimatedCost, {
          tags: [
            `prompt:${event.promptId ?? "unknown"}`,
            `model:${event.modelId ?? "unknown"}`,
            `tenant:${event.tenantId ?? "unknown"}`,
          ],
        });
        if (event.durationMs != null) {
          await datadogClient.histogram("llm.duration_ms", event.durationMs, {
            tags: [`prompt:${event.promptId ?? "unknown"}`],
          });
        }
      },
    }),
  ],
});

Pass tenant context per request

import { createSessionId } from "@use-crux/core";
import { generate } from "@use-crux/ai";
import { openai } from "@ai-sdk/openai";
import { propagateAttributes } from "@use-crux/core/observability";
import { summarize } from "./prompts";

export async function summarizeForTenant(tenantId: string, text: string) {
  return propagateAttributes(
    {
      sessionId: createSessionId(),
      metadata: { tenantId },
    },
    () =>
      generate(summarize, {
        model: openai("gpt-4o"),
        input: { text },
      }),
  );
}

propagateAttributes stamps sessionId / userId onto every record and projects metadata as meta.<key> attributes (for example meta.tenantId). It does not add graph nodes, only correlators.

How it works

Devtools transport + withTelemetry() + custom plugin compose. Each plugin installs independently. OTel and your cost tracker both observe the same graph-record spine rather than separate per-event callback maps.
Devtools is explicit local/tunnel visibility. When observability.serverUrl is set, Crux installs the HTTP observability transport. Leave it unset outside deliberate local or staging tunnel sessions.
withTelemetry emits OTel spans. Every generate, memory op, compaction, judge, flow step, and tool call becomes a span with crux.* / GenAI attributes. Your existing OTel collector ingests them.
Your custom plugin subscribes to graph records. subscribeObservability(['span:end'], …) receives completed generation spans with metrics (tokens, cost, duration) already normalized.
Correlators group multi-tenant work. Use propagateAttributes({ sessionId, userId, metadata }, fn) so tags stay attached across nested spans without inventing plugin-specific event fields.

Variations

Devtools via tunnel

For staging environments or controlled debugging sessions, run crux dev --tunnel on your local machine and point DEVTOOLS_URL at the public tunnel origin. Set CRUX_DEVTOOLS_TOKEN to the scoped ingest token printed by crux dev. Treat this as an explicit temporary visibility channel, not default production telemetry.

Sampling

If you have high volume and don't want to push every event:

withCostTracker({
  onCost: async (event) => {
    if (Math.random() > 0.05) return; // 5% sample
    await datadogClient.gauge(/* ... */);
  },
});

Observability stack

Primitives used

When to reach for this pattern

Full code

`lib/ai/plugins/cost-tracker.ts`

`crux.config.ts`

Pass tenant context per request

How it works

Variations

Devtools via tunnel

Sampling

Trace context propagation across services

Where to next

Observability guide

Plugins guide

Telemetry guide

Runtime setup

On this page