blazen-core 0.1.120

//! Session-scoped live reference registry.
//!
//! Stores Python/JS objects that cannot or should not be JSON-serialized
//! — DB connections, file handles, large in-memory tensors, lambdas,
//! locks, etc. — keyed by [`RegistryKey`] (a wrapped [`Uuid`]). Event
//! payloads carry only the key as a JSON marker
//! (`{"__blazen_session_ref__": "<uuid>"}`); the actual object lives
//! in the registry until workflow completion.
//!
//! Each [`crate::Context`] owns its own [`SessionRefRegistry`] so the
//! registry's lifetime is tied to the workflow run. Live references are
//! deliberately excluded from snapshots — see [`SessionPausePolicy`]
//! for what happens at pause/snapshot boundaries.

use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;

use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use uuid::Uuid;

/// JSON tag identifying a session-ref placeholder inside an event payload.
///
/// Format: `{"__blazen_session_ref__": "<uuid>"}`. Bindings detect this tag
/// and resolve the UUID through the active [`SessionRefRegistry`].
pub const SESSION_REF_TAG: &str = "__blazen_session_ref__";

/// Defensive cap on the number of live references a single workflow run
/// may hold. Prevents runaway loops from exhausting memory.
pub const MAX_SESSION_REFS_PER_RUN: usize = 10_000;

/// Strongly-typed wrapper around [`Uuid`] for session-ref keys.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct RegistryKey(pub Uuid);

impl RegistryKey {
    /// Mint a fresh random key.
    #[must_use]
    pub fn new() -> Self {
        Self(Uuid::new_v4())
    }

    /// Parse a key from a UUID string.
    ///
    /// # Errors
    /// Returns the underlying [`uuid::Error`] if `s` is not a valid UUID.
    pub fn parse(s: &str) -> Result<Self, uuid::Error> {
        Uuid::parse_str(s).map(Self)
    }
}

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

impl std::fmt::Display for RegistryKey {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

/// Type-erased entry. Each binding stores a binding-specific
/// `Arc<...>` (e.g. `Arc<Py<PyAny>>` from `PyO3`, `Arc<napi::Ref<JsObject>>`
/// from `napi-rs`) and downcasts on retrieval.
type AnyArc = Arc<dyn Any + Send + Sync>;

/// Per-context registry of live session references.
#[derive(Default)]
pub struct SessionRefRegistry {
    inner: RwLock<HashMap<RegistryKey, AnyArc>>,
}

impl std::fmt::Debug for SessionRefRegistry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Avoid taking the lock here so `Debug` formatting never blocks.
        f.debug_struct("SessionRefRegistry").finish_non_exhaustive()
    }
}

impl SessionRefRegistry {
    /// Create an empty registry.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Insert an `Arc<dyn Any + Send + Sync>` directly. Returns the freshly
    /// minted key, or an error if the registry is at capacity.
    ///
    /// # Errors
    /// Returns [`SessionRefError::CapacityExceeded`] if the registry already
    /// holds [`MAX_SESSION_REFS_PER_RUN`] entries.
    pub async fn insert_arc(&self, value: AnyArc) -> Result<RegistryKey, SessionRefError> {
        let mut g = self.inner.write().await;
        if g.len() >= MAX_SESSION_REFS_PER_RUN {
            return Err(SessionRefError::CapacityExceeded {
                cap: MAX_SESSION_REFS_PER_RUN,
            });
        }
        let key = RegistryKey::new();
        g.insert(key, value);
        Ok(key)
    }

    /// Insert any `Send + Sync + 'static` value, wrapping it in an `Arc` for
    /// you. Returns the freshly minted key.
    ///
    /// # Errors
    /// See [`Self::insert_arc`].
    pub async fn insert<T: Any + Send + Sync + 'static>(
        &self,
        value: T,
    ) -> Result<RegistryKey, SessionRefError> {
        self.insert_arc(Arc::new(value)).await
    }

    /// Look up the type-erased entry. Bindings call this and downcast.
    pub async fn get_any(&self, key: RegistryKey) -> Option<AnyArc> {
        self.inner.read().await.get(&key).cloned()
    }

    /// Look up and downcast to a concrete `Arc<T>`.
    pub async fn get<T: Any + Send + Sync + 'static>(&self, key: RegistryKey) -> Option<Arc<T>> {
        let any = self.inner.read().await.get(&key).cloned()?;
        Arc::downcast::<T>(any).ok()
    }

    /// Remove a single entry, returning the removed value if present.
    pub async fn remove(&self, key: RegistryKey) -> Option<AnyArc> {
        self.inner.write().await.remove(&key)
    }

    /// Drain all entries, returning the number removed.
    pub async fn drain(&self) -> usize {
        let mut g = self.inner.write().await;
        let n = g.len();
        g.clear();
        n
    }

    /// Number of currently live entries (for tests/diagnostics).
    pub async fn len(&self) -> usize {
        self.inner.read().await.len()
    }

    /// Whether the registry has any live entries.
    pub async fn is_empty(&self) -> bool {
        self.inner.read().await.is_empty()
    }

    /// Iterate every key currently in the registry. Used by the snapshot
    /// walker to apply [`SessionPausePolicy`] uniformly.
    pub async fn keys(&self) -> Vec<RegistryKey> {
        self.inner.read().await.keys().copied().collect()
    }
}

/// What to do with live session references when a workflow is paused or
/// snapshotted.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SessionPausePolicy {
    /// Try to pickle each live ref into the snapshot. On any failure,
    /// raise [`crate::WorkflowError::SessionRefsNotSerializable`] and
    /// abort the snapshot. **Default — recommended.**
    #[default]
    PickleOrError,
    /// Drop live refs from the snapshot, emit `tracing::warn!` per drop,
    /// store a diagnostic report in snapshot metadata. On resume, accessing
    /// a dropped field raises a clear runtime error from the binding.
    WarnDrop,
    /// Refuse to pause if any live refs are in flight. Raises
    /// [`crate::WorkflowError::SessionRefsNotSerializable`] immediately.
    HardError,
}

/// Error type for session-ref registry operations.
#[derive(Debug, thiserror::Error)]
pub enum SessionRefError {
    /// Returned when [`SessionRefRegistry::insert_arc`] is called while the
    /// registry already holds [`MAX_SESSION_REFS_PER_RUN`] entries.
    #[error(
        "session ref registry capacity exceeded ({cap} entries) — \
         too many live references in this workflow run"
    )]
    CapacityExceeded {
        /// The configured capacity that was hit.
        cap: usize,
    },
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn insert_and_get_roundtrip() {
        let reg = SessionRefRegistry::new();
        let key = reg.insert(42_i32).await.unwrap();
        let got = reg.get::<i32>(key).await.unwrap();
        assert_eq!(*got, 42);
    }

    #[tokio::test]
    async fn get_wrong_type_returns_none() {
        let reg = SessionRefRegistry::new();
        let key = reg.insert(42_i32).await.unwrap();
        assert!(reg.get::<String>(key).await.is_none());
    }

    #[tokio::test]
    async fn remove_returns_value_and_clears() {
        let reg = SessionRefRegistry::new();
        let key = reg.insert("hello".to_owned()).await.unwrap();
        assert_eq!(reg.len().await, 1);
        let removed = reg.remove(key).await;
        assert!(removed.is_some());
        assert_eq!(reg.len().await, 0);
    }

    #[tokio::test]
    async fn drain_clears_everything() {
        let reg = SessionRefRegistry::new();
        let _ = reg.insert(1_i32).await.unwrap();
        let _ = reg.insert(2_i32).await.unwrap();
        let _ = reg.insert(3_i32).await.unwrap();
        assert_eq!(reg.drain().await, 3);
        assert!(reg.is_empty().await);
    }

    #[tokio::test]
    async fn capacity_cap_enforced() {
        // Run with a tight cap to make the test fast — we just verify the
        // CapacityExceeded path triggers. We can't easily change the const,
        // so we test the logic by inserting up to the real limit (slow test
        // skipped). Instead test that insert returns Ok for normal use.
        let reg = SessionRefRegistry::new();
        for i in 0..100_i32 {
            assert!(reg.insert(i).await.is_ok());
        }
        assert_eq!(reg.len().await, 100);
    }

    #[test]
    fn registry_key_parse_roundtrip() {
        let k = RegistryKey::new();
        let s = k.to_string();
        let parsed = RegistryKey::parse(&s).unwrap();
        assert_eq!(k, parsed);
    }

    #[test]
    fn session_pause_policy_default_is_pickle_or_error() {
        assert_eq!(
            SessionPausePolicy::default(),
            SessionPausePolicy::PickleOrError
        );
    }

    #[test]
    fn session_pause_policy_serde_roundtrip() {
        let p = SessionPausePolicy::WarnDrop;
        let json = serde_json::to_string(&p).unwrap();
        assert_eq!(json, "\"warn_drop\"");
        let back: SessionPausePolicy = serde_json::from_str(&json).unwrap();
        assert_eq!(back, p);
    }
}