laser_sdk/agent/
router.rs

1use crate::agent::{AgentRegistry, RegisteredCard};
2use crate::error::LaserError;
3use crate::provenance::{AgentTopic, Provenance};
4use crate::types::{AgentId, PrincipalId};
5use laser_wire::agent::CapabilityDescriptor;
6
7/// Where a message is addressed: one specific agent, every listener on the topic,
8/// or resolved by capability against the card registry (one capable agent, or all
9/// of them for fan-out).
10#[derive(Debug, Clone)]
11pub enum Router {
12    To(AgentId),
13    /// Route to one named agent only when its live connection authenticated as
14    /// the required principal.
15    ToPrincipal {
16        agent: AgentId,
17        principal: PrincipalId,
18    },
19    Broadcast,
20    /// Route to the one capable agent the policy picks.
21    ToCapable(CapabilitySelector),
22    /// Route to every capable agent (fan-out: a diagnostic pool, a verifier panel).
23    AllCapable(CapabilitySelector),
24}
25
26/// A capability route: which skill to resolve and how to pick among the agents
27/// that advertise it.
28#[derive(Debug, Clone)]
29pub struct CapabilitySelector {
30    pub skill: String,
31    pub policy: RoutePolicy,
32    /// Require the selected live agent connection to authenticate as this
33    /// principal. `None` preserves claim-only discovery.
34    pub principal: Option<PrincipalId>,
35}
36
37impl CapabilitySelector {
38    /// A selector for `skill`, picking by `policy`.
39    pub fn new(skill: impl Into<String>, policy: RoutePolicy) -> Self {
40        Self {
41            skill: skill.into(),
42            policy,
43            principal: None,
44        }
45    }
46
47    /// Bind this capability route to a server-authenticated principal.
48    pub fn principal(mut self, principal: PrincipalId) -> Self {
49        self.principal = Some(principal);
50        self
51    }
52}
53
54/// How a resolved target agent is turned into the topic its work is sent on. The
55/// substrate has no single well-known inbox, on purpose: a real deployment scopes
56/// each user to its own stream and each workflow to its own topic, and a topic
57/// may be created for one process and dropped when done. So routing resolves a
58/// target to where that agent actually consumes, never a shared name baked into
59/// the SDK that breaks the moment two users share a server.
60///
61/// **Stream scope.** A route resolves a *topic*, addressed within the caller's
62/// current stream (`laser.with_stream`). That stream is the isolation boundary: each
63/// user owns a stream, and its agents are addressed by topic inside it, gated
64/// per topic by the streaming server's access control. Addressing an agent on a *different*
65/// stream is cross-stream federation, which needs a reply-to address on the wire
66/// so the target can reply across streams. The routing contract has no such
67/// address, so a route never crosses streams silently.
68///
69/// Set as a default at agent-build time ([`Agent::inbox_route`](crate::agent::Agent)),
70/// where it sticks as the agent's routing convention.
71#[derive(Debug, Clone, Default)]
72pub enum InboxRoute {
73    /// Resolve each target to the `inbox` topic it advertises in its live presence
74    /// (`AGDX_SET_CLIENT_METADATA`, the registry fusion of presence and cards).
75    /// The default. A target advertising no inbox is a per-target failure, surfaced,
76    /// never silently rerouted to a shared topic.
77    #[default]
78    Advertised,
79    /// Send every routed message to this fixed topic regardless of per-agent
80    /// presence. For a workflow where the caller already owns one topic that every
81    /// participant joined, or any custom convention the caller enforces. The
82    /// explicit escape hatch, honored ahead of presence.
83    Fixed(AgentTopic<'static>),
84}
85
86impl InboxRoute {
87    /// Resolve `agent`'s inbox topic under this route, returning the owned Iggy
88    /// `Identifier` to address. `advertised` is the agent's live-presence inbox
89    /// (from [`AgentRegistry::inbox_for`](crate::agent::AgentRegistry::inbox_for)),
90    /// consulted only by the [`Advertised`](Self::Advertised) path. The caller
91    /// wraps the result as [`AgentTopic::Custom`](crate::provenance::AgentTopic::Custom)
92    /// in its own scope (the topic enum borrows the identifier, so it cannot be
93    /// returned). Errors with [`LaserError::NoInbox`] when an advertised route
94    /// finds no inbox for the agent, rather than inventing a destination.
95    pub fn resolve(
96        &self,
97        agent: &AgentId,
98        advertised: Option<&str>,
99    ) -> Result<iggy::prelude::Identifier, LaserError> {
100        let topic = match self {
101            Self::Fixed(topic) => topic.topic_string(),
102            Self::Advertised => advertised
103                .ok_or_else(|| LaserError::NoInbox {
104                    agent: agent.as_str().to_owned(),
105                })?
106                .to_owned(),
107        };
108        iggy::prelude::Identifier::named(&topic).map_err(|error| {
109            LaserError::HandlerConfig(format!("inbox topic `{topic}` is invalid: {error}"))
110        })
111    }
112}
113
114/// How a capability route picks one agent from the agents advertising the skill.
115#[derive(Clone, Default)]
116pub enum RoutePolicy {
117    /// The lowest advertised cost class.
118    Cheapest,
119    /// The lowest advertised latency class.
120    Fastest,
121    /// The lowest advertised load.
122    LeastLoaded,
123    /// Prefer this agent when it is among the candidates, else fall back to
124    /// [`Any`](Self::Any). Keeps a conversation pinned to one agent.
125    Sticky(AgentId),
126    /// Any candidate (the first resolved). The default.
127    #[default]
128    Any,
129    /// Your ranking over exactly the candidate view the shipped policies read,
130    /// so a custom scorer can never see less than a built-in.
131    Custom(std::sync::Arc<dyn RouteScorer>),
132}
133
134// Manual because `Arc<dyn RouteScorer>` has no useful `Debug` (and the derive
135// would demand one of every scorer).
136impl std::fmt::Debug for RoutePolicy {
137    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
138        match self {
139            Self::Cheapest => write!(f, "Cheapest"),
140            Self::Fastest => write!(f, "Fastest"),
141            Self::LeastLoaded => write!(f, "LeastLoaded"),
142            Self::Sticky(agent) => f.debug_tuple("Sticky").field(agent).finish(),
143            Self::Any => write!(f, "Any"),
144            Self::Custom(_) => write!(f, "Custom(..)"),
145        }
146    }
147}
148
149/// A user ranking over capability-route candidates, plugged in with
150/// [`RoutePolicy::Custom`]. `select` returns the index of the chosen candidate
151/// (out of bounds or `None` means no pick, and the route errors with
152/// [`LaserError::NoCapableAgent`](crate::error::LaserError::NoCapableAgent)
153/// exactly as an empty candidate list does).
154pub trait RouteScorer: Send + Sync {
155    /// Pick one of `candidates` for `skill_id`, or `None` to refuse them all.
156    fn select(&self, skill_id: &str, candidates: &[RouteCandidate<'_>]) -> Option<usize>;
157}
158
159/// One capability-route candidate as every policy sees it: the agent, its
160/// registered card, and the descriptor it advertises for the routed skill.
161pub struct RouteCandidate<'a> {
162    /// The candidate agent.
163    pub agent: &'a AgentId,
164    /// The agent's registered card (freshness-filtered by the registry).
165    pub card: &'a RegisteredCard,
166    /// The candidate's descriptor for the routed skill, when its card names
167    /// the skill explicitly. The advisory class fields (`cost_class`,
168    /// `latency_class`, `load`) live here.
169    pub capability: Option<CapabilityDescriptor>,
170}
171
172/// Pick one agent for `skill_id` from `candidates` (a registry
173/// [`resolve`](crate::agent::AgentRegistry::resolve) result) by `policy`, or
174/// `None` when there are no candidates. The class fields are advisory `u*` where
175/// lower is better, and a candidate that does not advertise the ranked field
176/// sorts last so an agent that publishes its class always beats one that omits it.
177fn select(skill_id: &str, candidates: &[&RegisteredCard], policy: &RoutePolicy) -> Option<AgentId> {
178    // Every policy, shipped or custom, ranks over the same candidate view, so
179    // the `Custom` seam can never see less than a built-in.
180    let view: Vec<RouteCandidate<'_>> = candidates
181        .iter()
182        .map(|card| RouteCandidate {
183            agent: &card.agent,
184            card,
185            capability: card
186                .card
187                .capabilities
188                .iter()
189                .find(|capability| capability.skill_id == skill_id)
190                .cloned(),
191        })
192        .collect();
193    // The advisory class fields are `u*` where lower is better, and a candidate
194    // that does not advertise the ranked field sorts last, so an agent that
195    // publishes its class always beats one that omits it.
196    let chosen = match policy {
197        RoutePolicy::Any => view.first(),
198        RoutePolicy::Sticky(agent) => view
199            .iter()
200            .find(|candidate| candidate.agent == agent)
201            .or_else(|| view.first()),
202        RoutePolicy::Cheapest => view.iter().min_by_key(|candidate| {
203            candidate
204                .capability
205                .as_ref()
206                .and_then(|d| d.cost_class)
207                .unwrap_or(u8::MAX)
208        }),
209        RoutePolicy::Fastest => view.iter().min_by_key(|candidate| {
210            candidate
211                .capability
212                .as_ref()
213                .and_then(|d| d.latency_class)
214                .unwrap_or(u8::MAX)
215        }),
216        RoutePolicy::LeastLoaded => view.iter().min_by_key(|candidate| {
217            candidate
218                .capability
219                .as_ref()
220                .and_then(|d| d.load)
221                .unwrap_or(u16::MAX)
222        }),
223        RoutePolicy::Custom(scorer) => scorer.select(skill_id, &view).and_then(|i| view.get(i)),
224    };
225    chosen.map(|candidate| candidate.agent.clone())
226}
227
228impl Router {
229    /// Route to one agent (stamps `agdx.to`).
230    pub fn to(agent: AgentId) -> Self {
231        Self::To(agent)
232    }
233
234    /// Route to `agent` only when its live presence is bound to `principal`.
235    pub fn to_principal(agent: AgentId, principal: PrincipalId) -> Self {
236        Self::ToPrincipal { agent, principal }
237    }
238
239    /// Clear any target so every consumer-group member may handle the message.
240    pub fn broadcast() -> Self {
241        Self::Broadcast
242    }
243
244    /// Route to the one agent advertising `skill` that `policy` picks.
245    pub fn to_capable(skill: impl Into<String>, policy: RoutePolicy) -> Self {
246        Self::ToCapable(CapabilitySelector::new(skill, policy))
247    }
248
249    /// Route to every agent advertising `skill` (fan-out).
250    pub fn all_capable(skill: impl Into<String>, policy: RoutePolicy) -> Self {
251        Self::AllCapable(CapabilitySelector::new(skill, policy))
252    }
253
254    /// Stamp (or clear) the target for the direct variants. The capability
255    /// variants are resolved against the registry by
256    /// [`resolve_targets`](Self::resolve_targets), not stamped directly, so here
257    /// they leave the target unset (a safe broadcast) rather than mis-target.
258    pub fn apply(&self, provenance: &mut Provenance) {
259        match self {
260            Self::To(agent) | Self::ToPrincipal { agent, .. } => {
261                provenance.target_agent_id = Some(agent.clone());
262            }
263            Self::Broadcast | Self::ToCapable(_) | Self::AllCapable(_) => {
264                provenance.target_agent_id = None;
265            }
266        }
267    }
268
269    /// Resolve to the concrete target agent(s) against `registry` at
270    /// `now_micros`: `To` is the one agent, `Broadcast` is the empty list (no
271    /// target), `ToCapable` is the one the policy picks, and `AllCapable` is every
272    /// capable agent. Errors with [`LaserError::NoCapableAgent`] when a capability
273    /// variant matches no live agent.
274    pub fn resolve_targets(
275        &self,
276        registry: &AgentRegistry,
277        now_micros: u64,
278    ) -> Result<Vec<AgentId>, LaserError> {
279        match self {
280            Self::To(agent) => Ok(vec![agent.clone()]),
281            Self::ToPrincipal { agent, principal } => {
282                require_principal(registry, agent, *principal)?;
283                Ok(vec![agent.clone()])
284            }
285            Self::Broadcast => Ok(Vec::new()),
286            Self::ToCapable(selector) => {
287                let candidates = trusted_candidates(registry, selector, now_micros)?;
288                select(&selector.skill, &candidates, &selector.policy)
289                    .map(|agent| vec![agent])
290                    .ok_or_else(|| LaserError::NoCapableAgent {
291                        skill: selector.skill.clone(),
292                    })
293            }
294            Self::AllCapable(selector) => {
295                let agents: Vec<AgentId> = trusted_candidates(registry, selector, now_micros)?
296                    .iter()
297                    .map(|card| card.agent.clone())
298                    .collect();
299                if agents.is_empty() {
300                    Err(LaserError::NoCapableAgent {
301                        skill: selector.skill.clone(),
302                    })
303                } else {
304                    Ok(agents)
305                }
306            }
307        }
308    }
309
310    /// Whether resolving this route requires live, principal-bearing presence.
311    #[cfg(feature = "query")]
312    pub(crate) fn requires_presence(&self) -> bool {
313        matches!(self, Self::ToPrincipal { .. })
314            || matches!(self, Self::ToCapable(selector) | Self::AllCapable(selector) if selector.principal.is_some())
315    }
316
317    /// The authenticated principal a constrained route requires. Contract reply
318    /// verification uses the same identity, joining discovery and authorship.
319    #[cfg(any(feature = "sign", test))]
320    pub(crate) fn required_principal(&self) -> Option<PrincipalId> {
321        match self {
322            Self::ToPrincipal { principal, .. } => Some(*principal),
323            Self::ToCapable(selector) | Self::AllCapable(selector) => selector.principal,
324            Self::To(_) | Self::Broadcast => None,
325        }
326    }
327}
328
329fn trusted_candidates<'a>(
330    registry: &'a AgentRegistry<'_>,
331    selector: &CapabilitySelector,
332    now_micros: u64,
333) -> Result<Vec<&'a RegisteredCard>, LaserError> {
334    let candidates = registry.resolve(&selector.skill, now_micros);
335    let Some(principal) = selector.principal else {
336        return Ok(candidates);
337    };
338    let trusted = filter_candidates_by_principal(&candidates, principal, |agent| {
339        registry.principal_for(agent)
340    });
341    if trusted.is_empty()
342        && let Some(card) = candidates.first()
343    {
344        return Err(principal_mismatch(registry, &card.agent, principal));
345    }
346    Ok(trusted)
347}
348
349fn filter_candidates_by_principal<'a>(
350    candidates: &[&'a RegisteredCard],
351    required: PrincipalId,
352    principal_for: impl Fn(&AgentId) -> Option<PrincipalId>,
353) -> Vec<&'a RegisteredCard> {
354    candidates
355        .iter()
356        .copied()
357        .filter(|card| principal_for(&card.agent) == Some(required))
358        .collect()
359}
360
361fn require_principal(
362    registry: &AgentRegistry<'_>,
363    agent: &AgentId,
364    principal: PrincipalId,
365) -> Result<(), LaserError> {
366    if registry.principal_for(agent) == Some(principal) {
367        Ok(())
368    } else {
369        Err(principal_mismatch(registry, agent, principal))
370    }
371}
372
373fn principal_mismatch(
374    registry: &AgentRegistry<'_>,
375    agent: &AgentId,
376    expected: PrincipalId,
377) -> LaserError {
378    LaserError::RoutePrincipalMismatch {
379        agent: agent.to_string(),
380        expected: expected.get(),
381        actual: registry.principal_for(agent).map(PrincipalId::get),
382    }
383}
384
385#[cfg(test)]
386mod tests {
387    use super::*;
388    use crate::types::ConversationId;
389
390    #[test]
391    fn given_a_route_to_an_agent_when_applied_then_should_set_the_target() {
392        let mut provenance = Provenance::builder()
393            .conversation_id(ConversationId::new())
394            .build();
395        Router::to("executor".parse().expect("executor is a valid agent id"))
396            .apply(&mut provenance);
397        assert_eq!(
398            provenance
399                .target_agent_id
400                .expect("the target should be set")
401                .as_str(),
402            "executor"
403        );
404    }
405
406    #[test]
407    fn given_a_broadcast_route_when_applied_then_should_clear_the_target() {
408        let mut provenance = Provenance::builder()
409            .conversation_id(ConversationId::new())
410            .target_agent_id("executor".parse().expect("executor is a valid agent id"))
411            .build();
412        Router::broadcast().apply(&mut provenance);
413        assert!(provenance.target_agent_id.is_none());
414    }
415
416    fn card(
417        agent: &str,
418        cost: Option<u8>,
419        latency: Option<u8>,
420        load: Option<u16>,
421    ) -> RegisteredCard {
422        use laser_wire::agent::{AgentCard, CapabilityDescriptor};
423        RegisteredCard {
424            agent: agent.parse().expect("valid agent id"),
425            card: AgentCard {
426                name: None,
427                version: None,
428                capabilities: vec![CapabilityDescriptor {
429                    skill_id: "diagnose".to_owned(),
430                    input: None,
431                    output: None,
432                    cost_class: cost,
433                    latency_class: latency,
434                    max_concurrency: None,
435                    health: None,
436                    load,
437                }],
438                ttl_micros: None,
439            },
440            observed_at_micros: 0,
441        }
442    }
443
444    #[test]
445    fn given_candidates_when_selected_by_policy_then_should_pick_the_best_by_that_class() {
446        let cheap = card("cheap", Some(1), Some(9), Some(900));
447        let fast = card("fast", Some(9), Some(1), Some(500));
448        let idle = card("idle", Some(5), Some(5), Some(10));
449        let candidates = [&cheap, &fast, &idle];
450
451        let pick = |policy| select("diagnose", &candidates, &policy).map(|a| a.as_str().to_owned());
452        assert_eq!(pick(RoutePolicy::Cheapest).as_deref(), Some("cheap"));
453        assert_eq!(pick(RoutePolicy::Fastest).as_deref(), Some("fast"));
454        assert_eq!(pick(RoutePolicy::LeastLoaded).as_deref(), Some("idle"));
455        assert_eq!(
456            pick(RoutePolicy::Sticky("fast".parse().expect("valid agent id"))).as_deref(),
457            Some("fast")
458        );
459        // Sticky to an absent agent falls back to a candidate.
460        assert!(pick(RoutePolicy::Sticky("gone".parse().expect("valid agent id"))).is_some());
461    }
462
463    #[test]
464    fn given_a_principal_bound_selector_when_filtering_then_should_drop_foreign_claims() {
465        let trusted = card("trusted", Some(1), None, None);
466        let foreign = card("foreign", Some(2), None, None);
467        let candidates = [&trusted, &foreign];
468
469        let filtered = filter_candidates_by_principal(&candidates, PrincipalId::new(7), |agent| {
470            match agent.as_str() {
471                "trusted" => Some(PrincipalId::new(7)),
472                "foreign" => Some(PrincipalId::new(9)),
473                _ => None,
474            }
475        });
476
477        assert_eq!(filtered.len(), 1);
478        assert_eq!(filtered[0].agent.as_str(), "trusted");
479    }
480
481    #[test]
482    fn given_a_principal_bound_route_when_reading_required_identity_then_should_return_principal() {
483        let route = Router::to_principal(
484            "metrics".parse().expect("metrics is a valid agent id"),
485            PrincipalId::new(42),
486        );
487
488        assert_eq!(route.required_principal(), Some(PrincipalId::new(42)));
489    }
490
491    struct HighestLoad;
492
493    impl RouteScorer for HighestLoad {
494        fn select(&self, _skill_id: &str, candidates: &[RouteCandidate<'_>]) -> Option<usize> {
495            (0..candidates.len()).max_by_key(|&i| {
496                candidates[i]
497                    .capability
498                    .as_ref()
499                    .and_then(|d| d.load)
500                    .unwrap_or(0)
501            })
502        }
503    }
504
505    struct RefuseAll;
506
507    impl RouteScorer for RefuseAll {
508        fn select(&self, _skill_id: &str, _candidates: &[RouteCandidate<'_>]) -> Option<usize> {
509            None
510        }
511    }
512
513    #[test]
514    fn given_a_custom_scorer_when_selected_then_should_rank_over_the_same_candidate_view() {
515        let cheap = card("cheap", Some(1), Some(9), Some(900));
516        let idle = card("idle", Some(5), Some(5), Some(10));
517        let candidates = [&cheap, &idle];
518
519        let busiest = select(
520            "diagnose",
521            &candidates,
522            &RoutePolicy::Custom(std::sync::Arc::new(HighestLoad)),
523        );
524        assert_eq!(
525            busiest.map(|a| a.as_str().to_owned()).as_deref(),
526            Some("cheap")
527        );
528
529        // A scorer that refuses every candidate picks nobody, exactly like an
530        // empty candidate list.
531        let refused = select(
532            "diagnose",
533            &candidates,
534            &RoutePolicy::Custom(std::sync::Arc::new(RefuseAll)),
535        );
536        assert!(refused.is_none());
537        // No candidates resolves to None.
538        assert!(select("diagnose", &[], &RoutePolicy::Any).is_none());
539    }
540
541    #[test]
542    fn given_a_fixed_inbox_route_when_resolved_then_should_use_that_topic_ignoring_presence() {
543        let agent: AgentId = "metrics".parse().expect("valid agent id");
544        let route = InboxRoute::Fixed(AgentTopic::Sessions);
545        // Fixed ignores the advertised inbox entirely.
546        let id = route
547            .resolve(&agent, Some("some.other.topic"))
548            .expect("a fixed route resolves");
549        assert_eq!(id.to_string(), "agent.sessions");
550    }
551
552    #[test]
553    fn given_an_advertised_route_when_an_inbox_is_present_then_should_resolve_to_it() {
554        let agent: AgentId = "metrics".parse().expect("valid agent id");
555        let id = InboxRoute::Advertised
556            .resolve(&agent, Some("metrics.work.2026"))
557            .expect("an advertised inbox resolves");
558        assert_eq!(id.to_string(), "metrics.work.2026");
559    }
560
561    #[test]
562    fn given_an_advertised_route_when_no_inbox_then_should_error_without_a_fallback() {
563        let agent: AgentId = "metrics".parse().expect("valid agent id");
564        let error = InboxRoute::Advertised
565            .resolve(&agent, None)
566            .expect_err("no inbox fails");
567        assert!(
568            matches!(error, LaserError::NoInbox { agent } if agent == "metrics"),
569            "advertised route with no inbox must fail loud, never fall back to a shared topic",
570        );
571    }
572}
laser_sdk/agent/router.rs

laser_sdk/agent/
router.rs
