fix: harden orjson import and use explicit UTF-8 file encoding

py/src/braintrust/bt_json.py

@@ -12,7 +12,9 @@
     import orjson
 
     _HAS_ORJSON = True
-except ImportError:
+# Not just ImportError: orjson 3.12.0 raises SystemError on import under
+# Python 3.15's `-X lazy_imports=all`.
+except Exception:
     _HAS_ORJSON = False
 
 

py/src/braintrust/btx/spec_loader.py

@@ -164,7 +164,7 @@ def load_specs(
         if providers is not None and provider_dir not in providers:
             continue
 
-        with open(yaml_path) as f:
+        with open(yaml_path, encoding="utf-8") as f:
             data = yaml.load(f, Loader=loader_cls)  # noqa: S506 — intentional, custom loader
 
         spec = LlmSpanSpec.from_dict(data, source_path=yaml_path)

py/src/braintrust/cli/push.py

@@ -31,7 +31,7 @@ def _pkg_install_arg(pkg) -> str | None:
         dist = importlib.metadata.distribution(pkg)
         direct_url = dist._path / "direct_url.json"  # type: ignore
         if direct_url.exists():
-            with open(direct_url) as f:
+            with open(direct_url, encoding="utf-8") as f:
                 j = json.loads(f.read())
                 if "url" in j:
                     return j["url"]

py/src/braintrust/integrations/harbor/compat.py

@@ -210,9 +210,9 @@ def load_backfill_snapshot(job_dir: str | Path) -> tuple[JobSnapshot, Any]:
     from harbor.models.trial.result import TrialResult
 
     directory = Path(job_dir).expanduser().resolve()
-    config = JobConfig.model_validate_json((directory / "config.json").read_text())
-    lock = JobLock.model_validate_json((directory / "lock.json").read_text())
-    job_result = JobResult.model_validate_json((directory / "result.json").read_text())
+    config = JobConfig.model_validate_json((directory / "config.json").read_text(encoding="utf-8"))
+    lock = JobLock.model_validate_json((directory / "lock.json").read_text(encoding="utf-8"))
+    job_result = JobResult.model_validate_json((directory / "result.json").read_text(encoding="utf-8"))
     results: list[Any] = []
     result_paths = {
         result_path.parent: result_path
@@ -221,7 +221,7 @@ def load_backfill_snapshot(job_dir: str | Path) -> tuple[JobSnapshot, Any]:
     }
     for result_path in result_paths.values():
         try:
-            trial_result = TrialResult.model_validate_json(result_path.read_text())
+            trial_result = TrialResult.model_validate_json(result_path.read_text(encoding="utf-8"))
             # A job directory may be moved before backfill. Resolve trial files
             # relative to the directory being backfilled, not the old jobs_dir.
             trial_result.config.trials_dir = directory
@@ -245,7 +245,7 @@ def load_backfill_snapshot(job_dir: str | Path) -> tuple[JobSnapshot, Any]:
     for result in results:
         trial_lock_path = trial_directory(result) / "lock.json"
         if trial_lock_path.exists():
-            trial_lock = TrialLock.model_validate_json(trial_lock_path.read_text())
+            trial_lock = TrialLock.model_validate_json(trial_lock_path.read_text(encoding="utf-8"))
         else:
             candidates = lock_by_name.get(result.task_name) or []
             trial_lock = candidates.pop(0) if candidates else None

py/src/braintrust/integrations/harbor/plugin.py

@@ -829,7 +829,7 @@ def _sync_final_result(self, result: Any) -> None:
         extra_instructions: list[str] = []
         for path in getattr(result.config, "extra_instruction_paths", []) or []:
             try:
-                extra_instructions.append(Path(path).read_text())
+                extra_instructions.append(Path(path).read_text(encoding="utf-8"))
             except OSError:
                 continue
         if extra_instructions:
@@ -1029,15 +1029,15 @@ def _persist_disabled_manifest(self) -> None:
         }
         path = self._snapshot.job_dir / "braintrust-sync.json"
         temp_path = path.with_suffix(".json.tmp")
-        temp_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n")
+        temp_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
         os.replace(temp_path, path)
         self._manifest = manifest
 
     @staticmethod
     def _load_manifest(job_dir: Path) -> dict[str, Any]:
         path = job_dir / "braintrust-sync.json"
         try:
-            data = json.loads(path.read_text())
+            data = json.loads(path.read_text(encoding="utf-8"))
             return data if isinstance(data, dict) else {}
         except (OSError, json.JSONDecodeError):
             return {}
@@ -1085,7 +1085,7 @@ def _persist_manifest(self, completed: bool) -> None:
         }
         path = snapshot.job_dir / "braintrust-sync.json"
         temp_path = path.with_suffix(".json.tmp")
-        temp_path.write_text(json.dumps(manifest, indent=2, sort_keys=True, default=str) + "\n")
+        temp_path.write_text(json.dumps(manifest, indent=2, sort_keys=True, default=str) + "\n", encoding="utf-8")
         os.replace(temp_path, path)
         self._manifest = manifest
 

py/src/braintrust/logger.py

@@ -1483,7 +1483,7 @@ def _write_payload_to_dir(payload_dir, payload, debug_logging_adjective=None):
         payload_file = os.path.join(payload_dir, f"payload_{time.time()}_{str(uuid.uuid4())[:8]}.json")
         try:
             os.makedirs(payload_dir, exist_ok=True)
-            with open(payload_file, "w") as f:
+            with open(payload_file, "w", encoding="utf-8") as f:
                 f.write(payload)
         except Exception as e:
             eprint(f"Failed to write failed payload to output file {payload_file}:\n", e)

py/src/braintrust/span_cache.py

@@ -172,7 +172,7 @@ def _ensure_initialized(self) -> None:
             self._cache_file_path = os.path.join(tempfile.gettempdir(), f"braintrust-span-cache-{unique_id}.jsonl")
 
             # Create the file
-            with open(self._cache_file_path, "w") as f:
+            with open(self._cache_file_path, "w", encoding="utf-8") as f:
                 pass
 
             self._initialized = True
@@ -239,7 +239,7 @@ def _flush_write_buffer(self) -> None:
             return
 
         try:
-            with open(self._cache_file_path, "a") as f:
+            with open(self._cache_file_path, "a", encoding="utf-8") as f:
                 for record in self._write_buffer:
                     f.write(json.dumps(record.to_dict()) + "\n")
             self._write_buffer.clear()
@@ -280,7 +280,7 @@ def get_by_root_span_id(self, root_span_id: str) -> list[CachedSpan] | None:
         # Read from disk if initialized
         if self._initialized and self._cache_file_path and os.path.exists(self._cache_file_path):
             try:
-                with open(self._cache_file_path, "r") as f:
+                with open(self._cache_file_path, "r", encoding="utf-8") as f:
                     for line in f:
                         line = line.strip()
                         if not line:

py/src/braintrust/test_bt_json.py

@@ -2,10 +2,13 @@
 # pyright: reportUnknownArgumentType=false
 # pyright: reportPrivateUsage=false
 import json
+import subprocess
+import sys
 import warnings
 from typing import Any
 from unittest import TestCase
 
+import pytest
 from braintrust.bt_json import bt_dumps, bt_safe_deep_copy
 from braintrust.logger import Attachment, ExternalAttachment
 from braintrust.test_helpers import with_memory_logger  # noqa: F401
@@ -690,3 +693,43 @@ def test_bt_safe_deep_copy_circular_in_plain_dict_is_caught(self):
         # bt_dumps succeeds because the circular ref was sanitized
         json_str = bt_dumps(result)
         self.assertIn("<circular reference>", json_str)
+
+
+_BROKEN_ORJSON_SCRIPT = """
+import importlib.abc
+import sys
+
+class BrokenOrjson(importlib.abc.MetaPathFinder):
+    def find_spec(self, fullname, path=None, target=None):
+        if fullname == "orjson" or fullname.startswith("orjson."):
+            raise SystemError("simulated orjson import failure")
+        return None
+
+sys.meta_path.insert(0, BrokenOrjson())
+from braintrust.bt_json import _HAS_ORJSON, bt_dumps
+
+assert not _HAS_ORJSON
+print(bt_dumps({"b": 1, "a": "\u00e9t\u00e9"}))
+"""
+
+
+@pytest.mark.parametrize(
+    "lazy_imports",
+    [
+        False,
+        pytest.param(
+            True,
+            marks=pytest.mark.skipif(sys.version_info < (3, 15), reason="global lazy imports require Python 3.15+"),
+        ),
+    ],
+)
+def test_bt_json_falls_back_to_json_when_orjson_import_fails(lazy_imports):
+    # orjson 3.12.0 raises SystemError (not ImportError) on import under
+    # Python 3.15's `-X lazy_imports=all`.
+    flags = ["-X", "lazy_imports=all"] if lazy_imports else []
+    result = subprocess.run(
+        [sys.executable, *flags, "-c", _BROKEN_ORJSON_SCRIPT], check=False, capture_output=True, text=True
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert json.loads(result.stdout) == {"a": "\u00e9t\u00e9", "b": 1}

py/src/braintrust/test_logger.py

@@ -9,6 +9,7 @@
 import logging
 import operator
 import os
+import subprocess
 import sys
 import threading
 import time
@@ -3670,3 +3671,30 @@ def test_proxy_conn_base_url(proxy_url, expected_base_url):
     state = BraintrustState()
     state.proxy_url = proxy_url
     assert state.proxy_conn().base_url == expected_base_url
+
+
+_WRITE_PAYLOAD_SCRIPT = """
+import sys
+import warnings
+
+from braintrust.logger import _HTTPBackgroundLogger
+
+warnings.simplefilter("error", EncodingWarning)
+_HTTPBackgroundLogger._write_payload_to_dir(payload_dir=sys.argv[1], payload='{"input":"\u4f60\u597d \u00e9t\u00e9"}')
+"""
+
+
+def test_write_payload_to_dir_uses_utf8(tmp_path):
+    # orjson output keeps non-ASCII characters, so the dump must not depend on
+    # the locale encoding (e.g. cp1252 on Windows before Python 3.15).
+    result = subprocess.run(
+        [sys.executable, "-X", "warn_default_encoding", "-c", _WRITE_PAYLOAD_SCRIPT, str(tmp_path)],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert result.stderr == ""
+    [payload_file] = tmp_path.iterdir()
+    assert payload_file.read_text(encoding="utf-8") == '{"input":"\u4f60\u597d \u00e9t\u00e9"}'