fix(tokenizer): keep the tiktoken daemon thread out of the dynamic linker

CHANGELOG.md

@@ -7,6 +7,42 @@ the GitHub Release body, so a release with no entry here fails.
 
 Versioning follows [docs/versioning.md](docs/versioning.md).
 
+## [0.8.1] - 2026-09-05
+
+### Fixed
+
+- `runtime/loop/tokenizer.py` no longer runs `import tiktoken` on the daemon
+  thread it spawns. The import is hoisted to the calling thread; only
+  `tiktoken.get_encoding()` stays on the thread. An `atexit` hook now joins an
+  in-flight load (1 s cap), following `providers/nonblocking_stream.py`.
+
+  Rationale: `tiktoken._tiktoken` is a Rust extension, so the import `dlopen`s a
+  shared object, and CPython kills daemon threads mid-flight at finalization
+  (`pthread_exit` at the next GIL acquisition). Killed inside the dynamic linker
+  that is not survivable — losing glibc's `_dl_load_lock` surfaces as a later
+  SIGSEGV, unwinding through a Rust / `extern "C"` frame calls `abort()`. A
+  2026-09-05 investigation in ApodexHarness found the thread parked inside that
+  import at interpreter exit on *every* short run (the import takes 24-29 ms and
+  the first `get_encoding_nonblocking()` call lands late in a run), with two
+  observed deaths *after* a fully correct protocol stream: `-11` in
+  `test_stateless_across_invocations`, `-6` in `test_serve_subprocess_e2e`. The
+  window is wider wherever `TIKTOKEN_CACHE_DIR` is unset, because the load then
+  takes the no-timeout network path.
+
+  The module's loop-protection intent is unchanged: what it defends against is a
+  minutes-long, no-timeout HTTP fetch inside `get_encoding()`, and that is still
+  on the daemon thread. The import is local and needs no network.
+
+### Consumer action
+
+None required — `get_encoding_nonblocking()` keeps its signature and its
+contract (returns `None` while loading; callers still fall back to a heuristic).
+One timing difference worth knowing: the *first* call for an encoding name now
+spends 24-29 ms importing tiktoken on the calling thread, where before it
+returned instantly. In exchange the encoder becomes available sooner, which
+shortens the heuristic-fallback window `tokens.py` documents as the opening
+turns of every process.
+
 ## [0.8.0] - 2026-09-05
 
 ### Added

agent_core/runtime/loop/tokenizer.py

@@ -17,16 +17,34 @@
    ``None``; callers fall back to a chars/4 heuristic until the encoder
    lands. The loop thread NEVER blocks on tiktoken, cache-baked or not.
 
-NOTE: ``llm_client.py`` keeps an equivalent private copy (with an
-``o200k_base`` → ``cl100k_base`` preference) that predates this module
-and is left untouched to avoid churning tested code. New callers should
-use :func:`get_encoding_nonblocking`.
+What is deliberately NOT on that daemon thread is ``import tiktoken``.
+``tiktoken._tiktoken`` is a Rust extension, so the import ``dlopen``s a
+shared object; CPython kills daemon threads mid-flight during
+finalization (``pthread_exit`` at the next GIL acquisition), and being
+killed inside the dynamic linker is not survivable — losing glibc's
+``_dl_load_lock`` shows up as a later SIGSEGV, and unwinding through a
+Rust / ``extern "C"`` frame calls ``abort()``. The 2026-09-05
+investigation found the thread parked inside that import at
+interpreter-exit on *every* short run, with two observed deaths after a
+fully correct protocol stream (``-11`` in
+``test_stateless_across_invocations``, ``-6`` in
+``test_serve_subprocess_e2e``). The import is a local dlopen — 24-29 ms,
+no network, nothing the docstring above is defending against — so it
+belongs on the caller thread. Only ``get_encoding()`` (140 ms+, and
+unbounded on a cache miss) needs the thread.
+
+Belt and braces: an ``atexit`` hook joins any in-flight init, following
+``providers/nonblocking_stream.py``. CPython runs ``atexit`` callbacks
+before it starts killing daemon threads, so that is the last point at
+which the load can be drained cleanly.
 """
 
 from __future__ import annotations
 
+import atexit
 import logging
 import threading
+import time
 from typing import Any
 
 logger = logging.getLogger(__name__)
@@ -43,39 +61,86 @@
 _encoders: dict[str, Any] = {}
 _lock = threading.Lock()
 
+# In-flight init threads, keyed by encoding name, so ``_join_pending`` can
+# drain them at exit. Each thread removes its own entry when it finishes.
+_threads: dict[str, threading.Thread] = {}
+_atexit_registered = False
+
+# Long enough for a cache-hit ``get_encoding`` (~140 ms) to land, short
+# enough that a wedged network fetch cannot hold up process exit. A load
+# that misses this deadline is left where it is — the thread no longer
+# touches the dynamic linker, which is what made the mid-flight kill
+# dangerous in the first place.
+_JOIN_TIMEOUT_S = 1.0
+
+
+def _join_pending() -> None:
+    """atexit: drain in-flight inits before the interpreter kills them."""
+    with _lock:
+        pending = list(_threads.values())
+    deadline = time.monotonic() + _JOIN_TIMEOUT_S
+    for thread in pending:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        thread.join(timeout=remaining)
+
 
-def _load(name: str) -> None:
-    """Blocking tiktoken init — only ever runs on a daemon thread."""
+def _load(name: str, tiktoken: Any) -> None:
+    """Blocking ``get_encoding`` — only ever runs on a daemon thread.
+
+    ``tiktoken`` is passed in already imported: this function must not
+    import anything, see the module docstring.
+    """
     enc: Any
     try:
-        import tiktoken  # pyright: ignore[reportMissingImports]
-
         enc = tiktoken.get_encoding(name)  # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
-    except Exception:  # not installed / fetch failure / bad name
+    except Exception:  # fetch failure / bad name
         enc = False
         logger.debug("tiktoken encoding %r unavailable; using heuristic", name)
     with _lock:
         _encoders[name] = enc
+        _threads.pop(name, None)
 
 
 def get_encoding_nonblocking(name: str = "cl100k_base") -> Any | None:
     """Return the cached tiktoken encoder for ``name`` without ever blocking.
 
-    The first call schedules a daemon-thread init and returns ``None``;
-    later calls return the encoder once it has loaded, or ``None`` while
-    it is still loading. Returns ``None`` permanently when tiktoken is
-    unavailable — callers MUST fall back to a heuristic on ``None``.
+    The first call for a name imports tiktoken on the calling thread (a
+    local dlopen, no network), schedules the encoder init on a daemon
+    thread and returns ``None``; later calls return the encoder once it
+    has loaded, or ``None`` while it is still loading. Returns ``None``
+    permanently when tiktoken is unavailable — callers MUST fall back to
+    a heuristic on ``None``.
     """
+    global _atexit_registered
+
     enc = _encoders.get(name, _MISSING)
     if enc is not _MISSING:
         return enc or None  # None (loading) and False (failed) both collapse to None
+
+    # On the caller thread, deliberately — never on the daemon thread.
+    try:
+        import tiktoken  # pyright: ignore[reportMissingImports]
+    except Exception:  # not installed
+        with _lock:
+            _encoders[name] = False
+        logger.debug("tiktoken unavailable; using heuristic")
+        return None
+
     with _lock:
-        if _encoders.get(name, _MISSING) is _MISSING:  # still unclaimed under the lock
-            _encoders[name] = None  # mark loading so concurrent callers don't re-spawn
-            threading.Thread(
-                target=_load,
-                args=(name,),
-                name=f"tiktoken-init-{name}",
-                daemon=True,
-            ).start()
+        if _encoders.get(name, _MISSING) is not _MISSING:  # claimed while we imported
+            return _encoders[name] or None
+        _encoders[name] = None  # mark loading so concurrent callers don't re-spawn
+        thread = threading.Thread(
+            target=_load,
+            args=(name, tiktoken),
+            name=f"tiktoken-init-{name}",
+            daemon=True,
+        )
+        _threads[name] = thread
+        if not _atexit_registered:
+            atexit.register(_join_pending)
+            _atexit_registered = True
+        thread.start()
     return None

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "apodex-agent-core"
-version = "0.8.0"
+version = "0.8.1"
 description = "Shared, product-neutral runtime primitives for Apodex agents"
 readme = "README.md"
 license = "Apache-2.0"

tests/test_tokenizer_nonblocking.py

@@ -0,0 +1,193 @@
+"""The tiktoken loader must not leave a daemon thread inside an import.
+
+``import tiktoken`` dlopens a Rust extension. A daemon thread killed
+mid-import at interpreter finalization takes glibc's ``_dl_load_lock``
+with it (later SIGSEGV) or unwinds through an ``extern "C"`` frame
+(``abort()``); both were observed in production as a signal death
+*after* a fully correct protocol stream. So the import belongs on the
+caller thread and only ``get_encoding`` on the daemon thread — these
+tests pin which thread runs which, and that the exit hook drains the
+load.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import threading
+import types
+from typing import Any
+
+import pytest
+
+from agent_core.runtime.loop import tokenizer
+
+
+class _RecordingLoader:
+    """Stands in for the real ``tiktoken``, recording the loading thread."""
+
+    def __init__(self, gate: threading.Event | None = None) -> None:
+        self.import_thread: str | None = None
+        self.encode_thread: str | None = None
+        self.encoder = object()
+        self._gate = gate
+
+    # -- meta_path finder -------------------------------------------------
+    def find_spec(self, fullname: str, path: Any = None, target: Any = None) -> Any:
+        if fullname != "tiktoken":
+            return None
+        return importlib.util.spec_from_loader(fullname, self)
+
+    def create_module(self, spec: Any) -> types.ModuleType:
+        return types.ModuleType(spec.name)
+
+    def exec_module(self, module: types.ModuleType) -> None:
+        self.import_thread = threading.current_thread().name
+        module.get_encoding = self._get_encoding  # pyright: ignore[reportAttributeAccessIssue]
+
+    # -- the fake tiktoken API -------------------------------------------
+    def _get_encoding(self, name: str) -> object:
+        self.encode_thread = threading.current_thread().name
+        if self._gate is not None:
+            self._gate.wait(timeout=5.0)
+        return self.encoder
+
+
+@pytest.fixture
+def clean_tokenizer():
+    """Reset the module cache and unhook any real/fake tiktoken."""
+    saved_module = sys.modules.pop("tiktoken", None)
+    saved_meta = list(sys.meta_path)
+    saved_atexit_registered = tokenizer._atexit_registered
+    tokenizer._encoders.clear()
+    tokenizer._threads.clear()
+    tokenizer._atexit_registered = True  # don't leak a real atexit hook per test
+    try:
+        yield
+    finally:
+        tokenizer._join_pending()
+        tokenizer._encoders.clear()
+        tokenizer._threads.clear()
+        tokenizer._atexit_registered = saved_atexit_registered
+        sys.meta_path[:] = saved_meta
+        sys.modules.pop("tiktoken", None)
+        if saved_module is not None:
+            sys.modules["tiktoken"] = saved_module
+
+
+def test_import_runs_on_caller_thread_get_encoding_on_daemon(clean_tokenizer) -> None:
+    loader = _RecordingLoader()
+    sys.meta_path.insert(0, loader)
+
+    assert tokenizer.get_encoding_nonblocking("cl100k_base") is None
+    # The import must already have happened, synchronously, right here.
+    assert loader.import_thread == threading.current_thread().name
+    assert "tiktoken" in sys.modules
+
+    tokenizer._join_pending()
+    assert loader.encode_thread == "tiktoken-init-cl100k_base"
+    assert tokenizer.get_encoding_nonblocking("cl100k_base") is loader.encoder
+
+
+def test_atexit_hook_registers_once_and_drains_the_in_flight_load(
+    clean_tokenizer, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    gate = threading.Event()
+    loader = _RecordingLoader(gate=gate)
+    sys.meta_path.insert(0, loader)
+    registered: list[Any] = []
+    monkeypatch.setattr(tokenizer.atexit, "register", registered.append)
+    tokenizer._atexit_registered = False
+
+    tokenizer.get_encoding_nonblocking("cl100k_base")
+    tokenizer.get_encoding_nonblocking("o200k_base")
+    assert "cl100k_base" in tokenizer._threads  # in flight
+    assert registered == [tokenizer._join_pending]
+
+    gate.set()
+    registered[0]()
+    assert tokenizer._threads == {}  # thread finished and deregistered itself
+    assert tokenizer._encoders["cl100k_base"] is loader.encoder
+
+
+def test_atexit_join_timeout_is_shared_across_threads(
+    clean_tokenizer, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    class _Clock:
+        now = 10.0
+
+        def monotonic(self) -> float:
+            return self.now
+
+    class _PendingThread:
+        def __init__(self, clock: _Clock) -> None:
+            self.clock = clock
+            self.timeouts: list[float] = []
+
+        def join(self, timeout: float | None = None) -> None:
+            assert timeout is not None
+            self.timeouts.append(timeout)
+            self.clock.now += 0.4
+
+    clock = _Clock()
+    pending = [_PendingThread(clock) for _ in range(4)]
+    tokenizer._threads.update({str(i): thread for i, thread in enumerate(pending)})  # type: ignore[arg-type]
+    monkeypatch.setattr(tokenizer.time, "monotonic", clock.monotonic)
+
+    tokenizer._join_pending()
+
+    assert [thread.timeouts for thread in pending] == [
+        [pytest.approx(1.0)],
+        [pytest.approx(0.6)],
+        [pytest.approx(0.2)],
+        [],
+    ]
+
+
+def test_caller_never_blocks_on_a_wedged_get_encoding(clean_tokenizer) -> None:
+    gate = threading.Event()
+    loader = _RecordingLoader(gate=gate)
+    sys.meta_path.insert(0, loader)
+    try:
+        # Returns immediately even though get_encoding is parked.
+        assert tokenizer.get_encoding_nonblocking("cl100k_base") is None
+        assert tokenizer.get_encoding_nonblocking("cl100k_base") is None
+    finally:
+        gate.set()
+
+
+def test_missing_tiktoken_is_terminal_and_spawns_no_thread(clean_tokenizer) -> None:
+    class _Blocker:
+        def find_spec(self, fullname: str, path: Any = None, target: Any = None) -> Any:
+            if fullname == "tiktoken":
+                raise ImportError("no tiktoken")
+            return None
+
+    sys.meta_path.insert(0, _Blocker())
+
+    assert tokenizer.get_encoding_nonblocking("cl100k_base") is None
+    assert tokenizer._threads == {}
+    assert tokenizer._encoders["cl100k_base"] is False
+    assert tokenizer.get_encoding_nonblocking("cl100k_base") is None
+
+
+def test_concurrent_callers_spawn_one_thread(clean_tokenizer) -> None:
+    gate = threading.Event()
+    loader = _RecordingLoader(gate=gate)
+    sys.meta_path.insert(0, loader)
+    try:
+        start = threading.Barrier(4)
+
+        def call() -> None:
+            start.wait(timeout=5.0)
+            tokenizer.get_encoding_nonblocking("cl100k_base")
+
+        callers = [threading.Thread(target=call) for _ in range(4)]
+        for t in callers:
+            t.start()
+        for t in callers:
+            t.join(timeout=5.0)
+
+        assert len(tokenizer._threads) == 1
+    finally:
+        gate.set()