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..75508b3d37 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, @@ -40,6 +59,16 @@ struct IngressRouteContext<'a> { targets: &'a election::ModelTargets, affinity: &'a affinity::AffinityRouter, plugin_manager: Option<&'a crate::plugin::PluginManager>, + /// Explicit exchange channel for the remote-mesh publish pair. + /// When `None`, falls back to `plugin_manager` as the channel + /// (production path). Set to `Some` in tests to inject a recording double, + /// because the production call sites set `plugin_manager: None` in tests + /// and the publish calls are guarded by `if let Some(plugin_manager) = + /// ctx.plugin_manager` — making them invisible to a test that passes + /// `plugin_manager: None`. This field lets a test inject a channel without + /// requiring a live `PluginManager`. + #[cfg(test)] + exchange_channel: Option<&'a dyn OpenAiExchangeChannel>, } struct ProxyConnectionContext<'a> { @@ -523,7 +552,46 @@ async fn route_missing_local_model( ) -> proxy::RouteDispatchOutcome { // Try remote mesh first. if let Some(mesh_targets) = remote_mesh_targets(ctx, model_name).await { - return proxy::route_model_request( + // 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); + // In tests, `exchange_channel` may be injected directly so the publish + // pair is observable even when `plugin_manager` is `None`. In + // production (and in non-test builds) `plugin_manager` is the channel. + #[cfg(test)] + let channel: Option<&dyn OpenAiExchangeChannel> = ctx.exchange_channel.or_else(|| { + ctx.plugin_manager + .map(|pm| pm as &dyn OpenAiExchangeChannel) + }); + #[cfg(not(test))] + let channel: Option<&dyn OpenAiExchangeChannel> = ctx + .plugin_manager + .map(|pm| pm as &dyn OpenAiExchangeChannel); + if let Some(ch) = channel { + ch.publish(&OpenAiExchangeEnvelope::effective_remote_mesh( + exchange_id.clone(), + model_name, + forwarded_nonce.clone(), + nonce_source, + )) + .await; + } + let outcome = proxy::route_model_request( ctx.node.clone(), tcp_stream, &mesh_targets, @@ -536,6 +604,17 @@ async fn route_missing_local_model( }, ) .await; + if let Some(ch) = channel { + ch.publish(&OpenAiExchangeEnvelope::terminal_remote_mesh( + exchange_id, + model_name, + plugin_route_status(&outcome), + forwarded_nonce, + nonce_source, + )) + .await; + } + return outcome; } // Check if the model is known locally but unavailable @@ -1184,6 +1263,8 @@ async fn handle_api_proxy_connection( targets: &targets, affinity: &affinity, plugin_manager: plugin_manager.as_ref(), + #[cfg(test)] + exchange_channel: None, }; handle_buffered_api_request( tcp_stream, 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..d5275b2da7 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 @@ -1,13 +1,29 @@ -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use crate::logging::{ LoggingService, OpenAiLifecycleAttachment, RawMeshLifecycleOwners, RawMeshRequestLifecycle, TerminalOutcome, }; +use crate::plugin::openai_exchange::{OpenAiExchangeChannel, OpenAiExchangeEnvelope}; +use async_trait::async_trait; use mesh_llm_events::logging::{events::LifecycleEvent, identifiers::RequestId}; use super::*; +/// Recording double for `OpenAiExchangeChannel` — accumulates every published +/// envelope so tests can assert on count, dispatch path, exchange id, and nonce. +#[derive(Default)] +struct RecordingChannel { + events: Mutex>, +} + +#[async_trait] +impl OpenAiExchangeChannel for RecordingChannel { + async fn publish(&self, event: &OpenAiExchangeEnvelope) { + self.events.lock().unwrap().push(event.clone()); + } +} + fn large_tokenize_request(model: &str) -> proxy::BufferedHttpRequest { proxy::BufferedHttpRequest { raw: b"unchanged tokenizer wire".to_vec(), @@ -245,6 +261,7 @@ async fn moa_single_worker_stays_in_gateway() { targets: &targets, affinity: &affinity, plugin_manager: None, + exchange_channel: None, }, }; let lifecycle = OpenAiLifecycleAttachment::unowned(); @@ -484,6 +501,7 @@ async fn api_proxy_tokenizer_route_ignores_generation_context_budget() { targets: &targets, affinity: &affinity, plugin_manager: None, + exchange_channel: None, }; let raw_before_decision = request.raw.clone(); @@ -909,3 +927,397 @@ fn disconnect_is_dropped_and_cannot_audit_model_access_as_success() { proxy::RouteDispatchOutcome::Responded(200) )); } + +// --- #1668 round-2: call-site test for route_missing_local_model --- + +/// Seed a `mesh::Node` with one admitted `Host` peer serving `model` at a +/// fake address. `hosts_for_model` returns the peer's `EndpointId` without +/// any gossip round trip, so `remote_mesh_targets` returns `Some` and +/// `route_missing_local_model` takes the remote-mesh branch. +fn test_remote_peer(seed: u32, model: &str) -> mesh::PeerInfo { + // Deterministic fake id derived from seed so callers can build multiple + // non-colliding peers. + let secret = { + let mut bytes = [0u8; 32]; + let seed_bytes = seed.to_le_bytes(); + bytes[..4].copy_from_slice(&seed_bytes); + bytes[4] = 0xde; + bytes[5] = 0xad; + iroh::SecretKey::from(bytes) + }; + let peer_id = iroh::EndpointId::from(secret.public()); + mesh::PeerInfo { + id: peer_id, + addr: iroh::EndpointAddr { + id: peer_id, + addrs: Default::default(), + }, + mesh_id: None, + mesh_policy_hash: None, + genesis_policy: None, + // Host role is required for `accepts_http_inference()` and + // therefore for `routes_http_model()` and `hosts_for_model()`. + role: mesh::NodeRole::Host { http_port: 9337 }, + first_joined_mesh_ts: None, + models: vec![model.to_string()], + vram_bytes: 16 * 1024 * 1024 * 1024, + rtt_ms: None, + model_source: None, + // admitted: true required for `is_admitted()`. + 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: crate::crypto::OwnershipSummary::default(), + inference_admission_state: None, + } +} + +/// Verifies that `route_missing_local_model` enters the remote-mesh branch +/// when at least one admitted peer serves the requested model, AND that the +/// function publishes BOTH the effective-request and terminal envelopes on +/// that branch via `IngressRouteContext::exchange_channel`. +/// +/// This test was previously defective (erlich review): it passed +/// `plugin_manager: None`, so the two publish calls — both guarded by +/// `if let Some(plugin_manager) = ctx.plugin_manager` — were never executed, +/// and the test could not observe the publish pair it claimed to prove. The +/// fix adds a `#[cfg(test)]` `exchange_channel` field to `IngressRouteContext` +/// that accepts a recording double injected here without needing a live +/// `PluginManager`. +/// +/// erlich's four required assertions: +/// (1) TWO messages published (effective + terminal) +/// (2) `dispatch_path: RemoteMesh` on both envelopes +/// (3) matching `exchange_id` across the pair +/// (4) the SAME nonce on both (matches the nonce in the forwarded request) +/// +/// Failure proof: if either publish call in `route_missing_local_model` is +/// deleted, `events.len()` drops to 1 and assertion (1) fails. Run: +/// `cargo test -p mesh-llm-host-runtime route_missing_local_model 2>&1` +/// with one publish removed to confirm the test catches the defect. +#[tokio::test] +async fn route_missing_local_model_enters_remote_mesh_branch_when_peer_serves_model() { + use crate::plugin::openai_exchange::{OpenAiExchangeDispatchPath, OpenAiExchangePhase}; + + let model = "acme/remote-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(crate::mesh::NodeRole::Worker) + .await + .expect("test node"); + node.insert_test_peer(test_remote_peer(1, model)).await; + + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + + // Recording double — accumulates every envelope the publish calls emit. + let recording = RecordingChannel::default(); + + // Loopback TCP pair — the server side becomes the ClientStream. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback listener"); + let addr = listener.local_addr().expect("local addr"); + let client_connect = tokio::net::TcpStream::connect(addr); + let server_accept = async { listener.accept().await.map(|(s, _)| s) }; + let (_client_side, server_side) = tokio::join!(client_connect, server_accept); + let tcp_stream = server_side.expect("accept server side"); + + // Build a minimal chat-completion request stamped with a client nonce so + // `capsule_nonce_headers()` returns `Some`. The nonce must survive into + // both published envelopes unchanged (assertion 4). + let body = + br#"{"model":"acme/remote-model:Q4_K_M","messages":[{"role":"user","content":"hi"}]}"#; + let nonce = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"; + let raw = format!( + "POST /v1/chat/completions HTTP/1.1\r\nHost: t\r\nContent-Type: application/json\r\nContent-Length: {len}\r\nx-capsule-client-nonce: {nonce}\r\n\r\n", + len = body.len(), + nonce = nonce, + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect::>(); + 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.len(), + completion_tokens: None, + stream: None, + model_name: Some(model.to_owned()), + request_object_request_ids: Vec::new(), + response_adapter: proxy::ResponseAdapter::OpenAiChatCompletionsJson, + correlation_id: None, + }; + + // Confirm the nonce header round-trips through the raw bytes before the + // function reads it — a prerequisite for assertion (4). + let (parsed_nonce, _origin) = request.capsule_nonce_headers(); + assert_eq!( + parsed_nonce.as_deref(), + Some(nonce), + "nonce header must be readable from the raw request bytes" + ); + + let ctx = IngressRouteContext { + node: &node, + targets: &targets, + affinity: &affinity, + plugin_manager: None, + // Inject the recording double so both publish calls are observable + // even though plugin_manager is None. + exchange_channel: Some(&recording), + }; + let lifecycle = OpenAiLifecycleAttachment::unowned(); + + let outcome = route_missing_local_model( + tcp_stream.into(), + &request, + &ctx, + model, + None, + lifecycle.route_observer(), + ) + .await; + + // A 404 would mean remote_mesh_targets returned None (peer not seen). + // Any other outcome — Failed, Dropped, or a non-404 status — proves the + // remote-mesh branch was entered, which is what this test pins. + assert!( + !matches!(outcome, proxy::RouteDispatchOutcome::Responded(404)), + "expected remote-mesh branch (not 404), got {outcome:?} — \ + the test peer may not be visible to hosts_for_model()" + ); + + let events = recording.events.lock().unwrap(); + + // (1) TWO messages published — effective + terminal. + // If either publish call is deleted this assertion is the first to fail. + assert_eq!( + events.len(), + 2, + "route_missing_local_model must publish both the effective-request \ + and terminal envelopes on the remote-mesh branch; got {} event(s)", + events.len() + ); + + // (2) Both envelopes carry `RemoteMesh` as the dispatch path. + assert_eq!( + events[0].dispatch_path, + OpenAiExchangeDispatchPath::RemoteMesh, + "effective envelope must carry RemoteMesh dispatch path" + ); + assert_eq!( + events[1].dispatch_path, + OpenAiExchangeDispatchPath::RemoteMesh, + "terminal envelope must carry RemoteMesh dispatch path" + ); + + // Sanity: correct phases. + assert_eq!(events[0].phase, OpenAiExchangePhase::EffectiveRequest); + assert_eq!(events[1].phase, OpenAiExchangePhase::Terminal); + + // (3) Matching exchange_id across the pair. + assert!( + !events[0].exchange_id.is_empty(), + "exchange_id must be non-empty" + ); + assert_eq!( + events[0].exchange_id, events[1].exchange_id, + "effective and terminal envelopes must share the same exchange_id" + ); + + // (4) The SAME nonce on both envelopes, matching the request's nonce header. + assert_eq!( + events[0].nonce.as_deref(), + Some(nonce), + "effective envelope must carry the forwarded client nonce" + ); + assert_eq!( + events[0].nonce, events[1].nonce, + "effective and terminal envelopes must carry the same nonce" + ); +} + +/// Verifies that `route_missing_local_model` sets `nonce_source = +/// Some(SidecarGeneratedFallback)` on both published envelopes when the +/// request carries BOTH `x-capsule-client-nonce` AND `x-capsule-nonce-origin`. +/// +/// The `remote_mesh_nonce_source` helper (ingress.rs lines 35-46) maps: +/// - nonce=Some, nonce_origin=None → `ClientSupplied` (covered by the +/// sibling test above) +/// - nonce=Some, nonce_origin=Some → `SidecarGeneratedFallback` ← this test +/// +/// This test exercises the second branch by stamping both headers on the raw +/// request, then asserting that every published envelope reports +/// `nonce_source == Some(SidecarGeneratedFallback)`. +#[tokio::test] +async fn route_missing_local_model_sidecar_generated_nonce_origin_sets_sidecar_fallback_source() { + use crate::plugin::openai_exchange::{ + ClientNonceSource, OpenAiExchangeDispatchPath, OpenAiExchangePhase, + }; + + let model = "acme/remote-model:Q4_K_M"; + let node = mesh::Node::new_for_tests(crate::mesh::NodeRole::Worker) + .await + .expect("test node"); + node.insert_test_peer(test_remote_peer(1, model)).await; + + let targets = election::ModelTargets::default(); + let affinity = affinity::AffinityRouter::new(); + + let recording = RecordingChannel::default(); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind loopback listener"); + let addr = listener.local_addr().expect("local addr"); + let client_connect = tokio::net::TcpStream::connect(addr); + let server_accept = async { listener.accept().await.map(|(s, _)| s) }; + let (_client_side, server_side) = tokio::join!(client_connect, server_accept); + let tcp_stream = server_side.expect("accept server side"); + + // Build a request stamped with BOTH x-capsule-client-nonce AND + // x-capsule-nonce-origin. The presence of x-capsule-nonce-origin signals + // that the frontend generated the nonce as a fallback (SidecarGeneratedFallback). + let body = + br#"{"model":"acme/remote-model:Q4_K_M","messages":[{"role":"user","content":"hi"}]}"#; + let nonce = "b2c3d4e5-f6a7-8901-bcde-f12345678901"; + let raw = format!( + "POST /v1/chat/completions HTTP/1.1\r\nHost: t\r\nContent-Type: application/json\r\nContent-Length: {len}\r\nx-capsule-client-nonce: {nonce}\r\nx-capsule-nonce-origin: frontend\r\n\r\n", + len = body.len(), + nonce = nonce, + ) + .into_bytes() + .into_iter() + .chain(body.iter().copied()) + .collect::>(); + 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.len(), + completion_tokens: None, + stream: None, + model_name: Some(model.to_owned()), + request_object_request_ids: Vec::new(), + response_adapter: proxy::ResponseAdapter::OpenAiChatCompletionsJson, + correlation_id: None, + }; + + // Confirm both headers are readable before the routing function runs. + let (parsed_nonce, parsed_origin) = request.capsule_nonce_headers(); + assert_eq!( + parsed_nonce.as_deref(), + Some(nonce), + "nonce header must be readable from the raw request bytes" + ); + assert!( + parsed_origin.is_some(), + "nonce-origin header must be readable from the raw request bytes" + ); + + let ctx = IngressRouteContext { + node: &node, + targets: &targets, + affinity: &affinity, + plugin_manager: None, + exchange_channel: Some(&recording), + }; + let lifecycle = OpenAiLifecycleAttachment::unowned(); + + let outcome = route_missing_local_model( + tcp_stream.into(), + &request, + &ctx, + model, + None, + lifecycle.route_observer(), + ) + .await; + + assert!( + !matches!(outcome, proxy::RouteDispatchOutcome::Responded(404)), + "expected remote-mesh branch (not 404), got {outcome:?}" + ); + + let events = recording.events.lock().unwrap(); + + assert_eq!( + events.len(), + 2, + "route_missing_local_model must publish both the effective-request \ + and terminal envelopes on the remote-mesh branch; got {} event(s)", + events.len() + ); + + assert_eq!( + events[0].dispatch_path, + OpenAiExchangeDispatchPath::RemoteMesh, + "effective envelope must carry RemoteMesh dispatch path" + ); + assert_eq!( + events[1].dispatch_path, + OpenAiExchangeDispatchPath::RemoteMesh, + "terminal envelope must carry RemoteMesh dispatch path" + ); + + assert_eq!(events[0].phase, OpenAiExchangePhase::EffectiveRequest); + assert_eq!(events[1].phase, OpenAiExchangePhase::Terminal); + + // Both envelopes must report SidecarGeneratedFallback because + // x-capsule-nonce-origin was present on the request. + assert_eq!( + events[0].nonce_source, + Some(ClientNonceSource::SidecarGeneratedFallback), + "effective envelope must carry SidecarGeneratedFallback nonce_source \ + when x-capsule-nonce-origin is present" + ); + assert_eq!( + events[1].nonce_source, + Some(ClientNonceSource::SidecarGeneratedFallback), + "terminal envelope must carry SidecarGeneratedFallback nonce_source \ + when x-capsule-nonce-origin is present" + ); +} 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..f9744c741a 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`] @@ -84,6 +90,19 @@ pub struct OpenAiExchangeEnvelope { pub nonce: Option, /// Which side contributed `nonce` — see [`ClientNonceSource`]. `None` /// exactly when `nonce` is `None` (no marker minted). + /// + /// **Asymmetry across routing-node pairs:** when node A minted the + /// fallback nonce (the client sent none), A reads its own + /// `x-capsule-nonce-origin` header and reports + /// `SidecarGeneratedFallback`. Node B strips that header deliberately + /// (anti-smuggling, `request_parse.rs:582`) so it sees a + /// well-formed nonce with no origin marker and reports `ClientSupplied` + /// for the same nonce. Both are locally correct: A reports what it + /// minted; B cannot trust the origin claim. A consumer joining both + /// halves on the same nonce will observe two different `nonce_source` + /// values — this is NOT a bug. Use the routing node's own envelope to + /// judge whether the nonce was client-supplied or sidecar-generated; + /// do not compare across nodes. #[serde(skip_serializing_if = "Option::is_none")] pub nonce_source: Option, } @@ -125,6 +144,74 @@ 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. + /// + /// **`nonce_source` asymmetry:** when the routing node (node A) minted + /// the fallback nonce, it reports `SidecarGeneratedFallback` here. + /// The receiving peer (node B) strips the `x-capsule-nonce-origin` + /// header (anti-smuggling) and therefore reports `ClientSupplied` for + /// the same nonce on its own envelope. Both are locally correct; a + /// consumer joining both envelopes will see two different `nonce_source` + /// values for the same nonce — see the field-level doc on + /// [`OpenAiExchangeEnvelope::nonce_source`] for the full explanation. + 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. + /// + /// **`nonce_source` asymmetry:** same as [`Self::effective_remote_mesh`] + /// — node A reports `SidecarGeneratedFallback` when it minted the nonce; + /// node B strips the origin header (anti-smuggling) and reports + /// `ClientSupplied` for the identical nonce. See + /// [`OpenAiExchangeEnvelope::nonce_source`] for the full explanation. + 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 +651,121 @@ mod tests { "slow exchange's terminal event pairs with its own effective event" ); } + + // --- #1668 review round: RemoteMesh / RawProxy envelope shapes --- + // + // Shape tests only — these call envelope constructors directly and assert + // on the fields they set. They do NOT invoke `route_missing_local_model` + // or `try_route_plugin_model`, so they would still pass if the publish + // calls inside those routing functions were deleted. Real end-to-end + // publish coverage (including both envelopes being emitted and their + // nonce_source values) lives in `ingress_tests::tests`. + + /// Verifies the envelope constructor shape for the RemoteMesh effective + + /// terminal pair: both envelopes carry `RemoteMesh` dispatch path, the + /// nonce and nonce_source are threaded onto both, and `capsule_id` is + /// absent on both (this node mints nothing on the remote-mesh path). + /// + /// // Shape test only — does not invoke the routing function. + /// // Real publish coverage is in ingress_tests::tests. + #[tokio::test] + async fn envelope_shape_remote_mesh_effective_and_terminal_carry_nonce_fields() { + 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()); + } + + /// Verifies the envelope constructor shape for the RawProxy effective + + /// terminal pair: both envelopes carry `RawProxy` dispatch path, and + /// nonce/nonce_source/capsule_id are all absent (the raw-proxy path never + /// runs through `openai-frontend`'s `OpenAiHookPolicy`, so no marker is + /// minted). + /// + /// // Shape test only — does not invoke the routing function. + /// // Real publish coverage is in ingress_tests::tests. + #[tokio::test] + async fn envelope_shape_raw_proxy_effective_and_terminal_have_no_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()); + } }