schwab_cli/commands/

ui_cmd.rs

use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use anyhow::{Context, Result};

use crate::agent::{daemon_status, run_agent_loop};
use crate::cli::RulesCommands;
use crate::config::RuntimeConfig;
use crate::rules::RulesConfig;
use crate::safety::require_trading_approval;
use crate::ui::agent_health::{new_shared_health, update_health, SharedAgentHealth};
use crate::ui::discover::{list_rules_files_display, resolve_rules_file};
use crate::ui::market_status::{self, MarketSnapshot};
use crate::ui::menu::{list_rules_files, show_dashboard, show_rules};
use crate::ui::spread_feed::{new_spread_snapshot, spawn_spread_mark_feed};
use crate::ui::watch::{run_watch_tui, WatchAgentMode, WatchConfig};
use crate::market_conditions::{spawn_market_conditions_feed, MarketConditionsSnapshot};

pub async fn run_dashboard(runtime: &RuntimeConfig, file: Option<PathBuf>) -> Result<()> {
    show_dashboard(runtime, file).await
}

pub async fn run_watch(
    runtime: &RuntimeConfig,
    file: Option<PathBuf>,
    monitor_only: bool,
) -> Result<()> {
    if runtime.output == crate::output::OutputFormat::Json {
        anyhow::bail!("watch is interactive only — omit --json");
    }

    let rules_path = resolve_rules_file(file, runtime.is_interactive())?;
    let daemon = daemon_status(&rules_path);

    let agent_mode = if daemon.running {
        WatchAgentMode::External
    } else if monitor_only {
        WatchAgentMode::MonitorOnly
    } else {
        WatchAgentMode::Embedded
    };

    let will_spawn_agent = matches!(agent_mode, WatchAgentMode::Embedded);

    if will_spawn_agent && !runtime.dry_run && !runtime.simulate {
        let agent_id = RulesConfig::load(&rules_path)
            .map(|r| r.agent_id)
            .unwrap_or_else(|_| "agent".into());
        require_trading_approval(
            runtime,
            "watch",
            &format!("Run and watch options agent `{agent_id}`"),
        )?;
    }

    let agent_runtime = RuntimeConfig::for_agent_trading(
        runtime.output,
        runtime.yes,
        runtime.dry_run,
        runtime.simulate,
        runtime.trust,
        true,
        runtime.no_audio,
        runtime.safety.clone(),
        runtime.sink.clone(),
    );

    let market_snapshot = Arc::new(Mutex::new(MarketSnapshot::default()));
    market_status::refresh_market_snapshot(runtime, &market_snapshot).await;
    let snapshot_for_refresh = market_snapshot.clone();
    let refresh_runtime = runtime.clone();
    let _market_refresh = tokio::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(120));
        interval.tick().await;
        loop {
            interval.tick().await;
            market_status::refresh_market_snapshot(&refresh_runtime, &snapshot_for_refresh).await;
        }
    });

    let market_conditions = Arc::new(Mutex::new(MarketConditionsSnapshot::default()));
    if let Ok(market_api) = runtime.build_market_api() {
        if let Ok(mut guard) = market_conditions.lock() {
            crate::market_conditions::refresh_market_conditions(&market_api, &mut guard).await;
        }
        let _conditions_feed =
            spawn_market_conditions_feed(market_api, market_conditions.clone());
    }

    let agent_health: Option<SharedAgentHealth> = if will_spawn_agent {
        Some(new_shared_health())
    } else {
        None
    };

    let agent_handle = if will_spawn_agent {
        let path = rules_path.clone();
        let health = agent_health.clone().expect("health when spawning");
        Some(tokio::spawn(async move {
            supervise_options_agent(agent_runtime, path, health).await;
        }))
    } else {
        None
    };

    // Brief pause so the first tick can start before the TUI loads.
    if will_spawn_agent {
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    }

    let spread_snapshot = new_spread_snapshot();
    if let (Ok(market), Ok(trader)) = (
        runtime.build_market_api(),
        runtime.build_api(),
    ) {
        let _spread_feed = spawn_spread_mark_feed(
            rules_path.clone(),
            runtime.simulate,
            spread_snapshot.clone(),
            market,
            trader,
        );
    }

    let watch_config = WatchConfig {
        rules_path: rules_path.clone(),
        agent_mode,
        market_snapshot,
        market_conditions,
        agent_health,
        spread_snapshot,
    };

    let watch_result = tokio::task::spawn_blocking(move || run_watch_tui(&watch_config))
        .await
        .context("watch UI thread panicked")?;

    if let Some(handle) = agent_handle {
        handle.abort();
        let _ = handle.await;
    }

    watch_result
}

/// Belt-and-suspenders: if `run_agent_loop` ever returns, restart with backoff.
async fn supervise_options_agent(
    rt: RuntimeConfig,
    path: PathBuf,
    health: SharedAgentHealth,
) {
    let mut restart_backoff_secs: u64 = 5;
    loop {
        update_health(&health, |g| {
            g.loop_running = true;
        });
        let result = run_agent_loop(&rt, &path, false, Some(health.clone())).await;
        match result {
            Ok(()) => {
                update_health(&health, |g| g.record_loop_stopped(None));
                let _ = crate::agent::paths::append_agent_log(
                    &path,
                    "agent supervisor: loop returned Ok — stopping",
                );
                break;
            }
            Err(e) => {
                let msg = format!("agent loop exited: {e:#}");
                let _ = crate::agent::paths::append_agent_log(&path, &msg);
                update_health(&health, |g| {
                    g.record_supervisor_restart();
                    g.record_loop_stopped(Some(msg));
                });
                let restarts = health.lock().map(|g| g.restart_count).unwrap_or(0);
                let _ = crate::agent::paths::append_agent_log(
                    &path,
                    &format!(
                        "agent supervisor: restarting in {restart_backoff_secs}s (restart #{restarts})"
                    ),
                );
                tokio::time::sleep(Duration::from_secs(restart_backoff_secs)).await;
                restart_backoff_secs = (restart_backoff_secs.saturating_mul(2)).min(300);
                update_health(&health, |g| {
                    g.loop_running = true;
                    g.healthy = false;
                });
            }
        }
    }
}

pub async fn run_rules(runtime: &RuntimeConfig, command: RulesCommands) -> Result<()> {
    match command {
        RulesCommands::Show { file } => show_rules(runtime, file).await,
        RulesCommands::List => {
            if runtime.output == crate::output::OutputFormat::Json {
                list_rules_files(runtime);
            } else {
                println!("\n{}", list_rules_files_display());
                println!();
            }
            Ok(())
        }
    }
}