fix(client): Join split surrogate pairs when encoding task parameters

clients/python/src/taskbroker_client/task.py

@@ -40,6 +40,21 @@
 R = TypeVar("R")
 
 
+def _join_surrogates(value: Any) -> Any:
+    """
+    Rebuild strings that hold a UTF-16 surrogate pair as two separate code points
+    (e.g. an emoji parsed with `ast.literal_eval`), which msgpack can't encode as
+    UTF-8. Pairs are joined into one character and lone halves become U+FFFD.
+    """
+    if isinstance(value, str):
+        return value.encode("utf-16", "surrogatepass").decode("utf-16", "replace")
+    if isinstance(value, dict):
+        return {_join_surrogates(k): _join_surrogates(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_join_surrogates(v) for v in value]
+    return value
+
+
 def assert_typed_kwarg(
     func: Callable[..., Any],
     param_name: str,
@@ -290,6 +305,13 @@ def _signal_send(self, task: Task[Any, Any], args: Any, kwargs: Any) -> None:
         """
         pass
 
+    def _producer_metric_tags(self) -> dict[str, str]:
+        return {
+            "namespace": self._namespace.name,
+            "taskname": self.name,
+            "topic": self._namespace.topic,
+        }
+
     def create_activation(
         self,
         args: Collection[Any],
@@ -342,7 +364,14 @@ def create_activation(
                 )
 
         data = {"args": args, "kwargs": kwargs}
-        msgpack_bytes = msgpack.packb(data, use_bin_type=True)
+        try:
+            msgpack_bytes = msgpack.packb(data, use_bin_type=True)
+        except UnicodeEncodeError:
+            msgpack_bytes = msgpack.packb(_join_surrogates(data), use_bin_type=True)
+            self.namespace.metrics.incr(
+                "taskworker.producer.surrogates_repaired",
+                tags=self._producer_metric_tags(),
+            )
 
         should_compress = (
             self.compression_type == CompressionType.ZSTD
@@ -355,11 +384,7 @@ def create_activation(
             parameters_bytes_val = zstd.compress(msgpack_bytes)
             elapsed = time.perf_counter() - start_time
 
-            metric_tags = {
-                "namespace": self._namespace.name,
-                "taskname": self.name,
-                "topic": self._namespace.topic,
-            }
+            metric_tags = self._producer_metric_tags()
             self.namespace.metrics.distribution(
                 "taskworker.producer.compressed_parameters_size",
                 len(parameters_bytes_val),

clients/python/tests/test_task.py

@@ -297,6 +297,71 @@ def with_parameters(one: str, two: int, org_id: int) -> None:
     assert activation.parameters == ""
 
 
+EMOJI = chr(0x1F600)
+# The same emoji as two separate surrogate code points, which isn't valid UTF-8.
+SPLIT_EMOJI = chr(0xD83D) + chr(0xDE00)
+
+
+def test_create_activation_joins_split_surrogate_pair(task_namespace: TaskNamespace) -> None:
+    @task_namespace.register(name="test.surrogates")
+    def with_message(message: str) -> None:
+        raise NotImplementedError
+
+    with patch.object(task_namespace.metrics, "incr") as mock_incr:
+        activation = with_message.create_activation([], {"message": "hi " + SPLIT_EMOJI})
+
+    params = msgpack.unpackb(activation.parameters_bytes, raw=False)
+    assert params["kwargs"] == {"message": "hi " + EMOJI}
+    mock_incr.assert_called_once_with(
+        "taskworker.producer.surrogates_repaired",
+        tags={
+            "namespace": task_namespace.name,
+            "taskname": "test.surrogates",
+            "topic": task_namespace.topic,
+        },
+    )
+
+
+def test_create_activation_joins_nested_surrogate_pairs(task_namespace: TaskNamespace) -> None:
+    @task_namespace.register(name="test.surrogates_nested")
+    def with_nested(data: dict[str, Any], count: int) -> None:
+        raise NotImplementedError
+
+    activation = with_nested.create_activation(
+        [{SPLIT_EMOJI: [SPLIT_EMOJI, b"raw", 1]}],
+        {"count": 2},
+    )
+
+    params = msgpack.unpackb(activation.parameters_bytes, raw=False)
+    assert params["args"] == [{EMOJI: [EMOJI, b"raw", 1]}]
+    assert params["kwargs"] == {"count": 2}
+
+
+def test_create_activation_replaces_lone_surrogate(task_namespace: TaskNamespace) -> None:
+    @task_namespace.register(name="test.surrogates_lone")
+    def with_message(message: str) -> None:
+        raise NotImplementedError
+
+    activation = with_message.create_activation([], {"message": "hi " + chr(0xD83D)})
+
+    params = msgpack.unpackb(activation.parameters_bytes, raw=False)
+    assert params["kwargs"] == {"message": "hi " + chr(0xFFFD)}
+
+
+def test_create_activation_valid_unicode_unchanged(task_namespace: TaskNamespace) -> None:
+    @task_namespace.register(name="test.unicode")
+    def with_message(message: str) -> None:
+        raise NotImplementedError
+
+    kwargs = {"message": "hi " + EMOJI}
+    with patch.object(task_namespace.metrics, "incr") as mock_incr:
+        activation = with_message.create_activation([], kwargs)
+
+    expected = msgpack.packb({"args": [], "kwargs": kwargs}, use_bin_type=True)
+    assert activation.parameters_bytes == expected
+    mock_incr.assert_not_called()
+
+
 def test_create_activation_tracing(task_namespace: TaskNamespace) -> None:
     @task_namespace.register(name="test.parameters")
     def with_parameters(one: str, two: int, org_id: int) -> None: