hyper_agent_ai/

agent_loop.rs

use std::collections::HashMap;
use std::fs;
use std::io::Write;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};

use serde::{Deserialize, Serialize};
use tokio::sync::Mutex as TokioMutex;
use tokio::sync::Notify;

use crate::agent_config::{load_agent_config, AgentConfig, TradingMode};
use crate::agent_history::AgentHistoryStore;
use crate::claude::AgentDecision;
use crate::tools::{
    CancelOrderTool, DoNothingTool, GetOpenOrdersTool, GetPositionsTool, GetTradeHistoryTool,
    PlaceOrderTool, SetStopLossTool, SetTakeProfitTool,
};
use hyper_agent_core::account_state_ext::{account_state_from_exchange, account_state_from_paper};
use hyper_agent_core::config::AppConfig;
use hyper_agent_core::executor::{DryRunExecutor, PaperExecutor};
use hyper_agent_core::live_executor::{AssetMetaCache, LiveExecutor};
use hyper_agent_core::pipeline::{OrderPipeline, PipelineContext, PipelineError};
use hyper_agent_core::pipeline_stages::{LoggerStage, OrderRouterStage, RiskGuardStage};
use hyper_agent_core::position_manager::PositionManager;
use hyper_agent_core::signal::{Side, SignalAction, SignalSource, TradeSignal};

use hyper_exchange::{ExchangeClient, Signer};
use hyper_risk::risk::{AccountState, RiskGuard};
use hyper_strategy::strategy_composer::ComposerProfile;
#[allow(deprecated)]
use hyper_ta::technical_analysis::{calculate_indicators, format_technical_summary};
use motosan_chat_agent::{AgentLoop as MotosanAgentLoop, Message as AgentMessage};
use motosan_chat_ai::MotosanAiClient;
use motosan_chat_core::{Channel, ChatType, IncomingEvent, IncomingEventKind, Thread};

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

/// Status of a single agent loop.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum LoopState {
    Running,
    Paused,
    Stopped,
    Error,
}

/// Externally visible status of an agent loop.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LoopStatus {
    pub agent_id: String,
    pub state: LoopState,
    pub interval_minutes: u64,
    pub iteration_count: u64,
    pub last_iteration_at: Option<String>,
    pub last_error: Option<String>,
}

/// Persisted configuration for an agent loop (restored on app restart).
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct LoopConfig {
    pub agent_id: String,
    pub interval_minutes: u64,
    pub auto_start: bool,
}

/// Log entry for a single loop iteration.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct AgentLoopLog {
    pub agent_id: String,
    pub iteration: u64,
    pub timestamp: String,
    pub phase: String,
    pub success: bool,
    pub message: String,
    pub decision_summary: Option<String>,
    pub tool_turn_count: u32,
    pub risk_blocked: bool,
    pub order_executed: bool,
}

// ---------------------------------------------------------------------------
// Wallet context for live order execution
// ---------------------------------------------------------------------------

/// Signing context required for live order execution on Hyperliquid L1.
///
/// Contains the agent wallet address and a signer that can produce EIP-712
/// signatures. The private key is never exposed — only the `Signer` trait
/// is used, which accepts a 32-byte hash and returns a 65-byte signature.
#[derive(Clone)]
pub struct WalletContext {
    /// The agent wallet address (0x-prefixed hex, e.g. from onboarding).
    pub agent_address: String,
    /// Whether to target mainnet (true) or testnet (false).
    pub is_mainnet: bool,
    /// Optional vault address for vault-based trading.
    pub vault_address: Option<String>,
    /// Signer that holds the private key behind an opaque trait.
    pub signer: Arc<dyn Signer>,
}

// ---------------------------------------------------------------------------
// Internal handle for a running loop
// ---------------------------------------------------------------------------

struct LoopHandle {
    status: Arc<TokioMutex<LoopStatus>>,
    /// Signal to cancel the loop task.
    cancel: tokio::sync::watch::Sender<bool>,
    /// Notify to wake up a paused loop or trigger an immediate run.
    trigger: Arc<Notify>,
}

// ---------------------------------------------------------------------------
// AgentLoopManager
// ---------------------------------------------------------------------------

/// Manages multiple agent loops (one per agent_id).
pub struct AgentLoopManager {
    loops: TokioMutex<HashMap<String, LoopHandle>>,
}

impl AgentLoopManager {
    pub fn new() -> Self {
        Self {
            loops: TokioMutex::new(HashMap::new()),
        }
    }

    /// Start an agent loop. If one is already running for this agent, returns an error.
    ///
    /// `wallet_context` is required for `TradingMode::Live`. Pass `None` for paper trading.
    pub async fn start_loop(
        &self,
        agent_id: String,
        interval_minutes: u64,
        wallet_context: Option<WalletContext>,
    ) -> Result<(), String> {
        let mut loops = self.loops.lock().await;

        // If a loop exists and is not stopped, reject.
        if let Some(handle) = loops.get(&agent_id) {
            let status = handle.status.lock().await;
            if status.state == LoopState::Running || status.state == LoopState::Paused {
                return Err(format!(
                    "Agent loop for '{}' is already {} -- stop it first",
                    agent_id,
                    match status.state {
                        LoopState::Running => "running",
                        LoopState::Paused => "paused",
                        _ => "active",
                    }
                ));
            }
        }

        let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
        let trigger = Arc::new(Notify::new());

        let status = Arc::new(TokioMutex::new(LoopStatus {
            agent_id: agent_id.clone(),
            state: LoopState::Running,
            interval_minutes,
            iteration_count: 0,
            last_iteration_at: None,
            last_error: None,
        }));

        // Persist config
        save_loop_config(&LoopConfig {
            agent_id: agent_id.clone(),
            interval_minutes,
            auto_start: true,
        });

        // Spawn the loop task
        let status_clone = Arc::clone(&status);
        let trigger_clone = Arc::clone(&trigger);
        let aid = agent_id.clone();
        let wc = wallet_context.clone();
        tokio::spawn(async move {
            run_loop(
                aid,
                interval_minutes,
                status_clone,
                cancel_rx,
                trigger_clone,
                wc,
            )
            .await;
        });

        loops.insert(
            agent_id,
            LoopHandle {
                status,
                cancel: cancel_tx,
                trigger,
            },
        );

        Ok(())
    }

    /// Stop an agent loop.
    pub async fn stop_loop(&self, agent_id: &str) -> Result<(), String> {
        let mut loops = self.loops.lock().await;
        let handle = loops
            .remove(agent_id)
            .ok_or_else(|| format!("No loop found for agent '{}'", agent_id))?;

        // Signal cancellation
        let _ = handle.cancel.send(true);
        // Wake the task so it can observe the cancellation immediately
        handle.trigger.notify_one();

        // Update status
        {
            let mut status = handle.status.lock().await;
            status.state = LoopState::Stopped;
        }

        // Update persisted config
        save_loop_config(&LoopConfig {
            agent_id: agent_id.to_string(),
            interval_minutes: handle.status.lock().await.interval_minutes,
            auto_start: false,
        });

        Ok(())
    }

    /// Pause a running loop.
    pub async fn pause_loop(&self, agent_id: &str) -> Result<(), String> {
        let loops = self.loops.lock().await;
        let handle = loops
            .get(agent_id)
            .ok_or_else(|| format!("No loop found for agent '{}'", agent_id))?;

        let mut status = handle.status.lock().await;
        if status.state != LoopState::Running {
            return Err(format!(
                "Cannot pause: agent loop is {:?}, not Running",
                status.state
            ));
        }
        status.state = LoopState::Paused;
        Ok(())
    }

    /// Resume a paused loop.
    pub async fn resume_loop(&self, agent_id: &str) -> Result<(), String> {
        let loops = self.loops.lock().await;
        let handle = loops
            .get(agent_id)
            .ok_or_else(|| format!("No loop found for agent '{}'", agent_id))?;

        let mut status = handle.status.lock().await;
        if status.state != LoopState::Paused {
            return Err(format!(
                "Cannot resume: agent loop is {:?}, not Paused",
                status.state
            ));
        }
        status.state = LoopState::Running;
        // Wake the loop so it starts the next iteration immediately
        handle.trigger.notify_one();
        Ok(())
    }

    /// Trigger a single immediate iteration (regardless of running state).
    /// If no loop exists, creates a temporary one that runs once and stops.
    pub async fn trigger_once(&self, agent_id: &str) -> Result<(), String> {
        let loops = self.loops.lock().await;
        if let Some(handle) = loops.get(agent_id) {
            // Trigger the existing loop to run one iteration now
            handle.trigger.notify_one();
            Ok(())
        } else {
            // No active loop -- run a single iteration inline (paper mode, no wallet)
            drop(loops);
            let log = execute_single_iteration(agent_id, 0, None).await;
            write_log_entry(&log);
            if log.success {
                Ok(())
            } else {
                Err(log.message)
            }
        }
    }

    /// Get the status of a specific agent loop.
    pub async fn get_status(&self, agent_id: &str) -> Result<LoopStatus, String> {
        let loops = self.loops.lock().await;
        if let Some(handle) = loops.get(agent_id) {
            Ok(handle.status.lock().await.clone())
        } else {
            // Check persisted config for a stopped loop
            Ok(LoopStatus {
                agent_id: agent_id.to_string(),
                state: LoopState::Stopped,
                interval_minutes: 15,
                iteration_count: 0,
                last_iteration_at: None,
                last_error: None,
            })
        }
    }
}

// ---------------------------------------------------------------------------
// Loop runner
// ---------------------------------------------------------------------------

async fn run_loop(
    agent_id: String,
    interval_minutes: u64,
    status: Arc<TokioMutex<LoopStatus>>,
    mut cancel_rx: tokio::sync::watch::Receiver<bool>,
    trigger: Arc<Notify>,
    wallet_context: Option<WalletContext>,
) {
    let interval = tokio::time::Duration::from_secs(interval_minutes * 60);

    loop {
        // Check cancellation
        if *cancel_rx.borrow() {
            break;
        }

        // Check if paused -- if so, wait for a trigger/cancel
        {
            let s = status.lock().await;
            if s.state == LoopState::Paused {
                drop(s);
                // Wait until either triggered (resume) or cancelled
                tokio::select! {
                    _ = trigger.notified() => continue,
                    _ = cancel_rx.changed() => break,
                }
            }
        }

        // Execute one iteration
        let iteration = {
            let s = status.lock().await;
            s.iteration_count
        };

        let log = execute_single_iteration(&agent_id, iteration, wallet_context.clone()).await;
        write_log_entry(&log);

        // Update status
        {
            let mut s = status.lock().await;
            if *cancel_rx.borrow() {
                break;
            }
            s.iteration_count += 1;
            s.last_iteration_at = Some(log.timestamp.clone());
            if log.success {
                s.last_error = None;
            } else {
                s.last_error = Some(log.message.clone());
            }
        }

        // Wait for the next interval, or an early trigger, or cancellation
        tokio::select! {
            _ = tokio::time::sleep(interval) => {},
            _ = trigger.notified() => {},
            _ = cancel_rx.changed() => break,
        }
    }

    // Mark as stopped when we exit
    let mut s = status.lock().await;
    s.state = LoopState::Stopped;
}

// ---------------------------------------------------------------------------
// Single iteration execution (pipeline-based)
// ---------------------------------------------------------------------------

/// Convert an AgentDecision (from Claude) into a TradeSignal for the pipeline.
///
/// Scans tool_calls for place_order and converts to an Open signal.
/// Returns None for do_nothing or if no recognised trading tool call is found.
fn decision_to_trade_signal(
    decision: &AgentDecision,
    symbol: &str,
    agent_id: &str,
) -> Option<TradeSignal> {
    for tc in &decision.tool_calls {
        match tc.tool_name.as_str() {
            "place_order" => {
                let side = tc
                    .tool_input
                    .get("is_buy")
                    .and_then(|v| v.as_bool())
                    .map(|b| if b { Side::Buy } else { Side::Sell })
                    .unwrap_or(Side::Buy);
                let size = tc
                    .tool_input
                    .get("size")
                    .and_then(|v| {
                        v.as_str()
                            .and_then(|s| s.parse().ok())
                            .or_else(|| v.as_f64())
                    })
                    .unwrap_or(0.0);
                let price = tc.tool_input.get("price").and_then(|v| {
                    v.as_str()
                        .and_then(|s| s.parse().ok())
                        .or_else(|| v.as_f64())
                });

                return Some(TradeSignal {
                    id: uuid::Uuid::new_v4().to_string(),
                    timestamp: std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap_or_default()
                        .as_secs(),
                    source: SignalSource::Playbook {
                        strategy_id: agent_id.to_string(),
                        regime: "claude".to_string(),
                    },
                    symbol: symbol.to_string(),
                    action: SignalAction::Open { side, size, price },
                    reason: decision.reasoning.clone(),
                });
            }
            "do_nothing" => return None,
            _ => continue,
        }
    }
    None
}

/// Create an error AgentLoopLog with minimal boilerplate.
fn error_log(
    agent_id: &str,
    iteration: u64,
    timestamp: String,
    phase: &str,
    message: &str,
) -> AgentLoopLog {
    AgentLoopLog {
        agent_id: agent_id.to_string(),
        iteration,
        timestamp,
        phase: phase.to_string(),
        success: false,
        message: message.to_string(),
        decision_summary: None,
        tool_turn_count: 0,
        risk_blocked: false,
        order_executed: false,
    }
}

/// Pipeline-based single iteration: fetches market data, calls Claude, then
/// routes the resulting signal through the unified OrderPipeline.
async fn execute_single_iteration(
    agent_id: &str,
    iteration: u64,
    wallet_context: Option<WalletContext>,
) -> AgentLoopLog {
    let timestamp = chrono::Utc::now().to_rfc3339();

    // --- Phase 0: Load agent config ---
    let config = load_agent_config(agent_id).unwrap_or_else(AgentConfig::default);

    let is_mainnet = true;
    let fetcher = hyper_market::market_data::MarketDataFetcher::new(is_mainnet);

    // --- Phase 1: Fetch real candle data ---
    let trading_pairs = if config.trading_pairs.is_empty() {
        vec!["BTC-PERP".to_string()]
    } else {
        config.trading_pairs.clone()
    };
    let primary_symbol = trading_pairs[0].clone();

    let candles = match fetcher
        .fetch_candles_with_params(&primary_symbol, "1h", 200)
        .await
    {
        Ok(c) if !c.is_empty() => c,
        Ok(_) => {
            return error_log(
                agent_id,
                iteration,
                timestamp,
                "fetch_market_data",
                "No candles returned from exchange",
            );
        }
        Err(e) => {
            return error_log(
                agent_id,
                iteration,
                timestamp,
                "fetch_market_data",
                &format!("Failed to fetch market data: {}", e),
            );
        }
    };

    // --- Phase 2: Calculate technical indicators ---
    #[allow(deprecated)]
    let indicators = calculate_indicators(&candles);
    let current_price = candles.last().map(|c| c.close);
    let ta_summary = format_technical_summary(&primary_symbol, &indicators, current_price);

    // --- Phase 3: Compose strategy signals ---
    let composer_analysis = if let Some(ref profile_str) = config.composer_profile {
        if let Some(profile) = parse_composer_profile(profile_str) {
            let composer =
                hyper_strategy::strategy_composer::StrategyComposer::new(profile, &primary_symbol);
            let volume_context =
                hyper_risk::risk_defaults::analyze_volume_context(&candles, &indicators);
            let signal = composer.compose_signals(&indicators, None, &volume_context);
            Some(composer.format_for_claude(&signal))
        } else {
            None
        }
    } else {
        None
    };

    // --- Phase 3b: Fetch funding rate ---
    let funding_rate_info =
        match crate::tools::fetch_funding_rate(&primary_symbol, is_mainnet).await {
            Ok(info) => Some(info),
            Err(e) => {
                eprintln!(
                    "[agent_loop] Failed to fetch funding rate for {}: {}",
                    primary_symbol, e
                );
                None
            }
        };

    // --- Phase 4: Build prompt for Claude ---
    let mut user_message = String::with_capacity(4096);
    user_message.push_str(&format!("## Market Analysis for {}\n\n", primary_symbol));
    user_message.push_str(&ta_summary);

    if let Some(ref composer_text) = composer_analysis {
        user_message.push_str("\n\n## Strategy Composer Analysis\n");
        user_message.push_str(composer_text);
    }

    if let Some(price) = current_price {
        user_message.push_str(&format!("\n\nCurrent Price: ${:.2}", price));
    }

    if let Some(ref fr_info) = funding_rate_info {
        user_message.push_str("\n\n## Funding Rate");
        if let Some(rate_pct) = fr_info.get("funding_rate_pct").and_then(|v| v.as_str()) {
            user_message.push_str(&format!("\nCurrent Funding Rate: {}", rate_pct));
        }
        if let Some(annual_pct) = fr_info.get("annualized_rate_pct").and_then(|v| v.as_str()) {
            user_message.push_str(&format!("\nAnnualized Rate: {}", annual_pct));
        }
        if let Some(direction) = fr_info.get("direction").and_then(|v| v.as_str()) {
            let direction_desc = match direction {
                "longs_pay_shorts" => "Longs pay shorts (bullish sentiment)",
                "shorts_pay_longs" => "Shorts pay longs (bearish sentiment)",
                _ => "Neutral",
            };
            user_message.push_str(&format!("\nDirection: {}", direction_desc));
        }
    }

    user_message.push_str("\n\n## Current Positions\nNo position data available in loop context.");

    let mode_str = match config.trading_mode {
        TradingMode::Paper => "PAPER TRADING (simulated)",
        TradingMode::Live => "LIVE TRADING (real orders)",
    };
    user_message.push_str(&format!("\n\nTrading Mode: {}", mode_str));
    user_message.push_str(&format!(
        "\nMax Position Size: ${:.2}",
        config.max_position_size_usd
    ));
    user_message.push_str(
        "\n\nPlease analyze the market and decide: use place_order to trade, or do_nothing if conditions are not favorable.",
    );

    let system_prompt = config.system_prompt.clone();

    // --- Phase 5: Call Claude via motosan-chat-agent ---
    let llm_result = match call_claude_real(
        agent_id,
        &system_prompt,
        &user_message,
        config.max_retries,
        config.max_tool_turns,
    )
    .await
    {
        Ok(outcome) => outcome,
        Err(e) => {
            eprintln!("[agent_loop] Claude API call failed: {}", e);
            return error_log(
                agent_id,
                iteration,
                timestamp,
                "call_claude",
                &format!("Claude API error: {}", e),
            );
        }
    };

    let decision = &llm_result.decision;

    let decision_summary = format!(
        "reasoning={} tool_calls={} confidence={:.2}",
        &decision.reasoning[..decision.reasoning.len().min(200)],
        decision.tool_calls.len(),
        decision.confidence,
    );

    // --- Phase 6: Convert decision to TradeSignal and run through pipeline ---
    let signal = decision_to_trade_signal(decision, &primary_symbol, agent_id);

    match signal {
        Some(sig) => {
            // Build position manager with a deterministic per-agent DB path
            let db_dir = {
                let mut p = dirs::data_dir().unwrap_or_else(|| PathBuf::from("."));
                p.push("hyper-agent");
                p.push("positions");
                let _ = fs::create_dir_all(&p);
                p
            };
            let db_path = db_dir.join(format!("{}.db", agent_id));
            let pm = match PositionManager::new(db_path.to_str().unwrap_or("positions.db")) {
                Ok(pm) => Arc::new(pm),
                Err(e) => {
                    return error_log(
                        agent_id,
                        iteration,
                        timestamp,
                        "pipeline_init",
                        &format!("Failed to open position DB: {}", e),
                    );
                }
            };

            let paper_exec = Arc::new(PaperExecutor::new(Arc::clone(&pm)));
            let dry_run = Arc::new(DryRunExecutor);

            // Build live executor from wallet_context when available
            let live_exec: Option<Arc<dyn hyper_agent_core::executor::OrderSubmitter>> =
                if matches!(config.trading_mode, TradingMode::Live) {
                    if let Some(ref wc) = wallet_context {
                        let client = ExchangeClient::new(wc.is_mainnet);
                        match AssetMetaCache::fetch(&client).await {
                            Ok(meta_cache) => {
                                let executor = LiveExecutor::new(
                                    Arc::clone(&wc.signer),
                                    wc.agent_address.clone(),
                                    wc.vault_address.clone(),
                                    wc.is_mainnet,
                                    client,
                                    meta_cache,
                                );
                                Some(Arc::new(executor))
                            }
                            Err(e) => {
                                eprintln!(
                                    "[agent_loop] Failed to fetch asset meta for live mode, \
                                     falling back to paper: {}",
                                    e
                                );
                                None
                            }
                        }
                    } else {
                        None
                    }
                } else {
                    None
                };

            let pipeline = OrderPipeline::new(vec![
                Box::new(RiskGuardStage::new()),
                Box::new(OrderRouterStage::new(
                    paper_exec,
                    live_exec.clone(),
                    dry_run,
                )),
                Box::new(LoggerStage),
            ]);

            // Use exchange account state in live mode, paper otherwise
            let (account_state, mode) = if live_exec.is_some() {
                if let Some(ref wc) = wallet_context {
                    let client = ExchangeClient::new(wc.is_mainnet);
                    match account_state_from_exchange(&client, &wc.agent_address).await {
                        Ok(state) => (state, hyper_agent_core::pipeline::TradingMode::Live),
                        Err(e) => {
                            eprintln!(
                                "[agent_loop] Failed to fetch exchange account state, \
                                     falling back to paper: {}",
                                e
                            );
                            (
                                account_state_from_paper(&pm).await,
                                hyper_agent_core::pipeline::TradingMode::Paper,
                            )
                        }
                    }
                } else {
                    (
                        account_state_from_paper(&pm).await,
                        hyper_agent_core::pipeline::TradingMode::Paper,
                    )
                }
            } else {
                (
                    account_state_from_paper(&pm).await,
                    hyper_agent_core::pipeline::TradingMode::Paper,
                )
            };

            let mut ctx = PipelineContext {
                mode,
                account_state,
                position_manager: pm,
                config: Arc::new(AppConfig::default()),
                execution_results: vec![],
            };

            match pipeline.execute(sig, &mut ctx).await {
                Ok(_) => AgentLoopLog {
                    agent_id: agent_id.to_string(),
                    iteration,
                    timestamp,
                    phase: "complete".to_string(),
                    success: true,
                    message: "Order executed successfully via pipeline".to_string(),
                    decision_summary: Some(decision_summary),
                    tool_turn_count: llm_result.tool_turn_count,
                    risk_blocked: false,
                    order_executed: true,
                },
                Err(PipelineError::RiskBlocked(msg)) => AgentLoopLog {
                    agent_id: agent_id.to_string(),
                    iteration,
                    timestamp,
                    phase: "complete".to_string(),
                    success: true,
                    message: format!("Risk check blocked execution: {}", msg),
                    decision_summary: Some(decision_summary),
                    tool_turn_count: llm_result.tool_turn_count,
                    risk_blocked: true,
                    order_executed: false,
                },
                Err(PipelineError::CircuitBreakerTripped) => AgentLoopLog {
                    agent_id: agent_id.to_string(),
                    iteration,
                    timestamp,
                    phase: "complete".to_string(),
                    success: true,
                    message: "Circuit breaker tripped -- positions auto-closed".to_string(),
                    decision_summary: Some(decision_summary),
                    tool_turn_count: llm_result.tool_turn_count,
                    risk_blocked: true,
                    order_executed: false,
                },
                Err(e) => error_log(agent_id, iteration, timestamp, "pipeline", &e.to_string()),
            }
        }
        None => {
            // Claude decided to do nothing (hold)
            let truncated_reasoning = &decision.reasoning[..decision.reasoning.len().min(200)];
            AgentLoopLog {
                agent_id: agent_id.to_string(),
                iteration,
                timestamp,
                phase: "complete".to_string(),
                success: true,
                message: "No trade signal -- hold".to_string(),
                decision_summary: Some(truncated_reasoning.to_string()),
                tool_turn_count: llm_result.tool_turn_count,
                risk_blocked: false,
                order_executed: false,
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Parse a composer profile string into a `ComposerProfile` enum.
pub fn parse_composer_profile(profile: &str) -> Option<ComposerProfile> {
    match profile.to_lowercase().as_str() {
        "conservative" => Some(ComposerProfile::Conservative),
        "all_weather" | "allweather" => Some(ComposerProfile::AllWeather),
        "turtle_system" | "turtlesystem" | "turtle" => Some(ComposerProfile::TurtleSystem),
        _ => None,
    }
}

fn extract_confidence_from_text(text: &str) -> Option<f64> {
    let lower = text.to_lowercase();
    for line in lower.lines() {
        let line = line.trim();
        if !line.contains("confidence") {
            continue;
        }
        for part in line.split_whitespace() {
            if let Ok(value) = part
                .trim_matches(|c: char| !c.is_ascii_digit() && c != '.')
                .parse::<f64>()
            {
                if (0.0..=1.0).contains(&value) {
                    return Some(value);
                }
            }
        }
    }
    None
}

// ---------------------------------------------------------------------------
// Thinking event emitter (used by Tauri)
// ---------------------------------------------------------------------------

type ThinkingEventEmitter = Arc<dyn Fn(String, String) + Send + Sync>;
static THINKING_EVENT_EMITTER: OnceLock<ThinkingEventEmitter> = OnceLock::new();

pub fn set_thinking_event_emitter(emitter: ThinkingEventEmitter) {
    let _ = THINKING_EVENT_EMITTER.set(emitter);
}

fn emit_thinking_event(agent_id: &str, chunk: &str) {
    if let Some(emitter) = THINKING_EVENT_EMITTER.get() {
        emitter(agent_id.to_string(), chunk.to_string());
    }
}

// ---------------------------------------------------------------------------
// Global agent history store (SQLite)
// ---------------------------------------------------------------------------

static HISTORY_STORE: OnceLock<std::sync::Mutex<AgentHistoryStore>> = OnceLock::new();

/// Initialise the global agent-history SQLite store.
///
/// Safe to call multiple times -- only the first call takes effect.
/// Typically called once at daemon startup.
pub fn init_history_store(db_path: &str) {
    let _ = HISTORY_STORE.get_or_init(|| {
        let store = AgentHistoryStore::new(db_path).expect("Failed to open agent history database");
        std::sync::Mutex::new(store)
    });
}

/// Insert a log entry into the global history store (if initialised).
fn persist_history(log: &AgentLoopLog) {
    if let Some(store) = HISTORY_STORE.get() {
        if let Ok(guard) = store.lock() {
            if let Err(e) = guard.insert(log) {
                tracing::warn!(error = %e, "Failed to persist agent history entry");
            }
        }
    }
}

/// Query the global history store. Returns an empty vec if not initialised.
pub fn query_history(agent_id: &str, limit: usize) -> Vec<AgentLoopLog> {
    HISTORY_STORE
        .get()
        .and_then(|store| store.lock().ok())
        .and_then(|guard| guard.list(agent_id, limit).ok())
        .unwrap_or_default()
}

// ---------------------------------------------------------------------------
// No-op channel for agent loop (used by call_claude_real)
// ---------------------------------------------------------------------------

struct LoopNoopChannel {
    agent_id: String,
}

#[async_trait::async_trait]
impl Channel for LoopNoopChannel {
    fn name(&self) -> &str {
        "agent-loop"
    }

    async fn listen(
        &self,
        _tx: tokio::sync::mpsc::Sender<IncomingEvent>,
    ) -> motosan_chat_core::Result<()> {
        Ok(())
    }

    async fn stream_chunk(&self, _user_id: &str, _chunk: &str) -> motosan_chat_core::Result<()> {
        emit_thinking_event(&self.agent_id, _chunk);
        Ok(())
    }

    async fn stream_done(&self, _user_id: &str) -> motosan_chat_core::Result<()> {
        Ok(())
    }

    async fn send(&self, _user_id: &str, _text: &str) -> motosan_chat_core::Result<()> {
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Real Claude API call
// ---------------------------------------------------------------------------

#[derive(Debug, Clone)]
struct LlmCallResult {
    decision: AgentDecision,
    tool_turn_count: u32,
}

/// Run motosan-chat-agent AgentLoop and convert output to AgentDecision.
async fn call_claude_real(
    agent_id: &str,
    system_prompt: &str,
    user_message: &str,
    max_retries: u32,
    max_tool_turns: u32,
) -> Result<LlmCallResult, String> {
    let cred_resolver = hyper_agent_core::credentials::CredentialResolver::new(
        hyper_agent_core::config::CredentialsSection::default(),
    );
    let api_key = cred_resolver
        .anthropic_key()
        .ok_or_else(|| "No Anthropic API key found (set ANTHROPIC_API_KEY or ANTHROPIC_OAUTH_TOKEN env var, or configure in TOML)".to_string())?;

    let retry_policy = motosan_ai::RetryPolicy::new()
        .max_retries(max_retries)
        .base_delay_ms(1_000)
        .max_delay_ms(30_000)
        .jitter(true)
        .respect_retry_after(true);

    let ai_client = motosan_ai::Client::builder()
        .provider(motosan_ai::Provider::Anthropic)
        .api_key(api_key)
        .model(crate::claude::ClaudeModel::Sonnet4.as_str())
        .retry_policy(retry_policy)
        .build()
        .map_err(|e| format!("Claude client build failed: {}", e))?;

    let llm_client = MotosanAiClient::new(ai_client).with_system(system_prompt);

    let risk_config = hyper_risk::risk::get_risk_config_sync();
    let account_state = AccountState::default();

    // Build a PositionManager for tools that need position data
    let tool_pm = {
        let mut p = dirs::data_dir().unwrap_or_else(|| PathBuf::from("."));
        p.push("hyper-agent");
        p.push("positions");
        let _ = std::fs::create_dir_all(&p);
        p.push(format!("{}.db", agent_id));
        Arc::new(
            PositionManager::new(p.to_str().unwrap_or("positions.db"))
                .map_err(|e| format!("Failed to open position DB for tools: {}", e))?,
        )
    };

    let loop_engine = MotosanAgentLoop::builder()
        .tool(GetPositionsTool::new(Arc::clone(&tool_pm)))
        .tool(PlaceOrderTool::new(
            RiskGuard::new(risk_config.clone()),
            account_state.clone(),
        ))
        .tool(CancelOrderTool::new(
            RiskGuard::new(risk_config),
            account_state,
        ))
        .tool(DoNothingTool)
        .tool(SetStopLossTool)
        .tool(SetTakeProfitTool)
        .tool(GetOpenOrdersTool::new(true))
        .tool(GetTradeHistoryTool::new(true))
        .max_iterations(max_tool_turns as usize)
        .build();

    let event = IncomingEvent {
        id: uuid::Uuid::new_v4().to_string(),
        platform: "hyper-agent".to_string(),
        user_id: "agent-loop".to_string(),
        channel_id: "agent-loop".to_string(),
        text: user_message.to_string(),
        reply_token: None,
        attachments: vec![],
        chat_type: ChatType::DirectMessage,
        kind: IncomingEventKind::Message,
    };
    let thread = Thread::new(
        event,
        Arc::new(LoopNoopChannel {
            agent_id: agent_id.to_string(),
        }),
    );

    let result = loop_engine
        .run(&llm_client, &thread, vec![AgentMessage::user(user_message)])
        .await
        .map_err(|e| format!("AgentLoop failed: {}", e))?;

    let tool_calls = result
        .tool_calls
        .into_iter()
        .map(|(tool_name, tool_input)| crate::claude::ToolCall {
            tool_name,
            tool_input,
        })
        .collect();

    Ok(LlmCallResult {
        decision: AgentDecision {
            reasoning: result.answer.clone(),
            tool_calls,
            confidence: extract_confidence_from_text(&result.answer).unwrap_or(0.5),
            raw_content: vec![],
        },
        tool_turn_count: result.iterations.saturating_sub(1) as u32,
    })
}

// ---------------------------------------------------------------------------
// Persistence helpers
// ---------------------------------------------------------------------------

fn loop_config_dir() -> PathBuf {
    let mut path = dirs::config_dir().unwrap_or_else(|| PathBuf::from("."));
    path.push("hyper-agent");
    path.push("agent-loops");
    let _ = fs::create_dir_all(&path);
    path
}

fn loop_config_path(agent_id: &str) -> PathBuf {
    loop_config_dir().join(format!("{}.json", agent_id))
}

fn save_loop_config(config: &LoopConfig) {
    let path = loop_config_path(&config.agent_id);
    if let Ok(json) = serde_json::to_string_pretty(config) {
        let _ = fs::write(&path, json);
    }
}

/// Load all persisted loop configs.
pub fn load_all_loop_configs() -> Vec<LoopConfig> {
    let dir = loop_config_dir();
    let entries = match fs::read_dir(&dir) {
        Ok(e) => e,
        Err(_) => return Vec::new(),
    };
    let mut configs = Vec::new();
    for entry in entries.flatten() {
        let path = entry.path();
        if path.extension().is_some_and(|ext| ext == "json") {
            if let Ok(data) = fs::read_to_string(&path) {
                if let Ok(config) = serde_json::from_str::<LoopConfig>(&data) {
                    configs.push(config);
                }
            }
        }
    }
    configs
}

/// Log directory for agent loop iterations.
fn log_dir() -> PathBuf {
    let mut path = dirs::data_dir().unwrap_or_else(|| PathBuf::from("."));
    path.push("hyper-agent");
    path.push("agent-logs");
    let _ = fs::create_dir_all(&path);
    path
}

fn log_file_path(agent_id: &str) -> PathBuf {
    log_dir().join(format!("{}.jsonl", agent_id))
}

/// Append a log entry to the agent's JSONL log file and persist to SQLite.
pub fn write_log_entry(log: &AgentLoopLog) {
    // JSONL file log (existing behaviour)
    let path = log_file_path(&log.agent_id);
    if let Ok(json) = serde_json::to_string(log) {
        if let Ok(mut file) = fs::OpenOptions::new().create(true).append(true).open(&path) {
            let _ = writeln!(file, "{}", json);
        }
    }
    // SQLite history store
    persist_history(log);
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

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

    #[test]
    fn test_loop_state_serialization() {
        let state = LoopState::Running;
        let json = serde_json::to_string(&state).unwrap();
        assert_eq!(json, "\"running\"");

        let state = LoopState::Paused;
        let json = serde_json::to_string(&state).unwrap();
        assert_eq!(json, "\"paused\"");

        let state = LoopState::Stopped;
        let json = serde_json::to_string(&state).unwrap();
        assert_eq!(json, "\"stopped\"");

        let state = LoopState::Error;
        let json = serde_json::to_string(&state).unwrap();
        assert_eq!(json, "\"error\"");
    }

    #[test]
    fn test_loop_state_deserialization() {
        let state: LoopState = serde_json::from_str("\"running\"").unwrap();
        assert_eq!(state, LoopState::Running);

        let state: LoopState = serde_json::from_str("\"paused\"").unwrap();
        assert_eq!(state, LoopState::Paused);

        let state: LoopState = serde_json::from_str("\"stopped\"").unwrap();
        assert_eq!(state, LoopState::Stopped);
    }

    #[test]
    fn test_loop_status_serialization() {
        let status = LoopStatus {
            agent_id: "agent-1".to_string(),
            state: LoopState::Running,
            interval_minutes: 15,
            iteration_count: 5,
            last_iteration_at: Some("2026-03-09T10:00:00Z".to_string()),
            last_error: None,
        };
        let json = serde_json::to_value(&status).unwrap();
        assert_eq!(json["agentId"], "agent-1");
        assert_eq!(json["state"], "running");
        assert_eq!(json["intervalMinutes"], 15);
        assert_eq!(json["iterationCount"], 5);
    }

    #[test]
    fn test_loop_config_serialization() {
        let config = LoopConfig {
            agent_id: "agent-1".to_string(),
            interval_minutes: 15,
            auto_start: true,
        };
        let json = serde_json::to_value(&config).unwrap();
        assert_eq!(json["agentId"], "agent-1");
        assert_eq!(json["intervalMinutes"], 15);
        assert_eq!(json["autoStart"], true);
    }

    #[test]
    fn test_agent_loop_log_serialization() {
        let log = AgentLoopLog {
            agent_id: "agent-1".to_string(),
            iteration: 3,
            timestamp: "2026-03-09T10:30:00Z".to_string(),
            phase: "complete".to_string(),
            success: true,
            message: "All good".to_string(),
            decision_summary: Some("Buy BTC".to_string()),
            tool_turn_count: 2,
            risk_blocked: false,
            order_executed: true,
        };
        let json = serde_json::to_value(&log).unwrap();
        assert_eq!(json["agentId"], "agent-1");
        assert_eq!(json["iteration"], 3);
        assert_eq!(json["phase"], "complete");
        assert_eq!(json["success"], true);
        assert_eq!(json["toolTurnCount"], 2);
        assert_eq!(json["riskBlocked"], false);
        assert_eq!(json["orderExecuted"], true);
    }

    #[test]
    fn test_agent_loop_manager_creation() {
        let _manager = AgentLoopManager::new();
    }

    #[tokio::test]
    async fn test_start_and_stop_loop() {
        let manager = AgentLoopManager::new();
        let result = manager.start_loop("test-agent".to_string(), 1, None).await;
        assert!(result.is_ok());

        let status = manager.get_status("test-agent").await.unwrap();
        assert_eq!(status.state, LoopState::Running);
        assert_eq!(status.interval_minutes, 1);

        let result = manager.stop_loop("test-agent").await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_start_duplicate_loop_fails() {
        let manager = AgentLoopManager::new();
        manager
            .start_loop("dup-agent".to_string(), 1, None)
            .await
            .unwrap();

        let result = manager.start_loop("dup-agent".to_string(), 1, None).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("already"));

        manager.stop_loop("dup-agent").await.unwrap();
    }

    #[tokio::test]
    async fn test_stop_nonexistent_loop_fails() {
        let manager = AgentLoopManager::new();
        let result = manager.stop_loop("nonexistent").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_pause_and_resume_loop() {
        let manager = AgentLoopManager::new();
        manager
            .start_loop("pr-agent".to_string(), 60, None)
            .await
            .unwrap();

        let result = manager.pause_loop("pr-agent").await;
        assert!(result.is_ok());

        let status = manager.get_status("pr-agent").await.unwrap();
        assert_eq!(status.state, LoopState::Paused);

        let result = manager.resume_loop("pr-agent").await;
        assert!(result.is_ok());

        let status = manager.get_status("pr-agent").await.unwrap();
        assert_eq!(status.state, LoopState::Running);

        manager.stop_loop("pr-agent").await.unwrap();
    }

    #[tokio::test]
    async fn test_get_status_nonexistent_returns_stopped() {
        let manager = AgentLoopManager::new();
        let status = manager.get_status("no-such-agent").await.unwrap();
        assert_eq!(status.state, LoopState::Stopped);
    }

    #[test]
    fn test_parse_composer_profile() {
        assert!(matches!(
            parse_composer_profile("conservative"),
            Some(ComposerProfile::Conservative)
        ));
        assert!(matches!(
            parse_composer_profile("all_weather"),
            Some(ComposerProfile::AllWeather)
        ));
        assert!(matches!(
            parse_composer_profile("turtle_system"),
            Some(ComposerProfile::TurtleSystem)
        ));
        assert!(matches!(
            parse_composer_profile("turtle"),
            Some(ComposerProfile::TurtleSystem)
        ));
        assert!(parse_composer_profile("unknown").is_none());
    }
}