bamboo-engine 2026.10.16

//! Event forwarding from MPSC to broadcast channels.
//!
//! Creates an MPSC channel for agent loop events and spawns a background task
//! that relays events to the session's broadcast sender while tracking runner
//! diagnostic state (budget events, tool execution, round progress).

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

use chrono::Utc;
use tokio::sync::{broadcast, mpsc, watch, RwLock};
use tokio_util::sync::CancellationToken;

use bamboo_agent_core::AgentEvent;

use super::runner_state::AgentRunner;

fn visible_terminal_reason(event: &AgentEvent) -> Option<&'static str> {
    match event {
        AgentEvent::Complete { .. } => Some("complete"),
        AgentEvent::Cancelled { .. } => Some("cancelled"),
        AgentEvent::Error { .. } => Some("error"),
        _ => None,
    }
}

/// Inbox to the account-wide change feed: `(session_id, event)` before the
/// writer assigns a seq. Threaded as `Option` so engine-internal callers that
/// have no feed (tests, standalone embeddings) can pass `None`. Defined here so
/// the engine stays free of any `bamboo-server` dependency.
pub type AccountFeedInbox = mpsc::Sender<(Option<String>, AgentEvent)>;

#[cfg(test)]
#[path = "root_actor_event_forwarder_tests.rs"]
mod root_actor_tests;

/// Completion signal for the durable history barrier emitted at the end of a
/// run. The runner must not become replaceable until the forwarder has actually
/// published `SessionHistoryCommitted`; merely enqueueing the event leaves a
/// race where a successor can retire this generation first.
#[derive(Clone, Debug)]
pub struct HistoryCommitBarrier {
    revision: watch::Receiver<u64>,
}

/// Forwarder-owned half of [`HistoryCommitBarrier`].
#[derive(Clone, Debug)]
pub struct HistoryCommitAcknowledger {
    revision: watch::Sender<u64>,
}

/// Create the acknowledgment pair shared by an event producer and forwarder.
pub fn history_commit_barrier() -> (HistoryCommitAcknowledger, HistoryCommitBarrier) {
    let (revision_tx, revision_rx) = watch::channel(0);
    (
        HistoryCommitAcknowledger {
            revision: revision_tx,
        },
        HistoryCommitBarrier {
            revision: revision_rx,
        },
    )
}

impl HistoryCommitAcknowledger {
    /// Mark one history barrier as synchronously published to every configured
    /// sink. Call this only after the publication fence accepts the event.
    pub fn acknowledge(&self) {
        self.revision
            .send_modify(|revision| *revision = revision.saturating_add(1));
    }
}

impl HistoryCommitBarrier {
    /// Enqueue a history barrier and wait until the matching forwarder has
    /// published it. Returns `false` if the forwarder disappeared first.
    pub async fn send_and_wait(
        &mut self,
        event_tx: &mpsc::Sender<AgentEvent>,
        session_id: String,
    ) -> bool {
        let observed = *self.revision.borrow_and_update();
        if event_tx
            .send(AgentEvent::SessionHistoryCommitted { session_id })
            .await
            .is_err()
        {
            return false;
        }

        loop {
            if *self.revision.borrow() > observed {
                return true;
            }
            if self.revision.changed().await.is_err() {
                return false;
            }
        }
    }
}

/// Forward a durable change event onto the account feed, if an inbox is wired.
///
/// Ephemeral events (tokens, heartbeats, …) are filtered out before any clone,
/// so this is near-free on the hot path. `session_id` is supplied explicitly so
/// terminal events (which carry no id) still route to the right session.
fn mirror_to_account_feed(inbox: &Option<AccountFeedInbox>, session_id: &str, event: &AgentEvent) {
    if let Some(inbox) = inbox {
        if event.is_durable_change() {
            let route_session_id = event.session_id().unwrap_or(session_id);
            let _ = inbox.try_send((Some(route_session_id.to_string()), event.clone()));
        }
    }
}

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

    use bamboo_agent_core::ToolResult;
    use serde_json::json;

    #[tokio::test]
    async fn generic_forwarder_projects_native_download_without_changing_ordinary_results() {
        let session_id = "generic-download-display";
        let (broadcast_tx, mut broadcast_rx) = broadcast::channel(16);
        let mut runner = AgentRunner::new();
        runner.status = super::super::runner_state::AgentStatus::Running;
        runner.event_sender = broadcast_tx.clone();
        let run_id = runner.run_id.clone();
        let runners = Arc::new(RwLock::new(HashMap::from([(
            session_id.to_string(),
            runner,
        )])));
        let (input, forwarder) =
            create_event_forwarder(session_id.to_string(), run_id, broadcast_tx, runners, None);
        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::ExecutionStarted { .. }
        ));

        let private = "private-file-url-and-bytes";
        input
            .send(AgentEvent::ToolStart {
                tool_call_id: "download".into(),
                tool_name: "browser".into(),
                arguments: json!({"action":"download","selector":private,"expected_epoch":9}),
            })
            .await
            .unwrap();
        input
            .send(AgentEvent::ToolError {
                tool_call_id: "download".into(),
                error: private.into(),
            })
            .await
            .unwrap();
        input
            .send(AgentEvent::ToolComplete {
                tool_call_id: "download".into(),
                result: ToolResult::text(false, private),
            })
            .await
            .unwrap();
        input
            .send(AgentEvent::ToolStart {
                tool_call_id: "read".into(),
                tool_name: "Read".into(),
                arguments: json!({"path":"readme.md"}),
            })
            .await
            .unwrap();
        input
            .send(AgentEvent::ToolComplete {
                tool_call_id: "read".into(),
                result: ToolResult::text(true, "ordinary Read result"),
            })
            .await
            .unwrap();
        input
            .send(AgentEvent::HookLifecycle {
                hook_name: private.into(),
                point: bamboo_domain::AgentHookPoint::AfterToolExecution,
                phase: private.into(),
                duration_ms: 7,
                decision: bamboo_domain::HookResult::WithContext {
                    result: Box::new(bamboo_domain::HookResult::Deny {
                        reason: private.into(),
                    }),
                    text: private.into(),
                },
            })
            .await
            .unwrap();
        drop(input);

        let mut visible = Vec::new();
        for _ in 0..6 {
            visible.push(
                tokio::time::timeout(std::time::Duration::from_secs(5), broadcast_rx.recv())
                    .await
                    .unwrap()
                    .unwrap(),
            );
        }
        for event in &visible[..3] {
            assert!(!serde_json::to_string(event).unwrap().contains(private));
        }
        assert!(
            matches!(&visible[2], AgentEvent::ToolComplete { result, .. }
            if result.result == "Tool result hidden" && result.images.is_empty())
        );
        assert!(
            matches!(&visible[4], AgentEvent::ToolComplete { result, .. }
            if result.result == "ordinary Read result")
        );
        assert!(!serde_json::to_string(&visible[5])
            .unwrap()
            .contains(private));
        assert!(matches!(
            &visible[5],
            AgentEvent::HookLifecycle {
                point: bamboo_domain::AgentHookPoint::AfterToolExecution,
                duration_ms: 7,
                decision: bamboo_domain::HookResult::WithContext { .. },
                ..
            }
        ));
        forwarder.await.unwrap();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn hundreds_of_child_streams_progress_while_global_registry_is_locked() {
        let runners = Arc::new(RwLock::new(HashMap::new()));
        let mut streams = Vec::new();
        for index in 0..512 {
            let id = format!("child-{index}");
            let mut runner = AgentRunner::new();
            runner.status = super::super::runner_state::AgentStatus::Running;
            let mut receiver = runner.event_sender.subscribe();
            let sender = runner.event_sender.clone();
            let run_id = runner.run_id.clone();
            runners.write().await.insert(id.clone(), runner);
            let (input, task) = create_event_forwarder(id, run_id, sender, runners.clone(), None);
            assert!(matches!(
                receiver.recv().await.unwrap(),
                AgentEvent::ExecutionStarted { .. }
            ));
            streams.push((input, receiver, task));
        }
        let held_registry = runners.write().await;
        let started = std::time::Instant::now();
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            futures::future::join_all(streams.iter_mut().map(|(input, receiver, _)| async move {
                for _ in 0..64 {
                    input
                        .send(AgentEvent::Token {
                            content: "delta".into(),
                        })
                        .await
                        .unwrap();
                }
                for _ in 0..64 {
                    assert!(matches!(
                        receiver.recv().await.unwrap(),
                        AgentEvent::Token { .. }
                    ));
                }
            }))
            .await;
        })
        .await
        .expect("independent token streams must not wait for registry ownership");
        eprintln!(
            "512 child streams / 32768 tokens with held registry: {:?}",
            started.elapsed()
        );
        assert!(held_registry
            .values()
            .all(|runner| runner.last_activity_at().is_some()));
        drop(held_registry);
        for (input, _, task) in streams {
            drop(input);
            task.await.unwrap();
        }
    }

    #[tokio::test]
    async fn child_approval_change_routes_to_parent_account_envelope() {
        let (tx, mut rx) = mpsc::channel(4);
        let event = AgentEvent::ChildApprovalChanged {
            parent_session_id: "parent-1".into(),
            child_session_id: "child-1".into(),
            child_attempt: 1,
            request_id: "req-1".into(),
            version: 2,
            status: "approved".into(),
            reason: None,
            tool_name: "Bash".into(),
            permission: "execute".into(),
            resource: "/tmp/x".into(),
            created_at: "2026-01-01T00:00:00Z".into(),
            resolved_at: Some("2026-01-01T00:00:01Z".into()),
        };

        mirror_to_account_feed(&Some(tx), "child-1", &event);
        let (session_id, mirrored) = rx.recv().await.unwrap();
        assert_eq!(session_id.as_deref(), Some("parent-1"));
        assert!(matches!(mirrored, AgentEvent::ChildApprovalChanged { .. }));
    }

    #[tokio::test]
    async fn child_browser_approval_is_redacted_before_broadcast_replay_and_account_feed() {
        let session_id = "child-browser-private";
        let private = "private-download-selector";
        let (broadcast_tx, mut broadcast_rx) = broadcast::channel(8);
        let mut runner = AgentRunner::new();
        runner.status = super::super::runner_state::AgentStatus::Running;
        runner.event_sender = broadcast_tx.clone();
        let run_id = runner.run_id.clone();
        let runners = Arc::new(RwLock::new(HashMap::from([(
            session_id.to_string(),
            runner,
        )])));
        let (account_tx, mut account_rx) = mpsc::channel(8);
        let (event_tx, forwarder) = create_event_forwarder(
            session_id.into(),
            run_id,
            broadcast_tx,
            runners.clone(),
            Some(account_tx),
        );
        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::ExecutionStarted { .. }
        ));
        event_tx
            .send(AgentEvent::ChildApprovalChanged {
                parent_session_id: "parent-private".into(),
                child_session_id: session_id.into(),
                child_attempt: 4,
                request_id: "request-private".into(),
                version: 7,
                status: "denied".into(),
                reason: Some(private.into()),
                tool_name: "browser".into(),
                permission: private.into(),
                resource: format!("browser:17:download:css:{private}"),
                created_at: "2026-09-24T00:00:00Z".into(),
                resolved_at: Some("2026-09-24T00:00:01Z".into()),
            })
            .await
            .unwrap();
        let live = broadcast_rx.recv().await.unwrap();
        assert!(!serde_json::to_string(&live).unwrap().contains(private));
        let _started = account_rx.recv().await.unwrap();
        let (route, feed) = account_rx.recv().await.unwrap();
        assert_eq!(route.as_deref(), Some("parent-private"));
        assert!(!serde_json::to_string(&feed).unwrap().contains(private));
        assert!(
            !serde_json::to_string(&runners.read().await[session_id].last_critical_events)
                .unwrap()
                .contains(private)
        );
        drop(event_tx);
        forwarder.await.unwrap();
    }

    #[tokio::test]
    async fn history_commit_barrier_acknowledges_actual_publication() {
        let session_id = "session-history-barrier";
        let (broadcast_tx, mut broadcast_rx) = broadcast::channel(16);
        let mut runner = AgentRunner::new();
        runner.status = super::super::runner_state::AgentStatus::Running;
        runner.event_sender = broadcast_tx.clone();
        let run_id = runner.run_id.clone();
        let runners = Arc::new(RwLock::new(HashMap::from([(
            session_id.to_string(),
            runner,
        )])));
        let (event_tx, forwarder, mut barrier) = create_event_forwarder_with_history_commit_barrier(
            session_id.to_string(),
            run_id,
            broadcast_tx,
            runners,
            None,
        );

        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::ExecutionStarted { .. }
        ));
        assert!(
            barrier
                .send_and_wait(&event_tx, session_id.to_string())
                .await,
            "producer must not continue until the forwarder publishes the barrier"
        );
        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::SessionHistoryCommitted { session_id: id } if id == session_id
        ));

        drop(event_tx);
        forwarder.await.unwrap();
    }

    #[tokio::test]
    async fn delayed_old_forwarder_cannot_publish_after_successor_reservation() {
        let session_id = "session-generation";
        let (broadcast_tx, mut broadcast_rx) = broadcast::channel(16);
        let mut successor = AgentRunner::new();
        successor.run_id = "run-new".to_string();
        successor.status = super::super::runner_state::AgentStatus::Running;
        successor.event_sender = broadcast_tx.clone();
        let runners = Arc::new(RwLock::new(HashMap::from([(
            session_id.to_string(),
            successor,
        )])));

        // The successor is already visible on the shared transport before an
        // old forwarder task finally gets CPU time.
        broadcast_tx
            .send(AgentEvent::ExecutionStarted {
                run_id: "run-new".to_string(),
                session_id: session_id.to_string(),
                started_at: Utc::now().to_rfc3339(),
            })
            .unwrap();
        let (old_tx, old_forwarder) = create_event_forwarder(
            session_id.to_string(),
            "run-old".to_string(),
            broadcast_tx.clone(),
            runners,
            None,
        );
        let _ = old_tx
            .send(AgentEvent::NeedClarification {
                question: "stale".to_string(),
                options: Some(vec!["A".to_string()]),
                tool_call_id: Some("old-tool".to_string()),
                tool_name: Some("ConclusionWithOptions".to_string()),
                allow_custom: false,
                source: Some(bamboo_agent_core::PendingQuestionSource::PauseTool),
            })
            .await;
        drop(old_tx);
        old_forwarder.await.unwrap();

        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::ExecutionStarted { ref run_id, .. } if run_id == "run-new"
        ));
        assert!(
            tokio::time::timeout(std::time::Duration::from_millis(50), broadcast_rx.recv())
                .await
                .is_err(),
            "old Started/Need/Complete must all be suppressed"
        );
    }

    #[tokio::test]
    async fn wrapped_child_lifecycle_reaches_bounded_legacy_replay_consumer() {
        let session_id = "root";
        let (broadcast_tx, mut broadcast_rx) = broadcast::channel(8);
        let mut runner = AgentRunner::new();
        runner.status = super::super::runner_state::AgentStatus::Running;
        runner.event_sender = broadcast_tx.clone();
        let run_id = runner.run_id.clone();
        let runners = Arc::new(RwLock::new(HashMap::from([(session_id.into(), runner)])));
        let (input, forwarder) = create_event_forwarder(
            session_id.into(),
            run_id,
            broadcast_tx,
            runners.clone(),
            None,
        );
        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::ExecutionStarted { .. }
        ));
        let legacy = AgentEvent::SubAgentEvent {
            parent_session_id: "root".into(),
            child_session_id: "parent".into(),
            event: Box::new(AgentEvent::SubAgentStarted {
                parent_session_id: "parent".into(),
                child_session_id: "grandchild".into(),
                title: Some("work".into()),
            }),
        };
        input.send(legacy).await.unwrap();
        assert!(matches!(
            broadcast_rx.recv().await.unwrap(),
            AgentEvent::SubAgentEvent { .. }
        ));
        let guard = runners.read().await;
        let cached = &guard[session_id].last_critical_events;
        assert_eq!(cached.len(), 1);
        assert!(matches!(&cached[0], AgentEvent::SubAgentEvent { .. }));
        drop(guard);
        drop(input);
        forwarder.await.unwrap();
    }
}

/// Create an MPSC channel for agent events and spawn a forwarding task
/// that relays events to the broadcast sender while tracking runner
/// diagnostic fields for live visibility.
///
/// `account_feed_inbox`, when present, also mirrors durable change events onto
/// the account-wide feed for resumable multi-client sync.
///
/// Returns `(mpsc_tx, forwarder_handle)`.
pub fn create_event_forwarder(
    session_id: String,
    run_id: String,
    broadcast_tx: broadcast::Sender<AgentEvent>,
    runners: Arc<RwLock<HashMap<String, AgentRunner>>>,
    account_feed_inbox: Option<AccountFeedInbox>,
) -> (mpsc::Sender<AgentEvent>, tokio::task::JoinHandle<()>) {
    let (event_tx, forwarder, _history_commit_barrier) =
        create_event_forwarder_with_history_commit_barrier(
            session_id,
            run_id,
            broadcast_tx,
            runners,
            account_feed_inbox,
        );
    (event_tx, forwarder)
}

/// Create an event forwarder plus a producer-side durable-history barrier.
/// Runtime paths that finalize and replace runner generations must use this
/// variant so they can await actual barrier publication before finalization.
pub fn create_event_forwarder_with_history_commit_barrier(
    session_id: String,
    run_id: String,
    broadcast_tx: broadcast::Sender<AgentEvent>,
    runners: Arc<RwLock<HashMap<String, AgentRunner>>>,
    account_feed_inbox: Option<AccountFeedInbox>,
) -> (
    mpsc::Sender<AgentEvent>,
    tokio::task::JoinHandle<()>,
    HistoryCommitBarrier,
) {
    create_event_forwarder_with_root_actor(
        session_id,
        run_id,
        broadcast_tx,
        runners,
        account_feed_inbox,
        None,
    )
}

async fn publish_forwarded_event(
    root_actor: Option<&crate::events::RootActorEventPublication>,
    session_id: &str,
    event: &AgentEvent,
    publish: Box<dyn FnOnce() -> bool + Send>,
) -> std::io::Result<()> {
    if let Some(owner) = root_actor {
        owner
            .publish(event.session_id().unwrap_or(session_id), event, publish)
            .await
    } else if publish() {
        Ok(())
    } else {
        Err(std::io::Error::other("event publication was retired"))
    }
}

const ROOT_AUTHORITY_LOST: &str =
    "This run was interrupted because its execution ownership was lost. Reload the conversation before retrying.";

fn record_publication_rejection(
    error: &std::io::Error,
    session_id: &str,
    run_id: &str,
    event: Option<&AgentEvent>,
    root_actor: Option<&crate::events::RootActorEventPublication>,
    phase: &'static str,
) -> String {
    use super::workspace_diagnostics::{record_failure, WorkspaceLogContext};
    let diagnostic_id = format!("RPA-{}", uuid::Uuid::new_v4().simple());
    let request_id = event.and_then(|event| match event {
        AgentEvent::ToolStart { tool_call_id, .. }
        | AgentEvent::ToolComplete { tool_call_id, .. }
        | AgentEvent::ToolError { tool_call_id, .. }
        | AgentEvent::ToolLifecycle { tool_call_id, .. } => Some(tool_call_id.as_str()),
        _ => None,
    });
    if let Some(owner) = root_actor {
        let (lease_epoch, attempt, fence_id) = owner.diagnostic_identity();
        tracing::warn!(target: "bamboo.execution_publication", %diagnostic_id,
            lease_epoch, attempt, %fence_id, phase,
            "Root publication rejected under captured execution fence");
    }
    record_failure(
        WorkspaceLogContext {
            session_id,
            run_id: Some(run_id),
            request_id,
            diagnostic_id: Some(&diagnostic_id),
        },
        phase,
        error,
    )
}

struct RootInterruption<'a> {
    session_id: &'a str,
    run_id: &'a str,
    runners: &'a Arc<RwLock<HashMap<String, AgentRunner>>>,
    publication: &'a super::event_publication::EventPublication,
    visible: &'a super::visible_messages::VisibleMessageStream,
    cancel_token: &'a CancellationToken,
}

impl RootInterruption<'_> {
    async fn close(self, diagnostic_id: &str) {
        // Retain the original exact-owner cancellation/publication behavior.
        // A diagnostic number never permits mutation or retries a tool.
        self.cancel_token.cancel();
        self.publication.retire().await;
        self.visible.mark_terminal("error");
        let guard = self.runners.read().await;
        if let Some(runner) = guard
            .get(self.session_id)
            .filter(|runner| runner.run_id == self.run_id)
        {
            let _ = runner.event_sender.send(AgentEvent::Error {
                message: format!(
                    "{ROOT_AUTHORITY_LOST} [actor_authority_conflict; {diagnostic_id}]"
                ),
            });
        }
    }
}

/// Close only the obsolete run's live transport. This is not a runtime-state
/// publication: it never enters the account journal, replay cache, or storage.
/// The exact local runner guard keeps this fixed error ahead of any successor's
/// Started frame on the shared session channel. A successor already installed
/// in the registry must not receive the obsolete run's error or cancellation.
/// Server adapters share this boundary with the generic engine forwarder.
pub async fn interrupt_root_on_authority_loss(
    error: &std::io::Error,
    session_id: &str,
    run_id: &str,
    runners: &Arc<RwLock<HashMap<String, AgentRunner>>>,
    publication: &super::event_publication::EventPublication,
    visible: &super::visible_messages::VisibleMessageStream,
    cancel_token: &CancellationToken,
) {
    if !error
        .get_ref()
        .is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>())
    {
        return;
    }
    let diagnostic_id = record_publication_rejection(
        error,
        session_id,
        run_id,
        None,
        None,
        "root_authority_interrupt",
    );
    RootInterruption {
        session_id,
        run_id,
        runners,
        publication,
        visible,
        cancel_token,
    }
    .close(&diagnostic_id)
    .await;
}

/// The Root adapter passes its immutable capability before any Started frame.
/// Legacy Child/Supervisor/standalone callers keep the existing queue behavior.
pub fn create_event_forwarder_with_root_actor(
    session_id: String,
    run_id: String,
    broadcast_tx: broadcast::Sender<AgentEvent>,
    runners: Arc<RwLock<HashMap<String, AgentRunner>>>,
    account_feed_inbox: Option<AccountFeedInbox>,
    root_actor: Option<crate::events::RootActorEventPublication>,
) -> (
    mpsc::Sender<AgentEvent>,
    tokio::task::JoinHandle<()>,
    HistoryCommitBarrier,
) {
    let (mpsc_tx, mut mpsc_rx) = mpsc::channel::<AgentEvent>(100);
    let (history_commit_acknowledger, history_commit_barrier) = history_commit_barrier();
    let forward = async move {
        if root_actor
            .as_ref()
            .is_some_and(|owner| !owner.matches_execution(&session_id, &run_id))
        {
            tracing::warn!(%session_id, %run_id, "Root forwarder handoff does not match its execution");
            return;
        }
        // A bound Root never enters the legacy account queue.
        let legacy_feed = if root_actor.is_some() {
            None
        } else {
            account_feed_inbox
        };
        let started_event = AgentEvent::ExecutionStarted {
            run_id: run_id.clone(),
            session_id: session_id.clone(),
            started_at: Utc::now().to_rfc3339(),
        };
        let (publication, visible_messages, cancel_token) = {
            let guard = runners.clone().read_owned().await;
            let Some(runner) = guard
                .get(&session_id)
                .filter(|runner| runner.run_id == run_id)
            else {
                return;
            };
            let publication = runner.event_publication.clone();
            let visible_messages = runner.visible_messages.clone();
            let cancel_token = runner.cancel_token.clone();
            let feed = legacy_feed.clone();
            let sid = session_id.clone();
            let tx = broadcast_tx.clone();
            let frame = started_event.clone();
            if let Err(error) = publish_forwarded_event(
                root_actor.as_ref(),
                &session_id,
                &frame,
                Box::new(move || {
                    let _guard = guard;
                    mirror_to_account_feed(&feed, &sid, &started_event);
                    let _ = tx.send(started_event);
                    true
                }),
            )
            .await
            {
                let diagnostic_id = record_publication_rejection(
                    &error,
                    &session_id,
                    &run_id,
                    Some(&frame),
                    root_actor.as_ref(),
                    "root_started_publication",
                );
                if error
                    .get_ref()
                    .is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>())
                {
                    RootInterruption {
                        session_id: &session_id,
                        run_id: &run_id,
                        runners: &runners,
                        publication: &publication,
                        visible: &visible_messages,
                        cancel_token: &cancel_token,
                    }
                    .close(&diagnostic_id)
                    .await;
                }
                return;
            }
            (publication, visible_messages, cancel_token)
        };
        let mut tool_event_display = bamboo_agent_core::NativeToolEventDisplay::default();
        while let Some(event) = mpsc_rx.recv().await {
            match &event {
                AgentEvent::VisibleMessageStart {
                    message_id,
                    created_at,
                } => {
                    if !publication.publish(|| {
                        visible_messages.start(message_id.clone(), created_at.to_owned())
                    }) {
                        return;
                    }
                    continue;
                }
                AgentEvent::VisibleMessageDiscard { message_id } => {
                    if !publication.publish(|| visible_messages.discard(message_id)) {
                        return;
                    }
                    continue;
                }
                _ => {}
            }
            let event = tool_event_display.project(event);
            let event = crate::external_agents::live::approval_event_for_display(event);
            let needs_runner_update = event.is_replayable_session_state()
                || matches!(
                    &event,
                    AgentEvent::TokenBudgetUpdated { .. }
                        | AgentEvent::ToolStart { .. }
                        | AgentEvent::ToolLifecycle { .. }
                        | AgentEvent::RunnerProgress { .. }
                );
            let is_history_commit = matches!(&event, AgentEvent::SessionHistoryCommitted { .. });
            let frame = event.clone();
            let feed = legacy_feed.clone();
            let sid = session_id.clone();
            let tx = broadcast_tx.clone();
            let result = if !needs_runner_update {
                let visible_token = match &event {
                    AgentEvent::Token { content } => Some(content.clone()),
                    _ => None,
                };
                let terminal_reason = visible_terminal_reason(&event);
                // Ephemeral token traffic retains its independent hot path.
                let owner = root_actor
                    .as_ref()
                    .filter(|_| event.is_durable_change() || terminal_reason.is_some());
                let publication = publication.clone();
                let visible = visible_messages.clone();
                publish_forwarded_event(
                    owner,
                    &session_id,
                    &frame,
                    Box::new(move || {
                        publication.publish(|| {
                            if let Some(content) = visible_token {
                                visible.append(content);
                            }
                            mirror_to_account_feed(&feed, &sid, &event);
                            let _ = tx.send(event);
                            if let Some(reason) = terminal_reason {
                                visible.mark_terminal(reason);
                            }
                            if is_history_commit {
                                visible.history_committed();
                            }
                        })
                    }),
                )
                .await
            } else {
                let mut guard = runners.clone().write_owned().await;
                if !guard
                    .get(&session_id)
                    .is_some_and(|runner| runner.run_id == run_id)
                {
                    return;
                }
                let publication = publication.clone();
                publish_forwarded_event(
                    root_actor.as_ref(),
                    &session_id,
                    &frame,
                    Box::new(move || {
                        let runner = guard.get_mut(&sid).expect("retained exact runner guard");
                        runner.last_event_at = Some(Utc::now());
                        publication.touch();
                        if event.is_replayable_session_state() {
                            runner.push_critical_event(event.clone());
                        }
                        match &event {
                            AgentEvent::TokenBudgetUpdated { .. } => {
                                runner.last_budget_event = Some(event.clone())
                            }
                            AgentEvent::ToolStart { tool_name, .. } => {
                                runner.last_tool_name = Some(tool_name.clone());
                                runner.last_tool_phase = Some("begin".into());
                            }
                            AgentEvent::ToolLifecycle {
                                tool_name, phase, ..
                            } => {
                                runner.last_tool_name = Some(tool_name.clone());
                                runner.last_tool_phase = Some(phase.clone());
                            }
                            AgentEvent::RunnerProgress { round_count, .. } => {
                                runner.round_count = *round_count;
                                runner.visible_messages.begin_round(*round_count);
                            }
                            _ => {}
                        }
                        mirror_to_account_feed(&feed, &sid, &event);
                        let _ = tx.send(event);
                        true
                    }),
                )
                .await
            };
            if let Err(error) = result {
                let diagnostic_id = record_publication_rejection(
                    &error,
                    &session_id,
                    &run_id,
                    Some(&frame),
                    root_actor.as_ref(),
                    "root_runtime_publication",
                );
                if error
                    .get_ref()
                    .is_some_and(|cause| cause.is::<bamboo_domain::SessionAuthorityConflict>())
                {
                    RootInterruption {
                        session_id: &session_id,
                        run_id: &run_id,
                        runners: &runners,
                        publication: &publication,
                        visible: &visible_messages,
                        cancel_token: &cancel_token,
                    }
                    .close(&diagnostic_id)
                    .await;
                }
                return;
            }
            if is_history_commit {
                history_commit_acknowledger.acknowledge();
            }
        }
    };
    (mpsc_tx, tokio::spawn(forward), history_commit_barrier)
}