review: batch-end quality round

control-plane/web/src/App.tsx

@@ -27,6 +27,27 @@ export function useAuth(): AuthState {
   return value;
 }
 
+interface NavItem {
+  to: string;
+  label: string;
+  icon: string;
+  minRole: Role;
+  page: ReactNode;
+  end?: boolean;
+}
+
+const roleRank: Record<Role, number> = { member: 0, tenant_admin: 1, system_admin: 2 };
+
+const navItems: NavItem[] = [
+  { to: "/", label: "Overview", icon: "◫", end: true, minRole: "member", page: <OverviewPage /> },
+  { to: "/usage", label: "Usage & cost", icon: "⌁", minRole: "member", page: <UsagePage /> },
+  { to: "/availability", label: "Availability", icon: "◉", minRole: "member", page: <AvailabilityPage /> },
+  { to: "/keys", label: "Access keys", icon: "⌘", minRole: "tenant_admin", page: <KeysPage /> },
+  { to: "/audit", label: "Audit", icon: "≣", minRole: "tenant_admin", page: <AuditPage /> },
+  { to: "/users", label: "Users & roles", icon: "♙", minRole: "system_admin", page: <UsersPage /> },
+  { to: "/configuration", label: "Configuration", icon: "⌗", minRole: "system_admin", page: <ConfigPage /> },
+];
+
 export default function App() {
   const [session, updateSession] = useState<Session | null>(null);
   const [loading, setLoading] = useState(true);
@@ -69,27 +90,6 @@ export default function App() {
   );
 }
 
-interface NavItem {
-  to: string;
-  label: string;
-  icon: string;
-  minRole: Role;
-  page: ReactNode;
-  end?: boolean;
-}
-
-const roleRank: Record<Role, number> = { member: 0, tenant_admin: 1, system_admin: 2 };
-
-const navItems: NavItem[] = [
-  { to: "/", label: "Overview", icon: "◫", end: true, minRole: "member", page: <OverviewPage /> },
-  { to: "/usage", label: "Usage & cost", icon: "⌁", minRole: "member", page: <UsagePage /> },
-  { to: "/availability", label: "Availability", icon: "◉", minRole: "member", page: <AvailabilityPage /> },
-  { to: "/keys", label: "Access keys", icon: "⌘", minRole: "tenant_admin", page: <KeysPage /> },
-  { to: "/audit", label: "Audit", icon: "≣", minRole: "tenant_admin", page: <AuditPage /> },
-  { to: "/users", label: "Users & roles", icon: "♙", minRole: "system_admin", page: <UsersPage /> },
-  { to: "/configuration", label: "Configuration", icon: "⌗", minRole: "system_admin", page: <ConfigPage /> },
-];
-
 function Shell() {
   const { session, logout } = useAuth();
   const role = session.user.role;

control-plane/web/src/pages/ConfigPage.tsx

@@ -5,13 +5,9 @@ import { dateTime } from "../format";
 import { useAPI, useAction } from "../hooks";
 import type { ConfigDocument, ConfigVersion } from "../types";
 
-interface VersionList {
-  versions: ConfigVersion[];
-}
-
 export default function ConfigPage() {
   const current = useAPI<ConfigDocument>("/api/v1/admin/config");
-  const history = useAPI<VersionList>("/api/v1/admin/config/versions");
+  const history = useAPI<{ versions: ConfigVersion[] }>("/api/v1/admin/config/versions");
   const [yaml, setYAML] = useState("");
   const [message, setMessage] = useState("");
   const { run, busy, error } = useAction();

crates/config/src/lib.rs

@@ -347,6 +347,19 @@ pub struct StabilityConf {
     pub availability_min_samples: u64,
 }
 
+impl Default for StabilityConf {
+    fn default() -> Self {
+        Self {
+            failure_threshold: default_failure_threshold(),
+            cooldown_seconds: default_cooldown_seconds(),
+            availability_window_minutes: default_availability_window_minutes(),
+            unstable_error_rate: default_unstable_error_rate(),
+            unavailable_error_rate: default_unavailable_error_rate(),
+            availability_min_samples: default_availability_min_samples(),
+        }
+    }
+}
+
 fn default_failure_threshold() -> usize {
     3
 }
@@ -366,19 +379,6 @@ fn default_cooldown_seconds() -> u64 {
     30
 }
 
-impl Default for StabilityConf {
-    fn default() -> Self {
-        Self {
-            failure_threshold: default_failure_threshold(),
-            cooldown_seconds: default_cooldown_seconds(),
-            availability_window_minutes: default_availability_window_minutes(),
-            unstable_error_rate: default_unstable_error_rate(),
-            unavailable_error_rate: default_unavailable_error_rate(),
-            availability_min_samples: default_availability_min_samples(),
-        }
-    }
-}
-
 /// One automatic-suspension tier: this many admission rejections in a day
 /// suspend the key for `suspend_hours`.
 #[derive(Debug, Clone, Deserialize)]
@@ -754,7 +754,6 @@ impl GatewayConfig {
                 protocols: preset.wires.iter().map(|w| (*w).to_owned()).collect(),
             });
         }
-        // normalize the global policy and every tenant override once at load
         compile_security(&mut self.security);
         for t in &mut self.tenants {
             if let Some(sec) = t.security.as_mut() {

crates/dag/src/nodes.rs

@@ -208,17 +208,18 @@ impl DagNode for CacheLookup {
         &["variant_select"]
     }
     async fn execute(&self, ctx: &mut DagContext) -> GResult<()> {
-        let param = match ctx.request.model_param_v2.as_ref() {
-            Some(p) => p,
-            None => return Ok(()),
+        let Some(param) = ctx.request.model_param_v2.as_ref() else {
+            return Ok(());
         };
         // online non-streaming cache_ttl models only; batch items bypass —
         // a hit is free (unbilled) and batches promise per-item billing
-        let ttl = ctx
+        let Some(ttl) = ctx
             .cfg
             .find_model(&param.model_name)
-            .and_then(|m| m.cache_ttl_seconds);
-        let Some(ttl) = ttl else { return Ok(()) };
+            .and_then(|m| m.cache_ttl_seconds)
+        else {
+            return Ok(());
+        };
         if ctx.request.stream || !ctx.request.is_online {
             return Ok(());
         }

crates/engines/src/bespoke.rs

@@ -494,11 +494,13 @@ fn dashscope_raw_usage(resp: &GatewayResponse) -> Vec<u8> {
 
 #[cfg(test)]
 mod tests {
-    use super::*;
-    use crate::transport::{MockTransport, SharedTransport};
+    use std::sync::Arc;
+
     use gw_consts::Protocol;
     use gw_models::{ChatMsg, GatewayRequest, ModelParamV2};
-    use std::sync::Arc;
+
+    use super::*;
+    use crate::transport::{MockTransport, SharedTransport};
 
     fn req(mt: Protocol, name: &str) -> GatewayRequest {
         GatewayRequest {

crates/engines/src/families.rs

@@ -818,7 +818,10 @@ impl ResponsesEngine {
             } else {
                 Some(Value::Array(tool_calls))
             },
-            model: v["model"].as_str().unwrap_or(&self.model_name()).to_owned(),
+            model: v["model"]
+                .as_str()
+                .map(str::to_owned)
+                .unwrap_or_else(|| self.model_name()),
             finish_reason: v["status"].as_str().unwrap_or("completed").to_owned(),
             prompt_tokens: input,
             completion_tokens: output,

crates/engines/src/http_transport.rs

@@ -83,13 +83,6 @@ impl HttpTransport {
         let p = self.policies.load();
         p.per_account.get(account).copied().unwrap_or(p.default)
     }
-
-    fn set_policies(&self, default: UpstreamPolicy, per_account: HashMap<String, UpstreamPolicy>) {
-        self.policies.store(std::sync::Arc::new(Policies {
-            default,
-            per_account,
-        }));
-    }
 }
 
 #[async_trait::async_trait]
@@ -99,7 +92,10 @@ impl Transport for HttpTransport {
         default: UpstreamPolicy,
         per_account: HashMap<String, UpstreamPolicy>,
     ) {
-        self.set_policies(default, per_account);
+        self.policies.store(std::sync::Arc::new(Policies {
+            default,
+            per_account,
+        }));
     }
 
     async fn send(&self, req: UpstreamRequest) -> GResult<UpstreamResponse> {

crates/handler/src/lib.rs

@@ -268,11 +268,10 @@ impl OnlineHandler {
         Ok(ctx)
     }
 
-    /// Run the wired moderator over raw text; `Some(reason)` to deny, `None` to
-    /// allow. The seam the realtime surface uses (it has no `DagContext`); the
-    /// caller records the security event on its own surface.
-    /// [`Self::moderation`] narrowed to the realtime surface: a live session
-    /// can't switch models, so `Degrade` denies there.
+    /// Run the wired moderator over raw text; the seam the realtime surface
+    /// uses (it has no `DagContext`), so the caller records the security event
+    /// on its own surface. [`Self::moderation`] narrowed to this surface: a
+    /// live session can't switch models, so `Degrade` denies here instead.
     pub async fn moderate_rt(&self, sec: &gw_config::SecurityConf, text: &str) -> RtModeration {
         match self.moderation(sec, text).await {
             Moderation::Allow => RtModeration::Allow,
@@ -349,10 +348,7 @@ pub enum RtModeration {
 /// A per-request correlation id: `req-<epoch_ms>-<seq>`, time-sortable and
 /// unique within the process (the seq disambiguates same-millisecond requests).
 pub fn new_request_id() -> String {
-    let ms = std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .map(|d| d.as_millis())
-        .unwrap_or(0);
+    let ms = gw_state::epoch_millis();
     format!("req-{ms}-{}", REQ_SEQ.fetch_add(1, Ordering::Relaxed))
 }
 

crates/handler/src/offline.rs

@@ -49,7 +49,7 @@ impl OfflineHandler {
                 .batch_create(&ak.ak, &ak.tenant, &model, items.len())
                 .await?;
             let this = self.clone();
-            let (id, model) = (job.id.clone(), model);
+            let id = job.id.clone();
             // items are captured HERE: an erasure landing after this instant
             // must stop them, so the marker comparison point is submission,
             // not the spawned executor's first poll

crates/handler/src/plugins.rs

@@ -33,13 +33,7 @@ pub fn security_check(sec: &SecurityConf, request: &mut GatewayRequest) -> ScanO
         return ScanOutcome::default();
     }
     let mut counts = ScanCounts::new(sec);
-    let mut visit = |s: &mut String| counts.visit(s);
-    for msg in &mut request.message {
-        for_each_message_text(msg, &mut visit);
-    }
-    if let Some(param) = request.model_param_v2.as_mut() {
-        for_each_param_text(param, &mut visit);
-    }
+    for_each_request_text(request, &mut |s| counts.visit(s));
     counts.outcome()
 }
 
@@ -98,22 +92,16 @@ impl<'a> ScanCounts<'a> {
 
 /// All inbound text a request carries — message content, multimodal text
 /// parts, assistant tool_calls, the Responses raw body, and the family typed
-/// params — collected via the same traversals the blocklist scan and DLP run,
+/// params — collected via the same traversal the blocklist scan and DLP run,
 /// so the field lists cannot drift apart. The one text view moderation and
-/// content retention operate on. `&mut` only to share those traversals;
-/// nothing is rewritten.
+/// content retention operate on. `&mut` only to share that traversal; nothing
+/// is rewritten.
 pub fn inbound_text(request: &mut GatewayRequest) -> String {
     let mut out = String::new();
-    let mut collect = |s: &mut String| {
+    for_each_request_text(request, &mut |s| {
         push_text(&mut out, s);
         0
-    };
-    for m in &mut request.message {
-        for_each_message_text(m, &mut collect);
-    }
-    if let Some(param) = request.model_param_v2.as_mut() {
-        for_each_param_text(param, &mut collect);
-    }
+    });
     out
 }
 
@@ -130,12 +118,7 @@ fn push_text(out: &mut String, s: &str) {
 /// the request's text slots. Returns the number of replaced ranges.
 pub fn apply_mask_spans(request: &mut GatewayRequest, spans: &[std::ops::Range<usize>]) -> usize {
     let mut masker = SpanMasker::new(spans);
-    for msg in &mut request.message {
-        for_each_message_text(msg, &mut |s| masker.apply(s));
-    }
-    if let Some(param) = request.model_param_v2.as_mut() {
-        for_each_param_text(param, &mut |s| masker.apply(s));
-    }
+    for_each_request_text(request, &mut |s| masker.apply(s));
     masker.hits
 }
 
@@ -252,6 +235,24 @@ fn walk_json_strings(v: &mut serde_json::Value, f: &mut impl FnMut(&mut String)
     }
 }
 
+/// Every text-bearing field of a whole request — each message plus the tail
+/// params — the ONE per-request traversal the blocklist scan, inbound-text
+/// view, mask application, and DLP redaction all share, so they can't drift
+/// apart field by field.
+fn for_each_request_text(
+    request: &mut GatewayRequest,
+    f: &mut impl FnMut(&mut String) -> usize,
+) -> usize {
+    let mut n = 0;
+    for msg in &mut request.message {
+        n += for_each_message_text(msg, f);
+    }
+    if let Some(param) = request.model_param_v2.as_mut() {
+        n += for_each_param_text(param, f);
+    }
+    n
+}
+
 /// The text-bearing fields of one chat turn — flat content, multimodal parts,
 /// and assistant tool_calls, all forwarded to the vendor by the engines — the
 /// ONE per-message field list the blocklist scan, DLP, and the moderation
@@ -431,17 +432,8 @@ pub fn dlp_redact_request(sec: &SecurityConf, request: &mut GatewayRequest) -> u
     if !sec.dlp_redact && !sec.detect_secrets {
         return 0;
     }
-    let secrets = sec.detect_secrets;
-    let pii = sec.dlp_redact;
-    let mut redact_field = |s: &mut String| redact_in_place(s, pii, secrets);
-    let mut hits = 0;
-    for msg in &mut request.message {
-        hits += for_each_message_text(msg, &mut redact_field);
-    }
-    if let Some(param) = request.model_param_v2.as_mut() {
-        hits += for_each_param_text(param, &mut redact_field);
-    }
-    hits
+    let (pii, secrets) = (sec.dlp_redact, sec.detect_secrets);
+    for_each_request_text(request, &mut |s| redact_in_place(s, pii, secrets))
 }
 
 /// A privacy-safe copy of `text` for content retention: PII and secrets are

crates/server/src/main.rs

@@ -223,18 +223,18 @@ fn select_transport() -> anyhow::Result<gw_engines::SharedTransport> {
     Ok(match env::var("GW_TRANSPORT").as_deref() {
         Ok("mock") => {
             tracing::info!("transport = mock (zero egress)");
-            std::sync::Arc::new(gw_engines::MockTransport)
+            Arc::new(gw_engines::MockTransport)
         }
         Ok("http") => {
             tracing::info!("transport = http (accounts without an endpoint fail)");
-            std::sync::Arc::new(gw_engines::http_transport::HttpTransport::with_policies(
+            Arc::new(gw_engines::http_transport::HttpTransport::with_policies(
                 Default::default(),
                 Default::default(),
             )?)
         }
         _ => {
             tracing::info!("transport = auto (mock:// in-process, real URLs over HTTP)");
-            std::sync::Arc::new(
+            Arc::new(
                 gw_engines::http_transport::DispatchTransport::with_policies(
                     Default::default(),
                     Default::default(),