lenso_kernel/

runtime.rs

use super::{
    AppAdmission, AppReadyGate, BTreeMap, CancellationToken, Cell, DiagnosticEvent,
    DiagnosticShutdownOutcome, DiagnosticSource, DriverControl, DriverTask, Duration,
    EventCapability, ExecutionAdapterCatalog, InvocationContext, LocalBoxFuture,
    ManagedResourceScope, ManagedTask, ManagedTaskScope, NativeEventBindingTable,
    NativeEventEndpointStateTable, NativeEventHandle, NativeRequestEndpoint, NativeRequestHandle,
    NativeStreamBindingTable, NativeStreamEndpoint, NativeStreamEndpointStateTable,
    NativeStreamHandle, PluginCriticality, PluginDependencies, PluginLifecycle, Rc, RefCell,
    RequestAdmission, RequestCapability, RequestId, ResolvedAppPlan, RestartPolicy,
    RuntimeDiagnostics, RuntimeFailure, ShutdownOutcome, StreamCapability,
    begin_plugin_supervision, event, handle_supervision_schedule_failure, oneshot,
    schedule_plugin_supervision, shutdown_native_plugins,
};

#[derive(Clone, Debug)]
pub(super) struct NativeEndpointSnapshot {
    pub(super) endpoint: Rc<dyn NativeRequestEndpoint>,
    pub(super) generation: u64,
    pub(super) cancellation: CancellationToken,
}

#[derive(Debug)]
pub(super) struct NativeEndpointState {
    pub(super) capability_id: &'static str,
    pub(super) descriptor_version: &'static str,
    pub(super) operations: &'static [&'static str],
    pub(super) endpoint: RefCell<Option<Rc<dyn NativeRequestEndpoint>>>,
    pub(super) generation: Cell<u64>,
    pub(super) cancellation: RefCell<CancellationToken>,
}

#[derive(Clone, Debug)]
pub(crate) struct NativeStreamEndpointSnapshot {
    pub(crate) endpoint: Rc<dyn NativeStreamEndpoint>,
    pub(crate) generation: u64,
    pub(crate) cancellation: CancellationToken,
}

#[derive(Debug)]
pub(crate) struct NativeStreamEndpointState {
    pub(super) capability_id: &'static str,
    pub(super) descriptor_version: &'static str,
    pub(super) operations: &'static [&'static str],
    pub(super) endpoint: RefCell<Option<Rc<dyn NativeStreamEndpoint>>>,
    pub(super) generation: Cell<u64>,
    pub(super) cancellation: RefCell<CancellationToken>,
}

impl NativeStreamEndpointState {
    pub(crate) fn new(endpoint: Rc<dyn NativeStreamEndpoint>, generation: u64) -> Self {
        Self {
            capability_id: endpoint.capability_id(),
            descriptor_version: endpoint.descriptor_version(),
            operations: endpoint.operations(),
            endpoint: RefCell::new(Some(endpoint)),
            generation: Cell::new(generation),
            cancellation: RefCell::new(CancellationToken::new()),
        }
    }

    pub(crate) fn snapshot(&self) -> Option<NativeStreamEndpointSnapshot> {
        self.endpoint
            .borrow()
            .clone()
            .map(|endpoint| NativeStreamEndpointSnapshot {
                endpoint,
                generation: self.generation.get(),
                cancellation: self.cancellation.borrow().clone(),
            })
    }

    pub(crate) fn mark_unavailable(&self) {
        self.cancellation.borrow().cancel();
        self.endpoint.borrow_mut().take();
    }

    pub(crate) fn cancel(&self) {
        self.cancellation.borrow().cancel();
    }

    pub(crate) fn install(&self, endpoint: Rc<dyn NativeStreamEndpoint>, generation: u64) {
        self.generation.set(generation);
        self.cancellation.replace(CancellationToken::new());
        self.endpoint.replace(Some(endpoint));
    }

    pub(crate) fn is_current(&self, generation: u64) -> bool {
        self.generation.get() == generation && self.endpoint.borrow().is_some()
    }
}

impl NativeEndpointState {
    pub(super) fn new(endpoint: Rc<dyn NativeRequestEndpoint>, generation: u64) -> Self {
        Self {
            capability_id: endpoint.capability_id(),
            descriptor_version: endpoint.descriptor_version(),
            operations: endpoint.operations(),
            endpoint: RefCell::new(Some(endpoint)),
            generation: Cell::new(generation),
            cancellation: RefCell::new(CancellationToken::new()),
        }
    }

    pub(super) fn snapshot(&self) -> Option<NativeEndpointSnapshot> {
        self.endpoint
            .borrow()
            .clone()
            .map(|endpoint| NativeEndpointSnapshot {
                endpoint,
                generation: self.generation.get(),
                cancellation: self.cancellation.borrow().clone(),
            })
    }

    pub(super) fn mark_unavailable(&self) {
        self.cancellation.borrow().cancel();
        self.endpoint.borrow_mut().take();
    }

    pub(super) fn cancel(&self) {
        self.cancellation.borrow().cancel();
    }

    pub(super) fn install(&self, endpoint: Rc<dyn NativeRequestEndpoint>, generation: u64) {
        self.generation.set(generation);
        self.cancellation.replace(CancellationToken::new());
        self.endpoint.replace(Some(endpoint));
    }

    pub(super) fn is_current(&self, generation: u64) -> bool {
        self.generation.get() == generation && self.endpoint.borrow().is_some()
    }
}

#[derive(Clone, Debug)]
pub(super) struct NativeEndpointBinding {
    pub(super) requirement_id: String,
    pub(super) plugin_instance: String,
    pub(super) state: Rc<NativeEndpointState>,
    pub(super) admissions: BTreeMap<String, RequestAdmission>,
}

impl NativeEndpointBinding {
    pub(super) fn admission(&self, operation: &str) -> Option<&RequestAdmission> {
        self.admissions.get(operation)
    }
}

#[derive(Clone, Debug)]
pub(crate) struct NativeStreamEndpointBinding {
    pub(super) requirement_id: String,
    pub(crate) plugin_instance: String,
    pub(crate) state: Rc<NativeStreamEndpointState>,
    pub(super) admissions: BTreeMap<String, RequestAdmission>,
}

impl NativeStreamEndpointBinding {
    pub(crate) fn admission(&self, operation: &str) -> Option<&RequestAdmission> {
        self.admissions.get(operation)
    }
}

#[derive(Debug)]
pub(super) struct NativePluginGeneration {
    pub(super) lifecycle: Rc<dyn PluginLifecycle>,
    pub(super) tasks: ManagedTaskScope,
    pub(super) resources: ManagedResourceScope,
    pub(super) stop_attempted: bool,
    pub(super) cleanup_timed_out: bool,
    pub(super) staged_admission: Option<AppAdmission>,
}

pub(super) enum GenerationPreparationFailure {
    Lifecycle,
    Cleanup { primary: RuntimeFailure },
}

#[derive(Debug)]
pub(super) struct NativePluginRuntime {
    pub(super) generation: RefCell<Option<NativePluginGeneration>>,
}

impl NativePluginRuntime {
    pub(super) fn take_generation(&self) -> Option<NativePluginGeneration> {
        self.generation.borrow_mut().take()
    }

    pub(super) fn install_generation(&self, generation: NativePluginGeneration) {
        debug_assert!(self.generation.borrow().is_none());
        self.generation.replace(Some(generation));
    }

    pub(super) fn generation_parts(
        &self,
    ) -> Option<(
        Rc<dyn PluginLifecycle>,
        ManagedTaskScope,
        ManagedResourceScope,
    )> {
        self.generation.borrow().as_ref().map(|generation| {
            (
                generation.lifecycle.clone(),
                generation.tasks.clone(),
                generation.resources.clone(),
            )
        })
    }
}

#[derive(Clone, Debug)]
pub(super) struct PluginSupervision {
    pub(super) policy: RestartPolicy,
    pub(super) criticality: PluginCriticality,
    pub(super) required_path: bool,
    pub(super) generation: u64,
    pub(super) attempts: Vec<Duration>,
    pub(super) stable_since: Option<Duration>,
    pub(super) restarting: bool,
}

#[derive(Debug, Default)]
pub(super) struct ShutdownCoordinator {
    pub(super) started: Cell<bool>,
    pub(super) cleanup_started_at: Cell<Option<Duration>>,
    pub(super) completed: Cell<bool>,
    pub(super) outcome: RefCell<Option<ShutdownOutcome>>,
    pub(super) waiters: RefCell<Vec<oneshot::Sender<ShutdownOutcome>>>,
}

impl ShutdownCoordinator {
    pub(super) fn start(&self, started_at: Duration) -> bool {
        if self.started.replace(true) {
            return false;
        }
        self.cleanup_started_at.set(Some(started_at));
        true
    }

    pub(super) fn begin_completion(&self) -> bool {
        !self.completed.replace(true)
    }

    pub(super) fn publish(&self, outcome: &ShutdownOutcome) {
        self.outcome.replace(Some(outcome.clone()));
        for waiter in self.waiters.borrow_mut().drain(..) {
            let _ = waiter.send(outcome.clone());
        }
    }

    pub(super) fn wait(&self) -> LocalBoxFuture<'static, ShutdownOutcome> {
        if let Some(outcome) = self.outcome.borrow().clone() {
            return Box::pin(futures::future::ready(outcome));
        }
        let (complete, waiter) = oneshot::channel();
        self.waiters.borrow_mut().push(complete);
        Box::pin(async move {
            waiter.await.unwrap_or(ShutdownOutcome::RuntimeFailure {
                error: RuntimeFailure::Internal {
                    detail: "shutdown coordinator terminated before publishing an outcome"
                        .to_owned(),
                },
            })
        })
    }
}

pub(super) struct NativeAppRuntime {
    pub(super) startup_context: RefCell<Option<InvocationContext>>,
    pub(super) startup_cleanup: Option<super::cleanup::StartupCleanupBudget>,
    pub(super) cleanup_timeout: Option<Duration>,
    pub(super) executions: Rc<super::settlement::ExecutionLedger>,
    pub(super) plan: Rc<ResolvedAppPlan>,
    // Compiled topology above remains immutable; only a validated identical
    // topology successor can replace the active execution/configuration data.
    pub(super) snapshot: RefCell<Rc<ResolvedAppPlan>>,
    pub(super) transition_adapters: RefCell<BTreeMap<String, Rc<dyn super::ExecutionAdapter>>>,
    pub(super) transition_pending: Cell<bool>,
    pub(super) transition_calls: Rc<Cell<usize>>,
    pub(super) retired_generations: RefCell<Vec<NativePluginGeneration>>,
    pub(super) last_transition: RefCell<Option<super::TransitionOutcome>>,
    pub(super) adapters: Rc<ExecutionAdapterCatalog>,
    pub(super) plugins: BTreeMap<String, NativePluginRuntime>,
    pub(super) dependencies: BTreeMap<String, PluginDependencies>,
    pub(super) endpoint_states: BTreeMap<(String, String), Rc<NativeEndpointState>>,
    pub(super) stream_endpoint_states: NativeStreamEndpointStateTable,
    pub(super) event_endpoint_states: NativeEventEndpointStateTable,
    pub(super) supervision: RefCell<BTreeMap<String, PluginSupervision>>,
    pub(super) supervision_tasks: RefCell<BTreeMap<String, ManagedTask>>,
    pub(super) activation_order: Vec<String>,
    pub(super) ready_gate: AppReadyGate,
    pub(super) admission: AppAdmission,
    pub(super) driver: DriverControl,
    pub(super) diagnostics: RuntimeDiagnostics,
    pub(super) request_ids: Rc<Cell<RequestId>>,
    pub(super) supervision_cancellation: CancellationToken,
    pub(super) shutdown_started: Cell<bool>,
    pub(super) shutdown: ShutdownCoordinator,
    pub(super) shutdown_task: RefCell<Option<DriverTask>>,
    pub(super) terminal_failure: RefCell<Option<RuntimeFailure>>,
    pub(super) cleanup_failure: RefCell<Option<RuntimeFailure>>,
}

impl std::fmt::Debug for NativeAppRuntime {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("NativeAppRuntime")
            .field("plugin_count", &self.plugins.len())
            .field("endpoint_count", &self.endpoint_states.len())
            .field("stream_endpoint_count", &self.stream_endpoint_states.len())
            .field("event_endpoint_count", &self.event_endpoint_states.len())
            .field("ready", &self.ready_gate.is_open())
            .field("accepting", &self.admission.is_open())
            .field("next_request_id", &self.request_ids.get())
            .field("shutdown_started", &self.shutdown_started.get())
            .field("cleanup_started", &self.shutdown.started.get())
            .field("cleanup_completed", &self.shutdown.completed.get())
            .field(
                "terminal_failure",
                &self.terminal_failure.borrow().is_some(),
            )
            .finish_non_exhaustive()
    }
}

impl NativeAppRuntime {
    pub(super) fn record_cleanup_failure(&self, error: &RuntimeFailure) {
        // Optional Plugin failure need not close admission, but retirement must
        // retain cleanup evidence even after that generation has been removed.
        self.cleanup_failure
            .borrow_mut()
            .get_or_insert_with(|| error.clone());
    }

    pub(super) fn mark_plugin_endpoints_unavailable(&self, instance_key: &str) {
        for ((provider, _), endpoint) in &self.endpoint_states {
            if provider == instance_key {
                endpoint.mark_unavailable();
            }
        }
        for ((provider, _), endpoint) in &self.stream_endpoint_states {
            if provider == instance_key {
                endpoint.mark_unavailable();
            }
        }
        for ((provider, _), endpoint) in &self.event_endpoint_states {
            if provider == instance_key {
                endpoint.mark_unavailable();
            }
        }
    }

    pub(super) fn begin_shutdown(&self) {
        let admission_closed_at = (self.driver.now)();
        if self.shutdown_started.replace(true) {
            return;
        }
        self.admission.close();
        self.supervision_cancellation.cancel();
        for endpoint in self.endpoint_states.values() {
            endpoint.cancel();
        }
        for endpoint in self.stream_endpoint_states.values() {
            endpoint.cancel();
        }
        for endpoint in self.event_endpoint_states.values() {
            endpoint.cancel();
        }
        for plugin in self.plugins.values() {
            if let Some(generation) = plugin.generation.borrow().as_ref()
                && let Some(admission) = &generation.staged_admission
            {
                admission.close();
            }
            if let Some((_, tasks, resources)) = plugin.generation_parts() {
                tasks.close();
                resources.close();
            }
        }
        self.diagnostics
            .emit(DiagnosticSource::Shutdown, admission_closed_at, |_| {
                DiagnosticEvent::ShutdownAdmissionClosed
            });
    }

    pub(super) fn complete_shutdown(&self, outcome: &ShutdownOutcome) {
        if !self.shutdown.begin_completion() {
            return;
        }
        let completed_at = (self.driver.now)();
        let started_at = self
            .shutdown
            .cleanup_started_at
            .get()
            .unwrap_or(completed_at);
        let diagnostic_outcome = match outcome {
            ShutdownOutcome::Clean => DiagnosticShutdownOutcome::Clean,
            ShutdownOutcome::RuntimeFailure { .. } => DiagnosticShutdownOutcome::RuntimeFailure,
            ShutdownOutcome::Timeout => DiagnosticShutdownOutcome::Timeout,
        };
        self.diagnostics
            .emit(DiagnosticSource::Shutdown, completed_at, |_| {
                DiagnosticEvent::ShutdownCompleted {
                    outcome: diagnostic_outcome,
                    elapsed: completed_at.saturating_sub(started_at),
                }
            });
        if let ShutdownOutcome::RuntimeFailure { error } = outcome {
            self.diagnostics
                .emit_runtime_failure(completed_at, None, error);
        }
        self.shutdown.publish(outcome);
    }
}

/// A started native App whose generated clients can invoke resolved bindings.
#[derive(Clone, Debug)]
pub struct NativeApp {
    pub(super) bindings: BTreeMap<(String, &'static str), Vec<NativeEndpointBinding>>,
    pub(super) stream_bindings: NativeStreamBindingTable,
    pub(super) event_bindings: NativeEventBindingTable,
    pub(super) diagnostics: RuntimeDiagnostics,
    pub(super) runtime: Rc<NativeAppRuntime>,
}

impl NativeApp {
    /// Returns the active immutable snapshot, including applied config/revisions.
    pub fn plan_snapshot(&self) -> ResolvedAppPlan {
        self.runtime.snapshot.borrow().as_ref().clone()
    }

    /// Returns the most recent committed transition and its retirement outcome.
    pub fn last_transition(&self) -> Option<super::TransitionOutcome> {
        self.runtime.last_transition.borrow().clone()
    }
    fn diagnostic_failure<T>(
        &self,
        instance_key: Option<&str>,
        error: RuntimeFailure,
    ) -> Result<T, RuntimeFailure> {
        let instance_key = instance_key
            .filter(|instance_key| self.runtime.plan.plugin_instance(instance_key).is_some());
        self.runtime.diagnostics.emit_runtime_failure(
            (self.runtime.driver.now)(),
            instance_key,
            &error,
        );
        Err(error)
    }

    /// Confirms that a generated client has one resolved binding before use.
    pub fn ensure_binding<C: RequestCapability>(
        &self,
        caller_instance: &str,
    ) -> Result<(), RuntimeFailure> {
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        if self
            .endpoints::<C>(caller_instance)
            .is_some_and(|endpoints| !endpoints.is_empty())
        {
            return Ok(());
        }
        self.diagnostic_failure(
            Some(caller_instance),
            RuntimeFailure::Unavailable { capability: C::ID },
        )
    }

    /// Materializes one typed handle from the immutable binding selected by the Plan.
    pub fn handle<C: RequestCapability>(
        &self,
        caller_instance: &str,
    ) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
        self.validate_requirement_lookup(caller_instance, C::ID)?;
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        let Some(endpoints) = self
            .endpoints::<C>(caller_instance)
            .filter(|endpoints| !endpoints.is_empty())
        else {
            return self.diagnostic_failure(
                Some(caller_instance),
                RuntimeFailure::Unavailable { capability: C::ID },
            );
        };
        Ok(NativeRequestHandle::from_endpoints(
            endpoints,
            self.runtime.clone(),
            caller_instance,
            false,
        ))
    }

    /// Materializes an optional typed handle; an absent binding remains `None`.
    pub fn optional_handle<C: RequestCapability>(
        &self,
        caller_instance: &str,
    ) -> Option<NativeRequestHandle<C>> {
        self.validate_requirement_lookup(caller_instance, C::ID)
            .ok()?;
        let caller_instance = caller_instance.to_owned();
        self.endpoints::<C>(&caller_instance)
            .filter(|endpoints| !endpoints.is_empty())
            .map(|endpoints| {
                NativeRequestHandle::from_endpoints(
                    endpoints,
                    self.runtime.clone(),
                    &caller_instance,
                    false,
                )
            })
    }

    /// Materializes a typed handle whose endpoints may be empty for a `many` requirement.
    pub fn many_handle<C: RequestCapability>(
        &self,
        caller_instance: &str,
    ) -> Result<NativeRequestHandle<C>, RuntimeFailure> {
        self.validate_requirement_lookup(caller_instance, C::ID)?;
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        let endpoints = self.endpoints::<C>(caller_instance).unwrap_or(&[]);
        Ok(NativeRequestHandle::from_endpoints(
            endpoints,
            self.runtime.clone(),
            caller_instance,
            false,
        ))
    }

    /// Returns the number of immutable provider endpoints bound to one requirement.
    pub fn binding_count<C: RequestCapability>(&self, caller_instance: &str) -> usize {
        self.endpoints::<C>(caller_instance).map_or(0, <[_]>::len)
    }

    /// Returns whether every declared Plugin has completed activation.
    pub fn is_ready(&self) -> bool {
        self.runtime.ready_gate.is_open()
    }

    /// Returns the App-wide readiness signal observed by Plugin tasks.
    pub fn ready_gate(&self) -> AppReadyGate {
        self.runtime.ready_gate.clone()
    }

    /// Returns whether new externally triggered work may be admitted.
    pub fn is_accepting(&self) -> bool {
        self.runtime.admission.is_open()
    }

    /// Returns the App-wide admission state.
    pub fn admission(&self) -> AppAdmission {
        self.runtime.admission.clone()
    }

    /// Returns the opt-in Runtime Diagnostics port for this App.
    pub fn diagnostics(&self) -> RuntimeDiagnostics {
        self.diagnostics.clone()
    }

    /// Returns the exact resolved Capability dependencies for one Plugin Instance.
    ///
    /// Runners use this host-facing view when they must preserve provider identity
    /// while transferring one binding across an Execution Lane. Ordinary Plugin
    /// code receives the same Interface through its lifecycle context.
    pub fn dependencies(
        &self,
        caller_instance: &str,
    ) -> Result<PluginDependencies, RuntimeFailure> {
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        self.runtime
            .dependencies
            .get(caller_instance)
            .cloned()
            .ok_or_else(|| RuntimeFailure::InvalidResolvedPlan {
                detail: format!(
                    "Plugin Instance `{caller_instance}` has no resolved dependency table"
                ),
            })
    }

    /// Returns current bounded request queue depths grouped by provider Instance.
    ///
    /// This structural snapshot contains no request payloads and is intended for
    /// Runner-owned placement diagnostics.
    pub fn instance_queue_depths(&self) -> BTreeMap<String, usize> {
        let mut depths = BTreeMap::new();
        for endpoints in self.bindings.values() {
            for endpoint in endpoints {
                let depth = endpoint
                    .admissions
                    .values()
                    .map(RequestAdmission::queue_depth)
                    .sum::<usize>();
                *depths.entry(endpoint.plugin_instance.clone()).or_insert(0) += depth;
            }
        }
        depths
    }

    /// Returns the terminal supervision failure, when a critical App path exhausted its budget.
    pub fn terminal_failure(&self) -> Option<RuntimeFailure> {
        self.runtime.terminal_failure.borrow().clone()
    }

    /// Returns whether supervision has produced a terminal App failure.
    pub fn is_failed(&self) -> bool {
        self.runtime.terminal_failure.borrow().is_some()
    }

    /// Returns the current ready generation for one Plugin Instance, when it is available.
    pub fn plugin_generation(&self, instance_key: &str) -> Option<u64> {
        self.runtime
            .supervision
            .borrow()
            .get(instance_key)
            .and_then(|state| {
                let request_current =
                    self.runtime
                        .endpoint_states
                        .iter()
                        .any(|((plugin, _), endpoint)| {
                            plugin == instance_key && endpoint.is_current(state.generation)
                        });
                let stream_current =
                    self.runtime
                        .stream_endpoint_states
                        .iter()
                        .any(|((plugin, _), endpoint)| {
                            plugin == instance_key && endpoint.is_current(state.generation)
                        });
                let event_current =
                    self.runtime
                        .event_endpoint_states
                        .iter()
                        .any(|((plugin, _), endpoint)| {
                            plugin == instance_key && endpoint.is_current(state.generation)
                        });
                (request_current || stream_current || event_current).then_some(state.generation)
            })
    }

    /// Reports a Plugin Instance failure and schedules its finite supervision policy.
    pub fn report_plugin_failure(&self, instance_key: &str) -> Result<(), RuntimeFailure> {
        if !begin_plugin_supervision(&self.runtime, instance_key)? {
            return Ok(());
        }
        schedule_plugin_supervision(&self.runtime, instance_key).map_err(|error| {
            handle_supervision_schedule_failure(&self.runtime, instance_key, error)
        })
    }

    /// Starts shutdown admission closure and cooperative cancellation.
    pub fn request_shutdown(&self) {
        self.runtime.begin_shutdown();
    }

    /// Performs bounded graceful shutdown using one global deadline.
    pub async fn shutdown(&self, timeout: Duration) -> ShutdownOutcome {
        self.shutdown_with_budget(super::cleanup::CleanupBudget::after(
            &self.runtime.driver,
            timeout,
        ))
        .await
    }

    pub(super) async fn shutdown_with_budget(
        &self,
        budget: super::cleanup::CleanupBudget,
    ) -> ShutdownOutcome {
        self.runtime.begin_shutdown();
        let cleanup_started_at = (self.runtime.driver.now)();
        if self.runtime.shutdown.start(cleanup_started_at) {
            let timeout = budget.remaining();
            self.runtime
                .diagnostics
                .emit(DiagnosticSource::Shutdown, cleanup_started_at, |_| {
                    DiagnosticEvent::ShutdownCleanupStarted { timeout }
                });
            let runtime = self.runtime.clone();
            let worker_runtime = runtime.clone();
            match (runtime.driver.spawn_local)(Box::pin(async move {
                let outcome = shutdown_native_plugins(&worker_runtime, budget).await;
                worker_runtime.complete_shutdown(&outcome);
            })) {
                Ok(task) => {
                    runtime.shutdown_task.replace(Some(task));
                }
                Err(error) => {
                    runtime.complete_shutdown(&ShutdownOutcome::RuntimeFailure {
                        error: RuntimeFailure::Internal {
                            detail: format!("failed to schedule App shutdown: {error:?}"),
                        },
                    });
                }
            }
        }
        self.runtime.shutdown.wait().await
    }

    /// Invokes a generated request Operation through the caller's resolved binding.
    pub async fn invoke<C: RequestCapability>(
        &self,
        caller_instance: &str,
        operation: &str,
        request: C::Request,
    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
        self.handle::<C>(caller_instance)?
            .invoke(operation, request)
            .await
    }

    /// Creates a request context with a fresh Kernel Request ID.
    ///
    /// `deadline` is an absolute instant returned by the selected
    /// [`RuntimeDriver`]'s monotonic clock.
    pub fn invocation_context(
        &self,
        deadline: Option<Duration>,
        cancellation: CancellationToken,
    ) -> InvocationContext {
        InvocationContext::new(self.next_request_id(), deadline, cancellation)
    }

    /// Creates a request context whose deadline is relative to the Driver's clock.
    pub fn invocation_context_after(
        &self,
        timeout: Duration,
        cancellation: CancellationToken,
    ) -> InvocationContext {
        self.invocation_context(
            Some((self.runtime.driver.now)().saturating_add(timeout)),
            cancellation,
        )
    }

    /// Invokes a request with an explicit propagated Invocation Context.
    pub async fn invoke_with_context<C: RequestCapability>(
        &self,
        caller_instance: &str,
        operation: &str,
        context: InvocationContext,
        request: C::Request,
    ) -> Result<Result<C::Response, C::DomainError>, RuntimeFailure> {
        self.handle::<C>(caller_instance)?
            .invoke_with_context(operation, context, request)
            .await
    }

    /// Materializes one typed bidirectional stream handle from the resolved Plan.
    pub fn stream_handle<C: StreamCapability>(
        &self,
        caller_instance: &str,
    ) -> Result<NativeStreamHandle<C>, RuntimeFailure> {
        self.validate_requirement_lookup(caller_instance, C::ID)?;
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        let Some(endpoints) = self
            .stream_endpoints::<C>(caller_instance)
            .filter(|endpoints| !endpoints.is_empty())
        else {
            return self.diagnostic_failure(
                Some(caller_instance),
                RuntimeFailure::Unavailable { capability: C::ID },
            );
        };
        Ok(NativeStreamHandle::from_endpoints(
            endpoints,
            self.runtime.clone(),
            caller_instance,
            false,
        ))
    }

    /// Materializes an optional typed bidirectional stream handle.
    pub fn optional_stream_handle<C: StreamCapability>(
        &self,
        caller_instance: &str,
    ) -> Option<NativeStreamHandle<C>> {
        self.validate_requirement_lookup(caller_instance, C::ID)
            .ok()?;
        let caller_instance = caller_instance.to_owned();
        self.stream_endpoints::<C>(&caller_instance)
            .filter(|endpoints| !endpoints.is_empty())
            .map(|endpoints| {
                NativeStreamHandle::from_endpoints(
                    endpoints,
                    self.runtime.clone(),
                    &caller_instance,
                    false,
                )
            })
    }

    /// Returns the number of immutable stream endpoints bound to one requirement.
    pub fn stream_binding_count<C: StreamCapability>(&self, caller_instance: &str) -> usize {
        self.stream_endpoints::<C>(caller_instance)
            .map_or(0, <[_]>::len)
    }

    /// Materializes a typed Event handle and requires at least one subscriber.
    pub fn event_handle<C: EventCapability>(
        &self,
        caller_instance: &str,
    ) -> Result<NativeEventHandle<C>, RuntimeFailure> {
        self.validate_requirement_lookup(caller_instance, C::ID)?;
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        let Some(endpoints) = self
            .event_endpoints::<C>(caller_instance)
            .filter(|endpoints| !endpoints.is_empty())
        else {
            return self.diagnostic_failure(
                Some(caller_instance),
                RuntimeFailure::Unavailable { capability: C::ID },
            );
        };
        Ok(NativeEventHandle::from_endpoints(
            endpoints,
            self.runtime.clone(),
            caller_instance,
            false,
        ))
    }

    /// Materializes an optional typed Event handle.
    pub fn optional_event_handle<C: EventCapability>(
        &self,
        caller_instance: &str,
    ) -> Option<NativeEventHandle<C>> {
        self.validate_requirement_lookup(caller_instance, C::ID)
            .ok()?;
        let caller_instance = caller_instance.to_owned();
        self.event_endpoints::<C>(&caller_instance)
            .filter(|endpoints| !endpoints.is_empty())
            .map(|endpoints| {
                NativeEventHandle::from_endpoints(
                    endpoints,
                    self.runtime.clone(),
                    &caller_instance,
                    false,
                )
            })
    }

    /// Materializes a typed Event handle whose endpoint set may be empty.
    pub fn many_event_handle<C: EventCapability>(
        &self,
        caller_instance: &str,
    ) -> Result<NativeEventHandle<C>, RuntimeFailure> {
        self.validate_requirement_lookup(caller_instance, C::ID)?;
        if self.runtime.admission.is_closed() {
            return self.diagnostic_failure(Some(caller_instance), RuntimeFailure::AdmissionClosed);
        }
        let endpoints = self.event_endpoints::<C>(caller_instance).unwrap_or(&[]);
        Ok(NativeEventHandle::from_endpoints(
            endpoints,
            self.runtime.clone(),
            caller_instance,
            false,
        ))
    }

    /// Returns the number of immutable Event subscriber endpoints bound to a requirement.
    pub fn event_binding_count<C: EventCapability>(&self, caller_instance: &str) -> usize {
        self.event_endpoints::<C>(caller_instance)
            .map_or(0, <[_]>::len)
    }

    fn validate_requirement_lookup(
        &self,
        caller: &str,
        capability: &'static str,
    ) -> Result<(), RuntimeFailure> {
        let declarations = self
            .runtime
            .plan
            .plugin_instance(caller)
            .map_or(0, |instance| {
                instance
                    .required_capabilities()
                    .iter()
                    .filter(|requirement| requirement.capability_id() == capability)
                    .count()
            });
        if declarations > 1 {
            return Err(RuntimeFailure::AmbiguousBinding {
                capability,
                providers: declarations,
            });
        }
        Ok(())
    }

    pub(super) fn next_request_id(&self) -> RequestId {
        let request_id = self.runtime.request_ids.get();
        self.runtime.request_ids.set(request_id.saturating_add(1));
        request_id
    }

    pub(super) fn endpoints<C: RequestCapability>(
        &self,
        caller_instance: &str,
    ) -> Option<&[NativeEndpointBinding]> {
        self.bindings
            .get(&(caller_instance.to_owned(), C::ID))
            .map(Vec::as_slice)
    }

    pub(super) fn stream_endpoints<C: StreamCapability>(
        &self,
        caller_instance: &str,
    ) -> Option<&[NativeStreamEndpointBinding]> {
        self.stream_bindings
            .get(&(caller_instance.to_owned(), C::ID))
            .map(Vec::as_slice)
    }

    pub(super) fn event_endpoints<C: EventCapability>(
        &self,
        caller_instance: &str,
    ) -> Option<&[event::NativeEventEndpointBinding]> {
        self.event_bindings
            .get(&(caller_instance.to_owned(), C::ID))
            .map(Vec::as_slice)
    }
}