node-app-sdk-rust 7.5.11

//! # Node-App SDK for Rust
//!
//! Build native [Node-App] plugins as shared libraries (`cdylib`) with a
//! safe, ergonomic Rust API. The SDK targets the **Node Host API v1**
//! (the canonical C header is `core/host-abi-v1/include/node-host-api-v1.h`
//! in the host repository).
//!
//! [Node-App]: https://github.com/econ-v1/node-app-distribution
//!
//! ## What this crate provides
//!
//! - The [`NodeApp`] trait — implement it on your plugin type to get HTTP,
//!   event, and capability handling.
//! - The [`declare_node_app!`] macro — generates all FFI boilerplate
//!   (vtable, panic guards, JSON serialization) from your trait impl.
//! - Helpers for talking to the host: [`log`], [`invoke_capability`],
//!   [`publish_event`], [`get_config`], [`get_storage`], [`set_storage`].
//! - Distributed-trace propagation across capability invocations
//!   (thread-local span context — see [`CurrentTrace`]).
//! - Hard limits matching the host: [`MAX_CAPABILITY_RESPONSE_SIZE`]
//!   (16 MiB), [`MAX_EVENT_NAME_LEN`] (256 B), [`MAX_EVENT_DATA_LEN`]
//!   (64 KiB).
//!
//! ## Stability
//!
//! This crate is published as **`1.0.1-experimental`**. The trait surface
//! and the underlying C ABI are FROZEN for the v1 series, but breaking
//! changes between `1.0.x-experimental` releases are possible until at
//! least two first-party packages have proven the surface in production.
//! Pin to an exact version in `Cargo.toml`.
//!
//! ## Quick start
//!
//! ```toml
//! # Cargo.toml
//! [lib]
//! crate-type = ["cdylib"]
//!
//! [dependencies]
//! node-app-sdk-rust = "1.0.1-experimental"
//! serde_json = "1.0"
//! ```
//!
//! ```rust,ignore
//! use node_app_sdk_rust::*;
//!
//! #[derive(Default)]
//! pub struct MyApp;
//!
//! impl NodeApp for MyApp {
//!     fn metadata() -> NodeAppInfo {
//!         NodeAppInfo {
//!             name: "my-app".into(),
//!             version: "0.1.0".into(),
//!             author: "Me".into(),
//!             description: "Hello-world Node-App".into(),
//!             capabilities: vec!["http_handler".into()],
//!         }
//!     }
//!
//!     fn handle_request(&self, _req: AppRequest) -> Result<AppResponse, NodeAppError> {
//!         Ok(AppResponse {
//!             status: 200,
//!             headers: Default::default(),
//!             body: serde_json::json!({ "hello": "world" }),
//!         })
//!     }
//! }
//!
//! declare_node_app!(MyApp);
//! ```
//!
//! Build with `cargo build --release`; the resulting shared library plus a
//! `manifest.json` is installed into the host's app directory.
//!
//! ## Calling host capabilities
//!
//! Use [`invoke_capability`] to call any capability registered with the
//! host's capability router (e.g. `core.storage.get`,
//! `core.lightning.payment.send`):
//!
//! ```rust,ignore
//! use node_app_sdk_rust::{invoke_capability, CapabilityRequest};
//!
//! let response = invoke_capability(&CapabilityRequest {
//!     id: "req-1".into(),
//!     capability: "core.storage.get".into(),
//!     payload: serde_json::json!({ "key": "user_pref" }),
//!     caller_node_id: None,
//!     trace_id: None,
//!     span_id: None,
//!     parent_span_id: None,
//!     trace_depth: None,
//! })?;
//! # Ok::<(), node_app_sdk_rust::NodeAppError>(())
//! ```
//!
//! Trace context (`trace_id`, `span_id`, `trace_depth`) is propagated
//! automatically when invoking from inside `handle_capability` — you do
//! not need to thread it manually.
//!
//! ## Publishing events
//!
//! ```rust,ignore
//! use node_app_sdk_rust::publish_event;
//!
//! publish_event(
//!     "my-app.something_happened",
//!     &serde_json::json!({ "user_id": 42 }),
//! )?;
//! # Ok::<(), node_app_sdk_rust::NodeAppError>(())
//! ```
//!
//! Event names **must** be namespaced with the app name (`my-app.*`); the
//! host rejects un-namespaced events.

#![warn(missing_docs)]
#![warn(rustdoc::broken_intra_doc_links)]

pub use node_app_api::types::{
    AppEvent, AppRequest, AppResponse, Capabilities, CapabilityExample, CapabilityRequest,
    CapabilityResponse, NodeAppInfo, ProvidedCapability,
};
pub use node_app_api::context::NodeAppContext;
pub use node_app_api::ffi::{FfiResult, NodeAppMetadata, NodeAppVTable};
pub use node_app_api::API_VERSION;

use std::cell::RefCell;
use std::ffi::CString;
use std::sync::atomic::{AtomicPtr, Ordering};
use std::sync::{Condvar, Mutex, MutexGuard, PoisonError};
use std::time::Duration;

/// Maximum response size for capability handlers (16 MiB).
///
/// The host enforces this limit on every capability response. Apps that
/// produce a larger response will see [`FfiResult::error`] returned with
/// error code `-6`. Use streaming or pagination for larger payloads.
pub const MAX_CAPABILITY_RESPONSE_SIZE: usize = 16 * 1024 * 1024;

/// Trace fields for the currently-executing capability span (per-thread).
///
/// Set at the start of `handle_capability` by the [`declare_node_app!`]
/// macro and cleared on return. Allows [`invoke_capability`] to propagate
/// distributed trace context automatically without the caller having to
/// thread `trace_id` / `span_id` through every call site.
#[doc(hidden)]
#[derive(Clone)]
pub struct CurrentTrace {
    /// Trace ID inherited from the inbound capability request.
    pub trace_id: String,
    /// Span ID of the current execution — becomes `parent_span_id` for
    /// sub-invocations made from this thread.
    pub span_id: String,
    /// Depth of the current span in the trace tree (root = 0).
    pub depth: u8,
}

thread_local! {
    /// Current trace context for the executing capability (set by
    /// [`declare_node_app!`]).
    #[doc(hidden)]
    pub static CURRENT_TRACE: RefCell<Option<CurrentTrace>> = const { RefCell::new(None) };
    #[doc(hidden)]
    pub static CURRENT_INVOCATION_CONTEXT: RefCell<Option<String>> = const { RefCell::new(None) };
}

#[doc(hidden)]
pub struct CurrentInvocationContextGuard(Option<String>);

impl CurrentInvocationContextGuard {
    #[doc(hidden)]
    pub fn enter(invocation_context_id: Option<String>) -> Self {
        let previous = CURRENT_INVOCATION_CONTEXT.with(|current| {
            std::mem::replace(&mut *current.borrow_mut(), invocation_context_id)
        });
        Self(previous)
    }
}

impl Drop for CurrentInvocationContextGuard {
    fn drop(&mut self) {
        let previous = self.0.take();
        CURRENT_INVOCATION_CONTEXT.with(|current| {
            *current.borrow_mut() = previous;
        });
    }
}

/// Global storage for the host context pointer.
///
/// Uses `AtomicPtr` instead of `OnceLock` so that the pointer can be
/// **updated on every init call**. On macOS, `dlclose` does not actually
/// unload user libraries (`man dlclose`: "Mac OS X does not support
/// dynamic unloading"), so the same library instance is reused across
/// hot-reloads. With a `OnceLock` the first-load context pointer would
/// be retained permanently; after the first-load `HostData` is dropped
/// that pointer becomes dangling, causing a SIGSEGV on the next reload
/// when any SDK function (e.g. `log`) tries to read it.
///
/// `AtomicPtr` allows `__store_context` to atomically replace the pointer
/// on each init, ensuring it always points to the current live `HostData`.
///
/// Safety invariant: the pointer is set to a valid `NodeAppContext` during
/// `__node_app_init` and is only read while the app is alive. The host
/// (NativeLoader) keeps `HostData` + `NodeAppContext` alive for the entire
/// lifetime of the loaded app instance.
static APP_CONTEXT: AtomicPtr<NodeAppContext> = AtomicPtr::new(std::ptr::null_mut());

/// Error code a call gets when it arrives after the app's shutdown started.
///
/// From the moment the host calls the app's shutdown entry, the
/// [`declare_node_app!`] wrappers refuse new `handle_request`,
/// `handle_event` and `handle_capability` calls with this code instead of
/// running them. The next `init` accepts calls again. See ADR-083.
pub const ERROR_SHUTTING_DOWN: i32 = -11;

/// How long the shutdown entry waits for in-flight calls before it logs a
/// warning that names how many are still running.
///
/// The entry keeps waiting after this. It must not return while a call is
/// still running inside the library: the host releases the library owner
/// when the shutdown entry returns, and it calls the app without holding
/// that owner. The host's own shutdown budget bounds the whole wait. See
/// ADR-083.
pub const SHUTDOWN_DRAIN_WARN_AFTER: Duration = Duration::from_secs(2);

/// Admission gate for calls into the app (generated by [`declare_node_app!`]).
///
/// Counts the calls that run inside the app and refuses new ones after
/// [`CallGate::close`], so the shutdown entry can tell the app to stop at
/// once and take the write lock only after the in-flight calls return.
#[doc(hidden)]
pub struct CallGate {
    state: Mutex<CallGateState>,
    drained: Condvar,
}

struct CallGateState {
    closed: bool,
    in_flight: usize,
}

impl CallGate {
    #[doc(hidden)]
    pub const fn new() -> Self {
        Self {
            state: Mutex::new(CallGateState {
                closed: false,
                in_flight: 0,
            }),
            drained: Condvar::new(),
        }
    }

    fn lock(&self) -> MutexGuard<'_, CallGateState> {
        // No code panics while it holds this mutex, so poison carries no
        // information; recover the state rather than wedge every call.
        self.state.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Admit one call, or `None` once the gate is closed. The call counts
    /// as in flight until the returned permit drops.
    #[doc(hidden)]
    pub fn enter(&self) -> Option<CallPermit<'_>> {
        let mut state = self.lock();
        if state.closed {
            return None;
        }
        state.in_flight += 1;
        Some(CallPermit(self))
    }

    /// Accept calls again (the app is being initialized).
    #[doc(hidden)]
    pub fn open(&self) {
        self.lock().closed = false;
    }

    /// Refuse every call from now on.
    #[doc(hidden)]
    pub fn close(&self) {
        self.lock().closed = true;
    }

    /// Wait until no call is in flight, or until `timeout` passes when it is
    /// `Some`. Returns the number of calls still in flight.
    #[doc(hidden)]
    pub fn wait_drained(&self, timeout: Option<Duration>) -> usize {
        let state = self.lock();
        let state = match timeout {
            Some(timeout) => {
                self.drained
                    .wait_timeout_while(state, timeout, |state| state.in_flight > 0)
                    .unwrap_or_else(PoisonError::into_inner)
                    .0
            }
            None => self
                .drained
                .wait_while(state, |state| state.in_flight > 0)
                .unwrap_or_else(PoisonError::into_inner),
        };
        state.in_flight
    }
}

impl Default for CallGate {
    fn default() -> Self {
        Self::new()
    }
}

/// One admitted call; see [`CallGate::enter`].
#[doc(hidden)]
pub struct CallPermit<'a>(&'a CallGate);

impl Drop for CallPermit<'_> {
    fn drop(&mut self) {
        let mut state = self.0.lock();
        state.in_flight -= 1;
        if state.in_flight == 0 {
            self.0.drained.notify_all();
        }
    }
}

// ── Public api-store LLM SDK (feature 472) ──────────────────────────────────
//
// Type-safe surface for the host's `/api/v2/public/api_store/llm/{call,stream}`
// endpoints. Lives in its own submodule so plugins that don't talk to the
// public LLM endpoint never see these symbols in their import surface.
pub mod llm;

/// Log levels accepted by the [`log`] function.
///
/// Use these constants instead of magic numbers when calling
/// [`log`] directly. The convenience macros ([`log_info!`], etc.)
/// take care of this for you.
pub mod log_level {
    /// Most verbose level — use for fine-grained tracing.
    pub const TRACE: u32 = 0;
    /// Debug-level diagnostics, typically not shown in production.
    pub const DEBUG: u32 = 1;
    /// Informational messages indicating normal operation.
    pub const INFO: u32 = 2;
    /// Warnings — recoverable issues or unusual conditions.
    pub const WARN: u32 = 3;
    /// Errors — operations that failed and require attention.
    pub const ERROR: u32 = 4;
}

/// Log a message to the host using the stored context.
///
/// Logs are written to the host's per-app log file
/// (`{log_dir}/{app_name}.log`) and forwarded to the host's tracing
/// subscriber, so they appear in the daemon's main log output too.
///
/// This function is a no-op if the context was not provided during init
/// or if the message contains invalid UTF-8.
///
/// # Arguments
/// * `level` - Log level (0=trace, 1=debug, 2=info, 3=warn, 4=error). Use the [`log_level`] constants.
/// * `message` - The log message (must not contain interior NUL bytes).
///
/// # Example
/// ```ignore
/// use node_app_sdk_rust::{log, log_level};
///
/// log(log_level::INFO, "App initialized successfully");
/// log(log_level::ERROR, "Something went wrong!");
/// ```
///
/// Most callers should use the convenience macros ([`log_info!`],
/// [`log_error!`], etc.) which accept `format!`-style arguments.
pub fn log(level: u32, message: &str) {
    let ctx_ptr = APP_CONTEXT.load(Ordering::Acquire);
    if ctx_ptr.is_null() {
        return;
    }

    let c_message = match CString::new(message) {
        Ok(s) => s,
        Err(_) => return, // Invalid message (contains null byte)
    };

    // Safety: ctx_ptr is valid for the app's lifetime, and host_log is a valid function pointer
    unsafe {
        let ctx = &*ctx_ptr;
        (ctx.host_log)(ctx.host_data, level, c_message.as_ptr());
    }
}

/// Invoke a capability on the host via the capability router.
///
/// This is the primary mechanism for app-to-app communication. The host
/// resolves the capability name to the providing app, dispatches the
/// request, and returns the response. The provider may itself be a
/// different app, the host kernel, or a remote node (transparent to the
/// caller).
///
/// Trace context is propagated automatically: if this call happens
/// inside `handle_capability` and the inbound request carried a
/// `trace_id`, the same trace ID is injected on outbound calls. Callers
/// may override this by setting `trace_id` explicitly on the request.
///
/// # Errors
///
/// Returns [`NodeAppError::CapabilityError`] when:
/// - The host context is not available (called before init).
/// - The host's invoke callback is not wired up.
/// - The capability is not registered, the provider rejects the call,
///   or the response cannot be deserialized.
pub fn invoke_capability(request: &CapabilityRequest) -> Result<CapabilityResponse, NodeAppError> {
    let ctx_ptr = APP_CONTEXT.load(Ordering::Acquire);
    if ctx_ptr.is_null() {
        return Err(NodeAppError::CapabilityError(
            "Host context not available".into(),
        ));
    }

    // Propagate distributed trace context if a parent span is active on this thread.
    // Only inject when the caller hasn't already set trace fields.
    let active_invocation_context =
        CURRENT_INVOCATION_CONTEXT.with(|current| current.borrow().clone());
    let active_trace = if request.trace_id.is_none() {
        CURRENT_TRACE.with(|current| current.borrow().clone())
    } else {
        None
    };
    let effective_request: std::borrow::Cow<CapabilityRequest> =
        if active_trace.is_some() || active_invocation_context.is_some() {
            let mut injected = request.clone();
            if let Some(trace) = active_trace {
                injected.trace_id = Some(trace.trace_id);
                injected.span_id = Some(trace.span_id);
                injected.parent_span_id = None;
                injected.trace_depth = Some(trace.depth);
            }
            if let Some(context_id) = active_invocation_context {
                injected.invocation_context_id = Some(context_id);
            }
            std::borrow::Cow::Owned(injected)
        } else {
            std::borrow::Cow::Borrowed(request)
        };

    // Serialize the request to JSON
    let request_json = serde_json::to_vec(effective_request.as_ref())?;

    // Safety: ctx_ptr is valid for the app's lifetime
    unsafe {
        let ctx = &*ctx_ptr;

        // Check if the callback is available
        if ctx.host_invoke_capability as usize == 0 {
            return Err(NodeAppError::CapabilityError(
                "host_invoke_capability callback not available".into(),
            ));
        }

        let result = (ctx.host_invoke_capability)(
            ctx.host_data,
            request_json.as_ptr(),
            request_json.len(),
        );

        if result.success && !result.data.is_null() && result.data_len > 0 {
            let response_slice = std::slice::from_raw_parts(result.data, result.data_len);
            let response: CapabilityResponse = serde_json::from_slice(response_slice)
                .map_err(|e| NodeAppError::CapabilityError(format!("Response deserialization error: {}", e)))?;
            // Free the host-allocated data
            // Note: The host is responsible for freeing this memory via its own allocator
            Ok(response)
        } else if !result.success {
            Err(NodeAppError::CapabilityError(format!(
                "Host capability invocation failed with error code {}",
                result.error_code
            )))
        } else {
            Err(NodeAppError::CapabilityError(
                "Empty response from host".into(),
            ))
        }
    }
}

/// Maximum event name length in bytes (256).
///
/// Names that exceed this limit are rejected by [`publish_event`] with
/// [`NodeAppError::EventFailed`]. The host enforces the same limit on
/// the receiving side.
pub const MAX_EVENT_NAME_LEN: usize = 256;

/// Maximum event data length in bytes (64 KiB).
///
/// Payloads that exceed this limit are rejected by [`publish_event`]
/// with [`NodeAppError::EventFailed`]. For larger artifacts, store them
/// (e.g. via `core.storage.insert`) and emit an event referencing the
/// storage key instead.
pub const MAX_EVENT_DATA_LEN: usize = 64 * 1024;

/// Publish a domain event to the host event bus.
///
/// The event is queued asynchronously (fire-and-forget). The event
/// name **must** be namespaced with the app name prefix
/// (e.g. `lightning.payment_received`, `my-app.user_created`); the host
/// rejects events whose name does not start with `{app_name}.`.
///
/// # Errors
///
/// Returns [`NodeAppError::EventFailed`] if the host context is not
/// available, the event name or data exceeds size limits
/// ([`MAX_EVENT_NAME_LEN`] / [`MAX_EVENT_DATA_LEN`]), or the host
/// rejects the event.
pub fn publish_event(name: &str, data: &serde_json::Value) -> Result<(), NodeAppError> {
    let ctx_ptr = APP_CONTEXT.load(Ordering::Acquire);
    if ctx_ptr.is_null() {
        return Err(NodeAppError::EventFailed(
            "Host context not available".into(),
        ));
    }

    let name_bytes = name.as_bytes();
    if name_bytes.len() > MAX_EVENT_NAME_LEN {
        return Err(NodeAppError::EventFailed(format!(
            "Event name exceeds {} byte limit (got {})",
            MAX_EVENT_NAME_LEN,
            name_bytes.len()
        )));
    }

    let data_json = serde_json::to_vec(data)?;
    if data_json.len() > MAX_EVENT_DATA_LEN {
        return Err(NodeAppError::EventFailed(format!(
            "Event data exceeds {} byte limit (got {})",
            MAX_EVENT_DATA_LEN,
            data_json.len()
        )));
    }

    unsafe {
        let ctx = &*ctx_ptr;

        if ctx.host_publish_event as usize == 0 {
            return Err(NodeAppError::EventFailed(
                "host_publish_event callback not available".into(),
            ));
        }

        let result = (ctx.host_publish_event)(
            ctx.host_data,
            name_bytes.as_ptr(),
            name_bytes.len(),
            data_json.as_ptr(),
            data_json.len(),
        );

        if result == 0 {
            Ok(())
        } else {
            Err(NodeAppError::EventFailed(format!(
                "host_publish_event returned error code {}",
                result
            )))
        }
    }
}

/// Log a TRACE-level message using `format!`-style arguments.
///
/// No-op when the host context has not been wired up yet. See [`log`]
/// for details about delivery semantics.
#[macro_export]
macro_rules! log_trace {
    ($($arg:tt)*) => {
        $crate::log($crate::log_level::TRACE, &format!($($arg)*))
    };
}

/// Log a DEBUG-level message using `format!`-style arguments.
#[macro_export]
macro_rules! log_debug {
    ($($arg:tt)*) => {
        $crate::log($crate::log_level::DEBUG, &format!($($arg)*))
    };
}

/// Log an INFO-level message using `format!`-style arguments.
#[macro_export]
macro_rules! log_info {
    ($($arg:tt)*) => {
        $crate::log($crate::log_level::INFO, &format!($($arg)*))
    };
}

/// Log a WARN-level message using `format!`-style arguments.
#[macro_export]
macro_rules! log_warn {
    ($($arg:tt)*) => {
        $crate::log($crate::log_level::WARN, &format!($($arg)*))
    };
}

/// Log an ERROR-level message using `format!`-style arguments.
#[macro_export]
macro_rules! log_error {
    ($($arg:tt)*) => {
        $crate::log($crate::log_level::ERROR, &format!($($arg)*))
    };
}

/// Store the context pointer for use by host helper functions.
///
/// Called automatically by [`declare_node_app!`] during init. App code
/// must not call this directly.
///
/// Uses `AtomicPtr::store` so the pointer is replaced on every init —
/// required on macOS where `dlclose` never unloads user libraries and the
/// same app instance is reused across hot-reloads.
#[doc(hidden)]
pub fn __store_context(ctx: *const NodeAppContext) {
    APP_CONTEXT.store(ctx as *mut NodeAppContext, Ordering::Release);
}

/// Get a configuration value from the host by key.
///
/// Common keys include `data_dir`, `host_port`, `app_name`, and
/// `app_id`. The full set of available keys is determined by the host
/// at app-init time.
///
/// Returns `None` if the key is not registered or the context is not
/// available (called before init).
pub fn get_config(key: &str) -> Option<String> {
    let ctx_ptr = APP_CONTEXT.load(Ordering::Acquire);
    if ctx_ptr.is_null() {
        return None;
    }
    let c_key = CString::new(key).ok()?;
    unsafe {
        let ctx = &*ctx_ptr;
        let result = (ctx.host_get_config)(ctx.host_data, c_key.as_ptr());
        if result.is_null() {
            return None;
        }
        Some(std::ffi::CStr::from_ptr(result).to_string_lossy().into_owned())
    }
}

/// Set a storage value scoped to this app.
///
/// Storage is persisted by the host across app restarts and is isolated
/// per-app: other apps cannot read or write keys you set here. Useful
/// for small bits of configuration or state; for larger or relational
/// data, prefer the `core.storage.*` capabilities.
///
/// No-op when the host context is not available or `key`/`value`
/// contain interior NUL bytes.
pub fn set_storage(key: &str, value: &str) {
    let ctx_ptr = APP_CONTEXT.load(Ordering::Acquire);
    if ctx_ptr.is_null() {
        return;
    }
    let c_key = match CString::new(key) {
        Ok(s) => s,
        Err(_) => return,
    };
    let c_value = match CString::new(value) {
        Ok(s) => s,
        Err(_) => return,
    };
    unsafe {
        let ctx = &*ctx_ptr;
        (ctx.host_set_storage)(ctx.host_data, c_key.as_ptr(), c_value.as_ptr());
    }
}

/// Get a storage value by key, scoped to this app.
///
/// See [`set_storage`] for scoping semantics. Returns `None` if the key
/// is not found or the context is not available.
pub fn get_storage(key: &str) -> Option<String> {
    let ctx_ptr = APP_CONTEXT.load(Ordering::Acquire);
    if ctx_ptr.is_null() {
        return None;
    }
    let c_key = CString::new(key).ok()?;
    unsafe {
        let ctx = &*ctx_ptr;
        let result = (ctx.host_get_storage)(ctx.host_data, c_key.as_ptr());
        if result.is_null() {
            return None;
        }
        Some(std::ffi::CStr::from_ptr(result).to_string_lossy().into_owned())
    }
}

/// Error type returned by app operations.
///
/// Variants surface specific failure modes from each lifecycle hook plus
/// transport-level errors (serialization, capability dispatch). The
/// `#[from] serde_json::Error` impl on [`NodeAppError::SerializationError`]
/// allows the `?` operator to propagate JSON errors directly.
#[derive(Debug, thiserror::Error)]
pub enum NodeAppError {
    /// Returned from [`NodeApp::init`] when initialization failed.
    #[error("Initialization failed: {0}")]
    InitFailed(String),
    /// Returned from [`NodeApp::handle_request`] when handling failed.
    #[error("Request handling failed: {0}")]
    RequestFailed(String),
    /// Returned from [`NodeApp::handle_event`] or [`publish_event`].
    #[error("Event handling failed: {0}")]
    EventFailed(String),
    /// Returned from [`NodeApp::shutdown`] when graceful shutdown failed.
    #[error("Shutdown failed: {0}")]
    ShutdownFailed(String),
    /// JSON (de)serialization error — auto-converted from
    /// [`serde_json::Error`] via the `?` operator.
    #[error("Serialization error: {0}")]
    SerializationError(#[from] serde_json::Error),
    /// Returned from [`NodeApp::handle_capability`] or [`invoke_capability`].
    #[error("Capability error: {0}")]
    CapabilityError(String),
}

/// Trait implemented by node-app plugins.
///
/// Pair this with [`declare_node_app!`] to generate all the FFI
/// boilerplate. Implementors must be `Default + Send + Sync + 'static`
/// because the macro stores the singleton instance behind a static
/// `OnceLock<Mutex<T>>`.
///
/// All trait methods have sensible default implementations; override
/// only the hooks the app uses. Apps that opt into a hook must also
/// declare the matching capability in their manifest, e.g.
/// `"http_handler"` for HTTP request routing.
pub trait NodeApp: Default + Send + Sync + 'static {
    /// Return app metadata (name, version, author, description, capabilities).
    ///
    /// Called once when the host loads the shared library. The values
    /// are cached for the lifetime of the process.
    fn metadata() -> NodeAppInfo;

    /// Initialize the app with the host context.
    ///
    /// Called once after loading. The default impl is a no-op — override
    /// to set up state (DB connections, caches, etc.). The context may
    /// be `None` if the host has not wired up callbacks yet (very early
    /// in bootstrap or in unit tests).
    fn init(&mut self, _ctx: Option<&NodeAppContext>) -> Result<(), NodeAppError> {
        Ok(())
    }

    /// Start shutting down: called at once when the host stops the app,
    /// while earlier calls may still be running.
    ///
    /// Runs under a shared borrow and does not wait for in-flight calls.
    /// From this point the SDK refuses new calls with
    /// [`ERROR_SHUTTING_DOWN`]. Override it to make in-flight work return
    /// soon: cancel a token, set a flag, close a channel. Keep it short and
    /// do not block in it. [`NodeApp::shutdown`] runs after it, once every
    /// in-flight call has returned. Default is a no-op. See ADR-083.
    fn begin_shutdown(&self) {}

    /// Shut down the app gracefully.
    ///
    /// Called before unloading, after [`NodeApp::begin_shutdown`] and after
    /// every in-flight call has returned, so it has exclusive access.
    /// Default is a no-op. Override to flush pending writes, close
    /// connections, etc.
    fn shutdown(&mut self) -> Result<(), NodeAppError> {
        Ok(())
    }

    /// Handle an incoming HTTP request proxied from the host.
    ///
    /// Only invoked if the app declared the `http_handler` capability.
    /// Default returns `501 Not Implemented`.
    fn handle_request(&self, _request: AppRequest) -> Result<AppResponse, NodeAppError> {
        Ok(AppResponse {
            status: 501,
            headers: Default::default(),
            body: serde_json::json!({"error": "Not implemented"}),
        })
    }

    /// Handle a domain event from the host event bus.
    ///
    /// Only invoked if the app declared the `event_listener` capability
    /// and subscribed to the event's namespace via the manifest.
    fn handle_event(&self, _event: AppEvent) -> Result<(), NodeAppError> {
        Ok(())
    }

    /// Return the list of service capabilities this app provides.
    ///
    /// Override to declare capabilities for the host's capability
    /// registry. Default returns an empty list (no capabilities
    /// provided). Each entry must use the namespace `{app_name}.{domain}.{action}`
    /// — the `core.*` namespace is reserved for first-party apps.
    fn provided_capabilities() -> Vec<ProvidedCapability> {
        Vec::new()
    }

    /// Handle a capability invocation from another app via the
    /// capability router.
    ///
    /// Override to implement capability handling logic. The default
    /// returns [`NodeAppError::CapabilityError`] indicating the
    /// capability is not implemented.
    ///
    /// Trace context is automatically captured into thread-local state
    /// for the duration of this call so [`invoke_capability`] can
    /// propagate it on outbound calls without explicit threading.
    fn handle_capability(
        &self,
        _request: CapabilityRequest,
    ) -> Result<CapabilityResponse, NodeAppError> {
        Err(NodeAppError::CapabilityError(
            "Capability handling not implemented".into(),
        ))
    }
}

/// Generate all FFI boilerplate for a [`NodeApp`] implementation.
///
/// This macro creates:
/// - A `OnceLock<RwLock<T>>` instance for the app (sound concurrent access).
/// - A call gate: the shutdown entry calls [`NodeApp::begin_shutdown`] at
///   once, refuses new calls with [`ERROR_SHUTTING_DOWN`], and takes the
///   write lock for [`NodeApp::shutdown`] only after in-flight calls return.
/// - The `_node_app_entry` export symbol returning a [`NodeAppVTable`].
/// - FFI wrapper functions for `init`, `shutdown`, `handle_request`,
///   `handle_event`, `handle_capability`, and `free`.
/// - `catch_unwind` guards on every FFI boundary to prevent UB from
///   panics crossing the Rust/C ABI.
/// - Thread-local trace span management around `handle_capability`.
///
/// Invoke once at the crate root after defining the app type:
///
/// ```ignore
/// declare_node_app!(MyApp);
/// ```
#[macro_export]
macro_rules! declare_node_app {
    ($app_type:ty) => {
        // Feature 463 B5 fix — RwLock instead of Mutex so capability calls
        // can run CONCURRENTLY (read-locked). The previous Mutex serialized
        // every call, causing a deadlock when one capability (e.g. agent.prompt)
        // invokes another capability on the SAME app re-entrantly through the
        // graph engine (agent.prompt → graph.engine.send_event → graph node →
        // graph.tasks.storage.* → core.agent_storage.* → BACK to agent app).
        // init still takes the write lock. Shutdown takes it only after the
        // call gate has drained, so begin_shutdown runs at once (ADR-083).
        static APP_INSTANCE: std::sync::OnceLock<std::sync::RwLock<$app_type>> =
            std::sync::OnceLock::new();
        static CALL_GATE: $crate::CallGate = $crate::CallGate::new();
        static VTABLE: std::sync::OnceLock<$crate::NodeAppVTable> =
            std::sync::OnceLock::new();

        // OnceLock-backed CStrings for metadata (valid for process lifetime)
        static META_NAME: std::sync::OnceLock<std::ffi::CString> = std::sync::OnceLock::new();
        static META_VERSION: std::sync::OnceLock<std::ffi::CString> = std::sync::OnceLock::new();
        static META_AUTHOR: std::sync::OnceLock<std::ffi::CString> = std::sync::OnceLock::new();
        static META_DESCRIPTION: std::sync::OnceLock<std::ffi::CString> =
            std::sync::OnceLock::new();

        unsafe extern "C" fn __node_app_init(
            ctx: *const std::os::raw::c_void,
        ) -> $crate::FfiResult {
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                let ctx_opt = if ctx.is_null() {
                    None
                } else {
                    // Store context pointer for use by log() helper
                    let ctx_typed = ctx as *const $crate::NodeAppContext;
                    $crate::__store_context(ctx_typed);
                    Some(unsafe { &*ctx_typed })
                };
                let app = APP_INSTANCE.get_or_init(|| {
                    std::sync::RwLock::new(<$app_type>::default())
                });
                let mut guard = match app.write() {
                    Ok(g) => g,
                    Err(e) => {
                        eprintln!("[node-app] rwlock poisoned in init: {}", e);
                        return $crate::FfiResult::error(-10);
                    }
                };
                // A cached library is initialized again after a shutdown
                // (reload), so accept calls again.
                CALL_GATE.open();
                match guard.init(ctx_opt) {
                    Ok(()) => $crate::FfiResult::ok(),
                    Err(e) => {
                        let msg = format!("init error: {}", e);
                        $crate::log($crate::log_level::ERROR, &msg);
                        eprintln!("[node-app] {}", msg);
                        $crate::FfiResult::error(-1)
                    }
                }
            })) {
                Ok(result) => result,
                Err(_) => {
                    eprintln!("[node-app] panic in init");
                    $crate::FfiResult::error(-99)
                }
            }
        }

        unsafe extern "C" fn __node_app_shutdown() -> $crate::FfiResult {
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                if let Some(app) = APP_INSTANCE.get() {
                    // ADR-083: refuse new calls, tell the app at once, then
                    // take the write lock only after in-flight calls return.
                    // Never return before they do: the host may unload the
                    // library when this entry returns.
                    CALL_GATE.close();
                    let hook = match app.read() {
                        Ok(app) => std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                            <$app_type as $crate::NodeApp>::begin_shutdown(&app)
                        })),
                        Err(e) => {
                            eprintln!("[node-app] rwlock poisoned in shutdown: {}", e);
                            return $crate::FfiResult::error(-10);
                        }
                    };
                    // A panicking hook must not skip the drain below.
                    if hook.is_err() {
                        let msg = "panic in begin_shutdown; still waiting for in-flight calls";
                        $crate::log($crate::log_level::ERROR, msg);
                        eprintln!("[node-app] {}", msg);
                    }
                    let in_flight =
                        CALL_GATE.wait_drained(Some($crate::SHUTDOWN_DRAIN_WARN_AFTER));
                    if in_flight > 0 {
                        let msg = format!(
                            "shutdown: {} call(s) still in flight after {} ms; waiting for them to return",
                            in_flight,
                            $crate::SHUTDOWN_DRAIN_WARN_AFTER.as_millis()
                        );
                        $crate::log($crate::log_level::WARN, &msg);
                        eprintln!("[node-app] {}", msg);
                        CALL_GATE.wait_drained(None);
                    }
                    let mut guard = match app.write() {
                        Ok(g) => g,
                        Err(e) => {
                            eprintln!("[node-app] rwlock poisoned in shutdown: {}", e);
                            return $crate::FfiResult::error(-10);
                        }
                    };
                    match guard.shutdown() {
                        Ok(()) => $crate::FfiResult::ok(),
                        Err(e) => {
                            let msg = format!("shutdown error: {}", e);
                            $crate::log($crate::log_level::ERROR, &msg);
                            eprintln!("[node-app] {}", msg);
                            $crate::FfiResult::error(-1)
                        }
                    }
                } else {
                    $crate::FfiResult::ok()
                }
            })) {
                Ok(result) => result,
                Err(_) => {
                    eprintln!("[node-app] panic in shutdown");
                    $crate::FfiResult::error(-99)
                }
            }
        }

        unsafe extern "C" fn __node_app_handle_request(
            request_json: *const u8,
            request_len: usize,
        ) -> $crate::FfiResult {
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                let json_slice =
                    unsafe { std::slice::from_raw_parts(request_json, request_len) };
                let request: $crate::AppRequest = match serde_json::from_slice(json_slice) {
                    Ok(r) => r,
                    Err(e) => {
                        eprintln!("[node-app] request deserialization error: {}", e);
                        return $crate::FfiResult::error(-2);
                    }
                };

                let app = match APP_INSTANCE.get() {
                    Some(a) => a,
                    None => return $crate::FfiResult::error(-3),
                };
                // ADR-083 — refused once shutdown starts; counted in flight
                // until this permit drops (after the read guard below).
                let _call = match CALL_GATE.enter() {
                    Some(call) => call,
                    None => return $crate::FfiResult::error($crate::ERROR_SHUTTING_DOWN),
                };
                // Feature 463 B5 — read-lock allows concurrent calls
                let guard = match app.read() {
                    Ok(g) => g,
                    Err(e) => {
                        eprintln!("[node-app] rwlock poisoned in handle_request: {}", e);
                        return $crate::FfiResult::error(-10);
                    }
                };

                match guard.handle_request(request) {
                    Ok(response) => match serde_json::to_vec(&response) {
                        Ok(bytes) => {
                            let len = bytes.len();
                            let boxed = bytes.into_boxed_slice();
                            let ptr = Box::into_raw(boxed) as *mut u8;
                            $crate::FfiResult {
                                success: true,
                                error_code: 0,
                                data: ptr,
                                data_len: len,
                            }
                        }
                        Err(e) => {
                            eprintln!("[node-app] response serialization error: {}", e);
                            $crate::FfiResult::error(-4)
                        }
                    },
                    Err(e) => {
                        eprintln!("[node-app] handle_request error: {}", e);
                        $crate::FfiResult::error(-5)
                    }
                }
            })) {
                Ok(result) => result,
                Err(_) => {
                    eprintln!("[node-app] panic in handle_request");
                    $crate::FfiResult::error(-99)
                }
            }
        }

        unsafe extern "C" fn __node_app_handle_event(
            event_json: *const u8,
            event_len: usize,
        ) -> $crate::FfiResult {
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                let json_slice =
                    unsafe { std::slice::from_raw_parts(event_json, event_len) };
                let event: $crate::AppEvent = match serde_json::from_slice(json_slice) {
                    Ok(e) => e,
                    Err(e) => {
                        eprintln!("[node-app] event deserialization error: {}", e);
                        return $crate::FfiResult::error(-2);
                    }
                };

                let app = match APP_INSTANCE.get() {
                    Some(a) => a,
                    None => return $crate::FfiResult::error(-3),
                };
                // ADR-083 — refused once shutdown starts; declared before the
                // read guard so the guard drops first.
                let _call = match CALL_GATE.enter() {
                    Some(call) => call,
                    None => return $crate::FfiResult::error($crate::ERROR_SHUTTING_DOWN),
                };
                // Feature 463 B5 — read-lock allows concurrent calls
                let guard = match app.read() {
                    Ok(g) => g,
                    Err(e) => {
                        eprintln!("[node-app] rwlock poisoned in handle_event: {}", e);
                        return $crate::FfiResult::error(-10);
                    }
                };

                match guard.handle_event(event) {
                    Ok(()) => $crate::FfiResult::ok(),
                    Err(e) => {
                        eprintln!("[node-app] handle_event error: {}", e);
                        $crate::FfiResult::error(-5)
                    }
                }
            })) {
                Ok(result) => result,
                Err(_) => {
                    eprintln!("[node-app] panic in handle_event");
                    $crate::FfiResult::error(-99)
                }
            }
        }

        unsafe extern "C" fn __node_app_handle_capability(
            request_json: *const u8,
            request_len: usize,
        ) -> $crate::FfiResult {
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                let json_slice =
                    unsafe { std::slice::from_raw_parts(request_json, request_len) };
                let request: $crate::CapabilityRequest = match serde_json::from_slice(json_slice) {
                    Ok(r) => r,
                    Err(e) => {
                        eprintln!("[node-app] capability request deserialization error: {}", e);
                        return $crate::FfiResult::error(-2);
                    }
                };

                let app = match APP_INSTANCE.get() {
                    Some(a) => a,
                    None => return $crate::FfiResult::error(-3),
                };
                // ADR-083 — refused once shutdown starts; declared before the
                // read guard so the guard drops first.
                let _call = match CALL_GATE.enter() {
                    Some(call) => call,
                    None => return $crate::FfiResult::error($crate::ERROR_SHUTTING_DOWN),
                };
                // Feature 463 B5 — read-lock allows the SAME app to handle
                // re-entrant capability calls (e.g. agent.prompt outer call
                // and an inner core.agent_storage.write from a graph node)
                // concurrently. handle_capability is &self so this is sound.
                let guard = match app.read() {
                    Ok(g) => g,
                    Err(e) => {
                        eprintln!("[node-app] rwlock poisoned in handle_capability: {}", e);
                        return $crate::FfiResult::error(-10);
                    }
                };

                // Set thread-local trace context for duration of this capability call.
                // This allows invoke_capability() to propagate trace automatically.
                $crate::CURRENT_TRACE.with(|tl| {
                    *tl.borrow_mut() = if let Some(ref trace_id) = request.trace_id {
                        Some($crate::CurrentTrace {
                            trace_id: trace_id.clone(),
                            span_id: request.span_id.clone().unwrap_or_default(),
                            depth: request.trace_depth.unwrap_or(0),
                        })
                    } else {
                        None
                    };
                });

                let _invocation_context_guard =
                    $crate::CurrentInvocationContextGuard::enter(
                        request.invocation_context_id.clone(),
                    );

                let cap_result = guard.handle_capability(request);

                // Clear trace context after call completes.
                $crate::CURRENT_TRACE.with(|tl| {
                    *tl.borrow_mut() = None;
                });

                match cap_result {
                    Ok(response) => match serde_json::to_vec(&response) {
                        Ok(bytes) => {
                            // Enforce 16MB response limit
                            if bytes.len() > $crate::MAX_CAPABILITY_RESPONSE_SIZE {
                                eprintln!(
                                    "[node-app] capability response exceeds 16MB limit ({} bytes)",
                                    bytes.len()
                                );
                                return $crate::FfiResult::error(-6);
                            }
                            let len = bytes.len();
                            let boxed = bytes.into_boxed_slice();
                            let ptr = Box::into_raw(boxed) as *mut u8;
                            $crate::FfiResult {
                                success: true,
                                error_code: 0,
                                data: ptr,
                                data_len: len,
                            }
                        }
                        Err(e) => {
                            eprintln!("[node-app] capability response serialization error: {}", e);
                            $crate::FfiResult::error(-4)
                        }
                    },
                    Err(e) => {
                        eprintln!("[node-app] handle_capability error: {}", e);
                        $crate::FfiResult::error(-5)
                    }
                }
            })) {
                Ok(result) => result,
                Err(_) => {
                    eprintln!("[node-app] panic in handle_capability");
                    $crate::FfiResult::error(-99)
                }
            }
        }

        unsafe extern "C" fn __node_app_free(ptr: *mut u8, len: usize) {
            if !ptr.is_null() && len > 0 {
                let _ = unsafe { Box::from_raw(std::slice::from_raw_parts_mut(ptr, len)) };
            }
        }

        #[no_mangle]
        pub unsafe extern "C" fn _node_app_entry(
            _ctx: *const std::os::raw::c_void,
        ) -> *const $crate::NodeAppVTable {
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                let info = <$app_type as $crate::NodeApp>::metadata();
                let caps = info.capability_flags();

                let name = META_NAME
                    .get_or_init(|| std::ffi::CString::new(info.name).unwrap_or_default());
                let version = META_VERSION
                    .get_or_init(|| std::ffi::CString::new(info.version).unwrap_or_default());
                let author = META_AUTHOR
                    .get_or_init(|| std::ffi::CString::new(info.author).unwrap_or_default());
                let description = META_DESCRIPTION
                    .get_or_init(|| std::ffi::CString::new(info.description).unwrap_or_default());

                let metadata = $crate::NodeAppMetadata {
                    api_version: $crate::API_VERSION,
                    name: name.as_ptr(),
                    version: version.as_ptr(),
                    author: author.as_ptr(),
                    description: description.as_ptr(),
                    capabilities: caps.bits(),
                };

                VTABLE.get_or_init(|| $crate::NodeAppVTable {
                    metadata,
                    init: __node_app_init,
                    shutdown: __node_app_shutdown,
                    handle_request: __node_app_handle_request,
                    handle_event: __node_app_handle_event,
                    handle_capability: __node_app_handle_capability,
                    free: __node_app_free,
                }) as *const $crate::NodeAppVTable
            })) {
                Ok(ptr) => ptr,
                Err(_) => {
                    eprintln!("[node-app] panic in _node_app_entry");
                    std::ptr::null()
                }
            }
        }
    };
}

#[cfg(test)]
mod invocation_context_tests {
    use std::ffi::{c_char, c_void};
    use std::sync::Mutex;

    use super::*;

    pub(super) static TEST_LOCK: Mutex<()> = Mutex::new(());

    unsafe extern "C" fn noop_log(_: *const c_void, _: u32, _: *const c_char) {}
    unsafe extern "C" fn no_config(_: *const c_void, _: *const c_char) -> *const c_char {
        std::ptr::null()
    }
    unsafe extern "C" fn noop_storage(_: *const c_void, _: *const c_char, _: *const c_char) {}
    unsafe extern "C" fn no_storage(_: *const c_void, _: *const c_char) -> *const c_char {
        std::ptr::null()
    }
    unsafe extern "C" fn noop_publish(
        _: *const c_void,
        _: *const u8,
        _: usize,
        _: *const u8,
        _: usize,
    ) -> i32 {
        0
    }
    unsafe extern "C" fn capture_invoke(
        host_data: *const c_void,
        request_json: *const u8,
        request_len: usize,
    ) -> FfiResult {
        let requests = &*(host_data as *const Mutex<Vec<CapabilityRequest>>);
        let bytes = std::slice::from_raw_parts(request_json, request_len);
        requests
            .lock()
            .unwrap()
            .push(serde_json::from_slice(bytes).unwrap());
        let response = CapabilityResponse {
            id: "nested-1".into(),
            success: true,
            payload: serde_json::json!({"ok": true}),
        };
        let bytes = serde_json::to_vec(&response).unwrap().into_boxed_slice();
        let len = bytes.len();
        let data = Box::into_raw(bytes) as *mut u8;
        FfiResult {
            success: true,
            error_code: 0,
            data,
            data_len: len,
        }
    }

    #[test]
    fn nested_invocation_propagates_opaque_context() {
        let _serial = TEST_LOCK.lock().unwrap();
        let captured = Mutex::new(Vec::<CapabilityRequest>::new());
        let context = NodeAppContext {
            host_data: &captured as *const _ as *const c_void,
            host_log: noop_log,
            host_get_config: no_config,
            host_set_storage: noop_storage,
            host_get_storage: no_storage,
            host_invoke_capability: capture_invoke,
            host_publish_event: noop_publish,
        };
        __store_context(&context);
        CURRENT_TRACE.with(|current| {
            *current.borrow_mut() = Some(CurrentTrace {
                trace_id: "trace-1".into(),
                span_id: "span-1".into(),
                depth: 4,
            });
        });
        let _invocation =
            CurrentInvocationContextGuard::enter(Some("opaque-context-1".into()));

        invoke_capability(&CapabilityRequest {
            id: "nested-1".into(),
            capability: "example.nested".into(),
            payload: serde_json::json!({}),
            ..Default::default()
        })
        .unwrap();

        let requests = captured.lock().unwrap();
        assert_eq!(requests.len(), 1);
        assert_eq!(
            requests[0].invocation_context_id.as_deref(),
            Some("opaque-context-1")
        );
        assert_eq!(requests[0].trace_id.as_deref(), Some("trace-1"));
        assert_eq!(requests[0].span_id.as_deref(), Some("span-1"));
        drop(requests);
        CURRENT_TRACE.with(|current| *current.borrow_mut() = None);
        __store_context(std::ptr::null());
    }
}

#[cfg(test)]
mod shutdown_gate_tests {
    use std::sync::atomic::AtomicBool;
    use std::sync::{Condvar, Mutex, PoisonError};
    use std::time::Duration;

    use super::*;

    struct Signal {
        set: Mutex<bool>,
        changed: Condvar,
    }

    impl Signal {
        const fn new() -> Self {
            Self {
                set: Mutex::new(false),
                changed: Condvar::new(),
            }
        }

        fn set(&self) {
            *self.set.lock().unwrap() = true;
            self.changed.notify_all();
        }

        fn reset(&self) {
            *self.set.lock().unwrap() = false;
        }

        fn is_set(&self) -> bool {
            *self.set.lock().unwrap()
        }

        fn wait(&self, timeout: Duration) -> bool {
            let set = self.set.lock().unwrap();
            *self
                .changed
                .wait_timeout_while(set, timeout, |set| !*set)
                .unwrap()
                .0
        }
    }

    // Tests that use the app reset these first and hold TEST_LOCK: the macro
    // allows one app per binary, so they share it.
    static CALL_ENTERED: Signal = Signal::new();
    static RELEASE_CALL: Signal = Signal::new();
    static HOOK_RAN: Signal = Signal::new();
    static SHUTDOWN_RAN: Signal = Signal::new();
    static PANIC_IN_HOOK: AtomicBool = AtomicBool::new(false);

    #[derive(Default)]
    struct SlowApp;

    impl NodeApp for SlowApp {
        fn metadata() -> NodeAppInfo {
            NodeAppInfo {
                name: "slow".into(),
                version: "0.0.0".into(),
                author: "test".into(),
                description: "holds a capability call open".into(),
                capabilities: Vec::new(),
            }
        }

        fn begin_shutdown(&self) {
            HOOK_RAN.set();
            if PANIC_IN_HOOK.load(Ordering::SeqCst) {
                panic!("begin_shutdown test panic");
            }
        }

        fn shutdown(&mut self) -> Result<(), NodeAppError> {
            SHUTDOWN_RAN.set();
            Ok(())
        }

        fn handle_capability(
            &self,
            request: CapabilityRequest,
        ) -> Result<CapabilityResponse, NodeAppError> {
            CALL_ENTERED.set();
            // Holds the app's read lock until the test lets go. The bound only
            // keeps a failing test from hanging.
            if !RELEASE_CALL.wait(Duration::from_secs(10)) {
                return Err(NodeAppError::CapabilityError("never released".into()));
            }
            Ok(CapabilityResponse {
                id: request.id,
                success: true,
                payload: serde_json::json!({}),
            })
        }
    }

    crate::declare_node_app!(SlowApp);

    /// 0 on success, else the wrapper's error code. Frees any response.
    fn code(result: FfiResult) -> i32 {
        if !result.success {
            return result.error_code;
        }
        if !result.data.is_null() {
            unsafe { __node_app_free(result.data, result.data_len) };
        }
        0
    }

    fn call() -> i32 {
        let request = serde_json::to_vec(&CapabilityRequest {
            id: "call-1".into(),
            capability: "slow.wait".into(),
            payload: serde_json::json!({}),
            ..Default::default()
        })
        .unwrap();
        code(unsafe { __node_app_handle_capability(request.as_ptr(), request.len()) })
    }

    fn request() -> i32 {
        let request = serde_json::to_vec(&AppRequest {
            id: "request-1".into(),
            method: "GET".into(),
            path: "/".into(),
            headers: Default::default(),
            body: serde_json::Value::Null,
            caller: None,
        })
        .unwrap();
        code(unsafe { __node_app_handle_request(request.as_ptr(), request.len()) })
    }

    fn event() -> i32 {
        let event = serde_json::to_vec(&AppEvent {
            name: "slow.tick".into(),
            data: serde_json::Value::Null,
        })
        .unwrap();
        code(unsafe { __node_app_handle_event(event.as_ptr(), event.len()) })
    }

    /// Stop the app while a capability call holds its read lock.
    fn stop_while_a_call_is_in_flight(panic_in_hook: bool) {
        for signal in [&CALL_ENTERED, &RELEASE_CALL, &HOOK_RAN, &SHUTDOWN_RAN] {
            signal.reset();
        }
        PANIC_IN_HOOK.store(panic_in_hook, Ordering::SeqCst);
        assert!(unsafe { __node_app_init(std::ptr::null()) }.success);

        let in_flight = std::thread::spawn(call);
        assert!(
            CALL_ENTERED.wait(Duration::from_secs(5)),
            "call never started"
        );

        let shutdown = std::thread::spawn(|| unsafe { __node_app_shutdown() }.success);
        // The in-flight call cannot return before RELEASE_CALL, so the hook
        // runs while that call still holds the read lock.
        assert!(
            HOOK_RAN.wait(Duration::from_secs(1)),
            "begin_shutdown waited behind the in-flight call"
        );
        assert_eq!(call(), ERROR_SHUTTING_DOWN, "capability call not refused");
        assert_eq!(request(), ERROR_SHUTTING_DOWN, "request not refused");
        assert_eq!(event(), ERROR_SHUTTING_DOWN, "event not refused");
        assert!(
            !SHUTDOWN_RAN.is_set(),
            "shutdown(&mut self) ran while a call was in flight"
        );

        RELEASE_CALL.set();
        assert_eq!(in_flight.join().unwrap(), 0, "in-flight call failed");
        assert!(shutdown.join().unwrap(), "shutdown entry failed");
        assert!(SHUTDOWN_RAN.is_set());
    }

    #[test]
    fn shutdown_hook_runs_while_a_call_is_in_flight() {
        let _serial = super::invocation_context_tests::TEST_LOCK
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        stop_while_a_call_is_in_flight(false);

        // A cached library is initialized again on reload; it takes calls again.
        assert!(unsafe { __node_app_init(std::ptr::null()) }.success);
        assert_eq!(call(), 0, "capability call refused after init");
        assert_eq!(request(), 0, "request refused after init");
        assert_eq!(event(), 0, "event refused after init");
    }

    #[test]
    fn panicking_shutdown_hook_still_waits_for_in_flight_calls() {
        let _serial = super::invocation_context_tests::TEST_LOCK
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        stop_while_a_call_is_in_flight(true);
    }

    #[test]
    fn call_gate_counts_each_call_until_its_permit_drops() {
        let gate = CallGate::new();
        let first = gate.enter().expect("open gate admits");
        let second = gate.enter().expect("open gate admits");
        gate.close();
        assert!(gate.enter().is_none(), "closed gate admitted a call");
        assert_eq!(gate.wait_drained(Some(Duration::from_millis(10))), 2);
        drop(first);
        assert_eq!(gate.wait_drained(Some(Duration::from_millis(10))), 1);
        drop(second);
        assert_eq!(gate.wait_drained(None), 0);
        gate.open();
        assert!(gate.enter().is_some(), "reopened gate refused a call");
    }
}