//! Always-on conversational [`Bot`] abstraction.
//!
//! A [`Bot`] is the keystone of Blazen's always-on feature: a persistent,
//! event-driven agent that sits idle until you [`send`](Bot::send) it a
//! message, runs an LLM + tool agentic turn, remembers the conversation
//! across turns, and streams its replies back via [`responses`](Bot::responses).
//!
//! Internally it is a thin, opinionated wrapper around a `.live()`
//! [`Workflow`](crate::workflow::Workflow) with a single step that:
//!
//! 1. Accepts an [`InboundMessage`].
//! 2. Loads the persisted [`ChatWindow`](blazen_llm::ChatWindow) conversation
//!    memory from the workflow [`Context`].
//! 3. Appends the user's message, optionally compacts the history via
//!    summarization, and runs
//!    [`run_agent_with_callback`](blazen_llm::run_agent_with_callback).
//! 4. Persists the updated window, folds the agent's usage/cost into the
//!    workflow's cost-budget accumulator, and writes a [`BotResponse`] to the
//!    event stream.
//! 5. Returns [`StepOutput::None`] to re-arm the loop for the next message.
//!
//! The lifecycle controls of the underlying live workflow are exposed
//! directly: [`idle_timeout`](BotBuilder::idle_timeout),
//! [`cost_budget_usd`](BotBuilder::cost_budget_usd),
//! [`shutdown`](Bot::shutdown), and [`snapshot`](Bot::snapshot).

use std::any::Any;
use std::sync::Arc;
use std::time::Duration;

use blazen_events::{AnyEvent, Event, StopEvent, UsageEvent, register_event_deserializer};
use blazen_llm::{
    AgentConfig, ChatMessage, ChatWindow, CurrentTimeTool, Model, SummarizationConfig, Tool,
    current_time_system_note, run_agent_with_callback,
};
use futures_util::Stream;
use serde::{Deserialize, Serialize};
use tokio_stream::StreamExt;

use crate::builder::WorkflowBuilder;
use crate::context::Context;
use crate::error::WorkflowError;
use crate::handler::WorkflowHandler;
use crate::snapshot::WorkflowSnapshot;
use crate::step::{StepFn, StepOutput, StepRegistration};

/// Context key under which the [`ChatWindow`] conversation memory is persisted
/// between turns.
const HISTORY_KEY: &str = "__bot_history";

/// Default `ChatWindow` token budget when none is configured.
const DEFAULT_HISTORY_TOKENS: usize = 8192;

/// Default maximum agentic tool-call iterations per turn.
const DEFAULT_MAX_ITERATIONS: u32 = 10;

// ===========================================================================
// Events
// ===========================================================================

/// A message sent *into* the bot, driving one agentic turn.
///
/// Emit one with [`Bot::send`]. The bot's live loop downcasts it, runs the
/// agent against the conversation history, and emits a [`BotResponse`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct InboundMessage {
    /// The user-supplied message text.
    pub text: String,
}

impl Event for InboundMessage {
    fn event_type() -> &'static str {
        static REGISTER: std::sync::Once = std::sync::Once::new();
        REGISTER.call_once(|| {
            register_event_deserializer("blazen::InboundMessage", |value| {
                serde_json::from_value::<InboundMessage>(value)
                    .ok()
                    .map(|e| Box::new(e) as _)
            });
        });
        "blazen::InboundMessage"
    }

    fn event_type_id(&self) -> &'static str {
        "blazen::InboundMessage"
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn clone_boxed(&self) -> Box<dyn AnyEvent> {
        Box::new(self.clone())
    }

    fn to_json(&self) -> serde_json::Value {
        serde_json::to_value(self).expect("InboundMessage serialization should never fail")
    }
}

/// The bot's reply, written to the event stream after each turn.
///
/// Consume these via [`Bot::responses`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BotResponse {
    /// The assistant's reply text for this turn.
    pub text: String,
}

impl Event for BotResponse {
    fn event_type() -> &'static str {
        static REGISTER: std::sync::Once = std::sync::Once::new();
        REGISTER.call_once(|| {
            register_event_deserializer("blazen::BotResponse", |value| {
                serde_json::from_value::<BotResponse>(value)
                    .ok()
                    .map(|e| Box::new(e) as _)
            });
        });
        "blazen::BotResponse"
    }

    fn event_type_id(&self) -> &'static str {
        "blazen::BotResponse"
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn clone_boxed(&self) -> Box<dyn AnyEvent> {
        Box::new(self.clone())
    }

    fn to_json(&self) -> serde_json::Value {
        serde_json::to_value(self).expect("BotResponse serialization should never fail")
    }
}

// ===========================================================================
// Builder
// ===========================================================================

/// Fluent builder for a [`Bot`].
///
/// Construct one with [`Bot::builder`]. All knobs have sensible defaults;
/// the only required input is the [`Model`].
pub struct BotBuilder {
    model: Arc<dyn Model>,
    system_prompt: Option<String>,
    tools: Vec<Arc<dyn Tool>>,
    max_iterations: u32,
    history_tokens: usize,
    idle_timeout: Option<Duration>,
    cost_budget_usd: Option<f64>,
    inject_time: bool,
    summarize: bool,
}

impl BotBuilder {
    /// Start a builder for the given chat [`Model`].
    #[must_use]
    fn new(model: Arc<dyn Model>) -> Self {
        Self {
            model,
            system_prompt: None,
            tools: Vec::new(),
            max_iterations: DEFAULT_MAX_ITERATIONS,
            history_tokens: DEFAULT_HISTORY_TOKENS,
            idle_timeout: None,
            cost_budget_usd: None,
            inject_time: true,
            summarize: false,
        }
    }

    /// Set the system prompt prepended to every agentic turn.
    #[must_use]
    pub fn system_prompt(mut self, prompt: impl Into<String>) -> Self {
        self.system_prompt = Some(prompt.into());
        self
    }

    /// Register a tool the bot's agent may call.
    #[must_use]
    pub fn tool(mut self, tool: Arc<dyn Tool>) -> Self {
        self.tools.push(tool);
        self
    }

    /// Cap the number of agentic tool-call rounds per turn. Default: 10.
    #[must_use]
    pub fn max_iterations(mut self, n: u32) -> Self {
        self.max_iterations = n;
        self
    }

    /// Cap the conversation-memory [`ChatWindow`] at this many estimated
    /// tokens. Default: 8192.
    #[must_use]
    pub fn history_tokens(mut self, n: usize) -> Self {
        self.history_tokens = n;
        self
    }

    /// Shut the bot down after this much wall-time elapses with no message
    /// processed. Surfaces as
    /// [`WorkflowError::IdleTimeout`](crate::error::WorkflowError::IdleTimeout)
    /// on the run's terminal result.
    #[must_use]
    pub fn idle_timeout(mut self, dur: Duration) -> Self {
        self.idle_timeout = Some(dur);
        self
    }

    /// Abort the bot once accumulated LLM cost exceeds this USD ceiling.
    /// Surfaces as
    /// [`WorkflowError::CostBudgetExceeded`](crate::error::WorkflowError::CostBudgetExceeded).
    #[must_use]
    pub fn cost_budget_usd(mut self, ceiling: f64) -> Self {
        self.cost_budget_usd = Some(ceiling);
        self
    }

    /// When `true` (the default), prepend the current UTC time to the system
    /// prompt each turn and add the
    /// [`CurrentTimeTool`](blazen_llm::CurrentTimeTool) so the model can also
    /// fetch the time on demand.
    #[must_use]
    pub fn inject_time(mut self, enabled: bool) -> Self {
        self.inject_time = enabled;
        self
    }

    /// When `true`, compact the conversation history via LLM summarization
    /// before each turn (keeping recent messages verbatim) so long-running
    /// chats retain salient context instead of merely dropping old messages.
    /// Default: `false`.
    #[must_use]
    pub fn summarize(mut self, enabled: bool) -> Self {
        self.summarize = enabled;
        self
    }

    /// Build the bot: wire the live workflow and start its event loop.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowError`] if the underlying workflow fails to build or
    /// start.
    pub async fn build(self) -> Result<Bot, WorkflowError> {
        // Ensure the event deserializers are registered (idempotent) so the
        // events survive a snapshot/resume round-trip through the registry.
        let _ = InboundMessage::event_type();
        let _ = BotResponse::event_type();

        let model = self.model;
        let base_system_prompt = self.system_prompt;
        let configured_tools = self.tools;
        let max_iterations = self.max_iterations;
        let history_tokens = self.history_tokens;
        let inject_time = self.inject_time;
        let summarize = self.summarize;

        let handler: StepFn = Arc::new(move |event, ctx| {
            let model = Arc::clone(&model);
            let base_system_prompt = base_system_prompt.clone();
            let configured_tools = configured_tools.clone();
            Box::pin(async move {
                let inbound = event
                    .as_any()
                    .downcast_ref::<InboundMessage>()
                    .ok_or_else(|| WorkflowError::EventDowncastFailed {
                        expected: InboundMessage::event_type(),
                        got: event.event_type_id().to_string(),
                    })?
                    .clone();

                run_turn(
                    &ctx,
                    model.as_ref(),
                    base_system_prompt.as_deref(),
                    &configured_tools,
                    inbound.text,
                    max_iterations,
                    history_tokens,
                    inject_time,
                    summarize,
                )
                .await
            })
        });

        let step = StepRegistration::new(
            "bot_turn".to_owned(),
            vec![InboundMessage::event_type()],
            vec![BotResponse::event_type(), StopEvent::event_type()],
            handler,
            // Serialize turns so the persisted ChatWindow is updated
            // atomically across overlapping messages.
            1,
        );

        let mut builder = WorkflowBuilder::new("blazen-bot").step(step).live();
        if let Some(dur) = self.idle_timeout {
            builder = builder.idle_timeout(dur);
        }
        if let Some(ceiling) = self.cost_budget_usd {
            builder = builder.cost_budget_usd(ceiling);
        }
        let workflow = builder.build()?;

        // Start the loop. The initial StartEvent has no handler, so under
        // live/lenient routing it is dropped and the loop idles waiting for
        // the first InboundMessage emitted via `Bot::send`.
        let handler = workflow.run(serde_json::Value::Null).await?;

        Ok(Bot { handler })
    }
}

/// Run a single agentic turn against the persisted conversation memory.
///
/// Loads (or creates) the [`ChatWindow`], appends the user message, optionally
/// summarizes, runs the agent, persists the window, folds usage into the
/// cost-budget accumulator, and writes the [`BotResponse`] to the stream.
#[allow(clippy::too_many_arguments)]
async fn run_turn(
    ctx: &Context,
    model: &dyn Model,
    base_system_prompt: Option<&str>,
    configured_tools: &[Arc<dyn Tool>],
    user_text: String,
    max_iterations: u32,
    history_tokens: usize,
    inject_time: bool,
    summarize: bool,
) -> Result<StepOutput, WorkflowError> {
    // 1. Load or create the conversation window and append the user turn.
    let mut window = ctx
        .get::<ChatWindow>(HISTORY_KEY)
        .await
        .unwrap_or_else(|| ChatWindow::new(history_tokens));
    window.add(ChatMessage::user(user_text));

    // 2. Optionally compact the history (summarize the older prefix).
    let mut messages = window.messages().to_vec();
    if summarize {
        messages = blazen_llm::compact(model, &messages, &SummarizationConfig::default())
            .await
            .map_err(to_workflow_error)?;
    }

    // 3. Assemble the agent config: tools + (optional) time tool, and the
    //    (optionally time-prefixed) system prompt.
    let mut tools: Vec<Arc<dyn Tool>> = configured_tools.to_vec();
    if inject_time {
        tools.push(Arc::new(CurrentTimeTool) as Arc<dyn Tool>);
    }

    let system_prompt = build_system_prompt(base_system_prompt, inject_time);

    let mut config = AgentConfig::new(tools).with_max_iterations(max_iterations);
    if let Some(prompt) = system_prompt {
        config = config.with_system_prompt(prompt);
    }

    // 4. Run the agentic loop. The callback is a no-op: per-turn streaming of
    //    intermediate agent events is intentionally out of scope here.
    let result = run_agent_with_callback(model, messages, config, |_evt| {})
        .await
        .map_err(to_workflow_error)?;

    // 5. Extract the assistant reply text and persist it to the window.
    let reply = result.response.content.clone().unwrap_or_default();
    window.add(ChatMessage::assistant(reply.clone()));
    ctx.set(HISTORY_KEY, window).await;

    // 6. Fold the turn's usage/cost into the workflow's running totals so the
    //    cost-budget guard (and usage rollups) observe it.
    if result.total_usage.is_some() || result.total_cost.is_some() {
        let usage = result.total_usage.unwrap_or_default();
        ctx.emit_usage(UsageEvent {
            provider: "bot".to_owned(),
            model: model.model_id().to_owned(),
            modality: blazen_events::Modality::Llm,
            prompt_tokens: usage.prompt_tokens,
            completion_tokens: usage.completion_tokens,
            total_tokens: usage.total_tokens,
            reasoning_tokens: usage.reasoning_tokens,
            cached_input_tokens: usage.cached_input_tokens,
            audio_input_tokens: usage.audio_input_tokens,
            audio_output_tokens: usage.audio_output_tokens,
            image_count: 0,
            audio_seconds: 0.0,
            video_seconds: 0.0,
            cost_usd: result.total_cost,
            latency_ms: result.timing.as_ref().and_then(|t| t.total_ms).unwrap_or(0),
            run_id: ctx.run_id().await,
        })
        .await;
    }

    // 7. Publish the reply to the broadcast stream so `responses()` sees it.
    ctx.write_event_to_stream(BotResponse { text: reply }).await;

    // 8. Re-arm: no routed output, so the loop returns to idle awaiting the
    //    next InboundMessage.
    Ok(StepOutput::None)
}

/// Compose the per-turn system prompt from the configured base prompt and the
/// optional time note. Returns `None` when neither is present.
fn build_system_prompt(base: Option<&str>, inject_time: bool) -> Option<String> {
    match (base, inject_time) {
        (Some(base), true) => Some(format!("{}\n\n{}", current_time_system_note(), base)),
        (Some(base), false) => Some(base.to_owned()),
        (None, true) => Some(current_time_system_note()),
        (None, false) => None,
    }
}

/// Map a [`blazen_llm::BlazenError`] into a [`WorkflowError`], preserving the
/// step name so caller-side errors are attributable.
fn to_workflow_error(err: blazen_llm::BlazenError) -> WorkflowError {
    WorkflowError::StepFailed {
        step_name: "bot_turn".to_owned(),
        source: Box::new(err),
    }
}

// ===========================================================================
// Bot
// ===========================================================================

/// A running, always-on conversational agent.
///
/// Created via [`Bot::builder`]. Drive it with [`send`](Self::send), consume
/// replies via [`responses`](Self::responses), and tear it down with
/// [`shutdown`](Self::shutdown).
pub struct Bot {
    handler: WorkflowHandler,
}

impl Bot {
    /// Start building a bot for the given chat [`Model`].
    #[must_use]
    pub fn builder(model: Arc<dyn Model>) -> BotBuilder {
        BotBuilder::new(model)
    }

    /// Send a message into the bot, driving one agentic turn.
    ///
    /// Non-blocking: the turn runs on the bot's event loop; its reply arrives
    /// on the [`responses`](Self::responses) stream.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowError::ChannelClosed`] if the bot's event loop has
    /// already exited (e.g. after [`shutdown`](Self::shutdown), an idle
    /// timeout, or a cost-budget breach).
    pub fn send(&self, text: impl Into<String>) -> Result<(), WorkflowError> {
        self.handler.emit(InboundMessage { text: text.into() })
    }

    /// Subscribe to the bot's [`BotResponse`] replies.
    ///
    /// Each call returns a fresh stream starting from the current point in
    /// time (replies emitted before subscription are not replayed). Borrows
    /// `&self`, so you can hold a stream and still call other methods.
    pub fn responses(&self) -> impl Stream<Item = BotResponse> + Send + Unpin + use<> {
        self.handler
            .stream_events()
            .filter_map(|boxed| boxed.as_any().downcast_ref::<BotResponse>().cloned())
    }

    /// Shut the bot down: terminate the live event loop.
    ///
    /// Emits a [`StopEvent`], which the live loop treats as its terminal
    /// event. After this, [`send`](Self::send) will fail with
    /// [`WorkflowError::ChannelClosed`].
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowError::ChannelClosed`] if the event loop has already
    /// exited.
    pub fn shutdown(&self) -> Result<(), WorkflowError> {
        self.handler.emit(StopEvent {
            result: serde_json::Value::Null,
        })
    }

    /// Capture a [`WorkflowSnapshot`] of the bot's current state (including
    /// its persisted conversation memory) without stopping the loop.
    ///
    /// # Errors
    ///
    /// Returns [`WorkflowError::ChannelClosed`] if the event loop has already
    /// exited.
    pub async fn snapshot(&self) -> Result<WorkflowSnapshot, WorkflowError> {
        self.handler.snapshot().await
    }
}

// ===========================================================================
// Tests
// ===========================================================================

#[cfg(test)]
mod tests {
    use std::pin::Pin;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use async_trait::async_trait;
    use blazen_llm::Model;
    use blazen_llm::error::BlazenError;
    use blazen_llm::types::{ModelRequest, ModelResponse, Role, StreamChunk};
    use futures_util::Stream;
    use tokio_stream::StreamExt;

    use super::*;

    /// A deterministic mock model: echoes the last user message back, prefixed
    /// with how many user messages it saw in the request — so the test can
    /// assert that conversation memory accumulates across turns.
    struct EchoModel {
        id: String,
        calls: AtomicUsize,
    }

    impl EchoModel {
        fn new() -> Self {
            Self {
                id: "echo-mock".to_owned(),
                calls: AtomicUsize::new(0),
            }
        }
    }

    #[async_trait]
    impl Model for EchoModel {
        fn model_id(&self) -> &str {
            &self.id
        }

        async fn complete(&self, request: ModelRequest) -> Result<ModelResponse, BlazenError> {
            self.calls.fetch_add(1, Ordering::SeqCst);

            let user_count = request
                .messages
                .iter()
                .filter(|m| m.role == Role::User)
                .count();
            let last_user = request
                .messages
                .iter()
                .rev()
                .find(|m| m.role == Role::User)
                .and_then(|m| m.content.text_content())
                .unwrap_or_default();

            Ok(ModelResponse {
                content: Some(format!("[seen {user_count} user msgs] {last_user}")),
                tool_calls: vec![],
                reasoning: None,
                citations: vec![],
                artifacts: vec![],
                usage: None,
                model: "echo-mock".to_owned(),
                finish_reason: Some("stop".to_owned()),
                cost: None,
                timing: None,
                images: vec![],
                audio: vec![],
                videos: vec![],
                metadata: serde_json::Value::Null,
            })
        }

        async fn stream(
            &self,
            _request: ModelRequest,
        ) -> Result<Pin<Box<dyn Stream<Item = Result<StreamChunk, BlazenError>> + Send>>, BlazenError>
        {
            Ok(Box::pin(futures_util::stream::empty()))
        }
    }

    /// Drive a real `send -> responses` round-trip across two turns, asserting
    /// the reply arrives AND that conversation memory persisted (the second
    /// turn's request carries more user messages than the first).
    #[tokio::test]
    async fn bot_send_receive_and_memory_persists() {
        let model = Arc::new(EchoModel::new());
        let bot = Bot::builder(model)
            .system_prompt("You are a test bot.")
            // Keep the turn cheap and deterministic; no time tool needed.
            .inject_time(false)
            .build()
            .await
            .expect("bot should build");

        // Subscribe BEFORE sending so we never miss the reply.
        let mut responses = bot.responses();

        // --- Turn 1 ---
        bot.send("hello").expect("send 1");
        let first = tokio::time::timeout(Duration::from_secs(5), responses.next())
            .await
            .expect("turn 1 should not time out")
            .expect("turn 1 should yield a BotResponse");

        // The mock echoes "[seen 1 user msgs] hello" — exactly one user message
        // (this turn's) was present in the request.
        assert!(
            first.text.contains("hello"),
            "first reply should echo the input, got: {}",
            first.text
        );
        assert!(
            first.text.contains("seen 1 user msgs"),
            "first turn should see exactly 1 user message, got: {}",
            first.text
        );

        // --- Turn 2 ---
        bot.send("how are you").expect("send 2");
        let second = tokio::time::timeout(Duration::from_secs(5), responses.next())
            .await
            .expect("turn 2 should not time out")
            .expect("turn 2 should yield a BotResponse");

        assert!(
            second.text.contains("how are you"),
            "second reply should echo the second input, got: {}",
            second.text
        );
        // Memory persisted: turn 2's request includes turn 1's user message,
        // turn 1's assistant reply, AND turn 2's user message -> 2 user msgs.
        assert!(
            second.text.contains("seen 2 user msgs"),
            "second turn should see 2 user messages (memory persisted), got: {}",
            second.text
        );

        bot.shutdown().expect("shutdown should succeed");
    }

    /// `build_system_prompt` composes the base prompt and time note correctly.
    #[test]
    fn system_prompt_composition() {
        // No base, no time -> None.
        assert!(super::build_system_prompt(None, false).is_none());

        // Base only -> base verbatim.
        assert_eq!(
            super::build_system_prompt(Some("hi"), false).as_deref(),
            Some("hi")
        );

        // Time only -> time note present.
        let time_only = super::build_system_prompt(None, true).expect("some");
        assert!(time_only.contains("Current time"));

        // Both -> time note then base.
        let both = super::build_system_prompt(Some("base prompt"), true).expect("some");
        assert!(both.contains("Current time"));
        assert!(both.contains("base prompt"));
    }

    /// A turn with no system prompt and no time injection still produces a
    /// reply (covers the `system_prompt == None` agent-config branch).
    #[tokio::test]
    async fn bot_works_without_system_prompt() {
        let model = Arc::new(EchoModel::new());
        let bot = Bot::builder(model)
            .inject_time(false)
            .build()
            .await
            .expect("bot should build");

        let mut responses = bot.responses();
        bot.send("ping").expect("send");
        let reply = tokio::time::timeout(Duration::from_secs(5), responses.next())
            .await
            .expect("should not time out")
            .expect("should yield a reply");
        assert!(reply.text.contains("ping"));

        bot.shutdown().expect("shutdown");
    }
}
