diff --git a/crates/python/src/py_plugin.rs b/crates/python/src/py_plugin.rs
index 2cf76a4a8..281ae0eca 100644
--- a/crates/python/src/py_plugin.rs
+++ b/crates/python/src/py_plugin.rs
@@ -3,11 +3,11 @@
 
 //! Python-facing generic plugin configuration and registration helpers.
 
+#[cfg(test)]
+use std::collections::HashSet;
 use std::future::Future;
 use std::pin::Pin;
-use std::sync::{Arc, Mutex};
-#[cfg(test)]
-use std::{collections::HashSet, sync::LazyLock};
+use std::sync::{Arc, LazyLock, Mutex};
 
 use pyo3::prelude::*;
 use serde_json::{Map, Value as Json};
@@ -685,6 +685,7 @@ fn initialize_plugins_py<'py>(
         .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))?;
     pyo3_async_runtimes::tokio::future_into_py(py, async move {
         let report = initialize_plugins(config).await.map_err(to_py_err)?;
+        reset_plugin_configuration_clear_state();
         Python::attach(|py| {
             let report = serde_json::to_value(&report)
                 .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(e.to_string()))?;
@@ -756,6 +757,35 @@ impl PluginTeardownError {
 
 type PluginTeardownResult = std::result::Result<(), PluginTeardownError>;
 
+struct PluginTeardownCompletion {
+    result: tokio::sync::watch::Sender<Option<PluginTeardownResult>>,
+}
+
+impl PluginTeardownCompletion {
+    fn new() -> Self {
+        let (result, _) = tokio::sync::watch::channel(None);
+        Self { result }
+    }
+
+    fn finish(&self, result: PluginTeardownResult) {
+        self.result.send_replace(Some(result));
+    }
+
+    async fn wait(&self, operation: &'static str) -> PluginTeardownResult {
+        let mut result = self.result.subscribe();
+        loop {
+            if let Some(result) = result.borrow().clone() {
+                return result;
+            }
+            if result.changed().await.is_err() {
+                return Err(PluginTeardownError::runtime(format!(
+                    "{operation} result channel closed unexpectedly"
+                )));
+            }
+        }
+    }
+}
+
 enum PluginHostCloseStatus {
     Active(Option<PluginHostActivation>),
     Closing,
@@ -764,15 +794,14 @@ enum PluginHostCloseStatus {
 
 struct PluginHostCloseState {
     status: Mutex<PluginHostCloseStatus>,
-    completion: tokio::sync::watch::Sender<Option<PluginTeardownResult>>,
+    completion: PluginTeardownCompletion,
 }
 
 impl PluginHostCloseState {
     fn new(activation: PluginHostActivation) -> Self {
-        let (completion, _) = tokio::sync::watch::channel(None);
         Self {
             status: Mutex::new(PluginHostCloseStatus::Active(Some(activation))),
-            completion,
+            completion: PluginTeardownCompletion::new(),
         }
     }
 
@@ -858,24 +887,84 @@ impl PluginHostCloseState {
             .status
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner()) = PluginHostCloseStatus::Closed;
-        self.completion.send_replace(Some(result));
+        reset_plugin_configuration_clear_state();
+        self.completion.finish(result);
     }
 
     async fn wait_for_close(&self) -> PluginTeardownResult {
-        let mut completion = self.completion.subscribe();
-        loop {
-            if let Some(result) = completion.borrow().clone() {
-                return result;
-            }
-            if completion.changed().await.is_err() {
-                return Err(PluginTeardownError::runtime(
-                    "dynamic plugin teardown result channel closed unexpectedly",
-                ));
+        self.completion.wait("dynamic plugin teardown").await
+    }
+}
+
+struct PluginConfigurationClearState {
+    started: Mutex<bool>,
+    completion: PluginTeardownCompletion,
+}
+
+impl PluginConfigurationClearState {
+    fn new() -> Self {
+        Self {
+            started: Mutex::new(false),
+            completion: PluginTeardownCompletion::new(),
+        }
+    }
+
+    fn begin_clear(self: &Arc<Self>) {
+        let should_start = {
+            let mut started = self
+                .started
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if *started {
+                false
+            } else {
+                *started = true;
+                true
             }
+        };
+        if !should_start {
+            return;
         }
+
+        let clear_state = Arc::clone(self);
+        let spawn = std::thread::Builder::new()
+            .name("nemo-relay-python-plugin-clear".into())
+            .spawn(move || {
+                let result = std::panic::catch_unwind(clear_plugin_configuration)
+                    .map_err(|_| PluginTeardownError::runtime("plugin teardown task panicked"))
+                    .and_then(|result| result.map_err(PluginTeardownError::from_plugin_error));
+                clear_state.completion.finish(result);
+            });
+        if let Err(error) = spawn {
+            self.completion
+                .finish(Err(PluginTeardownError::runtime(format!(
+                    "failed to start plugin teardown task: {error}"
+                ))));
+        }
+    }
+
+    async fn wait_for_clear(&self) -> PluginTeardownResult {
+        self.completion.wait("plugin teardown").await
     }
 }
 
+static PLUGIN_CONFIGURATION_CLEAR_STATE: LazyLock<Mutex<Arc<PluginConfigurationClearState>>> =
+    LazyLock::new(|| Mutex::new(Arc::new(PluginConfigurationClearState::new())));
+
+fn plugin_configuration_clear_state() -> Arc<PluginConfigurationClearState> {
+    PLUGIN_CONFIGURATION_CLEAR_STATE
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .clone()
+}
+
+fn reset_plugin_configuration_clear_state() {
+    *PLUGIN_CONFIGURATION_CLEAR_STATE
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner()) =
+        Arc::new(PluginConfigurationClearState::new());
+}
+
 #[pymethods]
 impl PyPluginHostActivation {
     /// Return the activation report captured during initialization.
@@ -935,6 +1024,7 @@ fn initialize_with_dynamic_plugins_py<'py>(
             PluginHostActivation::activate_with_discovered_config(config, dynamic_plugins)
                 .await
                 .map_err(plugin_error_to_py_err)?;
+        reset_plugin_configuration_clear_state();
         Python::attach(|py| {
             Py::new(
                 py,
@@ -953,6 +1043,19 @@ fn clear_plugin_configuration_py(py: Python<'_>) -> PyResult<()> {
     py.detach(clear_plugin_configuration).map_err(to_py_err)
 }
 
+#[pyfunction(name = "clear_plugin_configuration_async")]
+#[pyo3(signature = () -> "object", text_signature = "() -> object")]
+fn clear_plugin_configuration_async_py<'py>(py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+    let clear_state = plugin_configuration_clear_state();
+    clear_state.begin_clear();
+    pyo3_async_runtimes::tokio::future_into_py(py, async move {
+        clear_state
+            .wait_for_clear()
+            .await
+            .map_err(|error| error.to_py_err())
+    })
+}
+
 #[pyfunction(name = "active_plugin_report")]
 #[pyo3(signature = () -> "object", text_signature = "() -> object")]
 fn active_plugin_report_py(py: Python<'_>) -> PyResult<Py<PyAny>> {
@@ -997,6 +1100,7 @@ pub fn register(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_function(wrap_pyfunction!(initialize_plugins_py, m)?)?;
     m.add_function(wrap_pyfunction!(initialize_with_dynamic_plugins_py, m)?)?;
     m.add_function(wrap_pyfunction!(clear_plugin_configuration_py, m)?)?;
+    m.add_function(wrap_pyfunction!(clear_plugin_configuration_async_py, m)?)?;
     m.add_function(wrap_pyfunction!(active_plugin_report_py, m)?)?;
     m.add_function(wrap_pyfunction!(list_plugin_kinds_py, m)?)?;
     m.add_function(wrap_pyfunction!(register_plugin_py, m)?)?;
diff --git a/crates/python/tests/coverage/nemo_guardrails_coverage_tests.rs b/crates/python/tests/coverage/nemo_guardrails_coverage_tests.rs
index ce3a4f274..9a9ed65da 100644
--- a/crates/python/tests/coverage/nemo_guardrails_coverage_tests.rs
+++ b/crates/python/tests/coverage/nemo_guardrails_coverage_tests.rs
@@ -709,7 +709,7 @@ async def run_case():
     else:
         raise AssertionError("expected streamed output block")
 
-    nemo_relay.plugin.clear()
+    await nemo_relay.plugin.clear_async()
     await nemo_relay.plugin.initialize(plugin_config("stream_first_false"))
     try:
         await run_stream(request)
diff --git a/crates/python/tests/coverage/py_plugin_coverage_tests.rs b/crates/python/tests/coverage/py_plugin_coverage_tests.rs
index 83bc2c9b5..9353d0bcf 100644
--- a/crates/python/tests/coverage/py_plugin_coverage_tests.rs
+++ b/crates/python/tests/coverage/py_plugin_coverage_tests.rs
@@ -112,6 +112,7 @@ fn register_adds_plugin_management_bindings() {
             "initialize_plugins",
             "initialize_with_dynamic_plugins",
             "clear_plugin_configuration",
+            "clear_plugin_configuration_async",
             "active_plugin_report",
             "list_plugin_kinds",
             "register_plugin",
@@ -159,6 +160,30 @@ fn register_adds_plugin_management_bindings() {
     });
 }
 
+#[test]
+fn async_clear_binding_completes_on_python_event_loop() {
+    let _python = crate::test_support::init_python_test();
+    let _plugin_test_state = lock_plugin_test_state_for_tests();
+    Python::attach(|py| {
+        let module = PyModule::new(py, "_plugin_async_clear").unwrap();
+        register(&module).unwrap();
+        let helpers = load_module(
+            py,
+            r#"
+async def clear(module):
+    await module.clear_plugin_configuration_async()
+"#,
+        );
+        with_event_loop(py, |event_loop| {
+            let clear = helpers.getattr("clear").unwrap().call1((module,)).unwrap();
+            event_loop
+                .call_method1("run_until_complete", (clear,))
+                .unwrap();
+        });
+        assert!(active_plugin_report_py(py).unwrap().bind(py).is_none());
+    });
+}
+
 #[test]
 fn plugin_context_registers_all_runtime_hooks_and_drains_registrations() {
     let _python = crate::test_support::init_python_test();
diff --git a/docs/build-plugins/language-binding/code-examples.mdx b/docs/build-plugins/language-binding/code-examples.mdx
index 5ebe2edca..776483b53 100644
--- a/docs/build-plugins/language-binding/code-examples.mdx
+++ b/docs/build-plugins/language-binding/code-examples.mdx
@@ -85,7 +85,7 @@ async def main() -> None:
         print("Activation report:", active_report)
         # Run instrumented application work here.
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 
 if __name__ == "__main__":
@@ -216,7 +216,7 @@ async def main() -> None:
         print("Activation report:", active_report)
         # Run managed tool or LLM work here to log lifecycle events.
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 
 if __name__ == "__main__":
@@ -335,7 +335,7 @@ async def main() -> None:
         await nemo_relay.plugin.initialize(config)
         # Run managed tool or LLM work here.
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 
 if __name__ == "__main__":
diff --git a/docs/build-plugins/language-binding/register-behavior.mdx b/docs/build-plugins/language-binding/register-behavior.mdx
index f84967af8..205d03a86 100644
--- a/docs/build-plugins/language-binding/register-behavior.mdx
+++ b/docs/build-plugins/language-binding/register-behavior.mdx
@@ -278,7 +278,7 @@ async def main() -> None:
         print("Activation report:", active_report)
         print("Available kinds:", nemo_relay.plugin.list_kinds())
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 
 if __name__ == "__main__":
diff --git a/docs/configure-plugins/adaptive/acg.mdx b/docs/configure-plugins/adaptive/acg.mdx
index a0f5f536b..f70fb2396 100644
--- a/docs/configure-plugins/adaptive/acg.mdx
+++ b/docs/configure-plugins/adaptive/acg.mdx
@@ -92,7 +92,7 @@ async def main():
         # Run instrumented application work here.
         pass
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 asyncio.run(main())
 ```
diff --git a/docs/configure-plugins/adaptive/adaptive-hints.mdx b/docs/configure-plugins/adaptive/adaptive-hints.mdx
index 4f2530ddf..e35b5c2ce 100644
--- a/docs/configure-plugins/adaptive/adaptive-hints.mdx
+++ b/docs/configure-plugins/adaptive/adaptive-hints.mdx
@@ -89,7 +89,7 @@ async def main():
         # Run instrumented application work here.
         pass
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 asyncio.run(main())
 ```
diff --git a/docs/configure-plugins/adaptive/response-cache.mdx b/docs/configure-plugins/adaptive/response-cache.mdx
index b9299bc0f..f1cb692cb 100644
--- a/docs/configure-plugins/adaptive/response-cache.mdx
+++ b/docs/configure-plugins/adaptive/response-cache.mdx
@@ -127,7 +127,7 @@ async def main():
         await nemo_relay.llm.execute("openai", request, call_model)
         await nemo_relay.llm.execute("openai", request, call_model)
     finally:
-        nemo_relay.plugin.clear()
+        await nemo_relay.plugin.clear_async()
 
 asyncio.run(main())
 ```
diff --git a/docs/configure-plugins/observability/about.mdx b/docs/configure-plugins/observability/about.mdx
index 7fcde0eb6..0519cea22 100644
--- a/docs/configure-plugins/observability/about.mdx
+++ b/docs/configure-plugins/observability/about.mdx
@@ -85,7 +85,8 @@ Use the barrier that matches how you manage the exporter lifecycle:
 - For a manual exporter, call its documented `force_flush()`, `export()`, or
   `shutdown()` operation.
 - For plugin-managed export, call `plugin.clear()` or
-  `clear_plugin_configuration()` during graceful shutdown.
+  `clear_plugin_configuration()` during graceful shutdown. From a Python
+  `asyncio` task, await `plugin.clear_async()` instead.
 
 If the process terminates before those operations complete, the output can omit
 queued telemetry even when the instrumented application work completed
diff --git a/docs/configure-plugins/observability/atif.mdx b/docs/configure-plugins/observability/atif.mdx
index b8e071656..817874ad8 100644
--- a/docs/configure-plugins/observability/atif.mdx
+++ b/docs/configure-plugins/observability/atif.mdx
@@ -323,7 +323,7 @@ async def main():
         # Run instrumented application work here.
         pass
     finally:
-        plugin.clear()
+        await plugin.clear_async()
 
 asyncio.run(main())
 ```
diff --git a/docs/configure-plugins/observability/atof.mdx b/docs/configure-plugins/observability/atof.mdx
index 05227ad13..ff1d87536 100644
--- a/docs/configure-plugins/observability/atof.mdx
+++ b/docs/configure-plugins/observability/atof.mdx
@@ -149,8 +149,10 @@ events delivered after the exporter closes. A timeout is logged and does not by
 itself cause `shutdown()` to return an error.
 
 During graceful shutdown, `plugin.clear()` or `clear_plugin_configuration()`
-uses that same terminal stream-sink operation. It does not guarantee that a
-worker which timed out drained or closed.
+uses that same terminal stream-sink operation. From a Python `asyncio` task,
+await `plugin.clear_async()` so queued sanitizers can finish on the running
+event loop. Teardown does not guarantee that a worker which timed out drained
+or closed.
 
 ## Migrate From Version 1
 
@@ -260,7 +262,7 @@ async def main():
         # Run instrumented application work here.
         pass
     finally:
-        plugin.clear()
+        await plugin.clear_async()
 
 asyncio.run(main())
 ```
diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi
index 790fa9619..e5ed22496 100644
--- a/python/nemo_relay/_native.pyi
+++ b/python/nemo_relay/_native.pyi
@@ -2385,6 +2385,17 @@ def clear_plugin_configuration() -> None:
     """
     ...
 
+def clear_plugin_configuration_async() -> Awaitable[None]:
+    """Clear active plugin configuration without blocking the Python event loop.
+
+    Returns:
+        Awaitable resolving when native teardown completes.
+
+    Exceptional flow:
+        Native cleanup and teardown worker errors propagate through the awaitable.
+    """
+    ...
+
 def active_plugin_report() -> Optional[_JsonObject]:
     """Return the active plugin report.
 
diff --git a/python/nemo_relay/plugin.py b/python/nemo_relay/plugin.py
index 8cc927f36..ce9e0b430 100644
--- a/python/nemo_relay/plugin.py
+++ b/python/nemo_relay/plugin.py
@@ -10,6 +10,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field, fields, is_dataclass
 from typing import TYPE_CHECKING, AsyncIterator, Callable, Literal, Protocol, Self, TypedDict, cast
@@ -38,6 +39,9 @@
 from nemo_relay._native import (
     clear_plugin_configuration as _clear_plugin_configuration,
 )
+from nemo_relay._native import (
+    clear_plugin_configuration_async as _clear_plugin_configuration_async,
+)
 from nemo_relay._native import (
     deregister_plugin as _deregister_plugin,
 )
@@ -467,10 +471,29 @@ def clear() -> None:
     Behavior:
         This removes active component registrations but leaves the plugin kind
         registry intact for future validation or initialization.
+
+    Raises:
+        RuntimeError: If called while an ``asyncio`` event loop is running on
+            the current thread. Use :func:`clear_async` instead.
     """
+    try:
+        asyncio.get_running_loop()
+    except RuntimeError:
+        pass
+    else:
+        raise RuntimeError("plugin.clear() cannot block a running asyncio event loop; use 'await plugin.clear_async()'")
     _clear_plugin_configuration()
 
 
+async def clear_async() -> None:
+    """Clear the active plugin configuration without blocking ``asyncio``.
+
+    Native teardown runs outside the Python event-loop thread so queued event
+    sanitizers can finish before their plugin-owned registrations are removed.
+    """
+    await _clear_plugin_configuration_async()
+
+
 @asynccontextmanager
 async def plugin(config: PluginConfig | JsonObject, *, clear_on_exit: bool = True) -> AsyncIterator[ConfigReport]:
     """Context manager for plugin initialization and cleanup.
@@ -493,7 +516,7 @@ async def plugin(config: PluginConfig | JsonObject, *, clear_on_exit: bool = Tru
             await subscribers.flush_async()
         finally:
             if clear_on_exit:
-                clear()
+                await clear_async()
 
 
 def report() -> ConfigReport | None:
@@ -569,6 +592,7 @@ def deregister(plugin_kind: str) -> bool:
     "Plugin",
     "initialize_with_dynamic_plugins",
     "clear",
+    "clear_async",
     "initialize",
     "deregister",
     "list_kinds",
diff --git a/python/nemo_relay/plugin.pyi b/python/nemo_relay/plugin.pyi
index d5d387ff5..174b7f921 100644
--- a/python/nemo_relay/plugin.pyi
+++ b/python/nemo_relay/plugin.pyi
@@ -156,6 +156,7 @@ async def initialize_with_dynamic_plugins(
     dynamic_plugins: list[DynamicPluginActivationSpec | JsonObject],
 ) -> PluginHostActivation: ...
 def clear() -> None: ...
+async def clear_async() -> None: ...
 def plugin(config: PluginConfig | JsonObject) -> AsyncContextManager[ConfigReport]: ...
 def report() -> ConfigReport | None: ...
 def list_kinds() -> list[str]: ...
diff --git a/python/tests/test_adaptive.py b/python/tests/test_adaptive.py
index 899c3b2e5..d3386d38f 100644
--- a/python/tests/test_adaptive.py
+++ b/python/tests/test_adaptive.py
@@ -251,7 +251,7 @@ async def test_configure_report_and_clear(self):
             assert report["diagnostics"] == []
             assert plugin.report() == report
         finally:
-            plugin.clear()
+            await plugin.clear_async()
 
     async def test_configure_allows_normal_llm_call(self):
         await plugin.initialize(
@@ -277,7 +277,7 @@ def my_llm(_request: LLMRequest):
             result = await llm.execute("test-model", request, my_llm)
             assert result["response"] == "ok"
         finally:
-            plugin.clear()
+            await plugin.clear_async()
 
     async def test_python_plugin_is_called_from_core_plugin_system(self):
         class HeaderPlugin:
@@ -399,7 +399,7 @@ def finalizer():
                 assert chunk["x-python-llm-stream-exec"] == "priority:17"
             assert collected[0]["x-python-llm-stream-exec"] == "priority:17"
         finally:
-            plugin.clear()
+            await plugin.clear_async()
             plugin.deregister("python.test_plugin")
 
     def test_list_kinds_includes_registered_plugin(self):
diff --git a/python/tests/test_dynamic_plugin_host.py b/python/tests/test_dynamic_plugin_host.py
index 2f5ea351b..61f4be9ae 100644
--- a/python/tests/test_dynamic_plugin_host.py
+++ b/python/tests/test_dynamic_plugin_host.py
@@ -279,7 +279,7 @@ async def test_empty_dynamic_specs_preserve_static_initialization_path():
     assert plugin.report() is None
     report = await plugin.initialize(plugin.PluginConfig())
     assert report == {"diagnostics": []}
-    plugin.clear()
+    await plugin.clear_async()
 
 
 async def test_native_activation_context_owns_callbacks_and_close_is_idempotent(
@@ -417,7 +417,7 @@ async def test_activation_reports_conflicts_and_rolls_back_partial_loads(
         with pytest.raises(RuntimeError, match="active dynamic plugin host"):
             await plugin.initialize({})
         with pytest.raises(RuntimeError, match="active dynamic plugin host"):
-            plugin.clear()
+            await plugin.clear_async()
     finally:
         await activation.close()
 
diff --git a/python/tests/test_event_sanitizers.py b/python/tests/test_event_sanitizers.py
index 0c6cf56fb..2b692b5cc 100644
--- a/python/tests/test_event_sanitizers.py
+++ b/python/tests/test_event_sanitizers.py
@@ -5,6 +5,9 @@
 
 import asyncio
 import contextvars
+import subprocess
+import sys
+import textwrap
 from collections.abc import Iterator
 from concurrent.futures import ThreadPoolExecutor
 from typing import cast
@@ -24,6 +27,95 @@ def capture_events_fixture() -> Iterator[tuple[str, list[nemo_relay.Event]]]:
     subscribers.deregister(name)
 
 
+def test_plugin_clear_remains_available_without_running_event_loop():
+    plugin.clear()
+
+
+def test_plugin_clear_is_asyncio_safe_with_pending_sanitizer(tmp_path):
+    script = textwrap.dedent(
+        """
+        import asyncio
+
+        from nemo_relay import plugin, scope
+
+        delivered = []
+
+        async def main():
+            teardown_started = asyncio.Event()
+            allow_teardown = asyncio.Event()
+
+            class SanitizedSubscriberPlugin:
+                def validate(self, _config):
+                    return None
+
+                def register(self, _config, context):
+                    async def sanitize(_event, fields):
+                        teardown_started.set()
+                        await allow_teardown.wait()
+                        fields["data"] = {"sanitized": True}
+                        return fields
+
+                    context.register_mark_sanitize_guardrail("sanitize", 0, sanitize)
+                    context.register_subscriber("capture", delivered.append)
+
+            kind = "python.test_async_clear"
+            plugin.register(kind, SanitizedSubscriberPlugin())
+            try:
+                await plugin.initialize(
+                    plugin.PluginConfig(components=[plugin.ComponentSpec(kind=kind)])
+                )
+                scope.event("pending-clear", data={"raw": True})
+                try:
+                    plugin.clear()
+                except RuntimeError as error:
+                    assert "await plugin.clear_async()" in str(error)
+                else:
+                    raise AssertionError("plugin.clear() did not reject a running event loop")
+
+                first_clear = asyncio.create_task(plugin.clear_async())
+                await asyncio.wait_for(teardown_started.wait(), timeout=2)
+                first_clear.cancel()
+                try:
+                    await first_clear
+                except asyncio.CancelledError:
+                    pass
+                else:
+                    raise AssertionError("plugin.clear_async() did not propagate cancellation")
+
+                second_clear = asyncio.create_task(plugin.clear_async())
+                await asyncio.sleep(0.05)
+                assert not second_clear.done()
+
+                allow_teardown.set()
+                await asyncio.wait_for(second_clear, timeout=2)
+                assert plugin.report() is None
+                assert len(delivered) == 1
+                assert delivered[0].data == {"sanitized": True}
+
+                await plugin.initialize(plugin.PluginConfig())
+                assert plugin.report() is not None
+                await plugin.clear_async()
+                assert plugin.report() is None
+            finally:
+                allow_teardown.set()
+                if plugin.report() is not None:
+                    await plugin.clear_async()
+                plugin.deregister(kind)
+
+        asyncio.run(main())
+        """
+    )
+    completed = subprocess.run(
+        [sys.executable, "-c", script],
+        check=False,
+        capture_output=True,
+        cwd=tmp_path,
+        text=True,
+        timeout=5,
+    )
+    assert completed.returncode == 0, completed.stderr
+
+
 def test_global_mark_sanitizers_order_convert_fields_and_remove_values(capture_events):
     _capture_name, events = capture_events
     calls: list[tuple[str, object]] = []
@@ -416,11 +508,11 @@ def sanitize(_event: nemo_relay.Event, fields: EventSanitizeFields) -> EventSani
         await plugin.initialize(plugin.PluginConfig(components=[plugin.ComponentSpec(kind=kind)]))
         scope.event("configured", data={"raw": True})
         await subscribers.flush_async()
-        plugin.clear()
+        await plugin.clear_async()
         scope.event("cleared", data={"raw": True})
         await subscribers.flush_async()
     finally:
-        plugin.clear()
+        await plugin.clear_async()
         plugin.deregister(kind)
 
     marks = {event.name: event for event in events if event.kind == "mark"}
@@ -454,5 +546,5 @@ def sanitize(_event: nemo_relay.Event, fields: EventSanitizeFields) -> EventSani
         await subscribers.flush_async()
         assert events[-1].data == {"raw": True}
     finally:
-        plugin.clear()
+        await plugin.clear_async()
         plugin.deregister(kind)
diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py
index 5e90648d3..68b7dd3b6 100644
--- a/python/tests/test_observability_plugin.py
+++ b/python/tests/test_observability_plugin.py
@@ -223,7 +223,7 @@ async def test_atof_stream_sink_snapshots_header_env(
                 with scope.scope("python-header-env-agent", ScopeType.Agent) as handle:
                     scope.event("python-header-env-mark", handle=handle, data={"step": 1})
             finally:
-                plugin.clear()
+                await plugin.clear_async()
             requests = capture.wait_for_requests(3)
 
         assert len(requests) == 3
@@ -295,7 +295,7 @@ def _inner():
             try:
                 handle = _inner()
             finally:
-                plugin.clear()
+                await plugin.clear_async()
 
         lines = (tmp_path / "events.jsonl").read_text().strip().splitlines()
         assert len(lines) == 3
@@ -400,7 +400,7 @@ def release_response() -> None:
             response_thread.start()
             teardown_started.set()
             started_at = time.monotonic()
-            plugin.clear()
+            await plugin.clear_async()
             cleared = True
             assert time.monotonic() - started_at < 2
             response_thread.join(timeout=2)
@@ -419,7 +419,7 @@ def release_response() -> None:
         finally:
             allow_response.set()
             if not cleared:
-                plugin.clear()
+                await plugin.clear_async()
             server.shutdown()
             server_thread.join(timeout=5)
             server.server_close()
@@ -434,7 +434,7 @@ async def test_atif_flushes_open_agent_on_clear(self, tmp_path):
         )
         handle = scope.push("python-open-agent", ScopeType.Agent)
         try:
-            plugin.clear()
+            await plugin.clear_async()
             assert (tmp_path / f"nemo-relay-atif-{handle.uuid}.json").exists()
         finally:
             scope.pop(handle)
@@ -464,7 +464,7 @@ async def test_atif_splits_multiple_top_level_agent_scopes(self, tmp_path):
             with scope.scope("python-second-agent", ScopeType.Agent) as second:
                 scope.event("python-second-mark", handle=second, data={"agent": "second"})
         finally:
-            plugin.clear()
+            await plugin.clear_async()
 
         files = sorted(tmp_path.glob("trajectory-*.json"))
         assert len(files) == 2