Add request trace classification and pseudonymous ids

deploy/.env.example

@@ -128,3 +128,10 @@ FILE_EXPIRATION_CHECK_INTERVAL_HOURS=0
 # - /app/configs/config_web_default_llamaindex.yml
 # - /app/configs/config_web_frag.yml
 # BACKEND_CONFIG=
+#
+# Optional trace enrichment for aiq_api auth middleware:
+# AIQ_TRACE_USER_IDENTITY_MODE=none
+# AIQ_TRACE_USER_IDENTITY_HMAC_SECRET=
+# AIQ_TRACE_CLIENT_ID_MODE=none
+# AIQ_TRACE_CLIENT_ID_HMAC_SECRET=
+# AIQ_TRACE_CLIENT_IP_HEADERS=x-real-ip,x-forwarded-for

docs/source/customization/configuration-reference.md

@@ -78,6 +78,16 @@ general:
 | `front_end.expiry_seconds` | `int` | `86400` | How long completed jobs remain in the database (seconds). |
 | `front_end.cors` | `object` | -- | CORS settings for the API server. |
 
+For `aiq_api`, request-trace enrichment is configured via environment variables
+rather than YAML fields. See `frontends/aiq_api/README.md` and the
+[Observability](../deployment/observability.md) guide for:
+
+- `AIQ_TRACE_USER_IDENTITY_MODE`
+- `AIQ_TRACE_USER_IDENTITY_HMAC_SECRET`
+- `AIQ_TRACE_CLIENT_ID_MODE`
+- `AIQ_TRACE_CLIENT_ID_HMAC_SECRET`
+- `AIQ_TRACE_CLIENT_IP_HEADERS`
+
 ---
 
 ## `llms` Section

docs/source/deployment/observability.md

@@ -206,6 +206,52 @@ general:
 | `redaction_headers` | Request headers checked to determine whether to redact. |
 | `resource_attributes` | Custom OTEL resource attributes attached to all spans. |
 
+### Request Classification Tags
+
+When the `aiq_api` auth middleware is enabled, request spans can include a set
+of low-risk request tags plus optional pseudonymous identity tags.
+
+Always-on request tags:
+
+- `aiq.caller.type` -- resolved caller type from auth middleware
+- `aiq.auth.transport` -- `bearer`, `cookie`, or `none`
+- `aiq.auth.verified` -- whether the request resolved to a verified principal
+- `aiq.access.channel` -- inferred request channel or the explicit `X-AIQ-Access-Channel` header
+
+Optional pseudonymous tags:
+
+- `enduser.id`, `aiq.user.id`, `aiq.auth.type` -- controlled by `AIQ_TRACE_USER_IDENTITY_MODE`
+- `aiq.user.email`, `aiq.user.name` -- added only in `full` mode
+- `aiq.client.id` -- controlled by `AIQ_TRACE_CLIENT_ID_MODE=ip`
+
+Environment variables:
+
+- `AIQ_TRACE_USER_IDENTITY_MODE=none|id|full`
+- `AIQ_TRACE_USER_IDENTITY_HMAC_SECRET=<secret>`
+- `AIQ_TRACE_CLIENT_ID_MODE=none|ip`
+- `AIQ_TRACE_CLIENT_ID_HMAC_SECRET=<secret>`
+- `AIQ_TRACE_CLIENT_IP_HEADERS=x-real-ip,x-forwarded-for`
+
+The `id` and `ip` modes emit HMAC-derived pseudonymous identifiers rather than
+raw subjects or raw IP addresses.
+
+### Access Channel Header
+
+Deployment-specific wrappers can set the optional request header
+`X-AIQ-Access-Channel` to separate traffic sources without modifying public
+middleware. Supported values are:
+
+- `ui`
+- `skill`
+- `api`
+- `headless`
+- `anonymous`
+- `internal`
+- `unknown`
+
+When absent, the middleware falls back to generic inference from auth transport
+and request shape.
+
 ### Batch Configuration
 
 The exporter supports standard OTEL batch settings:

docs/source/deployment/production.md

@@ -120,6 +120,19 @@ Set `LOG_LEVEL=DEBUG` for verbose output during troubleshooting. Use `LOG_LEVEL=
 
 The backend supports OpenTelemetry-compatible tracing. See [Observability](./observability.md) for setup guides covering Phoenix, LangSmith, Weave, and the OTEL Collector with privacy redaction.
 
+If you are deploying the `aiq_api` front-end and want request correlation in
+traces, set the relevant environment variables at deploy time rather than
+hardcoding them in code:
+
+- `AIQ_TRACE_USER_IDENTITY_MODE`
+- `AIQ_TRACE_USER_IDENTITY_HMAC_SECRET`
+- `AIQ_TRACE_CLIENT_ID_MODE`
+- `AIQ_TRACE_CLIENT_ID_HMAC_SECRET`
+- `AIQ_TRACE_CLIENT_IP_HEADERS`
+
+Deployment-specific wrappers may also set `X-AIQ-Access-Channel` to distinguish
+UI, skill, and API traffic at the request level.
+
 ### Metrics to Watch
 
 | Metric | Source | What to look for |

frontends/aiq_api/README.md

@@ -198,6 +198,11 @@ is disabled by default and opt-in via the `REQUIRE_AUTH` environment variable.
 | Variable | Purpose | Default |
 |----------|---------|---------|
 | `REQUIRE_AUTH` | Enforce authentication on all non-exempt routes | `false` |
+| `AIQ_TRACE_USER_IDENTITY_MODE` | Control whether verified user identity is attached to request traces: `none`, `id`, or `full` | `none` |
+| `AIQ_TRACE_USER_IDENTITY_HMAC_SECRET` | Secret used to pseudonymize verified user IDs in traces when identity tagging is enabled | unset |
+| `AIQ_TRACE_CLIENT_ID_MODE` | Control whether a pseudonymous client identifier is attached to request traces: `none` or `ip` | `none` |
+| `AIQ_TRACE_CLIENT_ID_HMAC_SECRET` | Secret used to pseudonymize client IDs when client tagging is enabled; falls back to the user-identity secret | unset |
+| `AIQ_TRACE_CLIENT_IP_HEADERS` | Ordered comma-separated client IP headers to trust before falling back to the ASGI client address | `x-real-ip,x-forwarded-for` |
 
 ### Enabling authentication
 
@@ -207,6 +212,59 @@ no validators are registered.
 
 Two registration mechanisms are supported — pick whichever fits your deployment:
 
+### Trace identity tagging
+
+Auth middleware can enrich active Datadog / OpenTelemetry request spans with
+low-risk request tags and optional pseudonymous identity tags.
+
+Always-on request tags:
+
+- `aiq.caller.type`: resolved caller type from the auth middleware
+- `aiq.auth.transport`: `bearer`, `cookie`, or `none`
+- `aiq.auth.verified`: whether the request resolved to a verified principal
+- `aiq.access.channel`: inferred or explicitly supplied access channel
+
+Optional user identity tags are controlled by `AIQ_TRACE_USER_IDENTITY_MODE`:
+
+- `none`: disable user identity tagging entirely
+- `id`: tag only pseudonymous stable identifiers (`enduser.id`, `aiq.user.id`, `aiq.auth.type`)
+- `full`: tag pseudonymous identifiers plus `aiq.user.email` and `aiq.user.name` when present
+
+Only verified principals are tagged. Anonymous, internal, and unverified JWT
+fallback callers are never attached to traces.
+Set `AIQ_TRACE_USER_IDENTITY_HMAC_SECRET` when using `id` or `full`; otherwise
+identity tagging is skipped.
+
+Optional client correlation is controlled by `AIQ_TRACE_CLIENT_ID_MODE`:
+
+- `none`: disable client correlation
+- `ip`: add `aiq.client.id` as an HMAC-derived pseudonymous identifier from the
+  client IP
+
+Set `AIQ_TRACE_CLIENT_ID_HMAC_SECRET` when using `ip`. If unset, the middleware
+falls back to `AIQ_TRACE_USER_IDENTITY_HMAC_SECRET`. `AIQ_TRACE_CLIENT_IP_HEADERS`
+controls which ingress headers are consulted first.
+
+### Access Channel Header
+
+Upstream middleware accepts an optional low-cardinality request header:
+
+- `X-AIQ-Access-Channel: <channel>`
+
+Supported values are:
+
+- `ui`
+- `skill`
+- `api`
+- `headless`
+- `anonymous`
+- `internal`
+- `unknown`
+
+When absent, the middleware falls back to generic inference from auth transport
+and request shape. Deployment-specific wrappers can set this header explicitly
+to separate UI, skill, and API traffic without modifying public middleware.
+
 #### 1. Programmatic (`register_validator`)
 
 Call `register_validator()` in any code that runs before `nat serve`:

frontends/aiq_api/src/aiq_api/auth/middleware.py

@@ -55,6 +55,36 @@
 ``AIQ_JWT_AUDIENCE``
     Optional ``aud`` claim to verify.  Leave unset to skip audience
     verification.
+
+``AIQ_TRACE_USER_IDENTITY_MODE``
+    Controls whether verified user identity is attached to active trace spans.
+    Supported values:
+    - ``none``: do not attach any user identity tags
+    - ``id``: attach pseudonymous stable identifiers only (``enduser.id``, ``aiq.user.id``,
+      ``aiq.auth.type``)
+    - ``full``: attach identifiers plus ``aiq.user.email`` and
+      ``aiq.user.name`` when present
+    Defaults to ``none``.
+
+``AIQ_TRACE_USER_IDENTITY_HMAC_SECRET``
+    Secret used to derive pseudonymous trace user IDs from verified subjects
+    when ``AIQ_TRACE_USER_IDENTITY_MODE`` is ``id`` or ``full``. If unset,
+    user identity tagging is disabled even when a mode is configured.
+
+``AIQ_TRACE_CLIENT_ID_MODE``
+    Controls whether a pseudonymous client identifier is attached to trace spans.
+    Supported values:
+    - ``none``: do not attach a client identifier
+    - ``ip``: derive a pseudonymous client ID from the request IP
+    Defaults to ``none``.
+
+``AIQ_TRACE_CLIENT_ID_HMAC_SECRET``
+    Secret used to derive pseudonymous client IDs. Falls back to
+    ``AIQ_TRACE_USER_IDENTITY_HMAC_SECRET`` when unset.
+
+``AIQ_TRACE_CLIENT_IP_HEADERS``
+    Comma-separated header names to check for the client IP before falling back
+    to the ASGI client address. Defaults to ``x-real-ip,x-forwarded-for``.
 """
 
 import json
@@ -69,8 +99,21 @@
 from starlette.types import Scope
 from starlette.types import Send
 
+from . import utils as auth_utils
+from .utils import _load_trace_client_id_mode
+from .utils import _load_trace_client_id_secret
+from .utils import _load_trace_client_ip_headers
+from .utils import _load_trace_user_identity_mode
+from .utils import _load_trace_user_identity_secret
+from .utils import attach_request_to_active_trace
+from .utils import is_headless_request
+
 logger = logging.getLogger(__name__)
 
+# Backwards-compatible aliases for tests and internal helper imports.
+_build_pseudonymous_trace_user_id = auth_utils._build_pseudonymous_trace_user_id
+_build_pseudonymous_trace_client_id = auth_utils._build_pseudonymous_trace_client_id
+
 # ---------------------------------------------------------------------------
 # ContextVar — carries the resolved caller identity through the call stack
 # ---------------------------------------------------------------------------
@@ -136,11 +179,6 @@ def is_external_request(headers: dict[bytes, bytes], external_hostnames: set[str
     return host in (external_hostnames or _load_external_hostnames())
 
 
-def is_headless_request(headers: dict[bytes, bytes]) -> bool:
-    """Return ``True`` for headless callers that should skip the clarifier."""
-    return headers.get(b"x-aiq-mode", b"").decode().lower() == "headless"
-
-
 def extract_auth_token(headers: dict[bytes, bytes]) -> str | None:
     """Extract a bearer token or idToken cookie from ASGI headers."""
     auth = headers.get(b"authorization", b"").decode()
@@ -245,6 +283,11 @@ def __init__(
         self.require_auth = require_auth
         self._validators: list = validators or []
         self._external_hostnames: set[str] = external_hostnames or _load_external_hostnames()
+        self._trace_user_identity_mode: str = _load_trace_user_identity_mode()
+        self._trace_user_identity_secret: str | None = _load_trace_user_identity_secret()
+        self._trace_client_id_mode: str = _load_trace_client_id_mode()
+        self._trace_client_id_secret: str | None = _load_trace_client_id_secret()
+        self._trace_client_ip_headers: list[str] = _load_trace_client_ip_headers()
 
         if require_auth and not self._validators:
             logger.warning(
@@ -257,6 +300,17 @@ def __init__(
             [type(v).__name__ for v in self._validators],
             self._external_hostnames,
         )
+        logger.info(
+            "AuthMiddleware trace user identity mode=%s secret_configured=%s",
+            self._trace_user_identity_mode,
+            bool(self._trace_user_identity_secret),
+        )
+        logger.info(
+            "AuthMiddleware trace client id mode=%s secret_configured=%s ip_headers=%s",
+            self._trace_client_id_mode,
+            bool(self._trace_client_id_secret),
+            self._trace_client_ip_headers,
+        )
 
     # ------------------------------------------------------------------
     # ASGI entry point
@@ -278,7 +332,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 
         if is_external and path in AUTH_EXEMPT_PATHS:
             user = {"type": "anonymous", "skip_clarifier": True}
-            await self._call_app(scope, receive, send, user)
+            await self._call_app(scope, receive, send, headers, user)
             return
 
         user, error_status, _ = await resolve_request_user(
@@ -292,7 +346,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             await self._send_json(send, error_status or 401, {"detail": detail})
             return
 
-        await self._call_app(scope, receive, send, user)
+        await self._call_app(scope, receive, send, headers, user)
 
     # ------------------------------------------------------------------
     # Internal helpers
@@ -303,11 +357,25 @@ async def _call_app(
         scope: Scope,
         receive: Receive,
         send: Send,
+        headers: dict[bytes, bytes],
         user: dict[str, Any],
     ) -> None:
         if "state" not in scope:
             scope["state"] = {}
         scope["state"]["user"] = user
+        is_external = self._is_external(headers)
+        trust_access_channel_override = (not is_external) or bool(user.get("sub"))
+        attach_request_to_active_trace(
+            headers,
+            scope,
+            user,
+            trust_access_channel_override=trust_access_channel_override,
+            user_identity_mode=self._trace_user_identity_mode,
+            user_identity_secret=self._trace_user_identity_secret,
+            client_id_mode=self._trace_client_id_mode,
+            client_id_secret=self._trace_client_id_secret,
+            client_ip_headers=self._trace_client_ip_headers,
+        )
 
         with user_context(user):
             await self.app(scope, receive, send)