⎇ 72758e75ce515f355f27615671c95ecaf8c2a8bc
crates/core/brahman-broker/src/lib.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
feat: Crossreferencia — Card.references como grafo del fractal
1//! `brahman-broker` — empareja productores y consumidores por tipo de flujo.
2//!
3//! El broker indexa [`brahman_card::Card`]s registradas por `SessionId` y,
4//! para cada `flow.input` de un consumidor, busca el `flow.output`
5//! compatible de mejor calidad entre los demás. Tres ejes:
6//!
7//! 1. **Estrategia de matching** ([`MatchStrategy`]):
8//! - `Exact`: igualdad estricta de [`brahman_card::TypeRef`].
9//! - `Structural`: misma forma (mismo `package` + `name` para Wit;
10//! ignora `interface`).
11//! - `ExactThenStructural`: prefiere exact; cae en structural si no hay.
12//!
13//! 2. **Override `pin_to`**: si el consumidor declara `pin_to = "label"`,
14//! el broker prefiere productores cuya Card tenga ese `label` (siempre
15//! que el tipo siga matcheando). Si la pista no resuelve, cae en
16//! matching por tipo normal.
17//!
18//! 3. **Prioridad**: empate de tipo se resuelve por
19//! [`brahman_card::Priority`] del productor (mayor gana). Empate de
20//! prioridad se resuelve lexicográficamente por `label` (estable y
21//! determinista).
22//!
23//! El broker es **stateless w.r.t. routes**: cada `find_producer_for` o
24//! `all_matches` se calcula bajo demanda. La única persistencia es el
25//! índice de Cards registradas. Esto permite re-evaluar matches cuando
26//! cambia el set sin invalidar caches.
27
28#![forbid(unsafe_code)]
29#![warn(rust_2018_idioms)]
30
31use std::collections::BTreeMap;
32use std::path::PathBuf;
33
34use brahman_card::{
35 Card, CardKind, CardReference, ContextBias, DataFacet, Flow, Lifecycle, Priority, TypeRef,
36 WitInterface,
37};
38use serde::{Deserialize, Serialize};
39use ulid::Ulid;
40
41/// Identificador de sesión emitido por el handshake. Idéntico al usado por
42/// `brahman-handshake` (no es un re-export para evitar la dependencia).
43pub type SessionId = Ulid;
44
45/// Estrategia de matching de tipos.
46#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
47#[serde(rename_all = "kebab-case")]
48pub enum MatchStrategy {
49 /// Igualdad estricta de `TypeRef`.
50 Exact,
51 /// Misma forma: para `Wit`, mismo `package` + `name`; para
52 /// `Primitive`, mismo `name`.
53 Structural,
54 /// Híbrido: intenta `Exact` primero; si no matchea, `Structural`.
55 /// Reporta cuál estrategia ganó en [`Match::via`].
56 #[default]
57 ExactThenStructural,
58}
59
60/// Configuración del broker.
61#[derive(Debug, Clone, Default)]
62pub struct BrokerConfig {
63 pub strategy: MatchStrategy,
64 /// Contexto operativo activo. Si una Card declara un
65 /// `priority_contexts.<this>`, ese bias se aplica durante el match.
66 /// `None` = sin biases per-contexto, sólo se usa lo estático.
67 pub current_context: Option<String>,
68}
69
70/// Vista mínima de una Card que el broker necesita.
71#[derive(Debug, Clone, Serialize, Deserialize)]
72pub struct BrokeredCard {
73 pub session: SessionId,
74 pub label: String,
75 pub lifecycle: Lifecycle,
76 pub priority: Priority,
77 pub inputs: Vec<Flow>,
78 pub outputs: Vec<Flow>,
79 /// Interfaz WIT extraída si el módulo es "consciente"; `None` si agnóstico.
80 pub wit: Option<WitInterface>,
81 /// Biases per-contexto, propagados desde `Card.priority_contexts`.
82 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
83 pub priority_contexts: BTreeMap<String, ContextBias>,
84 /// Naturaleza de la entidad. Diferencia procesos (Ente) de
85 /// agrupaciones de datos (Data — p. ej. Mónadas Nouser).
86 #[serde(default)]
87 pub kind: CardKind,
88 /// Faceta de datos cuando `kind != Ente`.
89 #[serde(default, skip_serializing_if = "Option::is_none")]
90 pub data: Option<DataFacet>,
91 /// Socket de servicio (data plane) si lo declara la Card.
92 #[serde(default, skip_serializing_if = "Option::is_none")]
93 pub service_socket: Option<PathBuf>,
94 /// Referencias a otras Cards (relaciones declaradas por esta Card).
95 #[serde(default, skip_serializing_if = "Vec::is_empty")]
96 pub references: Vec<CardReference>,
97}
98
99impl BrokeredCard {
100 fn from_card(session: SessionId, card: &Card, wit: Option<WitInterface>) -> Self {
101 Self {
102 session,
103 label: card.label.clone(),
104 lifecycle: card.lifecycle,
105 priority: card.priority,
106 inputs: card.flow.input.clone(),
107 outputs: card.flow.output.clone(),
108 wit,
109 priority_contexts: card.priority_contexts.clone(),
110 kind: card.kind,
111 data: card.data.clone(),
112 service_socket: card.service_socket.clone(),
113 references: card.references.clone(),
114 }
115 }
116}
117
118/// Punto extremo de un flujo: qué sesión + nombre del flow dentro de su Card.
119#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
120pub struct Endpoint {
121 pub session: SessionId,
122 pub flow_name: String,
123}
124
125/// Match concreto entre un consumidor y un productor.
126#[derive(Debug, Clone, Serialize, Deserialize)]
127pub struct Match {
128 pub consumer: Endpoint,
129 pub consumer_label: String,
130 pub producer: Endpoint,
131 pub producer_label: String,
132 /// Tipo del flow (lado consumidor — lado productor coincide en
133 /// estrategia Exact, puede diferir en `interface` en Structural).
134 pub ty: TypeRef,
135 /// Estrategia que efectivamente matcheó.
136 pub via: MatchStrategy,
137 /// `true` si el match fue resuelto por `pin_to` y no por type-search.
138 pub pinned: bool,
139}
140
141// =====================================================================
142// Broker
143// =====================================================================
144
145/// El broker. Registra Cards por SessionId, computa matches bajo demanda.
146#[derive(Debug, Clone, Default)]
147pub struct Broker {
148 cards: BTreeMap<SessionId, BrokeredCard>,
149 config: BrokerConfig,
150}
151
152impl Broker {
153 pub fn new(config: BrokerConfig) -> Self {
154 Self {
155 cards: BTreeMap::new(),
156 config,
157 }
158 }
159
160 /// Registra una Card con su WIT opcional. Devuelve `Some(prev)` si
161 /// reemplazó una existente. Pasar `None` en `wit` indica módulo
162 /// agnóstico (sin contrato WIT extraído).
163 pub fn register(
164 &mut self,
165 session: SessionId,
166 card: &Card,
167 wit: Option<WitInterface>,
168 ) -> Option<BrokeredCard> {
169 self.cards
170 .insert(session, BrokeredCard::from_card(session, card, wit))
171 }
172
173 /// Quita una Card por sesión.
174 pub fn unregister(&mut self, session: SessionId) -> Option<BrokeredCard> {
175 self.cards.remove(&session)
176 }
177
178 /// Cardinalidad del registro.
179 pub fn len(&self) -> usize {
180 self.cards.len()
181 }
182
183 pub fn is_empty(&self) -> bool {
184 self.cards.is_empty()
185 }
186
187 /// Iterador sobre las sesiones registradas.
188 pub fn sessions(&self) -> impl Iterator<Item = SessionId> + '_ {
189 self.cards.keys().copied()
190 }
191
192 /// Iterador sobre las Cards registradas (vista compartida).
193 pub fn cards(&self) -> impl Iterator<Item = &BrokeredCard> + '_ {
194 self.cards.values()
195 }
196
197 /// Busca el mejor productor para un input específico de un consumidor.
198 ///
199 /// Algoritmo:
200 /// 1. Resuelve el flow input en el consumidor.
201 /// 2. Si tiene `pin_to`, prefiere productores con ese `label` que
202 /// matcheen el tipo (cualquier estrategia configurada).
203 /// 3. Si no hay pin_to o la pista falló, escanea todos los outputs
204 /// de las otras Cards. Filtra por compatibilidad de tipo.
205 /// 4. Ordena por (priority desc, label asc) y devuelve el primero.
206 pub fn find_producer_for(&self, consumer: SessionId, input_name: &str) -> Option<Match> {
207 let cons = self.cards.get(&consumer)?;
208 let input = cons.inputs.iter().find(|f| f.name == input_name)?;
209
210 // pin_to efectivo: bias del contexto activo (si la Card declara
211 // override consumer-side) > pin_to estático del Flow.
212 let context_pin = self
213 .context_bias(cons)
214 .and_then(|b| b.pin_to.as_deref());
215 let effective_pin = context_pin.or(input.pin_to.as_deref());
216
217 if let Some(pin) = effective_pin {
218 for prod in self.cards.values() {
219 if prod.session == consumer || prod.label != pin {
220 continue;
221 }
222 for out in &prod.outputs {
223 if let Some(via) = self.types_match(&input.ty, &out.ty) {
224 return Some(self.make_match(cons, prod, input, out, via, true));
225 }
226 }
227 }
228 // Fall through: pin no resuelto, type-search general.
229 }
230
231 let mut candidates: Vec<(&BrokeredCard, &Flow, MatchStrategy)> = Vec::new();
232 for prod in self.cards.values() {
233 if prod.session == consumer {
234 continue;
235 }
236 for out in &prod.outputs {
237 if let Some(via) = self.types_match(&input.ty, &out.ty) {
238 candidates.push((prod, out, via));
239 }
240 }
241 }
242
243 // Sort por (effective priority desc, label asc). El bias del
244 // contexto puede subir o bajar la priority del productor.
245 candidates.sort_by(|(a, _, _), (b, _, _)| {
246 self.effective_priority(b)
247 .cmp(&self.effective_priority(a))
248 .then_with(|| a.label.cmp(&b.label))
249 });
250
251 let (prod, out, via) = candidates.into_iter().next()?;
252 Some(self.make_match(cons, prod, input, out, via, false))
253 }
254
255 /// Devuelve el `ContextBias` que aplica a este Card en el contexto
256 /// activo (si lo hay).
257 fn context_bias<'a>(&self, card: &'a BrokeredCard) -> Option<&'a ContextBias> {
258 self.config
259 .current_context
260 .as_ref()
261 .and_then(|ctx| card.priority_contexts.get(ctx))
262 }
263
264 /// Priority efectiva del Card como productor, considerando el bias
265 /// del contexto activo. El offset se clampa a `[Low=0, Critical=3]`.
266 fn effective_priority(&self, card: &BrokeredCard) -> i16 {
267 let base = priority_value(card.priority);
268 let offset = self
269 .context_bias(card)
270 .map(|b| b.priority_offset as i16)
271 .unwrap_or(0);
272 (base + offset).clamp(0, 3)
273 }
274
275 /// Calcula todos los matches consumer→producer en el set actual.
276 /// Útil para introspección o para que el Admin emita rutas en lote.
277 pub fn all_matches(&self) -> Vec<Match> {
278 let mut out = Vec::new();
279 for cons in self.cards.values() {
280 for input in &cons.inputs {
281 if let Some(m) = self.find_producer_for(cons.session, &input.name) {
282 out.push(m);
283 }
284 }
285 }
286 out
287 }
288
289 fn types_match(&self, consumer_ty: &TypeRef, producer_ty: &TypeRef) -> Option<MatchStrategy> {
290 match self.config.strategy {
291 MatchStrategy::Exact => exact_match(consumer_ty, producer_ty).then_some(MatchStrategy::Exact),
292 MatchStrategy::Structural => {
293 structural_match(consumer_ty, producer_ty).then_some(MatchStrategy::Structural)
294 }
295 MatchStrategy::ExactThenStructural => {
296 if exact_match(consumer_ty, producer_ty) {
297 Some(MatchStrategy::Exact)
298 } else if structural_match(consumer_ty, producer_ty) {
299 Some(MatchStrategy::Structural)
300 } else {
301 None
302 }
303 }
304 }
305 }
306
307 fn make_match(
308 &self,
309 cons: &BrokeredCard,
310 prod: &BrokeredCard,
311 input: &Flow,
312 output: &Flow,
313 via: MatchStrategy,
314 pinned: bool,
315 ) -> Match {
316 Match {
317 consumer: Endpoint {
318 session: cons.session,
319 flow_name: input.name.clone(),
320 },
321 consumer_label: cons.label.clone(),
322 producer: Endpoint {
323 session: prod.session,
324 flow_name: output.name.clone(),
325 },
326 producer_label: prod.label.clone(),
327 ty: input.ty.clone(),
328 via,
329 pinned,
330 }
331 }
332}
333
334// =====================================================================
335// Predicados de matching (libres, testeables aislados)
336// =====================================================================
337
338fn priority_value(p: Priority) -> i16 {
339 match p {
340 Priority::Low => 0,
341 Priority::Normal => 1,
342 Priority::High => 2,
343 Priority::Critical => 3,
344 }
345}
346
347fn exact_match(a: &TypeRef, b: &TypeRef) -> bool {
348 a == b
349}
350
351fn structural_match(a: &TypeRef, b: &TypeRef) -> bool {
352 match (a, b) {
353 (TypeRef::Primitive { name: na }, TypeRef::Primitive { name: nb }) => na == nb,
354 (
355 TypeRef::Wit {
356 package: pa, name: na, ..
357 },
358 TypeRef::Wit {
359 package: pb, name: nb, ..
360 },
361 ) => pa == pb && na == nb,
362 _ => false,
363 }
364}
365
366// =====================================================================
367// Tests
368// =====================================================================
369
370#[cfg(test)]
371mod tests {
372 use super::*;
373 use brahman_card::{Card, Flows, Payload, Supervision, CARD_SCHEMA_VERSION};
374
375 fn card(label: &str, priority: Priority, flows: Flows) -> Card {
376 Card {
377 schema_version: CARD_SCHEMA_VERSION,
378 id: Ulid::new(),
379 label: label.into(),
380 payload: Payload::Virtual,
381 supervision: Supervision::OneShot,
382 priority,
383 flow: flows,
384 ..Default::default()
385 }
386 }
387
388 fn prim(name: &str) -> TypeRef {
389 TypeRef::Primitive { name: name.into() }
390 }
391
392 fn wit(pkg: &str, iface: Option<&str>, name: &str) -> TypeRef {
393 TypeRef::Wit {
394 package: pkg.into(),
395 interface: iface.map(|s| s.into()),
396 name: name.into(),
397 }
398 }
399
400 fn flow(name: &str, ty: TypeRef, pin: Option<&str>) -> Flow {
401 Flow {
402 name: name.into(),
403 ty,
404 pin_to: pin.map(|s| s.into()),
405 }
406 }
407
408 #[test]
409 fn exact_match_same_typeref() {
410 let mut b = Broker::new(BrokerConfig {
411 strategy: MatchStrategy::Exact,
412 current_context: None,
413 });
414 let producer = card(
415 "dht",
416 Priority::Normal,
417 Flows {
418 input: vec![],
419 output: vec![flow("results", prim("string"), None)],
420 },
421 );
422 let consumer = card(
423 "ui",
424 Priority::Normal,
425 Flows {
426 input: vec![flow("query", prim("string"), None)],
427 output: vec![],
428 },
429 );
430 let s_prod = Ulid::new();
431 let s_cons = Ulid::new();
432 b.register(s_prod, &producer, None);
433 b.register(s_cons, &consumer, None);
434
435 let m = b.find_producer_for(s_cons, "query").expect("match");
436 assert_eq!(m.producer_label, "dht");
437 assert_eq!(m.via, MatchStrategy::Exact);
438 assert!(!m.pinned);
439 }
440
441 #[test]
442 fn structural_ignores_interface() {
443 let mut b = Broker::new(BrokerConfig {
444 strategy: MatchStrategy::Structural,
445 current_context: None,
446 });
447 let producer = card(
448 "dht",
449 Priority::Normal,
450 Flows {
451 input: vec![],
452 output: vec![flow(
453 "out",
454 wit("brahman:dht", Some("v1"), "entity-result"),
455 None,
456 )],
457 },
458 );
459 let consumer = card(
460 "ui",
461 Priority::Normal,
462 Flows {
463 input: vec![flow(
464 "in",
465 wit("brahman:dht", Some("v2"), "entity-result"),
466 None,
467 )],
468 output: vec![],
469 },
470 );
471 let s_prod = Ulid::new();
472 let s_cons = Ulid::new();
473 b.register(s_prod, &producer, None);
474 b.register(s_cons, &consumer, None);
475
476 let m = b.find_producer_for(s_cons, "in").expect("match");
477 assert_eq!(m.via, MatchStrategy::Structural);
478 }
479
480 #[test]
481 fn exact_strategy_rejects_interface_mismatch() {
482 let mut b = Broker::new(BrokerConfig {
483 strategy: MatchStrategy::Exact,
484 current_context: None,
485 });
486 let producer = card(
487 "dht",
488 Priority::Normal,
489 Flows {
490 input: vec![],
491 output: vec![flow(
492 "out",
493 wit("brahman:dht", Some("v1"), "entity-result"),
494 None,
495 )],
496 },
497 );
498 let consumer = card(
499 "ui",
500 Priority::Normal,
501 Flows {
502 input: vec![flow(
503 "in",
504 wit("brahman:dht", Some("v2"), "entity-result"),
505 None,
506 )],
507 output: vec![],
508 },
509 );
510 b.register(Ulid::new(), &producer, None);
511 let s_cons = Ulid::new();
512 b.register(s_cons, &consumer, None);
513
514 assert!(b.find_producer_for(s_cons, "in").is_none());
515 }
516
517 #[test]
518 fn exact_then_structural_prefers_exact() {
519 let mut b = Broker::new(BrokerConfig {
520 strategy: MatchStrategy::ExactThenStructural,
521 current_context: None,
522 });
523 // Productor 1: match estructural (interface diferente)
524 let p_struct = card(
525 "dht-cache",
526 Priority::Normal,
527 Flows {
528 input: vec![],
529 output: vec![flow(
530 "out",
531 wit("brahman:dht", Some("v2"), "entity-result"),
532 None,
533 )],
534 },
535 );
536 // Productor 2: match exact (interface igual)
537 let p_exact = card(
538 "dht",
539 Priority::Normal,
540 Flows {
541 input: vec![],
542 output: vec![flow(
543 "out",
544 wit("brahman:dht", Some("v1"), "entity-result"),
545 None,
546 )],
547 },
548 );
549 let consumer = card(
550 "ui",
551 Priority::Normal,
552 Flows {
553 input: vec![flow(
554 "in",
555 wit("brahman:dht", Some("v1"), "entity-result"),
556 None,
557 )],
558 output: vec![],
559 },
560 );
561 b.register(Ulid::new(), &p_struct, None);
562 b.register(Ulid::new(), &p_exact, None);
563 let s_cons = Ulid::new();
564 b.register(s_cons, &consumer, None);
565
566 let m = b.find_producer_for(s_cons, "in").expect("match");
567 // El exact gana incluso si tiene priority igual: por estrategia.
568 assert_eq!(m.producer_label, "dht");
569 assert_eq!(m.via, MatchStrategy::Exact);
570 }
571
572 #[test]
573 fn pin_to_overrides_type_search() {
574 let mut b = Broker::new(BrokerConfig::default());
575 // Dos productores que producen el mismo tipo.
576 let p1 = card(
577 "dht-prod",
578 Priority::Normal,
579 Flows {
580 input: vec![],
581 output: vec![flow("out", prim("string"), None)],
582 },
583 );
584 let p2 = card(
585 "dht-test",
586 Priority::Normal,
587 Flows {
588 input: vec![],
589 output: vec![flow("out", prim("string"), None)],
590 },
591 );
592 let consumer = card(
593 "ui",
594 Priority::Normal,
595 Flows {
596 input: vec![flow("in", prim("string"), Some("dht-test"))],
597 output: vec![],
598 },
599 );
600 b.register(Ulid::new(), &p1, None);
601 b.register(Ulid::new(), &p2, None);
602 let s_cons = Ulid::new();
603 b.register(s_cons, &consumer, None);
604
605 let m = b.find_producer_for(s_cons, "in").expect("match");
606 assert_eq!(m.producer_label, "dht-test");
607 assert!(m.pinned);
608 }
609
610 #[test]
611 fn pin_to_unresolvable_falls_back_to_type_match() {
612 let mut b = Broker::new(BrokerConfig::default());
613 let p = card(
614 "real-dht",
615 Priority::Normal,
616 Flows {
617 input: vec![],
618 output: vec![flow("out", prim("string"), None)],
619 },
620 );
621 let consumer = card(
622 "ui",
623 Priority::Normal,
624 Flows {
625 input: vec![flow("in", prim("string"), Some("nonexistent"))],
626 output: vec![],
627 },
628 );
629 b.register(Ulid::new(), &p, None);
630 let s_cons = Ulid::new();
631 b.register(s_cons, &consumer, None);
632
633 let m = b.find_producer_for(s_cons, "in").expect("match");
634 assert_eq!(m.producer_label, "real-dht");
635 assert!(!m.pinned);
636 }
637
638 #[test]
639 fn priority_breaks_ties() {
640 let mut b = Broker::new(BrokerConfig::default());
641 let p_low = card(
642 "z-dht",
643 Priority::Low,
644 Flows {
645 input: vec![],
646 output: vec![flow("out", prim("string"), None)],
647 },
648 );
649 let p_high = card(
650 "a-dht",
651 Priority::High,
652 Flows {
653 input: vec![],
654 output: vec![flow("out", prim("string"), None)],
655 },
656 );
657 let consumer = card(
658 "ui",
659 Priority::Normal,
660 Flows {
661 input: vec![flow("in", prim("string"), None)],
662 output: vec![],
663 },
664 );
665 b.register(Ulid::new(), &p_low, None);
666 b.register(Ulid::new(), &p_high, None);
667 let s_cons = Ulid::new();
668 b.register(s_cons, &consumer, None);
669
670 let m = b.find_producer_for(s_cons, "in").expect("match");
671 assert_eq!(m.producer_label, "a-dht"); // priority High > Low
672 }
673
674 #[test]
675 fn label_alpha_breaks_priority_ties() {
676 let mut b = Broker::new(BrokerConfig::default());
677 let p1 = card(
678 "z-dht",
679 Priority::Normal,
680 Flows {
681 input: vec![],
682 output: vec![flow("out", prim("string"), None)],
683 },
684 );
685 let p2 = card(
686 "a-dht",
687 Priority::Normal,
688 Flows {
689 input: vec![],
690 output: vec![flow("out", prim("string"), None)],
691 },
692 );
693 let consumer = card(
694 "ui",
695 Priority::Normal,
696 Flows {
697 input: vec![flow("in", prim("string"), None)],
698 output: vec![],
699 },
700 );
701 b.register(Ulid::new(), &p1, None);
702 b.register(Ulid::new(), &p2, None);
703 let s_cons = Ulid::new();
704 b.register(s_cons, &consumer, None);
705
706 let m = b.find_producer_for(s_cons, "in").expect("match");
707 assert_eq!(m.producer_label, "a-dht"); // alfabético gana
708 }
709
710 #[test]
711 fn unregister_removes_producer() {
712 let mut b = Broker::new(BrokerConfig::default());
713 let p = card(
714 "dht",
715 Priority::Normal,
716 Flows {
717 input: vec![],
718 output: vec![flow("out", prim("string"), None)],
719 },
720 );
721 let consumer = card(
722 "ui",
723 Priority::Normal,
724 Flows {
725 input: vec![flow("in", prim("string"), None)],
726 output: vec![],
727 },
728 );
729 let s_p = Ulid::new();
730 b.register(s_p, &p, None);
731 let s_c = Ulid::new();
732 b.register(s_c, &consumer, None);
733
734 assert!(b.find_producer_for(s_c, "in").is_some());
735 b.unregister(s_p);
736 assert!(b.find_producer_for(s_c, "in").is_none());
737 }
738
739 #[test]
740 fn no_self_loops() {
741 let mut b = Broker::new(BrokerConfig::default());
742 let same = card(
743 "echo",
744 Priority::Normal,
745 Flows {
746 input: vec![flow("in", prim("string"), None)],
747 output: vec![flow("out", prim("string"), None)],
748 },
749 );
750 let s = Ulid::new();
751 b.register(s, &same, None);
752
753 // Solo una Card registrada — no hay otra que produzca string.
754 assert!(b.find_producer_for(s, "in").is_none());
755 }
756
757 #[test]
758 fn all_matches_lists_pairs() {
759 let mut b = Broker::new(BrokerConfig::default());
760 let dht = card(
761 "dht",
762 Priority::Normal,
763 Flows {
764 input: vec![flow("query", prim("string"), None)],
765 output: vec![flow("results", prim("bytes"), None)],
766 },
767 );
768 let ui = card(
769 "ui",
770 Priority::Normal,
771 Flows {
772 input: vec![flow("data", prim("bytes"), None)],
773 output: vec![flow("user-input", prim("string"), None)],
774 },
775 );
776 b.register(Ulid::new(), &dht, None);
777 b.register(Ulid::new(), &ui, None);
778
779 let matches = b.all_matches();
780 assert_eq!(matches.len(), 2);
781 // dht.query ← ui.user-input y ui.data ← dht.results
782 let pairs: Vec<_> = matches
783 .iter()
784 .map(|m| (m.consumer_label.as_str(), m.producer_label.as_str()))
785 .collect();
786 assert!(pairs.contains(&("dht", "ui")));
787 assert!(pairs.contains(&("ui", "dht")));
788 }
789
790 // ===========================================================
791 // Priority contexts
792 // ===========================================================
793
794 #[test]
795 fn context_priority_offset_lifts_producer_above_alphabetic_winner() {
796 // Sin contexto, "a-prod" gana contra "b-prod" (alfabético).
797 // En contexto "test", b-prod tiene offset +1 → debería ganar.
798 let mut a_prod = card(
799 "a-prod",
800 Priority::Normal,
801 Flows {
802 input: vec![],
803 output: vec![flow("out", prim("string"), None)],
804 },
805 );
806 a_prod.priority_contexts = std::collections::BTreeMap::new(); // explícito vacío
807
808 let mut b_prod = card(
809 "b-prod",
810 Priority::Normal,
811 Flows {
812 input: vec![],
813 output: vec![flow("out", prim("string"), None)],
814 },
815 );
816 b_prod.priority_contexts.insert(
817 "test".into(),
818 ContextBias {
819 pin_to: None,
820 priority_offset: 1,
821 },
822 );
823
824 let consumer = card(
825 "ui",
826 Priority::Normal,
827 Flows {
828 input: vec![flow("in", prim("string"), None)],
829 output: vec![],
830 },
831 );
832
833 let s_cons = Ulid::new();
834
835 // Caso 1: sin contexto → a-prod gana (alfabético).
836 let mut b = Broker::new(BrokerConfig {
837 strategy: MatchStrategy::default(),
838 current_context: None,
839 });
840 b.register(Ulid::new(), &a_prod, None);
841 b.register(Ulid::new(), &b_prod, None);
842 b.register(s_cons, &consumer, None);
843 let m = b.find_producer_for(s_cons, "in").unwrap();
844 assert_eq!(m.producer_label, "a-prod");
845
846 // Caso 2: contexto "test" → b-prod gana por offset +1.
847 let mut b = Broker::new(BrokerConfig {
848 strategy: MatchStrategy::default(),
849 current_context: Some("test".into()),
850 });
851 b.register(Ulid::new(), &a_prod, None);
852 b.register(Ulid::new(), &b_prod, None);
853 b.register(s_cons, &consumer, None);
854 let m = b.find_producer_for(s_cons, "in").unwrap();
855 assert_eq!(m.producer_label, "b-prod");
856 }
857
858 #[test]
859 fn context_pin_to_overrides_static_pin() {
860 // Consumer pinea estático a "real-dht", pero en contexto "test"
861 // declara override a "mock-dht".
862 let real = card(
863 "real-dht",
864 Priority::Normal,
865 Flows {
866 input: vec![],
867 output: vec![flow("out", prim("string"), None)],
868 },
869 );
870 let mock = card(
871 "mock-dht",
872 Priority::Normal,
873 Flows {
874 input: vec![],
875 output: vec![flow("out", prim("string"), None)],
876 },
877 );
878 let mut consumer = card(
879 "ui",
880 Priority::Normal,
881 Flows {
882 input: vec![flow("in", prim("string"), Some("real-dht"))],
883 output: vec![],
884 },
885 );
886 consumer.priority_contexts.insert(
887 "test".into(),
888 ContextBias {
889 pin_to: Some("mock-dht".into()),
890 priority_offset: 0,
891 },
892 );
893
894 let s_cons = Ulid::new();
895
896 // Caso 1: sin contexto → static pin gana ("real-dht").
897 let mut b = Broker::new(BrokerConfig::default());
898 b.register(Ulid::new(), &real, None);
899 b.register(Ulid::new(), &mock, None);
900 b.register(s_cons, &consumer, None);
901 let m = b.find_producer_for(s_cons, "in").unwrap();
902 assert_eq!(m.producer_label, "real-dht");
903 assert!(m.pinned);
904
905 // Caso 2: contexto "test" → context override gana ("mock-dht").
906 let mut b = Broker::new(BrokerConfig {
907 strategy: MatchStrategy::default(),
908 current_context: Some("test".into()),
909 });
910 b.register(Ulid::new(), &real, None);
911 b.register(Ulid::new(), &mock, None);
912 b.register(s_cons, &consumer, None);
913 let m = b.find_producer_for(s_cons, "in").unwrap();
914 assert_eq!(m.producer_label, "mock-dht");
915 assert!(m.pinned);
916 }
917
918 #[test]
919 fn unknown_context_no_op() {
920 // Si la Card declara biases para "test" pero el broker está en
921 // "prod", los biases no aplican.
922 let mut b_prod = card(
923 "b-prod",
924 Priority::Normal,
925 Flows {
926 input: vec![],
927 output: vec![flow("out", prim("string"), None)],
928 },
929 );
930 b_prod.priority_contexts.insert(
931 "test".into(),
932 ContextBias {
933 pin_to: None,
934 priority_offset: 5,
935 },
936 );
937 let a_prod = card(
938 "a-prod",
939 Priority::Normal,
940 Flows {
941 input: vec![],
942 output: vec![flow("out", prim("string"), None)],
943 },
944 );
945 let consumer = card(
946 "ui",
947 Priority::Normal,
948 Flows {
949 input: vec![flow("in", prim("string"), None)],
950 output: vec![],
951 },
952 );
953
954 let mut b = Broker::new(BrokerConfig {
955 strategy: MatchStrategy::default(),
956 current_context: Some("prod".into()),
957 });
958 let s_cons = Ulid::new();
959 b.register(Ulid::new(), &a_prod, None);
960 b.register(Ulid::new(), &b_prod, None);
961 b.register(s_cons, &consumer, None);
962
963 // En contexto "prod" sin biases declarados, gana por alfabético.
964 let m = b.find_producer_for(s_cons, "in").unwrap();
965 assert_eq!(m.producer_label, "a-prod");
966 }
967
968 #[test]
969 fn priority_offset_clamps_to_critical() {
970 // Offset enorme no debe hacer overflow ni saltar fuera del rango.
971 let mut prod = card(
972 "p",
973 Priority::Normal,
974 Flows {
975 input: vec![],
976 output: vec![flow("out", prim("string"), None)],
977 },
978 );
979 prod.priority_contexts.insert(
980 "x".into(),
981 ContextBias {
982 pin_to: None,
983 priority_offset: 100,
984 },
985 );
986
987 let b = Broker::new(BrokerConfig {
988 strategy: MatchStrategy::default(),
989 current_context: Some("x".into()),
990 });
991 let bc = BrokeredCard::from_card(Ulid::new(), &prod, None);
992 // effective_priority debe estar clampada a 3 (Critical), no 101.
993 assert_eq!(b.effective_priority(&bc), 3);
994 }
995}