//! Cron payload executor.
//!
//! Executes the payload of a cron job after `run()` marks it as started.
//! Supports `SystemEvent` (broadcast) and `AgentTurn` (spawn agent run).

use std::sync::Arc;

use crate::agent::AgentConfigurationError;
use crate::cron::CronPayload;
use crate::messages::outbound::{
    MessageContent, MessageMetadata, OutboundContext, OutboundMessage,
};
use crate::server::ws::{SystemEvent, WsServerState};
use serde_json::Value;
use std::time::Duration;
use tokio_util::sync::CancellationToken;

pub const NO_LLM_PROVIDER_CONFIGURED_ERROR: &str = "no LLM provider configured";

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CronExecuteError {
    LlmNotConfigured,
    Configuration(AgentConfigurationError),
    Other(String),
}

impl std::fmt::Display for CronExecuteError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            CronExecuteError::LlmNotConfigured => f.write_str(NO_LLM_PROVIDER_CONFIGURED_ERROR),
            CronExecuteError::Configuration(error) => write!(f, "{error}"),
            CronExecuteError::Other(message) => f.write_str(message),
        }
    }
}

impl std::error::Error for CronExecuteError {}

impl From<String> for CronExecuteError {
    fn from(value: String) -> Self {
        CronExecuteError::Other(value)
    }
}
/// Outcome of executing a cron payload.
#[derive(Debug)]
pub enum CronRunOutcome {
    /// A SystemEvent was enqueued and broadcast.
    Broadcast,
    /// An AgentTurn was spawned.
    Spawned { run_id: String },
}

/// Optional execution limits enforced before payload dispatch.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ExecutionLimits {
    pub max_turns: Option<u32>,
    pub max_timeout_seconds: Option<u32>,
}

/// Execute a cron job payload.
///
/// For `SystemEvent`: enqueues the event into system event history.
/// For `AgentTurn`: creates a session, registers an agent run, and spawns execution.
pub async fn execute_payload(
    job_id: &str,
    payload: &CronPayload,
    state: &Arc<WsServerState>,
    limits: ExecutionLimits,
) -> Result<CronRunOutcome, CronExecuteError> {
    match payload {
        CronPayload::SystemEvent { text } => execute_system_event(job_id, text, state),
        // Forward-compat sentinel: jobs.json carried a payload kind this
        // binary does not recognize (typically: a downgrade after a newer
        // daemon wrote unknown payloads). Surface as a non-fatal error
        // for this job so the run is recorded as failed and the operator
        // can clear the job — rather than silently dropping every cron
        // by aborting the whole jobs.json parse.
        CronPayload::Unknown => Err(CronExecuteError::Other(format!(
            "cron job '{job_id}' has an unknown payload kind \
             (likely written by a newer daemon); clear or update the job"
        ))),
        CronPayload::AgentTurn {
            message,
            model,
            route,
            thinking,
            timeout_seconds,
            allow_unsafe_external_content,
            deliver,
            channel,
            to,
            best_effort_deliver,
        } => {
            // Reject if both route and model are set — requests must be unambiguous.
            if route.is_some() && model.is_some() {
                return Err(CronExecuteError::Other(
                    "cannot set both `route` and `model` in the same cron task".to_string(),
                ));
            }
            execute_agent_turn(
                job_id,
                state,
                AgentTurnParams {
                    message,
                    model,
                    route,
                    thinking,
                    timeout_seconds: *timeout_seconds,
                    allow_unsafe_external_content: *allow_unsafe_external_content,
                    deliver: *deliver,
                    channel,
                    to,
                    best_effort_deliver: *best_effort_deliver,
                    execution_limits: limits,
                },
            )
            .await
        }
    }
}

struct AgentTurnParams<'a> {
    message: &'a str,
    model: &'a Option<String>,
    route: &'a Option<String>,
    thinking: &'a Option<String>,
    timeout_seconds: Option<u32>,
    allow_unsafe_external_content: Option<bool>,
    deliver: Option<bool>,
    channel: &'a Option<String>,
    to: &'a Option<String>,
    best_effort_deliver: Option<bool>,
    execution_limits: ExecutionLimits,
}

fn execute_system_event(
    job_id: &str,
    text: &str,
    state: &Arc<WsServerState>,
) -> Result<CronRunOutcome, CronExecuteError> {
    let now = crate::cron::now_ms();
    state.enqueue_system_event(SystemEvent {
        ts: now,
        text: text.to_string(),
        host: None,
        ip: None,
        device_id: None,
        instance_id: Some(format!("cron:{}", job_id)),
        reason: Some("cron".to_string()),
    });
    Ok(CronRunOutcome::Broadcast)
}

async fn execute_agent_turn(
    job_id: &str,
    state: &Arc<WsServerState>,
    params: AgentTurnParams<'_>,
) -> Result<CronRunOutcome, CronExecuteError> {
    let session_key = format!("cron:{}", job_id);
    let run_id = uuid::Uuid::new_v4().to_string();
    let normalized_channel = normalize_channel(params.channel);
    let normalized_to = normalize_recipient(params.to);

    let timeout_seconds = apply_timeout_limit(
        params.timeout_seconds,
        params.execution_limits.max_timeout_seconds,
    )?;

    let metadata = build_session_metadata(
        &normalized_channel,
        &normalized_to,
        params.thinking,
        params.model,
    );
    let has_metadata_updates = has_metadata_updates(
        &normalized_channel,
        &normalized_to,
        params.thinking,
        params.model,
    );

    let session =
        load_or_create_cron_session(state, &session_key, metadata, has_metadata_updates).await?;
    append_user_message(state, &session.id, params.message).await?;

    let config = build_agent_config(
        params.model,
        params.route,
        params.allow_unsafe_external_content,
        params.deliver,
        params.execution_limits.max_turns,
    );

    // Validate provider and model before registering the run to avoid orphan entries.
    let provider = state
        .llm_provider()
        .ok_or(CronExecuteError::LlmNotConfigured)?;
    if config.model.trim().is_empty() {
        let error = AgentConfigurationError::missing_model();
        error.log_operator_hint();
        return Err(CronExecuteError::Configuration(error));
    }

    let cancel_token = CancellationToken::new();
    register_agent_run(state, &run_id, &session_key, params.message, &cancel_token)?;
    spawn_timeout_cancellation(state, &run_id, timeout_seconds, cancel_token.clone());

    spawn_delivery_waiter_if_enabled(
        state,
        &run_id,
        job_id,
        params.deliver.unwrap_or(false),
        params.best_effort_deliver.unwrap_or(false),
        normalized_channel,
        normalized_to,
    );

    crate::agent::spawn_run(
        run_id.clone(),
        session_key,
        config,
        state.clone(),
        provider,
        cancel_token,
    );

    Ok(CronRunOutcome::Spawned { run_id })
}

fn normalize_channel(channel: &Option<String>) -> Option<String> {
    channel
        .as_ref()
        .map(|s| s.trim())
        .filter(|s| !s.is_empty())
        .map(|s| s.to_ascii_lowercase())
}

fn apply_timeout_limit(
    requested_timeout_seconds: Option<u32>,
    max_timeout_seconds: Option<u32>,
) -> Result<Option<u32>, CronExecuteError> {
    let Some(limit) = max_timeout_seconds else {
        return Ok(requested_timeout_seconds);
    };
    if limit == 0 {
        return Err(CronExecuteError::Other(
            "objective policy violation: maxRunTimeoutSeconds must be greater than 0".to_string(),
        ));
    }

    match requested_timeout_seconds {
        Some(requested) if requested > limit => Err(CronExecuteError::Other(format!(
            "objective policy violation: timeoutSeconds {requested} exceeds maxRunTimeoutSeconds {limit}"
        ))),
        // In bounded objective mode, an explicit zero timeout is treated as
        // "use the policy cap", not "unbounded execution".
        Some(0) => Ok(Some(limit)),
        Some(requested) => Ok(Some(requested)),
        None => Ok(Some(limit)),
    }
}

fn normalize_recipient(to: &Option<String>) -> Option<String> {
    to.as_ref()
        .map(|s| s.trim())
        .filter(|s| !s.is_empty())
        .map(|s| s.to_string())
}

fn build_session_metadata(
    normalized_channel: &Option<String>,
    normalized_to: &Option<String>,
    thinking: &Option<String>,
    model: &Option<String>,
) -> crate::sessions::SessionMetadata {
    crate::sessions::SessionMetadata {
        channel: normalized_channel.clone(),
        chat_id: normalized_to.clone(),
        thinking_level: thinking.clone(),
        model: model.clone(),
        ..Default::default()
    }
}

fn has_metadata_updates(
    normalized_channel: &Option<String>,
    normalized_to: &Option<String>,
    thinking: &Option<String>,
    model: &Option<String>,
) -> bool {
    normalized_channel.is_some() || normalized_to.is_some() || thinking.is_some() || model.is_some()
}

async fn load_or_create_cron_session(
    state: &Arc<WsServerState>,
    session_key: &str,
    metadata: crate::sessions::SessionMetadata,
    has_metadata_updates: bool,
) -> Result<crate::sessions::Session, String> {
    let store = state.session_store().clone();
    match crate::sessions::get_session_by_key_blocking(store, session_key.to_string()).await {
        Ok(existing) => {
            if has_metadata_updates {
                let store = state.session_store().clone();
                let session_id = existing.id.clone();
                tokio::task::spawn_blocking(move || store.patch_session(&session_id, metadata))
                    .await
                    .map_err(|e| format!("session patch task failed: {e}"))?
                    .map_err(|e| format!("failed to update session: {e}"))
            } else {
                Ok(existing)
            }
        }
        Err(crate::sessions::SessionStoreError::NotFound(_)) => {
            let store = state.session_store().clone();
            let session_key = session_key.to_string();
            tokio::task::spawn_blocking(move || store.get_or_create_session(&session_key, metadata))
                .await
                .map_err(|e| format!("session create task failed: {e}"))?
                .map_err(|e| format!("failed to create session: {e}"))
        }
        Err(err) => Err(format!("failed to load session: {}", err)),
    }
}

async fn append_user_message(
    state: &Arc<WsServerState>,
    session_id: &str,
    message: &str,
) -> Result<(), String> {
    let msg = crate::sessions::ChatMessage::user(session_id, message);
    crate::sessions::append_message_blocking(state.session_store().clone(), msg)
        .await
        .map_err(|e| format!("failed to append message: {}", e))
}

fn build_agent_config(
    model: &Option<String>,
    route: &Option<String>,
    allow_unsafe_external_content: Option<bool>,
    deliver: Option<bool>,
    max_turns_limit: Option<u32>,
) -> crate::agent::AgentConfig {
    let cfg = crate::config::load_config().unwrap_or(Value::Object(serde_json::Map::new()));
    let mut config = crate::agent::AgentConfig::default();
    // Resolve model through route resolver; cron-level model/route override is
    // passed as request_model/request_route so it takes highest precedence.
    if let Err(e) = crate::agent::resolve_agent_model(
        &mut config,
        &cfg,
        None,
        &crate::agent::ModelResolutionOverrides {
            request_route: route.as_deref(),
            request_model: model.as_deref(),
            ..Default::default()
        },
    ) {
        tracing::warn!(error = %e, "cron agent config: model resolution failed");
    }
    crate::agent::apply_agent_config_from_settings(&mut config, &cfg, None);
    if let Some(allow) = allow_unsafe_external_content {
        config.exfiltration_guard = !allow;
    }
    if let Some(deliver) = deliver {
        // Delivery for cron runs is handled via a completion waiter below.
        config.deliver = deliver;
    }
    // Task policy is a per-objective ceiling. It can only reduce turns versus
    // global/runtime config, never raise them.
    if let Some(max_turns_limit) = max_turns_limit.filter(|max_turns| *max_turns > 0) {
        config.max_turns = config.max_turns.min(max_turns_limit);
    }
    config
}

fn register_agent_run(
    state: &Arc<WsServerState>,
    run_id: &str,
    session_key: &str,
    message: &str,
    cancel_token: &CancellationToken,
) -> Result<(), CronExecuteError> {
    use crate::server::ws::{
        current_agents_max_concurrent, AgentRunRegisterOutcome, AgentRunStatus,
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64;
    let cap = current_agents_max_concurrent();
    let mut registry = state.agent_run_registry.lock();
    match registry.try_register_with_cap(
        crate::server::ws::AgentRun {
            run_id: run_id.to_string(),
            session_key: session_key.to_string(),
            delivery_recipient_id: None,
            typing_context: None,
            status: AgentRunStatus::Queued,
            message: message.to_string(),
            response: String::new(),
            error: None,
            created_at: now,
            started_at: None,
            completed_at: None,
            cancel_token: cancel_token.clone(),
            waiters: Vec::new(),
        },
        cap,
    ) {
        AgentRunRegisterOutcome::Registered | AgentRunRegisterOutcome::DuplicateActive => Ok(()),
        AgentRunRegisterOutcome::AtCap { active, cap } => Err(CronExecuteError::Other(format!(
            "agents.defaults.maxConcurrent reached: {active}/{cap} runs active. \
             Skipping cron-spawned agent run; cancel an existing run or raise the cap."
        ))),
    }
}

fn spawn_timeout_cancellation(
    state: &Arc<WsServerState>,
    run_id: &str,
    timeout_seconds: Option<u32>,
    cancel_token: CancellationToken,
) {
    let Some(timeout) = timeout_seconds.filter(|timeout| *timeout > 0) else {
        return;
    };
    let run_id = run_id.to_string();
    let waiter = {
        let mut registry = state.agent_run_registry.lock();
        registry.add_waiter(&run_id)
    };
    if let Some(waiter) = waiter {
        tokio::spawn(async move {
            tokio::select! {
                _ = tokio::time::sleep(Duration::from_secs(timeout as u64)) => {
                    tracing::warn!(
                        run_id = %run_id,
                        timeout_seconds = timeout,
                        "cron agent run exceeded timeout; cancelling"
                    );
                    cancel_token.cancel();
                }
                _ = waiter => {}
            }
        });
    }
}

fn spawn_delivery_waiter_if_enabled(
    state: &Arc<WsServerState>,
    run_id: &str,
    job_id: &str,
    deliver: bool,
    retry_enabled: bool,
    normalized_channel: Option<String>,
    normalized_to: Option<String>,
) {
    if !deliver {
        return;
    }

    match (normalized_channel, normalized_to) {
        (Some(channel_id), Some(recipient_id)) => {
            let delivery_pipeline = Arc::clone(state.message_pipeline());
            let delivery_run_id = run_id.to_string();
            let delivery_message_id = format!("cron-deliver:{delivery_run_id}");
            let waiter = {
                let mut registry = state.agent_run_registry.lock();
                registry.add_waiter(run_id)
            };
            if let Some(waiter) = waiter {
                tokio::spawn(async move {
                    let result = match waiter.await {
                        Ok(result) => result,
                        Err(_) => {
                            tracing::warn!(
                                run_id = %delivery_run_id,
                                "cron delivery waiter dropped before completion"
                            );
                            return;
                        }
                    };
                    if result.status != crate::server::ws::AgentRunStatus::Completed {
                        return;
                    }
                    let Some(content) = result.response else {
                        return;
                    };
                    let metadata = MessageMetadata {
                        recipient_id: Some(recipient_id),
                        ..Default::default()
                    };
                    let outbound = OutboundMessage::new(channel_id, MessageContent::text(content))
                        .with_metadata(metadata);
                    let mut ctx = OutboundContext::new().with_trace_id(&delivery_message_id);
                    if retry_enabled {
                        ctx = ctx.with_retries(3);
                    }
                    if let Err(err) = delivery_pipeline.queue_with_idempotency(
                        outbound,
                        ctx,
                        Some(&delivery_message_id),
                    ) {
                        tracing::warn!(
                            run_id = %delivery_run_id,
                            error = %err,
                            "failed to queue cron delivery message"
                        );
                    }
                });
            }
        }
        (None, _) => {
            tracing::warn!(
                job_id = %job_id,
                "cron delivery requested without channel; skipping"
            );
        }
        (_, None) => {
            tracing::warn!(
                job_id = %job_id,
                "cron delivery requested without recipient; skipping"
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::cron::CronPayload;
    use crate::server::ws::{WsServerConfig, WsServerState};
    use crate::sessions;
    use async_trait::async_trait;
    use std::sync::Arc;
    use tokio::sync::mpsc;
    use tokio_util::sync::CancellationToken;

    struct NeverCalledProvider;

    #[async_trait]
    impl crate::agent::LlmProvider for NeverCalledProvider {
        async fn complete(
            &self,
            _request: crate::agent::provider::CompletionRequest,
            _cancel_token: CancellationToken,
        ) -> Result<mpsc::Receiver<crate::agent::StreamEvent>, crate::agent::AgentError> {
            panic!("cron missing-model guard should run before provider.complete");
        }
    }

    /// Create a WsServerState backed by a temp directory so tests work on all
    /// platforms (including Windows CI where writing to ~/.config/carapace may fail).
    fn make_test_state() -> (Arc<WsServerState>, tempfile::TempDir) {
        let tmp = tempfile::tempdir().unwrap();
        let store = Arc::new(sessions::SessionStore::with_base_path(
            tmp.path().join("sessions"),
        ));
        let state = WsServerState::new(WsServerConfig::default()).with_session_store(store);
        (Arc::new(state), tmp)
    }

    #[tokio::test]
    async fn test_execute_system_event() {
        let (state, _tmp) = make_test_state();

        let payload = CronPayload::SystemEvent {
            text: "test cron event".to_string(),
        };

        let result = execute_payload("job-1", &payload, &state, ExecutionLimits::default()).await;
        assert!(result.is_ok());
        assert!(matches!(result.unwrap(), CronRunOutcome::Broadcast));
    }

    #[tokio::test]
    async fn test_execute_agent_turn_no_provider() {
        // Without an LLM provider, agent turn should fail
        let (state, _tmp) = make_test_state();

        let payload = CronPayload::AgentTurn {
            message: "do something".to_string(),
            model: None,
            route: None,
            thinking: None,
            timeout_seconds: None,
            allow_unsafe_external_content: None,
            deliver: None,
            channel: None,
            to: None,
            best_effort_deliver: None,
        };

        let result = execute_payload("job-2", &payload, &state, ExecutionLimits::default()).await;
        assert!(result.is_err());
        assert_eq!(result.unwrap_err(), CronExecuteError::LlmNotConfigured);
    }

    #[tokio::test]
    async fn test_execute_agent_turn_missing_model_returns_safe_error() {
        let (state, _tmp) = make_test_state();
        state.set_llm_provider(Some(Arc::new(NeverCalledProvider)));

        let payload = CronPayload::AgentTurn {
            message: "do something".to_string(),
            model: None,
            route: None,
            thinking: None,
            timeout_seconds: None,
            allow_unsafe_external_content: None,
            deliver: None,
            channel: None,
            to: None,
            best_effort_deliver: None,
        };

        let result = execute_payload(
            "job-missing-model",
            &payload,
            &state,
            ExecutionLimits::default(),
        )
        .await;
        let error = result.unwrap_err();
        assert!(matches!(error, CronExecuteError::Configuration(_)));

        let message = error.to_string();
        assert_eq!(message, "agent model is not configured");
        assert!(!message.contains("agents.defaults.model"), "got: {message}");
        assert!(!message.contains("anthropic:"), "got: {message}");
    }

    #[tokio::test]
    async fn test_agent_turn_applies_session_metadata() {
        let (state, _tmp) = make_test_state();

        let payload = CronPayload::AgentTurn {
            message: "hello".to_string(),
            model: Some("model-x".to_string()),
            route: None,
            thinking: Some("deep".to_string()),
            timeout_seconds: Some(10),
            allow_unsafe_external_content: Some(false),
            deliver: Some(true),
            channel: Some("Signal".to_string()),
            to: Some("123".to_string()),
            best_effort_deliver: Some(true),
        };

        let result =
            execute_payload("job-meta", &payload, &state, ExecutionLimits::default()).await;
        assert!(result.is_err());

        let session = state
            .session_store()
            .get_session_by_key("cron:job-meta")
            .unwrap();
        assert_eq!(session.metadata.channel, Some("signal".to_string()));
        assert_eq!(session.metadata.chat_id, Some("123".to_string()));
        assert_eq!(session.metadata.thinking_level, Some("deep".to_string()));
        assert_eq!(session.metadata.model, Some("model-x".to_string()));
    }

    #[tokio::test]
    async fn test_agent_turn_updates_existing_metadata() {
        let (state, _tmp) = make_test_state();

        let initial = sessions::SessionMetadata {
            channel: Some("signal".to_string()),
            chat_id: Some("111".to_string()),
            thinking_level: Some("shallow".to_string()),
            model: Some("old-model".to_string()),
            ..Default::default()
        };
        state
            .session_store()
            .get_or_create_session("cron:job-update", initial)
            .unwrap();

        let payload = CronPayload::AgentTurn {
            message: "hello".to_string(),
            model: Some("new-model".to_string()),
            route: None,
            thinking: Some("deep".to_string()),
            timeout_seconds: Some(10),
            allow_unsafe_external_content: Some(false),
            deliver: Some(true),
            channel: Some("Discord".to_string()),
            to: Some("999".to_string()),
            best_effort_deliver: Some(true),
        };

        let result =
            execute_payload("job-update", &payload, &state, ExecutionLimits::default()).await;
        assert!(result.is_err());

        let session = state
            .session_store()
            .get_session_by_key("cron:job-update")
            .unwrap();
        assert_eq!(session.metadata.channel, Some("discord".to_string()));
        assert_eq!(session.metadata.chat_id, Some("999".to_string()));
        assert_eq!(session.metadata.thinking_level, Some("deep".to_string()));
        assert_eq!(session.metadata.model, Some("new-model".to_string()));
    }

    #[test]
    fn test_apply_timeout_limit_enforces_policy_budget() {
        let result = apply_timeout_limit(Some(30), Some(10));
        assert_eq!(
            result,
            Err(CronExecuteError::Other(
                "objective policy violation: timeoutSeconds 30 exceeds maxRunTimeoutSeconds 10"
                    .to_string()
            ))
        );
    }

    #[test]
    fn test_apply_timeout_limit_uses_budget_when_timeout_missing_or_zero() {
        assert_eq!(apply_timeout_limit(None, Some(12)).unwrap(), Some(12));
        assert_eq!(apply_timeout_limit(Some(0), Some(12)).unwrap(), Some(12));
    }

    #[tokio::test]
    async fn test_execute_agent_turn_rejects_timeout_over_execution_limits() {
        let (state, _tmp) = make_test_state();
        let payload = CronPayload::AgentTurn {
            message: "hello".to_string(),
            model: None,
            route: None,
            thinking: None,
            timeout_seconds: Some(45),
            allow_unsafe_external_content: None,
            deliver: None,
            channel: None,
            to: None,
            best_effort_deliver: None,
        };
        let result = execute_payload(
            "job-timeout-limit",
            &payload,
            &state,
            ExecutionLimits {
                max_turns: Some(10),
                max_timeout_seconds: Some(10),
            },
        )
        .await;
        assert!(matches!(
            result,
            Err(CronExecuteError::Other(message))
            if message
                == "objective policy violation: timeoutSeconds 45 exceeds maxRunTimeoutSeconds 10"
        ));
    }
}
