ThreadCache workers leaking spawner's Context on py314+ free-threaded builds (#3473)

* Fix ThreadCache workers leaking spawner's Context on py314+ free-threaded builds

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix typo

* add newsfragments

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix typing and reformat newfragments entry

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* add a regression test

* update newfragments entry

* update _thread_cache

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* update the test to be skipped on pypi

* update the test

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* simply regression test

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: EmmanuelNiyonshuti

newsfragments/3472.bugfix.rst

@@ -0,0 +1 @@
+Fixed ``trio.to_thread.run_sync``` workers leaking spawner's ``Context`` on ``py314t+`` free-threaded builds.

src/trio/_core/_thread_cache.py

@@ -5,6 +5,7 @@
 import os
 import sys
 import traceback
+from contextvars import Context
 from functools import partial
 from itertools import count
 from threading import Lock, Thread
@@ -151,8 +152,17 @@ def __init__(self, thread_cache: ThreadCache) -> None:
         self._worker_lock = Lock()
         self._worker_lock.acquire()
         self._default_name = f"Trio thread {next(name_counter)}"
-
-        self._thread = Thread(target=self._work, name=self._default_name, daemon=True)
+        if sys.version_info >= (3, 14):
+            self._thread = Thread(
+                target=self._work,
+                name=self._default_name,
+                daemon=True,
+                context=Context(),
+            )
+        else:
+            self._thread = Thread(
+                target=self._work, name=self._default_name, daemon=True
+            )
 
         if set_os_thread_name:
             set_os_thread_name(self._thread.ident, self._default_name)

src/trio/_tests/test_threads.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import contextvars
+import gc
 import queue as stdlib_queue
 import re
 import sys
@@ -687,6 +688,28 @@ def g() -> tuple[str, str, threading.Thread]:
     assert sniffio.current_async_library() == "trio"
 
 
+async def test_worker_thread_context_not_leaked() -> None:
+    # Regression test for: https://github.com/python-trio/trio/issues/3472
+
+    class Foo:
+        pass
+
+    def sync_fn() -> None:
+        pass
+
+    cvar: contextvars.ContextVar[Foo] = contextvars.ContextVar("cvar")
+    contextval = Foo()
+    ref = weakref.ref(contextval)
+    cvar.set(contextval)
+    await to_thread_run_sync(sync_fn)
+    cvar.set(Foo())
+
+    del contextval
+    gc.collect()
+
+    assert ref() is None
+
+
 async def test_trio_from_thread_run_sync() -> None:
     # Test that to_thread_run_sync correctly "hands off" the trio token to
     # trio.from_thread.run_sync()