fix: interceptor sync bridge + async detection, example fixes

.lint_baselines/falsey_clobber.json

@@ -41,20 +41,6 @@
     "axonflow/client.py:845:20",
     "axonflow/client.py:931:20",
     "axonflow/execution.py:205:19",
-    "axonflow/interceptors/anthropic.py:134:43",
-    "axonflow/interceptors/anthropic.py:161:43",
-    "axonflow/interceptors/bedrock.py:175:43",
-    "axonflow/interceptors/bedrock.py:206:43",
-    "axonflow/interceptors/gemini.py:141:39",
-    "axonflow/interceptors/gemini.py:167:43",
-    "axonflow/interceptors/gemini.py:228:39",
-    "axonflow/interceptors/gemini.py:252:43",
-    "axonflow/interceptors/ollama.py:121:43",
-    "axonflow/interceptors/ollama.py:150:43",
-    "axonflow/interceptors/ollama.py:202:43",
-    "axonflow/interceptors/ollama.py:228:43",
-    "axonflow/interceptors/openai.py:110:43",
-    "axonflow/interceptors/openai.py:137:43",
     "axonflow/masfeat.py:296:23",
     "axonflow/masfeat.py:297:24",
     "axonflow/masfeat.py:298:25",

CHANGELOG.md

@@ -9,6 +9,48 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
  and tag v{X.Y.Z}. The release workflow's preflight checks the section
  header matches the tag. -->
 
+## [8.5.1] - 2026-07-09 — Interceptor sync bridge + async-client detection + example fixes
+
+Hostile-testing sweep ahead of the BukuWarung integration
+(getaxonflow/axonflow-enterprise#2861).
+
+### Fixed
+
+- **Interceptor sync wrap paths no longer crash inside a running event
+  loop.** All five provider interceptors (openai, anthropic, bedrock,
+  gemini, ollama) ran the async governance check with
+  `loop.run_until_complete`, which raises `RuntimeError: This event loop is
+  already running` whenever the caller sits inside a running loop (FastAPI
+  handler, Jupyter, an async app driving a sync provider client) — the
+  async-adapter-bypass class: the governance check crashed instead of
+  completing. The shared `run_coroutine_sync` bridge now executes
+  governance on one persistent background loop (daemon thread) and blocks
+  the call site until the verdict is in; repeated calls reuse the same
+  loop, so the HTTP pool stays valid (a throwaway-loop bridge broke the
+  second call with "Event loop is closed").
+- **`AsyncOpenAI`/`AsyncAnthropic` clients are detected as async again.**
+  openai>=1 and anthropic decorate their async `create` methods with
+  plain-`def` wrappers (`@required_args`), so `asyncio.iscoroutinefunction`
+  returned False and async clients were silently given the SYNC wrap path.
+  `is_async_callable` follows `__wrapped__` (`inspect.unwrap`) to the real
+  `async def`.
+- **Examples pass `AXONFLOW_USER_TOKEN` and fail honestly.** Enterprise
+  stacks (`DEPLOYMENT_MODE=enterprise`) validate user tokens as JWTs;
+  `quickstart`, `gateway_mode` and `openai_integration` passed hardcoded
+  non-JWT literals and 401'd. All three now read `AXONFLOW_USER_TOKEN`.
+  `gateway_mode`'s blocked-request demo uses a stacked-SQLi query (blocked
+  on every stack posture — PII policies default to redact, not block) and
+  exits non-zero if unexpectedly approved; `openai_integration` prints the
+  outcome of its policy-block probe instead of ending silently.
+
+### Added
+
+- `runtime-e2e/interceptor_sync_bridge/` — live-agent assertion driving a
+  REAL `openai.OpenAI` client through `wrap_openai_client` from inside a
+  running loop and from plain sync code (blocked verdict enforced both
+  ways, no RuntimeError), plus async-detection assertions for
+  `AsyncOpenAI`.
+
 ## [8.5.0] - 2026-06-09 — Decision Mode PEP: decide → fulfill → forward
 
 Adds the SDK analog of the platform PEP client (`platform/shared/pep`,

axonflow/_version.py

@@ -1,3 +1,3 @@
 """Single source of truth for the AxonFlow SDK version."""
 
-__version__ = "8.5.0"
+__version__ = "8.5.1"

axonflow/interceptors/_sync_bridge.py

@@ -0,0 +1,90 @@
+"""Shared sync-over-async bridge for interceptor governance calls.
+
+Every provider interceptor has a sync wrap path that must run the async
+governance check (``AxonFlow.proxy_llm_call``) to completion BEFORE the
+provider call. The old per-interceptor pattern::
+
+    loop = asyncio.get_event_loop()
+    response = loop.run_until_complete(...)
+
+raises ``RuntimeError: This event loop is already running`` whenever the
+caller sits inside a running loop (FastAPI handler, Jupyter, an async app
+driving a sync provider client). A governance check that crashes instead of
+completing is the async-adapter-bypass failure class: callers who catch and
+continue would proceed ungoverned.
+
+``run_coroutine_sync`` executes the coroutine on ONE persistent background
+event loop (daemon thread, created lazily) and blocks the caller until the
+verdict is in — whether or not the calling thread has a running loop. A
+persistent loop matters: the AxonFlow client's underlying HTTP pool is
+loop-affine, so running each governance call on a throwaway loop
+(``asyncio.run`` per call) breaks the SECOND call with "Event loop is
+closed". Either way the governance result — allow, block, or error — is
+returned/raised synchronously and can never be skipped.
+
+Sharing note: an ``AxonFlow`` instance's HTTP pool binds to the first loop
+that uses it. An instance driven through this bridge (sync-wrapped provider
+clients) must not ALSO be awaited on a caller's own loop — construct one
+instance per context (one for sync-wrapped clients, one for async-wrapped).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import threading
+from collections.abc import Coroutine
+from typing import Any, TypeVar
+
+T = TypeVar("T")
+
+
+class _BridgeState:
+    """Holder for the shared background loop (avoids module-global rebinding)."""
+
+    lock = threading.Lock()
+    loop: asyncio.AbstractEventLoop | None = None
+
+
+def is_async_callable(fn: Any) -> bool:
+    """Detect coroutine functions THROUGH ``functools.wraps`` decorators.
+
+    Modern provider SDKs (openai>=1, anthropic) decorate their async
+    ``create`` methods with plain-``def`` wrappers (e.g. ``@required_args``)
+    that return the coroutine. ``asyncio.iscoroutinefunction`` on the bound
+    method is then False, so an ``AsyncOpenAI`` client silently got the SYNC
+    wrap path. ``inspect.unwrap`` follows ``__wrapped__`` to the real
+    ``async def``.
+    """
+    if inspect.iscoroutinefunction(fn):
+        return True
+    try:
+        return inspect.iscoroutinefunction(inspect.unwrap(fn))
+    except ValueError:  # __wrapped__ cycle — be conservative
+        return False
+
+
+def _get_bridge_loop() -> asyncio.AbstractEventLoop:
+    """Lazily start the shared background loop (daemon thread)."""
+    with _BridgeState.lock:
+        if _BridgeState.loop is None or _BridgeState.loop.is_closed():
+            loop = asyncio.new_event_loop()
+            thread = threading.Thread(
+                target=loop.run_forever,
+                name="axonflow-interceptor-bridge",
+                daemon=True,
+            )
+            thread.start()
+            _BridgeState.loop = loop
+        return _BridgeState.loop
+
+
+def run_coroutine_sync(coro: Coroutine[Any, Any, T]) -> T:
+    """Run *coro* to completion from sync code, inside or outside a loop.
+
+    Always executes on the shared background loop so repeated calls reuse
+    the same loop (and therefore the same HTTP connection pool), and blocks
+    the calling thread until the result — the verdict cannot be skipped.
+    """
+    loop = _get_bridge_loop()
+    return asyncio.run_coroutine_threadsafe(coro, loop).result()

axonflow/interceptors/anthropic.py

@@ -24,12 +24,12 @@
 
 from __future__ import annotations
 
-import asyncio
 from collections.abc import Callable
 from functools import wraps
 from typing import TYPE_CHECKING, Any, TypeVar
 
 from axonflow.exceptions import PolicyViolationError
+from axonflow.interceptors._sync_bridge import is_async_callable, run_coroutine_sync
 from axonflow.interceptors.base import BaseInterceptor
 
 if TYPE_CHECKING:
@@ -99,16 +99,7 @@ def _extract_prompt(kwargs: dict[str, Any]) -> str:
                             parts.append(block.get("text", ""))
         return " ".join(parts)
 
-    def _get_loop() -> asyncio.AbstractEventLoop:
-        """Get or create event loop."""
-        try:
-            return asyncio.get_event_loop()
-        except RuntimeError:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            return loop
-
-    if asyncio.iscoroutinefunction(original_create):
+    if is_async_callable(original_create):
 
         @wraps(original_create)
         async def async_wrapped_create(*args: Any, **kwargs: Any) -> Any:
@@ -131,7 +122,11 @@ async def async_wrapped_create(*args: Any, **kwargs: Any) -> Any:
             )
 
             if response.blocked:
-                raise PolicyViolationError(response.block_reason or "Request blocked by policy")
+                raise PolicyViolationError(
+                    response.block_reason
+                    if response.block_reason is not None
+                    else "Request blocked by policy"
+                )
 
             # Call original
             return await original_create(*args, **kwargs)
@@ -144,8 +139,7 @@ def sync_wrapped_create(*args: Any, **kwargs: Any) -> Any:
             prompt = _extract_prompt(kwargs)
 
             # Check with AxonFlow (sync)
-            loop = _get_loop()
-            response = loop.run_until_complete(
+            response = run_coroutine_sync(
                 axonflow.proxy_llm_call(
                     user_token=user_token,
                     query=prompt,
@@ -158,7 +152,11 @@ def sync_wrapped_create(*args: Any, **kwargs: Any) -> Any:
             )
 
             if response.blocked:
-                raise PolicyViolationError(response.block_reason or "Request blocked by policy")
+                raise PolicyViolationError(
+                    response.block_reason
+                    if response.block_reason is not None
+                    else "Request blocked by policy"
+                )
 
             # Call original
             return original_create(*args, **kwargs)

axonflow/interceptors/bedrock.py

@@ -26,13 +26,13 @@
 
 from __future__ import annotations
 
-import asyncio
 import json
 from collections.abc import Callable
 from functools import wraps
 from typing import TYPE_CHECKING, Any, TypeVar
 
 from axonflow.exceptions import PolicyViolationError
+from axonflow.interceptors._sync_bridge import run_coroutine_sync
 from axonflow.interceptors.base import BaseInterceptor
 
 if TYPE_CHECKING:
@@ -139,15 +139,6 @@ def _extract_prompt(body: Any, _model_id: str) -> str:
                 pass
         return ""
 
-    def _get_loop() -> asyncio.AbstractEventLoop:
-        """Get or create event loop."""
-        try:
-            return asyncio.get_event_loop()
-        except RuntimeError:
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            return loop
-
     # Wrap invoke_model
     if hasattr(bedrock_client, "invoke_model"):
         original_invoke = bedrock_client.invoke_model
@@ -158,8 +149,7 @@ def sync_wrapped_invoke(*args: Any, **kwargs: Any) -> Any:
             body = kwargs.get("body", "")
             prompt = _extract_prompt(body, model_id)
 
-            loop = _get_loop()
-            response = loop.run_until_complete(
+            response = run_coroutine_sync(
                 axonflow.proxy_llm_call(
                     user_token=user_token,
                     query=prompt,
@@ -172,7 +162,11 @@ def sync_wrapped_invoke(*args: Any, **kwargs: Any) -> Any:
             )
 
             if response.blocked:
-                raise PolicyViolationError(response.block_reason or "Request blocked by policy")
+                raise PolicyViolationError(
+                    response.block_reason
+                    if response.block_reason is not None
+                    else "Request blocked by policy"
+                )
 
             return original_invoke(*args, **kwargs)
 
@@ -188,8 +182,7 @@ def sync_wrapped_stream(*args: Any, **kwargs: Any) -> Any:
             body = kwargs.get("body", "")
             prompt = _extract_prompt(body, model_id)
 
-            loop = _get_loop()
-            response = loop.run_until_complete(
+            response = run_coroutine_sync(
                 axonflow.proxy_llm_call(
                     user_token=user_token,
                     query=prompt,
@@ -203,7 +196,11 @@ def sync_wrapped_stream(*args: Any, **kwargs: Any) -> Any:
             )
 
             if response.blocked:
-                raise PolicyViolationError(response.block_reason or "Request blocked by policy")
+                raise PolicyViolationError(
+                    response.block_reason
+                    if response.block_reason is not None
+                    else "Request blocked by policy"
+                )
 
             return original_stream(*args, **kwargs)