liminal_sdk/

embedded.rs

use alloc::format;
use alloc::string::{String, ToString};
use alloc::sync::Arc;
use core::fmt;
use core::marker::PhantomData;
use core::pin::Pin;
use core::task::{Context, Poll};

use futures_core::Stream;
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::{
    ChannelHandle, ConversationEvent, ConversationHandle, ConversationId, PressureResponse,
    SchemaMetadata, SchemaValidate, SdkError,
};

#[cfg(test)]
mod tests;

/// Type-erased view of a typed channel message passed to an embedded backend.
///
/// The embedded SDK path hands this reference directly to the in-process backend;
/// it never asks serde to produce bytes, protocol envelopes, or wire frames.
pub trait EmbeddedChannelMessage {
    /// Returns the compile-time schema metadata declared by the message type.
    #[must_use]
    fn schema_metadata(&self) -> SchemaMetadata;

    /// Returns the Rust type name used for diagnostics and backend routing.
    #[must_use]
    fn type_name(&self) -> &'static str;
}

impl<M> EmbeddedChannelMessage for M
where
    M: Serialize + SchemaValidate,
{
    fn schema_metadata(&self) -> SchemaMetadata {
        M::schema_metadata()
    }

    fn type_name(&self) -> &'static str {
        core::any::type_name::<M>()
    }
}

/// Type-erased view of a typed conversation message passed to an embedded backend.
pub trait EmbeddedConversationMessage {
    /// Returns the Rust type name used for diagnostics and backend routing.
    #[must_use]
    fn type_name(&self) -> &'static str;
}

impl<M> EmbeddedConversationMessage for M
where
    M: Serialize,
{
    fn type_name(&self) -> &'static str {
        core::any::type_name::<M>()
    }
}

/// Stream used by SDK handles when no typed messages are buffered locally.
pub struct SdkSubscription<M> {
    pending_error: Option<SdkError>,
    message: PhantomData<M>,
}

impl<M> SdkSubscription<M> {
    /// Creates an empty typed subscription stream.
    #[must_use]
    pub const fn empty() -> Self {
        Self {
            pending_error: None,
            message: PhantomData,
        }
    }

    /// Creates a subscription stream that surfaces a setup error to the caller.
    #[must_use]
    pub const fn error(error: SdkError) -> Self {
        Self {
            pending_error: Some(error),
            message: PhantomData,
        }
    }

    /// Returns true when this local subscription has no pending setup error.
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.pending_error.is_none()
    }
}

impl<M> Default for SdkSubscription<M> {
    fn default() -> Self {
        Self::empty()
    }
}

impl<M> fmt::Debug for SdkSubscription<M> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SdkSubscription")
            .field("has_pending_error", &self.pending_error.is_some())
            .finish()
    }
}

impl<M> Unpin for SdkSubscription<M> {}

impl<M> Stream for SdkSubscription<M> {
    type Item = Result<M, SdkError>;

    fn poll_next(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        core::hint::black_box(context.waker());
        Poll::Ready(self.pending_error.take().map(Err))
    }
}

/// Stream used when a conversation has no lifecycle events buffered locally.
#[derive(Clone, Debug, Default)]
pub struct EmptyLifecycleStream;

impl Unpin for EmptyLifecycleStream {}

impl Stream for EmptyLifecycleStream {
    type Item = ConversationEvent;

    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        core::hint::black_box((self, context.waker()));
        Poll::Ready(None)
    }
}

/// Ready future used by SDK handles that complete an operation synchronously.
pub struct ReadyResult<T> {
    result: Option<Result<T, SdkError>>,
}

impl<T> fmt::Debug for ReadyResult<T> {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ReadyResult")
            .field("is_ready", &self.result.is_some())
            .finish()
    }
}

impl<T> ReadyResult<T> {
    /// Creates a ready future from a result.
    #[must_use]
    pub const fn new(result: Result<T, SdkError>) -> Self {
        Self {
            result: Some(result),
        }
    }
}

impl<T> Unpin for ReadyResult<T> {}

impl<T> core::future::Future for ReadyResult<T> {
    type Output = Result<T, SdkError>;

    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
        core::hint::black_box(context.waker());
        let Some(result) = self.result.take() else {
            return Poll::Ready(Err(SdkError::Protocol {
                description: "ready future polled after completion".to_string(),
            }));
        };

        Poll::Ready(result)
    }
}

/// Direct in-process channel backend used by embedded handles.
///
/// Publish only. There is no subscribe or deliver method here and never has
/// been, which is why [`ChannelHandle::subscribe`] on an
/// [`EmbeddedChannelHandle`] has no delivery path to reach and refuses instead
/// of returning a stream nothing can feed.
pub trait EmbeddedChannelBackend: fmt::Debug + Send + Sync {
    /// Publishes a typed message reference without protocol framing or wire encoding.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if the in-process backend rejects the publish attempt.
    fn publish(&self, message: &dyn EmbeddedChannelMessage) -> Result<PressureResponse, SdkError>;
}

/// Direct in-process conversation backend used by embedded handles.
pub trait EmbeddedConversationBackend: fmt::Debug + Send + Sync {
    /// Sends a typed message reference without protocol framing or wire encoding.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError`] if the in-process backend rejects the send attempt.
    fn send(&self, message: &dyn EmbeddedConversationMessage) -> Result<(), SdkError>;
}

/// Ledger row that records why embedded mode has no in-process bus behind it.
const B1_SEAM: &str = "seam B1 — Embedded/frame-conv integration \
                       (docs/stack-review/liminal-ledger.md)";

/// The refusal embedded typed `subscribe` returns instead of an instantly-dry
/// stream.
///
/// Embedded typed subscribe delivery is unimplemented, and unimplemented in a
/// stronger sense than the publish seam beside it: publish at least has a
/// backend trait a caller can implement, while [`EmbeddedChannelBackend`]
/// declares `publish` and nothing else. There is no subscribe or deliver method
/// on it, so no backend installed through
/// [`EmbeddedConfig::with_channel_backend`] can feed a subscription either.
/// Nothing in this crate delivers a message to an embedded subscriber.
///
/// No ledger row records this gap. B1 — the row named by [`B1_SEAM`], and the
/// row this module already cites for the default backends — is about the SDK's
/// no-op embedded *backend*, i.e. the publish/send seam that a shared-scheduler
/// `liminal`-core integration would fill; read at
/// `docs/stack-review/liminal-ledger.md` it says nothing about subscription
/// delivery, and the backend it describes has no subscription surface to fill.
/// This refusal says so rather than borrowing a row that does not cover it.
fn unwired_embedded_subscribe() -> SdkError {
    SdkError::Unwired {
        surface: "typed subscribe delivery (ChannelHandle::subscribe on an embedded handle)"
            .to_string(),
        seam: "no ledger row records this gap; B1 covers the no-op embedded backend, which is \
               publish-only and has no subscription surface at all"
            .to_string(),
        alternative: "embedded mode has no in-process delivery path to point at. For real \
                      delivery use a remote handle against a running server: \
                      SubscriptionStream::open plus SubscriptionStream::recv_timeout, or \
                      WebSocketSubscriptionStream::open on the WebSocket transport"
            .to_string(),
    }
}

/// The channel backend [`EmbeddedConfig::new`] installs by default — and it
/// REFUSES every publish.
///
/// Embedded mode has no in-process bus behind it. That absence is deliberate
/// and recorded: it is seam **B1**, "Embedded/frame-conv integration", in
/// `docs/stack-review/liminal-ledger.md`, which makes shared-scheduler
/// construction on the `liminal` core crate the blessed path and leaves this
/// backend as a labelled seam until then.
///
/// What it must NOT be is a seam that lies. It used to discard the message and
/// return [`PressureResponse::Accept`], so every embedded publish reported
/// success while nothing was delivered to anyone. It now returns
/// [`SdkError::Unwired`] naming the B1 seam and
/// [`EmbeddedConfig::with_channel_backend`], which is how a caller installs a
/// backend that genuinely delivers.
#[derive(Clone, Debug, Default)]
pub struct DirectEmbeddedChannelBackend;

impl EmbeddedChannelBackend for DirectEmbeddedChannelBackend {
    fn publish(&self, message: &dyn EmbeddedChannelMessage) -> Result<PressureResponse, SdkError> {
        Err(SdkError::Unwired {
            surface: format!(
                "the default embedded channel backend (publish of {})",
                message.type_name()
            ),
            seam: B1_SEAM.to_string(),
            alternative: "install an in-process backend that genuinely delivers with \
                          EmbeddedConfig::with_channel_backend"
                .to_string(),
        })
    }
}

/// The conversation backend [`EmbeddedConfig::new`] installs by default — and it
/// REFUSES every send.
///
/// The sibling of [`DirectEmbeddedChannelBackend`] and the same seam: it used to
/// discard the message and return `Ok(())`, reporting a delivery that never
/// happened. It now returns [`SdkError::Unwired`] naming the B1 seam and
/// [`EmbeddedConfig::with_conversation_backend`].
#[derive(Clone, Debug, Default)]
pub struct DirectEmbeddedConversationBackend;

impl EmbeddedConversationBackend for DirectEmbeddedConversationBackend {
    fn send(&self, message: &dyn EmbeddedConversationMessage) -> Result<(), SdkError> {
        Err(SdkError::Unwired {
            surface: format!(
                "the default embedded conversation backend (send of {})",
                message.type_name()
            ),
            seam: B1_SEAM.to_string(),
            alternative: "install an in-process backend that genuinely delivers with \
                          EmbeddedConfig::with_conversation_backend"
                .to_string(),
        })
    }
}

/// Configuration for embedded SDK handles.
#[derive(Clone, Debug)]
pub struct EmbeddedConfig {
    /// Application-visible channel name.
    pub channel_name: String,
    /// Application-visible conversation identifier.
    pub conversation_id: ConversationId,
    /// Direct channel backend used for in-process publication.
    pub channel_backend: Arc<dyn EmbeddedChannelBackend>,
    /// Direct conversation backend used for in-process conversation sends.
    pub conversation_backend: Arc<dyn EmbeddedConversationBackend>,
}

impl EmbeddedConfig {
    /// Creates embedded configuration without requiring a server address.
    ///
    /// **The default backends REFUSE.** Both
    /// [`DirectEmbeddedChannelBackend`] and [`DirectEmbeddedConversationBackend`]
    /// return [`SdkError::Unwired`] naming seam B1: embedded mode has no
    /// in-process bus behind it, and this configuration will not pretend
    /// otherwise by reporting an `Accept` nothing delivered. Install real
    /// backends with [`with_channel_backend`](Self::with_channel_backend) and
    /// [`with_conversation_backend`](Self::with_conversation_backend).
    #[must_use]
    pub fn new(
        channel_name: impl Into<String>,
        conversation_id: impl Into<ConversationId>,
    ) -> Self {
        Self {
            channel_name: channel_name.into(),
            conversation_id: conversation_id.into(),
            channel_backend: Arc::new(DirectEmbeddedChannelBackend),
            conversation_backend: Arc::new(DirectEmbeddedConversationBackend),
        }
    }

    /// Replaces the direct in-process channel backend.
    ///
    /// This installs a publish path and nothing else: [`EmbeddedChannelBackend`]
    /// has no subscribe or deliver method, so a backend installed here does not
    /// give [`ChannelHandle::subscribe`] a delivery path — that surface refuses
    /// either way.
    #[must_use]
    pub fn with_channel_backend(mut self, backend: Arc<dyn EmbeddedChannelBackend>) -> Self {
        self.channel_backend = backend;
        self
    }

    /// Replaces the direct in-process conversation backend.
    #[must_use]
    pub fn with_conversation_backend(
        mut self,
        backend: Arc<dyn EmbeddedConversationBackend>,
    ) -> Self {
        self.conversation_backend = backend;
        self
    }
}

/// Channel handle that publishes through direct in-process references.
///
/// Publishes — and only publishes. [`ChannelHandle::subscribe`] on this handle
/// REFUSES with [`SdkError::Unwired`]: embedded mode has no in-process delivery
/// path, and this handle will not hand back an empty stream that looks like a
/// live subscription with nothing on it yet.
#[derive(Clone, Debug)]
pub struct EmbeddedChannelHandle {
    channel_name: String,
    backend: Arc<dyn EmbeddedChannelBackend>,
}

impl EmbeddedChannelHandle {
    /// Creates an embedded channel handle from direct in-process configuration.
    #[must_use]
    pub fn new(config: &EmbeddedConfig) -> Self {
        Self {
            channel_name: config.channel_name.clone(),
            backend: Arc::clone(&config.channel_backend),
        }
    }

    /// Returns the application-visible channel name.
    #[must_use]
    pub fn channel_name(&self) -> &str {
        self.channel_name.as_str()
    }
}

impl ChannelHandle for EmbeddedChannelHandle {
    type Subscription<M>
        = SdkSubscription<M>
    where
        M: DeserializeOwned;

    type ReplyFuture<'a, Resp>
        = ReadyResult<Resp>
    where
        Self: 'a,
        Resp: DeserializeOwned + 'a;

    fn publish<M>(&self, message: M) -> Result<PressureResponse, SdkError>
    where
        M: Serialize + SchemaValidate,
    {
        self.backend.publish(&message)
    }

    /// REFUSES. Embedded typed subscription delivery is unimplemented.
    ///
    /// The returned stream yields exactly one item — `Err(SdkError::Unwired)` —
    /// and then ends. It touches no backend, because there is no backend method
    /// it could touch.
    ///
    /// What this replaces reported a success it could not have had. The old body
    /// returned [`SdkSubscription::empty`], a stream whose `poll_next` is
    /// `Ready(None)` immediately and forever, so a caller's
    /// `while let Some(m) = sub.next().await` exited on the first poll and read
    /// as "subscribed, nothing published yet" — indistinguishable from a quiet
    /// channel, and permanent.
    ///
    /// There is no in-process alternative to point at: [`EmbeddedChannelBackend`]
    /// is publish-only, so a backend installed through
    /// [`EmbeddedConfig::with_channel_backend`] gives this surface nothing. For
    /// real delivery use a remote handle against a running server —
    /// `SubscriptionStream::open` plus `SubscriptionStream::recv_timeout`, or
    /// `WebSocketSubscriptionStream::open` on the WebSocket transport.
    fn subscribe<M>(&self) -> Self::Subscription<M>
    where
        M: DeserializeOwned,
    {
        SdkSubscription::error(unwired_embedded_subscribe())
    }

    fn request_reply<Req, Resp>(&self, request: Req) -> ReadyResult<Resp>
    where
        Req: Serialize + SchemaValidate,
        Resp: DeserializeOwned,
    {
        let schema = Req::schema_metadata();
        core::hint::black_box((&request, &schema));
        ReadyResult::new(Err(SdkError::Protocol {
            description: "embedded request/reply requires an in-process responder backend"
                .to_string(),
        }))
    }
}

/// Conversation handle that sends through direct in-process references.
#[derive(Clone, Debug)]
pub struct EmbeddedConversationHandle {
    conversation_id: ConversationId,
    backend: Arc<dyn EmbeddedConversationBackend>,
}

impl EmbeddedConversationHandle {
    /// Creates an embedded conversation handle from direct in-process configuration.
    #[must_use]
    pub fn new(config: &EmbeddedConfig) -> Self {
        Self {
            conversation_id: config.conversation_id.clone(),
            backend: Arc::clone(&config.conversation_backend),
        }
    }

    /// Returns the application-visible conversation identifier.
    #[must_use]
    pub const fn conversation_id(&self) -> &ConversationId {
        &self.conversation_id
    }
}

impl ConversationHandle for EmbeddedConversationHandle {
    type ReceiveFuture<'a, M>
        = ReadyResult<M>
    where
        Self: 'a,
        M: DeserializeOwned + 'a;

    type LifecycleStream = EmptyLifecycleStream;

    fn send<M>(&self, message: M) -> Result<(), SdkError>
    where
        M: Serialize,
    {
        self.backend.send(&message)
    }

    fn receive<M>(&self) -> ReadyResult<M>
    where
        M: DeserializeOwned,
    {
        ReadyResult::new(Err(SdkError::Conversation {
            conversation_id: self.conversation_id.as_str().to_string(),
            description: "embedded receive requires an in-process inbox backend".to_string(),
        }))
    }

    fn lifecycle(&self) -> Self::LifecycleStream {
        EmptyLifecycleStream
    }
}