feat(seer-explorer): add rpc to fetch trace by id

src/sentry/seer/endpoints/seer_rpc.py

@@ -70,6 +70,7 @@
     rpc_get_issues_for_transaction,
     rpc_get_profiles_for_trace,
     rpc_get_trace_for_transaction,
+    rpc_get_trace_from_id,
     rpc_get_transactions_for_project,
 )
 from sentry.seer.fetch_issues import by_error_type, by_function_name, by_text_query, utils
@@ -999,6 +1000,7 @@ def send_seer_webhook(*, event_name: str, organization_id: int, payload: dict) -
     "get_trace_for_transaction": rpc_get_trace_for_transaction,
     "get_profiles_for_trace": rpc_get_profiles_for_trace,
     "get_issues_for_transaction": rpc_get_issues_for_transaction,
+    "get_trace_from_id": rpc_get_trace_from_id,
     #
     # Replays
     "get_replay_summary_logs": rpc_get_replay_summary_logs,

src/sentry/seer/explorer/index_data.py

@@ -548,6 +548,92 @@ def get_issues_for_transaction(transaction_name: str, project_id: int) -> Transa
     )
 
 
+def get_trace_from_id(trace_id: str, project_id: int) -> TraceData | None:
+    """
+    Get a trace from an ID.
+
+    Args:
+        trace_id: The ID of the trace to fetch. This can be shortened as the first 8 chars.
+        project_id: The ID of the project
+
+    Returns:
+        TraceData with all spans and relationships, or None if no trace found
+    """
+    try:
+        project = Project.objects.get(id=project_id)
+    except Project.DoesNotExist:
+        logger.exception(
+            "Project does not exist; cannot fetch trace",
+            extra={"project_id": project_id, "trace_id": trace_id},
+        )
+        return None
+
+    end_time = datetime.now(UTC)
+    start_time = end_time - timedelta(hours=24)
+
+    snuba_params = SnubaParams(
+        start=start_time,
+        end=end_time,
+        projects=[project],
+        organization=project.organization,
+    )
+    config = SearchResolverConfig(
+        auto_fields=True,
+    )
+
+    # Get all spans in the trace
+    spans_result = Spans.run_table_query(
+        params=snuba_params,
+        query_string=f"trace:{trace_id}",
+        selected_columns=[
+            "span_id",
+            "parent_span",
+            "span.op",
+            "span.description",
+            "transaction",
+            "trace",  # Full trace ID
+            "precise.start_ts",
+        ],
+        orderby=["precise.start_ts"],
+        offset=0,
+        limit=5000,
+        referrer=Referrer.SEER_RPC,
+        config=config,
+        sampling_mode="NORMAL",
+    )
+
+    if not spans_result.get("data", []):
+        logger.info(
+            "No spans found for trace",
+            extra={"trace_id": trace_id, "project_id": project_id},
+        )
+        return None
+
+    # Build span objects
+    spans = []
+    for row in spans_result["data"]:
+        if span_id := row.get("span_id"):
+            spans.append(
+                Span(
+                    span_id=span_id,
+                    parent_span_id=row.get("parent_span"),
+                    span_op=row.get("span.op"),
+                    span_description=row.get("span.description") or "",
+                )
+            )
+
+    transaction_name = spans_result["data"][0]["transaction"]
+    full_trace_id = spans_result["data"][0]["trace"]
+
+    return TraceData(
+        trace_id=full_trace_id,
+        project_id=project_id,
+        transaction_name=transaction_name,
+        total_spans=len(spans),
+        spans=spans,
+    )
+
+
 # RPC wrappers
 
 
@@ -570,3 +656,8 @@ def rpc_get_profiles_for_trace(trace_id: str, project_id: int) -> dict[str, Any]
 def rpc_get_issues_for_transaction(transaction_name: str, project_id: int) -> dict[str, Any]:
     issues = get_issues_for_transaction(transaction_name, project_id)
     return issues.dict() if issues else {}
+
+
+def rpc_get_trace_from_id(trace_id: str, project_id: int) -> dict[str, Any]:
+    trace = get_trace_from_id(trace_id, project_id)
+    return trace.dict() if trace else {}

tests/sentry/seer/explorer/test_index_data.py

@@ -9,6 +9,7 @@
     get_issues_for_transaction,
     get_profiles_for_trace,
     get_trace_for_transaction,
+    get_trace_from_id,
     get_transactions_for_project,
 )
 from sentry.testutils.cases import APITransactionTestCase, SnubaTestCase, SpanTestCase
@@ -775,3 +776,74 @@ def test_get_issues_for_transaction_with_quotes(self) -> None:
         issue2 = result2.issues[0]
         assert issue2.id == event2.group.id
         assert issue2.transaction == transaction_name_in
+
+    def test_get_trace_from_id(self) -> None:
+        transaction_name = "api/users/profile"
+        trace_id = uuid.uuid4().hex
+        spans = []
+        for i in range(5):
+            # Create spans for this trace
+            span = self.create_span(
+                {
+                    **({"description": f"span-{i}"} if i != 4 else {}),
+                    "sentry_tags": {"transaction": transaction_name},
+                    "trace_id": trace_id,
+                    "parent_span_id": None if i == 0 else f"parent-{i-1}",
+                    "is_segment": i == 0,  # First span is the transaction span
+                },
+                start_ts=self.ten_mins_ago + timedelta(minutes=i),
+            )
+            spans.append(span)
+
+        self.store_spans(spans, is_eap=True)
+
+        # Call our function with shortened trace ID
+        result = get_trace_from_id(trace_id[:8], self.project.id)
+
+        # Verify basic structure
+        assert result is not None
+        assert result.transaction_name == transaction_name
+        assert result.project_id == self.project.id
+        # assert result.trace_id == trace_id
+        assert len(result.spans) == len(spans)
+
+        # Verify all spans have correct structure and belong to the chosen trace
+        for i, result_span in enumerate(result.spans):
+            assert hasattr(result_span, "span_id")
+            assert hasattr(result_span, "span_description")
+            assert hasattr(result_span, "parent_span_id")
+            assert hasattr(result_span, "span_op")
+            assert result_span.span_description is not None
+            if i != 4:
+                assert result_span.span_description.startswith("span-")
+
+        # Verify parent-child relationships are preserved
+        root_spans = [s for s in result.spans if s.parent_span_id is None]
+        assert len(root_spans) == 1  # Should have exactly one root span
+
+    def test_get_trace_from_id_wrong_project(self) -> None:
+        transaction_name = "api/users/profile"
+        trace_id = uuid.uuid4().hex
+        other_project = self.create_project(organization=self.organization)
+
+        spans = []
+        for i in range(2):
+            # Create spans in the wrong project
+            span = self.create_span(
+                {
+                    "project_id": other_project.id,
+                    "description": f"span-{i}",
+                    "sentry_tags": {"transaction": transaction_name},
+                    "trace_id": trace_id,
+                    "parent_span_id": None if i == 0 else f"parent-{i-1}",
+                    "is_segment": i == 0,  # First span is the transaction span
+                },
+                start_ts=self.ten_mins_ago + timedelta(minutes=i),
+            )
+            spans.append(span)
+
+        self.store_spans(spans, is_eap=True)
+
+        # Call our function with shortened trace ID
+        result = get_trace_from_id(trace_id[:8], self.project.id)
+        assert result is None