diff --git a/CHANGELOG.md b/CHANGELOG.md index 2eaf36a..4e59953 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/agent_core/runtime/loop/tokenizer.py b/agent_core/runtime/loop/tokenizer.py index 13974b6..8c7bea0 100644 --- a/agent_core/runtime/loop/tokenizer.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 1895479..0371fde 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/test_tokenizer_nonblocking.py b/tests/test_tokenizer_nonblocking.py new file mode 100644 index 0000000..e460343 --- /dev/null +++ b/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() diff --git a/uv.lock b/uv.lock index 24876fa..167511e 100644 --- a/uv.lock +++ b/uv.lock @@ -50,7 +50,7 @@ wheels = [ [[package]] name = "apodex-agent-core" -version = "0.8.0" +version = "0.8.1" source = { editable = "." } dependencies = [ { name = "anthropic", extra = ["bedrock"] },