[mq] [skip ddci] working branch - merge 7a432ec on top of main at 3a825cc

{"baseBranch":"main","baseCommit":"3a825ccafb6026572c5132fd3a6058544f1e205d","createdAt":"2026-08-13T13:33:12.423411Z","headSha":"7a432ec3cf262476295f7296cc79a56b392e687f","id":"9d476dfb-5b9f-4da4-a987-d5a01bb7fd38","priority":"200","pullRequestNumber":"19431","queuedAt":"2026-08-13T13:33:12.422283Z","status":"STATUS_QUEUED"}

Author: gh-worker-dd-mergequeue-cf854d[bot]

ddtrace/internal/native/_native.pyi

@@ -174,6 +174,14 @@ def store_metadata(data: PyTracerMetadata) -> PyAnonymousFileHandle:
     """
     ...
 
+if sys.implementation.name == "cpython" and sys.version_info >= (3, 14):
+    def register_context_watcher() -> bool:
+        """Register the Python context watcher if a watcher slot is available."""
+        ...
+    def is_context_watcher_registered() -> bool:
+        """Return whether the Python context watcher is registered."""
+        ...
+
 if sys.platform == "linux":
     def update_otel_thread_context(span: SpanData, local_root: Optional[SpanData], trace_flags: int) -> None:
         """

ddtrace/internal/opentelemetry/thread_context.py

@@ -17,31 +17,43 @@ def context_provider(self) -> BaseContextProvider: ...
 
 
 _ContextActivationListener = Callable[[BaseContextProvider, Optional[Union[Context, Span]]], None]
+_ContextSwitchListener = Callable[[], None]
+_ThreadContextListeners = tuple[_ContextActivationListener, _ContextSwitchListener]
 
 
 if sys.platform == "linux":
     from ddtrace.internal.native._native import detach_otel_thread_context
     from ddtrace.internal.native._native import update_otel_thread_context
 
-    def register_otel_thread_context_listener(tracer: TracerProtocol) -> Optional[_ContextActivationListener]:
+    def register_otel_thread_context_listener(tracer: TracerProtocol) -> Optional[_ThreadContextListeners]:
         if not config._otel_thread_context_enabled:
             return None
 
-        def _sync_otel_thread_context(provider: BaseContextProvider, ctx: Optional[Union[Context, Span]]) -> None:
-            if provider is not tracer.context_provider:
-                return
-
+        def _sync_otel_thread_context(ctx: Optional[Union[Context, Span]]) -> None:
             if type(ctx) is Span:
                 sampling_priority = ctx._local_root.context.sampling_priority
                 trace_flags = 1 if sampling_priority is not None and sampling_priority > 0 else 0
                 update_otel_thread_context(ctx, ctx._local_root_value, trace_flags)
             else:
                 detach_otel_thread_context()
 
-        core.on("ddtrace.context_provider.activate", _sync_otel_thread_context)
-        return _sync_otel_thread_context
+        def _sync_active_otel_thread_context() -> None:
+            _sync_otel_thread_context(tracer.context_provider.active())
+
+        def _on_context_provider_activate(provider: BaseContextProvider, ctx: Optional[Union[Context, Span]]) -> None:
+            if provider is tracer.context_provider:
+                _sync_otel_thread_context(ctx)
+
+        core.on("ddtrace.context_provider.activate", _on_context_provider_activate)
+        core.on("python.context.switch", _sync_active_otel_thread_context)
+
+        if sys.implementation.name == "cpython" and sys.version_info >= (3, 14):
+            from ddtrace.internal.native._native import register_context_watcher
+
+            register_context_watcher()
+        return _on_context_provider_activate, _sync_active_otel_thread_context
 
 else:
 
-    def register_otel_thread_context_listener(tracer: TracerProtocol) -> Optional[_ContextActivationListener]:
+    def register_otel_thread_context_listener(tracer: TracerProtocol) -> Optional[_ThreadContextListeners]:
         return None

releasenotes/notes/python-314-otel-thread-context-switches-38bdb87440442bbc.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - |
+    tracing: Adds Python 3.14+ support for keeping OpenTelemetry thread-context records synchronized across asynchronous context switches.
+    This behavior is enabled by default and can be disabled by setting ``DD_TRACE_OTEL_CTX_ENABLED`` to ``false``.

src/native/context_watcher.rs

@@ -0,0 +1,103 @@
+use pyo3::ffi;
+use pyo3::prelude::*;
+use std::ffi::{c_int, c_uint};
+use std::matches;
+use std::panic::{catch_unwind, AssertUnwindSafe};
+use std::sync::OnceLock;
+
+const CONTEXT_SWITCH_EVENT: &str = "python.context.switch";
+
+type PyContextEvent = c_uint;
+const PY_CONTEXT_SWITCHED: PyContextEvent = 1;
+type PyContextWatchCallback =
+    unsafe extern "C" fn(event: PyContextEvent, object: *mut ffi::PyObject) -> c_int;
+
+unsafe extern "C" {
+    fn PyContext_AddWatcher(callback: PyContextWatchCallback) -> c_int;
+}
+
+static WATCHER_ID: OnceLock<Option<c_int>> = OnceLock::new();
+
+#[pyfunction]
+pub fn register_context_watcher(py: Python<'_>) -> bool {
+    WATCHER_ID
+        .get_or_init(|| {
+            // SAFETY: This module is only compiled for CPython 3.14+ with the
+            // GIL enabled, and the callback signature matches
+            // PyContext_WatchCallback from cpython/context.h.
+            let watcher_id = unsafe { PyContext_AddWatcher(context_watcher) };
+            if watcher_id == -1 {
+                // Context-switch publication is optional. If no watcher slot
+                // is available, clear the C-API error and leave it disabled.
+                drop(PyErr::fetch(py));
+                None
+            } else {
+                Some(watcher_id)
+            }
+        })
+        .is_some()
+}
+
+#[pyfunction]
+pub fn is_context_watcher_registered() -> bool {
+    matches!(WATCHER_ID.get(), Some(Some(_)))
+}
+
+pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
+    m.add_function(wrap_pyfunction!(register_context_watcher, m)?)?;
+    m.add_function(wrap_pyfunction!(is_context_watcher_registered, m)?)
+}
+
+unsafe extern "C" fn context_watcher(event: PyContextEvent, object: *mut ffi::PyObject) -> c_int {
+    // CPython may invoke watcher callbacks with an exception already set. Clear
+    // it temporarily so listeners can use regular Python APIs, then restore it;
+    // losing it makes Context.run raise SystemError instead of the original error.
+    let pending_exception = unsafe { ffi::PyErr_GetRaisedException() };
+    let callback_result = match catch_unwind(AssertUnwindSafe(|| {
+        // CPython invokes context watchers on an attached thread, but entering
+        // through the C API bypasses PyO3's attachment bookkeeping.
+        Python::attach(|py| {
+            if event == PY_CONTEXT_SWITCHED {
+                // Listeners must not enter another Context: CPython context watchers
+                // are reentrant. The OTel listener does not enter a Context.
+                if let Err(error) =
+                    crate::event_hub::dispatch(py, CONTEXT_SWITCH_EVENT, None, false)
+                {
+                    error.restore(py);
+                    return -1;
+                }
+            }
+
+            0
+        })
+    })) {
+        Ok(result) => result,
+        Err(_) => {
+            // Keep panics from crossing the C boundary even if attaching itself
+            // fails before a Python token is available.
+            unsafe {
+                ffi::PyErr_SetString(
+                    ffi::PyExc_RuntimeError,
+                    c"panic in Python context watcher".as_ptr(),
+                )
+            };
+            -1
+        }
+    };
+
+    if pending_exception.is_null() {
+        return callback_result;
+    }
+
+    // A new callback error must not replace the exception which was pending on
+    // entry. Report it as unraisable before restoring the original exception.
+    unsafe {
+        if callback_result == -1 {
+            ffi::PyErr_WriteUnraisable(object);
+        }
+        ffi::PyErr_Clear();
+        ffi::PyErr_SetRaisedException(pending_exception);
+    }
+
+    0
+}

src/native/lib.rs

@@ -4,6 +4,8 @@ mod crashtracker;
 pub use datadog_profiling_ffi::*;
 mod config;
 mod context_provider;
+#[cfg(all(Py_3_14, not(any(PyPy, GraalPy))))]
+mod context_watcher;
 mod contextvar;
 mod data_pipeline;
 #[cfg(feature = "stats")]
@@ -102,5 +104,8 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> {
     // Add tracer_flare submodule
     m.add_wrapped(pyo3::wrap_pymodule!(tracer_flare::native_flare))?;
 
+    #[cfg(all(Py_3_14, not(any(PyPy, GraalPy))))]
+    context_watcher::register(m)?;
+
     Ok(())
 }

tests/internal/events/test_context_watcher.py

@@ -0,0 +1,143 @@
+import contextvars
+import sys
+import threading
+
+import pytest
+
+from ddtrace.internal import core
+
+
+pytestmark = pytest.mark.skipif(
+    sys.implementation.name != "cpython" or sys.version_info < (3, 14),
+    reason="requires the CPython 3.14 context watcher",
+)
+
+
+@pytest.fixture(autouse=True)
+def _register_context_watcher():
+    from ddtrace.internal.native._native import is_context_watcher_registered
+    from ddtrace.internal.native._native import register_context_watcher
+
+    assert register_context_watcher()
+    assert is_context_watcher_registered()
+
+
+def test_context_watcher_dispatches_events_and_releases_listener_snapshot():
+    value = contextvars.ContextVar("value", default="outer")
+    inner_context = contextvars.copy_context()
+    inner_context.run(value.set, "inner")
+    observed = []
+    test_thread_id = threading.get_ident()
+
+    def record_context_switch():
+        if threading.get_ident() == test_thread_id:
+            observed.append(value.get())
+
+    core.on("python.context.switch", record_context_switch)
+    try:
+        reference_count = sys.getrefcount(record_context_switch)
+        inner_context.run(lambda: None)
+        assert sys.getrefcount(record_context_switch) == reference_count
+    finally:
+        core.reset_listeners("python.context.switch", record_context_switch)
+
+    assert observed == ["inner", "outer"]
+
+
+def test_context_watcher_preserves_pending_exception_over_listener_failure():
+    inner_context = contextvars.copy_context()
+    observed = []
+    unraisable = []
+    test_thread_id = threading.get_ident()
+
+    class ExpectedError(Exception):
+        pass
+
+    class ListenerError(BaseException):
+        pass
+
+    def record_context_switch():
+        if threading.get_ident() != test_thread_id:
+            return
+        observed.append(contextvars.copy_context())
+        if len(observed) == 2:
+            raise ListenerError
+
+    def raise_expected_error():
+        raise ExpectedError
+
+    original_unraisablehook = sys.unraisablehook
+    sys.unraisablehook = lambda args: unraisable.append(args.exc_value)
+    core.on("python.context.switch", record_context_switch)
+    try:
+        with pytest.raises(ExpectedError):
+            inner_context.run(raise_expected_error)
+    finally:
+        core.reset_listeners("python.context.switch", record_context_switch)
+        sys.unraisablehook = original_unraisablehook
+
+    assert len(observed) == 2
+    assert len(unraisable) == 1
+    assert isinstance(unraisable[0], ListenerError)
+
+
+@pytest.mark.subprocess(env={"_DD_GLOBAL_TRACER_INIT": "false"})
+def test_context_watcher_slot_exhaustion_disables_watcher():
+    """Registration failure is non-fatal and remains cached after watcher slots are freed."""
+    import ctypes
+    import sys
+
+    assert "ddtrace.internal.native._native" not in sys.modules
+
+    callback_type = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_uint, ctypes.py_object)
+    callback = callback_type(lambda event, obj: 0)
+    add_watcher = ctypes.pythonapi.PyContext_AddWatcher
+    add_watcher.argtypes = [callback_type]
+    add_watcher.restype = ctypes.c_int
+    clear_watcher = ctypes.pythonapi.PyContext_ClearWatcher
+    clear_watcher.argtypes = [ctypes.c_int]
+    clear_watcher.restype = ctypes.c_int
+
+    watcher_ids = []
+    for _ in range(64):
+        try:
+            watcher_ids.append(add_watcher(callback))
+        except RuntimeError:
+            break
+    else:
+        raise AssertionError("context watcher slots were not exhausted")
+
+    try:
+        from ddtrace.internal.native._native import is_context_watcher_registered
+        from ddtrace.internal.native._native import register_context_watcher
+
+        assert register_context_watcher() is False
+        assert is_context_watcher_registered() is False
+    finally:
+        for watcher_id in watcher_ids:
+            assert clear_watcher(watcher_id) == 0
+
+    assert register_context_watcher() is False
+    assert is_context_watcher_registered() is False
+
+
+@pytest.mark.subprocess(env={"_DD_GLOBAL_TRACER_INIT": "false"})
+def test_context_watcher_registration_is_idempotent():
+    """Repeated registration uses one watcher; tracer startup is disabled so this test owns the first call."""
+    from contextvars import Context
+
+    from ddtrace.internal import core
+    from ddtrace.internal.native._native import is_context_watcher_registered
+    from ddtrace.internal.native._native import register_context_watcher
+
+    assert is_context_watcher_registered() is False
+
+    for _ in range(16):
+        assert register_context_watcher() is True
+
+    assert is_context_watcher_registered() is True
+
+    observed = []
+    core.on("python.context.switch", lambda: observed.append(None))
+    Context().run(lambda: None)
+    assert observed == [None, None]