DecisionTarget: optional client reuse and retries

README.md

@@ -965,7 +965,7 @@ r = results[0]
 r.judgment["choice"], r.judgment["confidence"], r.judgment["probabilities"]
 ```
 
-For Jev on OpenRouter, use `DecisionTarget.openrouter("typesafe/jev-1.13")`, with the key in `OPENROUTER_API_KEY`. Provider routing, for example `{"provider": {"zdr": True}}`, goes in `extra_body`.
+For Jev on OpenRouter, use `DecisionTarget.openrouter("typesafe/jev-1.13")`, with the key in `OPENROUTER_API_KEY`. Provider routing, for example `{"provider": {"zdr": True}}`, goes in `extra_body`. Each response's `raw` gains an `endpoint` block (scheme, host, port, and whether the standard proxy settings route it through a proxy), and a plain `http://` URL to a host other than loopback raises a warning.
 
 `DecisionTarget`:
 - checks the endpoint's limits before sending: 2–26 options per question, and 64 KiB per request for Ollama. A scenario over a limit is recorded as an error; its documents are never shortened.
@@ -999,6 +999,8 @@ Use `max_turns=1` with `choice_match`. Follow-up turns need an `auditor_model` o
 local servers can use the same target. No timestamps are required. Provider/model
 support for languages, formats and optional parameters varies; unsupported
 requests are reported as errors rather than silently changing the input.
+The file is uploaded as `audio` plus its lowercased extension, so the local
+filename is not sent with the upload.
 
 ```python
 from simpleaudit import Auditor, ASRTarget

pyproject.toml

@@ -91,6 +91,8 @@ target-version = ["py311", "py312", "py313"]
 testpaths = ["tests"]
 addopts = "--cov=simpleaudit --cov-report=html --cov-report=term-missing"
 asyncio_mode = "auto"
+# Tests use plain-http stub hosts; the targets' cleartext warning is expected there.
+filterwarnings = ['ignore:target endpoint http\S+ uses plain http:UserWarning']
 
 [tool.coverage.run]
 source = ["simpleaudit"]

simpleaudit/targets/_http.py

@@ -0,0 +1,77 @@
+"""HTTP plumbing shared by the targets: client creation and endpoint provenance.
+
+``describe_endpoint`` records where a request went (scheme, host, port, and
+whether the standard proxy settings route it through a proxy) so an audit
+record can say which endpoint answered. It never includes the path, query,
+userinfo, headers or keys.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import urllib.request
+import warnings
+from typing import Any
+from urllib.parse import urlsplit
+
+_DEFAULT_PORTS = {"http": 80, "https": 443}
+
+
+def make_client(timeout: float) -> Any:
+    """A fresh ``httpx.AsyncClient(timeout=timeout)``, as the targets always created."""
+    import httpx
+
+    return httpx.AsyncClient(timeout=timeout)
+
+
+def _is_loopback(host: str) -> bool:
+    if host.lower() == "localhost":
+        return True
+    try:
+        return ipaddress.ip_address(host).is_loopback
+    except ValueError:
+        return False
+
+
+def describe_endpoint(url: str) -> dict[str, Any]:
+    """``{"scheme", "host", "port", "via_proxy"}`` for ``url``.
+
+    ``via_proxy`` follows ``urllib.request.getproxies()`` / ``proxy_bypass()``
+    (the environment's proxy variables); an injected client's own proxy
+    configuration is not visible here.
+    """
+    parts = urlsplit(url)
+    scheme = parts.scheme.lower()
+    host = parts.hostname or ""
+    proxies = urllib.request.getproxies()
+    proxied = bool(proxies.get(scheme) or proxies.get("all"))
+    return {
+        "scheme": scheme,
+        "host": host,
+        "port": parts.port or _DEFAULT_PORTS.get(scheme),
+        "via_proxy": proxied and not urllib.request.proxy_bypass(host),
+    }
+
+
+def warn_if_cleartext(url: str) -> None:
+    """Warn when ``url`` is plain http to a host other than loopback."""
+    parts = urlsplit(url)
+    if parts.scheme.lower() == "http" and not _is_loopback(parts.hostname or ""):
+        warnings.warn(
+            f"target endpoint http://{parts.hostname} uses plain http: requests "
+            "and responses to it are not encrypted",
+            UserWarning,
+            stacklevel=3,
+        )
+
+
+def with_endpoint(raw: Any, url: str) -> dict[str, Any]:
+    """``raw`` with an ``endpoint`` block added; provider fields are never overwritten.
+
+    A dict without an ``endpoint`` key gets one; anything else is wrapped as
+    ``{"response": raw, "endpoint": ...}``.
+    """
+    endpoint = describe_endpoint(url)
+    if isinstance(raw, dict) and "endpoint" not in raw:
+        return {**raw, "endpoint": endpoint}
+    return {"response": raw, "endpoint": endpoint}

simpleaudit/targets/asr.py

@@ -8,6 +8,7 @@
 from typing import Any
 from urllib.parse import unquote, urlparse
 
+from ._http import make_client, warn_if_cleartext, with_endpoint
 from .base import TargetContext, TargetResponse
 
 
@@ -35,6 +36,7 @@ def __init__(
     ) -> None:
         self.model = model
         self.url = base_url.rstrip("/") + "/audio/transcriptions"
+        warn_if_cleartext(self.url)
         self.api_key = api_key
         self.timeout = timeout
         self.options = dict(options or {})
@@ -63,8 +65,6 @@ async def send(
         documents: list[str | dict[str, Any]] | None = None,
         response_format: dict[str, Any] | None = None,
     ) -> TargetResponse:
-        import httpx
-
         if file_uri is None:
             file_uri = next(
                 (
@@ -102,9 +102,7 @@ async def send(
         headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
         if context:
             headers.update(context.trace_headers)
-        client = (
-            self._client if self._client is not None else httpx.AsyncClient(timeout=self.timeout)
-        )
+        client = self._client if self._client is not None else make_client(self.timeout)
         try:
             with path.open("rb") as audio:
                 response = await client.post(
@@ -113,7 +111,8 @@ async def send(
                     headers=headers,
                     files={
                         "file": (
-                            path.name,
+                            # Neutral name: the local filename stays local.
+                            "audio" + path.suffix.lower(),
                             audio,
                             mimetypes.guess_type(path.name)[0] or "application/octet-stream",
                         )
@@ -124,7 +123,7 @@ async def send(
             payload = response.json()
             if not isinstance(payload, dict) or not isinstance(payload.get("text"), str):
                 raise ValueError("Transcription response must contain a string 'text' field")
-            return TargetResponse(content=payload["text"], raw=payload)
+            return TargetResponse(content=payload["text"], raw=with_endpoint(payload, self.url))
         finally:
             if self._client is None:
                 await client.aclose()

simpleaudit/targets/decision.py

@@ -34,21 +34,48 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import os
 from collections.abc import Mapping
+from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
 from typing import Any
 
 from ..context_marks import parse_documents
+from ._http import make_client, warn_if_cleartext, with_endpoint
 from .base import TargetContext, TargetResponse, history_documents
 
 #: Ollama's request-body cap for System One requests without images.
 OLLAMA_MAX_BODY_BYTES = 64 * 1024
 
+#: Statuses worth another attempt: rate limiting and server-side failures.
+RETRY_STATUSES = frozenset({429, 500, 502, 503, 504})
+
+#: Statuses whose ``Retry-After`` header (RFC 9110 section 10.2.3) is honoured.
+RETRY_AFTER_STATUSES = frozenset({429, 503})
+
+#: Longest wait, in seconds, that a ``Retry-After`` header can ask for.
+_RETRY_AFTER_MAX = 60.0
+
 #: OpenRouter's System One endpoint (``POST /api/v1/systemone``).
 OPENROUTER_SYSTEMONE_URL = "https://openrouter.ai/api/v1/systemone"
 
 
+def _retry_after_seconds(resp: Any) -> float | None:
+    """Seconds a ``Retry-After`` header asks for (delay-seconds or HTTP-date), else None."""
+    value = (resp.headers.get("Retry-After") or "").strip()
+    if value.isdigit():
+        return float(value)
+    try:
+        when = parsedate_to_datetime(value)
+    except (TypeError, ValueError):
+        return None
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=timezone.utc)
+    return max(0.0, (when - datetime.now(timezone.utc)).total_seconds())
+
+
 class DecisionTarget:
     """Send a scenario's decision question to a System One endpoint."""
 
@@ -68,8 +95,16 @@ def __init__(
         min_options: int = 2,
         max_options: int | None = 26,
         max_body_bytes: int | None = None,
+        max_retries: int = 0,
+        retry_backoff: float = 0.5,
+        reuse_client: bool = False,
     ) -> None:
+        if max_retries < 0:
+            raise ValueError("max_retries must be 0 or more")
+        if retry_backoff < 0:
+            raise ValueError("retry_backoff must be 0 or more")
         self.url = url
+        warn_if_cleartext(url)
         self.model = model
         self.headers = dict(headers or {})
         if api_key:
@@ -80,6 +115,73 @@ def __init__(
         self.min_options = min_options
         self.max_options = max_options
         self.max_body_bytes = max_body_bytes
+        #: Extra attempts after a transport error or a status in ``RETRY_STATUSES``;
+        #: 0 (the default) sends each request once. Attempt ``n`` waits
+        #: ``retry_backoff * 2**n`` seconds before the next one, as in ``ModelTarget``.
+        #: A ``Retry-After`` header on a 429 or 503 replaces that wait, capped at
+        #: ``_RETRY_AFTER_MAX`` (60 s).
+        self.max_retries = max_retries
+        self.retry_backoff = retry_backoff
+        self._reuse_client = reuse_client
+        self._shared_client: Any | None = None
+        self._shared_loop: asyncio.AbstractEventLoop | None = None
+
+    def _acquire_client(self) -> tuple[Any, bool]:
+        """The client for one request and whether the request must close it.
+
+        An injected client is never closed here. With ``reuse_client`` one
+        client serves every request on the running event loop. Each sync
+        ``run()`` has its own loop, and a client cannot be used or closed on
+        another loop, so a client from an earlier loop is dropped and replaced.
+        """
+        if self._client is not None:
+            return self._client, False
+        if not self._reuse_client:
+            return make_client(self.timeout), True
+        loop = asyncio.get_running_loop()
+        if self._shared_client is None or self._shared_loop is not loop:
+            self._shared_client = make_client(self.timeout)
+            self._shared_loop = loop
+        return self._shared_client, False
+
+    async def aclose(self) -> None:
+        """Close the client kept by ``reuse_client``; an injected client is the caller's."""
+        client, loop = self._shared_client, self._shared_loop
+        self._shared_client = self._shared_loop = None
+        if client is not None and loop is asyncio.get_running_loop():
+            await client.aclose()
+
+    async def __aenter__(self) -> DecisionTarget:
+        return self
+
+    async def __aexit__(self, *exc: Any) -> None:
+        await self.aclose()
+
+    async def _post(self, payload: bytes, headers: dict[str, str]) -> Any:
+        """POST ``payload``, retrying transient failures up to ``max_retries`` times."""
+        import httpx
+
+        client, owns_client = self._acquire_client()
+        try:
+            for attempt in range(self.max_retries + 1):
+                last = attempt == self.max_retries
+                delay = self.retry_backoff * 2**attempt
+                try:
+                    resp = await client.post(self.url, content=payload, headers=headers)
+                except httpx.TransportError:
+                    if last:
+                        raise
+                else:
+                    if resp.status_code not in RETRY_STATUSES or last:
+                        return resp
+                    if resp.status_code in RETRY_AFTER_STATUSES:
+                        asked = _retry_after_seconds(resp)
+                        if asked is not None:
+                            delay = min(asked, _RETRY_AFTER_MAX)
+                await asyncio.sleep(delay)
+        finally:
+            if owns_client:
+                await client.aclose()
 
     @classmethod
     def ollama(
@@ -173,23 +275,12 @@ async def send(
         if context is not None:
             headers.update(context.trace_headers)
 
-        owns_client = self._client is None
-        if client := self._client:
-            pass
-        else:
-            import httpx
-
-            client = httpx.AsyncClient(timeout=self.timeout)
-        try:
-            resp = await client.post(self.url, content=payload, headers=headers)
-            if resp.status_code >= 400:
-                raise RuntimeError(
-                    f"decision endpoint returned HTTP {resp.status_code}: {resp.text[:300]}"
-                )
-            data = resp.json()
-        finally:
-            if owns_client:
-                await client.aclose()
+        resp = await self._post(payload, headers)
+        if resp.status_code >= 400:
+            raise RuntimeError(
+                f"decision endpoint returned HTTP {resp.status_code}: {resp.text[:300]}"
+            )
+        data = resp.json()
 
         answer = (data.get("answers") or {}).get(decision["id"]) if isinstance(data, dict) else None
         if not isinstance(answer, dict):
@@ -204,7 +295,7 @@ async def send(
         usage = data.get("usage") or {}
         return TargetResponse(
             content=f"{choice}: {description}" if description else choice,
-            raw=data,
+            raw=with_endpoint(data, self.url),
             input_tokens=usage.get("input_tokens"),
             output_tokens=usage.get("output_tokens"),
             decision=answer,

simpleaudit/targets/http.py

@@ -24,6 +24,7 @@
 from typing import Any, Callable, Dict, List, Optional, Union
 
 from ..context_marks import document_texts
+from ._http import make_client, warn_if_cleartext, with_endpoint
 from .base import (
     Target,
     TargetContext,
@@ -113,6 +114,7 @@ def __init__(
         system_field: Optional[str] = None,
     ) -> None:
         self.url = url
+        warn_if_cleartext(url)
         self.method = method.upper()
         self.headers = dict(headers or {})
         self.request_template = dict(request_template or {})
@@ -215,9 +217,7 @@ async def send(
         if client := self._client:
             pass
         else:
-            import httpx
-
-            client = httpx.AsyncClient(timeout=self.timeout)
+            client = make_client(self.timeout)
         try:
             resp = await client.request(self.method, self.url, json=body, headers=headers)
             resp.raise_for_status()
@@ -242,7 +242,7 @@ async def send(
 
         return TargetResponse(
             content=content,
-            raw=data,
+            raw=with_endpoint(data, self.url),
             input_tokens=int(in_tok) if in_tok is not None else None,
             output_tokens=int(out_tok) if out_tok is not None else None,
         )