Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion crates/switchyard-nemo-relay-plugin/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,7 @@ retains ownership of the outer LLM lifecycle.
When Relay is configured with OTLP logs and metrics exporters, the plugin emits
typed telemetry through Relay's native plugin runtime:

- Routing request, decision, and overhead marks are Info logs.
- Routing request, privacy decision, routing decision, and overhead marks are Info logs.
- Per-model call marks are Debug logs with `call_role`, outcome, and latency,
but no token usage. Streaming marks cover stream creation; later failures are
reported separately.
Expand Down Expand Up @@ -267,6 +267,7 @@ meaning requires a new schema version.
| Mark | Data fields |
| --- | --- |
| `switchyard.routing.requested` | `algorithm` |
| `switchyard.privacy.decision` | `lane`, `source`, `reason_code`, optional `clear_score`, and optional `clear_threshold` |
| `switchyard.routing.llm_call` | `call_index`, `selected_model`, `call_role`, `outcome`, `latency_ms` |
| `switchyard.routing.overhead` | `latency_ms` |
| `switchyard.routing.decision` | `algorithm`, optional `outcome_id`, `selected_model`, nullable `served_model`, nullable `fallback_used`, and optional `evidence` |
Expand All @@ -277,6 +278,10 @@ meaning requires a new schema version.
`evidence` is an object containing the supported string fields `source`, `verdict`,
`trigger`, and `reason_code`, and numeric fields `score`, `confidence`, and `threshold`.
String values longer than 64 bytes are omitted and should be stable, non-sensitive labels.
After successful lane selection, privacy-enabled routes record one privacy decision mark. The
mark is retained if the selected lane later fails. `clear_score` is the classifier probability
assigned to `no_sensitive_content`; `clear_threshold` is the minimum score required for the
standard lane. Routes without privacy configuration do not emit this mark.

## Failure policy

Expand Down
264 changes: 262 additions & 2 deletions crates/switchyard-nemo-relay-plugin/src/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use switchyard_protocol::{
Request, Response, Usage, WireFormat,
};
use switchyard_runner::{
ProviderKeyRedactor, Route, RouteErrorSummary, Runner, stream_error_summary,
PrivacyDecision, ProviderKeyRedactor, Route, RouteErrorSummary, Runner, stream_error_summary,
};
use switchyard_translation::{TranslationEngine, encode_stream_with_extensions};

Expand Down Expand Up @@ -254,7 +254,12 @@ impl SwitchyardRuntime {
.unwrap_or_else(|poisoned| poisoned.into_inner())
.push(observation);
});
match route.execute(request, Some(observer)).await {
let result = route
.execute_with_privacy_observer(request, Some(observer), |decision| {
events.push(privacy_decision_mark(decision, metadata.clone()));
})
.await;
match result {
Ok(output) => {
let outcome_fields = self.emit_observations(
&mut events,
Expand Down Expand Up @@ -666,6 +671,29 @@ fn route_execution_error_mark(
}
}

fn privacy_decision_mark(decision: &PrivacyDecision, metadata: Json) -> RoutingEvent {
let mut data = Map::from_iter([
("lane".into(), Json::String(decision.lane().into())),
("source".into(), Json::String(decision.source().into())),
(
"reason_code".into(),
Json::String(decision.reason_code().into()),
),
]);
if let Some(score) = decision.clear_score() {
data.insert("clear_score".into(), json!(score));
}
if let Some(threshold) = decision.clear_threshold() {
data.insert("clear_threshold".into(), json!(threshold));
}
RoutingEvent::Mark(RoutingMark {
name: "switchyard.privacy.decision".into(),
data: Json::Object(data),
metadata,
severity: Some(LogSeverity::Info),
})
}

fn route_execution_error_events(
summary: &RouteErrorSummary,
metadata: Json,
Expand Down Expand Up @@ -958,6 +986,47 @@ mod tests {
.expect("cross-format runtime should load")
}

fn privacy_runtime(base_url: &str) -> SwitchyardRuntime {
let deployment = json!({
"schema_version": 1,
"llm_clients": {
"target": {
"format": "openai_chat",
"base_url": base_url,
"max_retries": 0,
}
},
"targets": {
"standard": {"id": "standard/model", "llm_client": "target"},
"restricted": {"id": "restricted/model", "llm_client": "target", "private": true},
"backup": {"id": "backup/model", "llm_client": "target", "private": true},
},
"routes": {
"privacy": {
"id": "switchyard/privacy",
"type": "random",
"targets": ["standard", "restricted", "backup"],
"weights": [100.0, 10.0, 1.0],
"seed": 42,
"privacy": {
"accept_external_signal": true,
},
}
}
});
SwitchyardRuntime::new(crate::config::SwitchyardConfig {
priority: 0,
switchyard_config_path: None,
switchyard_config: Some(
deployment
.as_object()
.expect("deployment should be an object")
.clone(),
),
})
.expect("privacy runtime should load")
}

fn namespaced_responses_request() -> RelayRequest {
RelayRequest {
headers: Map::new(),
Expand Down Expand Up @@ -988,6 +1057,197 @@ mod tests {
assert!(!runtime.manages_model("other"));
}

#[tokio::test]
async fn privacy_marks_survive_lane_failure() {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(body_partial_json(json!({"model": "standard/model"})))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"id": "chatcmpl_privacy",
"object": "chat.completion",
"model": "standard/model",
"choices": [{
"index": 0,
"message": {"role": "assistant", "content": "ok"},
"finish_reason": "stop",
}],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
})))
.expect(1)
.mount(&server)
.await;
for model in ["restricted/model", "backup/model"] {
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.and(body_partial_json(json!({"model": model})))
.respond_with(ResponseTemplate::new(503).set_body_string("unavailable"))
.expect(1)
.mount(&server)
.await;
}
let runtime = privacy_runtime(&format!("{}/v1", server.uri()));

for (restricted, expected, succeeds) in [
(
false,
json!({"lane": "standard", "source": "policy", "reason_code": "all_clear"}),
true,
),
(
true,
json!({"lane": "restricted", "source": "external_signal", "reason_code": "restricted"}),
false,
),
] {
let mut request = runtime
.decode_request(
WireFormat::OpenAiChat,
RelayRequest {
headers: Map::new(),
content: json!({
"model": "switchyard/privacy",
"messages": [{"role": "user", "content": "hello"}],
}),
},
false,
)
.expect("privacy request should decode");
if restricted {
switchyard_runner::mark_privacy_restricted(&mut request);
}
let execution = runtime
.execute_buffered(WireFormat::OpenAiChat, request)
.await;
assert_eq!(execution.result.is_ok(), succeeds);
let marks = execution
.events
.iter()
.filter_map(|event| match event {
RoutingEvent::Mark(mark) => Some(mark),
RoutingEvent::Metric(_) => None,
})
.collect::<Vec<_>>();
let privacy = marks
.iter()
.position(|mark| mark.name == "switchyard.privacy.decision")
.expect("privacy mark should be emitted");
assert_eq!(marks[privacy].data, expected);
if !succeeds {
let error = marks
.iter()
.position(|mark| mark.name == "switchyard.routing.error")
.expect("failure should emit an error mark");
assert!(privacy < error);
}
}
}

// Answer failures must not restore a public candidate, including on the streaming path.
#[tokio::test]
async fn private_target_flags_restrict_buffered_and_streaming_fallbacks() {
for streaming in [false, true] {
let server = MockServer::start().await;
Mock::given(body_partial_json(json!({"model": "restricted/model"})))
.respond_with(ResponseTemplate::new(503).set_body_string("unavailable"))
.expect(1)
.mount(&server)
.await;
let response = json!({
"id": "chatcmpl_private",
"object": if streaming { "chat.completion.chunk" } else { "chat.completion" },
"model": "backup/model",
"choices": [{
"index": 0,
(if streaming { "delta" } else { "message" }): {"role": "assistant", "content": "ok"},
"finish_reason": "stop",
}],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
});
let response = if streaming {
ResponseTemplate::new(200).set_body_raw(
format!("data: {response}\n\ndata: [DONE]\n\n"),
"text/event-stream",
)
} else {
ResponseTemplate::new(200).set_body_json(response)
};
Mock::given(body_partial_json(json!({"model": "backup/model"})))
.respond_with(response)
.expect(1)
.mount(&server)
.await;
let runtime = privacy_runtime(&format!("{}/v1", server.uri()));
let mut request = runtime
.decode_request(
WireFormat::OpenAiChat,
RelayRequest {
headers: Map::new(),
content: json!({
"model": "switchyard/privacy",
"messages": [{"role": "user", "content": "hello"}],
"stream": streaming,
}),
},
streaming,
)
.expect("privacy request should decode");
switchyard_runner::mark_privacy_restricted(&mut request);
let events = if streaming {
let execution = runtime
.execute_stream(WireFormat::OpenAiChat, request, Arc::new(|_| {}))
.await;
let chunks = execution
.result
.expect("private fallback should stream")
.collect::<Vec<_>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()
.expect("stream should encode");
assert!(
chunks
.iter()
.any(|chunk| chunk["choices"][0]["delta"]["content"] == "ok")
);
execution.events
} else {
let execution = runtime
.execute_buffered(WireFormat::OpenAiChat, request)
.await;
assert_eq!(
execution.result.expect("private fallback should answer")["model"],
"backup/model"
);
execution.events
};
let decision = events
.iter()
.find_map(|event| match event {
RoutingEvent::Mark(mark) if mark.name == "switchyard.routing.decision" => {
Some(&mark.data)
}
_ => None,
})
.expect("routing decision should be emitted");
assert_eq!(decision["selected_model"], "restricted/model");
assert_eq!(decision["served_model"], "backup/model");
assert_eq!(decision["fallback_used"], true);
let requests = server
.received_requests()
.await
.expect("requests should be recorded");
let models = requests
.iter()
.map(|request| {
serde_json::from_slice::<Json>(&request.body).expect("request JSON")["model"]
.clone()
})
.collect::<Vec<_>>();
assert_eq!(models, ["restricted/model", "backup/model"]);
}
}

#[tokio::test]
async fn responses_requests_use_the_configured_target_format() {
for (format, endpoint, upstream_response) in [
Expand Down
2 changes: 1 addition & 1 deletion crates/switchyard-runner/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ pub use algorithm::{
pub use failure::{RouteErrorKind, RouteErrorPhase, RouteErrorSummary, stream_error_summary};
// Re-exported because `Route::new` takes it, so a host wiring routes does not need a libsy dep.
pub use libsy::RuntimeModels;
pub use privacy::mark_privacy_restricted;
pub use privacy::{PrivacyDecision, mark_privacy_restricted};
pub use provider_key_redactor::ProviderKeyRedactor;
pub use route::{
AuxiliaryTarget, CallerAuthKind, ModelCapabilities, Route, RunOutput, RunnerError,
Expand Down
12 changes: 6 additions & 6 deletions crates/switchyard-runner/src/privacy.rs
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ pub(crate) fn selected_lane(request: &Request) -> Option<PrivacyLane> {
}

/// Bounded explanation of the privacy lane selected for one request.
pub(crate) struct PrivacyDecision {
pub struct PrivacyDecision {
lane: PrivacyLane,
source: PrivacySource,
reason_code: &'static str,
Expand All @@ -93,27 +93,27 @@ impl PrivacySource {

impl PrivacyDecision {
/// Returns the selected execution lane.
pub(crate) fn lane(&self) -> &'static str {
pub fn lane(&self) -> &'static str {
self.lane.as_str()
}

/// Returns the policy component that selected the lane.
pub(crate) fn source(&self) -> &'static str {
pub fn source(&self) -> &'static str {
self.source.as_str()
}

/// Returns a bounded reason for the decision.
pub(crate) const fn reason_code(&self) -> &'static str {
pub const fn reason_code(&self) -> &'static str {
self.reason_code
}

/// Returns the classifier probability assigned to clear content, when available.
pub(crate) const fn clear_score(&self) -> Option<f64> {
pub const fn clear_score(&self) -> Option<f64> {
self.clear_score
}

/// Returns the minimum clear score required for the standard lane, when applicable.
pub(crate) const fn clear_threshold(&self) -> Option<f64> {
pub const fn clear_threshold(&self) -> Option<f64> {
self.clear_threshold
}

Expand Down
Loading
Loading