//! Event handling for dirigent_core event stream.
//!
//! The EventHandler subscribes to dirigent_core's global event stream and writes
//! to the archive in real-time, accumulating streaming message chunks into complete
//! messages.

use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;
use uuid::Uuid;
use chrono::Utc;

use dirigent_protocol::{
    ContentBlock, Event, Message, MessagePart, MessageRole, Session, SessionUpdate,
    TurnCompleteTrigger,
};
use dirigent_protocol::streaming::{BusEvent, BusReceiver};

/// Closure that publishes a `BusEvent` to the shared event bus.
///
/// The archivist cannot depend on `dirigent_core::SharingBus` directly
/// (that would introduce a dependency cycle). Callers in `api::archivist`
/// and `web::server` wrap the core's `SharingBus::publish` call in a
/// closure that matches this signature and install it via
/// [`EventHandler::set_bus_publisher`]. Implementations must spawn the
/// publish onto a tokio task (the call site here is synchronous).
pub type BusPublishFn = Arc<dyn Fn(BusEvent) + Send + Sync>;

use crate::accumulator::MessageAccumulator;
use crate::coordinator::Archivist;
use crate::error::Result;
use crate::types::{
    MessageRecord, MetaEventRecord, MetaEventType, RegisterSessionRequest,
    RegisterSessionResponse, RegisterStatus, SessionCompleteness,
};

/// Event handler for subscribing to dirigent_core events and archiving them
pub struct EventHandler {
    archivist: Arc<Archivist>,
    accumulator: Mutex<MessageAccumulator>,
    /// Track which message IDs have been archived to prevent duplicates
    archived_messages: Mutex<HashSet<String>>,
    /// Sessions excluded from archiving (per-session toggle support)
    /// Key is "connector_id:session_id" composite key
    excluded_sessions: Mutex<HashSet<String>>,
    /// Connectors whose sessions should be excluded from archiving by default
    /// This is used for Gateway connectors where users can opt-in to archiving
    excluded_connector_ids: Mutex<HashSet<String>>,
    /// Sessions currently being replayed via session/load.
    /// SessionUpdate events for these sessions are suppressed to avoid
    /// re-archiving messages that already exist in the archive.
    replaying_sessions: Mutex<HashSet<String>>,
    /// Bus publisher used to emit `SessionRegistered` back onto the shared
    /// event bus (`dirigent_core::SharingBus`). Installed via
    /// [`Self::set_bus_publisher`] from the boot wiring in the server
    /// crate. Without it, downstream consumers (e.g. the web UI's session
    /// list refresh) won't receive the signal that a session is durably
    /// registered — but archival still functions.
    bus_publisher: Option<BusPublishFn>,
}

impl EventHandler {
    /// Create a new event handler
    pub fn new(archivist: Arc<Archivist>) -> Self {
        Self {
            archivist,
            accumulator: Mutex::new(
                MessageAccumulator::new().expect("Failed to create accumulator"),
            ),
            archived_messages: Mutex::new(HashSet::new()),
            excluded_sessions: Mutex::new(HashSet::new()),
            excluded_connector_ids: Mutex::new(HashSet::new()),
            replaying_sessions: Mutex::new(HashSet::new()),
            bus_publisher: None,
        }
    }

    /// Install a bus publisher so the archivist can emit its
    /// `SessionRegistered` events onto the shared event bus.
    ///
    /// The callback receives a [`BusEvent`] and is expected to forward it
    /// to `dirigent_core::SharingBus::publish` (likely by spawning a tokio
    /// task).
    pub fn set_bus_publisher(&mut self, publisher: BusPublishFn) {
        self.bus_publisher = Some(publisher);
    }

    /// Add a connector ID to the list of connectors whose sessions should be excluded by default
    ///
    /// Sessions from these connectors will start with archiving disabled. Users can still
    /// enable archiving for individual sessions using the toggle.
    ///
    /// This is useful for Gateway connectors where most sessions are transient
    /// and don't need to be archived.
    pub async fn add_excluded_connector(&self, connector_id: &str) {
        let mut excluded = self.excluded_connector_ids.lock().await;
        excluded.insert(connector_id.to_string());
        tracing::info!(
            "Connector '{}' added to excluded list - sessions will not archive by default",
            connector_id
        );
    }

    /// Remove a connector ID from the list of excluded connectors
    ///
    /// New sessions from this connector will archive by default (existing exclusions are not removed).
    pub async fn remove_excluded_connector(&self, connector_id: &str) {
        let mut excluded = self.excluded_connector_ids.lock().await;
        excluded.remove(connector_id);
        tracing::info!(
            "Connector '{}' removed from excluded list - sessions will archive by default",
            connector_id
        );
    }

    /// Check if a connector is in the excluded list
    pub async fn is_connector_excluded(&self, connector_id: &str) -> bool {
        let excluded = self.excluded_connector_ids.lock().await;
        excluded.contains(connector_id)
    }

    /// Check if a session is excluded from archiving
    ///
    /// A session is excluded if:
    /// 1. It's explicitly in the excluded_sessions set, OR
    /// 2. Its connector is in excluded_connector_ids (e.g., Gateway) and the session
    ///    hasn't been explicitly enabled (which would have removed it from excluded_sessions)
    ///
    /// For Gateway connectors, sessions default to excluded until explicitly enabled.
    pub async fn is_session_excluded(&self, connector_id: &str, session_id: &str) -> bool {
        let key = format!("{}:{}", connector_id, session_id);
        let excluded = self.excluded_sessions.lock().await;

        // If session is explicitly in excluded set, it's excluded
        if excluded.contains(&key) {
            return true;
        }

        // Check if connector defaults to excluded (e.g., Gateway)
        // For these connectors, sessions are excluded by default
        let connector_excluded = {
            let excluded_connectors = self.excluded_connector_ids.lock().await;
            excluded_connectors.contains(connector_id)
        };

        if connector_excluded {
            // Connector defaults to excluded. Session is excluded UNLESS it has been
            // explicitly enabled. We track enabled sessions by checking if they were
            // previously in excluded_sessions and removed (toggle enabled them).
            // Since they're not in excluded_sessions now, check included_sessions.
            // For simplicity, we use a marker: if connector is excluded but session
            // is not in excluded_sessions, check if we have an "enabled" marker.
            // Since we don't have a separate set, we'll treat "not in excluded_sessions"
            // for an excluded connector as still excluded (the default state).
            // The toggle will add an "enabled:key" marker to excluded_sessions to indicate
            // the session was explicitly enabled.
            let enabled_key = format!("enabled:{}", key);
            if excluded.contains(&enabled_key) {
                return false; // Explicitly enabled
            }
            return true; // Default excluded for Gateway
        }

        false
    }

    /// Exclude a session from archiving (disable archiving for this session)
    ///
    /// Returns true if the session was newly excluded, false if already excluded.
    pub async fn exclude_session(&self, connector_id: &str, session_id: &str) -> bool {
        let key = format!("{}:{}", connector_id, session_id);
        let mut excluded = self.excluded_sessions.lock().await;
        let newly_excluded = excluded.insert(key.clone());
        if newly_excluded {
            tracing::info!(
                "Archiving disabled for session {} (connector: {})",
                session_id,
                connector_id
            );
        }
        newly_excluded
    }

    /// Include a session in archiving (enable archiving for this session)
    ///
    /// Returns true if the session was previously excluded, false if it wasn't excluded.
    pub async fn include_session(&self, connector_id: &str, session_id: &str) -> bool {
        let key = format!("{}:{}", connector_id, session_id);
        let mut excluded = self.excluded_sessions.lock().await;
        let was_excluded = excluded.remove(&key);
        if was_excluded {
            tracing::info!(
                "Archiving enabled for session {} (connector: {})",
                session_id,
                connector_id
            );
        }
        was_excluded
    }

    /// Toggle session archiving status
    ///
    /// Returns the new status: true = archiving enabled, false = archiving disabled
    ///
    /// For Gateway connectors (in excluded_connector_ids), sessions default to excluded.
    /// Toggling uses an "enabled:" marker to track explicitly enabled sessions.
    pub async fn toggle_session_archiving(&self, connector_id: &str, session_id: &str) -> bool {
        let key = format!("{}:{}", connector_id, session_id);
        let enabled_key = format!("enabled:{}", key);

        // Check if connector defaults to excluded (e.g., Gateway)
        let connector_excluded = {
            let excluded_connectors = self.excluded_connector_ids.lock().await;
            excluded_connectors.contains(connector_id)
        };

        let mut excluded = self.excluded_sessions.lock().await;

        if connector_excluded {
            // Gateway connector: sessions default to excluded
            if excluded.contains(&enabled_key) {
                // Currently enabled, disable by removing the enabled marker
                excluded.remove(&enabled_key);
                tracing::info!(
                    "Archiving disabled for session {} (connector: {}, Gateway default)",
                    session_id,
                    connector_id
                );
                false // archiving now disabled
            } else {
                // Currently excluded (default), enable by adding the enabled marker
                // Also remove from explicit excluded set if present
                excluded.remove(&key);
                excluded.insert(enabled_key);
                tracing::info!(
                    "Archiving enabled for session {} (connector: {}, Gateway override)",
                    session_id,
                    connector_id
                );
                true // archiving now enabled
            }
        } else {
            // Non-Gateway connector: normal toggle logic
            if excluded.contains(&key) {
                excluded.remove(&key);
                tracing::info!(
                    "Archiving enabled for session {} (connector: {})",
                    session_id,
                    connector_id
                );
                true // archiving now enabled
            } else {
                excluded.insert(key);
                tracing::info!(
                    "Archiving disabled for session {} (connector: {})",
                    session_id,
                    connector_id
                );
                false // archiving now disabled
            }
        }
    }

    /// Canonicalize a finalized message record and write it to the archive.
    ///
    /// This is the SINGLE write path for all finalization triggers (TurnComplete,
    /// SessionIdle, stale timeout, shutdown flush). Resolves connector_id to
    /// connector_uid, then to the canonical scroll_id, rewrites the record's
    /// session field, and appends.
    async fn canonicalize_and_write(
        &self,
        record: &mut MessageRecord,
        connector_id: &str,
        native_session_id: &str,
        message_id: &str,
        finalized_via: &str,
    ) -> Result<()> {
        // 1. Resolve connector_id to connector_uid
        let connector_uid = match self.archivist.resolve_connector_uid(connector_id).await {
            Ok(uid) => uid,
            Err(e) => {
                tracing::warn!(
                    "canonicalize_and_write: Failed to resolve connector_uid for '{}': {}. \
                     Message {} will not be archived.",
                    connector_id,
                    e,
                    message_id
                );
                return Ok(());
            }
        };

        // 2. Resolve or lazy-register session to get canonical scroll_id
        let scroll_id = self
            .resolve_or_register_session(
                connector_uid,
                native_session_id,
                &format!(
                    "{}: Session mapping missing during message write",
                    finalized_via
                ),
            )
            .await?;

        // 3. Rewrite record to canonical identity
        record.session = scroll_id;

        tracing::info!(
            "canonicalize_and_write: Writing message {} ({} bytes) for session {} -> scroll_id {} via {}",
            message_id,
            record.content_md.len(),
            native_session_id,
            scroll_id,
            finalized_via,
        );

        // 4. Write to archive
        self.archivist
            .append_messages(scroll_id, vec![record.clone()], None)
            .await?;

        Ok(())
    }

    /// Resolve session with lazy registration fallback
    ///
    /// Attempts to resolve the session mapping. If resolution fails, performs
    /// lazy registration with placeholder metadata to ensure messages can be written.
    async fn resolve_or_register_session(
        &self,
        connector_uid: Uuid,
        native_session_id: &str,
        reason: &str,
    ) -> Result<Uuid> {
        match self
            .archivist
            .resolve_session(connector_uid, native_session_id, None)
            .await
        {
            Ok(scroll_id) => Ok(scroll_id),
            Err(e) => {
                tracing::warn!(
                    "Failed to resolve session {} for connector {}: {}. Attempting lazy registration.",
                    native_session_id,
                    connector_uid,
                    e
                );

                // Attempt lazy registration using available metadata
                let register_req = RegisterSessionRequest {
                    connector_uid,
                    native_session_id: native_session_id.to_string(),
                    title: Some("Lazy-registered session".to_string()),
                    custom_scroll_id: None,
                    metadata: serde_json::json!({
                        "lazy_registered": true,
                        "reason": reason
                    }),
                    completeness: SessionCompleteness::Discovered,
                    parent_scroll_id: None,
                    is_subagent: false,
                    continuation: None,
                    agent_id: None,
                    subagent_type: None,
                    spawning_tool_use_id: None,
                };

                match self.archivist.register_session(register_req, None).await {
                    Ok(response) => {
                        tracing::info!(
                            "Lazy registration successful for session {} (scroll_id: {})",
                            native_session_id,
                            response.scroll_id
                        );
                        Ok(response.scroll_id)
                    }
                    Err(reg_error) => {
                        tracing::error!(
                            "Lazy registration failed for session {}: {}. Message will be lost.",
                            native_session_id,
                            reg_error
                        );
                        Err(reg_error)
                    }
                }
            }
        }
    }

    /// Create or get an existing meta session for an ACP client.
    ///
    /// If a meta session already exists for this client_id, returns its scroll_id.
    /// Otherwise creates a new meta session with SessionKind::AcpConnection.
    async fn get_or_create_meta_session(
        &self,
        client_id: &str,
        connector_uid: Uuid,
        connected_at: &str,
    ) -> Result<Uuid> {
        // First, try to find existing meta session
        if let Some(existing) = self.archivist.find_meta_session_by_client(client_id, None).await? {
            tracing::debug!(
                "Found existing meta session {} for client {}",
                existing.scroll_id,
                client_id
            );
            return Ok(existing.scroll_id);
        }

        // No existing session, create new one
        let req = RegisterSessionRequest {
            connector_uid,
            native_session_id: format!("acp-meta-{}", client_id),
            title: Some(format!("ACP Connection: {}", client_id)),
            custom_scroll_id: None,
            completeness: SessionCompleteness::Complete,
            metadata: serde_json::json!({
                "kind": "ACP_CONNECTION",
                "acp_client_id": client_id,
                "is_connected": true,
                "connected_at": connected_at,
            }),
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        let response = self.archivist.register_session(req, None).await?;

        tracing::info!(
            "Created meta session {} for ACP client {}",
            response.scroll_id,
            client_id
        );

        Ok(response.scroll_id)
    }

    /// Finalize stale message buffers that have been inactive for too long
    ///
    /// This prevents data loss when MessageCompleted events are missed or delayed.
    /// Buffers inactive for longer than the threshold are finalized and written to archive.
    async fn finalize_stale_buffers(&self, threshold: Duration) -> Result<()> {
        let now = Utc::now();

        // Collect stale message IDs
        let stale_ids = {
            let acc = self.accumulator.lock().await;
            acc.get_stale_message_ids(now, threshold)
        };

        if stale_ids.is_empty() {
            return Ok(());
        }

        tracing::warn!(
            "Finalizing {} stale message buffer(s) (inactive > {:?})",
            stale_ids.len(),
            threshold
        );

        // Finalize each stale buffer
        for message_id in stale_ids {
            // Finalize buffer
            let finalized_record = {
                let mut acc = self.accumulator.lock().await;
                acc.finalize(&message_id)
            };

            if let Some((mut record, connector_id, native_session_id)) = finalized_record {
                // Check if already archived
                let already_archived = {
                    let mut archived = self.archived_messages.lock().await;
                    if archived.contains(&message_id) {
                        true
                    } else {
                        archived.insert(message_id.clone());
                        false
                    }
                };

                if already_archived {
                    tracing::debug!(
                        "Stale message {} already archived, skipping",
                        message_id
                    );
                    continue;
                }

                // Tag with finalization method
                if let Some(metadata_obj) = record.metadata.as_object_mut() {
                    metadata_obj.insert(
                        "finalized_via".to_string(),
                        serde_json::Value::String("stale_timeout".to_string()),
                    );
                }

                // Canonicalize identity and write to archive
                if let Err(e) = self.canonicalize_and_write(
                    &mut record,
                    &connector_id,
                    &native_session_id,
                    &message_id,
                    "stale_timeout",
                ).await {
                    tracing::error!(
                        "Failed to write stale message {} to archive: {}",
                        message_id,
                        e
                    );
                    // Continue processing other stale buffers
                }
            }
        }

        Ok(())
    }

    /// Gracefully shut down the event handler, flushing all active buffers
    ///
    /// This ensures no messages are lost when the server is shut down.
    /// All active message buffers are finalized and written to archive.
    pub async fn shutdown_and_flush(&self) -> Result<()> {
        tracing::info!("Shutting down EventHandler, flushing active buffers");

        // Get all active buffer message IDs
        let message_ids = {
            let acc = self.accumulator.lock().await;
            acc.get_all_message_ids()
        };

        if message_ids.is_empty() {
            tracing::info!("No active buffers to flush on shutdown");
            return Ok(());
        }

        tracing::warn!(
            "Flushing {} active message buffer(s) on shutdown",
            message_ids.len()
        );

        let mut flushed_count = 0;
        let mut already_archived_count = 0;

        // Finalize each buffer
        for message_id in message_ids {
            // Finalize buffer
            let finalized_record = {
                let mut acc = self.accumulator.lock().await;
                acc.finalize(&message_id)
            };

            if let Some((mut record, connector_id, native_session_id)) = finalized_record {
                // Check if already archived (deduplication)
                let already_archived = {
                    let mut archived = self.archived_messages.lock().await;
                    if archived.contains(&message_id) {
                        true
                    } else {
                        archived.insert(message_id.clone());
                        false
                    }
                };

                if already_archived {
                    tracing::debug!("Shutdown: Message {} already archived, skipping", message_id);
                    already_archived_count += 1;
                    continue;
                }

                // Tag with finalization method
                if let Some(metadata_obj) = record.metadata.as_object_mut() {
                    metadata_obj.insert(
                        "finalized_via".to_string(),
                        serde_json::Value::String("shutdown_flush".to_string()),
                    );
                }

                // Canonicalize identity and write to archive
                if let Err(e) = self.canonicalize_and_write(
                    &mut record,
                    &connector_id,
                    &native_session_id,
                    &message_id,
                    "shutdown_flush",
                ).await {
                    tracing::error!(
                        "Shutdown: Failed to write message {} to archive: {}",
                        message_id,
                        e
                    );
                    // Continue processing other buffers despite error
                } else {
                    flushed_count += 1;
                }
            }
        }

        tracing::info!(
            "Shutdown complete: {} messages flushed, {} already archived",
            flushed_count,
            already_archived_count
        );

        Ok(())
    }

    /// Run the event loop, consuming events from the bus receiver.
    ///
    /// Accepts a [`BusReceiver`] from `dirigent_core::SharingBus`. Each
    /// received `BusEvent` is unwrapped to its inner `Event` and dispatched
    /// to [`Self::handle_event`]. The `routing` metadata is currently
    /// ignored by the archivist dispatcher — it continues to operate on the
    /// pre-existing `Event` enum — but the scroll_id hints are already
    /// attached at publish time for future consumers.
    pub async fn run(&self, mut bus_rx: BusReceiver) {
        let mut interval = tokio::time::interval(Duration::from_secs(1));
        // Use a long stale threshold (300s / 5 minutes) to match the general tool execution
        // limit. Task agents and other slow tools can run for minutes. TurnComplete is the
        // authoritative finalization signal; this is just a safety net for missed events.
        let stale_threshold = Duration::from_secs(300);

        // Sample the lagged counter periodically so operators see drops in logs.
        let lagged = Arc::clone(&bus_rx.lagged);
        let mut last_lagged: u64 = 0;

        loop {
            tokio::select! {
                maybe_bus_event = bus_rx.rx.recv() => {
                    match maybe_bus_event {
                        Some(bus_event) => {
                            // Unwrap the Arc<Event>; clone the inner only when
                            // we're the last holder so the common path is a
                            // move rather than a deep clone.
                            let event = Arc::try_unwrap(bus_event.event)
                                .unwrap_or_else(|shared| (*shared).clone());
                            if let Err(e) = self.handle_event(event).await {
                                tracing::error!("Failed to archive event: {}", e);
                            }
                        }
                        None => {
                            tracing::info!("Bus event stream closed, stopping event handler");
                            break;
                        }
                    }
                }
                _ = interval.tick() => {
                    // Log any newly-dropped events since last tick.
                    let current = lagged.load(std::sync::atomic::Ordering::Relaxed);
                    if current > last_lagged {
                        tracing::warn!(
                            dropped = current - last_lagged,
                            total = current,
                            "archivist bus subscriber is lagging; events were dropped"
                        );
                        last_lagged = current;
                    }
                    if let Err(e) = self.finalize_stale_buffers(stale_threshold).await {
                        tracing::error!("Failed to finalize stale buffers: {}", e);
                    }
                }
            }
        }
    }

    /// Dispatch event to appropriate handler
    async fn handle_event(&self, event: Event) -> Result<()> {
        match event {
            Event::SessionCreated {
                connector_id,
                session,
            } => {
                // Always register sessions - we want to track them even if archiving is disabled
                // This allows enabling archiving later
                self.handle_session_created(connector_id, session).await?;
            }
            Event::SessionsListed {
                connector_id,
                sessions,
            } => {
                // Register all discovered sessions — archivist deduplicates via ALIASED status
                self.handle_sessions_listed(connector_id, sessions).await?;
            }
            Event::TurnComplete {
                connector_id,
                session_id,
                message_id,
                trigger,
            } => {
                // Check if session is excluded from archiving
                if self.is_session_excluded(&connector_id, &session_id).await {
                    tracing::debug!(
                        "Skipping TurnComplete for excluded session {} (connector: {})",
                        session_id,
                        connector_id
                    );
                    return Ok(());
                }
                tracing::debug!(
                    "TurnComplete received for message {} in session {} (trigger: {:?})",
                    message_id,
                    session_id,
                    trigger
                );
                self.handle_turn_complete(connector_id, session_id, message_id, trigger).await?;
            }
            Event::MessageCompleted {
                connector_id,
                message,
            } => {
                // MessageCompleted is now informational only - metadata is ready
                // Finalization happens on TurnComplete
                tracing::debug!(
                    "MessageCompleted received for message {} in session {} (connector: {}) - metadata ready, awaiting TurnComplete for finalization",
                    message.id,
                    message.session_id,
                    connector_id
                );
                // We could update message metadata in accumulator here if needed,
                // but for now we keep this as a no-op for backward compatibility
            }
            Event::SessionUpdate {
                connector_id,
                session_id,
                update,
            } => {
                // Skip re-archiving for sessions being replayed (already Complete)
                if self.replaying_sessions.lock().await.contains(&session_id) {
                    tracing::debug!(
                        "Suppressing SessionUpdate for replaying session {} (already archived)",
                        session_id
                    );
                    return Ok(());
                }
                // Check if session is excluded from archiving
                if self.is_session_excluded(&connector_id, &session_id).await {
                    tracing::debug!(
                        "Skipping SessionUpdate for excluded session {} (connector: {})",
                        session_id,
                        connector_id
                    );
                    return Ok(());
                }
                self.handle_session_update(connector_id, session_id, update)
                    .await?;
            }
            Event::SessionIdle { connector_id, session_id } => {
                // Clear replay state if this session was being replayed
                if self.replaying_sessions.lock().await.remove(&session_id) {
                    tracing::debug!(
                        "Session {} replay complete, resuming normal archiving",
                        session_id
                    );
                }
                self.handle_session_idle(connector_id, session_id).await?;
            }
            Event::SessionMetadataUpdated {
                connector_id,
                session_id,
                title,
                total_messages: _,
                model,
            } => {
                // Check if session is excluded from archiving
                if self.is_session_excluded(&connector_id, &session_id).await {
                    tracing::debug!(
                        "Skipping SessionMetadataUpdated for excluded session {} (connector: {})",
                        session_id,
                        connector_id
                    );
                    return Ok(());
                }
                self.handle_session_metadata_updated(connector_id, session_id, title, model)
                    .await?;
            }
            Event::AcpClientConnected {
                client_id,
                connected_at,
                capabilities,
                connector_uid,
            } => {
                self.handle_acp_client_connected(client_id, connected_at, capabilities, connector_uid).await?;
            }
            Event::AcpClientDisconnected {
                client_id,
                disconnected_at,
                reason,
            } => {
                self.handle_acp_client_disconnected(client_id, disconnected_at, reason).await?;
            }
            Event::AcpClientSessionOpened {
                client_id,
                gateway_session_id,
                client_session_id: _,
                timestamp: _,
            } => {
                self.handle_acp_session_opened(client_id, gateway_session_id).await?;
            }
            Event::AcpClientSessionRouted {
                client_id,
                from_session_id,
                to_session_id,
                connector_id,
                connector_title: _,
                connector_kind: _,
                model: _,
                agent_info: _,
                timestamp: _,
            } => {
                self.handle_acp_session_routed(client_id, from_session_id, to_session_id, connector_id).await?;
            }
            Event::SessionMetadataReceived {
                connector_id,
                session_id,
                models,
                modes,
                config_options,
            } => {
                // Check if session is excluded from archiving
                if self.is_session_excluded(&connector_id, &session_id).await {
                    tracing::debug!(
                        "Skipping SessionMetadataReceived for excluded session {} (connector: {})",
                        session_id,
                        connector_id
                    );
                    return Ok(());
                }
                self.handle_session_acp_metadata_received(connector_id, session_id, models, modes, config_options).await?;
            }
            Event::SessionClosed { connector_id, session_id } => {
                tracing::debug!("SessionClosed: session '{}' from connector '{}'", session_id, connector_id);
                // No action needed — session remains in archive, just not actively connected
            }
            _ => {
                // Ignore other event types for MVP
            }
        }
        Ok(())
    }

    /// Handle SessionCreated event
    /// Register a single session with the archivist, given an already-resolved connector_uid.
    ///
    /// Returns the registration response (Accepted/Aliased/Rejected).
    /// Callers that process batches should resolve connector_uid once and reuse it.
    async fn register_single_session(
        &self,
        connector_uid: Uuid,
        session: &Session,
        completeness: SessionCompleteness,
    ) -> Result<RegisterSessionResponse> {
        // Never use external session ID as scroll_id, even if it's a valid UUID.
        // The archivist MUST generate a fresh UUID7 for storage consistency.
        // External IDs (including UUID4s from connectors like claude-code-acp)
        // are stored in session metadata for reverse lookup via sessions.jsonl.
        let mut metadata = serde_json::json!({
            "project_path": session.metadata.project_path,
            "model": session.metadata.model,
            "project_id": session.metadata.project_id.map(|u| u.to_string()),
        });

        // Propagate tool_configuration from _meta.extra into session metadata.
        // This preserves the connector's tool configuration in the archived session
        // so it can be restored when the session is loaded later.
        if let Some(ref meta) = session.metadata._meta {
            if let Some(tool_config) = meta.extra.get("tool_configuration") {
                if let Some(obj) = metadata.as_object_mut() {
                    obj.insert("tool_configuration".to_string(), tool_config.clone());
                }
            }
        }

        let req = RegisterSessionRequest {
            connector_uid,
            native_session_id: session.id.clone(),
            title: Some(session.title.clone()),
            custom_scroll_id: None,
            metadata,
            completeness,
            parent_scroll_id: None,
            is_subagent: false,
            continuation: None,
            agent_id: None,
            subagent_type: None,
            spawning_tool_use_id: None,
        };

        self.archivist.register_session(req, None).await
    }

    async fn handle_session_created(&self, connector_id: String, session: Session) -> Result<()> {
        if session.id.is_empty() {
            tracing::warn!(
                "SessionCreated event has empty session ID — session will not be resumable"
            );
        }

        // Resolve connector_id to connector_uid using the archivist
        // This handles both UUID connector IDs and human-readable IDs like "gateway-1"
        let connector_uid = match self.archivist.resolve_connector_uid(&connector_id).await {
            Ok(uid) => uid,
            Err(e) => {
                tracing::warn!(
                    "Failed to resolve connector_uid for connector_id '{}': {}. \
                     This connector may not be registered with the archivist.",
                    connector_id,
                    e
                );
                return Ok(()); // Skip this event, don't fail the handler
            }
        };

        let response = self.register_single_session(connector_uid, &session, SessionCompleteness::Complete).await?;

        // If the session was already known (Aliased), check if it's a replay of a Complete session.
        if response.status == RegisterStatus::Aliased {
            // Check if the existing session is Complete — if so, this is a replay
            // and we should suppress re-archiving of replayed messages.
            if let Ok(meta) = self.archivist.get_session_metadata(response.scroll_id, None).await {
                if meta.completeness == SessionCompleteness::Complete {
                    self.replaying_sessions.lock().await.insert(session.id.clone());
                    tracing::debug!(
                        "Session {} marked as replaying (already Complete)",
                        session.id
                    );
                }
            }
            // Upgrade completeness to Complete (idempotent for already-Complete sessions)
            if let Err(e) = self.archivist.update_session_completeness(
                response.scroll_id,
                SessionCompleteness::Complete,
                None,
            ).await {
                tracing::debug!(
                    "Failed to upgrade completeness for session {}: {}",
                    session.id, e
                );
            }
        }

        tracing::info!(
            "Registered session {} for connector {}",
            session.id,
            connector_id
        );

        // Emit SessionRegistered to signal that this session is durably
        // registered and the frontend can refresh the session list with
        // confidence. Published onto the SharingBus.
        let registered_event = Event::SessionRegistered {
            connector_id: connector_id.clone(),
            session_id: session.id.clone(),
            scroll_id: response.scroll_id.to_string(),
        };
        if let Some(ref publisher) = self.bus_publisher {
            let bus_event = BusEvent::from_archivist_event(
                registered_event,
                &connector_id,
                &session.id,
                Some(response.scroll_id),
            );
            publisher(bus_event);
        }

        // T015: Check if this connector is in the excluded list
        // If so, automatically exclude the session from archiving
        if self.is_connector_excluded(&connector_id).await {
            self.exclude_session(&connector_id, &session.id).await;
            tracing::info!(
                "Session {} auto-excluded from archiving (connector '{}' is excluded by default)",
                session.id,
                connector_id
            );
        }

        Ok(())
    }

    /// Handle SessionsListed event — register all discovered sessions.
    ///
    /// Sessions already known return ALIASED (fast, from cache) with metadata refresh.
    /// Newly discovered sessions return ACCEPTED and get a fresh scroll_id.
    /// Errors on individual sessions are logged and skipped.
    async fn handle_sessions_listed(
        &self,
        connector_id: String,
        sessions: Vec<Session>,
    ) -> Result<()> {
        if sessions.is_empty() {
            return Ok(());
        }

        let connector_uid = match self.archivist.resolve_connector_uid(&connector_id).await {
            Ok(uid) => uid,
            Err(e) => {
                tracing::warn!(
                    "SessionsListed: Failed to resolve connector '{}': {}. Skipping {} sessions.",
                    connector_id, e, sessions.len()
                );
                return Ok(());
            }
        };

        let is_excluded = self.is_connector_excluded(&connector_id).await;
        let mut accepted = 0u32;
        let mut aliased = 0u32;
        let mut errors = 0u32;

        for session in &sessions {
            match self.register_single_session(connector_uid, session, SessionCompleteness::Discovered).await {
                Ok(response) => match response.status {
                    RegisterStatus::Accepted => {
                        accepted += 1;
                        tracing::info!(
                            "SessionsListed: New session '{}' registered (scroll: {}) for connector '{}'",
                            session.id, response.scroll_id, connector_id
                        );
                        // Emit SessionRegistered to signal durable registration
                        // (published onto the SharingBus).
                        let registered_event = Event::SessionRegistered {
                            connector_id: connector_id.clone(),
                            session_id: session.id.clone(),
                            scroll_id: response.scroll_id.to_string(),
                        };
                        if let Some(ref publisher) = self.bus_publisher {
                            let bus_event = BusEvent::from_archivist_event(
                                registered_event,
                                &connector_id,
                                &session.id,
                                Some(response.scroll_id),
                            );
                            publisher(bus_event);
                        }
                        if is_excluded {
                            self.exclude_session(&connector_id, &session.id).await;
                        }
                        // Refresh metadata — title/model from session/list may differ from
                        // what was stored at registration (e.g., session was created via
                        // SessionCreated with empty title, now session/list has the real title)
                        let title = if session.title.is_empty() { None } else { Some(session.title.clone()) };
                        let model = session.metadata.model.clone();
                        if title.is_some() || model.is_some() {
                            if let Err(e) = self.archivist.update_session_metadata(
                                response.scroll_id, title, model, None,
                            ).await {
                                tracing::debug!(
                                    "SessionsListed: metadata refresh after accept failed for '{}': {}",
                                    session.id, e
                                );
                            }
                        }
                    }
                    RegisterStatus::Aliased => {
                        aliased += 1;
                        // Refresh metadata — title/model may have changed upstream
                        let title = if session.title.is_empty() { None } else { Some(session.title.clone()) };
                        let model = session.metadata.model.clone();
                        if title.is_some() || model.is_some() {
                            if let Err(e) = self.archivist.update_session_metadata(
                                response.scroll_id, title, model, None,
                            ).await {
                                tracing::debug!(
                                    "SessionsListed: metadata refresh failed for '{}': {}",
                                    session.id, e
                                );
                            }
                        }
                    }
                    RegisterStatus::Rejected => {
                        errors += 1;
                        tracing::warn!(
                            "SessionsListed: Registration rejected for session '{}' (connector '{}')",
                            session.id, connector_id
                        );
                    }
                },
                Err(e) => {
                    errors += 1;
                    tracing::warn!(
                        "SessionsListed: Failed to register '{}': {}. Continuing.",
                        session.id, e
                    );
                }
            }
        }

        tracing::info!(
            "SessionsListed: {} sessions for '{}': {} new, {} known, {} errors",
            sessions.len(), connector_id, accepted, aliased, errors
        );

        Ok(())
    }

    /// Handle TurnComplete event - triggers finalization and archiving
    ///
    /// This is THE primary signal that all content for a turn has been received
    /// and the message is ready to be finalized and archived.
    async fn handle_turn_complete(
        &self,
        connector_id: String,
        session_id: String,
        message_id: String,
        trigger: TurnCompleteTrigger,
    ) -> Result<()> {
        tracing::info!(
            "TurnComplete: Finalizing message {} in session {} (connector: {}, trigger: {:?})",
            message_id,
            session_id,
            connector_id,
            trigger
        );

        // ALWAYS finalize to remove buffer, even if already archived
        let finalized_record = {
            let mut acc = self.accumulator.lock().await;
            acc.finalize(&message_id)
        };

        // Check if we have a buffer for this message
        let Some((mut record, _acc_connector_id, _acc_native_session_id)) = finalized_record else {
            tracing::warn!(
                "TurnComplete: No accumulated content for message {} in session {} - message may not have been streamed",
                message_id,
                session_id
            );
            // This could happen if:
            // 1. Message was sent without streaming (but MessageCompleted should have handled it)
            // 2. TurnComplete arrived before any chunks (timing issue)
            // 3. Message was already finalized by SessionIdle or stale timeout
            // For now, we log and skip - the message may already be archived
            return Ok(());
        };

        // Check if we've already archived this message AFTER finalization
        // This prevents duplicate writes while ensuring buffer cleanup
        {
            let mut archived = self.archived_messages.lock().await;
            if archived.contains(&message_id) {
                tracing::debug!(
                    "TurnComplete: Message {} already archived, buffer cleaned but not re-written",
                    message_id
                );
                return Ok(());
            }
            // Mark as archived before processing to prevent race conditions
            archived.insert(message_id.clone());
        }

        // Tag with TurnComplete-specific metadata
        if let Some(metadata_obj) = record.metadata.as_object_mut() {
            metadata_obj.insert(
                "finalized_via".to_string(),
                serde_json::Value::String("turn_complete".to_string()),
            );
            metadata_obj.insert(
                "turn_complete_trigger".to_string(),
                serde_json::to_value(&trigger).unwrap_or(serde_json::json!(null)),
            );
        }

        // Canonicalize identity and write to archive
        self.canonicalize_and_write(
            &mut record,
            &connector_id,
            &session_id,
            &message_id,
            "turn_complete",
        )
        .await?;

        Ok(())
    }

    /// Handle SessionIdle event
    /// Finalizes and writes any buffered messages for the session that haven't been completed yet.
    /// This provides a safety net for the race condition where MessageCompleted arrives before
    /// chunks have been accumulated, or when MessageCompleted is missed entirely.
    async fn handle_session_idle(&self, connector_id: String, session_id: String) -> Result<()> {
        tracing::debug!("SessionIdle received for session {} (connector: {})", session_id, connector_id);

        // Get all message IDs that have buffers for this session
        let message_ids_for_session = {
            let acc = self.accumulator.lock().await;
            acc.get_message_ids_for_session(&session_id)
        };

        if message_ids_for_session.is_empty() {
            tracing::debug!(
                "SessionIdle: No active buffers for session {}, nothing to finalize",
                session_id
            );
            return Ok(());
        }

        tracing::info!(
            "SessionIdle: Finalizing {} buffered message(s) for session {}",
            message_ids_for_session.len(),
            session_id
        );

        // Finalize each message and write to archive
        for message_id in message_ids_for_session {
            // ALWAYS finalize to remove buffer, even if already archived
            let finalized_record = {
                let mut acc = self.accumulator.lock().await;
                acc.finalize(&message_id)
            };

            // If no buffer existed, skip (already cleaned up)
            let Some((mut record, _acc_connector_id, _acc_native_session_id)) = finalized_record else {
                tracing::debug!("SessionIdle: No buffer for message {}", message_id);
                continue;
            };

            // Check if already archived to prevent duplicate writes
            let already_archived = {
                let archived = self.archived_messages.lock().await;
                archived.contains(&message_id)
            };

            if already_archived {
                tracing::debug!(
                    "SessionIdle: Message {} already archived, buffer cleaned but not re-written",
                    message_id
                );
                continue;
            }

            // Mark as archived
            {
                let mut archived = self.archived_messages.lock().await;
                archived.insert(message_id.clone());
            }

            // Tag with finalization method
            if let Some(metadata_obj) = record.metadata.as_object_mut() {
                metadata_obj.insert(
                    "finalized_via".to_string(),
                    serde_json::Value::String("session_idle".to_string()),
                );
            }

            // Canonicalize identity and write to archive
            self.canonicalize_and_write(
                &mut record,
                &connector_id,
                &session_id,
                &message_id,
                "session_idle",
            )
            .await?;
        }

        Ok(())
    }

    /// Write message to archive using connector_uid to resolve scroll_id
    ///
    /// This method is kept for potential future use with non-streaming messages,
    /// though currently unused since TurnComplete handles all finalization.
    #[allow(dead_code)]
    async fn write_message_to_archive(&self, connector_uid: Uuid, message: Message) -> Result<()> {
        // Convert dirigent_protocol::Message to markdown
        let mut content_md = String::new();

        for part in &message.content {
            match part {
                MessagePart::Text { text } => {
                    content_md.push_str(text);
                }
                MessagePart::Thinking { text } => {
                    content_md.push_str(&format!("<thinking>{}</thinking>", text));
                }
                MessagePart::Code { language, code } => {
                    content_md.push_str(&format!("\n```{}\n{}\n```\n", language, code));
                }
                MessagePart::Tool {
                    tool,
                    tool_call_id: _,
                    input,
                    output,
                } => {
                    content_md.push_str(&format!("\n**Tool**: {}\n", tool));
                    content_md.push_str(&format!(
                        "```json\n{}\n```\n",
                        serde_json::to_string_pretty(input).unwrap_or_else(|_| "{}".to_string())
                    ));
                    if let Some(output_val) = output {
                        content_md.push_str(&format!(
                            "\n**Output**:\n```json\n{}\n```\n",
                            serde_json::to_string_pretty(output_val)
                                .unwrap_or_else(|_| "{}".to_string())
                        ));
                    }
                }
                MessagePart::File { path, content } => {
                    content_md.push_str(&format!("\n**File**: {}\n```\n{}\n```\n", path, content));
                }
            }
        }

        // Parse message_id
        let message_uuid = Uuid::parse_str(&message.id).unwrap_or_else(|_| {
            tracing::warn!("Failed to parse message_id as UUID: {}", message.id);
            Uuid::now_v7()
        });

        // Resolve scroll_id from native session_id BEFORE creating MessageRecord
        // This ensures we use the canonical session identifier
        // If resolution fails, attempt lazy registration as a fallback
        let scroll_id = self
            .resolve_or_register_session(
                connector_uid,
                &message.session_id,
                "Session mapping missing during message write (non-streaming path)",
            )
            .await?;

        // Convert MessageRole to string
        let role = match message.role {
            MessageRole::User => "user".to_string(),
            MessageRole::Assistant => "assistant".to_string(),
        };

        // Serialize metadata
        let metadata = message
            .metadata
            .map(|m| serde_json::to_value(m).unwrap_or(serde_json::json!({})))
            .unwrap_or(serde_json::json!({}));

        // Serialize original content parts for rich UI rendering
        let content_parts = serde_json::to_value(&message.content).ok();

        let record = MessageRecord {
            version: 1,
            message_id: message_uuid,
            session: scroll_id, // Use canonical scroll_id instead of native session UUID
            parent_id: None,
            ts: message.created_at,
            role,
            author: None,
            content_md,
            content_parts,
            attachments: Vec::new(),
            metadata,
        };

        // Write message to archive
        self.archivist
            .append_messages(scroll_id, vec![record], None)
            .await?;

        tracing::info!(
            "Wrote message {} to archive for session {} (scroll_id: {})",
            message.id,
            message.session_id,
            scroll_id
        );

        Ok(())
    }

    /// Handle SessionUpdate event
    async fn handle_session_update(
        &self,
        connector_id: String,
        session_id: String,
        update: SessionUpdate,
    ) -> Result<()> {
        let mut acc = self.accumulator.lock().await;

        match update {
            SessionUpdate::AgentMessageChunk {
                message_id,
                content,
                ..
            } => {
                acc.add_chunk(message_id, session_id, connector_id, "assistant".to_string(), content);
            }
            SessionUpdate::UserMessageChunk {
                message_id,
                content,
                ..
            } => {
                acc.add_chunk(message_id, session_id, connector_id, "user".to_string(), content);
            }
            SessionUpdate::AgentThoughtChunk {
                message_id,
                content,
                ..
            } => {
                // Extract text from ContentBlock
                if let ContentBlock::Text { text } = content {
                    acc.add_thinking(message_id, session_id, connector_id, text);
                }
            }
            SessionUpdate::ToolCall {
                message_id,
                tool_call,
                ..
            } => {
                // Convert to internal ToolCallData and add/update
                // This handles both initial tool call and any subsequent updates
                let tool_call_data = crate::accumulator::ToolCallData {
                    id: tool_call.id.clone(),
                    tool_name: tool_call.tool_name.clone(),
                    input: tool_call.raw_input.unwrap_or_else(|| serde_json::json!({})),
                    output: tool_call.raw_output,
                };

                acc.add_or_update_tool_call(message_id, tool_call_data);
            }
            SessionUpdate::ToolCallUpdate {
                message_id,
                tool_call_id,
                tool_call,
                ..
            } => {
                // Convert to ToolCallData and merge with existing tool call
                let tool_call_data = crate::accumulator::ToolCallData {
                    id: tool_call_id,
                    tool_name: tool_call.tool_name.clone(),
                    input: tool_call.raw_input.unwrap_or_else(|| serde_json::json!({})),
                    output: tool_call.raw_output,
                };

                acc.add_or_update_tool_call(message_id, tool_call_data);
            }
            SessionUpdate::Unknown { .. } => {
                // Ignore unknown update types - forward compatibility
            }
        }

        Ok(())
    }

    /// Handle SessionMetadataUpdated event
    async fn handle_session_metadata_updated(
        &self,
        connector_id: String,
        session_id: String,
        title: Option<String>,
        model: Option<String>,
    ) -> Result<()> {
        tracing::debug!(
            "SessionMetadataUpdated event: connector={}, session={}, title={:?}, model={:?}",
            connector_id,
            session_id,
            title,
            model
        );

        // Resolve connector_id to connector_uid using the archivist
        let connector_uid = match self.archivist.resolve_connector_uid(&connector_id).await {
            Ok(uid) => uid,
            Err(e) => {
                tracing::warn!(
                    "Failed to resolve connector_uid for connector_id '{}': {}. \
                     Skipping metadata update.",
                    connector_id,
                    e
                );
                return Ok(()); // Skip this event, don't fail the handler
            }
        };

        // Resolve session to scroll_id
        let scroll_id = match self
            .archivist
            .resolve_session(connector_uid, &session_id, None)
            .await
        {
            Ok(scroll_id) => scroll_id,
            Err(e) => {
                tracing::warn!(
                    "Failed to resolve session {} for connector {}: {}. Skipping metadata update.",
                    session_id,
                    connector_uid,
                    e
                );
                return Ok(()); // Skip this event if session not found
            }
        };

        // Update session metadata in archive
        self.archivist
            .update_session_metadata(scroll_id, title.clone(), model.clone(), None)
            .await?;

        tracing::info!(
            "Updated session metadata: scroll_id={}, title={:?}, model={:?}",
            scroll_id,
            title,
            model
        );

        Ok(())
    }

    /// Handle SessionMetadataReceived event (ACP-specific models/modes/config_options metadata)
    async fn handle_session_acp_metadata_received(
        &self,
        connector_id: String,
        session_id: String,
        models: Option<dirigent_protocol::SessionModelState>,
        modes: Option<dirigent_protocol::SessionModeState>,
        config_options: Option<Vec<dirigent_protocol::ConfigOption>>,
    ) -> Result<()> {
        tracing::debug!(
            "SessionMetadataReceived event: connector={}, session={}, has_models={}, has_modes={}, has_config_options={}",
            connector_id,
            session_id,
            models.is_some(),
            modes.is_some(),
            config_options.is_some()
        );

        // Resolve connector_id to connector_uid using the archivist
        let connector_uid = match self.archivist.resolve_connector_uid(&connector_id).await {
            Ok(uid) => uid,
            Err(e) => {
                tracing::warn!(
                    "Failed to resolve connector_uid for connector_id '{}': {}. \
                     Skipping ACP metadata update.",
                    connector_id,
                    e
                );
                return Ok(()); // Skip this event, don't fail the handler
            }
        };

        // Resolve session to scroll_id
        let scroll_id = match self
            .archivist
            .resolve_session(connector_uid, &session_id, None)
            .await
        {
            Ok(scroll_id) => scroll_id,
            Err(e) => {
                tracing::warn!(
                    "Failed to resolve session {} for connector {}: {}. Skipping ACP metadata update.",
                    session_id,
                    connector_uid,
                    e
                );
                return Ok(()); // Skip this event if session not found
            }
        };

        // Convert protocol types to JSON for storage
        let models_json = models.and_then(|m| serde_json::to_value(m).ok());
        let modes_json = modes.and_then(|m| serde_json::to_value(m).ok());
        let config_options_json = config_options.and_then(|co| serde_json::to_value(co).ok());

        // Update session ACP metadata in archive
        self.archivist
            .update_session_acp_metadata(scroll_id, models_json.clone(), modes_json.clone(), config_options_json.clone(), None)
            .await?;

        tracing::info!(
            "Updated session ACP metadata: scroll_id={}, has_models={}, has_modes={}, has_config_options={}",
            scroll_id,
            models_json.is_some(),
            modes_json.is_some(),
            config_options_json.is_some()
        );

        Ok(())
    }

    /// Handle AcpClientConnected event
    async fn handle_acp_client_connected(
        &self,
        client_id: String,
        connected_at: String,
        _capabilities: Option<serde_json::Value>,
        connector_uid_str: String,
    ) -> Result<()> {
        tracing::info!("ACP client connected: {}", client_id);

        // Parse the connector_uid from the event
        // This is the Acceptor connector's UID, used to create meta sessions under the right connector
        let connector_uid = match Uuid::parse_str(&connector_uid_str) {
            Ok(uid) if uid != Uuid::nil() => uid,
            Ok(_) | Err(_) => {
                // If parsing fails or we get a nil UUID, log a warning and skip meta session creation
                tracing::warn!(
                    "Invalid connector_uid '{}' for ACP client {}. Meta session will not be created.",
                    connector_uid_str,
                    client_id
                );
                return Ok(());
            }
        };

        // Create or get meta session
        let scroll_id = self.get_or_create_meta_session(&client_id, connector_uid, &connected_at).await?;

        // Append ClientConnected event
        let event = MetaEventRecord {
            version: 1,
            event_id: Uuid::now_v7(),
            session: scroll_id,
            ts: chrono::DateTime::parse_from_rfc3339(&connected_at)
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or_else(|_| Utc::now()),
            event_type: MetaEventType::ClientConnected,
            description: format!("Client {} connected", client_id),
            linked_session_id: None,
            linked_connector_id: None,
            linked_connector_title: None,
            metadata: serde_json::json!({}),
        };

        self.archivist.append_meta_events(scroll_id, vec![event], None).await?;

        // Update connection status
        self.archivist.update_meta_session_status(scroll_id, true, None, None).await?;

        Ok(())
    }

    /// Handle AcpClientDisconnected event
    async fn handle_acp_client_disconnected(
        &self,
        client_id: String,
        disconnected_at: String,
        reason: Option<String>,
    ) -> Result<()> {
        tracing::info!("ACP client disconnected: {} (reason: {:?})", client_id, reason);

        // Find the meta session for this client
        let Some(meta_session) = self.archivist.find_meta_session_by_client(&client_id, None).await? else {
            tracing::warn!("No meta session found for disconnecting client {}", client_id);
            return Ok(());
        };

        let scroll_id = meta_session.scroll_id;

        // Append ClientDisconnected event
        let event = MetaEventRecord {
            version: 1,
            event_id: Uuid::now_v7(),
            session: scroll_id,
            ts: chrono::DateTime::parse_from_rfc3339(&disconnected_at)
                .map(|dt| dt.with_timezone(&Utc))
                .unwrap_or_else(|_| Utc::now()),
            event_type: MetaEventType::ClientDisconnected,
            description: format!(
                "Client {} disconnected{}",
                client_id,
                reason.as_ref().map(|r| format!(" ({})", r)).unwrap_or_default()
            ),
            linked_session_id: None,
            linked_connector_id: None,
            linked_connector_title: None,
            metadata: serde_json::json!({
                "reason": reason
            }),
        };

        self.archivist.append_meta_events(scroll_id, vec![event], None).await?;

        // Update connection status to disconnected
        self.archivist.update_meta_session_status(scroll_id, false, None, None).await?;

        Ok(())
    }

    /// Handle AcpClientSessionOpened event
    async fn handle_acp_session_opened(
        &self,
        client_id: String,
        gateway_session_id: String,
    ) -> Result<()> {
        tracing::info!("ACP client {} opened session {}", client_id, gateway_session_id);

        // Find the meta session for this client
        let Some(meta_session) = self.archivist.find_meta_session_by_client(&client_id, None).await? else {
            tracing::warn!("No meta session found for client {} opening session", client_id);
            return Ok(());
        };

        let scroll_id = meta_session.scroll_id;

        // Try to parse gateway_session_id as UUID for linking
        let linked_session_id = Uuid::parse_str(&gateway_session_id).ok();

        // Append SessionOpened event
        let event = MetaEventRecord {
            version: 1,
            event_id: Uuid::now_v7(),
            session: scroll_id,
            ts: Utc::now(),
            event_type: MetaEventType::SessionOpened,
            description: format!("Opened session {}", gateway_session_id),
            linked_session_id,
            linked_connector_id: None,
            linked_connector_title: None,
            metadata: serde_json::json!({
                "gateway_session_id": gateway_session_id
            }),
        };

        self.archivist.append_meta_events(scroll_id, vec![event], None).await?;

        // Update current session
        self.archivist.update_meta_session_status(scroll_id, true, linked_session_id, None).await?;

        Ok(())
    }

    /// Handle AcpClientSessionRouted event
    async fn handle_acp_session_routed(
        &self,
        client_id: String,
        from_session_id: String,
        to_session_id: String,
        to_connector_id: String,
    ) -> Result<()> {
        tracing::info!(
            "ACP client {} session routed: {} -> {} (connector: {})",
            client_id, from_session_id, to_session_id, to_connector_id
        );

        // Find the meta session for this client
        let Some(meta_session) = self.archivist.find_meta_session_by_client(&client_id, None).await? else {
            tracing::warn!("No meta session found for client {} during routing", client_id);
            return Ok(());
        };

        let scroll_id = meta_session.scroll_id;

        // Try to parse to_session_id as UUID for linking
        let linked_session_id = Uuid::parse_str(&to_session_id).ok();

        // Append SessionSwitched event
        let event = MetaEventRecord {
            version: 1,
            event_id: Uuid::now_v7(),
            session: scroll_id,
            ts: Utc::now(),
            event_type: MetaEventType::SessionSwitched,
            description: format!("Switched to {} via {}", to_session_id, to_connector_id),
            linked_session_id,
            linked_connector_id: Some(to_connector_id.clone()),
            linked_connector_title: None, // Could be resolved later
            metadata: serde_json::json!({
                "from_session_id": from_session_id,
                "to_session_id": to_session_id,
                "to_connector_id": to_connector_id
            }),
        };

        self.archivist.append_meta_events(scroll_id, vec![event], None).await?;

        // Update current session
        self.archivist.update_meta_session_status(scroll_id, true, linked_session_id, None).await?;

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;
    use dirigent_protocol::MessageStatus;

    /// Create a real `Archivist` backed by a unique temp directory.
    /// Previously these tests used a hand-rolled `MockArchivist`, but the
    /// trait-object interface was removed in Phase 2 and EventHandler now
    /// requires a concrete coordinator.
    ///
    /// Uses `from_single_backend` (not `new_with_single_archive`) so each test
    /// is fully isolated — otherwise tests running in parallel race on the
    /// shared `.archives.json` that `new_with_single_archive` writes into the
    /// tempdir's parent.
    async fn mk_test_archivist() -> Arc<Archivist> {
        let tmp = std::env::temp_dir()
            .join(format!("events_test_{}", Uuid::now_v7()));
        let backend = Arc::new(
            crate::backends::JsonlBackend::new(tmp)
                .await
                .expect("create test backend"),
        );
        Arc::new(
            Archivist::from_single_backend("main".into(), backend)
                .await
                .expect("create test archivist"),
        )
    }


    #[tokio::test]
    async fn test_event_handler_creation() {
        let archivist = mk_test_archivist().await;
        let _handler = EventHandler::new(archivist);
    }

    #[tokio::test]
    async fn test_handle_session_update_agent_chunk() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let update = SessionUpdate::AgentMessageChunk {
            message_id: "msg_123".to_string(),
            content: ContentBlock::Text {
                text: "Hello".to_string(),
            },
            _meta: None,
        };

        let result = handler
            .handle_session_update(
                "connector_123".to_string(),
                "session_456".to_string(),
                update,
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_handle_session_update_thinking_chunk() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let update = SessionUpdate::AgentThoughtChunk {
            message_id: "msg_789".to_string(),
            content: ContentBlock::Text {
                text: "Thinking...".to_string(),
            },
            _meta: None,
        };

        let result = handler
            .handle_session_update(
                "connector_456".to_string(),
                "session_abc".to_string(),
                update,
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_write_message_to_archive() {
        use uuid::Uuid;

        let archivist = mk_test_archivist().await;

        // Pre-register the connector so lazy session registration can attach to it.
        // (The old MockArchivist answered every call unconditionally; the real
        // coordinator requires the connector exist first.)
        let connector_uid = Uuid::now_v7();
        archivist
            .register_connector(
                crate::types::RegisterConnectorRequest {
                    r#type: "Test".into(),
                    title: "t".into(),
                    client_native_id: format!("t@{}", Uuid::now_v7()),
                    custom_uid: Some(connector_uid),
                    metadata: serde_json::json!({}),
                    fingerprint: None,
                },
                None,
            )
            .await
            .unwrap();

        let handler = EventHandler::new(archivist);

        let message = Message {
            id: "01936e8f-e5a7-7000-8000-000000000001".to_string(),
            session_id: "01936e8f-e5a7-7000-8000-000000000002".to_string(),
            role: MessageRole::User,
            created_at: Utc::now(),
            content: vec![MessagePart::Text {
                text: "Hello, world!".to_string(),
            }],
            status: MessageStatus::Completed,
            metadata: None,
        };

        let result = handler
            .write_message_to_archive(connector_uid, message)
            .await;
        assert!(result.is_ok(), "write_message_to_archive failed: {:?}", result);
    }

    #[tokio::test]
    async fn test_user_message_turn_complete_event() {
        use uuid::Uuid;

        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let connector_id = Uuid::now_v7().to_string();
        let session_id = "01936e8f-e5a7-7000-8000-000000000102".to_string();
        let message_id = "01936e8f-e5a7-7000-8000-000000000101".to_string();

        // First, add some chunks to accumulator (simulating streaming)
        let update = SessionUpdate::UserMessageChunk {
            message_id: message_id.clone(),
            content: ContentBlock::Text {
                text: "What is the capital of France?".to_string(),
            },
            _meta: None,
        };

        let _ = handler
            .handle_session_update(
                connector_id.clone(),
                session_id.clone(),
                update,
            )
            .await;

        // Now handle TurnComplete event for user message
        let result = handler
            .handle_turn_complete(
                connector_id,
                session_id,
                message_id,
                TurnCompleteTrigger::ExplicitSignal,
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_assistant_message_turn_complete_event() {
        use uuid::Uuid;

        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let connector_id = Uuid::now_v7().to_string();
        let session_id = "01936e8f-e5a7-7000-8000-000000000202".to_string();
        let message_id = "01936e8f-e5a7-7000-8000-000000000201".to_string();

        // First, add some chunks to accumulator (simulating streaming)
        let update = SessionUpdate::AgentMessageChunk {
            message_id: message_id.clone(),
            content: ContentBlock::Text {
                text: "The capital of France is Paris.".to_string(),
            },
            _meta: None,
        };

        let _ = handler
            .handle_session_update(
                connector_id.clone(),
                session_id.clone(),
                update,
            )
            .await;

        // Now handle TurnComplete event for assistant message
        let result = handler
            .handle_turn_complete(
                connector_id,
                session_id,
                message_id,
                TurnCompleteTrigger::ExplicitSignal,
            )
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_session_idle_finalizes_buffered_messages() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let session_id = "01936e8f-e5a7-7000-8000-000000000010".to_string();
        let message_id = "01936e8f-e5a7-7000-8000-000000000011".to_string();

        // Add chunks to accumulator (simulating streaming)
        let update = SessionUpdate::AgentMessageChunk {
            message_id: message_id.clone(),
            content: ContentBlock::Text {
                text: "Hello from stream".to_string(),
            },
            _meta: None,
        };

        let result = handler
            .handle_session_update(
                "connector_123".to_string(),
                session_id.clone(),
                update,
            )
            .await;
        assert!(result.is_ok());

        // Verify buffer exists
        let has_buffer = {
            let acc = handler.accumulator.lock().await;
            !acc.get_message_ids_for_session(&session_id).is_empty()
        };
        assert!(has_buffer, "Buffer should exist before SessionIdle");

        // Handle SessionIdle
        let result = handler.handle_session_idle("connector_123".to_string(), session_id.clone()).await;
        assert!(result.is_ok());

        // Verify buffer was finalized and cleared
        let has_buffer_after = {
            let acc = handler.accumulator.lock().await;
            !acc.get_message_ids_for_session(&session_id).is_empty()
        };
        assert!(!has_buffer_after, "Buffer should be cleared after SessionIdle");

        // Verify message was marked as archived
        let is_archived = {
            let archived = handler.archived_messages.lock().await;
            archived.contains(&message_id)
        };
        assert!(is_archived, "Message should be marked as archived");
    }

    #[tokio::test]
    async fn test_session_idle_with_no_buffers() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let session_id = "01936e8f-e5a7-7000-8000-000000000020".to_string();

        // Handle SessionIdle with no buffers (should be no-op)
        let result = handler.handle_session_idle("test-connector".to_string(), session_id).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_session_idle_skips_already_archived() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let session_id = "01936e8f-e5a7-7000-8000-000000000030".to_string();
        let message_id = "01936e8f-e5a7-7000-8000-000000000031".to_string();

        // Add chunks to accumulator
        let update = SessionUpdate::AgentMessageChunk {
            message_id: message_id.clone(),
            content: ContentBlock::Text {
                text: "Test message".to_string(),
            },
            _meta: None,
        };

        handler
            .handle_session_update(
                "connector_456".to_string(),
                session_id.clone(),
                update,
            )
            .await
            .unwrap();

        // Verify buffer exists before marking as archived
        let has_buffer_before = {
            let acc = handler.accumulator.lock().await;
            !acc.get_message_ids_for_session(&session_id).is_empty()
        };
        assert!(has_buffer_before, "Buffer should exist before marking as archived");

        // Mark message as already archived
        {
            let mut archived = handler.archived_messages.lock().await;
            archived.insert(message_id.clone());
        }

        // Handle SessionIdle - should clean buffer but skip writing already archived message
        let result = handler.handle_session_idle("connector_456".to_string(), session_id.clone()).await;
        assert!(result.is_ok());

        // Buffer SHOULD be cleared even though message was already archived
        // This is the fix: finalize() is called before checking archived_messages
        let has_buffer = {
            let acc = handler.accumulator.lock().await;
            !acc.get_message_ids_for_session(&session_id).is_empty()
        };
        assert!(!has_buffer, "Buffer should be cleared even when message is already archived (prevents leak)");
    }

    fn make_test_session(id: &str, title: &str) -> Session {
        Session {
            id: id.to_string(),
            title: title.to_string(),
            created_at: Utc::now(),
            updated_at: Utc::now(),
            metadata: dirigent_protocol::SessionMetadata {
                project_path: "/test/project".to_string(),
                model: Some("claude-4".to_string()),
                total_messages: 0,
                system_message: None,
                current_mode_id: None,
                _meta: None,
                project_id: None,
            },
            cwd: None,
            models: None,
            modes: None,
            config_options: None,
            acp_client_id: None,
        }
    }

    #[tokio::test]
    async fn test_handle_sessions_listed_registers_sessions() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let connector_id = Uuid::now_v7().to_string();
        let sessions = vec![
            make_test_session("session-1", "First Session"),
            make_test_session("session-2", "Second Session"),
        ];

        let result = handler
            .handle_sessions_listed(connector_id, sessions)
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_handle_sessions_listed_empty_list() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let result = handler
            .handle_sessions_listed("some-connector".to_string(), vec![])
            .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_handle_event_sessions_listed() {
        let archivist = mk_test_archivist().await;
        let handler = EventHandler::new(archivist);

        let connector_id = Uuid::now_v7().to_string();
        let event = Event::SessionsListed {
            connector_id,
            sessions: vec![
                make_test_session("sess-a", "Session A"),
                make_test_session("sess-b", "Session B"),
                make_test_session("sess-c", "Session C"),
            ],
        };

        let result = handler.handle_event(event).await;
        assert!(result.is_ok());
    }
}
