From dd87be35801cfe190000a8785278fddf311ab6c7 Mon Sep 17 00:00:00 2001 From: ayushag Date: Fri, 10 Jul 2026 01:01:48 -0700 Subject: [PATCH 1/7] feat(libsy): add agent-aware model pool routing Signed-off-by: ayushag --- Cargo.lock | 18 + Cargo.toml | 1 + crates/libsy-protocol/src/envelope.rs | 19 +- crates/libsy/Cargo.toml | 1 + crates/libsy/README.md | 32 ++ crates/libsy/src/agentic.rs | 728 ++++++++++++++++++++++++++ crates/libsy/src/lib.rs | 8 +- demo/libsy-proxy/Cargo.toml | 26 + demo/libsy-proxy/README.md | 36 ++ demo/libsy-proxy/src/main.rs | 415 +++++++++++++++ 10 files changed, 1280 insertions(+), 4 deletions(-) create mode 100644 crates/libsy/src/agentic.rs create mode 100644 demo/libsy-proxy/Cargo.toml create mode 100644 demo/libsy-proxy/README.md create mode 100644 demo/libsy-proxy/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 18749f29..5a06fd77 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -856,6 +856,7 @@ dependencies = [ "async-trait", "futures", "rand 0.8.6", + "serde", "serde_json", "switchyard-protocol", "tokio", @@ -874,6 +875,23 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "libsy-proxy" +version = "0.1.0" +dependencies = [ + "async-trait", + "axum", + "libsy", + "serde_json", + "switchyard-components", + "switchyard-components-v2", + "switchyard-core", + "switchyard-protocol", + "switchyard-server", + "switchyard-translation", + "tokio", +] + [[package]] name = "libyaml-rs" version = "0.3.0" diff --git a/Cargo.toml b/Cargo.toml index 582ea8df..a7847a15 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,7 @@ members = [ "crates/switchyard-server", "crates/switchyard-skill-distillation", "crates/switchyard-translation", + "demo/libsy-proxy", ] [workspace.package] diff --git a/crates/libsy-protocol/src/envelope.rs b/crates/libsy-protocol/src/envelope.rs index 9906bd86..eef02e89 100644 --- a/crates/libsy-protocol/src/envelope.rs +++ b/crates/libsy-protocol/src/envelope.rs @@ -20,7 +20,7 @@ pub struct Context { /// All fields are optional; algorithms and observers use whichever are present /// (e.g. to key per-session state or emit correlated telemetry). `extra_metadata` /// is a free-form escape hatch for host-specific keys. -#[derive(Clone)] +#[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct Metadata { /// Stable id for a multi-request session/conversation. pub session_id: Option, @@ -28,6 +28,8 @@ pub struct Metadata { pub agent_id: Option, /// Id of the task the request belongs to. pub task_id: Option, + /// Agent-specific lineage and semantic routing signals. + pub agent_context: Option>, /// External trace/request id for joining with the host's telemetry. pub correlation_id: Option, /// Arbitrary host-defined key/value metadata. @@ -38,6 +40,21 @@ pub struct Metadata { pub wire_format: Option, } +/// Optional lineage and semantic signals for agent-aware routing. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct AgentContext { + /// Id of the parent agent, when this request comes from a child agent. + pub parent_agent_id: Option, + /// Harness-defined kind of agent call, such as `collab_spawn` or `review`. + pub agent_kind: Option, + /// Semantic agent role, such as `explorer`, `worker`, or `reviewer`. + pub agent_role: Option, + /// Semantic task class supplied by the harness or a prior classifier. + pub task_kind: Option, + /// Id of the current agent turn. + pub turn_id: Option, +} + /// A request an algorithm routes: the normalized [`LlmRequest`] plus the original /// provider payload and correlation [`Metadata`]. #[derive(Clone)] diff --git a/crates/libsy/Cargo.toml b/crates/libsy/Cargo.toml index df830383..810629cf 100644 --- a/crates/libsy/Cargo.toml +++ b/crates/libsy/Cargo.toml @@ -16,6 +16,7 @@ async-trait = "0.1" serde_json = "1" futures = "0.3" rand = "0.8" +serde = { version = "1", features = ["derive"] } switchyard-protocol = { path = "../libsy-protocol" } tokio = { version = "1", features = ["full"] } tokio-stream = "0.1" diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 36f28495..463775bf 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -40,6 +40,38 @@ println!("answer: {}", completion_text(&response.llm_response.aggregate().await? Runnable: [`research_agent`](../libsy-examples/examples/research_agent.rs) (in the `libsy-examples` crate). +## Agent- and subtask-aware routing + +`agentic::AgentAwareOrchAlgo` classifies the first request for a stable agent/task +against an arbitrary model pool, then reuses that assignment. This keeps a child +agent and its prompt-cache locality on one model while allowing sibling agents in +the same root session to use different models. Invalid or failed classifier calls +fall back without caching the failure. + +```rust +use libsy::agentic::{AgentAwareOrchAlgo, AgentRoutingCandidate}; + +let algo: Arc = Arc::new(AgentAwareOrchAlgo::new( + "classifier", + vec![ + AgentRoutingCandidate::new("frontier", "planning, synthesis, and review"), + AgentRoutingCandidate::new("fast", "bounded research and mechanical edits"), + ], + "frontier", // fail-open target + targets, +)); +``` + +`metadata_from_headers` converts harness-specific identity into neutral `Metadata`. +Supported inputs include Codex `session-id`, `thread-id`, +`x-codex-parent-thread-id`, `x-openai-subagent`, and `x-codex-turn-metadata`; +NeMo Relay `x-nemo-relay-session-id` / `x-nemo-relay-subagent-id`; Dynamo +`x-dynamo-session-id` / `x-dynamo-parent-session-id`; and explicit +`x-switchyard-*` overrides. The adapter stays local and dependency-free because +Relay's matching gateway normalizer is not exposed through its public Rust library API. + +See [`demo/libsy-proxy`](../../demo/libsy-proxy) for a runnable Switchyard HTTP proxy. + ## Requests & responses ```rust diff --git a/crates/libsy/src/agentic.rs b/crates/libsy/src/agentic.rs new file mode 100644 index 00000000..1125b64a --- /dev/null +++ b/crates/libsy/src/agentic.rs @@ -0,0 +1,728 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Agent- and subtask-aware model routing. +//! +//! [`metadata_from_headers`] normalizes identity signals emitted by Codex, NeMo Relay, +//! Dynamo, or explicit Switchyard headers into neutral [`Metadata`]. +//! [`AgentAwareOrchAlgo`] classifies a stable agent/task once, then reuses that model +//! assignment to retain model and prompt-cache locality. + +use std::collections::{BTreeMap, HashMap}; +use std::error::Error; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use serde::Deserialize; +use switchyard_protocol::{completion_text, prompt_text, text_request, AgentContext}; + +use crate::{ + Algorithm, Context, Decision, Driver, LlmTargetSet, Metadata, Request, Response, Signals, +}; + +const CODEX_TURN_METADATA_HEADER: &str = "x-codex-turn-metadata"; +const MAX_ASSIGNMENTS: usize = 4096; + +type BoxErr = Box; + +/// One model available to the agent-aware classifier. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AgentRoutingCandidate { + /// Semantic target name used to look up the model in [`LlmTargetSet`]. + pub target: String, + /// Capability/cost guidance shown to the classifier. + pub description: String, +} + +impl AgentRoutingCandidate { + /// Creates a model-pool entry from a target name and classifier guidance. + pub fn new(target: impl Into, description: impl Into) -> Self { + Self { + target: target.into(), + description: description.into(), + } + } +} + +/// Inspectable decision emitted for classifier and routed calls. +pub struct AgentRoutingDecision { + /// Target selected from the configured model pool. + pub selected_model: String, + /// Human-readable classifier, cache, or fallback explanation. + pub reason: String, + /// Classifier-inferred task class, when supplied. + pub task_kind: Option, + /// Classifier confidence in `[0, 1]`, when supplied. + pub confidence: Option, + /// Stable agent id used for the assignment, when available. + pub agent_id: Option, + /// Whether the selection came from an existing per-agent assignment. + pub cache_hit: bool, +} + +impl Decision for AgentRoutingDecision { + fn selected_model(&self) -> &str { + &self.selected_model + } + + fn reasoning(&self) -> Option<&str> { + Some(&self.reason) + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } +} + +/// LLM classifier over a model pool with per-agent/task assignment affinity. +pub struct AgentAwareOrchAlgo { + classifier_model: String, + candidates: Vec, + fallback_model: String, + target_set: LlmTargetSet, + assignments: Mutex>, +} + +impl AgentAwareOrchAlgo { + /// Configures the classifier target, candidate pool, fail-open target, and targets. + pub fn new( + classifier_model: impl Into, + candidates: Vec, + fallback_model: impl Into, + target_set: LlmTargetSet, + ) -> Self { + Self { + classifier_model: classifier_model.into(), + candidates, + fallback_model: fallback_model.into(), + target_set, + assignments: Mutex::new(HashMap::new()), + } + } + + /// Classifies a request, failing open without caching invalid or failed results. + async fn classify( + &self, + ctx: &Context, + driver: &Driver, + request: &Request, + ) -> Result<(Assignment, bool), BoxErr> { + let classifier_decision: Arc = Arc::new(AgentRoutingDecision { + selected_model: self.classifier_model.clone(), + reason: format!("classifying agent/subtask via {}", self.classifier_model), + task_kind: request + .metadata + .as_ref() + .and_then(|metadata| metadata.agent_context.as_deref()) + .and_then(|agent| agent.task_kind.clone()), + confidence: None, + agent_id: request + .metadata + .as_ref() + .and_then(|metadata| metadata.agent_id.clone()), + cache_hit: false, + }); + driver + .info(ctx.clone(), Arc::clone(&classifier_decision)) + .await?; + + let classify_request = Request { + llm_request: text_request( + request.llm_request.model.clone(), + classifier_prompt( + &self.candidates, + request.metadata.as_ref(), + &prompt_text(&request.llm_request), + ), + ), + // Classifier calls receive neutral text and metadata, never the provider + // payload with its tools or continuation state. + raw_request: None, + metadata: request.metadata.clone(), + }; + + let output = match self.target_set.get_target(&self.classifier_model) { + Ok(target) => { + driver + .call_llm_target(ctx.clone(), &target, classify_request, classifier_decision) + .await + } + Err(error) => Err(error), + }; + + match output { + Ok(response) => match response.llm_response.into_agg().await { + Ok(response) => { + match parse_classifier_output(&completion_text(&response), &self.candidates) { + Some(assignment) => Ok((assignment, true)), + None => Ok(( + self.fallback_assignment( + "classifier returned an invalid model-pool decision", + ), + false, + )), + } + } + Err(error) => Ok(( + self.fallback_assignment(&format!("classifier response failed: {error}")), + false, + )), + }, + Err(error) => Ok(( + self.fallback_assignment(&format!("classifier call failed: {error}")), + false, + )), + } + } + + fn fallback_assignment(&self, reason: &str) -> Assignment { + Assignment { + model: self.fallback_model.clone(), + reason: format!("{reason}; fell back to {}", self.fallback_model), + task_kind: None, + confidence: None, + } + } + + fn cached_assignment(&self, key: &AssignmentKey) -> Option { + self.assignments + .lock() + .ok() + .and_then(|assignments| assignments.get(key).cloned()) + } + + fn store_assignment(&self, key: AssignmentKey, assignment: Assignment) { + if let Ok(mut assignments) = self.assignments.lock() { + if !assignments.contains_key(&key) && assignments.len() >= MAX_ASSIGNMENTS { + if let Some(evicted) = assignments.keys().next().cloned() { + assignments.remove(&evicted); + } + } + assignments.insert(key, assignment); + } + } +} + +#[async_trait] +impl Algorithm for AgentAwareOrchAlgo { + async fn create_run_task( + self: Arc, + ctx: Context, + driver: Driver, + request: Request, + ) -> Result { + let key = assignment_key(request.metadata.as_ref()); + let cached = key.as_ref().and_then(|key| self.cached_assignment(key)); + let (assignment, cache_hit) = match cached { + Some(assignment) => (assignment, true), + None => { + let (assignment, cacheable) = self.classify(&ctx, &driver, &request).await?; + if cacheable { + if let Some(key) = key { + self.store_assignment(key, assignment.clone()); + } + } + (assignment, false) + } + }; + + let routed_target = self.target_set.get_target(&assignment.model)?; + let route_decision: Arc = Arc::new(AgentRoutingDecision { + selected_model: assignment.model.clone(), + reason: if cache_hit { + format!("reused stable agent/task assignment: {}", assignment.reason) + } else { + assignment.reason.clone() + }, + task_kind: assignment.task_kind, + confidence: assignment.confidence, + agent_id: request + .metadata + .as_ref() + .and_then(|metadata| metadata.agent_id.clone()), + cache_hit, + }); + driver + .info(ctx.clone(), Arc::clone(&route_decision)) + .await?; + driver + .call_llm_target(ctx, &routed_target, request, route_decision) + .await + } + + async fn process_signals(self: Arc, _signals: Signals) -> Result<(), BoxErr> { + Ok(()) + } +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct AssignmentKey { + session_id: Option, + agent_id: String, + task_id: Option, +} + +#[derive(Clone, Debug)] +struct Assignment { + model: String, + reason: String, + task_kind: Option, + confidence: Option, +} + +#[derive(Deserialize)] +struct ClassifierOutput { + model: String, + #[serde(default)] + task_kind: Option, + #[serde(default)] + confidence: Option, + #[serde(default)] + reason: Option, +} + +#[derive(Default, Deserialize)] +struct CodexTurnMetadata { + session_id: Option, + thread_id: Option, + parent_thread_id: Option, + turn_id: Option, + subagent_kind: Option, + agent_role: Option, + task_id: Option, + task_kind: Option, +} + +/// Normalizes harness-specific request headers into libsy metadata. +/// +/// Explicit `x-switchyard-*` headers win. NeMo Relay and Dynamo correlation +/// headers are accepted without linking either runtime. Codex's structured turn +/// metadata is preferred over its compatibility projections. +pub fn metadata_from_headers(headers: &BTreeMap>) -> Metadata { + let codex = header(headers, CODEX_TURN_METADATA_HEADER) + .and_then(|value| serde_json::from_str::(&value).ok()) + .unwrap_or_default(); + + let agent_context = AgentContext { + parent_agent_id: first_some([ + header(headers, "x-switchyard-parent-agent-id"), + header(headers, "x-dynamo-parent-session-id"), + codex.parent_thread_id, + header(headers, "x-codex-parent-thread-id"), + ]), + agent_kind: first_some([ + header(headers, "x-switchyard-agent-kind"), + codex.subagent_kind, + header(headers, "x-openai-subagent"), + ]), + agent_role: first_some([header(headers, "x-switchyard-agent-role"), codex.agent_role]), + task_kind: first_some([header(headers, "x-switchyard-task-kind"), codex.task_kind]), + turn_id: first_some([header(headers, "x-switchyard-turn-id"), codex.turn_id]), + }; + let agent_context = (agent_context != AgentContext::default()).then(|| Box::new(agent_context)); + + Metadata { + session_id: first_some([ + header(headers, "x-switchyard-session-id"), + header(headers, "x-nemo-relay-session-id"), + codex.session_id, + header(headers, "session-id"), + ]), + agent_id: first_some([ + header(headers, "x-switchyard-agent-id"), + header(headers, "x-nemo-relay-subagent-id"), + header(headers, "x-dynamo-session-id"), + codex.thread_id, + header(headers, "thread-id"), + ]), + task_id: first_some([ + header(headers, "x-switchyard-task-id"), + codex.task_id, + header(headers, "x-task-id"), + ]), + agent_context, + correlation_id: first_some([ + header(headers, "x-request-id"), + header(headers, "x-client-request-id"), + ]), + ..Metadata::default() + } +} + +fn assignment_key(metadata: Option<&Metadata>) -> Option { + let metadata = metadata?; + Some(AssignmentKey { + session_id: metadata.session_id.clone(), + agent_id: metadata.agent_id.clone()?, + task_id: metadata.task_id.clone(), + }) +} + +fn classifier_prompt( + candidates: &[AgentRoutingCandidate], + metadata: Option<&Metadata>, + prompt: &str, +) -> String { + let metadata = metadata.cloned().unwrap_or_default(); + let agent = metadata + .agent_context + .as_deref() + .cloned() + .unwrap_or_default(); + let candidate_lines = candidates + .iter() + .map(|candidate| format!("- {}: {}", candidate.target, candidate.description)) + .collect::>() + .join("\n"); + format!( + "You route one model call inside an agent hierarchy. Select exactly one target from the \n\ + model pool. Prefer the cheapest candidate likely to complete the subtask correctly. \n\ + Return JSON only: {{\"model\":\"\",\"task_kind\":\"\",\ + \"confidence\":<0..1>,\"reason\":\"\"}}.\n\n\ + Agent context:\n\ + session_id={:?}\nagent_id={:?}\nparent_agent_id={:?}\nagent_kind={:?}\n\ + agent_role={:?}\ntask_id={:?}\ntask_kind={:?}\nturn_id={:?}\n\n\ + Model pool:\n{}\n\nSubtask:\n\n{}\n", + metadata.session_id, + metadata.agent_id, + agent.parent_agent_id, + agent.agent_kind, + agent.agent_role, + metadata.task_id, + agent.task_kind, + agent.turn_id, + candidate_lines, + prompt, + ) +} + +fn parse_classifier_output( + output: &str, + candidates: &[AgentRoutingCandidate], +) -> Option { + let trimmed = output.trim(); + if let Some(candidate) = candidates + .iter() + .find(|candidate| candidate.target == trimmed) + { + return Some(Assignment { + model: candidate.target.clone(), + reason: "classifier selected the target by name".to_string(), + task_kind: None, + confidence: None, + }); + } + + let start = trimmed.find('{')?; + let end = trimmed.rfind('}')?; + let parsed = serde_json::from_str::(&trimmed[start..=end]).ok()?; + let model = parsed.model.trim(); + if !candidates.iter().any(|candidate| candidate.target == model) { + return None; + } + Some(Assignment { + model: model.to_string(), + reason: parsed + .reason + .filter(|reason| !reason.trim().is_empty()) + .unwrap_or_else(|| "classifier selected the target".to_string()), + task_kind: parsed.task_kind.filter(|kind| !kind.trim().is_empty()), + confidence: parsed + .confidence + .filter(|value| value.is_finite()) + .map(|value| value.clamp(0.0, 1.0)), + }) +} + +fn header(headers: &BTreeMap>, name: &str) -> Option { + headers + .iter() + .find(|(key, _)| key.eq_ignore_ascii_case(name)) + .and_then(|(_, values)| values.iter().find(|value| !value.trim().is_empty())) + .map(|value| value.trim().to_string()) +} + +fn first_some(values: [Option; N]) -> Option { + values.into_iter().flatten().next() +} + +#[cfg(test)] +mod tests { + use std::collections::VecDeque; + + use super::*; + use crate::{LlmClient, LlmResponse, LlmTarget, RoutedRequest}; + use switchyard_protocol::{text_response, Metadata}; + + struct RoutingClient { + classifier_calls: Arc>, + responses: Arc>>, + } + + #[async_trait] + impl LlmClient for RoutingClient { + async fn call(&self, routed: RoutedRequest) -> Result { + let selected = routed.decision.selected_model().to_string(); + let completion = if selected == "classifier" { + let mut calls = self.classifier_calls.lock().map_err(|_| "lock poisoned")?; + *calls += 1; + self.responses + .lock() + .map_err(|_| "lock poisoned")? + .pop_front() + .ok_or("missing classifier response")? + } else { + selected + }; + Ok(Response { + llm_response: LlmResponse::Agg(text_response(None, completion)), + metadata: routed.request.metadata, + }) + } + } + + fn algo(responses: Vec<&str>) -> (Arc, Arc>) { + let classifier_calls = Arc::new(Mutex::new(0)); + let client = Arc::new(RoutingClient { + classifier_calls: Arc::clone(&classifier_calls), + responses: Arc::new(Mutex::new( + responses.into_iter().map(str::to_string).collect(), + )), + }) as Arc; + let target = |name: &str| LlmTarget { + semantic_name: name.to_string(), + llm_client: Some(Arc::clone(&client)), + }; + let targets = LlmTargetSet::new(vec![ + target("classifier"), + target("frontier"), + target("fast"), + ]); + let algo: Arc = Arc::new(AgentAwareOrchAlgo::new( + "classifier", + vec![ + AgentRoutingCandidate::new("frontier", "complex planning and final review"), + AgentRoutingCandidate::new("fast", "bounded research and mechanical edits"), + ], + "frontier", + targets, + )); + (algo, classifier_calls) + } + + fn request(agent_id: Option<&str>, task_id: Option<&str>, prompt: &str) -> Request { + Request { + llm_request: text_request(Some("auto".to_string()), prompt), + raw_request: Some(serde_json::json!({"input": prompt})), + metadata: Some(Metadata { + session_id: Some("session-1".to_string()), + agent_id: agent_id.map(str::to_string), + task_id: task_id.map(str::to_string), + agent_context: Some(Box::new(AgentContext { + agent_role: Some("explorer".to_string()), + ..AgentContext::default() + })), + ..Metadata::default() + }), + } + } + + fn decision(model: &str, task_kind: &str, confidence: f64, reason: &str) -> String { + serde_json::json!({ + "model": model, + "task_kind": task_kind, + "confidence": confidence, + "reason": reason, + }) + .to_string() + } + + fn response_text(response: &Response) -> String { + response + .llm_response + .as_agg() + .map(completion_text) + .unwrap_or_default() + } + + #[test] + fn normalizes_codex_turn_metadata_and_compatibility_headers() { + let mut headers = BTreeMap::new(); + headers.insert("session-id".to_string(), vec!["compat-session".to_string()]); + headers.insert("thread-id".to_string(), vec!["compat-agent".to_string()]); + headers.insert( + CODEX_TURN_METADATA_HEADER.to_string(), + vec![serde_json::json!({ + "session_id": "root-session", + "thread_id": "child-agent", + "parent_thread_id": "root-agent", + "turn_id": "turn-7", + "subagent_kind": "collab_spawn", + }) + .to_string()], + ); + + let metadata = metadata_from_headers(&headers); + assert_eq!(metadata.session_id.as_deref(), Some("root-session")); + assert_eq!(metadata.agent_id.as_deref(), Some("child-agent")); + let agent = metadata.agent_context.as_deref(); + assert_eq!( + agent.and_then(|value| value.parent_agent_id.as_deref()), + Some("root-agent") + ); + assert_eq!( + agent.and_then(|value| value.agent_kind.as_deref()), + Some("collab_spawn") + ); + assert_eq!( + agent.and_then(|value| value.turn_id.as_deref()), + Some("turn-7") + ); + } + + #[test] + fn explicit_switchyard_headers_override_relay_and_codex() { + let headers = BTreeMap::from([ + ( + "x-switchyard-session-id".to_string(), + vec!["canonical-session".to_string()], + ), + ( + "x-switchyard-agent-id".to_string(), + vec!["canonical-agent".to_string()], + ), + ( + "x-nemo-relay-session-id".to_string(), + vec!["relay-session".to_string()], + ), + ( + "x-dynamo-session-id".to_string(), + vec!["relay-agent".to_string()], + ), + ]); + + let metadata = metadata_from_headers(&headers); + assert_eq!(metadata.session_id.as_deref(), Some("canonical-session")); + assert_eq!(metadata.agent_id.as_deref(), Some("canonical-agent")); + } + + #[test] + fn normalizes_relay_and_dynamo_headers_without_runtime_dependency() { + let headers = BTreeMap::from([ + ( + "x-nemo-relay-session-id".to_string(), + vec!["relay-session".to_string()], + ), + ( + "x-nemo-relay-subagent-id".to_string(), + vec!["relay-child".to_string()], + ), + ( + "x-dynamo-parent-session-id".to_string(), + vec!["relay-parent".to_string()], + ), + ]); + + let metadata = metadata_from_headers(&headers); + assert_eq!(metadata.session_id.as_deref(), Some("relay-session")); + assert_eq!(metadata.agent_id.as_deref(), Some("relay-child")); + assert_eq!( + metadata + .agent_context + .as_deref() + .and_then(|agent| agent.parent_agent_id.as_deref()), + Some("relay-parent") + ); + } + + #[tokio::test] + async fn valid_classifier_output_routes_from_the_pool() -> Result<(), BoxErr> { + let (algo, calls) = algo(vec![&decision("fast", "research", 0.9, "bounded lookup")]); + let (trace, response) = algo + .run( + Context::default(), + request(Some("agent-a"), None, "survey the API"), + ) + .await?; + + assert_eq!(response_text(&response), "fast"); + assert_eq!(trace.len(), 2); + let routed = trace[1] + .as_any() + .downcast_ref::() + .ok_or("expected AgentRoutingDecision")?; + assert_eq!(routed.task_kind.as_deref(), Some("research")); + assert_eq!(routed.confidence, Some(0.9)); + assert!(!routed.cache_hit); + assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 1); + Ok(()) + } + + #[tokio::test] + async fn stable_agent_assignment_avoids_reclassification() -> Result<(), BoxErr> { + let (algo, calls) = algo(vec![&decision("fast", "research", 0.8, "explorer")]); + Arc::clone(&algo) + .run( + Context::default(), + request(Some("agent-a"), None, "first turn"), + ) + .await?; + let (trace, response) = algo + .run( + Context::default(), + request(Some("agent-a"), None, "second turn"), + ) + .await?; + + assert_eq!(response_text(&response), "fast"); + assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 1); + assert_eq!(trace.len(), 1); + let routed = trace[0] + .as_any() + .downcast_ref::() + .ok_or("expected AgentRoutingDecision")?; + assert!(routed.cache_hit); + Ok(()) + } + + #[tokio::test] + async fn explicit_task_change_reclassifies_the_same_agent() -> Result<(), BoxErr> { + let first = decision("fast", "research", 0.8, "lookup"); + let second = decision("frontier", "review", 0.9, "final review"); + let (algo, calls) = algo(vec![&first, &second]); + let (_, first_response) = Arc::clone(&algo) + .run( + Context::default(), + request(Some("agent-a"), Some("task-1"), "collect evidence"), + ) + .await?; + let (_, second_response) = algo + .run( + Context::default(), + request(Some("agent-a"), Some("task-2"), "adjudicate findings"), + ) + .await?; + + assert_eq!(response_text(&first_response), "fast"); + assert_eq!(response_text(&second_response), "frontier"); + assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 2); + Ok(()) + } + + #[tokio::test] + async fn invalid_decision_falls_back_without_poisoning_affinity() -> Result<(), BoxErr> { + let valid = decision("fast", "research", 0.7, "retry succeeded"); + let (algo, calls) = algo(vec!["not-json", &valid]); + let (_, first) = Arc::clone(&algo) + .run(Context::default(), request(Some("agent-a"), None, "first")) + .await?; + let (_, second) = algo + .run(Context::default(), request(Some("agent-a"), None, "second")) + .await?; + + assert_eq!(response_text(&first), "frontier"); + assert_eq!(response_text(&second), "fast"); + assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 2); + Ok(()) + } +} diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 9d107233..0176342a 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -55,12 +55,14 @@ //! ## Algorithms //! //! [`RandomAlgo`] provides uniform random routing in the core crate. Worked -//! implementations of an LLM classifier and a stateful ensemble, plus runnable -//! agents, live in the `libsy-examples` crate. +//! [`agentic::AgentAwareOrchAlgo`] provides sub-agent-aware model-pool routing and +//! neutral harness-header normalization. Worked implementations of an LLM classifier +//! and a stateful ensemble, plus runnable agents, live in the `libsy-examples` crate. mod algorithms; pub use algorithms::noop::{NoopAlgo, NoopDecision}; pub use algorithms::rand::{RandomAlgo, RandomDecision}; +pub mod agentic; mod driver; @@ -74,7 +76,7 @@ use futures::{Stream, StreamExt}; /// [`LlmResponseChunk`] is one streaming event; [`LlmResponse`] is the streamed response /// (a live [`LlmResponseStream`] or the terminal aggregate). pub use switchyard_protocol::{ - AggLlmResponse, Context, Decision, LlmRequest, LlmResponse, LlmResponseChunk, + AgentContext, AggLlmResponse, Context, Decision, LlmRequest, LlmResponse, LlmResponseChunk, LlmResponseStream, Metadata, Request, Response, RoutedLlmClient, }; diff --git a/demo/libsy-proxy/Cargo.toml b/demo/libsy-proxy/Cargo.toml new file mode 100644 index 00000000..cb09ab40 --- /dev/null +++ b/demo/libsy-proxy/Cargo.toml @@ -0,0 +1,26 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +[package] +name = "libsy-proxy" +version = "0.1.0" +description = "Demo LLM proxy with libsy agent-aware routing" +authors.workspace = true +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true + +[dependencies] +libsy = { path = "../../crates/libsy" } +switchyard-components = { path = "../../crates/switchyard-components" } +switchyard-components-v2 = { path = "../../crates/switchyard-components-v2" } +switchyard-core = { path = "../../crates/switchyard-core" } +switchyard-protocol = { path = "../../crates/libsy-protocol" } +switchyard-server = { path = "../../crates/switchyard-server" } +switchyard-translation = { path = "../../crates/switchyard-translation" } + +async-trait = "0.1" +axum = "0.8" +serde_json = "1" +tokio = { version = "1", features = ["macros", "rt-multi-thread", "net", "signal"] } diff --git a/demo/libsy-proxy/README.md b/demo/libsy-proxy/README.md new file mode 100644 index 00000000..0b2aa39f --- /dev/null +++ b/demo/libsy-proxy/README.md @@ -0,0 +1,36 @@ +# Agent-aware libsy proxy POC + +This demo embeds `libsy`'s agent-aware model-pool router in the PR #17 Switchyard server. It accepts +OpenAI Chat, Anthropic Messages, and OpenAI Responses request shapes, normalizes Codex/Relay lineage +headers, asks the classifier for a model assignment, and reuses valid assignments per stable +`(session, agent, explicit task)` key. + +Run it with an NVIDIA Inference Hub bearer token: + +```bash +ANTHROPIC_API_KEY="$INFERENCE_HUB_SY_API_KEY" cargo run -p libsy-proxy +``` + +Send two buffered calls for the same Codex child thread. The first call classifies; the second reuses +the assignment: + +```bash +curl http://127.0.0.1:4000/v1/chat/completions \ + -H 'content-type: application/json' \ + -H 'session-id: root-session-1' \ + -H 'thread-id: child-thread-1' \ + -H 'x-codex-parent-thread-id: root-thread-1' \ + -H 'x-openai-subagent: collab_spawn' \ + -H 'x-codex-turn-metadata: {"session_id":"root-session-1","thread_id":"child-thread-1","parent_thread_id":"root-thread-1","turn_id":"turn-1","subagent_kind":"collab_spawn"}' \ + -d '{"model":"libsy-agent-aware","messages":[{"role":"user","content":"Survey the request protocol and report evidence."}],"stream":false}' +``` + +The response exposes the chosen logical target and rationale in `x-model-router-*` headers. Explicit +`x-switchyard-session-id`, `x-switchyard-agent-id`, `x-switchyard-parent-agent-id`, +`x-switchyard-agent-role`, `x-switchyard-task-id`, and `x-switchyard-task-kind` headers override +harness-derived values. + +The demo currently requires buffered upstream responses (`"stream": false`). Header normalization +and route selection work for Responses-shaped requests, but running an unmodified streaming Codex +CLI through this demo requires a follow-up adapter between Switchyard's SSE response and libsy's +neutral response stream. diff --git a/demo/libsy-proxy/src/main.rs b/demo/libsy-proxy/src/main.rs new file mode 100644 index 00000000..2992271a --- /dev/null +++ b/demo/libsy-proxy/src/main.rs @@ -0,0 +1,415 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Demo proxy combining Switchyard HTTP/translation with libsy agent-aware routing. +//! +//! The server accepts OpenAI Chat, Anthropic Messages, and OpenAI Responses requests. +//! libsy normalizes harness identity headers, assigns a stable agent/task to a model +//! pool target, and makes calls through Switchyard's OpenAI-compatible backend. + +use std::collections::BTreeMap; +use std::error::Error; +use std::net::SocketAddr; +use std::sync::Arc; + +use async_trait::async_trait; +use serde_json::Value; + +use libsy::agentic::{ + metadata_from_headers, AgentAwareOrchAlgo, AgentRoutingCandidate, AgentRoutingDecision, +}; +use libsy::{ + Algorithm, Context, Decision, LlmClient, LlmResponse, LlmTarget, LlmTargetSet, Request, + Response, RoutedRequest, +}; +use switchyard_components::OpenAiPassthroughBackend; +use switchyard_components_v2::{Profile, ProfileInput, ProfileResponse, RoutingMetadata}; +use switchyard_core::{ + ChatRequest, ChatRequestType, ChatResponse, EndpointConfig, LlmBackend, ModelId, ProxyContext, + Result, SwitchyardError, +}; +use switchyard_server::{build_switchyard_router, ProfileRegistry, ServerState}; +use switchyard_translation::{TranslationEngine, TranslationPolicy, WireFormat}; + +const CLASSIFIER_MODEL: &str = "nvidia/deepseek-ai/deepseek-v4-flash"; +const FRONTIER_MODEL: &str = "aws/anthropic/bedrock-claude-opus-4-7"; +const FAST_MODEL: &str = "nvidia/deepseek-ai/deepseek-v4-flash"; +const CLASSIFIER_TARGET: &str = "classifier"; +const FRONTIER_TARGET: &str = "frontier"; +const FAST_TARGET: &str = "fast"; + +const DEFAULT_BASE_URL: &str = "https://inference-api.nvidia.com/v1"; +const DEFAULT_ADDR: &str = "127.0.0.1:4000"; +const PROFILE_MODEL_ID: &str = "libsy-agent-aware"; + +type BoxErr = Box; + +/// libsy client backed by Switchyard's OpenAI-compatible backend. +struct SwitchyardBackendClient { + backend: Arc, + model_ids: BTreeMap, + translation: Arc, +} + +#[async_trait] +impl LlmClient for SwitchyardBackendClient { + async fn call(&self, routed: RoutedRequest) -> std::result::Result { + let target = routed.decision.selected_model(); + let model = self + .model_ids + .get(target) + .ok_or_else(|| format!("no provider model configured for target '{target}'"))?; + let chat_request = + chat_request_for_call(&routed.request, model, self.translation.as_ref())?; + + let mut ctx = ProxyContext::new(); + let response = self + .backend + .call(&mut ctx, &chat_request) + .await + .map_err(|error| BoxErr::from(error.to_string()))?; + let body = response.body().cloned().ok_or_else(|| { + BoxErr::from("the libsy proxy demo currently requires buffered upstream responses") + })?; + let decoded = self + .translation + .decode_response(WireFormat::OpenAiChat, &body, &TranslationPolicy::default()) + .map_err(|error| BoxErr::from(error.to_string()))?; + + Ok(Response { + llm_response: LlmResponse::Agg(decoded.response), + metadata: routed.request.metadata, + }) + } +} + +/// Switchyard profile that delegates all routing decisions to libsy. +struct LibsyAgentAwareProfile { + algorithm: Arc, + translation: Arc, +} + +#[async_trait] +impl Profile for LibsyAgentAwareProfile { + async fn run(&self, input: ProfileInput) -> Result { + let request = libsy_request(&input, self.translation.as_ref())?; + let (trace, response) = Arc::clone(&self.algorithm) + .run(Context::default(), request) + .await + .map_err(|error| SwitchyardError::Other(error.to_string()))?; + let aggregate = response + .llm_response + .into_agg() + .await + .map_err(|error| SwitchyardError::Other(error.to_string()))?; + let body = self + .translation + .encode_response( + WireFormat::OpenAiChat, + &aggregate, + &TranslationPolicy::default(), + ) + .map_err(|error| SwitchyardError::Other(error.to_string()))? + .body; + + Ok(ProfileResponse::with_routing_metadata( + ChatResponse::openai_completion(body), + routing_metadata(&trace), + )) + } +} + +/// Decode the inbound wire body and attach normalized harness metadata. +fn libsy_request(input: &ProfileInput, translation: &TranslationEngine) -> Result { + let format = wire_format(input.request.request_type()); + let decoded = translation + .decode_request(format, input.request.body(), &TranslationPolicy::default()) + .map_err(|error| SwitchyardError::InvalidRequest(error.to_string()))?; + let mut metadata = metadata_from_headers(&input.metadata.headers); + metadata + .extra_metadata + .get_or_insert_with(Default::default) + .insert( + "inbound_format".to_string(), + inbound_format(input.request.request_type()).to_string(), + ); + + Ok(Request { + llm_request: decoded.request, + raw_request: Some(input.request.body().clone()), + metadata: Some(metadata), + }) +} + +/// Preserve routed provider payloads; encode classifier calls from the neutral IR. +fn chat_request_for_call( + request: &Request, + model: &str, + translation: &TranslationEngine, +) -> std::result::Result { + let Some(mut body) = request.raw_request.clone() else { + let mut classifier_request = request.llm_request.clone(); + classifier_request.model = Some(model.to_string()); + classifier_request.stream = false; + let body = translation + .encode_request( + WireFormat::OpenAiChat, + &classifier_request, + &TranslationPolicy::default(), + ) + .map_err(|error| BoxErr::from(error.to_string()))? + .body; + return Ok(ChatRequest::openai_chat(body)); + }; + + if let Some(object) = body.as_object_mut() { + object.insert("model".to_string(), Value::String(model.to_string())); + } + let request = match request + .metadata + .as_ref() + .and_then(|metadata| metadata.extra_metadata.as_ref()) + .and_then(|extra| extra.get("inbound_format")) + .map(String::as_str) + { + Some("anthropic") => ChatRequest::anthropic(body), + Some("openai_responses") => ChatRequest::openai_responses(body), + _ => ChatRequest::openai_chat(body), + }; + Ok(request) +} + +fn inbound_format(request_type: ChatRequestType) -> &'static str { + match request_type { + ChatRequestType::OpenAiChat => "openai_chat", + ChatRequestType::Anthropic => "anthropic", + ChatRequestType::OpenAiResponses => "openai_responses", + } +} + +fn wire_format(request_type: ChatRequestType) -> WireFormat { + match request_type { + ChatRequestType::OpenAiChat => WireFormat::OpenAiChat, + ChatRequestType::Anthropic => WireFormat::AnthropicMessages, + ChatRequestType::OpenAiResponses => WireFormat::OpenAiResponses, + } +} + +/// Surface libsy's routing decision as `x-model-router-*` response headers. +fn routing_metadata(trace: &[Arc]) -> RoutingMetadata { + let decision = trace.last(); + let agent = + decision.and_then(|decision| decision.as_any().downcast_ref::()); + let selected_target = decision.map(|decision| decision.selected_model()); + RoutingMetadata { + selected_model: selected_target.map(|target| match target { + FRONTIER_TARGET => FRONTIER_MODEL.to_string(), + FAST_TARGET => FAST_MODEL.to_string(), + other => other.to_string(), + }), + selected_tier: selected_target.map(str::to_string), + confidence: agent.and_then(|decision| decision.confidence), + router_version: Some("libsy-agent-aware-v1".to_string()), + tolerance: None, + rationale: decision.and_then(|decision| decision.reasoning().map(str::to_string)), + } +} + +fn build_algorithm() -> Result> { + let base_url = + std::env::var("LIBSY_PROXY_BASE_URL").unwrap_or_else(|_| DEFAULT_BASE_URL.to_string()); + let api_key = std::env::var("ANTHROPIC_API_KEY").map_err(|_| { + SwitchyardError::InvalidConfig( + "ANTHROPIC_API_KEY must be set to the upstream bearer key".to_string(), + ) + })?; + let backend = Arc::new(OpenAiPassthroughBackend::new(EndpointConfig { + base_url: Some(base_url), + api_key: Some(api_key), + timeout_secs: Some(120.0), + })?); + let model_ids = BTreeMap::from([ + (CLASSIFIER_TARGET.to_string(), CLASSIFIER_MODEL.to_string()), + (FRONTIER_TARGET.to_string(), FRONTIER_MODEL.to_string()), + (FAST_TARGET.to_string(), FAST_MODEL.to_string()), + ]); + let translation = Arc::new(TranslationEngine::default()); + let client = Arc::new(SwitchyardBackendClient { + backend, + model_ids, + translation, + }) as Arc; + let target = |name: &str| LlmTarget { + semantic_name: name.to_string(), + llm_client: Some(Arc::clone(&client)), + }; + let targets = LlmTargetSet::new(vec![ + target(CLASSIFIER_TARGET), + target(FRONTIER_TARGET), + target(FAST_TARGET), + ]); + + Ok(Arc::new(AgentAwareOrchAlgo::new( + CLASSIFIER_TARGET, + vec![ + AgentRoutingCandidate::new( + FRONTIER_TARGET, + "frontier model for planning, ambiguous implementation, synthesis, and review", + ), + AgentRoutingCandidate::new( + FAST_TARGET, + "efficient model for bounded exploration, retrieval, and mechanical edits", + ), + ], + FRONTIER_TARGET, + targets, + ))) +} + +#[tokio::main] +async fn main() -> Result<()> { + let translation = Arc::new(TranslationEngine::default()); + let profile = Arc::new(LibsyAgentAwareProfile { + algorithm: build_algorithm()?, + translation, + }) as Arc; + let registry = ProfileRegistry::from_profiles([( + ModelId::new(PROFILE_MODEL_ID)?, + profile, + PROFILE_MODEL_ID.to_string(), + )])?; + let state = ServerState::new(registry); + let addr: SocketAddr = std::env::var("LIBSY_PROXY_ADDR") + .unwrap_or_else(|_| DEFAULT_ADDR.to_string()) + .parse() + .map_err(|error: std::net::AddrParseError| { + SwitchyardError::InvalidConfig(error.to_string()) + })?; + let listener = tokio::net::TcpListener::bind(addr) + .await + .map_err(|error| SwitchyardError::Other(error.to_string()))?; + let bound_addr = listener + .local_addr() + .map_err(|error| SwitchyardError::Other(error.to_string()))?; + + println!("libsy-proxy listening on http://{bound_addr}"); + println!(" routing (libsy agent-aware): classifier={CLASSIFIER_MODEL}"); + println!(" frontier={FRONTIER_MODEL}"); + println!(" fast={FAST_MODEL}"); + println!( + " send model \"{PROFILE_MODEL_ID}\" to /v1/chat/completions, /v1/messages, or /v1/responses" + ); + + axum::serve(listener, build_switchyard_router(state)) + .with_graceful_shutdown(async { + let _ = tokio::signal::ctrl_c().await; + }) + .await + .map_err(|error| SwitchyardError::Other(error.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use switchyard_components_v2::RequestMetadata; + use switchyard_protocol::{prompt_text, text_request, Metadata}; + + #[test] + fn routed_call_preserves_provider_body_and_rewrites_model() -> std::result::Result<(), BoxErr> { + let request = Request { + llm_request: text_request(Some("libsy-agent-aware".to_string()), "inspect"), + raw_request: Some(json!({ + "model": "libsy-agent-aware", + "input": "inspect", + "tools": [{"type": "function", "name": "shell"}], + "stream": false, + })), + metadata: Some(Metadata { + extra_metadata: Some(BTreeMap::from([( + "inbound_format".to_string(), + "openai_responses".to_string(), + )])), + ..Metadata::default() + }), + }; + + let routed = + chat_request_for_call(&request, "provider/model", &TranslationEngine::default())?; + assert_eq!(routed.request_type(), ChatRequestType::OpenAiResponses); + assert_eq!( + routed.body().get("model").and_then(Value::as_str), + Some("provider/model") + ); + assert!(routed.body().get("tools").is_some()); + Ok(()) + } + + #[test] + fn decodes_responses_input_and_normalizes_headers() -> Result<()> { + let input = ProfileInput { + request: ChatRequest::openai_responses(json!({ + "model": "libsy-agent-aware", + "input": [{ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "current subtask"}] + }], + "stream": false, + })), + metadata: RequestMetadata { + headers: BTreeMap::from([( + "thread-id".to_string(), + vec!["child-agent".to_string()], + )]), + ..RequestMetadata::default() + }, + }; + + let request = libsy_request(&input, &TranslationEngine::default())?; + assert_eq!(prompt_text(&request.llm_request), "current subtask"); + assert_eq!( + request + .metadata + .as_ref() + .and_then(|metadata| metadata.agent_id.as_deref()), + Some("child-agent") + ); + Ok(()) + } + + #[test] + fn classifier_call_uses_a_buffered_synthetic_request() -> std::result::Result<(), BoxErr> { + let request = Request { + llm_request: text_request(Some("auto".to_string()), "classify this"), + raw_request: None, + metadata: None, + }; + + let classifier = + chat_request_for_call(&request, "classifier/model", &TranslationEngine::default())?; + assert_eq!(classifier.request_type(), ChatRequestType::OpenAiChat); + assert_ne!( + classifier.body().get("stream").and_then(Value::as_bool), + Some(true) + ); + Ok(()) + } + + #[test] + fn response_metadata_exposes_provider_model_and_logical_tier() { + let decision: Arc = Arc::new(AgentRoutingDecision { + selected_model: FAST_TARGET.to_string(), + reason: "bounded lookup".to_string(), + task_kind: Some("research".to_string()), + confidence: Some(0.9), + agent_id: Some("child-1".to_string()), + cache_hit: false, + }); + + let metadata = routing_metadata(&[decision]); + assert_eq!(metadata.selected_model.as_deref(), Some(FAST_MODEL)); + assert_eq!(metadata.selected_tier.as_deref(), Some(FAST_TARGET)); + assert_eq!(metadata.confidence, Some(0.9)); + } +} From 519797ecfb0ba368195d55e73f25eb4ebe3db959 Mon Sep 17 00:00:00 2001 From: ayushag Date: Thu, 16 Jul 2026 15:21:39 -0700 Subject: [PATCH 2/7] feat(libsy): add pluggable routing affinity --- crates/libsy-protocol/src/envelope.rs | 2 + crates/libsy/README.md | 44 +- crates/libsy/src/affinity.rs | 430 +++++++++++++++ crates/libsy/src/agentic.rs | 728 -------------------------- crates/libsy/src/algorithms/rand.rs | 236 ++++++++- crates/libsy/src/lib.rs | 10 +- crates/libsy/tests/affinity.rs | 45 ++ demo/libsy-proxy/README.md | 18 +- demo/libsy-proxy/src/main.rs | 116 ++-- 9 files changed, 786 insertions(+), 843 deletions(-) create mode 100644 crates/libsy/src/affinity.rs delete mode 100644 crates/libsy/src/agentic.rs create mode 100644 crates/libsy/tests/affinity.rs diff --git a/crates/libsy-protocol/src/envelope.rs b/crates/libsy-protocol/src/envelope.rs index eef02e89..36a39054 100644 --- a/crates/libsy-protocol/src/envelope.rs +++ b/crates/libsy-protocol/src/envelope.rs @@ -43,6 +43,8 @@ pub struct Metadata { /// Optional lineage and semantic signals for agent-aware routing. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct AgentContext { + /// Whether the harness explicitly identified this request as coming from a child agent. + pub is_subagent: bool, /// Id of the parent agent, when this request comes from a child agent. pub parent_agent_id: Option, /// Harness-defined kind of agent call, such as `collab_spawn` or `review`. diff --git a/crates/libsy/README.md b/crates/libsy/README.md index 463775bf..6d9a52b6 100644 --- a/crates/libsy/README.md +++ b/crates/libsy/README.md @@ -40,35 +40,35 @@ println!("answer: {}", completion_text(&response.llm_response.aggregate().await? Runnable: [`research_agent`](../libsy-examples/examples/research_agent.rs) (in the `libsy-examples` crate). -## Agent- and subtask-aware routing +## Composable affinity -`agentic::AgentAwareOrchAlgo` classifies the first request for a stable agent/task -against an arbitrary model pool, then reuses that assignment. This keeps a child -agent and its prompt-cache locality on one model while allowing sibling agents in -the same root session to use different models. Invalid or failed classifier calls -fall back without caching the failure. +Routing algorithms can consume any `affinity::Affinity` policy at the point where +their final model is known. `SessionAffinity` retains one model per session; +`SubAgentAffinity` retains one model per explicitly identified child agent while +ordinary requests continue through the algorithm on every turn. ```rust -use libsy::agentic::{AgentAwareOrchAlgo, AgentRoutingCandidate}; - -let algo: Arc = Arc::new(AgentAwareOrchAlgo::new( - "classifier", - vec![ - AgentRoutingCandidate::new("frontier", "planning, synthesis, and review"), - AgentRoutingCandidate::new("fast", "bounded research and mechanical edits"), - ], - "frontier", // fail-open target - targets, -)); +use libsy::affinity::SubAgentAffinity; +use libsy::RandomAlgo; + +let algo: Arc = Arc::new( + RandomAlgo::new(targets) + .with_affinity(Arc::new(SubAgentAffinity::new())), +); ``` -`metadata_from_headers` converts harness-specific identity into neutral `Metadata`. +The first request from a child agent runs the random algorithm and atomically retains +its choice; later requests for the same `(session_id, agent_id)` reuse that target. +With one configured target, that target is retained directly. `metadata_from_headers` +converts harness-specific identity into neutral `Metadata`. Supported inputs include Codex `session-id`, `thread-id`, `x-codex-parent-thread-id`, `x-openai-subagent`, and `x-codex-turn-metadata`; NeMo Relay `x-nemo-relay-session-id` / `x-nemo-relay-subagent-id`; Dynamo `x-dynamo-session-id` / `x-dynamo-parent-session-id`; and explicit -`x-switchyard-*` overrides. The adapter stays local and dependency-free because -Relay's matching gateway normalizer is not exposed through its public Rust library API. +`x-switchyard-*` overrides, including `x-switchyard-is-subagent`. Header parsing is +separate from affinity keying, so non-HTTP embedders can populate the same metadata +directly. The adapter stays local and dependency-free because Relay's matching gateway +normalizer is not exposed through its public Rust library API. See [`demo/libsy-proxy`](../../demo/libsy-proxy) for a runnable Switchyard HTTP proxy. @@ -289,8 +289,8 @@ with `cargo test -p libsy-examples`). **Reference algorithms** — implementations to read and route with: -- [`RandomAlgo`](src/algorithms/rand.rs) — uniform random over the set - (one call). +- [`RandomAlgo`](src/algorithms/rand.rs) — uniform random over the set, optionally + retaining selections through an `Affinity` policy (one call). - [`LlmClassifierOrchAlgo`](../libsy-examples/src/llm_class.rs) — classify, then route strong/weak; fail open to strong. - [`EnsembleOrchAlgo`](../libsy-examples/src/ensemble.rs) — stateful: fan out to diff --git a/crates/libsy/src/affinity.rs b/crates/libsy/src/affinity.rs new file mode 100644 index 00000000..7cc6f92f --- /dev/null +++ b/crates/libsy/src/affinity.rs @@ -0,0 +1,430 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Reusable model-affinity policies and harness metadata normalization. +//! +//! Algorithms remain responsible for choosing a model. An [`Affinity`] policy only +//! decides whether that choice should be retained and which stable request identity +//! owns the assignment. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Mutex; + +use serde::Deserialize; +use switchyard_protocol::{AgentContext, Metadata, Request}; + +const CODEX_TURN_METADATA_HEADER: &str = "x-codex-turn-metadata"; +const MAX_ASSIGNMENTS: usize = 4096; + +/// Namespaced stable identity used to retain a model assignment. +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +pub struct AffinityKey { + namespace: String, + components: Vec, +} + +impl AffinityKey { + /// Creates a key whose namespace prevents collisions with other affinity policies. + pub fn new(namespace: impl Into, components: Vec) -> Self { + Self { + namespace: namespace.into(), + components, + } + } +} + +/// Bounded, process-local storage shared by an affinity policy. +#[derive(Default)] +pub struct AffinityState { + assignments: Mutex>, +} + +impl AffinityState { + /// Returns the retained model for `key`, failing open if the state lock is poisoned. + fn get(&self, key: &AffinityKey) -> Option { + self.assignments + .lock() + .ok() + .and_then(|assignments| assignments.get(key).cloned()) + } + + /// Atomically retains the first model assigned to `key` and returns that model. + fn get_or_insert(&self, key: AffinityKey, proposed: String) -> String { + let Ok(mut assignments) = self.assignments.lock() else { + return proposed; + }; + if let Some(assigned) = assignments.get(&key) { + return assigned.clone(); + } + if assignments.len() >= MAX_ASSIGNMENTS { + if let Some(evicted) = assignments.keys().next().cloned() { + assignments.remove(&evicted); + } + } + assignments.insert(key, proposed.clone()); + proposed + } +} + +/// Policy that maps eligible requests to stable assignment keys. +/// +/// Algorithms consume this trait at the point where their final model is known: +/// check [`assignment`](Self::assignment), run normal selection on a miss, then call +/// [`retain`](Self::retain) before routing the request. +pub trait Affinity: Send + Sync { + /// Returns the stable key for an eligible request, or `None` when affinity does + /// not apply. + fn key(&self, request: &Request) -> Option; + + /// Returns this policy's process-local assignment state. + fn state(&self) -> &AffinityState; + + /// Returns an existing model assignment for `request`. + fn assignment(&self, request: &Request) -> Option { + self.key(request).and_then(|key| self.state().get(&key)) + } + + /// Retains `proposed` for an eligible request and returns the canonical model. + /// + /// When concurrent requests propose different models, the first insertion wins. + /// Ineligible requests simply receive `proposed` without storing it. + fn retain(&self, request: &Request, proposed: String) -> String { + let Some(key) = self.key(request) else { + return proposed; + }; + self.state().get_or_insert(key, proposed) + } +} + +/// Retains one model for every request carrying the same session id. +#[derive(Default)] +pub struct SessionAffinity { + state: AffinityState, +} + +impl SessionAffinity { + /// Creates an empty process-local session-affinity policy. + pub fn new() -> Self { + Self::default() + } +} + +impl Affinity for SessionAffinity { + fn key(&self, request: &Request) -> Option { + Some(AffinityKey::new( + "session", + vec![request.metadata.as_ref()?.session_id.clone()?], + )) + } + + fn state(&self) -> &AffinityState { + &self.state + } +} + +/// Retains one model for each explicitly identified child agent in a session. +#[derive(Default)] +pub struct SubAgentAffinity { + state: AffinityState, +} + +impl SubAgentAffinity { + /// Creates an empty process-local child-agent-affinity policy. + pub fn new() -> Self { + Self::default() + } +} + +impl Affinity for SubAgentAffinity { + fn key(&self, request: &Request) -> Option { + let metadata = request.metadata.as_ref()?; + if !metadata.agent_context.as_deref()?.is_subagent { + return None; + } + Some(AffinityKey::new( + "subagent", + vec![metadata.session_id.clone()?, metadata.agent_id.clone()?], + )) + } + + fn state(&self) -> &AffinityState { + &self.state + } +} + +#[derive(Default, Deserialize)] +struct CodexTurnMetadata { + session_id: Option, + thread_id: Option, + parent_thread_id: Option, + turn_id: Option, + subagent_kind: Option, + agent_role: Option, + task_id: Option, + task_kind: Option, +} + +/// Normalizes harness-specific request headers into libsy metadata. +/// +/// Explicit `x-switchyard-*` headers win. NeMo Relay and Dynamo correlation +/// headers are accepted without linking either runtime. Codex's structured turn +/// metadata is preferred over its compatibility projections. +pub fn metadata_from_headers(headers: &BTreeMap>) -> Metadata { + let codex = header(headers, CODEX_TURN_METADATA_HEADER) + .and_then(|value| serde_json::from_str::(&value).ok()) + .unwrap_or_default(); + + let switchyard_parent = header(headers, "x-switchyard-parent-agent-id"); + let relay_subagent = header(headers, "x-nemo-relay-subagent-id"); + let dynamo_parent = header(headers, "x-dynamo-parent-session-id"); + let codex_parent = header(headers, "x-codex-parent-thread-id"); + let openai_subagent = header(headers, "x-openai-subagent"); + let inferred_subagent = switchyard_parent.is_some() + || relay_subagent.is_some() + || dynamo_parent.is_some() + || codex.parent_thread_id.is_some() + || codex.subagent_kind.is_some() + || codex_parent.is_some() + || openai_subagent.is_some(); + let is_subagent = header(headers, "x-switchyard-is-subagent") + .as_deref() + .and_then(parse_bool) + .unwrap_or(inferred_subagent); + + let agent_context = AgentContext { + is_subagent, + parent_agent_id: first_some([ + switchyard_parent, + dynamo_parent, + codex.parent_thread_id, + codex_parent, + ]), + agent_kind: first_some([ + header(headers, "x-switchyard-agent-kind"), + codex.subagent_kind, + openai_subagent, + ]), + agent_role: first_some([header(headers, "x-switchyard-agent-role"), codex.agent_role]), + task_kind: first_some([header(headers, "x-switchyard-task-kind"), codex.task_kind]), + turn_id: first_some([header(headers, "x-switchyard-turn-id"), codex.turn_id]), + }; + let agent_context = (agent_context != AgentContext::default()).then(|| Box::new(agent_context)); + + Metadata { + session_id: first_some([ + header(headers, "x-switchyard-session-id"), + header(headers, "x-nemo-relay-session-id"), + codex.session_id, + header(headers, "session-id"), + ]), + agent_id: first_some([ + header(headers, "x-switchyard-agent-id"), + relay_subagent, + header(headers, "x-dynamo-session-id"), + codex.thread_id, + header(headers, "thread-id"), + ]), + task_id: first_some([ + header(headers, "x-switchyard-task-id"), + codex.task_id, + header(headers, "x-task-id"), + ]), + agent_context, + correlation_id: first_some([ + header(headers, "x-request-id"), + header(headers, "x-client-request-id"), + ]), + ..Metadata::default() + } +} + +fn header(headers: &BTreeMap>, name: &str) -> Option { + headers + .iter() + .find(|(key, _)| key.eq_ignore_ascii_case(name)) + .and_then(|(_, values)| values.iter().find(|value| !value.trim().is_empty())) + .map(|value| value.trim().to_string()) +} + +fn parse_bool(value: &str) -> Option { + match value.trim().to_ascii_lowercase().as_str() { + "1" | "true" | "yes" | "on" => Some(true), + "0" | "false" | "no" | "off" => Some(false), + _ => None, + } +} + +fn first_some(values: [Option; N]) -> Option { + values.into_iter().flatten().next() +} + +#[cfg(test)] +mod tests { + use super::*; + use switchyard_protocol::{text_request, AgentContext}; + + fn request(metadata: Metadata) -> Request { + Request { + llm_request: text_request(Some("auto".to_string()), "hi"), + raw_request: None, + metadata: Some(metadata), + } + } + + fn child_request(task_id: &str) -> Request { + request(Metadata { + session_id: Some("session-1".to_string()), + agent_id: Some("child-1".to_string()), + task_id: Some(task_id.to_string()), + agent_context: Some(Box::new(AgentContext { + is_subagent: true, + ..AgentContext::default() + })), + ..Metadata::default() + }) + } + + #[test] + fn session_affinity_keys_all_requests_by_session() { + let affinity = SessionAffinity::new(); + let first = request(Metadata { + session_id: Some("session-1".to_string()), + agent_id: Some("agent-a".to_string()), + ..Metadata::default() + }); + let second = request(Metadata { + session_id: Some("session-1".to_string()), + agent_id: Some("agent-b".to_string()), + ..Metadata::default() + }); + + assert_eq!(affinity.retain(&first, "model-a".to_string()), "model-a"); + assert_eq!(affinity.assignment(&second).as_deref(), Some("model-a")); + } + + #[test] + fn subagent_affinity_ignores_task_changes() { + let affinity = SubAgentAffinity::new(); + assert_eq!( + affinity.retain(&child_request("task-1"), "model-a".to_string()), + "model-a" + ); + assert_eq!( + affinity.assignment(&child_request("task-2")).as_deref(), + Some("model-a") + ); + } + + #[test] + fn subagent_affinity_does_not_apply_to_root_agents() { + let affinity = SubAgentAffinity::new(); + let root = request(Metadata { + session_id: Some("session-1".to_string()), + agent_id: Some("root-1".to_string()), + agent_context: Some(Box::new(AgentContext::default())), + ..Metadata::default() + }); + + assert_eq!(affinity.retain(&root, "model-a".to_string()), "model-a"); + assert!(affinity.assignment(&root).is_none()); + } + + #[test] + fn normalizes_codex_child_metadata() { + let headers = BTreeMap::from([( + CODEX_TURN_METADATA_HEADER.to_string(), + vec![serde_json::json!({ + "session_id": "root-session", + "thread_id": "child-agent", + "parent_thread_id": "root-agent", + "turn_id": "turn-7", + "subagent_kind": "collab_spawn", + }) + .to_string()], + )]); + + let metadata = metadata_from_headers(&headers); + assert_eq!(metadata.session_id.as_deref(), Some("root-session")); + assert_eq!(metadata.agent_id.as_deref(), Some("child-agent")); + let agent = metadata.agent_context.as_deref(); + assert_eq!(agent.map(|value| value.is_subagent), Some(true)); + assert_eq!( + agent.and_then(|value| value.parent_agent_id.as_deref()), + Some("root-agent") + ); + } + + #[test] + fn root_codex_metadata_is_not_inferred_as_a_subagent() { + let headers = BTreeMap::from([( + CODEX_TURN_METADATA_HEADER.to_string(), + vec![serde_json::json!({ + "session_id": "root-session", + "thread_id": "root-agent", + "turn_id": "turn-1", + }) + .to_string()], + )]); + + let metadata = metadata_from_headers(&headers); + assert_eq!( + metadata + .agent_context + .as_deref() + .map(|agent| agent.is_subagent), + Some(false) + ); + } + + #[test] + fn explicit_switchyard_subagent_flag_overrides_inference() { + let headers = BTreeMap::from([ + ( + "x-switchyard-is-subagent".to_string(), + vec!["false".to_string()], + ), + ( + "x-switchyard-parent-agent-id".to_string(), + vec!["parent".to_string()], + ), + ]); + + let metadata = metadata_from_headers(&headers); + assert_eq!( + metadata + .agent_context + .as_deref() + .map(|agent| agent.is_subagent), + Some(false) + ); + } + + #[test] + fn normalizes_relay_and_dynamo_child_headers() { + let headers = BTreeMap::from([ + ( + "x-nemo-relay-session-id".to_string(), + vec!["relay-session".to_string()], + ), + ( + "x-nemo-relay-subagent-id".to_string(), + vec!["relay-child".to_string()], + ), + ( + "x-dynamo-parent-session-id".to_string(), + vec!["relay-parent".to_string()], + ), + ]); + + let metadata = metadata_from_headers(&headers); + assert_eq!(metadata.session_id.as_deref(), Some("relay-session")); + assert_eq!(metadata.agent_id.as_deref(), Some("relay-child")); + assert_eq!( + metadata + .agent_context + .as_deref() + .map(|agent| agent.is_subagent), + Some(true) + ); + } +} diff --git a/crates/libsy/src/agentic.rs b/crates/libsy/src/agentic.rs deleted file mode 100644 index 1125b64a..00000000 --- a/crates/libsy/src/agentic.rs +++ /dev/null @@ -1,728 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -//! Agent- and subtask-aware model routing. -//! -//! [`metadata_from_headers`] normalizes identity signals emitted by Codex, NeMo Relay, -//! Dynamo, or explicit Switchyard headers into neutral [`Metadata`]. -//! [`AgentAwareOrchAlgo`] classifies a stable agent/task once, then reuses that model -//! assignment to retain model and prompt-cache locality. - -use std::collections::{BTreeMap, HashMap}; -use std::error::Error; -use std::sync::{Arc, Mutex}; - -use async_trait::async_trait; -use serde::Deserialize; -use switchyard_protocol::{completion_text, prompt_text, text_request, AgentContext}; - -use crate::{ - Algorithm, Context, Decision, Driver, LlmTargetSet, Metadata, Request, Response, Signals, -}; - -const CODEX_TURN_METADATA_HEADER: &str = "x-codex-turn-metadata"; -const MAX_ASSIGNMENTS: usize = 4096; - -type BoxErr = Box; - -/// One model available to the agent-aware classifier. -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct AgentRoutingCandidate { - /// Semantic target name used to look up the model in [`LlmTargetSet`]. - pub target: String, - /// Capability/cost guidance shown to the classifier. - pub description: String, -} - -impl AgentRoutingCandidate { - /// Creates a model-pool entry from a target name and classifier guidance. - pub fn new(target: impl Into, description: impl Into) -> Self { - Self { - target: target.into(), - description: description.into(), - } - } -} - -/// Inspectable decision emitted for classifier and routed calls. -pub struct AgentRoutingDecision { - /// Target selected from the configured model pool. - pub selected_model: String, - /// Human-readable classifier, cache, or fallback explanation. - pub reason: String, - /// Classifier-inferred task class, when supplied. - pub task_kind: Option, - /// Classifier confidence in `[0, 1]`, when supplied. - pub confidence: Option, - /// Stable agent id used for the assignment, when available. - pub agent_id: Option, - /// Whether the selection came from an existing per-agent assignment. - pub cache_hit: bool, -} - -impl Decision for AgentRoutingDecision { - fn selected_model(&self) -> &str { - &self.selected_model - } - - fn reasoning(&self) -> Option<&str> { - Some(&self.reason) - } - - fn as_any(&self) -> &dyn std::any::Any { - self - } -} - -/// LLM classifier over a model pool with per-agent/task assignment affinity. -pub struct AgentAwareOrchAlgo { - classifier_model: String, - candidates: Vec, - fallback_model: String, - target_set: LlmTargetSet, - assignments: Mutex>, -} - -impl AgentAwareOrchAlgo { - /// Configures the classifier target, candidate pool, fail-open target, and targets. - pub fn new( - classifier_model: impl Into, - candidates: Vec, - fallback_model: impl Into, - target_set: LlmTargetSet, - ) -> Self { - Self { - classifier_model: classifier_model.into(), - candidates, - fallback_model: fallback_model.into(), - target_set, - assignments: Mutex::new(HashMap::new()), - } - } - - /// Classifies a request, failing open without caching invalid or failed results. - async fn classify( - &self, - ctx: &Context, - driver: &Driver, - request: &Request, - ) -> Result<(Assignment, bool), BoxErr> { - let classifier_decision: Arc = Arc::new(AgentRoutingDecision { - selected_model: self.classifier_model.clone(), - reason: format!("classifying agent/subtask via {}", self.classifier_model), - task_kind: request - .metadata - .as_ref() - .and_then(|metadata| metadata.agent_context.as_deref()) - .and_then(|agent| agent.task_kind.clone()), - confidence: None, - agent_id: request - .metadata - .as_ref() - .and_then(|metadata| metadata.agent_id.clone()), - cache_hit: false, - }); - driver - .info(ctx.clone(), Arc::clone(&classifier_decision)) - .await?; - - let classify_request = Request { - llm_request: text_request( - request.llm_request.model.clone(), - classifier_prompt( - &self.candidates, - request.metadata.as_ref(), - &prompt_text(&request.llm_request), - ), - ), - // Classifier calls receive neutral text and metadata, never the provider - // payload with its tools or continuation state. - raw_request: None, - metadata: request.metadata.clone(), - }; - - let output = match self.target_set.get_target(&self.classifier_model) { - Ok(target) => { - driver - .call_llm_target(ctx.clone(), &target, classify_request, classifier_decision) - .await - } - Err(error) => Err(error), - }; - - match output { - Ok(response) => match response.llm_response.into_agg().await { - Ok(response) => { - match parse_classifier_output(&completion_text(&response), &self.candidates) { - Some(assignment) => Ok((assignment, true)), - None => Ok(( - self.fallback_assignment( - "classifier returned an invalid model-pool decision", - ), - false, - )), - } - } - Err(error) => Ok(( - self.fallback_assignment(&format!("classifier response failed: {error}")), - false, - )), - }, - Err(error) => Ok(( - self.fallback_assignment(&format!("classifier call failed: {error}")), - false, - )), - } - } - - fn fallback_assignment(&self, reason: &str) -> Assignment { - Assignment { - model: self.fallback_model.clone(), - reason: format!("{reason}; fell back to {}", self.fallback_model), - task_kind: None, - confidence: None, - } - } - - fn cached_assignment(&self, key: &AssignmentKey) -> Option { - self.assignments - .lock() - .ok() - .and_then(|assignments| assignments.get(key).cloned()) - } - - fn store_assignment(&self, key: AssignmentKey, assignment: Assignment) { - if let Ok(mut assignments) = self.assignments.lock() { - if !assignments.contains_key(&key) && assignments.len() >= MAX_ASSIGNMENTS { - if let Some(evicted) = assignments.keys().next().cloned() { - assignments.remove(&evicted); - } - } - assignments.insert(key, assignment); - } - } -} - -#[async_trait] -impl Algorithm for AgentAwareOrchAlgo { - async fn create_run_task( - self: Arc, - ctx: Context, - driver: Driver, - request: Request, - ) -> Result { - let key = assignment_key(request.metadata.as_ref()); - let cached = key.as_ref().and_then(|key| self.cached_assignment(key)); - let (assignment, cache_hit) = match cached { - Some(assignment) => (assignment, true), - None => { - let (assignment, cacheable) = self.classify(&ctx, &driver, &request).await?; - if cacheable { - if let Some(key) = key { - self.store_assignment(key, assignment.clone()); - } - } - (assignment, false) - } - }; - - let routed_target = self.target_set.get_target(&assignment.model)?; - let route_decision: Arc = Arc::new(AgentRoutingDecision { - selected_model: assignment.model.clone(), - reason: if cache_hit { - format!("reused stable agent/task assignment: {}", assignment.reason) - } else { - assignment.reason.clone() - }, - task_kind: assignment.task_kind, - confidence: assignment.confidence, - agent_id: request - .metadata - .as_ref() - .and_then(|metadata| metadata.agent_id.clone()), - cache_hit, - }); - driver - .info(ctx.clone(), Arc::clone(&route_decision)) - .await?; - driver - .call_llm_target(ctx, &routed_target, request, route_decision) - .await - } - - async fn process_signals(self: Arc, _signals: Signals) -> Result<(), BoxErr> { - Ok(()) - } -} - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -struct AssignmentKey { - session_id: Option, - agent_id: String, - task_id: Option, -} - -#[derive(Clone, Debug)] -struct Assignment { - model: String, - reason: String, - task_kind: Option, - confidence: Option, -} - -#[derive(Deserialize)] -struct ClassifierOutput { - model: String, - #[serde(default)] - task_kind: Option, - #[serde(default)] - confidence: Option, - #[serde(default)] - reason: Option, -} - -#[derive(Default, Deserialize)] -struct CodexTurnMetadata { - session_id: Option, - thread_id: Option, - parent_thread_id: Option, - turn_id: Option, - subagent_kind: Option, - agent_role: Option, - task_id: Option, - task_kind: Option, -} - -/// Normalizes harness-specific request headers into libsy metadata. -/// -/// Explicit `x-switchyard-*` headers win. NeMo Relay and Dynamo correlation -/// headers are accepted without linking either runtime. Codex's structured turn -/// metadata is preferred over its compatibility projections. -pub fn metadata_from_headers(headers: &BTreeMap>) -> Metadata { - let codex = header(headers, CODEX_TURN_METADATA_HEADER) - .and_then(|value| serde_json::from_str::(&value).ok()) - .unwrap_or_default(); - - let agent_context = AgentContext { - parent_agent_id: first_some([ - header(headers, "x-switchyard-parent-agent-id"), - header(headers, "x-dynamo-parent-session-id"), - codex.parent_thread_id, - header(headers, "x-codex-parent-thread-id"), - ]), - agent_kind: first_some([ - header(headers, "x-switchyard-agent-kind"), - codex.subagent_kind, - header(headers, "x-openai-subagent"), - ]), - agent_role: first_some([header(headers, "x-switchyard-agent-role"), codex.agent_role]), - task_kind: first_some([header(headers, "x-switchyard-task-kind"), codex.task_kind]), - turn_id: first_some([header(headers, "x-switchyard-turn-id"), codex.turn_id]), - }; - let agent_context = (agent_context != AgentContext::default()).then(|| Box::new(agent_context)); - - Metadata { - session_id: first_some([ - header(headers, "x-switchyard-session-id"), - header(headers, "x-nemo-relay-session-id"), - codex.session_id, - header(headers, "session-id"), - ]), - agent_id: first_some([ - header(headers, "x-switchyard-agent-id"), - header(headers, "x-nemo-relay-subagent-id"), - header(headers, "x-dynamo-session-id"), - codex.thread_id, - header(headers, "thread-id"), - ]), - task_id: first_some([ - header(headers, "x-switchyard-task-id"), - codex.task_id, - header(headers, "x-task-id"), - ]), - agent_context, - correlation_id: first_some([ - header(headers, "x-request-id"), - header(headers, "x-client-request-id"), - ]), - ..Metadata::default() - } -} - -fn assignment_key(metadata: Option<&Metadata>) -> Option { - let metadata = metadata?; - Some(AssignmentKey { - session_id: metadata.session_id.clone(), - agent_id: metadata.agent_id.clone()?, - task_id: metadata.task_id.clone(), - }) -} - -fn classifier_prompt( - candidates: &[AgentRoutingCandidate], - metadata: Option<&Metadata>, - prompt: &str, -) -> String { - let metadata = metadata.cloned().unwrap_or_default(); - let agent = metadata - .agent_context - .as_deref() - .cloned() - .unwrap_or_default(); - let candidate_lines = candidates - .iter() - .map(|candidate| format!("- {}: {}", candidate.target, candidate.description)) - .collect::>() - .join("\n"); - format!( - "You route one model call inside an agent hierarchy. Select exactly one target from the \n\ - model pool. Prefer the cheapest candidate likely to complete the subtask correctly. \n\ - Return JSON only: {{\"model\":\"\",\"task_kind\":\"\",\ - \"confidence\":<0..1>,\"reason\":\"\"}}.\n\n\ - Agent context:\n\ - session_id={:?}\nagent_id={:?}\nparent_agent_id={:?}\nagent_kind={:?}\n\ - agent_role={:?}\ntask_id={:?}\ntask_kind={:?}\nturn_id={:?}\n\n\ - Model pool:\n{}\n\nSubtask:\n\n{}\n", - metadata.session_id, - metadata.agent_id, - agent.parent_agent_id, - agent.agent_kind, - agent.agent_role, - metadata.task_id, - agent.task_kind, - agent.turn_id, - candidate_lines, - prompt, - ) -} - -fn parse_classifier_output( - output: &str, - candidates: &[AgentRoutingCandidate], -) -> Option { - let trimmed = output.trim(); - if let Some(candidate) = candidates - .iter() - .find(|candidate| candidate.target == trimmed) - { - return Some(Assignment { - model: candidate.target.clone(), - reason: "classifier selected the target by name".to_string(), - task_kind: None, - confidence: None, - }); - } - - let start = trimmed.find('{')?; - let end = trimmed.rfind('}')?; - let parsed = serde_json::from_str::(&trimmed[start..=end]).ok()?; - let model = parsed.model.trim(); - if !candidates.iter().any(|candidate| candidate.target == model) { - return None; - } - Some(Assignment { - model: model.to_string(), - reason: parsed - .reason - .filter(|reason| !reason.trim().is_empty()) - .unwrap_or_else(|| "classifier selected the target".to_string()), - task_kind: parsed.task_kind.filter(|kind| !kind.trim().is_empty()), - confidence: parsed - .confidence - .filter(|value| value.is_finite()) - .map(|value| value.clamp(0.0, 1.0)), - }) -} - -fn header(headers: &BTreeMap>, name: &str) -> Option { - headers - .iter() - .find(|(key, _)| key.eq_ignore_ascii_case(name)) - .and_then(|(_, values)| values.iter().find(|value| !value.trim().is_empty())) - .map(|value| value.trim().to_string()) -} - -fn first_some(values: [Option; N]) -> Option { - values.into_iter().flatten().next() -} - -#[cfg(test)] -mod tests { - use std::collections::VecDeque; - - use super::*; - use crate::{LlmClient, LlmResponse, LlmTarget, RoutedRequest}; - use switchyard_protocol::{text_response, Metadata}; - - struct RoutingClient { - classifier_calls: Arc>, - responses: Arc>>, - } - - #[async_trait] - impl LlmClient for RoutingClient { - async fn call(&self, routed: RoutedRequest) -> Result { - let selected = routed.decision.selected_model().to_string(); - let completion = if selected == "classifier" { - let mut calls = self.classifier_calls.lock().map_err(|_| "lock poisoned")?; - *calls += 1; - self.responses - .lock() - .map_err(|_| "lock poisoned")? - .pop_front() - .ok_or("missing classifier response")? - } else { - selected - }; - Ok(Response { - llm_response: LlmResponse::Agg(text_response(None, completion)), - metadata: routed.request.metadata, - }) - } - } - - fn algo(responses: Vec<&str>) -> (Arc, Arc>) { - let classifier_calls = Arc::new(Mutex::new(0)); - let client = Arc::new(RoutingClient { - classifier_calls: Arc::clone(&classifier_calls), - responses: Arc::new(Mutex::new( - responses.into_iter().map(str::to_string).collect(), - )), - }) as Arc; - let target = |name: &str| LlmTarget { - semantic_name: name.to_string(), - llm_client: Some(Arc::clone(&client)), - }; - let targets = LlmTargetSet::new(vec![ - target("classifier"), - target("frontier"), - target("fast"), - ]); - let algo: Arc = Arc::new(AgentAwareOrchAlgo::new( - "classifier", - vec![ - AgentRoutingCandidate::new("frontier", "complex planning and final review"), - AgentRoutingCandidate::new("fast", "bounded research and mechanical edits"), - ], - "frontier", - targets, - )); - (algo, classifier_calls) - } - - fn request(agent_id: Option<&str>, task_id: Option<&str>, prompt: &str) -> Request { - Request { - llm_request: text_request(Some("auto".to_string()), prompt), - raw_request: Some(serde_json::json!({"input": prompt})), - metadata: Some(Metadata { - session_id: Some("session-1".to_string()), - agent_id: agent_id.map(str::to_string), - task_id: task_id.map(str::to_string), - agent_context: Some(Box::new(AgentContext { - agent_role: Some("explorer".to_string()), - ..AgentContext::default() - })), - ..Metadata::default() - }), - } - } - - fn decision(model: &str, task_kind: &str, confidence: f64, reason: &str) -> String { - serde_json::json!({ - "model": model, - "task_kind": task_kind, - "confidence": confidence, - "reason": reason, - }) - .to_string() - } - - fn response_text(response: &Response) -> String { - response - .llm_response - .as_agg() - .map(completion_text) - .unwrap_or_default() - } - - #[test] - fn normalizes_codex_turn_metadata_and_compatibility_headers() { - let mut headers = BTreeMap::new(); - headers.insert("session-id".to_string(), vec!["compat-session".to_string()]); - headers.insert("thread-id".to_string(), vec!["compat-agent".to_string()]); - headers.insert( - CODEX_TURN_METADATA_HEADER.to_string(), - vec![serde_json::json!({ - "session_id": "root-session", - "thread_id": "child-agent", - "parent_thread_id": "root-agent", - "turn_id": "turn-7", - "subagent_kind": "collab_spawn", - }) - .to_string()], - ); - - let metadata = metadata_from_headers(&headers); - assert_eq!(metadata.session_id.as_deref(), Some("root-session")); - assert_eq!(metadata.agent_id.as_deref(), Some("child-agent")); - let agent = metadata.agent_context.as_deref(); - assert_eq!( - agent.and_then(|value| value.parent_agent_id.as_deref()), - Some("root-agent") - ); - assert_eq!( - agent.and_then(|value| value.agent_kind.as_deref()), - Some("collab_spawn") - ); - assert_eq!( - agent.and_then(|value| value.turn_id.as_deref()), - Some("turn-7") - ); - } - - #[test] - fn explicit_switchyard_headers_override_relay_and_codex() { - let headers = BTreeMap::from([ - ( - "x-switchyard-session-id".to_string(), - vec!["canonical-session".to_string()], - ), - ( - "x-switchyard-agent-id".to_string(), - vec!["canonical-agent".to_string()], - ), - ( - "x-nemo-relay-session-id".to_string(), - vec!["relay-session".to_string()], - ), - ( - "x-dynamo-session-id".to_string(), - vec!["relay-agent".to_string()], - ), - ]); - - let metadata = metadata_from_headers(&headers); - assert_eq!(metadata.session_id.as_deref(), Some("canonical-session")); - assert_eq!(metadata.agent_id.as_deref(), Some("canonical-agent")); - } - - #[test] - fn normalizes_relay_and_dynamo_headers_without_runtime_dependency() { - let headers = BTreeMap::from([ - ( - "x-nemo-relay-session-id".to_string(), - vec!["relay-session".to_string()], - ), - ( - "x-nemo-relay-subagent-id".to_string(), - vec!["relay-child".to_string()], - ), - ( - "x-dynamo-parent-session-id".to_string(), - vec!["relay-parent".to_string()], - ), - ]); - - let metadata = metadata_from_headers(&headers); - assert_eq!(metadata.session_id.as_deref(), Some("relay-session")); - assert_eq!(metadata.agent_id.as_deref(), Some("relay-child")); - assert_eq!( - metadata - .agent_context - .as_deref() - .and_then(|agent| agent.parent_agent_id.as_deref()), - Some("relay-parent") - ); - } - - #[tokio::test] - async fn valid_classifier_output_routes_from_the_pool() -> Result<(), BoxErr> { - let (algo, calls) = algo(vec![&decision("fast", "research", 0.9, "bounded lookup")]); - let (trace, response) = algo - .run( - Context::default(), - request(Some("agent-a"), None, "survey the API"), - ) - .await?; - - assert_eq!(response_text(&response), "fast"); - assert_eq!(trace.len(), 2); - let routed = trace[1] - .as_any() - .downcast_ref::() - .ok_or("expected AgentRoutingDecision")?; - assert_eq!(routed.task_kind.as_deref(), Some("research")); - assert_eq!(routed.confidence, Some(0.9)); - assert!(!routed.cache_hit); - assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 1); - Ok(()) - } - - #[tokio::test] - async fn stable_agent_assignment_avoids_reclassification() -> Result<(), BoxErr> { - let (algo, calls) = algo(vec![&decision("fast", "research", 0.8, "explorer")]); - Arc::clone(&algo) - .run( - Context::default(), - request(Some("agent-a"), None, "first turn"), - ) - .await?; - let (trace, response) = algo - .run( - Context::default(), - request(Some("agent-a"), None, "second turn"), - ) - .await?; - - assert_eq!(response_text(&response), "fast"); - assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 1); - assert_eq!(trace.len(), 1); - let routed = trace[0] - .as_any() - .downcast_ref::() - .ok_or("expected AgentRoutingDecision")?; - assert!(routed.cache_hit); - Ok(()) - } - - #[tokio::test] - async fn explicit_task_change_reclassifies_the_same_agent() -> Result<(), BoxErr> { - let first = decision("fast", "research", 0.8, "lookup"); - let second = decision("frontier", "review", 0.9, "final review"); - let (algo, calls) = algo(vec![&first, &second]); - let (_, first_response) = Arc::clone(&algo) - .run( - Context::default(), - request(Some("agent-a"), Some("task-1"), "collect evidence"), - ) - .await?; - let (_, second_response) = algo - .run( - Context::default(), - request(Some("agent-a"), Some("task-2"), "adjudicate findings"), - ) - .await?; - - assert_eq!(response_text(&first_response), "fast"); - assert_eq!(response_text(&second_response), "frontier"); - assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 2); - Ok(()) - } - - #[tokio::test] - async fn invalid_decision_falls_back_without_poisoning_affinity() -> Result<(), BoxErr> { - let valid = decision("fast", "research", 0.7, "retry succeeded"); - let (algo, calls) = algo(vec!["not-json", &valid]); - let (_, first) = Arc::clone(&algo) - .run(Context::default(), request(Some("agent-a"), None, "first")) - .await?; - let (_, second) = algo - .run(Context::default(), request(Some("agent-a"), None, "second")) - .await?; - - assert_eq!(response_text(&first), "frontier"); - assert_eq!(response_text(&second), "fast"); - assert_eq!(*calls.lock().map_err(|_| "lock poisoned")?, 2); - Ok(()) - } -} diff --git a/crates/libsy/src/algorithms/rand.rs b/crates/libsy/src/algorithms/rand.rs index adc2b366..f75c51b2 100644 --- a/crates/libsy/src/algorithms/rand.rs +++ b/crates/libsy/src/algorithms/rand.rs @@ -14,6 +14,7 @@ use std::sync::Arc; use async_trait::async_trait; use rand::seq::SliceRandom; +use crate::affinity::Affinity; use crate::{Algorithm, Context, Decision, Driver, LlmTargetSet, Request, Response}; /// Decision produced by [`RandomAlgo`]: which target was chosen and why. @@ -41,6 +42,7 @@ impl Decision for RandomDecision { /// Uniform random router over a target set. pub struct RandomAlgo { target_set: LlmTargetSet, + affinity: Option>, } impl RandomAlgo { @@ -49,7 +51,35 @@ impl RandomAlgo { /// Wrap it in an [`Arc`] and drive it with [`Algorithm::run`] or /// [`Algorithm::run_stream`]. pub fn new(target_set: LlmTargetSet) -> Self { - Self { target_set } + Self { + target_set, + affinity: None, + } + } + + /// Retains selections according to `affinity` while preserving per-request + /// random routing for requests the policy does not key. + pub fn with_affinity(mut self, affinity: Arc) -> Self { + self.affinity = Some(affinity); + self + } + + /// Selects one configured target name uniformly at random. + fn choose_target(&self) -> Result> { + if let [target] = self.target_set.targets() { + return Ok(target.semantic_name.clone()); + } + // Scope the non-Send ThreadRng so callers may await after this returns. + let selected = { + let mut rng = rand::thread_rng(); + self.target_set + .targets() + .choose(&mut rng) + .ok_or("no targets available")? + .semantic_name + .clone() + }; + Ok(selected) } } @@ -61,19 +91,35 @@ impl Algorithm for RandomAlgo { driver: Driver, request: Request, ) -> Result> { - // Scope the non-Send ThreadRng before the await so the future remains Send. - let target = { - let mut rng = rand::thread_rng(); - self.target_set - .targets() - .choose(&mut rng) - .ok_or("no targets available")? - .clone() + let (selected, reasoning) = match self.affinity.as_ref() { + Some(affinity) => match affinity.assignment(&request) { + Some(selected) => { + let reasoning = format!("affinity reused target '{selected}'"); + (selected, reasoning) + } + None => { + let proposed = self.choose_target()?; + let applies = affinity.key(&request).is_some(); + let selected = affinity.retain(&request, proposed.clone()); + let reasoning = if !applies { + format!("random routing selected target '{selected}'") + } else if selected == proposed { + format!("random routing selected and retained target '{selected}'") + } else { + format!("affinity retained concurrent target '{selected}'") + }; + (selected, reasoning) + } + }, + None => { + let selected = self.choose_target()?; + let reasoning = format!("random routing selected target '{selected}'"); + (selected, reasoning) + } }; - - let selected = target.semantic_name.clone(); + let target = self.target_set.get_target(&selected)?; let decision: Arc = Arc::new(RandomDecision { - reasoning: format!("random routing selected target '{selected}'"), + reasoning, selected_model: selected, }); @@ -91,7 +137,8 @@ mod tests { use switchyard_protocol::{completion_text, text_request, text_response}; - use crate::{LlmResponse, LlmTarget, Request, RoutedLlmClient, Signals}; + use crate::affinity::{Affinity, SessionAffinity, SubAgentAffinity}; + use crate::{AgentContext, LlmResponse, LlmTarget, Metadata, Request, RoutedLlmClient, Signals}; /// Echoes the selected target so tests can inspect which target was called. struct EchoClient; @@ -135,6 +182,30 @@ mod tests { Arc::new(algorithm(names)) } + fn shared_algorithm_with_affinity( + names: &[&str], + affinity: Arc, + ) -> Arc { + Arc::new(algorithm(names).with_affinity(affinity)) + } + + fn affinity_request(is_subagent: bool, agent_id: &str, task_id: &str) -> Request { + Request { + llm_request: text_request(Some("auto".to_string()), "hi"), + raw_request: None, + metadata: Some(Metadata { + session_id: Some("session-1".to_string()), + agent_id: Some(agent_id.to_string()), + task_id: Some(task_id.to_string()), + agent_context: Some(Box::new(AgentContext { + is_subagent, + ..AgentContext::default() + })), + ..Metadata::default() + }), + } + } + #[tokio::test] async fn single_target_is_always_selected_and_called( ) -> Result<(), Box> { @@ -234,4 +305,143 @@ mod tests { assert_eq!(concrete.selected_model, "only/model"); Ok(()) } + + #[tokio::test] + async fn subagent_affinity_reuses_selection_across_task_changes( + ) -> Result<(), Box> { + let algorithm = shared_algorithm_with_affinity( + &["a/model", "b/model"], + Arc::new(SubAgentAffinity::new()), + ); + let (_, first) = Arc::clone(&algorithm) + .run( + Context::default(), + affinity_request(true, "child-1", "task-1"), + ) + .await?; + let (trace, second) = algorithm + .run( + Context::default(), + affinity_request(true, "child-1", "task-2"), + ) + .await?; + + assert_eq!(first.selected_model(), second.selected_model()); + assert!(trace[0] + .reasoning() + .unwrap_or_default() + .contains("affinity reused")); + Ok(()) + } + + #[tokio::test] + async fn single_target_subagent_is_retained_without_reselection( + ) -> Result<(), Box> { + let algorithm = + shared_algorithm_with_affinity(&["only/model"], Arc::new(SubAgentAffinity::new())); + Arc::clone(&algorithm) + .run( + Context::default(), + affinity_request(true, "child-1", "task-1"), + ) + .await?; + let (trace, response) = algorithm + .run( + Context::default(), + affinity_request(true, "child-1", "task-2"), + ) + .await?; + + assert_eq!( + response + .llm_response + .as_agg() + .map(completion_text) + .unwrap_or_default(), + "only/model" + ); + assert!(trace[0] + .reasoning() + .unwrap_or_default() + .contains("affinity reused")); + Ok(()) + } + + #[tokio::test] + async fn subagent_affinity_does_not_retain_root_selection( + ) -> Result<(), Box> { + let affinity = Arc::new(SubAgentAffinity::new()); + let algorithm = shared_algorithm_with_affinity( + &["a/model", "b/model"], + Arc::clone(&affinity) as Arc, + ); + Arc::clone(&algorithm) + .run( + Context::default(), + affinity_request(false, "root-1", "task-1"), + ) + .await?; + + assert!(affinity + .assignment(&affinity_request(false, "root-1", "task-2")) + .is_none()); + let (trace, _) = algorithm + .run( + Context::default(), + affinity_request(false, "root-1", "task-2"), + ) + .await?; + assert!(trace[0] + .reasoning() + .unwrap_or_default() + .starts_with("random routing selected")); + Ok(()) + } + + #[tokio::test] + async fn session_affinity_reuses_selection_for_different_agents( + ) -> Result<(), Box> { + let algorithm = shared_algorithm_with_affinity( + &["a/model", "b/model"], + Arc::new(SessionAffinity::new()), + ); + let (_, first) = Arc::clone(&algorithm) + .run( + Context::default(), + affinity_request(false, "agent-a", "task-1"), + ) + .await?; + let (_, second) = algorithm + .run( + Context::default(), + affinity_request(false, "agent-b", "task-2"), + ) + .await?; + + assert_eq!(first.selected_model(), second.selected_model()); + Ok(()) + } + + #[tokio::test] + async fn concurrent_first_subagent_turns_use_one_assignment( + ) -> Result<(), Box> { + let algorithm = shared_algorithm_with_affinity( + &["a/model", "b/model"], + Arc::new(SubAgentAffinity::new()), + ); + let first = Arc::clone(&algorithm).run( + Context::default(), + affinity_request(true, "child-1", "task-1"), + ); + let second = algorithm.run( + Context::default(), + affinity_request(true, "child-1", "task-1"), + ); + let (first, second) = tokio::join!(first, second); + let (_, first) = first?; + let (_, second) = second?; + + assert_eq!(first.selected_model(), second.selected_model()); + Ok(()) + } } diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 0176342a..0cd7b751 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -54,15 +54,15 @@ //! //! ## Algorithms //! -//! [`RandomAlgo`] provides uniform random routing in the core crate. Worked -//! [`agentic::AgentAwareOrchAlgo`] provides sub-agent-aware model-pool routing and -//! neutral harness-header normalization. Worked implementations of an LLM classifier -//! and a stateful ensemble, plus runnable agents, live in the `libsy-examples` crate. +//! [`RandomAlgo`] provides uniform random routing in the core crate and can retain +//! selections through an [`affinity::Affinity`] policy. Worked implementations of an +//! LLM classifier and a stateful ensemble, plus runnable agents, live in the +//! `libsy-examples` crate. mod algorithms; pub use algorithms::noop::{NoopAlgo, NoopDecision}; pub use algorithms::rand::{RandomAlgo, RandomDecision}; -pub mod agentic; +pub mod affinity; mod driver; diff --git a/crates/libsy/tests/affinity.rs b/crates/libsy/tests/affinity.rs new file mode 100644 index 00000000..6226bacb --- /dev/null +++ b/crates/libsy/tests/affinity.rs @@ -0,0 +1,45 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Public affinity extension-point contract tests. + +use libsy::affinity::{Affinity, AffinityKey, AffinityState}; +use libsy::{Metadata, Request}; +use switchyard_protocol::text_request; + +/// Example policy implemented outside the `libsy` crate. +#[derive(Default)] +struct TaskAffinity { + state: AffinityState, +} + +impl Affinity for TaskAffinity { + fn key(&self, request: &Request) -> Option { + let metadata = request.metadata.as_ref()?; + Some(AffinityKey::new( + "task", + vec![metadata.session_id.clone()?, metadata.task_id.clone()?], + )) + } + + fn state(&self) -> &AffinityState { + &self.state + } +} + +#[test] +fn external_policy_can_reuse_the_public_affinity_state() { + let affinity = TaskAffinity::default(); + let request = Request { + llm_request: text_request(Some("auto".to_string()), "hi"), + raw_request: None, + metadata: Some(Metadata { + session_id: Some("session-1".to_string()), + task_id: Some("task-1".to_string()), + ..Metadata::default() + }), + }; + + assert_eq!(affinity.retain(&request, "model-a".to_string()), "model-a"); + assert_eq!(affinity.assignment(&request).as_deref(), Some("model-a")); +} diff --git a/demo/libsy-proxy/README.md b/demo/libsy-proxy/README.md index 0b2aa39f..f5962304 100644 --- a/demo/libsy-proxy/README.md +++ b/demo/libsy-proxy/README.md @@ -1,9 +1,9 @@ -# Agent-aware libsy proxy POC +# Affinity-aware libsy proxy POC -This demo embeds `libsy`'s agent-aware model-pool router in the PR #17 Switchyard server. It accepts +This demo embeds `libsy`'s `RandomAlgo` with `SubAgentAffinity` in the Switchyard server. It accepts OpenAI Chat, Anthropic Messages, and OpenAI Responses request shapes, normalizes Codex/Relay lineage -headers, asks the classifier for a model assignment, and reuses valid assignments per stable -`(session, agent, explicit task)` key. +headers, randomly routes ordinary requests on every turn, and retains the first selection for each +stable `(session_id, agent_id)` child-agent key. Run it with an NVIDIA Inference Hub bearer token: @@ -11,8 +11,8 @@ Run it with an NVIDIA Inference Hub bearer token: ANTHROPIC_API_KEY="$INFERENCE_HUB_SY_API_KEY" cargo run -p libsy-proxy ``` -Send two buffered calls for the same Codex child thread. The first call classifies; the second reuses -the assignment: +Send two buffered calls for the same Codex child thread. The first call randomly selects a target; +the second reuses the assignment: ```bash curl http://127.0.0.1:4000/v1/chat/completions \ @@ -22,13 +22,13 @@ curl http://127.0.0.1:4000/v1/chat/completions \ -H 'x-codex-parent-thread-id: root-thread-1' \ -H 'x-openai-subagent: collab_spawn' \ -H 'x-codex-turn-metadata: {"session_id":"root-session-1","thread_id":"child-thread-1","parent_thread_id":"root-thread-1","turn_id":"turn-1","subagent_kind":"collab_spawn"}' \ - -d '{"model":"libsy-agent-aware","messages":[{"role":"user","content":"Survey the request protocol and report evidence."}],"stream":false}' + -d '{"model":"libsy-random-affinity","messages":[{"role":"user","content":"Survey the request protocol and report evidence."}],"stream":false}' ``` The response exposes the chosen logical target and rationale in `x-model-router-*` headers. Explicit `x-switchyard-session-id`, `x-switchyard-agent-id`, `x-switchyard-parent-agent-id`, -`x-switchyard-agent-role`, `x-switchyard-task-id`, and `x-switchyard-task-kind` headers override -harness-derived values. +`x-switchyard-is-subagent`, `x-switchyard-agent-role`, and `x-switchyard-task-kind` headers override +harness-derived values. A root-agent request is not retained and runs random routing on every turn. The demo currently requires buffered upstream responses (`"stream": false`). Header normalization and route selection work for Responses-shaped requests, but running an unmodified streaming Codex diff --git a/demo/libsy-proxy/src/main.rs b/demo/libsy-proxy/src/main.rs index 2992271a..a735fb43 100644 --- a/demo/libsy-proxy/src/main.rs +++ b/demo/libsy-proxy/src/main.rs @@ -1,11 +1,11 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! Demo proxy combining Switchyard HTTP/translation with libsy agent-aware routing. +//! Demo proxy combining Switchyard HTTP/translation with libsy random routing and affinity. //! //! The server accepts OpenAI Chat, Anthropic Messages, and OpenAI Responses requests. -//! libsy normalizes harness identity headers, assigns a stable agent/task to a model -//! pool target, and makes calls through Switchyard's OpenAI-compatible backend. +//! libsy normalizes harness identity headers, randomly routes ordinary requests, and +//! retains the first model selected for each stable child agent. use std::collections::BTreeMap; use std::error::Error; @@ -15,12 +15,10 @@ use std::sync::Arc; use async_trait::async_trait; use serde_json::Value; -use libsy::agentic::{ - metadata_from_headers, AgentAwareOrchAlgo, AgentRoutingCandidate, AgentRoutingDecision, -}; +use libsy::affinity::{metadata_from_headers, SubAgentAffinity}; use libsy::{ - Algorithm, Context, Decision, LlmClient, LlmResponse, LlmTarget, LlmTargetSet, Request, - Response, RoutedRequest, + Algorithm, Context, Decision, LlmClient, LlmResponse, LlmTarget, LlmTargetSet, RandomAlgo, + Request, Response, RoutedRequest, }; use switchyard_components::OpenAiPassthroughBackend; use switchyard_components_v2::{Profile, ProfileInput, ProfileResponse, RoutingMetadata}; @@ -31,16 +29,14 @@ use switchyard_core::{ use switchyard_server::{build_switchyard_router, ProfileRegistry, ServerState}; use switchyard_translation::{TranslationEngine, TranslationPolicy, WireFormat}; -const CLASSIFIER_MODEL: &str = "nvidia/deepseek-ai/deepseek-v4-flash"; const FRONTIER_MODEL: &str = "aws/anthropic/bedrock-claude-opus-4-7"; const FAST_MODEL: &str = "nvidia/deepseek-ai/deepseek-v4-flash"; -const CLASSIFIER_TARGET: &str = "classifier"; const FRONTIER_TARGET: &str = "frontier"; const FAST_TARGET: &str = "fast"; const DEFAULT_BASE_URL: &str = "https://inference-api.nvidia.com/v1"; const DEFAULT_ADDR: &str = "127.0.0.1:4000"; -const PROFILE_MODEL_ID: &str = "libsy-agent-aware"; +const PROFILE_MODEL_ID: &str = "libsy-random-affinity"; type BoxErr = Box; @@ -84,13 +80,13 @@ impl LlmClient for SwitchyardBackendClient { } /// Switchyard profile that delegates all routing decisions to libsy. -struct LibsyAgentAwareProfile { +struct LibsyAffinityProfile { algorithm: Arc, translation: Arc, } #[async_trait] -impl Profile for LibsyAgentAwareProfile { +impl Profile for LibsyAffinityProfile { async fn run(&self, input: ProfileInput) -> Result { let request = libsy_request(&input, self.translation.as_ref())?; let (trace, response) = Arc::clone(&self.algorithm) @@ -141,20 +137,20 @@ fn libsy_request(input: &ProfileInput, translation: &TranslationEngine) -> Resul }) } -/// Preserve routed provider payloads; encode classifier calls from the neutral IR. +/// Preserve routed provider payloads; encode neutral requests when no raw body exists. fn chat_request_for_call( request: &Request, model: &str, translation: &TranslationEngine, ) -> std::result::Result { let Some(mut body) = request.raw_request.clone() else { - let mut classifier_request = request.llm_request.clone(); - classifier_request.model = Some(model.to_string()); - classifier_request.stream = false; + let mut neutral_request = request.llm_request.clone(); + neutral_request.model = Some(model.to_string()); + neutral_request.stream = false; let body = translation .encode_request( WireFormat::OpenAiChat, - &classifier_request, + &neutral_request, &TranslationPolicy::default(), ) .map_err(|error| BoxErr::from(error.to_string()))? @@ -198,8 +194,6 @@ fn wire_format(request_type: ChatRequestType) -> WireFormat { /// Surface libsy's routing decision as `x-model-router-*` response headers. fn routing_metadata(trace: &[Arc]) -> RoutingMetadata { let decision = trace.last(); - let agent = - decision.and_then(|decision| decision.as_any().downcast_ref::()); let selected_target = decision.map(|decision| decision.selected_model()); RoutingMetadata { selected_model: selected_target.map(|target| match target { @@ -208,8 +202,8 @@ fn routing_metadata(trace: &[Arc]) -> RoutingMetadata { other => other.to_string(), }), selected_tier: selected_target.map(str::to_string), - confidence: agent.and_then(|decision| decision.confidence), - router_version: Some("libsy-agent-aware-v1".to_string()), + confidence: None, + router_version: Some("libsy-random-affinity-v1".to_string()), tolerance: None, rationale: decision.and_then(|decision| decision.reasoning().map(str::to_string)), } @@ -229,7 +223,6 @@ fn build_algorithm() -> Result> { timeout_secs: Some(120.0), })?); let model_ids = BTreeMap::from([ - (CLASSIFIER_TARGET.to_string(), CLASSIFIER_MODEL.to_string()), (FRONTIER_TARGET.to_string(), FRONTIER_MODEL.to_string()), (FAST_TARGET.to_string(), FAST_MODEL.to_string()), ]); @@ -243,33 +236,16 @@ fn build_algorithm() -> Result> { semantic_name: name.to_string(), llm_client: Some(Arc::clone(&client)), }; - let targets = LlmTargetSet::new(vec![ - target(CLASSIFIER_TARGET), - target(FRONTIER_TARGET), - target(FAST_TARGET), - ]); + let targets = LlmTargetSet::new(vec![target(FRONTIER_TARGET), target(FAST_TARGET)]); + let algorithm = RandomAlgo::new(targets).with_affinity(Arc::new(SubAgentAffinity::new())); - Ok(Arc::new(AgentAwareOrchAlgo::new( - CLASSIFIER_TARGET, - vec![ - AgentRoutingCandidate::new( - FRONTIER_TARGET, - "frontier model for planning, ambiguous implementation, synthesis, and review", - ), - AgentRoutingCandidate::new( - FAST_TARGET, - "efficient model for bounded exploration, retrieval, and mechanical edits", - ), - ], - FRONTIER_TARGET, - targets, - ))) + Ok(Arc::new(algorithm)) } #[tokio::main] async fn main() -> Result<()> { let translation = Arc::new(TranslationEngine::default()); - let profile = Arc::new(LibsyAgentAwareProfile { + let profile = Arc::new(LibsyAffinityProfile { algorithm: build_algorithm()?, translation, }) as Arc; @@ -293,9 +269,9 @@ async fn main() -> Result<()> { .map_err(|error| SwitchyardError::Other(error.to_string()))?; println!("libsy-proxy listening on http://{bound_addr}"); - println!(" routing (libsy agent-aware): classifier={CLASSIFIER_MODEL}"); - println!(" frontier={FRONTIER_MODEL}"); - println!(" fast={FAST_MODEL}"); + println!(" routing (libsy random + sub-agent affinity):"); + println!(" frontier={FRONTIER_MODEL}"); + println!(" fast={FAST_MODEL}"); println!( " send model \"{PROFILE_MODEL_ID}\" to /v1/chat/completions, /v1/messages, or /v1/responses" ); @@ -318,9 +294,9 @@ mod tests { #[test] fn routed_call_preserves_provider_body_and_rewrites_model() -> std::result::Result<(), BoxErr> { let request = Request { - llm_request: text_request(Some("libsy-agent-aware".to_string()), "inspect"), + llm_request: text_request(Some(PROFILE_MODEL_ID.to_string()), "inspect"), raw_request: Some(json!({ - "model": "libsy-agent-aware", + "model": PROFILE_MODEL_ID, "input": "inspect", "tools": [{"type": "function", "name": "shell"}], "stream": false, @@ -349,7 +325,7 @@ mod tests { fn decodes_responses_input_and_normalizes_headers() -> Result<()> { let input = ProfileInput { request: ChatRequest::openai_responses(json!({ - "model": "libsy-agent-aware", + "model": PROFILE_MODEL_ID, "input": [{ "type": "message", "role": "user", @@ -358,10 +334,14 @@ mod tests { "stream": false, })), metadata: RequestMetadata { - headers: BTreeMap::from([( - "thread-id".to_string(), - vec!["child-agent".to_string()], - )]), + headers: BTreeMap::from([ + ("session-id".to_string(), vec!["session-1".to_string()]), + ("thread-id".to_string(), vec!["child-agent".to_string()]), + ( + "x-openai-subagent".to_string(), + vec!["collab_spawn".to_string()], + ), + ]), ..RequestMetadata::default() }, }; @@ -375,22 +355,30 @@ mod tests { .and_then(|metadata| metadata.agent_id.as_deref()), Some("child-agent") ); + assert_eq!( + request + .metadata + .as_ref() + .and_then(|metadata| metadata.agent_context.as_deref()) + .map(|agent| agent.is_subagent), + Some(true) + ); Ok(()) } #[test] - fn classifier_call_uses_a_buffered_synthetic_request() -> std::result::Result<(), BoxErr> { + fn neutral_call_uses_a_buffered_synthetic_request() -> std::result::Result<(), BoxErr> { let request = Request { llm_request: text_request(Some("auto".to_string()), "classify this"), raw_request: None, metadata: None, }; - let classifier = - chat_request_for_call(&request, "classifier/model", &TranslationEngine::default())?; - assert_eq!(classifier.request_type(), ChatRequestType::OpenAiChat); + let neutral = + chat_request_for_call(&request, "provider/model", &TranslationEngine::default())?; + assert_eq!(neutral.request_type(), ChatRequestType::OpenAiChat); assert_ne!( - classifier.body().get("stream").and_then(Value::as_bool), + neutral.body().get("stream").and_then(Value::as_bool), Some(true) ); Ok(()) @@ -398,18 +386,14 @@ mod tests { #[test] fn response_metadata_exposes_provider_model_and_logical_tier() { - let decision: Arc = Arc::new(AgentRoutingDecision { + let decision: Arc = Arc::new(libsy::RandomDecision { selected_model: FAST_TARGET.to_string(), - reason: "bounded lookup".to_string(), - task_kind: Some("research".to_string()), - confidence: Some(0.9), - agent_id: Some("child-1".to_string()), - cache_hit: false, + reasoning: "random routing selected target 'fast'".to_string(), }); let metadata = routing_metadata(&[decision]); assert_eq!(metadata.selected_model.as_deref(), Some(FAST_MODEL)); assert_eq!(metadata.selected_tier.as_deref(), Some(FAST_TARGET)); - assert_eq!(metadata.confidence, Some(0.9)); + assert_eq!(metadata.confidence, None); } } From e61b9c6358048b8699a583725649529718256b7d Mon Sep 17 00:00:00 2001 From: ayushag Date: Thu, 16 Jul 2026 15:24:05 -0700 Subject: [PATCH 3/7] docs(libsy): refresh affinity terminology Signed-off-by: ayushag --- crates/libsy-protocol/src/envelope.rs | 4 ++-- demo/libsy-proxy/src/main.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/libsy-protocol/src/envelope.rs b/crates/libsy-protocol/src/envelope.rs index 36a39054..aa8f0591 100644 --- a/crates/libsy-protocol/src/envelope.rs +++ b/crates/libsy-protocol/src/envelope.rs @@ -40,7 +40,7 @@ pub struct Metadata { pub wire_format: Option, } -/// Optional lineage and semantic signals for agent-aware routing. +/// Optional lineage and semantic signals for affinity-aware routing. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct AgentContext { /// Whether the harness explicitly identified this request as coming from a child agent. @@ -51,7 +51,7 @@ pub struct AgentContext { pub agent_kind: Option, /// Semantic agent role, such as `explorer`, `worker`, or `reviewer`. pub agent_role: Option, - /// Semantic task class supplied by the harness or a prior classifier. + /// Semantic task class supplied by the harness. pub task_kind: Option, /// Id of the current agent turn. pub turn_id: Option, diff --git a/demo/libsy-proxy/src/main.rs b/demo/libsy-proxy/src/main.rs index a735fb43..4886fcb2 100644 --- a/demo/libsy-proxy/src/main.rs +++ b/demo/libsy-proxy/src/main.rs @@ -369,7 +369,7 @@ mod tests { #[test] fn neutral_call_uses_a_buffered_synthetic_request() -> std::result::Result<(), BoxErr> { let request = Request { - llm_request: text_request(Some("auto".to_string()), "classify this"), + llm_request: text_request(Some("auto".to_string()), "route this"), raw_request: None, metadata: None, }; From c2cefcd66f4ec60e28d1285176cc069cc7725a57 Mon Sep 17 00:00:00 2001 From: ayushag Date: Thu, 16 Jul 2026 15:28:03 -0700 Subject: [PATCH 4/7] ci: rerun checks after GitHub outage Signed-off-by: ayushag From 55f9d5ab426140d348a992035db96bc3d0b1f504 Mon Sep 17 00:00:00 2001 From: ayushag Date: Thu, 16 Jul 2026 15:29:44 -0700 Subject: [PATCH 5/7] ci: pin uv setup version Signed-off-by: ayushag --- .github/workflows/ci.yml | 4 ++++ .github/workflows/perf.yml | 1 + 2 files changed, 5 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5f8d5aa7..50f79114 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -22,6 +22,7 @@ jobs: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v6 with: + version: "0.11.29" python-version: "3.12" enable-cache: true cache-dependency-glob: "uv.lock" @@ -92,6 +93,7 @@ jobs: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v6 with: + version: "0.11.29" python-version: "3.12" enable-cache: true cache-dependency-glob: "uv.lock" @@ -113,6 +115,7 @@ jobs: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v6 with: + version: "0.11.29" python-version: ${{ matrix.python-version }} enable-cache: true cache-dependency-glob: "uv.lock" @@ -136,6 +139,7 @@ jobs: - uses: actions/checkout@v4 - uses: astral-sh/setup-uv@v6 with: + version: "0.11.29" python-version: "3.12" enable-cache: true cache-dependency-glob: "uv.lock" diff --git a/.github/workflows/perf.yml b/.github/workflows/perf.yml index a7131216..1ece41e2 100644 --- a/.github/workflows/perf.yml +++ b/.github/workflows/perf.yml @@ -41,6 +41,7 @@ jobs: - uses: astral-sh/setup-uv@v6 with: + version: "0.11.29" python-version: "3.12" enable-cache: true cache-dependency-glob: "uv.lock" From adfc7bc67e1aabffb82df7a2d9dddfebbf201795 Mon Sep 17 00:00:00 2001 From: Greg Clark Date: Fri, 17 Jul 2026 10:07:33 -0400 Subject: [PATCH 6/7] fix(libsy-proxy): adapt demo client to renamed RoutedLlmClient trait Signed-off-by: Greg Clark --- crates/libsy/src/algorithms/rand.rs | 4 +++- demo/libsy-proxy/src/main.rs | 22 +++++++++++++--------- 2 files changed, 16 insertions(+), 10 deletions(-) diff --git a/crates/libsy/src/algorithms/rand.rs b/crates/libsy/src/algorithms/rand.rs index f75c51b2..f4f1bdf5 100644 --- a/crates/libsy/src/algorithms/rand.rs +++ b/crates/libsy/src/algorithms/rand.rs @@ -138,7 +138,9 @@ mod tests { use switchyard_protocol::{completion_text, text_request, text_response}; use crate::affinity::{Affinity, SessionAffinity, SubAgentAffinity}; - use crate::{AgentContext, LlmResponse, LlmTarget, Metadata, Request, RoutedLlmClient, Signals}; + use crate::{ + AgentContext, LlmResponse, LlmTarget, Metadata, Request, RoutedLlmClient, Signals, + }; /// Echoes the selected target so tests can inspect which target was called. struct EchoClient; diff --git a/demo/libsy-proxy/src/main.rs b/demo/libsy-proxy/src/main.rs index 4886fcb2..1a10150d 100644 --- a/demo/libsy-proxy/src/main.rs +++ b/demo/libsy-proxy/src/main.rs @@ -17,8 +17,8 @@ use serde_json::Value; use libsy::affinity::{metadata_from_headers, SubAgentAffinity}; use libsy::{ - Algorithm, Context, Decision, LlmClient, LlmResponse, LlmTarget, LlmTargetSet, RandomAlgo, - Request, Response, RoutedRequest, + Algorithm, Context, Decision, LlmResponse, LlmTarget, LlmTargetSet, RandomAlgo, Request, + Response, RoutedLlmClient, }; use switchyard_components::OpenAiPassthroughBackend; use switchyard_components_v2::{Profile, ProfileInput, ProfileResponse, RoutingMetadata}; @@ -48,15 +48,19 @@ struct SwitchyardBackendClient { } #[async_trait] -impl LlmClient for SwitchyardBackendClient { - async fn call(&self, routed: RoutedRequest) -> std::result::Result { - let target = routed.decision.selected_model(); +impl RoutedLlmClient for SwitchyardBackendClient { + async fn call( + &self, + _ctx: Context, + request: Request, + decision: Arc, + ) -> std::result::Result { + let target = decision.selected_model(); let model = self .model_ids .get(target) .ok_or_else(|| format!("no provider model configured for target '{target}'"))?; - let chat_request = - chat_request_for_call(&routed.request, model, self.translation.as_ref())?; + let chat_request = chat_request_for_call(&request, model, self.translation.as_ref())?; let mut ctx = ProxyContext::new(); let response = self @@ -74,7 +78,7 @@ impl LlmClient for SwitchyardBackendClient { Ok(Response { llm_response: LlmResponse::Agg(decoded.response), - metadata: routed.request.metadata, + metadata: request.metadata, }) } } @@ -231,7 +235,7 @@ fn build_algorithm() -> Result> { backend, model_ids, translation, - }) as Arc; + }) as Arc; let target = |name: &str| LlmTarget { semantic_name: name.to_string(), llm_client: Some(Arc::clone(&client)), From c6291e7280e6d43961021b97fe72efe3e6434ffb Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Fri, 17 Jul 2026 16:07:45 -0700 Subject: [PATCH 7/7] docs: add sub-agent routing proposal --- docs/internal/subagent_routing.md | 269 ++++++++++++++++++++++++++++++ 1 file changed, 269 insertions(+) create mode 100644 docs/internal/subagent_routing.md diff --git a/docs/internal/subagent_routing.md b/docs/internal/subagent_routing.md new file mode 100644 index 00000000..54750e04 --- /dev/null +++ b/docs/internal/subagent_routing.md @@ -0,0 +1,269 @@ +# Sub-Agent Routing Infrastructure + +**Status:** Proposed + +## Why this matters + +Frontier models are becoming orchestrators, not only answer generators. Anthropic describes +[Claude Fable 5](https://www.anthropic.com/claude/fable) as capable of planning across stages, +delegating to sub-agents, and checking their work over long-running tasks. Its +[system card](https://www-cdn.anthropic.com/2f9323abbcc4abe219577539efe19a623c9ca2bd/Claude%20Fable%205%20%26%20Claude%20Mythos%205%20System%20Card.pdf) +also evaluates blocking sub-agents, fixed-agent teams, and asynchronous sub-agents. In those +evaluations, the multi-agent variants outperformed the single-agent variants, and asynchronous +workers improved latency at the cost of using more tokens. + +This changes the role of the proxy. Once the harness declares that a request belongs to an isolated +worker, the proxy should preserve that boundary instead of treating it as an ordinary turn and +allowing the normal pipeline to reinterpret it. The orchestrator and worker may intentionally use +different models, even when the worker model is not cheaper. + +The routing opportunity exists because a sub-agent task crosses a real inference boundary: + +- it receives a bounded delegation prompt and its own context; +- it issues separate model requests with its own agent identity; +- it can run concurrently with the parent and other workers; and +- it returns a result or summary to the orchestrator rather than extending the parent's reasoning + inline. + +The tasks are isolated at the model-request boundary, even when they are not logically independent. +A workflow may still contain dependencies, such as research before implementation or review after +implementation. The harness owns that task graph; Switchyard only sees separately attributable +inference requests. That is enough to route each worker without understanding or reimplementing the +orchestration plan. + +### Benefits for Switchyard + +Switchyard is the proxy between the harness and the model provider. That position creates two +primary responsibilities when the request is marked as sub-agent work: + +1. **Preserve input-cache continuity.** Prompt/KV caches are normally scoped to a model and often + to an endpoint or replica. If Switchyard ignores the sub-agent signal and lets every worker turn + pass through an ordinary per-request router, one sub-agent loop may hop between models or + targets. The growing worker prefix then has to be processed again instead of reusing the cache + created by earlier turns. A configured `subagent_target` keeps worker requests on an intentional, + stable target at the Switchyard layer. +2. **Honor the user's model expectation.** A user or harness may deliberately assign one model to + orchestration and another to delegated execution, review, or exploration. Sending the marked + worker request through the normal router can override that division of responsibility. The + profile-level `subagent_target` makes the expected worker model an explicit deployment contract. + +This infrastructure does not implement a new prompt cache. It preserves the conditions under which +the upstream cache can work: stable model/target selection and a cache-compatible wire format. If a +target itself spreads requests across replicas that do not share KV state, provider-side or +load-balancer affinity is still required. + +The current repository already has the pieces needed to enforce those responsibilities: + +- **Existing cache semantics.** Switchyard already preserves Anthropic `cache_control` on native + Anthropic paths, injects cache breakpoints when an OpenAI-shaped client is routed to an Anthropic + target in supported profiles, and records cached and cache-creation tokens per model/tier. +- **Stable target bypass.** For a profile that normally runs a classifier, random choice, or + multi-stage router, a recognized sub-agent request can take the existing passthrough path directly + instead of recomputing the worker's model on every turn. +- **Protocol-independent behavior.** Codex enters through OpenAI Responses and Claude Code through + Anthropic Messages, but both endpoints already create provider-neutral `ChatRequest` values and + retain normalized request headers for the profile. +- **Reuse of production behavior.** Existing target resolution, backend-format selection, + translation, streaming, error handling, lifecycle, and usage accounting remain on the request + path. Sub-agent routing selects an existing branch rather than creating a parallel serving stack. +- **Capacity isolation.** A `subagent_target` can point at a dedicated model or separately scaled + endpoint, keeping bursts of parallel worker traffic from consuming the orchestrator's capacity. +- **Measurable rollout.** Existing profile routing metadata and shared statistics can attribute + target selection, cached tokens, cache-creation tokens, latency, and errors to the sub-agent + branch without labeling metrics with high-cardinality agent IDs. +- **Safe opt-in.** Because the behavior belongs to each profile and is absent by default, operators + can enable it route by route. Profiles without `subagent_target` retain their current behavior. + +Lower inference cost or routing latency may follow from choosing a different worker target, but +neither is required. The core value is preserving cache locality and honoring the model boundary +expressed by the user, harness, and profile configuration. + +## Objective + +Let a Switchyard profile send sub-agent work to a dedicated model while leaving its normal +routing behavior unchanged. + +The design should: + +- recognize sub-agent requests from Codex and Claude Code; +- preserve a stable, cache-compatible target for a sub-agent loop; +- honor the profile's explicit worker-model choice; +- reuse Switchyard targets, profiles, backends, translation, streaming, and statistics; +- work with any routing profile rather than being implemented separately by every router; and +- be inactive when the selected profile does not configure a sub-agent target. + +## Task-type handling + +Treat every recognized delegated user task equally. All of them use the profile's single +`subagent_target`; Switchyard does not classify prompts or select targets by task type. This keeps +the worker target stable for input-cache continuity and honors the user's orchestrator-versus-worker +model choice. + +Codex exposes partial subtypes such as `review` and `collab_spawn`, while Claude Code does not expose +a portable semantic task type. Switchyard may retain an explicit subtype as metadata, but it does +not affect routing. Codex `review` and `collab_spawn` therefore route identically. + +Codex `compact` and `memory_consolidation` maintain harness context rather than execute delegated +user work, so they remain on normal routing. Differentiated worker routing can be reconsidered only +if a harness supplies a stable explicit signal and production evidence shows that users need a +different model. The initial contract remains one binary sub-agent signal and one target. + +## Capability contract + +When Codex or Claude Code identifies a request as sub-agent work, and the selected Switchyard +route appoints a sub-agent model, Switchyard sends the request to that model. Requests without a +recognized signal or without an appointment continue through normal routing. + +One possible configuration surface is an optional common profile field: + +```yaml +profiles: + coding: + type: stage_router + capable: anthropic/claude-opus-4.7 + efficient: moonshotai/kimi-k2.6 + subagent_target: moonshotai/kimi-k2.6 +``` + +In this example, `subagent_target` refers to an existing entry in `targets`. The final field name +and schema may evolve with the profile system; the contract is that the appointment is explicit, +optional, and resolved through existing Switchyard targets. + +At runtime, the selected profile behaves as follows: + +```text +request + -> select route from request model + -> normalize harness agent metadata + -> route appoints a sub-agent model and request is recognized sub-agent work? + yes -> send to the appointed sub-agent model + no -> existing route behavior + -> existing response translation and delivery +``` + +This is shared routing infrastructure. Individual routers do not need to parse Codex or Claude +Code headers. The implementation may use a direct target override, a common wrapper, a +router-native policy, affinity-aware selection, or another shared mechanism. The observable +behavior above is the design constraint. + +## Relationship to agent-aware routing + +The agent-aware routing work on the base branch provides two useful foundations: + +- normalized agent context separates harness-specific header parsing from routing policy; and +- optional sub-agent affinity can keep requests from one child agent on a stable selection. + +This proposal builds on those primitives but defines a different layer: an operator-facing +capability to appoint a model for sub-agent work across Codex and Claude Code. The random-router +affinity example is one possible implementation path, not the product contract. + +Affinity is not required when the appointment resolves to one fixed target. It becomes useful when +the appointed path is itself a router or model pool whose concrete selection could otherwise vary +across turns in the same worker loop. + +The remaining follow-up work is to expose the appointment through the profile system, normalize +Claude Code's child-agent signal alongside Codex, and preserve production streaming, translation, +and statistics behavior on the appointed path. Header normalization should remain shared request +metadata infrastructure so it is useful even when no affinity policy is enabled. + +## Configuration behavior + +The appointment is opt-in per route and should be available to every routing profile without +adding router-specific implementations. It resolves through the existing target system. An +unknown appointed target is a configuration error, while an absent appointment produces the same +runtime route as today. + +The request body's `model` still selects the top-level route first. A request that names a target +directly continues to use that target's normal passthrough route; it does not inherit an unrelated +route's sub-agent appointment. + +## Request detection + +Detection is a small, deterministic function over normalized request metadata. Shared ingress +code may derive that metadata from case-insensitive, trimmed headers, but routing policies should +not parse harness-specific headers independently. + +| Client | Signal | Sub-agent request | +| --- | --- | --- | +| Codex | `x-openai-subagent` | `collab_spawn` or `review` | +| Claude Code | `x-claude-code-agent-id` | Any non-empty value | +| Explicit Switchyard client | `x-switchyard-is-subagent` | `true` | + +Codex values such as `compact` and `memory_consolidation` are not treated as sub-agent work in +the initial implementation. Unknown or malformed values fall through to normal profile routing. +New client values should be added deliberately with captured request fixtures and tests rather +than inferred from header presence alone. + +Claude Code documents `x-claude-code-agent-id` as present only for requests from an in-process +sub-agent or teammate. The initial policy treats both as isolated worker traffic. +`x-claude-code-parent-agent-id` is retained for lineage and cost attribution but is not required to +make the binary routing decision. + +Codex may also provide session, thread, turn, and parent identifiers. These are useful for +observability and optional affinity, but the recognized `x-openai-subagent` subtype remains the +binary routing signal. Agent and parent-agent identifiers must not become high-cardinality metric +labels. + +These client signals should be verified against the upstream implementations when detection is +implemented: + +- [Codex request headers](https://github.com/openai/codex/blob/main/codex-rs/codex-api/src/requests/headers.rs) +- [Claude Code gateway request headers](https://code.claude.com/docs/en/llm-gateway) + +## Runtime behavior + +| Route appointment | Sub-agent signal | Result | +| --- | --- | --- | +| None | Absent or present | Existing route behavior | +| Configured | Absent or unrecognized | Existing route behavior | +| Configured | Recognized | Appointed sub-agent model | + +The decision runs after inbound request translation has produced the normal `ChatRequest`, so it +does not depend on the client wire format. It must work for the endpoints used by both harnesses: + +- Codex through OpenAI Responses; +- Claude Code through Anthropic Messages; and +- OpenAI Chat Completions when used by another compatible client. + +The appointed path must not buffer or rewrite the request or response. Streaming and non-streaming +requests continue through the existing target backend and translation engine. + +If the configured sub-agent target fails, Switchyard returns the normal target error. It must not +silently send the request back through the main router, because that would make cost, capability, +and failure behavior unpredictable. + +## Fast activation benchmark + +[OpenThoughts-TBLite](https://huggingface.co/datasets/open-thoughts/OpenThoughts-TBLite) +provides 100 difficulty-calibrated, TB2-style tasks intended for faster iteration. It is a good +end-to-end smoke test for this infrastructure because it runs through Harbor with the real Codex or +Claude Code harness, but the benchmark alone does not guarantee that either harness will delegate. + +For the activation run, use a Harbor prompt template that encourages native delegation without +prescribing a decomposition: + +```jinja2 +Consider using subagents for independent parts of this task when doing so would improve speed or +quality. Choose the decomposition and number of subagents based on the work, then coordinate their +results before completing the task. + +{{ instruction }} +``` + +Run one task first and repeat the same command with `claude-code`: + +```bash +harbor run \ + --dataset openthoughts-tblite \ + --n-tasks 1 \ + --n-concurrent 1 \ + --agent codex \ + --model \ + --ak prompt_template_path= +``` + +The initial signal is not the benchmark score. It is whether the harness delegates and, when it +does, whether the parent request follows normal profile routing while marked child requests use +the appointed sub-agent model. Run the same task once without the prompt template as a negative +control. After this path is stable, expand to a fixed TBLite subset for repeatable comparisons and +retain full TB2 as the higher-difficulty validation.