feat: route normalized OTLP signals to project destinations

server/cmd/gram/streams.go

@@ -516,7 +516,6 @@ func newStreamsCommand() *cli.Command {
 				replicaDB,
 				encryptionClient,
 				guardianPolicy,
-				featureFlags,
 			)
 
 			metricRelayHandler := otelsvc.NewMetricRelayHandler(

server/internal/dataexports/queries.sql

@@ -1,5 +1,25 @@
 -- Every query pins both organization_id and project_id. Resource UUIDs and
 -- tenant-pinned foreign keys are integrity controls, not authorization bounds.
+
+-- Resolve the active OTEL destination for one project data source.
+-- name: GetActiveOtelRouteDestination :one
+SELECT
+  destination.endpoint_url,
+  destination.headers_encrypted,
+  COALESCE(destination.sensitive_data, 'exclude') = 'include' AS include_sensitive_data
+FROM data_export_routes AS route
+JOIN otel_destinations AS destination
+  ON destination.organization_id = route.organization_id
+ AND destination.project_id = route.project_id
+ AND destination.id = route.otel_destination_id
+WHERE route.organization_id = @organization_id
+  AND route.project_id = @project_id
+  AND route.data_source = @data_source
+  AND route.enabled IS TRUE
+  AND route.deleted IS FALSE
+  AND route.otel_destination_id IS NOT NULL
+  AND destination.deleted IS FALSE;
+
 -- List active destinations in stable creation order for the management API.
 -- name: ListOtelDestinations :many
 SELECT *

server/internal/feature/flags.go

@@ -118,13 +118,6 @@ const (
 	// plugins.canaryHooksOrgSlugs), independent of this flag, so a PostHog outage
 	// can't strand it on stale hooks.
 	FlagHooksRollout Flag = "hooks-rollout"
-
-	// FlagOTELLogCustomerRelay controls which organizations have normalized
-	// OTEL logs relayed to customer-defined destinations. It is targeted by
-	// PostHog organization group (org slug) and fails closed per organization,
-	// so one unavailable evaluation never changes another organization's
-	// delivery.
-	FlagOTELLogCustomerRelay Flag = "otel-log-customer-relay"
 )
 
 // Variants of FlagAssistantPlatformMCP. Anything else — no variant, an

server/internal/otel/dialect/log_claude_code.go

@@ -12,7 +12,7 @@ func (ClaudeCodeLog) AppliesTo(record *otelv1.InboundLogRecord) bool {
 }
 
 func (ClaudeCodeLog) InputContent(record *otelv1.InboundLogRecord) (string, genaiconv.InputMessages, error) {
-	key, value := getOneLogAttr(record, "user_prompt")
+	key, value := getOneLogAttr(record, claudeCodeUserPromptKey)
 	if key == "" || value == "" {
 		return "", nil, nil
 	}
@@ -37,12 +37,12 @@ func (ClaudeCodeLog) SessionID(record *otelv1.InboundLogRecord) (string, string,
 }
 
 func (ClaudeCodeLog) ExternalUserEmail(record *otelv1.InboundLogRecord) (string, string, error) {
-	key, value := getOneLogAttr(record, "user.email")
+	key, value := getOneLogAttr(record, userEmailKey)
 	return key, value, nil
 }
 
 func (ClaudeCodeLog) ExternalUserID(record *otelv1.InboundLogRecord) (string, string, error) {
-	key, value := getOneLogAttr(record, "user.account_id")
+	key, value := getOneLogAttr(record, vendorUserAccountIDKey)
 	return key, value, nil
 }
 

server/internal/otel/dialect/log_codex.go

@@ -20,7 +20,7 @@ func (CodexLog) AppliesTo(record *otelv1.InboundLogRecord) bool {
 func (CodexLog) InputContent(record *otelv1.InboundLogRecord) (string, genaiconv.InputMessages, error) {
 	switch record.GetEventName() {
 	case codexUserPromptEvent:
-		key, prompt := getOneLogAttr(record, "prompt")
+		key, prompt := getOneLogAttr(record, codexPromptKey)
 		if key == "" || prompt == "" || prompt == codexRedactedUserPrompt {
 			return "", nil, nil
 		}
@@ -48,12 +48,12 @@ func (CodexLog) SessionID(record *otelv1.InboundLogRecord) (string, string, erro
 }
 
 func (CodexLog) ExternalUserEmail(record *otelv1.InboundLogRecord) (string, string, error) {
-	key, value := getOneLogAttr(record, "user.email")
+	key, value := getOneLogAttr(record, userEmailKey)
 	return key, value, nil
 }
 
 func (CodexLog) ExternalUserID(record *otelv1.InboundLogRecord) (string, string, error) {
-	key, value := getOneLogAttr(record, "user.account_id")
+	key, value := getOneLogAttr(record, vendorUserAccountIDKey)
 	return key, value, nil
 }
 

server/internal/otel/dialect/log_semconv.go

@@ -13,11 +13,11 @@ type SemconvLog struct{}
 func (SemconvLog) AppliesTo(*otelv1.InboundLogRecord) bool { return true }
 
 func (SemconvLog) InputContent(record *otelv1.InboundLogRecord) (string, genaiconv.InputMessages, error) {
-	return semconvLogContent[genaiconv.InputMessages](record, "gen_ai.input.messages")
+	return semconvLogContent[genaiconv.InputMessages](record, semconvInputMessagesKey)
 }
 
 func (SemconvLog) OutputContent(record *otelv1.InboundLogRecord) (string, genaiconv.OutputMessages, error) {
-	return semconvLogContent[genaiconv.OutputMessages](record, "gen_ai.output.messages")
+	return semconvLogContent[genaiconv.OutputMessages](record, semconvOutputMessagesKey)
 }
 
 func (SemconvLog) SessionID(record *otelv1.InboundLogRecord) (string, string, error) {
@@ -26,12 +26,12 @@ func (SemconvLog) SessionID(record *otelv1.InboundLogRecord) (string, string, er
 }
 
 func (SemconvLog) ExternalUserEmail(record *otelv1.InboundLogRecord) (string, string, error) {
-	key, value := getOneLogAttr(record, "user.email")
+	key, value := getOneLogAttr(record, userEmailKey)
 	return key, value, nil
 }
 
 func (SemconvLog) ExternalUserID(record *otelv1.InboundLogRecord) (string, string, error) {
-	key, value := getOneLogAttr(record, "user.id")
+	key, value := getOneLogAttr(record, semconvUserIDKey)
 	return key, value, nil
 }
 

server/internal/otel/dialect/metric_claude_code.go

@@ -18,12 +18,12 @@ func (ClaudeCodeMetric) SessionID(point MetricDataPoint) (string, string, error)
 }
 
 func (ClaudeCodeMetric) ExternalUserID(point MetricDataPoint) (string, string, error) {
-	key, value := getOneMetricPointAttr(point, "user.account_id")
+	key, value := getOneMetricPointAttr(point, vendorUserAccountIDKey)
 	return key, value, nil
 }
 
 func (ClaudeCodeMetric) ExternalUserEmail(point MetricDataPoint) (string, string, error) {
-	key, value := getOneMetricPointAttr(point, "user.email")
+	key, value := getOneMetricPointAttr(point, userEmailKey)
 	return key, value, nil
 }
 

server/internal/otel/dialect/metric_codex.go

@@ -20,12 +20,12 @@ func (CodexMetric) SessionID(point MetricDataPoint) (string, string, error) {
 }
 
 func (CodexMetric) ExternalUserID(point MetricDataPoint) (string, string, error) {
-	key, value := getOneMetricPointAttr(point, "user.account_id")
+	key, value := getOneMetricPointAttr(point, vendorUserAccountIDKey)
 	return key, value, nil
 }
 
 func (CodexMetric) ExternalUserEmail(point MetricDataPoint) (string, string, error) {
-	key, value := getOneMetricPointAttr(point, "user.email")
+	key, value := getOneMetricPointAttr(point, userEmailKey)
 	return key, value, nil
 }
 

server/internal/otel/dialect/metric_semconv.go

@@ -12,12 +12,12 @@ func (SemconvMetric) SessionID(point MetricDataPoint) (string, string, error) {
 }
 
 func (SemconvMetric) ExternalUserID(point MetricDataPoint) (string, string, error) {
-	key, value := getOneMetricPointAttr(point, "user.id")
+	key, value := getOneMetricPointAttr(point, semconvUserIDKey)
 	return key, value, nil
 }
 
 func (SemconvMetric) ExternalUserEmail(point MetricDataPoint) (string, string, error) {
-	key, value := getOneMetricPointAttr(point, "user.email")
+	key, value := getOneMetricPointAttr(point, userEmailKey)
 	return key, value, nil
 }
 

server/internal/otel/dialect/sensitivity.go

@@ -0,0 +1,50 @@
+package dialect
+
+import "strings"
+
+// Sensitive attribute keys are shared by dialect extraction and relay redaction.
+// Dialect keys stay in the exact set even when a prefix also covers them.
+const (
+	claudeCodeUserPromptKey  = "user_prompt"
+	codexPromptKey           = "prompt"
+	semconvInputMessagesKey  = "gen_ai.input.messages"
+	semconvOutputMessagesKey = "gen_ai.output.messages"
+	semconvUserIDKey         = "user.id"
+	userEmailKey             = "user.email"
+	vendorUserAccountIDKey   = "user.account_id"
+)
+
+var sensitiveDataExactKeys = map[string]struct{}{
+	claudeCodeUserPromptKey:      {},
+	codexPromptKey:               {},
+	semconvInputMessagesKey:      {},
+	semconvOutputMessagesKey:     {},
+	semconvUserIDKey:             {},
+	userEmailKey:                 {},
+	vendorUserAccountIDKey:       {},
+	"assistant":                  {},
+	"content":                    {},
+	"gen_ai.system_instructions": {},
+	"tool.args":                  {},
+	"tool_result":                {},
+}
+
+var sensitiveDataPrefixes = [...]string{
+	"gen_ai.input.",
+	"gen_ai.output.",
+	"gen_ai.tool.call.",
+	"enduser.",
+	"user.",
+}
+
+func IsSensitiveDataKey(key string) bool {
+	if _, ok := sensitiveDataExactKeys[key]; ok {
+		return true
+	}
+	for _, prefix := range sensitiveDataPrefixes {
+		if strings.HasPrefix(key, prefix) {
+			return true
+		}
+	}
+	return false
+}

server/internal/otel/dialect/sensitivity_test.go

@@ -0,0 +1,41 @@
+package dialect
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsSensitiveDataKey(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		key       string
+		sensitive bool
+	}{
+		{key: "gen_ai.input.messages", sensitive: true},
+		{key: "gen_ai.output.messages", sensitive: true},
+		{key: "gen_ai.tool.call.arguments", sensitive: true},
+		{key: "gen_ai.tool.call.result", sensitive: true},
+		{key: "gen_ai.system_instructions", sensitive: true},
+		{key: "user_prompt", sensitive: true},
+		{key: "prompt", sensitive: true},
+		{key: "tool.args", sensitive: true},
+		{key: "tool_result", sensitive: true},
+		{key: "content", sensitive: true},
+		{key: "assistant", sensitive: true},
+		{key: "enduser.id", sensitive: true},
+		{key: "enduser.email", sensitive: true},
+		{key: "user.id", sensitive: true},
+		{key: "user.account_id", sensitive: true},
+		{key: "user.email", sensitive: true},
+		{key: "gen_ai.input", sensitive: false},
+		{key: "gen_ai.system", sensitive: false},
+		{key: "username", sensitive: false},
+		{key: "model", sensitive: false},
+	}
+
+	for _, test := range tests {
+		require.Equalf(t, test.sensitive, IsSensitiveDataKey(test.key), "key %q", test.key)
+	}
+}

server/internal/otel/dialect/span_claude_code.go

@@ -12,7 +12,7 @@ func (e ClaudeCodeSpan) AppliesTo(span *otelv1.InboundSpan) bool {
 }
 
 func (e ClaudeCodeSpan) InputContent(span *otelv1.InboundSpan) (key string, val genaiconv.InputMessages, err error) {
-	k, v := getOneAttr(span, "user_prompt")
+	k, v := getOneAttr(span, claudeCodeUserPromptKey)
 	if k == "" || v == "" {
 		return "", nil, nil
 	}
@@ -41,12 +41,12 @@ func (e ClaudeCodeSpan) SessionID(span *otelv1.InboundSpan) (key string, val str
 }
 
 func (e ClaudeCodeSpan) ExternalUserEmail(span *otelv1.InboundSpan) (key string, val string, err error) {
-	key, val = getOneAttr(span, "user.email")
+	key, val = getOneAttr(span, userEmailKey)
 	return key, val, nil
 }
 
 func (e ClaudeCodeSpan) ExternalUserID(span *otelv1.InboundSpan) (key string, val string, err error) {
-	key, val = getOneAttr(span, "user.account_id")
+	key, val = getOneAttr(span, vendorUserAccountIDKey)
 	return key, val, nil
 }
 

server/internal/otel/dialect/span_semconv.go

@@ -15,11 +15,11 @@ func (e SemconvSpan) AppliesTo(span *otelv1.InboundSpan) bool {
 }
 
 func (e SemconvSpan) InputContent(span *otelv1.InboundSpan) (string, genaiconv.InputMessages, error) {
-	return semconvContent[genaiconv.InputMessages](span, "gen_ai.input.messages")
+	return semconvContent[genaiconv.InputMessages](span, semconvInputMessagesKey)
 }
 
 func (e SemconvSpan) OutputContent(span *otelv1.InboundSpan) (string, genaiconv.OutputMessages, error) {
-	return semconvContent[genaiconv.OutputMessages](span, "gen_ai.output.messages")
+	return semconvContent[genaiconv.OutputMessages](span, semconvOutputMessagesKey)
 }
 
 func semconvContent[T any](span *otelv1.InboundSpan, desired string) (string, T, error) {
@@ -99,12 +99,12 @@ func (e SemconvSpan) SessionID(span *otelv1.InboundSpan) (key string, val string
 }
 
 func (e SemconvSpan) ExternalUserEmail(span *otelv1.InboundSpan) (key string, val string, err error) {
-	key, val = getOneAttr(span, "user.email")
+	key, val = getOneAttr(span, userEmailKey)
 	return key, val, nil
 }
 
 func (e SemconvSpan) ExternalUserID(span *otelv1.InboundSpan) (key string, val string, err error) {
-	key, val = getOneAttr(span, "user.id")
+	key, val = getOneAttr(span, semconvUserIDKey)
 	return key, val, nil
 }
 

server/internal/otel/handler_ilog_transforms_test.go

@@ -157,7 +157,7 @@ func TestMaxSizeLogRecordFitsRelayExportAfterFullEnrichment(t *testing.T) {
 	require.Positive(t, attributes[string(TokensCountKey)].GetIntValue())
 	require.NotEmpty(t, attributes[string(TokensCodecKey)].GetStringValue())
 
-	request, err := newLogRelayExportRequest([]*otelv1.LogRecord{published})
+	request, err := newLogRelayExportRequest([]*otelv1.LogRecord{published}, true)
 	require.NoError(t, err)
 	require.LessOrEqual(t, proto.Size(request), maxLogRelayExportBytes)
 }