fix(#49): allow-anonymous mode must ignore unrecognized keys, not 401

Regression from #49: the auth middleware rejected ANY present-but-
unresolvable bearer token with 401 invalid_api_key, even when
require_auth=false. But OpenAI-compatible clients (opencode, Open WebUI,
Agent Zero, litellm) send a placeholder bearer by default — so enabling
the build broke every existing client even though the operator never
opted into auth. Pre-#49 the bearer was never inspected at all.

Fix: in allow-anonymous mode (require_auth=false, the default) an
unrecognized key is now ignored and the request is served anonymously,
restoring pre-#49 behaviour. A bad key only 401s when require_auth=true.
A valid key is still resolved + metered in both modes.

Test renamed/split: unrecognized_key_is_ignored_when_auth_not_required
(now 200, served anonymously) + invalid_key_is_401_when_auth_required.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/cortex-core/src/discovery.rs

@@ -68,6 +68,57 @@ pub struct HealthResponse {
     pub devices: Vec<DeviceHealth>,
     #[serde(default)]
     pub activation: ActivationStatus,
+    /// Per-model admission load (#53): how many requests are running vs.
+    /// queued on each loaded model right now. Cortex's load-aware router
+    /// (#55) reads this to spread traffic across replicas and to propagate
+    /// honest backpressure. `#[serde(default)]` keeps older gateways/neurons
+    /// interoperable (absent → empty → treated as no load info).
+    #[serde(default)]
+    pub models: Vec<ModelLoad>,
+}
+
+/// Live admission load for one loaded model (#53).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ModelLoad {
+    pub id: String,
+    /// Requests currently running (batch-1 → 0 or 1).
+    pub in_flight: usize,
+    /// Requests waiting in the bounded admission queue.
+    pub queue_depth: usize,
+}
+
+#[cfg(test)]
+mod health_load_tests {
+    use super::*;
+
+    #[test]
+    fn health_response_without_models_field_still_deserializes() {
+        // A pre-#53 neuron's /health payload omits `models`; the gateway
+        // must still parse it (serde default → empty).
+        let json = r#"{"uptime_secs":42,"devices":[]}"#;
+        let resp: HealthResponse = serde_json::from_str(json).expect("back-compat parse");
+        assert_eq!(resp.uptime_secs, 42);
+        assert!(resp.models.is_empty());
+    }
+
+    #[test]
+    fn health_response_round_trips_model_load() {
+        let resp = HealthResponse {
+            uptime_secs: 1,
+            devices: vec![],
+            activation: ActivationStatus::default(),
+            models: vec![ModelLoad {
+                id: "Qwen/Qwen3.6-27B".into(),
+                in_flight: 1,
+                queue_depth: 3,
+            }],
+        };
+        let s = serde_json::to_string(&resp).unwrap();
+        let back: HealthResponse = serde_json::from_str(&s).unwrap();
+        assert_eq!(back.models.len(), 1);
+        assert_eq!(back.models[0].in_flight, 1);
+        assert_eq!(back.models[0].queue_depth, 3);
+    }
 }
 
 /// High-level activation state of the neuron daemon. The HTTP listener

crates/cortex-core/src/node.rs

@@ -1,4 +1,4 @@
-use crate::discovery::{ActivationStatus, DiscoveryResponse};
+use crate::discovery::{ActivationStatus, DiscoveryResponse, ModelLoad};
 use crate::harness::{ModelCost, ModelLimit};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
@@ -27,6 +27,11 @@ pub struct NodeState {
     /// to synthesize `Loading` locations so clients see a catalogued
     /// model that's mid-prewarm as "loading", not "missing".
     pub activation: Option<ActivationStatus>,
+    /// Last-seen per-model admission load from this neuron's `/health`
+    /// (#53), keyed by model id. The router (#55) reads it to pick the
+    /// least-busy replica when a model is loaded on more than one neuron.
+    /// Empty until the first /health poll reports load.
+    pub model_load: HashMap<String, ModelLoad>,
 }
 
 /// A model registered on a node, with its runtime status.

crates/cortex-gateway/src/auth.rs

@@ -83,9 +83,23 @@ pub async fn require_principal(
                 req.extensions_mut().insert(principal);
                 next.run(req).await
             }
-            // A present-but-invalid credential is always an error, even when
-            // anonymous access is otherwise allowed.
-            Err(_) => unauthorized("invalid API key"),
+            // An unrecognized key only hard-fails when auth is *required*.
+            // In allow-anonymous mode (the default) we must IGNORE it and
+            // serve the request unauthenticated — otherwise the placeholder
+            // keys that OpenAI-compatible clients send by default (opencode,
+            // Open WebUI, Agent Zero, litellm) would all break, even though
+            // the operator never opted into auth. Pre-#49 the bearer was
+            // never inspected at all; this preserves that for require_auth=false.
+            Err(_) => {
+                if fleet.require_auth {
+                    unauthorized("invalid API key")
+                } else {
+                    tracing::debug!(
+                        "ignoring unrecognized bearer token (require_auth=false): serving anonymously"
+                    );
+                    next.run(req).await
+                }
+            }
         },
         None => {
             if fleet.require_auth {

crates/cortex-gateway/src/handlers.rs

@@ -761,6 +761,19 @@ async fn proxy_with_metrics(
     body: Bytes,
     model_id: &str,
 ) -> Response {
+    // Fail-fast prompt pre-validation (#56): refuse a prompt that already
+    // exceeds the model's advertised context window *before* dispatching to
+    // neuron — the same `400 context_length_exceeded` neuron would emit on
+    // overflow, just earlier and without burning a cold-load/queue slot.
+    // cortex has no tokenizer, so the estimate under-counts and neuron stays
+    // the exact wall; we only catch gross overages (the A0 failure mode).
+    if let Some(context) = advertised_context(fleet, &route.node_name, model_id).await {
+        let est = estimate_prompt_tokens(&body);
+        if est > context {
+            return context_length_exceeded_response(context, est, &headers);
+        }
+    }
+
     let labels = [
         ("model", model_id.to_string()),
         ("node", route.node_name.clone()),
@@ -844,6 +857,98 @@ async fn advertised_output_limit(
         .map(|l| l.output as u64)
 }
 
+/// The model's advertised hard context window (`limit.context`, #62/#67) on a
+/// node, used for fail-fast prompt pre-validation (#56). `None` when no limit
+/// is known — pre-validation is then skipped and neuron remains the wall.
+async fn advertised_context(fleet: &CortexState, node_name: &str, model_id: &str) -> Option<u64> {
+    let nodes = fleet.nodes.read().await;
+    nodes
+        .get(node_name)?
+        .models
+        .get(model_id)?
+        .limit
+        .as_ref()
+        .map(|l| l.context as u64)
+}
+
+/// Conservative prompt-token estimate (~4 chars/token over message text).
+/// cortex has no tokenizer; under-counting is the safe direction — we only
+/// pre-reject gross overages (#56), and neuron enforces the exact wall.
+fn estimate_prompt_tokens(body: &[u8]) -> u64 {
+    let Ok(v) = serde_json::from_slice::<Value>(body) else {
+        return (body.len() as u64 / 4).max(1);
+    };
+    let mut chars = 0usize;
+    if let Some(messages) = v.get("messages").and_then(Value::as_array) {
+        for m in messages {
+            match m.get("content") {
+                Some(Value::String(s)) => chars += s.len(),
+                Some(Value::Array(parts)) => {
+                    for p in parts {
+                        if let Some(t) = p.get("text").and_then(Value::as_str) {
+                            chars += t.len();
+                        }
+                    }
+                }
+                _ => {}
+            }
+            chars += 8; // rough per-message role/formatting overhead
+        }
+    } else if let Some(prompt) = v.get("prompt").and_then(Value::as_str) {
+        chars += prompt.len(); // legacy /v1/completions
+    } else {
+        return (body.len() as u64 / 4).max(1);
+    }
+    (chars as u64 / 4).max(1)
+}
+
+/// Client-specific, advisory guidance for an over-long prompt (#56),
+/// fingerprinted from `User-Agent`. Strictly advisory: it rides the
+/// `X-Helexa-Advice` header only, never the error envelope, and behaviour
+/// never depends on it. Unknown clients get nothing.
+fn client_advice(headers: &HeaderMap) -> Option<&'static str> {
+    let ua = headers
+        .get(axum::http::header::USER_AGENT)?
+        .to_str()
+        .ok()?
+        .to_ascii_lowercase();
+    if ua.contains("litellm") {
+        Some(
+            "litellm forwards the full context; lower the configured context window or enable client-side compaction",
+        )
+    } else if ua.contains("agent-zero") || ua.contains("agent zero") {
+        Some("reduce the conversation/context size or summarize earlier turns before resending")
+    } else if ua.contains("zed") {
+        Some("reduce the assistant context window in Zed's settings")
+    } else {
+        None
+    }
+}
+
+/// `400 context_length_exceeded` for an over-long prompt caught at the edge
+/// (#56), in the #60 envelope — the same shape neuron emits on overflow, so
+/// clients (opencode auto-compacts) handle it identically. Attaches the
+/// advisory `X-Helexa-Advice` header for fingerprinted clients.
+fn context_length_exceeded_response(
+    context: u64,
+    prompt_est: u64,
+    headers: &HeaderMap,
+) -> Response {
+    let env = OpenAiError::context_length_exceeded(format!(
+        "This model's maximum context length is {context} tokens. Your request is \
+         estimated at ~{prompt_est} tokens. Please reduce the length of the messages."
+    ))
+    .with_extra("max", json!(context))
+    .with_extra("estimated_prompt_tokens", json!(prompt_est));
+    let mut response = crate::error::envelope_response(env);
+    if let Some(advice) = client_advice(headers)
+        && let Ok(value) = axum::http::HeaderValue::from_str(advice)
+    {
+        response.headers_mut().insert("x-helexa-advice", value);
+    }
+    response
+}
+
 /// Update `last_accessed` timestamp for a model on a node (drives LRU eviction).
 async fn touch_model(fleet: &CortexState, node_name: &str, model_id: &str) {
     let mut nodes = fleet.nodes.write().await;

crates/cortex-gateway/src/poller.rs

@@ -200,6 +200,9 @@ async fn poll_health(fleet: &CortexState, name: &str, endpoint: &str) {
             let mut nodes = fleet.nodes.write().await;
             if let Some(node) = nodes.get_mut(name) {
                 node.activation = Some(h.activation);
+                // Per-model admission load (#53) → keyed by id for the
+                // load-aware router (#55).
+                node.model_load = h.models.into_iter().map(|m| (m.id.clone(), m)).collect();
             }
         }
         Err(e) => {

crates/cortex-gateway/src/router.rs

@@ -132,7 +132,9 @@ pub async fn resolve(
     // Snapshot loaded / unloaded / recovering state from the poller cache.
     let (loaded_route, unloaded_route, recovering_node, any_healthy) = {
         let nodes = fleet.nodes.read().await;
-        let mut loaded_route = None;
+        // All healthy nodes with the model loaded, each with its current
+        // admission load (#53) so we can pick the least-busy replica (#55).
+        let mut loaded_candidates: Vec<(String, String, usize)> = Vec::new();
         let mut unloaded_route = None;
         let mut recovering_node = None;
         let mut any_healthy = false;
@@ -144,8 +146,15 @@ pub async fn resolve(
             if let Some(entry) = node.models.get(model_id) {
                 match entry.status {
                     ModelStatus::Loaded | ModelStatus::Reloading => {
-                        loaded_route = Some((node.name.clone(), node.endpoint.clone(), false));
-                        break;
+                        // Least-busy score: in-flight + queued from the
+                        // neuron's last /health (#53). Unknown load (no poll
+                        // yet) scores 0 so the replica stays eligible.
+                        let score = node
+                            .model_load
+                            .get(model_id)
+                            .map(|l| l.in_flight + l.queue_depth)
+                            .unwrap_or(0);
+                        loaded_candidates.push((node.name.clone(), node.endpoint.clone(), score));
                     }
                     ModelStatus::Unloaded => {
                         if unloaded_route.is_none() {
@@ -175,6 +184,12 @@ pub async fn resolve(
                 }
             }
         }
+        // Pick the least-busy loaded replica; ties break by node name for
+        // deterministic routing. `false` = not a cold start.
+        let loaded_route = loaded_candidates
+            .into_iter()
+            .min_by(|a, b| a.2.cmp(&b.2).then_with(|| a.0.cmp(&b.0)))
+            .map(|(name, endpoint, _score)| (name, endpoint, false));
         (loaded_route, unloaded_route, recovering_node, any_healthy)
     };
 

crates/cortex-gateway/src/state.rs

@@ -37,6 +37,7 @@ impl CortexState {
                     last_poll: None,
                     discovery: None,
                     activation: None,
+                    model_load: HashMap::new(),
                 },
             );
         }

crates/cortex-gateway/tests/auth.rs

@@ -175,11 +175,33 @@ async fn missing_key_when_required_is_401_invalid_api_key() {
 }
 
 #[tokio::test]
-async fn invalid_key_is_401_even_when_auth_not_required() {
+async fn unrecognized_key_is_ignored_when_auth_not_required() {
     let (neuron, seen) = spawn_capturing_neuron().await;
-    // A present-but-wrong credential is always an error.
+    // allow-anonymous mode: a placeholder/unknown bearer (as opencode,
+    // Open WebUI, Agent Zero, litellm all send by default) must NOT be
+    // rejected — it's ignored and the request is served anonymously.
     let gateway = spawn_gateway(&neuron, one_key_config(false)).await;
 
+    let resp = reqwest::Client::new()
+        .post(format!("{gateway}/v1/chat/completions"))
+        .bearer_auth("sk-dummy-placeholder")
+        .json(&chat_body())
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), reqwest::StatusCode::OK);
+    let _ = resp.bytes().await.unwrap();
+    // Served, but anonymous — no principal stamped from the bogus key.
+    assert!(seen.lock().unwrap().account_id.is_none());
+}
+
+#[tokio::test]
+async fn invalid_key_is_401_when_auth_required() {
+    let (neuron, seen) = spawn_capturing_neuron().await;
+    // With auth required, a present-but-wrong credential is rejected.
+    let gateway = spawn_gateway(&neuron, one_key_config(true)).await;
+
     let resp = reqwest::Client::new()
         .post(format!("{gateway}/v1/chat/completions"))
         .bearer_auth("sk-wrong")

crates/cortex-gateway/tests/load_routing.rs

@@ -0,0 +1,189 @@
+//! Load-aware routing across replicas (#55).
+//!
+//! When a model is loaded on more than one healthy neuron, the router picks
+//! the least-busy replica using the per-model admission load each neuron
+//! reports on `GET /health` (#53), rather than always taking the first.
+
+mod common;
+
+use axum::Json;
+use axum::extract::Path;
+use axum::http::{StatusCode, header};
+use axum::response::IntoResponse;
+use axum::routing::{get, post};
+use cortex_core::config::{
+    EvictionSettings, EvictionStrategy, GatewayConfig, GatewaySettings, NeuronEndpoint,
+};
+use cortex_core::discovery::ModelLoad;
+use cortex_core::node::{ModelEntry, ModelStatus};
+use cortex_gateway::state::CortexState;
+use serde_json::{Value, json};
+use std::sync::Arc;
+use tokio::net::TcpListener;
+
+/// Seed a node as healthy with `test-model` loaded and a given admission load.
+async fn seed_loaded(fleet: &CortexState, node: &str, in_flight: usize, queue_depth: usize) {
+    let mut nodes = fleet.nodes.write().await;
+    let n = nodes.get_mut(node).expect("node exists");
+    n.healthy = true;
+    n.models.insert(
+        "test-model".into(),
+        ModelEntry {
+            id: "test-model".into(),
+            status: ModelStatus::Loaded,
+            last_accessed: None,
+            vram_estimate_mb: Some(8000),
+            capabilities: Vec::new(),
+            tool_call: false,
+            reasoning: false,
+            limit: None,
+        },
+    );
+    n.model_load.insert(
+        "test-model".into(),
+        ModelLoad {
+            id: "test-model".into(),
+            in_flight,
+            queue_depth,
+        },
+    );
+}
+
+/// Build a gateway state over two mock neurons (no poller; we seed state).
+async fn two_neuron_fleet(endpoint_a: &str, endpoint_b: &str) -> Arc<CortexState> {
+    let config = GatewayConfig {
+        gateway: GatewaySettings {
+            listen: "127.0.0.1:0".into(),
+            metrics_listen: "127.0.0.1:0".into(),
+        },
+        eviction: EvictionSettings {
+            strategy: EvictionStrategy::Lru,
+            defrag_after_cycles: 0,
+        },
+        neurons: vec![
+            NeuronEndpoint {
+                name: "node-a".into(),
+                endpoint: endpoint_a.to_string(),
+            },
+            NeuronEndpoint {
+                name: "node-b".into(),
+                endpoint: endpoint_b.to_string(),
+            },
+        ],
+        models_config: "/dev/null".into(),
+        entitlements: Default::default(),
+    };
+    Arc::new(CortexState::from_config(&config))
+}
+
+#[tokio::test]
+async fn routes_to_least_busy_replica() {
+    let neuron_a = common::spawn_mock_neuron().await;
+    let neuron_b = common::spawn_mock_neuron().await;
+    let fleet = two_neuron_fleet(&neuron_a, &neuron_b).await;
+
+    // A is busy (1 running + 3 queued), B is idle.
+    seed_loaded(&fleet, "node-a", 1, 3).await;
+    seed_loaded(&fleet, "node-b", 0, 0).await;
+
+    let route = cortex_gateway::router::resolve(&fleet, "test-model")
+        .await
+        .expect("model is loaded on both nodes");
+    assert_eq!(route.node_name, "node-b", "should pick the idle replica");
+
+    // Flip the load: now B is the busy one.
+    seed_loaded(&fleet, "node-a", 0, 0).await;
+    seed_loaded(&fleet, "node-b", 1, 5).await;
+    let route = cortex_gateway::router::resolve(&fleet, "test-model")
+        .await
+        .expect("still loaded");
+    assert_eq!(route.node_name, "node-a", "should follow the lighter load");
+}
+
+/// Mock neuron whose inference endpoint always returns a #63 backpressure
+/// envelope (503 + Retry-After) — simulating a saturated neuron.
+async fn spawn_busy_neuron() -> String {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let base_url = format!("http://{addr}");
+    let inference_url = base_url.clone();
+    let app = axum::Router::new()
+        .route(
+            "/models/{model_id}/endpoint",
+            get(move |Path(_): Path<String>| {
+                let url = inference_url.clone();
+                async move { Json(json!({ "url": url })) }
+            }),
+        )
+        .route(
+            "/v1/chat/completions",
+            post(|| async {
+                let body = json!({"error": {
+                    "message": "model is busy (admission queue full); retry shortly",
+                    "type": "rate_limit_error",
+                    "code": "rate_limit_exceeded",
+                    "param": null
+                }});
+                (
+                    StatusCode::SERVICE_UNAVAILABLE,
+                    [(header::RETRY_AFTER, "6")],
+                    Json(body),
+                )
+                    .into_response()
+            }),
+        );
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    base_url
+}
+
+#[tokio::test]
+async fn neuron_backpressure_is_propagated_intact() {
+    // A saturated neuron's 503 + Retry-After + envelope must reach the client
+    // verbatim — not unwrapped, remapped, or stripped (#55 / #63).
+    let neuron = spawn_busy_neuron().await;
+    let fleet = two_neuron_fleet(&neuron, &neuron).await;
+    seed_loaded(&fleet, "node-a", 1, 8).await;
+
+    let app = cortex_gateway::build_app(Arc::clone(&fleet));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+
+    let resp = reqwest::Client::new()
+        .post(format!("http://{addr}/v1/chat/completions"))
+        .json(&json!({"model": "test-model", "messages": [{"role": "user", "content": "hi"}]}))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), reqwest::StatusCode::SERVICE_UNAVAILABLE);
+    assert_eq!(
+        resp.headers()
+            .get(reqwest::header::RETRY_AFTER)
+            .and_then(|v| v.to_str().ok()),
+        Some("6"),
+        "Retry-After must survive the proxy"
+    );
+    let body: Value = resp.json().await.unwrap();
+    assert_eq!(body["error"]["code"], "rate_limit_exceeded");
+}
+
+#[tokio::test]
+async fn ties_break_deterministically_by_name() {
+    let neuron_a = common::spawn_mock_neuron().await;
+    let neuron_b = common::spawn_mock_neuron().await;
+    let fleet = two_neuron_fleet(&neuron_a, &neuron_b).await;
+
+    // Equal load on both → stable pick (lowest node name).
+    seed_loaded(&fleet, "node-a", 0, 0).await;
+    seed_loaded(&fleet, "node-b", 0, 0).await;
+
+    let route = cortex_gateway::router::resolve(&fleet, "test-model")
+        .await
+        .expect("loaded");
+    assert_eq!(route.node_name, "node-a", "ties break by name");
+}

crates/cortex-gateway/tests/prompt_prevalidation.rs

@@ -0,0 +1,174 @@
+//! Fail-fast prompt pre-validation + advisory client hints (#56).
+//!
+//! cortex refuses a prompt that already exceeds the model's advertised
+//! context window before dispatching to neuron — the same #60
+//! `context_length_exceeded` envelope neuron would emit, just earlier — and
+//! attaches an advisory `X-Helexa-Advice` header for fingerprinted clients.
+
+use axum::Json;
+use axum::extract::Path;
+use axum::routing::{get, post};
+use cortex_core::config::{
+    EvictionSettings, EvictionStrategy, GatewayConfig, GatewaySettings, NeuronEndpoint,
+};
+use cortex_core::harness::ModelLimit;
+use cortex_core::node::{ModelEntry, ModelStatus};
+use cortex_gateway::state::CortexState;
+use serde_json::{Value, json};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use tokio::net::TcpListener;
+
+/// Mock neuron with a hit counter, so a test can prove a request was (or
+/// wasn't) dispatched past the gateway's pre-validation.
+async fn spawn_counting_neuron() -> (String, Arc<AtomicU64>) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let base_url = format!("http://{addr}");
+    let inference_url = base_url.clone();
+    let hits = Arc::new(AtomicU64::new(0));
+    let sink = Arc::clone(&hits);
+    let app = axum::Router::new()
+        .route(
+            "/models/{model_id}/endpoint",
+            get(move |Path(_): Path<String>| {
+                let url = inference_url.clone();
+                async move { Json(json!({ "url": url })) }
+            }),
+        )
+        .route(
+            "/v1/chat/completions",
+            post(move || {
+                let sink = Arc::clone(&sink);
+                async move {
+                    sink.fetch_add(1, Ordering::SeqCst);
+                    Json(json!({
+                        "id": "c", "object": "chat.completion", "created": 1_700_000_000_u64,
+                        "model": "test-model",
+                        "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
+                        "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4}
+                    }))
+                }
+            }),
+        );
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    (base_url, hits)
+}
+
+/// Gateway over one neuron with `test-model` loaded and a tiny advertised
+/// context window (so a modest prompt overflows it).
+async fn spawn_gateway(neuron: &str, context: usize) -> String {
+    let config = GatewayConfig {
+        gateway: GatewaySettings {
+            listen: "127.0.0.1:0".into(),
+            metrics_listen: "127.0.0.1:0".into(),
+        },
+        eviction: EvictionSettings {
+            strategy: EvictionStrategy::Lru,
+            defrag_after_cycles: 0,
+        },
+        neurons: vec![NeuronEndpoint {
+            name: "mock-node".into(),
+            endpoint: neuron.to_string(),
+        }],
+        models_config: "/dev/null".into(),
+        entitlements: Default::default(),
+    };
+    let fleet = Arc::new(CortexState::from_config(&config));
+    {
+        let mut nodes = fleet.nodes.write().await;
+        let n = nodes.get_mut("mock-node").unwrap();
+        n.healthy = true;
+        n.models.insert(
+            "test-model".into(),
+            ModelEntry {
+                id: "test-model".into(),
+                status: ModelStatus::Loaded,
+                last_accessed: None,
+                vram_estimate_mb: Some(8000),
+                capabilities: Vec::new(),
+                tool_call: false,
+                reasoning: false,
+                limit: Some(ModelLimit {
+                    context,
+                    input: None,
+                    output: 16,
+                }),
+            },
+        );
+    }
+    let app = cortex_gateway::build_app(Arc::clone(&fleet));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+    format!("http://{addr}")
+}
+
+#[tokio::test]
+async fn over_long_prompt_is_rejected_before_dispatch() {
+    let (neuron, hits) = spawn_counting_neuron().await;
+    let gateway = spawn_gateway(&neuron, 50).await; // tiny 50-token window
+
+    // ~1200 chars → ~300 est tokens, well over 50.
+    let big = "word ".repeat(240);
+    let resp = reqwest::Client::new()
+        .post(format!("{gateway}/v1/chat/completions"))
+        .header("user-agent", "litellm/1.0")
+        .json(&json!({"model": "test-model", "messages": [{"role": "user", "content": big}]}))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST);
+    // Advisory hint for the fingerprinted client (header only, never body).
+    assert!(
+        resp.headers().get("x-helexa-advice").is_some(),
+        "litellm should get advice"
+    );
+    let body: Value = resp.json().await.unwrap();
+    assert_eq!(body["error"]["code"], "context_length_exceeded");
+    assert_eq!(body["error"]["max"], 50);
+    // Refused at the edge — neuron never saw it.
+    assert_eq!(hits.load(Ordering::SeqCst), 0);
+}
+
+#[tokio::test]
+async fn within_context_passes_through() {
+    let (neuron, hits) = spawn_counting_neuron().await;
+    let gateway = spawn_gateway(&neuron, 4096).await;
+
+    let resp = reqwest::Client::new()
+        .post(format!("{gateway}/v1/chat/completions"))
+        .json(&json!({"model": "test-model", "messages": [{"role": "user", "content": "hi"}]}))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), reqwest::StatusCode::OK);
+    let _ = resp.bytes().await.unwrap();
+    assert_eq!(hits.load(Ordering::SeqCst), 1, "served by neuron");
+}
+
+#[tokio::test]
+async fn unknown_client_gets_no_advice_header() {
+    let (neuron, _hits) = spawn_counting_neuron().await;
+    let gateway = spawn_gateway(&neuron, 50).await;
+
+    let big = "word ".repeat(240);
+    let resp = reqwest::Client::new()
+        .post(format!("{gateway}/v1/chat/completions"))
+        // no/unknown User-Agent → no advice, but still a clean 400
+        .json(&json!({"model": "test-model", "messages": [{"role": "user", "content": big}]}))
+        .send()
+        .await
+        .unwrap();
+
+    assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST);
+    assert!(resp.headers().get("x-helexa-advice").is_none());
+    let body: Value = resp.json().await.unwrap();
+    assert_eq!(body["error"]["code"], "context_length_exceeded");
+}

crates/neuron/src/api.rs

@@ -13,6 +13,7 @@ use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::{IntoResponse, Json};
 use axum::routing::{get, post};
 use cortex_core::discovery::{DiscoveryResponse, HealthResponse};
+use cortex_core::entitlements::{HEADER_ACCOUNT_ID, HEADER_KEY_ID};
 use cortex_core::harness::ModelSpec;
 use cortex_core::openai::{ChatCompletionRequest, MessageContent};
 use cortex_core::responses::{ResponsesRequest, ResponsesUsage};
@@ -71,6 +72,12 @@ async fn health_handler(State(state): State<Arc<NeuronState>>) -> Json<HealthRes
     // know about activation lifecycle.
     let mut snapshot = state.health_cache.snapshot().await;
     snapshot.activation = state.activation.snapshot().await;
+    // Per-model admission load (#53) — read live from the candle harness so
+    // cortex's load-aware router (#55) can spread traffic and propagate
+    // backpressure. Absent when no candle harness is present.
+    if let Some(candle) = &state.candle {
+        snapshot.models = candle.load_snapshot().await;
+    }
     Json(snapshot)
 }
 
@@ -228,6 +235,17 @@ fn default_enable_thinking(req: &mut ChatCompletionRequest, include_thinking: bo
     }
 }
 
+/// The request's principal for fair-share admission (#54), reconstructed
+/// from the internal headers cortex stamps (#49). cortex strips any
+/// client-supplied copy and asserts the authoritative value, so over the
+/// trusted WireGuard link these are safe to key fair-share on. `None` for an
+/// unauthenticated/direct request — exempt from the per-principal cap.
+fn principal_key(headers: &axum::http::HeaderMap) -> Option<String> {
+    let account = headers.get(HEADER_ACCOUNT_ID)?.to_str().ok()?;
+    let key = headers.get(HEADER_KEY_ID)?.to_str().ok()?;
+    Some(format!("{account}/{key}"))
+}
+
 /// OpenAI-compatible chat completions. Dispatches to streaming SSE when
 /// `stream: true` is set on the request; otherwise returns a single
 /// `ChatCompletionResponse`.
@@ -271,8 +289,14 @@ async fn chat_completions(
     // true`) keep reasoning on.
     default_enable_thinking(&mut req, include_thinking);
 
+    // Fair-share admission principal (#54), from cortex's stamped headers.
+    let principal = principal_key(&headers);
+
     if req.stream.unwrap_or(false) {
-        match candle.chat_completion_stream_with(req, chat_config).await {
+        match candle
+            .chat_completion_stream_with(req, chat_config, principal)
+            .await
+        {
             Ok(rx) => {
                 // Each chunk → one SSE `data: {json}` line. After the
                 // channel closes, append the OpenAI [DONE] terminator.
@@ -289,7 +313,7 @@ async fn chat_completions(
             Err(e) => inference_error_response(e),
         }
     } else {
-        match candle.chat_completion(req).await {
+        match candle.chat_completion(req, principal).await {
             Ok(resp) => Json(resp).into_response(),
             Err(e) => inference_error_response(e),
         }
@@ -302,6 +326,7 @@ async fn chat_completions(
 /// event stream into the Responses event family.
 async fn responses(
     State(state): State<Arc<NeuronState>>,
+    headers: axum::http::HeaderMap,
     Json(req): Json<ResponsesRequest>,
 ) -> impl IntoResponse {
     let Some(candle) = state.candle.as_ref().map(Arc::clone) else {
@@ -336,9 +361,12 @@ async fn responses(
     };
     chat_req.stream = Some(stream_requested);
 
+    // Fair-share admission principal (#54), from cortex's stamped headers.
+    let principal = principal_key(&headers);
+
     if stream_requested {
         match candle
-            .responses_stream(chat_req, response_id, message_item_id)
+            .responses_stream(chat_req, response_id, message_item_id, principal)
             .await
         {
             Ok(rx) => {
@@ -362,7 +390,7 @@ async fn responses(
         // and translate the result. We don't currently re-tokenise
         // to compute usage; the harness returns it via the chat
         // response and we pass it through.
-        match candle.chat_completion(chat_req).await {
+        match candle.chat_completion(chat_req, principal).await {
             Ok(chat_resp) => {
                 // Extract the assistant text (chat completions
                 // always emits one choice on the candle path).
@@ -486,8 +514,8 @@ fn inference_error_response(err: InferenceError) -> axum::response::Response {
             "template_render_failed",
             format!("chat template could not render this request: {detail}"),
         ),
-        // Admission control refused (#53): a fast, retryable "busy" signal.
-        // 503 (service busy) + Retry-After; opencode/AI SDK back off.
+        // Admission control refused on load (#53): a fast, retryable "busy"
+        // signal. 503 (service busy) + Retry-After; opencode/AI SDK back off.
         InferenceError::Overloaded { retry_after_secs } => OpenAiError::new(
             503,
             "rate_limit_error",
@@ -495,6 +523,15 @@ fn inference_error_response(err: InferenceError) -> axum::response::Response {
             "model is busy (admission queue full); retry shortly",
         )
         .with_retry_after(retry_after_secs),
+        // Per-principal fair-share cap (#54): 429 rate_limit_exceeded +
+        // Retry-After — the caller is sending too many concurrent requests.
+        InferenceError::PerPrincipalLimit { retry_after_secs } => OpenAiError::new(
+            429,
+            "rate_limit_error",
+            "rate_limit_exceeded",
+            "too many concurrent requests for this key; retry shortly",
+        )
+        .with_retry_after(retry_after_secs),
         InferenceError::Other(e) => OpenAiError::without_code(500, "api_error", format!("{e:#}")),
     };
     envelope_response(env)

crates/neuron/src/config.rs

@@ -113,6 +113,13 @@ pub struct AdmissionConfig {
     /// honest signal).
     #[serde(default = "default_admission_max_wait_secs")]
     pub max_wait_secs: u64,
+    /// Per-principal fair-share cap (#54): max in-flight + queued requests
+    /// for any single principal (resolved from the `x-helexa-*` headers
+    /// cortex stamps), so one client can't monopolize the queue while others
+    /// wait. Over-cap → `429 rate_limit_exceeded` + `Retry-After`. `0`
+    /// disables the cap; anonymous requests are always exempt.
+    #[serde(default = "default_admission_max_per_principal")]
+    pub max_per_principal: usize,
 }
 
 impl Default for AdmissionConfig {
@@ -121,6 +128,7 @@ impl Default for AdmissionConfig {
             max_in_flight: default_admission_max_in_flight(),
             max_queue_depth: default_admission_max_queue_depth(),
             max_wait_secs: default_admission_max_wait_secs(),
+            max_per_principal: default_admission_max_per_principal(),
         }
     }
 }
@@ -137,6 +145,10 @@ fn default_admission_max_wait_secs() -> u64 {
     30
 }
 
+fn default_admission_max_per_principal() -> usize {
+    2
+}
+
 /// `[harness.candle.prefix_cache]` settings.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PrefixCacheConfig {

crates/neuron/src/harness/admission.rs

@@ -21,40 +21,56 @@
 //! `/health` can read live load without contending with inference.
 
 use crate::config::AdmissionConfig;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 
-/// Why admission was refused. Both map to the #63 backpressure envelope
-/// (`429`/`503` + `rate_limit_exceeded` + `Retry-After`); they differ only
-/// in cause, for logging.
+/// Why admission was refused. All map to the #63 backpressure envelope
+/// (`rate_limit_exceeded` + `Retry-After`); they differ in cause (and HTTP
+/// status — load → `503`, per-principal → `429`).
 #[derive(Debug, Clone, Copy)]
 pub enum AdmissionRejection {
-    /// The bounded wait queue was already full.
+    /// The bounded wait queue was already full (server-side load).
     QueueFull { retry_after_secs: u64 },
     /// A queue slot was taken but the in-flight slot didn't free within
-    /// `max_wait`.
+    /// `max_wait` (server-side load).
     Timeout { retry_after_secs: u64 },
+    /// This principal already has `max_per_principal` requests in flight or
+    /// queued (#54 fair-share) — one principal can't monopolize the model.
+    PrincipalCap { retry_after_secs: u64 },
 }
 
 impl AdmissionRejection {
     pub fn retry_after_secs(&self) -> u64 {
         match self {
             AdmissionRejection::QueueFull { retry_after_secs }
-            | AdmissionRejection::Timeout { retry_after_secs } => *retry_after_secs,
+            | AdmissionRejection::Timeout { retry_after_secs }
+            | AdmissionRejection::PrincipalCap { retry_after_secs } => *retry_after_secs,
         }
     }
 }
 
-/// Bounded batch-1 scheduler for one loaded model.
+/// Admission accounting, mutated under a brief lock (never held across an
+/// await). `pending` is queued + in-flight overall; `per_principal` is the
+/// same count keyed by principal for fair-share (#54).
+#[derive(Default, Debug)]
+struct AdmissionState {
+    pending: usize,
+    per_principal: HashMap<String, usize>,
+}
+
+/// Bounded batch-1 scheduler for one loaded model, with per-principal
+/// fair-share.
 pub struct AdmissionController {
     /// In-flight slots — `max_in_flight` permits (1 for batch-1).
     slots: Arc<Semaphore>,
-    /// Queued + in-flight count, for fast rejection and load reporting.
-    pending: Arc<AtomicUsize>,
-    /// `max_in_flight + max_queue_depth` — the rejection threshold.
+    /// Queued + in-flight accounting (overall + per principal).
+    state: Arc<Mutex<AdmissionState>>,
+    /// `max_in_flight + max_queue_depth` — the overall rejection threshold.
     max_pending: usize,
+    /// Max in-flight + queued for any single principal (#54). `0` disables.
+    max_per_principal: usize,
     max_in_flight: usize,
     max_wait: Duration,
 }
@@ -65,42 +81,69 @@ impl AdmissionController {
         let max_in_flight = cfg.max_in_flight.max(1);
         Self {
             slots: Arc::new(Semaphore::new(max_in_flight)),
-            pending: Arc::new(AtomicUsize::new(0)),
+            state: Arc::new(Mutex::new(AdmissionState::default())),
             max_pending: max_in_flight + cfg.max_queue_depth,
+            max_per_principal: cfg.max_per_principal,
             max_in_flight,
             max_wait: Duration::from_secs(cfg.max_wait_secs),
         }
     }
 
-    /// Admit a request: reserve a queue slot (fast-rejecting if full), then
-    /// wait up to `max_wait` for an in-flight slot. The returned permit must
-    /// be held for the request's lifetime; dropping it frees both slots.
-    pub async fn enter(&self) -> Result<AdmissionPermit, AdmissionRejection> {
-        // Reserve a pending slot up front so concurrent callers can't all
-        // slip past the threshold check. Roll back if we're over capacity.
-        let prev = self.pending.fetch_add(1, Ordering::AcqRel);
-        if prev >= self.max_pending {
-            self.pending.fetch_sub(1, Ordering::AcqRel);
-            return Err(AdmissionRejection::QueueFull {
-                retry_after_secs: self.retry_hint(),
-            });
+    /// Admit a request for `principal` (`None` = anonymous, exempt from the
+    /// per-principal cap). Reserves a queue slot — fast-rejecting if the
+    /// overall queue is full or the principal is over its fair-share cap —
+    /// then waits up to `max_wait` for an in-flight slot. The returned permit
+    /// must be held for the request's lifetime; dropping it frees the slots.
+    pub async fn enter(
+        &self,
+        principal: Option<&str>,
+    ) -> Result<AdmissionPermit, AdmissionRejection> {
+        // Decision + reservation under one brief lock so concurrent callers
+        // can't both slip past the thresholds. No await is held here.
+        {
+            let mut st = self.state.lock().expect("admission state poisoned");
+            if st.pending >= self.max_pending {
+                return Err(AdmissionRejection::QueueFull {
+                    retry_after_secs: self.retry_hint(st.pending),
+                });
+            }
+            if let Some(p) = principal
+                && self.max_per_principal > 0
+                && st.per_principal.get(p).copied().unwrap_or(0) >= self.max_per_principal
+            {
+                return Err(AdmissionRejection::PrincipalCap {
+                    retry_after_secs: self.retry_hint(st.pending),
+                });
+            }
+            st.pending += 1;
+            if let Some(p) = principal {
+                *st.per_principal.entry(p.to_string()).or_insert(0) += 1;
+            }
         }
 
         match tokio::time::timeout(self.max_wait, Arc::clone(&self.slots).acquire_owned()).await {
             Ok(Ok(permit)) => Ok(AdmissionPermit {
                 _permit: permit,
-                pending: Arc::clone(&self.pending),
+                state: Arc::clone(&self.state),
+                principal: principal.map(str::to_string),
             }),
             // Semaphore is never closed; treat a closed/elapsed wait the same.
             Ok(Err(_)) | Err(_) => {
-                self.pending.fetch_sub(1, Ordering::AcqRel);
+                self.release(principal);
                 Err(AdmissionRejection::Timeout {
-                    retry_after_secs: self.retry_hint(),
+                    retry_after_secs: self.retry_hint(self.max_pending),
                 })
             }
         }
     }
 
+    /// Roll back a reserved-but-not-admitted slot (wait timed out).
+    fn release(&self, principal: Option<&str>) {
+        let mut st = self.state.lock().expect("admission state poisoned");
+        st.pending = st.pending.saturating_sub(1);
+        decrement_principal(&mut st.per_principal, principal);
+    }
+
     /// Requests currently running (holding an in-flight slot).
     pub fn in_flight(&self) -> usize {
         self.max_in_flight
@@ -109,29 +152,45 @@ impl AdmissionController {
 
     /// Requests waiting for an in-flight slot.
     pub fn queue_depth(&self) -> usize {
-        self.pending
-            .load(Ordering::Acquire)
-            .saturating_sub(self.in_flight())
+        let pending = self.state.lock().expect("admission state poisoned").pending;
+        pending.saturating_sub(self.in_flight())
     }
 
     /// Rough `Retry-After`: scale with how backed-up the model is, clamped to
     /// a sane band. Without per-request timing this is a heuristic, but it
     /// gives well-behaved clients (opencode/AI SDK) a sensible backoff.
-    fn retry_hint(&self) -> u64 {
-        ((self.queue_depth() as u64 + 1) * 2).clamp(1, 120)
+    fn retry_hint(&self, pending: usize) -> u64 {
+        let queued = pending.saturating_sub(self.max_in_flight) as u64;
+        ((queued + 1) * 2).clamp(1, 120)
     }
 }
 
-/// Held for a request's lifetime; frees the in-flight + queue slot on drop.
+/// Decrement (and prune at zero) a principal's outstanding count.
+fn decrement_principal(map: &mut HashMap<String, usize>, principal: Option<&str>) {
+    if let Some(p) = principal
+        && let Some(count) = map.get_mut(p)
+    {
+        *count -= 1;
+        if *count == 0 {
+            map.remove(p);
+        }
+    }
+}
+
+/// Held for a request's lifetime; frees the in-flight + queue slot (and the
+/// principal's fair-share slot) on drop.
 #[derive(Debug)]
 pub struct AdmissionPermit {
     _permit: OwnedSemaphorePermit,
-    pending: Arc<AtomicUsize>,
+    state: Arc<Mutex<AdmissionState>>,
+    principal: Option<String>,
 }
 
 impl Drop for AdmissionPermit {
     fn drop(&mut self) {
-        self.pending.fetch_sub(1, Ordering::AcqRel);
+        let mut st = self.state.lock().expect("admission state poisoned");
+        st.pending = st.pending.saturating_sub(1);
+        decrement_principal(&mut st.per_principal, self.principal.as_deref());
     }
 }
 
@@ -139,11 +198,14 @@ impl Drop for AdmissionPermit {
 mod tests {
     use super::*;
 
+    /// Config with the per-principal cap disabled (0) — most tests exercise
+    /// the overall queue with anonymous (`None`) callers.
     fn cfg(max_in_flight: usize, max_queue_depth: usize, max_wait_secs: u64) -> AdmissionConfig {
         AdmissionConfig {
             max_in_flight,
             max_queue_depth,
             max_wait_secs,
+            max_per_principal: 0,
         }
     }
 
@@ -151,7 +213,7 @@ mod tests {
     async fn admits_up_to_in_flight_and_reports_load() {
         let ctrl = AdmissionController::new(&cfg(1, 4, 30));
         assert_eq!(ctrl.in_flight(), 0);
-        let p = ctrl.enter().await.expect("first admits");
+        let p = ctrl.enter(None).await.expect("first admits");
         assert_eq!(ctrl.in_flight(), 1);
         assert_eq!(ctrl.queue_depth(), 0);
         drop(p);
@@ -162,17 +224,17 @@ mod tests {
     async fn rejects_when_queue_full() {
         // 1 in-flight + 1 queue slot = capacity 2; the 3rd is refused fast.
         let ctrl = Arc::new(AdmissionController::new(&cfg(1, 1, 30)));
-        let _running = ctrl.enter().await.expect("admit running");
+        let _running = ctrl.enter(None).await.expect("admit running");
 
         // Fill the single queue slot with a waiter that parks on the semaphore.
         let ctrl2 = Arc::clone(&ctrl);
-        let waiter = tokio::spawn(async move { ctrl2.enter().await.map(|p| drop(p)) });
+        let waiter = tokio::spawn(async move { ctrl2.enter(None).await.map(|p| drop(p)) });
         // Give the waiter a moment to occupy the queue slot.
         tokio::time::sleep(Duration::from_millis(50)).await;
         assert_eq!(ctrl.queue_depth(), 1);
 
         // Queue full → immediate QueueFull with a Retry-After hint.
-        match ctrl.enter().await {
+        match ctrl.enter(None).await {
             Err(AdmissionRejection::QueueFull { retry_after_secs }) => {
                 assert!(retry_after_secs >= 1)
             }
@@ -190,13 +252,47 @@ mod tests {
         // request can't even queue, so it's QueueFull, not Timeout. Use a
         // queue of 1 and a tiny max_wait to exercise the timeout path.
         let ctrl = Arc::new(AdmissionController::new(&cfg(1, 1, 0)));
-        let _running = ctrl.enter().await.expect("admit running");
+        let _running = ctrl.enter(None).await.expect("admit running");
         // max_wait 0 → the queued request times out almost immediately.
-        match ctrl.enter().await {
+        match ctrl.enter(None).await {
             Err(AdmissionRejection::Timeout { .. }) => {}
             other => panic!("expected Timeout, got {other:?}"),
         }
         // The timed-out request released its queue slot.
         assert_eq!(ctrl.queue_depth(), 0);
     }
+
+    #[tokio::test]
+    async fn per_principal_cap_protects_other_principals() {
+        // Generous overall queue, but each principal capped at 1 in-flight+
+        // queued. Principal A holds the running slot; A's second request is
+        // refused (PrincipalCap) rather than occupying the queue, so B's
+        // single request still gets a queue slot and proceeds.
+        let cfg = AdmissionConfig {
+            max_in_flight: 1,
+            max_queue_depth: 8,
+            max_wait_secs: 30,
+            max_per_principal: 1,
+        };
+        let ctrl = Arc::new(AdmissionController::new(&cfg));
+
+        let _a1 = ctrl.enter(Some("acct-a/key-a")).await.expect("A admits");
+
+        // A is over its fair-share cap → fast PrincipalCap, no queue slot taken.
+        match ctrl.enter(Some("acct-a/key-a")).await {
+            Err(AdmissionRejection::PrincipalCap { retry_after_secs }) => {
+                assert!(retry_after_secs >= 1)
+            }
+            other => panic!("expected PrincipalCap, got {other:?}"),
+        }
+
+        // B (a different principal) is admitted to the queue and proceeds
+        // once A releases — it was never stuck behind A's backlog.
+        let ctrl2 = Arc::clone(&ctrl);
+        let b = tokio::spawn(async move { ctrl2.enter(Some("acct-b/key-b")).await.map(drop) });
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        assert_eq!(ctrl.queue_depth(), 1, "B is queued, not rejected");
+        drop(_a1);
+        b.await.unwrap().expect("B is served after A releases");
+    }
 }

crates/neuron/src/harness/candle.rs

@@ -149,6 +149,16 @@ impl LoadedHandle {
         }
     }
 
+    /// Current admission load (#53): `(in_flight, queue_depth)`. Lock-free,
+    /// so `/health` can read it without contending with inference.
+    pub fn load(&self) -> (usize, usize) {
+        match self {
+            LoadedHandle::Single(m) => (m.admission.in_flight(), m.admission.queue_depth()),
+            #[cfg(feature = "cuda")]
+            LoadedHandle::Tp(m) => (m.admission.in_flight(), m.admission.queue_depth()),
+        }
+    }
+
     /// Modalities the loaded model supports. Stage B7 (single-GPU) +
     /// TP-vision (#12) — both single-GPU and TP loads advertise
     /// `"vision"` when a replicated vision tower materialised.
@@ -2018,6 +2028,7 @@ impl CandleHarness {
     pub async fn chat_completion(
         &self,
         request: ChatCompletionRequest,
+        principal: Option<String>,
     ) -> Result<ChatCompletionResponse, InferenceError> {
         let handle = {
             let models = self.models.read().await;
@@ -2042,7 +2053,7 @@ impl CandleHarness {
             LoadedHandle::Single(m) => m,
             #[cfg(feature = "cuda")]
             LoadedHandle::Tp(m) => {
-                return self.chat_completion_tp(m, request).await;
+                return self.chat_completion_tp(m, request, principal).await;
             }
         };
 
@@ -2076,7 +2087,7 @@ impl CandleHarness {
         // The permit is held for the whole request (released on drop).
         let _admit = loaded
             .admission
-            .enter()
+            .enter(principal.as_deref())
             .await
             .map_err(InferenceError::from)?;
 
@@ -2399,9 +2410,14 @@ impl CandleHarness {
     pub async fn chat_completion_stream(
         &self,
         request: ChatCompletionRequest,
+        principal: Option<String>,
     ) -> Result<mpsc::Receiver<ChatCompletionChunk>, InferenceError> {
-        self.chat_completion_stream_with(request, wire_chat::ChatProjectionConfig::default())
-            .await
+        self.chat_completion_stream_with(
+            request,
+            wire_chat::ChatProjectionConfig::default(),
+            principal,
+        )
+        .await
     }
 
     /// Same as [`Self::chat_completion_stream`] but lets the caller
@@ -2412,8 +2428,9 @@ impl CandleHarness {
         &self,
         request: ChatCompletionRequest,
         mut config: wire_chat::ChatProjectionConfig,
+        principal: Option<String>,
     ) -> Result<mpsc::Receiver<ChatCompletionChunk>, InferenceError> {
-        let stream = self.inference_stream(request).await?;
+        let stream = self.inference_stream(request, principal).await?;
         // Fill in the model's reasoning markers if the caller
         // didn't pre-populate them — they're a property of the
         // loaded model (which the HTTP handler doesn't reach into
@@ -2440,9 +2457,10 @@ impl CandleHarness {
         request: ChatCompletionRequest,
         response_id: String,
         message_item_id: String,
+        principal: Option<String>,
     ) -> Result<mpsc::Receiver<crate::wire::openai_responses::ResponseStreamFrame>, InferenceError>
     {
-        let stream = self.inference_stream(request).await?;
+        let stream = self.inference_stream(request, principal).await?;
         let meta = crate::wire::openai_responses::ResponseMeta {
             response_id,
             created_at: stream.created,
@@ -2463,6 +2481,7 @@ impl CandleHarness {
     async fn inference_stream(
         &self,
         request: ChatCompletionRequest,
+        principal: Option<String>,
     ) -> Result<InferenceStream, InferenceError> {
         let handle = {
             let models = self.models.read().await;
@@ -2487,7 +2506,7 @@ impl CandleHarness {
             LoadedHandle::Single(m) => m,
             #[cfg(feature = "cuda")]
             LoadedHandle::Tp(m) => {
-                return self.inference_tp_stream(m, request).await;
+                return self.inference_tp_stream(m, request, principal).await;
             }
         };
 
@@ -2636,7 +2655,7 @@ impl CandleHarness {
         // into the inference task and is held until it completes.
         let admit = loaded
             .admission
-            .enter()
+            .enter(principal.as_deref())
             .await
             .map_err(InferenceError::from)?;
 
@@ -2811,6 +2830,24 @@ pub struct InferenceStream {
 /// Auto-recovery (#17) — rebuild a poisoned model's device context
 /// automatically instead of leaving it bricked until a human reloads.
 impl CandleHarness {
+    /// Per-model admission load for `GET /health` (#53): in-flight + queued
+    /// counts for every resident model. Lock-free per-model reads, so this
+    /// only briefly holds the registry read lock to enumerate handles.
+    pub async fn load_snapshot(&self) -> Vec<cortex_core::discovery::ModelLoad> {
+        let models = self.models.read().await;
+        models
+            .values()
+            .map(|handle| {
+                let (in_flight, queue_depth) = handle.load();
+                cortex_core::discovery::ModelLoad {
+                    id: handle.model_id().to_string(),
+                    in_flight,
+                    queue_depth,
+                }
+            })
+            .collect()
+    }
+
     /// True while `model_id` is being auto-recovered (its slot is briefly
     /// absent from the registry during the reload).
     pub async fn is_recovering(&self, model_id: &str) -> bool {
@@ -3472,6 +3509,7 @@ impl CandleHarness {
         &self,
         tp: Arc<TpLoadedModel>,
         request: ChatCompletionRequest,
+        principal: Option<String>,
     ) -> Result<ChatCompletionResponse, InferenceError> {
         // Tag every line of this request with a short req_id so a
         // grep over journalctl reconstructs one request even when
@@ -3508,7 +3546,8 @@ impl CandleHarness {
         }
 
         let tp_for_marker = Arc::clone(&tp);
-        let handle = tokio::spawn(chat_completion_tp_inner(tp, request).instrument(span.clone()));
+        let handle =
+            tokio::spawn(chat_completion_tp_inner(tp, request, principal).instrument(span.clone()));
         match handle.await {
             Ok(Ok(resp)) => Ok(resp),
             Ok(Err(e)) => {
@@ -3579,6 +3618,7 @@ impl CandleHarness {
         &self,
         tp: Arc<TpLoadedModel>,
         request: ChatCompletionRequest,
+        principal: Option<String>,
     ) -> Result<InferenceStream, InferenceError> {
         if tp.poisoned.load(Ordering::Acquire) {
             return Err(self.trigger_recovery(&request.model).await);
@@ -3726,7 +3766,11 @@ impl CandleHarness {
 
         // Admission control (#53): refuse before opening the stream; the
         // permit moves into the orchestration task and is held for its life.
-        let admit = tp.admission.enter().await.map_err(InferenceError::from)?;
+        let admit = tp
+            .admission
+            .enter(principal.as_deref())
+            .await
+            .map_err(InferenceError::from)?;
 
         let tool_schemas = build_tool_schemas(&request);
         let tp_for_task = Arc::clone(&tp);
@@ -4235,6 +4279,7 @@ impl CandleHarness {
 async fn chat_completion_tp_inner(
     tp: Arc<TpLoadedModel>,
     request: ChatCompletionRequest,
+    principal: Option<String>,
 ) -> Result<ChatCompletionResponse, InferenceError> {
     let req_start = std::time::Instant::now();
     let model_id = request.model.clone();
@@ -4325,7 +4370,11 @@ async fn chat_completion_tp_inner(
 
     // Admission control (#53): bounded queue + fast reject before joining
     // the pool-lock wait. Held for the whole request (released on drop).
-    let _admit = tp.admission.enter().await.map_err(InferenceError::from)?;
+    let _admit = tp
+        .admission
+        .enter(principal.as_deref())
+        .await
+        .map_err(InferenceError::from)?;
 
     // Acquire the pool lock for the duration of the request. After
     // Phase 3 the leader's TpLeaderModel lives in the device worker
@@ -4869,19 +4918,31 @@ pub enum InferenceError {
     /// failure mode that hid several client-compat bugs. Maps to 422.
     #[error("chat template could not render this request: {detail}")]
     TemplateRenderFailed { detail: String },
-    /// Admission control (#53) refused the request: the model's bounded
-    /// queue is full or the wait elapsed. Maps to `429 rate_limit_exceeded`
-    /// + `Retry-After` — a fast, retryable "busy" signal, not a stall.
+    /// Admission control (#53) refused on load: the model's bounded queue is
+    /// full or the wait elapsed. Maps to `503 rate_limit_exceeded` +
+    /// `Retry-After` — a fast, retryable "busy" signal, not a stall.
     #[error("model is busy; retry after {retry_after_secs}s")]
     Overloaded { retry_after_secs: u64 },
+    /// Per-principal fair-share cap (#54) exceeded: this principal already
+    /// has its max requests in flight/queued. Maps to `429
+    /// rate_limit_exceeded` + `Retry-After`; a well-behaved client self-paces.
+    #[error("per-principal in-flight limit reached; retry after {retry_after_secs}s")]
+    PerPrincipalLimit { retry_after_secs: u64 },
     #[error(transparent)]
     Other(#[from] anyhow::Error),
 }
 
 impl From<super::admission::AdmissionRejection> for InferenceError {
     fn from(rejection: super::admission::AdmissionRejection) -> Self {
-        InferenceError::Overloaded {
-            retry_after_secs: rejection.retry_after_secs(),
+        use super::admission::AdmissionRejection;
+        match rejection {
+            AdmissionRejection::QueueFull { retry_after_secs }
+            | AdmissionRejection::Timeout { retry_after_secs } => {
+                InferenceError::Overloaded { retry_after_secs }
+            }
+            AdmissionRejection::PrincipalCap { retry_after_secs } => {
+                InferenceError::PerPrincipalLimit { retry_after_secs }
+            }
         }
     }
 }

crates/neuron/src/health.rs

@@ -30,6 +30,9 @@ impl HealthCache {
                 // direct read from the cache stays a well-typed
                 // HealthResponse on the wire.
                 activation: Default::default(),
+                // Per-model admission load is overlaid by the api handler
+                // from the candle harness (#53); the cache doesn't own it.
+                models: Vec::new(),
             }),
             has_gpus: RwLock::new(false),
         }

crates/neuron/tests/api.rs

@@ -114,6 +114,12 @@ async fn test_health_endpoint() {
 
     let body: serde_json::Value = resp.json().await.unwrap();
     assert_eq!(body["uptime_secs"], 0);
+    // Per-model admission load (#53) is always present, even with no models
+    // loaded (empty array) — cortex's load-aware router (#55) relies on it.
+    assert!(
+        body["models"].is_array(),
+        "/health must expose a models load array"
+    );
 }
 
 #[tokio::test]