pjson_rs/application/handlers/

command_handlers.rs

//! Command handlers implementing business use cases

use crate::{
    application::{ApplicationError, ApplicationResult, commands::*, handlers::CommandHandlerGat},
    domain::{
        aggregates::StreamSession,
        config::limits::{MAX_FRAMES_PER_REQUEST, MAX_SESSION_TIMEOUT_SECONDS},
        entities::Frame,
        ports::{
            DictionaryStore, EventPublisherGat, FrameStoreGat, NoopDictionaryStore,
            StreamRepositoryGat,
        },
        value_objects::{SessionId, StreamId},
    },
    infrastructure::adapters::InMemoryFrameStore,
};
use std::sync::Arc;

/// Handler for session management commands.
///
/// Holds an optional [`DictionaryStore`] (defaulting to [`NoopDictionaryStore`])
/// so that frame-generating commands can feed accepted frame payloads into the
/// per-session training corpus. Without this wiring the
/// `GET /pjs/sessions/{id}/dictionary` endpoint would be unreachable end-to-end.
///
/// Also holds a [`FrameStoreGat`] (defaulting to [`InMemoryFrameStore`]) so
/// frames produced by `GenerateFramesCommand` / `BatchGenerateFramesCommand`
/// remain queryable through `GET /pjs/sessions/{id}/streams/{id}/frames`.
pub struct SessionCommandHandler<R, P, F = InMemoryFrameStore>
where
    R: StreamRepositoryGat + 'static,
    P: EventPublisherGat + 'static,
    F: FrameStoreGat + 'static,
{
    repository: Arc<R>,
    event_publisher: Arc<P>,
    #[cfg_attr(
        not(all(feature = "compression", not(target_arch = "wasm32"))),
        allow(dead_code)
    )]
    dictionary_store: Arc<dyn DictionaryStore>,
    frame_store: Arc<F>,
}

impl<R, P, F> std::fmt::Debug for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + 'static,
    P: EventPublisherGat + 'static,
    F: FrameStoreGat + 'static,
{
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SessionCommandHandler")
            .finish_non_exhaustive()
    }
}

impl<R, P> SessionCommandHandler<R, P, InMemoryFrameStore>
where
    R: StreamRepositoryGat + 'static,
    P: EventPublisherGat + 'static,
{
    /// Create a handler with the no-op [`DictionaryStore`] and a fresh
    /// [`InMemoryFrameStore`].
    ///
    /// The dictionary endpoint will return `404 Not Found` until the handler is
    /// constructed with [`SessionCommandHandler::with_stores`] and a concrete
    /// [`DictionaryStore`] such as
    /// [`crate::infrastructure::repositories::InMemoryDictionaryStore`].
    pub fn new(repository: Arc<R>, event_publisher: Arc<P>) -> Self {
        Self::with_dictionary_store(repository, event_publisher, Arc::new(NoopDictionaryStore))
    }

    /// Create a handler with a custom [`DictionaryStore`] and a fresh
    /// [`InMemoryFrameStore`].
    pub fn with_dictionary_store(
        repository: Arc<R>,
        event_publisher: Arc<P>,
        dictionary_store: Arc<dyn DictionaryStore>,
    ) -> Self {
        Self::with_stores(
            repository,
            event_publisher,
            dictionary_store,
            Arc::new(InMemoryFrameStore::new()),
        )
    }
}

impl<R, P, F> SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + 'static,
    P: EventPublisherGat + 'static,
    F: FrameStoreGat + 'static,
{
    /// Create a handler that feeds accepted frames into both the dictionary
    /// training corpus and the [`FrameStoreGat`] used by the frames query
    /// endpoint.
    pub fn with_stores(
        repository: Arc<R>,
        event_publisher: Arc<P>,
        dictionary_store: Arc<dyn DictionaryStore>,
        frame_store: Arc<F>,
    ) -> Self {
        Self {
            repository,
            event_publisher,
            dictionary_store,
            frame_store,
        }
    }

    /// Feed each accepted frame's serialized payload into the per-session
    /// training corpus.
    ///
    /// Errors are intentionally swallowed: training is best-effort and a
    /// transient failure must not poison the frame-generation response. The
    /// `OnceCell` inside `InMemoryDictionaryStore` is not poisoned on error
    /// either, so the next sample will retry.
    #[cfg(all(feature = "compression", not(target_arch = "wasm32")))]
    async fn train_from_frames(&self, session_id: SessionId, frames: &[Frame]) {
        for frame in frames {
            if let Ok(bytes) = serde_json::to_vec(frame.payload()) {
                let _ = self
                    .dictionary_store
                    .train_if_ready(session_id, bytes)
                    .await;
            }
        }
    }

    /// Persist a mutated session and publish its accumulated domain events.
    ///
    /// Events are drained from `session` before the persisted clone is taken,
    /// so the stored copy's `pending_events` buffer is empty and each
    /// subsequent call only publishes events produced since the last save.
    /// If `save_session` fails after the drain, those events are discarded
    /// rather than retried — there is no path back into `pending_events`.
    async fn save_and_publish(&self, session: &mut StreamSession) -> ApplicationResult<()>
    where
        R: StreamRepositoryGat + Send + Sync,
        P: EventPublisherGat + Send + Sync,
    {
        let events: Vec<_> = session.take_events().into_iter().collect();
        self.repository
            .save_session(session.clone())
            .await
            .map_err(ApplicationError::Domain)?;
        self.event_publisher
            .publish_batch(events)
            .await
            .map_err(ApplicationError::Domain)?;
        Ok(())
    }

    /// Persist generated frames into the [`FrameStoreGat`], grouping by
    /// `stream_id` so a single batch may span multiple streams.
    async fn persist_frames_grouped_by_stream(&self, frames: &[Frame]) -> ApplicationResult<()>
    where
        F: FrameStoreGat + Send + Sync,
    {
        if frames.is_empty() {
            return Ok(());
        }
        let mut buckets: std::collections::HashMap<StreamId, Vec<Frame>> =
            std::collections::HashMap::new();
        for frame in frames {
            buckets
                .entry(frame.stream_id())
                .or_default()
                .push(frame.clone());
        }
        for (stream_id, group) in buckets {
            self.frame_store
                .append_frames(stream_id, group)
                .await
                .map_err(ApplicationError::Domain)?;
        }
        Ok(())
    }
}

impl<R, P, F> CommandHandlerGat<CreateSessionCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = SessionId;

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: CreateSessionCommand) -> Self::HandleFuture<'_> {
        async move {
            CommandValidator::validate_create_session(&command)
                .map_err(|errors| ApplicationError::Validation(errors.join("; ")))?;

            // Create new session
            let mut session = StreamSession::new(command.config);

            // Set client information
            if let (Some(client_info), user_agent, ip_address) =
                (command.client_info, command.user_agent, command.ip_address)
            {
                session.set_client_info(client_info, user_agent, ip_address);
            }

            // Activate session
            session.activate().map_err(ApplicationError::Domain)?;

            let session_id = session.id();

            self.save_and_publish(&mut session).await?;

            Ok(session_id)
        }
    }
}

impl<R, P, F> CommandHandlerGat<CreateStreamCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = StreamId;

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: CreateStreamCommand) -> Self::HandleFuture<'_> {
        async move {
            CommandValidator::validate_create_stream(&command)
                .map_err(|errors| ApplicationError::Validation(errors.join("; ")))?;

            // Atomic per-session read-modify-write (#457): a plain load +
            // mutate + save cycle can race under a concurrent mutation of
            // the same session (e.g. a completion) and silently drop one
            // side's update to SessionStats.
            let (stream_id, events) = self
                .repository
                .create_stream_atomic(
                    command.session_id.into(),
                    command.source_data,
                    command.config,
                )
                .await
                .map_err(|e| match e {
                    crate::domain::DomainError::SessionNotFound(_) => ApplicationError::NotFound(
                        format!("Session {} not found", command.session_id),
                    ),
                    other => ApplicationError::Domain(other),
                })?;

            self.event_publisher
                .publish_batch(events)
                .await
                .map_err(ApplicationError::Domain)?;

            Ok(stream_id)
        }
    }
}

impl<R, P, F> CommandHandlerGat<StartStreamCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = ();

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: StartStreamCommand) -> Self::HandleFuture<'_> {
        async move {
            // Atomic per-session read-modify-write (#457): see CreateStreamCommand.
            let events = self
                .repository
                .start_stream_atomic(command.session_id.into(), command.stream_id.into())
                .await
                .map_err(|e| match e {
                    crate::domain::DomainError::SessionNotFound(_) => ApplicationError::NotFound(
                        format!("Session {} not found", command.session_id),
                    ),
                    other => ApplicationError::Domain(other),
                })?;

            self.event_publisher
                .publish_batch(events)
                .await
                .map_err(ApplicationError::Domain)?;

            Ok(())
        }
    }
}

impl<R, P, F> CommandHandlerGat<CompleteStreamCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = ();

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: CompleteStreamCommand) -> Self::HandleFuture<'_> {
        async move {
            // Atomic per-session read-modify-write (#457): a plain load +
            // mutate + save cycle can race under concurrent completions for
            // the same session and silently drop one update to SessionStats.
            let events = self
                .repository
                .complete_stream_atomic(command.session_id.into(), command.stream_id.into())
                .await
                .map_err(|e| match e {
                    crate::domain::DomainError::SessionNotFound(_) => ApplicationError::NotFound(
                        format!("Session {} not found", command.session_id),
                    ),
                    other => ApplicationError::Domain(other),
                })?;

            self.event_publisher
                .publish_batch(events)
                .await
                .map_err(ApplicationError::Domain)?;

            Ok(())
        }
    }
}

impl<R, P, F> CommandHandlerGat<GenerateFramesCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = Vec<Frame>;

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: GenerateFramesCommand) -> Self::HandleFuture<'_> {
        async move {
            CommandValidator::validate_generate_frames(&command)
                .map_err(|errors| ApplicationError::Validation(errors.join("; ")))?;

            // Generate frames through the aggregate root so session-level
            // stats and events stay consistent with the child stream mutation.
            let priority = command
                .priority_threshold
                .try_into()
                .map_err(ApplicationError::Domain)?;

            // Atomic per-session read-modify-write (#457): see CreateStreamCommand.
            let (frames, events) = self
                .repository
                .create_stream_patch_frames_atomic(
                    command.session_id.into(),
                    command.stream_id.into(),
                    priority,
                    command.max_frames,
                )
                .await
                .map_err(|e| match e {
                    crate::domain::DomainError::SessionNotFound(_) => ApplicationError::NotFound(
                        format!("Session {} not found", command.session_id),
                    ),
                    crate::domain::DomainError::StreamNotFound(_) => ApplicationError::NotFound(
                        format!("Stream {} not found", command.stream_id),
                    ),
                    other => ApplicationError::Domain(other),
                })?;

            // Publish immediately after the atomic call, before any
            // fallible I/O below: `create_stream_patch_frames_atomic`
            // already drained these events from the session's buffer inside
            // its lock, so if a later step (e.g. `append_frames`) fails and
            // returns early, there is no buffer left to recover them from —
            // publishing first ensures they are never silently lost.
            self.event_publisher
                .publish_batch(events)
                .await
                .map_err(ApplicationError::Domain)?;

            // WebSocket frame production runs through a disjoint session model
            // (`infrastructure/websocket`) and does not increment this counter, so
            // `pjs_frames_total` reflects HTTP throughput only — see #239.
            #[cfg(feature = "metrics")]
            metrics::counter!("pjs_frames_total").increment(frames.len() as u64);

            // Feed accepted frame payloads into the per-session training corpus
            // so the dictionary endpoint becomes reachable end-to-end.
            #[cfg(all(feature = "compression", not(target_arch = "wasm32")))]
            self.train_from_frames(command.session_id.into(), &frames)
                .await;

            // Persist generated frames so GET /streams/{id}/frames can return them.
            self.frame_store
                .append_frames(command.stream_id.into(), frames.clone())
                .await
                .map_err(ApplicationError::Domain)?;

            Ok(frames)
        }
    }
}

impl<R, P, F> CommandHandlerGat<BatchGenerateFramesCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = Vec<Frame>;

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: BatchGenerateFramesCommand) -> Self::HandleFuture<'_> {
        async move {
            // Atomic per-session read-modify-write (#457, #477): see
            // GenerateFramesCommand for why events publish before any
            // fallible step below. Trade-off here specifically: if
            // `persist_frames_grouped_by_stream` fails partway through a
            // multi-stream batch, `FramesBatched` and session stats are
            // already committed while `GET .../frames` may return fewer
            // frames than were reported — accepted as consistent with
            // GenerateFramesCommand rather than left silently divergent.
            let priority = command
                .priority_threshold
                .try_into()
                .map_err(ApplicationError::Domain)?;

            let (frames, events) = self
                .repository
                .batch_generate_frames_atomic(
                    command.session_id.into(),
                    priority,
                    command.max_frames,
                )
                .await
                .map_err(|e| match e {
                    crate::domain::DomainError::SessionNotFound(_) => ApplicationError::NotFound(
                        format!("Session {} not found", command.session_id),
                    ),
                    other => ApplicationError::Domain(other),
                })?;

            self.event_publisher
                .publish_batch(events)
                .await
                .map_err(ApplicationError::Domain)?;

            // WebSocket frame production runs through a disjoint session model
            // (`infrastructure/websocket`) and does not increment this counter, so
            // `pjs_frames_total` reflects HTTP throughput only — see #239.
            #[cfg(feature = "metrics")]
            metrics::counter!("pjs_frames_total").increment(frames.len() as u64);

            // Feed accepted frame payloads into the per-session training corpus
            // so the dictionary endpoint becomes reachable end-to-end.
            #[cfg(all(feature = "compression", not(target_arch = "wasm32")))]
            self.train_from_frames(command.session_id.into(), &frames)
                .await;

            // Persist generated frames so GET /streams/{id}/frames can return them.
            // Frames may span multiple streams in a batch; group by stream_id.
            self.persist_frames_grouped_by_stream(&frames).await?;

            Ok(frames)
        }
    }
}

impl<R, P, F> CommandHandlerGat<CloseSessionCommand> for SessionCommandHandler<R, P, F>
where
    R: StreamRepositoryGat + Send + Sync,
    P: EventPublisherGat + Send + Sync,
    F: FrameStoreGat + Send + Sync,
{
    type Response = ();

    type HandleFuture<'a>
        = impl std::future::Future<Output = ApplicationResult<Self::Response>> + Send + 'a
    where
        Self: 'a;

    fn handle(&self, command: CloseSessionCommand) -> Self::HandleFuture<'_> {
        async move {
            // Atomic per-session read-modify-write (#457): see CreateStreamCommand.
            let events = self
                .repository
                .close_session_atomic(command.session_id.into())
                .await
                .map_err(|e| match e {
                    crate::domain::DomainError::SessionNotFound(_) => ApplicationError::NotFound(
                        format!("Session {} not found", command.session_id),
                    ),
                    other => ApplicationError::Domain(other),
                })?;

            self.event_publisher
                .publish_batch(events)
                .await
                .map_err(ApplicationError::Domain)?;

            Ok(())
        }
    }
}

/// Validates commands at the application boundary, before they reach the
/// domain layer.
///
/// Invoked as the first statement of [`CommandHandlerGat::handle`] for
/// [`CreateSessionCommand`], [`CreateStreamCommand`], and
/// [`GenerateFramesCommand`] on [`SessionCommandHandler`] — ahead of any
/// session lookup, so a malformed request fails with
/// [`ApplicationError::Validation`] (HTTP 400) instead of a later domain
/// error or a misleading 404/500.
pub struct CommandValidator;

impl CommandValidator {
    /// Validate CreateSessionCommand
    pub fn validate_create_session(command: &CreateSessionCommand) -> Result<(), Vec<String>> {
        let mut errors = Vec::new();

        if command.config.max_concurrent_streams == 0 {
            errors.push("max_concurrent_streams must be greater than 0".to_string());
        }

        if command.config.session_timeout_seconds == 0 {
            errors.push("session_timeout_seconds must be greater than 0".to_string());
        }

        if command.config.session_timeout_seconds > MAX_SESSION_TIMEOUT_SECONDS {
            errors.push(format!(
                "session_timeout_seconds cannot exceed {MAX_SESSION_TIMEOUT_SECONDS}"
            ));
        }

        if errors.is_empty() {
            Ok(())
        } else {
            Err(errors)
        }
    }

    /// Validate CreateStreamCommand
    pub fn validate_create_stream(command: &CreateStreamCommand) -> Result<(), Vec<String>> {
        let mut errors = Vec::new();

        if command.source_data.is_null() {
            errors.push("source_data cannot be null".to_string());
        }

        if errors.is_empty() {
            Ok(())
        } else {
            Err(errors)
        }
    }

    /// Validate GenerateFramesCommand
    pub fn validate_generate_frames(command: &GenerateFramesCommand) -> Result<(), Vec<String>> {
        let mut errors = Vec::new();

        if command.max_frames == 0 {
            errors.push("max_frames must be greater than 0".to_string());
        }

        if command.max_frames > MAX_FRAMES_PER_REQUEST {
            errors.push(format!("max_frames cannot exceed {MAX_FRAMES_PER_REQUEST}"));
        }

        if errors.is_empty() {
            Ok(())
        } else {
            Err(errors)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::{
        aggregates::stream_session::SessionConfig, events::DomainEvent, ports::EventPublisherGat,
    };
    use crate::test_support::MockRepository;

    struct MockEventPublisher;

    impl EventPublisherGat for MockEventPublisher {
        type PublishFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        type PublishBatchFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        fn publish(&self, _event: DomainEvent) -> Self::PublishFuture<'_> {
            async move { Ok(()) }
        }

        fn publish_batch(&self, _events: Vec<DomainEvent>) -> Self::PublishBatchFuture<'_> {
            async move { Ok(()) }
        }
    }

    /// Event publisher that records each `publish_batch` call as its own
    /// batch, so a test can inspect exactly which events each command's
    /// `save_and_publish` call emitted.
    struct RecordingEventPublisher {
        batches: parking_lot::Mutex<Vec<Vec<DomainEvent>>>,
    }

    impl RecordingEventPublisher {
        fn new() -> Self {
            Self {
                batches: parking_lot::Mutex::new(Vec::new()),
            }
        }

        fn batches(&self) -> Vec<Vec<DomainEvent>> {
            self.batches.lock().clone()
        }
    }

    impl EventPublisherGat for RecordingEventPublisher {
        type PublishFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        type PublishBatchFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        fn publish(&self, event: DomainEvent) -> Self::PublishFuture<'_> {
            async move {
                self.batches.lock().push(vec![event]);
                Ok(())
            }
        }

        fn publish_batch(&self, events: Vec<DomainEvent>) -> Self::PublishBatchFuture<'_> {
            async move {
                self.batches.lock().push(events);
                Ok(())
            }
        }
    }

    /// Event publisher that records every published event, for asserting on
    /// what actually got published rather than merely whether the handler
    /// returned `Ok`.
    struct TrackingEventPublisher {
        published: parking_lot::Mutex<Vec<DomainEvent>>,
    }

    impl TrackingEventPublisher {
        fn new() -> Self {
            Self {
                published: parking_lot::Mutex::new(Vec::new()),
            }
        }

        fn published_events(&self) -> Vec<DomainEvent> {
            self.published.lock().clone()
        }
    }

    impl EventPublisherGat for TrackingEventPublisher {
        type PublishFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        type PublishBatchFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        fn publish(&self, event: DomainEvent) -> Self::PublishFuture<'_> {
            async move {
                self.published.lock().push(event);
                Ok(())
            }
        }

        fn publish_batch(&self, events: Vec<DomainEvent>) -> Self::PublishBatchFuture<'_> {
            async move {
                self.published.lock().extend(events);
                Ok(())
            }
        }
    }

    /// Frame store whose `append_frames` always fails, for testing that a
    /// downstream persistence failure doesn't discard events already
    /// drained from the session's atomic mutation.
    struct FailingFrameStore;

    impl crate::domain::ports::FrameStoreGat for FailingFrameStore {
        type AppendFramesFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        type GetFramesFuture<'a>
            = impl std::future::Future<
                Output = crate::domain::DomainResult<crate::domain::ports::FrameStorePage>,
            > + Send
            + 'a
        where
            Self: 'a;

        type DeleteFramesForStreamFuture<'a>
            = impl std::future::Future<Output = crate::domain::DomainResult<()>> + Send + 'a
        where
            Self: 'a;

        fn append_frames(
            &self,
            _stream_id: StreamId,
            _frames: Vec<Frame>,
        ) -> Self::AppendFramesFuture<'_> {
            async move {
                Err(crate::domain::DomainError::InvalidInput(
                    "simulated frame store failure".to_string(),
                ))
            }
        }

        fn get_frames(
            &self,
            _stream_id: StreamId,
            _since_sequence: Option<u64>,
            _priority_filter: Option<crate::domain::value_objects::Priority>,
            _limit: Option<usize>,
        ) -> Self::GetFramesFuture<'_> {
            async move {
                Ok(crate::domain::ports::FrameStorePage {
                    frames: Vec::new(),
                    total_matching: 0,
                })
            }
        }

        fn delete_frames_for_stream(
            &self,
            _stream_id: StreamId,
        ) -> Self::DeleteFramesForStreamFuture<'_> {
            async move { Ok(()) }
        }
    }
    #[tokio::test]
    async fn test_create_session_command() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        let command = CreateSessionCommand {
            config: SessionConfig::default(),
            client_info: Some("test-client".to_string()),
            user_agent: None,
            ip_address: None,
        };

        let result = handler.handle(command).await;
        assert!(result.is_ok());

        let session_id = result.unwrap();

        // Verify session was saved
        let saved_session = repository.find_session(session_id).await.unwrap();
        assert!(saved_session.is_some());
    }

    #[tokio::test]
    async fn test_session_command_handler_creation() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher.clone());

        assert!(std::ptr::eq(
            handler.repository.as_ref(),
            repository.as_ref()
        ));
        assert!(std::ptr::eq(
            handler.event_publisher.as_ref(),
            event_publisher.as_ref()
        ));
    }

    #[tokio::test]
    async fn test_create_session_with_full_client_info() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        let command = CreateSessionCommand {
            config: SessionConfig::default(),
            client_info: Some("test-client".to_string()),
            user_agent: Some("Mozilla/5.0".to_string()),
            ip_address: Some("192.168.1.1".to_string()),
        };

        let result = handler.handle(command).await;
        assert!(result.is_ok());

        let session_id = result.unwrap();
        let saved_session = repository.find_session(session_id).await.unwrap();
        assert!(saved_session.is_some());
    }

    #[tokio::test]
    async fn test_create_session_without_client_info() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let command = CreateSessionCommand {
            config: SessionConfig::default(),
            client_info: None,
            user_agent: None,
            ip_address: None,
        };

        let result = handler.handle(command).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_create_stream_command() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        // First create a session
        let create_session_cmd = CreateSessionCommand {
            config: SessionConfig::default(),
            client_info: None,
            user_agent: None,
            ip_address: None,
        };

        let session_id = handler.handle(create_session_cmd).await.unwrap();

        // Then create a stream
        let create_stream_cmd = CreateStreamCommand {
            session_id: session_id.into(),
            source_data: serde_json::json!({"test": "data"}).into(),
            config: None,
        };

        let result = handler.handle(create_stream_cmd).await;
        assert!(result.is_ok());

        let stream_id = result.unwrap();
        assert_ne!(stream_id, StreamId::new()); // Should be a valid stream ID
    }

    #[tokio::test]
    async fn test_create_stream_with_config() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        // Create session first
        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        // Create stream with config
        let stream_config = crate::domain::entities::stream::StreamConfig::default();
        let create_stream_cmd = CreateStreamCommand {
            session_id: session_id.into(),
            source_data: serde_json::json!({"test": "data"}).into(),
            config: Some(stream_config),
        };

        let result = handler.handle(create_stream_cmd).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_create_stream_session_not_found() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let non_existent_session_id = SessionId::new();
        let create_stream_cmd = CreateStreamCommand {
            session_id: non_existent_session_id.into(),
            source_data: serde_json::json!({"test": "data"}).into(),
            config: None,
        };

        let result = handler.handle(create_stream_cmd).await;
        assert!(result.is_err());
        assert!(matches!(
            result.err().unwrap(),
            ApplicationError::NotFound(_)
        ));
    }

    #[tokio::test]
    async fn test_start_stream_command() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        // Create session and stream first
        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"test": "data"}).into(),
                config: None,
            })
            .await
            .unwrap();

        // Start the stream
        let start_stream_cmd = StartStreamCommand {
            session_id: session_id.into(),
            stream_id: stream_id.into(),
        };

        let result = handler.handle(start_stream_cmd).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_start_stream_session_not_found() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let start_stream_cmd = StartStreamCommand {
            session_id: SessionId::new().into(),
            stream_id: StreamId::new().into(),
        };

        let result = handler.handle(start_stream_cmd).await;
        assert!(result.is_err());
        assert!(matches!(
            result.err().unwrap(),
            ApplicationError::NotFound(_)
        ));
    }

    #[tokio::test]
    async fn test_complete_stream_command() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        // Create session, stream, and start it
        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"test": "data"}).into(),
                config: None,
            })
            .await
            .unwrap();

        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        // Complete the stream
        let complete_stream_cmd = CompleteStreamCommand {
            session_id: session_id.into(),
            stream_id: stream_id.into(),
            checksum: Some("abc123".to_string()),
        };

        let result = handler.handle(complete_stream_cmd).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_complete_stream_without_checksum() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        // Create and start stream
        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"test": "data"}).into(),
                config: None,
            })
            .await
            .unwrap();

        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        // Complete without checksum
        let complete_stream_cmd = CompleteStreamCommand {
            session_id: session_id.into(),
            stream_id: stream_id.into(),
            checksum: None,
        };

        let result = handler.handle(complete_stream_cmd).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_close_session_command() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher);

        // Create session first
        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        // Close the session
        let close_session_cmd = CloseSessionCommand {
            session_id: session_id.into(),
        };

        let result = handler.handle(close_session_cmd).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_close_session_not_found() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let close_session_cmd = CloseSessionCommand {
            session_id: SessionId::new().into(),
        };

        let result = handler.handle(close_session_cmd).await;
        assert!(result.is_err());
        assert!(matches!(
            result.err().unwrap(),
            ApplicationError::NotFound(_)
        ));
    }

    #[cfg(all(feature = "compression", not(target_arch = "wasm32")))]
    mod dictionary_wiring {
        //! Regression tests for issue #224 — frame-ingest must feed the per-session
        //! training corpus so that `GET /pjs/sessions/{id}/dictionary` becomes
        //! reachable end-to-end.
        use super::*;
        use crate::{
            compression::zstd::N_TRAIN,
            domain::{
                entities::Frame,
                ports::{DictionaryFuture, DictionaryStore},
            },
            infrastructure::repositories::InMemoryDictionaryStore,
            security::CompressionBombDetector,
        };
        use pjson_rs_domain::value_objects::{JsonData, StreamId};
        use std::sync::atomic::{AtomicUsize, Ordering};

        /// Counts every `train_if_ready` invocation so a test can verify the
        /// command handler reaches the dictionary store at all.
        struct CountingDictionaryStore {
            inner: InMemoryDictionaryStore,
            calls: AtomicUsize,
        }

        impl CountingDictionaryStore {
            fn new() -> Self {
                Self {
                    inner: InMemoryDictionaryStore::new(
                        Arc::new(CompressionBombDetector::default()),
                        64 * 1024,
                    ),
                    calls: AtomicUsize::new(0),
                }
            }

            fn call_count(&self) -> usize {
                self.calls.load(Ordering::SeqCst)
            }
        }

        impl DictionaryStore for CountingDictionaryStore {
            fn get_dictionary<'a>(
                &'a self,
                session_id: SessionId,
            ) -> DictionaryFuture<'a, Option<Arc<crate::compression::zstd::ZstdDictionary>>>
            {
                self.inner.get_dictionary(session_id)
            }

            fn train_if_ready<'a>(
                &'a self,
                session_id: SessionId,
                sample: Vec<u8>,
            ) -> DictionaryFuture<'a, ()> {
                self.calls.fetch_add(1, Ordering::SeqCst);
                self.inner.train_if_ready(session_id, sample)
            }
        }

        fn make_patch_frame(stream_id: StreamId, sequence: u64, n: usize) -> Frame {
            let patch = crate::domain::entities::frame::FramePatch::set(
                pjson_rs_domain::value_objects::JsonPath::new(format!("$.items[{n}]")).unwrap(),
                JsonData::Integer(n as i64),
            );
            Frame::patch(
                stream_id,
                sequence,
                pjson_rs_domain::value_objects::Priority::HIGH,
                vec![patch],
            )
            .unwrap()
        }

        #[tokio::test]
        async fn test_train_from_frames_records_each_payload() {
            let store = Arc::new(CountingDictionaryStore::new());
            let handler = SessionCommandHandler::with_dictionary_store(
                Arc::new(MockRepository::new()),
                Arc::new(MockEventPublisher),
                store.clone(),
            );

            let session_id = SessionId::new();
            let stream_id = StreamId::new();
            let frames: Vec<Frame> = (0..5)
                .map(|i| make_patch_frame(stream_id, i as u64, i))
                .collect();

            handler.train_from_frames(session_id, &frames).await;

            assert_eq!(
                store.call_count(),
                5,
                "every accepted frame must feed train_if_ready"
            );
        }

        #[tokio::test]
        async fn test_train_from_frames_fires_dictionary_after_threshold() {
            let store = Arc::new(InMemoryDictionaryStore::new(
                Arc::new(CompressionBombDetector::default()),
                64 * 1024,
            ));
            let handler = SessionCommandHandler::with_dictionary_store(
                Arc::new(MockRepository::new()),
                Arc::new(MockEventPublisher),
                store.clone(),
            );

            let session_id = SessionId::new();
            let stream_id = StreamId::new();
            let frames: Vec<Frame> = (0..N_TRAIN)
                .map(|i| make_patch_frame(stream_id, i as u64, i))
                .collect();

            handler.train_from_frames(session_id, &frames).await;

            let dict = store.get_dictionary(session_id).await.unwrap();
            assert!(
                dict.is_some(),
                "dictionary must be trained once N_TRAIN frame payloads have been ingested"
            );
        }
    }

    #[tokio::test]
    async fn test_generate_frames_persists_into_frame_store() {
        use crate::domain::ports::FrameStoreGat;
        use crate::infrastructure::adapters::InMemoryFrameStore;

        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let frame_store = Arc::new(InMemoryFrameStore::new());

        let handler = SessionCommandHandler::with_stores(
            repository.clone(),
            event_publisher,
            Arc::new(crate::domain::ports::NoopDictionaryStore),
            frame_store.clone(),
        );

        // Bring up an active session with one stream.
        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"items": [1, 2, 3, 4]}).into(),
                config: None,
            })
            .await
            .unwrap();

        // Frames can only be produced from a streaming stream.
        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        // GenerateFrames must (a) return the frames and (b) leave them in the
        // frame store so the GET endpoint can find them.
        let frames = handler
            .handle(GenerateFramesCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 8,
            })
            .await
            .unwrap();

        assert!(
            !frames.is_empty(),
            "command must produce at least one frame"
        );

        let page = frame_store
            .get_frames(stream_id, None, None, None)
            .await
            .unwrap();
        assert_eq!(
            page.frames.len(),
            frames.len(),
            "every frame returned by the command must be persisted",
        );
        assert_eq!(page.total_matching, frames.len());
    }

    #[tokio::test]
    async fn test_generate_frames_publishes_events_before_frame_store_failure() {
        // Regression test: create_stream_patch_frames_atomic drains the
        // session's event buffer inside its lock, so once it returns there
        // is no buffer left to recover those events from. The handler must
        // publish them before any later fallible step (append_frames) can
        // discard them on error.
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(TrackingEventPublisher::new());
        let frame_store = Arc::new(FailingFrameStore);

        let handler = SessionCommandHandler::with_stores(
            repository.clone(),
            event_publisher.clone(),
            Arc::new(crate::domain::ports::NoopDictionaryStore),
            frame_store,
        );

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"items": [1, 2, 3, 4]}).into(),
                config: None,
            })
            .await
            .unwrap();

        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        // Every event published above (SessionActivated, StreamCreated,
        // StreamStarted) is irrelevant to this assertion; only what happens
        // from here matters.
        let events_before = event_publisher.published_events().len();

        let result = handler
            .handle(GenerateFramesCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 8,
            })
            .await;

        assert!(
            result.is_err(),
            "FailingFrameStore must cause the command to fail"
        );

        let events_after = event_publisher.published_events().len();
        assert!(
            events_after > events_before,
            "FramesBatched must be published even though append_frames failed \
             (was {events_before}, now {events_after})"
        );
    }

    #[tokio::test]
    async fn test_create_session_rejects_zero_max_concurrent_streams() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let config = SessionConfig {
            max_concurrent_streams: 0,
            ..Default::default()
        };

        let result = handler
            .handle(CreateSessionCommand {
                config,
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    #[tokio::test]
    async fn test_create_session_rejects_zero_session_timeout() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let config = SessionConfig {
            session_timeout_seconds: 0,
            ..Default::default()
        };

        let result = handler
            .handle(CreateSessionCommand {
                config,
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    #[tokio::test]
    async fn test_create_stream_rejects_null_source_data_before_session_lookup() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        // A nonexistent session id proves validation runs before load_session
        // — otherwise this would fail with `ApplicationError::NotFound`.
        let result = handler
            .handle(CreateStreamCommand {
                session_id: SessionId::new().into(),
                source_data: serde_json::Value::Null.into(),
                config: None,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    #[tokio::test]
    async fn test_generate_frames_rejects_zero_max_frames() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let result = handler
            .handle(GenerateFramesCommand {
                session_id: SessionId::new().into(),
                stream_id: StreamId::new().into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 0,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    #[tokio::test]
    async fn test_generate_frames_rejects_max_frames_over_limit() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let result = handler
            .handle(GenerateFramesCommand {
                session_id: SessionId::new().into(),
                stream_id: StreamId::new().into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: MAX_FRAMES_PER_REQUEST + 1,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    #[tokio::test]
    async fn test_batch_generate_frames_persists_into_frame_store() {
        use crate::domain::ports::FrameStoreGat;
        use crate::infrastructure::adapters::InMemoryFrameStore;

        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let frame_store = Arc::new(InMemoryFrameStore::new());

        let handler = SessionCommandHandler::with_stores(
            repository.clone(),
            event_publisher,
            Arc::new(crate::domain::ports::NoopDictionaryStore),
            frame_store.clone(),
        );

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"items": [1, 2, 3, 4]}).into(),
                config: None,
            })
            .await
            .unwrap();

        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        // BatchGenerateFrames must (a) route through the atomic repository
        // call (#477) and (b) leave the frames in the frame store so the GET
        // endpoint can find them.
        let frames = handler
            .handle(BatchGenerateFramesCommand {
                session_id: session_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 8,
            })
            .await
            .unwrap();

        assert!(
            !frames.is_empty(),
            "command must produce at least one frame"
        );

        let page = frame_store
            .get_frames(stream_id, None, None, None)
            .await
            .unwrap();
        assert_eq!(
            page.frames.len(),
            frames.len(),
            "every frame returned by the command must be persisted",
        );
    }

    #[tokio::test]
    async fn test_batch_generate_frames_session_not_found() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let result = handler
            .handle(BatchGenerateFramesCommand {
                session_id: SessionId::new().into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 8,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::NotFound(_))));
    }

    #[tokio::test]
    async fn test_batch_generate_frames_no_streaming_streams_returns_empty() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"a": 1}).into(),
                config: None,
            })
            .await
            .unwrap();
        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();
        handler
            .handle(CompleteStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
                checksum: None,
            })
            .await
            .unwrap();

        // No Streaming streams left — the command must succeed with an empty
        // result, not error.
        let frames = handler
            .handle(BatchGenerateFramesCommand {
                session_id: session_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 8,
            })
            .await
            .unwrap();

        assert!(frames.is_empty());
    }

    /// Regression test for #498: `BatchGenerateFramesCommand::priority_threshold`
    /// must actually filter which patches are emitted, not be silently
    /// replaced by a hardcoded `Priority::BACKGROUND` inside
    /// `batch_generate_frames_atomic`. `"logs"` is heuristically classified
    /// `BACKGROUND` (see `services::priority`), so a `HIGH` threshold must
    /// drop it entirely.
    #[tokio::test]
    async fn test_batch_generate_frames_priority_threshold_filters_low_priority_patches() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"logs": "background noise"}).into(),
                config: None,
            })
            .await
            .unwrap();
        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        let frames = handler
            .handle(BatchGenerateFramesCommand {
                session_id: session_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(
                    crate::domain::value_objects::Priority::HIGH.value(),
                )
                .unwrap(),
                max_frames: 8,
            })
            .await
            .unwrap();

        assert!(
            frames.is_empty(),
            "a HIGH priority_threshold must filter out the BACKGROUND-priority \
             `logs` patch instead of falling back to Priority::BACKGROUND"
        );
    }

    #[tokio::test]
    async fn test_batch_generate_frames_max_frames_zero_returns_empty() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"a": 1}).into(),
                config: None,
            })
            .await
            .unwrap();
        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        // `BatchGenerateFramesCommand::max_frames` is deliberately left
        // unvalidated (#438) — zero is accepted and truncates to nothing
        // rather than being rejected.
        let frames = handler
            .handle(BatchGenerateFramesCommand {
                session_id: session_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 0,
            })
            .await
            .unwrap();

        assert!(frames.is_empty());
    }

    /// Regression test: a session already closed by the time
    /// `BatchGenerateFramesCommand` reaches `batch_generate_frames_atomic`
    /// must surface as `ApplicationError::Domain(InvalidSessionState)`, not
    /// `NotFound` — the session still exists, it is just no longer active.
    /// This is the same outcome a `CloseSessionCommand` racing ahead of a
    /// concurrent `BatchGenerateFramesCommand` on the same session would
    /// produce.
    #[tokio::test]
    async fn test_batch_generate_frames_closed_session_returns_domain_error() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        handler
            .handle(CloseSessionCommand {
                session_id: session_id.into(),
            })
            .await
            .unwrap();

        let result = handler
            .handle(BatchGenerateFramesCommand {
                session_id: session_id.into(),
                priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                max_frames: 8,
            })
            .await;

        assert!(matches!(
            result,
            Err(ApplicationError::Domain(
                crate::domain::DomainError::InvalidSessionState(_)
            ))
        ));
    }

    /// Regression test for #477: `BatchGenerateFramesCommand` must not lose
    /// `SessionStats` updates to a concurrent atomic command on the same
    /// session, the way the old `find_session` + mutate + `save_session`
    /// path could. Routing through `batch_generate_frames_atomic` means both
    /// commands now share the same per-session lock as every other
    /// `*_atomic` method.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_batch_generate_frames_concurrent_with_complete_stream_loses_no_update() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = Arc::new(SessionCommandHandler::new(repository, event_publisher));

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        // `frame_stream_id` stays active for the whole test so every
        // concurrent batch call has data to draw from, regardless of
        // ordering relative to `complete_stream_id`'s completion —
        // `create_priority_frames` skips inactive streams, so completing the
        // *only* stream mid-race would make batch calls racing after it
        // legitimately (not due to a lost update) return zero frames.
        let frame_stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"a": 1, "b": 2}).into(),
                config: None,
            })
            .await
            .unwrap();
        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: frame_stream_id.into(),
            })
            .await
            .unwrap();

        let complete_stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"c": 3}).into(),
                config: None,
            })
            .await
            .unwrap();
        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: complete_stream_id.into(),
            })
            .await
            .unwrap();

        const N: usize = 25;
        let barrier = Arc::new(tokio::sync::Barrier::new(N + 1));

        let mut handles = Vec::with_capacity(N + 1);
        for _ in 0..N {
            let handler = Arc::clone(&handler);
            let barrier = Arc::clone(&barrier);
            handles.push(tokio::spawn(async move {
                barrier.wait().await;
                handler
                    .handle(BatchGenerateFramesCommand {
                        session_id: session_id.into(),
                        priority_threshold: crate::application::dto::PriorityDto::new(1).unwrap(),
                        max_frames: 2,
                    })
                    .await
                    .unwrap()
            }));
        }
        {
            let handler = Arc::clone(&handler);
            let barrier = Arc::clone(&barrier);
            handles.push(tokio::spawn(async move {
                barrier.wait().await;
                handler
                    .handle(CompleteStreamCommand {
                        session_id: session_id.into(),
                        stream_id: complete_stream_id.into(),
                        checksum: None,
                    })
                    .await
                    .unwrap();
                vec![]
            }));
        }

        let mut total_frames = 0usize;
        for handle in handles {
            total_frames += handle.await.unwrap().len();
        }
        assert_eq!(total_frames, N * 2);

        let session = handler
            .repository
            .find_session(session_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(session.stats().total_frames, (N * 2) as u64);
        assert_eq!(session.stats().completed_streams, 1);
    }

    #[tokio::test]
    async fn test_create_session_rejects_session_timeout_that_would_panic_chrono() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        // Reproduces the S1 panic: `session_timeout_seconds as i64` overflowed
        // `chrono::Duration::seconds`'s valid range and panicked inside
        // `StreamSession::with_time_provider` before this bound existed.
        let config = SessionConfig {
            session_timeout_seconds: 9_223_372_036_854_776,
            ..Default::default()
        };

        let result = handler
            .handle(CreateSessionCommand {
                config,
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    #[tokio::test]
    async fn test_create_session_rejects_session_timeout_that_would_wrap_negative() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(MockEventPublisher);
        let handler = SessionCommandHandler::new(repository, event_publisher);

        // Reproduces the S1 silent-wraparound case: `u64::MAX as i64` is -1,
        // which previously created a session already expired on arrival.
        let config = SessionConfig {
            session_timeout_seconds: u64::MAX,
            ..Default::default()
        };

        let result = handler
            .handle(CreateSessionCommand {
                config,
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await;

        assert!(matches!(result, Err(ApplicationError::Validation(_))));
    }

    /// Regression test for #466: `save_and_publish` must drain
    /// `pending_events` before persisting the session, so a later command on
    /// the same session neither finds stale events still queued in the
    /// persisted copy nor republishes an earlier command's events.
    #[tokio::test]
    async fn test_save_and_publish_does_not_leak_pending_events_across_commands() {
        let repository = Arc::new(MockRepository::new());
        let event_publisher = Arc::new(RecordingEventPublisher::new());
        let handler = SessionCommandHandler::new(repository.clone(), event_publisher.clone());

        let session_id = handler
            .handle(CreateSessionCommand {
                config: SessionConfig::default(),
                client_info: None,
                user_agent: None,
                ip_address: None,
            })
            .await
            .unwrap();

        let after_create_session = repository.find_session(session_id).await.unwrap().unwrap();
        assert!(
            after_create_session.pending_events().is_empty(),
            "persisted session must not carry undrained events after CreateSessionCommand"
        );

        let stream_id = handler
            .handle(CreateStreamCommand {
                session_id: session_id.into(),
                source_data: serde_json::json!({"test": "data"}).into(),
                config: None,
            })
            .await
            .unwrap();

        let after_create_stream = repository.find_session(session_id).await.unwrap().unwrap();
        assert!(
            after_create_stream.pending_events().is_empty(),
            "persisted session must not carry undrained events after CreateStreamCommand"
        );

        handler
            .handle(StartStreamCommand {
                session_id: session_id.into(),
                stream_id: stream_id.into(),
            })
            .await
            .unwrap();

        let after_start_stream = repository.find_session(session_id).await.unwrap().unwrap();
        assert!(
            after_start_stream.pending_events().is_empty(),
            "persisted session must not carry undrained events after StartStreamCommand"
        );

        let batches = event_publisher.batches();
        assert_eq!(
            batches.len(),
            3,
            "each command's save_and_publish call must publish exactly one batch"
        );
        let (create_session_events, create_stream_events, start_stream_events) =
            (&batches[0], &batches[1], &batches[2]);

        // Without the fix, each persisted clone still carried its
        // predecessor's undrained events, so every later batch grew to
        // include every earlier command's events too.
        assert_eq!(
            create_session_events.len(),
            1,
            "CreateSessionCommand must publish only its own SessionActivated event"
        );
        assert_eq!(
            create_stream_events.len(),
            1,
            "CreateStreamCommand must publish only its own StreamCreated event, not a republish of CreateSessionCommand's"
        );
        assert_eq!(
            start_stream_events.len(),
            1,
            "StartStreamCommand must publish only its own StreamStarted event, not a republish of earlier commands'"
        );

        for event in create_stream_events {
            assert!(
                !create_session_events.contains(event),
                "CreateStreamCommand republished an event from CreateSessionCommand: {event:?}"
            );
        }
        for event in start_stream_events {
            assert!(
                !create_session_events.contains(event) && !create_stream_events.contains(event),
                "StartStreamCommand republished an earlier command's event: {event:?}"
            );
        }
    }
}