@ai-craft/agent-observer

Provider-agnostic agent observability harness. Zero external runtime dependencies.

Wraps any agent call and emits structured AgentRunEvent objects to pluggable sinks — with concrete adapters for Anthropic, OpenAI, Gemini, Snowflake Cortex, and agent-loop.

Installation

pnpm add @ai-craft/agent-observer

Quick start

import { EventBus, RunSession, ConsoleSink, TokentrackerSink, mapAnthropicEvent } from '@ai-craft/agent-observer';

const bus = new EventBus();
bus.register(new ConsoleSink());
bus.register(new TokentrackerSink()); // → ~/.ai-craft/usage.jsonl

const session = new RunSession(bus, 'anthropic', 'claude-haiku-4-5-20251001', {
  sessionId: 'my-conversation-id', // optional — groups runs in the ledger
});

// Pipe any Anthropic streaming response through mapAnthropicEvent
let inputTokens;
for await (const raw of stream) {
  const event = mapAnthropicEvent(raw, { run_id: session.run_id, turn: 1, provider: 'anthropic', model: 'claude-haiku-4-5-20251001', input_tokens: inputTokens });
  if (!event) continue;
  if (event.event_type === 'run_start') inputTokens = event.usage?.input_tokens;
  bus.publish(event);
}

Core concepts

EventBus

The central hub. All adapters publish to it; all sinks consume from it.

// Standard — process all events
const bus = new EventBus();

// High-traffic sampling — only observe 10% of runs
const bus = new EventBus({ sampleRate: 0.1 });

RunSession

Tracks a single agent run. Auto-generates run_id, tracks TTFT, and propagates session_id.

const session = new RunSession(bus, provider, model, { sessionId, parentRunId });

session.emit({ event_type: 'run_start' });
session.emit({ event_type: 'text_delta', text: 'Hello' });
session.complete(latency_ms, usage);  // emits run_complete with ttft_ms injected automatically
session.error(err);                   // emits event_type: error
session.retry();                      // re-emits run_start, incrementing retry_count

AgentRunEvent schema

interface AgentRunEvent {
  event_type: 'run_start' | 'text_delta' | 'thinking_delta' | 'status'
             | 'tool_use' | 'tool_result' | 'run_complete' | 'error';
  run_id: string;
  session_id?: string;        // groups runs by conversation
  parent_run_id?: string;     // links child run to parent in multi-agent setups
  turn: number;
  ts: string;                 // ISO 8601 UTC
  provider: string;
  model: string | null;
  usage?: {
    input_tokens: number | null;
    output_tokens: number | null;
    total_tokens: number | null;
    token_source: 'api' | 'estimated' | 'unavailable';
    cache_write_tokens?: number | null;  // Anthropic prompt cache write tokens
    cache_read_tokens?: number | null;   // Anthropic prompt cache read tokens
  };
  latency_ms?: number;        // set on run_complete
  ttft_ms?: number;           // time-to-first-token, set on run_complete when text_delta fired
  cost_usd?: number;          // computed from PRICE_MAP; undefined for unknown models
  text?: string;
  tool_calls?: ToolCallRecord[];
  tool_result_content?: unknown;  // payload of tool_result events
  status_message?: string;
  error?: { code?: string; message: string };
  retry_count?: number;       // retry attempt counter; present on run_start when > 0
}

Adapters

| Adapter | Function | Source | |---------|----------|--------| | Anthropic | mapAnthropicEvent(raw, ctx) | Anthropic Messages streaming SSE | | OpenAI | mapOpenAIChunk(raw, ctx) | OpenAI Chat Completions streaming chunks | | Gemini / Vertex | mapGeminiChunk(raw, ctx) | Gemini generateContentStream chunks | | Snowflake Cortex | mapCortexEvent(raw, ctx) | Cortex :run SSE events | | agent-loop | mapLoopEvent(event, provider?) | @ai-craft/agent-loop LoopEvents (event.type or event.kind) |

Sinks

TokentrackerSink

Appends one JSONL line per run_complete to ~/.ai-craft/usage.jsonl.

{
  "ts": "2026-07-10T10:00:00.000Z",
  "run_id": "run-1720605600-abc123",
  "session_id": "my-conversation",     // present when sessionId was set
  "provider": "anthropic",
  "model": "claude-haiku-4-5-20251001",
  "input_tokens": 19,
  "output_tokens": 8,
  "token_source": "api",
  "latency_ms": 834,
  "ttft_ms": 312,                      // present when text_delta fired
  "cost_usd": 0.0000474                // present for known models
}

Override the ledger path:

new TokentrackerSink('/my/custom/path.jsonl')
// or via env:
AICRAFT_USAGE_LOG=/my/custom/path.jsonl

token_source can be 'estimated' for providers with no API-reported usage (e.g. Snowflake Cortex, whose SSE stream never returns token counts). In that case input_tokens/output_tokens are computed via estimateTokenCount() from the request_sent event's outbound messages and the accumulated response text — a directional ~4-characters-per-token heuristic, not a real tokenizer. cost_usd is only populated for estimated runs when a per-1K-token rate is configured, via the optional AGENT_OBSERVER_COST_PER_1K_TOKENS env var or TokentrackerSink's 2nd constructor arg (new TokentrackerSink(path, ratePerThousandTokens)); otherwise cost_usd is omitted. The DebuggerSink dashboard labels estimated figures (est.) to distinguish them from real API-reported usage.

ConsoleSink

Prints every event to stdout with icons and compact formatting.

HttpSink

POSTs every event as JSON to a webhook URL.

bus.register(new HttpSink('https://my-telemetry-endpoint.example.com/events'));

AlertSink

Fires a callback and/or webhook when latency or token thresholds are exceeded.

bus.register(new AlertSink({
  latencyThresholdMs: 5000,
  tokenThreshold: 10_000,
  onAlert: (event, reason) => console.error('[ALERT]', reason),
  webhook: 'https://hooks.slack.com/...',  // optional — fire-and-forget POST
}));

Cost model

import { computeCost, PRICE_MAP, CORTEX_PRICE_MAP, registerPrices } from '@ai-craft/agent-observer';

// Direct USD pricing (Anthropic, OpenAI, Google)
const cost = computeCost('anthropic', 'claude-sonnet-4-6', inputTokens, outputTokens);
// → number (USD) | null (unknown model or missing token counts)

// Snowflake Cortex credit-based pricing
// cost = (tokens / 1M) × creditsPerMillion × creditCostUsd
const cortexCost = computeCost('cortex', 'claude-3-5-sonnet', inputTokens, outputTokens, {
  creditCostUsd: 3.0,  // USD per Snowflake credit
});
// or set AGENT_OBSERVER_SNOWFLAKE_CREDIT_COST_USD=3.0 as a process env fallback

// Register prices for new or private models at runtime
registerPrices({
  'cortex/my-private-model': { inputCredits: 5, outputCredits: 5 },
  'myprovider/gpt-custom':   { input: 2 / 1e6, output: 8 / 1e6 },
});

Supported providers in PRICE_MAP: Anthropic Claude 3.x / 4.x, OpenAI GPT-4o / o1, Google Gemini 1.5 / 2.x.

CORTEX_PRICE_MAP covers Snowflake Cortex models: claude-3-5-sonnet, claude-3-haiku, mistral-large2, llama3-70b, llama3-8b, mixtral-8x7b, snowflake-arctic. Keys use the cortex/${model} format.

Usage CLI

View the JSONL ledger as a formatted table:

# From the dist dir after build, or via npx when published:
node dist/packages/agent-observer/bin/usage.mjs
node dist/packages/agent-observer/bin/usage.mjs --tail 10
node dist/packages/agent-observer/bin/usage.mjs --session my-conversation-id

xfuse / Snowflake Cortex integration

Set env vars before starting entity-service:

AGENT_OBSERVER_DIST=/path/to/dist/packages/agent-observer/index.js
AGENT_OBSERVER_SAMPLE_RATE=1  # 0.0–1.0; default 1 (observe all)

The withCortexObserver wrapper in ai-chat-observer.helper.ts handles the rest:

• Lazy singleton load of the dist
• Per-agent turn counter (increments per SendMessage call)
• Transparent yield* passthrough — stream is never affected by observer errors

Running tests

nx test agent-observer
nx run agent-observer:typecheck
nx run agent-observer:typecheck-spec
nx build agent-observer

121 unit + integration tests. Zero external runtime dependencies.

Examples

# Live Anthropic stream (needs ANTHROPIC_API_KEY)
node packages/agent-observer/examples/anthropic-stream.mjs

# agent-loop tap (needs ANTHROPIC_API_KEY + nx build agent-loop agent-llm)
node packages/agent-observer/examples/agent-loop-tap.mjs