lenso_kernel/
request.rs

1use std::{any::Any, time::Duration};
2
3use super::{
4    CancellationToken, EventCapability, InvocationContext, LocalBoxFuture, NativeAppRuntime,
5    NativeEndpointBinding, NativeEventHandle, NativeRequestEndpoint, NativeRequestHandle,
6    NativeStreamEndpointBinding, NativeStreamHandle, PluginEventDependencyHandle, Rc, RefCell,
7    StreamCapability, Weak,
8};
9
10pub trait RequestCapability: 'static {
11    /// Typed request value.
12    type Request: 'static;
13    /// Typed success value.
14    type Response: 'static;
15    /// Typed Capability-defined error value.
16    type DomainError: 'static;
17    /// Stable Capability series identity.
18    const ID: &'static str;
19    /// Exact generated Descriptor version.
20    const DESCRIPTOR_VERSION: &'static str;
21
22    /// Invokes one native endpoint using the most specific binding available.
23    ///
24    /// Generated bindings override this hook with a typed path. Older bindings retain the
25    /// type-erased compatibility path without requiring regeneration.
26    #[doc(hidden)]
27    fn invoke_native(
28        endpoint: &dyn NativeRequestEndpoint,
29        operation: &str,
30        request: Self::Request,
31        context: InvocationContext,
32    ) -> NativeRequestFuture<Self>
33    where
34        Self: Sized,
35    {
36        invoke_typed_or_erased_native_request::<Self>(endpoint, operation, request, context)
37    }
38}
39
40/// A typed native request result before Kernel cancellation and supervision are applied.
41#[doc(hidden)]
42pub type NativeRequestFuture<C> = LocalBoxFuture<
43    'static,
44    Result<
45        Result<<C as RequestCapability>::Response, <C as RequestCapability>::DomainError>,
46        RuntimeFailure,
47    >,
48>;
49
50type TypedNativeRequestFn<C> =
51    dyn Fn(&str, <C as RequestCapability>::Request, InvocationContext) -> NativeRequestFuture<C>;
52
53/// Runtime-provided typed endpoint used when a request crosses an execution boundary.
54///
55/// Generated in-process endpoints may expose a more specific endpoint type. This generic
56/// carrier lets execution adapters preserve typed request, response, and domain-error values
57/// without routing them through `Box<dyn Any>`.
58#[doc(hidden)]
59pub struct TypedNativeRequestEndpoint<C: RequestCapability> {
60    invoke: Rc<TypedNativeRequestFn<C>>,
61}
62
63impl<C: RequestCapability> TypedNativeRequestEndpoint<C> {
64    /// Creates a typed endpoint around one runtime-owned dispatcher.
65    pub fn new(
66        invoke: impl Fn(&str, C::Request, InvocationContext) -> NativeRequestFuture<C> + 'static,
67    ) -> Self {
68        Self {
69            invoke: Rc::new(invoke),
70        }
71    }
72
73    /// Dispatches one request without type erasure.
74    pub fn invoke(
75        &self,
76        operation: &str,
77        request: C::Request,
78        context: InvocationContext,
79    ) -> NativeRequestFuture<C> {
80        (self.invoke)(operation, request, context)
81    }
82}
83
84impl<C: RequestCapability> std::fmt::Debug for TypedNativeRequestEndpoint<C> {
85    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86        formatter
87            .debug_struct("TypedNativeRequestEndpoint")
88            .field("capability", &C::ID)
89            .finish_non_exhaustive()
90    }
91}
92
93/// Uses a runtime-provided typed endpoint when present, retaining erased compatibility.
94#[doc(hidden)]
95pub fn invoke_typed_or_erased_native_request<C: RequestCapability>(
96    endpoint: &dyn NativeRequestEndpoint,
97    operation: &str,
98    request: C::Request,
99    context: InvocationContext,
100) -> NativeRequestFuture<C> {
101    if let Some(endpoint) = endpoint
102        .typed_endpoint()
103        .and_then(|endpoint| endpoint.downcast_ref::<TypedNativeRequestEndpoint<C>>())
104    {
105        endpoint.invoke(operation, request, context)
106    } else {
107        invoke_erased_native_request::<C>(endpoint, operation, request, context)
108    }
109}
110
111/// Compatibility dispatcher used by generated bindings when a typed endpoint is unavailable.
112#[doc(hidden)]
113pub fn invoke_erased_native_request<C: RequestCapability>(
114    endpoint: &dyn NativeRequestEndpoint,
115    operation: &str,
116    request: C::Request,
117    context: InvocationContext,
118) -> NativeRequestFuture<C> {
119    let invocation = endpoint.invoke(operation, Box::new(request), context);
120    Box::pin(async move {
121        match invocation.await? {
122            Ok(value) => value
123                .downcast::<C::Response>()
124                .map(|value| Ok(*value))
125                .map_err(|_| RuntimeFailure::ProtocolViolation { capability: C::ID }),
126            Err(value) => value
127                .downcast::<C::DomainError>()
128                .map(|value| Err(*value))
129                .map_err(|_| RuntimeFailure::ProtocolViolation { capability: C::ID }),
130        }
131    })
132}
133
134/// Kernel-generated identity for one logical request invocation.
135pub type RequestId = u64;
136
137/// Runtime-owned failure, kept separate from Capability-defined Domain Errors.
138#[derive(Clone, Debug, Eq, PartialEq)]
139pub enum RuntimeFailure {
140    /// The consumer has no resolved binding for this Capability.
141    Unavailable { capability: &'static str },
142    /// The bound provider does not declare the requested Operation.
143    UnknownOperation {
144        capability: &'static str,
145        operation: String,
146    },
147    /// A singular generated client was used for a requirement with many providers.
148    AmbiguousBinding {
149        capability: &'static str,
150        providers: usize,
151    },
152    /// Generated native types disagreed with the prepared endpoint.
153    ProtocolViolation { capability: &'static str },
154    /// A package selected by the Plan was not linked into the native App.
155    MissingPluginFactory {
156        instance: String,
157        package_id: String,
158    },
159    /// No installed Execution Adapter provides the class selected by one Instance.
160    UnavailableExecutionClass {
161        instance_key: String,
162        execution_class: String,
163    },
164    /// The Resolved Plan or prepared endpoint set is internally inconsistent.
165    InvalidResolvedPlan { detail: String },
166    /// New request admission was closed because the App is shutting down.
167    AdmissionClosed,
168    /// The request could not enter a full bounded admission queue.
169    ResourceExhausted {
170        capability: &'static str,
171        operation: String,
172    },
173    /// The invocation deadline expired before the request completed.
174    DeadlineExceeded { request_id: RequestId },
175    /// The caller cancelled the invocation before it completed.
176    Cancelled { request_id: RequestId },
177    /// The Runtime Driver or Adapter reported an internal execution failure.
178    Internal { detail: String },
179    /// A Plugin generation reported a failure that should trigger supervision.
180    PluginFailure { detail: String },
181    /// A Plugin Instance exhausted its finite restart budget.
182    PluginRestartExhausted { instance: String, attempts: usize },
183}
184
185/// The lifecycle phase represented by a Plugin context.
186#[derive(Clone, Copy, Debug, Eq, PartialEq)]
187pub enum PluginLifecyclePhase {
188    /// The Plugin may validate configuration and reserve reversible resources.
189    Prepare,
190    /// Adapter lowering constructs the complete inert Plugin object.
191    Construct,
192    /// The Plugin may initialize against already prepared dependencies.
193    Activate,
194    /// The App Ready Gate has opened and externally triggered work may begin.
195    Ready,
196    /// The Plugin must release work and resources owned by this generation.
197    Deactivate,
198}
199
200#[cfg(test)]
201mod typed_endpoint_tests {
202    use std::any::Any;
203
204    use super::*;
205
206    #[derive(Debug)]
207    struct Echo;
208
209    impl RequestCapability for Echo {
210        type Request = u64;
211        type Response = u64;
212        type DomainError = ();
213        const ID: &'static str = "test.echo@1";
214        const DESCRIPTOR_VERSION: &'static str = "1.0.0";
215    }
216
217    #[derive(Debug)]
218    struct Endpoint {
219        typed: TypedNativeRequestEndpoint<Echo>,
220    }
221
222    impl NativeRequestEndpoint for Endpoint {
223        fn capability_id(&self) -> &'static str {
224            Echo::ID
225        }
226
227        fn descriptor_version(&self) -> &'static str {
228            Echo::DESCRIPTOR_VERSION
229        }
230
231        fn operations(&self) -> &'static [&'static str] {
232            &["echo"]
233        }
234
235        fn typed_endpoint(&self) -> Option<&dyn Any> {
236            Some(&self.typed)
237        }
238
239        fn invoke(
240            &self,
241            _operation: &str,
242            _request: Box<dyn Any>,
243            _context: InvocationContext,
244        ) -> LocalBoxFuture<'static, Result<crate::ErasedDomainResult, RuntimeFailure>> {
245            panic!("typed dispatch must not call the erased endpoint")
246        }
247    }
248
249    #[test]
250    fn default_dispatch_uses_runtime_typed_endpoint() {
251        let endpoint = Endpoint {
252            typed: TypedNativeRequestEndpoint::new(|_, request, _| {
253                Box::pin(futures::future::ready(Ok(Ok(request + 1))))
254            }),
255        };
256        let context = InvocationContext::new(1, None, CancellationToken::new());
257
258        let result =
259            futures::executor::block_on(Echo::invoke_native(&endpoint, "echo", 41, context));
260
261        assert_eq!(result, Ok(Ok(42)));
262    }
263}
264
265/// A deterministic dependency visible to one Plugin Instance.
266#[derive(Clone, Debug)]
267pub struct PluginDependency {
268    pub(super) requirement_id: String,
269    pub(super) capability_id: String,
270    pub(super) provider_instance: String,
271    pub(super) provider_order: usize,
272    pub(super) handle: Option<PluginDependencyHandle>,
273    pub(super) stream_handle: Option<PluginStreamDependencyHandle>,
274    pub(super) event_handle: Option<PluginEventDependencyHandle>,
275}
276
277impl PluginDependency {
278    pub(super) fn new(
279        requirement_id: impl Into<String>,
280        capability_id: impl Into<String>,
281        provider_instance: impl Into<String>,
282        provider_order: usize,
283        handle: Option<PluginDependencyHandle>,
284        stream_handle: Option<PluginStreamDependencyHandle>,
285        event_handle: Option<PluginEventDependencyHandle>,
286    ) -> Self {
287        Self {
288            requirement_id: requirement_id.into(),
289            capability_id: capability_id.into(),
290            provider_instance: provider_instance.into(),
291            provider_order,
292            handle,
293            stream_handle,
294            event_handle,
295        }
296    }
297
298    /// Returns the Capability required by this dependency.
299    pub fn requirement_id(&self) -> &str {
300        &self.requirement_id
301    }
302
303    /// Returns the Capability required by this dependency.
304    pub fn capability_id(&self) -> &str {
305        &self.capability_id
306    }
307
308    /// Returns the App-local provider Instance key.
309    pub fn provider_instance(&self) -> &str {
310        &self.provider_instance
311    }
312
313    /// Returns the deterministic provider order for a `many` binding.
314    pub const fn provider_order(&self) -> usize {
315        self.provider_order
316    }
317
318    /// Returns the resolved native endpoint handle when the Adapter supplied one.
319    pub fn handle(&self) -> Option<PluginDependencyHandle> {
320        self.handle.clone()
321    }
322
323    /// Returns the resolved native stream endpoint handle when the Adapter supplied one.
324    pub fn stream_handle(&self) -> Option<PluginStreamDependencyHandle> {
325        self.stream_handle.clone()
326    }
327
328    /// Returns the resolved native Event endpoint handle when the Adapter supplied one.
329    pub fn event_handle(&self) -> Option<PluginEventDependencyHandle> {
330        self.event_handle.clone()
331    }
332}
333
334/// An opaque, Adapter-resolved Capability endpoint passed to lifecycle code.
335#[derive(Clone, Debug)]
336pub struct PluginDependencyHandle {
337    pub(super) binding: NativeEndpointBinding,
338    pub(super) caller_instance: String,
339    pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
340}
341
342/// An opaque, Adapter-resolved stream Capability endpoint passed to lifecycle code.
343#[derive(Clone, Debug)]
344pub struct PluginStreamDependencyHandle {
345    pub(super) binding: NativeStreamEndpointBinding,
346    pub(super) caller_instance: String,
347    pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
348}
349
350impl PluginStreamDependencyHandle {
351    /// Returns the Capability implemented by this stream handle.
352    pub fn capability_id(&self) -> &'static str {
353        self.binding.state.capability_id
354    }
355
356    /// Returns the exact Descriptor version implemented by this stream handle.
357    pub fn descriptor_version(&self) -> &'static str {
358        self.binding.state.descriptor_version
359    }
360
361    /// Returns the exact stream Operation table implemented by this handle.
362    pub fn operations(&self) -> &'static [&'static str] {
363        self.binding.state.operations
364    }
365
366    /// Derives a uniquely identified child context whose cancellation observes
367    /// its parent without allowing the child Stream to cancel that parent.
368    pub fn child_context(
369        &self,
370        context: InvocationContext,
371    ) -> Result<InvocationContext, RuntimeFailure> {
372        let runtime = self
373            .runtime
374            .borrow()
375            .upgrade()
376            .ok_or(RuntimeFailure::AdmissionClosed)?;
377        let request_id = runtime.request_ids.get();
378        runtime.request_ids.set(request_id.saturating_add(1));
379        Ok(context.for_child_request(request_id))
380    }
381
382    /// Converts this resolved dependency into its generated typed stream handle.
383    pub fn typed<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
384        if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
385            return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
386        }
387        let runtime = self
388            .runtime
389            .borrow()
390            .upgrade()
391            .ok_or(RuntimeFailure::AdmissionClosed)?;
392        Ok(NativeStreamHandle::from_endpoints(
393            std::slice::from_ref(&self.binding),
394            runtime,
395            &self.caller_instance,
396            true,
397        ))
398    }
399}
400
401impl PluginDependencyHandle {
402    /// Derives a uniquely identified child Request whose cancellation observes
403    /// its parent without allowing the child to cancel that parent.
404    pub fn child_context(
405        &self,
406        context: InvocationContext,
407    ) -> Result<InvocationContext, RuntimeFailure> {
408        let runtime = self
409            .runtime
410            .borrow()
411            .upgrade()
412            .ok_or(RuntimeFailure::AdmissionClosed)?;
413        let request_id = runtime.request_ids.get();
414        runtime.request_ids.set(request_id.saturating_add(1));
415        Ok(context.for_child_request(request_id))
416    }
417
418    /// Returns the Capability implemented by this handle.
419    pub fn capability_id(&self) -> &'static str {
420        self.binding.state.capability_id
421    }
422
423    /// Returns the exact Descriptor version implemented by this handle.
424    pub fn descriptor_version(&self) -> &'static str {
425        self.binding.state.descriptor_version
426    }
427
428    /// Returns the exact Operation table implemented by this handle.
429    pub fn operations(&self) -> &'static [&'static str] {
430        self.binding.state.operations
431    }
432
433    /// Invokes this exact resolved dependency through Kernel admission and supervision.
434    ///
435    /// Language Execution Adapters use this after their generated codec has decoded a
436    /// portable request into the provider's native erased value.
437    pub fn invoke_erased(
438        &self,
439        operation: &str,
440        request: Box<dyn Any>,
441        context: InvocationContext,
442    ) -> LocalBoxFuture<'static, Result<crate::ErasedDomainResult, RuntimeFailure>> {
443        let Some(runtime) = self.runtime.borrow().upgrade() else {
444            return Box::pin(futures::future::ready(Err(RuntimeFailure::AdmissionClosed)));
445        };
446        crate::request_handle::invoke_erased_dependency(
447            self.binding.clone(),
448            runtime,
449            self.caller_instance.clone(),
450            operation.to_owned(),
451            context,
452            request,
453        )
454    }
455
456    /// Converts this resolved dependency into its generated typed request handle.
457    pub fn typed<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
458        if self.capability_id() != C::ID || self.descriptor_version() != C::DESCRIPTOR_VERSION {
459            return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
460        }
461        let runtime = self
462            .runtime
463            .borrow()
464            .upgrade()
465            .ok_or(RuntimeFailure::AdmissionClosed)?;
466        Ok(NativeRequestHandle::from_endpoints(
467            std::slice::from_ref(&self.binding),
468            runtime,
469            &self.caller_instance,
470            true,
471        ))
472    }
473}
474
475/// The explicit Capability dependencies available during Plugin lifecycle.
476#[derive(Clone, Debug, Default)]
477pub struct PluginDependencies {
478    pub(super) requirements: Vec<lenso_app_plan::CapabilityRequirementPlan>,
479    pub(super) bindings: Vec<PluginDependency>,
480    pub(super) caller_instance: Rc<str>,
481    pub(super) runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
482}
483
484impl PluginDependencies {
485    pub(super) fn new(
486        caller_instance: impl Into<String>,
487        runtime: Rc<RefCell<Weak<NativeAppRuntime>>>,
488        requirements: Vec<lenso_app_plan::CapabilityRequirementPlan>,
489    ) -> Self {
490        Self {
491            requirements,
492            bindings: Vec::new(),
493            caller_instance: Rc::from(caller_instance.into()),
494            runtime,
495        }
496    }
497
498    /// Returns dependencies in the order materialized by the Resolved App Plan.
499    pub fn requirements(&self) -> &[lenso_app_plan::CapabilityRequirementPlan] {
500        &self.requirements
501    }
502
503    /// Narrows this view to one declared dependency, retaining optional absence.
504    pub fn requirement(&self, id: &str) -> Result<Self, RuntimeFailure> {
505        let requirement = self
506            .requirements
507            .iter()
508            .find(|requirement| requirement.requirement_id() == id)
509            .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
510                detail: format!(
511                    "consumer `{}` has no requirement `{id}`",
512                    self.caller_instance
513                ),
514            })?;
515        Ok(Self {
516            requirements: vec![requirement.clone()],
517            bindings: self
518                .bindings
519                .iter()
520                .filter(|binding| binding.requirement_id() == id)
521                .cloned()
522                .collect(),
523            caller_instance: self.caller_instance.clone(),
524            runtime: self.runtime.clone(),
525        })
526    }
527
528    fn validate_lookup(
529        &self,
530        capability: &'static str,
531        version: &str,
532    ) -> Result<(), RuntimeFailure> {
533        let declarations = self
534            .requirements
535            .iter()
536            .filter(|requirement| requirement.capability_id() == capability)
537            .collect::<Vec<_>>();
538        match declarations.as_slice() {
539            [] => Err(RuntimeFailure::Unavailable { capability }),
540            [declaration] if declaration.descriptor_version() == version => Ok(()),
541            [_] => Err(RuntimeFailure::ProtocolViolation { capability }),
542            declarations => Err(RuntimeFailure::AmbiguousBinding {
543                capability,
544                providers: declarations.len(),
545            }),
546        }
547    }
548
549    /// Returns dependencies in the order materialized by the Resolved App Plan.
550    pub fn bindings(&self) -> &[PluginDependency] {
551        &self.bindings
552    }
553
554    /// Returns the number of explicit dependencies.
555    pub fn len(&self) -> usize {
556        self.bindings.len()
557    }
558
559    /// Returns whether this Plugin has no explicit dependencies.
560    pub fn is_empty(&self) -> bool {
561        self.bindings.is_empty()
562    }
563
564    /// Creates a Kernel Invocation Context for work initiated by this Plugin.
565    ///
566    /// The request identity and monotonic deadline come from the same Runtime
567    /// Driver as the App. The context is still owned by the caller and its
568    /// cancellation token remains explicit.
569    pub fn invocation_context(
570        &self,
571        deadline: Option<Duration>,
572        cancellation: CancellationToken,
573    ) -> Result<InvocationContext, RuntimeFailure> {
574        let runtime = self
575            .runtime
576            .borrow()
577            .upgrade()
578            .ok_or(RuntimeFailure::AdmissionClosed)?;
579        let request_id = runtime.request_ids.get();
580        runtime.request_ids.set(request_id.saturating_add(1));
581        Ok(InvocationContext::new(request_id, deadline, cancellation)
582            .with_shared_caller_instance(self.caller_instance.clone()))
583    }
584
585    /// Creates a Plugin Invocation Context with a Driver-relative deadline.
586    pub fn invocation_context_after(
587        &self,
588        timeout: Duration,
589        cancellation: CancellationToken,
590    ) -> Result<InvocationContext, RuntimeFailure> {
591        let runtime = self
592            .runtime
593            .borrow()
594            .upgrade()
595            .ok_or(RuntimeFailure::AdmissionClosed)?;
596        let deadline = (runtime.driver.now)().saturating_add(timeout);
597        drop(runtime);
598        self.invocation_context(Some(deadline), cancellation)
599    }
600
601    pub(super) fn shutdown_invocation_context(
602        &self,
603        deadline: Option<Duration>,
604        cancellation: CancellationToken,
605    ) -> Result<InvocationContext, RuntimeFailure> {
606        self.invocation_context(deadline, cancellation)
607            .map(InvocationContext::for_shutdown_dependency_call)
608    }
609
610    /// Returns the one explicitly bound typed dependency.
611    pub fn one<C: RequestCapability>(&self) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
612        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
613        let handles: Vec<_> = self
614            .bindings
615            .iter()
616            .filter(|binding| binding.capability_id() == C::ID)
617            .filter_map(PluginDependency::handle)
618            .collect();
619        match handles.as_slice() {
620            [handle] => handle.typed::<C>(),
621            [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
622            handles => Err(RuntimeFailure::AmbiguousBinding {
623                capability: C::ID,
624                providers: handles.len(),
625            }),
626        }
627    }
628
629    /// Returns an optional explicitly bound typed dependency.
630    pub fn optional<C: RequestCapability>(
631        &self,
632    ) -> Result<Option<NativeRequestHandle<C>>, RuntimeFailure> {
633        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
634        match self
635            .bindings
636            .iter()
637            .filter(|binding| binding.capability_id() == C::ID)
638            .filter_map(PluginDependency::handle)
639            .collect::<Vec<_>>()
640            .as_slice()
641        {
642            [] => Ok(None),
643            [handle] => handle.typed::<C>().map(Some),
644            handles => Err(RuntimeFailure::AmbiguousBinding {
645                capability: C::ID,
646                providers: handles.len(),
647            }),
648        }
649    }
650
651    /// Returns all explicitly bound typed dependencies in resolved provider order.
652    pub fn many<C: RequestCapability>(
653        &self,
654    ) -> Result<Vec<NativeRequestHandle<C>>, RuntimeFailure> {
655        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
656        self.bindings
657            .iter()
658            .filter(|binding| binding.capability_id() == C::ID)
659            .filter_map(PluginDependency::handle)
660            .map(|handle| handle.typed::<C>())
661            .collect()
662    }
663
664    /// Returns the one explicitly bound typed stream dependency.
665    pub fn one_stream<C: StreamCapability>(&self) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
666        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
667        let handles: Vec<_> = self
668            .bindings
669            .iter()
670            .filter(|binding| binding.capability_id() == C::ID)
671            .filter_map(PluginDependency::stream_handle)
672            .collect();
673        match handles.as_slice() {
674            [handle] => handle.typed::<C>(),
675            [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
676            handles => Err(RuntimeFailure::AmbiguousBinding {
677                capability: C::ID,
678                providers: handles.len(),
679            }),
680        }
681    }
682
683    /// Returns an optional explicitly bound typed stream dependency.
684    pub fn optional_stream<C: StreamCapability>(
685        &self,
686    ) -> Result<Option<NativeStreamHandle<C>>, RuntimeFailure> {
687        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
688        match self
689            .bindings
690            .iter()
691            .filter(|binding| binding.capability_id() == C::ID)
692            .filter_map(PluginDependency::stream_handle)
693            .collect::<Vec<_>>()
694            .as_slice()
695        {
696            [] => Ok(None),
697            [handle] => handle.typed::<C>().map(Some),
698            handles => Err(RuntimeFailure::AmbiguousBinding {
699                capability: C::ID,
700                providers: handles.len(),
701            }),
702        }
703    }
704
705    /// Returns all explicitly bound typed stream dependencies in Plan order.
706    pub fn many_stream<C: StreamCapability>(
707        &self,
708    ) -> Result<Vec<NativeStreamHandle<C>>, RuntimeFailure> {
709        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
710        self.bindings
711            .iter()
712            .filter(|binding| binding.capability_id() == C::ID)
713            .filter_map(PluginDependency::stream_handle)
714            .map(|handle| handle.typed::<C>())
715            .collect()
716    }
717
718    /// Returns one typed Event handle over every explicit binding in Plan order.
719    pub fn many_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
720        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
721        let handles: Vec<_> = self
722            .bindings
723            .iter()
724            .filter(|binding| binding.capability_id() == C::ID)
725            .filter_map(PluginDependency::event_handle)
726            .collect();
727        if handles.iter().any(|handle| {
728            handle.capability_id() != C::ID || handle.descriptor_version() != C::DESCRIPTOR_VERSION
729        }) {
730            return Err(RuntimeFailure::ProtocolViolation { capability: C::ID });
731        }
732        let runtime = self
733            .runtime
734            .borrow()
735            .upgrade()
736            .ok_or(RuntimeFailure::AdmissionClosed)?;
737        let endpoints = handles
738            .iter()
739            .map(|handle| handle.binding.clone())
740            .collect::<Vec<_>>();
741        Ok(NativeEventHandle::from_endpoints(
742            &endpoints,
743            runtime,
744            &self.caller_instance,
745            true,
746        ))
747    }
748
749    /// Returns the one explicitly bound typed Event dependency.
750    pub fn one_event<C: EventCapability>(&self) -> Result<NativeEventHandle<C>, RuntimeFailure> {
751        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
752        match self
753            .bindings
754            .iter()
755            .filter(|binding| binding.capability_id() == C::ID)
756            .filter_map(PluginDependency::event_handle)
757            .collect::<Vec<_>>()
758            .as_slice()
759        {
760            [handle] => handle.typed::<C>(),
761            [] => Err(RuntimeFailure::Unavailable { capability: C::ID }),
762            handles => Err(RuntimeFailure::AmbiguousBinding {
763                capability: C::ID,
764                providers: handles.len(),
765            }),
766        }
767    }
768
769    /// Returns an optional explicitly bound typed Event dependency.
770    pub fn optional_event<C: EventCapability>(
771        &self,
772    ) -> Result<Option<NativeEventHandle<C>>, RuntimeFailure> {
773        self.validate_lookup(C::ID, C::DESCRIPTOR_VERSION)?;
774        match self
775            .bindings
776            .iter()
777            .filter(|binding| binding.capability_id() == C::ID)
778            .filter_map(PluginDependency::event_handle)
779            .collect::<Vec<_>>()
780            .as_slice()
781        {
782            [] => Ok(None),
783            [handle] => handle.typed::<C>().map(Some),
784            handles => Err(RuntimeFailure::AmbiguousBinding {
785                capability: C::ID,
786                providers: handles.len(),
787            }),
788        }
789    }
790}
lenso_kernel/request.rs

lenso_kernel/
request.rs
