diff --git a/crates/mesh-llm-host-runtime/src/network/openai/forwarded_request.rs b/crates/mesh-llm-host-runtime/src/network/openai/forwarded_request.rs index 65838d308c..d08188ce7a 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/forwarded_request.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/forwarded_request.rs @@ -162,15 +162,26 @@ pub(super) fn finalize_forwarded_request( Ok(forwarded) } -/// Rebuild a request for a remote peer without forwarding ingress credentials. +/// Rebuild a request for a remote peer without forwarding ingress credentials +/// or the routing headers the ROUTING node already consumed. +/// +/// `x-mesh-target` / `x-mesh-exclude` express the client's routing intent to +/// the node the client is talking to; they must never survive onto a +/// peer-to-peer hop. Without this, a peer that re-enters `route_request` +/// still carrying the original target/exclude can re-route the request +/// itself -- and since nothing here carries a hop count or visited-peer +/// list, a stale advertisement can bounce a request between peers with no +/// bound (CodeRabbit / erlich, PR #1671 round 2). pub(super) fn prepare_peer_forwarded_request(raw: &[u8]) -> Result> { - const CALLER_CREDENTIAL_HEADERS: &[&str] = &[ + const OMITTED_ON_PEER_FORWARD: &[&str] = &[ "authorization", "proxy-authorization", "x-api-key", "api-key", + super::request_parse::MESH_TARGET_HEADER, + super::request_parse::MESH_EXCLUDE_HEADER, ]; - finalize_forwarded_request(raw, false, None, None, CALLER_CREDENTIAL_HEADERS) + finalize_forwarded_request(raw, false, None, None, OMITTED_ON_PEER_FORWARD) } #[cfg(test)] @@ -198,6 +209,36 @@ mod tests { assert_eq!(forwarded, expected); } + /// Regression (CodeRabbit / erlich, PR #1671 round 2): routing headers + /// must never survive a peer-to-peer hop, or a peer that re-enters + /// `route_request` can re-route the request again with no hop bound. + #[test] + fn peer_forwarding_strips_mesh_routing_headers() { + let body = b"{}"; + let raw = format!( + "POST /v1/chat/completions HTTP/1.1\r\nHost: localhost\r\nx-mesh-target: aabbcc\r\nx-mesh-exclude: ddeeff,001122\r\nX-Keep: yes\r\nContent-Length: {}\r\n\r\n", + body.len() + ); + let mut raw = raw.into_bytes(); + raw.extend_from_slice(body); + + let forwarded = prepare_peer_forwarded_request(&raw).unwrap(); + let text = String::from_utf8_lossy(&forwarded); + + assert!( + !text.to_ascii_lowercase().contains("x-mesh-target"), + "x-mesh-target must not reach the peer: {text}" + ); + assert!( + !text.to_ascii_lowercase().contains("x-mesh-exclude"), + "x-mesh-exclude must not reach the peer: {text}" + ); + assert!( + text.contains("X-Keep: yes"), + "unrelated headers must survive: {text}" + ); + } + #[test] fn peer_forwarding_rejects_incomplete_headers() { let raw = b"GET /v1/models HTTP/1.1\r\nAuthorization: Bearer caller-secret\r\n"; diff --git a/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs b/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs index dc9e8266b6..3d97e9a286 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/ingress.rs @@ -9,7 +9,7 @@ use crate::network::openai::client_stream::ClientStream; use crate::network::openai::transport as proxy; use crate::network::router; use crate::plugin::openai_exchange::{ - OpenAiExchangeChannel, OpenAiExchangeDispatchPath, OpenAiExchangeEnvelope, + ClientNonceSource, OpenAiExchangeChannel, OpenAiExchangeDispatchPath, OpenAiExchangeEnvelope, }; use mesh_llm_events::audit::{audit_events, emit_audit}; use mesh_llm_events::{OutputEvent, emit_event}; @@ -27,6 +27,25 @@ fn plugin_route_status(outcome: &proxy::RouteDispatchOutcome) -> Option { } } +/// Map a `RemoteMesh` forwarded nonce's origin marker (see +/// [`proxy::BufferedHttpRequest::capsule_nonce_headers`]) to the tri-state a +/// plugin uses to judge trust. An origin marker present means THIS frontend +/// minted the nonce (the client sent none, so a fallback was generated at +/// ingress); its absence means the value came from the client unchanged. +/// `None` exactly when there is no nonce to report at all. +fn remote_mesh_nonce_source( + nonce: &Option, + nonce_origin: &Option, +) -> Option { + nonce.as_ref().map(|_| { + if nonce_origin.is_some() { + ClientNonceSource::SidecarGeneratedFallback + } else { + ClientNonceSource::ClientSupplied + } + }) +} + enum AutoRouteResolution { Continue { effective_model: Option, @@ -52,12 +71,16 @@ struct AutoRouteDecision { required_tokens: Option, } +/// Convert a [`proxy::RouteDispatchOutcome`] to the [`crate::logging::TerminalOutcome`] +/// variant used by structured terminal-event logging at request boundaries. fn terminal_outcome_for_dispatch( outcome: proxy::RouteDispatchOutcome, ) -> crate::logging::TerminalOutcome { outcome.terminal_outcome() } +/// Return `true` when the dispatch outcome carries a 2xx status code, +/// indicating the model inference request was served successfully. fn model_access_succeeded(outcome: proxy::RouteDispatchOutcome) -> bool { matches!( outcome, @@ -69,6 +92,10 @@ fn model_access_succeeded(outcome: proxy::RouteDispatchOutcome) -> bool { ) } +/// Lift a raw I/O result from a response-send helper into a typed +/// [`proxy::RouteDispatchOutcome`]: `Ok(())` becomes `Responded(status_code)`; +/// any write error becomes `Dropped` (the connection closed before the client +/// could read the response). fn response_outcome(status_code: u16, result: std::io::Result<()>) -> proxy::RouteDispatchOutcome { match result { Ok(()) => proxy::RouteDispatchOutcome::Responded(status_code), @@ -513,31 +540,124 @@ fn warn_pipeline_fallback(strong_name: &str) { tracing::warn!("pipeline: falling back to direct proxy for {strong_name}"); } +/// Route a request whose model is not being served from local candidates +/// (either genuinely absent locally, or forced away from local by +/// `x-mesh-target`/`x-mesh-exclude` -- see `route_request`), trying remote +/// mesh peers, then local-unavailable/plugin fallbacks, then 404. +#[allow(clippy::too_many_arguments)] async fn route_missing_local_model( tcp_stream: ClientStream, request: &proxy::BufferedHttpRequest, ctx: &IngressRouteContext<'_>, model_name: &str, + target: Option, + excluded: &[iroh::EndpointId], required_tokens: Option, route_observer: OpenAiRouteObserver<'_>, ) -> proxy::RouteDispatchOutcome { - // Try remote mesh first. - if let Some(mesh_targets) = remote_mesh_targets(ctx, model_name).await { - return proxy::route_model_request( - ctx.node.clone(), + // An explicit self-target can never be found by `resolve_remote_mesh_route` + // below -- it only searches OTHER peers' advertised hosts -- so it always + // failed closed with a spurious 409 even when this node serves the model + // itself via a plugin (ndizazzo P2a, PR #1671 round 2). The caller only + // reaches this function for a self-target once local HOST-served + // candidates already came up empty (see `route_request`), so what's left + // to check here is plugin-served availability. + if target == Some(ctx.node.id()) { + return route_self_targeted_model( tcp_stream, - &mesh_targets, - model_name, request, - proxy::RouteModelRequestContext { - required_tokens, - affinity: ctx.affinity, - route_observer, - }, + ctx, + model_name, + excluded, + route_observer, ) .await; } + // Try remote mesh first. + match resolve_remote_mesh_route(ctx, model_name, target, excluded).await { + RemoteMeshRoute::TargetUnavailable { target_hex } => { + // Fail closed: never substitute another peer for an explicitly + // named `x-mesh-target` that doesn't (or no longer) serve this + // model -- that would silently defeat the live-twin check the + // header exists for. + return response_outcome( + 409, + proxy::send_error_observed( + tcp_stream, + 409, + &format!( + "x-mesh-target '{target_hex}' does not serve model '{model_name}' -- refusing to fall back to another peer" + ), + route_observer, + ) + .await, + ); + } + RemoteMeshRoute::Targets(mesh_targets) => { + // This node is routing the exchange to a peer, not serving it -- + // publish the same effective/terminal pair try_route_plugin_model + // already does for its own dispatch below, with `RemoteMesh` in + // place of `RawProxy`, so a plugin on the ROUTING node can observe + // this exchange too (previously it observed nothing at all for a + // routed exchange). No marker exists on this path yet -- a peer's + // `X-Capsule-Id` response header is not read back here -- so + // capsule_id stays absent, same as the plugin-served terminal + // event just below. + let exchange_id = uuid::Uuid::new_v4().to_string(); + // The client-contributed capsule nonce, already stabilized (and, + // if the client sent none, minted) by `finalize_forwarded_request` + // at ingress -- read back off the already-buffered request rather + // than minted here: a second fallback minted on THIS node would + // not match whatever was already stamped into the request this + // node forwards to the peer byte-for-byte, breaking "same nonce + // both sides." + let (forwarded_nonce, nonce_origin) = request.capsule_nonce_headers(); + let nonce_source = remote_mesh_nonce_source(&forwarded_nonce, &nonce_origin); + if let Some(plugin_manager) = ctx.plugin_manager { + plugin_manager + .publish(&OpenAiExchangeEnvelope::effective_remote_mesh( + exchange_id.clone(), + model_name, + forwarded_nonce.clone(), + nonce_source, + )) + .await; + } + // Only echoed when the client asked for a specific peer via + // `x-mesh-target` -- absent headers must produce today's + // response byte-for-byte, with no `x-mesh-served-by` added. + let served_by_hex = target.map(|id| hex::encode(id.as_bytes())); + let outcome = proxy::route_model_request( + ctx.node.clone(), + tcp_stream, + &mesh_targets, + model_name, + request, + proxy::RouteModelRequestContext { + required_tokens, + affinity: ctx.affinity, + route_observer, + served_by_header: served_by_hex.as_deref(), + }, + ) + .await; + if let Some(plugin_manager) = ctx.plugin_manager { + plugin_manager + .publish(&OpenAiExchangeEnvelope::terminal_remote_mesh( + exchange_id, + model_name, + plugin_route_status(&outcome), + forwarded_nonce, + nonce_source, + )) + .await; + } + return outcome; + } + RemoteMeshRoute::NoRemoteHost => {} + } + // Check if the model is known locally but unavailable // (e.g., loading/draining/failed with all-None candidates). Return 503 for these cases so // clients can retry; return 404 only when the model truly doesn't exist anywhere. @@ -553,6 +673,30 @@ async fn route_missing_local_model( ); } + // `x-mesh-exclude` naming this node must block LOCAL PLUGIN fallback too + // -- otherwise the client's explicit "not this node" is honored for + // host-served candidates (via `mesh_headers_force_remote`, which is why + // execution reached this function at all) but silently ignored the + // moment the model turns out to be plugin-served instead, and the + // excluded node serves the request anyway (ndizazzo P2b, PR #1671 + // round 2). Fail closed with 409, matching `x-mesh-target`'s contract, + // rather than the generic "not found anywhere" 404. + if excluded.contains(&ctx.node.id()) { + return response_outcome( + 409, + proxy::send_error_observed( + tcp_stream, + 409, + &format!( + "this node ({}) is excluded via x-mesh-exclude and no other candidate serves model '{model_name}'", + hex::encode(ctx.node.id().as_bytes()) + ), + route_observer, + ) + .await, + ); + } + // Try plugin dispatch (admission-checked inside). if ctx.plugin_manager.is_some() { return try_route_plugin_model(ctx, tcp_stream, request, model_name, route_observer).await; @@ -572,6 +716,82 @@ async fn route_missing_local_model( ) } +/// Resolve an explicit `x-mesh-target` naming THIS node against local +/// availability -- host-served or plugin-served -- instead of asking +/// `resolve_remote_mesh_route` to find "self" among other peers, which it +/// never will (ndizazzo P2a, PR #1671 round 2). The caller guarantees this +/// node does not currently host-serve `model_name` (see +/// `route_missing_local_model`), so only plugin-served availability remains +/// to check. Fails closed with 409 -- the same contract `x-mesh-target` uses +/// for a peer that turns out not to serve the model -- when this node +/// doesn't serve it either, or when `x-mesh-exclude` names this node too (a +/// self-target contradicted by a self-exclude). +async fn route_self_targeted_model( + tcp_stream: ClientStream, + request: &proxy::BufferedHttpRequest, + ctx: &IngressRouteContext<'_>, + model_name: &str, + excluded: &[iroh::EndpointId], + route_observer: OpenAiRouteObserver<'_>, +) -> proxy::RouteDispatchOutcome { + let self_id = ctx.node.id(); + if let Some(plugin_manager) = ctx.plugin_manager { + match plugin_manager + .inference_endpoint_for_model(model_name) + .await + { + Err(error) => { + // Transient resolution failure — degrade gracefully with 503, + // not 409. A 409 would mis-state the reason: the peer did not + // refuse to serve the model; we failed to ask it. The caller + // (client) should retry; the peer is not at fault. + tracing::warn!( + %error, + "route_self_targeted_model: failed to resolve plugin endpoint for '{model_name}'" + ); + return response_outcome( + 503, + proxy::send_503_observed( + tcp_stream, + &format!( + "plugin endpoint for model '{model_name}' unavailable (resolution error)" + ), + route_observer, + ) + .await, + ); + } + Ok(Some(_endpoint)) => { + // Self is the target and serves the model via plugin. + if !excluded.contains(&self_id) { + return try_route_plugin_model( + ctx, + tcp_stream, + request, + model_name, + route_observer, + ) + .await; + } + } + Ok(None) => {} + } + } + response_outcome( + 409, + proxy::send_error_observed( + tcp_stream, + 409, + &format!( + "x-mesh-target '{}' does not serve model '{model_name}' -- refusing to fall back to another peer", + hex::encode(self_id.as_bytes()) + ), + route_observer, + ) + .await, + ) +} + /// Check whether the model is known locally but currently unavailable — all local candidates are None. fn has_local_unavailable_candidates(targets: &election::ModelTargets, model_name: &str) -> bool { let cands = targets.candidates(model_name); @@ -581,13 +801,60 @@ fn has_local_unavailable_candidates(targets: &election::ModelTargets, model_name .all(|t| matches!(t, election::InferenceTarget::None)) } -async fn remote_mesh_targets( +/// Outcome of resolving `x-mesh-target` / `x-mesh-exclude` against the +/// current `hosts_for_model()` candidate set for one request. +enum RemoteMeshRoute { + /// Route normally (either the ordinary multi-candidate remote-mesh pool, + /// or -- when `x-mesh-target` was given -- a forced single-candidate pool + /// containing only that peer). + Targets(election::ModelTargets), + /// `x-mesh-target` named a peer that isn't in the (possibly + /// `x-mesh-exclude`-filtered) candidate set for this model. The caller + /// must fail closed, never substitute a different peer. + TargetUnavailable { target_hex: String }, + /// No remote host serves this model (after exclusion) -- fall through to + /// local/plugin/404 handling exactly as when neither header is present. + NoRemoteHost, +} + +/// Build the [`RemoteMeshRoute`] for `model_name` given the optional +/// `x-mesh-target` and the `x-mesh-exclude` set. Queries `hosts_for_model` +/// from the node to get the live remote candidate list, applies the exclude +/// filter, and either forces a single-peer target pool (when `target` is +/// `Some`) or returns the full filtered pool. Returns +/// [`RemoteMeshRoute::TargetUnavailable`] when an explicit `target` is not +/// in the candidate set -- the caller must fail closed, never substitute. +async fn resolve_remote_mesh_route( ctx: &IngressRouteContext<'_>, model_name: &str, -) -> Option { - let remote_hosts = ctx.node.hosts_for_model(model_name).await; + target: Option, + excluded: &[iroh::EndpointId], +) -> RemoteMeshRoute { + let remote_hosts: Vec = ctx + .node + .hosts_for_model(model_name) + .await + .into_iter() + .filter(|id| !excluded.contains(id)) + .collect(); + + if let Some(target) = target { + return if remote_hosts.contains(&target) { + let mut mesh_targets = ctx.targets.clone(); + mesh_targets.targets.insert( + model_name.to_string(), + vec![election::InferenceTarget::Remote(target)], + ); + RemoteMeshRoute::Targets(mesh_targets) + } else { + RemoteMeshRoute::TargetUnavailable { + target_hex: hex::encode(target.as_bytes()), + } + }; + } + if remote_hosts.is_empty() { - return None; + return RemoteMeshRoute::NoRemoteHost; } let mut mesh_targets = ctx.targets.clone(); mesh_targets.targets.insert( @@ -597,9 +864,97 @@ async fn remote_mesh_targets( .map(election::InferenceTarget::Remote) .collect(), ); - Some(mesh_targets) + RemoteMeshRoute::Targets(mesh_targets) } +/// Decode a hex-encoded 32-byte `EndpointId` from a header value string. +/// Leading/trailing whitespace is trimmed before decoding. Returns `None` on +/// any decode or length error — the caller is responsible for turning `None` +/// into an appropriate rejection (400 or 409). +fn parse_endpoint_id_hex(value: &str) -> Option { + let bytes = hex::decode(value.trim()).ok()?; + let bytes: [u8; 32] = bytes.as_slice().try_into().ok()?; + iroh::EndpointId::from_bytes(&bytes).ok() +} + +/// Parse the (possibly repeated) `x-mesh-target` header values. Zero values +/// is a no-op; exactly one must decode as an `EndpointId`; more than one is +/// ambiguous and rejected rather than silently picking one. +fn parse_mesh_target_header(values: &[String]) -> Result, String> { + match values { + [] => Ok(None), + [only] => parse_endpoint_id_hex(only) + .map(Some) + .ok_or_else(|| format!("invalid x-mesh-target value '{only}'")), + _ => Err("multiple x-mesh-target headers are ambiguous".to_string()), + } +} + +/// Parse the `x-mesh-exclude` header value(s), each a comma-separated list of +/// `EndpointId`s. Any unparseable OR empty entry rejects the whole request +/// rather than silently dropping an exclusion the client asked for. +fn parse_mesh_exclude_header(values: &[String]) -> Result, String> { + let mut excluded = Vec::new(); + for value in values { + for part in value.split(',') { + let part = part.trim(); + if part.is_empty() { + return Err("x-mesh-exclude contains an empty entry".to_string()); + } + let id = parse_endpoint_id_hex(part) + .ok_or_else(|| format!("invalid x-mesh-exclude value '{part}'"))?; + excluded.push(id); + } + } + Ok(excluded) +} + +/// Parse and validate the raw `x-mesh-target` / `x-mesh-exclude` header +/// values off `request` in one place, so every caller enforces them the same +/// way before making a routing decision. +fn parse_mesh_routing_headers( + request: &proxy::BufferedHttpRequest, +) -> Result<(Option, Vec), String> { + let (target_values, exclude_values) = request.mesh_routing_header_values()?; + let target = parse_mesh_target_header(&target_values)?; + let excluded = parse_mesh_exclude_header(&exclude_values)?; + Ok((target, excluded)) +} + +/// Whether `request` carries a non-trivial `x-mesh-target`/`x-mesh-exclude` +/// ask. Malformed headers are already rejected with 400 wherever this +/// request is validated before reaching MoA, so a parse failure here (which +/// should not happen at this point) is treated as "no ask" rather than +/// re-rejecting. Shared by both MoA call sites (`try_handle_moa_intercept` +/// and the passive mesh-request path in `transport.rs`) so "does this +/// request want routing headers honored" is answered the same way in both +/// places. +pub(crate) fn mesh_routing_headers_requested(request: &proxy::BufferedHttpRequest) -> bool { + matches!( + parse_mesh_routing_headers(request), + Ok((target, excluded)) if target.is_some() || !excluded.is_empty() + ) +} + +/// Whether `x-mesh-target`/`x-mesh-exclude` must force this request away from +/// local candidates: an exclude naming this node, or a target naming some +/// other peer. A target naming this node is not forcing -- it is allowed to +/// serve locally (checked by the ordinary local-candidate path) and only +/// changes whether `x-mesh-served-by` is echoed. +fn mesh_headers_force_remote( + self_id: iroh::EndpointId, + target: Option, + excluded: &[iroh::EndpointId], +) -> bool { + excluded.contains(&self_id) || target.is_some_and(|id| id != self_id) +} + +/// Dispatch an inference request to an out-of-process plugin endpoint that +/// serves `model_name` (path 2 / `RawProxy` dispatch). Checks activity policy +/// admission first; resolves the plugin endpoint via `plugin_manager`; emits +/// an effective and a terminal OpenAI exchange event to the plugin bus so any +/// observing plugin can track the full lifecycle of each exchange, including +/// cases where the backend returns an error or the connection is dropped. async fn try_route_plugin_model( ctx: &IngressRouteContext<'_>, mut tcp_stream: ClientStream, @@ -740,6 +1095,8 @@ async fn try_route_plugin_model( } } +/// Route a model-bearing or model-less request: local candidates unless +/// `x-mesh-target`/`x-mesh-exclude` force this node out of consideration. async fn route_request( tcp_stream: ClientStream, request: &mut proxy::BufferedHttpRequest, @@ -750,20 +1107,42 @@ async fn route_request( ) -> proxy::RouteDispatchOutcome { prepare_cache_routing_body(request, effective_model); if let Some(model_name) = effective_model { - // Model explicitly requested. Check local candidates first. - if !has_available_candidates(ctx.targets, model_name) { + // Model explicitly requested. Parse and enforce `x-mesh-target` / + // `x-mesh-exclude` BEFORE the local-candidate check below -- a + // targeted or excluded request must never be silently served from + // local candidates without ever consulting these headers. + let (target, excluded) = match parse_mesh_routing_headers(request) { + Ok(parsed) => parsed, + Err(message) => { + return response_outcome( + 400, + proxy::send_400_observed(tcp_stream, &message, route_observer).await, + ); + } + }; + let self_id = ctx.node.id(); + let forced_remote = mesh_headers_force_remote(self_id, target, &excluded); + + if forced_remote || !has_available_candidates(ctx.targets, model_name) { return route_missing_local_model( tcp_stream, request, ctx, model_name, + target, + &excluded, required_tokens, route_observer, ) .await; } - // Local candidates available — route normally. + // Local candidates available and this node was neither excluded nor + // targeted at a different peer — route locally. Echo + // `x-mesh-served-by` only when the client explicitly named this node. + let served_by_hex = target + .filter(|id| *id == self_id) + .map(|id| hex::encode(id.as_bytes())); proxy::route_model_request( ctx.node.clone(), tcp_stream, @@ -774,6 +1153,7 @@ async fn route_request( required_tokens, affinity: ctx.affinity, route_observer, + served_by_header: served_by_hex.as_deref(), }, ) .await @@ -796,6 +1176,11 @@ async fn route_request( } } +/// Ensure the request body is parsed as JSON when an effective model is known, +/// so cache routing and provider-confirmed local receipts use a stable prefix +/// key even when only one eligible target exists. No-op for tokenize requests +/// (which use a different body shape). The body is already bounded and +/// buffered at ingress; parsing here does not change the forwarded bytes. fn prepare_cache_routing_body( request: &mut proxy::BufferedHttpRequest, effective_model: Option<&str>, @@ -809,6 +1194,11 @@ fn prepare_cache_routing_body( } } +/// Run model-name resolution for a `model: "auto"` request, returning an +/// [`AutoRouteDecision`] on success or `Err(())` when no served model can +/// satisfy the media inputs in the request body. Side-effects: enables auto +/// route hooks on the buffered request if a model is selected, and records +/// the model hit on the node for activity tracking. async fn prepare_auto_route_decision( request: &mut proxy::BufferedHttpRequest, ctx: &IngressRouteContext<'_>, @@ -844,6 +1234,10 @@ async fn prepare_auto_route_decision( } } +/// Respond with 422 when the auto-route resolver determines no served model +/// can satisfy the media inputs (e.g., audio/image) in the request. The +/// response body names the constraint so the client knows to re-send without +/// the unsupported media. async fn send_media_unsupported( tcp_stream: ClientStream, route_observer: OpenAiRouteObserver<'_>, @@ -860,6 +1254,9 @@ async fn send_media_unsupported( ) } +/// Build the sorted list of model names visible to the `/v1/models` endpoint: +/// the remote-mesh callable set from `targets` merged with `local_models` +/// (plugin-served and locally-launched models) with duplicates removed. fn callable_models_with_local_served( targets: &election::ModelTargets, local_models: Vec, @@ -912,6 +1309,110 @@ fn pipeline_route_model<'a>( use_pipeline.then_some(routing_model).flatten() } +/// Which non-`route_request` dispatch kind, if any, this request will take — +/// each one bypasses `route_request`'s own header enforcement entirely, so a +/// caller holding `x-mesh-target`/`x-mesh-exclude` must reject before +/// dispatching into one of them rather than silently ignoring the headers. +/// `None` means the request will reach `route_request`'s ordinary +/// model-bearing path, where the headers are enforced against real +/// candidates. Mirrors the same checks `try_pipeline_route` makes, evaluated +/// one step earlier and side-effect free. +/// +/// `model: "mesh"` is deliberately NOT covered here: whether MoA can honor +/// these headers depends on whether committee routing actually convenes for +/// this request, which is not known until `try_handle_moa_intercept` calls +/// into `moa_gateway::try_handle_moa` -- that function degrades `model: +/// "mesh"` to a single concrete model when no committee can be formed, and a +/// degraded request can and should honor the headers downstream. Rejecting +/// here, before MoA ever runs, would reject requests MoA was about to make +/// routable. See `moa_gateway::try_handle_moa`'s `mesh_routing_requested` +/// parameter for where that rejection now lives. +fn mesh_routing_unsupported_dispatch_kind( + request: &proxy::BufferedHttpRequest, + decision: &AutoRouteDecision, + routing_model: Option<&str>, +) -> Option<&'static str> { + if pipeline_route_model(request, decision, routing_model).is_some() { + Some("pipeline") + } else if decision.effective_model.is_none() { + Some("no model specified") + } else { + None + } +} + +/// Enforce `x-mesh-target`/`x-mesh-exclude` before ANY dispatch decision is +/// made -- not just the ordinary model-bearing path inside `route_request`. +/// Pipeline and the model-less fallback both run before `route_request` ever +/// parses the headers, so a malformed header on one of those paths used to +/// reach whatever status that dispatch kind happens to fail with instead of +/// 400, and a *valid* header was silently ignored rather than being honored +/// or explicitly rejected (CodeRabbit + ndizazzo P1, PR #1671 round 2 -- +/// "where are these headers enforced" is answered once, here, rather than +/// once per dispatch kind). Returns the stream to continue dispatch when the +/// headers are absent or compatible with where this request is headed; +/// returns the terminal outcome (already written to the stream) otherwise. +/// `route_request` re-parses the same immutable request headers for its own +/// model-bearing enforcement; that second parse is cheap and keeps this +/// function from having to thread the parsed values through. +/// +/// Malformed-header validation (400, via `parse_mesh_routing_headers` below) +/// still runs unconditionally here, ahead of MoA -- only the *unsupported +/// dispatch kind* rejection (409) excludes `model: "mesh"`; see +/// `mesh_routing_unsupported_dispatch_kind`. +async fn enforce_mesh_routing_headers_before_dispatch( + tcp_stream: ClientStream, + request: &proxy::BufferedHttpRequest, + decision: &AutoRouteDecision, + routing_model: Option<&str>, + route_observer: OpenAiRouteObserver<'_>, +) -> Result { + let (target, excluded) = match parse_mesh_routing_headers(request) { + Ok(parsed) => parsed, + Err(message) => { + return Err(response_outcome( + 400, + proxy::send_400_observed(tcp_stream, &message, route_observer).await, + )); + } + }; + if target.is_none() && excluded.is_empty() { + return Ok(tcp_stream); + } + match mesh_routing_unsupported_dispatch_kind(request, decision, routing_model) { + Some(kind) => Err(response_outcome( + 409, + proxy::send_409_observed( + tcp_stream, + &format!( + "routing headers present but this dispatch does not support them ({kind})" + ), + route_observer, + ) + .await, + )), + None => Ok(tcp_stream), + } +} + +/// Apply activity-policy admission to an inference request that has already +/// passed the control-plane gate. Returns the stream to continue dispatch, or +/// an outcome (already written to the stream) when admission is denied. +async fn admit_buffered_api_request( + tcp_stream: ClientStream, + ctx: &ProxyConnectionContext<'_>, + ingress_type: crate::runtime::IngressType, + lifecycle: &OpenAiLifecycleAttachment, +) -> Result { + check_activity_admission( + tcp_stream, + &ctx.route.node.activity_policy_guard, + ingress_type, + lifecycle.route_observer(), + ) + .await +} + async fn try_pipeline_route( tcp_stream: &mut ClientStream, request: &mut proxy::BufferedHttpRequest, @@ -952,6 +1453,11 @@ async fn try_handle_moa_intercept( if decision.effective_model.as_deref() != Some(moa::VIRTUAL_MODEL_NAME) { return MoaInterceptResult::NotMoa(tcp_stream); } + // Whether the caller asked `x-mesh-target`/`x-mesh-exclude` to be + // honored. `try_handle_moa` rejects with 409 only if it decides to + // actually convene a committee -- a degrade to a single concrete model + // continues and the headers are honored downstream by `route_request`. + let mesh_routing_requested = mesh_routing_headers_requested(request); // `try_handle_moa` self-gates on the model name and consumes the // stream when it accepts. The outer gate above guarantees the gate // matches, so the inner call always returns `None` here — the stream @@ -964,10 +1470,11 @@ async fn try_handle_moa_intercept( tcp_stream, request, decision.effective_model.as_deref(), - super::moa_gateway::MoaRoutingContext { + crate::network::openai::moa_gateway::MoaRoutingContext { targets: Some(ctx.route.targets), required_tokens: decision.required_tokens, affinity: ctx.route.affinity, + mesh_routing_requested, }, route_observer, ) @@ -1020,6 +1527,13 @@ async fn try_handle_moa_intercept( } } +/// Drive one buffered OpenAI-shaped request through every ingress stage in +/// order: control-plane/admission gates, auto-route resolution, mesh routing +/// header enforcement, MoA, pipeline, and finally `route_request`'s ordinary +/// dispatch. Each stage either hands the stream to the next one or writes a +/// terminal response and returns -- this function owns the single terminal +/// lifecycle event for the request no matter which stage ends it. +#[allow(clippy::cognitive_complexity)] async fn handle_buffered_api_request( tcp_stream: ClientStream, mut request: proxy::BufferedHttpRequest, @@ -1072,10 +1586,31 @@ async fn handle_buffered_api_request( proxy::rewrite_public_model_alias(&mut request, &callable, &descriptors); // Admission applies to inference work after control-path rejection. - let tcp_stream = match check_activity_admission( + let tcp_stream = + match admit_buffered_api_request(tcp_stream, &ctx, ingress_type, &lifecycle).await { + Ok(stream) => stream, + Err(outcome) => { + lifecycle.terminal(terminal_outcome_for_dispatch(outcome)); + return; + } + }; + + let decision = match prepare_auto_route_decision(&mut request, &ctx.route, &descriptors).await { + Ok(decision) => decision, + Err(()) => { + let outcome = send_media_unsupported(tcp_stream, lifecycle.route_observer()).await; + lifecycle.terminal(terminal_outcome_for_dispatch(outcome)); + return; + } + }; + + let mut routing_model = decision.effective_model.clone(); + + let tcp_stream = match enforce_mesh_routing_headers_before_dispatch( tcp_stream, - &ctx.route.node.activity_policy_guard, - ingress_type, + &request, + &decision, + routing_model.as_deref(), lifecycle.route_observer(), ) .await @@ -1087,16 +1622,6 @@ async fn handle_buffered_api_request( } }; - let decision = match prepare_auto_route_decision(&mut request, &ctx.route, &descriptors).await { - Ok(decision) => decision, - Err(()) => { - let outcome = send_media_unsupported(tcp_stream, lifecycle.route_observer()).await; - lifecycle.terminal(terminal_outcome_for_dispatch(outcome)); - return; - } - }; - - let mut routing_model = decision.effective_model.clone(); let tcp_stream = match try_handle_moa_intercept( tcp_stream, &mut request, diff --git a/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs b/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs index 688e00c587..19161962f4 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/ingress_tests/tests.rs @@ -314,6 +314,207 @@ fn moa_degraded_model_is_consumed_by_pipeline_dispatch() { ); } +fn dispatch_kind_request( + model: Option<&str>, + response_adapter: proxy::ResponseAdapter, +) -> proxy::BufferedHttpRequest { + proxy::BufferedHttpRequest { + raw: Vec::new(), + method: "POST".to_owned(), + path: "/v1/chat/completions".to_owned(), + client_path: "/v1/chat/completions".to_owned(), + request_id: RequestId::default(), + body_json: None, + body_json_attempted: false, + body_bytes: None, + body_len_bytes: 0, + completion_tokens: None, + stream: None, + model_name: model.map(str::to_owned), + request_object_request_ids: Vec::new(), + response_adapter, + correlation_id: None, + } +} + +/// Regression (CodeRabbit, PR #1671 round 2 follow-up): `model: "mesh"` must +/// NOT be flagged here -- whether the routing headers can be honored depends +/// on whether `try_handle_moa` actually convenes a committee, which this +/// pure, side-effect-free function cannot know. Rejecting eagerly here (the +/// old behavior) blocked requests that MoA was about to degrade to a +/// concrete model and route with the headers honored. See +/// `moa_gateway::mesh_routing_tests` for the two outcomes this now defers +/// to (`try_handle_moa` rejects only when a committee actually convenes). +#[test] +fn mesh_routing_unsupported_dispatch_kind_does_not_flag_moa_dispatch() { + let request = + dispatch_kind_request(Some(moa::VIRTUAL_MODEL_NAME), proxy::ResponseAdapter::None); + let decision = AutoRouteDecision { + effective_model: Some(moa::VIRTUAL_MODEL_NAME.to_owned()), + classification: None, + required_tokens: None, + }; + assert_eq!( + mesh_routing_unsupported_dispatch_kind( + &request, + &decision, + decision.effective_model.as_deref() + ), + None + ); +} + +/// Regression (CodeRabbit + ndizazzo P1, PR #1671 round 2): same gap for +/// pipeline dispatch, which also runs before `route_request`. +#[test] +fn mesh_routing_unsupported_dispatch_kind_flags_pipeline_dispatch() { + use crate::network::router::{Category, Classification, Complexity}; + + let request = dispatch_kind_request( + Some("local/only-model:Q4_K_M"), + proxy::ResponseAdapter::None, + ); + let decision = AutoRouteDecision { + effective_model: Some("local/only-model:Q4_K_M".to_owned()), + classification: Some(Classification { + category: Category::Code, + complexity: Complexity::Deep, + needs_tools: true, + has_media_inputs: false, + }), + required_tokens: None, + }; + assert_eq!( + mesh_routing_unsupported_dispatch_kind( + &request, + &decision, + decision.effective_model.as_deref() + ), + Some("pipeline") + ); +} + +/// Regression (CodeRabbit, PR #1671 round 1/2): the model-less fallback +/// (no `model` in the request at all) also bypasses `route_request`'s +/// model-bearing branch and must not silently ignore a routing header. +#[test] +fn mesh_routing_unsupported_dispatch_kind_flags_model_less_dispatch() { + let request = dispatch_kind_request(None, proxy::ResponseAdapter::None); + let decision = AutoRouteDecision { + effective_model: None, + classification: None, + required_tokens: None, + }; + assert_eq!( + mesh_routing_unsupported_dispatch_kind(&request, &decision, None), + Some("no model specified") + ); +} + +/// The ordinary model-bearing path (no MoA, no pipeline, a real model) must +/// remain unaffected -- `route_request` enforces the headers itself there. +#[test] +fn mesh_routing_unsupported_dispatch_kind_is_none_for_ordinary_dispatch() { + let request = dispatch_kind_request( + Some("local/only-model:Q4_K_M"), + proxy::ResponseAdapter::None, + ); + let decision = AutoRouteDecision { + effective_model: Some("local/only-model:Q4_K_M".to_owned()), + classification: None, + required_tokens: None, + }; + assert_eq!( + mesh_routing_unsupported_dispatch_kind( + &request, + &decision, + decision.effective_model.as_deref() + ), + None + ); +} + +/// Regression (ndizazzo, PR #1671 -- "the model-less bypass remains"): +/// `enforce_mesh_routing_headers_before_dispatch` calls +/// `parse_mesh_routing_headers` unconditionally, before it ever asks whether +/// this dispatch kind supports the headers -- so a malformed `x-mesh-target` +/// must 400 even on a request that carries no `model` at all. The unit tests +/// above exercise `mesh_routing_unsupported_dispatch_kind` and the header +/// parsers in isolation; this test proves the composition end to end through +/// the actual gate function, on the wire. +#[tokio::test] +async fn enforce_mesh_routing_headers_before_dispatch_rejects_malformed_header_without_model() { + use tokio::io::AsyncReadExt; + + let body = serde_json::json!({ "messages": [{ "role": "user", "content": "hi" }] }); + let body_bytes = serde_json::to_vec(&body).expect("serialize body"); + let mut raw = format!( + "POST /v1/chat/completions HTTP/1.1\r\nHost: t\r\nContent-Type: application/json\r\n\ + x-mesh-target: not-a-valid-endpoint-id\r\nContent-Length: {}\r\n\r\n", + body_bytes.len() + ) + .into_bytes(); + raw.extend_from_slice(&body_bytes); + let request = proxy::BufferedHttpRequest { + raw, + method: "POST".to_owned(), + path: "/v1/chat/completions".to_owned(), + client_path: "/v1/chat/completions".to_owned(), + request_id: RequestId::default(), + body_json: None, + body_json_attempted: false, + body_bytes: None, + body_len_bytes: body_bytes.len(), + completion_tokens: None, + stream: None, + model_name: None, + request_object_request_ids: Vec::new(), + response_adapter: proxy::ResponseAdapter::None, + correlation_id: None, + }; + let decision = AutoRouteDecision { + effective_model: None, + classification: None, + required_tokens: None, + }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + let client = tokio::net::TcpStream::connect(addr); + let server = async { listener.accept().await.map(|(stream, _)| stream) }; + let (client_side, server_side) = tokio::join!(client, server); + let mut client_side = client_side.expect("connect"); + let tcp_stream: ClientStream = server_side.expect("accept").into(); + + let result = enforce_mesh_routing_headers_before_dispatch( + tcp_stream, + &request, + &decision, + None, + OpenAiRouteObserver::default(), + ) + .await; + + let rejected_with_400 = matches!(result, Err(proxy::RouteDispatchOutcome::Responded(400))); + assert!( + rejected_with_400, + "expected a malformed x-mesh-target to 400 even on a model-less request" + ); + + let mut response = Vec::new(); + client_side + .read_to_end(&mut response) + .await + .expect("read response"); + let response_text = String::from_utf8_lossy(&response); + assert!( + response_text.starts_with("HTTP/1.1 400"), + "expected 400 on the wire, got: {response_text}" + ); +} + // --- Routing behavior tests for model-independent daemon support --- #[test] @@ -909,3 +1110,541 @@ fn disconnect_is_dropped_and_cannot_audit_model_access_as_success() { proxy::RouteDispatchOutcome::Responded(200) )); } + +// --- x-mesh-target / x-mesh-exclude header parsing and fail-closed routing --- + +fn test_endpoint_id(seed: u8) -> iroh::EndpointId { + iroh::EndpointId::from(iroh::SecretKey::from_bytes(&[seed; 32]).public()) +} + +/// A minimal admitted, HTTP-routable peer serving exactly `model` — enough to +/// exercise `hosts_for_model()` without a real gossip round trip. Field-for- +/// field mirrors `moa_gateway::fleet_sim_tests::fleet_peer`'s shape. +fn test_remote_peer(seed: u8, model: &str) -> mesh::PeerInfo { + let id = test_endpoint_id(seed); + mesh::PeerInfo { + id, + addr: iroh::EndpointAddr { + id, + addrs: Default::default(), + }, + mesh_id: None, + mesh_policy_hash: None, + genesis_policy: None, + role: mesh::NodeRole::Host { http_port: 9337 }, + first_joined_mesh_ts: None, + models: vec![model.to_string()], + vram_bytes: 0, + rtt_ms: None, + model_source: None, + admitted: true, + serving_models: vec![model.to_string()], + hosted_models: vec![model.to_string()], + hosted_models_known: true, + available_models: vec![], + requested_models: vec![], + explicit_model_interests: vec![], + last_seen: std::time::Instant::now(), + last_mentioned: std::time::Instant::now(), + version: None, + gpu_name: None, + hostname: None, + is_soc: None, + gpu_vram: None, + gpu_reserved_bytes: None, + memory: None, + gpu_mem_bandwidth_gbps: None, + gpu_compute_tflops_fp32: None, + gpu_compute_tflops_fp16: None, + available_model_metadata: vec![], + experts_summary: None, + available_model_sizes: std::collections::HashMap::new(), + served_model_descriptors: vec![], + served_model_runtime: vec![], + owner_attestation: None, + release_attestation_summary: crate::ReleaseAttestationSummary::default(), + artifact_transfer_supported: false, + stage_protocol_generation_supported: false, + stage_status_list_supported: false, + local_gguf_content_id_supported: false, + advertised_model_throughput: vec![], + cache_affinity: None, + display_rtt: None, + selected_path: None, + propagated_latency: None, + owner_summary: Default::default(), + inference_admission_state: None, + } +} + +fn remote_mesh_test_ctx<'a>( + node: &'a mesh::Node, + targets: &'a election::ModelTargets, + affinity: &'a affinity::AffinityRouter, +) -> IngressRouteContext<'a> { + IngressRouteContext { + node, + targets, + affinity, + plugin_manager: None, + } +} + +#[test] +fn parse_mesh_target_header_absent_is_a_no_op() { + assert_eq!(parse_mesh_target_header(&[]), Ok(None)); +} + +#[test] +fn parse_mesh_target_header_parses_one_valid_value() { + let id = test_endpoint_id(0x42); + let value = hex::encode(id.as_bytes()); + assert_eq!(parse_mesh_target_header(&[value]), Ok(Some(id))); +} + +#[test] +fn parse_mesh_target_header_rejects_malformed_value() { + assert!(parse_mesh_target_header(&["not-a-hex-endpoint-id".to_string()]).is_err()); +} + +#[test] +fn parse_mesh_target_header_rejects_multiple_values_as_ambiguous() { + let value = hex::encode(test_endpoint_id(0x11).as_bytes()); + assert!(parse_mesh_target_header(&[value.clone(), value]).is_err()); +} + +#[test] +fn parse_mesh_exclude_header_absent_is_empty() { + assert_eq!(parse_mesh_exclude_header(&[]), Ok(vec![])); +} + +#[test] +fn parse_mesh_exclude_header_splits_comma_separated_list() { + let id_a = test_endpoint_id(0x11); + let id_b = test_endpoint_id(0x22); + let combined = format!( + "{},{}", + hex::encode(id_a.as_bytes()), + hex::encode(id_b.as_bytes()) + ); + assert_eq!(parse_mesh_exclude_header(&[combined]), Ok(vec![id_a, id_b])); +} + +#[test] +fn parse_mesh_exclude_header_rejects_malformed_entry() { + assert!(parse_mesh_exclude_header(&["not-a-hex-endpoint-id".to_string()]).is_err()); +} + +#[test] +fn parse_mesh_exclude_header_rejects_an_empty_entry() { + assert!(parse_mesh_exclude_header(&["".to_string()]).is_err()); +} + +#[test] +fn parse_mesh_exclude_header_rejects_an_empty_entry_between_valid_ones() { + let id_a = hex::encode(test_endpoint_id(0x11).as_bytes()); + let id_b = hex::encode(test_endpoint_id(0x22).as_bytes()); + assert!(parse_mesh_exclude_header(&[format!("{id_a},,{id_b}")]).is_err()); +} + +#[test] +fn parse_mesh_exclude_header_rejects_a_trailing_comma() { + let id_a = hex::encode(test_endpoint_id(0x11).as_bytes()); + assert!(parse_mesh_exclude_header(&[format!("{id_a},")]).is_err()); +} + +// --- the local-first bypass: `mesh_headers_force_remote` must be consulted +// before a request is ever handed to the local-candidate path --- + +#[test] +fn mesh_headers_force_remote_is_false_with_no_headers() { + let self_id = test_endpoint_id(0x01); + assert!(!mesh_headers_force_remote(self_id, None, &[])); +} + +#[test] +fn mesh_headers_force_remote_is_true_when_target_names_another_peer() { + let self_id = test_endpoint_id(0x01); + let peer_id = test_endpoint_id(0x02); + assert!( + mesh_headers_force_remote(self_id, Some(peer_id), &[]), + "a target naming a remote peer must force routing away from local candidates" + ); +} + +#[test] +fn mesh_headers_force_remote_is_false_when_target_names_this_node() { + let self_id = test_endpoint_id(0x01); + assert!( + !mesh_headers_force_remote(self_id, Some(self_id), &[]), + "a target naming this node is allowed to serve locally" + ); +} + +#[test] +fn mesh_headers_force_remote_is_true_when_excluding_this_node() { + let self_id = test_endpoint_id(0x01); + let other_id = test_endpoint_id(0x02); + assert!( + mesh_headers_force_remote(self_id, None, &[other_id, self_id]), + "excluding this node must remove the local candidate" + ); +} + +#[test] +fn mesh_headers_force_remote_is_false_when_excluding_a_different_peer() { + let self_id = test_endpoint_id(0x01); + let other_id = test_endpoint_id(0x02); + assert!(!mesh_headers_force_remote(self_id, None, &[other_id])); +} + +#[tokio::test] +async fn resolve_remote_mesh_route_forces_single_candidate_for_serving_target() { + let model = "acme/code-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let peer = test_remote_peer(0x10, model); + let target_id = peer.id; + node.insert_test_peer(peer).await; + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let ctx = remote_mesh_test_ctx(&node, &targets, &affinity); + + match resolve_remote_mesh_route(&ctx, model, Some(target_id), &[]).await { + RemoteMeshRoute::Targets(mesh_targets) => { + assert_eq!( + mesh_targets.targets.get(model), + Some(&vec![election::InferenceTarget::Remote(target_id)]), + "x-mesh-target must force exactly one candidate, never a pool" + ); + } + _ => panic!("a target that serves the model must route, not fail closed"), + } +} + +#[tokio::test] +async fn resolve_remote_mesh_route_fails_closed_for_a_target_that_does_not_serve_the_model() { + let model = "acme/code-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + node.insert_test_peer(test_remote_peer(0x10, model)).await; + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let ctx = remote_mesh_test_ctx(&node, &targets, &affinity); + + // A syntactically valid EndpointId that no peer advertises for this model. + let stray_target = test_endpoint_id(0x99); + let route = resolve_remote_mesh_route(&ctx, model, Some(stray_target), &[]).await; + assert!( + matches!(route, RemoteMeshRoute::TargetUnavailable { .. }), + "a target that doesn't serve the model must fail closed, never substitute another peer" + ); +} + +#[tokio::test] +async fn resolve_remote_mesh_route_exclude_removes_a_peer_from_the_candidate_set() { + let model = "acme/code-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let peer_a = test_remote_peer(0x10, model); + let peer_b = test_remote_peer(0x20, model); + let (id_a, id_b) = (peer_a.id, peer_b.id); + node.insert_test_peer(peer_a).await; + node.insert_test_peer(peer_b).await; + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let ctx = remote_mesh_test_ctx(&node, &targets, &affinity); + + match resolve_remote_mesh_route(&ctx, model, None, &[id_a]).await { + RemoteMeshRoute::Targets(mesh_targets) => { + assert_eq!( + mesh_targets.targets.get(model), + Some(&vec![election::InferenceTarget::Remote(id_b)]), + "the excluded peer must be gone; the other peer must remain" + ); + } + other => panic!( + "expected the non-excluded peer to remain routable, got a different route: {}", + match other { + RemoteMeshRoute::Targets(_) => unreachable!(), + RemoteMeshRoute::TargetUnavailable { target_hex } => target_hex, + RemoteMeshRoute::NoRemoteHost => "NoRemoteHost".to_string(), + } + ), + } +} + +#[tokio::test] +async fn resolve_remote_mesh_route_excluding_the_named_target_fails_closed() { + let model = "acme/code-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let peer = test_remote_peer(0x10, model); + let target_id = peer.id; + node.insert_test_peer(peer).await; + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let ctx = remote_mesh_test_ctx(&node, &targets, &affinity); + + // Excluding the very peer named by x-mesh-target is a contradiction; it + // must fail closed rather than silently ignore the exclude and route + // there anyway. + let route = resolve_remote_mesh_route(&ctx, model, Some(target_id), &[target_id]).await; + assert!(matches!(route, RemoteMeshRoute::TargetUnavailable { .. })); +} + +#[tokio::test] +async fn resolve_remote_mesh_route_with_no_headers_and_no_remote_host_falls_through() { + let model = "acme/code-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let ctx = remote_mesh_test_ctx(&node, &targets, &affinity); + + let route = resolve_remote_mesh_route(&ctx, model, None, &[]).await; + assert!( + matches!(route, RemoteMeshRoute::NoRemoteHost), + "absent headers with no serving peer must fall through exactly as before" + ); +} + +#[tokio::test] +async fn resolve_remote_mesh_route_with_no_headers_pools_every_serving_peer() { + let model = "acme/code-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let peer_a = test_remote_peer(0x10, model); + let peer_b = test_remote_peer(0x20, model); + let (id_a, id_b) = (peer_a.id, peer_b.id); + node.insert_test_peer(peer_a).await; + node.insert_test_peer(peer_b).await; + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let ctx = remote_mesh_test_ctx(&node, &targets, &affinity); + + match resolve_remote_mesh_route(&ctx, model, None, &[]).await { + RemoteMeshRoute::Targets(mesh_targets) => { + let mut pool: Vec = mesh_targets + .targets + .get(model) + .expect("model present") + .iter() + .map(|target| match target { + election::InferenceTarget::Remote(id) => *id, + other => panic!("expected only remote candidates, got {other:?}"), + }) + .collect(); + pool.sort(); + let mut expected = vec![id_a, id_b]; + expected.sort(); + assert_eq!( + pool, expected, + "absent headers must route today's full multi-candidate pool" + ); + } + _ => panic!("expected the ordinary multi-candidate pool"), + } +} + +fn plugin_only_request(model: &str) -> proxy::BufferedHttpRequest { + let body = b"{}"; + let raw = format!( + "POST /v1/chat/completions HTTP/1.1\r\nHost: t\r\nContent-Length: {}\r\n\r\n", + body.len() + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect::>(); + proxy::BufferedHttpRequest { + raw, + method: "POST".to_owned(), + path: "/v1/chat/completions".to_owned(), + client_path: "/v1/chat/completions".to_owned(), + request_id: RequestId::default(), + body_json: None, + body_json_attempted: false, + body_bytes: None, + body_len_bytes: body.len(), + completion_tokens: None, + stream: None, + model_name: Some(model.to_owned()), + request_object_request_ids: Vec::new(), + response_adapter: proxy::ResponseAdapter::None, + correlation_id: None, + } +} + +fn empty_test_plugin_manager() -> crate::plugin::PluginManager { + crate::plugin::PluginManager::for_test_summaries(Vec::new()) +} + +/// Regression (ndizazzo P2a, PR #1671 round 2): an explicit `x-mesh-target` +/// naming THIS node must resolve against LOCAL PLUGIN availability instead +/// of failing closed with a spurious 409 because `resolve_remote_mesh_route` +/// only ever searches OTHER peers. Proven here by dialing a plugin endpoint +/// that refuses the connection (nothing bound to it): if the fix holds, the +/// outcome is the plugin-dispatch-attempted 503 `try_route_plugin_model` +/// itself produces on a failed endpoint -- never the fail-closed 409 that +/// never even looks at the plugin manager. +#[tokio::test] +async fn route_self_targeted_model_attempts_a_registered_plugin_instead_of_failing_closed() { + use tokio::io::AsyncReadExt; + + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let model = "acme/plugin-model:Q4_K_M"; + + // Bind then immediately drop the listener: the port is real but nothing + // answers, so a dial there deterministically refuses instead of racing a + // live server this test doesn't need. + let reserved = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind reserved port"); + let refusing_addr = reserved.local_addr().expect("addr"); + drop(reserved); + + let plugin_manager = empty_test_plugin_manager(); + plugin_manager + .set_test_inference_endpoints(vec![crate::plugin::InferenceEndpointRoute { + plugin_name: "acme".to_string(), + endpoint_id: "ep1".to_string(), + address: format!("http://{refusing_addr}"), + models: vec![model.to_string()], + }]) + .await; + + let ctx = IngressRouteContext { + node: &node, + targets: &targets, + affinity: &affinity, + plugin_manager: Some(&plugin_manager), + }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + let client = tokio::net::TcpStream::connect(addr); + let server = async { listener.accept().await.map(|(stream, _)| stream) }; + let (client_side, server_side) = tokio::join!(client, server); + let mut client_side = client_side.expect("connect"); + let tcp_stream: ClientStream = server_side.expect("accept").into(); + + let request = plugin_only_request(model); + let outcome = route_self_targeted_model( + tcp_stream, + &request, + &ctx, + model, + &[], + OpenAiRouteObserver::default(), + ) + .await; + + let mut response = Vec::new(); + client_side + .read_to_end(&mut response) + .await + .expect("read response"); + let response_text = String::from_utf8_lossy(&response); + + assert!( + !response_text.starts_with("HTTP/1.1 409"), + "self-target with a registered plugin model must not fail closed with the \ + 'refusing to fall back to another peer' 409: {response_text}" + ); + assert!( + response_text.contains("plugin endpoint"), + "expected the plugin-dispatch-attempted failure message, got: {response_text}" + ); + assert!( + matches!(outcome, proxy::RouteDispatchOutcome::Responded(503)), + "expected the plugin attempt's own 503, got {outcome:?}" + ); +} + +/// Regression (ndizazzo P2b, PR #1671 round 2): `x-mesh-exclude` naming this +/// node must block LOCAL PLUGIN fallback too. Proven by registering a +/// plugin endpoint at an address that would hang/fail if dialed, and +/// asserting the exclude check still produces 409 -- i.e. the plugin is +/// never even attempted once this node is excluded. +#[tokio::test] +async fn route_missing_local_model_excluding_self_blocks_local_plugin_fallback() { + use tokio::io::AsyncReadExt; + + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let self_id = node.id(); + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + let model = "acme/plugin-model:Q4_K_M"; + + let plugin_manager = empty_test_plugin_manager(); + plugin_manager + .set_test_inference_endpoints(vec![crate::plugin::InferenceEndpointRoute { + plugin_name: "acme".to_string(), + endpoint_id: "ep1".to_string(), + // Never dialed if the exclude check runs first, as it must. + address: "http://127.0.0.1:1".to_string(), + models: vec![model.to_string()], + }]) + .await; + + let ctx = IngressRouteContext { + node: &node, + targets: &targets, + affinity: &affinity, + plugin_manager: Some(&plugin_manager), + }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + let client = tokio::net::TcpStream::connect(addr); + let server = async { listener.accept().await.map(|(stream, _)| stream) }; + let (client_side, server_side) = tokio::join!(client, server); + let mut client_side = client_side.expect("connect"); + let tcp_stream: ClientStream = server_side.expect("accept").into(); + + let request = plugin_only_request(model); + let outcome = route_missing_local_model( + tcp_stream, + &request, + &ctx, + model, + None, + &[self_id], + None, + OpenAiRouteObserver::default(), + ) + .await; + + let mut response = Vec::new(); + client_side + .read_to_end(&mut response) + .await + .expect("read response"); + let response_text = String::from_utf8_lossy(&response); + + assert!( + response_text.starts_with("HTTP/1.1 409"), + "x-mesh-exclude naming this node must block local plugin fallback with 409, got: {response_text}" + ); + assert!( + matches!(outcome, proxy::RouteDispatchOutcome::Responded(409)), + "expected 409, got {outcome:?}" + ); +} diff --git a/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mesh_routing_tests.rs b/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mesh_routing_tests.rs new file mode 100644 index 0000000000..7a24103414 --- /dev/null +++ b/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mesh_routing_tests.rs @@ -0,0 +1,170 @@ +//! Regression (CodeRabbit, PR #1671 round 2 follow-up): `x-mesh-target`/ +//! `x-mesh-exclude` must not be rejected before `try_handle_moa` knows +//! whether a committee will actually be convened for `model: "mesh"`. +//! +//! Before this fix, the ingress-level pre-check +//! (`enforce_mesh_routing_headers_before_dispatch` / +//! `mesh_routing_unsupported_dispatch_kind`) rejected every `model: "mesh"` +//! request carrying either header with 409, before `try_handle_moa` ever +//! ran -- including the case where `try_handle_moa` was about to degrade +//! `model: "mesh"` to a single concrete model it could have routed with the +//! headers honored. These tests exercise `try_handle_moa` directly: a real +//! committee (fabricated via `fleet_sim_tests::node_with_fleet`, no sockets) +//! must still reject, but a degrade (zero admitted workers) must not. + +use super::fleet_sim_tests::{BIG_MODELS, node_with_fleet}; +use super::{MoaRoutingContext, try_handle_moa}; +use crate::inference::election; +use crate::mesh; +use crate::network::openai::client_stream::ClientStream; +use crate::network::openai::transport as proxy; +use crate::network::openai::transport::ResponseAdapter; +use mesh_llm_events::logging::identifiers::RequestId; + +fn moa_request_with_target(target_hex: &str) -> proxy::BufferedHttpRequest { + let body = serde_json::json!({ + "model": "mesh", + "messages": [{ "role": "user", "content": "hello" }], + }); + let body_bytes = serde_json::to_vec(&body).expect("serialize body"); + let mut raw = format!( + "POST /v1/chat/completions HTTP/1.1\r\nHost: t\r\nContent-Type: application/json\r\n\ + x-mesh-target: {target_hex}\r\nContent-Length: {}\r\n\r\n", + body_bytes.len() + ) + .into_bytes(); + raw.extend_from_slice(&body_bytes); + proxy::BufferedHttpRequest { + raw, + method: "POST".to_owned(), + path: "/v1/chat/completions".to_owned(), + client_path: "/v1/chat/completions".to_owned(), + request_id: RequestId::default(), + body_json: None, + body_json_attempted: false, + body_bytes: None, + body_len_bytes: body_bytes.len(), + completion_tokens: None, + stream: None, + model_name: Some("mesh".to_owned()), + request_object_request_ids: Vec::new(), + response_adapter: ResponseAdapter::OpenAiChatCompletionsJson, + correlation_id: None, + } +} + +async fn test_stream_pair() -> (ClientStream, tokio::net::TcpStream) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let addr = listener.local_addr().expect("addr"); + let client = tokio::net::TcpStream::connect(addr); + let server = async { listener.accept().await.map(|(stream, _)| stream) }; + let (client_side, server_side) = tokio::join!(client, server); + let client_side = client_side.expect("connect"); + let tcp_stream: ClientStream = server_side.expect("accept").into(); + (tcp_stream, client_side) +} + +/// A real committee (>=1 admitted worker) must still refuse to honor +/// `x-mesh-target`/`x-mesh-exclude` -- a fan-out across every admitted +/// worker cannot single out or exclude one peer. Deleting the +/// `mesh_routing_requested` guard in `try_handle_moa` (or hardcoding it to +/// `false`) turns this red: the request would instead run `run_moa_turn` +/// and hang/fail on a real dial to the fabricated peer. +#[tokio::test] +async fn try_handle_moa_rejects_routing_headers_once_a_committee_is_convened() { + use tokio::io::AsyncReadExt; + + let node = node_with_fleet(&[(BIG_MODELS[0], 1), (BIG_MODELS[1], 1)]).await; + let targets = election::ModelTargets::default(); + let affinity = crate::network::affinity::AffinityRouter::default(); + let mut request = moa_request_with_target("aabbccdd"); + let (tcp_stream, mut client_side) = test_stream_pair().await; + + let result = try_handle_moa( + &node, + tcp_stream, + &mut request, + Some("mesh"), + MoaRoutingContext { + targets: Some(&targets), + required_tokens: None, + affinity: &affinity, + mesh_routing_requested: true, + }, + crate::logging::OpenAiRouteObserver::default(), + ) + .await; + + assert!( + matches!(result, super::MoaDispatchResult::Responded(409)), + "expected a 409 once a committee is convened, got a different MoaDispatchResult variant" + ); + + let mut response = Vec::new(); + client_side + .read_to_end(&mut response) + .await + .expect("read response"); + let response_text = String::from_utf8_lossy(&response); + assert!( + response_text.starts_with("HTTP/1.1 409"), + "expected 409 on the wire, got: {response_text}" + ); + assert!( + response_text.contains("committee"), + "expected the committee-specific rejection message, got: {response_text}" + ); +} + +/// Zero admitted workers must degrade `model: "mesh"` to a single concrete +/// model and hand the stream back as `Passthrough` regardless of +/// `mesh_routing_requested` -- the caller (`route_request`) re-parses and +/// honors the headers against the rewritten model. Restoring the old +/// eager rejection (checking `mesh_routing_requested` before knowing +/// whether a committee will form) turns this red: the request would be +/// rejected with 409 even though no committee was ever going to run. +#[tokio::test] +async fn try_handle_moa_degrades_and_continues_when_no_committee_can_form() { + let node = mesh::Node::new_for_tests(mesh::NodeRole::Client) + .await + .expect("test node should start"); + let mut targets = election::ModelTargets::default(); + targets.targets.insert( + "solo/only-model:Q4_K_M".to_string(), + vec![election::InferenceTarget::Remote(iroh::EndpointId::from( + iroh::SecretKey::from_bytes(&[7u8; 32]).public(), + ))], + ); + let affinity = crate::network::affinity::AffinityRouter::default(); + let mut request = moa_request_with_target("aabbccdd"); + let (tcp_stream, _client_side) = test_stream_pair().await; + + let result = try_handle_moa( + &node, + tcp_stream, + &mut request, + Some("mesh"), + MoaRoutingContext { + targets: Some(&targets), + required_tokens: None, + affinity: &affinity, + mesh_routing_requested: true, // must NOT block the degrade + }, + crate::logging::OpenAiRouteObserver::default(), + ) + .await; + + assert!( + matches!(result, super::MoaDispatchResult::Passthrough(_)), + "expected Passthrough on degrade even with routing headers present, got a different \ + MoaDispatchResult variant" + ); + assert_eq!( + request.model_name.as_deref(), + Some("solo/only-model:Q4_K_M"), + "degrade must rewrite the virtual model so route_request can honor the routing headers \ + against the real target" + ); +} diff --git a/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mod.rs b/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mod.rs index ca9daf337b..407b9032a3 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mod.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/moa_gateway/mod.rs @@ -147,11 +147,17 @@ fn committee_admission( CommitteeAdmission::Admitted } -/// Per-request routing inputs shared with ordinary single-model routing. +/// Per-request routing inputs shared with ordinary single-model routing: the +/// local targets table (for worker-pool assembly), the caller's token +/// budget, cache affinity, and whether `x-mesh-target`/`x-mesh-exclude` were +/// asked to be honored. Grouped so `try_handle_moa` stays under clippy's +/// argument-count lint -- see its doc comment for what +/// `mesh_routing_requested` controls. pub struct MoaRoutingContext<'a> { pub targets: Option<&'a election::ModelTargets>, pub required_tokens: Option, pub affinity: &'a crate::network::affinity::AffinityRouter, + pub mesh_routing_requested: bool, } /// Detect `model: "mesh"`, build a mesh-wide MoA config, run the turn, @@ -168,6 +174,20 @@ pub struct MoaRoutingContext<'a> { /// successful Mesh gateway response, a 503 (when no worker is admitted), or /// a 400 (when the request body wasn't JSON) was already written. The caller /// must *not* attempt to respond again. +/// +/// `routing.mesh_routing_requested` is whether the caller asked +/// `x-mesh-target` / `x-mesh-exclude` to be honored. Those headers cannot be +/// honored once a real committee is convened -- a committee fans a turn out +/// across every admitted worker, not one peer -- so this function rejects +/// with 409 only at the point it actually decides to convene one (right +/// before `run_moa_turn`, below). Degrading to a single concrete model +/// instead (`degrade_to_single_model`, and +/// `CommitteeAdmission::ServeFromSingleModel`) returns `Passthrough` +/// unconditionally: the caller falls through to ordinary routing, which +/// re-parses and honors the same headers against the rewritten model. +/// Rejecting earlier -- before knowing whether a committee will actually +/// form -- would reject requests that MoA was about to make routable (PR +/// #1671 round 2 follow-up). pub async fn try_handle_moa( node: &mesh::Node, tcp_stream: ClientStream, @@ -176,6 +196,12 @@ pub async fn try_handle_moa( routing: MoaRoutingContext<'_>, route_observer: OpenAiRouteObserver<'_>, ) -> MoaDispatchResult { + let MoaRoutingContext { + targets, + required_tokens, + affinity, + mesh_routing_requested, + } = routing; if !effective_model.is_some_and(automatic::is_directive) { return MoaDispatchResult::Passthrough(tcp_stream); } @@ -220,20 +246,14 @@ pub async fn try_handle_moa( let enable_thinking = effective_enable_thinking_for_moa(&body_json); - let Some(mut config) = admitted_gateway_config( - node, - routing.targets, - routing.required_tokens, - routing.affinity, - ) - .await + let Some(mut config) = admitted_gateway_config(node, targets, required_tokens, affinity).await else { // Zero admitted workers cannot produce a turn, so degrade through the // ordinary selector (which returns 503 if no model exists). return degrade_to_single_model( node, - routing.targets, - routing.required_tokens, + targets, + required_tokens, tcp_stream, request, route_observer, @@ -242,6 +262,28 @@ pub async fn try_handle_moa( }; config.enable_thinking = enable_thinking; + if mesh_routing_requested { + // A committee is about to be convened for real (candidates were + // admitted above) -- `x-mesh-target`/`x-mesh-exclude` name or + // exclude a single peer, which a fan-out across every admitted + // worker cannot honor. Reject here, at the point the committee + // decision is actually made, rather than before `try_handle_moa` + // knew whether it would degrade instead. + return match proxy::send_409_observed( + tcp_stream, + "x-mesh-target/x-mesh-exclude are not honored once a committee is convened for \ + model \"mesh\" (multi-agent orchestration fans out across every admitted worker, \ + not one peer); retry without the routing headers, or with a concrete model name \ + once the fleet has degraded to one", + route_observer, + ) + .await + { + Ok(()) => MoaDispatchResult::Responded(409), + Err(_) => MoaDispatchResult::Dropped("moa_response_write_failed"), + }; + } + run_moa_turn( tcp_stream, body_json, @@ -450,6 +492,10 @@ mod usage_tests { #[path = "fleet_sim_tests.rs"] mod fleet_sim_tests; +#[cfg(test)] +#[path = "mesh_routing_tests.rs"] +mod mesh_routing_tests; + #[cfg(test)] #[path = "fleet_fairness_tests.rs"] mod fleet_fairness_tests; diff --git a/crates/mesh-llm-host-runtime/src/network/openai/request_parse.rs b/crates/mesh-llm-host-runtime/src/network/openai/request_parse.rs index 996957a219..d81526d08f 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/request_parse.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/request_parse.rs @@ -18,6 +18,12 @@ pub(crate) const MAX_HEADER_BYTES: usize = 64 * 1024; /// lifecycle parent, so ordinary API clients cannot opt into target-owner /// suppression by sending it themselves. pub(crate) const RAW_LIFECYCLE_OWNER_HEADER: &str = "x-mesh-llm-raw-lifecycle"; +/// Force remote-mesh dispatch to exactly one peer (fail closed if it doesn't +/// serve the requested model). See `ingress.rs`'s remote-mesh routing. +pub(crate) const MESH_TARGET_HEADER: &str = "x-mesh-target"; +/// Remove one or more peers from the remote-mesh candidate set before +/// selection. Comma-separated within one header value. +pub(crate) const MESH_EXCLUDE_HEADER: &str = "x-mesh-exclude"; pub(super) const MAX_BODY_BYTES: usize = 8 * 1024 * 1024; const MAX_OBJECT_UPLOAD_BODY_BYTES: usize = 64 * 1024 * 1024; const MAX_CHUNKED_WIRE_BYTES: usize = MAX_BODY_BYTES * 6 + 64 * 1024; @@ -163,6 +169,23 @@ impl BufferedHttpRequest { capsule_nonce_headers_from_raw(&self.raw) } + /// Raw (unparsed) values of the `x-mesh-target` / `x-mesh-exclude` mesh + /// routing headers, read back off the already-buffered raw request. + /// + /// Every occurrence of each header name is returned verbatim, including + /// duplicates — the router (not this parser) decides whether more than + /// one `x-mesh-target` value is an error. These headers are opaque to + /// this layer: no endpoint-id parsing happens here. A header value with + /// non-UTF-8 bytes is rejected outright rather than silently dropped, so + /// an attacker can't smuggle a routing decision past invalid bytes. + pub fn mesh_routing_header_values(&self) -> Result<(Vec, Vec), String> { + let target = header_values_from_raw(&self.raw, MESH_TARGET_HEADER) + .map_err(|()| format!("{MESH_TARGET_HEADER} header contains invalid UTF-8"))?; + let exclude = header_values_from_raw(&self.raw, MESH_EXCLUDE_HEADER) + .map_err(|()| format!("{MESH_EXCLUDE_HEADER} header contains invalid UTF-8"))?; + Ok((target, exclude)) + } + /// The only semantic request media kind trusted by artifact capture. /// /// This derives from the closed OpenAI ingress route vocabulary and a @@ -855,6 +878,35 @@ fn capsule_nonce_headers_from_raw(raw: &[u8]) -> (Option, Option (find(nonce_header), find(origin_header)) } +/// Every value of a given header name, read back off an already-rebuilt raw +/// HTTP request. Only the request-header block is scanned. Order matches the +/// wire order; duplicates are returned as separate entries. `Err(())` means +/// at least one occurrence of `name` had non-UTF-8 bytes -- the caller must +/// reject the request rather than silently drop that occurrence. +fn header_values_from_raw(raw: &[u8], name: &str) -> Result, ()> { + let header_end = raw + .windows(4) + .position(|window| window == b"\r\n\r\n") + .unwrap_or(raw.len()); + let mut headers_buf = [httparse::EMPTY_HEADER; MAX_HEADERS]; + let mut req = httparse::Request::new(&mut headers_buf); + if req + .parse(&raw[..header_end.saturating_add(4).min(raw.len())]) + .is_err() + { + return Ok(Vec::new()); + } + req.headers + .iter() + .filter(|header| header.name.eq_ignore_ascii_case(name)) + .map(|header| { + std::str::from_utf8(header.value) + .map(|value| value.trim().to_string()) + .map_err(|_| ()) + }) + .collect() +} + fn client_nonce_from_headers(headers: &[httparse::Header<'_>]) -> (String, Option<&'static str>) { let nonce_header = openai_frontend::lifecycle::CLIENT_NONCE_HEADER.as_str(); let inbound = headers diff --git a/crates/mesh-llm-host-runtime/src/network/openai/request_parse_tests.rs b/crates/mesh-llm-host-runtime/src/network/openai/request_parse_tests.rs index 85d12e7408..88bb288596 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/request_parse_tests.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/request_parse_tests.rs @@ -1038,3 +1038,97 @@ fn capsule_nonce_headers_from_raw_returns_none_without_headers() { assert_eq!(nonce, None); assert_eq!(origin, None); } + +fn request_with_raw(raw: &[u8]) -> BufferedHttpRequest { + BufferedHttpRequest { + raw: raw.to_vec(), + method: "POST".to_owned(), + path: "/v1/chat/completions".to_owned(), + client_path: "/v1/chat/completions".to_owned(), + request_id: RequestId::default(), + body_json: None, + body_json_attempted: false, + body_bytes: None, + body_len_bytes: 0, + completion_tokens: None, + stream: None, + model_name: None, + request_object_request_ids: Vec::new(), + response_adapter: ResponseAdapter::None, + correlation_id: None, + } +} + +#[test] +fn mesh_routing_header_values_absent_is_empty() { + let request = request_with_raw( + concat!( + "POST /v1/chat/completions HTTP/1.1\r\n", + "host: 127.0.0.1\r\n", + "\r\n", + "{}", + ) + .as_bytes(), + ); + let (target, exclude) = request.mesh_routing_header_values().unwrap(); + assert!(target.is_empty()); + assert!(exclude.is_empty()); +} + +#[test] +fn mesh_routing_header_values_reads_both_headers_verbatim() { + let request = request_with_raw( + concat!( + "POST /v1/chat/completions HTTP/1.1\r\n", + "host: 127.0.0.1\r\n", + "x-mesh-target: aabbcc\r\n", + "x-mesh-exclude: 112233,445566\r\n", + "\r\n", + "{}", + ) + .as_bytes(), + ); + let (target, exclude) = request.mesh_routing_header_values().unwrap(); + assert_eq!(target, vec!["aabbcc".to_string()]); + assert_eq!(exclude, vec!["112233,445566".to_string()]); +} + +#[test] +fn mesh_routing_header_values_surfaces_every_duplicate_x_mesh_target() { + // Ambiguity (more than one value) is the router's call, not this + // parser's -- it must see every occurrence, not just the first. + let request = request_with_raw( + concat!( + "POST /v1/chat/completions HTTP/1.1\r\n", + "host: 127.0.0.1\r\n", + "x-mesh-target: aabbcc\r\n", + "x-mesh-target: ddeeff\r\n", + "\r\n", + "{}", + ) + .as_bytes(), + ); + let (target, exclude) = request.mesh_routing_header_values().unwrap(); + assert_eq!(target, vec!["aabbcc".to_string(), "ddeeff".to_string()]); + assert!(exclude.is_empty()); +} + +#[test] +fn mesh_routing_header_values_rejects_non_utf8_x_mesh_target() { + let mut raw = + b"POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\nx-mesh-target: ".to_vec(); + raw.extend_from_slice(&[0xff, 0xfe]); + raw.extend_from_slice(b"\r\n\r\n{}"); + let request = request_with_raw(&raw); + assert!(request.mesh_routing_header_values().is_err()); +} + +#[test] +fn mesh_routing_header_values_rejects_non_utf8_x_mesh_exclude() { + let mut raw = + b"POST /v1/chat/completions HTTP/1.1\r\nhost: 127.0.0.1\r\nx-mesh-exclude: ".to_vec(); + raw.extend_from_slice(&[0xff, 0xfe]); + raw.extend_from_slice(b"\r\n\r\n{}"); + let request = request_with_raw(&raw); + assert!(request.mesh_routing_header_values().is_err()); +} diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response.rs b/crates/mesh-llm-host-runtime/src/network/openai/response.rs index 52d02a4bf2..88f711a3d5 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response.rs @@ -24,7 +24,7 @@ pub(crate) use models::send_models_list_with_descriptors; pub use pipeline::{PipelineCapsuleNonce, PipelineProxyResult, pipeline_proxy_local}; pub(super) use routing::{route_local_attempt, route_remote_attempt}; pub(crate) use send::{ - append_safe_header, is_valid_header_name, send_400, send_400_observed, send_503, - send_503_observed, send_error_observed, send_json_ok_with_headers, + append_safe_header, is_valid_header_name, send_400, send_400_observed, send_409_observed, + send_503, send_503_observed, send_error_observed, send_json_ok_with_headers, send_json_with_status_and_headers_observed, }; diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/common.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/common.rs index 7eac909b81..1208b02e3a 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/common.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/common.rs @@ -29,6 +29,9 @@ pub(in crate::network::openai) struct RouteAttemptLoggingContext<'a> { pub(in crate::network::openai) retry_policy: ResponseRetryPolicy, pub(in crate::network::openai) response_adapter: ResponseAdapter, pub(in crate::network::openai) route_observer: OpenAiRouteObserver<'a>, + /// Hex-encoded `EndpointId` to echo back as `x-mesh-served-by` on + /// delivery. See `RouteModelRequestContext::served_by_header`. + pub(in crate::network::openai) served_by: Option<&'a str>, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/dispatch.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/dispatch.rs index efa8f27163..071bf36aca 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/dispatch.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/dispatch.rs @@ -18,9 +18,12 @@ pub(in crate::network::openai::response) struct RelayAttemptContext<'a> { pub(in crate::network::openai::response) request_id: RequestId, pub(in crate::network::openai::response) disconnect_message: &'a str, pub(in crate::network::openai::response) commit_message: &'a str, + pub(in crate::network::openai::response) served_by: Option<&'a str>, pub(in crate::network::openai::response) route_observer: OpenAiRouteObserver<'a>, } +/// Relay one probed upstream response through the adapter, error, or success path. +#[allow(clippy::too_many_arguments)] pub(in crate::network::openai::response) async fn relay_probed_response( tcp_stream: &mut ClientStream, reader: &mut R, @@ -28,6 +31,7 @@ pub(in crate::network::openai::response) async fn relay_probed_response, route_observer: OpenAiRouteObserver<'_>, ) -> Result { if let Some(result) = relay_adapted_response( @@ -36,6 +40,7 @@ pub(in crate::network::openai::response) async fn relay_probed_response( probe: ResponseProbe, retry_policy: ResponseRetryPolicy, response_adapter: ResponseAdapter, + served_by: Option<&str>, route_observer: OpenAiRouteObserver<'_>, ) -> Result> { match response_adapter { @@ -78,6 +85,7 @@ async fn relay_adapted_response( reader, probe, retry_policy, + served_by, route_observer, ) .await?, @@ -88,6 +96,7 @@ async fn relay_adapted_response( reader, probe, retry_policy, + served_by, route_observer, ) .await?, @@ -98,6 +107,7 @@ async fn relay_adapted_response( reader, probe, retry_policy, + served_by, route_observer, ) .await?, @@ -108,6 +118,7 @@ async fn relay_adapted_response( reader, probe, retry_policy, + served_by, route_observer, ) .await?, @@ -132,6 +143,7 @@ pub(in crate::network::openai::response) async fn relay_attempted_response target, @@ -113,6 +114,7 @@ async fn route_http_endpoint_attempt_after_forward( request_id, disconnect_message: "API proxy (external endpoint): downstream client disconnected during relay", commit_message: "API proxy (external endpoint) ended after commit", + served_by: None, route_observer, }, retry_policy, diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/json_adaptation.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/json_adaptation.rs index a6869fceaa..509eada61e 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/json_adaptation.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/json_adaptation.rs @@ -5,7 +5,7 @@ use super::common::{ }; use super::probe::{ ResponseBodyReadLimits, ResponseProbe, append_capsule_nonce_headers, - read_transformed_response_body, try_parse_response_headers, + append_mesh_served_by_header, read_transformed_response_body, try_parse_response_headers, }; use super::relay::relay_error_response; use crate::logging::OpenAiRouteObserver; @@ -23,6 +23,7 @@ const TRANSFORMED_RESPONSE_READ_LIMITS: ResponseBodyReadLimits = ResponseBodyRea idle_timeout: TRANSFORMED_RESPONSE_BODY_IDLE_TIMEOUT, }; +/// Relay a chat-completions upstream response translated into Responses-API JSON. pub(in crate::network::openai::response) async fn relay_translated_responses_json< R: AsyncRead + Unpin, >( @@ -30,6 +31,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_jso reader: &mut R, probe: ResponseProbe, retry_policy: ResponseRetryPolicy, + served_by: Option<&str>, route_observer: OpenAiRouteObserver<'_>, ) -> Result { if retry_policy.context_overflow && probe.retryable_context_overflow { @@ -37,7 +39,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_jso } if !(200..300).contains(&probe.status_code) { - return relay_error_response(tcp_stream, reader, probe, route_observer).await; + return relay_error_response(tcp_stream, reader, probe, served_by, route_observer).await; } let mut buffered = probe.buffered; let parsed = try_parse_response_headers(&buffered)? @@ -66,6 +68,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_jso parsed.client_nonce.as_deref(), parsed.nonce_origin.as_deref(), ); + append_mesh_served_by_header(&mut header, served_by); header.push_str("Connection: close\r\n\r\n"); tcp_stream.write_all(header.as_bytes()).await?; tcp_stream.write_all(&translated_body).await?; @@ -78,6 +81,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_jso }) } +/// Relay a chat-completions upstream response through JSON body normalization. pub(in crate::network::openai::response) async fn relay_normalized_chat_completion_json< R: AsyncRead + Unpin, >( @@ -85,6 +89,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi reader: &mut R, probe: ResponseProbe, retry_policy: ResponseRetryPolicy, + served_by: Option<&str>, route_observer: OpenAiRouteObserver<'_>, ) -> Result { if retry_policy.context_overflow && probe.retryable_context_overflow { @@ -92,7 +97,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi } if !(200..300).contains(&probe.status_code) { - return relay_error_response(tcp_stream, reader, probe, route_observer).await; + return relay_error_response(tcp_stream, reader, probe, served_by, route_observer).await; } let mut buffered = probe.buffered; let parsed = try_parse_response_headers(&buffered)? @@ -122,6 +127,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi parsed.client_nonce.as_deref(), parsed.nonce_origin.as_deref(), ); + append_mesh_served_by_header(&mut header, served_by); header.push_str("Connection: close\r\n\r\n"); tcp_stream.write_all(header.as_bytes()).await?; tcp_stream.write_all(&normalized_body).await?; @@ -200,6 +206,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -274,6 +281,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -328,6 +336,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/probe.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/probe.rs index 1a4db2fff7..331f610bf9 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/probe.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/probe.rs @@ -42,6 +42,96 @@ pub(in crate::network::openai::response) fn append_capsule_nonce_headers( } } +/// The header the routing node echoes the resolved `x-mesh-target` peer back +/// on, so the client can seal which peer answered without parsing +/// provenance. See `ingress.rs`'s remote-mesh routing. +pub(in crate::network::openai::response) const MESH_SERVED_BY_HEADER: &str = "x-mesh-served-by"; + +/// Append the `x-mesh-served-by` header (when present) to a hand-built +/// response header string. Sibling of `append_capsule_nonce_headers` for +/// response adapters that rebuild headers instead of relaying them raw. +pub(in crate::network::openai::response) fn append_mesh_served_by_header( + header: &mut String, + served_by: Option<&str>, +) { + if let Some(served_by) = served_by { + header.push_str(&format!("{MESH_SERVED_BY_HEADER}: {served_by}\r\n")); + } +} + +/// Remove an existing `name: ...` header line (case-insensitive) from the +/// header block `buf[..header_end]`, if one is present, and return how many +/// bytes were removed (0 if none matched). +/// +/// Belt-and-braces for [`insert_header_before_body`]: a value this codebase +/// already meant to be single-valued (e.g. `x-mesh-served-by`) must never +/// appear twice just because some upstream response already carried one — +/// canonicalize to one value instead of appending a second. +fn remove_existing_header(buf: &mut Vec, header_end: usize, name: &str) -> usize { + let mut offset = 0usize; + let mut total_removed = 0usize; + while offset < header_end.saturating_sub(total_removed) { + let remaining_end = header_end - total_removed; + let Some(rel) = buf[offset..remaining_end] + .windows(2) + .position(|w| w == b"\r\n") + else { + break; + }; + let line_end = offset + rel + 2; + let is_match = buf[offset..line_end] + .iter() + .position(|&b| b == b':') + .is_some_and(|colon| buf[offset..offset + colon].eq_ignore_ascii_case(name.as_bytes())); + if is_match { + let removed = line_end - offset; + buf.drain(offset..line_end); + total_removed += removed; + // don't advance offset — the next line now starts at the same offset + } else { + offset = line_end; + } + } + total_removed +} + +/// Splice a `name: value` header line into an already-buffered raw HTTP +/// response, immediately before the blank line that terminates the header +/// block, replacing any existing header of the same name (case-insensitive) +/// rather than appending a duplicate. Returns the net change in buffer +/// length (new line inserted minus any old line removed) so callers can +/// adjust byte offsets computed against the pre-splice buffer. Used by relay +/// paths that forward upstream response bytes verbatim and have no other +/// rebuild step. +/// +/// `header_end` must be the byte offset of the first body byte (the +/// convention `httparse::Status::Complete` and `ParsedResponseHeaders` +/// already use) — i.e. it points just past the terminating `\r\n\r\n`. +pub(in crate::network::openai::response) fn insert_header_before_body( + buf: &mut Vec, + header_end: usize, + name: &str, + value: &str, +) -> isize { + if header_end < 2 || header_end > buf.len() { + return 0; + } + let removed = remove_existing_header(buf, header_end, name); + let header_end = header_end - removed; + let mut line = Vec::with_capacity(name.len() + value.len() + 4); + line.extend_from_slice(name.as_bytes()); + line.extend_from_slice(b": "); + line.extend( + value + .bytes() + .filter(|byte| *byte != b'\r' && *byte != b'\n'), + ); + line.extend_from_slice(b"\r\n"); + let inserted = line.len(); + buf.splice(header_end - 2..header_end - 2, line); + inserted as isize - removed as isize +} + #[derive(Clone, Copy)] pub(in crate::network::openai::response) struct ResponseBodyReadLimits { pub(in crate::network::openai::response) max_body_bytes: usize, @@ -329,6 +419,127 @@ mod tests { use crate::network::openai::response::common::is_timeout_error; use tokio::io::AsyncWriteExt; + #[test] + fn insert_header_before_body_splices_before_the_blank_line() { + let body = b"{}"; + let header = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n"; + let mut buf = header.to_vec(); + buf.extend_from_slice(body); + let header_end = header.len(); + + let inserted = insert_header_before_body(&mut buf, header_end, "x-mesh-served-by", "ab12"); + + let expected = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nx-mesh-served-by: ab12\r\n\r\n{}"; + assert_eq!(buf, expected); + assert_eq!( + inserted, + (expected.len() - (header.len() + body.len())) as isize + ); + // The body bytes themselves must be untouched, just shifted. + assert_eq!(&buf[buf.len() - body.len()..], body); + } + + /// Regression (erlich review, 2026-09-07 round 2 / ndizazzo P2c): a + /// response that already carries the header (e.g. echoed back verbatim + /// from a peer) must end up with exactly one value, not two appended + /// copies that a client combining repeated fields would read as `id, id`. + #[test] + fn insert_header_before_body_replaces_an_existing_value_instead_of_duplicating() { + let body = b"{}"; + let header = b"HTTP/1.1 200 OK\r\nx-mesh-served-by: stale\r\nContent-Length: 2\r\n\r\n"; + let mut buf = header.to_vec(); + buf.extend_from_slice(body); + let header_end = header.len(); + + insert_header_before_body(&mut buf, header_end, "x-mesh-served-by", "fresh"); + + let text = String::from_utf8_lossy(&buf); + assert_eq!( + text.matches("x-mesh-served-by:").count(), + 1, + "expected exactly one served-by header, got: {text}" + ); + assert!( + text.contains("x-mesh-served-by: fresh\r\n"), + "expected the new value to win: {text}" + ); + assert!(!text.contains("stale"), "stale value must be gone: {text}"); + assert!(buf.ends_with(body), "body must survive the replace: {text}"); + } + + /// Regression (CodeRabbit, PR #1671: "the header-removal loop stops + /// after the first case-insensitive match"): a response carrying TWO + /// pre-existing `x-mesh-served-by` occurrences (mixed case, as an + /// upstream that itself failed to canonicalize might produce) must end + /// up with exactly one value after the splice, not one stale duplicate + /// left behind by a loop that returns after its first removal. + #[test] + fn insert_header_before_body_removes_every_duplicate_not_just_the_first() { + let body = b"{}"; + let header = b"HTTP/1.1 200 OK\r\nx-mesh-served-by: first\r\nContent-Length: 2\r\nX-Mesh-Served-By: second\r\n\r\n"; + let mut buf = header.to_vec(); + buf.extend_from_slice(body); + let header_end = header.len(); + + insert_header_before_body(&mut buf, header_end, "x-mesh-served-by", "fresh"); + + let text = String::from_utf8_lossy(&buf); + assert_eq!( + text.to_ascii_lowercase() + .matches("x-mesh-served-by:") + .count(), + 1, + "expected every duplicate collapsed to one header, got: {text}" + ); + assert!( + text.contains("x-mesh-served-by: fresh\r\n"), + "expected the new value to win: {text}" + ); + assert!( + !text.contains("first"), + "first duplicate must be gone: {text}" + ); + assert!( + !text.contains("second"), + "second duplicate must be gone: {text}" + ); + assert!(buf.ends_with(body), "body must survive the replace: {text}"); + } + + #[test] + fn insert_header_before_body_replace_is_case_insensitive() { + let header = b"HTTP/1.1 200 OK\r\nX-Mesh-Served-By: stale\r\n\r\n"; + let mut buf = header.to_vec(); + let header_end = header.len(); + + insert_header_before_body(&mut buf, header_end, "x-mesh-served-by", "fresh"); + + let text = String::from_utf8_lossy(&buf); + assert_eq!( + text.to_ascii_lowercase() + .matches("x-mesh-served-by:") + .count(), + 1, + "expected exactly one served-by header regardless of case, got: {text}" + ); + assert!(text.contains("fresh")); + } + + #[test] + fn insert_header_before_body_strips_crlf_from_the_value() { + let header = b"HTTP/1.1 200 OK\r\n\r\n"; + let mut buf = header.to_vec(); + let header_end = header.len(); + + insert_header_before_body(&mut buf, header_end, "x-mesh-served-by", "ab\r\n12"); + + assert!( + String::from_utf8_lossy(&buf).contains("x-mesh-served-by: ab12\r\n"), + "CR/LF in the value must not let it smuggle extra header lines: {}", + String::from_utf8_lossy(&buf) + ); + } + #[tokio::test] async fn transformed_response_rejects_oversized_content_length_before_reading() { let (_writer, mut reader) = tokio::io::duplex(64); diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/relay.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/relay.rs index 61adfe5e44..189ba31df8 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/relay.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/relay.rs @@ -3,9 +3,10 @@ use super::common::{ ResponseRetryPolicy, RouteAttemptResult, parse_token_usage_from_json_body, retryable_quality_result, }; +use super::probe::MESH_SERVED_BY_HEADER; use super::probe::{ - ParsedResponseHeaders, ResponseProbe, append_capsule_nonce_headers, read_response_chunk, - try_parse_response_headers, + ParsedResponseHeaders, ResponseProbe, append_capsule_nonce_headers, insert_header_before_body, + read_response_chunk, try_parse_response_headers, }; use crate::logging::{ArtifactUnavailableReason, OpenAiRouteObserver}; use crate::network::openai::client_stream::ClientStream; @@ -90,10 +91,41 @@ fn oversized_error_http_response(status_code: u16) -> Vec { .into_bytes() } +/// Byte offset just past the terminating `\r\n\r\n` of a freshly-built +/// response, for splicing an extra header into a buffer whose header block +/// wasn't tracked through the branch that produced it (oversized / remapped / +/// passthrough error bodies each build `outgoing` differently). +/// +/// `None` when no `\r\n\r\n` terminator is found (e.g. an upstream that ends +/// its header block with a bare LF) -- returning `response.len()` here used +/// to look like a valid offset to `insert_header_before_body`, which would +/// then splice two bytes before the end of an already-complete response, +/// corrupting it silently instead of skipping the insert. +fn response_header_end(response: &[u8]) -> Option { + response + .windows(4) + .position(|window| window == b"\r\n\r\n") + .map(|pos| pos + 4) +} + +/// Splice `x-mesh-served-by` into an already-built error response, when set. +fn append_served_by_to_error_response(outgoing: &mut Vec, served_by: Option<&str>) { + let Some(served_by) = served_by else { return }; + let Some(header_end) = response_header_end(outgoing) else { + tracing::debug!( + "no header terminator found while echoing x-mesh-served-by on an error response; skipping insert" + ); + return; + }; + insert_header_before_body(outgoing, header_end, MESH_SERVED_BY_HEADER, served_by); +} + +/// Relay a non-2xx upstream response, echoing `x-mesh-served-by` when set. pub(in crate::network::openai::response) async fn relay_error_response( tcp_stream: &mut ClientStream, reader: &mut R, probe: ResponseProbe, + served_by: Option<&str>, route_observer: OpenAiRouteObserver<'_>, ) -> Result { let status_code = probe.status_code; @@ -103,7 +135,7 @@ pub(in crate::network::openai::response) async fn relay_error_response MAX_ERROR_RESPONSE_BYTES { + let mut outgoing = if buffered.len().saturating_sub(header_end) > MAX_ERROR_RESPONSE_BYTES { tracing::warn!( "upstream error body exceeded {} bytes for status {}", MAX_ERROR_RESPONSE_BYTES, @@ -113,6 +145,7 @@ pub(in crate::network::openai::response) async fn relay_error_response, route_observer: OpenAiRouteObserver<'_>, ) -> Result { if let Some(content_length) = parsed.content_length { @@ -152,10 +186,24 @@ pub(in crate::network::openai::response) async fn relay_success_response( tcp_stream: &mut ClientStream, upstream: &mut U, @@ -78,6 +81,7 @@ async fn route_local_attempt_after_forward, route_observer: OpenAiRouteObserver<'_>, ) -> RouteAttemptResult { match probe_with_downstream_disconnect(tcp_stream, probe_http_response_local(upstream)).await { @@ -94,6 +98,7 @@ async fn route_local_attempt_after_forward( tcp_stream: &mut ClientStream, quic_recv: &mut R, @@ -141,6 +147,7 @@ async fn route_remote_attempt_after_forward, route_observer: OpenAiRouteObserver<'_>, ) -> RouteAttemptResult { match probe_with_downstream_disconnect(tcp_stream, probe_http_response(quic_recv)).await { @@ -160,6 +167,7 @@ async fn route_remote_attempt_after_forward tunnel, @@ -216,6 +225,7 @@ pub(in crate::network::openai) async fn route_remote_attempt( request_id, retry_policy, response_adapter, + served_by, route_observer, ) .await @@ -361,6 +371,7 @@ mod tests { RequestId::new(), ResponseRetryPolicy::next_target_available(false), ResponseAdapter::None, + None, OpenAiRouteObserver::default(), ) .await @@ -405,6 +416,7 @@ mod tests { RequestId::new(), ResponseRetryPolicy::next_target_available(false), ResponseAdapter::None, + None, OpenAiRouteObserver::default(), ) .await @@ -456,6 +468,7 @@ mod tests { RequestId::new(), ResponseRetryPolicy::next_target_available(false), ResponseAdapter::None, + None, OpenAiRouteObserver::default(), ) .await @@ -541,6 +554,7 @@ mod tests { RequestId::new(), ResponseRetryPolicy::next_target_available(false), ResponseAdapter::None, + None, OpenAiRouteObserver::default(), ) .await @@ -577,6 +591,7 @@ mod tests { RequestId::new(), ResponseRetryPolicy::next_target_available(false), ResponseAdapter::None, + None, OpenAiRouteObserver::default(), ) .await diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/send.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/send.rs index 560df2e903..7d97c13701 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/send.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/send.rs @@ -148,6 +148,14 @@ pub(crate) async fn send_400_observed( send_error_observed(stream, 400, msg, route_observer).await } +pub(crate) async fn send_409_observed( + stream: ClientStream, + msg: &str, + route_observer: OpenAiRouteObserver<'_>, +) -> std::io::Result<()> { + send_error_observed(stream, 409, msg, route_observer).await +} + pub(crate) async fn send_503_observed( stream: ClientStream, reason: &str, diff --git a/crates/mesh-llm-host-runtime/src/network/openai/response/stream_translation.rs b/crates/mesh-llm-host-runtime/src/network/openai/response/stream_translation.rs index fbc51fd864..a28f057938 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/response/stream_translation.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/response/stream_translation.rs @@ -1,8 +1,8 @@ use super::cache_cost::{CacheCostObservation, parse_cache_cost_from_json_body}; use super::common::{ResponseRetryPolicy, RouteAttemptResult, parse_token_usage_from_json_body}; use super::probe::{ - ResponseProbe, append_capsule_nonce_headers, response_is_event_stream, - try_parse_response_headers, + ResponseProbe, append_capsule_nonce_headers, append_mesh_served_by_header, + response_is_event_stream, try_parse_response_headers, }; use super::relay::{relay_error_response, relay_success_response}; use crate::logging::{OpenAiRouteObserver, OpenAiStreamArtifactCapture}; @@ -67,6 +67,7 @@ impl ResponsesStreamRelayState { } } +/// Relay a streaming chat-completions upstream response, normalizing tool-call ids. pub(in crate::network::openai::response) async fn relay_normalized_chat_completion_stream< R: AsyncRead + Unpin, >( @@ -74,6 +75,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi reader: &mut R, probe: ResponseProbe, retry_policy: ResponseRetryPolicy, + served_by: Option<&str>, route_observer: OpenAiRouteObserver<'_>, ) -> Result { if retry_policy.context_overflow && probe.retryable_context_overflow { @@ -82,7 +84,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi if !(200..300).contains(&probe.status_code) { route_observer.stream_error("upstream_status"); - return relay_error_response(tcp_stream, reader, probe, route_observer).await; + return relay_error_response(tcp_stream, reader, probe, served_by, route_observer).await; } let parsed = try_parse_response_headers(&probe.buffered)? @@ -94,6 +96,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi probe, parsed, retry_policy, + served_by, route_observer, ) .await; @@ -111,6 +114,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi parsed.client_nonce.as_deref(), parsed.nonce_origin.as_deref(), ); + append_mesh_served_by_header(&mut header, served_by); header.push_str("Connection: close\r\n\r\n"); tcp_stream.write_all(header.as_bytes()).await?; let mut response_capture = route_observer.begin_stream_response_capture(); @@ -209,6 +213,7 @@ pub(in crate::network::openai::response) async fn relay_normalized_chat_completi }) } +/// Relay a streaming chat-completions upstream response translated into Responses-API SSE. pub(in crate::network::openai::response) async fn relay_translated_responses_stream< R: AsyncRead + Unpin, >( @@ -216,6 +221,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_str reader: &mut R, probe: ResponseProbe, retry_policy: ResponseRetryPolicy, + served_by: Option<&str>, route_observer: OpenAiRouteObserver<'_>, ) -> Result { fn should_parse_stream_chunk(data: &str, model_missing: bool, usage_missing: bool) -> bool { @@ -276,7 +282,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_str if !(200..300).contains(&probe.status_code) { route_observer.stream_error("upstream_status"); - return relay_error_response(tcp_stream, reader, probe, route_observer).await; + return relay_error_response(tcp_stream, reader, probe, served_by, route_observer).await; } let parsed = try_parse_response_headers(&probe.buffered)? @@ -291,6 +297,7 @@ pub(in crate::network::openai::response) async fn relay_translated_responses_str parsed.client_nonce.as_deref(), parsed.nonce_origin.as_deref(), ); + append_mesh_served_by_header(&mut header, served_by); header.push_str("Connection: close\r\n\r\n"); tcp_stream.write_all(header.as_bytes()).await?; let mut response_capture = route_observer.begin_stream_response_capture(); @@ -724,6 +731,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -812,6 +820,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::capture_test_observer(RequestId::new(), &observer_capture), ) .await @@ -887,6 +896,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -940,6 +950,7 @@ mod tests { request_id: RequestId::new(), disconnect_message: "test client disconnected", commit_message: "test stream relay failed", + served_by: None, route_observer: OpenAiRouteObserver::default(), }, ResponseRetryPolicy::next_target_available(false), @@ -989,6 +1000,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -1032,6 +1044,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -1091,6 +1104,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await @@ -1148,6 +1162,7 @@ mod tests { &mut upstream_reader, probe, ResponseRetryPolicy::next_target_available(false), + None, OpenAiRouteObserver::default(), ) .await diff --git a/crates/mesh-llm-host-runtime/src/network/openai/transport.rs b/crates/mesh-llm-host-runtime/src/network/openai/transport.rs index 896b6e9e2f..95bd316504 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/transport.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/transport.rs @@ -26,8 +26,9 @@ pub use super::request_parse::{ }; pub(crate) use super::response::{ PipelineCapsuleNonce, PipelineProxyResult, append_safe_header, pipeline_proxy_local, - send_400_observed, send_503_observed, send_error_observed, send_json_ok_with_headers, - send_json_with_status_and_headers_observed, send_models_list_with_descriptors, + send_400_observed, send_409_observed, send_503_observed, send_error_observed, + send_json_ok_with_headers, send_json_with_status_and_headers_observed, + send_models_list_with_descriptors, }; pub(crate) use super::routing_rank::{ capabilities_for_model, descriptor_metadata_for_model, request_budget_tokens_from_parts, @@ -430,15 +431,18 @@ async fn route_mesh_moa_or_passthrough( let moa_model_name = request.model_name.clone(); let moa_required_tokens = request_context_budget(request); let adapter = request.response_adapter; + let mesh_routing_requested = + crate::network::openai::ingress::mesh_routing_headers_requested(request); let result = match crate::network::openai::moa_gateway::try_handle_moa( node, tcp_stream, request, moa_model_name.as_deref(), - super::moa_gateway::MoaRoutingContext { + crate::network::openai::moa_gateway::MoaRoutingContext { targets: None, // passive path has no local targets table required_tokens: moa_required_tokens, affinity, + mesh_routing_requested, }, route_observer, ) @@ -728,6 +732,10 @@ async fn route_mesh_request_attempts( retry_policy: ResponseRetryPolicy::next_target_available(idx + 1 < total_targets), response_adapter: request.response_adapter, route_observer, + // This is the separate "mesh" auto-plan fan-out across many + // candidate hosts, not the `x-mesh-target` forced single-peer + // path -- there is no one chosen target to echo here. + served_by: None, }, ) .await; @@ -1515,6 +1523,7 @@ pub async fn route_to_target( retry_policy, response_adapter, route_observer, + served_by: None, }, ) .await; @@ -1606,6 +1615,7 @@ pub async fn route_http_endpoint_request( retry_policy: ResponseRetryPolicy::next_target_available(false), response_adapter: request.response_adapter, route_observer, + served_by: None, }, ) .await; diff --git a/crates/mesh-llm-host-runtime/src/network/openai/transport_route_model.rs b/crates/mesh-llm-host-runtime/src/network/openai/transport_route_model.rs index adbdaae522..286b72b5d5 100644 --- a/crates/mesh-llm-host-runtime/src/network/openai/transport_route_model.rs +++ b/crates/mesh-llm-host-runtime/src/network/openai/transport_route_model.rs @@ -6,6 +6,12 @@ pub(crate) struct RouteModelRequestContext<'a> { pub(crate) required_tokens: Option, pub(crate) affinity: &'a AffinityRouter, pub(crate) route_observer: OpenAiRouteObserver<'a>, + /// Hex-encoded `EndpointId` to echo back as `x-mesh-served-by` once the + /// response is delivered. Set only when the caller (an `x-mesh-target` + /// forced single-candidate dispatch) already knows the exact peer that + /// will serve the request; `None` everywhere else, including ordinary + /// multi-candidate remote-mesh routing. + pub(crate) served_by_header: Option<&'a str>, } pub async fn route_model_request( @@ -25,6 +31,7 @@ pub async fn route_model_request( required_tokens: context.required_tokens, affinity: context.affinity, route_observer: context.route_observer, + served_by_header: context.served_by_header, }; route_model_request_inner(args).await } @@ -38,6 +45,7 @@ struct RouteModelRequestArgs<'a> { required_tokens: Option, affinity: &'a AffinityRouter, route_observer: OpenAiRouteObserver<'a>, + served_by_header: Option<&'a str>, } struct RouteModelState { @@ -92,6 +100,7 @@ async fn route_model_request_inner(args: RouteModelRequestArgs<'_>) -> RouteDisp required_tokens, affinity, route_observer, + served_by_header, } = args; let route_started = Instant::now(); let mut tcp_stream = tcp_stream; @@ -153,6 +162,7 @@ async fn route_model_request_inner(args: RouteModelRequestArgs<'_>) -> RouteDisp retry_policy, response_adapter: request.response_adapter, route_observer, + served_by: served_by_header, }, ) .await; diff --git a/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs b/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs index 73f7ffa7c9..ae83410336 100644 --- a/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs +++ b/crates/mesh-llm-host-runtime/src/plugin/openai_exchange.rs @@ -29,6 +29,12 @@ pub enum OpenAiExchangeDispatchPath { /// The raw-proxy ingress (`network/openai/ingress.rs`), used for /// plugin-served models; never sees a typed `ChatCompletionRequest`. RawProxy, + /// The raw-proxy ingress routes this exchange to a peer on the mesh + /// rather than serving it locally (`route_missing_local_model`'s + /// remote-mesh branch). This node is the requester/router, not the + /// server, for the exchange this envelope describes — a downstream + /// plugin must not treat it as the served-side event. + RemoteMesh, } /// Which moment in an exchange's lifecycle an [`OpenAiExchangeEnvelope`] @@ -125,6 +131,59 @@ impl OpenAiExchangeEnvelope { nonce_source, } } + + /// Effective-request envelope for the `RemoteMesh` dispatch path, + /// carrying the nonce this node is about to forward to the peer + /// unchanged — so a plugin observing only the effective event already + /// knows what a later client ack must sign over, rather than having to + /// wait for the terminal event. `capsule_id` stays absent: this node + /// mints nothing on this path. + pub fn effective_remote_mesh( + exchange_id: impl Into, + model: impl Into, + nonce: Option, + nonce_source: Option, + ) -> Self { + Self { + exchange_id: exchange_id.into(), + dispatch_path: OpenAiExchangeDispatchPath::RemoteMesh, + phase: OpenAiExchangePhase::EffectiveRequest, + model: model.into(), + status: None, + capsule_id: None, + nonce, + nonce_source, + } + } + + /// Terminal envelope for the `RemoteMesh` dispatch path — a routing node + /// observing (not serving) an exchange it forwarded to a peer. + /// + /// Unlike [`Self::terminal`]'s `marker`, which bundles a capsule_id this + /// node minted together with the nonce that capsule is correlated + /// against, a routing node mints nothing here: `nonce` is the same + /// client-contributed value forwarded to the peer unchanged (present + /// only when the request already carries a stabilized nonce). No peer + /// response header is read back on this path, so `capsule_id` stays + /// absent, same as the plugin-served terminal event. + pub fn terminal_remote_mesh( + exchange_id: impl Into, + model: impl Into, + status: Option, + nonce: Option, + nonce_source: Option, + ) -> Self { + Self { + exchange_id: exchange_id.into(), + dispatch_path: OpenAiExchangeDispatchPath::RemoteMesh, + phase: OpenAiExchangePhase::Terminal, + model: model.into(), + status, + capsule_id: None, + nonce, + nonce_source, + } + } } /// Publishes [`OpenAiExchangeEnvelope`]s to whatever is subscribed on @@ -564,4 +623,119 @@ mod tests { "slow exchange's terminal event pairs with its own effective event" ); } + + // --- #1668 review round: RemoteMesh / RawProxy publish pairs --- + // + // Neither `route_missing_local_model`'s remote-mesh branch nor + // `try_route_plugin_model` (see `network::openai::ingress`) is economical + // to invoke directly in a unit test -- both need a live TCP stream and a + // real `mesh::Node`/`PluginManager` (see `ingress_tests`'s own comment + // on `plugin_route_status`). These test the pure envelope pair each call + // site publishes instead. + + /// `route_missing_local_model`'s remote-mesh branch publishes the same + /// effective/terminal pair `try_route_plugin_model` does for its own + /// dispatch, with `RemoteMesh` in place of `RawProxy` -- and, unlike that + /// path, carrying the nonce forwarded to the peer unchanged on BOTH + /// envelopes, not just the terminal one, so a plugin observing only the + /// effective event already knows what a later client ack must sign over. + /// `capsule_id` stays absent on both: this node mints nothing here. + #[tokio::test] + async fn remote_mesh_branch_publishes_effective_and_terminal_with_the_forwarded_nonce() { + let channel = RecordingChannel::default(); + let nonce = Some("6d7d8d2e-3f4a-4b5c-8d9e-0a1b2c3d4e5f".to_string()); + let nonce_source = Some(ClientNonceSource::ClientSupplied); + + channel + .publish(&OpenAiExchangeEnvelope::effective_remote_mesh( + "exch-rm-1", + "hermes-2-pro-mistral-7b", + nonce.clone(), + nonce_source, + )) + .await; + channel + .publish(&OpenAiExchangeEnvelope::terminal_remote_mesh( + "exch-rm-1", + "hermes-2-pro-mistral-7b", + Some(200), + nonce.clone(), + nonce_source, + )) + .await; + + let events = channel.events.lock().unwrap(); + assert_eq!(events.len(), 2, "one effective-request, one terminal"); + + assert_eq!( + events[0].dispatch_path, + OpenAiExchangeDispatchPath::RemoteMesh + ); + assert_eq!(events[0].phase, OpenAiExchangePhase::EffectiveRequest); + assert_eq!(events[0].nonce, nonce); + assert_eq!(events[0].nonce_source, nonce_source); + assert!(events[0].capsule_id.is_none()); + + assert_eq!( + events[1].dispatch_path, + OpenAiExchangeDispatchPath::RemoteMesh + ); + assert_eq!(events[1].phase, OpenAiExchangePhase::Terminal); + assert_eq!(events[1].exchange_id, events[0].exchange_id); + assert_eq!(events[1].status, Some(200)); + assert_eq!(events[1].nonce, nonce); + assert_eq!(events[1].nonce_source, nonce_source); + assert!(events[1].capsule_id.is_none()); + } + + /// The sibling of the test above for `try_route_plugin_model`'s own + /// effective/terminal pair (`RawProxy`) -- the pattern the remote-mesh + /// branch mirrors, previously uncovered at this level for the same + /// "not economical to invoke directly" reason. No marker exists on this + /// path (it never runs through `openai-frontend`'s `OpenAiHookPolicy`), + /// so nonce/nonce_source/capsule_id all stay absent on both envelopes. + #[tokio::test] + async fn raw_proxy_plugin_route_publishes_effective_and_terminal_without_a_marker() { + let channel = RecordingChannel::default(); + + channel + .publish(&OpenAiExchangeEnvelope::effective( + "exch-rp-1", + OpenAiExchangeDispatchPath::RawProxy, + "acme/plugin-model", + )) + .await; + channel + .publish(&OpenAiExchangeEnvelope::terminal( + "exch-rp-1", + OpenAiExchangeDispatchPath::RawProxy, + "acme/plugin-model", + Some(200), + None, + None, + )) + .await; + + let events = channel.events.lock().unwrap(); + assert_eq!(events.len(), 2, "one effective-request, one terminal"); + + assert_eq!( + events[0].dispatch_path, + OpenAiExchangeDispatchPath::RawProxy + ); + assert_eq!(events[0].phase, OpenAiExchangePhase::EffectiveRequest); + assert!(events[0].nonce.is_none()); + assert!(events[0].capsule_id.is_none()); + + assert_eq!( + events[1].dispatch_path, + OpenAiExchangeDispatchPath::RawProxy + ); + assert_eq!(events[1].phase, OpenAiExchangePhase::Terminal); + assert_eq!(events[1].exchange_id, events[0].exchange_id); + assert_eq!(events[1].status, Some(200)); + assert!(events[1].nonce.is_none()); + assert!(events[1].nonce_source.is_none()); + assert!(events[1].capsule_id.is_none()); + } } diff --git a/crates/openai-frontend/README.md b/crates/openai-frontend/README.md index 733d36a316..5fbea171b1 100644 --- a/crates/openai-frontend/README.md +++ b/crates/openai-frontend/README.md @@ -52,6 +52,7 @@ For the concrete benchy command and contract, see | Client nonce | Supported | Accepts `x-capsule-client-nonce` only when it is exactly one valid UUIDv4; a missing, invalid, non-UUIDv4, or duplicated value is replaced with a freshly minted UUIDv4. When this frontend mints the value it stamps `x-capsule-nonce-origin: frontend`; a forwarded (client-supplied) nonce carries no origin marker, and any inbound `x-capsule-nonce-origin` is always stripped so a caller cannot forge it. Both headers are echoed on covered responses: the axum router and, via the host runtime's forwarding rebuild, the public `:9337` proxy JSON/SSE paths (including the pipeline/MoA strong-model path and remapped upstream error responses). Locally synthesized error responses (e.g. no-target `503`s and `/v1/models` listings) do not yet carry the headers; threading the request nonce onto those senders is a cross-cutting signature change tracked as a follow-up. The origin marker asserts only that *this* frontend minted the value, not that it is the original ingress for a remote-routed request. | | Backend timeout | Supported | Configurable via `OpenAiFrontendConfig` or the `MESH_OPENAI_BACKEND_TIMEOUT_SECS` environment variable; defaults to 600 seconds (`0` disables it) and maps timeouts to OpenAI-shaped 504 errors. | | Agent session header | Supported | Set `MESH_AGENT_SESSION_HEADER` to accept a trusted upstream header as the stable agent-session identity. | +| Mesh routing headers | Supported (remote-mesh routing only) | Implemented by the host runtime's public `:9337` proxy, not this crate, but documented here as part of the same header contract. `x-mesh-target: ` forces dispatch to exactly that peer if it currently advertises the requested model; otherwise the request fails closed with a `409` naming the mismatch — it is never silently rerouted to another peer or served locally. `x-mesh-exclude: [,...]` removes one or more peers from the candidate set before selection (comma-separated within one header, or repeated). The routing node echoes the resolved peer back as `x-mesh-served-by: ` on the response when `x-mesh-target` resolves to a remote peer or a locally-served (non-plugin) model; plugin-served self-targets do not yet echo this header. Both request headers are no-ops when absent, and an unparseable value is a `400`, never a silently ignored one. | | embeddings/rerank/infill/audio/vision | Out of scope | Not needed for staged text benchmark entrypoints. | ## Shape