meerkat_runtime/handles/

interaction_stream.rs

//! Runtime impl of [`meerkat_core::handles::InteractionStreamHandle`] (U6).
//!
//! Routes the interaction-stream lifecycle (`Reserved` / `Attached` /
//! `Completed` / `Expired` / `ClosedEarly` / `Abandoned`) into the session's MeerkatMachine
//! DSL `interaction_streams` substate map and fans the emitted
//! `InteractionStreamCleanup` effects out to the installed shell-side
//! observer, so the comms runtime's `interaction_stream_registry` becomes a
//! pure projection of DSL truth — no shadow `state` field, no shell-side
//! CAS, no TTL meaning hidden in registry maps.

use std::sync::{Arc, RwLock, Weak};

use meerkat_core::handles::{
    DslTransitionError, InteractionStreamCleanupObserver, InteractionStreamHandle,
};
use meerkat_core::peer_correlation::{
    InteractionStreamState as CoreInteractionStreamState, PeerCorrelationId,
};

use super::HandleDslAuthority;
use crate::meerkat_machine::dsl as mm_dsl;

/// Runtime-backed [`InteractionStreamHandle`] impl.
///
/// Every trait method routes to the corresponding DSL input on the session's
/// shared MeerkatMachine authority. After the transition lands, emitted
/// effects are scanned for `InteractionStreamCleanup` and dispatched to the
/// installed [`InteractionStreamCleanupObserver`] (if any), closing the
/// "terminal transition → effect → shell projection cleanup" loop.
///
/// The observer is held as a `Weak` reference because the canonical owner is
/// the session's `CommsRuntime`, which in turn holds a strong handle pointer;
/// storing the observer strongly would create a cycle preventing
/// `CommsRuntime::drop` from firing on session teardown. `Weak::upgrade`
/// returning `None` after teardown makes cleanup dispatch a no-op, which is
/// the desired post-shutdown semantics.
pub struct RuntimeInteractionStreamHandle {
    dsl: Arc<HandleDslAuthority>,
    cleanup_observer: RwLock<Option<Weak<dyn InteractionStreamCleanupObserver>>>,
}

impl std::fmt::Debug for RuntimeInteractionStreamHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let observer_tag = self
            .cleanup_observer
            .read()
            .ok()
            .as_deref()
            .and_then(|o| o.as_ref().map(|_| "<observer>"));
        f.debug_struct("RuntimeInteractionStreamHandle")
            .field("dsl", &self.dsl)
            .field("cleanup_observer", &observer_tag)
            .finish()
    }
}

impl RuntimeInteractionStreamHandle {
    /// Construct a handle backed by the session's shared DSL authority.
    pub fn new(dsl: Arc<HandleDslAuthority>) -> Self {
        Self {
            dsl,
            cleanup_observer: RwLock::new(None),
        }
    }

    /// Construct a handle backed by an ephemeral DSL authority for tests and
    /// minimal hosts that explicitly opt into machine-owned semantics.
    pub fn ephemeral() -> Self {
        Self::new(Arc::new(HandleDslAuthority::ephemeral()))
    }

    fn apply_input_and_dispatch_cleanup(
        &self,
        input: mm_dsl::MeerkatMachineInput,
        context: &'static str,
    ) -> Result<(), DslTransitionError> {
        // Sample observer UNDER the DSL lock, dispatch OUTSIDE — same
        // race-closing pattern as `session_context.rs`.
        type CleanupTarget = Result<
            (
                PeerCorrelationId,
                Option<meerkat_core::InteractionStreamAbandonReason>,
            ),
            String,
        >;
        let dispatch: Option<(
            Arc<dyn InteractionStreamCleanupObserver>,
            Vec<CleanupTarget>,
        )> = self
            .dsl
            .apply_input_with_effects_and_sample(input, context, |effects| {
                let observer_opt = self
                    .cleanup_observer
                    .read()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .as_ref()
                    .and_then(Weak::upgrade);
                let observer = observer_opt?;
                let targets: Vec<CleanupTarget> = effects
                    .iter()
                    .filter_map(|effect| match effect {
                        mm_dsl::MeerkatMachineEffect::InteractionStreamCleanup {
                            corr_id,
                            abandon_reason,
                        } => Some(match dsl_corr_id_to_core(corr_id.clone()) {
                            Some(core_id) => Ok((core_id, (*abandon_reason).map(Into::into))),
                            None => Err(corr_id.0.clone()),
                        }),
                        _ => None,
                    })
                    .collect();
                Some((observer, targets))
            })?;
        if let Some((observer, targets)) = dispatch {
            for target in targets {
                match target {
                    Ok((core_id, abandon_reason)) => {
                        observer.on_interaction_stream_cleanup(core_id, abandon_reason);
                    }
                    Err(raw) => tracing::error!(
                        raw = %raw,
                        context = context,
                        "InteractionStreamCleanup: DSL emitted a corr_id that is not a valid UUID — broken invariant; skipping observer dispatch"
                    ),
                }
            }
        }
        Ok(())
    }
}

fn dsl_corr_id_to_core(dsl_id: mm_dsl::PeerCorrelationId) -> Option<PeerCorrelationId> {
    uuid::Uuid::parse_str(&dsl_id.0)
        .ok()
        .map(PeerCorrelationId::from_uuid)
}

impl InteractionStreamHandle for RuntimeInteractionStreamHandle {
    fn reserved(&self, corr_id: PeerCorrelationId) -> Result<(), DslTransitionError> {
        self.apply_input_and_dispatch_cleanup(
            mm_dsl::MeerkatMachineInput::InteractionStreamReserved {
                corr_id: corr_id.into(),
            },
            "InteractionStreamHandle::reserved",
        )
    }

    fn attached(&self, corr_id: PeerCorrelationId) -> Result<(), DslTransitionError> {
        self.apply_input_and_dispatch_cleanup(
            mm_dsl::MeerkatMachineInput::InteractionStreamAttached {
                corr_id: corr_id.into(),
            },
            "InteractionStreamHandle::attached",
        )
    }

    fn completed(&self, corr_id: PeerCorrelationId) -> Result<(), DslTransitionError> {
        self.apply_input_and_dispatch_cleanup(
            mm_dsl::MeerkatMachineInput::InteractionStreamCompleted {
                corr_id: corr_id.into(),
            },
            "InteractionStreamHandle::completed",
        )
    }

    fn expired(&self, corr_id: PeerCorrelationId) -> Result<(), DslTransitionError> {
        self.apply_input_and_dispatch_cleanup(
            mm_dsl::MeerkatMachineInput::InteractionStreamExpired {
                corr_id: corr_id.into(),
            },
            "InteractionStreamHandle::expired",
        )
    }

    fn closed_early(&self, corr_id: PeerCorrelationId) -> Result<(), DslTransitionError> {
        self.apply_input_and_dispatch_cleanup(
            mm_dsl::MeerkatMachineInput::InteractionStreamClosedEarly {
                corr_id: corr_id.into(),
            },
            "InteractionStreamHandle::closed_early",
        )
    }

    fn abandoned(
        &self,
        corr_id: PeerCorrelationId,
        reason: meerkat_core::InteractionStreamAbandonReason,
    ) -> Result<(), DslTransitionError> {
        self.apply_input_and_dispatch_cleanup(
            mm_dsl::MeerkatMachineInput::InteractionStreamAbandoned {
                corr_id: corr_id.into(),
                reason: reason.into(),
            },
            "InteractionStreamHandle::abandoned",
        )
    }

    fn state(&self, corr_id: PeerCorrelationId) -> Option<CoreInteractionStreamState> {
        let dsl_key: mm_dsl::PeerCorrelationId = corr_id.into();
        let snapshot = self.dsl.snapshot_state();
        // Disjoint-set encoding (matches the DSL `interaction_stream_disjoint`
        // discipline): a corr_id is in at most one of the two active sets.
        // Terminal states (`Completed` / `Expired` / `ClosedEarly` /
        // `Abandoned`) leave both
        // sets and are never observable here — they surface only via the
        // `InteractionStreamStateChanged` effect, like the peer-correlation
        // sibling enums.
        if snapshot.attached_interaction_streams.contains(&dsl_key) {
            Some(CoreInteractionStreamState::Attached)
        } else if snapshot.reserved_interaction_streams.contains(&dsl_key) {
            Some(CoreInteractionStreamState::Reserved)
        } else {
            None
        }
    }

    fn install_cleanup_observer(&self, observer: Arc<dyn InteractionStreamCleanupObserver>) {
        *self
            .cleanup_observer
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(Arc::downgrade(&observer));
    }
}

#[cfg(test)]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    fn new_handle() -> RuntimeInteractionStreamHandle {
        let mut authority = mm_dsl::MeerkatMachineAuthority::new();
        authority
            .apply_signal(mm_dsl::MeerkatMachineSignal::Initialize)
            .unwrap();
        mm_dsl::MeerkatMachineMutator::apply(
            &mut authority,
            mm_dsl::MeerkatMachineInput::RegisterSession {
                session_id: mm_dsl::SessionId::from("interaction-stream-test".to_string()),
            },
        )
        .unwrap();
        let dsl = Arc::new(HandleDslAuthority::from_shared(Arc::new(Mutex::new(
            authority,
        ))));
        RuntimeInteractionStreamHandle::new(dsl)
    }

    struct CleanupRecorder(
        Mutex<
            Vec<(
                PeerCorrelationId,
                Option<meerkat_core::InteractionStreamAbandonReason>,
            )>,
        >,
    );

    impl InteractionStreamCleanupObserver for CleanupRecorder {
        fn on_interaction_stream_cleanup(
            &self,
            corr_id: PeerCorrelationId,
            abandon_reason: Option<meerkat_core::InteractionStreamAbandonReason>,
        ) {
            self.0.lock().unwrap().push((corr_id, abandon_reason));
        }
    }

    #[test]
    fn abandoned_is_typed_terminal_from_reserved_or_attached() {
        let handle = new_handle();
        let recorder = Arc::new(CleanupRecorder(Mutex::new(Vec::new())));
        handle.install_cleanup_observer(
            recorder.clone() as Arc<dyn InteractionStreamCleanupObserver>
        );

        let reserved = PeerCorrelationId::new();
        handle.reserved(reserved).unwrap();
        handle
            .abandoned(
                reserved,
                meerkat_core::InteractionStreamAbandonReason::SendFailed,
            )
            .unwrap();
        assert_eq!(handle.state(reserved), None);

        let attached = PeerCorrelationId::new();
        handle.reserved(attached).unwrap();
        handle.attached(attached).unwrap();
        handle
            .abandoned(
                attached,
                meerkat_core::InteractionStreamAbandonReason::TerminalDeliveryFailed,
            )
            .unwrap();
        assert_eq!(handle.state(attached), None);
        assert_eq!(
            *recorder.0.lock().unwrap(),
            vec![
                (
                    reserved,
                    Some(meerkat_core::InteractionStreamAbandonReason::SendFailed),
                ),
                (
                    attached,
                    Some(meerkat_core::InteractionStreamAbandonReason::TerminalDeliveryFailed,),
                ),
            ]
        );
    }
}