feat(auth): configurable runtime-token header (server + SDK) to avoid gateway Authorization collision [HYBIM-866]

sdks/python/src/agent_control/__init__.py

@@ -113,6 +113,7 @@ async def handle_input(user_message: str) -> str:
     write_events,
 )
 from .otel_sink import control_event_to_otel_span
+from .runtime_auth import validate_http_field_name
 from .tracing import (
     get_current_span_id,
     get_current_trace_id,
@@ -458,6 +459,7 @@ def init(
     policy_refresh_interval_seconds: int = 60,
     target_type: str | None = None,
     target_id: str | None = None,
+    runtime_token_header: str | None = None,
     **kwargs: object
 ) -> Agent:
     """
@@ -506,6 +508,12 @@ def init(
             returned set when both are present.
         target_id: Optional opaque target identifier. Required iff target_type
             is also supplied.
+        runtime_token_header: Optional HTTP header the runtime token is sent on
+            for evaluation requests (defaults to AGENT_CONTROL_RUNTIME_TOKEN_HEADER
+            env var or Authorization). Point this at a dedicated header (e.g.
+            X-Agent-Control-Runtime-Token) when the server runs behind a gateway
+            that reserves Authorization for its own identity JWT. The server must
+            be configured to read the same header.
         **kwargs: Additional metadata to store with the agent
 
     Returns:
@@ -555,6 +563,16 @@ async def handle(message: str):
         raise ValueError(
             "target_type and target_id must be supplied together."
         )
+    # Validate the runtime-token header up front, before stopping the refresh
+    # loop or mutating shared session state. An explicit blank or a
+    # syntactically-invalid header must fail here rather than on the first
+    # later evaluation. (A None/unset value is fine; AgentControlClient resolves
+    # the env-var fallback and default when the eval client is built.)
+    if runtime_token_header is not None:
+        if not runtime_token_header.strip():
+            raise ValueError("runtime_token_header must not be blank.")
+        validate_http_field_name(runtime_token_header.strip())
+
     resolved_api_key_header = (
         api_key_header
         or os.getenv(AgentControlClient.API_KEY_HEADER_ENV_VAR)
@@ -588,6 +606,9 @@ async def handle(message: str):
         state.server_url = server_url or os.getenv('AGENT_CONTROL_URL') or 'http://localhost:8000'
         state.api_key = api_key
         state.api_key_header = resolved_api_key_header
+        # Stored raw (may be None); AgentControlClient resolves the env-var
+        # fallback and blank-handling so the rule stays in one place.
+        state.runtime_token_header = runtime_token_header
         state.runtime_token_cache.clear()
         state.target_type = target_type
         state.target_id = target_id
@@ -746,6 +767,7 @@ def _reset_state() -> None:
         state.server_url = None
         state.api_key = None
         state.api_key_header = None
+        state.runtime_token_header = None
         state.runtime_token_cache.clear()
         state.target_type = None
         state.target_id = None

sdks/python/src/agent_control/_state.py

@@ -27,6 +27,7 @@ def __init__(self) -> None:
         self.server_url: str | None = None
         self.api_key: str | None = None
         self.api_key_header: str | None = None
+        self.runtime_token_header: str | None = None
         self.runtime_token_cache = RuntimeTokenCache()
         # Optional target context fixed at init() time; both fields are set
         # together or both remain None.

sdks/python/src/agent_control/client.py

@@ -15,11 +15,14 @@
     RuntimeTokenCache,
     normalize_runtime_auth_mode,
     parse_runtime_token_exchange_response,
+    resolve_runtime_token_header,
 )
 
 _logger = logging.getLogger(__name__)
 
 _RUNTIME_AUTH_MODE_ENV_VAR = "AGENT_CONTROL_RUNTIME_AUTH_MODE"
+_RUNTIME_TOKEN_HEADER_ENV_VAR = "AGENT_CONTROL_RUNTIME_TOKEN_HEADER"
+_DEFAULT_RUNTIME_TOKEN_HEADER = "Authorization"
 _DEFAULT_RUNTIME_TOKEN_REFRESH_MARGIN_SECONDS = 30
 _AUTO_RUNTIME_TOKEN_FALLBACK_STATUSES = {404, 500, 502, 503, 504}
 _GLOBAL_RUNTIME_TOKEN_FALLBACK_STATUSES = {404}
@@ -35,19 +38,34 @@ def _runtime_cache_identity(api_key: str | None, api_key_header: str) -> str:
 
 
 class _AgentControlAuth(httpx.Auth):
-    """Attach local API-key credentials unless a request already has Bearer auth."""
+    """Attach local API-key credentials unless the request already carries them.
+
+    The API key is suppressed only when the request already presents a bearer
+    credential on ``Authorization`` or already carries the API key on its own
+    header. When the runtime token rides a dedicated header, the API key is
+    left in place: it may be the outer credential the request needs to
+    authenticate at a gateway before Agent Control verifies the runtime token,
+    and it rides a different header so there is no collision.
+    """
 
-    def __init__(self, api_key: str | None, header_name: str = "X-API-Key") -> None:
+    def __init__(
+        self,
+        api_key: str | None,
+        header_name: str = "X-API-Key",
+    ) -> None:
         self._api_key = api_key
         self._header_name = header_name
 
     def auth_flow(
         self,
         request: httpx.Request,
     ) -> Generator[httpx.Request, httpx.Response, None]:
-        if self._api_key and "Authorization" not in request.headers:
-            if self._header_name not in request.headers:
-                request.headers[self._header_name] = self._api_key
+        if (
+            self._api_key
+            and "Authorization" not in request.headers
+            and self._header_name not in request.headers
+        ):
+            request.headers[self._header_name] = self._api_key
         yield request
 
 
@@ -102,6 +120,7 @@ def __init__(
         runtime_token_cache: RuntimeTokenCache | None = None,
         runtime_token_refresh_margin_seconds: int = (_DEFAULT_RUNTIME_TOKEN_REFRESH_MARGIN_SECONDS),
         transport: httpx.AsyncBaseTransport | None = None,
+        runtime_token_header: str | None = None,
     ):
         """
         Initialize the client.
@@ -125,6 +144,13 @@ def __init__(
             runtime_token_refresh_margin_seconds: Refresh cached runtime tokens
                 before this many seconds of validity remain.
             transport: Optional httpx transport, primarily for tests.
+            runtime_token_header: HTTP header the runtime token is sent on.
+                Defaults to ``Authorization``; the
+                AGENT_CONTROL_RUNTIME_TOKEN_HEADER environment variable
+                overrides the default. Point this at a dedicated header (e.g.
+                ``X-Agent-Control-Runtime-Token``) when the server runs behind
+                a gateway that reserves ``Authorization`` for its own identity
+                JWT. The server must be configured to read the same header.
         """
         resolved_base_url = base_url or os.environ.get(
             self.BASE_URL_ENV_VAR, "http://localhost:8000"
@@ -140,6 +166,19 @@ def __init__(
         self._runtime_cache_identity = _runtime_cache_identity(self._api_key, self._api_key_header)
         configured_runtime_mode = runtime_auth_mode or os.environ.get(_RUNTIME_AUTH_MODE_ENV_VAR)
         self._runtime_auth_mode = normalize_runtime_auth_mode(configured_runtime_mode)
+        # Explicit blank param is a hard error; a blank env var falls back to
+        # the default (mirrors the server's _resolve_runtime_token_header).
+        self._runtime_token_header = resolve_runtime_token_header(
+            runtime_token_header,
+            os.environ.get(_RUNTIME_TOKEN_HEADER_ENV_VAR),
+            default=_DEFAULT_RUNTIME_TOKEN_HEADER,
+        )
+        # Bearer only on Authorization; a dedicated header carries the raw token
+        # so it can't collide with the gateway's Authorization JWT. Must stay in
+        # sync with LocalJwtVerifyProvider._require_bearer on the server.
+        self._runtime_token_use_bearer = (
+            self._runtime_token_header.lower() == _DEFAULT_RUNTIME_TOKEN_HEADER.lower()
+        )
         if runtime_token_refresh_margin_seconds < 0:
             raise ValueError("runtime_token_refresh_margin_seconds must be >= 0.")
         self._runtime_token_refresh_margin_seconds = runtime_token_refresh_margin_seconds
@@ -198,7 +237,10 @@ async def __aenter__(self) -> "AgentControlClient":
             base_url=self.base_url,
             timeout=self.timeout,
             headers=self._get_headers(),
-            auth=_AgentControlAuth(self._api_key, self._api_key_header),
+            auth=_AgentControlAuth(
+                self._api_key,
+                self._api_key_header,
+            ),
             transport=self._transport,
             event_hooks={"response": [self._check_server_version]},
         )
@@ -259,7 +301,14 @@ async def post_runtime_evaluation(
             headers=request_headers,
         )
 
-        if _should_refresh_runtime_token(response) and runtime_authorization is not None:
+        runtime_token_on_dedicated_header = not self._runtime_token_use_bearer
+        if (
+            _should_refresh_runtime_token(
+                response,
+                runtime_token_on_dedicated_header=runtime_token_on_dedicated_header,
+            )
+            and runtime_authorization is not None
+        ):
             await response.aread()
             if target_type is not None and target_id is not None:
                 self._runtime_token_cache.remove(
@@ -281,7 +330,10 @@ async def post_runtime_evaluation(
                 headers=request_headers,
             )
             if (
-                _should_refresh_runtime_token(response)
+                _should_refresh_runtime_token(
+                    response,
+                    runtime_token_on_dedicated_header=runtime_token_on_dedicated_header,
+                )
                 and target_type is not None
                 and target_id is not None
             ):
@@ -295,18 +347,22 @@ async def post_runtime_evaluation(
 
         return response
 
+    def _format_runtime_token(self, token: str) -> str:
+        """Sole place the Bearer prefix is applied (see __init__ for the rule)."""
+        return f"Bearer {token}" if self._runtime_token_use_bearer else token
+
     def _merge_runtime_headers(
         self,
         headers: dict[str, str] | None,
         runtime_authorization: str | None,
     ) -> dict[str, str] | None:
-        """Merge caller headers with an optional Bearer token."""
+        """Merge caller headers with an optional runtime token header."""
         if headers is None and runtime_authorization is None:
             return None
 
         merged = dict(headers or {})
         if runtime_authorization is not None:
-            merged["Authorization"] = runtime_authorization
+            merged[self._runtime_token_header] = runtime_authorization
         return merged
 
     async def _runtime_authorization(
@@ -350,7 +406,7 @@ async def _runtime_authorization(
                 refresh_margin_seconds=self._runtime_token_refresh_margin_seconds,
             )
             if cached is not None:
-                return f"Bearer {cached.token}"
+                return self._format_runtime_token(cached.token)
 
         exchange_lock = self._runtime_token_cache.exchange_lock(
             self.base_url,
@@ -379,7 +435,7 @@ async def _runtime_authorization(
                     refresh_margin_seconds=self._runtime_token_refresh_margin_seconds,
                 )
                 if cached is not None:
-                    return f"Bearer {cached.token}"
+                    return self._format_runtime_token(cached.token)
 
             token = await self._exchange_runtime_token(
                 target_type=target_type,
@@ -388,7 +444,7 @@ async def _runtime_authorization(
             )
         if token is None:
             return None
-        return f"Bearer {token}"
+        return self._format_runtime_token(token)
 
     async def _exchange_runtime_token(
         self,
@@ -457,10 +513,39 @@ async def _exchange_runtime_token(
         return token.token
 
 
-def _should_refresh_runtime_token(response: httpx.Response) -> bool:
+def _authenticate_flags_runtime_token(response: httpx.Response) -> bool:
+    """True when the response's WWW-Authenticate marks the token invalid.
+
+    Looks for the RFC 6750 ``error="invalid_token"`` challenge the runtime
+    verifier emits, so a refresh is driven by the server naming the token as
+    the problem, not by a bare status code.
+    """
+    return "invalid_token" in response.headers.get("WWW-Authenticate", "").lower()
+
+
+def _should_refresh_runtime_token(
+    response: httpx.Response,
+    *,
+    runtime_token_on_dedicated_header: bool = False,
+) -> bool:
+    """Decide whether a runtime-token refresh is warranted for this response.
+
+    When the runtime token rides ``Authorization`` (the default), a 401 is
+    unambiguous: the only credential on that header is the runtime token, so
+    treat it as expired and refresh.
+
+    When the token rides a dedicated header (behind a gateway that owns
+    ``Authorization`` for its own identity JWT), a bare 401 is ambiguous: it
+    may be the gateway rejecting its own credential, not our runtime token.
+    Evicting a still-valid token there just forces another exchange that hits
+    the same gateway 401 and hides the original error. So in that case only
+    refresh when the server explicitly flags the runtime token as invalid via
+    WWW-Authenticate (or a 403 invalid_token challenge).
+    """
     if response.status_code == 401:
+        if runtime_token_on_dedicated_header:
+            return _authenticate_flags_runtime_token(response)
         return True
     if response.status_code != 403:
         return False
-    authenticate = response.headers.get("WWW-Authenticate", "")
-    return "invalid_token" in authenticate.lower()
+    return _authenticate_flags_runtime_token(response)

sdks/python/src/agent_control/control_decorators.py

@@ -280,6 +280,7 @@ async def _evaluate(
         base_url=server_url,
         api_key=state.api_key,
         api_key_header=state.api_key_header,
+        runtime_token_header=state.runtime_token_header,
         runtime_token_cache=state.runtime_token_cache,
     ) as client:
         # If we have controls, use local evaluation which handles both SDK and server controls

sdks/python/src/agent_control/evaluation.py

@@ -560,6 +560,7 @@ async def evaluate_controls(
         base_url=state.server_url,
         api_key=state.api_key,
         api_key_header=state.api_key_header,
+        runtime_token_header=state.runtime_token_header,
         runtime_token_cache=state.runtime_token_cache,
     ) as client:
         return await check_evaluation_with_local(

sdks/python/src/agent_control/runtime_auth.py

@@ -16,6 +16,52 @@
 _DEFAULT_MAX_CACHE_ENTRIES = 256
 _DEFAULT_JWT_UNAVAILABLE_TTL_SECONDS = 30
 
+DEFAULT_RUNTIME_TOKEN_HEADER = "Authorization"
+
+# RFC 7230 "token" characters, the grammar an HTTP header field name must
+# follow. Anything outside this set (spaces, colons, control chars, non-ASCII)
+# produces a header a compliant client or gateway cannot send, so reject it up
+# front rather than failing at request time.
+_HTTP_TOKEN_CHARS = frozenset(
+    "!#$%&'*+-.^_`|~0123456789"
+    "abcdefghijklmnopqrstuvwxyz"
+    "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+)
+
+
+def validate_http_field_name(name: str, *, field: str = "runtime_token_header") -> str:
+    """Validate ``name`` against the RFC 7230 header field-name grammar.
+
+    Returns the name unchanged when valid; raises ``ValueError`` otherwise.
+    Keeps the SDK and server rejecting the same set of invalid headers.
+    """
+    if not name or any(ch not in _HTTP_TOKEN_CHARS for ch in name):
+        raise ValueError(
+            f"{field} must be a valid HTTP header field name "
+            "(RFC 7230 token: letters, digits, and !#$%&'*+-.^_`|~ only, no spaces)."
+        )
+    return name
+
+
+def resolve_runtime_token_header(
+    value: str | None,
+    env_value: str | None,
+    *,
+    default: str = DEFAULT_RUNTIME_TOKEN_HEADER,
+) -> str:
+    """Resolve and validate the runtime-token header (param > env > default).
+
+    An explicitly-passed blank ``value`` is a hard error; a blank env value
+    falls back to the default. The resolved header is validated against the
+    HTTP field-name grammar so misconfiguration is caught before it is used.
+    """
+    if value is not None and not value.strip():
+        raise ValueError("runtime_token_header must not be blank.")
+    if env_value is not None and not env_value.strip():
+        env_value = None
+    resolved = (value or env_value or default).strip()
+    return validate_http_field_name(resolved)
+
 
 @dataclass(frozen=True)
 class RuntimeToken: