feat(auth): add cache-through token caching for azure_ad and gcp_adc

apis/Cargo.toml

@@ -40,7 +40,7 @@ serde_yaml = { workspace = true }
 sqlx = { workspace = true, optional = true }
 thiserror = { workspace = true }
 tiktoken-rs = { workspace = true }
-tokio = { workspace = true, features = ["rt", "time", "net"] }
+tokio = { workspace = true, features = ["rt", "sync", "time", "net"] }
 tracing = { workspace = true }
 url = { workspace = true }
 utoipa = { workspace = true }

apis/src/lib.rs

@@ -18,6 +18,7 @@ pub mod promotion;
 #[cfg(feature = "store")]
 pub mod store;
 pub mod subrequest;
+pub mod token_cache;
 pub(crate) mod web_search;
 
 /// Whether a `Content-Type` header value indicates `text/event-stream`,

apis/src/token_cache.rs

@@ -0,0 +1,281 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright (c) 2026 Praxis Contributors
+
+//! [`TokenCache`] — a cache-through, on-demand-refreshed cache for a
+//! single short-lived credential (an upstream bearer token, an access
+//! token, ...).
+//!
+//! # Semantics
+//!
+//! There is no background refresh. Every call to
+//! [`TokenCache::get_or_refresh`] checks the cache; if the cached value
+//! is still valid it is returned immediately (a shared read lock, no
+//! network call). If it is missing or past its safety margin, the
+//! caller takes an exclusive lock, checks again in case a concurrent
+//! caller already refreshed it while this one was waiting for the lock,
+//! and only then calls `fetch` — at most once per group of callers that
+//! all observe a stale cache at the same time. A failed fetch is
+//! propagated to the caller and nothing is cached, so the next call
+//! tries again; there is no server-side retry/backoff loop.
+
+use std::time::{Duration, Instant};
+
+use tokio::sync::RwLock;
+use tracing::warn;
+
+// -----------------------------------------------------------------------------
+// Margin
+// -----------------------------------------------------------------------------
+
+/// Safety margin to subtract from a TTL before treating a cached value
+/// as expired, capped at half the TTL so a short-lived credential stays
+/// usable for part of its life instead of being cached already-expired.
+fn effective_margin(ttl: Duration, margin: Duration) -> Duration {
+    margin.min(ttl / 2)
+}
+
+// -----------------------------------------------------------------------------
+// TokenCache
+// -----------------------------------------------------------------------------
+
+/// A cached value and the instant after which it must not be used.
+struct Entry<T> {
+    /// The cached value itself.
+    value: T,
+
+    /// Instant, already adjusted by the cache's margin, after which the
+    /// value must not be used.
+    expires_at: Instant,
+}
+
+/// Return a clone of `entry`'s value if it is still valid, `None`
+/// otherwise (missing or past its safety-margin-adjusted expiry).
+fn valid_cached_value<T: Clone>(entry: Option<&Entry<T>>) -> Option<T> {
+    entry
+        .filter(|entry| entry.expires_at > Instant::now())
+        .map(|entry| entry.value.clone())
+}
+
+/// Cache-through cache for one credential. See the module docs.
+pub struct TokenCache<T> {
+    /// Safety window subtracted from a fetched value's TTL (capped at
+    /// half the TTL) before it is treated as expired.
+    margin: Duration,
+
+    /// The cached entry, if any.
+    cache: RwLock<Option<Entry<T>>>,
+}
+
+impl<T: Clone + Send + Sync> TokenCache<T> {
+    /// Build an empty cache. `margin` is the safety window subtracted
+    /// from a fetched value's TTL (capped at half the TTL) before it is
+    /// treated as expired.
+    pub fn new(margin: Duration) -> Self {
+        Self {
+            margin,
+            cache: RwLock::new(None),
+        }
+    }
+
+    /// Return a cached, valid value — fetching a new one first if the
+    /// cache is empty or the cached value is within its safety margin
+    /// of expiry.
+    ///
+    /// At most one concurrent caller ever calls `fetch`: everyone else
+    /// who finds the cache stale queues on the exclusive lock behind
+    /// the first caller and, once it releases the lock, re-checks and
+    /// finds the value that caller just published.
+    ///
+    /// # Errors
+    ///
+    /// Returns whatever `fetch` returns on failure. Nothing is cached
+    /// in that case, so the next call tries again.
+    #[expect(
+        clippy::significant_drop_tightening,
+        reason = "guard's last use is immediately before each return; an explicit drop() would just \
+                   restate that, not shorten the actual hold time"
+    )]
+    pub async fn get_or_refresh<F, Fut, E>(&self, fetch: F) -> Result<T, E>
+    where
+        F: FnOnce() -> Fut + Send,
+        Fut: Future<Output = Result<(T, Duration), E>> + Send,
+    {
+        // Fast path: shared read lock, no fetch.
+        let fresh = valid_cached_value(self.cache.read().await.as_ref());
+        if let Some(value) = fresh {
+            return Ok(value);
+        }
+
+        // Slow path: exclusive lock, re-check (another caller may have
+        // already refreshed it while this one waited for the lock).
+        let mut guard = self.cache.write().await;
+        let fresh = valid_cached_value(guard.as_ref());
+        if let Some(value) = fresh {
+            return Ok(value);
+        }
+
+        let (value, ttl) = fetch().await?;
+        if ttl <= self.margin {
+            warn!(
+                ttl_secs = ttl.as_secs(),
+                margin_secs = self.margin.as_secs(),
+                "token TTL is unusually short; validity margin reduced"
+            );
+        }
+        let expires_at = Instant::now() + ttl.saturating_sub(effective_margin(ttl, self.margin));
+        *guard = Some(Entry {
+            value: value.clone(),
+            expires_at,
+        });
+        Ok(value)
+    }
+}
+
+// -----------------------------------------------------------------------------
+// Tests
+// -----------------------------------------------------------------------------
+
+#[cfg(test)]
+#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
+#[allow(clippy::unwrap_used, clippy::expect_used, reason = "tests")]
+mod tests {
+    use std::sync::atomic::{AtomicU32, Ordering};
+
+    use super::{TokenCache, effective_margin};
+
+    #[test]
+    fn effective_margin_caps_at_half_ttl() {
+        use std::time::Duration;
+        assert_eq!(
+            effective_margin(Duration::from_secs(3600), Duration::from_secs(30)),
+            Duration::from_secs(30)
+        );
+        assert_eq!(
+            effective_margin(Duration::from_secs(40), Duration::from_secs(30)),
+            Duration::from_secs(20)
+        );
+        assert_eq!(
+            effective_margin(Duration::from_secs(10), Duration::from_secs(30)),
+            Duration::from_secs(5)
+        );
+    }
+
+    #[tokio::test]
+    async fn first_call_fetches_and_caches() {
+        let cache: TokenCache<u32> = TokenCache::new(std::time::Duration::from_millis(5));
+        let calls = AtomicU32::new(0);
+
+        let value = cache
+            .get_or_refresh(|| async {
+                calls.fetch_add(1, Ordering::SeqCst);
+                Ok::<_, &str>((42_u32, std::time::Duration::from_secs(60)))
+            })
+            .await
+            .expect("fetch must succeed");
+
+        assert_eq!(value, 42);
+        assert_eq!(calls.load(Ordering::SeqCst), 1, "fetch must be called exactly once");
+    }
+
+    #[tokio::test]
+    async fn second_call_within_ttl_does_not_refetch() {
+        let cache: TokenCache<u32> = TokenCache::new(std::time::Duration::from_millis(5));
+        let calls = AtomicU32::new(0);
+        let fetch = || async {
+            calls.fetch_add(1, Ordering::SeqCst);
+            Ok::<_, &str>((7_u32, std::time::Duration::from_secs(60)))
+        };
+
+        let first = cache.get_or_refresh(fetch).await.expect("first fetch must succeed");
+        let second = cache.get_or_refresh(fetch).await.expect("second call must succeed");
+
+        assert_eq!(first, 7);
+        assert_eq!(second, 7);
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            1,
+            "a still-valid cache must not trigger a second fetch"
+        );
+    }
+
+    #[tokio::test]
+    async fn refetch_after_expiry() {
+        let cache: TokenCache<u32> = TokenCache::new(std::time::Duration::from_millis(1));
+        let calls = AtomicU32::new(0);
+        let fetch = || async {
+            let n = calls.fetch_add(1, Ordering::SeqCst) + 1;
+            Ok::<_, &str>((n, std::time::Duration::from_millis(5)))
+        };
+
+        let first = cache.get_or_refresh(fetch).await.expect("first fetch must succeed");
+        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
+        let second = cache.get_or_refresh(fetch).await.expect("second fetch must succeed");
+
+        assert_eq!(first, 1);
+        assert_eq!(second, 2, "an expired cache must trigger a fresh fetch");
+        assert_eq!(calls.load(Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test]
+    async fn concurrent_calls_with_empty_cache_fetch_exactly_once() {
+        let cache: TokenCache<u32> = TokenCache::new(std::time::Duration::from_millis(5));
+        let calls = std::sync::Arc::new(AtomicU32::new(0));
+        let cache = std::sync::Arc::new(cache);
+
+        let mut handles = Vec::new();
+        for _ in 0..10 {
+            let cache = std::sync::Arc::clone(&cache);
+            let calls = std::sync::Arc::clone(&calls);
+            handles.push(tokio::spawn(async move {
+                cache
+                    .get_or_refresh(|| async move {
+                        calls.fetch_add(1, Ordering::SeqCst);
+                        // Widen the race window so all 10 tasks are
+                        // guaranteed to observe the empty cache before
+                        // the first one publishes a value.
+                        tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+                        Ok::<_, &str>((99_u32, std::time::Duration::from_secs(60)))
+                    })
+                    .await
+                    .expect("fetch must succeed")
+            }));
+        }
+
+        for handle in handles {
+            assert_eq!(handle.await.expect("task must not panic"), 99);
+        }
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            1,
+            "10 concurrent callers against an empty cache must fetch exactly once"
+        );
+    }
+
+    #[tokio::test]
+    async fn fetch_error_is_propagated_and_not_cached() {
+        let cache: TokenCache<u32> = TokenCache::new(std::time::Duration::from_millis(5));
+        let calls = AtomicU32::new(0);
+
+        let err = cache
+            .get_or_refresh(|| async {
+                calls.fetch_add(1, Ordering::SeqCst);
+                Err::<(u32, std::time::Duration), _>("boom")
+            })
+            .await
+            .expect_err("failed fetch must propagate the error");
+
+        assert_eq!(err, "boom");
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+
+        // Nothing was cached, so the next call must try again.
+        let value = cache
+            .get_or_refresh(|| async {
+                calls.fetch_add(1, Ordering::SeqCst);
+                Ok::<_, &str>((5_u32, std::time::Duration::from_secs(60)))
+            })
+            .await
+            .expect("retry must succeed");
+        assert_eq!(value, 5);
+        assert_eq!(calls.load(Ordering::SeqCst), 2, "a failed fetch must not be cached");
+    }
+}

docs/filters/azure_ad.md

@@ -20,5 +20,4 @@ client_id: 11111111-1111-1111-1111-111111111111
 scope: https://cognitiveservices.azure.com/.default
 client_secret_env_var: AZURE_CLIENT_SECRET
 authority_host: login.microsoftonline.com   # optional, for sovereign clouds
-refresh_ratio: 0.75                          # optional, refresh at 75% of the usable lifetime
 ```

docs/filters/gcp_adc.md

@@ -9,15 +9,15 @@
 
 Experimental: requires the `gcp-adc-filter` cargo feature, which is off by default and activates the `experimental` marker. This filter is a work in progress and its configuration surface may change between releases.
 
-**Token acquisition is not implemented yet.** This filter currently establishes the configuration surface, credential-source resolution, and the fail-closed request path: the token cache is never populated, so every request is rejected with `503`. Fetch (metadata server, key file) arrives with the shared background-refresh primitive tracked in praxis#555, praxis#1042, and praxis#1043 — filters must not spawn their own refresher threads.
+Acquires a token via Application Default Credentials (GKE metadata server) and injects `Authorization: Bearer <token>` on every proxied request, keeping GCP credentials invisible to the downstream client. There is no background refresh thread: caching is cache-through, the same as [`crate::azure::azure_ad`] — see [`praxis_ai_apis::token_cache::TokenCache`] for the exact contract.
 
-Once implemented, the filter will acquire a token via Application Default Credentials (GKE metadata or a service-account key file) and inject `Authorization: Bearer <token>` on every proxied request, keeping GCP credentials invisible to the downstream client.
+**Service-account key file (`source: key_file`) token fetch is not implemented yet** — it needs `JWT` signing, which this workspace does not currently depend on. Config parsing, file resolution, and validation for `key_file` all work; `on_request` fails closed with a clear "not implemented" reason instead of silently 503ing forever.
 
 Credential-source resolution happens at construct time: `GOOGLE_APPLICATION_CREDENTIALS` is read once when the pipeline is built (reload the config to pick up changes), and a `gcloud` user credential file (`authorized_user`) is rejected as a configuration error rather than silently falling through the ADC chain.
 
 This filter only injects `Authorization`. Pointing the request at the correct Vertex endpoint (cluster `endpoints` + `tls.sni`) is the operator's responsibility.
 
-Until a token is cached, and whenever the cached token is missing or expired, requests are rejected with `503` rather than forwarded unauthenticated.
+Whenever no valid token can be produced — none cached and the inline fetch fails — the request is rejected with `503` rather than forwarded unauthenticated.
 
 ## Configuration
 
@@ -27,13 +27,12 @@
 | `scope` | string | no | `OAuth2` scope requested with the access token. |
 | `service_account` | string | no | Metadata service account email, or `default` (the default). Only used with the `metadata` and `adc` sources; rejected for `key_file`. Interpolated into the metadata path, so it must be `default` or a service-account email (letters, digits, `@`, `.`, `-`, `_`). |
 | `credentials_file` | string | no | Path to a service-account key JSON file. Required when `source` is `key_file`; rejected for the other sources (`adc` reads `GOOGLE_APPLICATION_CREDENTIALS` instead). |
-| `refresh_ratio` | number | no | Fraction of a token's TTL at which to refresh it. Must be in the open interval `(0, 1)`. |
+| `metadata_host` | string | no | GCE/GKE metadata server host. Defaults to the real metadata server; the only other accepted value is a `127.0.0.1` loopback address, to point tests at a local mock. The metadata endpoint is only safe to reach over plain HTTP because it never leaves the VM/host, so nothing else is accepted (not even `localhost`, which is a resolvable hostname rather than a fixed address). |
 
 ## Example
 
 ```yaml
 filter: gcp_adc
 source: adc
 scope: https://www.googleapis.com/auth/cloud-platform
-refresh_ratio: 0.75
 ```

examples/README.md

@@ -26,7 +26,7 @@ before sending requests.
 | [aws-sigv4.yaml](configs/aws-sigv4.yaml) | Signs outbound requests to an AWS service (Bedrock, in this example) using Signature Version 4. Credentials are static, sourced from environment variables — see the module docs on Sigv4SignFilter for the planned OIDC/default-credential-chain follow-up |
 | [azure-ad.yaml](configs/azure-ad.yaml) | Acquires an Entra ID bearer token via the client-credentials grant and injects "Authorization: Bearer <token>" on every proxied request to Azure OpenAI |
 | [credential-injection.yaml](configs/credential-injection.yaml) | Injects per-cluster API credentials into upstream requests and strips client-provided credentials to prevent forwarding |
-| [gcp-adc.yaml](configs/gcp-adc.yaml) | Establishes the gcp_adc filter's configuration surface and fail-closed behavior |
+| [gcp-adc.yaml](configs/gcp-adc.yaml) | Acquires an OAuth2 access token from the GCE/GKE metadata server (source: adc or metadata) and injects "Authorization: Bearer <token>" on every proxied request to Vertex AI |
 | [intelligent-route-all-capabilities.yaml](configs/intelligent-route-all-capabilities.yaml) | Demonstrates every candidate capability and selection input handled by intelligent_route today |
 | [intelligent-route-inference.yaml](configs/intelligent-route-inference.yaml) | Routes requests to different upstream clusters based on the inference model name extracted from a configured request header.  The header value is set by an earlier filter such as `json_body_field` |
 | [intelligent-route-mcp.yaml](configs/intelligent-route-mcp.yaml) | Routes MCP `tools/call` requests to the cluster that owns the requested tool, using the `mcp.name` metadata set by the `mcp` filter |

examples/configs/azure-ad.yaml

@@ -2,9 +2,9 @@
 #
 # Acquires an Entra ID bearer token via the client-credentials grant
 # and injects "Authorization: Bearer <token>" on every proxied
-# request to Azure OpenAI. The token is cached in memory and refreshed
-# in the background before it expires; the downstream client never
-# sees Entra ID.
+# request to Azure OpenAI. The token is cached in memory (cache-through:
+# a request that finds the cache stale fetches a fresh token inline
+# before proceeding); the downstream client never sees Entra ID.
 #
 # This filter only injects the Authorization header — it does NOT
 # manage the upstream Host. Point the backend cluster at your Azure
@@ -40,7 +40,6 @@ filter_chains:
         scope: https://cognitiveservices.azure.com/.default
         client_secret_env_var: AZURE_CLIENT_SECRET
         # authority_host: login.microsoftonline.com  # optional, for sovereign clouds
-        # refresh_ratio: 0.75                         # optional, refresh at 75% of TTL
 
 insecure_options:
   allow_private_endpoints: true # example proxies to a local backend