interceptor.py

"""httpx-level monkey-patch interceptor for LLM API calls.

Similar to OpenClaw's global fetch interception, this module patches
httpx.Client.send() to intercept all outbound HTTP requests made by
OpenAI / Anthropic / other LLM SDKs.

Capabilities:
  - Intercept LLM API requests before they leave the process
  - Block requests and return provider-compatible fake responses
  - Intercept non-streaming responses after receipt
  - Replace responses with security-policy messages
  - Fail-Open: any error in the interceptor allows the request through

Install via:
    interceptor = HttpxInterceptor()
    interceptor.install(auth_service)
"""
from __future__ import annotations

import json
import logging
import threading
import time
from typing import Any

try:
    import httpx
except ImportError:
    httpx = None  # type: ignore[assignment]

logger = logging.getLogger("hermes.plugin.security")

# ------------------------------------------------------------------
# Known LLM API patterns
# ------------------------------------------------------------------

_LLM_API_HOSTS: tuple[str, ...] = (
    "api.openai.com",
    "api.anthropic.com",
    "openrouter.ai",
    "api.deepseek.com",
    "generativelanguage.googleapis.com",
    "api.groq.com",
    "api.mistral.ai",
    "api.cohere.ai",
    "api.together.xyz",
    "api.fireworks.ai",
)

_CHAT_ENDPOINTS: tuple[str, ...] = (
    "/chat/completions",
    "/v1/messages",
    "/v1/responses",
)

# ------------------------------------------------------------------
# Module-level state (set by install / cleared by uninstall)
# ------------------------------------------------------------------

_original_send = None          # type: ignore[assignment]
_original_async_send = None    # type: ignore[assignment]
_interceptor = None            # type: HttpxInterceptor | None

# ------------------------------------------------------------------
# Install-time re-entrancy guard
# ------------------------------------------------------------------
# Marker attribute on the patched function to detect repeated install()
# calls (e.g. plugin hot-reload / module reimport).  True originals are
# stashed on the httpx class itself so they survive module reloads and
# _original_send always points to the real unpatched method.
_INTERCEPTOR_MARKER = "_hermes_security_interceptor"
_TRUE_ORIGINAL_SEND_ATTR = "_hermes_true_original_send"
_TRUE_ORIGINAL_ASYNC_SEND_ATTR = "_hermes_true_original_async_send"


# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------

def _is_llm_api_call(request: Any) -> bool:
    """Return True if *request* targets a known LLM provider endpoint."""
    try:
        host = (request.url.host or "").lower()
        path = str(getattr(request.url, "path", "")).lower()
        url_str = str(request.url).lower()
    except Exception:
        return False

    for pattern in _LLM_API_HOSTS:
        if pattern in host or pattern in url_str:
            return True

    for ep in _CHAT_ENDPOINTS:
        if ep in path:
            return True

    return False


def _extract_request_summary(request: Any) -> dict[str, Any]:
    """Pull key fields out of an httpx.Request for logging / APS payload."""
    info: dict[str, Any] = {
        "method": str(getattr(request, "method", "?")),
        "url": str(getattr(request, "url", "?")),
        "host": "",
    }
    try:
        info["host"] = str(request.url.host or "")
    except Exception:
        pass

    try:
        if request.content:
            body = json.loads(request.content.decode("utf-8"))
            info["model"] = body.get("model", "")
            info["stream"] = body.get("stream", False)
            msgs = body.get("messages") or []
            info["message_count"] = len(msgs)
    except Exception:
        pass

    return info


def _build_fake_openai_response(message: str, model: str = "security-policy") -> bytes:
    """Construct an OpenAI-compatible chat.completion JSON body."""
    return json.dumps(
        {
            "id": f"security-block-{int(time.time())}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": model,
            "choices": [
                {
                    "index": 0,
                    "message": {"role": "assistant", "content": message},
                    "finish_reason": "stop",
                }
            ],
            "usage": {
                "prompt_tokens": 0,
                "completion_tokens": 0,
                "total_tokens": 0,
            },
        },
        ensure_ascii=False,
    ).encode("utf-8")


def _build_fake_anthropic_response(message: str, model: str = "security-policy") -> bytes:
    """Construct an Anthropic-compatible messages JSON body."""
    return json.dumps(
        {
            "id": f"msg_security_{int(time.time())}",
            "type": "message",
            "role": "assistant",
            "content": [{"type": "text", "text": message}],
            "model": model,
            "stop_reason": "end_turn",
            "usage": {"input_tokens": 0, "output_tokens": 0},
        },
        ensure_ascii=False,
    ).encode("utf-8")


def _build_fake_openai_sse_response(message: str, model: str = "security-policy") -> bytes:
    """Construct an OpenAI-compatible streaming (SSE) response."""
    chunk_id = f"chatcmpl-security-{int(time.time())}"
    created = int(time.time())

    # First chunk: role
    chunk1 = json.dumps({
        "id": chunk_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [{
            "index": 0,
            "delta": {"role": "assistant", "content": ""},
            "finish_reason": None,
        }],
    }, ensure_ascii=False)

    # Second chunk: content
    chunk2 = json.dumps({
        "id": chunk_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [{
            "index": 0,
            "delta": {"content": message},
            "finish_reason": None,
        }],
    }, ensure_ascii=False)

    # Third chunk: finish
    chunk3 = json.dumps({
        "id": chunk_id,
        "object": "chat.completion.chunk",
        "created": created,
        "model": model,
        "choices": [{
            "index": 0,
            "delta": {},
            "finish_reason": "stop",
        }],
    }, ensure_ascii=False)

    sse_body = f"data: {chunk1}\n\ndata: {chunk2}\n\ndata: {chunk3}\n\ndata: [DONE]\n\n"
    return sse_body.encode("utf-8")


def _build_fake_anthropic_sse_response(message: str, model: str = "security-policy") -> bytes:
    """Construct an Anthropic-compatible streaming (SSE) response."""
    msg_id = f"msg_security_{int(time.time())}"

    events = []

    # message_start
    events.append('event: message_start\ndata: ' + json.dumps({
        "type": "message_start",
        "message": {
            "id": msg_id,
            "type": "message",
            "role": "assistant",
            "content": [],
            "model": model,
            "stop_reason": None,
            "usage": {"input_tokens": 0, "output_tokens": 0},
        },
    }, ensure_ascii=False))

    # content_block_start
    events.append('event: content_block_start\ndata: ' + json.dumps({
        "type": "content_block_start",
        "index": 0,
        "content_block": {"type": "text", "text": ""},
    }, ensure_ascii=False))

    # content_block_delta with the message
    events.append('event: content_block_delta\ndata: ' + json.dumps({
        "type": "content_block_delta",
        "index": 0,
        "delta": {"type": "text_delta", "text": message},
    }, ensure_ascii=False))

    # content_block_stop
    events.append('event: content_block_stop\ndata: ' + json.dumps({
        "type": "content_block_stop",
        "index": 0,
    }, ensure_ascii=False))

    # message_delta
    events.append('event: message_delta\ndata: ' + json.dumps({
        "type": "message_delta",
        "delta": {"stop_reason": "end_turn"},
        "usage": {"output_tokens": 1},
    }, ensure_ascii=False))

    # message_stop
    events.append('event: message_stop\ndata: ' + json.dumps({
        "type": "message_stop",
    }, ensure_ascii=False))

    sse_body = "\n\n".join(events) + "\n\n"
    return sse_body.encode("utf-8")


def _make_blocked_response(request: Any, reason: str, model: str, is_streaming: bool = False) -> Any:
    """Return a httpx.Response that the SDK will accept as a normal reply."""
    url_str = str(request.url).lower()
    is_anthropic = "anthropic" in url_str

    if is_streaming:
        if is_anthropic:
            body = _build_fake_anthropic_sse_response(f"[Security Policy] {reason}", model)
        else:
            body = _build_fake_openai_sse_response(f"[Security Policy] {reason}", model)
        content_type = "text/event-stream"
    else:
        if is_anthropic:
            body = _build_fake_anthropic_response(f"[Security Policy] {reason}", model)
        else:
            body = _build_fake_openai_response(f"[Security Policy] {reason}", model)
        content_type = "application/json"

    return httpx.Response(
        status_code=200,
        headers={"content-type": content_type},
        content=body,
        request=request,
    )


def _extract_text_from_response_body(body_text: str, is_anthropic: bool) -> str:
    """Extract assistant text content from a non-streaming JSON response body."""
    try:
        data = json.loads(body_text)
        if is_anthropic:
            # Anthropic: {"content": [{"type": "text", "text": "..."}]}
            content = data.get("content", [])
            if isinstance(content, list):
                return "".join(
                    block.get("text", "")
                    for block in content
                    if isinstance(block, dict) and block.get("type") == "text"
                )
        else:
            # OpenAI: {"choices": [{"message": {"content": "..."}}]}
            choices = data.get("choices", [])
            if choices:
                return choices[0].get("message", {}).get("content", "") or ""
    except Exception:
        pass
    return ""


def _extract_text_from_sse_body(body_text: str, is_anthropic: bool) -> str:
    """Extract assistant text content from an SSE (streaming) response body."""
    texts: list[str] = []
    for line in body_text.split("\n"):
        line = line.strip()
        if not line.startswith("data: ") or line == "data: [DONE]":
            continue
        try:
            data = json.loads(line[6:])
            if is_anthropic:
                # Anthropic: content_block_delta with text_delta
                if data.get("type") == "content_block_delta":
                    delta = data.get("delta", {})
                    if delta.get("type") == "text_delta":
                        text = delta.get("text", "")
                        if text:
                            texts.append(text)
            else:
                # OpenAI: choices[0].delta.content
                choices = data.get("choices", [])
                if choices:
                    content = choices[0].get("delta", {}).get("content")
                    if content:
                        texts.append(content)
        except Exception:
            continue
    return "".join(texts)


def _make_hint_response(request: Any, response: Any, original_body: str,
                       hint_content: str, model: str,
                       is_streaming: bool = False) -> Any:
    """Build a response that appends *hint_content* to the original response text.

    Unlike block (which replaces the entire body), hint preserves the original
    assistant output and only appends a security tip.
    Supports both non-streaming (JSON) and streaming (SSE) responses.
    """
    url_str = str(request.url).lower()
    is_anthropic = "anthropic" in url_str

    if is_streaming:
        original_text = _extract_text_from_sse_body(original_body, is_anthropic)
    else:
        original_text = _extract_text_from_response_body(original_body, is_anthropic)
    combined_text = original_text + hint_content

    if is_streaming:
        if is_anthropic:
            body = _build_fake_anthropic_sse_response(combined_text, model)
        else:
            body = _build_fake_openai_sse_response(combined_text, model)
        content_type = "text/event-stream"
    else:
        if is_anthropic:
            body = _build_fake_anthropic_response(combined_text, model)
        else:
            body = _build_fake_openai_response(combined_text, model)
        content_type = "application/json"

    return httpx.Response(
        status_code=response.status_code,
        headers={"content-type": content_type},
        content=body,
        request=request,
    )


# ------------------------------------------------------------------
# Core interceptor class
# ------------------------------------------------------------------

class HttpxInterceptor:
    """Manages the httpx monkey-patch lifecycle."""

    def __init__(self) -> None:
        self._installed = False
        self._lock = threading.Lock()
        self._auth_service: Any = None

    # ---------- public API ----------

    def install(self, auth_service: Any) -> None:
        """Patch httpx.Client.send (and AsyncClient.send) globally.

        Includes hot-reload safety:
        - If the patch is already in place (detected via marker attribute),
          we skip re-patching and only update the interceptor / auth refs.
        - The *true* original methods are stashed on the httpx classes so
          they survive module reloads and new HttpxInterceptor instances.
        """
        global _original_send, _original_async_send, _interceptor

        if httpx is None:
            logger.error("[SECURITY][INTERCEPT] httpx not available — interceptor NOT installed")
            return

        self._auth_service = auth_service

        # --- Hot-reload guard: already patched? ---
        already_patched = getattr(httpx.Client.send, _INTERCEPTOR_MARKER, False)
        if already_patched:
            # Patch is in place from a previous install() — just refresh refs.
            _original_send = getattr(httpx.Client, _TRUE_ORIGINAL_SEND_ATTR, _original_send)
            _original_async_send = getattr(
                httpx.AsyncClient, _TRUE_ORIGINAL_ASYNC_SEND_ATTR, _original_async_send,
            )
            _interceptor = self
            self._installed = True
            logger.warning(
                "[SECURITY][INTERCEPT] interceptor already patched (hot-reload?) — "
                "refs refreshed, skipping re-patch"
            )
            return

        if self._installed:
            logger.warning("[SECURITY][INTERCEPT] interceptor already installed, skipping")
            return

        # --- Sync patch ---
        # Stash the true original so it survives module reloads.
        if not hasattr(httpx.Client, _TRUE_ORIGINAL_SEND_ATTR):
            setattr(httpx.Client, _TRUE_ORIGINAL_SEND_ATTR, httpx.Client.send)
        _original_send = getattr(httpx.Client, _TRUE_ORIGINAL_SEND_ATTR)

        httpx.Client.send = _patched_send  # type: ignore[assignment]
        setattr(_patched_send, _INTERCEPTOR_MARKER, True)  # mark for detection

        # --- Async patch ---
        if not hasattr(httpx.AsyncClient, _TRUE_ORIGINAL_ASYNC_SEND_ATTR):
            setattr(httpx.AsyncClient, _TRUE_ORIGINAL_ASYNC_SEND_ATTR, httpx.AsyncClient.send)
        _original_async_send = getattr(httpx.AsyncClient, _TRUE_ORIGINAL_ASYNC_SEND_ATTR)

        httpx.AsyncClient.send = _patched_async_send  # type: ignore[assignment]
        setattr(_patched_async_send, _INTERCEPTOR_MARKER, True)

        _interceptor = self
        self._installed = True

        logger.info(
            "[SECURITY][INTERCEPT] httpx interceptor INSTALLED — "
            "all LLM API calls (sync+async) will be monitored"
        )

    def uninstall(self) -> None:
        """Restore original httpx methods."""
        global _original_send, _original_async_send, _interceptor

        if not self._installed:
            return

        if _original_send is not None:
            httpx.Client.send = _original_send  # type: ignore[assignment]
            _original_send = None
        if _original_async_send is not None:
            httpx.AsyncClient.send = _original_async_send  # type: ignore[assignment]
            _original_async_send = None

        _interceptor = None
        self._installed = False
        logger.info("[SECURITY][INTERCEPT] httpx interceptor UNINSTALLED — original methods restored")

    @property
    def installed(self) -> bool:
        return self._installed

    # ---------- APS check ----------

    def check_request(self, request: Any) -> dict[str, Any] | None:
        """Ask APS whether this LLM HTTP request should be allowed.

        Returns None (allow / fail-open), or {"action": "block", "content": "..."}.
        """
        if not self._auth_service or not self._auth_service.is_ready():
            logger.debug("[SECURITY][INTERCEPT] auth not ready -> allow (fail-open)")
            return None

        try:
            from .check import call_remote_security_check
            from .config import config
            from .utils import filter_sensitive_headers

            # Extract raw request info matching OpenClaw format
            url = str(request.url)
            method = str(request.method)

            # Headers: convert httpx headers to dict, filter sensitive ones
            raw_headers = dict(request.headers)
            safe_headers = filter_sensitive_headers(raw_headers)

            # Body: raw request body text
            req_body_text = ""
            try:
                if request.content:
                    req_body_text = request.content.decode("utf-8")
            except Exception:
                req_body_text = ""

            llm_payload = {
                "url": url,
                "method": method,
                "headers": safe_headers,
                "body": req_body_text,
            }

            payload = {
                "req_type": "llm",
                "agent_session_id": "",
                "llm_run_id": "",
                "llm_payload": llm_payload,
            }

            logger.info(
                "[SECURITY][INTERCEPT] -> APS pre-request check: url=%s method=%s",
                url, method,
            )
            logger.info(
                "[SECURITY][INTERCEPT] -> APS req payload headers: %s",
                json.dumps(safe_headers, ensure_ascii=False),
            )
            logger.info(
                "[SECURITY][INTERCEPT] -> APS req payload body (preview): %s",
                req_body_text[:2000] if req_body_text else "(empty)",
            )

            result = call_remote_security_check(
                endpoint="/v1/agent/llm_check",
                direction="req",
                payload=payload,
                timeout=config.check_timeout,
            )
            return result

        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] APS request check error (fail-open): %s", exc)
            return None

    def check_response(self, request: Any, response: Any) -> dict[str, Any] | None:
        """Ask APS whether this LLM HTTP response should be allowed."""
        if not self._auth_service or not self._auth_service.is_ready():
            return None

        try:
            from .check import call_remote_security_check
            from .config import config
            from .utils import filter_sensitive_headers

            url = str(request.url)

            # Response headers: filter sensitive ones
            resp_headers = dict(response.headers)
            safe_resp_headers = filter_sensitive_headers(resp_headers)

            # Response body: full text
            resp_body_text = ""
            try:
                resp_body_text = response.text
            except Exception:
                resp_body_text = ""

            # Note: response direction does NOT include "method" field
            llm_payload = {
                "url": url,
                "headers": safe_resp_headers,
                "body": resp_body_text,
            }

            payload = {
                "req_type": "llm",
                "agent_session_id": "",
                "llm_run_id": "",
                "llm_payload": llm_payload,
            }

            logger.info(
                "[SECURITY][INTERCEPT] -> APS post-response check: url=%s status=%d",
                url, response.status_code,
            )
            logger.info(
                "[SECURITY][INTERCEPT] -> APS resp payload headers: %s",
                json.dumps(safe_resp_headers, ensure_ascii=False),
            )
            logger.info(
                "[SECURITY][INTERCEPT] -> APS resp payload body (preview): %s",
                resp_body_text[:2000] if resp_body_text else "(empty)",
            )

            result = call_remote_security_check(
                endpoint="/v1/agent/llm_check",
                direction="resp",
                payload=payload,
                timeout=config.check_timeout,
            )
            return result

        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] APS response check error (fail-open): %s", exc)
            return None


# ------------------------------------------------------------------
# Patched send (sync)
# ------------------------------------------------------------------

def _patched_send(self: Any, request: Any, *args: Any, **kwargs: Any) -> Any:
    """Drop-in replacement for httpx.Client.send."""
    global _original_send, _interceptor

    # Fast path: not an LLM call -> pass through immediately
    if not _is_llm_api_call(request):
        return _original_send(self, request, *args, **kwargs)
    summary = _extract_request_summary(request)
    logger.info(
        "[SECURITY][INTERCEPT] >> Detected LLM API call: %s %s "
        "(model=%s stream=%s msgs=%s)",
        summary.get("method", "?"),
        summary.get("url", "?"),
        summary.get("model", "?"),
        summary.get("stream", "?"),
        summary.get("message_count", "?"),
    )

    # Log full request body for debugging
    try:
        if request.content:
            req_body_str = request.content.decode("utf-8")
            logger.info(
                "[SECURITY][INTERCEPT] >> Request body (%d bytes): %s",
                len(req_body_str),
                req_body_str[:3000],
            )
    except Exception:
        pass

    # ---- PRE-REQUEST CHECK ----
    req_action = "allow"
    req_content = None

    if _interceptor is not None:
        try:
            pre = _interceptor.check_request(request)
        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] pre-check exception (fail-open): %s", exc)
            pre = None

        if pre and pre.get("action") == "block":
            reason = pre.get("content", "Request blocked by security policy")
            # Detect if original request wants streaming
            is_streaming = summary.get("stream", False)
            logger.warning(
                "[SECURITY][INTERCEPT] !! REQUEST BLOCKED: %s %s reason=%s (streaming=%s)",
                summary.get("method", ""),
                summary.get("url", ""),
                reason,
                is_streaming,
            )
            fake = _make_blocked_response(
                request, reason,
                summary.get("model", "security-policy"),
                is_streaming=is_streaming,
            )
            logger.info(
                "[SECURITY][INTERCEPT] << Returning fake %s response (blocked, %d bytes)",
                "SSE" if is_streaming else "JSON",
                len(fake.content),
            )
            return fake

        if pre:
            req_action = pre.get("action", "allow")
            req_content = pre.get("content")
            logger.info(
                "[SECURITY][INTERCEPT] pre-check result: action=%s has_content=%s",
                pre.get("action", "?"),
                bool(req_content),
            )
            logger.debug(
                "[SECURITY][INTERCEPT] pre-check full response: %s",
                json.dumps(pre, ensure_ascii=False)[:500],
            )
        else:
            logger.info(
                "[SECURITY][INTERCEPT] pre-check returned None (allow/fail-open)",
            )

    # ---- FORWARD ORIGINAL REQUEST ----
    logger.info("[SECURITY][INTERCEPT] >> Forwarding to origin: %s", summary.get("url", ""))

    try:
        response = _original_send(self, request, *args, **kwargs)
    except Exception as exc:
        logger.info("[SECURITY][INTERCEPT] << Request failed with exception: %s", exc)
        raise

    logger.info(
        "[SECURITY][INTERCEPT] << Response: HTTP %d  content-type=%s  streaming=%s",
        response.status_code,
        response.headers.get("content-type", "?"),
        "text/event-stream" in response.headers.get("content-type", ""),
    )

    # ---- POST-RESPONSE CHECK ----
    is_streaming = "text/event-stream" in response.headers.get("content-type", "")

    # Log response body for non-streaming responses
    if not is_streaming:
        try:
            resp_text = response.text
            logger.info(
                "[SECURITY][INTERCEPT] << Response body (%d bytes): %s",
                len(resp_text),
                resp_text[:3000],
            )
        except Exception:
            pass

    resp_action = "allow"
    resp_content = None

    if _interceptor is not None and not is_streaming:
        try:
            post = _interceptor.check_response(request, response)
        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] post-check exception (fail-open): %s", exc)
            post = None

        if post and post.get("action") == "block":
            reason = post.get("content", "Response blocked by security policy")
            logger.warning(
                "[SECURITY][INTERCEPT] !! RESPONSE BLOCKED: reason=%s",
                reason,
            )
            fake = _make_blocked_response(
                request, reason,
                summary.get("model", "security-policy"),
                is_streaming=False,  # post-check only runs for non-streaming
            )
            logger.info("[SECURITY][INTERCEPT] << Replacing response with security message")
            return fake

        if post:
            resp_action = post.get("action", "allow")
            resp_content = post.get("content")
            logger.info(
                "[SECURITY][INTERCEPT] post-check result: action=%s has_content=%s",
                post.get("action", "?"),
                bool(resp_content),
            )
            logger.debug(
                "[SECURITY][INTERCEPT] post-check full response: %s",
                json.dumps(post, ensure_ascii=False)[:500],
            )
        else:
            logger.info(
                "[SECURITY][INTERCEPT] post-check returned None (allow/fail-open)",
            )
    elif is_streaming:
        logger.info("[SECURITY][INTERCEPT] << Streaming response — post-check skipped")

    # ---- HINT OR LOGIC: block > hint > allow ----
    # Priority: resp block (handled above) > resp hint > req hint > allow
    effective_action = (
        "hint" if resp_action == "hint"
        else "hint" if req_action == "hint"
        else "allow"
    )

    if effective_action == "hint":
        if resp_action == "hint":
            hint_content = resp_content or "\n\n[Security Hint]"
            log_action = "RESPONSE_HINT"
        else:
            hint_content = req_content or "\n\n[Security Hint]"
            log_action = "REQUEST_HINT_APPLIED"

        logger.warning(
            "[SECURITY][INTERCEPT] !! %s: %s %s (streaming=%s)",
            log_action,
            summary.get("method", ""),
            summary.get("url", ""),
            is_streaming,
        )

        try:
            if is_streaming:
                response.read()  # buffer full streaming body
            resp_body = response.text
            hint_resp = _make_hint_response(
                request, response, resp_body, hint_content,
                summary.get("model", "security-policy"),
                is_streaming=is_streaming,
            )
            logger.info(
                "[SECURITY][INTERCEPT] << Returning hint %s response (%d bytes)",
                "SSE" if is_streaming else "JSON",
                len(hint_resp.content),
            )
            return hint_resp
        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] hint response build failed (pass-through): %s", exc)

    return response


# ------------------------------------------------------------------
# Patched send (async)
# ------------------------------------------------------------------

async def _patched_async_send(self: Any, request: Any, *args: Any, **kwargs: Any) -> Any:
    """Drop-in replacement for httpx.AsyncClient.send."""
    global _original_async_send, _interceptor

    if not _is_llm_api_call(request):
        return await _original_async_send(self, request, *args, **kwargs)
    summary = _extract_request_summary(request)
    logger.info(
        "[SECURITY][INTERCEPT] >> (async) Detected LLM API call: %s %s "
        "(model=%s stream=%s msgs=%s)",
        summary.get("method", "?"),
        summary.get("url", "?"),
        summary.get("model", "?"),
        summary.get("stream", "?"),
        summary.get("message_count", "?"),
    )

    # ---- PRE-REQUEST CHECK (sync — APS calls are blocking) ----
    req_action = "allow"
    req_content = None

    if _interceptor is not None:
        try:
            pre = _interceptor.check_request(request)
        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] (async) pre-check exception (fail-open): %s", exc)
            pre = None

        if pre and pre.get("action") == "block":
            reason = pre.get("content", "Request blocked by security policy")
            is_streaming = summary.get("stream", False)
            logger.warning(
                "[SECURITY][INTERCEPT] !! (async) REQUEST BLOCKED: %s %s reason=%s (streaming=%s)",
                summary.get("method", ""),
                summary.get("url", ""),
                reason,
                is_streaming,
            )
            return _make_blocked_response(
                request, reason,
                summary.get("model", "security-policy"),
                is_streaming=is_streaming,
            )

        if pre:
            req_action = pre.get("action", "allow")
            req_content = pre.get("content")
            logger.info("[SECURITY][INTERCEPT] (async) pre-check: action=%s has_content=%s", pre.get("action", "?"),
                        bool(req_content))
            logger.debug(
                "[SECURITY][INTERCEPT] (async) pre-check full response: %s",
                json.dumps(pre, ensure_ascii=False)[:500],
            )
        else:
            logger.info(
                "[SECURITY][INTERCEPT] (async) pre-check returned None (allow/fail-open)",
            )

    # ---- FORWARD ----
    logger.info("[SECURITY][INTERCEPT] >> (async) Forwarding to origin: %s", summary.get("url", ""))

    try:
        response = await _original_async_send(self, request, *args, **kwargs)
    except Exception as exc:
        logger.info("[SECURITY][INTERCEPT] << (async) Request failed: %s", exc)
        raise

    logger.info(
        "[SECURITY][INTERCEPT] << (async) Response: HTTP %d  content-type=%s",
        response.status_code,
        response.headers.get("content-type", "?"),
    )

    # ---- POST-RESPONSE CHECK ----
    is_streaming = "text/event-stream" in response.headers.get("content-type", "")

    resp_action = "allow"
    resp_content = None

    if _interceptor is not None and not is_streaming:
        try:
            post = _interceptor.check_response(request, response)
        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] (async) post-check exception (fail-open): %s", exc)
            post = None

        if post and post.get("action") == "block":
            reason = post.get("content", "Response blocked by security policy")
            logger.warning("[SECURITY][INTERCEPT] !! (async) RESPONSE BLOCKED: reason=%s", reason)
            return _make_blocked_response(
                request, reason,
                summary.get("model", "security-policy"),
                is_streaming=False,
            )

        if post:
            resp_action = post.get("action", "allow")
            resp_content = post.get("content")
            logger.info("[SECURITY][INTERCEPT] (async) post-check: action=%s has_content=%s", post.get("action", "?"),
                        bool(resp_content))
            logger.debug(
                "[SECURITY][INTERCEPT] (async) post-check full response: %s",
                json.dumps(post, ensure_ascii=False)[:500],
            )
        else:
            logger.info(
                "[SECURITY][INTERCEPT] (async) post-check returned None (allow/fail-open)",
            )
    elif is_streaming:
        logger.info("[SECURITY][INTERCEPT] << (async) Streaming — post-check skipped")

    # ---- HINT OR LOGIC: block > hint > allow ----
    effective_action = (
        "hint" if resp_action == "hint"
        else "hint" if req_action == "hint"
        else "allow"
    )

    if effective_action == "hint":
        if resp_action == "hint":
            hint_content = resp_content or "\n\n[Security Hint]"
            log_action = "RESPONSE_HINT"
        else:
            hint_content = req_content or "\n\n[Security Hint]"
            log_action = "REQUEST_HINT_APPLIED"

        logger.warning(
            "[SECURITY][INTERCEPT] !! (async) %s: %s %s (streaming=%s)",
            log_action,
            summary.get("method", ""),
            summary.get("url", ""),
            is_streaming,
        )

        try:
            if is_streaming:
                await response.aread()  # buffer full streaming body
            resp_body = response.text
            hint_resp = _make_hint_response(
                request, response, resp_body, hint_content,
                summary.get("model", "security-policy"),
                is_streaming=is_streaming,
            )
            logger.info(
                "[SECURITY][INTERCEPT] << (async) Returning hint %s response (%d bytes)",
                "SSE" if is_streaming else "JSON",
                len(hint_resp.content),
            )
            return hint_resp
        except Exception as exc:
            logger.warning("[SECURITY][INTERCEPT] (async) hint response build failed (pass-through): %s", exc)

    return response