//! Process-wide singleton: cached policy + cooperative throttling.
//!
//! One sampler task refreshes [`Signals`] every 30s and recomputes the
//! [`Policy`]. Workers call [`current_policy`] for cheap reads or
//! [`wait_for_capacity`] to cooperatively block until the host is ready.

use std::sync::{Arc, OnceLock};
use std::time::Duration;

use parking_lot::RwLock;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

use crate::openhuman::config::{Config, SchedulerGateConfig};
use crate::openhuman::scheduler_gate::policy::{decide, Policy};
use crate::openhuman::scheduler_gate::signals::Signals;

/// Process-wide ceiling on concurrent LLM-bound work.
///
/// Held at 1 to keep concurrent local-Ollama / bge-m3 calls (8K context,
/// ~1.3 GB resident each) from saturating local RAM. See
/// `feedback_local_llm_load.md` — backfills with multiple
/// simultaneous Ollama requests have crashed the user's laptop twice.
///
/// Cloud-backend LLM calls bypass this semaphore at the worker layer
/// (see `memory_tree::jobs::worker::run_once`) because they're
/// bandwidth-bound, not RAM-bound, and the worker pool itself bounds
/// concurrency upstream. Keeping this at 1 preserves the laptop-RAM
/// contract regardless of backend.
const LLM_SLOTS: usize = 1;

static LLM_PERMITS: OnceLock<Arc<Semaphore>> = OnceLock::new();

fn llm_permits() -> &'static Arc<Semaphore> {
    LLM_PERMITS.get_or_init(|| Arc::new(Semaphore::new(LLM_SLOTS)))
}

/// RAII guard returned by [`wait_for_capacity`] / [`acquire_llm_permit`].
///
/// While the caller holds an `LlmPermit`, no other LLM-bound caller in
/// the process can acquire one (the global semaphore has a single slot).
/// Drop the permit as soon as the LLM request returns — holding it past
/// post-processing serialises unrelated work for no reason.
///
/// This type is intentionally opaque: callers can't reach into the
/// underlying [`OwnedSemaphorePermit`] and risk forgetting to release it.
#[must_use = "drop the LlmPermit only after the LLM call returns"]
pub struct LlmPermit {
    _permit: OwnedSemaphorePermit,
}

impl Drop for LlmPermit {
    fn drop(&mut self) {
        log::trace!("[scheduler_gate] llm permit released");
    }
}

struct State {
    cfg: SchedulerGateConfig,
    signals: Signals,
    policy: Policy,
}

static STATE: OnceLock<Arc<RwLock<State>>> = OnceLock::new();
static STARTED: std::sync::Once = std::sync::Once::new();

const SAMPLE_INTERVAL: Duration = Duration::from_secs(30);

/// Initialise the gate and spawn the background sampler.
///
/// Idempotent — repeat calls during bootstrap are no-ops. Subsequent config
/// reloads should call [`update_config`] instead.
pub fn init_global(config: &Config) {
    let cfg = config.scheduler_gate.clone();
    STARTED.call_once(|| {
        let signals = Signals::sample();
        let policy = decide(&signals, &cfg);
        log::info!(
            "[scheduler_gate] startup policy={} mode={} on_ac={} charge={:?} cpu={:.1}% server={}",
            policy.as_str(),
            cfg.mode.as_str(),
            signals.on_ac_power,
            signals.battery_charge,
            signals.cpu_usage_pct,
            signals.server_mode,
        );
        let state = Arc::new(RwLock::new(State {
            cfg,
            signals,
            policy,
        }));
        let _ = STATE.set(state.clone());

        tokio::spawn(async move {
            loop {
                tokio::time::sleep(SAMPLE_INTERVAL).await;
                // Sampling does a brief blocking sleep + sysinfo refresh —
                // push it off the async runtime.
                let signals = match tokio::task::spawn_blocking(Signals::sample).await {
                    Ok(s) => s,
                    Err(err) => {
                        log::warn!("[scheduler_gate] sampler join error: {err:#}");
                        continue;
                    }
                };
                let mut guard = state.write();
                let next = decide(&signals, &guard.cfg);
                if next != guard.policy {
                    log::info!(
                        "[scheduler_gate] policy {} -> {} (on_ac={} charge={:?} cpu={:.1}% server={})",
                        guard.policy.as_str(),
                        next.as_str(),
                        signals.on_ac_power,
                        signals.battery_charge,
                        signals.cpu_usage_pct,
                        signals.server_mode,
                    );
                }
                guard.signals = signals;
                guard.policy = next;
            }
        });
    });
}

/// Update the gate's view of user config (e.g. after a settings change).
pub fn update_config(cfg: SchedulerGateConfig) {
    if let Some(state) = STATE.get() {
        let mut guard = state.write();
        guard.cfg = cfg;
        guard.policy = decide(&guard.signals, &guard.cfg);
    }
}

/// Current policy. Defaults to [`Policy::Normal`] before [`init_global`] runs
/// (e.g. in unit tests) so callers don't deadlock waiting on a sampler that
/// will never start.
pub fn current_policy() -> Policy {
    STATE
        .get()
        .map(|s| s.read().policy)
        .unwrap_or(Policy::Normal)
}

/// Most recent sampled signals, or a neutral default if the sampler hasn't run.
pub fn current_signals() -> Signals {
    STATE.get().map(|s| s.read().signals).unwrap_or(Signals {
        on_ac_power: true,
        battery_charge: None,
        cpu_usage_pct: 0.0,
        server_mode: false,
    })
}

/// Cooperatively block a caller until the host is ready for LLM-bound
/// work, then hand back an [`LlmPermit`] that holds a slot in the global
/// LLM semaphore.
///
/// Policy-driven backoff happens **before** semaphore acquisition so a
/// `Paused` mode doesn't pile up tasks queued for the slot — they sit
/// in the pause-poll loop, not in the semaphore wait queue.
///
/// * **Aggressive / Normal** — wait for the global slot; return immediately
///   once granted.
/// * **Throttled** — sleep `throttled_backoff_ms` first so concurrent
///   workers serialise themselves, then acquire the slot.
/// * **Paused** — poll every `paused_poll_ms` until the policy changes,
///   then acquire the slot.
///
/// Drop the returned [`LlmPermit`] as soon as the LLM call returns.
///
/// Returns `None` only if the global LLM semaphore has been closed
/// (never happens in production — the semaphore lives for the lifetime
/// of the process). Callers can safely treat `None` as "skip the
/// gate" rather than propagating an error.
pub async fn wait_for_capacity() -> Option<LlmPermit> {
    loop {
        let (policy, throttled_ms, paused_ms) = match STATE.get() {
            Some(state) => {
                let g = state.read();
                (g.policy, g.cfg.throttled_backoff_ms, g.cfg.paused_poll_ms)
            }
            None => {
                // Gate not initialised (unit tests, early bootstrap).
                // Acquire directly — no policy to consult.
                return acquire_llm_permit_inner().await;
            }
        };
        match policy {
            Policy::Aggressive | Policy::Normal => {
                return acquire_llm_permit_inner().await;
            }
            Policy::Throttled => {
                log::trace!(
                    "[scheduler_gate] throttled — sleeping {throttled_ms}ms before permit acquire"
                );
                tokio::time::sleep(Duration::from_millis(throttled_ms)).await;
                return acquire_llm_permit_inner().await;
            }
            Policy::Paused { reason } => {
                log::debug!(
                    "[scheduler_gate] paused ({}); polling every {paused_ms}ms",
                    reason.as_str()
                );
                tokio::time::sleep(Duration::from_millis(paused_ms)).await;
                // re-evaluate; user may have toggled the gate back on.
            }
        }
    }
}

async fn acquire_llm_permit_inner() -> Option<LlmPermit> {
    let sem = llm_permits().clone();
    match sem.acquire_owned().await {
        Ok(permit) => {
            log::trace!("[scheduler_gate] llm permit acquired");
            Some(LlmPermit { _permit: permit })
        }
        Err(_) => {
            // Semaphore closed — should never happen since we never
            // close it. Log loudly and let the caller proceed without
            // a permit so the pipeline doesn't deadlock.
            log::warn!(
                "[scheduler_gate] llm semaphore closed unexpectedly — proceeding without a permit"
            );
            None
        }
    }
}

/// Test/diagnostic hook: try to grab a permit without consulting the
/// gate policy. Returns `None` if no slots are free. **Do not** call
/// from production code — production callers should use
/// [`wait_for_capacity`] so the policy backoff applies.
#[cfg(test)]
pub fn try_acquire_llm_permit() -> Option<LlmPermit> {
    let sem = llm_permits().clone();
    sem.try_acquire_owned()
        .ok()
        .map(|p| LlmPermit { _permit: p })
}

/// Number of permits currently available. Test-only diagnostic.
#[cfg(test)]
pub fn available_llm_permits() -> usize {
    llm_permits().available_permits()
}

#[cfg(test)]
mod tests {
    //! These tests share the **process-wide** `LLM_PERMITS` semaphore
    //! (which is intentional — that's what they're testing). They are
    //! serialised via a module-local mutex so two test threads can't
    //! both hold a permit at the same time and confuse each other's
    //! `available_permits` reads.
    use super::*;
    use std::sync::Mutex;
    use std::time::Instant;
    use tokio::time::{timeout, Duration as TokioDuration};

    static GATE_TEST_LOCK: Mutex<()> = Mutex::new(());

    fn lock() -> std::sync::MutexGuard<'static, ()> {
        // Tolerate poisoning so a panicking test doesn't block the rest.
        GATE_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner())
    }

    #[tokio::test]
    async fn wait_for_capacity_returns_permit_when_gate_uninit() {
        let _g = lock();
        let permit = wait_for_capacity().await;
        assert!(
            permit.is_some(),
            "uninit gate must still hand back a permit"
        );
        assert_eq!(
            available_llm_permits(),
            0,
            "permit must occupy the single LLM slot"
        );
        drop(permit);
        assert_eq!(available_llm_permits(), 1, "drop must release the slot");
    }

    #[tokio::test]
    async fn second_waiter_blocks_until_first_drops() {
        let _g = lock();
        let first = wait_for_capacity().await.expect("first permit");
        assert_eq!(available_llm_permits(), 0);

        // Spawn a second acquirer; it must block.
        let handle = tokio::spawn(async move {
            let started = Instant::now();
            let p = wait_for_capacity().await;
            (started.elapsed(), p)
        });

        // Give the second waiter a moment to start polling.
        tokio::time::sleep(TokioDuration::from_millis(40)).await;
        assert!(!handle.is_finished(), "second waiter must be blocked");

        // Release the first permit; the second should resolve.
        drop(first);
        let (elapsed, second) = timeout(TokioDuration::from_secs(1), handle)
            .await
            .unwrap()
            .unwrap();
        assert!(
            second.is_some(),
            "second waiter must eventually get a permit"
        );
        assert!(
            elapsed >= TokioDuration::from_millis(20),
            "second waiter should have actually waited (got {elapsed:?})"
        );
        drop(second);
    }

    #[tokio::test]
    async fn semaphore_size_is_one() {
        let _g = lock();
        let p1 = wait_for_capacity().await.expect("first permit");
        // Try-acquire must fail while the slot is held.
        assert!(
            try_acquire_llm_permit().is_none(),
            "semaphore must be size-1 — second try_acquire should fail"
        );
        drop(p1);
        // Now another should succeed.
        let p2 = try_acquire_llm_permit().expect("permit free after drop");
        drop(p2);
    }
}
