1pub mod route_table;
39
40use std::collections::HashMap;
41use std::fmt;
42use std::sync::Arc;
43
44use async_trait::async_trait;
45use meerkat_machine_schema::identity::{
46 CompositionId, EffectVariantId, FieldId, InputVariantId, MachineId, MachineInstanceId, RouteId,
47 SignalVariantId,
48};
49use thiserror::Error;
50
51pub use route_table::{RouteTable, RouteTableError, RoutedInputDescriptor, RoutedSignalDescriptor};
52
53#[derive(Debug, Clone, PartialEq, Eq, Hash)]
58pub struct ProducerInstance {
59 pub composition: CompositionId,
61 pub instance_id: MachineInstanceId,
63 pub machine: MachineId,
65}
66
67#[derive(Debug, Clone, PartialEq, Eq)]
75pub enum EffectPayload<E> {
76 Emitted {
78 variant: EffectVariantId,
80 body: E,
82 },
83}
84
85impl<E: ProducerEffect> EffectPayload<E> {
86 pub fn variant(&self) -> &EffectVariantId {
88 match self {
89 Self::Emitted { variant, .. } => variant,
90 }
91 }
92
93 pub fn body(&self) -> &E {
95 match self {
96 Self::Emitted { body, .. } => body,
97 }
98 }
99}
100
101#[derive(Debug, Clone, PartialEq, Eq)]
104pub enum SignalPayload<S> {
105 Emitted {
107 variant: EffectVariantId,
111 body: S,
113 },
114}
115
116impl<S: ProducerSignal> SignalPayload<S> {
117 pub fn variant(&self) -> &EffectVariantId {
119 match self {
120 Self::Emitted { variant, .. } => variant,
121 }
122 }
123
124 pub fn body(&self) -> &S {
126 match self {
127 Self::Emitted { body, .. } => body,
128 }
129 }
130}
131
132#[derive(Debug, Clone, PartialEq, Eq, Hash)]
134pub struct RouteKey {
135 pub composition: CompositionId,
136 pub route_id: RouteId,
137}
138
139pub trait ProducerEffect: fmt::Debug + Send + Sync + 'static {
144 fn variant_id(&self) -> EffectVariantId;
151
152 fn field(&self, id: &FieldId) -> Option<FieldValue<'_>>;
160}
161
162pub trait ProducerSignal: fmt::Debug + Send + Sync + 'static {
172 fn variant_id(&self) -> EffectVariantId;
174
175 fn field(&self, id: &FieldId) -> Option<FieldValue<'_>>;
177}
178
179#[derive(Debug, Clone)]
188pub enum FieldValue<'a> {
189 Str(&'a str),
191 U64(u64),
193 I64(i64),
195 Bool(bool),
197 Opaque(Arc<dyn std::any::Any + Send + Sync>),
202}
203
204#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct DispatchOutcome {
207 pub route: RouteKey,
209 pub consumer: MachineInstanceId,
211 pub applied_input: InputVariantId,
213}
214
215#[derive(Debug, Clone, PartialEq, Eq)]
217pub struct SignalDispatchOutcome {
218 pub route: RouteKey,
220 pub consumer: MachineInstanceId,
222 pub applied_signal: SignalVariantId,
224}
225
226#[derive(Debug, Clone, PartialEq, Eq, Error)]
232pub enum DispatchRefusal {
233 #[error("dispatcher composition {expected} does not match producer composition {actual}")]
235 CompositionMismatch {
236 expected: CompositionId,
237 actual: CompositionId,
238 },
239 #[error(
241 "no input route declared for producer {instance} effect variant {variant} in composition {composition}"
242 )]
243 UnresolvedRoute {
244 composition: CompositionId,
245 instance: MachineInstanceId,
246 variant: EffectVariantId,
247 },
248 #[error("route {route} requires producer field {field} on variant {variant}, not provided")]
251 MissingProducerField {
252 route: RouteId,
253 variant: EffectVariantId,
254 field: FieldId,
255 },
256 #[error(
260 "no consumer surface registered for target instance {instance} in composition {composition}"
261 )]
262 UnwiredConsumer {
263 composition: CompositionId,
264 instance: MachineInstanceId,
265 },
266 #[error("consumer {instance} refused input {variant}: {error}")]
273 ConsumerRefused {
274 instance: MachineInstanceId,
275 variant: InputVariantId,
276 error: ConsumerError,
277 },
278}
279
280#[derive(Debug, Clone, PartialEq, Eq, Error)]
282pub enum SignalDispatchRefusal {
283 #[error("dispatcher composition {expected} does not match producer composition {actual}")]
285 CompositionMismatch {
286 expected: CompositionId,
287 actual: CompositionId,
288 },
289 #[error(
291 "no signal route declared for producer {instance} variant {variant} in composition {composition}"
292 )]
293 UnresolvedRoute {
294 composition: CompositionId,
295 instance: MachineInstanceId,
296 variant: EffectVariantId,
297 },
298 #[error("route {route} requires producer field {field} on variant {variant}, not provided")]
301 MissingProducerField {
302 route: RouteId,
303 variant: EffectVariantId,
304 field: FieldId,
305 },
306 #[error(
309 "no signal consumer surface registered for target instance {instance} in composition {composition}"
310 )]
311 UnwiredConsumer {
312 composition: CompositionId,
313 instance: MachineInstanceId,
314 },
315 #[error("consumer {instance} refused signal {variant}: {error}")]
319 ConsumerRefused {
320 instance: MachineInstanceId,
321 variant: SignalVariantId,
322 error: ConsumerError,
323 },
324}
325
326#[derive(Debug, Clone, PartialEq, Eq, Error)]
337#[error("{message} [{error_code}]")]
338pub struct ConsumerError {
339 error_code: &'static str,
343 message: String,
345}
346
347impl ConsumerError {
348 pub fn new(error_code: &'static str, message: impl Into<String>) -> Self {
351 Self {
352 error_code,
353 message: message.into(),
354 }
355 }
356
357 pub fn error_code(&self) -> &'static str {
359 self.error_code
360 }
361
362 pub fn message(&self) -> &str {
364 &self.message
365 }
366}
367
368impl From<String> for ConsumerError {
369 fn from(message: String) -> Self {
375 Self::new("consumer_projection_failed", message)
376 }
377}
378
379#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
388#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
389pub trait ConsumerSurface: Send + Sync {
390 fn instance_id(&self) -> &MachineInstanceId;
393
394 async fn apply_routed_input(
399 &self,
400 variant: InputVariantId,
401 projected_fields: Vec<(FieldId, OwnedFieldValue)>,
402 ) -> Result<(), ConsumerError>;
403}
404
405#[async_trait]
407pub trait SignalConsumerSurface: Send + Sync {
408 fn instance_id(&self) -> &MachineInstanceId;
410
411 async fn receive_signal(
413 &self,
414 variant: SignalVariantId,
415 projected_fields: Vec<(FieldId, OwnedFieldValue)>,
416 ) -> Result<(), ConsumerError>;
417}
418
419#[derive(Debug, Clone)]
424pub enum OwnedFieldValue {
425 Str(String),
426 U64(u64),
427 I64(i64),
428 Bool(bool),
429 Opaque(Arc<dyn std::any::Any + Send + Sync>),
430}
431
432impl FieldValue<'_> {
433 pub fn to_owned_value(&self) -> OwnedFieldValue {
437 match self {
438 FieldValue::Str(s) => OwnedFieldValue::Str((*s).to_owned()),
439 FieldValue::U64(v) => OwnedFieldValue::U64(*v),
440 FieldValue::I64(v) => OwnedFieldValue::I64(*v),
441 FieldValue::Bool(v) => OwnedFieldValue::Bool(*v),
442 FieldValue::Opaque(handle) => OwnedFieldValue::Opaque(Arc::clone(handle)),
443 }
444 }
445}
446
447#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
455#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
456pub trait CompositionDispatcher: Send + Sync {
457 type Effect: ProducerEffect;
460
461 fn composition(&self) -> &CompositionId;
464
465 async fn dispatch(
468 &self,
469 producer: ProducerInstance,
470 effect: EffectPayload<Self::Effect>,
471 ) -> Result<DispatchOutcome, DispatchRefusal>;
472}
473
474#[async_trait]
476pub trait CompositionSignalDispatcher: Send + Sync {
477 type Signal: ProducerSignal;
479
480 fn composition(&self) -> &CompositionId;
482
483 async fn dispatch_signal(
486 &self,
487 producer: ProducerInstance,
488 signal: SignalPayload<Self::Signal>,
489 ) -> Result<SignalDispatchOutcome, SignalDispatchRefusal>;
490}
491
492pub trait ContextProvider<E: ProducerEffect>: Send + Sync {
516 fn provide_context(
529 &self,
530 producer: &ProducerInstance,
531 effect: &EffectPayload<E>,
532 ) -> Vec<(FieldId, OwnedFieldValue)>;
533}
534
535pub enum CompositionBinding<E: ProducerEffect> {
559 Standalone,
562 Wired(Arc<dyn CompositionDispatcher<Effect = E>>),
566 OwnerProvided {
572 dispatcher: Arc<dyn CompositionDispatcher<Effect = E>>,
573 context: Arc<dyn ContextProvider<E>>,
574 },
575}
576
577impl<E: ProducerEffect> fmt::Debug for CompositionBinding<E> {
578 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
579 match self {
580 Self::Standalone => f.debug_struct("CompositionBinding::Standalone").finish(),
581 Self::Wired(_) => f
582 .debug_struct("CompositionBinding::Wired")
583 .field("dispatcher", &"<dyn CompositionDispatcher>")
584 .finish(),
585 Self::OwnerProvided { .. } => f
586 .debug_struct("CompositionBinding::OwnerProvided")
587 .field("dispatcher", &"<dyn CompositionDispatcher>")
588 .field("context", &"<dyn ContextProvider>")
589 .finish(),
590 }
591 }
592}
593
594impl<E: ProducerEffect> CompositionBinding<E> {
595 pub fn standalone() -> Self {
602 Self::Standalone
603 }
604
605 pub fn wired_with(dispatcher: Arc<dyn CompositionDispatcher<Effect = E>>) -> Self {
611 Self::Wired(dispatcher)
612 }
613
614 pub fn owner_provided(
622 dispatcher: Arc<dyn CompositionDispatcher<Effect = E>>,
623 context: Arc<dyn ContextProvider<E>>,
624 ) -> Self {
625 Self::OwnerProvided {
626 dispatcher,
627 context,
628 }
629 }
630
631 pub fn is_standalone(&self) -> bool {
633 matches!(self, Self::Standalone)
634 }
635
636 pub fn wired(&self) -> Option<&Arc<dyn CompositionDispatcher<Effect = E>>> {
645 match self {
646 Self::Standalone => None,
647 Self::Wired(d) => Some(d),
648 Self::OwnerProvided { dispatcher, .. } => Some(dispatcher),
649 }
650 }
651
652 pub fn context_provider(&self) -> Option<&Arc<dyn ContextProvider<E>>> {
661 match self {
662 Self::Standalone | Self::Wired(_) => None,
663 Self::OwnerProvided { context, .. } => Some(context),
664 }
665 }
666}
667
668pub struct CatalogCompositionDispatcher<E: ProducerEffect> {
683 composition: CompositionId,
684 table: RouteTable,
685 consumers: HashMap<MachineInstanceId, Arc<dyn ConsumerSurface>>,
686 _effect: std::marker::PhantomData<fn(E)>,
687}
688
689pub struct CatalogCompositionSignalDispatcher<S: ProducerSignal> {
695 composition: CompositionId,
696 table: RouteTable,
697 consumers: HashMap<MachineInstanceId, Arc<dyn SignalConsumerSurface>>,
698 _signal: std::marker::PhantomData<fn(S)>,
699}
700
701impl<S: ProducerSignal> fmt::Debug for CatalogCompositionSignalDispatcher<S> {
702 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
703 f.debug_struct("CatalogCompositionSignalDispatcher")
704 .field("composition", &self.composition)
705 .field("signal_routes", &self.table.signal_route_count())
706 .field("consumers", &self.consumers.len())
707 .finish()
708 }
709}
710
711impl<E: ProducerEffect> fmt::Debug for CatalogCompositionDispatcher<E> {
712 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
713 f.debug_struct("CatalogCompositionDispatcher")
714 .field("composition", &self.composition)
715 .field("routes", &self.table.len())
716 .field("consumers", &self.consumers.len())
717 .finish()
718 }
719}
720
721impl<E: ProducerEffect> CatalogCompositionDispatcher<E> {
722 pub fn new(composition: CompositionId, table: RouteTable) -> Self {
725 Self {
726 composition,
727 table,
728 consumers: HashMap::new(),
729 _effect: std::marker::PhantomData,
730 }
731 }
732
733 pub fn with_consumer(mut self, surface: Arc<dyn ConsumerSurface>) -> Self {
740 self.consumers
741 .insert(surface.instance_id().clone(), surface);
742 self
743 }
744}
745
746impl<S: ProducerSignal> CatalogCompositionSignalDispatcher<S> {
747 pub fn new(composition: CompositionId, table: RouteTable) -> Self {
749 Self {
750 composition,
751 table,
752 consumers: HashMap::new(),
753 _signal: std::marker::PhantomData,
754 }
755 }
756
757 pub fn with_consumer(mut self, surface: Arc<dyn SignalConsumerSurface>) -> Self {
759 self.consumers
760 .insert(surface.instance_id().clone(), surface);
761 self
762 }
763}
764
765#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
766#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
767impl<E: ProducerEffect> CompositionDispatcher for CatalogCompositionDispatcher<E> {
768 type Effect = E;
769
770 fn composition(&self) -> &CompositionId {
771 &self.composition
772 }
773
774 async fn dispatch(
775 &self,
776 producer: ProducerInstance,
777 effect: EffectPayload<Self::Effect>,
778 ) -> Result<DispatchOutcome, DispatchRefusal> {
779 if producer.composition != self.composition {
780 return Err(DispatchRefusal::CompositionMismatch {
781 expected: self.composition.clone(),
782 actual: producer.composition,
783 });
784 }
785
786 let variant = effect.variant().clone();
787 let body = effect.body();
788
789 let descriptor = self
790 .table
791 .resolve(&producer.instance_id, &variant)
792 .ok_or_else(|| DispatchRefusal::UnresolvedRoute {
793 composition: self.composition.clone(),
794 instance: producer.instance_id.clone(),
795 variant: variant.clone(),
796 })?;
797
798 let mut projected: Vec<(FieldId, OwnedFieldValue)> =
799 Vec::with_capacity(descriptor.bindings.len());
800 for (from_field, to_field) in &descriptor.bindings {
801 let value =
802 body.field(from_field)
803 .ok_or_else(|| DispatchRefusal::MissingProducerField {
804 route: descriptor.route_id.clone(),
805 variant: variant.clone(),
806 field: from_field.clone(),
807 })?;
808 projected.push((to_field.clone(), value.to_owned_value()));
809 }
810
811 let consumer = self.consumers.get(&descriptor.instance_id).ok_or_else(|| {
812 DispatchRefusal::UnwiredConsumer {
813 composition: self.composition.clone(),
814 instance: descriptor.instance_id.clone(),
815 }
816 })?;
817
818 consumer
819 .apply_routed_input(descriptor.input_variant.clone(), projected)
820 .await
821 .map_err(|error| DispatchRefusal::ConsumerRefused {
822 instance: descriptor.instance_id.clone(),
823 variant: descriptor.input_variant.clone(),
824 error,
825 })?;
826
827 Ok(DispatchOutcome {
828 route: RouteKey {
829 composition: self.composition.clone(),
830 route_id: descriptor.route_id.clone(),
831 },
832 consumer: descriptor.instance_id.clone(),
833 applied_input: descriptor.input_variant.clone(),
834 })
835 }
836}
837
838#[async_trait]
839impl<S: ProducerSignal> CompositionSignalDispatcher for CatalogCompositionSignalDispatcher<S> {
840 type Signal = S;
841
842 fn composition(&self) -> &CompositionId {
843 &self.composition
844 }
845
846 async fn dispatch_signal(
847 &self,
848 producer: ProducerInstance,
849 signal: SignalPayload<Self::Signal>,
850 ) -> Result<SignalDispatchOutcome, SignalDispatchRefusal> {
851 if producer.composition != self.composition {
852 return Err(SignalDispatchRefusal::CompositionMismatch {
853 expected: self.composition.clone(),
854 actual: producer.composition,
855 });
856 }
857
858 let variant = signal.variant().clone();
859 let body = signal.body();
860
861 let descriptor = self
862 .table
863 .resolve_signal(&producer.instance_id, &variant)
864 .ok_or_else(|| SignalDispatchRefusal::UnresolvedRoute {
865 composition: self.composition.clone(),
866 instance: producer.instance_id.clone(),
867 variant: variant.clone(),
868 })?;
869
870 let mut projected: Vec<(FieldId, OwnedFieldValue)> =
871 Vec::with_capacity(descriptor.bindings.len());
872 for (from_field, to_field) in &descriptor.bindings {
873 let value = body.field(from_field).ok_or_else(|| {
874 SignalDispatchRefusal::MissingProducerField {
875 route: descriptor.route_id.clone(),
876 variant: variant.clone(),
877 field: from_field.clone(),
878 }
879 })?;
880 projected.push((to_field.clone(), value.to_owned_value()));
881 }
882
883 let consumer = self.consumers.get(&descriptor.instance_id).ok_or_else(|| {
884 SignalDispatchRefusal::UnwiredConsumer {
885 composition: self.composition.clone(),
886 instance: descriptor.instance_id.clone(),
887 }
888 })?;
889
890 consumer
891 .receive_signal(descriptor.signal_variant.clone(), projected)
892 .await
893 .map_err(|error| SignalDispatchRefusal::ConsumerRefused {
894 instance: descriptor.instance_id.clone(),
895 variant: descriptor.signal_variant.clone(),
896 error,
897 })?;
898
899 Ok(SignalDispatchOutcome {
900 route: RouteKey {
901 composition: self.composition.clone(),
902 route_id: descriptor.route_id.clone(),
903 },
904 consumer: descriptor.instance_id.clone(),
905 applied_signal: descriptor.signal_variant.clone(),
906 })
907 }
908}
909
910#[cfg(test)]
911mod tests {
912 use super::*;
913 use meerkat_machine_schema::catalog::meerkat_mob_seam_composition;
914
915 #[derive(Debug, Clone, PartialEq, Eq)]
921 enum SeamEffect {
922 Mob(MobEffect),
923 }
924
925 #[derive(Debug, Clone, PartialEq, Eq)]
926 enum MobEffect {
927 RequestRuntimeBinding {
928 agent_runtime_id: String,
929 fence_token: u64,
930 generation: u64,
931 session_id: String,
932 },
933 }
934
935 impl ProducerEffect for SeamEffect {
936 fn variant_id(&self) -> EffectVariantId {
937 match self {
938 Self::Mob(MobEffect::RequestRuntimeBinding { .. }) => {
939 EffectVariantId::parse("RequestRuntimeBinding").expect("slug")
940 }
941 }
942 }
943
944 fn field(&self, id: &FieldId) -> Option<FieldValue<'_>> {
945 match self {
946 Self::Mob(MobEffect::RequestRuntimeBinding {
947 agent_runtime_id,
948 fence_token,
949 generation,
950 session_id,
951 }) => match id.as_str() {
952 "agent_runtime_id" => Some(FieldValue::Str(agent_runtime_id)),
953 "fence_token" => Some(FieldValue::U64(*fence_token)),
954 "generation" => Some(FieldValue::U64(*generation)),
955 "session_id" => Some(FieldValue::Str(session_id)),
956 _ => None,
957 },
958 }
959 }
960 }
961
962 #[allow(clippy::enum_variant_names)]
966 #[derive(Debug, Clone, PartialEq, Eq)]
967 enum SeamSignal {
968 RuntimeBound {
969 agent_runtime_id: String,
970 fence_token: u64,
971 },
972 RuntimeRetired {
973 agent_runtime_id: String,
974 fence_token: u64,
975 },
976 RuntimeDestroyed {
977 agent_runtime_id: String,
978 fence_token: u64,
979 },
980 }
981
982 impl ProducerSignal for SeamSignal {
983 fn variant_id(&self) -> EffectVariantId {
984 let slug = match self {
985 Self::RuntimeBound { .. } => "RuntimeBound",
986 Self::RuntimeRetired { .. } => "RuntimeRetired",
987 Self::RuntimeDestroyed { .. } => "RuntimeDestroyed",
988 };
989 EffectVariantId::parse(slug).expect("signal source slug")
990 }
991
992 fn field(&self, id: &FieldId) -> Option<FieldValue<'_>> {
993 let (agent_runtime_id, fence_token) = match self {
994 Self::RuntimeBound {
995 agent_runtime_id,
996 fence_token,
997 }
998 | Self::RuntimeRetired {
999 agent_runtime_id,
1000 fence_token,
1001 }
1002 | Self::RuntimeDestroyed {
1003 agent_runtime_id,
1004 fence_token,
1005 } => (agent_runtime_id, fence_token),
1006 };
1007 match id.as_str() {
1008 "agent_runtime_id" => Some(FieldValue::Str(agent_runtime_id)),
1009 "fence_token" => Some(FieldValue::U64(*fence_token)),
1010 _ => None,
1011 }
1012 }
1013 }
1014
1015 #[derive(Default)]
1016 struct RecordingMeerkatSurface {
1017 log: tokio::sync::Mutex<Vec<(InputVariantId, Vec<(FieldId, OwnedFieldValue)>)>>,
1018 }
1019
1020 #[async_trait]
1021 impl ConsumerSurface for RecordingMeerkatSurface {
1022 fn instance_id(&self) -> &MachineInstanceId {
1023 static ID: std::sync::OnceLock<MachineInstanceId> = std::sync::OnceLock::new();
1026 ID.get_or_init(|| MachineInstanceId::parse("meerkat").unwrap())
1027 }
1028
1029 async fn apply_routed_input(
1030 &self,
1031 variant: InputVariantId,
1032 projected_fields: Vec<(FieldId, OwnedFieldValue)>,
1033 ) -> Result<(), ConsumerError> {
1034 self.log.lock().await.push((variant, projected_fields));
1035 Ok(())
1036 }
1037 }
1038
1039 #[derive(Default)]
1040 struct RecordingMobSignalSurface {
1041 log: tokio::sync::Mutex<Vec<(SignalVariantId, Vec<(FieldId, OwnedFieldValue)>)>>,
1042 }
1043
1044 #[async_trait]
1045 impl SignalConsumerSurface for RecordingMobSignalSurface {
1046 fn instance_id(&self) -> &MachineInstanceId {
1047 static ID: std::sync::OnceLock<MachineInstanceId> = std::sync::OnceLock::new();
1048 ID.get_or_init(|| MachineInstanceId::parse("mob").unwrap())
1049 }
1050
1051 async fn receive_signal(
1052 &self,
1053 variant: SignalVariantId,
1054 projected_fields: Vec<(FieldId, OwnedFieldValue)>,
1055 ) -> Result<(), ConsumerError> {
1056 self.log.lock().await.push((variant, projected_fields));
1057 Ok(())
1058 }
1059 }
1060
1061 fn mob_producer() -> ProducerInstance {
1062 ProducerInstance {
1063 composition: CompositionId::parse("meerkat_mob_seam").unwrap(),
1064 instance_id: MachineInstanceId::parse("mob").unwrap(),
1065 machine: MachineId::parse("MobMachine").unwrap(),
1066 }
1067 }
1068
1069 fn meerkat_producer() -> ProducerInstance {
1070 ProducerInstance {
1071 composition: CompositionId::parse("meerkat_mob_seam").unwrap(),
1072 instance_id: MachineInstanceId::parse("meerkat").unwrap(),
1073 machine: MachineId::parse("MeerkatMachine").unwrap(),
1074 }
1075 }
1076
1077 fn sample_effect() -> EffectPayload<SeamEffect> {
1078 EffectPayload::Emitted {
1079 variant: EffectVariantId::parse("RequestRuntimeBinding").unwrap(),
1080 body: SeamEffect::Mob(MobEffect::RequestRuntimeBinding {
1081 agent_runtime_id: "rt-1".into(),
1082 fence_token: 7,
1083 generation: 3,
1084 session_id: "019dbd3d-d7ad-75a1-96d0-8013927e78f8".into(),
1085 }),
1086 }
1087 }
1088
1089 fn build_dispatcher(
1090 consumer: Arc<RecordingMeerkatSurface>,
1091 ) -> CatalogCompositionDispatcher<SeamEffect> {
1092 let schema = meerkat_mob_seam_composition();
1093 let table = RouteTable::from_schema(&schema).expect("seam schema routes are well-formed");
1094 CatalogCompositionDispatcher::new(schema.name.clone(), table).with_consumer(consumer)
1095 }
1096
1097 fn sample_signal() -> SignalPayload<SeamSignal> {
1098 let body = SeamSignal::RuntimeBound {
1099 agent_runtime_id: "rt-1".into(),
1100 fence_token: 7,
1101 };
1102 SignalPayload::Emitted {
1103 variant: body.variant_id(),
1104 body,
1105 }
1106 }
1107
1108 fn build_signal_dispatcher(
1109 consumer: Arc<RecordingMobSignalSurface>,
1110 ) -> CatalogCompositionSignalDispatcher<SeamSignal> {
1111 let schema = meerkat_mob_seam_composition();
1112 let table = RouteTable::from_schema(&schema).expect("seam schema routes are well-formed");
1113 CatalogCompositionSignalDispatcher::new(schema.name.clone(), table).with_consumer(consumer)
1114 }
1115
1116 #[tokio::test]
1117 async fn dispatches_mob_routed_effect_to_meerkat_consumer() {
1118 let consumer = Arc::new(RecordingMeerkatSurface::default());
1119 let dispatcher = build_dispatcher(Arc::clone(&consumer));
1120
1121 let outcome = dispatcher
1122 .dispatch(mob_producer(), sample_effect())
1123 .await
1124 .expect("well-formed routed effect");
1125
1126 assert_eq!(outcome.consumer.as_str(), "meerkat");
1127 assert_eq!(outcome.applied_input.as_str(), "PrepareBindings");
1128 assert_eq!(
1129 outcome.route.route_id.as_str(),
1130 "binding_request_reaches_meerkat"
1131 );
1132
1133 let log = consumer.log.lock().await;
1134 assert_eq!(
1135 log.len(),
1136 1,
1137 "dispatcher must call the consumer exactly once"
1138 );
1139 let (variant, fields) = &log[0];
1140 assert_eq!(variant.as_str(), "PrepareBindings");
1141 let field_names: Vec<&str> = fields.iter().map(|(k, _)| k.as_str()).collect();
1142 assert_eq!(
1143 field_names,
1144 vec![
1145 "agent_runtime_id",
1146 "fence_token",
1147 "generation",
1148 "session_id"
1149 ]
1150 );
1151 match &fields[0].1 {
1152 OwnedFieldValue::Str(s) => assert_eq!(s, "rt-1"),
1153 other => panic!("expected Str, got {other:?}"),
1154 }
1155 match &fields[1].1 {
1156 OwnedFieldValue::U64(v) => assert_eq!(*v, 7),
1157 other => panic!("expected U64, got {other:?}"),
1158 }
1159 match &fields[2].1 {
1160 OwnedFieldValue::U64(v) => assert_eq!(*v, 3),
1161 other => panic!("expected U64, got {other:?}"),
1162 }
1163 match &fields[3].1 {
1164 OwnedFieldValue::Str(s) => assert_eq!(s, "019dbd3d-d7ad-75a1-96d0-8013927e78f8"),
1165 other => panic!("expected Str for session_id, got {other:?}"),
1166 }
1167 }
1168
1169 #[tokio::test]
1170 async fn dispatches_meerkat_routed_signal_to_mob_consumer() {
1171 let consumer = Arc::new(RecordingMobSignalSurface::default());
1172 let dispatcher = build_signal_dispatcher(Arc::clone(&consumer));
1173
1174 let outcome = dispatcher
1175 .dispatch_signal(meerkat_producer(), sample_signal())
1176 .await
1177 .expect("well-formed routed signal");
1178
1179 assert_eq!(outcome.consumer.as_str(), "mob");
1180 assert_eq!(outcome.applied_signal.as_str(), "ObserveRuntimeReady");
1181 assert_eq!(outcome.route.route_id.as_str(), "runtime_bound_reaches_mob");
1182
1183 let log = consumer.log.lock().await;
1184 assert_eq!(
1185 log.len(),
1186 1,
1187 "dispatcher must call the signal consumer exactly once"
1188 );
1189 let (variant, fields) = &log[0];
1190 assert_eq!(variant.as_str(), "ObserveRuntimeReady");
1191 let field_names: Vec<&str> = fields.iter().map(|(k, _)| k.as_str()).collect();
1192 assert_eq!(field_names, vec!["agent_runtime_id", "fence_token"]);
1193 match &fields[0].1 {
1194 OwnedFieldValue::Str(s) => assert_eq!(s, "rt-1"),
1195 other => panic!("expected Str, got {other:?}"),
1196 }
1197 match &fields[1].1 {
1198 OwnedFieldValue::U64(v) => assert_eq!(*v, 7),
1199 other => panic!("expected U64, got {other:?}"),
1200 }
1201 }
1202
1203 #[tokio::test]
1204 async fn signal_dispatch_refuses_input_route_typed() {
1205 let consumer = Arc::new(RecordingMobSignalSurface::default());
1206 let dispatcher = build_signal_dispatcher(consumer);
1207
1208 let payload = SignalPayload::Emitted {
1209 variant: EffectVariantId::parse("RequestRuntimeBinding").unwrap(),
1210 body: SeamSignal::RuntimeBound {
1211 agent_runtime_id: "rt-1".into(),
1212 fence_token: 7,
1213 },
1214 };
1215
1216 let err = dispatcher
1217 .dispatch_signal(mob_producer(), payload)
1218 .await
1219 .expect_err("input route is out of the signal surface");
1220
1221 assert!(matches!(err, SignalDispatchRefusal::UnresolvedRoute { .. }));
1222 }
1223
1224 #[tokio::test]
1225 async fn signal_dispatch_refuses_unwired_consumer_typed() {
1226 let schema = meerkat_mob_seam_composition();
1227 let table = RouteTable::from_schema(&schema).unwrap();
1228 let dispatcher: CatalogCompositionSignalDispatcher<SeamSignal> =
1229 CatalogCompositionSignalDispatcher::new(schema.name.clone(), table);
1230
1231 let err = dispatcher
1232 .dispatch_signal(meerkat_producer(), sample_signal())
1233 .await
1234 .expect_err("unwired signal consumer");
1235
1236 assert!(matches!(err, SignalDispatchRefusal::UnwiredConsumer { .. }));
1237 }
1238
1239 #[tokio::test]
1240 async fn signal_dispatch_refuses_missing_field_typed() {
1241 #[derive(Debug)]
1242 struct BrokenSignal;
1243
1244 impl ProducerSignal for BrokenSignal {
1245 fn variant_id(&self) -> EffectVariantId {
1246 EffectVariantId::parse("RuntimeBound").unwrap()
1247 }
1248
1249 fn field(&self, _id: &FieldId) -> Option<FieldValue<'_>> {
1250 None
1251 }
1252 }
1253
1254 let schema = meerkat_mob_seam_composition();
1255 let table = RouteTable::from_schema(&schema).unwrap();
1256 let consumer = Arc::new(RecordingMobSignalSurface::default());
1257 let dispatcher: CatalogCompositionSignalDispatcher<BrokenSignal> =
1258 CatalogCompositionSignalDispatcher::new(schema.name.clone(), table)
1259 .with_consumer(consumer);
1260
1261 let err = dispatcher
1262 .dispatch_signal(
1263 meerkat_producer(),
1264 SignalPayload::Emitted {
1265 variant: EffectVariantId::parse("RuntimeBound").unwrap(),
1266 body: BrokenSignal,
1267 },
1268 )
1269 .await
1270 .expect_err("missing producer field");
1271
1272 assert!(matches!(
1273 err,
1274 SignalDispatchRefusal::MissingProducerField { .. }
1275 ));
1276 }
1277
1278 #[tokio::test]
1279 async fn refuses_mismatched_composition() {
1280 let consumer = Arc::new(RecordingMeerkatSurface::default());
1281 let dispatcher = build_dispatcher(consumer);
1282
1283 let mut wrong = mob_producer();
1284 wrong.composition = CompositionId::parse("some_other_composition").unwrap();
1285
1286 let err = dispatcher
1287 .dispatch(wrong, sample_effect())
1288 .await
1289 .expect_err("composition mismatch");
1290
1291 assert!(matches!(err, DispatchRefusal::CompositionMismatch { .. }));
1292 }
1293
1294 #[tokio::test]
1295 async fn refuses_unrouted_effect_typed() {
1296 let consumer = Arc::new(RecordingMeerkatSurface::default());
1297 let dispatcher = build_dispatcher(consumer);
1298
1299 let payload = EffectPayload::Emitted {
1303 variant: EffectVariantId::parse("UnknownEffect").unwrap(),
1304 body: SeamEffect::Mob(MobEffect::RequestRuntimeBinding {
1305 agent_runtime_id: "rt".into(),
1306 fence_token: 0,
1307 generation: 0,
1308 session_id: "019dbd3d-d7ad-75a1-96d0-8013927e78f8".into(),
1309 }),
1310 };
1311
1312 let err = dispatcher
1313 .dispatch(mob_producer(), payload)
1314 .await
1315 .expect_err("unresolved route");
1316
1317 assert!(matches!(err, DispatchRefusal::UnresolvedRoute { .. }));
1318 }
1319
1320 #[tokio::test]
1321 async fn refuses_unwired_consumer_typed() {
1322 let schema = meerkat_mob_seam_composition();
1326 let table = RouteTable::from_schema(&schema).unwrap();
1327 let dispatcher: CatalogCompositionDispatcher<SeamEffect> =
1328 CatalogCompositionDispatcher::new(schema.name.clone(), table);
1329
1330 let err = dispatcher
1331 .dispatch(mob_producer(), sample_effect())
1332 .await
1333 .expect_err("unwired consumer");
1334
1335 assert!(matches!(err, DispatchRefusal::UnwiredConsumer { .. }));
1336 }
1337
1338 #[tokio::test]
1339 async fn standalone_binding_has_no_dispatcher() {
1340 let binding: CompositionBinding<SeamEffect> = CompositionBinding::Standalone;
1341 assert!(binding.is_standalone());
1342 assert!(binding.wired().is_none());
1343 }
1344
1345 #[tokio::test]
1346 async fn wired_binding_exposes_dispatcher() {
1347 let consumer = Arc::new(RecordingMeerkatSurface::default());
1348 let dispatcher = Arc::new(build_dispatcher(consumer));
1349 let binding: CompositionBinding<SeamEffect> = CompositionBinding::Wired(dispatcher);
1350 assert!(!binding.is_standalone());
1351 assert!(binding.wired().is_some());
1352 assert!(
1353 binding.context_provider().is_none(),
1354 "plain Wired binding has no owner-supplied context"
1355 );
1356 }
1357
1358 struct PinnedSessionContext {
1364 session_id: String,
1365 }
1366
1367 impl ContextProvider<SeamEffect> for PinnedSessionContext {
1368 fn provide_context(
1369 &self,
1370 _producer: &ProducerInstance,
1371 _effect: &EffectPayload<SeamEffect>,
1372 ) -> Vec<(FieldId, OwnedFieldValue)> {
1373 vec![(
1374 FieldId::parse("session_id").expect("field id"),
1375 OwnedFieldValue::Str(self.session_id.clone()),
1376 )]
1377 }
1378 }
1379
1380 #[tokio::test]
1381 async fn owner_provided_binding_exposes_both_dispatcher_and_context() {
1382 let consumer = Arc::new(RecordingMeerkatSurface::default());
1383 let dispatcher = Arc::new(build_dispatcher(consumer));
1384 let context = Arc::new(PinnedSessionContext {
1385 session_id: "session-abc".into(),
1386 });
1387 let binding: CompositionBinding<SeamEffect> =
1388 CompositionBinding::owner_provided(dispatcher, context);
1389
1390 assert!(!binding.is_standalone());
1391 assert!(
1392 binding.wired().is_some(),
1393 "OwnerProvided is a superset of Wired for dispatcher access"
1394 );
1395 assert!(
1396 binding.context_provider().is_some(),
1397 "OwnerProvided must expose the owner-supplied context"
1398 );
1399
1400 let provider = binding.context_provider().expect("context provider");
1405 let producer = mob_producer();
1406 let effect = sample_effect();
1407 let fields = provider.provide_context(&producer, &effect);
1408 assert_eq!(fields.len(), 1);
1409 assert_eq!(fields[0].0.as_str(), "session_id");
1410 match &fields[0].1 {
1411 OwnedFieldValue::Str(s) => assert_eq!(s, "session-abc"),
1412 other => panic!("expected Str context field, got {other:?}"),
1413 }
1414 }
1415
1416 #[tokio::test]
1417 async fn composition_binding_constructors_parallel_machine_halves() {
1418 let standalone: CompositionBinding<SeamEffect> = CompositionBinding::standalone();
1424 assert!(standalone.is_standalone());
1425 assert!(standalone.wired().is_none());
1426 assert!(standalone.context_provider().is_none());
1427
1428 let consumer = Arc::new(RecordingMeerkatSurface::default());
1429 let dispatcher: Arc<dyn CompositionDispatcher<Effect = SeamEffect>> =
1430 Arc::new(build_dispatcher(consumer));
1431 let wired: CompositionBinding<SeamEffect> =
1432 CompositionBinding::wired_with(Arc::clone(&dispatcher));
1433 assert!(!wired.is_standalone());
1434 assert!(wired.wired().is_some());
1435 assert!(wired.context_provider().is_none());
1436
1437 let context = Arc::new(PinnedSessionContext {
1438 session_id: "session-xyz".into(),
1439 });
1440 let owner_provided: CompositionBinding<SeamEffect> =
1441 CompositionBinding::owner_provided(dispatcher, context);
1442 assert!(!owner_provided.is_standalone());
1443 assert!(owner_provided.wired().is_some());
1444 assert!(owner_provided.context_provider().is_some());
1445 }
1446
1447 struct RefusingMeerkatSurface;
1450
1451 #[async_trait]
1452 impl ConsumerSurface for RefusingMeerkatSurface {
1453 fn instance_id(&self) -> &MachineInstanceId {
1454 static ID: std::sync::OnceLock<MachineInstanceId> = std::sync::OnceLock::new();
1455 ID.get_or_init(|| MachineInstanceId::parse("meerkat").unwrap())
1456 }
1457
1458 async fn apply_routed_input(
1459 &self,
1460 _variant: InputVariantId,
1461 _projected_fields: Vec<(FieldId, OwnedFieldValue)>,
1462 ) -> Result<(), ConsumerError> {
1463 Err(ConsumerError::new(
1464 "runtime_destroyed",
1465 "consumer machine no longer accepts inputs",
1466 ))
1467 }
1468 }
1469
1470 #[tokio::test]
1476 async fn consumer_refusal_preserves_typed_error_code_through_dispatcher() {
1477 let schema = meerkat_mob_seam_composition();
1478 let table = RouteTable::from_schema(&schema).expect("seam schema routes are well-formed");
1479 let dispatcher = CatalogCompositionDispatcher::new(schema.name.clone(), table)
1480 .with_consumer(Arc::new(RefusingMeerkatSurface));
1481
1482 let err = dispatcher
1483 .dispatch(mob_producer(), sample_effect())
1484 .await
1485 .expect_err("refusing consumer surface");
1486
1487 match err {
1488 DispatchRefusal::ConsumerRefused { error, .. } => {
1489 assert_eq!(
1490 error.error_code(),
1491 "runtime_destroyed",
1492 "typed consumer error_code must survive the dispatch seam, not be flattened to a string"
1493 );
1494 }
1495 other => {
1496 panic!("expected ConsumerRefused carrying a typed ConsumerError, got {other:?}")
1497 }
1498 }
1499 }
1500}