use futures::stream::StreamExt;
use std::sync::Arc;

use crate::server::SouveraineServer;

use crate::backend::{Backend, BackendEvent};

use super::LocalBackend;

#[async_trait::async_trait]
impl crate::core::nervous::handler::TurnInjector for LocalBackend {
    /// Heartbeat-driven turn injection. The cron loop pauses while
    /// `active_sessions > 0`, so by the time we get here the agent is
    /// idle. We grab the most recent conversation (or create a fresh one
    /// if the agent has none), append the scheduled prompt as a user
    /// message, and drain the resulting stream — the turn runs silently
    /// in the background. Anything subconscious surfaces lands in the inbox.
    async fn inject_background_turn(
        &self,
        agent_id: &str,
        text: &str,
    ) -> anyhow::Result<()> {
        let conv_id = match self.server.sessions.list_for_agent(agent_id).last().cloned() {
            Some(id) => id,
            None => self.ensure_conversation(agent_id).await?,
        };
        let stream = self.send(&conv_id, text).await?;
        // Drain the stream in the background — no UI is listening. But the
        // subconscious's N+1 pass runs inside this turn, and what she
        // surfaces (a commitment, a reflection, an archivist synthesis)
        // would otherwise vanish with the drained events. Collect those and
        // stash them so the next TUI/CLI session shows the human what
        // happened during the autonomous cycle.
        let server = self.server.clone();
        let agent_id = agent_id.to_string();
        tokio::spawn(async move {
            use crate::core::nervous::pending::PendingSurfacing;
            let mut s = stream;
            let mut stashed: Vec<PendingSurfacing> = Vec::new();
            while let Some(ev) = s.next().await {
                match ev {
                    Ok(BackendEvent::Surfacing { source, content, priority }) => {
                        // Skip the no-op heartbeat sentinel — the subconscious
                        // always queues a low "pass complete, no anomalies"
                        // item so the UI shows the pass ran. That is noise to
                        // resurface on connect; only stash real observations.
                        if priority.eq_ignore_ascii_case("low")
                            && content.contains("no anomalies detected")
                        {
                            continue;
                        }
                        stashed.push(PendingSurfacing {
                            kind: "surfacing".to_string(),
                            source,
                            content,
                            priority,
                            at: chrono::Utc::now(),
                        });
                    }
                    Ok(BackendEvent::Reflection(content)) => {
                        stashed.push(PendingSurfacing {
                            kind: "reflection".to_string(),
                            source: String::new(),
                            content,
                            priority: String::new(),
                            at: chrono::Utc::now(),
                        });
                    }
                    Ok(BackendEvent::Archivist { synthesis, .. }) => {
                        stashed.push(PendingSurfacing {
                            kind: "archivist".to_string(),
                            source: String::new(),
                            content: synthesis,
                            priority: String::new(),
                            at: chrono::Utc::now(),
                        });
                    }
                    _ => {}
                }
            }
            if !stashed.is_empty() {
                let dir = server.agents.agent_data_dir(&agent_id);
                match crate::core::nervous::pending::append(&dir, &stashed).await {
                    Ok(()) => tracing::info!(
                        agent = %agent_id,
                        count = stashed.len(),
                        "stashed heartbeat surfacings for next session"
                    ),
                    Err(e) => tracing::warn!(
                        "pending heartbeat surfacings stash failed: {}", e
                    ),
                }
            }
        });
        Ok(())
    }

    /// Surface-initiated turn injection. Called by the
    /// [`SensoriumInputHandler`] when a `sensorium:input` event arrives.
    ///
    /// Unlike background turns, the turn's output events are NOT drained
    /// here — `run_turn` already fires them onto the EventBus as `turn:*`
    /// events (via `TurnEventDispatcher`). The originating sensorium's
    /// `run` loop consumes those events for incremental rendering.
    ///
    /// We drain the stream only to prevent backpressure on the mpsc
    /// channel. The EventBus is the public event system; the stream is
    /// a TUI-internal detail.
    async fn inject_surface_turn(
        &self,
        agent_id: &str,
        conversation_id: &str,
        text: &str,
    ) -> anyhow::Result<()> {
        // Resolve the surface chat ID to a Souveraine conversation ID.
        // Matrix room IDs are not Souveraine conversation IDs — the
        // mapping survives for the lifetime of the surface session so
        // subsequent messages in the same room route to the same
        // conversation. The lock scope is carefully bounded to avoid
        // holding a !Send MutexGuard across the .await below.
        let conv_id = {
            let map = self.server.surface_conversations.lock().unwrap();
            map.get(conversation_id).cloned()
        };
        let conv_id = match conv_id {
            Some(id) => id,
            None => {
                let id = self.ensure_conversation(agent_id).await?;
                self.server
                    .surface_conversations
                    .lock()
                    .unwrap()
                    .insert(conversation_id.to_string(), id.clone());
                id
            }
        };

        let stream = self.send(&conv_id, text).await?;
        // Drain the stream in the background — the EventBus already carries
        // every `turn:*` event via TurnEventDispatcher. The sensorium
        // renders from the bus. We drain here so the mpsc channel doesn't
        // back up.
        tokio::spawn(async move {
            let mut s = stream;
            while let Some(ev) = s.next().await {
                if ev.is_err() {
                    break;
                }
            }
        });
        Ok(())
    }
}

/// Drain subconscious's intrusive box for the given agent and return formatted
/// `[ surfacing: ... ]` lines ready to prepend to the user's next message.
/// Marks each drained item as delivered (moved to `sent.md`). Mirrors
/// the `readSurfacingThoughts` + `clearSurfacingThoughts` pair — the substrate
/// reads the channel subconscious wrote to and lets the conscious mind see it
/// before she reads the user.
///
/// Critical urgency gets `[ surfacing — CRITICAL: ... ]`. High becomes
/// `[ surfacing — !: ... ]`. Low/none keep the bare form. The shape is a
/// gradient the agent feels, not a number she has to read.
pub(super) async fn drain_intrusive_surfacings(
    server: &Arc<SouveraineServer>,
    agent_id: &str,
) -> Vec<String> {
    use crate::core::subconscious::{SubconsciousInbox, Urgency};

    let sub_repo = server.agents.subconscious_memory_repo(agent_id);
    let primary_repo = server.agents.memory_repo(agent_id);
    let inbox = SubconsciousInbox::with_primary(sub_repo, primary_repo);

    let items = match inbox.get_intrusive().await {
        Ok(items) => items,
        Err(e) => {
            tracing::debug!("intrusive surfacing read failed (continuing without): {}", e);
            return Vec::new();
        }
    };

    if items.is_empty() {
        return Vec::new();
    }

    let mut lines = Vec::with_capacity(items.len());
    for item in &items {
        let prefix = match item.urgency {
            Urgency::Critical => "[ surfacing — CRITICAL:",
            Urgency::High => "[ surfacing — !:",
            Urgency::Low => "[ surfacing:",
        };
        let content = item.content.trim();
        lines.push(format!("{} {} ]", prefix, content));

        if let Err(e) = inbox.mark_delivered(&item.id).await {
            tracing::debug!("mark_delivered failed for {}: {}", item.id, e);
        }
    }
    lines
}
