diff --git a/crates/cli/src/configuration/mod.rs b/crates/cli/src/configuration/mod.rs index 8b0e06995..6840dcbb8 100644 --- a/crates/cli/src/configuration/mod.rs +++ b/crates/cli/src/configuration/mod.rs @@ -75,6 +75,7 @@ struct FileUpstreamConfig { openai_auth_header: Option, anthropic_base_url: Option, anthropic_auth_header: Option, + caller_credential_targets: Option>, } #[derive(Debug, Clone, Default, Deserialize)] @@ -297,6 +298,7 @@ fn persistent_bootstrap_fingerprint( "openai_auth_header": gateway.openai_auth_header, "anthropic_base_url": gateway.anthropic_base_url, "anthropic_auth_header": gateway.anthropic_auth_header, + "caller_credential_targets": gateway.caller_credential_targets, "metadata": gateway.metadata, "plugin_config": gateway.plugin_config, "max_hook_payload_bytes": gateway.max_hook_payload_bytes, @@ -1397,7 +1399,24 @@ fn apply_file_upstream_config( openai_auth_header, anthropic_base_url, anthropic_auth_header, + caller_credential_targets, } = upstream; + if let Some(targets) = caller_credential_targets { + for (name, target) in &targets { + let valid_url = reqwest::Url::parse(&target.url).ok().is_some_and(|url| { + matches!(url.scheme(), "http" | "https") + && url.host_str().is_some() + && url.username().is_empty() + && url.password().is_none() + && url.fragment().is_none() + }); + if name.trim().is_empty() || !valid_url { + return Err(CliError::Config("caller_credential_targets requires nonempty names and absolute HTTP(S) endpoint URLs without userinfo or fragments".into())); + } + } + // Replace as a policy unit: layering must not retain permissions removed by an override. + gateway.caller_credential_targets = targets; + } if let Some(value) = openai_base_url { gateway.openai_base_url = value; if openai_auth_header.is_none() { diff --git a/crates/cli/src/configuration/types.rs b/crates/cli/src/configuration/types.rs index 851bde012..cddcb490f 100644 --- a/crates/cli/src/configuration/types.rs +++ b/crates/cli/src/configuration/types.rs @@ -3,12 +3,14 @@ //! Resolved runtime configuration model. +use std::collections::BTreeMap; use std::net::SocketAddr; use std::path::PathBuf; use axum::http::HeaderMap; +use nemo_relay::api::runtime::provider::LlmProviderFormat; use nemo_relay::logging::LoggingConfig; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; use strum::{Display, IntoStaticStr}; @@ -18,6 +20,13 @@ use super::{ DEFAULT_MAX_HOOK_PAYLOAD_BYTES, DEFAULT_MAX_PASSTHROUGH_BODY_BYTES, header_json, header_string, }; +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct CallerCredentialTarget { + pub(crate) url: String, + pub(crate) format: LlmProviderFormat, +} + #[derive(Debug, Clone)] pub(crate) struct GatewayConfig { pub(crate) bind: SocketAddr, @@ -25,6 +34,7 @@ pub(crate) struct GatewayConfig { pub(crate) openai_auth_header: Option, pub(crate) anthropic_base_url: String, pub(crate) anthropic_auth_header: Option, + pub(crate) caller_credential_targets: BTreeMap, pub(crate) metadata: Option, pub(crate) plugin_config: Option, pub(crate) max_hook_payload_bytes: usize, @@ -115,6 +125,7 @@ impl Default for GatewayConfig { openai_auth_header: None, anthropic_base_url: "https://api.anthropic.com".into(), anthropic_auth_header: None, + caller_credential_targets: BTreeMap::new(), metadata: None, plugin_config: None, max_hook_payload_bytes: DEFAULT_MAX_HOOK_PAYLOAD_BYTES, diff --git a/crates/cli/src/gateway/mod.rs b/crates/cli/src/gateway/mod.rs index 4fa2a3aa4..c953120e8 100644 --- a/crates/cli/src/gateway/mod.rs +++ b/crates/cli/src/gateway/mod.rs @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 pub(crate) mod client; +mod provider; mod request; mod response; mod routes; @@ -408,6 +409,7 @@ async fn run_managed_buffered( codecs: RouteCodecs, operational: OperationalContext, ) -> Result, CliError> { + let dispatcher = provider::dispatcher(&state, &prepared); let upstream_failures = Arc::new(CapturedUpstreamFailures::default()); let func = build_buffered_func( state.clone(), @@ -441,7 +443,13 @@ async fn run_managed_buffered( .response_codec_opt(codecs.response) .build(); let result = TASK_SCOPE_STACK - .scope(scope_stack, async move { llm_call_execute(params).await }) + .scope( + scope_stack, + nemo_relay::api::runtime::provider::with_llm_provider_dispatcher( + dispatcher, + async move { llm_call_execute(params).await }, + ), + ) .await; match result { Ok(response_json) => { @@ -590,6 +598,7 @@ async fn run_managed_streaming( codecs: RouteCodecs, operational: OperationalContext, ) -> Result, CliError> { + let dispatcher = provider::dispatcher(&state, &prepared); let upstream_failures = Arc::new(CapturedUpstreamFailures::default()); let func = build_streaming_func( state.clone(), @@ -651,7 +660,10 @@ async fn run_managed_streaming( let json_stream_result = TASK_SCOPE_STACK .scope( scope_stack, - async move { llm_stream_call_execute(params).await }, + nemo_relay::api::runtime::provider::with_llm_provider_dispatcher( + dispatcher, + async move { llm_stream_call_execute(params).await }, + ), ) .await; let json_stream = match json_stream_result { diff --git a/crates/cli/src/gateway/provider.rs b/crates/cli/src/gateway/provider.rs new file mode 100644 index 000000000..a366efa1b --- /dev/null +++ b/crates/cli/src/gateway/provider.rs @@ -0,0 +1,233 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Host-owned provider transport. Credentials never cross the plugin ABI. + +use nemo_relay::api::runtime::provider::{ + LlmProviderDispatcher, LlmProviderFormat, LlmProviderRequest, +}; +use nemo_relay::codec::streaming::SseEventDecoder; + +use super::*; +use crate::configuration::CallerCredentialTarget; + +struct ProviderTransport { + client: reqwest::Client, + targets: BTreeMap, + source: ProviderRoute, + headers: HeaderMap, + credential_present: bool, + response_limit: usize, +} + +pub(super) fn dispatcher( + state: &AppState, + prepared: &PreparedGatewayRequest, +) -> LlmProviderDispatcher { + let transport = Arc::new(ProviderTransport { + client: state.http_no_redirect.clone(), + targets: state.config.caller_credential_targets.clone(), + source: prepared.provider, + headers: prepared.headers.clone(), + credential_present: prepared + .authorization + .source_credential + .provider_credential_present(), + response_limit: state.config.max_passthrough_body_bytes, + }); + let buffered = transport.clone(); + LlmProviderDispatcher::new( + Arc::new(move |request| { + let transport = buffered.clone(); + Box::pin(async move { transport.buffered(request).await }) + }), + Arc::new(move |request| { + let transport = transport.clone(); + Box::pin(async move { transport.streaming(request).await }) + }), + ) +} + +impl ProviderTransport { + fn headers_for(&self, target: &CallerCredentialTarget) -> Result { + let openai = matches!( + self.source, + ProviderRoute::OpenAiChatCompletions | ProviderRoute::OpenAiResponses + ); + let matching_family = match target.format { + LlmProviderFormat::OpenaiChat | LlmProviderFormat::OpenaiResponses => openai, + LlmProviderFormat::AnthropicMessages => { + matches!(self.source, ProviderRoute::AnthropicMessages) + } + }; + if !matching_family { + return Err(FlowError::InvalidArgument( + "provider target belongs to a different credential family".into(), + )); + } + let credential_names: &[&str] = if openai { + &["authorization", "api-key", "x-api-key"] + } else { + &["authorization", "x-api-key", "anthropic-api-key", "api-key"] + }; + let mut headers = HeaderMap::new(); + for name in credential_names { + if let Some(value) = self.headers.get(*name).filter(|value| !value.is_empty()) { + let mut value = value.clone(); + value.set_sensitive(true); + headers.insert(HeaderName::from_static(name), value); + } + } + if !self.credential_present || headers.is_empty() { + return Err(FlowError::InvalidArgument( + "private provider dispatch requires a caller provider credential".into(), + )); + } + let companion_names: &[&str] = if openai { + &["chatgpt-account-id", "x-openai-fedramp"] + } else { + &["anthropic-version", "anthropic-beta"] + }; + for name in companion_names { + if let Some(value) = self.headers.get(*name) { + headers.insert(HeaderName::from_static(name), value.clone()); + } + } + Ok(headers) + } + + async fn send( + &self, + mut request: LlmProviderRequest, + streaming: bool, + ) -> Result { + let target = self.targets.get(&request.target).ok_or_else(|| { + FlowError::InvalidArgument( + "provider target is not authorized by caller_credential_targets".into(), + ) + })?; + let headers = self.headers_for(target)?; + let content = request.content.as_object_mut().ok_or_else(|| { + FlowError::InvalidArgument("provider request content must be an object".into()) + })?; + content.insert("stream".into(), Value::Bool(streaming)); + let response = self + .client + .post(&target.url) + .headers(headers) + .json(&request.content) + .send() + .await + .map_err(|error| { + safe_failure( + None, + if error.is_timeout() { + UpstreamFailureClass::Timeout + } else { + UpstreamFailureClass::Connection + }, + ) + })?; + if !response.status().is_success() { + let status = response.status().as_u16(); + // Never return provider error bodies, redirect locations, or transport URLs: they + // can echo credentials. Status retains enough information for plugin retry policy. + let class = match status { + 401 | 403 => UpstreamFailureClass::Authentication, + 408 | 429 | 500..=599 => UpstreamFailureClass::RetryableStatus, + _ => UpstreamFailureClass::InvalidRequest, + }; + return Err(safe_failure(Some(status), class)); + } + Ok(response) + } + + async fn buffered(&self, request: LlmProviderRequest) -> Result { + let mut response = self.send(request, false).await?; + let mut bytes = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(|_| malformed_response())? { + if chunk.len() > self.response_limit.saturating_sub(bytes.len()) { + return Err(FlowError::InvalidArgument( + "provider response exceeds gateway body limit".into(), + )); + } + bytes.extend_from_slice(&chunk); + } + let mut value = serde_json::from_slice(&bytes).map_err(|_| malformed_response())?; + self.redact(&mut value); + Ok(value) + } + + async fn streaming( + self: Arc, + request: LlmProviderRequest, + ) -> Result { + let response = self.send(request, true).await?; + let mut bytes = response.bytes_stream(); + let mut decoder = SseEventDecoder::new(); + Ok(LlmJsonStream::new(stream! { + while let Some(chunk) = bytes.next().await { + let Ok(chunk) = chunk else { + yield Err(malformed_response()); + return; + }; + for result in decoder.push_bytes_results(&chunk) { + match result { + Ok(mut event) => { self.redact(&mut event.data); yield Ok(event.data); } + Err(_) => { yield Err(malformed_response()); return; } + } + } + } + match decoder.finish() { + Ok(Some(mut event)) => { self.redact(&mut event.data); yield Ok(event.data); } + Ok(None) => {} + Err(_) => yield Err(malformed_response()), + } + })) + } + + // Defense in depth for providers that echo header values in successful JSON or SSE data. + // Configured endpoints remain trusted recipients; this is not a sandbox for a malicious peer. + fn redact(&self, value: &mut Value) { + match value { + Value::String(text) => { + for name in ["authorization", "x-api-key", "api-key", "anthropic-api-key"] { + if let Some(secret) = self.headers.get(name).and_then(|v| v.to_str().ok()) { + let secret = secret.strip_prefix("Bearer ").unwrap_or(secret); + if !secret.is_empty() { + *text = text.replace(secret, "[REDACTED]"); + } + } + } + } + Value::Array(values) => values.iter_mut().for_each(|value| self.redact(value)), + Value::Object(values) => { + let original = std::mem::take(values); + for (key, mut value) in original { + let mut key = Value::String(key); + self.redact(&mut key); + self.redact(&mut value); + values.insert(key.as_str().expect("string key").to_owned(), value); + } + } + _ => {} + } + } +} + +fn malformed_response() -> FlowError { + FlowError::Internal("provider returned an unreadable response".into()) +} + +fn safe_failure(status: Option, class: UpstreamFailureClass) -> FlowError { + FlowError::Upstream(UpstreamFailure { + status, + body: "private provider call failed".into(), + headers: BTreeMap::new(), + class, + }) +} + +#[cfg(test)] +#[path = "../../tests/coverage/shared/private_provider_tests.rs"] +mod tests; diff --git a/crates/cli/tests/coverage/shared/config_tests.rs b/crates/cli/tests/coverage/shared/config_tests.rs index 437905018..927b26d8d 100644 --- a/crates/cli/tests/coverage/shared/config_tests.rs +++ b/crates/cli/tests/coverage/shared/config_tests.rs @@ -589,6 +589,7 @@ manifest = "plugins/acme/relay-plugin.toml" fn config() -> GatewayConfig { GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://openai".into(), openai_auth_header: None, @@ -4686,3 +4687,69 @@ fn dynamic_plugin_identity_allows_worker_without_manifest() { assert_eq!(identity["manifest"], Value::Null); assert_eq!(identity["lifecycle_generation"], 7); } + +#[test] +fn caller_credential_target_policy_is_explicit_validated_and_replaced() { + let mut gateway = GatewayConfig::default(); + assert!(gateway.caller_credential_targets.is_empty()); + let upstream: FileUpstreamConfig = toml::from_str( + r#" +[caller_credential_targets.answer] +url = "https://example.com/v1/responses?api-version=test" +format = "openai_responses" +"#, + ) + .unwrap(); + apply_file_upstream_config(&mut gateway, Some(upstream)).unwrap(); + assert_eq!(gateway.caller_credential_targets.len(), 1); + for url in [ + "/relative", + "ftp://example.com", + "https://secret@example.com", + "https://example.com/#fragment", + ] { + let raw = format!( + "[caller_credential_targets.invalid]\nurl = {url:?}\nformat = \"openai_chat\"\n" + ); + let config: FileUpstreamConfig = toml::from_str(&raw).unwrap(); + assert!(apply_file_upstream_config(&mut gateway, Some(config)).is_err()); + assert_eq!(gateway.caller_credential_targets.len(), 1); + } + let empty: FileUpstreamConfig = toml::from_str("caller_credential_targets = {}\n").unwrap(); + apply_file_upstream_config(&mut gateway, Some(empty)).unwrap(); + assert!(gateway.caller_credential_targets.is_empty()); +} + +#[test] +fn caller_credential_target_policy_changes_persistent_gateway_identity() { + let temp = tempfile::tempdir().unwrap(); + let xdg = temp.path().join("xdg"); + std::fs::create_dir_all(&xdg).unwrap(); + let _scope = PluginConfigDiscoveryScope::enter(temp.path(), &xdg); + let mut resolved = ResolvedConfig::default(); + let original = persistent_bootstrap_fingerprint(&resolved, &[]).unwrap(); + resolved.gateway.caller_credential_targets.insert( + "answer".into(), + CallerCredentialTarget { + url: "https://example.com/v1/responses".into(), + format: nemo_relay::api::runtime::provider::LlmProviderFormat::OpenaiResponses, + }, + ); + let authorized = persistent_bootstrap_fingerprint(&resolved, &[]).unwrap(); + assert_ne!(original, authorized); + resolved + .gateway + .caller_credential_targets + .get_mut("answer") + .unwrap() + .url = "https://other.example.com/v1/responses".into(); + assert_ne!( + authorized, + persistent_bootstrap_fingerprint(&resolved, &[]).unwrap() + ); + resolved.gateway.caller_credential_targets.clear(); + assert_eq!( + original, + persistent_bootstrap_fingerprint(&resolved, &[]).unwrap() + ); +} diff --git a/crates/cli/tests/coverage/shared/gateway_tests.rs b/crates/cli/tests/coverage/shared/gateway_tests.rs index f9a08cdd0..8298c00f0 100644 --- a/crates/cli/tests/coverage/shared/gateway_tests.rs +++ b/crates/cli/tests/coverage/shared/gateway_tests.rs @@ -495,6 +495,7 @@ fn provider_route_names_round_trip_through_alignment_routes() { #[test] fn provider_routes_preserve_path_query_and_choose_upstream() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://openai/v1/".into(), openai_auth_header: None, @@ -544,6 +545,7 @@ fn chatgpt_shaped_responses_path_is_a_responses_route() { #[test] fn openai_upstream_url_accepts_origin_or_v1_base() { let mut config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://openai".into(), openai_auth_header: None, @@ -578,6 +580,7 @@ fn openai_upstream_url_accepts_origin_or_v1_base() { #[test] fn anthropic_upstream_url_accepts_origin_or_v1_base() { let mut config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://openai".into(), openai_auth_header: None, @@ -2238,6 +2241,7 @@ fn chatgpt_backend_url_omits_v1_prefix() { #[tokio::test] async fn passthrough_rejects_unsupported_provider_path_directly() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://openai".into(), openai_auth_header: None, @@ -2277,6 +2281,7 @@ async fn passthrough_rejects_unsupported_provider_path_directly() { #[tokio::test] async fn models_rejects_non_get_requests_directly() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://openai".into(), openai_auth_header: None, @@ -2680,6 +2685,7 @@ fn a_refused_named_upstream_is_rejected_rather_than_rerouted() { #[tokio::test] async fn models_refuses_an_unusable_named_upstream() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), // Nothing must reach this. If the refusal fell back to configured routing, the request // would be sent here instead of failing. diff --git a/crates/cli/tests/coverage/shared/private_provider_tests.rs b/crates/cli/tests/coverage/shared/private_provider_tests.rs new file mode 100644 index 000000000..a40052997 --- /dev/null +++ b/crates/cli/tests/coverage/shared/private_provider_tests.rs @@ -0,0 +1,369 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use super::*; +use crate::configuration::GatewayConfig; +use axum::{Json, Router, routing::post}; +use nemo_relay::api::event::Event; +use nemo_relay::api::subscriber::{deregister_subscriber, flush_subscribers, register_subscriber}; +use nemo_relay::plugin::dynamic::{ + NativePluginLoadSpec, PluginHostActivation, load_native_plugins, +}; +use nemo_relay::plugin::{PluginComponentSpec, PluginConfig}; +use serde_json::json; + +type Captures = Arc>>; + +async fn capture(State(captures): State, request: Request) -> Response { + let (parts, body) = request.into_parts(); + let body: Value = + serde_json::from_slice(&axum::body::to_bytes(body, 1024 * 1024).await.unwrap()).unwrap(); + captures + .lock() + .unwrap() + .push((parts.uri.path().into(), parts.headers.clone(), body.clone())); + let path = parts.uri.path(); + if path == "/redirect" { + return Response::builder() + .status(307) + .header("location", "/stolen") + .body(Body::empty()) + .unwrap(); + } + if path == "/fail" { + return Response::builder() + .status(503) + .body(Body::from( + parts.headers["authorization"].to_str().unwrap().to_owned(), + )) + .unwrap(); + } + let model = body["model"].as_str().unwrap(); + let content = if path == "/echo" { + parts.headers["authorization"].to_str().unwrap().to_owned() + } else { + "ok".to_owned() + }; + let response = if path == "/responses" { + json!({"id":"resp_test","object":"response","status":"completed","model":model,"output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}) + } else if path == "/messages" { + json!({"id":"msg_test","type":"message","role":"assistant","model":model,"content":[{"type":"text","text":content}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}) + } else { + json!({"id":"chatcmpl_test","object":"chat.completion","created":1,"model":model,"choices":[{"index":0,"message":{"role":"assistant","content":content},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}) + }; + if body["stream"] == true { + let events = if path == "/responses" { + vec![json!({"type":"response.completed","response":response})] + } else if path == "/messages" { + vec![ + json!({"type":"message_start","message":response}), + json!({"type":"message_stop"}), + ] + } else { + vec![ + json!({"id":"chatcmpl_test","object":"chat.completion.chunk","created":1,"model":model,"choices":[{"index":0,"delta":{"role":"assistant","content":content},"finish_reason":"stop"}]}), + ] + }; + let text = events + .into_iter() + .map(|event| format!("data: {event}\n\n")) + .collect::(); + Response::builder() + .header("content-type", "text/event-stream") + .body(Body::from(text)) + .unwrap() + } else { + axum::response::IntoResponse::into_response(Json(response)) + } +} + +async fn upstream() -> (String, Captures, tokio::task::JoinHandle<()>) { + let captures = Arc::new(Mutex::new(Vec::new())); + let app = Router::new() + .route("/{*path}", post(capture)) + .with_state(captures.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (url, captures, server) +} + +fn config(url: &str) -> GatewayConfig { + // Ordinary forwarding would fail; success must come from the native capability. + let mut config = GatewayConfig { + openai_base_url: "http://127.0.0.1:1".into(), + anthropic_base_url: "http://127.0.0.1:1".into(), + ..Default::default() + }; + for (name, path, format) in [ + ("chat", "chat", LlmProviderFormat::OpenaiChat), + ("responses", "responses", LlmProviderFormat::OpenaiResponses), + ( + "anthropic", + "messages", + LlmProviderFormat::AnthropicMessages, + ), + ("redirect", "redirect", LlmProviderFormat::OpenaiChat), + ("fail", "fail", LlmProviderFormat::OpenaiChat), + ("echo", "echo", LlmProviderFormat::OpenaiChat), + ] { + config.caller_credential_targets.insert( + name.into(), + CallerCredentialTarget { + url: format!("{url}/{path}"), + format, + }, + ); + } + config +} + +async fn gateway_call( + state: AppState, + path: &str, + target: &str, + streaming: bool, + caller: usize, +) -> Result { + let payload = json!({"model":format!("caller-{caller}"),"messages":[{"role":"user","content":"hello"}],"input":"hello","max_tokens":8,"stream":streaming,"fixture_provider_targets": [target]}); + gateway_payload(state, path, payload, caller).await +} + +async fn gateway_payload( + state: AppState, + path: &str, + payload: Value, + caller: usize, +) -> Result { + let request = Request::builder() + .method("POST") + .uri(path) + .header("content-type", "application/json") + .header( + "authorization", + format!("Bearer synthetic-caller-secret-{caller}"), + ) + .header("chatgpt-account-id", format!("account-{caller}")) + .header("anthropic-version", "2023-06-01") + .header("cookie", "must-not-forward") + .body(Body::from(payload.to_string())) + .unwrap(); + let response = super::super::passthrough(State(state), request) + .await + .map_err(|e| e.to_string())?; + let bytes = axum::body::to_bytes(response.into_body(), 1024 * 1024) + .await + .map_err(|e| e.to_string())?; + Ok(String::from_utf8(bytes.to_vec()).unwrap()) +} + +// A real cdylib uses its own SDK Tokio runtime and crosses the C ABI in both directions. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn native_provider_dispatch_gateway_end_to_end() { + let library = std::env::var_os("NEMO_RELAY_TEST_NATIVE_PLUGIN") + .map(std::path::PathBuf::from) + .unwrap_or_else(|| { + let name = format!( + "{}nemo_relay_plugin_fixture{}", + std::env::consts::DLL_PREFIX, + std::env::consts::DLL_SUFFIX + ); + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../../target/test-plugin-fixtures/debug") + .join(name) + }); + assert!( + library.exists(), + "run just build-test-plugin-fixtures first: {}", + library.display() + ); + let temp = tempfile::tempdir().unwrap(); + let manifest = temp.path().join("relay-plugin.toml"); + let library = serde_json::to_string(&library.to_string_lossy()).unwrap(); + std::fs::write( + &manifest, + format!( + r#" +manifest_version = 1 +[plugin] +id = "fixture_native" +kind = "rust_dynamic" +[compat] +relay = ">=0.8.0,<1.0" +native_api = "1" +[defaults] +enabled = false +[capabilities] +items = ["plugin_native"] +[source] +artifact = {library} +[load] +library = {library} +symbol = "nemo_relay_fixture_native_plugin" +"# + ), + ) + .unwrap(); + let activation = load_native_plugins([NativePluginLoadSpec { + plugin_id: "fixture_native".into(), + manifest_ref: manifest.to_string_lossy().into(), + }]) + .unwrap(); + let mut plugins = PluginConfig::default(); + plugins.components.push(PluginComponentSpec { + kind: "fixture_native".into(), + enabled: true, + config: serde_json::Map::from_iter([("private_provider".into(), json!(true))]), + }); + let mut host = PluginHostActivation::initialize_exact(plugins) + .await + .unwrap(); + let events = Arc::new(Mutex::new(Vec::::new())); + let captured_events = events.clone(); + register_subscriber( + "private_provider_events", + Arc::new(move |event| captured_events.lock().unwrap().push(event.clone())), + ) + .unwrap(); + let (url, captures, server) = upstream().await; + let state = AppState::new(config(&url)); + let mut calls = Vec::new(); + for caller in 0..12 { + let state = state.clone(); + calls.push(tokio::spawn(async move { + let (path, target) = match caller % 3 { + 0 => ("/v1/chat/completions", "chat"), + 1 => ("/v1/responses", "responses"), + _ => ("/v1/messages", "anthropic"), + }; + gateway_call(state, path, target, caller % 2 == 0, caller) + .await + .unwrap() + })); + } + for (caller, call) in calls.into_iter().enumerate() { + let body = call.await.unwrap(); + assert!(body.contains(&format!("caller-{caller}")), "{body}"); + assert!(!body.contains("synthetic-caller-secret")); + assert!(!body.contains("nemo_relay_gateway_error"), "{body}"); + } + assert_eq!(captures.lock().unwrap().len(), 12); + for (_, headers, body) in captures.lock().unwrap().iter() { + let caller = body["model"] + .as_str() + .unwrap() + .strip_prefix("caller-") + .unwrap(); + assert_eq!( + headers["authorization"], + format!("Bearer synthetic-caller-secret-{caller}") + ); + assert!(!headers.contains_key("cookie")); + if caller.parse::().unwrap() % 3 != 2 { + assert_eq!(headers["chatgpt-account-id"], format!("account-{caller}")); + } + } + let before = captures.lock().unwrap().len(); + for target in ["unknown", "anthropic"] { + let error = gateway_call(state.clone(), "/v1/chat/completions", target, false, 50) + .await + .unwrap_err(); + assert!(!error.contains("synthetic-caller-secret")); + } + assert_eq!(captures.lock().unwrap().len(), before); + let error = gateway_call(state.clone(), "/v1/chat/completions", "redirect", false, 50) + .await + .unwrap_err(); + assert!(error.contains("307")); + assert!( + !captures + .lock() + .unwrap() + .iter() + .any(|(path, _, _)| path == "/stolen") + ); + let error = gateway_call(state.clone(), "/v1/chat/completions", "fail", false, 50) + .await + .unwrap_err(); + assert!(error.contains("503")); + assert!(!error.contains("synthetic-caller-secret")); + // A routing-model call, a repeated attempt, and a fallback all keep the same credential. + let payload = json!({"model":"caller-51","input":"hello","messages":[{"role":"user","content":"hello"}],"stream":true,"fixture_provider_probe":"chat","fixture_provider_targets":["fail","fail","responses"]}); + gateway_payload(state.clone(), "/v1/responses", payload, 51) + .await + .unwrap(); + let rows = captures.lock().unwrap().clone(); + assert_eq!( + rows.iter() + .filter(|(_, _, body)| body["model"] == "caller-51") + .count(), + 4 + ); + for (_, headers, body) in rows { + if body["model"] == "caller-51" { + assert_eq!( + headers["authorization"], + "Bearer synthetic-caller-secret-51" + ); + } + } + for streaming in [false, true] { + let body = gateway_call(state.clone(), "/v1/chat/completions", "echo", streaming, 52) + .await + .unwrap(); + assert!(!body.contains("synthetic-caller-secret-52")); + assert!(body.contains("[REDACTED]")); + } + flush_subscribers().unwrap(); + assert!(events.lock().unwrap().len() >= 24); + let event_json = serde_json::to_string(&*events.lock().unwrap()).unwrap(); + assert!(event_json.contains("caller-52")); + assert!(!event_json.contains("synthetic-caller-secret")); + assert!(!event_json.contains("must-not-forward")); + deregister_subscriber("private_provider_events").unwrap(); + host.close().unwrap(); + activation.clear(); + server.abort(); +} + +#[tokio::test] +async fn provider_dispatch_never_uses_invocation_or_deployment_credentials() { + let (url, captures, server) = upstream().await; + let mut cfg = config(&url); + cfg.openai_auth_header = Some("Bearer deployment-secret".into()); + let state = AppState::new(cfg); + let token = crate::provider_auth::TransparentProxyCredential::from_static("invocation-secret"); + for provider_header in [None, Some("api-key"), Some("x-api-key")] { + let mut headers = HeaderMap::new(); + headers.insert( + "authorization", + HeaderValue::from_static("Bearer invocation-secret"), + ); + if let Some(name) = provider_header { + headers.insert(name, HeaderValue::from_static("provider-secret")); + } + let source = token.consume(&mut headers).unwrap(); + let transport = ProviderTransport { + client: state.http_no_redirect.clone(), + targets: state.config.caller_credential_targets.clone(), + source: ProviderRoute::OpenAiChatCompletions, + headers, + credential_present: source.provider_credential_present(), + response_limit: 4096, + }; + let result = transport + .buffered(LlmProviderRequest { + target: "chat".into(), + content: json!({"model":"test"}), + }) + .await; + assert_eq!(result.is_ok(), provider_header.is_some()); + } + let captures = captures.lock().unwrap(); + assert_eq!(captures.len(), 2); + assert!(!captures[0].1.contains_key("authorization")); + assert_eq!(captures[0].1["api-key"], "provider-secret"); + assert!(!captures[1].1.contains_key("authorization")); + assert_eq!(captures[1].1["x-api-key"], "provider-secret"); + server.abort(); +} diff --git a/crates/cli/tests/coverage/shared/server_tests.rs b/crates/cli/tests/coverage/shared/server_tests.rs index 7cf007f06..af8abd731 100644 --- a/crates/cli/tests/coverage/shared/server_tests.rs +++ b/crates/cli/tests/coverage/shared/server_tests.rs @@ -338,6 +338,7 @@ impl Drop for TestServer { fn test_config() -> GatewayConfig { crate::test_support::enable_operational_logs(); GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, diff --git a/crates/cli/tests/coverage/shared/session_tests.rs b/crates/cli/tests/coverage/shared/session_tests.rs index dca962f16..00f5c7f29 100644 --- a/crates/cli/tests/coverage/shared/session_tests.rs +++ b/crates/cli/tests/coverage/shared/session_tests.rs @@ -1723,6 +1723,7 @@ async fn has_pending_alignment(manager: &SessionManager, session_id: &str) -> bo #[tokio::test] async fn nests_agent_subagent_and_tool_lifecycle() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -3586,6 +3587,7 @@ async fn writes_atif_on_session_end_from_plugin_config() { let atif_dir = temp.path().join("atif"); install_test_atif_plugin(&atif_dir).await; let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -4195,6 +4197,7 @@ async fn duplicate_agent_end_does_not_overwrite_atif_with_empty_session() { let atif_dir = temp.path().join("atif"); install_test_atif_plugin(&atif_dir).await; let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -4383,6 +4386,7 @@ async fn inferred_skill_load_hook_marks_use_the_stable_event_contract() { #[tokio::test] async fn handles_out_of_order_subagent_and_tool_end_events() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -4461,6 +4465,7 @@ async fn terminal_retry_for_unknown_session_is_ignored() { #[tokio::test] async fn out_of_order_started_subagent_end_does_not_leak_scope() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -4535,6 +4540,7 @@ async fn out_of_order_started_subagent_end_does_not_leak_scope() { #[tokio::test] async fn agent_end_closes_nested_active_subagents_lifo() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -4593,6 +4599,7 @@ async fn agent_end_closes_nested_active_subagents_lifo() { #[tokio::test] async fn llm_lifecycle_starts_implicit_gateway_session() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -5078,6 +5085,7 @@ async fn claude_orphan_subagent_stop_after_closed_turn_does_not_open_null_turn() #[tokio::test] async fn llm_lifecycle_uses_single_active_hook_session_when_header_is_missing() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -5207,6 +5215,7 @@ async fn unidentified_concurrent_gateway_calls_use_isolated_ephemeral_sessions() #[tokio::test] async fn single_pending_llm_hint_claims_next_gateway_llm() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -5306,6 +5315,7 @@ async fn single_pending_llm_hint_claims_next_gateway_llm() { #[tokio::test] async fn multiple_llm_hints_resolve_by_generation_id() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -5423,6 +5433,7 @@ async fn multiple_llm_hints_resolve_by_generation_id() { #[tokio::test] async fn ambiguous_llm_hints_fall_back_to_agent_scope() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -5518,6 +5529,7 @@ async fn ambiguous_llm_hints_fall_back_to_agent_scope() { #[tokio::test] async fn no_active_hint_reuses_last_llm_owner() { let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -7351,6 +7363,7 @@ fn merge_metadata_handles_objects_nulls_and_scalars() { fn session_test_config() -> GatewayConfig { crate::test_support::enable_operational_logs(); GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, @@ -7367,6 +7380,7 @@ fn session_test_config() -> GatewayConfig { async fn turn_ended_is_noop_without_active_turn_scope() { let temp = tempfile::tempdir().unwrap(); let config = GatewayConfig { + caller_credential_targets: Default::default(), bind: "127.0.0.1:0".parse().unwrap(), openai_base_url: "http://127.0.0.1".into(), openai_auth_header: None, diff --git a/crates/core/src/api/runtime.rs b/crates/core/src/api/runtime.rs index c59e7e5c7..bc3bda30c 100644 --- a/crates/core/src/api/runtime.rs +++ b/crates/core/src/api/runtime.rs @@ -6,6 +6,7 @@ pub mod callbacks; mod continuation_context; pub mod global; +pub mod provider; pub mod scope_stack; pub mod state; pub mod subscriber_dispatcher; diff --git a/crates/core/src/api/runtime/continuation_context.rs b/crates/core/src/api/runtime/continuation_context.rs index 7ce3a1c0c..109115f81 100644 --- a/crates/core/src/api/runtime/continuation_context.rs +++ b/crates/core/src/api/runtime/continuation_context.rs @@ -5,6 +5,10 @@ use std::future::Future; +use super::provider::{ + LlmProviderDispatcher, current_provider_dispatcher, scope_provider_dispatcher, +}; + use crate::api::optimization::{ LlmOptimizationRecorder, current_llm_optimization_recorder, scope_llm_optimization_recorder, }; @@ -31,6 +35,7 @@ pub struct MiddlewareContinuationContext { publication_context: Option, publication_buffer: Option, optimization_recorder: Option, + provider_dispatcher: Option, } impl MiddlewareContinuationContext { @@ -44,6 +49,7 @@ impl MiddlewareContinuationContext { publication_context: capture_publication_context(), publication_buffer: capture_nested_publication_buffer(), optimization_recorder: current_llm_optimization_recorder(), + provider_dispatcher: current_provider_dispatcher(), } } @@ -69,6 +75,7 @@ impl MiddlewareContinuationContext { publication_context: self.publication_context.clone(), publication_buffer: self.publication_buffer.clone(), optimization_recorder: self.optimization_recorder.clone(), + provider_dispatcher: self.provider_dispatcher.clone(), }) } @@ -87,6 +94,7 @@ impl MiddlewareContinuationContext { /// Poll `future` with the captured Relay task context restored. #[doc(hidden)] pub async fn run(&self, future: F) -> F::Output { + let future = scope_provider_dispatcher(self.provider_dispatcher.clone(), future); let scoped = TASK_SCOPE_STACK.scope(self.scope_stack.clone(), future); let published = with_task_publication_context(self.publication_context.clone(), scoped); let published = diff --git a/crates/core/src/api/runtime/provider.rs b/crates/core/src/api/runtime/provider.rs new file mode 100644 index 000000000..0733fe2a5 --- /dev/null +++ b/crates/core/src/api/runtime/provider.rs @@ -0,0 +1,72 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Request-local provider execution supplied by an embedding host. + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; + +pub use nemo_relay_types::api::provider::{LlmProviderFormat, LlmProviderRequest}; + +use crate::api::runtime::LlmJsonStream; +use crate::error::Result; +use crate::json::Json; + +/// Host callback for a buffered call to an authorized provider target. +pub type LlmProviderCallFn = Arc< + dyn Fn(LlmProviderRequest) -> Pin> + Send>> + Send + Sync, +>; + +/// Host callback for a streamed call to an authorized provider target. +pub type LlmProviderStreamFn = Arc< + dyn Fn(LlmProviderRequest) -> Pin> + Send>> + + Send + + Sync, +>; + +/// Private request-scoped provider execution supplied by the host. +/// +/// Callbacks must authorize every target, keep credentials out of returned +/// values and errors, disable credential-bearing redirects, and propagate +/// cancellation by dropping pending I/O. The runtime never serializes these +/// callbacks or places them in events. Native plugins access them only through +/// their live execution continuation. +#[derive(Clone)] +pub struct LlmProviderDispatcher { + pub(crate) call: LlmProviderCallFn, + pub(crate) stream: LlmProviderStreamFn, +} + +impl LlmProviderDispatcher { + /// Create a dispatcher bound to one inbound request's credentials and policy. + pub fn new(call: LlmProviderCallFn, stream: LlmProviderStreamFn) -> Self { + Self { call, stream } + } +} + +tokio::task_local! { + static PROVIDER_DISPATCHER: Option; +} + +/// Run a managed LLM invocation with a private host provider dispatcher. +/// +/// Each concurrent request must supply its own dispatcher. Ordinary LLM +/// execution remains unchanged when no dispatcher is installed. +pub async fn with_llm_provider_dispatcher( + dispatcher: LlmProviderDispatcher, + future: F, +) -> F::Output { + scope_provider_dispatcher(Some(dispatcher), future).await +} + +pub(crate) fn current_provider_dispatcher() -> Option { + PROVIDER_DISPATCHER.try_with(Clone::clone).ok().flatten() +} + +pub(crate) async fn scope_provider_dispatcher( + dispatcher: Option, + future: F, +) -> F::Output { + PROVIDER_DISPATCHER.scope(dispatcher, future).await +} diff --git a/crates/core/src/plugin/dynamic/native.rs b/crates/core/src/plugin/dynamic/native.rs index c7d7a4661..615f42915 100644 --- a/crates/core/src/plugin/dynamic/native.rs +++ b/crates/core/src/plugin/dynamic/native.rs @@ -65,19 +65,27 @@ use nemo_relay_plugin::{ NemoRelayNativeAsyncStreamMiddlewareCb, NemoRelayNativeConditionalMiddlewareCb, NemoRelayNativeEventSanitizeCb, NemoRelayNativeEventSubscriberCb, NemoRelayNativeFreeFn, NemoRelayNativeHostApiV1, NemoRelayNativeHostApiV3, NemoRelayNativeHostApiV4, - NemoRelayNativeHostApiV5, NemoRelayNativeLlmAsyncStream, NemoRelayNativeLlmCodecKind, - NemoRelayNativeLlmConditionalCb, NemoRelayNativeLlmExecutionCb, NemoRelayNativeLlmRequestCodec, - NemoRelayNativeLlmRequestInterceptCb, NemoRelayNativeLlmResponseCodec, - NemoRelayNativeLlmSanitizeRequestCb, NemoRelayNativeLlmSanitizeRequestContext, - NemoRelayNativeLlmSanitizeResponseCb, NemoRelayNativeLlmSanitizeResponseContext, - NemoRelayNativeLlmStreamExecutionCb, NemoRelayNativeLlmStreamV1, NemoRelayNativePluginContext, - NemoRelayNativePluginEntry, NemoRelayNativePluginRuntime, NemoRelayNativePluginV1, - NemoRelayNativeScopeHandle, NemoRelayNativeScopeStack, NemoRelayNativeScopeStackBinding, - NemoRelayNativeScopeType, NemoRelayNativeString, NemoRelayNativeToolConditionalCb, - NemoRelayNativeToolExecutionCb, NemoRelayNativeToolExecutionContextCb, - NemoRelayNativeToolJsonCb, NemoRelayNativeWithScopeStackCb, NemoRelayStatus, + NemoRelayNativeHostApiV5, NemoRelayNativeHostApiV6, NemoRelayNativeLlmAsyncStream, + NemoRelayNativeLlmCodecKind, NemoRelayNativeLlmConditionalCb, NemoRelayNativeLlmExecutionCb, + NemoRelayNativeLlmRequestCodec, NemoRelayNativeLlmRequestInterceptCb, + NemoRelayNativeLlmResponseCodec, NemoRelayNativeLlmSanitizeRequestCb, + NemoRelayNativeLlmSanitizeRequestContext, NemoRelayNativeLlmSanitizeResponseCb, + NemoRelayNativeLlmSanitizeResponseContext, NemoRelayNativeLlmStreamExecutionCb, + NemoRelayNativeLlmStreamV1, NemoRelayNativePluginContext, NemoRelayNativePluginEntry, + NemoRelayNativePluginRuntime, NemoRelayNativePluginV1, NemoRelayNativeScopeHandle, + NemoRelayNativeScopeStack, NemoRelayNativeScopeStackBinding, NemoRelayNativeScopeType, + NemoRelayNativeString, NemoRelayNativeToolConditionalCb, NemoRelayNativeToolExecutionCb, + NemoRelayNativeToolExecutionContextCb, NemoRelayNativeToolJsonCb, + NemoRelayNativeWithScopeStackCb, NemoRelayStatus, }; use serde_json::{Map, Value as Json}; + +mod provider; +use crate::api::runtime::provider::{LlmProviderDispatcher, current_provider_dispatcher}; +use provider::{ + native_async_next_call_provider, native_async_next_has_provider, + native_async_next_stream_provider, +}; use sha2::{Digest, Sha256}; use tokio::runtime::Runtime; use tokio_stream::{Stream, StreamExt}; @@ -409,9 +417,13 @@ fn load_one_native_plugin( )) })?; let mut status = entry(native_host_api(), &mut plugin); - // Older SDKs reject newer tables. Negotiate from the current v5 table - // through separately frozen v4, v3, and v2 tables so their struct sizes and + // Older SDKs reject newer tables. Negotiate from the current v6 table + // through separately frozen v5, v4, v3, and v2 tables so their struct sizes and // function pointers do not change as the current ABI grows. + if status == NemoRelayStatus::InvalidArg { + drop_native_plugin_descriptor(&mut plugin); + status = entry(native_host_api_v5(), &mut plugin); + } if status == NemoRelayStatus::InvalidArg { drop_native_plugin_descriptor(&mut plugin); status = entry(native_host_api_v4(), &mut plugin); @@ -866,6 +878,11 @@ unsafe extern "C" fn native_llm_response_codec_decode( } fn native_host_api() -> *const NemoRelayNativeHostApiV1 { + static HOST_API: OnceLock = OnceLock::new(); + &HOST_API.get_or_init(build_native_host_api_v6).v5.v4.v3.v1 as *const NemoRelayNativeHostApiV1 +} + +fn native_host_api_v5() -> *const NemoRelayNativeHostApiV1 { static HOST_API: OnceLock = OnceLock::new(); &HOST_API.get_or_init(build_native_host_api_v5).v4.v3.v1 as *const NemoRelayNativeHostApiV1 } @@ -1007,7 +1024,7 @@ fn build_native_host_api_v4() -> NemoRelayNativeHostApiV4 { fn build_native_host_api_v5() -> NemoRelayNativeHostApiV5 { let mut v4 = build_native_host_api_v4(); - v4.v3.v1.abi_version = NEMO_RELAY_NATIVE_ABI_VERSION; + v4.v3.v1.abi_version = nemo_relay_plugin::NEMO_RELAY_NATIVE_ABI_VERSION_TOOL_EXECUTION_CONTEXT; v4.v3.v1.struct_size = std::mem::size_of::(); NemoRelayNativeHostApiV5 { v4, @@ -1016,6 +1033,18 @@ fn build_native_host_api_v5() -> NemoRelayNativeHostApiV5 { } } +fn build_native_host_api_v6() -> NemoRelayNativeHostApiV6 { + let mut v5 = build_native_host_api_v5(); + v5.v4.v3.v1.abi_version = NEMO_RELAY_NATIVE_ABI_VERSION; + v5.v4.v3.v1.struct_size = std::mem::size_of::(); + NemoRelayNativeHostApiV6 { + v5, + async_next_has_provider: native_async_next_has_provider, + async_next_call_provider: native_async_next_call_provider, + async_next_stream_provider: native_async_next_stream_provider, + } +} + fn read_native_string(value: *const NemoRelayNativeString) -> crate::plugin::Result { if value.is_null() { return Ok(String::new()); @@ -1698,6 +1727,7 @@ struct NativeAsyncNext { inner: NativeAsyncNextInner, runtime: tokio::runtime::Handle, context: MiddlewareContinuationContext, + provider_dispatcher: Option, owner: Option, // The native callback owns this handle independently of its completion. // Retaining the library here prevents an unload while it still uses `next`. @@ -1720,6 +1750,7 @@ impl NativeAsyncNext { inner, runtime, context: MiddlewareContinuationContext::capture(), + provider_dispatcher: current_provider_dispatcher(), owner: None, _callback_user_data: callback_user_data, } @@ -2818,6 +2849,15 @@ unsafe extern "C" fn native_async_next_invoke_result( return NemoRelayStatus::InvalidArg; } }; + spawn_native_unary(next, future, cb, user_data) +} + +fn spawn_native_unary( + next: &NativeAsyncNext, + future: Pin> + Send>>, + cb: NemoRelayNativeAsyncNextResultCb, + user_data: *mut c_void, +) -> NemoRelayStatus { let continuation_context = match next.context.isolated_for_current_invocation() { Ok(context) => context, Err(error) => return status_from_flow_error(error), @@ -2959,6 +2999,7 @@ struct NativePullLlmStream { runtime: tokio::runtime::Handle, context: MiddlewareContinuationContext, state: Mutex, + provider_owner: Option, _library_guard: Option>, } @@ -2998,6 +3039,12 @@ impl NativeCallbackGuard for NativePullOpenCallbackGuard { } } +impl NativeCallbackGuard for NativePullCallbackGuard { + fn suppress(&mut self) { + self.active = false; + } +} + impl NativePullCallbackGuard { fn deliver(&mut self, result: FlowResult>) { if self.active { @@ -3038,11 +3085,27 @@ unsafe extern "C" fn native_async_next_open_llm_stream( Ok(request) => request, Err(status) => return status, }; + let next_fn = next_fn.clone(); + spawn_native_stream( + next, + Box::pin(async move { next_fn(request).await }), + cb, + user_data, + false, + ) +} + +fn spawn_native_stream( + next: &NativeAsyncNext, + future: Pin> + Send>>, + cb: NemoRelayNativeAsyncLlmStreamOpenCb, + user_data: *mut c_void, + provider_call: bool, +) -> NemoRelayStatus { let context = match next.context.isolated_for_current_invocation() { Ok(context) => context, Err(error) => return status_from_flow_error(error), }; - let next_fn = next_fn.clone(); let runtime = next.runtime.clone(); let stream_runtime = runtime.clone(); let stream_context = context.clone(); @@ -3050,6 +3113,7 @@ unsafe extern "C" fn native_async_next_open_llm_stream( let callback_user_data = next._callback_user_data.clone(); let user_data = user_data as usize; let cleanup_owner = owner.clone(); + let provider_owner = provider_call.then(|| owner.clone()).flatten(); let (start_tx, start_rx) = tokio::sync::oneshot::channel(); let (callback_registration, callback_task) = NativeCallbackRegistration::new(NativePullOpenCallbackGuard { @@ -3063,9 +3127,7 @@ unsafe extern "C" fn native_async_next_open_llm_stream( return; } let mut callback_guard = callback_task.claim(); - let result = AssertUnwindSafe(context.run(next_fn(request))) - .catch_unwind() - .await; + let result = AssertUnwindSafe(context.run(future)).catch_unwind().await; remove_native_next_operation(&cleanup_owner, tokio::task::id()); match result { Ok(Ok(stream)) => { @@ -3074,6 +3136,7 @@ unsafe extern "C" fn native_async_next_open_llm_stream( runtime: stream_runtime, context: stream_context, state: Mutex::new(NativePullStreamState::Idle), + provider_owner, _library_guard: callback_guard.library_guard.take(), }); unsafe { @@ -3126,6 +3189,18 @@ unsafe extern "C" fn native_async_llm_stream_pull( let Some(stream) = (unsafe { (stream as *const NativePullLlmStream).as_ref() }) else { return NemoRelayStatus::NullPointer; }; + if stream.provider_owner.is_some() && !provider::owner_is_active(&stream.provider_owner) { + unsafe { + native_async_llm_stream_cancel( + stream as *const _ as *const NemoRelayNativeLlmAsyncStream, + ) + }; + if let Ok(mut guard) = stream.stream.try_lock() { + guard.take(); + } + set_native_last_error("provider stream belongs to a settled execution"); + return NemoRelayStatus::InvalidArg; + } let mut state = stream .state .lock() @@ -3144,12 +3219,15 @@ unsafe extern "C" fn native_async_llm_stream_pull( user_data, active: true, }; + let owner = stream.provider_owner.clone(); + let cleanup_owner = owner.clone(); + let (callback_registration, callback_task) = NativeCallbackRegistration::new(callback_guard); let (start_tx, start_rx) = tokio::sync::oneshot::channel(); let task = stream.runtime.spawn(async move { - let mut callback_guard = callback_guard; if start_rx.await.is_err() { return; } + let mut callback_guard = callback_task.claim(); let result = AssertUnwindSafe(context.run(async { let mut guard = task_stream.stream.lock().await; match guard.as_mut() { @@ -3175,6 +3253,7 @@ unsafe extern "C" fn native_async_llm_stream_pull( panic_payload_message(payload.as_ref()) ))) }); + remove_native_next_operation(&cleanup_owner, tokio::task::id()); let cancelled = { let mut state = task_stream .state @@ -3199,7 +3278,14 @@ unsafe extern "C" fn native_async_llm_stream_pull( ))); } }); - *state = NativePullStreamState::Pulling(task.abort_handle()); + let abort = task.abort_handle(); + if !register_native_next_operation(&owner, task.id(), abort.clone()) { + callback_registration.reject(); + abort.abort(); + return NemoRelayStatus::InvalidArg; + } + callback_registration.accept(); + *state = NativePullStreamState::Pulling(abort); drop(state); let _ = start_tx.send(()); NemoRelayStatus::Ok diff --git a/crates/core/src/plugin/dynamic/native/provider.rs b/crates/core/src/plugin/dynamic/native/provider.rs new file mode 100644 index 000000000..ddce0dbf2 --- /dev/null +++ b/crates/core/src/plugin/dynamic/native/provider.rs @@ -0,0 +1,101 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Private provider calls bound to a live native execution continuation. + +use nemo_relay_plugin::LlmProviderRequest; + +use super::*; + +pub(super) fn owner_is_active(owner: &Option) -> bool { + match owner { + Some(NativeAsyncNextOwner::Completion(owner)) => owner.upgrade().is_some_and(|owner| { + !owner.cancelled.load(Ordering::Acquire) + && owner + .sender + .lock() + .unwrap_or_else(|error| error.into_inner()) + .is_some() + }), + Some(NativeAsyncNextOwner::Stream(owner)) => owner.upgrade().is_some_and(|owner| { + !owner.cancelled.load(Ordering::Acquire) && !owner.settled.load(Ordering::Acquire) + }), + None => false, + } +} + +pub(super) unsafe extern "C" fn native_async_next_has_provider( + next: *const NemoRelayNativeAsyncNext, +) -> bool { + let Some(next) = (unsafe { (next as *const NativeAsyncNext).as_ref() }) else { + return false; + }; + matches!( + next.inner, + NativeAsyncNextInner::Llm(_) | NativeAsyncNextInner::LlmStream(_) + ) && next.provider_dispatcher.is_some() + && owner_is_active(&next.owner) +} + +fn provider_request( + next: &NativeAsyncNext, + request_json: *const NemoRelayNativeString, +) -> Result<(LlmProviderDispatcher, LlmProviderRequest), NemoRelayStatus> { + if matches!(next.inner, NativeAsyncNextInner::Tool(_)) || !owner_is_active(&next.owner) { + set_native_last_error("provider calls require a live LLM execution continuation"); + return Err(NemoRelayStatus::InvalidArg); + } + let Some(dispatcher) = next.provider_dispatcher.clone() else { + set_native_last_error("private provider dispatch is unavailable for this request"); + return Err(NemoRelayStatus::InvalidArg); + }; + let value = parse_json_arg(request_json, "provider request")?; + let request = serde_json::from_value(value).map_err(|_| { + set_native_last_error("invalid provider request: expected target and content"); + NemoRelayStatus::InvalidJson + })?; + Ok((dispatcher, request)) +} + +pub(super) unsafe extern "C" fn native_async_next_call_provider( + next: *const NemoRelayNativeAsyncNext, + request_json: *const NemoRelayNativeString, + cb: NemoRelayNativeAsyncNextResultCb, + user_data: *mut c_void, +) -> NemoRelayStatus { + let Some(next) = (unsafe { (next as *const NativeAsyncNext).as_ref() }) else { + return NemoRelayStatus::NullPointer; + }; + let (dispatcher, request) = match provider_request(next, request_json) { + Ok(input) => input, + Err(status) => return status, + }; + spawn_native_unary( + next, + Box::pin(async move { (dispatcher.call)(request).await }), + cb, + user_data, + ) +} + +pub(super) unsafe extern "C" fn native_async_next_stream_provider( + next: *const NemoRelayNativeAsyncNext, + request_json: *const NemoRelayNativeString, + cb: NemoRelayNativeAsyncLlmStreamOpenCb, + user_data: *mut c_void, +) -> NemoRelayStatus { + let Some(next) = (unsafe { (next as *const NativeAsyncNext).as_ref() }) else { + return NemoRelayStatus::NullPointer; + }; + let (dispatcher, request) = match provider_request(next, request_json) { + Ok(input) => input, + Err(status) => return status, + }; + spawn_native_stream( + next, + Box::pin(async move { (dispatcher.stream)(request).await }), + cb, + user_data, + true, + ) +} diff --git a/crates/core/tests/fixtures/native_plugin/src/lib.rs b/crates/core/tests/fixtures/native_plugin/src/lib.rs index fda6a7907..be8737707 100644 --- a/crates/core/tests/fixtures/native_plugin/src/lib.rs +++ b/crates/core/tests/fixtures/native_plugin/src/lib.rs @@ -10,10 +10,9 @@ use futures::StreamExt; use nemo_relay_plugin::{ CategoryProfile, ConfigDiagnostic, DiagnosticLevel, Event, EventCategory, EventSanitizeFields, Json, LlmJsonAsyncStream, LlmRequest, LlmRequestInterceptOutcome, MetricKind, - MetricMeasurement, MetricValueType, NEMO_RELAY_NATIVE_ABI_VERSION, - NEMO_RELAY_NATIVE_ABI_VERSION_ASYNC_MIDDLEWARE, NEMO_RELAY_NATIVE_ABI_VERSION_LEGACY, - NEMO_RELAY_NATIVE_ABI_VERSION_RUNTIME_CONTROL, NativeExecutorConfig, NativePlugin, - NemoRelayNativeAsyncCallbackState, + MetricMeasurement, MetricValueType, NEMO_RELAY_NATIVE_ABI_VERSION_ASYNC_MIDDLEWARE, + NEMO_RELAY_NATIVE_ABI_VERSION_LEGACY, NEMO_RELAY_NATIVE_ABI_VERSION_RUNTIME_CONTROL, + NativeExecutorConfig, NativePlugin, NemoRelayNativeAsyncCallbackState, NemoRelayNativeAsyncMiddlewareCb, NemoRelayNativeAsyncMiddlewareKind, NemoRelayNativeAsyncNext, NemoRelayNativeAsyncStream, NemoRelayNativeHostApiV1, NemoRelayNativeHostApiV3, NemoRelayNativeHostApiV4, NemoRelayNativeHostApiV5, NemoRelayNativePluginContext, @@ -64,6 +63,13 @@ impl NativePlugin for FixtureNativePlugin { plugin_config: &Map, ctx: &mut PluginContext<'_>, ) -> nemo_relay_plugin::Result<()> { + if plugin_config + .get("private_provider") + .and_then(Json::as_bool) + .unwrap_or(false) + { + return register_private_provider_fixture(ctx); + } let event_metadata_injector_error = plugin_config .get("event_metadata_injector_error") .and_then(Json::as_bool) @@ -528,7 +534,7 @@ pub unsafe extern "C" fn nemo_relay_fixture_native_plugin_v2( } } -/// Raw ABI-v5 entry used to verify the current table. +/// Raw ABI-v5 entry used to verify host fallback. #[unsafe(no_mangle)] pub unsafe extern "C" fn nemo_relay_fixture_native_plugin_v5( host: *const NemoRelayNativeHostApiV1, @@ -538,7 +544,7 @@ pub unsafe extern "C" fn nemo_relay_fixture_native_plugin_v5( fixture_compat_entry( host, out, - NEMO_RELAY_NATIVE_ABI_VERSION, + 5, std::mem::size_of::(), b"fixture_native_v5", ) @@ -1619,3 +1625,77 @@ unsafe fn raw_host_string_value( }; std::str::from_utf8(bytes).ok().map(str::to_owned) } + +fn provider_targets(request: &mut LlmRequest) -> Vec { + request + .content + .as_object_mut() + .unwrap() + .remove("fixture_provider_targets") + .and_then(|value| serde_json::from_value(value).ok()) + .unwrap_or_else(|| vec!["answer".into()]) +} + +fn register_private_provider_fixture(ctx: &mut PluginContext<'_>) -> nemo_relay_plugin::Result<()> { + if !ctx.supports_provider_dispatch() { + return Err("host does not support private provider dispatch".into()); + } + ctx.register_llm_execution_intercept( + "private_provider", + 0, + |_name, mut request, next| async move { + assert!(!request.headers.contains_key("authorization")); + assert!(!request.headers.contains_key("x-api-key")); + let provider = next.provider()?; + let targets = provider_targets(&mut request); + let mut last = Err("no provider targets".to_owned()); + for target in targets { + last = provider + .call(nemo_relay_plugin::LlmProviderRequest { + target, + content: request.content.clone(), + }) + .await; + if last.is_ok() { + break; + } + } + last + }, + )?; + ctx.register_llm_stream_execution_intercept( + "private_provider_stream", + 0, + |_name, mut request, next| async move { + assert!(!request.headers.contains_key("authorization")); + let provider = next.provider()?; + if let Some(probe) = request + .content + .as_object_mut() + .unwrap() + .remove("fixture_provider_probe") + { + provider + .call(nemo_relay_plugin::LlmProviderRequest { + target: probe.as_str().unwrap().into(), + content: request.content.clone(), + }) + .await?; + } + let targets = provider_targets(&mut request); + let mut last = Err("no provider targets".to_owned()); + for target in targets { + last = provider + .stream(nemo_relay_plugin::LlmProviderRequest { + target, + content: request.content.clone(), + }) + .await; + if last.is_ok() { + break; + } + } + last + }, + ) +} diff --git a/crates/core/tests/unit/native_plugin_tests.rs b/crates/core/tests/unit/native_plugin_tests.rs index 52047c057..361d8f0f8 100644 --- a/crates/core/tests/unit/native_plugin_tests.rs +++ b/crates/core/tests/unit/native_plugin_tests.rs @@ -633,6 +633,12 @@ fn assert_native_digest_edges() { fn assert_native_host_api_versions() { let current = native_host_api(); + let frozen_v5 = native_host_api_v5(); + assert_eq!(unsafe { (*frozen_v5).abi_version }, 5); + assert_eq!( + unsafe { (*frozen_v5).struct_size }, + std::mem::size_of::() + ); let frozen_v4 = native_host_api_v4(); let frozen_v3 = native_host_api_v3(); let legacy = native_host_api_v2(); @@ -655,7 +661,7 @@ fn assert_native_host_api_versions() { ); assert_eq!( unsafe { (*current).struct_size }, - std::mem::size_of::() + std::mem::size_of::() ); assert_eq!( unsafe { (*frozen_v4).struct_size }, @@ -1779,7 +1785,7 @@ fn assert_native_json_output_and_host_api() { assert_eq!(host_api.abi_version, NEMO_RELAY_NATIVE_ABI_VERSION); assert_eq!( host_api.struct_size, - std::mem::size_of::() + std::mem::size_of::() ); } @@ -7237,3 +7243,222 @@ fn native_stream_continuation_covers_success_and_error() { NemoRelayStatus::NullPointer ); } + +#[test] +fn private_provider_calls_are_cancelled_and_cannot_be_reused_after_completion() { + struct DropProbe(Arc); + + impl Drop for DropProbe { + fn drop(&mut self) { + self.0.store(true, Ordering::Release); + } + } + + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); + let completion = Arc::new(NativeAsyncCompletion { + sender: Mutex::new(Some(completion_tx)), + cancelled: AtomicBool::new(false), + next_invoked: AtomicBool::new(false), + next_abort: Mutex::new(None), + continuation_aborts: Mutex::new(HashMap::new()), + codec: None, + before_settlement_lock: None, + _callback_user_data: None, + }); + let wait = NativeAsyncWait { + completion: Arc::clone(&completion), + receiver: completion_rx, + completed: false, + }; + let started = Arc::new(AtomicBool::new(false)); + let dropped = Arc::new(AtomicBool::new(false)); + let dispatcher = LlmProviderDispatcher::new( + { + let started = Arc::clone(&started); + let dropped = Arc::clone(&dropped); + Arc::new(move |_value| { + let started = Arc::clone(&started); + let probe = DropProbe(Arc::clone(&dropped)); + Box::pin(async move { + started.store(true, Ordering::Release); + let _probe = probe; + std::future::pending::>().await + }) + }) + }, + Arc::new(|_| Box::pin(async { unreachable!() })), + ); + let mut next = NativeAsyncNext::with_completion_owner( + NativeAsyncNextInner::Llm(Arc::new(|_| Box::pin(async { unreachable!() }))), + runtime.handle().clone(), + None, + &completion, + ); + next.provider_dispatcher = Some(dispatcher); + let next = Arc::new(next); + let next_ref = Arc::into_raw(next) as *const NemoRelayNativeAsyncNext; + let invocation = native_string_from_json(&json!({"target":"pending","content":{}})).unwrap(); + let (result_tx, result_rx) = + tokio::sync::oneshot::channel::>(); + assert_eq!( + unsafe { + native_async_next_call_provider( + next_ref, + invocation, + complete_native_next_result, + Box::into_raw(Box::new(result_tx)).cast(), + ) + }, + NemoRelayStatus::Ok + ); + runtime.block_on(async { + while !started.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + }); + assert!(unsafe { native_async_next_has_provider(next_ref) }); + drop(wait); + let result = runtime + .block_on(result_rx) + .unwrap() + .expect_err("cancelled continuation should reject its result callback"); + assert!(result.contains("cancelled"), "{result}"); + assert!(dropped.load(Ordering::Acquire)); + assert!(completion.cancelled.load(Ordering::Acquire)); + assert!(!unsafe { native_async_next_has_provider(next_ref) }); + assert_eq!( + unsafe { + native_async_next_call_provider( + next_ref, + invocation, + complete_native_next_result, + ptr::null_mut(), + ) + }, + NemoRelayStatus::InvalidArg + ); + + unsafe { + native_string_free(invocation); + native_async_next_release(next_ref); + } +} + +#[test] +fn private_provider_stream_cancels_pending_pull_and_rejects_late_reads() { + struct DropProbe(Arc); + impl Drop for DropProbe { + fn drop(&mut self) { + self.0.store(true, Ordering::Release); + } + } + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let (sender, receiver) = tokio::sync::oneshot::channel(); + let completion = Arc::new(NativeAsyncCompletion { + sender: Mutex::new(Some(sender)), + cancelled: AtomicBool::new(false), + next_invoked: AtomicBool::new(false), + next_abort: Mutex::new(None), + continuation_aborts: Mutex::new(HashMap::new()), + codec: None, + before_settlement_lock: None, + _callback_user_data: None, + }); + let wait = NativeAsyncWait { + completion: completion.clone(), + receiver, + completed: false, + }; + let started = Arc::new(AtomicBool::new(false)); + let dropped = Arc::new(AtomicBool::new(false)); + let dispatcher = + LlmProviderDispatcher::new(Arc::new(|_| Box::pin(async { unreachable!() })), { + let started = started.clone(); + let dropped = dropped.clone(); + Arc::new(move |_| { + let started = started.clone(); + let probe = DropProbe(dropped.clone()); + Box::pin(async move { + Ok(LlmJsonStream::new(futures_util::stream::unfold( + probe, + move |probe| { + let started = started.clone(); + async move { + started.store(true, Ordering::Release); + std::future::pending::<()>().await; + Some((Ok(Json::Null), probe)) + } + }, + ))) + }) + }) + }); + let mut next = NativeAsyncNext::with_completion_owner( + NativeAsyncNextInner::Llm(Arc::new(|_| Box::pin(async { unreachable!() }))), + runtime.handle().clone(), + None, + &completion, + ); + next.provider_dispatcher = Some(dispatcher); + let next_ref = Arc::into_raw(Arc::new(next)) as *const NemoRelayNativeAsyncNext; + let request = native_string_from_json(&json!({"target":"answer","content":{}})).unwrap(); + let (open_tx, open_rx) = tokio::sync::oneshot::channel::(); + assert_eq!( + unsafe { + native_async_next_stream_provider( + next_ref, + request, + complete_pull_stream_open, + Box::into_raw(Box::new(open_tx)).cast(), + ) + }, + NemoRelayStatus::Ok + ); + let stream = + runtime.block_on(open_rx).unwrap().unwrap() as *const NemoRelayNativeLlmAsyncStream; + let (pull_tx, pull_rx) = tokio::sync::oneshot::channel::(); + assert_eq!( + unsafe { + native_async_llm_stream_pull( + stream, + complete_pull_stream_item, + Box::into_raw(Box::new(pull_tx)).cast(), + ) + }, + NemoRelayStatus::Ok + ); + runtime.block_on(async { + tokio::time::timeout(std::time::Duration::from_secs(2), async { + while !started.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + }); + drop(wait); + assert!( + runtime + .block_on(pull_rx) + .unwrap() + .unwrap_err() + .contains("cancelled") + ); + assert_eq!( + unsafe { native_async_llm_stream_pull(stream, complete_pull_stream_item, ptr::null_mut()) }, + NemoRelayStatus::InvalidArg + ); + assert!(dropped.load(Ordering::Acquire)); + unsafe { + native_async_llm_stream_release(stream); + native_string_free(request); + native_async_next_release(next_ref); + } +} diff --git a/crates/plugin/src/async_sdk.rs b/crates/plugin/src/async_sdk.rs index 332ae714b..fcab7e5b9 100644 --- a/crates/plugin/src/async_sdk.rs +++ b/crates/plugin/src/async_sdk.rs @@ -148,7 +148,44 @@ impl Drop for NativeExecutor { } #[derive(Clone, Copy)] -struct HostV4(NemoRelayNativeHostApiV4); +struct HostV4(NemoRelayNativeHostApiV4, Option); + +type NativeUnaryCall = unsafe extern "C" fn( + *const NemoRelayNativeAsyncNext, + *const NemoRelayNativeString, + NemoRelayNativeAsyncNextResultCb, + *mut c_void, +) -> NemoRelayStatus; +type NativeStreamCall = unsafe extern "C" fn( + *const NemoRelayNativeAsyncNext, + *const NemoRelayNativeString, + NemoRelayNativeAsyncLlmStreamOpenCb, + *mut c_void, +) -> NemoRelayStatus; + +#[derive(Clone, Copy)] +struct ProviderApi { + available: unsafe extern "C" fn(*const NemoRelayNativeAsyncNext) -> bool, + call: NativeUnaryCall, + stream: NativeStreamCall, +} + +impl ProviderApi { + fn from_host(host: &NemoRelayNativeHostApiV1) -> Option { + if host.abi_version < NEMO_RELAY_NATIVE_ABI_VERSION_PROVIDER_DISPATCH + || host.struct_size < std::mem::size_of::() + { + return None; + } + // SAFETY: the ABI and advertised table size cover the complete v6 tail. + let host = unsafe { &*(host as *const _ as *const NemoRelayNativeHostApiV6) }; + Some(Self { + available: host.async_next_has_provider, + call: host.async_next_call_provider, + stream: host.async_next_stream_provider, + }) + } +} unsafe impl Send for HostV4 {} unsafe impl Sync for HostV4 {} @@ -269,6 +306,13 @@ impl ToolNext { pub struct LlmNext(Arc); impl LlmNext { + /// Resolve private provider execution for this live request. + /// + /// Returns an error on older hosts or calls without a provider dispatcher. + pub fn provider(&self) -> Result { + LlmProvider::new(Arc::clone(&self.0)) + } + /// Continues the LLM chain with a replacement request. pub async fn call(&self, request: LlmRequest) -> Result { invoke_unary_next(&self.0, &request).await @@ -283,42 +327,99 @@ pub type LlmJsonAsyncStream = Pin> + Send>>; pub struct LlmStreamNext(Arc); impl LlmStreamNext { + /// Resolve buffered and streamed provider execution for this live request. + pub fn provider(&self) -> Result { + LlmProvider::new(Arc::clone(&self.0)) + } + /// Opens an independent pull-based downstream stream. pub async fn call(&self, request: LlmRequest) -> Result { - let request = HostString::from_json(&self.0.host.0.v3.v1, &request) - .ok_or_else(|| "failed to serialize LLM stream request".to_string())?; - let (sender, receiver) = futures::channel::oneshot::channel(); - let callback_state = Box::into_raw(Box::new(OpenState { - sender, - host: self.0.host, - })); - let status = unsafe { - (self.0.host.0.async_next_open_llm_stream)( - self.0.raw, - request.as_ptr(), - open_stream_callback, - callback_state.cast(), - ) - }; - if status != NemoRelayStatus::Ok { - drop(unsafe { Box::from_raw(callback_state) }); - return Err(status_message( - &self.0.host.0.v3.v1, - status, - "open LLM stream", - )); + open_next_stream(&self.0, &request, self.0.host.0.async_next_open_llm_stream).await + } +} + +/// Request-scoped capability for calls to host-authorized provider targets. +/// +/// The host owns credentials and destination policy. Both methods may be used +/// repeatedly or concurrently for routing, answers, retries, and fallbacks. +/// The capability expires when its execution interceptor or output stream +/// settles; cloning it does not extend that lifetime. Calls bypass the normal +/// continuation chain and do not re-enter plugin routing. +#[derive(Clone)] +pub struct LlmProvider(Arc); + +impl LlmProvider { + fn new(next: Arc) -> Result { + if next + .host + .1 + .is_some_and(|api| unsafe { (api.available)(next.raw) }) + { + Ok(Self(next)) + } else { + Err("private provider dispatch is unavailable for this request (requires a supporting host and native ABI v6)".into()) } - let mut opened = receiver - .await - .map_err(|_| "LLM stream open callback was dropped".to_string())??; - let raw = opened.take(); - Ok(Box::pin(PullStream { - host: self.0.host, - raw, - pending: None, - finished: false, - })) } + + /// Execute one buffered request using the caller credential held by the host. + pub async fn call(&self, request: LlmProviderRequest) -> Result { + let api = self + .0 + .host + .1 + .ok_or("private provider dispatch is unavailable")?; + invoke_unary(&self.0, &request, api.call).await + } + + /// Open an independent provider stream; dropping it cancels the host stream. + pub async fn stream(&self, request: LlmProviderRequest) -> Result { + let api = self + .0 + .host + .1 + .ok_or("private provider dispatch is unavailable")?; + open_next_stream(&self.0, &request, api.stream).await + } +} + +async fn open_next_stream( + next: &NextInner, + request: &T, + open: NativeStreamCall, +) -> Result { + let request = HostString::from_json(&next.host.0.v3.v1, request) + .ok_or_else(|| "failed to serialize LLM stream request".to_string())?; + let (sender, receiver) = futures::channel::oneshot::channel(); + let callback_state = Box::into_raw(Box::new(OpenState { + sender, + host: next.host, + })); + let status = unsafe { + open( + next.raw, + request.as_ptr(), + open_stream_callback, + callback_state.cast(), + ) + }; + if status != NemoRelayStatus::Ok { + drop(unsafe { Box::from_raw(callback_state) }); + return Err(status_message( + &next.host.0.v3.v1, + status, + "open LLM stream", + )); + } + let mut opened = receiver + .await + .map_err(|_| "LLM stream open callback was dropped".to_string())??; + let raw = opened.take(); + Ok(Box::pin(PullStream { + host: next.host, + raw, + pending: None, + finished: false, + })) } struct OpenedStream { @@ -475,6 +576,14 @@ unsafe extern "C" fn pull_stream_callback( } async fn invoke_unary_next(next: &NextInner, value: &T) -> Result { + invoke_unary(next, value, next.host.0.v3.async_next_invoke_result).await +} + +async fn invoke_unary( + next: &NextInner, + value: &T, + invoke: NativeUnaryCall, +) -> Result { let value = HostString::from_json(&next.host.0.v3.v1, value) .ok_or_else(|| "failed to serialize native continuation input".to_string())?; let (sender, receiver) = futures::channel::oneshot::channel(); @@ -483,7 +592,7 @@ async fn invoke_unary_next(next: &NextInner, value: &T) -> Result< host: next.host, })); let status = unsafe { - (next.host.0.v3.async_next_invoke_result)( + invoke( next.raw, value.as_ptr(), unary_next_callback, @@ -1075,15 +1184,24 @@ impl CodecIdentityInvocation { } impl PluginContext<'_> { + /// Whether this host exposes native provider dispatch (ABI v6). + /// + /// A particular invocation also requires a host dispatcher and an + /// authorized target. Use `next.provider()` to check request availability. + pub fn supports_provider_dispatch(&self) -> bool { + ProviderApi::from_host(self.host).is_some() + } + fn host_v4(&self) -> Result { if self.host.abi_version < NEMO_RELAY_NATIVE_ABI_VERSION_RUNTIME_CONTROL || self.host.struct_size < std::mem::size_of::() { return Err("typed async native middleware requires Relay ABI v4".into()); } - Ok(HostV4(unsafe { - *(self.host as *const _ as *const NemoRelayNativeHostApiV4) - })) + Ok(HostV4( + unsafe { *(self.host as *const _ as *const NemoRelayNativeHostApiV4) }, + ProviderApi::from_host(self.host), + )) } fn register_unary_adapter( diff --git a/crates/plugin/src/lib.rs b/crates/plugin/src/lib.rs index 6e2fd9f76..69bcfd29b 100644 --- a/crates/plugin/src/lib.rs +++ b/crates/plugin/src/lib.rs @@ -11,7 +11,9 @@ mod async_sdk; -pub use async_sdk::{LlmJsonAsyncStream, LlmNext, LlmStreamNext, NativeExecutorConfig, ToolNext}; +pub use async_sdk::{ + LlmJsonAsyncStream, LlmNext, LlmProvider, LlmStreamNext, NativeExecutorConfig, ToolNext, +}; use std::ffi::{c_char, c_void}; use std::marker::{PhantomData, PhantomPinned}; @@ -26,6 +28,7 @@ pub use nemo_relay_types::api::event::{ MetricMeasurement, MetricValueType, PendingMarkSpec, ScopeCategory, }; pub use nemo_relay_types::api::llm::{LlmAttributes, LlmRequest, LlmRequestInterceptOutcome}; +pub use nemo_relay_types::api::provider::{LlmProviderFormat, LlmProviderRequest}; pub use nemo_relay_types::api::registry::{ RuntimeRegistrationIdentity, RuntimeRegistrationKind, RuntimeRegistrationOwner, RuntimeRegistrationOwnerKind, @@ -50,10 +53,12 @@ use serde_json::Map; /// Native plugin ABI version supported by this crate. /// -/// Version 5 adds a context-aware raw tool execution intercept registration. -/// Hosts retain frozen version-4, version-3, and version-2 tables for +/// Version 6 adds request-scoped, host-owned provider calls. +/// Hosts retain frozen version-5, version-4, version-3, and version-2 tables for /// already-built plugins that target those layouts. -pub const NEMO_RELAY_NATIVE_ABI_VERSION: u32 = 5; +pub const NEMO_RELAY_NATIVE_ABI_VERSION: u32 = 6; +/// ABI version that introduced private host-owned provider execution. +pub const NEMO_RELAY_NATIVE_ABI_VERSION_PROVIDER_DISPATCH: u32 = 6; /// ABI version that introduced context-aware raw tool execution intercepts. pub const NEMO_RELAY_NATIVE_ABI_VERSION_TOOL_EXECUTION_CONTEXT: u32 = 5; /// ABI version that introduced runtime diagnostics and dynamic gate control. @@ -1342,6 +1347,34 @@ pub struct NemoRelayNativeHostApiV5 { -> NemoRelayStatus, } +/// ABI-v6 host extension for request-scoped provider calls. +/// +/// All older tables remain frozen prefixes. Callbacks and stream handles use +/// the same ownership and cancellation contract as execution continuations. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct NemoRelayNativeHostApiV6 { + /// Frozen ABI-v5 compatibility prefix. + pub v5: NemoRelayNativeHostApiV5, + /// Whether this live LLM continuation has a host provider dispatcher. + pub async_next_has_provider: + unsafe extern "C" fn(next: *const NemoRelayNativeAsyncNext) -> bool, + /// Execute a buffered [`LlmProviderRequest`] using private host credentials. + pub async_next_call_provider: unsafe extern "C" fn( + next: *const NemoRelayNativeAsyncNext, + request_json: *const NemoRelayNativeString, + cb: NemoRelayNativeAsyncNextResultCb, + user_data: *mut c_void, + ) -> NemoRelayStatus, + /// Open a provider stream from [`LlmProviderRequest`] JSON. + pub async_next_stream_provider: unsafe extern "C" fn( + next: *const NemoRelayNativeAsyncNext, + request_json: *const NemoRelayNativeString, + cb: NemoRelayNativeAsyncLlmStreamOpenCb, + user_data: *mut c_void, + ) -> NemoRelayStatus, +} + unsafe impl Send for NemoRelayNativeHostApiV3 {} unsafe impl Sync for NemoRelayNativeHostApiV3 {} // SAFETY: the v4 host table is immutable after construction. Its function @@ -1352,6 +1385,9 @@ unsafe impl Sync for NemoRelayNativeHostApiV4 {} // same thread-safe host function table. unsafe impl Send for NemoRelayNativeHostApiV5 {} unsafe impl Sync for NemoRelayNativeHostApiV5 {} +// SAFETY: v6 extends the immutable thread-safe host function table. +unsafe impl Send for NemoRelayNativeHostApiV6 {} +unsafe impl Sync for NemoRelayNativeHostApiV6 {} // The host API table is immutable after construction. Function pointers and // the null-terminated version string pointer are safe to share across threads. @@ -3107,11 +3143,16 @@ enum OwnedHostApi { V3(NemoRelayNativeHostApiV3), V4(NemoRelayNativeHostApiV4), V5(NemoRelayNativeHostApiV5), + V6(NemoRelayNativeHostApiV6), } impl OwnedHostApi { unsafe fn copy_from(host: &NemoRelayNativeHostApiV1) -> Self { - if host.abi_version >= NEMO_RELAY_NATIVE_ABI_VERSION_TOOL_EXECUTION_CONTEXT + if host.abi_version >= NEMO_RELAY_NATIVE_ABI_VERSION_PROVIDER_DISPATCH + && host.struct_size >= std::mem::size_of::() + { + Self::V6(unsafe { *(host as *const _ as *const NemoRelayNativeHostApiV6) }) + } else if host.abi_version >= NEMO_RELAY_NATIVE_ABI_VERSION_TOOL_EXECUTION_CONTEXT && host.struct_size >= std::mem::size_of::() { Self::V5(unsafe { *(host as *const _ as *const NemoRelayNativeHostApiV5) }) @@ -3134,6 +3175,7 @@ impl OwnedHostApi { Self::V3(host) => &host.v1, Self::V4(host) => &host.v3.v1, Self::V5(host) => &host.v4.v3.v1, + Self::V6(host) => &host.v5.v4.v3.v1, } } } diff --git a/crates/plugin/tests/typed_callbacks.rs b/crates/plugin/tests/typed_callbacks.rs index b7a78cc1f..8ec6814ad 100644 --- a/crates/plugin/tests/typed_callbacks.rs +++ b/crates/plugin/tests/typed_callbacks.rs @@ -465,7 +465,7 @@ static UNAVAILABLE_CONTEXT_GATE_CALLS: AtomicUsize = AtomicUsize::new(0); #[test] fn native_abi_struct_sizes_are_self_describing() { - assert_eq!(NEMO_RELAY_NATIVE_ABI_VERSION, 5); + assert_eq!(NEMO_RELAY_NATIVE_ABI_VERSION, 6); assert_eq!( size_of::(), test_host().struct_size @@ -6310,3 +6310,50 @@ fn plugin_validate_and_register_panics_replace_last_error() { drop_exported_plugin(&host, register_plugin); } } + +#[test] +fn native_abi_v6_keeps_the_v5_prefix_frozen() { + use nemo_relay_plugin::NemoRelayNativeHostApiV6; + assert_eq!(offset_of!(NemoRelayNativeHostApiV6, v5), 0); + assert_eq!( + offset_of!(NemoRelayNativeHostApiV6, async_next_has_provider), + size_of::() + ); + assert_eq!( + size_of::(), + size_of::() + 3 * size_of::() + ); +} + +#[test] +fn provider_request_rejects_credentials_and_destination_overrides() { + use nemo_relay_plugin::LlmProviderRequest; + for extra in ["url", "headers", "authorization"] { + let mut value = json!({"target": "answer", "content": {"model": "test"}}); + value[extra] = json!("injected"); + assert!(serde_json::from_value::(value).is_err()); + } +} + +#[test] +fn provider_capability_discovery_requires_the_entire_v6_table() { + use nemo_relay_plugin::NemoRelayNativeHostApiV6; + let v4 = test_host_v4(); + assert!(!test_context(&v4.v3.v1).supports_provider_dispatch()); + let v5 = test_host_v5(); + assert!(!test_context(&v5.v4.v3.v1).supports_provider_dispatch()); + unsafe extern "C" fn available(_: *const NemoRelayNativeAsyncNext) -> bool { + true + } + let mut v6 = NemoRelayNativeHostApiV6 { + v5, + async_next_has_provider: available, + async_next_call_provider: v4.v3.async_next_invoke_result, + async_next_stream_provider: v4.async_next_open_llm_stream, + }; + v6.v5.v4.v3.v1.abi_version = 6; + v6.v5.v4.v3.v1.struct_size = size_of::(); + assert!(test_context(&v6.v5.v4.v3.v1).supports_provider_dispatch()); + v6.v5.v4.v3.v1.struct_size -= 1; + assert!(!test_context(&v6.v5.v4.v3.v1).supports_provider_dispatch()); +} diff --git a/crates/types/src/api/mod.rs b/crates/types/src/api/mod.rs index a47e4024c..734966dc2 100644 --- a/crates/types/src/api/mod.rs +++ b/crates/types/src/api/mod.rs @@ -7,6 +7,8 @@ pub mod event; /// LLM DTOs and attributes. pub mod llm; +/// Host-owned provider execution DTOs. +pub mod provider; /// Runtime-registration discovery DTOs. pub mod registry; /// Scope DTOs and attributes. diff --git a/crates/types/src/api/provider.rs b/crates/types/src/api/provider.rs new file mode 100644 index 000000000..ed3ae486c --- /dev/null +++ b/crates/types/src/api/provider.rs @@ -0,0 +1,33 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Credential-free requests for host-owned provider execution. + +use serde::{Deserialize, Serialize}; + +use crate::Json; + +/// Provider protocol accepted by a configured caller-credential target. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LlmProviderFormat { + /// OpenAI Chat Completions. + OpenaiChat, + /// OpenAI Responses. + OpenaiResponses, + /// Anthropic Messages. + AnthropicMessages, +} + +/// One provider call authorized and executed by the host. +/// +/// The target is a host-configured name, never a URL. Credentials, headers, +/// and credential-selection overrides are deliberately absent. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LlmProviderRequest { + /// Name of the host-authorized destination. + pub target: String, + /// Request body in the destination's configured provider format. + pub content: Json, +} diff --git a/docs/build-plugins/native/native-abi-reference.mdx b/docs/build-plugins/native/native-abi-reference.mdx index 6420750a1..f93d8553f 100644 --- a/docs/build-plugins/native/native-abi-reference.mdx +++ b/docs/build-plugins/native/native-abi-reference.mdx @@ -23,7 +23,7 @@ extern "C" fn nemo_relay_register_plugin( ) -> NemoRelayStatus ``` -The current host negotiates ABI v5, then the frozen v4, v3, and legacy v2 tables. +The current host negotiates ABI v6, then the frozen v5, v4, v3, and legacy v2 tables. ABI v4 extends the complete v3 prefix with completion-scoped [codecs](/about-nemo-relay/concepts/codecs) and pull-based downstream LLM streams, plus an activation-owned runtime capability for @@ -40,7 +40,8 @@ function signatures and field order are defined by the public | Frozen v1/v2 prefix | Version and struct-size negotiation; host version; string allocation, access, and release; thread-local error reporting; callback-scoped LLM request decode and encode plus response decode; subscriber, five tool, six LLM, and three event-sanitizer registrations; current scope, scope push and pop, mark emission, isolated stack creation and release, thread-stack set, capture, and restore, captured-binding release, active-stack inspection, and scoped binding. | | Frozen v3 extension | Completion resolve, reject, cancellation inspection, and release; one-shot completion-coupled continuation invocation; continuation release; generic async middleware registration; bounded output-stream push, finish, reject, cancellation inspection, and release; downstream stream invocation; stream-middleware registration; and repeated or concurrent unary continuation invocation with independent result callbacks. | | Frozen v4 extension | Completion-scoped LLM request decode and encode plus response decode; pull-based downstream LLM stream open, pull, cancel, and release; completion retain for typed codec facades; output-stream backpressure inspection; extended mark emission; runtime diagnostics; activation-owned runtime capability creation, retain, and release; global runtime-registration discovery; owned conditional middleware guardrail registration and deregistration; and activation-owned and runtime-discovered callback gate registration. The callback registration slots are appended after the original constant-reason slots. | -| Current v5 extension | Context-carrying tool execution intercept registration. The callback receives one JSON object containing `tool_name`, `args`, and `tool_call_id`. | +| Frozen v5 extension | Context-carrying tool execution intercept registration. The callback receives one JSON object containing `tool_name`, `args`, and `tool_call_id`. | +| Current v6 extension | Request-scoped private provider capability discovery, buffered calls, and streaming calls. | The prefix and descriptor layout are explicit. A plugin fills the descriptor with its stable kind, component multiplicity, opaque state, callbacks, and destructor. The host @@ -221,3 +222,26 @@ waits for completion and stream references to be released, deregisters component surfaces, runs component cleanup, and only then unloads the shared library. A plugin that retains a completion, continuation, stream, codec, or runtime handle indefinitely can therefore delay safe unload. + +## ABI v6 Private Provider Dispatch + +`NemoRelayNativeHostApiV6` embeds the complete frozen v5 table at offset zero and +appends `async_next_has_provider`, `async_next_call_provider`, and +`async_next_stream_provider`. Hosts negotiate v6, v5, v4, v3, then v2; an older plugin +that rejects a newer version receives its original frozen table. Check both +`abi_version >= 6` and `struct_size >= size_of::()` before +reading the appended function pointers. + +Provider calls reuse the retained `NemoRelayNativeAsyncNext` from a live LLM execution. +Tool, request-sanitizer, subscriber, and registration contexts do not grant provider +calls. `async_next_has_provider` reports request availability; actual calls also check +lifetime and host policy. The JSON input is exactly `{ "target": "name", "content": {} }`; +unknown fields are rejected. Secrets and transport URLs are not part of this contract. + +Unary and stream-open callbacks follow the existing acceptance rule: after an `Ok` +return, the host invokes the callback exactly once, including on cancellation; on a +non-`Ok` return, it does not invoke it and the caller retains its callback data. Opened +streams use the existing pull/cancel/release API. Every active provider operation is +registered with the owning execution for cancellation; later calls or pulls are rejected +after settlement. See [Wrap Execution](/build-plugins/native/wrap-execution) for the typed +SDK and gateway target configuration. diff --git a/docs/build-plugins/native/wrap-execution.mdx b/docs/build-plugins/native/wrap-execution.mdx index 22a3082ae..85794c0bb 100644 --- a/docs/build-plugins/native/wrap-execution.mdx +++ b/docs/build-plugins/native/wrap-execution.mdx @@ -155,3 +155,83 @@ behavior: Success means unary and stream continuations preserve scope, errors, and cancellation, while Relay-owned tool marks and LLM request accounting remain separate from application results. + +## Private Caller-Credential Provider Calls + +The gateway keeps provider credentials out of `LlmRequest.headers` and events. A +native execution plugin that performs its own provider calls can use +`next.provider()` on a host implementing native ABI v6. This returns a +request-scoped `LlmProvider` with buffered `call` and streaming `stream` methods. +Both methods are available in unary and streaming execution intercepts, allowing a +buffered routing-model call before a streamed answer. They call the provider directly; +they do not re-enter execution middleware or automatically emit a separate LLM event. + +Check `context.supports_provider_dispatch()` during registration when your configuration +requires this capability. Then use `next.provider()?` for each execution. Registration +support does not imply that a request has credentials or permission for a target. +Hosts with native ABI v2–v5 do not provide this capability; ordinary `next.call()` +continues to work. Plugins using these methods must rebuild against an SDK exposing +`LlmProviderRequest` and declare a Relay compatibility range covering hosts they test. +The ABI v6 requirement is independent of the manifest's unchanged `native_api = "1"`. + +```rust +use nemo_relay_plugin::LlmProviderRequest; + +if !context.supports_provider_dispatch() { + return Err("this configuration requires native ABI v6 provider dispatch".into()); +} +context.register_llm_execution_intercept( + "answer", 0, + |_name, request, next| async move { + next.provider()?.call(LlmProviderRequest { + target: "answer".into(), + content: request.content, + }).await + }, +)?; +``` + +The CLI gateway authorizes target names in its `config.toml`: + +```toml +[upstream.caller_credential_targets.answer] +url = "https://api.openai.com/v1/responses" +format = "openai_responses" + +[upstream.caller_credential_targets.router] +url = "https://api.openai.com/v1/chat/completions" +format = "openai_chat" +``` + +The URL is the complete endpoint. Supported formats are `openai_chat`, +`openai_responses`, and `anthropic_messages`. Targets default to an empty map. +A higher-precedence map replaces the entire lower-precedence map, including an +explicit empty map that revokes all destinations. URLs must be absolute HTTP(S) +URLs without userinfo or fragments. Only authorize endpoints you trust to receive +callers' provider credentials. HTTP is useful for local tests; use HTTPS for remote +providers. OpenAI Chat and Responses share a credential family; Anthropic is separate. +A target's format declares its credential family; the plugin supplies provider-specific +JSON and must adapt any routing-model or fallback request and response shapes itself. + +Each call checks the name and source family. The host sets the `stream` flag from the +chosen method and forwards only the caller's provider credential plus recognized +companion headers: OpenAI account/FedRAMP headers or Anthropic version/beta headers. +Gateway invocation tokens are consumed before this capability is created. Missing caller +credentials do not fall back to environment or deployment-owned credentials. Plugins +cannot override URLs or headers, and redirects are rejected, including same-origin ones. + +Retries and fallbacks must name authorized targets on every attempt and remain within +the request's provider family. Plugins own their retry limits and response selection. +HTTP failures expose status and a generic failure message, not upstream bodies or +redirect locations. Successful JSON and SSE values redact literal credential echoes; +configured endpoints are still trusted, and this does not prevent an endpoint from +encoding a secret in a response. Native plugins are in-process, unsandboxed extensions. + +The capability expires when its execution completes or is cancelled. Retaining its +handle does not authorize later calls. Pending calls and active stream pulls are aborted +with their owning execution; plugins must drop streams and handles they no longer need. +The ordinary continuation, deployment-owned authentication, and keyless targets retain +their existing behavior. Rust embedders can install equivalent request-specific host +callbacks with `with_llm_provider_dispatcher`; they own target policy, credential +provenance, transport cancellation, and response/error sanitization. Worker plugins and +Python, Node.js, Go, and raw application FFI bindings do not expose this capability.