fix(otel): Add Mastra OTEL ingestion support with provider transformers

- Detect Mastra by instrumentation scope name (@mastra/otel)
- Treat Mastra as v1 framework (all attributes in spans, no log merging)
- Mark v1 framework root spans as $ai_span instead of $ai_trace to fix tree hierarchy
- Add provider transformer pattern for framework-specific data transformations
- Filter out raw input/output attributes to prevent duplicate otel.input/otel.output

Fixes tree display issue where Mastra generations weren't appearing as children
under the trace. Root spans from v1 frameworks must be $ai_span (not $ai_trace)
since TraceQueryRunner filters out $ai_trace events from the events array.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: andrewm4894

products/llm_analytics/backend/api/otel/conventions/genai.py

@@ -4,12 +4,17 @@
 Implements the GenAI semantic conventions (gen_ai.*) as fallback
 when PostHog-native attributes are not present.
 
+Supports provider-specific transformations for frameworks like Mastra
+that use custom OTEL formats.
+
 Reference: https://opentelemetry.io/docs/specs/semconv/gen-ai/
 """
 
 from collections import defaultdict
 from typing import Any
 
+from .providers import PROVIDER_TRANSFORMERS
+
 
 def has_genai_attributes(span: dict[str, Any]) -> bool:
     """Check if span uses GenAI semantic conventions."""
@@ -60,16 +65,44 @@ def _extract_indexed_messages(attributes: dict[str, Any], prefix: str) -> list[d
     return messages if messages else None
 
 
-def extract_genai_attributes(span: dict[str, Any]) -> dict[str, Any]:
+def extract_genai_attributes(span: dict[str, Any], scope: dict[str, Any] | None = None) -> dict[str, Any]:
     """
     Extract GenAI semantic convention attributes from span.
 
     GenAI conventions use `gen_ai.*` prefix and are fallback
     when PostHog-native attributes are not present.
+
+    Supports provider-specific transformations for frameworks that use
+    custom OTEL formats (e.g., Mastra).
+
+    Args:
+        span: Parsed OTEL span
+        scope: Instrumentation scope info (for provider detection)
+
+    Returns:
+        Extracted attributes dict
     """
+    import structlog
+
+    logger = structlog.get_logger(__name__)
     attributes = span.get("attributes", {})
+    scope = scope or {}
     result: dict[str, Any] = {}
 
+    # Detect provider-specific transformer
+    provider_transformer = None
+    for transformer_class in PROVIDER_TRANSFORMERS:
+        transformer = transformer_class()
+        if transformer.can_handle(span, scope):
+            provider_transformer = transformer
+            logger.info(
+                "provider_transformer_detected",
+                provider=transformer.get_provider_name(),
+                scope_name=scope.get("name"),
+                span_name=span.get("name"),
+            )
+            break
+
     # Model (prefer request, fallback to response, then system)
     model = (
         attributes.get("gen_ai.request.model")
@@ -100,14 +133,62 @@ def extract_genai_attributes(span: dict[str, Any]) -> dict[str, Any]:
         result["prompt"] = prompts
     # Fallback to direct gen_ai.prompt attribute
     elif (prompt := attributes.get("gen_ai.prompt")) is not None:
-        result["prompt"] = prompt
+        # Try provider-specific transformation
+        if provider_transformer:
+            logger.info(
+                "provider_transform_prompt_attempt",
+                provider=provider_transformer.get_provider_name(),
+                prompt_type=type(prompt).__name__,
+                prompt_length=len(str(prompt)) if prompt else 0,
+            )
+            transformed = provider_transformer.transform_prompt(prompt)
+            if transformed is not None:
+                logger.info(
+                    "provider_transform_prompt_success",
+                    provider=provider_transformer.get_provider_name(),
+                    result_type=type(transformed).__name__,
+                    result_length=len(transformed) if isinstance(transformed, list) else 0,
+                )
+                result["prompt"] = transformed
+            else:
+                logger.info(
+                    "provider_transform_prompt_none",
+                    provider=provider_transformer.get_provider_name(),
+                )
+                result["prompt"] = prompt
+        else:
+            result["prompt"] = prompt
 
     completions = _extract_indexed_messages(attributes, "gen_ai.completion")
     if completions:
         result["completion"] = completions
     # Fallback to direct gen_ai.completion attribute
     elif (completion := attributes.get("gen_ai.completion")) is not None:
-        result["completion"] = completion
+        # Try provider-specific transformation
+        if provider_transformer:
+            logger.info(
+                "provider_transform_completion_attempt",
+                provider=provider_transformer.get_provider_name(),
+                completion_type=type(completion).__name__,
+                completion_length=len(str(completion)) if completion else 0,
+            )
+            transformed = provider_transformer.transform_completion(completion)
+            if transformed is not None:
+                logger.info(
+                    "provider_transform_completion_success",
+                    provider=provider_transformer.get_provider_name(),
+                    result_type=type(transformed).__name__,
+                    result_length=len(transformed) if isinstance(transformed, list) else 0,
+                )
+                result["completion"] = transformed
+            else:
+                logger.info(
+                    "provider_transform_completion_none",
+                    provider=provider_transformer.get_provider_name(),
+                )
+                result["completion"] = completion
+        else:
+            result["completion"] = completion
 
     # Model parameters
     if (temperature := attributes.get("gen_ai.request.temperature")) is not None:

products/llm_analytics/backend/api/otel/conventions/providers/__init__.py

@@ -0,0 +1,22 @@
+"""
+Provider-specific OTEL transformers.
+
+Each provider (Mastra, Langchain, LlamaIndex, etc.) handles their
+specific OTEL format quirks and normalizes to PostHog format.
+"""
+
+from .base import ProviderTransformer
+from .mastra import MastraTransformer
+
+# Registry of all available provider transformers
+# Add new providers here as they're implemented
+PROVIDER_TRANSFORMERS: list[type[ProviderTransformer]] = [
+    MastraTransformer,
+    # Future: LangchainTransformer, LlamaIndexTransformer, etc.
+]
+
+__all__ = [
+    "ProviderTransformer",
+    "MastraTransformer",
+    "PROVIDER_TRANSFORMERS",
+]

products/llm_analytics/backend/api/otel/conventions/providers/base.py

@@ -0,0 +1,67 @@
+"""
+Base provider transformer interface.
+
+Provider transformers handle framework/library-specific OTEL formats
+and normalize them to PostHog's standard format.
+"""
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class ProviderTransformer(ABC):
+    """
+    Base class for provider-specific OTEL transformers.
+
+    Each provider (Mastra, Langchain, LlamaIndex, etc.) can implement
+    a transformer to handle their specific OTEL format quirks.
+    """
+
+    @abstractmethod
+    def can_handle(self, span: dict[str, Any], scope: dict[str, Any]) -> bool:
+        """
+        Detect if this transformer can handle the given span.
+
+        Args:
+            span: Parsed OTEL span
+            scope: Instrumentation scope info
+
+        Returns:
+            True if this transformer recognizes and can handle this span
+        """
+        pass
+
+    @abstractmethod
+    def transform_prompt(self, prompt: Any) -> Any:
+        """
+        Transform provider-specific prompt format to standard format.
+
+        Args:
+            prompt: Raw prompt value from gen_ai.prompt attribute
+
+        Returns:
+            Normalized prompt (list of message dicts, string, or None if no transformation needed)
+        """
+        pass
+
+    @abstractmethod
+    def transform_completion(self, completion: Any) -> Any:
+        """
+        Transform provider-specific completion format to standard format.
+
+        Args:
+            completion: Raw completion value from gen_ai.completion attribute
+
+        Returns:
+            Normalized completion (list of message dicts, string, or None if no transformation needed)
+        """
+        pass
+
+    def get_provider_name(self) -> str:
+        """
+        Get the provider name for logging/debugging.
+
+        Returns:
+            Human-readable provider name
+        """
+        return self.__class__.__name__.replace("Transformer", "")

products/llm_analytics/backend/api/otel/conventions/providers/mastra.py

@@ -0,0 +1,119 @@
+"""
+Mastra provider transformer.
+
+Handles Mastra's OTEL format which wraps messages in custom structures:
+- Input: {"messages": [{"role": "user", "content": [...]}]}
+- Output: {"files": [], "text": "...", "warnings": [], ...}
+"""
+
+import json
+from typing import Any
+
+from .base import ProviderTransformer
+
+
+class MastraTransformer(ProviderTransformer):
+    """
+    Transform Mastra's OTEL format to PostHog standard format.
+
+    Mastra uses @mastra/otel instrumentation scope and wraps messages
+    in custom structures that need unwrapping.
+    """
+
+    def can_handle(self, span: dict[str, Any], scope: dict[str, Any]) -> bool:
+        """
+        Detect Mastra by instrumentation scope name.
+
+        Mastra sets scope.name to "@mastra/otel" in its span converter.
+        """
+        scope_name = scope.get("name", "")
+
+        # Primary detection: instrumentation scope
+        if scope_name == "@mastra/otel":
+            return True
+
+        # Fallback: check for mastra-prefixed attributes
+        attributes = span.get("attributes", {})
+        return any(key.startswith("mastra.") for key in attributes.keys())
+
+    def transform_prompt(self, prompt: Any) -> Any:
+        """
+        Transform Mastra's wrapped input format.
+
+        Mastra wraps messages as: {"messages": [{"role": "user", "content": [...]}]}
+        where content can be an array of objects like [{"type": "text", "text": "..."}]
+        """
+        import structlog
+
+        logger = structlog.get_logger(__name__)
+
+        if not isinstance(prompt, str):
+            logger.info("mastra_transform_prompt_skip_not_string", prompt_type=type(prompt).__name__)
+            return None  # No transformation needed
+
+        try:
+            parsed = json.loads(prompt)
+            logger.info(
+                "mastra_transform_prompt_parsed",
+                has_messages=("messages" in parsed) if isinstance(parsed, dict) else False,
+                parsed_type=type(parsed).__name__,
+            )
+
+            # Check for Mastra input format: {"messages": [...]}
+            if not isinstance(parsed, dict) or "messages" not in parsed:
+                return None  # Not Mastra format
+
+            messages = parsed["messages"]
+            if not isinstance(messages, list):
+                return None
+
+            # Transform Mastra messages to standard format
+            result = []
+            for msg in messages:
+                if not isinstance(msg, dict) or "role" not in msg:
+                    continue
+
+                # Handle Mastra's content array format: [{"type": "text", "text": "..."}]
+                if "content" in msg and isinstance(msg["content"], list):
+                    text_parts = []
+                    for content_item in msg["content"]:
+                        if isinstance(content_item, dict):
+                            if content_item.get("type") == "text" and "text" in content_item:
+                                text_parts.append(content_item["text"])
+
+                    if text_parts:
+                        result.append({"role": msg["role"], "content": " ".join(text_parts)})
+                    else:
+                        # Keep as-is if we can't extract text
+                        result.append(msg)
+                else:
+                    # Standard format message
+                    result.append(msg)
+
+            return result if result else None
+
+        except (json.JSONDecodeError, TypeError, KeyError):
+            return None
+
+    def transform_completion(self, completion: Any) -> Any:
+        """
+        Transform Mastra's wrapped output format.
+
+        Mastra wraps output as: {"files": [], "text": "...", "warnings": [], ...}
+        Extract just the text content.
+        """
+        if not isinstance(completion, str):
+            return None  # No transformation needed
+
+        try:
+            parsed = json.loads(completion)
+
+            # Check for Mastra output format: {"text": "...", ...}
+            if not isinstance(parsed, dict) or "text" not in parsed:
+                return None  # Not Mastra format
+
+            # Extract text content as assistant message
+            return [{"role": "assistant", "content": parsed["text"]}]
+
+        except (json.JSONDecodeError, TypeError, KeyError):
+            return None