---
title: "Monitoring Observability"
description: "Monitoring and observability patterns for Prometheus metrics, Grafana dashboards, Langfuse v4 LLM tracing (as_type, score_current_span, should_export_span, LangfuseMedia), and drift detection. Use when adding logging, metrics, distributed tracing, LLM cost tracking, or quality drift monitoring."
canonical: "https://orchestkit.yonyon.ai/docs/reference/skills/monitoring-observability"
---

# Monitoring Observability

Monitoring and observability patterns for Prometheus metrics, Grafana dashboards, Langfuse v4 LLM tracing (as_type, score_current_span, should_export_span, LangfuseMedia), and drift detection. Use when adding logging, metrics, distributed tracing, LLM cost tracking, or quality drift monitoring.

<span className="badge badge-gray">Reference</span> <span className="badge badge-yellow">medium</span>

> **Not directly invocable** — no slash command and no model auto-selection. An agent loads it explicitly via `Read()`.

<ContextualSkillSidebar slug="monitoring-observability" />

> **Monitoring Observability** Monitoring and observability patterns for Prometheus metrics, Grafana dashboards, Langfuse v4 LLM tracing (as_type, score_current_span, should_export_span, LangfuseMedia), and drift detection. Use when adding logging, metrics, distributed tracing, LLM cost tracking, or quality drift monitoring.


# Monitoring & Observability

A wrap around Prometheus, Grafana, OpenTelemetry and Langfuse, not a re-teaching of them. This
skill carries OrchestKit's delta (version floors, house decisions, scars) and points at the
vendor for everything else. Start at `references/ork-delta.md`.

## Upstream coverage (do not restate)

These topics are fully covered first-party. Read the source, do not add a local copy.

| Topic | First-party source |
|-------|--------------------|
| Prometheus metric types, RED method, cardinality, PromQL | &lt;https://prometheus.io/docs/practices/&gt; |
| Alertmanager grouping, inhibition, escalation, runbooks | &lt;https://prometheus.io/docs/alerting/latest/configuration/&gt; |
| Grafana dashboards, Loki and LogQL, Promtail | &lt;https://grafana.com/docs/&gt; |
| OpenTelemetry spans, sampling, context propagation | &lt;https://opentelemetry.io/docs/&gt; |
| Langfuse Python SDK (`@observe`, `as_type`, `score_current_span`, `should_export_span`, `LangfuseMedia`) | &lt;https://langfuse.com/docs/sdk/python&gt; |
| Langfuse v2 to v4 Python and v3 to v5 JS migration paths | &lt;https://langfuse.com/docs/sdk/python/v4-migration&gt; |
| Langfuse self-hosting (ClickHouse, Redis, S3, Helm) | &lt;https://langfuse.com/docs/deployment/self-host&gt; |
| Langfuse cost tracking, model pricing, Metrics API v2 | &lt;https://langfuse.com/docs/model-usage-and-cost&gt; |
| Langfuse scores, online evaluators, annotation queues, prompt management | &lt;https://langfuse.com/docs/scores/overview&gt; |
| Langfuse framework integrations (LangChain, LangGraph, CrewAI, Pydantic AI, Bedrock, LiveKit) | &lt;https://langfuse.com/docs/integrations&gt; |
| Agent Graphs, observation types, rendered tool calls | &lt;https://langfuse.com/docs/tracing-features/agent-graphs&gt; |
| PSI, KS test, KL and JS divergence, Wasserstein, embedding drift | &lt;https://www.evidentlyai.com/blog/data-drift-detection-large-datasets&gt; |
| EWMA control charts | &lt;https://www.itl.nist.gov/div898/handbook/pmc/section3/pmc324.htm&gt; |
| structlog, Winston, correlation IDs, log sampling | &lt;https://www.structlog.org/en/stable/&gt; |

## Quick Reference

| Category | Rules | Impact | When to Use |
|----------|-------|--------|-------------|
| [Infrastructure Monitoring](#infrastructure-monitoring) | 1 | CRITICAL | Grafana dashboards, Golden Signals, SLO/SLI |
| [LLM Observability](#llm-observability) | 1 | HIGH | Langfuse tracing, observation types, agent graphs |
| [Silent Failures](#silent-failures) | 3 | HIGH | Tool skipping, quality degradation, loop/token spike alerting |

**Total: 5 rules across 3 categories.** Drift detection, cost tracking, eval scoring, Prometheus
instrumentation and alert-rule authoring moved to the upstream sources listed above.

## Quick Start

```python
# Langfuse v4 LLM tracing: semantic as_type plus inline scoring
from langfuse import observe, get_client

@observe(as_type="generation", name="analyze_content")
async def analyze_content(content: str):
    get_client().update_current_trace(
        user_id="user_123", session_id="session_abc",
        tags=["production", "orchestkit"],
    )
    result = await llm.generate(content)
    get_client().score_current_span(name="response_quality", value=0.85)
    return result
```

```python
# Prometheus RED method, wired the way this repo expects (bounded labels only)
from prometheus_client import Counter, Histogram

http_requests = Counter('http_requests_total', 'Total requests', ['method', 'endpoint', 'status'])
http_duration = Histogram('http_request_duration_seconds', 'Request latency',
    buckets=[0.01, 0.05, 0.1, 0.5, 1, 2, 5])
```

## Infrastructure Monitoring

Dashboard and health-check patterns. Metric instrumentation and alert-rule syntax are upstream.

| Rule | File | Key Pattern |
|------|------|-------------|
| Grafana Dashboards | `rules/monitoring-grafana.md` | Golden Signals, SLO/SLI, health checks |

> **CC 2.1.161 — OTEL resource attributes as metric labels:** `OTEL_RESOURCE_ATTRIBUTES` values are now attached as labels on metric datapoints, so usage metrics can be sliced by custom dimensions (team, repo, environment). Add label selectors to dashboards for multi-tenant / per-team cost and usage tracking.

## LLM Observability

Langfuse-based tracing for LLM applications. Cost tracking, scoring and drift statistics are
upstream; what stays here is how this repo wires traces.

| Rule | File | Key Pattern |
|------|------|-------------|
| Langfuse Traces | `rules/llm-langfuse-traces.md` | @observe decorator, OTEL spans, agent graphs |

## Silent Failures

Detection and alerting for silent failures in LLM agents.

| Rule | File | Key Pattern |
|------|------|-------------|
| Tool Skipping | `rules/silent-tool-skipping.md` | Expected vs actual tool calls, Langfuse traces |
| Quality Degradation | `rules/silent-degraded-quality.md` | Heuristics + LLM-as-judge, z-score baselines |
| Silent Alerting | `rules/silent-alerting.md` | Loop detection, token spikes, escalation workflow |

> **CC 2.1.169 — OTEL client-cert paths require trust:** untrusted project settings can no longer set OTEL client-certificate paths without a trust confirmation. If your OTEL exporter uses client certs configured in project `.claude/settings.json`, expect a one-time trust prompt on first use in an untrusted project — telemetry silently not flowing after 2.1.169 is usually this gate, not the collector.

## Key Decisions

| Decision | Recommendation | Rationale |
|----------|----------------|-----------|
| Metric methodology | RED method (Rate, Errors, Duration) | Industry standard, covers essential service health |
| Log format | Structured JSON | Machine-parseable, supports log aggregation |
| Tracing | OpenTelemetry | Vendor-neutral, auto-instrumentation, broad ecosystem |
| LLM observability | Langfuse (not LangSmith) | Open-source, self-hosted, built-in prompt management |
| LLM tracing API | `@observe(as_type=...)` + `score_current_span()` | v4: semantic types, inline scoring, span filtering |
| Langfuse APIs | Observations API v2 + Metrics API v2 | v4 (Mar 2026): faster querying, aggregations at scale |
| Hook telemetry transport | JSONL under `~/.claude/analytics/`, never an SDK in-process | Hooks are per-event processes; SDK init would be paid on every spawn (`references/ork-delta.md`) |

## Detailed Documentation

| Resource | Description |
|----------|-------------|
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/ork-delta.md` | **Start here.** Floors, house decisions and scars that upstream docs do not carry |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/langfuse-js-v5.md` | **JS/TS SDK v5** delta from Python 4.x: package map, phantom packages, SpanProcessor vs exporter. Read before writing any JS Langfuse code |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/experiments-api.md` | Langfuse experiments and dataset runs as this repo uses them |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/evaluation-scores.md` | Score shapes and scoring pipeline wiring |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/session-tracking.md` | Session and user grouping across multi-step workflows |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/metrics-collection.md` | Claude Code OTEL metric inventory and collector-side joins |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/dashboards.md` | Dashboard layout conventions |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/structured-logging.md` | Structured log field conventions |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/references/dev-agent-lens.md` | LiteLLM proxy layer for API-boundary observability |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/examples/orchestkit-monitoring-dashboard.md` | Worked monitoring dashboard example |
| `$\{CLAUDE_PLUGIN_ROOT\}/skills/monitoring-observability/scripts` | Templates: Prometheus, OpenTelemetry, health checks, Langfuse |

## Related Skills

- `defense-in-depth` - Layer 8 observability as part of security architecture
- `devops-deployment` - Observability integration with CI/CD and Kubernetes
- `resilience-patterns` - Monitoring circuit breakers and failure scenarios
- `llm-evaluation` - Evaluation patterns that integrate with Langfuse scoring
- `caching` - Caching strategies that reduce costs tracked by Langfuse


---

## Rules (5)

### Trace LLM call chains with Langfuse for debugging slow or incorrect responses — HIGH


# Langfuse Traces

## Basic Tracing with @observe (v4)

```python
from langfuse import observe, get_client

@observe(as_type="chain")  # Auto-creates trace on first root span
async def analyze_content(content: str):
    get_client().update_current_observation(
        metadata={"content_length": len(content)}
    )
    return await llm.generate(content)
```

## Nested Spans

```python
from langfuse import observe, get_client

@observe(name="content_analysis")
async def analyze(content: str):
    # Nested span for retrieval
    @observe(as_type="retriever", name="retrieval")
    async def retrieve_context():
        chunks = await vector_db.search(content)
        get_client().update_current_observation(
            metadata={"chunks_retrieved": len(chunks)}
        )
        return chunks

    # Nested span for generation
    @observe(as_type="generation", name="generation")
    async def generate_analysis(context):
        response = await llm.generate(content)
        get_client().update_current_observation(
            model="claude-sonnet-5",
            usage={"input_tokens": 1500, "output_tokens": 1000},
        )
        return response

    context = await retrieve_context()
    return await generate_analysis(context)
```

Result in Langfuse UI:
```
content_analysis (2.3s, $0.045)
+-- retrieval (0.1s)
|   +-- metadata: {chunks_retrieved: 5}
+-- generation (2.2s, $0.045)
    +-- model: claude-sonnet-5
    +-- tokens: 1500 input, 1000 output
```

## Session & User Tracking

```python
from langfuse import observe, get_client

@observe()
async def analysis(content: str):
    get_client().update_current_trace(
        user_id="user_123",
        session_id="session_abc",
        metadata={"content_type": "article", "agent_count": 8},
        tags=["production", "orchestkit"],
    )
    return await run_pipeline(content)
```

## Observation Types for Agent Graphs

Use `as_type=` (v4) to assign semantic span types for Agent Graph rendering:

```python
@observe(as_type="agent", name="supervisor")
async def supervisor(query: str): ...     # Agent node in graph

@observe(as_type="generation", name="llm_call")
async def generate(query: str): ...       # LLM generation step

@observe(as_type="retriever", name="vector_search")
async def retrieve(query: str): ...       # Retrieval step

@observe(as_type="chain", name="prompt_chain")
async def chain(inputs: dict): ...        # Sequential processing

@observe(as_type="guardrail", name="pii_check")
async def check_pii(text: str): ...       # Safety check

@observe(as_type="embedding", name="embed")
async def embed(text: str): ...           # Vector generation

@observe(as_type="evaluator", name="quality_judge")
async def evaluate(output: str): ...      # Inspectable evaluator trace
```

## Inline Span Scoring (v4)

Use `score_current_span()` to attach scores directly to the active span:

```python
from langfuse import observe, get_client

@observe(as_type="chain", name="rag_pipeline")
async def rag_pipeline(query: str):
    context = await retrieve(query)
    response = await generate(query, context)

    get_client().score_current_span(
        name="relevance", value=0.85,
        comment="Good retrieval alignment",
    )
    return response
```

## Filtering Noisy OTel Spans

When using OpenTelemetry auto-instrumentation, many infra spans (HTTP clients,
DB drivers, DNS) are exported to Langfuse. Use `should_export_span` to keep
only the spans you care about:

```python
from langfuse import Langfuse

def span_filter(span) -> bool:
    """Only export LLM and application spans, skip infra noise."""
    dominated_libs = {"urllib3", "httpcore", "dns", "ssl"}
    lib = span.attributes.get("otel.library.name", "")
    return lib not in dominated_libs

# v4: should_export_span is a Langfuse client kwarg
langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    should_export_span=span_filter,
)
```

## OpenTelemetry Integration

```python
# v4: the Langfuse client installs its own OTEL span processor internally,
# so there is no LangfuseSpanProcessor to add to a TracerProvider.
from langfuse import Langfuse
from langfuse.span_filter import is_default_export_span

langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com",
    should_export_span=is_default_export_span,
)
```

## JavaScript/TypeScript Setup

```typescript
import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

// v5 ships a SpanProcessor, NOT an exporter: it goes in `spanProcessors`,
// never `traceExporter`. Keys fall back to LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY.
const sdk = new NodeSDK({
  spanProcessors: [
    new LangfuseSpanProcessor({
      publicKey: process.env.LANGFUSE_PUBLIC_KEY,
      secretKey: process.env.LANGFUSE_SECRET_KEY,
    }),
  ],
});
sdk.start();

// Short-lived processes MUST flush before exit or trailing spans are lost.
process.on("beforeExit", async () => {
  await sdk.shutdown();
});
```

## Best Practices

1. **Use `from langfuse import observe, get_client`** — NOT `from langfuse.decorators`
2. **Let `@observe()` auto-create traces** — no explicit `langfuse.trace()` needed
3. **Name spans descriptively** (e.g., "retrieval", "generation")
4. **Use `as_type=` parameter** (v4) for Agent Graph rendering
5. **Add metadata** for debugging (chunk counts, model params)
6. **Truncate large inputs/outputs** to 500-1000 chars
7. **Tag production vs staging** traces for environment filtering

**Incorrect — flat trace without nested spans:**
```python
@observe()
async def analyze(content: str):
    chunks = await retrieve(content)  # Not traced
    result = await generate(chunks)   # Not traced
    return result  # No visibility into sub-operations
```

**Correct — nested spans for full visibility:**
```python
@observe(name="content_analysis")
async def analyze(content: str):
    @observe(name="retrieval")
    async def retrieve_context():
        return await vector_db.search(content)

    @observe(name="generation")
    async def generate_analysis(chunks):
        return await llm.generate(chunks)

    chunks = await retrieve_context()
    return await generate_analysis(chunks)
```


### Design Grafana dashboards for actionable incident response and capacity planning — CRITICAL


# Grafana Dashboards

## The Four Golden Signals

| Signal | Metric | Description |
|--------|--------|-------------|
| **Latency** | Response time | How long requests take |
| **Traffic** | Requests/sec | Volume of demand |
| **Errors** | Error rate | Failures per second |
| **Saturation** | Resource usage | How full the service is |

### Dashboard Layout (Top Row)

```
+--------------+--------------+--------------+--------------+
|  Latency     |  Traffic     |  Errors      |  Saturation  |
|  (p50/p95)   |  (req/s)     |  (5xx rate)  |  (CPU/mem)   |
+--------------+--------------+--------------+--------------+
```

## Service Dashboard Structure

1. **Overview** (single row) — Traffic, errors, latency, saturation
2. **Request breakdown** — By endpoint, method, status code
3. **Dependencies** — Database, Redis, external APIs
4. **Resources** — CPU, memory, disk, network
5. **Business metrics** — Registrations, purchases, LLM costs

## RED Metrics for Dashboards

```text
# Rate
rate(http_requests_total[5m])

# Errors
sum(rate(http_requests_total{status=~"5.."}[5m]))

# Duration
histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))
```

## USE Metrics for Resources

- **Utilization** — % of resource used
- **Saturation** — Queue depth, wait time
- **Errors** — Error count

## SLO/SLI Definitions

### Service Level Indicators (SLIs)

```text
# Availability SLI: % of successful requests
sum(rate(http_requests_total{status!~"5.."}[30d])) /
sum(rate(http_requests_total[30d]))

# Latency SLI: % of requests < 1s
sum(rate(http_request_duration_seconds_bucket{le="1"}[30d])) /
sum(rate(http_request_duration_seconds_count[30d]))
```

### Service Level Objectives (SLOs)

| SLO | Target | Error Budget |
|-----|--------|--------------|
| Availability | 99.9% | 43 min downtime/month |
| Latency | 99% &lt; 1s | 1% of requests can be slow |

**Error Budget:** If consumed, freeze feature work and focus on reliability.

## Health Checks (Kubernetes)

| Probe | Purpose | Endpoint |
|-------|---------|----------|
| **Liveness** | Is app running? | `/health` |
| **Readiness** | Ready for traffic? | `/ready` |
| **Startup** | Finished starting? | `/startup` |

## Dashboard Best Practices

1. **Use time ranges** — Last 1h, 6h, 24h, 7d
2. **Percentiles over averages** — p50, p95, p99
3. **Color code thresholds** — green/yellow/red
4. **Include annotations** — deployments, incidents
5. **Link to runbooks** — from alert panels

**Incorrect — using average latency hides tail latency:**
```text
avg(http_request_duration_seconds)  # Misleading for user experience
```

**Correct — using percentiles shows tail latency:**
```text
histogram_quantile(0.95, rate(http_request_duration_seconds_bucket[5m]))
histogram_quantile(0.99, rate(http_request_duration_seconds_bucket[5m]))
```


### Alert on silent failures using statistical baselines and proactive health monitoring — HIGH


## Silent Failure Alerting

Set up alerting for failures that produce no errors but deliver wrong results.

**Incorrect — alerting only on exceptions:**
```python
try:
    result = await agent.run()
except Exception as e:
    alert(e)  # Only catches crashes, not silent failures
# Agent returned gibberish — no exception raised, no alert sent
```

**Correct — statistical baseline anomaly detection:**
```python
import numpy as np

class BaselineAnomalyDetector:
    def __init__(self, window_size=100, z_threshold=3.0):
        self.window_size = window_size
        self.z_threshold = z_threshold
        self.history = []

    def add_observation(self, value: float) -> dict:
        self.history.append(value)
        if len(self.history) > self.window_size:
            self.history = self.history[-self.window_size:]
        if len(self.history) < 10:
            return {"alert": False, "reason": "insufficient_data"}
        mean = np.mean(self.history[:-1])
        std = np.std(self.history[:-1])
        if std == 0:
            return {"alert": False}
        z_score = abs(value - mean) / std
        if z_score > self.z_threshold:
            return {"alert": True, "type": "statistical_anomaly",
                    "z_score": z_score, "value": value, "mean": mean}
        return {"alert": False, "z_score": z_score}
```

**Silent failure type priorities:**

| Type | Detection Method | Priority |
|------|------------------|----------|
| Tool Skipping | Expected vs actual tool calls | Critical |
| Infinite Loop | Iteration count + token spike | Critical |
| Gibberish Output | LLM-as-judge + heuristics | High |
| Quality Degradation | Score &lt; baseline | Medium |
| Latency Spike | p99 > threshold | Medium |

**Key rules:**
- Alert on silent failures (service up, logic broken), not just errors
- Use z-score > 3.0 (99.7% confidence) for anomaly detection
- Maintain rolling baselines with 100-observation windows
- Detection priority: tool skipping > loops > gibberish > anomalies
- Need minimum 10 observations before baseline alerting is reliable
- Combine statistical detection with proactive quality checks


### Detect silent quality degradation in agent outputs that pass basic error checks — HIGH


## Silent Quality Degradation Detection

Detect gibberish, repetitive, or low-quality LLM outputs that pass basic checks.

**Incorrect — checking only for emptiness:**
```python
if len(response) > 0:  # Not-empty is not correct
    return response     # Gibberish passes this check
```

**Correct — heuristic pre-filter + LLM-as-judge:**
```python
from langfuse import observe, get_client

@observe(name="quality_check")
async def detect_degraded_quality(response: str) -> dict:
    # Quick heuristics first (cheap, fast)
    if len(response) < 10:
        return {"alert": True, "type": "too_short"}

    # Repetition check: ratio of unique words to total words
    words = response.split()
    if len(words) > 0 and len(set(words)) / len(words) < 0.3:
        return {"alert": True, "type": "repetitive"}

    # LLM-as-judge for semantic quality (more expensive, run second)
    judge_prompt = f"""Rate this response quality (0-1):
    - 0: Gibberish, nonsensical, or completely wrong
    - 0.5: Partially correct but missing key information
    - 1: High quality, accurate, complete
    Response: {response[:1000]}
    Score (just the number):"""

    score = await llm.generate(judge_prompt)
    score_value = float(score.strip())
    get_client().score_current_trace(name="quality_check", value=score_value)

    if score_value < 0.5:
        return {"alert": True, "type": "low_quality", "score": score_value}
    return {"alert": False, "score": score_value}
```

**Loop and token spike detection:**
```python
class LoopDetector:
    def __init__(self, max_iterations=10, token_spike_multiplier=3.0):
        self.max_iterations = max_iterations
        self.token_spike_multiplier = token_spike_multiplier
        self.iteration_count = 0
        self.total_tokens = 0
        self.baseline_tokens = 2000

    def check(self, tokens_used: int) -> dict:
        self.iteration_count += 1
        self.total_tokens += tokens_used
        if self.iteration_count > self.max_iterations:
            return {"alert": True, "type": "max_iterations"}
        expected = self.baseline_tokens * self.iteration_count
        if self.total_tokens > expected * self.token_spike_multiplier:
            return {"alert": True, "type": "token_spike",
                    "tokens": self.total_tokens, "expected": expected}
        return {"alert": False}
```

**Key rules:**
- Layer detection: heuristics first (cheap), then LLM-as-judge (accurate)
- Track unique-word ratio — below 0.3 indicates repetitive/stuck output
- Monitor token consumption per iteration — 3x baseline indicates infinite loop
- Log quality scores to Langfuse for trend analysis and drift detection
- Not-empty and no-error are insufficient quality checks


### Detect when agents silently skip expected tool calls and produce incorrect results — CRITICAL


## Silent Tool Skipping Detection

Detect when LLM agents skip expected tool calls without raising errors.

**Incorrect — assuming success if no error:**
```python
result = await agent.run()
# No error raised, but agent skipped the search tool entirely
# Result is fabricated from training data, not real data
return result  # Wrong answer delivered confidently
```

**Correct — validate tool usage against expectations:**
```python
from langfuse import Langfuse

def check_tool_usage(trace_id: str, expected_tools: list[str]) -> dict:
    langfuse = Langfuse()
    trace = langfuse.fetch_trace(trace_id)

    actual_tools = [
        span.name for span in trace.observations
        if span.type == "tool"
    ]

    missing_tools = set(expected_tools) - set(actual_tools)

    if missing_tools:
        return {
            "alert": True,
            "type": "tool_skipping",
            "missing": list(missing_tools),
            "message": f"Agent skipped expected tools: {missing_tools}"
        }
    return {"alert": False}

# Usage
expected_tools = ["search", "calculate"]
tool_check = check_tool_usage(trace_id, expected_tools)
if tool_check["alert"]:
    alert(tool_check)
    fallback_to_manual_execution()
```

**Key rules:**
- Never assume success just because no error was raised
- Define expected tool lists per agent task and validate after execution
- Tool skipping is often caused by middleware interference or prompt changes
- Alert on tool skipping with Critical priority — it produces wrong results silently
- Always have a fallback path when expected tools are not called



---

## References (9)

### Dashboards

# Monitoring Dashboards

Grafana dashboard patterns and SLO/SLI definitions.

## The Four Golden Signals

| Signal | Metric | Description |
|--------|--------|-------------|
| **Latency** | Response time | How long requests take |
| **Traffic** | Requests/sec | Volume of demand |
| **Errors** | Error rate | Failures per second |
| **Saturation** | Resource usage | How full the service is |

## SLO/SLI Examples

```yaml
# SLO: 99.9% availability
SLI: availability = successful_requests / total_requests
Target: > 0.999

# SLO: 95% of requests < 500ms
SLI: latency_p95 = histogram_quantile(0.95, request_duration_seconds)
Target: < 0.5

# SLO: < 0.1% error rate
SLI: error_rate = failed_requests / total_requests
Target: < 0.001
```

## Grafana Dashboard Structure

1. **Overview row** - traffic, errors, latency
2. **Saturation row** - CPU, memory, disk
3. **Details row** - per-endpoint breakdown
4. **Database row** - query performance, connections

## Best Practices

1. **Use time ranges** - Last 1h, 6h, 24h, 7d
2. **Percentiles over averages** - p50, p95, p99
3. **Color code thresholds** - green/yellow/red
4. **Include annotations** - deployments, incidents

See Grafana dashboards in `backend/grafana/dashboards/`.


### Dev Agent Lens

# dev-agent-lens Integration

LiteLLM-based proxy that intercepts Claude API calls for cost tracking, latency monitoring, and model routing visibility. Complements OrchestKit's hook-level JSONL analytics with API-level observability.

## When to Use dev-agent-lens vs Other Layers

| Layer | What It Sees | Latency Impact | Setup |
|-------|-------------|----------------|-------|
| **dev-agent-lens** (proxy) | API calls, token counts, model routing, costs | +5-15ms per call | Docker compose, env vars |
| **OrchestKit JSONL** (hooks) | Hook timing, agent spawns, skill usage, team activity | Zero (async writes) | Already active |
| **CC Native OTLP** (telemetry) | Tool-level spans (Read, Write, Bash, Task) | Zero (built-in) | 3 env vars |

**Use dev-agent-lens when you need**: per-request cost breakdown, model version tracking, API error rates, prompt/completion token ratios, latency percentiles at the API boundary.

**Don't use dev-agent-lens when**: you only need hook/skill/agent-level data (use JSONL), or tool-level spans (use CC OTLP).

## Architecture

```
Claude Code CLI
  │
  ├─ ANTHROPIC_BASE_URL=http://localhost:4000 ──→ dev-agent-lens (LiteLLM proxy :4000)
  │                                                  │
  │                                                  ├─→ Anthropic API (actual model calls)
  │                                                  ├─→ Langfuse (traces, costs)
  │                                                  └─→ Prometheus (:9090, optional)
  │
  ├─ OTEL_EXPORTER_OTLP_ENDPOINT ──→ Langfuse OTEL (:3100/api/public/otel)
  │                                    (tool-level spans from CC native telemetry)
  │
  └─ ~/.claude/analytics/*.jsonl ──→ JSONL bridge script (optional)
                                      (hook timing, agent routing, skill usage)
```

## API Key Caveat

**Claude Code Free/Pro users**: Cannot use `ANTHROPIC_BASE_URL` — the CLI sends requests directly to Anthropic's API using your subscription. The proxy approach only works with **API key access** (pay-per-token via `ANTHROPIC_API_KEY`).

**Claude Code Max users**: Same limitation — Max plans route through Anthropic's managed infrastructure, not a configurable base URL.

This means dev-agent-lens is primarily useful for:
- Self-hosted/enterprise deployments using API keys
- Development environments where you control the API routing
- CI/CD pipelines calling Claude via API

## Docker Compose Template

```yaml
# Add to your project's docker-compose.yml
# Profile: observability (docker compose --profile observability up)

services:
  dev-agent-lens:
    image: ghcr.io/berriai/litellm:main-latest
    profiles: [observability]
    ports:
      - "4000:4000"
    environment:
      LITELLM_MASTER_KEY: "sk-dev-local"
      LANGFUSE_PUBLIC_KEY: "${LANGFUSE_PUBLIC_KEY:-pk-lf-dev}"
      LANGFUSE_SECRET_KEY: "${LANGFUSE_SECRET_KEY:-sk-lf-dev}"
      LANGFUSE_HOST: "${LANGFUSE_HOST:-http://langfuse-web:3100}"
    volumes:
      - ./litellm-config.yaml:/app/config.yaml
    command: ["--config", "/app/config.yaml"]
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
```

### LiteLLM Config (`litellm-config.yaml`)

```yaml
model_list:
  - model_name: claude-sonnet-5
    litellm_params:
      model: anthropic/claude-sonnet-5
      api_key: os.environ/ANTHROPIC_API_KEY

  - model_name: claude-opus-5
    litellm_params:
      model: anthropic/claude-opus-5
      api_key: os.environ/ANTHROPIC_API_KEY

  - model_name: claude-haiku-4-5-20251001
    litellm_params:
      model: anthropic/claude-haiku-4-5-20251001
      api_key: os.environ/ANTHROPIC_API_KEY

general_settings:
  master_key: sk-dev-local

litellm_settings:
  success_callback: ["langfuse"]
  failure_callback: ["langfuse"]
  cache: false
  set_verbose: false
```

### Shell Configuration (API key users only)

```bash
# ~/.zshrc or ~/.bashrc — only for API key access
export ANTHROPIC_BASE_URL="http://localhost:4000"
export ANTHROPIC_API_KEY="sk-ant-..."  # your actual key
```

## What You See in Langfuse

With dev-agent-lens forwarding to Langfuse, each Claude API call creates a trace with:

- **Model**: Exact model ID (`claude-sonnet-5`)
- **Tokens**: Input/output/cache token counts
- **Cost**: Per-request USD cost (Anthropic pricing)
- **Latency**: Time-to-first-token, total duration
- **Status**: Success/failure, error codes, rate limits
- **Metadata**: Request headers, retry counts

This is the **API boundary** layer — it sees what crosses the network. It does NOT see:
- Which OrchestKit agent spawned the request (use JSONL for that)
- Which tool CC executed (use CC OTLP for that)
- Hook execution timing (use JSONL for that)

## Complementary Data: Proxy + Hooks + OTLP

The three layers together give full observability:

```
Question: "Why was this session slow?"

Layer 1 (CC OTLP):    Tool spans show 47 Read calls, 12 Bash calls
Layer 2 (JSONL):       Hook timing shows pre-push hook took 8.3s
Layer 3 (Proxy):       API calls show 3 rate-limited retries, p99 latency 4.2s

Answer: Rate limiting + excessive file reads + slow pre-push hook
```

## References

- [LiteLLM Proxy docs](https://docs.litellm.ai/docs/proxy/quick_start)
- [Langfuse LiteLLM integration](https://langfuse.com/docs/integrations/litellm)
- [CC OTLP telemetry](https://docs.anthropic.com/en/docs/claude-code/telemetry)


### Evaluation Scores

# LLM Evaluation & Scoring

Track quality metrics with custom scores, automated evaluation, and evaluator execution tracing.

## Basic Scoring (v3)

```python
from langfuse import observe, get_client, Langfuse

langfuse = Langfuse()

@observe()
async def analyze_and_score(query: str):
    """Run analysis and score the result."""
    response = await llm.generate(query)

    # Score via get_client() within @observe context
    get_client().update_current_observation(
        output=response[:500],
    )

    # Score the trace
    get_client().score_current_trace(
        name="relevance",
        value=0.85,
        comment="Response addresses query but lacks depth",
    )
    return response


# Or score by trace_id directly
langfuse.create_score(
    trace_id="trace_123",
    name="factuality",
    value=0.92,
    data_type="NUMERIC",
)
```

## Evaluator Execution Tracing

In v3, each evaluator run creates its own inspectable trace:

```python
from langfuse import observe, get_client

@observe(as_type="evaluator", name="relevance_judge")
async def evaluate_relevance(query: str, response: str):
    """Each evaluator call creates an inspectable trace in Langfuse."""
    score = await llm_judge.evaluate(
        criteria="relevance",
        query=query,
        response=response,
    )

    get_client().update_current_observation(
        input={"query": query[:500], "response": response[:500]},
        output={"score": score, "criteria": "relevance"},
        model="claude-sonnet-5",
    )

    # The evaluator's own LLM calls are visible in its trace
    return score
```

Result in Langfuse UI:
```
evaluator:relevance_judge (0.8s, $0.01)
├── generation: judge_prompt → score: 0.85
└── metadata: {criteria: "relevance", model: "claude-sonnet-5"}
```

## Score Analytics

View multi-score comparisons in the Langfuse dashboard:

- **Score distributions**: Histogram of scores by criterion
- **Multi-score comparison**: Side-by-side comparison of relevance, depth, accuracy
- **Quality trends**: Track scores over time
- **Filter by threshold**: Show only low-scoring traces
- **Compare prompts**: Which prompt version scores higher?

## Mutable Score Configs

Configure score types and ranges in Langfuse settings:

```python
# Score configs can be updated without code changes
# In Langfuse UI: Settings → Score Configs

# Numeric scores
langfuse.create_score(trace_id="...", name="relevance", value=0.85, data_type="NUMERIC")

# Categorical scores
langfuse.create_score(trace_id="...", name="sentiment", value="positive", data_type="CATEGORICAL")

# Boolean scores
langfuse.create_score(trace_id="...", name="contains_pii", value=0, data_type="BOOLEAN")
```

## Automated Scoring with G-Eval

```python
from langfuse import observe, get_client
from app.shared.services.g_eval import GEvalScorer

scorer = GEvalScorer()

@observe()
async def analyze_with_scoring(query: str):
    response = await llm.generate(query)

    # Run G-Eval scoring
    scores = await scorer.score(
        query=query,
        response=response,
        criteria=["relevance", "coherence", "depth"],
    )

    # Record all scores
    for criterion, score in scores.items():
        get_client().score_current_trace(name=criterion, value=score)

    return response
```

## Quality Scores Trend Query

```sql
SELECT
    DATE(timestamp) as date,
    AVG(value) FILTER (WHERE name = 'relevance') as avg_relevance,
    AVG(value) FILTER (WHERE name = 'depth') as avg_depth,
    AVG(value) FILTER (WHERE name = 'factuality') as avg_factuality
FROM scores
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY DATE(timestamp)
ORDER BY date;
```

## Datasets for Evaluation

Create test datasets and run automated evaluations:

```python
from langfuse import Langfuse, observe, get_client

langfuse = Langfuse()

# Fetch dataset
dataset = langfuse.get_dataset("security_audit_test_set")

@observe()
async def evaluate_item(item):
    """Evaluate a single dataset item with tracing."""
    response = await llm.generate(item.input)

    get_client().update_current_observation(
        input=item.input,
        output=response,
    )

    # Score
    score = await evaluate_response(item.expected_output, response)
    get_client().score_current_trace(name="accuracy", value=score)

    return score

# Run evaluation
for item in dataset.items:
    await evaluate_item(item)
```

## Dataset Structure in UI

```
security_audit_test_set
├── item_1: XSS vulnerability test
│   ├── input: "Check this HTML for XSS..."
│   └── expected_output: "Found XSS in innerHTML..."
├── item_2: SQL injection test
│   ├── input: "Review this SQL query..."
│   └── expected_output: "SQL injection vulnerability in WHERE clause..."
└── item_3: CSRF protection test
    ├── input: "Analyze this form..."
    └── expected_output: "Missing CSRF token..."
```

## Evaluation Metrics

Common score types:

| Metric | Range | Description |
|--------|-------|-------------|
| **Relevance** | 0-1 | Does response address the query? |
| **Coherence** | 0-1 | Is response logically structured? |
| **Depth** | 0-1 | Level of detail and analysis |
| **Factuality** | 0-1 | Accuracy of claims |
| **Completeness** | 0-1 | All aspects of query covered? |
| **Toxicity** | 0-1 | Harmful or inappropriate content |

## Best Practices

1. **Score all production traces** for quality monitoring
2. **Use evaluator type** (`@observe(as_type="evaluator")`) for inspectable judge traces
3. **Use consistent criteria** across all evaluations
4. **Automate scoring** with G-Eval or similar
5. **Set quality thresholds** (e.g., avg_relevance > 0.7)
6. **Create test datasets** for regression testing
7. **Track scores by prompt version** to measure improvements
8. **Alert on quality drops** (e.g., avg_score &lt; 0.6 for 3 days)

## Integration with OrchestKit Quality Gate

```python
from langfuse import observe, get_client

@observe(name="quality_gate")
async def quality_gate_node(state: WorkflowState):
    """Quality gate with Langfuse scoring."""

    # Get scores from evaluators
    scores = await run_quality_evaluators(state)

    # Log scores to trace
    for criterion, score in scores.items():
        get_client().score_current_trace(name=criterion, value=score)

    # Check threshold
    avg_score = sum(scores.values()) / len(scores)
    passed = avg_score >= 0.7

    return {"quality_gate_passed": passed, "quality_scores": scores}
```

## References

- [Langfuse Scores](https://langfuse.com/docs/scores)
- [Evaluator Tracing](https://langfuse.com/docs/scores-and-evaluation/evaluator-tracing)
- [Score Analytics](https://langfuse.com/docs/analytics/scores)
- [Datasets Guide](https://langfuse.com/docs/datasets)


### Experiments Api

# Langfuse Experiments API

## Overview

The Experiments API enables systematic evaluation of LLM outputs across datasets. Use it for A/B testing prompts, comparing models, and tracking quality over time. v3 adds the Experiment Runner SDK, dataset item versioning, and corrected outputs.

```
┌─────────────────────────────────────────────────────────────────────┐
│                     Langfuse Experiments Flow                        │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│   ┌──────────┐     ┌────────────┐     ┌──────────────┐              │
│   │ Dataset  │────▶│ Experiment │────▶│ Runs (Items) │              │
│   │ (inputs) │     │ (config)   │     │ (executions) │              │
│   └──────────┘     └────────────┘     └──────┬───────┘              │
│                                              │                       │
│                                              ▼                       │
│                                    ┌──────────────────┐             │
│                                    │ Evaluators       │             │
│                                    │ (judge outputs)  │             │
│                                    └────────┬─────────┘             │
│                                             │                        │
│                                             ▼                        │
│                                    ┌──────────────────┐             │
│                                    │ Scores           │             │
│                                    │ (per run)        │             │
│                                    └──────────────────┘             │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘
```

## Experiment Runner SDK (v3)

The high-level API simplifies running experiments:

```python
from langfuse import Langfuse

langfuse = Langfuse()


async def my_pipeline(input_data: dict) -> str:
    """Your LLM pipeline to evaluate."""
    return await llm.generate(input_data["query"])


# Run experiment in one call
result = langfuse.run_experiment(
    dataset_name="golden-analysis-dataset",
    experiment_name="sonnet-v-gpt5",
    run_fn=my_pipeline,
    evaluators=[
        {"name": "relevance", "fn": relevance_evaluator},
        {"name": "depth", "fn": depth_evaluator},
    ],
)

# Result contains:
# - experiment_id
# - per-item scores
# - aggregate statistics
print(f"Avg relevance: {result.stats['relevance']['mean']:.2f}")
print(f"Avg depth: {result.stats['depth']['mean']:.2f}")
```

## Creating Datasets

### From Code

```python
from langfuse import Langfuse

langfuse = Langfuse()

# Create dataset with JSON schema enforcement
dataset = langfuse.create_dataset(
    name="golden-analysis-dataset",
    description="Curated analysis examples with expected outputs",
    metadata={"version": "v2", "schema_version": "1.0"},
)

# Add items with versioning
items = [
    {
        "input": {"url": "https://example.com/article1", "type": "article"},
        "expected_output": "Expected analysis for article 1...",
        "metadata": {"category": "tutorial", "difficulty": "beginner"},
    },
    {
        "input": {"url": "https://example.com/article2", "type": "article"},
        "expected_output": "Expected analysis for article 2...",
        "metadata": {"category": "reference", "difficulty": "advanced"},
    },
]

for item in items:
    langfuse.create_dataset_item(
        dataset_name="golden-analysis-dataset",
        input=item["input"],
        expected_output=item.get("expected_output"),
        metadata=item.get("metadata"),
    )
```

### From Existing Traces

```python
# Create dataset from production traces
traces = langfuse.get_traces(
    filter={
        "score_name": "human_verified",
        "score_value_gte": 0.9,  # Only high-quality
    },
    limit=100,
)

dataset = langfuse.create_dataset(name="production-golden-v1")

for trace in traces:
    langfuse.create_dataset_item(
        dataset_name="production-golden-v1",
        input=trace.input,
        expected_output=trace.output,
        metadata={"trace_id": trace.id},
    )
```

## Dataset Item Versioning

Track changes to dataset items over time:

```python
# Update an existing item — creates a new version
langfuse.update_dataset_item(
    dataset_name="golden-analysis-dataset",
    item_id="item_123",
    expected_output="Updated expected output with more detail...",
    metadata={"version": 2, "updated_by": "human_reviewer"},
)

# View item history in Langfuse UI:
# item_123 v1 (Jan 15) → v2 (Feb 01)
# Each experiment run records which version it evaluated against
```

## JSON Schema Enforcement

Enforce structure on dataset items:

```python
# Create dataset with schema
dataset = langfuse.create_dataset(
    name="structured-eval-dataset",
    description="Dataset with enforced input schema",
    metadata={
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "context": {"type": "array", "items": {"type": "string"}},
            },
            "required": ["query"],
        }
    },
)

# Items must match schema — invalid items are rejected
langfuse.create_dataset_item(
    dataset_name="structured-eval-dataset",
    input={"query": "What is XSS?", "context": ["OWASP guide..."]},
    expected_output="XSS is a web security vulnerability...",
)
```

## Dataset Folder Organization

Organize datasets into folders:

```
Datasets/
├── production/
│   ├── golden-v1
│   ├── golden-v2
│   └── regression-suite
├── experiments/
│   ├── prompt-variants
│   └── model-comparison
└── development/
    ├── unit-tests
    └── edge-cases
```

## Batch Add Observations to Datasets

Add multiple observations at once:

```python
# Batch add from production traces
observation_ids = ["obs_1", "obs_2", "obs_3", "obs_4", "obs_5"]

langfuse.batch_add_dataset_items(
    dataset_name="production-golden-v1",
    observation_ids=observation_ids,
    metadata={"batch": "2026-02-01", "source": "production"},
)
```

## Corrected Outputs for Fine-Tuning

Use corrected outputs to build fine-tuning datasets:

```python
# Add corrected output to existing dataset item
langfuse.update_dataset_item(
    dataset_name="golden-analysis-dataset",
    item_id="item_456",
    corrected_output="Human-corrected version of the analysis...",
    metadata={"corrected_by": "expert_reviewer", "correction_type": "factual"},
)

# Export for fine-tuning
items = langfuse.get_dataset("golden-analysis-dataset").items

fine_tuning_data = []
for item in items:
    if item.corrected_output:
        fine_tuning_data.append({
            "messages": [
                {"role": "user", "content": str(item.input)},
                {"role": "assistant", "content": item.corrected_output},
            ]
        })

# Export as JSONL for fine-tuning
import json
with open("fine_tuning.jsonl", "w") as f:
    for entry in fine_tuning_data:
        f.write(json.dumps(entry) + "\n")
```

## Running Experiments (Manual)

### Basic Experiment

```python
from langfuse import observe, get_client, Langfuse

langfuse = Langfuse()


@observe()
async def run_experiment(
    dataset_name: str,
    experiment_name: str,
    model_config: dict,
):
    """Run an experiment on a dataset."""
    dataset = langfuse.get_dataset(dataset_name)

    results = []

    for item in dataset.items:
        @observe(name="experiment_run")
        async def evaluate_item(item=item):
            output = await your_pipeline(item.input, model_config)

            get_client().update_current_observation(
                input=item.input,
                output=output,
                metadata={"dataset_item_id": item.id},
            )

            return output

        output = await evaluate_item()
        results.append({"item_id": item.id, "output": output})

    return results
```

### A/B Testing Models

```python
async def ab_test_models(dataset_name: str):
    """Compare two model configurations."""

    configs = {
        "sonnet": {"model": "claude-sonnet-5", "temperature": 0.7},
        "gpt5": {"model": "gpt-5.5", "temperature": 0.7},
    }

    for name, config in configs.items():
        result = langfuse.run_experiment(
            dataset_name=dataset_name,
            experiment_name=f"model-comparison-{name}",
            run_fn=lambda input_data: your_pipeline(input_data, config),
            evaluators=[
                {"name": "relevance", "fn": relevance_evaluator},
                {"name": "depth", "fn": depth_evaluator},
            ],
        )
        print(f"{name}: avg_relevance={result.stats['relevance']['mean']:.2f}")
```

## Experiment Compare View

Compare experiments side-by-side in Langfuse UI:

- **Aggregate scores**: Average, median, std per criterion
- **Per-item comparison**: See how each item scored across experiments
- **Annotations**: Add notes to individual items or experiments
- **Diff view**: See which items improved or regressed
- **Export**: Download comparison as CSV

## OrchestKit Integration

### Golden Dataset Experiment

```python
from langfuse import Langfuse

langfuse = Langfuse()


async def run_golden_experiment():
    """Run quality experiment on golden dataset."""

    # 1. Create dataset from golden analyses
    golden_analyses = await get_golden_analyses()

    dataset = langfuse.create_dataset(name="orchestkit-golden-v1")
    for analysis in golden_analyses:
        langfuse.create_dataset_item(
            dataset_name="orchestkit-golden-v1",
            input={"url": analysis.url},
            expected_output=analysis.synthesis,
            metadata={"analysis_id": str(analysis.id)},
        )

    # 2. Run experiment with Experiment Runner
    result = langfuse.run_experiment(
        dataset_name="orchestkit-golden-v1",
        experiment_name=f"quality-test-{datetime.now().isoformat()}",
        run_fn=run_analysis_pipeline,
        evaluators=[
            {"name": "depth", "fn": depth_evaluator},
            {"name": "accuracy", "fn": accuracy_evaluator},
            {"name": "overall", "fn": overall_evaluator},
        ],
    )

    return {
        "experiment_id": result.experiment_id,
        "stats": result.stats,
        "pass_rate": result.stats["overall"]["mean"] >= 0.7,
    }
```

### Prompt Variant Testing

```python
async def test_prompt_variants():
    """A/B test different prompt templates."""

    variants = {
        "detailed": "Provide a comprehensive, in-depth analysis...",
        "concise": "Provide a brief, focused analysis...",
        "structured": "Analyze using the following structure: 1) Overview...",
    }

    results = {}
    for name, prompt in variants.items():
        result = langfuse.run_experiment(
            dataset_name="orchestkit-golden-v1",
            experiment_name=f"prompt-variant-{name}",
            run_fn=lambda input_data: run_with_prompt(input_data, prompt),
            evaluators=[
                {"name": "overall", "fn": overall_evaluator},
            ],
        )
        results[name] = result.stats["overall"]["mean"]

    # Compare all variants
    winner = max(results, key=results.get)
    return {"winner": winner, "scores": results}
```

## Viewing Results

### Langfuse Dashboard

1. **Experiments Tab**: See all experiments with aggregate scores
2. **Compare View**: Side-by-side experiment comparison with annotations
3. **Runs Tab**: See individual executions with per-item scores
4. **Diff View**: Identify regressions between experiment versions

### Export Results

```python
import pandas as pd
from langfuse import Langfuse

langfuse = Langfuse()


def export_experiment_results(experiment_id: str) -> pd.DataFrame:
    """Export experiment results to DataFrame."""

    runs = langfuse.get_experiment_runs(experiment_id)

    data = []
    for run in runs:
        scores = langfuse.get_scores(trace_id=run.trace_id)
        score_dict = {s.name: s.value for s in scores}

        data.append({
            "run_id": run.id,
            "item_id": run.dataset_item_id,
            **run.input,
            **score_dict,
        })

    return pd.DataFrame(data)
```

## Best Practices

1. **Use Experiment Runner SDK** for simplified experiment execution
2. **Version your datasets** with semantic names like `golden-v1`, `golden-v2`
3. **Use corrected outputs** to build fine-tuning datasets from production data
4. **Include metadata**: Store model config, prompt version in experiment metadata
5. **Evaluate consistently**: Use same evaluators across experiments
6. **Track over time**: Run same experiment periodically to detect regression
7. **Use ground truth**: When available, compute similarity to expected output
8. **Organize datasets** in folders by purpose (production, experiments, dev)


### Langfuse Js V5

# Langfuse JS/TS SDK v5: the delta from the Python page

Wrap, not tutorial. Full API docs live at &lt;https://langfuse.com/docs/sdk/typescript&gt;. This page
carries only what a Python-shaped mental model gets wrong, plus the symbols this repo has
already been burned by.

The JS/TS SDK is a **different major** from Python and a different package layout. Python is on
4.x and imports everything from `langfuse`; JS/TS is on 5.x and splits across scoped packages.
Do not translate a Python snippet by guessing the JS name.

Every symbol below was taken from the published type declarations at 5.9.1
(`cdn.jsdelivr.net/npm/@langfuse/&lt;pkg&gt;@5.9.1/dist/index.d.ts`). If you add one, verify it the
same way. This file previously documented `LangfuseExporter`, a class that has never existed.

## Package map

| Package | Exports you actually use |
|---|---|
| `@langfuse/otel` | `LangfuseSpanProcessor`, `MaskFunction`, `ShouldExportSpan`, `isDefaultExportSpan`, `isGenAISpan`, `isKnownLLMInstrumentor`, `isLangfuseSpan` |
| `@langfuse/tracing` | `observe`, `startObservation`, `startActiveObservation`, `updateActiveObservation`, `createTraceId`, `getActiveTraceId`, `setActiveTraceIO`, `propagateAttributes` |
| `@langfuse/client` | `LangfuseClient`, `DatasetManager`, `ExperimentManager`, `ScoreManager`, `PromptManager`, `createEvaluatorFromAutoevals`, `RegressionError` |
| `@langfuse/langchain` | `CallbackHandler` |
| `@langfuse/openai` | OpenAI SDK auto-instrumentation |
| `@langfuse/vercel-ai-sdk` | `LangfuseVercelAiSdkIntegration` |

Two traps:

- **`@langfuse/core` is not the client.** It is an internal utility package ("Core functions and
  utilities for Langfuse packages") exporting API types and `LangfuseAPIClient`. It has no
  `Langfuse` export. Never install or import it directly.
- **`@langfuse/vercel` does not exist.** The successor to the old `langfuse-vercel` is
  `@langfuse/vercel-ai-sdk`. It was documented here for months and was never published.

## It is a SpanProcessor, not an exporter

The single most common porting mistake. It goes in `spanProcessors`, never `traceExporter`.

```typescript
import { NodeSDK } from "@opentelemetry/sdk-node";
import { LangfuseSpanProcessor } from "@langfuse/otel";

const sdk = new NodeSDK({
  spanProcessors: [new LangfuseSpanProcessor()],
});
sdk.start();

// Short-lived processes MUST flush, or trailing spans are dropped on exit.
main().finally(() => sdk.shutdown());
```

Credentials default to `LANGFUSE_PUBLIC_KEY`, `LANGFUSE_SECRET_KEY` and `LANGFUSE_BASE_URL`
(default `https://cloud.langfuse.com`), the same names the Python client reads.

## Masking and filtering: where the Python kwargs went

The Python client takes `should_export_span` as a constructor kwarg. In JS the equivalent levers
live on the processor as `mask` and `shouldExportSpan`, alongside `flushAt`, `flushInterval`,
`exportMode`, `environment` and `release`.

```typescript
new LangfuseSpanProcessor({
  // Redact secrets from span payloads before they leave the process.
  mask: ({ data }) =>
    typeof data === "string" ? data.replace(/secret_\w+/g, "secret_***") : data,

  // Drop noisy infra spans (DB drivers, DNS, HTTP clients) to cut ingestion cost.
  shouldExportSpan: ({ otelSpan }) => otelSpan.name.startsWith("my-service"),
});
```

`shouldExportSpan` is a **full override** of the default filtering, not a narrowing. To narrow,
compose with the shipped predicate instead of replacing it:

```typescript
import { LangfuseSpanProcessor, isDefaultExportSpan } from "@langfuse/otel";

new LangfuseSpanProcessor({
  shouldExportSpan: (params) =>
    isDefaultExportSpan(params) && !params.otelSpan.name.startsWith("pg."),
});
```

## Evaluator vs RunEvaluator

The JS client ships an experiment runner the Python-focused pages do not cover, and the two
evaluator shapes are easy to confuse.

| Type | Shape |
|---|---|
| `ExperimentTask` | `(params) => Promise&lt;any&gt;`, receives `input`, `expectedOutput`, `metadata` |
| `Evaluator` | `(params) => Promise&lt;Evaluation \| Evaluation[]&gt;`, scores ONE item |
| `RunEvaluator` | `(params) => Promise&lt;Evaluation \| Evaluation[]&gt;`, scores the WHOLE run |
| `Evaluation` | `\{ name, value, comment?, metadata?, dataType?, configId? \}` |

Use `Evaluator` for per-item quality and `RunEvaluator` for aggregate assertions (pass rate,
mean score). A per-item evaluator cannot see the other items, so an aggregate check written as
an `Evaluator` silently measures the wrong thing.

`createEvaluatorFromAutoevals` adapts an autoevals scorer instead of hand-writing one, and
`RegressionError` is thrown when a run regresses against a configured baseline. Catch it to fail
CI on quality drops rather than only on exceptions.

On the older AI SDK v6 line there is no integration package: set
`experimental_telemetry: \{ isEnabled: true \}` on the call and let `LangfuseSpanProcessor`
collect it.

## Cross-references

- Ork-specific floors, scars and house decisions: `references/ork-delta.md`
- Python v4 tracing rules: `rules/llm-langfuse-traces.md`
- Upstream migration guides: &lt;https://langfuse.com/docs/sdk/typescript/v4-migration&gt;


### Metrics Collection

# Metrics Collection

Application metrics best practices with Prometheus.

## Metric Types

### 1. Counter - Monotonically increasing value (resets to 0 on restart)
```python
http_requests_total = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

# Usage
http_requests_total.labels(method='GET', endpoint='/api/users', status=200).inc()
```
**Use cases:** Request counts, error counts, bytes processed

### 2. Gauge - Value that can go up or down
```python
active_connections = Gauge(
    'active_connections',
    'Number of active database connections'
)

# Usage
active_connections.set(25)  # Set to specific value
active_connections.inc()    # Increment by 1
active_connections.dec()    # Decrement by 1
```
**Use cases:** Queue length, memory usage, temperature

### 3. Histogram - Distribution of values (with buckets)
```python
request_duration = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1, 2, 5, 10]  # Choose meaningful buckets!
)

# Usage
with request_duration.labels(method='GET', endpoint='/api/users').time():
    # ... handle request
    pass
```
**Use cases:** Request latency, response size

### 4. Summary - Like Histogram but calculates quantiles on client side
```python
request_duration = Summary(
    'http_request_duration_seconds',
    'HTTP request duration',
    ['method', 'endpoint']
)
```

**Histogram vs Summary:**
- **Histogram**: Calculate quantiles on Prometheus server (recommended)
- **Summary**: Calculate quantiles on application side (higher client CPU, can't aggregate across instances)

## Cardinality Management

**Problem:** Too many unique label combinations

```python
# BAD: Unbounded cardinality (user_id can be millions of values)
http_requests_total = Counter(
    'http_requests_total',
    ['method', 'endpoint', 'user_id']  # user_id creates millions of time series!
)

# GOOD: Bounded cardinality
http_requests_total = Counter(
    'http_requests_total',
    ['method', 'endpoint', 'status']  # Limited to ~10 methods x 100 endpoints x 10 statuses = 10,000 series
)
```

**Cardinality limits:**
- Good: &lt; 10,000 unique time series per metric
- Acceptable: 10,000-100,000
- Bad: > 100,000 (Prometheus performance degrades)

**Rule:** Never use unbounded labels (user IDs, request IDs, timestamps)

## Custom Business Metrics

```python
# LLM token usage
llm_tokens_used = Counter(
    'llm_tokens_used_total',
    'Total LLM tokens consumed',
    ['model', 'operation']  # e.g., model='claude-sonnet', operation='analysis'
)

# LLM cost tracking
llm_cost_dollars = Counter(
    'llm_cost_dollars_total',
    'Total LLM cost in dollars',
    ['model']
)

# Cache hit rate
cache_operations = Counter(
    'cache_operations_total',
    'Cache operations',
    ['operation', 'result']  # operation='get', result='hit|miss'
)

# Cache hit rate query:
# sum(rate(cache_operations_total{result="hit"}[5m])) /
# sum(rate(cache_operations_total[5m]))
```

## LLM Cost Tracking Example

```python
from prometheus_client import Counter, Histogram

llm_tokens_used = Counter(
    'llm_tokens_used_total',
    'Total LLM tokens consumed',
    ['model', 'operation', 'token_type']
)

llm_cost_dollars = Counter(
    'llm_cost_dollars_total',
    'Total LLM cost in dollars',
    ['model', 'operation']
)

llm_request_duration = Histogram(
    'llm_request_duration_seconds',
    'LLM request duration',
    ['model', 'operation'],
    buckets=[0.5, 1, 2, 5, 10, 20, 30]
)

@observe(name="llm_call")
async def call_llm(prompt: str, model: str, operation: str) -> str:
    start_time = time.time()
    response = await anthropic_client.messages.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=1024
    )
    duration = time.time() - start_time

    input_tokens = response.usage.input_tokens
    output_tokens = response.usage.output_tokens

    llm_tokens_used.labels(model=model, operation=operation, token_type="input").inc(input_tokens)
    llm_tokens_used.labels(model=model, operation=operation, token_type="output").inc(output_tokens)

    # Cost calculation (Claude Sonnet 4.5 pricing)
    input_cost = (input_tokens / 1_000_000) * 3.00
    output_cost = (output_tokens / 1_000_000) * 15.00
    total_cost = input_cost + output_cost

    llm_cost_dollars.labels(model=model, operation=operation).inc(total_cost)
    llm_request_duration.labels(model=model, operation=operation).observe(duration)

    return response.content[0].text
```

**Grafana dashboard queries:**
```text
# Total cost per day
sum(increase(llm_cost_dollars_total[1d])) by (model)

# Token usage rate
sum(rate(llm_tokens_used_total[5m])) by (model, token_type)

# Cost per operation
sum(increase(llm_cost_dollars_total[1h])) by (operation)

# p95 LLM latency
histogram_quantile(0.95, rate(llm_request_duration_seconds_bucket[5m]))
```

See `scripts/prometheus-metrics.ts` for complete setup.

## Claude Code OTel Metrics — Notes for Dashboards

Claude Code emits OTel metrics under the `claude_code.*` namespace. Two metrics that frequently appear on team velocity dashboards:

| Metric | Counts |
|---|---|
| `claude_code.tool.use` | Each tool invocation, labelled by `tool` |
| `claude_code.pull_request.count` | PRs/MRs created during sessions |

### CC 2.1.129: `claude_code.pull_request.count` now counts MCP-filed PRs

Before CC 2.1.129, `claude_code.pull_request.count` only counted PRs/MRs created via shell commands run through the Bash tool (`gh pr create`, `glab mr create`, custom scripts). As of CC 2.1.129, the metric **also counts PRs/MRs filed via MCP tools** — e.g., GitHub MCP server's `create_pull_request`, GitLab MCP equivalents, and any custom MCP server exposing a PR/MR-creation tool.

**Impact on existing dashboards**: a step-function increase at the 2.1.129 cutover for teams where MCP-driven PR creation is non-trivial. The "spike" is a measurement-surface change, not a behavioral change. Annotate the dashboard with the version bump so it isn't misread as a productivity surge.

**Labels are unchanged**: the counter still emits `provider` and `result` labels; MCP-filed PRs are not specially flagged. If you need to distinguish MCP vs. shell origins:

```text
# Derive an MCP-origin counter by joining on tool-name in a separate stream
sum(rate(claude_code.pull_request.count[5m])) by (provider)
  - on(provider) sum(rate(claude_code.tool.use{tool=~"Bash"}[5m]))
```

(In practice, hold a separate counter in your collector that increments only on `claude_code.tool.use\{tool=~"mcp__.*"\}` co-occurring with a PR-creation event.)

**See also**: `$\{CLAUDE_PLUGIN_ROOT\}/skills/telemetry-inspect/SKILL.md` and `$\{CLAUDE_PLUGIN_ROOT\}/skills/configure/references/cc-version-settings.md` (CC 2.1.129 section) for the upstream changelog reference.

### Ork Delta

# OrchestKit Delta: Monitoring and Observability

What this repo knows that the vendor docs do not. Prometheus, Grafana, OpenTelemetry and
Langfuse tutorials belong upstream (see the "Upstream coverage" table in `SKILL.md`); this file
holds only the version floors, house decisions and scars that came out of OrchestKit itself.

Every entry is `rule / Why / Upstream`. An entry with no scar and no house decision does not
belong here, it belongs in the SKILL.md pointer table.

## Read `TRACEPARENT` from the environment in anything Claude Code spawns

Why: ork hook telemetry already forwards it. `src/hooks/src/lib/telemetry.ts:138` attaches
`process.env.TRACEPARENT` to every emitted event and `src/hooks/src/lib/http-sink.ts:225`
re-sends it as an HTTP `traceparent` header, so a downstream service that ignores the variable
silently breaks the CC-tool-span to service-trace join that the hooks already paid for. The
variable is only populated when OTEL tracing is enabled, so treat absence as "not traced", not
as an error.

Upstream: `src/skills/doctor/references/version-compatibility.md` (CC 2.1.97 row, "Bash OTEL
TRACEPARENT: subprocesses inherit W3C TRACEPARENT env var when OTEL tracing is enabled").

## Set `OTEL_*` at the subprocess invocation site, never in the shell that launched Claude Code

Why: CC 2.1.128 stopped propagating `OTEL_*` from the CLI process into spawned children (Bash
tool, hooks, MCP stdio and Streamable HTTP servers, LSP servers). Anything that relied on
inheritance goes silent with no error and no log line, so a step-down in span volume right
after a CC upgrade is this gate, not a broken collector. For MCP servers, declare the OTEL
variables in the server's `env` block in `.mcp.json` so the CLI collector and the server
collector stay independently configurable.

Upstream: `src/skills/configure/references/cc-version-settings.md`, section "Subprocesses No
Longer Inherit OTEL_* Env Vars", which is the canonical write-up for this repo.

## Never link the Langfuse SDK into an OrchestKit hook

Why: house decision. Hooks run as short-lived per-event processes, so SDK initialization
(roughly 50 to 150 ms) is paid on every single spawn, and requiring `LANGFUSE_*` would turn an
optional cloud account into a hard install dependency. ork hooks instead stay output-format
agnostic and append JSONL under `~/.claude/analytics/` (`hook-timing.jsonl`,
`agent-usage.jsonl`, `session-summary.jsonl`, `skill-usage.jsonl`, `task-usage.jsonl`,
`team-activity.jsonl`). Anything that wants traces bridges from those files in one long-lived
process, where SDK init is amortized instead of repeated.

Upstream: `src/skills/telemetry-inspect/SKILL.md` for the file inventory and health checks, and
`references/dev-agent-lens.md` for the proxy layer that sits at the API boundary instead.

## Treat Langfuse's three version numbers as three separate axes

Why: the Python SDK (4.x), the JS/TS SDK (5.x) and the self-hosted platform (v3, Postgres plus
ClickHouse plus Redis plus S3 or blob) move independently. "Correcting" one number into another
has repeatedly turned accurate documentation into wrong documentation here, which is why the
repo's skill-authoring rule calls out Langfuse by name under "Distinguish version axes". A doc
saying "Langfuse v3 requires ClickHouse" is talking about the platform and is correct. This
skill's `targets:` floor and `upstream-version-tested:` both describe the Python SDK axis only.

Upstream: `.claude/rules/skill-authoring.md`, section "Version and API Claims Must Be
Machine-Checkable".

## Confirm a `@langfuse/*` package and every imported symbol exist before writing the import

Why: two shipped scars. `LangfuseExporter` was documented across four files in this repo and has
never been exported by any published version of `@langfuse/otel`, and `@langfuse/vercel` was
documented for months and was never published at all (the real package is
`@langfuse/vercel-ai-sdk`). No version gate can catch a symbol that never existed, so the check
has to happen at authoring time: `curl registry.npmjs.org/&lt;pkg&gt;/latest` for the package, then
read the published `dist/index.d.ts` for the symbol. Related trap: `@langfuse/core` is an
internal utility package with no `Langfuse` export, so it is never the client.

Upstream: `references/langfuse-js-v5.md` holds the verified symbol table, and
`.claude/rules/skill-authoring.md` codifies the check.

## Do not present a Python service tree as OrchestKit's own observability implementation

Why: distilled from the retired orchestkit-langfuse-traces example (plus the multi-judge and
prompt-management references deleted alongside it); no traced incident. All three described a
different codebase that happened to share the name, citing `backend/app/shared/services/...`
modules, a `QUALITY_INITIATIVE_FIXES` doc and an 8-agent LangGraph analysis pipeline, none of
which exist at HEAD. The same content attributed a Jinja2
prompt-fallback design to "Issue #414"; `gh issue view 414` resolves to "fix(tests):
multi-instance-lock security tests skip, hook was deleted in #361", so the attribution was
false. OrchestKit ships a Claude Code plugin (skills, agents, TypeScript hooks); its own
observability surface is the JSONL analytics files above, not a FastAPI backend.

Upstream: `.claude/rules/skill-authoring.md` for the provenance bar, and `CLAUDE.md` for the
actual directory structure.


### Session Tracking

# Session & User Tracking

Group related traces, track performance by user, and filter with natural language.

## Session Tracking (v3)

Group related traces into user sessions using `get_client()`:

```python
from langfuse import observe, get_client

@observe(name="url_fetch")
async def fetch_url(url: str, session_id: str):
    get_client().update_current_trace(session_id=session_id)
    return await http.get(url)

@observe(name="content_analysis")
async def analyze(content: str, session_id: str):
    get_client().update_current_trace(session_id=session_id)
    return await run_agents(content)

@observe(name="quality_gate")
async def quality_check(result: str, session_id: str):
    get_client().update_current_trace(session_id=session_id)
    return await evaluate(result)


# Usage — all 3 traces grouped under one session
session_id = f"analysis_{analysis_id}"
url_content = await fetch_url(url, session_id)
result = await analyze(url_content, session_id)
final = await quality_check(result, session_id)
```

## Session View in UI

```
Session: analysis_abc123 (15.2s, $0.23)
├── url_fetch (1.0s, $0.02)
├── content_analysis (12.5s, $0.18)
│   ├── retrieval (0.5s, $0.01)
│   ├── security_audit (3.0s, $0.05)
│   ├── tech_comparison (2.5s, $0.04)
│   └── implementation_plan (6.5s, $0.08)
└── quality_gate (1.7s, $0.03)
```

## User Tracking

Track performance per user:

```python
from langfuse import observe, get_client

@observe()
async def analysis(content: str, user_id: str):
    get_client().update_current_trace(
        user_id=user_id,
        session_id="session_abc",
        metadata={
            "content_type": "article",
            "url": "https://example.com/post",
            "analysis_id": "abc123",
        },
    )
    return await run_pipeline(content)
```

## Natural Language Filtering

Langfuse v3 supports natural language queries to filter traces in the UI:

```
# Examples of natural language filters:
"show me traces with latency > 5s from yesterday"
"find all traces by user_123 with cost > $0.10"
"traces tagged 'production' with relevance score < 0.5"
"sessions with more than 3 traces in the last 24 hours"
```

This replaces manual filter construction for common queries.

## Metadata Tracking

Track custom metadata for filtering and analytics:

```python
from langfuse import observe, get_client

@observe()
async def analysis(content: str):
    get_client().update_current_trace(
        user_id="user_123",
        metadata={
            "content_type": "article",
            "url": "https://example.com/post",
            "analysis_id": "abc123",
            "agent_count": 8,
            "total_cost_usd": 0.15,
            "difficulty": "complex",
            "language": "en",
        },
        tags=["production", "orchestkit", "security"],
    )
    return await run_pipeline(content)
```

## Analytics Queries

### Performance by User

```sql
SELECT
    user_id,
    COUNT(*) as trace_count,
    AVG(latency_ms) as avg_latency,
    SUM(calculated_total_cost) as total_cost
FROM traces
WHERE timestamp > NOW() - INTERVAL '30 days'
GROUP BY user_id
ORDER BY total_cost DESC
LIMIT 10;
```

### v2 Metrics API Alternative

```python
from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse()

# Query session metrics via Metrics API instead of SQL
metrics = langfuse.get_metrics(
    metric_name="trace_count",
    from_timestamp=datetime.now() - timedelta(days=7),
    to_timestamp=datetime.now(),
    group_by="user_id",
    granularity="day",
)

for group in metrics.groups:
    print(f"User {group.key}: {group.values[0].value} traces")
```

### Performance by Content Type

```sql
SELECT
    metadata->>'content_type' as content_type,
    COUNT(*) as count,
    AVG(latency_ms) as avg_latency,
    AVG(calculated_total_cost) as avg_cost
FROM traces
WHERE metadata->>'content_type' IS NOT NULL
GROUP BY content_type
ORDER BY count DESC;
```

### Slowest Sessions

```sql
SELECT
    session_id,
    COUNT(*) as trace_count,
    SUM(latency_ms) as total_latency,
    SUM(calculated_total_cost) as total_cost
FROM traces
WHERE session_id IS NOT NULL
    AND timestamp > NOW() - INTERVAL '7 days'
GROUP BY session_id
ORDER BY total_latency DESC
LIMIT 10;
```

## Tags for Filtering

Use tags for environment and feature flags:

```python
from langfuse import observe, get_client

@observe()
async def production_analysis(content: str):
    get_client().update_current_trace(
        tags=["production", "v2-pipeline", "security-enabled"],
    )
    return await run_pipeline(content)

@observe()
async def staging_analysis(content: str):
    get_client().update_current_trace(
        tags=["staging", "experiment", "new-model"],
    )
    return await run_pipeline(content)
```

## Best Practices

1. **Always set session_id** for multi-step workflows
2. **Always set user_id** for user attribution
3. **Add meaningful metadata** (content_type, analysis_id, difficulty)
4. **Use consistent tag names** across environments
5. **Tag production vs staging** traces
6. **Use natural language filtering** for quick trace lookups
7. **Track business metrics** in metadata (conversion, revenue, user_tier)
8. **Filter by tags** in dashboards for environment-specific views

## OrchestKit Session Pattern

```python
from langfuse import observe, get_client

@observe(name="content_analysis_workflow")
async def run_content_analysis(analysis_id: str, content: str, user_id: str):
    """Full workflow with session tracking."""

    # Set session-level metadata
    get_client().update_current_trace(
        session_id=f"analysis_{analysis_id}",
        user_id=user_id,
        metadata={
            "analysis_id": analysis_id,
            "content_length": len(content),
            "agent_count": 8,
            "environment": "production",
        },
        tags=["orchestkit", "production", "content-analysis"],
    )

    # All nested @observe calls inherit session_id
    results = []
    for agent in agents:
        result = await execute_agent(agent, content)
        results.append(result)

    return results
```

## Identifying Slow or Expensive Users

```sql
-- Users with highest average latency
SELECT
    user_id,
    COUNT(*) as sessions,
    AVG(total_latency) as avg_session_latency,
    AVG(total_cost) as avg_session_cost
FROM (
    SELECT
        user_id,
        session_id,
        SUM(latency_ms) as total_latency,
        SUM(calculated_total_cost) as total_cost
    FROM traces
    WHERE timestamp > NOW() - INTERVAL '7 days'
    GROUP BY user_id, session_id
) sessions
GROUP BY user_id
HAVING COUNT(*) >= 5  -- At least 5 sessions
ORDER BY avg_session_latency DESC
LIMIT 10;
```

## References

- [Langfuse Sessions](https://langfuse.com/docs/tracing-features/sessions)
- [User Tracking](https://langfuse.com/docs/tracing-features/users)
- [Tags & Metadata](https://langfuse.com/docs/tracing)
- [Natural Language Filtering](https://langfuse.com/docs/tracing-features/filtering)


### Structured Logging

# Structured Logging

JSON logging best practices for production systems.

## Why Structured Logging?

- **Searchable** - query by fields (user_id, trace_id)
- **Machine-readable** - parse and aggregate easily
- **Contextual** - attach metadata to every log

## Python (structlog)

```python
import structlog

logger = structlog.get_logger()

logger.info("user_login", user_id="123", ip="192.168.1.1")
# Output: {"event": "user_login", "user_id": "123", "ip": "192.168.1.1", "timestamp": "2025-12-19T10:00:00Z"}
```

## Node.js (pino)

```typescript
import pino from 'pino';

const logger = pino();

logger.info({ userId: '123', action: 'login' }, 'User logged in');
// Output: {"level":30,"userId":"123","action":"login","msg":"User logged in","time":1702990800000}
```

## Log Levels

| Level | Use Case | Example |
|-------|----------|---------|
| **DEBUG** | Development only | Variable values, function calls |
| **INFO** | Normal operations | User actions, workflow steps |
| **WARN** | Recoverable issues | Retries, deprecated API usage |
| **ERROR** | Failures | Exceptions, failed requests |
| **CRITICAL** | System failure | Database down, out of memory |

## Best Practices

1. **Always include trace_id** - correlate across services
2. **Log at boundaries** - API requests/responses, DB queries
3. **Don't log secrets** - mask passwords, API keys
4. **Use correlation IDs** - track requests across microservices

See `scripts/structured-logging.ts` for implementation.



---

## Examples (1)

### Orchestkit Monitoring Dashboard

# OrchestKit Monitoring Dashboard - Real Implementation

This document shows OrchestKit's actual monitoring setup including metrics, dashboards, and alerting rules.

## Overview

**OrchestKit Monitoring Stack:**
- **Logs**: Structlog (JSON) → Loki
- **Metrics**: Prometheus (RED + business metrics)
- **Traces**: Langfuse (LLM observability)
- **Dashboards**: Grafana
- **Alerts**: Prometheus Alertmanager → Slack

**Key Metrics:**
- LLM costs: $35k/year → $2-5k/year (95% reduction via caching)
- Retrieval pass rate: 91.6% (target: >90%)
- Quality gate pass rate: 85% (target: >80%)
- Hybrid search latency: 5ms (HNSW index)

## Dashboard Structure

### 1. Service Overview Dashboard

**Top Row - Golden Signals:**
```
┌──────────────┬──────────────┬──────────────┬──────────────┐
│  Latency     │  Traffic     │  Errors      │  Saturation  │
│  p50: 245ms  │  12.5 req/s  │  0.3% (5xx)  │  CPU: 45%    │
│  p95: 680ms  │  (stable)    │  (good)      │  Mem: 62%    │
│  p99: 1.2s   │              │              │  (healthy)   │
└──────────────┴──────────────┴──────────────┴──────────────┘
```

**Prometheus Queries:**

```text
# p95 latency
histogram_quantile(0.95,
  rate(http_request_duration_seconds_bucket[5m])
)

# Request rate
sum(rate(http_requests_total[5m]))

# Error rate (5xx)
sum(rate(http_requests_total{status=~"5.."}[5m])) /
sum(rate(http_requests_total[5m]))

# CPU saturation
avg(rate(process_cpu_seconds_total[5m])) * 100
```

### 2. LLM Observability Dashboard

**Metrics Tracked:**
- Cost per model (Claude, Gemini, Voyage)
- Token usage (input/output)
- Cache hit rates (L1: Prompt Cache, L2: Semantic Cache)
- LLM latency distribution

**Cost Breakdown Panel:**
```text
# Total cost per day by model
sum(increase(llm_cost_dollars_total[1d])) by (model)

# Cost per operation
sum(increase(llm_cost_dollars_total[1h])) by (operation)
```

**Example Results:**
| Model | Daily Cost | Monthly (Projected) |
|-------|------------|---------------------|
| claude-sonnet-5 | $5.20 | $156 |
| gemini-3-flash | $1.80 | $54 |
| voyage-code-2 | $0.40 | $12 |
| **Total** | **$7.40** | **$222** |

**Cache Performance Panel:**
```text
# Cache hit rate
sum(rate(cache_operations_total{result="hit"}[5m])) /
sum(rate(cache_operations_total[5m]))

# Cost savings from cache (estimated)
sum(rate(cache_operations_total{result="hit"}[5m])) *
avg_over_time(llm_cost_dollars_total[1h])
```

**Results:**
| Cache Level | Hit Rate | Daily Savings |
|-------------|----------|---------------|
| L1 (Prompt Cache) | 90% | $90 |
| L2 (Semantic Cache) | 75% | $21 |
| **Total Savings** | - | **$111/day** |

### 3. Quality Metrics Dashboard

**Panels:**
1. Quality gate pass rate (target: >80%)
2. G-Eval scores by criterion (completeness, accuracy, coherence, depth)
3. Failed analyses count
4. Quality score distribution

**Quality Gate Pass Rate:**
```text
# Pass rate over last 24h
sum(rate(quality_gate_passed_total[24h])) /
sum(rate(quality_gate_total[24h]))
```

**G-Eval Scores (from Langfuse):**
```sql
-- Track quality trends
SELECT
    DATE(timestamp) as date,
    AVG(value) FILTER (WHERE name = 'quality_completeness') as completeness,
    AVG(value) FILTER (WHERE name = 'quality_accuracy') as accuracy,
    AVG(value) FILTER (WHERE name = 'quality_coherence') as coherence,
    AVG(value) FILTER (WHERE name = 'quality_depth') as depth
FROM langfuse.scores
WHERE timestamp > NOW() - INTERVAL '7 days'
GROUP BY DATE(timestamp);
```

**Example Results:**
| Date | Completeness | Accuracy | Coherence | Depth | Overall |
|------|--------------|----------|-----------|-------|---------|
| 2025-01-20 | 0.85 | 0.92 | 0.88 | 0.78 | 0.86 |
| 2025-01-21 | 0.83 | 0.91 | 0.87 | 0.76 | 0.84 |

### 4. Database Performance Dashboard

**Panels:**
1. Query latency (p50/p95/p99)
2. Connection pool usage
3. Slow queries (>500ms)
4. Cache hit ratio

**Query Latency:**
```text
# p95 query latency
histogram_quantile(0.95,
  rate(db_query_duration_seconds_bucket[5m])
) by (query_type)
```

**Connection Pool:**
```text
# Active connections
db_connections_active

# Connection pool saturation
db_connections_active / db_connections_max
```

**Real Metrics:**
| Metric | Value | Target |
|--------|-------|--------|
| p50 query latency | 8ms | &lt;100ms |
| p95 query latency | 45ms | &lt;500ms |
| Active connections | 12 | &lt;20 |
| Pool saturation | 60% | &lt;80% |

### 5. Retrieval Quality Dashboard

**Metrics from Golden Dataset (98 analyses, 415 chunks):**

**Pass Rate:**
```text
# Retrieval pass rate (expected chunk in top-k)
sum(retrieval_pass_total) / sum(retrieval_total)
```

**Results:** 186/203 queries passed = **91.6% pass rate** (target: >90%)

**MRR by Difficulty:**
```sql
-- Mean Reciprocal Rank by query difficulty
SELECT
    difficulty,
    COUNT(*) as queries,
    AVG(mrr) as avg_mrr
FROM retrieval_evaluation
GROUP BY difficulty;
```

**Results:**
| Difficulty | Queries | MRR | Pass Rate |
|------------|---------|-----|-----------|
| Easy | 78 | 0.892 | 96.2% |
| Medium | 89 | 0.745 | 91.0% |
| Hard | 36 | 0.686 | 83.3% |
| **Overall** | **203** | **0.777** | **91.6%** |

**Search Latency:**
```text
# Hybrid search latency (HNSW + BM25 RRF)
histogram_quantile(0.95,
  rate(search_duration_seconds_bucket[5m])
)
```

**Results:**
| Operation | p50 | p95 | p99 |
|-----------|-----|-----|-----|
| Vector search (HNSW) | 3ms | 5ms | 8ms |
| BM25 search | 4ms | 7ms | 12ms |
| RRF fusion | 1ms | 2ms | 3ms |
| **Total hybrid search** | **8ms** | **14ms** | **23ms** |

**Comparison to IVFFlat:**
- HNSW: 5ms
- IVFFlat: 85ms
- **Speedup: 17x faster**

## Structured Logging Examples

### Log Format

**OrchestKit uses structlog with JSON output:**
```json
{
  "event": "supervisor_routing",
  "level": "info",
  "timestamp": "2025-01-21T10:30:45.123Z",
  "correlation_id": "abc-123-def",
  "analysis_id": "550e8400-e29b-41d4-a716-446655440000",
  "workflow_step": "supervisor",
  "agent": "tech_comparator",
  "remaining_agents": 7,
  "content_length": 45823,
  "logger": "app.workflows.supervisor"
}
```

### Key Log Events

**1. Analysis Started:**
```json
{
  "event": "analysis_started",
  "level": "info",
  "analysis_id": "550e8400-...",
  "url": "https://example.com/article",
  "content_type": "article"
}
```

**2. Agent Execution:**
```json
{
  "event": "agent_execution_started",
  "level": "info",
  "agent_type": "security_auditor",
  "correlation_id": "abc-123-def",
  "analysis_id": "550e8400-..."
}
```

**3. LLM Call:**
```json
{
  "event": "llm_call_completed",
  "level": "info",
  "model": "claude-sonnet-5",
  "operation": "security_audit",
  "input_tokens": 1800,
  "output_tokens": 1200,
  "cost_dollars": 0.021,
  "duration_seconds": 2.3,
  "cache_hit": false
}
```

**4. Quality Gate:**
```json
{
  "event": "quality_gate_passed",
  "level": "info",
  "analysis_id": "550e8400-...",
  "quality_scores": {
    "completeness": 0.85,
    "accuracy": 0.92,
    "coherence": 0.88,
    "depth": 0.78
  },
  "overall_quality": 0.86,
  "passed": true
}
```

**5. Error Logging:**
```json
{
  "event": "analysis_failed",
  "level": "error",
  "analysis_id": "550e8400-...",
  "error_type": "ValidationError",
  "error_message": "Quality gate failed: depth score too low",
  "quality_scores": {
    "depth": 0.45
  },
  "traceback": "...",
  "correlation_id": "abc-123-def"
}
```

### Loki Queries (LogQL)

**Find all errors in last hour:**
```text
{app="orchestkit-backend"} |= "ERROR" | json
```

**Count errors by endpoint:**
```text
sum by (endpoint) (
  count_over_time({app="orchestkit-backend"} |= "ERROR" [5m])
)
```

**Search for specific analysis:**
```text
{app="orchestkit-backend"}
| json
| analysis_id="550e8400-e29b-41d4-a716-446655440000"
```

**p95 LLM latency from logs:**
```text
quantile_over_time(0.95,
  {app="orchestkit-backend"}
  | json
  | event="llm_call_completed"
  | unwrap duration_seconds [5m]
)
```

## Alerting Rules

### 1. Service Availability

**File:** `monitoring/prometheus/alerts/service.yml`

```yaml
groups:
- name: service-health
  interval: 30s
  rules:
  - alert: ServiceDown
    expr: up == 0
    for: 1m
    labels:
      severity: critical
      team: platform
    annotations:
      summary: "Service {{ $labels.job }} is down"
      description: "{{ $labels.instance }} has been down for 1 minute"
      runbook_url: "https://wiki.orchestkit.dev/runbooks/service-down"

  - alert: HighErrorRate
    expr: |
      sum(rate(http_requests_total{status=~"5.."}[5m])) /
      sum(rate(http_requests_total[5m])) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "High error rate detected"
      description: "Error rate is {{ $value | humanizePercentage }} (threshold: 5%)"
```

### 2. LLM Cost Alerts

**File:** `monitoring/prometheus/alerts/llm-cost.yml`

```yaml
groups:
- name: llm-costs
  interval: 1h
  rules:
  - alert: DailyCostExceeded
    expr: |
      sum(increase(llm_cost_dollars_total[24h])) > 20
    labels:
      severity: high
      team: ai-ml
    annotations:
      summary: "Daily LLM cost exceeded $20"
      description: "Current daily cost: ${{ $value }}"

  - alert: UnexpectedCostSpike
    expr: |
      sum(rate(llm_cost_dollars_total[1h])) >
      sum(rate(llm_cost_dollars_total[1h] offset 24h)) * 2
    for: 2h
    labels:
      severity: high
    annotations:
      summary: "LLM cost spike detected"
      description: "Current hourly cost is 2x yesterday's average"
```

### 3. Quality Degradation

**File:** `monitoring/prometheus/alerts/quality.yml`

```yaml
groups:
- name: quality-metrics
  interval: 5m
  rules:
  - alert: LowQualityGatePassRate
    expr: |
      sum(rate(quality_gate_passed_total[1h])) /
      sum(rate(quality_gate_total[1h])) < 0.80
    for: 30m
    labels:
      severity: high
      team: ml
    annotations:
      summary: "Quality gate pass rate below 80%"
      description: "Current pass rate: {{ $value | humanizePercentage }}"

  - alert: CacheHitRateDegraded
    expr: |
      sum(rate(cache_operations_total{result="hit"}[30m])) /
      sum(rate(cache_operations_total[30m])) < 0.70
    for: 1h
    labels:
      severity: medium
    annotations:
      summary: "Cache hit rate below 70%"
      description: "Cache performance degraded: {{ $value | humanizePercentage }}"
```

### 4. Database Performance

**File:** `monitoring/prometheus/alerts/database.yml`

```yaml
groups:
- name: database-performance
  interval: 1m
  rules:
  - alert: SlowQueries
    expr: |
      histogram_quantile(0.95,
        rate(db_query_duration_seconds_bucket[5m])
      ) > 0.5
    for: 10m
    labels:
      severity: high
    annotations:
      summary: "p95 query latency exceeded 500ms"
      description: "Current p95: {{ $value }}s"

  - alert: ConnectionPoolExhausted
    expr: db_connections_active / db_connections_max > 0.9
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Database connection pool near capacity"
      description: "{{ $value | humanizePercentage }} of connections in use"
```

## Alert Routing & Escalation

**File:** `monitoring/alertmanager/config.yml`

```yaml
route:
  group_by: ['alertname', 'cluster', 'service']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: slack-default

  routes:
  # Critical alerts → Slack + PagerDuty
  - match:
      severity: critical
    receiver: pagerduty-critical
    continue: true  # Also send to Slack

  # High severity → Slack
  - match:
      severity: high
    receiver: slack-high

  # Medium/low → Slack (throttled)
  - match_re:
      severity: (medium|low)
    receiver: slack-low
    group_interval: 1h

receivers:
- name: slack-default
  slack_configs:
  - api_url: <slack_webhook_url>
    channel: '#alerts'
    title: '{{ .GroupLabels.alertname }}'
    text: '{{ range .Alerts }}{{ .Annotations.summary }}\n{{ end }}'

- name: pagerduty-critical
  pagerduty_configs:
  - service_key: <pagerduty_service_key>
```

## Health Check Endpoints

### 1. Liveness Probe

**Endpoint:** `GET /health`
**Purpose:** Is the application running?

```python
@app.get("/health")
async def health_check():
    """Basic liveness check."""
    return {"status": "healthy"}
```

### 2. Readiness Probe

**Endpoint:** `GET /ready`
**Purpose:** Is the application ready to serve traffic?

```python
@app.get("/ready")
async def readiness_check():
    """Check if app can handle requests."""

    checks = {}

    # Database check
    try:
        await db.execute("SELECT 1")
        checks["database"] = {"status": "pass", "latency_ms": 5}
    except Exception as e:
        checks["database"] = {"status": "fail", "error": str(e)}

    # Redis check
    try:
        await redis.ping()
        checks["redis"] = {"status": "pass", "latency_ms": 2}
    except Exception as e:
        checks["redis"] = {"status": "fail", "error": str(e)}

    # Overall status
    all_healthy = all(c["status"] == "pass" for c in checks.values())
    status = "healthy" if all_healthy else "degraded"

    return {
        "status": status,
        "checks": checks,
        "version": "1.0.0",
        "uptime": int(time.time() - app.start_time)
    }
```

**Response:**
```json
{
  "status": "healthy",
  "checks": {
    "database": {"status": "pass", "latency_ms": 5},
    "redis": {"status": "pass", "latency_ms": 2}
  },
  "version": "1.0.0",
  "uptime": 3600
}
```

## References

- Template: `../scripts/structured-logging.ts`
- Template: `../scripts/prometheus-metrics.ts`
- Template: `../scripts/alerting-rules.yml`
- [OrchestKit Redis Connection](../../../../backend/app/shared/services/cache/redis_connection.py)
- [OrchestKit Quality Initiative](../../../../docs/QUALITY_INITIATIVE_FIXES.md)