weavegraph 0.7.0

//! High-level [`App`] entry point for workflow invocation.
//!
//! `App` manages node registration, graph compilation, and delegates execution
//! to an [`AppRunner`].
use rustc_hash::FxHashMap;
use std::sync::Arc;

use crate::channels::Channel;
use crate::channels::errors::{ErrorEvent, ErrorScope};
use crate::control::FrontierCommand;
use crate::event_bus::{ChannelSink, EventBus, EventStream};
use crate::message::*;
use crate::node::*;
use crate::reducers::ReducerRegistry;
use crate::runtimes::runner::RunnerError;
use crate::runtimes::{AppRunner, Checkpointer, CheckpointerType, RuntimeConfig, SessionInit};
use crate::state::*;
use crate::types::*;
use crate::utils::collections::new_extra_map;
use crate::utils::id_generator::IdGenerator;
use futures_util::stream::BoxStream;
use thiserror::Error;
use tokio::task::JoinHandle;
use tracing::instrument;

/// Central coordination point for workflow execution.
///
/// Holds the compiled graph topology (nodes, edges, conditional routing),
/// the reducer registry, and runtime configuration.
/// Construct via [`GraphBuilder`](crate::graphs::GraphBuilder).
///
/// # Examples
///
/// ```rust,no_run
/// use weavegraph::graphs::GraphBuilder;
/// use weavegraph::state::VersionedState;
/// use weavegraph::types::NodeKind;
/// use weavegraph::node::{Node, NodeContext, NodeError, NodePartial};
/// use async_trait::async_trait;
///
/// # struct MyNode;
/// # #[async_trait]
/// # impl Node for MyNode {
/// #     async fn run(&self, _: weavegraph::state::StateSnapshot, _: NodeContext) -> Result<NodePartial, NodeError> {
/// #         Ok(NodePartial::default())
/// #     }
/// # }
/// #
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// let app = GraphBuilder::new()
///     .add_node(NodeKind::Custom("process".into()), MyNode)
///     .add_edge(NodeKind::Start, NodeKind::Custom("process".into()))
///     .add_edge(NodeKind::Custom("process".into()), NodeKind::End)
///     .compile()?;
///
/// let final_state = app.invoke(VersionedState::new_with_user_message("Hello")).await?;
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct App {
    nodes: FxHashMap<NodeKind, Arc<dyn Node>>,
    edges: FxHashMap<NodeKind, Vec<NodeKind>>,
    conditional_edges: Vec<crate::graphs::ConditionalEdge>,
    reducer_registry: ReducerRegistry,
    runtime_config: RuntimeConfig,
}

/// Event bus handle and initial subscription, obtained from [`App::event_stream`].
///
/// Lets callers attach additional sinks before execution starts, or consume the
/// broadcast feed as an async stream, blocking iterator, or timed poll.
pub struct AppEventStream {
    event_bus: EventBus,
    event_stream: Option<EventStream>,
}

/// Error returned when an [`AppEventStream`] subscription is accessed after being consumed.
#[derive(Debug, Error)]
#[cfg_attr(feature = "diagnostics", derive(miette::Diagnostic))]
pub enum AppEventStreamError {
    /// The event stream has already been taken from this handle.
    #[error("event stream has already been taken")]
    #[cfg_attr(
        feature = "diagnostics",
        diagnostic(
            code(weavegraph::app::event_stream),
            help("Verify stream subscription and event channel capacity.")
        )
    )]
    AlreadyTaken,
}

type AppEventStreamResult<T> = Result<T, AppEventStreamError>;

/// Handle for a streaming workflow invocation.
///
/// Dropping the handle aborts the workflow task. Call [`join`](InvocationHandle::join)
/// to await graceful completion; the paired event stream emits a diagnostic with scope
/// [`STREAM_END_SCOPE`](crate::event_bus::STREAM_END_SCOPE) before closing.
pub struct InvocationHandle {
    join_handle: Option<JoinHandle<Result<VersionedState, RunnerError>>>,
}

/// Result of applying node partials at a barrier.
///
/// Aggregates channel and error information in deterministic order so downstream
/// consumers observe stable behaviour across executions.
#[derive(Debug, Clone, Default)]
pub struct BarrierOutcome {
    /// Channel identifiers updated during this barrier.
    pub updated_channels: Vec<&'static str>,
    /// Error events emitted by nodes in the superstep.
    pub errors: Vec<ErrorEvent>,
    /// Frontier commands emitted during the barrier.
    pub frontier_commands: Vec<(NodeKind, FrontierCommand)>,
}

/// Stable metadata describing a compiled graph definition.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub struct GraphMetadata {
    /// Weavegraph crate version used to build the graph.
    pub weavegraph_version: String,
    /// Deterministic hash of the graph definition surface.
    pub graph_hash: String,
    /// Number of registered executable nodes.
    pub node_count: usize,
    /// Number of unconditional edges.
    pub edge_count: usize,
    /// Number of conditional edge registrations.
    pub conditional_edge_count: usize,
    /// Reducer registration signature included in the hash.
    pub reducer_signature: Vec<String>,
}

fn fnv1a_hex(parts: &[String]) -> String {
    const OFFSET: u64 = 0xcbf29ce484222325;
    const PRIME: u64 = 0x100000001b3;
    let mut hash = OFFSET;
    for part in parts {
        for byte in part.bytes().chain([0xff]) {
            hash ^= u64::from(byte);
            hash = hash.wrapping_mul(PRIME);
        }
    }
    format!("{hash:016x}")
}

impl AppEventStream {
    fn new(event_bus: EventBus, event_stream: EventStream) -> Self {
        Self {
            event_bus,
            event_stream: Some(event_stream),
        }
    }

    /// Access the bus to add sinks before execution begins.
    pub fn event_bus(&self) -> &EventBus {
        &self.event_bus
    }

    /// Mutable reference to the underlying broadcast subscription.
    pub fn event_stream(&mut self) -> AppEventStreamResult<&mut EventStream> {
        self.event_stream
            .as_mut()
            .ok_or(AppEventStreamError::AlreadyTaken)
    }

    /// Consume the handle and return the raw event stream.
    pub fn into_stream(mut self) -> AppEventStreamResult<EventStream> {
        self.event_stream
            .take()
            .ok_or(AppEventStreamError::AlreadyTaken)
    }

    /// Consume the handle and return the event bus.
    pub fn into_event_bus(self) -> EventBus {
        self.event_bus
    }

    /// Split the handle into the bus and event stream.
    pub fn split(mut self) -> AppEventStreamResult<(EventBus, EventStream)> {
        let stream = self
            .event_stream
            .take()
            .ok_or(AppEventStreamError::AlreadyTaken)?;
        Ok((self.event_bus, stream))
    }

    /// Convert into a blocking iterator.
    pub fn into_blocking_iter(self) -> AppEventStreamResult<crate::event_bus::BlockingEventIter> {
        Ok(self.into_stream()?.into_blocking_iter())
    }

    /// Convert into a boxed async stream.
    pub fn into_async_stream(
        self,
    ) -> AppEventStreamResult<BoxStream<'static, crate::event_bus::Event>> {
        Ok(self.into_stream()?.into_async_stream())
    }

    /// Await the next event with a timeout, skipping lag notifications.
    pub async fn next_timeout(
        &mut self,
        duration: std::time::Duration,
    ) -> AppEventStreamResult<Option<crate::event_bus::Event>> {
        Ok(self.event_stream()?.next_timeout(duration).await)
    }
}

impl InvocationHandle {
    /// Abort the underlying workflow task.
    pub fn abort(&self) {
        if let Some(handle) = &self.join_handle {
            handle.abort();
        }
    }

    /// Returns `true` if the underlying task has finished or been aborted.
    #[must_use]
    pub fn is_finished(&self) -> bool {
        self.join_handle.as_ref().is_none_or(|h| h.is_finished())
    }

    /// Await the workflow result.
    ///
    /// Returns [`RunnerError::JoinHandleConsumed`] if called more than once,
    /// or [`RunnerError::Join`] if the task panicked or was cancelled.
    pub async fn join(mut self) -> Result<VersionedState, RunnerError> {
        let handle = self
            .join_handle
            .take()
            .ok_or(RunnerError::JoinHandleConsumed)?;
        handle.await.map_err(RunnerError::Join)?
    }
}

impl App {
    /// Build an `App` from pre-validated graph components.
    pub(crate) fn from_parts(
        nodes: FxHashMap<NodeKind, Arc<dyn Node>>,
        edges: FxHashMap<NodeKind, Vec<NodeKind>>,
        conditional_edges: Vec<crate::graphs::ConditionalEdge>,
        runtime_config: RuntimeConfig,
        reducer_registry: ReducerRegistry,
    ) -> Self {
        App {
            nodes,
            edges,
            conditional_edges,
            reducer_registry,
            runtime_config,
        }
    }

    /// Conditional edges registered in the graph.
    #[must_use]
    pub fn conditional_edges(&self) -> &[crate::graphs::ConditionalEdge] {
        &self.conditional_edges
    }

    /// Nodes registered in the graph, keyed by [`NodeKind`].
    #[must_use]
    pub fn nodes(&self) -> &FxHashMap<NodeKind, Arc<dyn Node>> {
        &self.nodes
    }

    /// Unconditional edges in the graph.
    #[must_use]
    pub fn edges(&self) -> &FxHashMap<NodeKind, Vec<NodeKind>> {
        &self.edges
    }

    /// Runtime configuration for this app instance.
    #[must_use]
    pub fn runtime_config(&self) -> &RuntimeConfig {
        &self.runtime_config
    }

    /// Weavegraph crate version compiled into this binary.
    #[must_use]
    pub fn weavegraph_version(&self) -> &'static str {
        env!("CARGO_PKG_VERSION")
    }

    /// Metadata describing this compiled graph definition.
    ///
    /// The hash covers node IDs, unconditional edges, conditional edge sources/counts,
    /// and reducer labels. Predicate closure bodies are opaque, so a predicate change
    /// with the same registration shape is undetectable.
    #[must_use]
    pub fn graph_metadata(&self) -> GraphMetadata {
        let mut parts = vec!["weavegraph-graph-v1".to_string()];

        let mut node_keys: Vec<String> = self.nodes.keys().map(NodeKind::encode).collect();
        node_keys.sort();
        parts.extend(node_keys.iter().map(|n| format!("node:{n}")));

        let mut edge_keys: Vec<String> = self
            .edges
            .iter()
            .flat_map(|(from, targets)| {
                targets
                    .iter()
                    .map(move |to| format!("edge:{}->{}", from.encode(), to.encode()))
            })
            .collect();
        edge_keys.sort();
        parts.extend(edge_keys);

        let mut conditional_sources: Vec<String> = self
            .conditional_edges
            .iter()
            .map(|e| e.from().encode())
            .collect();
        conditional_sources.sort();
        parts.extend(
            conditional_sources
                .iter()
                .enumerate()
                .map(|(i, from)| format!("conditional:{i}:{from}")),
        );

        let reducer_signature = self.reducer_registry.definition_signature();
        parts.extend(reducer_signature.iter().map(|r| format!("reducer:{r}")));

        GraphMetadata {
            weavegraph_version: self.weavegraph_version().to_string(),
            graph_hash: fnv1a_hex(&parts),
            node_count: self.nodes.len(),
            edge_count: self.edges.values().map(Vec::len).sum(),
            conditional_edge_count: self.conditional_edges.len(),
            reducer_signature,
        }
    }

    /// The graph definition hash without the full metadata.
    #[must_use]
    pub fn graph_definition_hash(&self) -> String {
        self.graph_metadata().graph_hash
    }

    /// Build a subscription to the configured event bus without starting execution.
    ///
    /// Useful for attaching additional sinks or observing events before running
    /// the workflow — for example in tests or custom server integrations.
    ///
    /// ```no_run
    /// use futures_util::StreamExt;
    /// use weavegraph::event_bus::{Event, MemorySink};
    /// use weavegraph::runtimes::{AppRunner, CheckpointerType};
    ///
    /// # async fn example(app: weavegraph::app::App, state: weavegraph::state::VersionedState) -> Result<(), Box<dyn std::error::Error>> {
    /// let mut handle = app.event_stream();
    /// handle.event_bus().add_sink(MemorySink::new());
    /// let (event_bus, event_stream) = handle
    ///     .split()
    ///     .expect("fresh event stream handle should still own the stream");
    ///
    /// let mut runner = AppRunner::builder()
    ///     .app(app.clone())
    ///     .checkpointer(CheckpointerType::InMemory)
    ///     .autosave(false)
    ///     .event_bus(event_bus)
    ///     .build()
    ///     .await;
    ///
    /// tokio::spawn(async move {
    ///     let mut stream = event_stream.into_async_stream();
    ///     while let Some(event) = stream.next().await {
    ///         if matches!(event, Event::LLM(llm) if llm.is_final()) {
    ///             tracing::info!("final streaming chunk delivered");
    ///         }
    ///     }
    /// });
    ///
    /// runner.create_session("demo".to_string(), state).await?;
    /// runner.run_until_complete("demo").await?;
    /// # Ok(())
    /// # }
    /// ```
    #[must_use]
    pub fn event_stream(&self) -> AppEventStream {
        let event_bus = self.runtime_config.event_bus.build_event_bus();
        let event_stream = event_bus.subscribe();
        AppEventStream::new(event_bus, event_stream)
    }

    fn resolve_checkpointer(
        &self,
        override_config: Option<CheckpointerType>,
    ) -> (CheckpointerType, Option<Arc<dyn Checkpointer>>) {
        let checkpointer_type = override_config.unwrap_or(CheckpointerType::InMemory);
        let custom_checkpointer = self.runtime_config.custom_checkpointer();
        (checkpointer_type, custom_checkpointer)
    }

    async fn build_runner(
        &self,
        event_bus: EventBus,
        autosave: bool,
        checkpointer_override: Option<CheckpointerType>,
    ) -> AppRunner {
        let (checkpointer_type, custom_checkpointer) =
            self.resolve_checkpointer(checkpointer_override);
        let builder = AppRunner::builder()
            .app(self.clone())
            .autosave(autosave)
            .event_bus(event_bus)
            .start_listener(true);
        let builder = match custom_checkpointer {
            Some(custom) => builder.checkpointer_custom(custom),
            None => builder.checkpointer(checkpointer_type),
        };
        builder.build().await
    }

    async fn invoke_with_bus_builder<R, F>(
        &self,
        initial_state: VersionedState,
        autosave: bool,
        checkpointer_override: Option<CheckpointerType>,
        build_event_bus: F,
    ) -> (Result<VersionedState, RunnerError>, R)
    where
        F: FnOnce() -> (EventBus, R),
    {
        let (event_bus, output) = build_event_bus();
        let runner = self
            .build_runner(event_bus, autosave, checkpointer_override)
            .await;
        let session_id = self.next_session_id();
        (
            Self::run_session(runner, session_id, initial_state).await,
            output,
        )
    }

    /// Invoke the workflow asynchronously while streaming events to the caller.
    ///
    /// Returns a join handle for the workflow outcome and an [`EventStream`] that yields
    /// every event emitted during execution. The stream closes after emitting a
    /// diagnostic with scope [`STREAM_END_SCOPE`](crate::event_bus::STREAM_END_SCOPE).
    ///
    /// Dropping the [`InvocationHandle`] stops the workflow. Dropping the stream alone
    /// does not; use the handle to interrupt execution on client disconnect.
    ///
    /// ```no_run
    /// use futures_util::StreamExt;
    /// use tokio::time::{sleep, Duration};
    /// use weavegraph::event_bus::STREAM_END_SCOPE;
    /// # async fn run(app: weavegraph::app::App, state: weavegraph::state::VersionedState) -> Result<(), Box<dyn std::error::Error>> {
    /// let (handle, events) = app.invoke_streaming(state).await;
    /// let mut handle_slot = Some(handle);
    ///
    /// let mut events = events.into_async_stream();
    /// tokio::spawn(async move {
    ///     while let Some(event) = events.next().await {
    ///         if event.scope_label() == Some(STREAM_END_SCOPE) {
    ///             tracing::info!("workflow finished");
    ///         }
    ///     }
    /// });
    ///
    /// tokio::select! {
    ///     result = async {
    ///         handle_slot
    ///             .take()
    ///             .expect("join branch must own the handle")
    ///             .join()
    ///             .await
    ///     } => {
    ///         if let Err(err) = result {
    ///             tracing::error!("workflow failed: {err}");
    ///         }
    ///     }
    ///     _ = sleep(Duration::from_secs(30)) => {
    ///         tracing::warn!("cancelling run after timeout");
    ///         if let Some(handle) = handle_slot.as_ref() {
    ///             handle.abort();
    ///         }
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// See `examples/streaming_events.rs` for a complete integration example.
    pub async fn invoke_streaming(
        &self,
        initial_state: VersionedState,
    ) -> (InvocationHandle, EventStream) {
        let (event_bus, event_stream) = self
            .event_stream()
            .split()
            .unwrap_or_else(|_| unreachable!("fresh event_stream() always owns its stream"));
        let runner = self.build_runner(event_bus, true, None).await;
        let session_id = self.next_session_id();
        let join = tokio::spawn(Self::run_session(runner, session_id, initial_state));
        (
            InvocationHandle {
                join_handle: Some(join),
            },
            event_stream,
        )
    }

    /// Execute the workflow to completion using the runtime-configured event bus.
    ///
    /// For streaming scenarios consider [`invoke_streaming`](Self::invoke_streaming),
    /// [`invoke_with_channel`](Self::invoke_with_channel), or
    /// [`invoke_with_sinks`](Self::invoke_with_sinks). For per-request isolation or
    /// custom lifecycle control use
    /// [`AppRunner::builder()`](crate::runtimes::runner::AppRunner::builder).
    ///
    /// ```rust,no_run
    /// use weavegraph::state::VersionedState;
    /// use weavegraph::channels::Channel;
    /// # use weavegraph::app::App;
    /// # async fn example(app: App) -> Result<(), Box<dyn std::error::Error>> {
    /// let final_state = app.invoke(VersionedState::new_with_user_message("Start")).await?;
    /// # Ok(())
    /// # }
    /// ```
    #[instrument(skip(self, initial_state), err)]
    pub async fn invoke(
        &self,
        initial_state: VersionedState,
    ) -> Result<VersionedState, RunnerError> {
        self.invoke_with_bus_builder(initial_state, true, None, || {
            (self.runtime_config.event_bus.build_event_bus(), ())
        })
        .await
        .0
    }

    /// Execute the workflow and stream events to a [`flume`] channel.
    ///
    /// Builds an `EventBus` from the runtime configuration and appends a
    /// [`ChannelSink`], then returns both the execution result and the receiver.
    ///
    /// ```rust,no_run
    /// use weavegraph::state::VersionedState;
    /// # use weavegraph::app::App;
    /// # async fn example(app: App) -> Result<(), Box<dyn std::error::Error>> {
    /// let (result, events) = app.invoke_with_channel(
    ///     VersionedState::new_with_user_message("Process this")
    /// ).await;
    ///
    /// tokio::spawn(async move {
    ///     while let Ok(event) = events.recv_async().await {
    ///         println!("Event: {event:?}");
    ///     }
    /// });
    ///
    /// let final_state = result?;
    /// # Ok(())
    /// # }
    /// ```
    #[instrument(skip(self, initial_state))]
    pub async fn invoke_with_channel(
        &self,
        initial_state: VersionedState,
    ) -> (
        Result<VersionedState, RunnerError>,
        flume::Receiver<crate::event_bus::Event>,
    ) {
        self.invoke_with_bus_builder(initial_state, false, None, || {
            let (tx, rx) = flume::unbounded();
            let event_bus = self.runtime_config.event_bus.build_event_bus();
            event_bus.add_sink(ChannelSink::new(tx));
            (event_bus, rx)
        })
        .await
    }

    /// Execute the workflow with additional event sinks.
    ///
    /// Runtime-configured sinks remain active; the provided sinks are appended.
    ///
    /// ```rust,no_run
    /// use weavegraph::event_bus::{ChannelSink, StdOutSink};
    /// use weavegraph::state::VersionedState;
    /// # use weavegraph::app::App;
    /// # async fn example(app: App) -> Result<(), Box<dyn std::error::Error>> {
    /// let (tx, rx) = flume::unbounded();
    ///
    /// let final_state = app.invoke_with_sinks(
    ///     VersionedState::new_with_user_message("Process data"),
    ///     vec![
    ///         Box::new(StdOutSink::default()),
    ///         Box::new(ChannelSink::new(tx)),
    ///     ],
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    #[instrument(skip(self, initial_state, sinks), err)]
    pub async fn invoke_with_sinks(
        &self,
        initial_state: VersionedState,
        sinks: Vec<Box<dyn crate::event_bus::EventSink>>,
    ) -> Result<VersionedState, RunnerError> {
        self.invoke_with_bus_builder(initial_state, false, None, move || {
            let event_bus = self.runtime_config.event_bus.build_event_bus();
            for sink in sinks {
                event_bus.add_boxed_sink(sink);
            }
            (event_bus, ())
        })
        .await
        .0
    }

    fn next_session_id(&self) -> String {
        self.runtime_config
            .session_id
            .clone()
            .unwrap_or_else(|| IdGenerator::new().generate_run_id())
    }

    async fn run_session(
        mut runner: AppRunner,
        session_id: String,
        initial_state: VersionedState,
    ) -> Result<VersionedState, RunnerError> {
        let init = runner
            .create_session(session_id.clone(), initial_state)
            .await?;

        if let SessionInit::Resumed { checkpoint_step } = init {
            tracing::info!(
                session = %session_id,
                checkpoint_step,
                "Resuming session from checkpoint"
            );
        }

        runner.run_until_complete(&session_id).await
    }

    /// Merge node outputs and apply state reductions after a superstep.
    ///
    /// Collects messages, extra data, errors, and frontier commands from each partial,
    /// runs the reducer registry over the merged update, and bumps channel versions only
    /// when content changes. Returns a [`BarrierOutcome`] with stable ordering so
    /// downstream consumers observe deterministic behaviour across executions.
    #[instrument(skip(self, state, run_ids, node_partials), err)]
    pub async fn apply_barrier(
        &self,
        state: &mut VersionedState,
        run_ids: &[NodeKind],
        node_partials: Vec<NodePartial>,
    ) -> Result<BarrierOutcome, Box<dyn std::error::Error + Send + Sync>> {
        let mut msgs_all: Vec<Message> = Vec::new();
        let mut extra_all = new_extra_map();
        let mut errors_all: Vec<ErrorEvent> = Vec::new();
        let mut frontier_commands: Vec<(NodeKind, FrontierCommand)> = Vec::new();

        let unknown = NodeKind::Custom("?".to_string());
        for (p, nid) in node_partials
            .iter()
            .zip(run_ids.iter().chain(std::iter::repeat(&unknown)))
        {
            if let Some(ms) = &p.messages
                && !ms.is_empty()
            {
                tracing::debug!(node = ?nid, count = ms.len(), "Node produced messages");
                msgs_all.extend(ms.clone());
            }

            if let Some(ex) = &p.extra
                && !ex.is_empty()
            {
                tracing::debug!(node = ?nid, keys = ex.len(), "Node produced extra data");
                let mut sorted: Vec<_> = ex.iter().collect();
                sorted.sort_by(|(a, _), (b, _)| a.cmp(b));
                for (k, v) in sorted {
                    extra_all.insert(k.clone(), v.clone());
                }
            }

            if let Some(errs) = &p.errors
                && !errs.is_empty()
            {
                tracing::debug!(node = ?nid, count = errs.len(), "Node produced errors");
                errors_all.extend(errs.clone());
            }

            if let Some(command) = &p.frontier {
                frontier_commands.push((nid.clone(), command.clone()));
            }
        }

        fn scope_sort_key(scope: &ErrorScope) -> (u8, &str, u64) {
            match scope {
                ErrorScope::Node { kind, step } => (0, kind.as_str(), *step),
                ErrorScope::Scheduler { step } => (1, "", *step),
                ErrorScope::Runner { session, step } => (2, session.as_str(), *step),
                ErrorScope::App => (3, "", 0),
            }
        }

        errors_all.sort_by(|a, b| {
            scope_sort_key(&a.scope)
                .cmp(&scope_sort_key(&b.scope))
                .then_with(|| a.when.cmp(&b.when))
                .then_with(|| a.error.message.cmp(&b.error.message))
        });

        let merged = NodePartial {
            messages: (!msgs_all.is_empty()).then_some(msgs_all),
            extra: (!extra_all.is_empty()).then_some(extra_all),
            errors: (!errors_all.is_empty()).then(|| errors_all.clone()),
            frontier: None,
        };

        let msgs_before_len = state.messages.len();
        let msgs_before_ver = state.messages.version();
        let extra_before = state.extra.snapshot();
        let extra_before_ver = state.extra.version();

        self.reducer_registry.apply_all(&mut *state, &merged)?;

        let mut updated: Vec<&'static str> = Vec::new();

        if state.messages.len() != msgs_before_len {
            state
                .messages
                .set_version(msgs_before_ver.saturating_add(1));
            tracing::info!(
                target: "weavegraph::app",
                channel = "messages",
                before_count = msgs_before_len,
                after_count = state.messages.len(),
                before_version = msgs_before_ver,
                after_version = state.messages.version(),
                "channel updated"
            );
            updated.push("messages");
        }

        let extra_after = state.extra.snapshot();
        if extra_after != extra_before {
            state.extra.set_version(extra_before_ver.saturating_add(1));
            tracing::info!(
                target: "weavegraph::app",
                channel = "extra",
                before_count = extra_before.len(),
                after_count = extra_after.len(),
                before_version = extra_before_ver,
                after_version = state.extra.version(),
                "channel updated"
            );
            updated.push("extra");
        }

        Ok(BarrierOutcome {
            updated_channels: updated,
            errors: errors_all,
            frontier_commands,
        })
    }
}