feat(core): extract pin_threads() helper for determinism

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/core/determinism.py

@@ -0,0 +1,30 @@
+"""Threading determinism: pin BLAS / OpenMP / pyarrow to single-threaded mode.
+
+Multi-threaded floating-point reductions reorder operands non-deterministically
+on each call, breaking the byte-identity guarantee in AGENTS.md §4 rule 3. Each
+pipeline calls `pin_threads()` at import time to lock the process to a single
+thread before any numerical work runs.
+
+Honors pre-set env vars: if the caller exported `OMP_NUM_THREADS=4` upstream,
+that value is preserved (we use `setdefault`, not `setitem`). The user is
+responsible for the determinism trade-off in that case.
+"""
+from __future__ import annotations
+
+import os
+
+import pyarrow as pa
+
+_ENV_VARS: tuple[str, ...] = (
+    "OMP_NUM_THREADS",
+    "OPENBLAS_NUM_THREADS",
+    "MKL_NUM_THREADS",
+)
+
+
+def pin_threads() -> None:
+    """Pin BLAS / OpenMP / pyarrow to single-threaded mode (idempotent)."""
+    for var in _ENV_VARS:
+        os.environ.setdefault(var, "1")
+    pa.set_cpu_count(1)
+    pa.set_io_thread_count(1)

tests/core/test_determinism.py

@@ -0,0 +1,75 @@
+"""Tests for src.core.determinism."""
+from __future__ import annotations
+
+import os
+
+import pyarrow as pa
+
+from src.core import determinism
+
+
+class TestPinThreads:
+    def test_sets_omp_env_var(self):
+        original = os.environ.pop("OMP_NUM_THREADS", None)
+        try:
+            determinism.pin_threads()
+            assert os.environ["OMP_NUM_THREADS"] == "1"
+        finally:
+            if original is None:
+                os.environ.pop("OMP_NUM_THREADS", None)
+            else:
+                os.environ["OMP_NUM_THREADS"] = original
+
+    def test_sets_openblas_env_var(self):
+        original = os.environ.pop("OPENBLAS_NUM_THREADS", None)
+        try:
+            determinism.pin_threads()
+            assert os.environ["OPENBLAS_NUM_THREADS"] == "1"
+        finally:
+            if original is None:
+                os.environ.pop("OPENBLAS_NUM_THREADS", None)
+            else:
+                os.environ["OPENBLAS_NUM_THREADS"] = original
+
+    def test_sets_mkl_env_var(self):
+        original = os.environ.pop("MKL_NUM_THREADS", None)
+        try:
+            determinism.pin_threads()
+            assert os.environ["MKL_NUM_THREADS"] == "1"
+        finally:
+            if original is None:
+                os.environ.pop("MKL_NUM_THREADS", None)
+            else:
+                os.environ["MKL_NUM_THREADS"] = original
+
+    def test_pins_pyarrow_cpu_count_to_1(self):
+        pa.set_cpu_count(4)
+        determinism.pin_threads()
+        assert pa.cpu_count() == 1
+
+    def test_pins_pyarrow_io_thread_count_to_1(self):
+        pa.set_io_thread_count(4)
+        determinism.pin_threads()
+        assert pa.io_thread_count() == 1
+
+    def test_does_not_override_existing_env(self):
+        """User explicitly setting OMP_NUM_THREADS=2 must win — pin_threads()
+        uses os.environ.setdefault so an upstream override is preserved."""
+        original = os.environ.get("OMP_NUM_THREADS")
+        os.environ["OMP_NUM_THREADS"] = "2"
+        try:
+            determinism.pin_threads()
+            assert os.environ["OMP_NUM_THREADS"] == "2"
+        finally:
+            if original is None:
+                os.environ.pop("OMP_NUM_THREADS", None)
+            else:
+                os.environ["OMP_NUM_THREADS"] = original
+
+    def test_idempotent(self):
+        determinism.pin_threads()
+        determinism.pin_threads()
+        assert pa.cpu_count() == 1
+        assert os.environ["OMP_NUM_THREADS"] == "1"
+        assert os.environ["OPENBLAS_NUM_THREADS"] == "1"
+        assert os.environ["MKL_NUM_THREADS"] == "1"