Observability

OverviewCopy Anchor Link

Cogitator ships two observability exporters in @cogitator-ai/core: Langfuse for AI-native tracing with cost tracking, and OpenTelemetry (OTLP) for standard distributed tracing. Neither is attached automatically — you connect them to the run callbacks (onRunStart, onToolCall, onToolResult, onSpan, onRunComplete, onRunError) that cog.run() accepts.

Both exporters are off unless you pass enabled: true; a disabled exporter accepts every call and does nothing, which makes it easy to toggle per environment.

SpansCopy Anchor Link

Every run produces spans, returned in result.trace.spans and streamed one by one to onSpan as they finish:

The root agent.run span is emitted last (when the run ends); a failed run emits it with status error and an error attribute.

Langfuse IntegrationCopy Anchor Link

Langfuse provides purpose-built observability for LLM applications — traces, generations, cost analysis, and evaluation scores. The langfuse SDK is an optional peer dependency:

pnpm add langfuse

import { createLangfuseExporter } from '@cogitator-ai/core';

const langfuse = createLangfuseExporter({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  baseUrl: 'https://cloud.langfuse.com',
  flushAt: 10,
  flushInterval: 5000,
  enabled: true,
});

await langfuse.init(); // throws when `langfuse` is not installed

Wiring a RunCopy Anchor Link

Each run becomes a Langfuse trace (id = run id, session = thread id, tagged cogitator and the agent name); tool calls become tool:<name> spans on it:

let currentRunId = '';

const result = await cog.run(agent, {
  input: 'Summarize recent papers on chain-of-thought reasoning',
  onRunStart: ({ runId, agentId, input, threadId }) => {
    currentRunId = runId;
    langfuse.onRunStart({
      runId,
      agentId,
      agentName: agent.name,
      input,
      threadId,
      model: agent.model,
    });
  },
  onToolCall: (call) => langfuse.onToolCall(currentRunId, call),
  onToolResult: (toolResult) => langfuse.onToolResult(currentRunId, toolResult),
  onRunComplete: (runResult) => langfuse.onRunComplete(runResult),
});

onToolCall and onToolResult do not carry the run id, so keep the one from onRunStart in a variable scoped to the run. onRunComplete writes the output, token usage and tool-call count to the trace and releases its state.

Generations and Custom SpansCopy Anchor Link

For LLM calls you make yourself (or to record finished onSpan spans), use the lower-level hooks:

const generationId = langfuse.onLLMCall({
  runId,
  model: 'openai/gpt-5.5',
  messages: [{ role: 'user', content: 'Summarize recent papers...' }],
  temperature: 0.7,
  maxTokens: 4096,
});

langfuse.onLLMResponse({
  generationId,
  output: 'Here is a summary...',
  inputTokens: 150,
  outputTokens: 820,
});

const span = {
  id: 'span_rerank',
  traceId: 'trace_abc123',
  name: 'rerank',
  kind: 'internal' as const,
  startTime: Date.now(),
  attributes: { candidates: 20 },
};
langfuse.onSpanStart(runId, span);
langfuse.onSpanEnd({
  ...span,
  status: 'ok',
  endTime: Date.now(),
  duration: 0,
  attributes: { ...span.attributes, output: ['doc-3', 'doc-7'] },
}); // the span's output is `attributes.output`

onLLMCall and onSpanStart take an optional parent (spanId / span.parentId) and fall back to the trace. Hooks for a run that was never started return quietly (onLLMCall returns '').

OpenTelemetry (OTLP)Copy Anchor Link

For teams using Jaeger, Grafana Tempo, Honeycomb or Datadog, the OTLP exporter sends spans as OTLP/HTTP JSON.

import { createOTLPExporter } from '@cogitator-ai/core';

const otlp = createOTLPExporter({
  endpoint: 'http://localhost:4318/v1/traces',
  serviceName: 'cogitator-production', // default 'cogitator'
  serviceVersion: '1.0.0', // default '1.0.0'
  headers: { Authorization: `Bearer ${process.env.OTLP_TOKEN}` },
  enabled: true,
});

otlp.start(); // flush every 5 seconds

To export every span of a run as it finishes:

let runId = '';
await cog.run(agent, {
  input: 'Hello',
  onRunStart: (data) => {
    runId = data.runId;
  },
  onSpan: (span) => otlp.exportSpan(runId, span),
});

You can also export spans after the fact — result.trace.spans.forEach((span) => otlp.exportSpan(result.runId, span)) — or build your own Span objects:

otlp.exportSpan('run_abc123', {
  id: 'span_001',
  traceId: 'trace_abc123',
  name: 'agent.run',
  kind: 'server',
  status: 'ok',
  startTime: Date.now() - 1500,
  endTime: Date.now(),
  duration: 1500,
  attributes: { 'agent.name': 'researcher', 'llm.model': 'openai/gpt-5.5' },
});

await otlp.flush();

Spans are batched and flushed every 5 seconds after start(), or as soon as 100 spans are pending.

OTLP needs hex ids (16 bytes for a trace, 8 for a span), while Cogitator ids look like trace_… and span_…. The exporter derives the OTLP ids from them with SHA-256 (ids that already are hex of the right length are kept), so the spans of a run share a trace and child spans point at their parents, also when they are exported after the run. The original ids and the run id are kept as the cogitator.trace_id, cogitator.span_id and cogitator.run_id attributes. Numbers, booleans and strings keep their types as OTLP attribute values; objects are sent as JSON strings.

A batch the collector refuses with a 4xx status is dropped and logged; on network errors, 408, 429 and 5xx it is kept and sent again with the next flush, up to 10,000 pending spans.

OTLPExporter also has onRunStart, onRunComplete and onRunError hooks. They do nothing — ids come from the spans themselves — and exist so you can wire both exporters to the same callbacks.

Span KindsCopy Anchor Link

Span status maps to OTLP OK, ERROR or UNSET.

Cost TrackingCopy Anchor Link

Every run reports its cost in result.usage.cost (USD, from the model registry prices) alongside inputTokens, outputTokens, cachedInputTokens and reasoningTokens. The llm.chat spans carry per-call token counts, so any OTLP backend can aggregate them by llm.model.

Langfuse computes cost from the token usage of generations; with onRunComplete the run's usage (including cost) is attached to the trace metadata. Group costs by agent (trace name), model, or session (thread) in the Langfuse dashboard.

ShutdownCopy Anchor Link

Flush exporters before process exit to avoid losing buffered data:

process.on('SIGTERM', async () => {
  await langfuse.flush();
  await langfuse.shutdown();
  otlp.stop();
  await otlp.flush();
});