feat(otel): OTLP/HTTP trace ingress

CHANGELOG.md

@@ -10,6 +10,18 @@ surface is governed by [`COMPATIBILITY.md`](COMPATIBILITY.md).
 ## [Unreleased]
 
 ### Added
+- **OTLP trace ingress.** RiskKernel can now act as an OTLP/HTTP trace endpoint
+  (`POST /v1/traces`), the consume side of the OpenTelemetry surface — point any
+  exporter at it (`OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:7070`) and the GenAI
+  model calls an app already traces (OpenLLMetry, the OpenAI Agents SDK, the Vercel
+  AI SDK) show up against governed runs with tokens and cost metered into the
+  ledger. Correlates by `riskkernel.run.id` (on the span or its resource); accepts
+  protobuf and JSON and replies with an OTLP `ExportTraceServiceResponse` (a span
+  without a run id is reported as a rejected partial-success span). Scope is observe
+  + meter — a consumed call is recorded (and marks the run halted if it crosses the
+  budget) but isn't blocked after the fact. Off by default; enable with
+  `RISKKERNEL_OTEL_INGRESS_ENABLED` and authenticated like the rest of the API. See
+  [`docs/OTLP_INGRESS.md`](docs/OTLP_INGRESS.md).
 - **CrewAI adapter (Python SDK).** `from riskkernel.adapters.crewai import
   RiskKernelStepCallback` — a `step_callback` you wire onto a CrewAI `Agent` or the
   whole `Crew` to bind it to a governed run with no other code change. One agent step

api/v1/otel-genai.md

@@ -69,7 +69,26 @@ conventions don't model. Names are stable per COMPATIBILITY.md.
 
 ## Consumption (ingress)
 
-When acting as an OTLP endpoint, RiskKernel reads incoming `gen_ai.usage.*` to feed
-the cost ledger and `gen_ai.request.model` / `gen_ai.system` to attribute spend,
-correlating by `riskkernel.run.id` when present (or a configured trace→run mapping
-otherwise). This lets RiskKernel govern apps it did not directly instrument.
+RiskKernel can act as an OTLP/HTTP trace endpoint at **`POST /v1/traces`** — the
+standard OTLP path, so any exporter targets it with one env var
+(`OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:7070`). It accepts both protobuf
+(`application/x-protobuf`, the OTLP default) and JSON (`application/json`) and
+replies with an OTLP `ExportTraceServiceResponse` in the same encoding.
+
+For each span carrying token usage (`gen_ai.usage.input_tokens` /
+`gen_ai.usage.output_tokens`), it reads `gen_ai.response.model` (falling back to
+`gen_ai.request.model`) and `gen_ai.system` and meters the call's tokens and cost
+into the ledger, correlating to a governed run by `riskkernel.run.id` (taken from
+the span, falling back to the resource). A span without a run id is observed and
+reported as a rejected span in the OTLP partial-success response, but not metered.
+Spans without usage (tool calls, retrieval, framework spans) are ignored.
+
+This lets RiskKernel make spend visible for apps it did not directly proxy. Scope
+is **observe + meter**: a consumed span records against the run's ledger (and marks
+the run halted if its budget is crossed) but does not retroactively block a call
+that already happened — governing consumed spans is a separate, future step.
+
+The receiver is **off by default**; no listener is mounted unless
+`RISKKERNEL_OTEL_INGRESS_ENABLED` is set. It is authenticated like the rest of the
+API, so an exporter must carry the bearer token via
+`OTEL_EXPORTER_OTLP_HEADERS=authorization=Bearer <token>` when a token is set.

cmd/riskkernel/main.go

@@ -140,7 +140,7 @@ func runServe(_ []string) error {
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 
-	srv := httpapi.New(cfg, deps.Gateway, deps.Runs, deps.Approvals, deps.Slack, deps.MCP, deps.Memory, deps.Log)
+	srv := httpapi.New(cfg, deps.Gateway, deps.Runs, deps.Approvals, deps.Slack, deps.MCP, deps.Memory, deps.Ingress, deps.Log)
 	addr := fmt.Sprintf(":%d", cfg.Port)
 	return srv.Serve(ctx, addr)
 }

docs/OTLP_INGRESS.md

@@ -0,0 +1,73 @@
+# OTLP trace ingress
+
+RiskKernel emits OpenTelemetry GenAI spans to your observability backend (the
+export side of Surface 3). The other half is **ingress**: RiskKernel can also *be*
+an OTLP endpoint that consumes GenAI spans from apps already instrumented with
+OpenLLMetry, the OpenAI Agents SDK, or the Vercel AI SDK — so it can make spend
+visible for agents it never directly proxied.
+
+Point an existing app's OTLP exporter at RiskKernel and the model calls it already
+traces show up against governed runs, with tokens and cost metered into the same
+ledger the proxy uses.
+
+## Enabling it
+
+The receiver is **off by default** — no listener is mounted unless you turn it on:
+
+```bash
+RISKKERNEL_OTEL_INGRESS_ENABLED=true riskkernel serve
+```
+
+On the app side, point any OpenTelemetry exporter at the daemon. The endpoint is
+the standard OTLP/HTTP traces path, so this is the usual one env var:
+
+```bash
+export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:7070
+# when RISKKERNEL_API_TOKEN is set, the exporter must carry it:
+export OTEL_EXPORTER_OTLP_HEADERS=authorization=Bearer $RISKKERNEL_API_TOKEN
+```
+
+The exporter appends `/v1/traces` itself. Both OTLP encodings work: protobuf
+(`application/x-protobuf`, the SDK default) and JSON (`application/json`).
+
+## Correlating spans to a run
+
+Set `riskkernel.run.id` on your spans (or once on the OTel resource) so consumed
+calls are attributed to a governed run:
+
+```python
+span.set_attribute("riskkernel.run.id", run_id)
+```
+
+The run is created lazily under the default budget if it doesn't exist yet, exactly
+like the proxy's run-id header. A GenAI usage span with **no** run id is observed
+and reported back as a rejected span in the OTLP partial-success response, but is
+not metered.
+
+## What gets metered
+
+For each span carrying token usage, RiskKernel reads the pinned GenAI attributes
+(see [`api/v1/otel-genai.md`](../api/v1/otel-genai.md)):
+
+| Attribute | Used for |
+|---|---|
+| `gen_ai.usage.input_tokens` / `gen_ai.usage.output_tokens` | Token counts and cost (priced via the same table the proxy uses). |
+| `gen_ai.response.model` (falls back to `gen_ai.request.model`) | The model, for pricing and attribution. |
+| `gen_ai.system` | The provider, recorded on the ledger entry. |
+| `riskkernel.run.id` | The run to attribute the call to. |
+
+Token counts emitted as a double or numeric string (some instrumenters do this)
+are accepted rather than dropped to zero. Spans without usage — tool calls,
+retrieval, framework spans — are ignored.
+
+## Scope: observe + meter
+
+Ingested calls already happened in the other app, so the budget can't block them
+before the fact. RiskKernel **observes and meters** them: the call is recorded
+against the run's ledger, and if the recorded usage crosses the run's budget the
+run is marked halted (visible in `riskkernel runs list`, the audit export, and
+`GET /v1/runs/{id}`). Actively governing consumed spans — gating or refusing the
+*next* call on a run that an external app is driving — is a separate, future step.
+
+For deterministic, before-the-fact enforcement, route calls through the proxy
+(Surface 1) or the SDK (Surface 2) instead.

go.mod

@@ -10,6 +10,8 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.44.0
 	go.opentelemetry.io/otel/sdk v1.44.0
 	go.opentelemetry.io/otel/trace v1.44.0
+	go.opentelemetry.io/proto/otlp v1.10.0
+	google.golang.org/protobuf v1.36.11
 	gopkg.in/yaml.v3 v3.0.1
 	modernc.org/sqlite v1.52.0
 )
@@ -29,7 +31,6 @@ require (
 	github.com/sethvargo/go-retry v0.3.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/otel/metric v1.44.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.55.0 // indirect
 	golang.org/x/sync v0.20.0 // indirect
@@ -38,7 +39,6 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/grpc v1.81.1 // indirect
-	google.golang.org/protobuf v1.36.11 // indirect
 	modernc.org/libc v1.72.3 // indirect
 	modernc.org/mathutil v1.7.1 // indirect
 	modernc.org/memory v1.11.0 // indirect

internal/app/bootstrap.go

@@ -40,6 +40,7 @@ type Deps struct {
 	Slack     *approval.SlackNotifier // nil when the Slack channel isn't configured
 	MCP       *mcp.Gateway            // nil when no upstream is configured
 	Memory    *memory.Reader
+	Ingress   *otel.Ingress // nil unless the OTLP trace receiver is enabled
 }
 
 // Close releases dependencies that hold resources (the tracer's buffered spans,
@@ -147,6 +148,13 @@ func Build(cfg *config.Config) (*Deps, error) {
 		log.Warn("RISKKERNEL_MEMORY_EMBEDDINGS is set but embeddings are not implemented in v0.1; using deterministic keyword search")
 	}
 
+	// OTLP trace receiver (Surface 3, consume side) — off unless explicitly enabled.
+	var ingress *otel.Ingress
+	if cfg.OTel.IngressEnabled {
+		ingress = otel.NewIngress(mgr, prices, log)
+		log.Info("otlp trace ingress enabled", "path", "POST /v1/traces")
+	}
+
 	return &Deps{
 		Config:    cfg,
 		Log:       log,
@@ -160,6 +168,7 @@ func Build(cfg *config.Config) (*Deps, error) {
 		Slack:     slackNotifier,
 		MCP:       mcpGW,
 		Memory:    memReader,
+		Ingress:   ingress,
 	}, nil
 }
 

internal/config/config.go

@@ -165,6 +165,12 @@ type OTelConfig struct {
 	// OTEL_EXPORTER_OTLP_HEADERS, as a comma-separated list of key=value pairs.
 	// Carries secrets — never logged.
 	Headers map[string]string
+
+	// IngressEnabled turns on the OTLP/HTTP trace receiver (Surface 3, consume
+	// side): RiskKernel becomes an OTLP endpoint at POST /v1/traces that meters
+	// GenAI spans from apps it didn't directly proxy. Off by default — no listener
+	// is mounted unless RISKKERNEL_OTEL_INGRESS_ENABLED is set truthy.
+	IngressEnabled bool
 }
 
 // BudgetConfig holds raw budget values (no governor dependency here so config
@@ -292,11 +298,12 @@ func loadOTel() OTelConfig {
 		headers = os.Getenv("OTEL_EXPORTER_OTLP_HEADERS")
 	}
 	return OTelConfig{
-		Endpoint:    endpoint,
-		Protocol:    getenvDefault("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc"),
-		Insecure:    insecure,
-		ServiceName: getenvDefault("OTEL_SERVICE_NAME", "riskkernel"),
-		Headers:     parseOTLPHeaders(headers),
+		Endpoint:       endpoint,
+		Protocol:       getenvDefault("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc"),
+		Insecure:       insecure,
+		ServiceName:    getenvDefault("OTEL_SERVICE_NAME", "riskkernel"),
+		Headers:        parseOTLPHeaders(headers),
+		IngressEnabled: envBoolDefault("RISKKERNEL_OTEL_INGRESS_ENABLED", false),
 	}
 }
 

internal/httpapi/memory_test.go

@@ -34,7 +34,7 @@ func newMemoryServer(t *testing.T) http.Handler {
 	t.Cleanup(func() { _ = store.Close() })
 	log := slog.New(slog.NewTextHandler(io.Discard, nil))
 	mgr := runs.NewManager(governor.Budget{}).WithStore(store, log)
-	srv := New(&config.Config{}, nil, mgr, nil, nil, nil, memory.NewReader(memDir), log)
+	srv := New(&config.Config{}, nil, mgr, nil, nil, nil, memory.NewReader(memDir), nil, log)
 	return srv.Handler()
 }
 

internal/httpapi/server.go

@@ -19,6 +19,7 @@ import (
 	"github.com/prashar32/riskkernel/internal/httpx"
 	"github.com/prashar32/riskkernel/internal/mcp"
 	"github.com/prashar32/riskkernel/internal/memory"
+	"github.com/prashar32/riskkernel/internal/otel"
 	"github.com/prashar32/riskkernel/internal/runs"
 	"github.com/prashar32/riskkernel/internal/storage"
 	"github.com/prashar32/riskkernel/internal/version"
@@ -33,13 +34,14 @@ type Server struct {
 	slack     *approval.SlackNotifier // nil when the Slack channel isn't configured
 	mcp       *mcp.Gateway
 	memory    *memory.Reader
+	ingress   *otel.Ingress // nil unless the OTLP trace receiver is enabled
 	log       *slog.Logger
 }
 
 // New constructs a Server.
 func New(cfg *config.Config, gw *gateway.Gateway, mgr *runs.Manager, gate *approval.Gate,
-	slack *approval.SlackNotifier, mcpGW *mcp.Gateway, mem *memory.Reader, log *slog.Logger) *Server {
-	return &Server{cfg: cfg, gateway: gw, runs: mgr, approvals: gate, slack: slack, mcp: mcpGW, memory: mem, log: log}
+	slack *approval.SlackNotifier, mcpGW *mcp.Gateway, mem *memory.Reader, ingress *otel.Ingress, log *slog.Logger) *Server {
+	return &Server{cfg: cfg, gateway: gw, runs: mgr, approvals: gate, slack: slack, mcp: mcpGW, memory: mem, ingress: ingress, log: log}
 }
 
 // Handler returns the root HTTP handler with all routes mounted.
@@ -107,6 +109,14 @@ func (s *Server) Handler() http.Handler {
 		mux.HandleFunc("GET /metrics", s.requireAuth(s.handleMetrics))
 	}
 
+	// Surface 3 (consume side) — the OTLP/HTTP trace receiver. Off unless enabled;
+	// authenticated like the rest of the API (OTLP exporters carry the bearer token
+	// via OTEL_EXPORTER_OTLP_HEADERS). Standard OTLP path so any exporter can target
+	// it with one env var.
+	if s.ingress != nil {
+		mux.HandleFunc("POST /v1/traces", s.requireAuth(s.ingress.HandleTraces))
+	}
+
 	return s.recoverer(mux)
 }
 

internal/httpapi/server_test.go

@@ -30,7 +30,7 @@ func newTestServer(t *testing.T, token string) (*Server, *runs.Manager, *approva
 	log := slog.New(slog.NewTextHandler(io.Discard, nil))
 	mgr := runs.NewManager(governor.Budget{Tokens: 100000}).WithStore(store, log)
 	gate := approval.NewGate(store, approval.Policy{DefaultSafe: true}, nil, log)
-	srv := New(&config.Config{APIToken: token}, nil, mgr, gate, nil, nil, memory.NewReader(t.TempDir()), log)
+	srv := New(&config.Config{APIToken: token}, nil, mgr, gate, nil, nil, memory.NewReader(t.TempDir()), nil, log)
 	return srv, mgr, gate
 }
 

internal/httpapi/slack_test.go

@@ -40,7 +40,7 @@ func newSlackTestServer(t *testing.T, secret string) (*Server, *runs.Manager, *a
 	mgr := runs.NewManager(governor.Budget{Tokens: 100000}).WithStore(store, log)
 	gate := approval.NewGate(store, approval.Policy{DefaultSafe: true}, nil, log)
 	slack := approval.NewSlackNotifier("xoxb-test", "C123", secret, log)
-	srv := New(&config.Config{}, nil, mgr, gate, slack, nil, memory.NewReader(t.TempDir()), log)
+	srv := New(&config.Config{}, nil, mgr, gate, slack, nil, memory.NewReader(t.TempDir()), nil, log)
 	return srv, mgr, gate
 }