fix(ci_visibility): xdist coverage upload with proper session id [backport 3.13] (#14551)

Backport 7e67963 from #14478 to 3.13.

CI Visibility: This fixes an issue where the coverage data from
pytest-xdist sessions were not using the proper session id.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Federico Mon <[email protected]>

Author: dd-octo-sts[bot]

ddtrace/internal/ci_visibility/encoder.py

@@ -264,6 +264,11 @@ def put(self, item):
             for span in item
             if COVERAGE_TAG_NAME in span.get_tags() or span.get_struct_tag(COVERAGE_TAG_NAME) is not None
         ]
+        # Also include session span for parent session ID lookup, even if it doesn't have coverage data
+        session_span = next((span for span in item if span.get_tag(EVENT_TYPE) == SESSION_TYPE), None)
+        if session_span and session_span not in spans_with_coverage:
+            spans_with_coverage.append(session_span)
+
         if not spans_with_coverage:
             raise NoEncodableSpansError()
         return super(CIVisibilityCoverageEncoderV02, self).put(spans_with_coverage)
@@ -294,8 +299,9 @@ def _build_body(self, data: bytes) -> List[bytes]:
         )
 
     def _build_data(self, traces: List[List[Span]]) -> Optional[bytes]:
+        new_parent_session_span_id = self._get_parent_session(traces)
         normalized_covs = [
-            self._convert_span(span)
+            self._convert_span(span, new_parent_session_span_id=new_parent_session_span_id)
             for trace in traces
             for span in trace
             if (COVERAGE_TAG_NAME in span.get_tags() or span.get_struct_tag(COVERAGE_TAG_NAME) is not None)
@@ -325,7 +331,7 @@ def _convert_span(
             files = json.loads(str(span.get_tag(COVERAGE_TAG_NAME)))["files"]
 
         converted_span = {
-            "test_session_id": int(span.get_tag(SESSION_ID) or "1"),
+            "test_session_id": new_parent_session_span_id or int(span.get_tag(SESSION_ID) or "1"),
             "test_suite_id": int(span.get_tag(SUITE_ID) or "1"),
             "files": files,
         }

releasenotes/notes/test-optimization-xdist-coverage-fix-b349ed080feeaa69.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    CI Visibility: This fix resolves an issue where coverage from sessions with pytest-xdist were not submitted with the proper session id, preventing Test Impact Analysis feature from working properly.

tests/ci_visibility/test_encoder.py

@@ -888,3 +888,74 @@ def test_full_encoding_with_parent_session_override():
     # Both should use the parent session ID (0xAAAAAA) instead of worker session ID
     assert test_event[b"content"][b"test_session_id"] == 0xAAAAAA
     assert session_event[b"content"][b"test_session_id"] == 0xAAAAAA
+
+
+def test_coverage_encoder_includes_session_span():
+    """Test that coverage encoder includes session span even if it doesn't have coverage data"""
+    # Create session span without coverage data
+    session_span = Span(name="test.session", span_id=0xAAAAAA, service="test")
+    session_span.set_tag(EVENT_TYPE, SESSION_TYPE)
+    session_span.set_tag(SESSION_ID, "12345")
+    session_span.parent_id = 0x999999  # Has parent ID for xdist
+
+    # Create test span with coverage data
+    coverage_data = {"files": [{"filename": "test.py", "segments": [[1, 0, 1, 0, -1]]}]}
+    test_span = Span(name="test.case", span_id=0xBBBBBB, service="test", span_type="test")
+    test_span.set_tag(COVERAGE_TAG_NAME, json.dumps(coverage_data))
+    test_span.set_tag(SESSION_ID, "12345")
+    test_span.set_tag(SUITE_ID, "67890")
+
+    # Create span without coverage data (should be filtered out)
+    regular_span = Span(name="regular.span", span_id=0xCCCCCC, service="test")
+
+    trace = [session_span, test_span, regular_span]
+
+    encoder = CIVisibilityCoverageEncoderV02(0, 0)
+    encoder.put(trace)
+
+    # Check what got stored in the buffer
+    assert len(encoder.buffer) == 1
+    stored_trace = encoder.buffer[0]
+
+    # Should include session span and coverage span, but not regular span
+    assert len(stored_trace) == 2
+    span_names = {span.name for span in stored_trace}
+    assert "test.session" in span_names
+    assert "test.case" in span_names
+    assert "regular.span" not in span_names
+
+
+def test_coverage_encoder_parent_session_id_propagation():
+    """Test that coverage encoder properly uses parent session ID from session span"""
+    # Create session span with parent_id (simulating xdist worker)
+    session_span = Span(name="worker.session", span_id=0xBBBBBB, service="test")
+    session_span.set_tag(EVENT_TYPE, SESSION_TYPE)
+    session_span.set_tag(SESSION_ID, "123")
+    session_span.parent_id = 0xAAAAAA  # Parent session ID from main process
+
+    # Create test span with coverage data
+    coverage_data = {"files": [{"filename": "test.py", "segments": [[1, 0, 1, 0, -1]]}]}
+    test_span = Span(name="test.case", span_id=0xCCCCCC, service="test", span_type="test")
+    test_span.set_tag(COVERAGE_TAG_NAME, json.dumps(coverage_data))
+    test_span.set_tag(SESSION_ID, "123")
+    test_span.set_tag(SUITE_ID, "67890")
+
+    traces = [[session_span, test_span]]
+
+    encoder = CIVisibilityCoverageEncoderV02(0, 0)
+    encoder.put(traces[0])
+
+    # Build coverage data
+    coverage_data_bytes = encoder._build_data(traces)
+    assert coverage_data_bytes is not None
+
+    # Decode the coverage data
+    decoded = msgpack.unpackb(coverage_data_bytes, raw=True, strict_map_key=False)
+    coverages = decoded[b"coverages"]
+
+    assert len(coverages) == 1
+    coverage_event = coverages[0]
+
+    # Should use parent session ID (0xAAAAAA) instead of worker session ID
+    assert coverage_event[b"test_session_id"] == 0xAAAAAA
+    assert coverage_event[b"test_suite_id"] == 67890