fix(scores): apply trace sampling to scores again

langfuse/_client/client.py

@@ -2022,15 +2022,15 @@ def create_score(
         try:
             new_body = ScoreBody(
                 id=score_id,
-                sessionId=session_id,
-                datasetRunId=dataset_run_id,
-                traceId=trace_id,
-                observationId=observation_id,
+                session_id=session_id,
+                dataset_run_id=dataset_run_id,
+                trace_id=trace_id,
+                observation_id=observation_id,
                 name=name,
                 value=value,
-                dataType=data_type,  # type: ignore
+                data_type=data_type,  # type: ignore
                 comment=comment,
-                configId=config_id,
+                config_id=config_id,
                 environment=environment or self._environment,
                 metadata=metadata,
             )

langfuse/_task_manager/score_ingestion_consumer.py

@@ -11,6 +11,7 @@
 from langfuse._utils.parse_error import handle_exception
 from langfuse._utils.request import APIError, LangfuseClient
 from langfuse._utils.serializer import EventSerializer
+from langfuse.api.core.serialization import convert_and_respect_annotation_metadata
 from langfuse.logger import langfuse_logger as logger
 
 from .._version import __version__ as langfuse_version
@@ -78,7 +79,11 @@ def _next(self) -> list:
 
                 # convert pydantic models to dicts
                 if "body" in event and isinstance(event["body"], BaseModel):
-                    event["body"] = event["body"].model_dump(exclude_none=True)
+                    event["body"] = convert_and_respect_annotation_metadata(
+                        object_=event["body"].model_dump(exclude_none=True),
+                        annotation=type(event["body"]),
+                        direction="write",
+                    )
 
                 item_size = self._get_item_size(event)
 

tests/unit/test_otel.py

@@ -1,6 +1,7 @@
 import json
 from datetime import datetime
 from hashlib import sha256
+from queue import Queue
 from typing import List, Sequence
 
 import pytest
@@ -1468,6 +1469,25 @@ def test_sampling(self, monkeypatch, tracer_provider, mock_processor_init):
         # Restore the original provider
         trace_api.set_tracer_provider(original_provider)
 
+    def test_score_sampling_follows_trace_sampling(self):
+        client = Langfuse(
+            public_key="test-public-key",
+            secret_key="test-secret-key",
+            base_url="http://test-host",
+            sample_rate=0.5,
+        )
+        # Detach from the consumer thread so it cannot drain the queue mid-test.
+        queue = Queue()
+        client._resources._score_ingestion_queue = queue
+
+        # The sampler keeps a trace when its low 64 bits fall below rate * 2^64.
+        client.create_score(name="sampled", value=1.0, trace_id="0" * 32)
+        client.create_score(name="dropped", value=1.0, trace_id="f" * 32)
+        client.create_score(name="session", value=1.0, session_id="session-1")
+
+        names = [queue.get_nowait()["body"].name for _ in range(queue.qsize())]
+        assert names == ["sampled", "session"]
+
     @pytest.mark.skip("Calling shutdown will pollute the global context")
     def test_shutdown_and_flush(self, langfuse_client, memory_exporter):
         """Test shutdown and flush operations."""

tests/unit/test_resource_manager.py

@@ -14,6 +14,7 @@
 from langfuse._task_manager.media_manager import MediaManager
 from langfuse._task_manager.media_upload_consumer import MediaUploadConsumer
 from langfuse._task_manager.score_ingestion_consumer import ScoreIngestionConsumer
+from langfuse.api.ingestion.types.score_body import ScoreBody
 from langfuse.types import MaskOtelSpansResult
 
 
@@ -156,6 +157,32 @@ def test_score_ingestion_consumer_pause_wakes_blocked_thread():
     assert not consumer.is_alive()
 
 
+def test_score_ingestion_consumer_serializes_body_by_alias():
+    queue = Queue()
+    queue.put(
+        {
+            "type": "score-create",
+            "body": ScoreBody(id="s1", trace_id="t" * 32, name="quality", value=1.0),
+        }
+    )
+    consumer = ScoreIngestionConsumer(
+        ingestion_queue=queue,
+        identifier=0,
+        client=Mock(),
+        public_key="pk-test",
+        flush_interval=0.01,
+    )
+
+    batch = consumer._next()
+
+    assert batch[0]["body"] == {
+        "id": "s1",
+        "traceId": "t" * 32,
+        "name": "quality",
+        "value": 1.0,
+    }
+
+
 def test_media_upload_consumer_signal_shutdown_wakes_blocked_thread():
     media_manager = MediaManager(
         api_client=Mock(),