Skip to main content

meerkat_runtime/composition/
mod.rs

1//! Composition dispatcher — THE typed execution path for routed effects.
2//!
3//! Wave-b V2 rebuilds composition dispatch as a typed, *mandatory* runtime seam.
4//! The deleted `composition_dispatch.rs` and `recompute_mob_peer_overlay*.rs`
5//! (wave-a tombstones `ce2dbe35e` / `f5e366f38`) were stringly-typed helpers
6//! that callers opted into. This module is their structural opposite:
7//!
8//! * **Typed end-to-end.** Producer identity is [`ProducerInstance`] carrying
9//!   typed [`CompositionId`], [`MachineInstanceId`], [`MachineId`]. Effects
10//!   travel as [`EffectPayload<E>`] where `E` is the producer composition's
11//!   typed seam-effect sum (the [`ProducerEffect`] trait bound). Route
12//!   resolution returns a typed [`RoutedInputDescriptor`] carrying
13//!   [`MachineInstanceId`] / [`InputVariantId`] / `Vec<(FieldId, FieldId)>`.
14//!   Signal-kind routes travel through the parallel [`SignalPayload<S>`] /
15//!   [`CompositionSignalDispatcher`] surface and resolve to typed
16//!   [`RoutedSignalDescriptor`] values.
17//! * **Mandatory, not optional.** The trait has no fallback surface. Routed
18//!   effects whose route is declared in the composition schema MUST resolve
19//!   through the dispatcher; unresolved routes are a typed
20//!   [`DispatchRefusal::UnresolvedRoute`] error, not a silent drop.
21//!   Signal-kind routes live on a separate index inside [`RouteTable`] and
22//!   MUST resolve through [`CompositionSignalDispatcher`].
23//! * **Compile-time presence/absence.** A `MeerkatMachine` either has a
24//!   composition dispatcher attached (via the `with_composition` constructor)
25//!   or it does not (the standalone / single-machine test path). The two
26//!   cases are distinguished by a typed [`CompositionBinding`] discriminant,
27//!   never by `Option<Arc<dyn CompositionDispatcher>>`.
28//!
29//! The default catalog-backed dispatcher ([`CatalogCompositionDispatcher`])
30//! consumes a [`RouteTable`] built from any
31//! [`meerkat_machine_schema::CompositionSchema`] and delivers each resolved
32//! [`RoutedInputDescriptor`] to a per-consumer-instance [`ConsumerSurface`]
33//! supplied at wire-up. The per-composition codegen module emitted by
34//! `meerkat-machine-codegen` (B-4 + B-4b) plugs in as the
35//! [`ProducerEffect`] implementation — `route_to_input` is equivalent to
36//! consulting the [`RouteTable`] built from the same schema.
37
38pub 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/// Typed identity of the producing machine instance inside a composition.
54///
55/// Unlike the deleted string-keyed helpers, every field is a typed newtype
56/// so cross-instance mixups are rejected at compile time.
57#[derive(Debug, Clone, PartialEq, Eq, Hash)]
58pub struct ProducerInstance {
59    /// Composition that contains the producer.
60    pub composition: CompositionId,
61    /// Instance id of the producer *within* `composition`.
62    pub instance_id: MachineInstanceId,
63    /// Underlying machine name (the schema this instance is an instance of).
64    pub machine: MachineId,
65}
66
67/// Typed effect payload. Generic over the producer composition's seam-effect
68/// sum (see the codegen-emitted `{Composition}Effect` enum).
69///
70/// The variant carries the typed [`EffectVariantId`] alongside the body so
71/// the dispatcher can look up a route without pattern-matching on the
72/// producer's effect enum (the [`ProducerEffect`] trait hides that under
73/// [`ProducerEffect::variant_id`]).
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub enum EffectPayload<E> {
76    /// Producer emitted a typed effect variant.
77    Emitted {
78        /// Typed variant id (matches the producer's effect enum tag).
79        variant: EffectVariantId,
80        /// The typed effect body.
81        body: E,
82    },
83}
84
85impl<E: ProducerEffect> EffectPayload<E> {
86    /// Borrow the typed variant id.
87    pub fn variant(&self) -> &EffectVariantId {
88        match self {
89            Self::Emitted { variant, .. } => variant,
90        }
91    }
92
93    /// Borrow the typed body.
94    pub fn body(&self) -> &E {
95        match self {
96            Self::Emitted { body, .. } => body,
97        }
98    }
99}
100
101/// Typed signal-route payload. Generic over the producer composition's
102/// seam-signal sum.
103#[derive(Debug, Clone, PartialEq, Eq)]
104pub enum SignalPayload<S> {
105    /// Producer emitted a typed signal-route source variant.
106    Emitted {
107        /// Typed source variant id. In the composition schema this is the
108        /// route's producer-side `effect_variant`; signal-kind routes still
109        /// originate from a producer effect and target a consumer signal.
110        variant: EffectVariantId,
111        /// The typed signal source body.
112        body: S,
113    },
114}
115
116impl<S: ProducerSignal> SignalPayload<S> {
117    /// Borrow the typed source variant id.
118    pub fn variant(&self) -> &EffectVariantId {
119        match self {
120            Self::Emitted { variant, .. } => variant,
121        }
122    }
123
124    /// Borrow the typed body.
125    pub fn body(&self) -> &S {
126        match self {
127            Self::Emitted { body, .. } => body,
128        }
129    }
130}
131
132/// Typed route key: `(composition, route)`.
133#[derive(Debug, Clone, PartialEq, Eq, Hash)]
134pub struct RouteKey {
135    pub composition: CompositionId,
136    pub route_id: RouteId,
137}
138
139/// Marker trait for the seam-effect sum emitted by
140/// `meerkat-machine-codegen::render_composition_driver`. Producer effect
141/// enums implement this to expose the typed variant id alongside their
142/// domain body — the dispatcher consults it without inspecting the enum.
143pub trait ProducerEffect: fmt::Debug + Send + Sync + 'static {
144    /// Typed variant id for this effect value.
145    ///
146    /// The codegen emits one arm per distinct `{producer_instance}::{variant}`
147    /// pair; implementers return the matching [`EffectVariantId`]. This is
148    /// the single handle the dispatcher needs to resolve the route without
149    /// case-matching on the producer's concrete enum.
150    fn variant_id(&self) -> EffectVariantId;
151
152    /// Borrow a field value by [`FieldId`].
153    ///
154    /// Used by the dispatcher to project producer fields into the typed
155    /// consumer input as declared by the composition's route bindings.
156    /// Returns `None` if the requested field is not present on this
157    /// variant. The dispatcher treats that as
158    /// [`DispatchRefusal::MissingProducerField`].
159    fn field(&self, id: &FieldId) -> Option<FieldValue<'_>>;
160}
161
162/// Marker trait for the seam-signal source sum consumed by
163/// [`CompositionSignalDispatcher`].
164///
165/// Signal-kind composition routes use the same producer-side
166/// `EffectVariantId` namespace as input routes, but their target is a
167/// consumer [`SignalVariantId`]. This trait mirrors [`ProducerEffect`] so
168/// signal dispatch has the same typed projection discipline without
169/// requiring callers to smuggle signal payloads through the input
170/// dispatcher.
171pub trait ProducerSignal: fmt::Debug + Send + Sync + 'static {
172    /// Typed producer-side variant id for this signal source value.
173    fn variant_id(&self) -> EffectVariantId;
174
175    /// Borrow a producer field by [`FieldId`].
176    fn field(&self, id: &FieldId) -> Option<FieldValue<'_>>;
177}
178
179/// Typed view over a producer-field value projected through a route binding.
180///
181/// The `ProducerEffect::field` implementation returns one of these so the
182/// dispatcher can move the value into the consumer input without a
183/// `serde_json::Value` round-trip. The variant set is intentionally small;
184/// richer shapes are expressed by the producer keeping the typed value
185/// inside its effect body and the consumer accepting it via the same
186/// typed enum (the codegen emits the matching types on both sides).
187#[derive(Debug, Clone)]
188pub enum FieldValue<'a> {
189    /// Borrowed string slice (owning producer retains the backing `String`).
190    Str(&'a str),
191    /// Unsigned 64-bit integer.
192    U64(u64),
193    /// Signed 64-bit integer.
194    I64(i64),
195    /// Boolean flag.
196    Bool(bool),
197    /// Opaque typed handle — producer and consumer agree on the Rust type.
198    /// The dispatcher moves the `Arc<dyn Any>` across without inspecting it.
199    /// This is *not* a `serde_json::Value` escape hatch: the contained Rust
200    /// type is determined by the typed route binding, not by ad-hoc JSON.
201    Opaque(Arc<dyn std::any::Any + Send + Sync>),
202}
203
204/// Outcome when a routed effect is successfully dispatched.
205#[derive(Debug, Clone, PartialEq, Eq)]
206pub struct DispatchOutcome {
207    /// Route that was resolved for this effect.
208    pub route: RouteKey,
209    /// Target consumer instance the typed input was delivered to.
210    pub consumer: MachineInstanceId,
211    /// Typed input variant applied on the consumer.
212    pub applied_input: InputVariantId,
213}
214
215/// Outcome when a routed signal is successfully dispatched.
216#[derive(Debug, Clone, PartialEq, Eq)]
217pub struct SignalDispatchOutcome {
218    /// Route that was resolved for this signal source.
219    pub route: RouteKey,
220    /// Target consumer instance the typed signal was delivered to.
221    pub consumer: MachineInstanceId,
222    /// Typed signal variant applied on the consumer.
223    pub applied_signal: SignalVariantId,
224}
225
226/// Reasons the dispatcher refuses a routed effect.
227///
228/// Unlike the deleted helper path, there is no "silently drop unknown
229/// effects" arm. Every failure is a typed variant so callers and RMAT
230/// audits can enumerate them without parsing error strings.
231#[derive(Debug, Clone, PartialEq, Eq, Error)]
232pub enum DispatchRefusal {
233    /// The producer is not registered for this dispatcher's composition.
234    #[error("dispatcher composition {expected} does not match producer composition {actual}")]
235    CompositionMismatch {
236        expected: CompositionId,
237        actual: CompositionId,
238    },
239    /// No input-kind route is declared for `(producer.instance_id, variant)`.
240    #[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    /// A route-binding references a producer field that the effect body did
249    /// not supply (via [`ProducerEffect::field`]).
250    #[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    /// No [`ConsumerSurface`] is registered for the resolved target
257    /// instance. This is a wiring bug at construction time, not a runtime
258    /// signal — the dispatcher refuses rather than queueing forever.
259    #[error(
260        "no consumer surface registered for target instance {instance} in composition {composition}"
261    )]
262    UnwiredConsumer {
263        composition: CompositionId,
264        instance: MachineInstanceId,
265    },
266    /// The consumer surface rejected the typed input (e.g. because the
267    /// consumer machine is no longer accepting inputs). The inner
268    /// [`ConsumerError`] is the consumer-side typed rejection and is opaque
269    /// to the dispatcher, but its stable `error_code` survives the seam so
270    /// callers and RMAT audits can enumerate refusals without parsing
271    /// strings.
272    #[error("consumer {instance} refused input {variant}: {error}")]
273    ConsumerRefused {
274        instance: MachineInstanceId,
275        variant: InputVariantId,
276        error: ConsumerError,
277    },
278}
279
280/// Reasons the signal dispatcher refuses a routed signal.
281#[derive(Debug, Clone, PartialEq, Eq, Error)]
282pub enum SignalDispatchRefusal {
283    /// The producer is not registered for this dispatcher's composition.
284    #[error("dispatcher composition {expected} does not match producer composition {actual}")]
285    CompositionMismatch {
286        expected: CompositionId,
287        actual: CompositionId,
288    },
289    /// No signal-kind route is declared for `(producer.instance_id, variant)`.
290    #[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    /// A route-binding references a producer field that the signal body
299    /// did not supply.
300    #[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    /// No [`SignalConsumerSurface`] is registered for the resolved target
307    /// instance.
308    #[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    /// The consumer surface rejected the typed signal. The inner
316    /// [`ConsumerError`] preserves the consumer's stable `error_code` across
317    /// the dispatch seam instead of flattening it into an untyped string.
318    #[error("consumer {instance} refused signal {variant}: {error}")]
319    ConsumerRefused {
320        instance: MachineInstanceId,
321        variant: SignalVariantId,
322        error: ConsumerError,
323    },
324}
325
326/// Opaque, typed consumer-side rejection.
327///
328/// The consumer surface owns a typed kernel error; the dispatcher must not
329/// flatten it into a bare string and lose the stable discriminant. This
330/// newtype is the dispatcher-facing projection of that typed error: it carries
331/// the consumer's stable [`ConsumerError::error_code`] (the same
332/// `&'static str` convention used by `meerkat_core::error`) alongside a
333/// human-readable message. The dispatcher moves it across the seam verbatim,
334/// so [`DispatchRefusal::ConsumerRefused`] / [`SignalDispatchRefusal::ConsumerRefused`]
335/// expose the typed code rather than re-parsing a flattened reason string.
336#[derive(Debug, Clone, PartialEq, Eq, Error)]
337#[error("{message} [{error_code}]")]
338pub struct ConsumerError {
339    /// Stable discriminant the consumer kernel owns (e.g. the consumer's
340    /// own typed `error_code()`). Opaque to the dispatcher, but enumerable
341    /// by callers and RMAT audits without parsing the message.
342    error_code: &'static str,
343    /// Human-readable rejection detail.
344    message: String,
345}
346
347impl ConsumerError {
348    /// Build a consumer rejection from the consumer kernel's stable
349    /// `error_code` and a human-readable message.
350    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    /// Stable typed discriminant preserved across the dispatch seam.
358    pub fn error_code(&self) -> &'static str {
359        self.error_code
360    }
361
362    /// Human-readable rejection detail.
363    pub fn message(&self) -> &str {
364        &self.message
365    }
366}
367
368impl From<String> for ConsumerError {
369    /// Consumer-surface input projection/refusal detail that the kernel
370    /// produced as a bare string is carried across the dispatch seam under a
371    /// stable `consumer_projection_failed` discriminant (the string stays as
372    /// human-readable detail, but the typed code is what the dispatcher and
373    /// RMAT audits observe — never a re-parsed message).
374    fn from(message: String) -> Self {
375        Self::new("consumer_projection_failed", message)
376    }
377}
378
379/// Delivery surface for one consumer instance inside a composition.
380///
381/// A consumer (e.g. the `meerkat` machine instance when mob routes
382/// `RequestRuntimeBinding` at it) implements this trait and registers an
383/// instance at composition wire-up. The dispatcher invokes it exactly once
384/// per resolved [`RoutedInput`]. The implementation is responsible for
385/// materializing the consumer-side typed input — the dispatcher only moves
386/// typed data across the seam.
387#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
388#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
389pub trait ConsumerSurface: Send + Sync {
390    /// Instance id this surface serves. The dispatcher matches against
391    /// [`RoutedInput::instance_id`] to pick the right surface.
392    fn instance_id(&self) -> &MachineInstanceId;
393
394    /// Apply a typed routed input. `projected_fields` carries the per-
395    /// consumer-field values resolved from the producer via the route's
396    /// field-bindings, owned so the surface can move them into its typed
397    /// input constructor.
398    async fn apply_routed_input(
399        &self,
400        variant: InputVariantId,
401        projected_fields: Vec<(FieldId, OwnedFieldValue)>,
402    ) -> Result<(), ConsumerError>;
403}
404
405/// Delivery surface for one signal-consuming instance inside a composition.
406#[async_trait]
407pub trait SignalConsumerSurface: Send + Sync {
408    /// Instance id this surface serves.
409    fn instance_id(&self) -> &MachineInstanceId;
410
411    /// Receive a typed routed signal.
412    async fn receive_signal(
413        &self,
414        variant: SignalVariantId,
415        projected_fields: Vec<(FieldId, OwnedFieldValue)>,
416    ) -> Result<(), ConsumerError>;
417}
418
419/// Owned counterpart of [`FieldValue`] used when delivering a routed input
420/// across the consumer-surface boundary. Moving owned values means the
421/// consumer can construct its typed input without re-borrowing the
422/// producer.
423#[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    /// Lift a borrowed field value into its owned counterpart, cloning the
434    /// backing `&str` when required. The [`Arc<dyn Any>`] path is shared,
435    /// not cloned.
436    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/// Composition dispatcher trait.
448///
449/// Monomorphized over the producer composition's seam-effect sum
450/// ([`CompositionDispatcher::Effect`]). Making the effect an associated type
451/// (rather than a generic on the method) keeps the trait dyn-safe — a
452/// `MeerkatMachine` can hold `Arc<dyn CompositionDispatcher<Effect = ...>>`
453/// without leaking the machine kernel's monomorphization concerns.
454#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
455#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
456pub trait CompositionDispatcher: Send + Sync {
457    /// Seam-effect sum this dispatcher handles. Matches the codegen-emitted
458    /// `{Composition}Effect` enum.
459    type Effect: ProducerEffect;
460
461    /// Composition id this dispatcher owns. Every [`ProducerInstance`]
462    /// passed to [`CompositionDispatcher::dispatch`] must match.
463    fn composition(&self) -> &CompositionId;
464
465    /// Dispatch a routed effect. Returns [`DispatchOutcome`] on success or
466    /// a typed [`DispatchRefusal`]. There is no silent-drop arm.
467    async fn dispatch(
468        &self,
469        producer: ProducerInstance,
470        effect: EffectPayload<Self::Effect>,
471    ) -> Result<DispatchOutcome, DispatchRefusal>;
472}
473
474/// Composition signal dispatcher trait.
475#[async_trait]
476pub trait CompositionSignalDispatcher: Send + Sync {
477    /// Seam-signal source sum this dispatcher handles.
478    type Signal: ProducerSignal;
479
480    /// Composition id this dispatcher owns.
481    fn composition(&self) -> &CompositionId;
482
483    /// Dispatch a routed signal. Returns [`SignalDispatchOutcome`] on
484    /// success or a typed [`SignalDispatchRefusal`].
485    async fn dispatch_signal(
486        &self,
487        producer: ProducerInstance,
488        signal: SignalPayload<Self::Signal>,
489    ) -> Result<SignalDispatchOutcome, SignalDispatchRefusal>;
490}
491
492/// Typed, owner-supplied context provider for an [`OwnerProvided`][op] binding.
493///
494/// Issue #342 — some routes need consumer-side fields that aren't in the
495/// producer's effect body (the canonical case is `session_id` on the
496/// `meerkat_mob_seam` composition: the mob effect doesn't carry it, but
497/// the consumer's applied input requires it). Rather than smuggle that
498/// state through a `serde_json::Value` side channel, the runtime that
499/// owns the dispatcher supplies it through a typed context provider.
500///
501/// **Exactly one method, no `serde_json::Value` in the signature.** The
502/// returned fields are typed [`OwnedFieldValue`]s keyed by
503/// [`FieldId`] — the same representation the route-binding table already
504/// uses for producer-field projections. The dispatcher can merge the
505/// provider's fields with producer-projected fields when constructing
506/// the typed input for a `ConsumerSurface`.
507///
508/// Implementations are synchronous and infallible: context retrieval
509/// should be an in-process lookup against state the runtime already
510/// owns (pinned session id, realm id, bind-epoch, …). Anything that
511/// could fail belongs on the producer effect body or on the consumer
512/// surface.
513///
514/// [op]: CompositionBinding::OwnerProvided
515pub trait ContextProvider<E: ProducerEffect>: Send + Sync {
516    /// Produce the owner-supplied typed context fields for a routed
517    /// `effect` emitted by `producer`.
518    ///
519    /// The returned vector's `FieldId`s must match the route's
520    /// [`BindingSource::ContextField`][bs] references declared in the
521    /// composition schema (#342). Missing ids surface as
522    /// [`DispatchRefusal::MissingProducerField`] at the dispatcher in
523    /// the same way unfulfilled producer fields do — the dispatcher
524    /// treats producer and owner-provided fields uniformly once
525    /// projection starts.
526    ///
527    /// [bs]: # "See issue #342: BindingSource gains ContextField(FieldId)"
528    fn provide_context(
529        &self,
530        producer: &ProducerInstance,
531        effect: &EffectPayload<E>,
532    ) -> Vec<(FieldId, OwnedFieldValue)>;
533}
534
535/// Typed binding attached to a runtime that holds a dispatcher.
536///
537/// Discriminates the "machine participates in a composition" case from the
538/// "machine is standalone" case *at the type level*: no
539/// `Option<Arc<dyn CompositionDispatcher>>`. Callers obtain the concrete
540/// dispatcher via [`CompositionBinding::wired`] and honor
541/// [`CompositionBinding::is_standalone`] to tell the two apart. The two
542/// constructor halves on `MeerkatMachine` (`with_composition(...)` vs
543/// `standalone(...)` / `ephemeral()` / `persistent()`) are the public
544/// face of this distinction.
545///
546/// **OwnerProvided (#342)**: some routes need consumer-side fields that
547/// aren't in the producer effect body — the canonical case is
548/// `session_id` on the `meerkat_mob_seam` composition. The
549/// `OwnerProvided` variant pairs a dispatcher with a typed
550/// [`ContextProvider`] so the runtime that owns the dispatcher supplies
551/// the missing fields from its own typed state at dispatch time.
552/// `OwnerProvided` is semantically a superset of `Wired`: callers that
553/// only need the dispatcher reach it through the same
554/// [`wired`](Self::wired) accessor; callers that need the context
555/// provider reach it through
556/// [`context_provider`](Self::context_provider), which returns `Some`
557/// only for `OwnerProvided`.
558pub enum CompositionBinding<E: ProducerEffect> {
559    /// Machine is not part of a composition. Routed-effect dispatch is not
560    /// available.
561    Standalone,
562    /// Machine participates in a composition and owns a typed dispatcher.
563    /// No owner-supplied context: all route bindings project from the
564    /// producer's effect body.
565    Wired(Arc<dyn CompositionDispatcher<Effect = E>>),
566    /// Machine participates in a composition that declares routes with
567    /// owner-supplied context (issue #342). The `context` is consulted
568    /// alongside the producer effect at dispatch time to fulfil route
569    /// bindings whose source is `ContextField` rather than
570    /// `ProducerField`.
571    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    /// Construct a `Standalone` binding.
596    ///
597    /// Mirrors `MeerkatMachine::standalone(...)` at the binding level so
598    /// call sites that wire a runtime without composition can say so
599    /// positively instead of spelling the enum variant. Equivalent to
600    /// `CompositionBinding::Standalone`.
601    pub fn standalone() -> Self {
602        Self::Standalone
603    }
604
605    /// Construct a `Wired` binding from a composition dispatcher.
606    ///
607    /// Use this when every route binding projects from the producer
608    /// effect body alone. If any route declares an owner-supplied
609    /// context field, use [`Self::owner_provided`] instead.
610    pub fn wired_with(dispatcher: Arc<dyn CompositionDispatcher<Effect = E>>) -> Self {
611        Self::Wired(dispatcher)
612    }
613
614    /// Construct an `OwnerProvided` binding from a composition
615    /// dispatcher and a typed context provider.
616    ///
617    /// Use this for compositions whose route bindings reference owner-
618    /// supplied context fields (per issue #342) — the provider is
619    /// consulted at dispatch time for each routed effect so the
620    /// missing fields can be fulfilled from the runtime's own state.
621    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    /// Report whether this machine is standalone (no composition attached).
632    pub fn is_standalone(&self) -> bool {
633        matches!(self, Self::Standalone)
634    }
635
636    /// Borrow the wired dispatcher, if any.
637    ///
638    /// Returns `None` only for [`CompositionBinding::Standalone`].
639    /// Both `Wired` and `OwnerProvided` expose their dispatcher through
640    /// this accessor so call sites that only need to dispatch a typed
641    /// effect don't have to branch on context-provider presence — the
642    /// type split exists so this is enforced at the construction
643    /// boundary, not re-checked at every call site.
644    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    /// Borrow the owner-supplied [`ContextProvider`], if any.
653    ///
654    /// Returns `Some` only for [`CompositionBinding::OwnerProvided`].
655    /// `Standalone` has no dispatcher; `Wired` has a dispatcher but no
656    /// owner-supplied context, so callers that walk route bindings and
657    /// encounter a `ContextField` source on a `Wired` binding should
658    /// surface a typed refusal rather than silently treat it as an
659    /// empty context.
660    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
668/// Default catalog-backed dispatcher.
669///
670/// Consumes a [`RouteTable`] (built from a
671/// [`meerkat_machine_schema::CompositionSchema`]) plus a map of consumer
672/// surfaces keyed by [`MachineInstanceId`]. Every routed effect goes
673/// through the same three steps:
674///
675/// 1. Look up the input-kind route for `(producer.instance_id, effect.variant)`.
676/// 2. Project the producer's field values into the consumer-field bindings.
677/// 3. Deliver via the consumer surface registered for the target instance.
678///
679/// No step has a silent-drop fallback. Unresolved routes, signal-kind
680/// targets, missing producer fields, and unwired consumers are all typed
681/// [`DispatchRefusal`] errors.
682pub 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
689/// Default catalog-backed signal dispatcher.
690///
691/// This is the signal-kind mirror of [`CatalogCompositionDispatcher`]:
692/// it consumes the same [`RouteTable`] but resolves through the signal
693/// index and delivers to [`SignalConsumerSurface`].
694pub 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    /// Build a new dispatcher for `composition`, using `table` as the typed
723    /// route index.
724    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    /// Register a consumer surface for a target instance.
734    ///
735    /// Panics are impossible — duplicate registrations replace the prior
736    /// entry. (Duplicate wiring is a construction bug; the callers in
737    /// wave-b prove registration happens exactly once per instance in the
738    /// composition schema.)
739    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    /// Build a new signal dispatcher for `composition`.
748    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    /// Register a signal consumer surface for a target instance.
758    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    /// Hand-written stand-in for the codegen-emitted `MeerkatMobSeamEffect`
916    /// sum. Matches the shape the B-4b tests pin for the live catalog:
917    /// one variant per producer instance, each wrapping a typed effect
918    /// body (we cover the `RequestRuntimeBinding` arm for the dispatcher
919    /// path).
920    #[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    /// Hand-written stand-in for the codegen-emitted signal source sum.
963    /// These are the MeerkatMachine routed lifecycle effects that the
964    /// `meerkat_mob_seam` schema routes to MobMachine signals.
965    #[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            // leak is fine in tests: we want a 'static reference; the
1024            // instance_id is stable for the lifetime of the test binary.
1025            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        // The schema has no route for `Mob::UnknownEffect`; use the well-
1300        // formed producer but label the variant with an id that has no
1301        // declared route.
1302        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        // Build a dispatcher with NO consumer surface registered. The route
1323        // resolves but the delivery step must return UnwiredConsumer, not
1324        // silently succeed.
1325        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    /// Owner-supplied context provider for routes that need typed
1359    /// fields not in the producer effect body. In production this would
1360    /// be a runtime-owned struct (e.g. one carrying a pinned
1361    /// `SessionId`); the test just returns a canned pair to exercise
1362    /// the trait's single-method signature.
1363    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        // The typed provider returns the expected single owner-supplied
1401        // field. Matches the #342 use case: `session_id` is absent from
1402        // the producer effect body but present in the projected fields
1403        // the consumer needs.
1404        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        // `CompositionBinding::standalone()` is the binding-level mirror
1419        // of `MeerkatMachine::standalone(...)`; `wired_with` and
1420        // `owner_provided` mirror `MeerkatMachine::with_composition(...)`.
1421        // The constructor split exists so call sites say positively which
1422        // half they are wiring, rather than spelling the enum variant.
1423        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    /// Consumer surface that always refuses with a typed [`ConsumerError`]
1448    /// carrying a known stable `error_code`. Row #33 gate fixture.
1449    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    /// Row #33 gate: a consumer refusal must preserve the consumer's typed
1471    /// `error_code` through the dispatcher. Under the OLD `Result<(), String>`
1472    /// contract the discriminant was flattened into an opaque message; the
1473    /// dispatcher could only re-parse a string. This asserts the typed code
1474    /// survives on `DispatchRefusal::ConsumerRefused`.
1475    #[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}