lenso_native_adapter/
lib.rs

1//! Native Rust Execution Adapter for statically linked Plugin packages.
2
3mod authoring;
4mod managed_tasks;
5
6use std::{collections::BTreeMap, rc::Rc};
7
8#[doc(hidden)]
9pub use authoring::{CompleteObjectLifecycle, ConstructionContext, LifecycleContext, PluginObject};
10#[doc(hidden)]
11pub use inventory as __inventory;
12use lenso_app_plan::{
13    ExecutionClassId, ResolvedAppPlan,
14    authoring::{HostCatalog, HostDefaultPlugin, HostPluginRelease, HostSlot, PluginDescriptor},
15};
16use lenso_kernel::{ActivateContext, DeactivateContext, PrepareContext};
17pub use lenso_kernel::{CancellationToken, RuntimeFailure};
18pub use lenso_native_adapter_macros::{PluginConfig, plugin, plugin_impl, provides};
19pub use lenso_runtime_codec::InstanceResources;
20pub use managed_tasks::{ManagedTasks, ManagedTasksError};
21
22/// Optional convention-based lifecycle hooks for a struct-level Plugin.
23///
24/// Add `#[plugin(lifecycle)]`, implement this trait, and override only the
25/// phases that own real work. The generated Adapter lifecycle still connects
26/// declared Capability ports before `activate`.
27#[allow(async_fn_in_trait)]
28pub trait Lifecycle: Clone + 'static {
29    async fn prepare(&self, _context: PrepareContext) -> Result<(), RuntimeFailure> {
30        Ok(())
31    }
32
33    async fn activate(&self, _context: ActivateContext) -> Result<(), RuntimeFailure> {
34        Ok(())
35    }
36
37    async fn deactivate(&self, _context: DeactivateContext) -> Result<(), RuntimeFailure> {
38        Ok(())
39    }
40}
41
42/// Implementation details referenced by generated Plugin glue.
43#[doc(hidden)]
44pub mod __private {
45    pub use crate::authoring::{ErasedConstructionFuture, LinkedPluginConstruction};
46    pub use crate::{
47        __inventory, CompleteObjectLifecycle, ConstructionContext, Lifecycle, LifecycleContext,
48        LinkedNativePluginFactory, NativePluginFactory, NativePluginFactoryContext,
49        NativePluginInstance, PluginObject, RuntimeFailure,
50    };
51    pub use futures;
52    pub use futures::future::LocalBoxFuture;
53    pub use lenso_kernel::{
54        ActivateContext, DeactivateContext, InvocationContext, NativeEventEndpoint,
55        NativeRequestEndpoint, NativeRequestFuture, NativeStreamEndpoint, NativeStreamSession,
56        PluginFuture, PluginLifecycle, PrepareContext,
57    };
58    pub use lenso_plugin_authoring::{
59        BoundCapabilityClient, CapabilityClient, CapabilityClientMany,
60    };
61    pub use lenso_runtime_codec::InstanceResources;
62    pub use serde_json;
63}
64
65use lenso_kernel::{
66    NativeEndpointSet, NativeEventEndpoint, NativeExecutionAdapter, NativeRequestEndpoint,
67    NativeStreamEndpoint, NoopPluginLifecycle, PluginLifecycle, PreparedBinding,
68    PreparedEventBinding, PreparedNativeApp, PreparedNativePlugin, PreparedStreamBinding,
69};
70
71/// One native Plugin factory contributed to the Host's link-time catalog.
72#[derive(Clone, Copy, Debug)]
73#[doc(hidden)]
74pub struct LinkedNativePluginFactory {
75    constructor: fn() -> Rc<dyn NativePluginFactory>,
76    descriptor: &'static str,
77}
78
79impl LinkedNativePluginFactory {
80    /// Creates a link-time catalog record. Intended for generated authoring glue.
81    #[doc(hidden)]
82    pub const fn new(
83        constructor: fn() -> Rc<dyn NativePluginFactory>,
84        descriptor: &'static str,
85    ) -> Self {
86        Self {
87            constructor,
88            descriptor,
89        }
90    }
91}
92
93inventory::collect!(LinkedNativePluginFactory);
94
95/// Endpoints created for one statically linked Plugin Instance generation.
96#[derive(Debug)]
97pub struct NativePluginInstance {
98    endpoints: NativeEndpointSet,
99    lifecycle: Rc<dyn PluginLifecycle>,
100}
101
102impl NativePluginInstance {
103    /// Creates a generation from its exact declared endpoint set.
104    pub fn new(endpoints: Vec<Rc<dyn NativeRequestEndpoint>>) -> Self {
105        Self::with_lifecycle(endpoints, NoopPluginLifecycle)
106    }
107
108    /// Creates a generation with its exact endpoints and lifecycle Interface.
109    pub fn with_lifecycle(
110        endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
111        lifecycle: impl PluginLifecycle,
112    ) -> Self {
113        Self {
114            endpoints: NativeEndpointSet::new(endpoints, Vec::new(), Vec::new()),
115            lifecycle: Rc::new(lifecycle),
116        }
117    }
118
119    /// Creates a generation with request and bidirectional stream endpoints.
120    pub fn with_endpoints(
121        endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
122        stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
123        lifecycle: impl PluginLifecycle,
124    ) -> Self {
125        Self {
126            endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, Vec::new()),
127            lifecycle: Rc::new(lifecycle),
128        }
129    }
130
131    /// Creates a generation containing only bidirectional stream endpoints.
132    pub fn with_stream_endpoints(
133        stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
134        lifecycle: impl PluginLifecycle,
135    ) -> Self {
136        Self::with_endpoints(Vec::new(), stream_endpoints, lifecycle)
137    }
138
139    /// Creates a generation containing only ephemeral Event endpoints.
140    pub fn with_event_endpoints(
141        event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
142        lifecycle: impl PluginLifecycle,
143    ) -> Self {
144        Self {
145            endpoints: NativeEndpointSet::new(Vec::new(), Vec::new(), event_endpoints),
146            lifecycle: Rc::new(lifecycle),
147        }
148    }
149
150    /// Creates a generation with request, stream, and ephemeral Event endpoints.
151    pub fn with_all_endpoints(
152        endpoints: Vec<Rc<dyn NativeRequestEndpoint>>,
153        stream_endpoints: Vec<Rc<dyn NativeStreamEndpoint>>,
154        event_endpoints: Vec<Rc<dyn NativeEventEndpoint>>,
155        lifecycle: impl PluginLifecycle,
156    ) -> Self {
157        Self {
158            endpoints: NativeEndpointSet::new(endpoints, stream_endpoints, event_endpoints),
159            lifecycle: Rc::new(lifecycle),
160        }
161    }
162
163    /// Returns the lifecycle Interface for this generation.
164    pub fn lifecycle(&self) -> Rc<dyn PluginLifecycle> {
165        self.lifecycle.clone()
166    }
167
168    /// Returns the exact endpoint set created for this generation.
169    pub fn endpoints(&self) -> &[Rc<dyn NativeRequestEndpoint>] {
170        self.endpoints.request()
171    }
172
173    /// Returns the exact bidirectional stream endpoint set created for this generation.
174    pub fn stream_endpoints(&self) -> &[Rc<dyn NativeStreamEndpoint>] {
175        self.endpoints.stream()
176    }
177
178    /// Returns the exact ephemeral Event endpoint set created for this Instance.
179    pub fn event_endpoints(&self) -> &[Rc<dyn NativeEventEndpoint>] {
180        self.endpoints.event()
181    }
182}
183
184impl Default for NativePluginInstance {
185    fn default() -> Self {
186        Self::new(Vec::new())
187    }
188}
189
190/// Adapter-specific factory for a statically linked native Rust Plugin.
191pub trait NativePluginFactory: std::fmt::Debug + 'static {
192    /// Package identity selected by the Resolved App Plan.
193    fn package_id(&self) -> &'static str;
194    /// Exact statically linked Cargo package version.
195    fn package_version(&self) -> &'static str {
196        ""
197    }
198    /// Exact authoring/runtime protocol implemented by this factory.
199    fn runtime_profile(&self) -> &'static str {
200        "lenso.native-authoring@1"
201    }
202    /// Immutable factory identity advertised by the exact Host Build Manifest.
203    ///
204    /// Plugin-resolved Plans carry this value as their package revision. The
205    /// default keeps ordinary statically linked factories unique by package and
206    /// version while allowing a factory to override the identity when its build
207    /// authority is more specific than a Cargo package version.
208    fn factory_identity(&self) -> String {
209        let version = self.package_version();
210        if version.is_empty() {
211            self.package_id().to_owned()
212        } else {
213            format!("{}@{version}", self.package_id())
214        }
215    }
216    /// Creates a fresh Plugin Instance generation.
217    fn instantiate(
218        &self,
219        context: NativePluginFactoryContext<'_>,
220    ) -> Result<NativePluginInstance, RuntimeFailure>;
221}
222
223/// Immutable Plan input supplied when a native factory creates one generation.
224#[derive(Clone, Copy, Debug)]
225pub struct NativePluginFactoryContext<'a> {
226    instance_key: &'a str,
227    entrypoint: &'a str,
228    configuration: &'a str,
229    resources: &'a InstanceResources,
230}
231
232impl<'a> NativePluginFactoryContext<'a> {
233    fn from_plan(
234        instance: &'a lenso_app_plan::PluginInstancePlan,
235        resources: &'a InstanceResources,
236    ) -> Self {
237        Self {
238            instance_key: instance.instance_key(),
239            entrypoint: instance.entrypoint(),
240            configuration: instance.configuration(),
241            resources,
242        }
243    }
244
245    /// Returns the App-local Plugin Instance key.
246    pub const fn instance_key(self) -> &'a str {
247        self.instance_key
248    }
249
250    /// Returns the exact package entrypoint selected before boot.
251    pub const fn entrypoint(self) -> &'a str {
252        self.entrypoint
253    }
254
255    /// Returns opaque Plugin-owned configuration selected before boot.
256    pub const fn configuration(self) -> &'a str {
257        self.configuration
258    }
259
260    /// Returns immutable supporting files snapshotted for this Generation.
261    pub const fn resources(self) -> &'a InstanceResources {
262        self.resources
263    }
264}
265
266/// Statically linked native Plugin factories available to an App binary.
267#[derive(Debug, Default)]
268pub struct NativePluginRegistry {
269    factories: Vec<Rc<dyn NativePluginFactory>>,
270    resources: lenso_runtime_codec::InstanceResourceCatalog,
271}
272
273type NativeInstances = BTreeMap<String, NativePluginInstance>;
274type PreparedGenerations = BTreeMap<String, PreparedNativePlugin>;
275type NativeBindings = (
276    Vec<PreparedBinding>,
277    Vec<PreparedStreamBinding>,
278    Vec<PreparedEventBinding>,
279);
280
281fn factory_matches(
282    factory: &dyn NativePluginFactory,
283    expected: &lenso_app_plan::PluginInstancePlan,
284) -> bool {
285    factory.package_id() == expected.package_id()
286        && factory.runtime_profile() == expected.runtime_profile()
287        && (expected.package_revision().is_empty()
288            || factory.package_version() == expected.package_revision()
289            || factory.factory_identity() == expected.package_revision())
290}
291
292impl NativePluginRegistry {
293    /// Creates an empty linked-factory registry.
294    pub fn new() -> Self {
295        Self::default()
296    }
297
298    /// Adds every Plugin factory contributed to this Host at link time.
299    ///
300    /// This catalog describes code available in the binary. The Resolved App
301    /// Plan remains the sole authority that selects and binds Plugin Instances.
302    #[must_use]
303    pub fn with_linked_factories(mut self) -> Self {
304        self.factories.extend(
305            inventory::iter::<LinkedNativePluginFactory>
306                .into_iter()
307                .map(|linked| (linked.constructor)()),
308        );
309        self.factories
310            .sort_by_key(|factory| factory.factory_identity());
311        self
312    }
313
314    /// Injects exact Generation-bound supporting files for selected Instances.
315    #[must_use]
316    pub fn with_resources(
317        mut self,
318        resources: lenso_runtime_codec::InstanceResourceCatalog,
319    ) -> Self {
320        self.resources = resources;
321        self
322    }
323
324    /// Returns the exact native factories available to this registry.
325    pub fn factories(&self) -> impl Iterator<Item = &dyn NativePluginFactory> {
326        self.factories.iter().map(std::convert::AsRef::as_ref)
327    }
328
329    /// Builds the immutable Host Catalog declared by this binary and Host policy.
330    pub fn host_catalog(
331        slots: impl IntoIterator<Item = HostSlot>,
332        defaults: impl IntoIterator<Item = HostDefaultPlugin>,
333    ) -> Result<HostCatalog, RuntimeFailure> {
334        let plugins = inventory::iter::<LinkedNativePluginFactory>
335            .into_iter()
336            .map(|linked| {
337                serde_json::from_str::<PluginDescriptor>(linked.descriptor)
338                    .map(HostPluginRelease::new)
339                    .map_err(|error| RuntimeFailure::InvalidResolvedPlan {
340                        detail: format!("invalid linked Plugin Descriptor: {error}"),
341                    })
342            })
343            .collect::<Result<Vec<_>, _>>()?;
344        Ok(HostCatalog::new(slots, plugins, defaults))
345    }
346    /// Adds one statically linked factory.
347    #[must_use]
348    pub fn with_factory(mut self, factory: impl NativePluginFactory) -> Self {
349        self.factories.push(Rc::new(factory));
350        self
351    }
352
353    fn prepare_instances(
354        &self,
355        plan: &ResolvedAppPlan,
356    ) -> Result<(NativeInstances, PreparedGenerations), RuntimeFailure> {
357        let mut instances = BTreeMap::new();
358        let mut generations = BTreeMap::new();
359        for expected in plan
360            .plugin_instances()
361            .iter()
362            .filter(|instance| instance.execution_class() == &ExecutionClassId::native_rust())
363        {
364            let matching_factories: Vec<_> = self
365                .factories
366                .iter()
367                .filter(|factory| factory_matches(factory.as_ref(), expected))
368                .collect();
369            let factory = match matching_factories.as_slice() {
370                [] => {
371                    return Err(RuntimeFailure::MissingPluginFactory {
372                        instance: expected.instance_key().to_owned(),
373                        package_id: expected.package_id().to_owned(),
374                    });
375                }
376                [factory] => *factory,
377                _ => {
378                    return invalid(format!(
379                        "multiple statically linked factories declare package `{}`",
380                        expected.package_id()
381                    ));
382                }
383            };
384            let generation = factory.instantiate(NativePluginFactoryContext::from_plan(
385                expected,
386                self.resources.for_instance(expected.instance_key()),
387            ))?;
388            generations.insert(
389                expected.instance_key().to_owned(),
390                PreparedNativePlugin::with_endpoint_set_lifecycle(
391                    generation.endpoints.clone(),
392                    generation.lifecycle(),
393                ),
394            );
395            if instances
396                .insert(expected.instance_key().to_owned(), generation)
397                .is_some()
398            {
399                return invalid(format!(
400                    "duplicate Plugin Instance `{}`",
401                    expected.instance_key()
402                ));
403            }
404        }
405        Ok((instances, generations))
406    }
407}
408
409impl NativeExecutionAdapter for NativePluginRegistry {
410    fn supports_runtime_profile(&self, authoring_version: u32, profile: &str) -> bool {
411        matches!(
412            (authoring_version, profile),
413            (1, "lenso.native-authoring@1" | "lenso.native-rust@1")
414                | (2, "lenso.native-authoring@2")
415        )
416    }
417
418    fn prepare(&self, plan: &ResolvedAppPlan) -> Result<PreparedNativeApp, RuntimeFailure> {
419        plan.validate()
420            .map_err(|error| RuntimeFailure::InvalidResolvedPlan {
421                detail: error.to_string(),
422            })?;
423
424        let (instances, generations) = self.prepare_instances(plan)?;
425        let (bindings, stream_bindings, event_bindings) = prepare_bindings(plan, &instances)?;
426        Ok(PreparedNativeApp::new(bindings, generations)
427            .with_stream_bindings(stream_bindings)
428            .with_event_bindings(event_bindings))
429    }
430
431    fn recreate(
432        &self,
433        plan: &ResolvedAppPlan,
434        instance_key: &str,
435    ) -> Result<PreparedNativePlugin, RuntimeFailure> {
436        let expected = plan
437            .plugin_instances()
438            .iter()
439            .find(|instance| instance.instance_key() == instance_key)
440            .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
441                detail: format!("unknown Plugin Instance `{instance_key}`"),
442            })?;
443        let matching_factories: Vec<_> = self
444            .factories
445            .iter()
446            .filter(|factory| factory_matches(factory.as_ref(), expected))
447            .collect();
448        let factory = match matching_factories.as_slice() {
449            [] => {
450                return Err(RuntimeFailure::MissingPluginFactory {
451                    instance: expected.instance_key().to_owned(),
452                    package_id: expected.package_id().to_owned(),
453                });
454            }
455            [factory] => *factory,
456            _ => {
457                return invalid(format!(
458                    "multiple statically linked factories declare package `{}`",
459                    expected.package_id()
460                ));
461            }
462        };
463        let generation = factory.instantiate(NativePluginFactoryContext::from_plan(
464            expected,
465            self.resources.for_instance(expected.instance_key()),
466        ))?;
467        Ok(PreparedNativePlugin::with_endpoint_set_lifecycle(
468            generation.endpoints.clone(),
469            generation.lifecycle(),
470        ))
471    }
472}
473
474fn prepare_bindings(
475    plan: &ResolvedAppPlan,
476    instances: &NativeInstances,
477) -> Result<NativeBindings, RuntimeFailure> {
478    let mut bindings = Vec::new();
479    let mut stream_bindings = Vec::new();
480    let mut event_bindings = Vec::new();
481    for binding in plan.capability_bindings() {
482        if !instances.contains_key(binding.provider_instance()) {
483            continue;
484        }
485        let provider = plan
486            .plugin_instance(binding.provider_instance())
487            .expect("validated binding provider should exist");
488        let descriptor = provider
489            .provided_capabilities()
490            .iter()
491            .find(|descriptor| descriptor.capability_id() == binding.capability_id())
492            .expect("validated binding descriptor should exist");
493        if !descriptor.request_operations().is_empty() {
494            let endpoint = instances
495                .get(binding.provider_instance())
496                .and_then(|instance| {
497                    instance.endpoints.request().iter().find(|endpoint| {
498                        endpoint.capability_id() == binding.capability_id()
499                            && endpoint.descriptor_version() == binding.descriptor_version()
500                    })
501                })
502                .cloned()
503                .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
504                    detail: format!(
505                        "Capability `{}` version `{}` has no request endpoint on provider `{}`",
506                        binding.capability_id(),
507                        binding.descriptor_version(),
508                        binding.provider_instance()
509                    ),
510                })?;
511            bindings.push(
512                PreparedBinding::new(
513                    binding.consumer_instance(),
514                    binding.provider_instance(),
515                    endpoint,
516                )
517                .with_requirement_id(binding.requirement_id()),
518            );
519        }
520        if !descriptor.stream_operations().is_empty() {
521            let endpoint = instances
522                .get(binding.provider_instance())
523                .and_then(|instance| {
524                    instance.endpoints.stream().iter().find(|endpoint| {
525                        endpoint.capability_id() == binding.capability_id()
526                            && endpoint.descriptor_version() == binding.descriptor_version()
527                    })
528                })
529                .cloned()
530                .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
531                    detail: format!(
532                        "Capability `{}` version `{}` has no stream endpoint on provider `{}`",
533                        binding.capability_id(),
534                        binding.descriptor_version(),
535                        binding.provider_instance()
536                    ),
537                })?;
538            stream_bindings.push(
539                PreparedStreamBinding::new(
540                    binding.consumer_instance(),
541                    binding.provider_instance(),
542                    endpoint,
543                )
544                .with_requirement_id(binding.requirement_id()),
545            );
546        }
547        if !descriptor.event_operations().is_empty() {
548            let endpoint = instances
549                .get(binding.provider_instance())
550                .and_then(|instance| {
551                    instance.endpoints.event().iter().find(|endpoint| {
552                        endpoint.capability_id() == binding.capability_id()
553                            && endpoint.descriptor_version() == binding.descriptor_version()
554                    })
555                })
556                .cloned()
557                .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
558                    detail: format!(
559                        "Capability `{}` version `{}` has no Event endpoint on provider `{}`",
560                        binding.capability_id(),
561                        binding.descriptor_version(),
562                        binding.provider_instance()
563                    ),
564                })?;
565            event_bindings.push(
566                PreparedEventBinding::new(
567                    binding.consumer_instance(),
568                    binding.provider_instance(),
569                    endpoint,
570                )
571                .with_requirement_id(binding.requirement_id()),
572            );
573        }
574    }
575    Ok((bindings, stream_bindings, event_bindings))
576}
577
578fn invalid<T>(detail: String) -> Result<T, RuntimeFailure> {
579    Err(RuntimeFailure::InvalidResolvedPlan { detail })
580}
lenso_native_adapter/lib.rs

lenso_native_adapter/
lib.rs
