Standard OTLP/HTTP trace receiver

README.md

@@ -224,6 +224,7 @@ Base URL: `http://localhost:8000`
 | `GET`    | `/v1/alerts/events`               | Alert events (`?state=open`)      |
 | `POST`   | `/v1/alerts/events/{id}/ack`      | Mark an alert as seen             |
 | `POST`   | `/v1/alerts/evaluate`             | Evaluate all rules now            |
+| `POST`   | `/v1/otlp/v1/traces`              | Standard OTLP/HTTP trace receiver |
 | `GET`    | `/v1/datasets`                    | List test-case datasets           |
 | `POST`   | `/v1/datasets`                    | Create a dataset                  |
 | `GET`    | `/v1/datasets/{id}`               | A dataset with its cases          |
@@ -280,6 +281,13 @@ Scores of the previous rubric are replaced when the turn is re-scored. A backgro
 backfills history. Evaluations are idempotent (the id derives from the trace and the rubric). The
 Evaluations page shows health by model, both axes, the grade and the per-criterion breakdown.
 
+### Standard OpenTelemetry (OTLP) ingestion
+
+Point any stock OpenTelemetry exporter at `OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:8100/v1/otlp`
+(protobuf or JSON, optionally gzipped) and its traces appear without the Observer SDK. GenAI
+conventions map to Observer span types, tokens and tools. Content-like attributes are dropped by
+default (`OBSERVATORY_OTLP_CAPTURE=metadata`). Mapping, privacy and limits: [docs/otlp.md](docs/otlp.md).
+
 ### Read-only MCP server
 
 `observer-mcp` lets Claude Code, Codex or any MCP client investigate Observer without screenshots:

backend/app/api/manitos_ingest.py

@@ -312,6 +312,14 @@ async def ingest_manitos_trace(
     """Upsert a complete integration trace with retry-safe semantics."""
 
     del request  # Required by SlowAPI's decorator contract.
+    return await apply_envelope(session, payload)
+
+
+async def apply_envelope(
+    session: AsyncSession, payload: ManitOSIngestRequest
+) -> ManitOSIngestResponse:
+    """The ingestion itself (also used by the OTLP receiver, which builds envelopes)."""
+
     digest = _payload_hash(payload)
     receipt = await _existing_receipt(
         session,

backend/app/api/otlp.py

@@ -0,0 +1,109 @@
+"""OTLP/HTTP trace receiver (``POST /v1/otlp/v1/traces``).
+
+Point a stock OpenTelemetry exporter at it:
+``OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:8100/v1/otlp`` (the exporter appends ``/v1/traces``),
+plus ``OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer <key>`` when a key is required.
+"""
+
+from __future__ import annotations
+
+import os
+from datetime import timezone
+from typing import Any
+
+from fastapi import APIRouter, Depends, Request, Response
+from pydantic import ValidationError
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.manitos_ingest import ManitOSIngestRequest, apply_envelope
+from app.db import get_session
+from app.models.trace import Span, Trace
+from app.otlp import (
+    MAX_BODY_BYTES,
+    OtlpError,
+    decompress,
+    request_from_json,
+    request_from_protobuf,
+    response_body,
+    translate,
+)
+from app.ratelimit import limiter
+
+router = APIRouter()
+
+_OTLP_LIMIT = os.getenv("RATE_LIMIT_OTLP", "600/minute")
+
+
+def _plain(status: int, message: str) -> Response:
+    return Response(content=message, status_code=status, media_type="text/plain")
+
+
+async def _link_and_widen(session: AsyncSession, envelope: dict[str, Any]) -> None:
+    """Spans of one trace can arrive in several batches: keep the trace's full time range and link
+    a span to a parent stored by an earlier batch (instead of failing or orphaning it)."""
+    trace_id = envelope["trace"]["id"]
+    external = {s["_external_parent"] for s in envelope["spans"] if s.get("_external_parent")}
+    stored: set[str] = set()
+    if external:
+        rows = await session.execute(
+            select(Span.id).where(Span.id.in_(external), Span.trace_id == trace_id)
+        )
+        stored = {str(i) for i in rows.scalars()}
+    for span in envelope["spans"]:
+        parent = span.pop("_external_parent", None)
+        if parent in stored:
+            span["parent_id"] = parent
+    existing = await session.get(Trace, trace_id)
+    if existing is not None and existing.start_time is not None:
+        trace = envelope["trace"]
+        start = existing.start_time
+        if start.tzinfo is None:
+            start = start.replace(tzinfo=timezone.utc)
+        trace["start_time"] = min(trace["start_time"], start.timestamp())
+        ends = [trace["end_time"]] if trace.get("end_time") else []
+        if existing.end_time is not None:
+            end = existing.end_time
+            if end.tzinfo is None:
+                end = end.replace(tzinfo=timezone.utc)
+            ends.append(end.timestamp())
+        trace["end_time"] = max(ends) if ends else None
+        trace["name"] = existing.name if existing.name else trace["name"]
+
+
+@router.post("/v1/traces")
+@limiter.limit(_OTLP_LIMIT)
+async def receive_traces(
+    request: Request, session: AsyncSession = Depends(get_session)
+) -> Response:
+    declared = request.headers.get("content-length")
+    if declared and declared.isdigit() and int(declared) > MAX_BODY_BYTES:
+        return _plain(413, "request too large")
+    content_type = request.headers.get("content-type", "").split(";")[0].strip().lower()
+    protobuf = content_type == "application/x-protobuf"
+    if not protobuf and content_type != "application/json":
+        return _plain(415, "use application/x-protobuf or application/json")
+    try:
+        body = decompress(await request.body(), request.headers.get("content-encoding"))
+        data = request_from_protobuf(body) if protobuf else request_from_json(body)
+        translated = translate(data, project_id=request.headers.get("x-observer-project"))
+    except OtlpError as exc:
+        return _plain(exc.status, str(exc))
+
+    rejected = translated.rejected
+    errors = list(translated.errors)
+    for envelope in translated.envelopes:
+        count = len(envelope["spans"])
+        try:
+            await _link_and_widen(session, envelope)
+            payload = ManitOSIngestRequest.model_validate(envelope)
+            async with session.begin_nested():  # a rejected trace leaves no receipt behind
+                await apply_envelope(session, payload)
+        except ValidationError as exc:
+            rejected += count
+            errors.append(f"trace rejected: {exc.errors()[0]['msg']}"[:200])
+        except Exception as exc:  # HTTPException from the shared ingestion: conflicts
+            rejected += count
+            errors.append(f"trace rejected: {getattr(exc, 'detail', exc.__class__.__name__)}"[:200])
+    content, media = response_body(rejected, "; ".join(errors[:3]), protobuf=protobuf)
+    return Response(content=content, media_type=media)

backend/app/main.py

@@ -15,6 +15,7 @@
 from app.api import analytics, evaluations, manitos_ingest, providers, traces
 from app.api import dataset_runs as dataset_runs_api
 from app.api import datasets as datasets_api
+from app.api import otlp as otlp_api
 from app.auth import require_api_key
 from app.db.postgres import engine
 from app.ratelimit import limiter
@@ -113,6 +114,12 @@ def _rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONRespons
     tags=["datasets"],
     dependencies=[_app_auth],
 )
+app.include_router(
+    otlp_api.router,
+    prefix="/v1/otlp",
+    tags=["otlp"],
+    dependencies=[_app_auth],
+)
 app.include_router(
     datasets_api.router,
     prefix="/v1/datasets",